feat(client): add incoming connection handling with lease validation
Refactor RelayClient to support accepting incoming connections on leased domains, including secure handshake management and channel-based notification. Renamed internal types for consistency and added goroutines for lease listening and updates. This enables bidirectional communication via the relay.
lemon-mint committed
Oct 27, 2025 at 13:14 UTC
c90d26137f409d9fe41c6c68456405f39d83adcf
4 files changed
+159
-47
relaydns/client.go
+148
-37
@@ -19,43 +19,57 @@ var (
19
ErrConnectionRejected = errors.New("connection rejected")
20
)
21
22
+type IncommingConn struct {
23
+ *cryptoops.SecureConnection
24
+ id string
25
+}
26
+
27
+func (i *IncommingConn) ID() string {
28
+ return i.id
29
+}
30
+
31
// RelayClient는 RelayServer에 연결하여 서비스를 요청하는 클라이언트입니다.
32
type RelayClient struct {
33
conn io.ReadWriteCloser
34
35
sess *yamux.Session
36
28
- streams map[uint32]*yamux.Stream
29
- streamsMu sync.Mutex
30
-
31
- leases map[string]*LeaseWithCred
37
+ leases map[string]*leaseWithCred
38
leasesMu sync.Mutex
39
40
stopCh chan struct{}
41
waitGroup sync.WaitGroup
42
+
43
+ incommingConnCh chan *IncommingConn
44
}
45
38
-type LeaseWithCred struct {
46
+type leaseWithCred struct {
47
Lease *rdverb.Lease
48
Cred *cryptoops.Credential
49
}
50
51
// NewRelayClient는 새로운 RelayClient 인스턴스를 생성합니다.
52
func NewRelayClient(conn io.ReadWriteCloser) *RelayClient {
45
- return &RelayClient{
46
- conn: conn,
47
- streams: make(map[uint32]*yamux.Stream),
48
- leases: make(map[string]*LeaseWithCred),
49
- stopCh: make(chan struct{}),
53
+ g := &RelayClient{
54
+ conn: conn,
55
+ leases: make(map[string]*leaseWithCred),
56
+ stopCh: make(chan struct{}),
57
+ incommingConnCh: make(chan *IncommingConn),
58
}
59
+
60
+ g.waitGroup.Add(1)
61
+ go g.leaseUpdateWorker()
62
+ go g.leaseListenWorker()
63
+
64
+ return g
65
}
66
67
// Close는 서버와의 연결을 종료합니다.
54
-func (c *RelayClient) Close() error {
55
- close(c.stopCh)
56
- c.waitGroup.Wait()
68
+func (g *RelayClient) Close() error {
69
+ close(g.stopCh)
70
+ g.waitGroup.Wait()
71
58
- err := c.conn.Close()
72
+ err := g.conn.Close()
73
if err != nil {
74
return err
75
}
@@ -63,38 +77,113 @@ func (c *RelayClient) Close() error {
77
}
78
79
// leaseUpdateWorker는 리스 업데이트를 처리하는 워커입니다.
66
-func (c *RelayClient) leaseUpdateWorker() {
80
+func (g *RelayClient) leaseUpdateWorker() {
81
+ defer g.waitGroup.Done()
82
+
83
ticker := time.NewTicker(5 * time.Second)
68
- var updateRequired = map[*LeaseWithCred]struct{}{}
84
+ var updateRequired = map[*leaseWithCred]struct{}{}
85
86
defer ticker.Stop()
87
for {
88
select {
73
- case <-c.stopCh:
89
+ case <-g.stopCh:
90
return
91
case <-ticker.C:
92
clear(updateRequired)
93
78
- c.leasesMu.Lock()
79
- for _, lease := range c.leases {
94
+ g.leasesMu.Lock()
95
+ for _, lease := range g.leases {
96
if lease.Lease.Expires < int64(time.Now().Add(30*time.Second).Unix()) {
97
updateRequired[lease] = struct{}{}
98
}
99
}
84
- c.leasesMu.Unlock()
100
+ g.leasesMu.Unlock()
101
102
for lease := range updateRequired {
103
lease.Lease.Expires = time.Now().Add(30 * time.Second).Unix()
88
- c.updateLease(context.Background(), lease.Cred, lease.Lease)
104
+ g.updateLease(context.Background(), lease.Cred, lease.Lease)
105
}
106
}
107
}
108
}
109
110
+func (g *RelayClient) leaseListenWorker() {
111
+ for {
112
+ stream, err := g.sess.AcceptStream()
113
+ if err != nil {
114
+ continue
115
+ }
116
+ go g.handleConnectionRequestStream(stream)
117
+ }
118
+}
119
+
120
+func (g *RelayClient) handleConnectionRequestStream(stream *yamux.Stream) {
121
+ pkt, err := readPacket(stream)
122
+ if err != nil {
123
+ stream.Close()
124
+ return
125
+ }
126
+
127
+ if pkt.Type != rdverb.PacketType_PACKET_TYPE_CONNECTION_REQUEST {
128
+ stream.Close()
129
+ return
130
+ }
131
+
132
+ req := &rdverb.ConnectionRequest{}
133
+ err = req.UnmarshalVT(pkt.Payload)
134
+ if err != nil {
135
+ stream.Close()
136
+ return
137
+ }
138
+
139
+ g.leasesMu.Lock()
140
+ lease, ok := g.leases[req.LeaseId]
141
+ g.leasesMu.Unlock()
142
+
143
+ resp := &rdverb.ConnectionResponse{}
144
+ if !ok {
145
+ resp.Code = rdverb.ResponseCode_RESPONSE_CODE_REJECTED
146
+ } else {
147
+ resp.Code = rdverb.ResponseCode_RESPONSE_CODE_ACCEPTED
148
+ }
149
+
150
+ respPayload, err := resp.MarshalVT()
151
+ if err != nil {
152
+ stream.Close()
153
+ return
154
+ }
155
+
156
+ err = writePacket(stream, &rdverb.Packet{
157
+ Type: rdverb.PacketType_PACKET_TYPE_CONNECTION_RESPONSE,
158
+ Payload: respPayload,
159
+ })
160
+ if err != nil {
161
+ stream.Close()
162
+ return
163
+ }
164
+
165
+ if !ok {
166
+ stream.Close()
167
+ return
168
+ }
169
+
170
+ handshaker := cryptoops.NewHandshaker(lease.Cred)
171
+ secConn, err := handshaker.ServerHandshake(stream, lease.Lease.Alpn)
172
+ if err != nil {
173
+ stream.Close()
174
+ return
175
+ }
176
+
177
+ g.incommingConnCh <- &IncommingConn{
178
+ SecureConnection: secConn,
179
+ id: req.LeaseId,
180
+ }
181
+}
182
+
183
// GetRelayInfo는 서버의 릴레이 정보를 요청합니다.
95
-func (c *RelayClient) GetRelayInfo(ctx context.Context) (*rdverb.RelayInfo, error) {
184
+func (g *RelayClient) GetRelayInfo(ctx context.Context) (*rdverb.RelayInfo, error) {
185
// 새 스트림 열기
97
- stream, err := c.sess.OpenStream()
186
+ stream, err := g.sess.OpenStream()
187
if err != nil {
188
return nil, err
189
}
@@ -136,9 +225,9 @@ func (c *RelayClient) GetRelayInfo(ctx context.Context) (*rdverb.RelayInfo, erro
225
}
226
227
// updateLease는 서버에 리스 업데이트를 요청합니다.
139
-func (c *RelayClient) updateLease(ctx context.Context, cred *cryptoops.Credential, lease *rdverb.Lease) (rdverb.ResponseCode, error) {
228
+func (g *RelayClient) updateLease(ctx context.Context, cred *cryptoops.Credential, lease *rdverb.Lease) (rdverb.ResponseCode, error) {
229
// 새 스트림 열기
141
- stream, err := c.sess.OpenStream()
230
+ stream, err := g.sess.OpenStream()
231
if err != nil {
232
return rdverb.ResponseCode_RESPONSE_CODE_UNKNOWN, err
233
}
@@ -202,9 +291,9 @@ func (c *RelayClient) updateLease(ctx context.Context, cred *cryptoops.Credentia
291
}
292
293
// deleteLease는 서버에 리스 삭제를 요청합니다.
205
-func (c *RelayClient) deleteLease(ctx context.Context, cred *cryptoops.Credential, identity *rdsec.Identity) (rdverb.ResponseCode, error) {
294
+func (g *RelayClient) deleteLease(ctx context.Context, cred *cryptoops.Credential, identity *rdsec.Identity) (rdverb.ResponseCode, error) {
295
// 새 스트림 열기
207
- stream, err := c.sess.OpenStream()
296
+ stream, err := g.sess.OpenStream()
297
if err != nil {
298
return rdverb.ResponseCode_RESPONSE_CODE_UNKNOWN, err
299
}
@@ -268,9 +357,9 @@ func (c *RelayClient) deleteLease(ctx context.Context, cred *cryptoops.Credentia
357
}
358
359
// requestConnection은 다른 클라이언트로의 연결을 요청합니다.
271
-func (c *RelayClient) requestConnection(ctx context.Context, leaseID string, alpn string, clientCred *cryptoops.Credential) (rdverb.ResponseCode, io.ReadWriteCloser, error) {
360
+func (g *RelayClient) requestConnection(ctx context.Context, leaseID string, alpn string, clientCred *cryptoops.Credential) (rdverb.ResponseCode, io.ReadWriteCloser, error) {
361
// 새 스트림 열기
273
- stream, err := c.sess.OpenStream()
362
+ stream, err := g.sess.OpenStream()
363
if err != nil {
364
return rdverb.ResponseCode_RESPONSE_CODE_UNKNOWN, nil, err
365
}
@@ -337,7 +426,7 @@ func (c *RelayClient) requestConnection(ctx context.Context, leaseID string, alp
426
return resp.Code, secConn, nil
427
}
428
340
-func (c *RelayClient) RegisterLease(ctx context.Context, cred *cryptoops.Credential, name string, alpns []string) error {
429
+func (g *RelayClient) RegisterLease(ctx context.Context, cred *cryptoops.Credential, name string, alpns []string) error {
430
identity := &rdsec.Identity{
431
Id: cred.ID(),
432
PublicKey: cred.PublicKey(),
@@ -350,20 +439,42 @@ func (c *RelayClient) RegisterLease(ctx context.Context, cred *cryptoops.Credent
439
Alpn: alpns,
440
}
441
353
- c.leasesMu.Lock()
354
- c.leases[identity.Id] = &LeaseWithCred{
442
+ g.leasesMu.Lock()
443
+ g.leases[identity.Id] = &leaseWithCred{
444
Lease: lease,
445
Cred: cred,
446
}
358
- c.leasesMu.Unlock()
447
+ g.leasesMu.Unlock()
448
360
- resp, err := c.updateLease(ctx, cred, lease)
449
+ resp, err := g.updateLease(ctx, cred, lease)
450
if err != nil || resp != rdverb.ResponseCode_RESPONSE_CODE_ACCEPTED {
362
- c.leasesMu.Lock()
363
- delete(c.leases, identity.Id)
364
- c.leasesMu.Unlock()
451
+ g.leasesMu.Lock()
452
+ delete(g.leases, identity.Id)
453
+ g.leasesMu.Unlock()
454
return err
455
}
456
457
return nil
458
}
459
+
460
+func (g *RelayClient) DeregisterLease(ctx context.Context, cred *cryptoops.Credential) error {
461
+ identity := &rdsec.Identity{
462
+ Id: cred.ID(),
463
+ PublicKey: cred.PublicKey(),
464
+ }
465
+
466
+ resp, err := g.deleteLease(ctx, cred, identity)
467
+ if err != nil || resp != rdverb.ResponseCode_RESPONSE_CODE_ACCEPTED {
468
+ return err
469
+ }
470
+
471
+ g.leasesMu.Lock()
472
+ delete(g.leases, identity.Id)
473
+ g.leasesMu.Unlock()
474
+
475
+ return nil
476
+}
477
+
478
+func (g *RelayClient) IncommingConnection() <-chan *IncommingConn {
479
+ return g.incommingConnCh
480
+}
relaydns/core/cryptoops/handshaker.go
+6
-5
@@ -8,6 +8,7 @@ import (
8
"errors"
9
"fmt"
10
"io"
11
+ "slices"
12
"time"
13
14
"golang.org/x/crypto/chacha20poly1305"
@@ -237,7 +238,7 @@ func (h *Handshaker) ClientHandshake(conn io.ReadWriteCloser, alpn string) (*Sec
238
}
239
240
// ServerHandshake performs the server-side of the handshake
240
-func (h *Handshaker) ServerHandshake(conn io.ReadWriteCloser, alpn string) (*SecureConnection, error) {
241
+func (h *Handshaker) ServerHandshake(conn io.ReadWriteCloser, alpns []string) (*SecureConnection, error) {
242
// Read client init message
243
clientInitBytes, err := readLengthPrefixed(conn)
244
if err != nil {
@@ -256,7 +257,7 @@ func (h *Handshaker) ServerHandshake(conn io.ReadWriteCloser, alpn string) (*Sec
257
}
258
259
// Validate client init
259
- if err := h.validateClientInit(clientInitSigned, clientInitPayload, alpn); err != nil {
260
+ if err := h.validateClientInit(clientInitSigned, clientInitPayload, alpns); err != nil {
261
// Silent failure: close connection and return error without sending response
262
conn.Close()
263
return nil, err
@@ -283,7 +284,7 @@ func (h *Handshaker) ServerHandshake(conn io.ReadWriteCloser, alpn string) (*Sec
284
Id: h.credential.ID(),
285
PublicKey: h.credential.PublicKey(),
286
},
286
- Alpn: alpn,
287
+ Alpn: clientInitPayload.Alpn,
288
SessionPublicKey: ephemeralPub,
289
}
290
@@ -325,7 +326,7 @@ func (h *Handshaker) ServerHandshake(conn io.ReadWriteCloser, alpn string) (*Sec
326
}
327
328
// validateClientInit validates the client init message
328
-func (h *Handshaker) validateClientInit(clientInitSigned *rdsec.SignedPayload, clientInitPayload *rdsec.ClientInitPayload, expectedAlpn string) error {
329
+func (h *Handshaker) validateClientInit(clientInitSigned *rdsec.SignedPayload, clientInitPayload *rdsec.ClientInitPayload, expectedAlpns []string) error {
330
if clientInitSigned == nil || clientInitPayload == nil {
331
return ErrInvalidProtocol
332
}
@@ -341,7 +342,7 @@ func (h *Handshaker) validateClientInit(clientInitSigned *rdsec.SignedPayload, c
342
}
343
344
// Check ALPN
344
- if clientInitPayload.GetAlpn() != expectedAlpn {
345
+ if !slices.Contains(expectedAlpns, clientInitPayload.GetAlpn()) {
346
return ErrHandshakeFailed
347
}
348
relaydns/core/cryptoops/handshaker_test.go
+2
-2
@@ -85,7 +85,7 @@ func TestHandshake(t *testing.T) {
85
// Start server handshake in a goroutine
86
go func() {
87
defer func() { done <- true }()
88
- serverSecureConn, serverErr = serverHandshaker.ServerHandshake(serverPipeConn, alpn)
88
+ serverSecureConn, serverErr = serverHandshaker.ServerHandshake(serverPipeConn, []string{alpn})
89
}()
90
91
// Give the server a moment to start waiting
@@ -326,7 +326,7 @@ func TestHandshakeWithInvalidALPN(t *testing.T) {
326
// Start server handshake in a goroutine
327
go func() {
328
defer func() { done <- true }()
329
- _, serverErr = serverHandshaker.ServerHandshake(serverPipeConn, serverALPN)
329
+ _, serverErr = serverHandshaker.ServerHandshake(serverPipeConn, []string{serverALPN})
330
}()
331
332
// Give the server a moment to start waiting
relaydns/helper.go
+3
-3
@@ -19,15 +19,15 @@ func decodeProtobuf[T interface {
19
}](
20
data []byte,
21
) (
22
- T,
22
+ *T,
23
error,
24
) {
25
var t T
26
err := t.UnmarshalVT(data)
27
if err != nil {
28
- return t, err
28
+ return nil, err
29
}
30
- return t, nil
30
+ return &t, nil
31
}
32
33
func writePacket(w io.Writer, packet *rdverb.Packet) error {