@samitouri / QOSAMI-X402Tunnel / commits / 9603e429

Refactor Listener to listener and update related methods

Kim committed Apr 14, 2026 at 19:02 UTC 9603e429b1c10b7039a6937a4bf2b07dbc38ae4a
7 files changed +492 -521
cmd/portal-tunnel/README.md
+1 -1
@@ -125,7 +125,7 @@ Legacy execution compatibility has been removed:
125 - The tunnel starts once relay URLs pass local validation. Remote compatibility checks, lease registration, and reconnects continue in the background until each relay becomes ready.
126 - With discovery enabled, the tunnel uses the public registry as discovery seed input and can expand through relay discovery. Explicit `--relays` values are always included separately from the auto-selected relay pool. With `--discovery=false`, only the explicit relay URLs are used. Published public URLs appear only for relays that have registered successfully.
127 - Explicit `--relays` listeners retry indefinitely with `RetryCount=0`. Auto-selected discovery relays are created with `RetryCount=10` and are dropped from the active set after that budget is exhausted.
128 -- `ListenerConfig.RetryCount` limits retries when positive. `RetryCount=0` retries indefinitely.
128 +- Retry count limits retries when positive. `RetryCount=0` retries indefinitely.
129 - Tenant TLS is provisioned automatically through the relay keyless signer. The SDK fetches the relay certificate chain and uses `/v1/sign` for remote signing.
130 - `portal expose` enables MITM strict enforcement by default. Use `--ban-mitm=false` to keep warning-only behavior when the TLS self-probe suspects relay termination.
131 - When the local service is unreachable, the tunnel returns an HTTP 503 page.
sdk/api_client.go
+120 -182
@@ -11,9 +11,7 @@ import (
11 "io"
12 "net"
13 "net/http"
14 - "net/url"
14 "strings"
16 - "sync"
15 "time"
16
17 "github.com/quic-go/quic-go"
@@ -35,189 +33,145 @@ const (
33
34 var errRelayIncompatible = errors.New("relay is incompatible")
35
38 -type apiClient struct {
39 - mu sync.RWMutex
40 - baseURL *url.URL
41 - httpClient *http.Client
42 - rawTLSConfig *tls.Config
43 - dialTimeout time.Duration
44 - requestTimeout time.Duration
45 - identity types.Identity
46 - accessToken string
47 - expiresAt time.Time
48 - metadata types.LeaseMetadata
49 - resolvedPublicIP string
50 - sniPort int
51 -}
52 -
53 -func newApiClient(relayURL string, cfg ListenerConfig) (*apiClient, error) {
54 - normalizedRelayURL, err := utils.NormalizeRelayURL(relayURL)
55 - if err != nil {
56 - return nil, err
57 - }
58 -
59 - baseURL, err := url.Parse(normalizedRelayURL)
60 - if err != nil {
61 - return nil, fmt.Errorf("parse relay url: %w", err)
62 - }
63 -
64 - dialTimeout := utils.DurationOrDefault(cfg.DialTimeout, defaultDialTimeout)
65 - requestTimeout := utils.DurationOrDefault(cfg.RequestTimeout, defaultRequestTimeout)
66 -
67 - return &apiClient{
68 - baseURL: baseURL,
69 - dialTimeout: dialTimeout,
70 - requestTimeout: requestTimeout,
71 - identity: cfg.Identity.Copy(),
72 - metadata: cfg.Metadata.Copy(),
73 - }, nil
74 -}
75 -
76 -func closeIdleHTTPClient(httpClient *http.Client) {
77 - if httpClient == nil {
36 +func closeIdleHTTPClient(controlHTTPClient *http.Client) {
37 + if controlHTTPClient == nil {
38 return
39 }
80 - if transport, ok := httpClient.Transport.(*http.Transport); ok {
40 + if transport, ok := controlHTTPClient.Transport.(*http.Transport); ok {
41 transport.CloseIdleConnections()
42 }
43 }
44
85 -func (a *apiClient) close() {
86 - if a == nil {
87 - return
88 - }
89 - a.mu.RLock()
90 - httpClient := a.httpClient
91 - a.mu.RUnlock()
92 - closeIdleHTTPClient(httpClient)
93 -}
94 -
45 // resetTransport tears down the cached HTTP client and TLS config so the next
46 // API call creates fresh TCP connections. Call this after detecting a system
47 // sleep/wake cycle where pooled connections are almost certainly dead.
98 -func (a *apiClient) resetTransport() {
99 - if a == nil {
48 +func (l *listener) resetTransport() {
49 + if l == nil {
50 return
51 }
102 - a.mu.Lock()
103 - httpClient := a.httpClient
104 - a.httpClient = nil
105 - a.rawTLSConfig = nil
106 - a.mu.Unlock()
107 - closeIdleHTTPClient(httpClient)
52 + l.mu.Lock()
53 + controlHTTPClient := l.controlHTTPClient
54 + l.controlHTTPClient = nil
55 + l.controlTLSConfig = nil
56 + l.mu.Unlock()
57 + closeIdleHTTPClient(controlHTTPClient)
58 }
59
110 -func (a *apiClient) registerLease(ctx context.Context, ttl time.Duration, udpEnabled, tcpEnabled bool) (types.RegisterResponse, error) {
111 - if err := a.ensureHTTPClient(ctx); err != nil {
60 +func (l *listener) registerLease(ctx context.Context, ttl time.Duration, udpEnabled, tcpEnabled bool) (types.RegisterResponse, error) {
61 + if err := l.ensureControlHTTPClient(ctx); err != nil {
62 return types.RegisterResponse{}, err
63 }
114 - a.mu.RLock()
115 - httpClient := a.httpClient
116 - a.mu.RUnlock()
117 - if httpClient == nil {
64 + l.mu.Lock()
65 + controlHTTPClient := l.controlHTTPClient
66 + relayURL := l.relayURL
67 + identity := l.identity.Copy()
68 + metadata := l.metadata.Copy()
69 + l.mu.Unlock()
70 + if controlHTTPClient == nil {
71 return types.RegisterResponse{}, errors.New("relay http client is unavailable")
72 }
73
74 var challenge types.RegisterChallengeResponse
75 challengeReq := types.RegisterChallengeRequest{
123 - Identity: a.identity.Copy(),
124 - Metadata: a.metadata.Copy(),
76 + Identity: identity,
77 + Metadata: metadata,
78 TTL: int(ttl / time.Second),
79 UDPEnabled: udpEnabled,
80 TCPEnabled: tcpEnabled,
81 }
129 - if err := utils.HTTPDoAPIPath(ctx, httpClient, a.baseURL, http.MethodPost, types.PathSDKRegisterChallenge, challengeReq, nil, &challenge); err != nil {
82 + if err := utils.HTTPDoAPIPath(ctx, controlHTTPClient, relayURL, http.MethodPost, types.PathSDKRegisterChallenge, challengeReq, nil, &challenge); err != nil {
83 return types.RegisterResponse{}, err
84 }
85
133 - signature, err := utils.SignEthereumPersonalMessage(challenge.SIWEMessage, a.identity.PrivateKey)
86 + signature, err := utils.SignEthereumPersonalMessage(challenge.SIWEMessage, identity.PrivateKey)
87 if err != nil {
88 return types.RegisterResponse{}, err
89 }
90
91 var resp types.RegisterResponse
139 - if err := utils.HTTPDoAPIPath(ctx, httpClient, a.baseURL, http.MethodPost, types.PathSDKRegister, types.RegisterRequest{
92 + if err := utils.HTTPDoAPIPath(ctx, controlHTTPClient, relayURL, http.MethodPost, types.PathSDKRegister, types.RegisterRequest{
93 ChallengeID: challenge.ChallengeID,
94 SIWEMessage: challenge.SIWEMessage,
95 SIWESignature: signature,
143 - ReportedIP: a.reportedIP(ctx),
96 + ReportedIP: l.reportedIP(ctx),
97 }, nil, &resp); err != nil {
98 return types.RegisterResponse{}, err
99 }
147 - resp.AccessToken = strings.TrimSpace(resp.AccessToken)
148 - if resp.AccessToken == "" {
149 - return types.RegisterResponse{}, errors.New("relay did not return access token")
150 - }
151 - registeredIdentity, err := utils.NormalizeIdentity(resp.Identity)
152 - if err != nil {
153 - return types.RegisterResponse{}, err
154 - }
155 - if registeredIdentity.Key() != a.identity.Key() {
156 - return types.RegisterResponse{}, errors.New("relay returned mismatched lease identity")
157 - }
158 - resp.Identity = registeredIdentity
159 -
160 - sniPort := 0
161 - if udpEnabled {
162 - if resp.SNIPort <= 0 {
163 - return types.RegisterResponse{}, errors.New("relay did not return sni port for udp transport")
164 - }
165 - sniPort = resp.SNIPort
166 - }
167 -
168 - a.mu.Lock()
169 - a.accessToken = resp.AccessToken
170 - a.expiresAt = resp.ExpiresAt
171 - a.sniPort = sniPort
172 - a.mu.Unlock()
100 return resp, nil
101 }
102
176 -func (a *apiClient) ensureHTTPClient(ctx context.Context) error {
177 - a.mu.RLock()
178 - if a.httpClient != nil && a.rawTLSConfig != nil {
179 - a.mu.RUnlock()
103 +func (l *listener) ensureControlHTTPClient(ctx context.Context) error {
104 + if l == nil {
105 + return errors.New("listener is unavailable")
106 + }
107 + l.mu.Lock()
108 + if l.controlHTTPClient != nil && l.controlTLSConfig != nil {
109 + l.mu.Unlock()
110 return nil
111 }
182 - a.mu.RUnlock()
112 + relayURL := l.relayURL
113 + requestTimeout := l.requestTimeout
114 + l.mu.Unlock()
115 + if relayURL == nil {
116 + return errors.New("relay url is unavailable")
117 + }
118
119 bootstrapCtx, cancel := context.WithTimeout(ctx, defaultDialTimeout+defaultHandshakeTimeout)
120 defer cancel()
121
187 - rawTLSConfig, httpClient, err := utils.NewHTTPTLSClient(bootstrapCtx, a.baseURL, a.requestTimeout)
122 + controlTLSConfig, controlHTTPClient, err := utils.NewHTTPTLSClient(bootstrapCtx, relayURL, requestTimeout)
123 if err != nil {
124 return err
125 }
191 - if err := a.ensureCompatible(ctx, httpClient); err != nil {
192 - closeIdleHTTPClient(httpClient)
126 + if err := l.ensureCompatible(ctx, controlHTTPClient); err != nil {
127 + closeIdleHTTPClient(controlHTTPClient)
128 return err
129 }
130
196 - a.mu.Lock()
197 - if a.httpClient != nil && a.rawTLSConfig != nil {
198 - a.mu.Unlock()
199 - closeIdleHTTPClient(httpClient)
131 + l.mu.Lock()
132 + if l.controlHTTPClient != nil && l.controlTLSConfig != nil {
133 + l.mu.Unlock()
134 + closeIdleHTTPClient(controlHTTPClient)
135 return nil
136 }
202 - oldHTTPClient := a.httpClient
203 - a.httpClient = httpClient
204 - a.rawTLSConfig = rawTLSConfig
205 - a.mu.Unlock()
206 - closeIdleHTTPClient(oldHTTPClient)
137 + oldControlHTTPClient := l.controlHTTPClient
138 + l.controlHTTPClient = controlHTTPClient
139 + l.controlTLSConfig = controlTLSConfig
140 + l.mu.Unlock()
141 + closeIdleHTTPClient(oldControlHTTPClient)
142
143 return nil
144 }
145
211 -func (a *apiClient) reportedIP(ctx context.Context) string {
212 - if a.resolvedPublicIP == "" {
213 - a.resolvedPublicIP = utils.ResolvePublicIP(ctx)
146 +func (l *listener) reportedIP(ctx context.Context) string {
147 + l.mu.Lock()
148 + if l.resolvedPublicIP != "" {
149 + ip := l.resolvedPublicIP
150 + l.mu.Unlock()
151 + return ip
152 + }
153 + l.mu.Unlock()
154 +
155 + ip := utils.ResolvePublicIP(ctx)
156 + l.mu.Lock()
157 + if l.resolvedPublicIP == "" {
158 + l.resolvedPublicIP = ip
159 }
215 - return a.resolvedPublicIP
160 + ip = l.resolvedPublicIP
161 + l.mu.Unlock()
162 + return ip
163 }
164
218 -func (a *apiClient) ensureCompatible(ctx context.Context, httpClient *http.Client) error {
165 +func (l *listener) ensureCompatible(ctx context.Context, controlHTTPClient *http.Client) error {
166 + l.mu.Lock()
167 + relayURL := l.relayURL
168 + l.mu.Unlock()
169 + if relayURL == nil {
170 + return errors.New("relay url is unavailable")
171 + }
172 +
173 var resp types.DomainResponse
220 - if err := utils.HTTPDoAPIPath(ctx, httpClient, a.baseURL, http.MethodGet, types.PathSDKDomain, nil, nil, &resp); err != nil {
174 + if err := utils.HTTPDoAPIPath(ctx, controlHTTPClient, relayURL, http.MethodGet, types.PathSDKDomain, nil, nil, &resp); err != nil {
175 err = fmt.Errorf("check relay compatibility: %w", err)
176 var netErr net.Error
177 var apiErr *types.APIRequestError
@@ -236,86 +190,73 @@ func (a *apiClient) ensureCompatible(ctx context.Context, httpClient *http.Clien
190 return nil
191 }
192
239 -func (a *apiClient) renewLease(ctx context.Context, ttl time.Duration) error {
240 - if err := a.ensureHTTPClient(ctx); err != nil {
241 - return err
193 +func (l *listener) renewRegisteredLease(ctx context.Context, ttl time.Duration, accessToken string) (types.RenewResponse, error) {
194 + if err := l.ensureControlHTTPClient(ctx); err != nil {
195 + return types.RenewResponse{}, err
196 }
197
244 - a.mu.RLock()
245 - httpClient := a.httpClient
246 - accessToken := a.accessToken
247 - a.mu.RUnlock()
248 - if httpClient == nil {
249 - return errors.New("relay http client is unavailable")
250 - }
251 - if strings.TrimSpace(accessToken) == "" {
252 - return errors.New("access token is not available")
198 + l.mu.Lock()
199 + controlHTTPClient := l.controlHTTPClient
200 + relayURL := l.relayURL
201 + l.mu.Unlock()
202 + if controlHTTPClient == nil {
203 + return types.RenewResponse{}, errors.New("relay http client is unavailable")
204 }
205
206 var resp types.RenewResponse
256 - if err := utils.HTTPDoAPIPath(ctx, httpClient, a.baseURL, http.MethodPost, types.PathSDKRenew, types.RenewRequest{
207 + if err := utils.HTTPDoAPIPath(ctx, controlHTTPClient, relayURL, http.MethodPost, types.PathSDKRenew, types.RenewRequest{
208 AccessToken: accessToken,
209 TTL: int(ttl / time.Second),
259 - ReportedIP: a.reportedIP(ctx),
210 + ReportedIP: l.reportedIP(ctx),
211 }, nil, &resp); err != nil {
261 - return err
262 - }
263 - resp.AccessToken = strings.TrimSpace(resp.AccessToken)
264 - if resp.AccessToken == "" {
265 - return errors.New("relay did not return renewed access token")
212 + return types.RenewResponse{}, err
213 }
267 -
268 - a.mu.Lock()
269 - if a.accessToken == accessToken {
270 - a.accessToken = resp.AccessToken
271 - a.expiresAt = resp.ExpiresAt
272 - }
273 - a.mu.Unlock()
274 - return nil
214 + return resp, nil
215 }
216
277 -func (a *apiClient) unregisterLease(ctx context.Context) error {
278 - if err := a.ensureHTTPClient(ctx); err != nil {
217 +func (l *listener) unregisterLease(ctx context.Context, accessToken string) error {
218 + if err := l.ensureControlHTTPClient(ctx); err != nil {
219 return err
220 }
281 - a.mu.RLock()
282 - httpClient := a.httpClient
283 - accessToken := a.accessToken
284 - a.mu.RUnlock()
285 - if httpClient == nil {
221 + l.mu.Lock()
222 + controlHTTPClient := l.controlHTTPClient
223 + relayURL := l.relayURL
224 + l.mu.Unlock()
225 + if controlHTTPClient == nil {
226 return errors.New("relay http client is unavailable")
227 }
288 - return utils.HTTPDoAPIPath(ctx, httpClient, a.baseURL, http.MethodPost, types.PathSDKUnregister, types.UnregisterRequest{
228 + return utils.HTTPDoAPIPath(ctx, controlHTTPClient, relayURL, http.MethodPost, types.PathSDKUnregister, types.UnregisterRequest{
229 AccessToken: accessToken,
230 }, nil, nil)
231 }
232
293 -func (a *apiClient) openReverseSession(ctx context.Context) (net.Conn, error) {
294 - if err := a.ensureHTTPClient(ctx); err != nil {
233 +func (l *listener) openReverseSession(ctx context.Context, accessToken string) (net.Conn, error) {
234 + if err := l.ensureControlHTTPClient(ctx); err != nil {
235 return nil, err
236 }
297 - a.mu.RLock()
298 - rawTLSConfig := a.rawTLSConfig
299 - accessToken := a.accessToken
300 - a.mu.RUnlock()
301 - if rawTLSConfig == nil {
237 + l.mu.Lock()
238 + controlTLSConfig := l.controlTLSConfig
239 + relayURL := l.relayURL
240 + dialTimeout := l.dialTimeout
241 + l.mu.Unlock()
242 + if controlTLSConfig == nil {
243 return nil, errors.New("relay tls config is unavailable")
244 }
245
246 dialer := &tls.Dialer{
306 - NetDialer: &net.Dialer{Timeout: a.dialTimeout},
307 - Config: rawTLSConfig.Clone(),
247 + NetDialer: &net.Dialer{Timeout: dialTimeout},
248 + Config: controlTLSConfig.Clone(),
249 }
250
310 - conn, err := dialer.DialContext(ctx, "tcp", utils.EnsurePort(a.baseURL.Host))
251 + conn, err := dialer.DialContext(ctx, "tcp", utils.EnsurePort(relayURL.Host))
252 if err != nil {
253 return nil, err
254 }
255
256 req := &http.Request{
257 Method: http.MethodGet,
317 - URL: utils.ResolveAPIURL(a.baseURL, types.PathSDKConnect),
318 - Host: a.baseURL.Host,
258 + URL: utils.ResolveAPIURL(relayURL, types.PathSDKConnect),
259 + Host: relayURL.Host,
260 Header: make(http.Header),
261 }
262 req.Header.Set(types.HeaderAccessToken, accessToken)
@@ -367,20 +308,20 @@ func (c *bufferedConn) Read(p []byte) (int, error) {
308 }
309
310 // openQUICSession opens a QUIC connection to the relay for datagram transport.
370 -func (a *apiClient) openQUICSession(ctx context.Context, accessToken string) (*quic.Conn, error) {
371 - if err := a.ensureHTTPClient(ctx); err != nil {
311 +func (l *listener) openQUICSession(ctx context.Context, accessToken string, sniPort int) (*quic.Conn, error) {
312 + if err := l.ensureControlHTTPClient(ctx); err != nil {
313 return nil, err
314 }
315
375 - a.mu.RLock()
376 - rawTLSConfig := a.rawTLSConfig
377 - sniPort := a.sniPort
378 - a.mu.RUnlock()
379 - if rawTLSConfig == nil {
316 + l.mu.Lock()
317 + controlTLSConfig := l.controlTLSConfig
318 + relayURL := l.relayURL
319 + l.mu.Unlock()
320 + if controlTLSConfig == nil {
321 return nil, errors.New("relay tls config is unavailable")
322 }
323
383 - tlsConf := rawTLSConfig.Clone()
324 + tlsConf := controlTLSConfig.Clone()
325 tlsConf.NextProtos = []string{"portal-tunnel"}
326
327 quicConf := &quic.Config{
@@ -389,12 +330,9 @@ func (a *apiClient) openQUICSession(ctx context.Context, accessToken string) (*q
330 MaxIdleTimeout: 60 * time.Second,
331 }
332
392 - if sniPort <= 0 {
393 - return nil, errors.New("sni port is not available")
394 - }
395 - host := strings.TrimSpace(a.baseURL.Hostname())
333 + host := strings.TrimSpace(relayURL.Hostname())
334 if host == "" {
397 - host = strings.TrimSpace(a.baseURL.Host)
335 + host = strings.TrimSpace(relayURL.Host)
336 }
337 dialAddr := net.JoinHostPort(host, fmt.Sprintf("%d", sniPort))
338 conn, err := quic.DialAddr(ctx, dialAddr, tlsConf, quicConf)
sdk/expose.go
+14 -11
@@ -44,7 +44,7 @@ type Exposure struct {
44
45 relaySet *discovery.RelaySet
46 listenerMu sync.RWMutex
47 - relayListeners map[string]*Listener
47 + relayListeners map[string]*listener
48
49 closeOnce sync.Once
50 connSeq atomic.Uint64
@@ -119,7 +119,7 @@ func Expose(ctx context.Context, cfg ExposeConfig) (*Exposure, error) {
119 accepted: make(chan net.Conn, max(len(relayURLs)*defaultReadyTarget*2, 1)),
120 datagrams: make(chan types.DatagramFrame, max(len(relayURLs)*32, 1)),
121 relaySet: discovery.NewRelaySet(relayURLs),
122 - relayListeners: make(map[string]*Listener, len(relayURLs)),
122 + relayListeners: make(map[string]*listener, len(relayURLs)),
123 }
124
125 if len(relayURLs) > 0 {
@@ -187,7 +187,7 @@ func (e *Exposure) SendDatagram(frame types.DatagramFrame) error {
187 if listener == nil {
188 return net.ErrClosed
189 }
190 - return listener.SendDatagram(frame)
190 + return listener.sendDatagram(frame)
191 }
192
193 func (e *Exposure) WaitDatagramReady(ctx context.Context) ([]string, error) {
@@ -208,7 +208,7 @@ func (e *Exposure) WaitDatagramReady(ctx context.Context) ([]string, error) {
208 continue
209 }
210
211 - udpAddr, ready, pending := listener.DatagramReady()
211 + udpAddr, ready, pending := listener.datagramReady()
212 if ready {
213 if _, ok := seen[udpAddr]; !ok {
214 seen[udpAddr] = struct{}{}
@@ -318,7 +318,7 @@ func (e *Exposure) Close() error {
318
319 e.listenerMu.Lock()
320 relayListeners := e.relayListeners
321 - e.relayListeners = make(map[string]*Listener)
321 + e.relayListeners = make(map[string]*listener)
322 e.listenerMu.Unlock()
323
324 relayURLs := make([]string, 0, len(relayListeners))
@@ -376,7 +376,7 @@ func (e *Exposure) reconcileRelayListeners(failOnError bool) error {
376 desiredRelayURLs := e.relaySet.PriorityRelays(clientState)
377
378 e.listenerMu.Lock()
379 - staleRelayListeners := make(map[string]*Listener)
379 + staleRelayListeners := make(map[string]*listener)
380 removedRelayURLs := make([]string, 0)
381 for relayURL, listener := range e.relayListeners {
382 if slices.Contains(desiredRelayURLs, relayURL) {
@@ -413,7 +413,7 @@ func (e *Exposure) reconcileRelayListeners(failOnError bool) error {
413 if slices.Contains(e.explicitRelays, relayURL) {
414 retryCount = 0
415 }
416 - listener, err := NewListener(context.Background(), relayURL, ListenerConfig{
416 + listener, err := newListener(context.Background(), relayURL, listenerConfig{
417 Identity: e.identity.Copy(),
418 UDPEnabled: e.udpEnabled,
419 TCPEnabled: e.tcpEnabled,
@@ -460,16 +460,19 @@ func (e *Exposure) reconcileRelayListeners(failOnError bool) error {
460 return nil
461 }
462
463 -func (e *Exposure) runListenerAcceptLoop(listener *Listener) {
463 +func (e *Exposure) runListenerAcceptLoop(listener *listener) {
464 if listener == nil {
465 return
466 }
467
468 - relayURL := listener.api.baseURL.String()
468 + relayURL := ""
469 + if listener.relayURL != nil {
470 + relayURL = listener.relayURL.String()
471 + }
472 if e.udpEnabled {
473 go func() {
474 for {
472 - frame, err := listener.AcceptDatagram()
475 + frame, err := listener.acceptDatagram()
476 if err != nil {
477 select {
478 case <-e.done:
@@ -482,7 +485,7 @@ func (e *Exposure) runListenerAcceptLoop(listener *Listener) {
485 log.Warn().
486 Err(err).
487 Str("relay_url", relayURL).
485 - Str("address", listener.Address()).
488 + Str("address", listener.address()).
489 Msg("datagram accept failed")
490 return
491 }
sdk/expose_test.go
+12 -12
@@ -29,17 +29,17 @@ func TestExposureReconcileRemovesBannedRelayFromActiveSet(t *testing.T) {
29
30 exposure := &Exposure{
31 relaySet: mustRelaySet(t, relayA, relayB),
32 - relayListeners: make(map[string]*Listener, 2),
32 + relayListeners: make(map[string]*listener, 2),
33 }
34 relayAClosed := make(chan struct{})
35 - exposure.relayListeners = map[string]*Listener{
35 + exposure.relayListeners = map[string]*listener{
36 relayA: {
37 - api: &apiClient{baseURL: relayURL},
38 - cancel: func() { close(relayAClosed) },
39 - doneCh: relayAClosed,
37 + relayURL: relayURL,
38 + cancel: func() { close(relayAClosed) },
39 + doneCh: relayAClosed,
40 },
41 relayB: {
42 - api: &apiClient{baseURL: relayBURL},
42 + relayURL: relayBURL,
43 },
44 }
45
@@ -84,16 +84,16 @@ func TestExposureReconcileRemovesStaleListener(t *testing.T) {
84 relayAClosed := make(chan struct{})
85 exposure := &Exposure{
86 relaySet: mustRelaySet(t, relayA, relayB),
87 - relayListeners: make(map[string]*Listener, 2),
87 + relayListeners: make(map[string]*listener, 2),
88 }
89 - exposure.relayListeners = map[string]*Listener{
89 + exposure.relayListeners = map[string]*listener{
90 relayA: {
91 - api: &apiClient{baseURL: relayAURL},
92 - cancel: func() { close(relayAClosed) },
93 - doneCh: relayAClosed,
91 + relayURL: relayAURL,
92 + cancel: func() { close(relayAClosed) },
93 + doneCh: relayAClosed,
94 },
95 relayB: {
96 - api: &apiClient{baseURL: relayBURL},
96 + relayURL: relayBURL,
97 },
98 }
99
sdk/listener.go
+296 -275
@@ -4,8 +4,10 @@ import (
4 "context"
5 "crypto/tls"
6 "errors"
7 + "fmt"
8 "io"
9 "net"
10 + "net/http"
11 "net/url"
12 "strings"
13 "sync"
@@ -20,7 +22,7 @@ import (
22 "github.com/gosuda/portal-tunnel/v2/utils"
23 )
24
23 -type ListenerConfig struct {
25 +type listenerConfig struct {
26 Identity types.Identity
27 UDPEnabled bool
28 TCPEnabled bool
@@ -37,8 +39,19 @@ type ListenerConfig struct {
39 relaySet *discovery.RelaySet
40 }
41
40 -type Listener struct {
41 - api *apiClient
42 +var errLeaseRefreshRequired = errors.New("lease refresh required")
43 +
44 +type listener struct {
45 + relayURL *url.URL
46 + controlHTTPClient *http.Client
47 + controlTLSConfig *tls.Config
48 + dialTimeout time.Duration
49 + requestTimeout time.Duration
50 + resolvedPublicIP string
51 + accessToken string
52 + expiresAt time.Time
53 + sniPort int
54 +
55 cancel context.CancelFunc
56 doneCh <-chan struct{}
57
@@ -52,10 +65,7 @@ type Listener struct {
65 datagram *transport.ClientDatagram
66 mitmManager *mitmManager
67
55 - registered chan struct{}
56 - closeOnce sync.Once
57 - registerOnce sync.Once
58 - streamCancel context.CancelFunc
68 + closeOnce sync.Once
69
70 banMITM bool
71 tcpEnabled bool
@@ -65,41 +75,50 @@ type Listener struct {
75 hostname string
76 udpAddr string
77 metadata types.LeaseMetadata
68 - tlsConfig *tls.Config
69 - tlsCloser io.Closer
78 +
79 + tenantTLSConfig *tls.Config
80 + tenantTLSCloser io.Closer
81 }
82
72 -// NewListener creates one relay listener and its dedicated relay transport for one relay URL.
83 +// newListener creates one relay listener and its dedicated relay transport for one relay URL.
84 // Only local config validation fails immediately; relay startup runs in the background until ready.
74 -func NewListener(ctx context.Context, relayURL string, cfg ListenerConfig) (*Listener, error) {
85 +func newListener(ctx context.Context, relayURL string, cfg listenerConfig) (*listener, error) {
86 listenerCtx, cancel := context.WithCancel(ctx)
87 readyTarget := utils.IntOrDefault(cfg.ReadyTarget, defaultReadyTarget)
88 leaseTTL := utils.DurationOrDefault(cfg.LeaseTTL, defaultLeaseTTL)
89 + dialTimeout := utils.DurationOrDefault(cfg.DialTimeout, defaultDialTimeout)
90 + requestTimeout := utils.DurationOrDefault(cfg.RequestTimeout, defaultRequestTimeout)
91 handshakeTimeout := utils.DurationOrDefault(cfg.HandshakeTimeout, defaultHandshakeTimeout)
92 renewBefore := utils.DurationOrDefault(cfg.RenewBefore, defaultRenewBefore)
93 retryWait := utils.DurationOrDefault(cfg.RetryWait, defaultRetryWait)
94
82 - api, err := newApiClient(relayURL, cfg)
95 + normalizedRelayURL, err := utils.NormalizeRelayURL(relayURL)
96 if err != nil {
97 cancel()
98 return nil, err
99 }
87 -
88 - l := &Listener{
89 - doneCh: listenerCtx.Done(),
90 - cancel: cancel,
91 - api: api,
92 - registered: make(chan struct{}),
93 - readyTarget: readyTarget,
94 - retryCount: cfg.RetryCount,
95 - retryWait: retryWait,
96 - leaseTTL: leaseTTL,
97 - renewBefore: renewBefore,
98 - identity: api.identity.Copy(),
99 - metadata: cfg.Metadata.Copy(),
100 - banMITM: cfg.BanMITM,
101 - tcpEnabled: cfg.TCPEnabled,
102 - relaySet: cfg.relaySet,
100 + relayurl, err := url.Parse(normalizedRelayURL)
101 + if err != nil {
102 + cancel()
103 + return nil, fmt.Errorf("parse relay url: %w", err)
104 + }
105 +
106 + l := &listener{
107 + doneCh: listenerCtx.Done(),
108 + cancel: cancel,
109 + relayURL: relayurl,
110 + dialTimeout: dialTimeout,
111 + requestTimeout: requestTimeout,
112 + readyTarget: readyTarget,
113 + retryCount: cfg.RetryCount,
114 + retryWait: retryWait,
115 + leaseTTL: leaseTTL,
116 + renewBefore: renewBefore,
117 + identity: cfg.Identity.Copy(),
118 + metadata: cfg.Metadata.Copy(),
119 + banMITM: cfg.BanMITM,
120 + tcpEnabled: cfg.TCPEnabled,
121 + relaySet: cfg.relaySet,
122 }
123 l.mitmManager = newMITMManager(listenerCtx, l)
124 l.stream = transport.NewClientStream(readyTarget, handshakeTimeout)
@@ -108,36 +127,22 @@ func NewListener(ctx context.Context, relayURL string, cfg ListenerConfig) (*Lis
127 log.Info().
128 Err(err).
129 Str("component", "sdk-datagram-plane").
111 - Str("address", l.Address()).
130 + Str("address", l.address()).
131 Msg("quic datagram plane disconnected; waiting to reconnect")
132 })
133 }
134
116 - go l.runStartup(listenerCtx)
135 + go l.run(listenerCtx)
136 return l, nil
137 }
138
120 -func (l *Listener) runStartup(ctx context.Context) {
139 +func (l *listener) run(ctx context.Context) {
140 var retries int
141
142 for {
143 err := l.registerAndConfigure(ctx)
144 switch {
145 case err == nil:
127 - l.startStreamLoops(ctx)
128 - if l.datagram != nil {
129 - go l.runDatagramLoop(ctx)
130 - }
131 - go l.runRenewLoop(ctx)
132 - publicURL := l.PublicURL()
133 - event := log.Info().Str("address", l.Address())
134 - if publicURL != "" {
135 - event.
136 - Msg("service ready at " + publicURL)
137 - return
138 - }
139 - event.Msg("relay listener registered")
140 - return
146 case errors.Is(err, context.Canceled), errors.Is(err, net.ErrClosed):
147 return
148 default:
@@ -146,55 +151,75 @@ func (l *Listener) runStartup(ctx context.Context) {
151 errors.Is(err, &types.APIRequestError{Code: types.APIErrorCodeTransportMismatch}) ||
152 errors.Is(err, &types.APIRequestError{Code: types.APIErrorCodeHostnameConflict}) ||
153 errors.Is(err, &types.APIRequestError{Code: types.APIErrorCodeIPBanned}) {
149 - if l.relaySet != nil && l.api != nil && l.api.baseURL != nil {
150 - l.relaySet.UnconfirmRelayURL(l.api.baseURL.String())
151 - l.relaySet.RecordRelayFailure(l.api.baseURL.String(), err, 1)
154 + relayURL := l.relayURL.String()
155 + if l.relaySet != nil && relayURL != "" {
156 + l.relaySet.UnconfirmRelayURL(relayURL)
157 + l.relaySet.RecordRelayFailure(relayURL, err, 1)
158 }
159 log.Error().
160 Err(err).
155 - Str("relay_url", l.api.baseURL.String()).
156 - Str("address", l.Address()).
161 + Str("relay_url", relayURL).
162 + Str("address", l.address()).
163 Msg("lease registration failed; closing listener")
164 _ = l.Close()
165 return
166 }
167 retries++
162 - if !l.retryOrClose(ctx, "lease registration", err, retries) {
168 + if !l.waitRetry(ctx, "lease registration", err, retries) {
169 + _ = l.Close()
170 return
171 }
172 + continue
173 + }
174 +
175 + retries = 0
176 + publicURL := l.publicURL()
177 + event := log.Info().Str("address", l.address())
178 + if publicURL != "" {
179 + event.Msg("service ready at " + publicURL)
180 + } else {
181 + event.Msg("relay listener registered")
182 + }
183 +
184 + err = l.runLease(ctx)
185 + if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, net.ErrClosed) {
186 + return
187 + }
188 +
189 + if errors.Is(err, errLeaseRefreshRequired) {
190 + _, _, _, tenantTLSCloser := l.clearLease("lease refresh required")
191 + if tenantTLSCloser != nil {
192 + _ = tenantTLSCloser.Close()
193 + }
194 + l.resetTransport()
195 + continue
196 }
197 +
198 + relayURL := l.relayURL.String()
199 + log.Error().
200 + Err(err).
201 + Str("relay_url", relayURL).
202 + Str("address", l.address()).
203 + Msg("listener connection retry budget exhausted; closing listener")
204 + _ = l.Close()
205 + return
206 }
207 }
208
169 -func (l *Listener) Close() error {
209 +func (l *listener) Close() error {
210 var closeErr error
211 l.closeOnce.Do(func() {
212 if l.cancel != nil {
213 l.cancel()
214 }
215
216 + identity, registered, accessToken, tenantTLSCloser := l.clearLease("")
217 +
218 l.mu.Lock()
177 - identity := l.identity.Copy()
178 - registered := l.hostname != ""
179 - tlsCloser := l.tlsCloser
180 - streamCancel := l.streamCancel
219 stream := l.stream
220 datagram := l.datagram
183 - api := l.api
184 - l.hostname = ""
185 - l.udpAddr = ""
186 - l.tlsConfig = nil
187 - l.tlsCloser = nil
188 - l.streamCancel = nil
221 l.mu.Unlock()
222
191 - if l.mitmManager != nil {
192 - l.mitmManager.reset()
193 - }
194 - if streamCancel != nil {
195 - streamCancel()
196 - }
197 -
223 if stream != nil {
224 stream.Drain()
225 }
@@ -202,22 +227,45 @@ func (l *Listener) Close() error {
227 datagram.Close()
228 }
229
205 - if api != nil && registered && identity.Key() != "" {
230 + if registered && identity.Key() != "" && strings.TrimSpace(accessToken) != "" {
231 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
207 - closeErr = errors.Join(closeErr, api.unregisterLease(ctx))
232 + closeErr = errors.Join(closeErr, l.unregisterLease(ctx, accessToken))
233 cancel()
234 }
210 - if tlsCloser != nil {
211 - closeErr = errors.Join(closeErr, tlsCloser.Close())
212 - }
213 - if api != nil {
214 - api.close()
235 + if tenantTLSCloser != nil {
236 + closeErr = errors.Join(closeErr, tenantTLSCloser.Close())
237 }
238 + l.resetTransport()
239 })
240 return closeErr
241 }
242
220 -func (l *Listener) Accept() (net.Conn, error) {
243 +func (l *listener) clearLease(reason string) (types.Identity, bool, string, io.Closer) {
244 + l.mu.Lock()
245 + identity := l.identity.Copy()
246 + registered := l.hostname != ""
247 + accessToken := l.accessToken
248 + tenantTLSCloser := l.tenantTLSCloser
249 + datagram := l.datagram
250 + l.hostname = ""
251 + l.udpAddr = ""
252 + l.tenantTLSConfig = nil
253 + l.tenantTLSCloser = nil
254 + l.accessToken = ""
255 + l.expiresAt = time.Time{}
256 + l.sniPort = 0
257 + l.mu.Unlock()
258 +
259 + if l.mitmManager != nil {
260 + l.mitmManager.reset()
261 + }
262 + if datagram != nil && reason != "" {
263 + datagram.Clear(reason)
264 + }
265 + return identity, registered, accessToken, tenantTLSCloser
266 +}
267 +
268 +func (l *listener) Accept() (net.Conn, error) {
269 if l.stream == nil {
270 return nil, net.ErrClosed
271 }
@@ -231,8 +279,8 @@ func (l *Listener) Accept() (net.Conn, error) {
279 if handleErr != nil {
280 log.Debug().
281 Err(handleErr).
234 - Str("relay_url", l.api.baseURL.String()).
235 - Str("address", l.Address()).
282 + Str("relay_url", l.relayURL.String()).
283 + Str("address", l.address()).
284 Msg("mitm self-probe handling failed")
285 }
286 if handled {
@@ -242,7 +290,7 @@ func (l *Listener) Accept() (net.Conn, error) {
290 }
291 }
292
245 -func (l *Listener) AcceptDatagram() (types.DatagramFrame, error) {
293 +func (l *listener) acceptDatagram() (types.DatagramFrame, error) {
294 if l == nil || l.datagram == nil {
295 return types.DatagramFrame{}, net.ErrClosed
296 }
@@ -256,14 +304,14 @@ func (l *Listener) AcceptDatagram() (types.DatagramFrame, error) {
304 l.mu.Lock()
305 frame.Address = l.identity.Address
306 frame.UDPAddr = l.udpAddr
259 - if l.api != nil && l.api.baseURL != nil {
260 - frame.RelayURL = l.api.baseURL.String()
307 + if l.relayURL != nil {
308 + frame.RelayURL = l.relayURL.String()
309 }
310 l.mu.Unlock()
311 return frame, nil
312 }
313
266 -func (l *Listener) SendDatagram(frame types.DatagramFrame) error {
314 +func (l *listener) sendDatagram(frame types.DatagramFrame) error {
315 if l == nil || l.datagram == nil {
316 return net.ErrClosed
317 }
@@ -282,121 +330,104 @@ func (l *Listener) SendDatagram(frame types.DatagramFrame) error {
330 return datagram.Send(frame.FlowID, frame.Payload)
331 }
332
285 -func (l *Listener) DatagramReady() (string, bool, bool) {
333 +func (l *listener) datagramReady() (string, bool, bool) {
334 if l == nil || l.datagram == nil {
335 return "", false, false
336 }
337
338 l.mu.Lock()
339 + hostname := l.hostname
340 udpAddr := l.udpAddr
341 datagram := l.datagram
342 l.mu.Unlock()
343
344 ready := datagram != nil && datagram.Connected() && udpAddr != ""
296 - select {
297 - case <-l.registered:
298 - return udpAddr, ready, udpAddr != "" && !ready
299 - default:
300 - return udpAddr, ready, !l.closed()
301 - }
302 -}
303 -
304 -func (l *Listener) Addr() net.Addr {
305 - l.mu.Lock()
306 - defer l.mu.Unlock()
307 - if l.identity.Address == "" {
308 - return listenerAddr("portal:closed")
309 - }
310 - return listenerAddr("portal:" + l.identity.Address)
345 + pending := !ready && !l.closed() && (hostname == "" || udpAddr != "")
346 + return udpAddr, ready, pending
347 }
348
313 -func (l *Listener) Address() string {
349 +func (l *listener) address() string {
350 l.mu.Lock()
351 defer l.mu.Unlock()
352 return l.identity.Address
353 }
354
319 -func (l *Listener) Hostname() string {
320 - l.mu.Lock()
321 - defer l.mu.Unlock()
322 - return l.hostname
323 -}
324 -
325 -func (l *Listener) Metadata() types.LeaseMetadata {
326 - l.mu.Lock()
327 - defer l.mu.Unlock()
328 - return l.metadata.Copy()
329 -}
330 -
331 -func (l *Listener) Identity() types.Identity {
332 - if l == nil {
333 - return types.Identity{}
334 - }
335 - return l.identity.Copy()
336 -}
337 -
338 -func (l *Listener) PublicURL() string {
339 - if l == nil || l.api == nil || l.api.baseURL == nil {
355 +func (l *listener) publicURL() string {
356 + if l == nil || l.relayURL == nil {
357 return ""
358 }
359
360 l.mu.Lock()
361 hostname := l.hostname
362 l.mu.Unlock()
346 -
363 if hostname == "" {
364 return ""
365 }
366
351 - if l.api.baseURL.Scheme == "" {
367 + if l.relayURL.Scheme == "" {
368 return "https://" + hostname
369 }
370
371 host := hostname
356 - if port := l.api.baseURL.Port(); port != "" {
372 + if port := l.relayURL.Port(); port != "" {
373 host = net.JoinHostPort(hostname, port)
374 }
375
376 return (&url.URL{
361 - Scheme: l.api.baseURL.Scheme,
377 + Scheme: l.relayURL.Scheme,
378 Host: host,
379 }).String()
380 }
381
366 -func (l *Listener) startStreamLoops(parentCtx context.Context) {
367 - if l.stream == nil || l.readyTarget <= 0 {
368 - return
369 - }
370 -
371 - streamCtx, cancel := context.WithCancel(parentCtx)
372 -
382 +func (l *listener) runLease(ctx context.Context) error {
383 l.mu.Lock()
374 - if l.streamCancel != nil {
375 - l.streamCancel()
376 - }
377 - l.streamCancel = cancel
384 + identity := l.identity.Copy()
385 + accessToken := l.accessToken
386 + sniPort := l.sniPort
387 + tlsConfig := l.tenantTLSConfig
388 readyTarget := l.readyTarget
389 l.mu.Unlock()
390
381 - for range readyTarget {
382 - go l.runReverseSessionLoop(streamCtx)
383 - }
384 -}
391 + leaseCtx, cancel := context.WithCancel(ctx)
392 + defer cancel()
393
386 -func (l *Listener) stopStreamLoops() {
387 - l.mu.Lock()
388 - cancel := l.streamCancel
389 - l.streamCancel = nil
390 - l.mu.Unlock()
394 + errCh := make(chan error, max(readyTarget, 1)+1)
395 + if l.stream != nil && readyTarget > 0 {
396 + for range readyTarget {
397 + go func() {
398 + if err := l.runReverseSessionLoop(leaseCtx, accessToken, tlsConfig); err != nil {
399 + select {
400 + case errCh <- err:
401 + case <-leaseCtx.Done():
402 + }
403 + }
404 + }()
405 + }
406 + }
407 + if l.datagram != nil {
408 + go l.runDatagramLoop(leaseCtx, identity, accessToken, sniPort)
409 + }
410 + go func() {
411 + if err := l.runRenewLoop(leaseCtx); err != nil {
412 + select {
413 + case errCh <- err:
414 + case <-leaseCtx.Done():
415 + }
416 + }
417 + }()
418
392 - if cancel != nil {
419 + select {
420 + case <-ctx.Done():
421 + return ctx.Err()
422 + case err := <-errCh:
423 cancel()
424 + return err
425 }
426 }
427
397 -func (l *Listener) runReverseSessionLoop(ctx context.Context) {
428 +func (l *listener) runReverseSessionLoop(ctx context.Context, accessToken string, tlsConfig *tls.Config) error {
429 if l.stream == nil {
399 - return
430 + return nil
431 }
432
433 var retries int
@@ -404,31 +435,32 @@ func (l *Listener) runReverseSessionLoop(ctx context.Context) {
435 claimed, err := l.stream.RunSession(
436 ctx,
437 func(ctx context.Context) (net.Conn, error) {
407 - return l.api.openReverseSession(ctx)
438 + if strings.TrimSpace(accessToken) == "" {
439 + return nil, errors.New("access token is not available")
440 + }
441 + return l.openReverseSession(ctx, accessToken)
442 },
443 func() *tls.Config {
410 - l.mu.Lock()
411 - defer l.mu.Unlock()
412 - return l.tlsConfig
444 + return tlsConfig
445 },
446 )
447 switch {
448 case err == nil:
449 retries = 0
450 case errors.Is(err, context.Canceled), errors.Is(err, net.ErrClosed):
419 - return
451 + return nil
452 case claimed:
453 retries = 0
454 default:
455 retries++
424 - if !l.retryOrClose(ctx, "reverse session connect", err, retries) {
425 - return
456 + if !l.waitRetry(ctx, "reverse session connect", err, retries) {
457 + return err
458 }
459 }
460 }
461 }
462
431 -func (l *Listener) runDatagramLoop(ctx context.Context) {
463 +func (l *listener) runDatagramLoop(ctx context.Context, identity types.Identity, accessToken string, sniPort int) {
464 if l.datagram == nil {
465 return
466 }
@@ -436,35 +468,12 @@ func (l *Listener) runDatagramLoop(ctx context.Context) {
468 for {
469 select {
470 case <-ctx.Done():
439 - l.datagram.Close()
471 + l.datagram.Clear("lease stopped")
472 return
473 default:
474 }
475
444 - l.mu.Lock()
445 - api := l.api
446 - identity := l.identity.Copy()
447 - udpAddr := l.udpAddr
448 - l.mu.Unlock()
449 - if api == nil || identity.Key() == "" || udpAddr == "" {
450 - if !utils.SleepOrDone(ctx, time.Second) {
451 - l.datagram.Close()
452 - return
453 - }
454 - continue
455 - }
456 - api.mu.RLock()
457 - accessToken := api.accessToken
458 - api.mu.RUnlock()
459 - if strings.TrimSpace(accessToken) == "" {
460 - if !utils.SleepOrDone(ctx, time.Second) {
461 - l.datagram.Close()
462 - return
463 - }
464 - continue
465 - }
466 -
467 - conn, err := api.openQUICSession(ctx, accessToken)
476 + conn, err := l.openQUICSession(ctx, accessToken, sniPort)
477 if err != nil {
478 log.Info().
479 Err(err).
@@ -472,7 +481,7 @@ func (l *Listener) runDatagramLoop(ctx context.Context) {
481 Str("address", identity.Address).
482 Msg("quic datagram plane unavailable; retrying")
483 if !utils.SleepOrDone(ctx, 2*time.Second) {
475 - l.datagram.Close()
484 + l.datagram.Clear("lease stopped")
485 return
486 }
487 continue
@@ -502,7 +511,7 @@ func (l *Listener) runDatagramLoop(ctx context.Context) {
511
512 select {
513 case <-ctx.Done():
505 - l.datagram.Close()
514 + l.datagram.Clear("lease stopped")
515 return
516 case <-recvDone:
517 }
@@ -513,7 +522,7 @@ func (l *Listener) runDatagramLoop(ctx context.Context) {
522 }
523 }
524
516 -func (l *Listener) runRenewLoop(ctx context.Context) {
525 +func (l *listener) runRenewLoop(ctx context.Context) error {
526 interval := l.leaseTTL / 2
527 if interval <= 0 {
528 interval = 30 * time.Second
@@ -534,7 +543,7 @@ func (l *Listener) runRenewLoop(ctx context.Context) {
543 // duration would equal the timer interval, not real time.
544 before := time.Now().Round(0)
545 if !utils.SleepOrDone(ctx, interval) {
537 - return
546 + return ctx.Err()
547 }
548 elapsed := time.Since(before)
549
@@ -546,27 +555,9 @@ func (l *Listener) runRenewLoop(ctx context.Context) {
555 log.Info().
556 Dur("expected", interval).
557 Dur("actual", elapsed).
549 - Str("address", l.Address()).
558 + Str("address", l.address()).
559 Msg("system sleep/wake detected; resetting transport and re-registering")
551 -
552 - l.stopStreamLoops()
553 - l.api.resetTransport()
554 -
555 - var retries int
556 - for {
557 - if err := l.registerAndConfigure(ctx); err == nil {
558 - l.startStreamLoops(ctx)
559 - break
560 - } else if errors.Is(err, context.Canceled) || errors.Is(err, net.ErrClosed) {
561 - return
562 - } else {
563 - retries++
564 - if !l.retryOrClose(ctx, "post-wake re-registration", err, retries) {
565 - return
566 - }
567 - }
568 - }
569 - continue
560 + return errLeaseRefreshRequired
561 }
562
563 var retries int
@@ -576,57 +567,96 @@ func (l *Listener) runRenewLoop(ctx context.Context) {
567 break
568 }
569 if errors.Is(err, context.Canceled) || errors.Is(err, net.ErrClosed) {
579 - return
570 + return err
571 + }
572 + if errors.Is(err, errLeaseRefreshRequired) {
573 + return err
574 }
575
576 retries++
583 - if !l.retryOrClose(ctx, "lease renewal", err, retries) {
584 - return
577 + if !l.waitRetry(ctx, "lease renewal", err, retries) {
578 + return err
579 }
580 }
581 }
582 }
583
590 -func (l *Listener) renewLease(ctx context.Context) error {
591 - if time.Now().Before(l.api.expiresAt) {
592 - requestCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
593 - err := l.api.renewLease(requestCtx, l.leaseTTL)
594 - cancel()
595 - if err == nil {
596 - return nil
597 - }
598 - if !errors.Is(err, &types.APIRequestError{Code: types.APIErrorCodeLeaseNotFound}) {
599 - return err
600 - }
584 +func (l *listener) renewLease(ctx context.Context) error {
585 + l.mu.Lock()
586 + expiresAt := l.expiresAt
587 + accessToken := strings.TrimSpace(l.accessToken)
588 + l.mu.Unlock()
589 +
590 + if accessToken == "" || !time.Now().Before(expiresAt) {
591 + return errLeaseRefreshRequired
592 }
593
594 requestCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
604 - defer cancel()
605 - return l.registerAndConfigure(requestCtx)
595 + resp, err := l.renewRegisteredLease(requestCtx, l.leaseTTL, accessToken)
596 + cancel()
597 + if err != nil {
598 + if errors.Is(err, &types.APIRequestError{Code: types.APIErrorCodeLeaseNotFound}) {
599 + return errLeaseRefreshRequired
600 + }
601 + return err
602 + }
603 +
604 + resp.AccessToken = strings.TrimSpace(resp.AccessToken)
605 + if resp.AccessToken == "" {
606 + return errors.New("relay did not return renewed access token")
607 + }
608 + l.mu.Lock()
609 + if l.accessToken == accessToken {
610 + l.accessToken = resp.AccessToken
611 + l.expiresAt = resp.ExpiresAt
612 + }
613 + l.mu.Unlock()
614 + return nil
615 }
616
608 -func (l *Listener) registerAndConfigure(ctx context.Context) error {
609 - resp, err := l.api.registerLease(ctx, l.leaseTTL, l.datagram != nil, l.tcpEnabled)
617 +func (l *listener) registerAndConfigure(ctx context.Context) error {
618 + resp, err := l.registerLease(ctx, l.leaseTTL, l.datagram != nil, l.tcpEnabled)
619 if err != nil {
620 return err
621 }
622 + resp.AccessToken = strings.TrimSpace(resp.AccessToken)
623 + if resp.AccessToken == "" {
624 + return errors.New("relay did not return access token")
625 + }
626 + registeredIdentity, err := utils.NormalizeIdentity(resp.Identity)
627 + if err != nil {
628 + _ = l.unregisterLease(context.Background(), resp.AccessToken)
629 + return err
630 + }
631 + l.mu.Lock()
632 + localIdentity := l.identity.Copy()
633 + l.mu.Unlock()
634 + if registeredIdentity.Key() != localIdentity.Key() {
635 + _ = l.unregisterLease(context.Background(), resp.AccessToken)
636 + return errors.New("relay returned mismatched lease identity")
637 + }
638 + resp.Identity = registeredIdentity
639 if l.datagram != nil && !resp.UDPEnabled {
614 - _ = l.api.unregisterLease(context.Background())
640 + _ = l.unregisterLease(context.Background(), resp.AccessToken)
641 return &types.APIRequestError{
642 Code: types.APIErrorCodeFeatureUnavailable,
643 Message: "relay did not enable required udp support",
644 }
645 }
620 - tlsConf, tlsCloser, err := keyless.BuildClientTLSConfig(l.api.baseURL.String(), []string{resp.Hostname})
646 + if l.datagram != nil && resp.SNIPort <= 0 {
647 + _ = l.unregisterLease(context.Background(), resp.AccessToken)
648 + return errors.New("relay did not return sni port for udp transport")
649 + }
650 + tlsConf, tenantTLSCloser, err := keyless.BuildClientTLSConfig(l.relayURL.String(), []string{resp.Hostname})
651 if err != nil {
622 - _ = l.api.unregisterLease(context.Background())
652 + _ = l.unregisterLease(context.Background(), resp.AccessToken)
653 return err
654 }
655
656 if ctx.Err() != nil {
627 - _ = l.api.unregisterLease(context.Background())
628 - if tlsCloser != nil {
629 - _ = tlsCloser.Close()
657 + _ = l.unregisterLease(context.Background(), resp.AccessToken)
658 + if tenantTLSCloser != nil {
659 + _ = tenantTLSCloser.Close()
660 }
661 return ctx.Err()
662 }
@@ -634,20 +664,27 @@ func (l *Listener) registerAndConfigure(ctx context.Context) error {
664 l.mu.Lock()
665 if ctx.Err() != nil {
666 l.mu.Unlock()
637 - _ = l.api.unregisterLease(context.Background())
638 - if tlsCloser != nil {
639 - _ = tlsCloser.Close()
667 + _ = l.unregisterLease(context.Background(), resp.AccessToken)
668 + if tenantTLSCloser != nil {
669 + _ = tenantTLSCloser.Close()
670 }
671 return ctx.Err()
672 }
643 - oldCloser := l.tlsCloser
673 + oldCloser := l.tenantTLSCloser
674 datagram := l.datagram
675 l.identity.Name = resp.Identity.Name
676 l.identity.Address = resp.Identity.Address
677 l.hostname = resp.Hostname
678 l.udpAddr = resp.UDPAddr
649 - l.tlsConfig = tlsConf
650 - l.tlsCloser = tlsCloser
679 + l.accessToken = resp.AccessToken
680 + l.expiresAt = resp.ExpiresAt
681 + if l.datagram != nil {
682 + l.sniPort = resp.SNIPort
683 + } else {
684 + l.sniPort = 0
685 + }
686 + l.tenantTLSConfig = tlsConf
687 + l.tenantTLSCloser = tenantTLSCloser
688 l.mu.Unlock()
689
690 if oldCloser != nil {
@@ -656,36 +693,39 @@ func (l *Listener) registerAndConfigure(ctx context.Context) error {
693 if datagram != nil {
694 datagram.Clear("lease updated")
695 }
659 - if l.relaySet != nil && l.api != nil && l.api.baseURL != nil {
660 - l.relaySet.ConfirmRelayURL(l.api.baseURL.String())
696 + relayURL := l.relayURL.String()
697 + if l.relaySet != nil && relayURL != "" {
698 + l.relaySet.ConfirmRelayURL(relayURL)
699 }
662 - l.registerOnce.Do(func() { close(l.registered) })
700 return nil
701 }
702
666 -func (l *Listener) retryOrClose(ctx context.Context, operation string, err error, retries int) bool {
703 +func (l *listener) waitRetry(ctx context.Context, operation string, err error, retries int) bool {
704 if ctx.Err() != nil {
705 return false
706 }
707
708 + relayURL := ""
709 + if l.relayURL != nil {
710 + relayURL = l.relayURL.String()
711 + }
712 logger := log.With().
672 - Str("relay_url", l.api.baseURL.String()).
713 + Str("relay_url", relayURL).
714 Str("operation", operation).
674 - Str("address", l.Address()).
715 + Str("address", l.address()).
716 Logger()
717
718 if l.retryCount > 0 && retries > l.retryCount {
678 - if l.relaySet != nil && l.api != nil && l.api.baseURL != nil {
679 - l.relaySet.UnconfirmRelayURL(l.api.baseURL.String())
680 - l.relaySet.RecordRelayFailure(l.api.baseURL.String(), err, 1)
719 + if l.relaySet != nil && relayURL != "" {
720 + l.relaySet.UnconfirmRelayURL(relayURL)
721 + l.relaySet.RecordRelayFailure(relayURL, err, 1)
722 }
723 if operation != "lease renewal" {
724 logger.Error().
725 Err(err).
726 Int("retry_count", l.retryCount).
686 - Msg("retry budget exhausted; closing listener")
727 + Msg("retry budget exhausted")
728 }
688 - _ = l.Close()
729 return false
730 }
731
@@ -698,23 +738,7 @@ func (l *Listener) retryOrClose(ctx context.Context, operation string, err error
738 Msg("operation failed; retrying")
739 }
740
701 - // Detect sleep/wake during the retry wait itself. If the OS
702 - // suspended the process, the renew loop will be re-registering
703 - // concurrently. Give it a moment to finish so the next attempt
704 - // uses a fresh access token and transport.
705 - // Round(0) forces wall-clock comparison (monotonic clock freezes
706 - // during macOS sleep).
707 - before := time.Now().Round(0)
708 - ok := utils.SleepOrDone(ctx, l.retryWait)
709 - if ok && time.Since(before) > l.retryWait+10*time.Second {
710 - logger.Info().
711 - Dur("expected", l.retryWait).
712 - Dur("actual", time.Since(before)).
713 - Msg("system sleep/wake detected during retry wait; pausing for re-registration")
714 - // Wait briefly for the renew loop to complete re-registration.
715 - utils.SleepOrDone(ctx, 3*time.Second)
716 - }
717 - return ok
741 + return utils.SleepOrDone(ctx, l.retryWait)
742 }
743
744 type listenerAddr string
@@ -722,7 +746,7 @@ type listenerAddr string
746 func (a listenerAddr) Network() string { return "portal" }
747 func (a listenerAddr) String() string { return string(a) }
748
725 -func (l *Listener) closed() bool {
749 +func (l *listener) closed() bool {
750 select {
751 case <-l.doneCh:
752 return true
@@ -731,17 +755,14 @@ func (l *Listener) closed() bool {
755 }
756 }
757
734 -func (l *Listener) ban() {
735 - if l.relaySet != nil && l.api != nil && l.api.baseURL != nil {
736 - l.relaySet.UnconfirmRelayURL(l.api.baseURL.String())
737 - l.relaySet.BanRelayURL(l.api.baseURL.String())
758 +func (l *listener) ban() {
759 + relayURL := ""
760 + if l.relayURL != nil {
761 + relayURL = l.relayURL.String()
762 }
739 - _ = l.Close()
740 -}
741 -
742 -func (l *Listener) BanMITM() bool {
743 - if l == nil {
744 - return false
763 + if l.relaySet != nil && relayURL != "" {
764 + l.relaySet.UnconfirmRelayURL(relayURL)
765 + l.relaySet.BanRelayURL(relayURL)
766 }
746 - return l.banMITM
767 + _ = l.Close()
768 }
sdk/mitm.go
+33 -22
@@ -51,7 +51,7 @@ type mitmProbeResult struct {
51
52 type mitmManager struct {
53 ctx context.Context
54 - listener *Listener
54 + listener *listener
55
56 mu sync.Mutex
57 pending map[string]*mitmProbePending
@@ -59,7 +59,7 @@ type mitmManager struct {
59 lastAt time.Time
60 }
61
62 -func newMITMManager(ctx context.Context, listener *Listener) *mitmManager {
62 +func newMITMManager(ctx context.Context, listener *listener) *mitmManager {
63 return &mitmManager{
64 ctx: ctx,
65 listener: listener,
@@ -77,24 +77,27 @@ func (m *mitmManager) reset() {
77
78 func (m *mitmManager) probeTLSPassthrough(ctx context.Context) (MITMProbeReport, error) {
79 l := m.listener
80 - if l == nil || l.api == nil || l.api.baseURL == nil {
80 + if l == nil || l.relayURL == nil {
81 return MITMProbeReport{}, errors.New("listener is not ready")
82 }
83
84 - publicURL := l.PublicURL()
84 + publicURL := l.publicURL()
85 if publicURL == "" {
86 return MITMProbeReport{}, errors.New("listener is not registered")
87 }
88
89 - hostname := l.Hostname()
89 + l.mu.Lock()
90 + hostname := l.hostname
91 + leaseTLSConfig := l.tenantTLSConfig
92 + l.mu.Unlock()
93 if hostname == "" {
94 return MITMProbeReport{}, errors.New("listener hostname is unavailable")
95 }
96
97 report := MITMProbeReport{
95 - RelayURL: l.api.baseURL.String(),
98 + RelayURL: l.relayURL.String(),
99 PublicURL: publicURL,
97 - Address: l.Address(),
100 + Address: l.address(),
101 }
102
103 probeCtx, cancel := context.WithTimeout(ctx, defaultMITMProbeTimeout)
@@ -115,18 +118,16 @@ func (m *mitmManager) probeTLSPassthrough(ctx context.Context) (MITMProbeReport,
118 ServerName: hostname,
119 InsecureSkipVerify: true,
120 }
118 - l.mu.Lock()
119 - if tlsConfig := l.tlsConfig; tlsConfig != nil {
120 - probeTLSConf.MinVersion = tlsConfig.MinVersion
121 - probeTLSConf.MaxVersion = tlsConfig.MaxVersion
122 - if len(tlsConfig.NextProtos) > 0 {
123 - probeTLSConf.NextProtos = append([]string(nil), tlsConfig.NextProtos...)
121 + if leaseTLSConfig != nil {
122 + probeTLSConf.MinVersion = leaseTLSConfig.MinVersion
123 + probeTLSConf.MaxVersion = leaseTLSConfig.MaxVersion
124 + if len(leaseTLSConfig.NextProtos) > 0 {
125 + probeTLSConf.NextProtos = append([]string(nil), leaseTLSConfig.NextProtos...)
126 }
127 }
126 - l.mu.Unlock()
128
129 dialer := &tls.Dialer{
129 - NetDialer: &net.Dialer{Timeout: l.api.dialTimeout},
130 + NetDialer: &net.Dialer{Timeout: l.dialTimeout},
131 Config: probeTLSConf,
132 }
133 conn, err := dialer.DialContext(probeCtx, "tcp", dialAddr)
@@ -180,15 +181,18 @@ func (m *mitmManager) probeTLSPassthrough(ctx context.Context) (MITMProbeReport,
181 }
182
183 func (m *mitmManager) probeDialAddress(publicURL string) (string, error) {
184 + l := m.listener
185 + if l == nil || l.relayURL == nil {
186 + return "", errors.New("listener is not ready")
187 + }
188 parsedURL, err := url.Parse(publicURL)
189 if err != nil {
190 return "", fmt.Errorf("parse public url: %w", err)
191 }
192
193 dialHost := parsedURL.Host
189 - l := m.listener
190 - if l != nil && l.api != nil && l.api.baseURL != nil && utils.IsLocalRelayHost(l.api.baseURL.Hostname()) {
191 - dialHost = l.api.baseURL.Host
194 + if utils.IsLocalRelayHost(l.relayURL.Hostname()) {
195 + dialHost = l.relayURL.Host
196 }
197 return utils.EnsurePort(dialHost), nil
198 }
@@ -222,6 +226,13 @@ func (m *mitmManager) maybeStart() {
226
227 func (m *mitmManager) logResult(report MITMProbeReport, err error) {
228 l := m.listener
229 + if l == nil {
230 + return
231 + }
232 + relayURL := ""
233 + if l.relayURL != nil {
234 + relayURL = l.relayURL.String()
235 + }
236 switch {
237 case l.closed():
238 return
@@ -231,8 +242,8 @@ func (m *mitmManager) logResult(report MITMProbeReport, err error) {
242 }
243 log.Warn().
244 Err(err).
234 - Str("relay_url", l.api.baseURL.String()).
235 - Str("address", l.Address()).
245 + Str("relay_url", relayURL).
246 + Str("address", l.address()).
247 Msg("tls passthrough self-probe failed")
248 case report.Reason == types.MITMProbeReasonProbeTimeout:
249 log.Warn().
@@ -242,12 +253,12 @@ func (m *mitmManager) logResult(report MITMProbeReport, err error) {
253 Msg("tls self-probe timed out before passthrough could be verified")
254 case report.Detected:
255 event := log.Warn().
245 - Bool("ban_mitm", l.BanMITM()).
256 + Bool("ban_mitm", l.banMITM).
257 Str("reason", report.Reason).
258 Str("relay_url", report.RelayURL).
259 Str("public_url", report.PublicURL).
260 Str("address", report.Address)
250 - if l.BanMITM() {
261 + if l.banMITM {
262 event.Msg("tls termination suspected by self-probe; banning relay")
263 l.ban()
264 return
sdk/mitm_test.go
+16 -18
@@ -26,7 +26,7 @@ func TestMITMProbeConnMatchesExporter(t *testing.T) {
26 defer closeMITMProbeTLSConn(clientConn)
27 defer closeMITMProbeTLSConn(serverConn)
28
29 - listener := &Listener{}
29 + listener := &listener{}
30 listener.mitmManager = newMITMManager(context.Background(), listener)
31
32 nonce := make([]byte, 16)
@@ -86,7 +86,7 @@ func TestMITMProbeConnDetectsExporterMismatch(t *testing.T) {
86 defer closeMITMProbeTLSConn(clientConn)
87 defer closeMITMProbeTLSConn(serverConn)
88
89 - listener := &Listener{}
89 + listener := &listener{}
90 listener.mitmManager = newMITMManager(context.Background(), listener)
91
92 nonce := make([]byte, 16)
@@ -144,7 +144,7 @@ func TestMITMProbeConnPassesThroughNormalTraffic(t *testing.T) {
144 defer closeMITMProbeTLSConn(clientConn)
145 defer closeMITMProbeTLSConn(serverConn)
146
147 - listener := &Listener{}
147 + listener := &listener{}
148 listener.mitmManager = newMITMManager(context.Background(), listener)
149
150 type handleResult struct {
@@ -206,8 +206,8 @@ func TestMITMProbeDetectionBansListener(t *testing.T) {
206 t.Fatalf("url.Parse() error = %v", err)
207 }
208
209 - listener := &Listener{
210 - api: &apiClient{baseURL: relayURL},
209 + listener := &listener{
210 + relayURL: relayURL,
211 relaySet: mustRelaySet(t, relayURL.String()),
212 cancel: func() {
213 select {
@@ -216,9 +216,8 @@ func TestMITMProbeDetectionBansListener(t *testing.T) {
216 close(doneCh)
217 }
218 },
219 - doneCh: doneCh,
220 - registered: make(chan struct{}),
221 - banMITM: true,
219 + doneCh: doneCh,
220 + banMITM: true,
221 }
222 listener.mitmManager = newMITMManager(context.Background(), listener)
223
@@ -245,12 +244,11 @@ func TestMITMProbeDetectionWarnsWithoutBanningListener(t *testing.T) {
244 t.Fatalf("url.Parse() error = %v", err)
245 }
246
248 - listener := &Listener{
249 - api: &apiClient{baseURL: relayURL},
250 - relaySet: mustRelaySet(t, relayURL.String()),
251 - doneCh: doneCh,
252 - registered: make(chan struct{}),
253 - banMITM: false,
247 + listener := &listener{
248 + relayURL: relayURL,
249 + relaySet: mustRelaySet(t, relayURL.String()),
250 + doneCh: doneCh,
251 + banMITM: false,
252 }
253 listener.mitmManager = newMITMManager(context.Background(), listener)
254
@@ -278,8 +276,8 @@ func TestMITMProbeDialAddressUsesRelayHostForLocalRelay(t *testing.T) {
276 t.Fatalf("url.Parse() error = %v", err)
277 }
278
281 - listener := &Listener{
282 - api: &apiClient{baseURL: relayURL},
279 + listener := &listener{
280 + relayURL: relayURL,
281 }
282 listener.mitmManager = newMITMManager(context.Background(), listener)
283
@@ -298,8 +296,8 @@ func TestMITMProbeDialAddressUsesPublicURLForRemoteRelay(t *testing.T) {
296 t.Fatalf("url.Parse() error = %v", err)
297 }
298
301 - listener := &Listener{
302 - api: &apiClient{baseURL: relayURL},
299 + listener := &listener{
300 + relayURL: relayURL,
301 }
302 listener.mitmManager = newMITMManager(context.Background(), listener)
303