tunnel.go raw
1 // Package tunnel carries a reliable, ordered, encrypted byte stream over
2 // NIP-01 events on a public Nostr relay.
3 //
4 // The point is to give a Bitcoin P2P peer (Neutrino) a net.Conn where the
5 // network would normally be a TCP socket. The bridge listens on the relay, and
6 // each accepted stream is connected to a chain node's P2P port. The relay sees
7 // only signed events with opaque NIP-44 content; the gateway sees the P2P
8 // bytes and nothing else.
9 //
10 // The stream is a windowed protocol: sequence numbers, cumulative
11 // acknowledgements, gap detection with negative acknowledgements, and
12 // retransmission. Relay delivery is normally ordered and reliable, so the
13 // retransmission path is a safety net for a relay that drops an event, not the
14 // common path. Frames are bounded by the relay's event size limit.
15 package tunnel
16
17 import (
18 "context"
19 "crypto/rand"
20 "encoding/hex"
21 "errors"
22 "fmt"
23 "net"
24 "sync"
25 "time"
26
27 "git.smesh.lol/nostrino/pkg/crypto/secp256k1"
28 "git.smesh.lol/nostrino/pkg/nip44"
29 "git.smesh.lol/nostrino/pkg/nostr"
30 "git.smesh.lol/nostrino/pkg/relay"
31 )
32
33 // Protocol version carried in the URI, HELLO and every envelope.
34 const protocolVersion = 1
35
36 // Event kinds. They are in the ephemeral range (20000..29999) on purpose: the
37 // stream is live traffic, not history, and a relay must not store it. A
38 // dropped session is restarted by the caller, which is also what a TCP reset
39 // means.
40 const (
41 // KindClient carries client-to-gateway frames.
42 KindClient = 23196
43 // KindGateway carries gateway-to-client frames.
44 KindGateway = 23197
45 )
46
47 // Bounds and timings.
48 const (
49 defaultMTU = 32 << 10
50 minMTU = 1024
51 maxMTU = 96 << 10
52 defaultWindow = 32
53 minWindow = 4
54 maxWindow = 256
55 defaultHandshake = 15 * time.Second
56 defaultHeartbeat = 15 * time.Second
57 defaultIdle = 90 * time.Second
58 ackEvery = 1
59 nackDelay = 300 * time.Millisecond
60 retransmitAfter = 1500 * time.Millisecond
61 maxFrameRetries = 8
62 sendQueueFrames = 64
63 inboxFrames = 128
64 )
65
66 var (
67 // ErrClosed is returned after the stream or the relay connection ends.
68 ErrClosed = errors.New("tunnel: stream closed")
69 // ErrTimeout is returned when a deadline expires.
70 ErrTimeout = timeoutError{}
71 // ErrFrameAhead means a peer sent a frame further ahead of the one this side expects
72 // than a whole window allows, which correct flow control cannot produce.
73 ErrFrameAhead = errors.New("tunnel: frame is too far ahead")
74 )
75
76 type timeoutError struct{}
77
78 func (timeoutError) Error() string { return "tunnel: i/o timeout" }
79 func (timeoutError) Timeout() bool { return true }
80 func (timeoutError) Temporary() bool { return true }
81
82 type side int
83
84 const (
85 sideClient side = iota
86 sideGateway
87 )
88
89 func (s side) String() string {
90 if s == sideGateway {
91 return "gateway"
92 }
93 return "client"
94 }
95
96 // DialConfig configures the client end of a tunnel.
97 type DialConfig struct {
98 // URI is required: it names the gateway pubkey and the relay.
99 URI *URI
100 // Key is the client's ephemeral transport key. The zero value generates
101 // one for this connection, which is the intended privacy model: the
102 // gateway and the relay do not learn a stable client identity.
103 Key [32]byte
104 // Chain is the chain name sent in HELLO, for example "btc" or "xbt".
105 Chain string
106 // MTU is the largest plaintext data frame. Zero means 32 KiB, which stays
107 // below a 60 KB encrypted-event limit with base64 expansion.
108 MTU int
109 // Window is the number of unacknowledged frames allowed in flight.
110 Window int
111 // HandshakeTimeout bounds connect, subscribe, EOSE and HELLO_ACK.
112 HandshakeTimeout time.Duration
113 // Heartbeat is the ping interval. Zero means 15s.
114 Heartbeat time.Duration
115 // IdleTimeout closes a stream that has received nothing. Zero means 90s.
116 IdleTimeout time.Duration
117 // Logf receives diagnostics. Nil discards them.
118 Logf func(string, ...any)
119 // TransportFactory overrides the relay WebSocket dialer, which is how the
120 // wasm build supplies its syscall/js transport and how tests inject one.
121 TransportFactory func(url string) (relay.Transport, error)
122 }
123
124 // Conn is one tunnel stream and implements net.Conn.
125 type Conn struct {
126 sess *session
127 local string
128 remote string
129 }
130
131 // Read implements net.Conn.
132 func (c *Conn) Read(p []byte) (int, error) { return c.sess.read(p) }
133
134 // Write implements net.Conn. It returns once the bytes are queued for
135 // transmission, subject to the tunnel window.
136 func (c *Conn) Write(p []byte) (int, error) { return c.sess.write(p) }
137
138 // Close closes the stream and, on the client end, its relay connection.
139 func (c *Conn) Close() error {
140 c.sess.closeWithErr(ErrClosed)
141 return nil
142 }
143
144 // Reset aborts the stream instead of closing it in order. It is the
145 // cancellation primitive: a browser worker that is being torn down, or a
146 // client abandoning a filter range and switching gateway, uses Reset so the
147 // peer releases the node connection without waiting for the stream to drain.
148 func (c *Conn) Reset() error {
149 c.sess.resetWithErr(errors.New("tunnel: reset by local end"))
150 return nil
151 }
152
153 // LocalAddr implements net.Conn.
154 func (c *Conn) LocalAddr() net.Addr { return netAddr(c.local) }
155
156 // RemoteAddr implements net.Conn.
157 func (c *Conn) RemoteAddr() net.Addr { return netAddr(c.remote) }
158
159 // SetDeadline implements net.Conn.
160 func (c *Conn) SetDeadline(t time.Time) error {
161 c.sess.setReadDeadline(t)
162 c.sess.setWriteDeadline(t)
163 return nil
164 }
165
166 // SetReadDeadline implements net.Conn.
167 func (c *Conn) SetReadDeadline(t time.Time) error {
168 c.sess.setReadDeadline(t)
169 return nil
170 }
171
172 // SetWriteDeadline implements net.Conn.
173 func (c *Conn) SetWriteDeadline(t time.Time) error {
174 c.sess.setWriteDeadline(t)
175 return nil
176 }
177
178 // Peer is the remote transport public key.
179 func (c *Conn) Peer() [32]byte { return c.sess.peerPub }
180
181 // PeerHex is Peer as lowercase hex.
182 func (c *Conn) PeerHex() string { return nostr.Hex(c.sess.peerPub[:]) }
183
184 // Chain is the chain name negotiated in HELLO.
185 func (c *Conn) Chain() string { return c.sess.chain }
186
187 type netAddr string
188
189 func (a netAddr) Network() string { return "nostr" }
190 func (a netAddr) String() string { return string(a) }
191
192 func defaultDialConfig(cfg DialConfig) (DialConfig, error) {
193 if cfg.URI == nil {
194 return cfg, errors.New("tunnel: nil uri")
195 }
196 if cfg.MTU == 0 {
197 cfg.MTU = defaultMTU
198 }
199 if cfg.MTU < minMTU || cfg.MTU > maxMTU {
200 return cfg, fmt.Errorf("tunnel: mtu %d outside [%d,%d]", cfg.MTU, minMTU, maxMTU)
201 }
202 if cfg.Window == 0 {
203 cfg.Window = defaultWindow
204 }
205 if cfg.Window < minWindow || cfg.Window > maxWindow {
206 return cfg, fmt.Errorf("tunnel: window %d outside [%d,%d]", cfg.Window, minWindow, maxWindow)
207 }
208 if cfg.HandshakeTimeout == 0 {
209 cfg.HandshakeTimeout = defaultHandshake
210 }
211 if cfg.Heartbeat == 0 {
212 cfg.Heartbeat = defaultHeartbeat
213 }
214 if cfg.IdleTimeout == 0 {
215 cfg.IdleTimeout = defaultIdle
216 }
217 if cfg.TransportFactory == nil {
218 cfg.TransportFactory = relay.DialTransport
219 }
220 if cfg.Logf == nil {
221 cfg.Logf = func(string, ...any) {}
222 }
223 return cfg, nil
224 }
225
226 // Dial opens a tunnel to the gateway named by cfg.URI.
227 func Dial(ctx context.Context, cfg DialConfig) (*Conn, error) {
228 cfg, err := defaultDialConfig(cfg)
229 if err != nil {
230 return nil, err
231 }
232 var key [32]byte
233 if cfg.Key == ([32]byte{}) {
234 if _, err := rand.Read(key[:]); err != nil {
235 return nil, fmt.Errorf("tunnel: generate key: %w", err)
236 }
237 } else {
238 key = cfg.Key
239 }
240 localPub, ok := secp256k1.PubKey(key)
241 if !ok {
242 return nil, errors.New("tunnel: invalid client key")
243 }
244 if !secp256k1.ValidatePubKey(cfg.URI.Gateway) {
245 return nil, errors.New("tunnel: invalid gateway key")
246 }
247
248 dialCtx, cancel := context.WithTimeout(ctx, cfg.HandshakeTimeout)
249 defer cancel()
250
251 sessID, err := newSessionID()
252 if err != nil {
253 return nil, err
254 }
255 var (
256 clients []*relay.Client
257 subs []*relay.Subscription
258 lastErr error
259 )
260 filter := nostr.Filter{
261 Kinds: []int{KindGateway},
262 Tags: map[string][]string{"p": {nostr.Hex(localPub[:])}, "s": {sessID}},
263 }
264 for _, relayURL := range cfg.URI.Relays {
265 t, err := cfg.TransportFactory(relayURL)
266 if err != nil {
267 lastErr = err
268 cfg.Logf("tunnel: relay %s: %v", relayURL, err)
269 continue
270 }
271 c, err := relay.New(t, key, relay.Options{RelayURL: relayURL})
272 if err != nil {
273 t.Close()
274 lastErr = err
275 cfg.Logf("tunnel: relay %s: %v", relayURL, err)
276 continue
277 }
278 sub, err := c.Subscribe(filter)
279 if err == nil {
280 err = c.WaitEOSE(dialCtx, sub)
281 }
282 if err != nil {
283 c.Close()
284 lastErr = err
285 cfg.Logf("tunnel: relay %s: %v", relayURL, err)
286 continue
287 }
288 clients = append(clients, c)
289 subs = append(subs, sub)
290 }
291 if len(clients) == 0 {
292 if lastErr == nil {
293 lastErr = errors.New("no relay")
294 }
295 return nil, fmt.Errorf("tunnel: connect relay: %w", lastErr)
296 }
297 closed := false
298 defer func() {
299 if !closed {
300 for _, c := range clients {
301 c.Close()
302 }
303 }
304 }()
305
306 s, err := newSession(sideClient, key, localPub, cfg.URI.Gateway, sessID, cfg)
307 if err != nil {
308 return nil, err
309 }
310 s.chain = cfg.Chain
311 s.setRelays(clients, subs)
312 s.start()
313 hello := &envelope{
314 V: protocolVersion, S: sessID, T: frameHello,
315 MTU: cfg.MTU, Win: cfg.Window, PV: protocolVersion,
316 Chain: cfg.Chain,
317 }
318 if err := s.sendEnvelope(hello); err != nil {
319 s.closeWithErr(err)
320 return nil, fmt.Errorf("tunnel: hello: %w", err)
321 }
322
323 select {
324 case <-s.established:
325 case <-dialCtx.Done():
326 s.closeWithErr(dialCtx.Err())
327 return nil, fmt.Errorf("tunnel: waiting for hello_ack: %w", dialCtx.Err())
328 case <-s.done:
329 return nil, fmt.Errorf("tunnel: handshake: %w", s.Err())
330 }
331 closed = true // the session owns the relay client now
332 return &Conn{
333 sess: s,
334 local: "client:" + nostr.Hex(localPub[:]),
335 remote: "gateway:" + nostr.Hex(cfg.URI.Gateway[:]),
336 }, nil
337 }
338
339 // DialURI is Dial with a parsed connection string.
340 func DialURI(ctx context.Context, raw string, cfg DialConfig) (*Conn, error) {
341 u, err := ParseURI(raw)
342 if err != nil {
343 return nil, err
344 }
345 cfg.URI = u
346 return Dial(ctx, cfg)
347 }
348
349 func newSessionID() (string, error) {
350 var b [8]byte
351 if _, err := rand.Read(b[:]); err != nil {
352 return "", fmt.Errorf("tunnel: session id: %w", err)
353 }
354 return hex.EncodeToString(b[:]), nil
355 }
356
357 // ListenConfig configures the gateway end of a tunnel.
358 type ListenConfig struct {
359 // Relays are all relays the gateway announces and listens on.
360 Relays []string
361 // Key is the gateway's persistent transport key.
362 Key [32]byte
363 // Chains are the chain names this gateway serves, for example
364 // ["btc","xbt"]. Empty serves any chain name.
365 Chains []string
366 // Allowlist is an optional set of client public keys (hex). Empty accepts
367 // any client, which is the public chain-data model.
368 Allowlist []string
369 // MaxSessions bounds concurrent streams. Zero means 64.
370 MaxSessions int
371 // MaxPerClient bounds one client pubkey's concurrent streams. Zero means 2.
372 MaxPerClient int
373 // MTU and Window are negotiated downward with the client.
374 MTU int
375 Window int
376 // HandshakeTimeout bounds relay connect and EOSE. Zero means 15s.
377 HandshakeTimeout time.Duration
378 // Heartbeat and IdleTimeout are per-session liveness bounds.
379 Heartbeat time.Duration
380 IdleTimeout time.Duration
381 Logf func(string, ...any)
382 TransportFactory func(url string) (relay.Transport, error)
383 }
384
385 // Listener accepts tunnel streams from any client that addresses the gateway
386 // public key on one of its relays.
387 type Listener struct {
388 cfg ListenConfig
389 key [32]byte
390 pub [32]byte
391 relays []*relay.Client
392 subs []*relay.Subscription
393
394 mu sync.Mutex
395 sessions map[string]*session
396 perClient map[string]int
397 accept chan *Conn
398 closed bool
399 done chan struct{}
400 closeErr error
401 wg sync.WaitGroup
402 }
403
404 // Listen connects to every configured relay and starts accepting streams.
405 func Listen(ctx context.Context, cfg ListenConfig) (*Listener, error) {
406 if cfg.Key == ([32]byte{}) {
407 return nil, errors.New("tunnel: gateway key is required")
408 }
409 pub, ok := secp256k1.PubKey(cfg.Key)
410 if !ok {
411 return nil, errors.New("tunnel: invalid gateway key")
412 }
413 if len(cfg.Relays) == 0 {
414 return nil, errors.New("tunnel: no relay configured")
415 }
416 if cfg.MTU == 0 {
417 cfg.MTU = defaultMTU
418 }
419 if cfg.Window == 0 {
420 cfg.Window = defaultWindow
421 }
422 if cfg.MaxSessions == 0 {
423 cfg.MaxSessions = 64
424 }
425 if cfg.MaxPerClient == 0 {
426 cfg.MaxPerClient = 2
427 }
428 if cfg.HandshakeTimeout == 0 {
429 cfg.HandshakeTimeout = defaultHandshake
430 }
431 if cfg.Heartbeat == 0 {
432 cfg.Heartbeat = defaultHeartbeat
433 }
434 if cfg.IdleTimeout == 0 {
435 cfg.IdleTimeout = defaultIdle
436 }
437 if cfg.TransportFactory == nil {
438 cfg.TransportFactory = relay.DialTransport
439 }
440 if cfg.Logf == nil {
441 cfg.Logf = func(string, ...any) {}
442 }
443 l := &Listener{
444 cfg: cfg,
445 key: cfg.Key,
446 pub: pub,
447 sessions: make(map[string]*session),
448 perClient: make(map[string]int),
449 accept: make(chan *Conn, 8),
450 done: make(chan struct{}),
451 }
452 handshakeCtx, cancel := context.WithTimeout(ctx, cfg.HandshakeTimeout)
453 ready := make(chan struct{}, len(cfg.Relays))
454 for _, rawURL := range cfg.Relays {
455 l.wg.Add(1)
456 go l.relayLoop(rawURL, ready)
457 }
458 select {
459 case <-ready:
460 cancel()
461 return l, nil
462 case <-handshakeCtx.Done():
463 cancel()
464 l.Close()
465 return nil, errors.New("tunnel: no relay connected")
466 }
467 }
468
469 // Publish sends a signed event on every connected relay. It returns nil when
470 // at least one relay accepted it.
471 func (l *Listener) Publish(ctx context.Context, ev *nostr.Event) error {
472 l.mu.Lock()
473 relays := append([]*relay.Client(nil), l.relays...)
474 l.mu.Unlock()
475 if len(relays) == 0 {
476 return errors.New("tunnel: no relay connected")
477 }
478 var lastErr error
479 sent := 0
480 for _, rc := range relays {
481 if err := rc.Publish(ctx, ev); err != nil {
482 lastErr = err
483 continue
484 }
485 sent++
486 }
487 if sent == 0 {
488 return fmt.Errorf("tunnel: publish: %w", lastErr)
489 }
490 return nil
491 }
492
493 // PubKey is the gateway transport public key.
494 func (l *Listener) PubKey() [32]byte { return l.pub }
495
496 // PubKeyHex is PubKey as lowercase hex.
497 func (l *Listener) PubKeyHex() string { return nostr.Hex(l.pub[:]) }
498
499 // Accept waits for the next handshaken stream.
500 func (l *Listener) Accept(ctx context.Context) (*Conn, error) {
501 select {
502 case c := <-l.accept:
503 return c, nil
504 case <-l.done:
505 if l.closeErr != nil {
506 return nil, l.closeErr
507 }
508 return nil, ErrClosed
509 case <-ctx.Done():
510 return nil, ctx.Err()
511 }
512 }
513
514 // Close ends every session and relay connection.
515 func (l *Listener) Close() error {
516 l.mu.Lock()
517 if l.closed {
518 l.mu.Unlock()
519 return nil
520 }
521 l.closed = true
522 l.closeErr = ErrClosed
523 close(l.done)
524 sessions := make([]*session, 0, len(l.sessions))
525 for _, s := range l.sessions {
526 sessions = append(sessions, s)
527 }
528 subs := append([]*relay.Subscription(nil), l.subs...)
529 relays := append([]*relay.Client(nil), l.relays...)
530 l.mu.Unlock()
531 for _, s := range sessions {
532 s.closeWithErr(ErrClosed)
533 }
534 for _, sub := range subs {
535 sub.Close()
536 }
537 for _, rc := range relays {
538 rc.Close()
539 }
540 l.wg.Wait()
541 return nil
542 }
543
544 // relayLoop maintains one relay connection: connect, subscribe, read, and
545 // reconnect with backoff when the relay or the network drops it. A public
546 // relay is not a durable socket, and a gateway that stops listening after the
547 // first idle timeout is not a gateway.
548 func (l *Listener) relayLoop(rawURL string, ready chan struct{}) {
549 defer l.wg.Done()
550 signaled := false
551 backoff := time.Second
552 for {
553 select {
554 case <-l.done:
555 return
556 default:
557 }
558 t, err := l.cfg.TransportFactory(rawURL)
559 if err != nil {
560 l.cfg.Logf("tunnel: relay %s: %v", rawURL, err)
561 if !l.sleep(backoff) {
562 return
563 }
564 backoff = nextBackoff(backoff)
565 continue
566 }
567 rc, err := relay.New(t, l.key, relay.Options{RelayURL: rawURL})
568 if err != nil {
569 t.Close()
570 if !l.sleep(backoff) {
571 return
572 }
573 backoff = nextBackoff(backoff)
574 continue
575 }
576 sub, err := rc.Subscribe(nostr.Filter{
577 Kinds: []int{KindClient},
578 Tags: map[string][]string{"p": {nostr.Hex(l.pub[:])}},
579 })
580 if err == nil {
581 ectx, ecancel := context.WithTimeout(context.Background(), l.cfg.HandshakeTimeout)
582 err = rc.WaitEOSE(ectx, sub)
583 ecancel()
584 }
585 if err != nil {
586 rc.Close()
587 l.cfg.Logf("tunnel: relay %s: %v", rawURL, err)
588 if !l.sleep(backoff) {
589 return
590 }
591 backoff = nextBackoff(backoff)
592 continue
593 }
594 l.mu.Lock()
595 if l.closed {
596 l.mu.Unlock()
597 rc.Close()
598 return
599 }
600 l.relays = append(l.relays, rc)
601 l.subs = append(l.subs, sub)
602 l.mu.Unlock()
603 l.cfg.Logf("tunnel: listening on %s as %s", rawURL, nostr.Hex(l.pub[:]))
604 backoff = time.Second
605 if !signaled {
606 signaled = true
607 select {
608 case ready <- struct{}{}:
609 default:
610 }
611 }
612 l.readLoop(rc, sub)
613 l.mu.Lock()
614 for i, c := range l.relays {
615 if c == rc {
616 l.relays = append(l.relays[:i], l.relays[i+1:]...)
617 break
618 }
619 }
620 for i, sb := range l.subs {
621 if sb == sub {
622 l.subs = append(l.subs[:i], l.subs[i+1:]...)
623 break
624 }
625 }
626 l.mu.Unlock()
627 rc.Close()
628 if !l.sleep(backoff) {
629 return
630 }
631 backoff = nextBackoff(backoff)
632 }
633 }
634
635 func (l *Listener) sleep(d time.Duration) bool {
636 select {
637 case <-time.After(d):
638 return true
639 case <-l.done:
640 return false
641 }
642 }
643
644 func nextBackoff(d time.Duration) time.Duration {
645 d *= 2
646 if d > 30*time.Second {
647 d = 30 * time.Second
648 }
649 return d
650 }
651
652 // readLoop dispatches one relay's client events to sessions.
653 func (l *Listener) readLoop(rc *relay.Client, sub *relay.Subscription) {
654 for {
655 select {
656 case <-l.done:
657 return
658 case <-sub.Done():
659 return
660 case ev := <-sub.Events:
661 if ev.Kind != KindClient || ev.PubKey == l.pub {
662 continue
663 }
664 peer := ev.PubKey
665 if !l.allowed(peer) {
666 continue
667 }
668 conv, err := nip44.ConversationKey(l.key, peer)
669 if err != nil {
670 continue
671 }
672 raw, err := nip44.Decrypt(conv, ev.Content)
673 if err != nil {
674 continue
675 }
676 env, err := parseEnvelope(raw)
677 if err != nil {
678 continue
679 }
680 if env.T == frameHello {
681 l.handleHello(rc, conv, peer, env)
682 continue
683 }
684 l.mu.Lock()
685 s := l.sessions[env.S]
686 l.mu.Unlock()
687 if s != nil {
688 s.addRelay(rc)
689 s.enqueue(env)
690 }
691 }
692 }
693 }
694
695 func (l *Listener) allowed(peer [32]byte) bool {
696 if len(l.cfg.Allowlist) == 0 {
697 return true
698 }
699 hexKey := nostr.Hex(peer[:])
700 for _, allowed := range l.cfg.Allowlist {
701 if allowed == hexKey {
702 return true
703 }
704 }
705 return false
706 }
707
708 func (l *Listener) chainAllowed(chain string) bool {
709 if len(l.cfg.Chains) == 0 {
710 return true
711 }
712 for _, c := range l.cfg.Chains {
713 if c == chain {
714 return true
715 }
716 }
717 return false
718 }
719
720 func (l *Listener) handleHello(rc *relay.Client, conv [32]byte, peer [32]byte, env *envelope) {
721 chain := env.Chain
722 if !l.chainAllowed(chain) {
723 l.rejectHello(rc, conv, peer, env, "chain not served: "+chain)
724 return
725 }
726 if env.PV != protocolVersion {
727 l.rejectHello(rc, conv, peer, env, "unsupported protocol version")
728 return
729 }
730 mtu := env.MTU
731 if mtu == 0 || mtu > l.cfg.MTU {
732 mtu = l.cfg.MTU
733 }
734 if mtu < minMTU {
735 mtu = minMTU
736 }
737 window := env.Win
738 if window == 0 || window > l.cfg.Window {
739 window = l.cfg.Window
740 }
741 if window < minWindow {
742 window = minWindow
743 }
744
745 l.mu.Lock()
746 if existing := l.sessions[env.S]; existing != nil {
747 l.mu.Unlock()
748 existing.addRelay(rc)
749 existing.sendHelloAck()
750 return
751 }
752 if len(l.sessions) >= l.cfg.MaxSessions {
753 l.mu.Unlock()
754 l.rejectHello(rc, conv, peer, env, "gateway at session limit")
755 return
756 }
757 peerHex := nostr.Hex(peer[:])
758 if l.perClient[peerHex] >= l.cfg.MaxPerClient {
759 l.mu.Unlock()
760 l.rejectHello(rc, conv, peer, env, "client at session limit")
761 return
762 }
763 s, err := newSession(sideGateway, l.key, l.pub, peer, env.S, DialConfig{
764 Chain: chain,
765 MTU: mtu,
766 Window: window,
767 Heartbeat: l.cfg.Heartbeat,
768 IdleTimeout: l.cfg.IdleTimeout,
769 HandshakeTimeout: l.cfg.HandshakeTimeout,
770 Logf: l.cfg.Logf,
771 })
772 if err != nil {
773 l.mu.Unlock()
774 l.cfg.Logf("tunnel: client %s: %v", nostr.Hex(peer[:]), err)
775 return
776 }
777 l.perClient[peerHex]++
778 s.chain = chain
779 s.setRelays([]*relay.Client{rc}, nil)
780 s.onClose = func() {
781 l.mu.Lock()
782 delete(l.sessions, env.S)
783 if l.perClient[peerHex] > 0 {
784 l.perClient[peerHex]--
785 }
786 l.mu.Unlock()
787 }
788 l.sessions[env.S] = s
789 l.mu.Unlock()
790
791 s.start()
792 s.sendHelloAck()
793 conn := &Conn{
794 sess: s,
795 local: "gateway:" + nostr.Hex(l.pub[:]),
796 remote: "client:" + nostr.Hex(peer[:]),
797 }
798 select {
799 case l.accept <- conn:
800 case <-l.done:
801 s.closeWithErr(ErrClosed)
802 default:
803 s.closeWithErr(errors.New("tunnel: accept backlog full"))
804 }
805 }
806
807 func (l *Listener) rejectHello(rc *relay.Client, conv [32]byte, peer [32]byte, env *envelope, reason string) {
808 s, err := newSession(sideGateway, l.key, l.pub, peer, env.S, DialConfig{
809 Heartbeat: l.cfg.Heartbeat,
810 IdleTimeout: l.cfg.IdleTimeout,
811 HandshakeTimeout: l.cfg.HandshakeTimeout,
812 Logf: l.cfg.Logf,
813 })
814 if err != nil {
815 return
816 }
817 s.setRelays([]*relay.Client{rc}, nil)
818 _ = s.sendEnvelope(&envelope{
819 V: protocolVersion, S: env.S, T: frameClose, Reason: reason,
820 })
821 s.closeWithErr(errors.New(reason))
822 }
823