@samitouri / QOSAMI-X402Tunnel / commits / c3459e6f

sdk: refactor listener, client

rabbitprincess committed Mar 7, 2026 at 20:54 UTC c3459e6fd558a250a8caa9d234a9ed2cfd37d8fa
12 files changed +749 -295
cmd/demo-app/main.go
+4 -6
@@ -18,7 +18,7 @@ import (
18
19 var (
20 flagServerURL string
21 - flagPort int
21 + flagAddr string
22 flagName string
23 flagDesc string
24 flagTags string
@@ -32,7 +32,7 @@ func main() {
32 logger := log.With().Str("component", "demo-app").Logger()
33
34 flag.StringVar(&flagServerURL, "server-url", "https://localhost:4017", "relay API URLs (comma-separated, https only)")
35 - flag.IntVar(&flagPort, "port", 8092, "local demo HTTP port")
35 + flag.StringVar(&flagAddr, "addr", "127.0.0.1:8092", "local demo HTTP listen address (disable if empty)")
36 flag.StringVar(&flagName, "name", "demo-app", "backend display name")
37 flag.StringVar(&flagDesc, "description", "Portal demo connectivity app", "lease description")
38 flag.StringVar(&flagTags, "tags", "demo,connectivity,activity,cloud,sun,morning", "comma-separated lease tags")
@@ -74,13 +74,11 @@ func runDemo() error {
74 }
75 defer listener.Close()
76
77 - if err := sdk.RunHTTP(ctx, listener, newHandler(), sdk.HTTPServeOptions{
78 - LocalAddr: fmt.Sprintf(":%d", flagPort),
79 - }); err != nil {
77 + logger.Info().Strs("public_urls", listener.PublicURLs()).Str("local_addr", flagAddr).Msg("demo app registered with relay")
78 + if err := sdk.RunHTTP(ctx, listener, newHandler(), flagAddr); err != nil {
79 return err
80 }
81
83 - logger.Info().Strs("public_urls", listener.PublicURLs()).Int("local_port", flagPort).Msg("demo app registered with relay")
82 if ctx.Err() != nil {
83 logger.Info().Msg("demo app shutting down")
84 }
portal/admin/handler.go
+25 -25
@@ -84,7 +84,7 @@ func (h *Handler) HandleRequest(w http.ResponseWriter, r *http.Request) {
84 }
85
86 if !h.isAuthenticated(r) {
87 - writeAPIError(w, http.StatusUnauthorized, "unauthorized", "unauthorized")
87 + writeAPIError(w, http.StatusUnauthorized, types.APIErrorCodeUnauthorized, "unauthorized")
88 return
89 }
90
@@ -111,17 +111,17 @@ func (h *Handler) HandleRequest(w http.ResponseWriter, r *http.Request) {
111
112 func (h *Handler) handleLogin(w http.ResponseWriter, r *http.Request) {
113 if r.Method != http.MethodPost {
114 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
114 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
115 return
116 }
117 if !h.auth.AuthEnabled() {
118 - writeAPIError(w, http.StatusServiceUnavailable, "auth_disabled", "admin authentication is not configured")
118 + writeAPIError(w, http.StatusServiceUnavailable, types.APIErrorCodeAuthDisabled, "admin authentication is not configured")
119 return
120 }
121
122 clientIP := policy.ExtractClientIP(r, h.trustProxy)
123 if h.auth.IsIPLocked(clientIP) {
124 - writeAPIErrorWithData(w, http.StatusTooManyRequests, "auth_locked", "Too many failed attempts. Please try again later.", types.AdminLoginResponse{
124 + writeAPIErrorWithData(w, http.StatusTooManyRequests, types.APIErrorCodeAuthLocked, "Too many failed attempts. Please try again later.", types.AdminLoginResponse{
125 Locked: true,
126 RemainingSeconds: h.auth.LockRemainingSeconds(clientIP),
127 })
@@ -130,7 +130,7 @@ func (h *Handler) handleLogin(w http.ResponseWriter, r *http.Request) {
130
131 var req types.AdminLoginRequest
132 if err := decodeJSON(w, r, &req); err != nil {
133 - writeAPIError(w, http.StatusBadRequest, "invalid_request", "invalid request body")
133 + writeAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidRequest, "invalid request body")
134 return
135 }
136 if !h.auth.ValidateKey(req.Key) {
@@ -139,14 +139,14 @@ func (h *Handler) handleLogin(w http.ResponseWriter, r *http.Request) {
139 if locked {
140 resp.RemainingSeconds = h.auth.LockRemainingSeconds(clientIP)
141 }
142 - writeAPIErrorWithData(w, http.StatusUnauthorized, "invalid_key", "Invalid key", resp)
142 + writeAPIErrorWithData(w, http.StatusUnauthorized, types.APIErrorCodeInvalidKey, "Invalid key", resp)
143 return
144 }
145
146 h.auth.ResetFailedLogin(clientIP)
147 token, err := h.auth.CreateSession()
148 if err != nil {
149 - writeAPIError(w, http.StatusInternalServerError, "session_create_failed", "failed to create admin session")
149 + writeAPIError(w, http.StatusInternalServerError, types.APIErrorCodeSessionCreateFailed, "failed to create admin session")
150 return
151 }
152
@@ -164,7 +164,7 @@ func (h *Handler) handleLogin(w http.ResponseWriter, r *http.Request) {
164
165 func (h *Handler) handleLogout(w http.ResponseWriter, r *http.Request) {
166 if r.Method != http.MethodPost {
167 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
167 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
168 return
169 }
170
@@ -185,7 +185,7 @@ func (h *Handler) handleLogout(w http.ResponseWriter, r *http.Request) {
185
186 func (h *Handler) handleAuthStatus(w http.ResponseWriter, r *http.Request) {
187 if r.Method != http.MethodGet {
188 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
188 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
189 return
190 }
191 writeAPIData(w, http.StatusOK, types.AdminAuthStatusResponse{
@@ -196,7 +196,7 @@ func (h *Handler) handleAuthStatus(w http.ResponseWriter, r *http.Request) {
196
197 func (h *Handler) handleLeases(w http.ResponseWriter, r *http.Request) {
198 if r.Method != http.MethodGet {
199 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
199 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
200 return
201 }
202 writeAPIData(w, http.StatusOK, h.buildLeaseRows(h.server, true))
@@ -204,7 +204,7 @@ func (h *Handler) handleLeases(w http.ResponseWriter, r *http.Request) {
204
205 func (h *Handler) handleBannedLeases(w http.ResponseWriter, r *http.Request) {
206 if r.Method != http.MethodGet {
207 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
207 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
208 return
209 }
210 writeAPIData(w, http.StatusOK, h.runtime.BannedLeases())
@@ -212,7 +212,7 @@ func (h *Handler) handleBannedLeases(w http.ResponseWriter, r *http.Request) {
212
213 func (h *Handler) handleSettings(w http.ResponseWriter, r *http.Request) {
214 if r.Method != http.MethodGet {
215 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
215 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
216 return
217 }
218 approver := h.runtime.Approver()
@@ -231,17 +231,17 @@ func (h *Handler) handleApprovalMode(w http.ResponseWriter, r *http.Request) {
231 case http.MethodPost:
232 var req types.AdminApprovalModeRequest
233 if err := decodeJSON(w, r, &req); err != nil {
234 - writeAPIError(w, http.StatusBadRequest, "invalid_request", "invalid request body")
234 + writeAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidRequest, "invalid request body")
235 return
236 }
237 if err := approver.SetMode(policy.Mode(req.Mode)); err != nil {
238 - writeAPIError(w, http.StatusBadRequest, "invalid_mode", "invalid mode (must be 'auto' or 'manual')")
238 + writeAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidMode, "invalid mode (must be 'auto' or 'manual')")
239 return
240 }
241 _ = h.settings.Save(h.runtime)
242 writeAPIData(w, http.StatusOK, types.AdminApprovalModeResponse{ApprovalMode: string(approver.Mode())})
243 default:
244 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
244 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
245 }
246 }
247
@@ -255,7 +255,7 @@ func (h *Handler) handleLeaseAction(w http.ResponseWriter, r *http.Request, path
255
256 leaseID, ok := decodeLeaseID(parts[0])
257 if !ok {
258 - writeAPIError(w, http.StatusBadRequest, "invalid_lease_id", "invalid lease ID")
258 + writeAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidLeaseID, "invalid lease ID")
259 return
260 }
261
@@ -263,7 +263,7 @@ func (h *Handler) handleLeaseAction(w http.ResponseWriter, r *http.Request, path
263 case "ban":
264 h.handleLeaseBan(w, r, leaseID)
265 case "bps":
266 - writeAPIError(w, http.StatusNotImplemented, "feature_unavailable", "bps control is not enabled in this build")
266 + writeAPIError(w, http.StatusNotImplemented, types.APIErrorCodeFeatureUnavailable, "bps control is not enabled in this build")
267 case "approve":
268 h.handleLeaseApproval(w, r, leaseID)
269 case "deny":
@@ -284,7 +284,7 @@ func (h *Handler) handleLeaseBan(w http.ResponseWriter, r *http.Request, leaseID
284 _ = h.settings.Save(h.runtime)
285 writeAPIOK(w, http.StatusOK)
286 default:
287 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
287 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
288 }
289 }
290
@@ -301,7 +301,7 @@ func (h *Handler) handleLeaseApproval(w http.ResponseWriter, r *http.Request, le
301 _ = h.settings.Save(h.runtime)
302 writeAPIOK(w, http.StatusOK)
303 default:
304 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
304 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
305 }
306 }
307
@@ -317,7 +317,7 @@ func (h *Handler) handleLeaseDenial(w http.ResponseWriter, r *http.Request, leas
317 _ = h.settings.Save(h.runtime)
318 writeAPIOK(w, http.StatusOK)
319 default:
320 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
320 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
321 }
322 }
323
@@ -329,7 +329,7 @@ func (h *Handler) handleIPBan(w http.ResponseWriter, r *http.Request, path strin
329 rawIP := strings.TrimSuffix(strings.TrimPrefix(path, types.PathAdminIPsPrefix), "/ban")
330 rawIP = strings.Trim(rawIP, "/")
331 if net.ParseIP(rawIP) == nil {
332 - writeAPIError(w, http.StatusBadRequest, "invalid_ip", "invalid IP address")
332 + writeAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidIP, "invalid IP address")
333 return
334 }
335
@@ -344,7 +344,7 @@ func (h *Handler) handleIPBan(w http.ResponseWriter, r *http.Request, path strin
344 _ = h.settings.Save(h.runtime)
345 writeAPIOK(w, http.StatusOK)
346 default:
347 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
347 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
348 }
349 }
350
@@ -379,7 +379,7 @@ func decodeLeaseID(encoded string) (string, bool) {
379 func writeAPIData(w http.ResponseWriter, status int, data any) {
380 w.Header().Set("Content-Type", "application/json")
381 w.WriteHeader(status)
382 - _ = json.NewEncoder(w).Encode(types.APIEnvelope{OK: true, Data: data})
382 + _ = json.NewEncoder(w).Encode(types.APIEnvelope[any]{OK: true, Data: data})
383 }
384
385 func writeAPIOK(w http.ResponseWriter, status int) {
@@ -389,7 +389,7 @@ func writeAPIOK(w http.ResponseWriter, status int) {
389 func writeAPIError(w http.ResponseWriter, status int, code, message string) {
390 w.Header().Set("Content-Type", "application/json")
391 w.WriteHeader(status)
392 - _ = json.NewEncoder(w).Encode(types.APIEnvelope{
392 + _ = json.NewEncoder(w).Encode(types.APIEnvelope[any]{
393 OK: false,
394 Error: &types.APIError{Code: code, Message: message},
395 })
@@ -398,7 +398,7 @@ func writeAPIError(w http.ResponseWriter, status int, code, message string) {
398 func writeAPIErrorWithData(w http.ResponseWriter, status int, code, message string, data any) {
399 w.Header().Set("Content-Type", "application/json")
400 w.WriteHeader(status)
401 - _ = json.NewEncoder(w).Encode(types.APIEnvelope{
401 + _ = json.NewEncoder(w).Encode(types.APIEnvelope[any]{
402 OK: false,
403 Data: data,
404 Error: &types.APIError{Code: code, Message: message},
portal/admin/handler_test.go
+2 -11
@@ -74,21 +74,12 @@ func TestLoginAndProtectedActions(t *testing.T) {
74 func decodeEnvelope[T any](t *testing.T, recorder *httptest.ResponseRecorder) T {
75 t.Helper()
76
77 - var envelope types.APIEnvelope
77 + var envelope types.APIEnvelope[T]
78 if err := json.NewDecoder(recorder.Body).Decode(&envelope); err != nil {
79 t.Fatalf("Decode envelope error = %v", err)
80 }
81 if !envelope.OK {
82 t.Fatalf("envelope not OK: %+v", envelope)
83 }
84 - data, err := json.Marshal(envelope.Data)
85 - if err != nil {
86 - t.Fatalf("Marshal envelope data error = %v", err)
87 - }
88 -
89 - var out T
90 - if err := json.Unmarshal(data, &out); err != nil {
91 - t.Fatalf("Unmarshal envelope data error = %v", err)
92 - }
93 - return out
84 + return envelope.Data
85 }
portal/server.go
+30 -30
@@ -273,7 +273,7 @@ func (s *Server) handleHealthz(w http.ResponseWriter, _ *http.Request) {
273
274 func (s *Server) handleDomain(w http.ResponseWriter, r *http.Request) {
275 if r.Method != http.MethodGet {
276 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
276 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
277 return
278 }
279 name := r.URL.Query().Get("name")
@@ -285,27 +285,27 @@ func (s *Server) handleDomain(w http.ResponseWriter, r *http.Request) {
285
286 func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) {
287 if r.Method != http.MethodPost {
288 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
288 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
289 return
290 }
291 clientIP := s.clientIPFromRequest(r)
292 if s.isClientIPBanned(clientIP) {
293 - writeAPIError(w, http.StatusForbidden, "ip_banned", "request denied because source IP is banned")
293 + writeAPIError(w, http.StatusForbidden, types.APIErrorCodeIPBanned, "request denied because source IP is banned")
294 return
295 }
296 var req types.RegisterRequest
297 if err := decodeJSONBody(w, r, &req); err != nil {
298 - writeAPIError(w, http.StatusBadRequest, "invalid_json", err.Error())
298 + writeAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidJSON, err.Error())
299 return
300 }
301 resp, err := s.registerLease(req, clientIP)
302 if err != nil {
303 - status, code := http.StatusBadRequest, "invalid_request"
303 + status, code := http.StatusBadRequest, types.APIErrorCodeInvalidRequest
304 if errors.Is(err, errHostnameConflict) {
305 - status, code = http.StatusConflict, "hostname_conflict"
305 + status, code = http.StatusConflict, types.APIErrorCodeHostnameConflict
306 }
307 if errors.Is(err, errIPBanned) {
308 - status, code = http.StatusForbidden, "ip_banned"
308 + status, code = http.StatusForbidden, types.APIErrorCodeIPBanned
309 }
310 writeAPIError(w, status, code, err.Error())
311 return
@@ -315,30 +315,30 @@ func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) {
315
316 func (s *Server) handleRenew(w http.ResponseWriter, r *http.Request) {
317 if r.Method != http.MethodPost {
318 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
318 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
319 return
320 }
321 clientIP := s.clientIPFromRequest(r)
322 if s.isClientIPBanned(clientIP) {
323 - writeAPIError(w, http.StatusForbidden, "ip_banned", "request denied because source IP is banned")
323 + writeAPIError(w, http.StatusForbidden, types.APIErrorCodeIPBanned, "request denied because source IP is banned")
324 return
325 }
326 var req types.RenewRequest
327 if err := decodeJSONBody(w, r, &req); err != nil {
328 - writeAPIError(w, http.StatusBadRequest, "invalid_json", err.Error())
328 + writeAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidJSON, err.Error())
329 return
330 }
331 resp, err := s.renewLease(req, clientIP)
332 if err != nil {
333 - status, code := http.StatusBadRequest, "invalid_request"
333 + status, code := http.StatusBadRequest, types.APIErrorCodeInvalidRequest
334 if errors.Is(err, errLeaseNotFound) {
335 - status, code = http.StatusNotFound, "lease_not_found"
335 + status, code = http.StatusNotFound, types.APIErrorCodeLeaseNotFound
336 }
337 if errors.Is(err, errUnauthorized) {
338 - status, code = http.StatusForbidden, "unauthorized"
338 + status, code = http.StatusForbidden, types.APIErrorCodeUnauthorized
339 }
340 if errors.Is(err, errIPBanned) {
341 - status, code = http.StatusForbidden, "ip_banned"
341 + status, code = http.StatusForbidden, types.APIErrorCodeIPBanned
342 }
343 writeAPIError(w, status, code, err.Error())
344 return
@@ -348,21 +348,21 @@ func (s *Server) handleRenew(w http.ResponseWriter, r *http.Request) {
348
349 func (s *Server) handleUnregister(w http.ResponseWriter, r *http.Request) {
350 if r.Method != http.MethodPost {
351 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
351 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
352 return
353 }
354 var req types.UnregisterRequest
355 if err := decodeJSONBody(w, r, &req); err != nil {
356 - writeAPIError(w, http.StatusBadRequest, "invalid_json", err.Error())
356 + writeAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidJSON, err.Error())
357 return
358 }
359 if err := s.unregisterLease(req); err != nil {
360 - status, code := http.StatusBadRequest, "invalid_request"
360 + status, code := http.StatusBadRequest, types.APIErrorCodeInvalidRequest
361 if errors.Is(err, errLeaseNotFound) {
362 - status, code = http.StatusNotFound, "lease_not_found"
362 + status, code = http.StatusNotFound, types.APIErrorCodeLeaseNotFound
363 }
364 if errors.Is(err, errUnauthorized) {
365 - status, code = http.StatusForbidden, "unauthorized"
365 + status, code = http.StatusForbidden, types.APIErrorCodeUnauthorized
366 }
367 writeAPIError(w, status, code, err.Error())
368 return
@@ -372,11 +372,11 @@ func (s *Server) handleUnregister(w http.ResponseWriter, r *http.Request) {
372
373 func (s *Server) handleConnect(w http.ResponseWriter, r *http.Request) {
374 if r.Method != http.MethodGet {
375 - writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
375 + writeAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")
376 return
377 }
378 if r.ProtoMajor != 1 {
379 - writeAPIError(w, http.StatusHTTPVersionNotSupported, "http11_only", "reverse connect requires HTTP/1.1")
379 + writeAPIError(w, http.StatusHTTPVersionNotSupported, types.APIErrorCodeHTTP11Only, "reverse connect requires HTTP/1.1")
380 return
381 }
382
@@ -384,33 +384,33 @@ func (s *Server) handleConnect(w http.ResponseWriter, r *http.Request) {
384 token := strings.TrimSpace(r.Header.Get(types.HeaderReverseToken))
385 clientIP := s.clientIPFromRequest(r)
386 if s.isClientIPBanned(clientIP) {
387 - writeAPIError(w, http.StatusForbidden, "ip_banned", "request denied because source IP is banned")
387 + writeAPIError(w, http.StatusForbidden, types.APIErrorCodeIPBanned, "request denied because source IP is banned")
388 return
389 }
390
391 lease, err := s.findLeaseByID(leaseID)
392 if err != nil {
393 - writeAPIError(w, http.StatusNotFound, "lease_not_found", err.Error())
393 + writeAPIError(w, http.StatusNotFound, types.APIErrorCodeLeaseNotFound, err.Error())
394 return
395 }
396 if !s.isLeaseRoutable(lease) {
397 - writeAPIError(w, http.StatusForbidden, "lease_rejected", "lease is not approved for routing")
397 + writeAPIError(w, http.StatusForbidden, types.APIErrorCodeLeaseRejected, "lease is not approved for routing")
398 return
399 }
400 if authErr := s.authorizeLeaseToken(lease, token); authErr != nil {
401 - writeAPIError(w, http.StatusForbidden, "unauthorized", authErr.Error())
401 + writeAPIError(w, http.StatusForbidden, types.APIErrorCodeUnauthorized, authErr.Error())
402 return
403 }
404
405 hijacker, ok := w.(http.Hijacker)
406 if !ok {
407 - writeAPIError(w, http.StatusInternalServerError, "hijack_unsupported", "hijacking is not supported")
407 + writeAPIError(w, http.StatusInternalServerError, types.APIErrorCodeHijackUnsupported, "hijacking is not supported")
408 return
409 }
410
411 conn, rw, err := hijacker.Hijack()
412 if err != nil {
413 - writeAPIError(w, http.StatusInternalServerError, "hijack_failed", err.Error())
413 + writeAPIError(w, http.StatusInternalServerError, types.APIErrorCodeHijackFailed, err.Error())
414 return
415 }
416
@@ -737,7 +737,7 @@ func (s *Server) wrapAPIHandler(base http.Handler) http.Handler {
737 var (
738 errLeaseNotFound = errors.New("lease not found")
739 errIPBanned = errors.New("request denied because source IP is banned")
740 - errUnauthorized = errors.New("unauthorized")
740 + errUnauthorized = errors.New(types.APIErrorCodeUnauthorized)
741 errHostnameConflict = errors.New("hostname already registered")
742 )
743
@@ -774,7 +774,7 @@ func tokenMatches(expected, actual string) bool {
774 func writeAPIData(w http.ResponseWriter, status int, data any) {
775 w.Header().Set("Content-Type", "application/json")
776 w.WriteHeader(status)
777 - _ = json.NewEncoder(w).Encode(types.APIEnvelope{OK: true, Data: data})
777 + _ = json.NewEncoder(w).Encode(types.APIEnvelope[any]{OK: true, Data: data})
778 }
779
780 func writeAPIOK(w http.ResponseWriter, status int) {
@@ -784,7 +784,7 @@ func writeAPIOK(w http.ResponseWriter, status int) {
784 func writeAPIError(w http.ResponseWriter, status int, code, message string) {
785 w.Header().Set("Content-Type", "application/json")
786 w.WriteHeader(status)
787 - _ = json.NewEncoder(w).Encode(types.APIEnvelope{
787 + _ = json.NewEncoder(w).Encode(types.APIEnvelope[any]{
788 OK: false,
789 Error: &types.APIError{Code: code, Message: message},
790 })
sdk/client.go
+76 -44
@@ -23,12 +23,14 @@ import (
23 )
24
25 const (
26 - defaultDialTimeout = 5 * time.Second
27 - defaultRequestTimeout = 15 * time.Second
28 - defaultHandshakeTimeout = 15 * time.Second
29 - defaultLeaseTTL = 2 * time.Minute
30 - defaultRenewBefore = 30 * time.Second
31 - defaultReadyTarget = 1
26 + defaultDialTimeout = 5 * time.Second
27 + defaultRequestTimeout = 15 * time.Second
28 + defaultHandshakeTimeout = 15 * time.Second
29 + defaultLeaseTTL = 2 * time.Minute
30 + defaultRenewBefore = 30 * time.Second
31 + defaultReadyTarget = 1
32 + defaultRetryDelay = 5 * time.Second
33 + defaultHTTPShutdownTimeout = 5 * time.Second
34 )
35
36 // ClientConfig configures the SDK client.
@@ -86,23 +88,18 @@ func (c *Client) Listen(ctx context.Context, req ListenRequest) (*Listener, erro
88 return nil, errors.New("no relay urls configured")
89 }
90
89 - listenerCtx, cancel := context.WithCancel(ctx)
90 - listener := &Listener{
91 - baseContext: func() context.Context { return listenerCtx },
92 - ctxDone: listenerCtx.Done(),
93 - cancel: cancel,
94 - }
91 + listener := newListener(ctx)
92
93 entries := make([]*listenerLease, 0, len(c.clients))
94 acceptedCap := 0
95 for _, client := range c.clients {
99 - entry, entryAcceptedCap, err := client.listenEntry(listener, req)
96 + entry, err := client.listenEntry(listener, req)
97 if err != nil {
101 - cancel()
98 + listener.cancel()
99 return nil, errors.Join(err, closeListenerEntries(entries))
100 }
101 entries = append(entries, entry)
105 - acceptedCap += entryAcceptedCap
102 + acceptedCap += entry.readyTarget
103 }
104
105 if acceptedCap <= 0 {
@@ -111,10 +108,9 @@ func (c *Client) Listen(ctx context.Context, req ListenRequest) (*Listener, erro
108
109 listener.accepted = make(chan acceptedConn, acceptedCap)
110 listener.entries = entries
111 + listener.activeCount = len(entries)
112 for _, entry := range entries {
115 - go entry.runSupervisor()
116 - go entry.runRenewLoop()
117 - entry.notify()
113 + entry.start()
114 }
115
116 return listener, nil
@@ -211,7 +207,7 @@ func (c *relayClient) Close() {
207 }
208 }
209
214 -func (c *relayClient) listenEntry(listener *Listener, req ListenRequest) (*listenerLease, int, error) {
210 +func (c *relayClient) listenEntry(listener *Listener, req ListenRequest) (*listenerLease, error) {
211 reverseToken := strings.TrimSpace(req.ReverseToken)
212 if reverseToken == "" {
213 reverseToken = randomToken()
@@ -225,7 +221,6 @@ func (c *relayClient) listenEntry(listener *Listener, req ListenRequest) (*liste
221 if leaseTTL <= 0 {
222 leaseTTL = defaultLeaseTTL
223 }
228 - acceptedCap := max(readyTarget*2, 1)
224
225 registerReq := types.RegisterRequest{
226 Name: req.Name,
@@ -237,14 +232,14 @@ func (c *relayClient) listenEntry(listener *Listener, req ListenRequest) (*liste
232 }
233
234 var registerResp types.RegisterResponse
240 - if err := c.doJSON(listener.baseContext(), http.MethodPost, types.PathSDKRegister, registerReq, &registerResp); err != nil {
241 - return nil, 0, err
235 + if err := c.doJSON(listener.ctx, http.MethodPost, types.PathSDKRegister, registerReq, &registerResp); err != nil {
236 + return nil, err
237 }
238
239 tlsConf, tlsCloser, err := keyless.BuildClientTLSConfig(c.baseURL.String(), registerResp.Hostnames)
240 if err != nil {
241 _ = c.unregisterLease(context.Background(), registerResp.LeaseID, reverseToken)
247 - return nil, 0, err
242 + return nil, err
243 }
244
245 return &listenerLease{
@@ -254,15 +249,15 @@ func (c *relayClient) listenEntry(listener *Listener, req ListenRequest) (*liste
249 RelayURL: c.baseURL.String(),
250 LeaseID: registerResp.LeaseID,
251 Hostnames: append([]string(nil), registerResp.Hostnames...),
257 - Metadata: registerResp.Metadata,
252 + Metadata: cloneLeaseMetadata(registerResp.Metadata),
253 },
254 reverseToken: reverseToken,
255 leaseTTL: leaseTTL,
256 readyTarget: readyTarget,
257 tlsConfig: tlsConf,
258 tlsCloser: tlsCloser,
264 - signal: make(chan struct{}, 1),
265 - }, acceptedCap, nil
259 + active: true,
260 + }, nil
261 }
262
263 func (c *relayClient) doJSON(ctx context.Context, method, path string, payload any, out any) error {
@@ -275,7 +270,8 @@ func (c *relayClient) doJSON(ctx context.Context, method, path string, payload a
270 body = bytes.NewReader(buf)
271 }
272
278 - req, err := http.NewRequestWithContext(ctx, method, c.resolve(path), body)
273 + ref, _ := url.Parse(path)
274 + req, err := http.NewRequestWithContext(ctx, method, c.baseURL.ResolveReference(ref).String(), body)
275 if err != nil {
276 return err
277 }
@@ -287,15 +283,22 @@ func (c *relayClient) doJSON(ctx context.Context, method, path string, payload a
283 }
284 defer resp.Body.Close()
285
290 - var envelope apiEnvelope
286 + var envelope types.APIEnvelope[json.RawMessage]
287 if err := json.NewDecoder(resp.Body).Decode(&envelope); err != nil {
288 return fmt.Errorf("decode response: %w", err)
289 }
290 if !envelope.OK {
291 if envelope.Error == nil {
296 - return fmt.Errorf("api request failed with status %d", resp.StatusCode)
292 + return &types.APIRequestError{
293 + StatusCode: resp.StatusCode,
294 + Message: fmt.Sprintf("api request failed with status %d", resp.StatusCode),
295 + }
296 + }
297 + return &types.APIRequestError{
298 + StatusCode: resp.StatusCode,
299 + Code: envelope.Error.Code,
300 + Message: envelope.Error.Message,
301 }
298 - return fmt.Errorf("%s: %s", envelope.Error.Code, envelope.Error.Message)
302 }
303 if out == nil {
304 return nil
@@ -329,11 +332,8 @@ func (c *relayClient) openReverseSession(ctx context.Context, leaseID, reverseTo
332 return nil, err
333 }
334
332 - connectURL, err := url.Parse(c.resolve(types.PathSDKConnect))
333 - if err != nil {
334 - _ = conn.Close()
335 - return nil, err
336 - }
335 + connectRef, _ := url.Parse(types.PathSDKConnect)
336 + connectURL := c.baseURL.ResolveReference(connectRef)
337 query := connectURL.Query()
338 query.Set("lease_id", leaseID)
339 connectURL.RawQuery = query.Encode()
@@ -361,23 +361,55 @@ func (c *relayClient) openReverseSession(ctx context.Context, leaseID, reverseTo
361 defer resp.Body.Close()
362
363 if resp.StatusCode != http.StatusOK {
364 - body, _ := io.ReadAll(io.LimitReader(resp.Body, 8<<10))
364 + apiErr := decodeAPIResponseError(resp)
365 _ = conn.Close()
366 - return nil, fmt.Errorf("reverse connect failed: %s", strings.TrimSpace(string(body)))
366 + return nil, apiErr
367 }
368
369 return wrapBufferedConn(conn, reader), nil
370 }
371
372 -func (c *relayClient) resolve(path string) string {
373 - ref, _ := url.Parse(path)
374 - return c.baseURL.ResolveReference(ref).String()
372 +func decodeAPIResponseError(resp *http.Response) error {
373 + if resp == nil {
374 + return &types.APIRequestError{Message: "empty api response"}
375 + }
376 +
377 + body, _ := io.ReadAll(io.LimitReader(resp.Body, 8<<10))
378 + var envelope types.APIEnvelope[json.RawMessage]
379 + if err := json.Unmarshal(body, &envelope); err == nil && envelope.Error != nil {
380 + return &types.APIRequestError{
381 + StatusCode: resp.StatusCode,
382 + Code: envelope.Error.Code,
383 + Message: envelope.Error.Message,
384 + }
385 + }
386 +
387 + return &types.APIRequestError{
388 + StatusCode: resp.StatusCode,
389 + Message: strings.TrimSpace(string(body)),
390 + }
391 +}
392 +
393 +func isLeaseResetError(err error) bool {
394 + var apiErr *types.APIRequestError
395 + if !errors.As(err, &apiErr) {
396 + return false
397 + }
398 + return apiErr.Code == types.APIErrorCodeLeaseNotFound
399 }
400
377 -type apiEnvelope struct {
378 - Error *types.APIError `json:"error"`
379 - Data json.RawMessage `json:"data"`
380 - OK bool `json:"ok"`
401 +func isTerminalEntryError(err error) bool {
402 + var apiErr *types.APIRequestError
403 + if !errors.As(err, &apiErr) {
404 + return false
405 + }
406 +
407 + switch apiErr.Code {
408 + case types.APIErrorCodeIPBanned, types.APIErrorCodeUnauthorized:
409 + return true
410 + default:
411 + return false
412 + }
413 }
414
415 func buildRootCAs(rootCAPEM []byte) (*x509.CertPool, error) {
sdk/client_test.go
+78
@@ -1,9 +1,17 @@
1 package sdk
2
3 import (
4 + "context"
5 + "crypto/tls"
6 + "encoding/json"
7 + "errors"
8 "net/http"
9 "net/http/httptest"
10 + "net/url"
11 "testing"
12 + "time"
13 +
14 + "github.com/gosuda/portal/v2/types"
15 )
16
17 func TestNewClientAutoTrustsLocalhostRelayCertificate(t *testing.T) {
@@ -71,3 +79,73 @@ func TestNewClientSupportsDedupedRelayURLs(t *testing.T) {
79 }
80 }
81 }
82 +
83 +func TestOpenReverseSessionPreservesAPIErrorCode(t *testing.T) {
84 + t.Parallel()
85 +
86 + server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
87 + if r.URL.Path != types.PathSDKConnect {
88 + t.Fatalf("request path = %q, want %q", r.URL.Path, types.PathSDKConnect)
89 + }
90 + if got := r.URL.Query().Get("lease_id"); got != "lease-123" {
91 + t.Fatalf("lease_id = %q, want %q", got, "lease-123")
92 + }
93 +
94 + w.Header().Set("Content-Type", "application/json")
95 + w.WriteHeader(http.StatusForbidden)
96 + if flusher, ok := w.(http.Flusher); ok {
97 + flusher.Flush()
98 + }
99 + time.Sleep(25 * time.Millisecond)
100 + _ = json.NewEncoder(w).Encode(types.APIEnvelope[any]{
101 + OK: false,
102 + Error: &types.APIError{Code: "unauthorized", Message: "bad reverse token"},
103 + })
104 + }))
105 + server.EnableHTTP2 = false
106 + server.StartTLS()
107 + defer server.Close()
108 +
109 + baseURL, err := url.Parse(server.URL)
110 + if err != nil {
111 + t.Fatalf("url.Parse() error = %v", err)
112 + }
113 +
114 + transport, ok := server.Client().Transport.(*http.Transport)
115 + if !ok {
116 + t.Fatalf("server client transport type = %T, want *http.Transport", server.Client().Transport)
117 + }
118 +
119 + relayClient := &relayClient{
120 + baseURL: baseURL,
121 + httpClient: &http.Client{
122 + Transport: transport.Clone(),
123 + Timeout: defaultRequestTimeout,
124 + },
125 + rawTLSConfig: &tls.Config{
126 + MinVersion: tls.VersionTLS12,
127 + ServerName: baseURL.Hostname(),
128 + RootCAs: transport.TLSClientConfig.RootCAs,
129 + NextProtos: []string{"http/1.1"},
130 + },
131 + }
132 +
133 + _, err = relayClient.openReverseSession(context.Background(), "lease-123", "tok_123")
134 + if err == nil {
135 + t.Fatal("openReverseSession() error = nil, want APIRequestError")
136 + }
137 +
138 + var apiErr *types.APIRequestError
139 + if !errors.As(err, &apiErr) {
140 + t.Fatalf("openReverseSession() error = %T, want *types.APIRequestError", err)
141 + }
142 + if apiErr.StatusCode != http.StatusForbidden {
143 + t.Fatalf("APIRequestError.StatusCode = %d, want %d", apiErr.StatusCode, http.StatusForbidden)
144 + }
145 + if apiErr.Code != "unauthorized" {
146 + t.Fatalf("APIRequestError.Code = %q, want %q", apiErr.Code, "unauthorized")
147 + }
148 + if apiErr.Message != "bad reverse token" {
149 + t.Fatalf("APIRequestError.Message = %q, want %q", apiErr.Message, "bad reverse token")
150 + }
151 +}
sdk/helper.go
+7 -20
@@ -7,37 +7,24 @@ import (
7 "net"
8 "net/http"
9 "strings"
10 - "time"
10
11 "golang.org/x/sync/errgroup"
12 )
13
15 -const defaultHTTPShutdownTimeout = 5 * time.Second
16 -
17 -type HTTPServeOptions struct {
18 - LocalAddr string
19 - ReadHeaderTimeout time.Duration
20 -}
21 -
22 -// RunHTTP serves one handler on the relay listener and, optionally, on a
23 -// local HTTP address for app-local access.
24 -func RunHTTP(ctx context.Context, relayListener net.Listener, handler http.Handler, opts HTTPServeOptions) error {
25 - readHeaderTimeout := opts.ReadHeaderTimeout
26 - if readHeaderTimeout <= 0 {
27 - readHeaderTimeout = defaultRequestTimeout
28 - }
29 -
14 +// RunHTTP serves one handler on the relay listener and, when localAddr is set,
15 +// on the provided local HTTP address for app-local access.
16 +func RunHTTP(ctx context.Context, relayListener net.Listener, handler http.Handler, localAddr string) error {
17 relaySrv := &http.Server{
18 Handler: handler,
32 - ReadHeaderTimeout: readHeaderTimeout,
19 + ReadHeaderTimeout: defaultRequestTimeout,
20 }
21
22 var localSrv *http.Server
36 - if opts.LocalAddr != "" {
23 + if strings.TrimSpace(localAddr) != "" {
24 localSrv = &http.Server{
38 - Addr: opts.LocalAddr,
25 + Addr: strings.TrimSpace(localAddr),
26 Handler: handler,
40 - ReadHeaderTimeout: readHeaderTimeout,
27 + ReadHeaderTimeout: defaultRequestTimeout,
28 }
29 }
30
sdk/helper_test.go
+2 -4
@@ -25,7 +25,7 @@ func TestRunHTTPAppRelayOnly(t *testing.T) {
25 go func() {
26 errCh <- RunHTTP(ctx, listener, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
27 _, _ = io.WriteString(w, "ok")
28 - }), HTTPServeOptions{})
28 + }), "")
29 }()
30
31 waitForHTTP(t, "http://"+listener.Addr().String())
@@ -64,9 +64,7 @@ func TestRunHTTPAppLocalAndRelay(t *testing.T) {
64 go func() {
65 errCh <- RunHTTP(ctx, relayListener, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
66 _, _ = io.WriteString(w, "ok")
67 - }), HTTPServeOptions{
68 - LocalAddr: localAddr,
69 - })
67 + }), localAddr)
68 }()
69
70 waitForHTTP(t, "http://"+relayListener.Addr().String())
sdk/listener.go
+273 -117
@@ -42,30 +42,36 @@ func (e ListenerEntry) PublicURLs() []string {
42
43 func (e ListenerEntry) clone() ListenerEntry {
44 e.Hostnames = append([]string(nil), e.Hostnames...)
45 + e.Metadata = cloneLeaseMetadata(e.Metadata)
46 return e
47 }
48
49 type Listener struct {
49 - baseContext func() context.Context
50 - ctxDone <-chan struct{}
51 - cancel context.CancelFunc
52 - accepted chan acceptedConn
53 - entries []*listenerLease
54 - closeOnce sync.Once
50 + ctx context.Context
51 + cancel context.CancelFunc
52 + accepted chan acceptedConn
53 + entries []*listenerLease
54 + workers errgroup.Group
55 + closeOnce sync.Once
56 +
57 + mu sync.RWMutex
58 + activeCount int
59 + terminalErr error
60 }
61
62 type listenerLease struct {
58 - tlsCloser io.Closer
59 - tlsConfig *tls.Config
60 - parent *Listener
61 - client *relayClient
62 - signal chan struct{}
63 - info ListenerEntry
64 - reverseToken string
65 - readyTarget int
66 - leaseTTL time.Duration
67 - activeSessions int
68 - mu sync.Mutex
63 + parent *Listener
64 + client *relayClient
65 + reverseToken string
66 + readyTarget int
67 + leaseTTL time.Duration
68 +
69 + mu sync.RWMutex
70 + info ListenerEntry
71 + tlsConfig *tls.Config
72 + tlsCloser io.Closer
73 + active bool
74 + terminalErr error
75 }
76
77 type acceptedConn struct {
@@ -73,6 +79,25 @@ type acceptedConn struct {
79 entry ListenerEntry
80 }
81
82 +type sessionSnapshot struct {
83 + info ListenerEntry
84 + tlsConfig *tls.Config
85 + active bool
86 + terminalErr error
87 +}
88 +
89 +func newListener(ctx context.Context) *Listener {
90 + if ctx == nil {
91 + ctx = context.Background()
92 + }
93 +
94 + listenerCtx, cancel := context.WithCancel(ctx)
95 + return &Listener{
96 + ctx: listenerCtx,
97 + cancel: cancel,
98 + }
99 +}
100 +
101 func (l *Listener) Accept() (net.Conn, error) {
102 conn, _, err := l.AcceptEntry()
103 return conn, err
@@ -81,14 +106,16 @@ func (l *Listener) Accept() (net.Conn, error) {
106 // AcceptEntry returns the next accepted connection plus relay-specific lease
107 // metadata for callers that need to distinguish which relay claimed it.
108 func (l *Listener) AcceptEntry() (net.Conn, ListenerEntry, error) {
84 - select {
85 - case <-l.ctxDone:
86 - return nil, ListenerEntry{}, net.ErrClosed
87 - case accepted := <-l.accepted:
88 - if accepted.conn == nil {
89 - return nil, ListenerEntry{}, net.ErrClosed
109 + for {
110 + select {
111 + case <-l.ctx.Done():
112 + return nil, ListenerEntry{}, l.closeError()
113 + case accepted := <-l.accepted:
114 + if accepted.conn == nil {
115 + return nil, ListenerEntry{}, l.closeError()
116 + }
117 + return accepted.conn, accepted.entry.clone(), nil
118 }
91 - return accepted.conn, accepted.entry.clone(), nil
119 }
120 }
121
@@ -96,7 +123,8 @@ func (l *Listener) Close() error {
123 var closeErr error
124 l.closeOnce.Do(func() {
125 l.cancel()
99 - closeErr = closeListenerEntries(l.entries)
126 + closeErr = errors.Join(l.workers.Wait(), closeListenerEntries(l.entries))
127 + l.drainAccepted()
128 })
129 return closeErr
130 }
@@ -112,23 +140,28 @@ func (l *Listener) Addr() net.Addr {
140 func (l *Listener) Entries() []ListenerEntry {
141 entries := make([]ListenerEntry, 0, len(l.entries))
142 for _, entry := range l.entries {
115 - entries = append(entries, entry.info.clone())
143 + info, ok := entry.snapshotInfo()
144 + if !ok {
145 + continue
146 + }
147 + entries = append(entries, info)
148 }
149 return entries
150 }
151
152 func (l *Listener) singleEntry() (ListenerEntry, bool) {
121 - if len(l.entries) != 1 {
153 + entries := l.Entries()
154 + if len(entries) != 1 {
155 return ListenerEntry{}, false
156 }
124 - return l.entries[0].info.clone(), true
157 + return entries[0], true
158 }
159
160 // PublicURLs returns all public HTTPS URLs exposed by the listener.
161 func (l *Listener) PublicURLs() []string {
162 var urls []string
130 - for _, entry := range l.entries {
131 - urls = append(urls, entry.info.PublicURLs()...)
163 + for _, entry := range l.Entries() {
164 + urls = append(urls, entry.PublicURLs()...)
165 }
166 return urls
167 }
@@ -147,6 +180,7 @@ func closeListenerEntries(entries []*listenerLease) error {
180 if entry == nil {
181 continue
182 }
183 + entry := entry
184 group.Go(func() error {
185 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
186 defer cancel()
@@ -165,66 +199,86 @@ func closeListenerEntries(entries []*listenerLease) error {
199 return closeErr
200 }
201
168 -func (l *listenerLease) runSupervisor() {
169 - for {
170 - select {
171 - case <-l.parent.ctxDone:
172 - return
173 - case <-l.signal:
174 - }
202 +func (l *listenerLease) start() {
203 + if l == nil || l.parent == nil {
204 + return
205 + }
206
176 - for l.reserveSessionSlot() {
177 - go l.runSession()
178 - }
207 + l.parent.workers.Go(func() error {
208 + l.runRenewLoop()
209 + return nil
210 + })
211 + for i := 0; i < l.readyTarget; i++ {
212 + l.parent.workers.Go(func() error {
213 + l.runSessionWorker()
214 + return nil
215 + })
216 }
217 }
218
219 func (l *listenerLease) runRenewLoop() {
183 - interval := l.leaseTTL / 2
184 - if interval <= 0 {
185 - interval = 30 * time.Second
186 - }
187 - if defaultRenewBefore > 0 && l.leaseTTL > defaultRenewBefore {
188 - interval = l.leaseTTL - defaultRenewBefore
189 - }
190 - if interval <= 0 {
191 - interval = 30 * time.Second
192 - }
193 -
194 - ticker := time.NewTicker(interval)
220 + ticker := time.NewTicker(l.renewInterval())
221 defer ticker.Stop()
222
223 for {
224 select {
199 - case <-l.parent.ctxDone:
225 + case <-l.parent.ctx.Done():
226 return
227 case <-ticker.C:
202 - ctx, cancel := context.WithTimeout(l.context(), 10*time.Second)
203 - _ = l.client.renewLease(ctx, l.info.LeaseID, l.reverseToken, l.leaseTTL)
228 + if l.shouldStop() {
229 + return
230 + }
231 +
232 + ctx, cancel := context.WithTimeout(l.parent.ctx, 10*time.Second)
233 + err := l.client.renewLease(ctx, l.leaseID(), l.reverseToken, l.leaseTTL)
234 cancel()
235 + if err == nil {
236 + continue
237 + }
238 + if isLeaseResetError(err) || isTerminalEntryError(err) {
239 + l.stop(err)
240 + return
241 + }
242 }
243 }
244 }
245
209 -func (l *listenerLease) runSession() {
210 - defer l.releaseSessionSlot()
246 +func (l *listenerLease) runSessionWorker() {
247 + for {
248 + if l.shouldStop() {
249 + return
250 + }
251
212 - sessionCtx := l.context()
213 - conn, err := l.client.openReverseSession(sessionCtx, l.info.LeaseID, l.reverseToken)
214 - if err != nil {
215 - sleepOrDone(sessionCtx, time.Second)
216 - return
217 - }
252 + snapshot := l.sessionSnapshot()
253 + if !snapshot.active {
254 + return
255 + }
256
219 - if err := l.awaitActivation(conn); err != nil {
220 - _ = conn.Close()
221 - if !errors.Is(err, context.Canceled) && !errors.Is(err, net.ErrClosed) {
222 - sleepOrDone(sessionCtx, time.Second)
257 + conn, err := l.client.openReverseSession(l.parent.ctx, snapshot.info.LeaseID, l.reverseToken)
258 + if err != nil {
259 + if isLeaseResetError(err) || isTerminalEntryError(err) {
260 + l.stop(err)
261 + return
262 + }
263 + sleepOrDone(l.parent.ctx, defaultRetryDelay)
264 + continue
265 + }
266 +
267 + if err := l.parent.awaitActivation(conn, snapshot.tlsConfig, snapshot.info); err != nil {
268 + _ = conn.Close()
269 + if errors.Is(err, context.Canceled) || errors.Is(err, net.ErrClosed) {
270 + return
271 + }
272 + if isLeaseResetError(err) || isTerminalEntryError(err) {
273 + l.stop(err)
274 + return
275 + }
276 + sleepOrDone(l.parent.ctx, defaultRetryDelay)
277 }
278 }
279 }
280
227 -func (l *listenerLease) awaitActivation(conn net.Conn) error {
281 +func (l *Listener) awaitActivation(conn net.Conn, tlsConfig *tls.Config, info ListenerEntry) error {
282 var marker [1]byte
283 for {
284 _ = conn.SetReadDeadline(time.Now().Add(2 * defaultHandshakeTimeout))
@@ -237,72 +291,190 @@ func (l *listenerLease) awaitActivation(conn net.Conn) error {
291 case types.MarkerKeepalive:
292 continue
293 case types.MarkerTLSStart:
240 - return l.activate(conn)
294 + return l.activate(conn, tlsConfig, info)
295 default:
296 return fmt.Errorf("unexpected reverse marker: 0x%02x", marker[0])
297 }
298 }
299 }
300
247 -func (l *listenerLease) activate(conn net.Conn) error {
248 - tlsConn := tls.Server(conn, l.tlsConfig.Clone())
249 - handshakeCtx, cancel := context.WithTimeout(l.context(), defaultHandshakeTimeout)
301 +func (l *Listener) activate(conn net.Conn, tlsConfig *tls.Config, info ListenerEntry) error {
302 + if tlsConfig == nil {
303 + return errors.New("missing tls config")
304 + }
305 +
306 + tlsConn := tls.Server(conn, tlsConfig.Clone())
307 + handshakeCtx, cancel := context.WithTimeout(l.ctx, defaultHandshakeTimeout)
308 defer cancel()
309 if err := tlsConn.HandshakeContext(handshakeCtx); err != nil {
310 return err
311 }
312
313 select {
256 - case <-l.parent.ctxDone:
314 + case <-l.ctx.Done():
315 _ = tlsConn.Close()
258 - return l.context().Err()
259 - case l.parent.accepted <- acceptedConn{conn: tlsConn, entry: l.info.clone()}:
316 + return l.closeError()
317 + case l.accepted <- acceptedConn{conn: tlsConn, entry: info.clone()}:
318 + return nil
319 + }
320 +}
321 +
322 +func (l *listenerLease) close(ctx context.Context) error {
323 + if l == nil {
324 return nil
325 }
326 +
327 + var closeErr error
328 + if l.client != nil {
329 + if err := l.client.unregisterLease(ctx, l.info.LeaseID, l.reverseToken); err != nil {
330 + closeErr = errors.Join(closeErr, err)
331 + }
332 + }
333 + if l.tlsCloser != nil {
334 + closeErr = errors.Join(closeErr, l.tlsCloser.Close())
335 + }
336 + return closeErr
337 +}
338 +
339 +func (l *listenerLease) renewInterval() time.Duration {
340 + interval := l.leaseTTL / 2
341 + if interval <= 0 {
342 + interval = 30 * time.Second
343 + }
344 + if defaultRenewBefore > 0 && l.leaseTTL > defaultRenewBefore {
345 + interval = l.leaseTTL - defaultRenewBefore
346 + }
347 + if interval <= 0 {
348 + interval = 30 * time.Second
349 + }
350 + return interval
351 }
352
264 -func (l *listenerLease) reserveSessionSlot() bool {
353 +func (l *Listener) fail(err error) {
354 l.mu.Lock()
266 - defer l.mu.Unlock()
267 - if l.isClosed() {
355 + if err != nil && l.terminalErr == nil {
356 + l.terminalErr = err
357 + }
358 + l.mu.Unlock()
359 + l.cancel()
360 +}
361 +
362 +func (l *Listener) closeError() error {
363 + l.mu.RLock()
364 + terminalErr := l.terminalErr
365 + l.mu.RUnlock()
366 + if terminalErr != nil {
367 + return terminalErr
368 + }
369 + if err := l.ctx.Err(); err != nil {
370 + if errors.Is(err, context.Canceled) {
371 + return net.ErrClosed
372 + }
373 + return err
374 + }
375 + return net.ErrClosed
376 +}
377 +
378 +func (l *Listener) isClosed() bool {
379 + if l == nil || l.ctx == nil {
380 return false
381 }
270 - if l.activeSessions >= l.readyTarget {
382 + select {
383 + case <-l.ctx.Done():
384 + return true
385 + default:
386 return false
387 }
273 - l.activeSessions++
274 - return true
388 }
389
277 -func (l *listenerLease) releaseSessionSlot() {
390 +func (l *Listener) drainAccepted() {
391 + for {
392 + select {
393 + case accepted := <-l.accepted:
394 + if accepted.conn != nil {
395 + _ = accepted.conn.Close()
396 + }
397 + default:
398 + return
399 + }
400 + }
401 +}
402 +
403 +func (l *Listener) entryStopped(_ *listenerLease, err error) {
404 l.mu.Lock()
279 - l.activeSessions--
405 + if l.activeCount > 0 {
406 + l.activeCount--
407 + }
408 + shouldFail := l.activeCount == 0
409 l.mu.Unlock()
281 - l.notify()
410 +
411 + if shouldFail {
412 + l.fail(err)
413 + }
414 }
415
284 -func (l *listenerLease) notify() {
285 - select {
286 - case l.signal <- struct{}{}:
287 - default:
416 +func (l *listenerLease) snapshotInfo() (ListenerEntry, bool) {
417 + snapshot := l.sessionSnapshot()
418 + if !snapshot.active {
419 + return ListenerEntry{}, false
420 }
421 + return snapshot.info, true
422 }
423
291 -func (l *listenerLease) close(ctx context.Context) error {
424 +func (l *listenerLease) leaseID() string {
425 + l.mu.RLock()
426 + defer l.mu.RUnlock()
427 + return l.info.LeaseID
428 +}
429 +
430 +func (l *listenerLease) sessionSnapshot() sessionSnapshot {
431 + l.mu.RLock()
432 + defer l.mu.RUnlock()
433 +
434 + return sessionSnapshot{
435 + info: l.info.clone(),
436 + tlsConfig: l.tlsConfig,
437 + active: l.active,
438 + terminalErr: l.terminalErr,
439 + }
440 +}
441 +
442 +func (l *listenerLease) shutdownState() (bool, error) {
443 + l.mu.RLock()
444 + defer l.mu.RUnlock()
445 + return !l.active, l.terminalErr
446 +}
447 +
448 +func (l *listenerLease) shouldStop() bool {
449 if l == nil {
293 - return nil
450 + return true
451 }
452 + if l.parent != nil && l.parent.isClosed() {
453 + return true
454 + }
455 + stopped, _ := l.shutdownState()
456 + return stopped
457 +}
458
296 - var closeErr error
297 - if l.client != nil {
298 - if err := l.client.unregisterLease(ctx, l.info.LeaseID, l.reverseToken); err != nil {
299 - closeErr = errors.Join(closeErr, err)
300 - }
459 +func (l *listenerLease) stop(err error) {
460 + if l == nil {
461 + return
462 }
302 - if l.tlsCloser != nil {
303 - closeErr = errors.Join(closeErr, l.tlsCloser.Close())
463 +
464 + l.mu.Lock()
465 + if !l.active {
466 + l.mu.Unlock()
467 + return
468 + }
469 + l.active = false
470 + if err != nil && l.terminalErr == nil {
471 + l.terminalErr = err
472 + }
473 + l.mu.Unlock()
474 +
475 + if l.parent != nil {
476 + l.parent.entryStopped(l, err)
477 }
305 - return closeErr
478 }
479
480 func sleepOrDone(ctx context.Context, d time.Duration) {
@@ -319,23 +491,7 @@ type listenerAddr string
491 func (a listenerAddr) Network() string { return "portal" }
492 func (a listenerAddr) String() string { return string(a) }
493
322 -func (l *listenerLease) context() context.Context {
323 - if l.parent != nil && l.parent.baseContext != nil {
324 - if ctx := l.parent.baseContext(); ctx != nil {
325 - return ctx
326 - }
327 - }
328 - return context.Background()
329 -}
330 -
331 -func (l *listenerLease) isClosed() bool {
332 - if l.parent == nil || l.parent.ctxDone == nil {
333 - return false
334 - }
335 - select {
336 - case <-l.parent.ctxDone:
337 - return true
338 - default:
339 - return false
340 - }
494 +func cloneLeaseMetadata(metadata types.LeaseMetadata) types.LeaseMetadata {
495 + metadata.Tags = append([]string(nil), metadata.Tags...)
496 + return metadata
497 }
sdk/listener_test.go
+186 -35
@@ -1,31 +1,35 @@
1 package sdk
2
3 import (
4 + "context"
5 + "errors"
6 "net"
7 "testing"
8 +
9 + "github.com/gosuda/portal/v2/types"
10 )
11
12 func TestListenerSingleEntryAccessors(t *testing.T) {
13 t.Parallel()
14
11 - listener := &Listener{
12 - entries: []*listenerLease{
13 - {
14 - info: ListenerEntry{
15 - RelayURL: "https://relay.example.com",
16 - LeaseID: "lease-1",
17 - Hostnames: []string{"app.relay.example.com"},
18 - },
15 + listener := newListener(context.Background())
16 + listener.entries = []*listenerLease{
17 + {
18 + info: ListenerEntry{
19 + RelayURL: "https://relay.example.com",
20 + LeaseID: "lease-1",
21 + Hostnames: []string{"app.relay.example.com"},
22 },
23 + active: true,
24 },
25 }
26
23 - entry, ok := listener.singleEntry()
24 - if !ok {
25 - t.Fatal("singleEntry() ok = false, want true")
27 + entries := listener.Entries()
28 + if len(entries) != 1 {
29 + t.Fatalf("Entries() len = %d, want 1", len(entries))
30 }
27 - if entry.LeaseID != "lease-1" {
28 - t.Fatalf("singleEntry().LeaseID = %q, want %q", entry.LeaseID, "lease-1")
31 + if listener.Addr().String() != "portal:lease-1" {
32 + t.Fatalf("Addr().String() = %q, want %q", listener.Addr().String(), "portal:lease-1")
33 }
34
35 publicURLs := listener.PublicURLs()
@@ -37,27 +41,28 @@ func TestListenerSingleEntryAccessors(t *testing.T) {
41 func TestListenerMultiEntryAccessors(t *testing.T) {
42 t.Parallel()
43
40 - listener := &Listener{
41 - entries: []*listenerLease{
42 - {
43 - info: ListenerEntry{
44 - RelayURL: "https://relay-a.example.com",
45 - LeaseID: "lease-a",
46 - Hostnames: []string{"a.example.com"},
47 - },
44 + listener := newListener(context.Background())
45 + listener.entries = []*listenerLease{
46 + {
47 + info: ListenerEntry{
48 + RelayURL: "https://relay-a.example.com",
49 + LeaseID: "lease-a",
50 + Hostnames: []string{"a.example.com"},
51 },
49 - {
50 - info: ListenerEntry{
51 - RelayURL: "https://relay-b.example.com",
52 - LeaseID: "lease-b",
53 - Hostnames: []string{"b.example.com"},
54 - },
52 + active: true,
53 + },
54 + {
55 + info: ListenerEntry{
56 + RelayURL: "https://relay-b.example.com",
57 + LeaseID: "lease-b",
58 + Hostnames: []string{"b.example.com"},
59 },
60 + active: true,
61 },
62 }
63
59 - if _, ok := listener.singleEntry(); ok {
60 - t.Fatal("singleEntry() ok = true, want false")
64 + if listener.Addr().String() != "portal:multi" {
65 + t.Fatalf("Addr().String() = %q, want %q", listener.Addr().String(), "portal:multi")
66 }
67
68 entries := listener.Entries()
@@ -71,19 +76,49 @@ func TestListenerMultiEntryAccessors(t *testing.T) {
76 }
77 }
78
79 +func TestListenerEntriesSkipInactiveLeases(t *testing.T) {
80 + t.Parallel()
81 +
82 + listener := newListener(context.Background())
83 + listener.entries = []*listenerLease{
84 + {
85 + info: ListenerEntry{
86 + RelayURL: "https://relay-a.example.com",
87 + LeaseID: "lease-a",
88 + Hostnames: []string{"a.example.com"},
89 + },
90 + active: true,
91 + },
92 + {
93 + info: ListenerEntry{
94 + RelayURL: "https://relay-b.example.com",
95 + LeaseID: "lease-b",
96 + Hostnames: []string{"b.example.com"},
97 + },
98 + active: false,
99 + },
100 + }
101 +
102 + entries := listener.Entries()
103 + if len(entries) != 1 {
104 + t.Fatalf("Entries() len = %d, want 1", len(entries))
105 + }
106 + if entries[0].LeaseID != "lease-a" {
107 + t.Fatalf("Entries()[0].LeaseID = %q, want %q", entries[0].LeaseID, "lease-a")
108 + }
109 +}
110 +
111 func TestListenerAcceptEntry(t *testing.T) {
112 t.Parallel()
113
77 - done := make(chan struct{})
114 + listener := newListener(context.Background())
115 + listener.accepted = make(chan acceptedConn, 2)
116 +
117 serverConn1, clientConn1 := net.Pipe()
118 defer clientConn1.Close()
119 serverConn2, clientConn2 := net.Pipe()
120 defer clientConn2.Close()
121
83 - listener := &Listener{
84 - ctxDone: done,
85 - accepted: make(chan acceptedConn, 1),
86 - }
122 listener.accepted <- acceptedConn{
123 conn: serverConn1,
124 entry: ListenerEntry{
@@ -92,6 +127,7 @@ func TestListenerAcceptEntry(t *testing.T) {
127 Hostnames: []string{"app.relay.example.com"},
128 },
129 }
130 + listener.accepted <- acceptedConn{conn: serverConn2}
131
132 conn, entry, err := listener.AcceptEntry()
133 if err != nil {
@@ -106,7 +142,6 @@ func TestListenerAcceptEntry(t *testing.T) {
142 t.Fatalf("AcceptEntry().LeaseID = %q, want %q", entry.LeaseID, "lease-1")
143 }
144
109 - listener.accepted <- acceptedConn{conn: serverConn2}
145 plainConn, err := listener.Accept()
146 if err != nil {
147 t.Fatalf("Accept() error = %v", err)
@@ -116,3 +151,119 @@ func TestListenerAcceptEntry(t *testing.T) {
151 t.Fatal("Accept() did not return the original connection")
152 }
153 }
154 +
155 +func TestListenerAcceptEntryReturnsTerminalError(t *testing.T) {
156 + t.Parallel()
157 +
158 + listener := newListener(context.Background())
159 + listener.accepted = make(chan acceptedConn, 1)
160 +
161 + wantErr := &types.APIRequestError{Code: types.APIErrorCodeUnauthorized, Message: "bad reverse token"}
162 + listener.fail(wantErr)
163 +
164 + conn, entry, err := listener.AcceptEntry()
165 + if conn != nil {
166 + t.Fatal("AcceptEntry() conn != nil, want nil")
167 + }
168 + if entry.RelayURL != "" || entry.LeaseID != "" || len(entry.Hostnames) != 0 || len(entry.Metadata.Tags) != 0 {
169 + t.Fatalf("AcceptEntry() entry = %#v, want zero value", entry)
170 + }
171 + if !errors.Is(err, wantErr) {
172 + t.Fatalf("AcceptEntry() error = %v, want %v", err, wantErr)
173 + }
174 +}
175 +
176 +func TestListenerStopEntryKeepsOtherLeasesActive(t *testing.T) {
177 + t.Parallel()
178 +
179 + listener := newListener(context.Background())
180 + entryA := &listenerLease{
181 + parent: listener,
182 + info: ListenerEntry{
183 + RelayURL: "https://relay-a.example.com",
184 + LeaseID: "lease-a",
185 + Hostnames: []string{"a.example.com"},
186 + },
187 + active: true,
188 + }
189 + entryB := &listenerLease{
190 + parent: listener,
191 + info: ListenerEntry{
192 + RelayURL: "https://relay-b.example.com",
193 + LeaseID: "lease-b",
194 + Hostnames: []string{"b.example.com"},
195 + },
196 + active: true,
197 + }
198 + listener.entries = []*listenerLease{entryA, entryB}
199 + listener.activeCount = 2
200 +
201 + entryA.stop(&types.APIRequestError{Code: types.APIErrorCodeUnauthorized, Message: "stopped"})
202 +
203 + if listener.isClosed() {
204 + t.Fatal("listener closed after stopping one entry, want active")
205 + }
206 +
207 + entries := listener.Entries()
208 + if len(entries) != 1 {
209 + t.Fatalf("Entries() len = %d, want 1", len(entries))
210 + }
211 + if entries[0].LeaseID != "lease-b" {
212 + t.Fatalf("Entries()[0].LeaseID = %q, want %q", entries[0].LeaseID, "lease-b")
213 + }
214 +}
215 +
216 +func TestListenerStopLastEntryCancelsListener(t *testing.T) {
217 + t.Parallel()
218 +
219 + listener := newListener(context.Background())
220 + entry := &listenerLease{
221 + parent: listener,
222 + info: ListenerEntry{
223 + RelayURL: "https://relay.example.com",
224 + LeaseID: "lease-1",
225 + Hostnames: []string{"app.relay.example.com"},
226 + },
227 + active: true,
228 + }
229 + listener.entries = []*listenerLease{entry}
230 + listener.activeCount = 1
231 + listener.accepted = make(chan acceptedConn, 1)
232 +
233 + wantErr := &types.APIRequestError{Code: types.APIErrorCodeLeaseNotFound, Message: "lease disappeared"}
234 + entry.stop(wantErr)
235 +
236 + if !listener.isClosed() {
237 + t.Fatal("listener is still active after stopping last entry")
238 + }
239 +
240 + _, _, err := listener.AcceptEntry()
241 + if !errors.Is(err, wantErr) {
242 + t.Fatalf("AcceptEntry() error = %v, want %v", err, wantErr)
243 + }
244 +}
245 +
246 +func TestListenerEntryCloneDeepCopiesMetadataTags(t *testing.T) {
247 + t.Parallel()
248 +
249 + entry := ListenerEntry{
250 + RelayURL: "https://relay.example.com",
251 + LeaseID: "lease-1",
252 + Hostnames: []string{"app.relay.example.com"},
253 + Metadata: types.LeaseMetadata{
254 + Owner: "alice",
255 + Tags: []string{"one", "two"},
256 + },
257 + }
258 +
259 + clone := entry.clone()
260 + clone.Hostnames[0] = "changed.example.com"
261 + clone.Metadata.Tags[0] = "changed"
262 +
263 + if entry.Hostnames[0] != "app.relay.example.com" {
264 + t.Fatalf("entry.Hostnames[0] = %q, want %q", entry.Hostnames[0], "app.relay.example.com")
265 + }
266 + if entry.Metadata.Tags[0] != "one" {
267 + t.Fatalf("entry.Metadata.Tags[0] = %q, want %q", entry.Metadata.Tags[0], "one")
268 + }
269 +}
types/api.go
+43 -3
@@ -1,6 +1,10 @@
1 package types
2
3 -import "time"
3 +import (
4 + "fmt"
5 + "strings"
6 + "time"
7 +)
8
9 const (
10 HeaderReverseToken = "X-Portal-Token"
@@ -8,8 +12,8 @@ const (
12 MarkerTLSStart = byte(0x02)
13 )
14
11 -type APIEnvelope struct {
12 - Data any `json:"data,omitempty"`
15 +type APIEnvelope[T any] struct {
16 + Data T `json:"data,omitempty"`
17 Error *APIError `json:"error,omitempty"`
18 OK bool `json:"ok"`
19 }
@@ -19,6 +23,42 @@ type APIError struct {
23 Message string `json:"message"`
24 }
25
26 +type APIRequestError struct {
27 + StatusCode int `json:"-"`
28 + Code string `json:"code,omitempty"`
29 + Message string `json:"message,omitempty"`
30 +}
31 +
32 +func (e *APIRequestError) Error() string {
33 + if e == nil {
34 + return ""
35 + }
36 + if strings.TrimSpace(e.Code) != "" {
37 + return e.Code + ": " + strings.TrimSpace(e.Message)
38 + }
39 + if strings.TrimSpace(e.Message) != "" {
40 + return strings.TrimSpace(e.Message)
41 + }
42 + if e.StatusCode > 0 {
43 + return fmt.Sprintf("api request failed with status %d", e.StatusCode)
44 + }
45 + return "api request failed"
46 +}
47 +
48 +func (e *APIRequestError) Is(target error) bool {
49 + other, ok := target.(*APIRequestError)
50 + if !ok {
51 + return false
52 + }
53 + if other.Code != "" && e.Code != other.Code {
54 + return false
55 + }
56 + if other.StatusCode != 0 && e.StatusCode != other.StatusCode {
57 + return false
58 + }
59 + return true
60 +}
61 +
62 type LeaseMetadata struct {
63 Description string `json:"description,omitempty"`
64 Owner string `json:"owner,omitempty"`
types/error_codes.go new
+23
@@ -0,0 +1,23 @@
1 +package types
2 +
3 +const (
4 + APIErrorCodeAuthDisabled = "auth_disabled"
5 + APIErrorCodeAuthLocked = "auth_locked"
6 + APIErrorCodeFeatureUnavailable = "feature_unavailable"
7 + APIErrorCodeHijackFailed = "hijack_failed"
8 + APIErrorCodeHijackUnsupported = "hijack_unsupported"
9 + APIErrorCodeHostnameConflict = "hostname_conflict"
10 + APIErrorCodeHTTP11Only = "http11_only"
11 + APIErrorCodeInvalidIP = "invalid_ip"
12 + APIErrorCodeInvalidJSON = "invalid_json"
13 + APIErrorCodeInvalidKey = "invalid_key"
14 + APIErrorCodeInvalidLeaseID = "invalid_lease_id"
15 + APIErrorCodeInvalidMode = "invalid_mode"
16 + APIErrorCodeInvalidRequest = "invalid_request"
17 + APIErrorCodeIPBanned = "ip_banned"
18 + APIErrorCodeLeaseNotFound = "lease_not_found"
19 + APIErrorCodeLeaseRejected = "lease_rejected"
20 + APIErrorCodeMethodNotAllowed = "method_not_allowed"
21 + APIErrorCodeSessionCreateFailed = "session_create_failed"
22 + APIErrorCodeUnauthorized = "unauthorized"
23 +)