refact: inline and consoliate unnecessary helpers
rabbitprincess committed
Apr 4, 2026 at 21:34 UTC
0468d9d53c958ee4082c18c4df7d762cb122dc37
7 files changed
+184
-213
AGENTS.md
+2
-1
@@ -20,5 +20,6 @@ These are mandates, not suggestions.
20
## Testing
21
22
- A test exists to catch real bugs. If deleting the test would not let a bug reach production, delete the test.
23
+- Do not add tests whose only purpose is generic regression prevention without a concrete bug, contract, or boundary at risk.
24
- Test contracts and boundaries: protocol compliance, error semantics, security invariants, integration across real I/O.
24
-- Do not test configuration shapes, constructor output fields, or struct assembly — the type system and constructors already guarantee those.
25
+- Do not test configuration shapes, constructor output fields, or struct assembly; the type system and constructors already guarantee those.
README.md
+1
-2
@@ -14,8 +14,7 @@
14
- **Self-hosted relays**: Connect to public relays or run your own
15
- **Relay discovery and pools**: Use discovered relays as a pool, with multi-relay access and failover
16
- **No login, no API keys**: Authenticate ownership using SIWE, with ENS-based identity support
17
-- **Raw TCP and UDP transport**: Native TCP reverse sessions with optional UDP (no SSH or WebSocket)
18
-- **TCP port routing**: Dedicated TCP ports for non-TLS services (e.g., Minecraft, game servers) without SNI-based routing
17
+- **Raw TCP/UDP + TCP port routing**: Native TCP reverse sessions, optional UDP, and dedicated TCP ports for non-TLS services
18
19
## Comparison
20
cmd/portal-tunnel/relays.go
+142
@@ -296,3 +296,145 @@ func proxyExposureDatagrams(ctx context.Context, exposure *sdk.Exposure, localAd
296
297
return nil
298
}
299
+
300
+type udpFlowKey struct {
301
+ flowID uint32
302
+ address string
303
+ relayURL string
304
+}
305
+
306
+type udpFlowEntry struct {
307
+ conn *net.UDPConn
308
+ lastSeen time.Time
309
+ frame types.DatagramFrame
310
+}
311
+
312
+type udpFlowManager struct {
313
+ target *net.UDPAddr
314
+ exposure *sdk.Exposure
315
+ mu sync.Mutex
316
+ flows map[udpFlowKey]*udpFlowEntry
317
+}
318
+
319
+func newUDPFlowManager(target *net.UDPAddr, exposure *sdk.Exposure) *udpFlowManager {
320
+ return &udpFlowManager{
321
+ target: target,
322
+ exposure: exposure,
323
+ flows: make(map[udpFlowKey]*udpFlowEntry),
324
+ }
325
+}
326
+
327
+func (m *udpFlowManager) runCleanup(ctx context.Context) {
328
+ ticker := time.NewTicker(15 * time.Second)
329
+ defer ticker.Stop()
330
+ for {
331
+ select {
332
+ case <-ctx.Done():
333
+ return
334
+ case <-ticker.C:
335
+ m.mu.Lock()
336
+ now := time.Now()
337
+ for key, f := range m.flows {
338
+ if now.Sub(f.lastSeen) > 30*time.Second {
339
+ _ = f.conn.Close()
340
+ delete(m.flows, key)
341
+ }
342
+ }
343
+ m.mu.Unlock()
344
+ }
345
+ }
346
+}
347
+
348
+func (m *udpFlowManager) getOrCreate(ctx context.Context, frame types.DatagramFrame) (*net.UDPConn, error) {
349
+ key := udpFlowKey{
350
+ flowID: frame.FlowID,
351
+ address: frame.Address,
352
+ relayURL: frame.RelayURL,
353
+ }
354
+
355
+ m.mu.Lock()
356
+ if f, ok := m.flows[key]; ok {
357
+ f.lastSeen = time.Now()
358
+ m.mu.Unlock()
359
+ return f.conn, nil
360
+ }
361
+ m.mu.Unlock()
362
+
363
+ localConn, err := net.DialUDP("udp", nil, m.target)
364
+ if err != nil {
365
+ return nil, err
366
+ }
367
+
368
+ m.mu.Lock()
369
+ if f, ok := m.flows[key]; ok {
370
+ m.mu.Unlock()
371
+ _ = localConn.Close()
372
+ f.lastSeen = time.Now()
373
+ return f.conn, nil
374
+ }
375
+ m.flows[key] = &udpFlowEntry{
376
+ conn: localConn,
377
+ lastSeen: time.Now(),
378
+ frame: types.DatagramFrame{
379
+ FlowID: frame.FlowID,
380
+ Address: frame.Address,
381
+ RelayURL: frame.RelayURL,
382
+ UDPAddr: frame.UDPAddr,
383
+ },
384
+ }
385
+ m.mu.Unlock()
386
+
387
+ go m.readLoop(ctx, key, localConn)
388
+ return localConn, nil
389
+}
390
+
391
+func (m *udpFlowManager) removeFlow(key udpFlowKey) {
392
+ m.mu.Lock()
393
+ if f, ok := m.flows[key]; ok {
394
+ _ = f.conn.Close()
395
+ delete(m.flows, key)
396
+ }
397
+ m.mu.Unlock()
398
+}
399
+
400
+func (m *udpFlowManager) readLoop(ctx context.Context, key udpFlowKey, conn *net.UDPConn) {
401
+ buf := make([]byte, 65535)
402
+ for {
403
+ n, err := conn.Read(buf)
404
+ if err != nil {
405
+ if ctx.Err() != nil {
406
+ return
407
+ }
408
+ log.Debug().
409
+ Err(err).
410
+ Uint32("flow_id", key.flowID).
411
+ Str("address", key.address).
412
+ Str("relay_url", key.relayURL).
413
+ Msg("local read ended")
414
+ m.removeFlow(key)
415
+ return
416
+ }
417
+
418
+ m.mu.Lock()
419
+ entry := m.flows[key]
420
+ if entry == nil {
421
+ m.mu.Unlock()
422
+ return
423
+ }
424
+ entry.lastSeen = time.Now()
425
+ replyFrame := entry.frame
426
+ replyFrame.Payload = append([]byte(nil), buf[:n]...)
427
+ m.mu.Unlock()
428
+
429
+ if sendErr := m.exposure.SendDatagram(replyFrame); sendErr != nil {
430
+ log.Debug().
431
+ Err(sendErr).
432
+ Uint32("flow_id", key.flowID).
433
+ Str("address", key.address).
434
+ Str("relay_url", key.relayURL).
435
+ Msg("send datagram to relay failed")
436
+ m.removeFlow(key)
437
+ return
438
+ }
439
+ }
440
+}
cmd/portal-tunnel/udp_flow.go
deleted
-155
@@ -1,155 +0,0 @@
1
-package main
2
-
3
-import (
4
- "context"
5
- "net"
6
- "sync"
7
- "time"
8
-
9
- "github.com/rs/zerolog/log"
10
-
11
- "github.com/gosuda/portal/v2/sdk"
12
- "github.com/gosuda/portal/v2/types"
13
-)
14
-
15
-type udpFlowKey struct {
16
- flowID uint32
17
- address string
18
- relayURL string
19
-}
20
-
21
-type udpFlowEntry struct {
22
- conn *net.UDPConn
23
- lastSeen time.Time
24
- frame types.DatagramFrame
25
-}
26
-
27
-type udpFlowManager struct {
28
- target *net.UDPAddr
29
- exposure *sdk.Exposure
30
- mu sync.Mutex
31
- flows map[udpFlowKey]*udpFlowEntry
32
-}
33
-
34
-func newUDPFlowManager(target *net.UDPAddr, exposure *sdk.Exposure) *udpFlowManager {
35
- return &udpFlowManager{
36
- target: target,
37
- exposure: exposure,
38
- flows: make(map[udpFlowKey]*udpFlowEntry),
39
- }
40
-}
41
-
42
-func (m *udpFlowManager) runCleanup(ctx context.Context) {
43
- ticker := time.NewTicker(15 * time.Second)
44
- defer ticker.Stop()
45
- for {
46
- select {
47
- case <-ctx.Done():
48
- return
49
- case <-ticker.C:
50
- m.mu.Lock()
51
- now := time.Now()
52
- for key, f := range m.flows {
53
- if now.Sub(f.lastSeen) > 30*time.Second {
54
- _ = f.conn.Close()
55
- delete(m.flows, key)
56
- }
57
- }
58
- m.mu.Unlock()
59
- }
60
- }
61
-}
62
-
63
-func (m *udpFlowManager) getOrCreate(ctx context.Context, frame types.DatagramFrame) (*net.UDPConn, error) {
64
- key := udpFlowKey{
65
- flowID: frame.FlowID,
66
- address: frame.Address,
67
- relayURL: frame.RelayURL,
68
- }
69
-
70
- m.mu.Lock()
71
- if f, ok := m.flows[key]; ok {
72
- f.lastSeen = time.Now()
73
- m.mu.Unlock()
74
- return f.conn, nil
75
- }
76
- m.mu.Unlock()
77
-
78
- localConn, err := net.DialUDP("udp", nil, m.target)
79
- if err != nil {
80
- return nil, err
81
- }
82
-
83
- m.mu.Lock()
84
- if f, ok := m.flows[key]; ok {
85
- m.mu.Unlock()
86
- _ = localConn.Close()
87
- f.lastSeen = time.Now()
88
- return f.conn, nil
89
- }
90
- m.flows[key] = &udpFlowEntry{
91
- conn: localConn,
92
- lastSeen: time.Now(),
93
- frame: types.DatagramFrame{
94
- FlowID: frame.FlowID,
95
- Address: frame.Address,
96
- RelayURL: frame.RelayURL,
97
- UDPAddr: frame.UDPAddr,
98
- },
99
- }
100
- m.mu.Unlock()
101
-
102
- go m.readLoop(ctx, key, localConn)
103
- return localConn, nil
104
-}
105
-
106
-func (m *udpFlowManager) removeFlow(key udpFlowKey) {
107
- m.mu.Lock()
108
- if f, ok := m.flows[key]; ok {
109
- _ = f.conn.Close()
110
- delete(m.flows, key)
111
- }
112
- m.mu.Unlock()
113
-}
114
-
115
-func (m *udpFlowManager) readLoop(ctx context.Context, key udpFlowKey, conn *net.UDPConn) {
116
- buf := make([]byte, 65535)
117
- for {
118
- n, err := conn.Read(buf)
119
- if err != nil {
120
- if ctx.Err() != nil {
121
- return
122
- }
123
- log.Debug().
124
- Err(err).
125
- Uint32("flow_id", key.flowID).
126
- Str("address", key.address).
127
- Str("relay_url", key.relayURL).
128
- Msg("local read ended")
129
- m.removeFlow(key)
130
- return
131
- }
132
-
133
- m.mu.Lock()
134
- entry := m.flows[key]
135
- if entry == nil {
136
- m.mu.Unlock()
137
- return
138
- }
139
- entry.lastSeen = time.Now()
140
- replyFrame := entry.frame
141
- replyFrame.Payload = append([]byte(nil), buf[:n]...)
142
- m.mu.Unlock()
143
-
144
- if sendErr := m.exposure.SendDatagram(replyFrame); sendErr != nil {
145
- log.Debug().
146
- Err(sendErr).
147
- Uint32("flow_id", key.flowID).
148
- Str("address", key.address).
149
- Str("relay_url", key.relayURL).
150
- Msg("send datagram to relay failed")
151
- m.removeFlow(key)
152
- return
153
- }
154
- }
155
-}
portal/api_server.go
+2
-2
@@ -529,7 +529,7 @@ func (s *Server) registerLease(req types.RegisterChallengeRequest, clientIP, rep
529
if !s.registry.policy.IsUDPEnabled() {
530
return types.RegisterResponse{}, errUDPDisabled
531
}
532
- if max := s.registry.policy.UDPMaxLeases(); max > 0 && s.registry.CountDatagramLeases() >= max {
532
+ if max := s.registry.policy.UDPMaxLeases(); max > 0 && s.registry.countDatagramLeases() >= max {
533
return types.RegisterResponse{}, errUDPCapacityExceeded
534
}
535
}
@@ -540,7 +540,7 @@ func (s *Server) registerLease(req types.RegisterChallengeRequest, clientIP, rep
540
if !s.registry.policy.IsTCPPortEnabled() {
541
return types.RegisterResponse{}, errTCPPortDisabled
542
}
543
- if max := s.registry.policy.TCPPortMaxLeases(); max > 0 && s.registry.CountTCPPortLeases() >= max {
543
+ if max := s.registry.policy.TCPPortMaxLeases(); max > 0 && s.registry.countTCPPortLeases() >= max {
544
return types.RegisterResponse{}, errTCPPortCapacityExceeded
545
}
546
}
portal/lease.go
+28
-38
@@ -62,9 +62,16 @@ func (r *leaseRegistry) Lookup(host string) (*leaseRecord, bool) {
62
r.mu.RLock()
63
defer r.mu.RUnlock()
64
65
- key, ok := routeLookup(r.routes, host)
65
+ key, ok := r.routes[host]
66
if !ok {
67
- return nil, false
67
+ parts := strings.Split(host, ".")
68
+ if len(parts) < 3 {
69
+ return nil, false
70
+ }
71
+ key, ok = r.routes["*."+strings.Join(parts[1:], ".")]
72
+ if !ok {
73
+ return nil, false
74
+ }
75
}
76
record, ok := r.leasesByKey[key]
77
return record, ok && record != nil
@@ -100,7 +107,7 @@ func (r *leaseRegistry) Register(record *leaseRecord) error {
107
record.Hostname = hostname
108
r.leasesByKey[key] = record
109
r.routes[hostname] = key
103
- r.setClientIPLocked(key, record.ClientIP)
110
+ r.policy.IPFilter().RegisterIdentityIP(key, record.ClientIP)
111
r.mu.Unlock()
112
113
if replaced != nil && replaced != record {
@@ -109,14 +116,6 @@ func (r *leaseRegistry) Register(record *leaseRecord) error {
116
return nil
117
}
118
112
-// setClientIPLocked updates the record's client IP and registers it with the
113
-// IP filter. Caller must hold r.mu.
114
-func (r *leaseRegistry) setClientIPLocked(identityKey, clientIP string) {
115
- if strings.TrimSpace(clientIP) != "" {
116
- r.policy.IPFilter().RegisterIdentityIP(identityKey, clientIP)
117
- }
118
-}
119
-
119
func (r *leaseRegistry) Renew(identity types.Identity, ttl time.Duration, clientIP, reportedIP string) (*leaseRecord, error) {
120
r.mu.Lock()
121
defer r.mu.Unlock()
@@ -135,7 +134,7 @@ func (r *leaseRegistry) Renew(identity types.Identity, ttl time.Duration, client
134
if strings.TrimSpace(reportedIP) != "" {
135
record.ReportedIP = reportedIP
136
}
138
- r.setClientIPLocked(record.Key(), clientIP)
137
+ r.policy.IPFilter().RegisterIdentityIP(record.Key(), clientIP)
138
return record, nil
139
}
140
@@ -171,7 +170,7 @@ func (r *leaseRegistry) issueRegisterChallenge(req types.RegisterChallengeReques
170
if !r.policy.IsUDPEnabled() {
171
return types.RegisterChallengeResponse{}, errUDPDisabled
172
}
174
- if max := r.policy.UDPMaxLeases(); max > 0 && r.CountDatagramLeases() >= max {
173
+ if max := r.policy.UDPMaxLeases(); max > 0 && r.countDatagramLeases() >= max {
174
return types.RegisterChallengeResponse{}, errUDPCapacityExceeded
175
}
176
}
@@ -179,7 +178,7 @@ func (r *leaseRegistry) issueRegisterChallenge(req types.RegisterChallengeReques
178
if !r.policy.IsTCPPortEnabled() {
179
return types.RegisterChallengeResponse{}, errTCPPortDisabled
180
}
182
- if max := r.policy.TCPPortMaxLeases(); max > 0 && r.CountTCPPortLeases() >= max {
181
+ if max := r.policy.TCPPortMaxLeases(); max > 0 && r.countTCPPortLeases() >= max {
182
return types.RegisterChallengeResponse{}, errTCPPortCapacityExceeded
183
}
184
}
@@ -239,7 +238,7 @@ func (r *leaseRegistry) Touch(identity types.Identity, clientIP string, now time
238
if strings.TrimSpace(clientIP) != "" {
239
record.ClientIP = clientIP
240
}
242
- r.setClientIPLocked(record.Key(), clientIP)
241
+ r.policy.IPFilter().RegisterIdentityIP(record.Key(), clientIP)
242
return record
243
}
244
@@ -264,25 +263,32 @@ func (r *leaseRegistry) cleanupExpired(now time.Time) []*leaseRecord {
263
return expired
264
}
265
267
-func (r *leaseRegistry) countActiveLeasesWhere(pred func(*leaseRecord) bool) int {
266
+func (r *leaseRegistry) countDatagramLeases() int {
267
r.mu.RLock()
268
defer r.mu.RUnlock()
269
+
270
now := time.Now()
271
count := 0
272
for _, record := range r.leasesByKey {
273
- if now.Before(record.ExpiresAt) && pred(record) {
273
+ if now.Before(record.ExpiresAt) && record.datagram != nil {
274
count++
275
}
276
}
277
return count
278
}
279
280
-func (r *leaseRegistry) CountDatagramLeases() int {
281
- return r.countActiveLeasesWhere(func(rec *leaseRecord) bool { return rec.datagram != nil })
282
-}
280
+func (r *leaseRegistry) countTCPPortLeases() int {
281
+ r.mu.RLock()
282
+ defer r.mu.RUnlock()
283
284
-func (r *leaseRegistry) CountTCPPortLeases() int {
285
- return r.countActiveLeasesWhere(func(rec *leaseRecord) bool { return rec.tcpPort != nil })
284
+ now := time.Now()
285
+ count := 0
286
+ for _, record := range r.leasesByKey {
287
+ if now.Before(record.ExpiresAt) && record.tcpPort != nil {
288
+ count++
289
+ }
290
+ }
291
+ return count
292
}
293
294
func (r *leaseRegistry) activeAdminSnapshots() []types.AdminLease {
@@ -393,19 +399,3 @@ func (r *leaseRecord) Close() {
399
}
400
}
401
}
396
-
397
-func routeLookup(routes map[string]string, host string) (string, bool) {
398
- if host == "" {
399
- return "", false
400
- }
401
- if identityKey, ok := routes[host]; ok {
402
- return identityKey, true
403
- }
404
- parts := strings.Split(host, ".")
405
- if len(parts) < 3 {
406
- return "", false
407
- }
408
- wildcard := "*." + strings.Join(parts[1:], ".")
409
- identityKey, ok := routes[wildcard]
410
- return identityKey, ok
411
-}
portal/server.go
+9
-15
@@ -257,10 +257,6 @@ func (s *Server) Wait() error {
257
return s.group.Wait()
258
}
259
260
-func (s *Server) Identity() types.Identity {
261
- return s.identity.Copy()
262
-}
263
-
260
func (s *Server) Shutdown(ctx context.Context) error {
261
var shutdownErr error
262
s.shutdownOnce.Do(func() {
@@ -529,25 +525,23 @@ func BridgeConns(left, right net.Conn) {
525
defer left.Close()
526
defer right.Close()
527
528
+ type closeWriter interface {
529
+ CloseWrite() error
530
+ }
531
var group errgroup.Group
532
group.Go(func() error {
533
_, err := io.Copy(right, left)
535
- closeWrite(right)
534
+ if cw, ok := right.(closeWriter); ok {
535
+ _ = cw.CloseWrite()
536
+ }
537
return err
538
})
539
group.Go(func() error {
540
_, err := io.Copy(left, right)
540
- closeWrite(left)
541
+ if cw, ok := left.(closeWriter); ok {
542
+ _ = cw.CloseWrite()
543
+ }
544
return err
545
})
546
_ = group.Wait()
547
}
545
-
546
-func closeWrite(conn net.Conn) {
547
- type closeWriter interface {
548
- CloseWrite() error
549
- }
550
- if cw, ok := conn.(closeWriter); ok {
551
- _ = cw.CloseWrite()
552
- }
553
-}