bridge.go raw
1 // Package bridge is the gateway side of the relay tunnel: it accepts an
2 // encrypted stream from a light client and connects it to a chain node's P2P
3 // port, forwarding only the message types a compact-filter client needs.
4 //
5 // Bridging P2P rather than exposing JSON-RPC is deliberate. The P2P protocol
6 // has no wallet surface, so there is nothing to spend and nothing to list; the
7 // whitelist keeps the stream to chain data, and the client verifies the header
8 // and filter chains itself, so a bridge can withhold but not forge.
9 package bridge
10
11 import (
12 "context"
13 "errors"
14 "fmt"
15 "net"
16 "net/url"
17 "strings"
18 "sync"
19 "time"
20
21 "git.smesh.lol/nostrino/pkg/btcwire"
22 "git.smesh.lol/nostrino/pkg/chain"
23 "git.smesh.lol/nostrino/pkg/nostr"
24 "git.smesh.lol/nostrino/pkg/p2p"
25 "git.smesh.lol/nostrino/pkg/registry"
26 "git.smesh.lol/nostrino/pkg/relay"
27 "git.smesh.lol/nostrino/pkg/tunnel"
28 "git.smesh.lol/nostrino/pkg/wsserver"
29 )
30
31 // Config configures a bridge.
32 type Config struct {
33 // Relays are the public relays the gateway announces on.
34 Relays []string
35 // NodeAddr is the chain node's P2P address, for example 127.0.0.1:18444.
36 NodeAddr string
37 // Key is the gateway's persistent transport key.
38 Key [32]byte
39 // Profile is the chain served. Magic is taken from it.
40 Profile *chain.Profile
41 // Chains is the chain name advertised to clients. Defaults to
42 // Profile.Name.
43 Chains []string
44 // Allowlist is an optional set of client pubkeys. Empty serves anyone.
45 Allowlist []string
46 // MaxSessions bounds concurrent streams. Zero means 32.
47 MaxSessions int
48 // NodeDialTimeout bounds the node connection. Zero means 10s.
49 NodeDialTimeout time.Duration
50 // IdleTimeout closes a stream with no traffic. Zero means 2 minutes.
51 IdleTimeout time.Duration
52 // MaxBytesPerSec bounds each direction of a stream. Zero means unlimited;
53 // the command defaults to 4 MiB/s.
54 MaxBytesPerSec int64
55 // MaxSessionsPerClient bounds one client pubkey's concurrent streams.
56 // Zero means 2.
57 MaxSessionsPerClient int
58 // Presence publishes a discoverable gateway announcement. Zero value
59 // disables it; the command enables it explicitly.
60 Presence bool
61 // PresenceInterval is the announcement period. Zero means 5 minutes.
62 PresenceInterval time.Duration
63 // Version is reported in presence and diagnostics.
64 Version string
65 // BridgeListen enables the WebSocket data plane on this TCP address.
66 // Empty disables it.
67 BridgeListen string
68 // BridgePath is the WebSocket request path. Empty means /nostrino.
69 BridgePath string
70 // BridgeURL is the advertised bridge URL. Empty derives it from
71 // BridgeListen and BridgePath; set it to a wss:// URL behind a proxy.
72 BridgeURL string
73 // BridgeToken, when set, is required as the token query parameter or as
74 // a bearer Authorization header on the WebSocket upgrade.
75 BridgeToken string
76 // AllowCommands adds to the default BIP157 whitelist.
77 AllowCommands []string
78 // Logf receives diagnostics; nil discards them.
79 Logf func(string, ...any)
80 // TransportFactory overrides the relay WebSocket dialer, which is how
81 // tests inject an in-process relay.
82 TransportFactory func(url string) (relay.Transport, error)
83 }
84
85 // Server is a running bridge.
86 type Server struct {
87 cfg Config
88 listener *tunnel.Listener
89 ws *wsserver.Server
90
91 mu sync.Mutex
92 conns map[net.Conn]struct{}
93 wsActive int
94 wsPerIP map[string]int
95 done chan struct{}
96 wg sync.WaitGroup
97 }
98
99 // MaxBlockPayload is the largest block this gateway will forward.
100 //
101 // The client has the same cap, and it belongs on both ends: a cap that lives only in the client
102 // protects the client from the node, and this one protects the gateway from a node that answers
103 // a getdata with more than anybody asked for.
104 const MaxBlockPayload = 8 << 20
105
106 // DefaultCommands is the P2P whitelist. Everything a compact-filter client
107 // needs is here; address discovery, bloom filters, mempool queries and BIP152
108 // block relay are not.
109 var DefaultCommands = []string{
110 btcwire.CmdVersion,
111 btcwire.CmdVerAck,
112 btcwire.CmdPing,
113 btcwire.CmdPong,
114 btcwire.CmdSendHeaders,
115 btcwire.CmdWtxidRelay,
116 btcwire.CmdSendAddrV2,
117 btcwire.CmdFeeFilter,
118 btcwire.CmdGetHeaders,
119 btcwire.CmdHeaders,
120 btcwire.CmdGetCFHeaders,
121 btcwire.CmdCFHeaders,
122 btcwire.CmdGetCFilters,
123 btcwire.CmdCFilter,
124 btcwire.CmdGetCFCheckpt,
125 btcwire.CmdCFCheckpt,
126 btcwire.CmdGetData,
127 btcwire.CmdTx,
128 btcwire.CmdInv,
129 btcwire.CmdNotFound,
130 // Core announces compact blocks straight after the handshake whether or not the
131 // client asked for BIP152. Dropping it closes every session 100ms in.
132 btcwire.CmdSendCmpct,
133 }
134
135 // Start connects the gateway to its relays and returns a running server.
136 func Start(ctx context.Context, cfg Config) (*Server, error) {
137 if cfg.Profile == nil {
138 return nil, errors.New("bridge: nil profile")
139 }
140 if err := cfg.Profile.Validate(); err != nil {
141 return nil, err
142 }
143 if cfg.NodeAddr == "" {
144 return nil, errors.New("bridge: no node address")
145 }
146 if cfg.Key == ([32]byte{}) {
147 return nil, errors.New("bridge: no gateway key")
148 }
149 if len(cfg.Relays) == 0 {
150 return nil, errors.New("bridge: no relay")
151 }
152 if cfg.MaxSessions == 0 {
153 cfg.MaxSessions = 32
154 }
155 if cfg.NodeDialTimeout == 0 {
156 cfg.NodeDialTimeout = 10 * time.Second
157 }
158 if cfg.IdleTimeout == 0 {
159 cfg.IdleTimeout = 2 * time.Minute
160 }
161 if cfg.MaxSessionsPerClient == 0 {
162 cfg.MaxSessionsPerClient = 2
163 }
164 if cfg.PresenceInterval == 0 {
165 cfg.PresenceInterval = 5 * time.Minute
166 }
167 if cfg.Version == "" {
168 cfg.Version = "nostrino"
169 }
170 if cfg.Logf == nil {
171 cfg.Logf = func(string, ...any) {}
172 }
173 chains := cfg.Chains
174 if len(chains) == 0 {
175 chains = []string{cfg.Profile.Name}
176 }
177 l, err := tunnel.Listen(ctx, tunnel.ListenConfig{
178 Relays: cfg.Relays,
179 Key: cfg.Key,
180 Chains: chains,
181 Allowlist: cfg.Allowlist,
182 MaxSessions: cfg.MaxSessions,
183 IdleTimeout: cfg.IdleTimeout,
184 Logf: cfg.Logf,
185 TransportFactory: cfg.TransportFactory,
186 })
187 if err != nil {
188 return nil, err
189 }
190 cfg.Chains = chains
191 s := &Server{
192 cfg: cfg,
193 listener: l,
194 conns: make(map[net.Conn]struct{}),
195 wsPerIP: make(map[string]int),
196 done: make(chan struct{}),
197 }
198 if cfg.BridgeListen != "" {
199 if cfg.BridgePath == "" {
200 cfg.BridgePath = "/nostrino"
201 }
202 ws, err := wsserver.Start(wsserver.Config{
203 Listen: cfg.BridgeListen,
204 Path: cfg.BridgePath,
205 Handler: s.handleWS,
206 Logf: cfg.Logf,
207 })
208 if err != nil {
209 l.Close()
210 return nil, err
211 }
212 s.ws = ws
213 if cfg.BridgeURL == "" {
214 cfg.BridgeURL = "ws://" + ws.Addr() + cfg.BridgePath
215 }
216 if cfg.BridgeToken != "" {
217 sep := "?"
218 if strings.Contains(cfg.BridgeURL, "?") {
219 sep = "&"
220 }
221 cfg.BridgeURL += sep + "token=" + url.QueryEscape(cfg.BridgeToken)
222 }
223 }
224 s.cfg = cfg
225 if cfg.Presence {
226 s.wg.Add(1)
227 go s.presenceLoop(ctx)
228 }
229 return s, nil
230 }
231
232 // PubKey is the gateway transport public key.
233 func (s *Server) PubKey() [32]byte { return s.listener.PubKey() }
234
235 // URI returns the connection URI a client should use.
236 func (s *Server) URI() *tunnel.URI {
237 u := &tunnel.URI{
238 Gateway: s.listener.PubKey(),
239 Relays: s.cfg.Relays,
240 Chains: s.cfg.Chains,
241 Version: 1,
242 }
243 if s.cfg.BridgeURL != "" {
244 u.Bridges = []string{s.cfg.BridgeURL}
245 }
246 return u
247 }
248
249 // BridgeAddr is the bound WebSocket address, or empty when the data plane is
250 // disabled.
251 func (s *Server) BridgeAddr() string {
252 if s.ws == nil {
253 return ""
254 }
255 return s.ws.Addr()
256 }
257
258 func (s *Server) handleWS(c *wsserver.Conn, req *wsserver.Request) {
259 if s.cfg.BridgeToken != "" {
260 token := req.Query.Get("token")
261 if token == "" {
262 auth := req.Header["authorization"]
263 if strings.HasPrefix(auth, "Bearer ") {
264 token = strings.TrimPrefix(auth, "Bearer ")
265 }
266 }
267 if token != s.cfg.BridgeToken {
268 s.cfg.Logf("bridge: websocket client %s rejected: bad token", c.RemoteAddr())
269 return
270 }
271 }
272 chainName := req.Query.Get("chain")
273 if chainName == "" && len(s.cfg.Chains) > 0 {
274 chainName = s.cfg.Chains[0]
275 }
276 allowed := false
277 for _, name := range s.cfg.Chains {
278 if name == chainName {
279 allowed = true
280 break
281 }
282 }
283 if !allowed {
284 s.cfg.Logf("bridge: websocket client asked for chain %q", chainName)
285 return
286 }
287 ip := c.RemoteAddr().String()
288 if host, _, err := net.SplitHostPort(ip); err == nil {
289 ip = host
290 }
291 s.mu.Lock()
292 if s.wsActive >= s.cfg.MaxSessions {
293 s.mu.Unlock()
294 s.cfg.Logf("bridge: websocket session limit reached")
295 return
296 }
297 if s.wsPerIP[ip] >= s.cfg.MaxSessionsPerClient {
298 s.mu.Unlock()
299 s.cfg.Logf("bridge: websocket per-client limit for %s", ip)
300 return
301 }
302 s.wsActive++
303 s.wsPerIP[ip]++
304 s.mu.Unlock()
305 defer func() {
306 s.mu.Lock()
307 s.wsActive--
308 if s.wsPerIP[ip] > 0 {
309 s.wsPerIP[ip]--
310 }
311 if s.wsPerIP[ip] == 0 {
312 delete(s.wsPerIP, ip)
313 }
314 s.mu.Unlock()
315 }()
316 s.handleConn(context.Background(), c, chainName, "ws:"+c.RemoteAddr().String())
317 }
318
319 // Run accepts streams until ctx ends or the server is closed.
320 func (s *Server) Run(ctx context.Context) error {
321 for {
322 conn, err := s.listener.Accept(ctx)
323 if err != nil {
324 select {
325 case <-s.done:
326 return nil
327 case <-ctx.Done():
328 return ctx.Err()
329 default:
330 return err
331 }
332 }
333 s.wg.Add(1)
334 go func() {
335 defer s.wg.Done()
336 s.handle(ctx, conn)
337 }()
338 }
339 }
340
341 func (s *Server) handle(ctx context.Context, c *tunnel.Conn) {
342 defer c.Close()
343 s.handleConn(ctx, c, c.Chain(), "nostr:"+c.PeerHex())
344 }
345
346 func (s *Server) handleConn(ctx context.Context, c net.Conn, chainName, peer string) {
347 allowedChain := false
348 for _, name := range s.cfg.Chains {
349 if name == chainName {
350 allowedChain = true
351 break
352 }
353 }
354 if !allowedChain {
355 s.cfg.Logf("bridge: client %s asked for chain %q", peer, chainName)
356 return
357 }
358
359 dialCtx, cancel := context.WithTimeout(ctx, s.cfg.NodeDialTimeout)
360 defer cancel()
361 var d net.Dialer
362 nodeConn, err := d.DialContext(dialCtx, "tcp", s.cfg.NodeAddr)
363 if err != nil {
364 s.cfg.Logf("bridge: node dial: %v", err)
365 return
366 }
367 defer nodeConn.Close()
368 s.track(nodeConn, true)
369 defer s.track(nodeConn, false)
370
371 s.cfg.Logf("bridge: session %s client=%s chain=%s node=%s", c.LocalAddr(), peer, chainName, s.cfg.NodeAddr)
372 start := time.Now()
373 var bytesUp, bytesDown int64
374 upLimiter := newLimiter(s.cfg.MaxBytesPerSec)
375 downLimiter := newLimiter(s.cfg.MaxBytesPerSec)
376
377 errc := make(chan error, 2)
378 go func() { errc <- s.pump(c, nodeConn, &bytesUp, upLimiter) }()
379 go func() { errc <- s.pump2(c, nodeConn, &bytesDown, downLimiter) }()
380 // The first error is the one that ended the session; the second is the refusal that
381 // follows the close. Reporting only the counters left a dead tunnel with no reason.
382 first := <-errc
383 c.Close()
384 nodeConn.Close()
385 <-errc
386 s.cfg.Logf("bridge: session %s closed after %s up=%d down=%d: %v",
387 c.LocalAddr(), time.Since(start).Round(time.Millisecond), bytesUp, bytesDown, first)
388 }
389
390 func (s *Server) track(c net.Conn, add bool) {
391 s.mu.Lock()
392 defer s.mu.Unlock()
393 if add {
394 s.conns[c] = struct{}{}
395 } else {
396 delete(s.conns, c)
397 }
398 }
399
400 // pump forwards tunnel -> node.
401 func (s *Server) pump(tunnelConn net.Conn, node net.Conn, bytes *int64, limit *limiter) error {
402 return s.copyFrames(tunnelConn, node, bytes, limit, "client")
403 }
404
405 // pump2 forwards node -> tunnel.
406 func (s *Server) pump2(tunnelConn net.Conn, node net.Conn, bytes *int64, limit *limiter) error {
407 return s.copyFrames(node, tunnelConn, bytes, limit, "node")
408 }
409
410 func (s *Server) copyFrames(src, dst net.Conn, bytes *int64, limit *limiter, from string) error {
411 reader := btcwire.NewReader(src, s.cfg.Profile.Magic)
412 for {
413 if s.cfg.IdleTimeout > 0 {
414 if d, ok := src.(interface{ SetReadDeadline(time.Time) error }); ok {
415 _ = d.SetReadDeadline(time.Now().Add(s.cfg.IdleTimeout))
416 }
417 }
418 msg, err := reader.Next()
419 if err != nil {
420 return err
421 }
422 if !s.allowed(msg.Command) {
423 return fmt.Errorf("bridge: %s sent disallowed command %q", from, msg.Command)
424 }
425 if msg.Command == btcwire.CmdBlock && len(msg.Payload) > MaxBlockPayload {
426 return fmt.Errorf("bridge: %s sent a %d-byte block, the cap is %d", from, len(msg.Payload), MaxBlockPayload)
427 }
428 frame, err := btcwire.EncodeMessage(s.cfg.Profile.Magic, msg.Command, msg.Payload)
429 if err != nil {
430 return err
431 }
432 limit.wait(int64(len(frame)))
433 if _, err := dst.Write(frame); err != nil {
434 return err
435 }
436 *bytes += int64(len(frame))
437 }
438 }
439
440 func (s *Server) allowed(command string) bool {
441 for _, cmd := range DefaultCommands {
442 if cmd == command {
443 return true
444 }
445 }
446 for _, cmd := range s.cfg.AllowCommands {
447 if cmd == command {
448 return true
449 }
450 }
451 return false
452 }
453
454 func (s *Server) presenceLoop(ctx context.Context) {
455 defer s.wg.Done()
456 ticker := time.NewTicker(s.cfg.PresenceInterval)
457 defer ticker.Stop()
458 for {
459 s.publishPresence(ctx)
460 select {
461 case <-s.done:
462 return
463 case <-ctx.Done():
464 return
465 case <-ticker.C:
466 }
467 }
468 }
469
470 func (s *Server) publishPresence(ctx context.Context) {
471 info := registry.Info{
472 Version: s.cfg.Version,
473 URI: s.URI().String(),
474 Chains: s.cfg.Chains,
475 Relays: s.cfg.Relays,
476 }
477 tipCtx, cancel := context.WithTimeout(ctx, 6*time.Second)
478 if node, err := CheckNode(tipCtx, s.cfg.NodeAddr, s.cfg.Profile, 5*time.Second); err == nil {
479 info.Tip = int64(node.TipHeight)
480 }
481 cancel()
482 ev, err := registry.Event(info, s.cfg.Key)
483 if err != nil {
484 s.cfg.Logf("bridge: presence: %v", err)
485 return
486 }
487 pctx, pcancel := context.WithTimeout(ctx, 10*time.Second)
488 defer pcancel()
489 if err := s.listener.Publish(pctx, ev); err != nil {
490 s.cfg.Logf("bridge: presence: %v", err)
491 return
492 }
493 s.cfg.Logf("bridge: announced gateway as %s", nostr.Hex(ev.PubKey[:]))
494 }
495
496 // Close stops the listener and closes every node connection.
497 func (s *Server) Close() error {
498 if s.ws != nil {
499 s.ws.Close()
500 }
501 select {
502 case <-s.done:
503 default:
504 close(s.done)
505 }
506 err := s.listener.Close()
507 s.mu.Lock()
508 for c := range s.conns {
509 c.Close()
510 }
511 s.mu.Unlock()
512 s.wg.Wait()
513 return err
514 }
515
516 // NodeInfo is the result of a node compatibility check.
517 type NodeInfo struct {
518 ProtocolVersion int32
519 Services uint64
520 UserAgent string
521 TipHeight int32
522 CompactFilters bool
523 }
524
525 // CheckNode connects to a node, handshakes, and reports whether it advertises
526 // compact filters. It is the bridge operator's preflight check.
527 func CheckNode(ctx context.Context, addr string, profile *chain.Profile, timeout time.Duration) (*NodeInfo, error) {
528 if profile == nil {
529 return nil, errors.New("bridge: nil profile")
530 }
531 if timeout == 0 {
532 timeout = 10 * time.Second
533 }
534 dctx, cancel := context.WithTimeout(ctx, timeout)
535 defer cancel()
536 peer, err := p2p.DialTCP(dctx, addr, p2p.Config{
537 Magic: profile.Magic,
538 UserAgent: p2p.DefaultUserAgent,
539 ReadTimeout: timeout,
540 RequireCompactFilters: false,
541 })
542 if err != nil {
543 return nil, err
544 }
545 defer peer.Close()
546 v, err := peer.Handshake(dctx)
547 if err != nil {
548 return nil, err
549 }
550 return &NodeInfo{
551 ProtocolVersion: v.ProtocolVersion,
552 Services: v.Services,
553 UserAgent: v.UserAgent,
554 TipHeight: v.LastBlock,
555 CompactFilters: v.Services&btcwire.ServiceNodeCompactFilters != 0,
556 }, nil
557 }
558