@samitouri / QOSAMI-X402Tunnel-1 / commits / a3af94e0

feat: Implement multi-hop support in the overlay and API client

Kim committed Apr 16, 2026 at 20:16 UTC a3af94e06a1e8fd9eb537fdd080cb417a8b0d07c
16 files changed +789 -803
portal/api_server.go
+122 -76
@@ -137,6 +137,18 @@ func (s *Server) apiHandler(base *http.ServeMux, keylessSignerHandler http.Handl
137 })
138 }
139
140 +func (s *Server) hopRegistryHandler() http.Handler {
141 + apiHandler := s.apiHandler(nil, nil)
142 + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
143 + switch strings.TrimSpace(r.URL.Path) {
144 + case types.PathSDKRegisterChallenge, types.PathSDKRegister, types.PathSDKRenew, types.PathSDKUnregister:
145 + apiHandler.ServeHTTP(w, r)
146 + default:
147 + utils.WriteAPIError(w, http.StatusNotFound, types.APIErrorCodeInvalidRequest, "unsupported hop registry path")
148 + }
149 + })
150 +}
151 +
152 func (s *Server) handleRoot(w http.ResponseWriter, _ *http.Request) {
153 utils.WriteAPIData(w, http.StatusOK, map[string]any{
154 "service": "portal-relay",
@@ -329,6 +341,30 @@ func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) {
341 if !ok {
342 return
343 }
344 + if req.Hop != nil {
345 + hop := req.Hop
346 + req.Hop = hop.Next
347 + status, body, err := s.forwardHopRegistry(r.Context(), hop, types.PathSDKRegister, req)
348 + if err != nil {
349 + writeAPIErrorResponse(w, err)
350 + return
351 + }
352 + if status >= http.StatusOK && status < http.StatusMultipleChoices {
353 + var resp struct {
354 + ExpiresAt time.Time `json:"expires_at"`
355 + }
356 + if err := utils.DecodeAPIData(body, &resp); err != nil {
357 + utils.WriteAPIError(w, http.StatusBadGateway, types.APIErrorCodeInvalidRequest, err.Error())
358 + return
359 + }
360 + if err := s.registry.RegisterHopRoute(hop, resp.ExpiresAt, time.Now()); err != nil {
361 + writeAPIErrorResponse(w, err)
362 + return
363 + }
364 + }
365 + utils.WriteRawAPIResponse(w, status, body)
366 + return
367 + }
368
369 challenge, err := s.registry.consumeVerifiedRegisterChallenge(req)
370 if err != nil {
@@ -367,6 +403,17 @@ func (s *Server) handleRegisterChallenge(w http.ResponseWriter, r *http.Request)
403 if !ok {
404 return
405 }
406 + if req.Hop != nil {
407 + hop := req.Hop
408 + req.Hop = hop.Next
409 + status, body, err := s.forwardHopRegistry(r.Context(), hop, types.PathSDKRegisterChallenge, req)
410 + if err != nil {
411 + writeAPIErrorResponse(w, err)
412 + return
413 + }
414 + utils.WriteRawAPIResponse(w, status, body)
415 + return
416 + }
417
418 scheme := "https"
419 if r.TLS == nil {
@@ -382,35 +429,11 @@ func (s *Server) handleRegisterChallenge(w http.ResponseWriter, r *http.Request)
429 Path: types.PathSDKRegister,
430 }).String()
431
385 - if len(req.MultiHop) > 0 && (s.hops == nil || s.overlay == nil || s.relaySet == nil) {
432 + req.HopToken = strings.TrimSpace(req.HopToken)
433 + if req.HopToken != "" && s.hopMux == nil {
434 utils.WriteAPIError(w, http.StatusServiceUnavailable, types.APIErrorCodeFeatureUnavailable, errFeatureUnavailable.Error())
435 return
436 }
389 - if len(req.MultiHop) > 0 {
390 - if len(req.MultiHop) < 2 {
391 - utils.InvalidRequestError(errors.New("multi-hop path requires entry and exit relays")).Write(w)
392 - return
393 - }
394 -
395 - multiHop := make([]types.RelayDescriptor, 0, len(req.MultiHop))
396 - for i, hop := range req.MultiHop {
397 - verified, err := auth.VerifyRelayDescriptor(hop)
398 - if err != nil {
399 - utils.InvalidRequestError(fmt.Errorf("hop %d: %w", i, err)).Write(w)
400 - return
401 - }
402 - multiHop = append(multiHop, verified)
403 - }
404 - if multiHop[0].APIHTTPSAddr == s.cfg.PortalURL {
405 - utils.InvalidRequestError(errors.New("entry relay must differ from exit relay")).Write(w)
406 - return
407 - }
408 - if multiHop[len(multiHop)-1].APIHTTPSAddr != s.cfg.PortalURL {
409 - utils.InvalidRequestError(errors.New("exit relay must match registration relay")).Write(w)
410 - return
411 - }
412 - req.MultiHop = multiHop
413 - }
437 if req.UDPEnabled && (!s.cfg.UDPEnabled || s.group != nil && s.quicTunnel == nil) {
438 utils.WriteAPIError(w, http.StatusServiceUnavailable, types.APIErrorCodeFeatureUnavailable, errFeatureUnavailable.Error())
439 return
@@ -443,6 +466,30 @@ func (s *Server) handleRenew(w http.ResponseWriter, r *http.Request) {
466 if !ok {
467 return
468 }
469 + if req.Hop != nil {
470 + hop := req.Hop
471 + req.Hop = hop.Next
472 + status, body, err := s.forwardHopRegistry(r.Context(), hop, types.PathSDKRenew, req)
473 + if err != nil {
474 + writeAPIErrorResponse(w, err)
475 + return
476 + }
477 + if status >= http.StatusOK && status < http.StatusMultipleChoices {
478 + var resp struct {
479 + ExpiresAt time.Time `json:"expires_at"`
480 + }
481 + if err := utils.DecodeAPIData(body, &resp); err != nil {
482 + utils.WriteAPIError(w, http.StatusBadGateway, types.APIErrorCodeInvalidRequest, err.Error())
483 + return
484 + }
485 + if err := s.registry.RegisterHopRoute(hop, resp.ExpiresAt, time.Now()); err != nil {
486 + writeAPIErrorResponse(w, err)
487 + return
488 + }
489 + }
490 + utils.WriteRawAPIResponse(w, status, body)
491 + return
492 + }
493
494 claims, err := auth.VerifyLeaseAccessToken(req.AccessToken, s.identity.PublicKey, s.cfg.PortalURL, time.Now().UTC())
495 if err != nil {
@@ -459,10 +506,6 @@ func (s *Server) handleRenew(w http.ResponseWriter, r *http.Request) {
506 writeAPIErrorResponse(w, err)
507 return
508 }
462 - if err := s.hops.renew(record); err != nil {
463 - utils.WriteAPIError(w, http.StatusServiceUnavailable, types.APIErrorCodeFeatureUnavailable, err.Error())
464 - return
465 - }
509 nextAccessToken, _, err := auth.IssueLeaseAccessToken(s.identity.PrivateKey, s.identity.Address, s.cfg.PortalURL, record.Copy(), ttl)
510 if err != nil {
511 utils.WriteAPIError(w, http.StatusInternalServerError, types.APIErrorCodeInternal, err.Error())
@@ -484,6 +527,20 @@ func (s *Server) handleUnregister(w http.ResponseWriter, r *http.Request) {
527 if !ok {
528 return
529 }
530 + if req.Hop != nil {
531 + hop := req.Hop
532 + req.Hop = hop.Next
533 + status, body, err := s.forwardHopRegistry(r.Context(), hop, types.PathSDKUnregister, req)
534 + if err != nil {
535 + writeAPIErrorResponse(w, err)
536 + return
537 + }
538 + if status >= http.StatusOK && status < http.StatusMultipleChoices {
539 + s.registry.DeleteHopRoute(hop)
540 + }
541 + utils.WriteRawAPIResponse(w, status, body)
542 + return
543 + }
544 claims, err := auth.VerifyLeaseAccessToken(req.AccessToken, s.identity.PublicKey, s.cfg.PortalURL, time.Now().UTC())
545 if err != nil {
546 utils.WriteAPIError(w, http.StatusForbidden, types.APIErrorCodeUnauthorized, errUnauthorized.Error())
@@ -495,7 +552,6 @@ func (s *Server) handleUnregister(w http.ResponseWriter, r *http.Request) {
552 writeAPIErrorResponse(w, err)
553 return
554 }
498 - s.hops.delete(record)
555 if record.isDirect() && s.acmeManager != nil {
556 deleteCtx, cancel := context.WithTimeout(context.Background(), defaultClaimTimeout)
557 if err := s.acmeManager.DeleteENSGaslessHostname(deleteCtx, record.Hostname); err != nil {
@@ -512,6 +568,37 @@ func (s *Server) handleUnregister(w http.ResponseWriter, r *http.Request) {
568 utils.WriteAPIData(w, http.StatusOK, map[string]any{})
569 }
570
571 +func (s *Server) forwardHopRegistry(ctx context.Context, route *types.HopRoute, path string, payload any) (int, []byte, error) {
572 + if s.hopMux == nil || s.overlay == nil || s.relaySet == nil {
573 + return 0, nil, errFeatureUnavailable
574 + }
575 + if route == nil {
576 + return 0, nil, errors.New("hop route is required")
577 + }
578 + forwardRelay, err := auth.VerifyRelayDescriptor(route.ForwardRelay)
579 + if err != nil {
580 + return 0, nil, fmt.Errorf("forward relay: %w", err)
581 + }
582 + if !forwardRelay.SupportsOverlayPeer ||
583 + strings.TrimSpace(forwardRelay.WireGuardPublicKey) == "" ||
584 + strings.TrimSpace(forwardRelay.WireGuardEndpoint) == "" ||
585 + strings.TrimSpace(forwardRelay.OverlayIPv4) == "" {
586 + return 0, nil, errors.New("forward relay wireguard overlay metadata is required")
587 + }
588 + if err := s.relaySet.InsertAnnounced(forwardRelay, time.Now().UTC()); err != nil {
589 + return 0, nil, fmt.Errorf("forward relay: %w", err)
590 + }
591 + if err := s.overlay.Sync(s.relaySet.OverlayPeerStates()); err != nil {
592 + return 0, nil, err
593 + }
594 +
595 + status, respBody, err := s.hopMux.OpenRegistry(ctx, forwardRelay.OverlayIPv4, path, payload)
596 + if err != nil {
597 + return 0, nil, &apiError{types.APIErrorCodeFeatureUnavailable, err.Error(), http.StatusBadGateway}
598 + }
599 + return status, respBody, nil
600 +}
601 +
602 func (s *Server) handleConnect(w http.ResponseWriter, r *http.Request) {
603 if !utils.RequireMethod(w, r, http.MethodGet) {
604 return
@@ -695,16 +782,10 @@ func (s *Server) registerLease(req types.RegisterChallengeRequest, clientIP, rep
782 }
783 issuedAt := claims.IssuedAt.Time().UTC()
784 expiresAt := claims.Expiry.Time().UTC()
698 - if len(req.MultiHop) > 0 && (s.hops == nil || s.overlay == nil || s.relaySet == nil) {
785 + req.HopToken = strings.TrimSpace(req.HopToken)
786 + if req.HopToken != "" && s.hopMux == nil {
787 return types.RegisterResponse{}, errFeatureUnavailable
788 }
701 - multiHop := append([]types.RelayDescriptor(nil), req.MultiHop...)
702 - if len(multiHop) > 0 {
703 - hostname, err = utils.LeaseHostname(identity.Name, utils.PortalRootHost(multiHop[0].APIHTTPSAddr))
704 - if err != nil {
705 - return types.RegisterResponse{}, err
706 - }
707 - }
789 identityKey := identity.Key()
790 stream := transport.NewRelayStream(identityKey, defaultIdleKeepalive, defaultReadyQueueLimit)
791 record := &leaseRecord{
@@ -718,14 +799,9 @@ func (s *Server) registerLease(req types.RegisterChallengeRequest, clientIP, rep
799 ReportedIP: utils.SanitizeReportedIP(reportedIP),
800 UDPEnabled: req.UDPEnabled,
801 TCPEnabled: req.TCPEnabled,
721 - multiHop: multiHop,
802 + hopToken: req.HopToken,
803 stream: stream,
804 }
724 - if record.isDirect() && s.hops != nil {
725 - if _, ok := s.hops.routeForHostname(record.Hostname, time.Now()); ok {
726 - return types.RegisterResponse{}, errHostnameConflict
727 - }
728 - }
805 if req.UDPEnabled {
806 if s.udpPorts == nil {
807 return types.RegisterResponse{}, errors.New("udp port allocation not available")
@@ -759,34 +835,7 @@ func (s *Server) registerLease(req types.RegisterChallengeRequest, clientIP, rep
835 return types.RegisterResponse{}, err
836 }
837
762 - replaced, err := s.registry.Register(record)
763 - if err != nil {
764 - record.Close()
765 - return types.RegisterResponse{}, err
766 - }
767 - s.hops.delete(replaced)
768 - if record.isDirect() && s.hops != nil {
769 - if _, ok := s.hops.routeForHostname(record.Hostname, time.Now()); ok {
770 - _, _ = s.registry.Unregister(record.Copy())
771 - record.Close()
772 - return types.RegisterResponse{}, errHostnameConflict
773 - }
774 - }
775 - if !record.isDirect() {
776 - now := time.Now().UTC()
777 - for i := 0; i < len(record.multiHop)-1; i++ {
778 - if err := s.relaySet.InsertAnnounced(record.multiHop[i], now); err != nil {
779 - _, _ = s.registry.Unregister(record.Copy())
780 - record.Close()
781 - return types.RegisterResponse{}, fmt.Errorf("sync hop %d peer: %w", i, err)
782 - }
783 - }
784 - if err := s.overlay.Sync(s.relaySet.OverlayPeerStates()); err != nil {
785 - log.Warn().Err(err).Msg("sync wireguard multi-hop peers")
786 - }
787 - }
788 - if err := s.hops.install(record); err != nil {
789 - _, _ = s.registry.Unregister(record.Copy())
838 + if err := s.registry.Register(record); err != nil {
839 record.Close()
840 return types.RegisterResponse{}, err
841 }
@@ -808,9 +857,6 @@ func (s *Server) registerLease(req types.RegisterChallengeRequest, clientIP, rep
857 UDPEnabled: record.UDPEnabled,
858 TCPEnabled: record.TCPEnabled,
859 }
811 - if !record.isDirect() {
812 - resp.KeylessURL = record.multiHop[0].APIHTTPSAddr
813 - }
860 if record.datagram != nil {
861 resp.SNIPort = s.cfg.SNIPort
862 resp.UDPAddr = fmt.Sprintf("%s:%d", s.identity.Name, record.datagram.UDPPort())
portal/auth/register_challenge.go
+1 -6
@@ -48,18 +48,13 @@ func NewRegisterChallenge(req types.RegisterChallengeRequest, domain, uri string
48 return nil, fmt.Errorf("build siwe message: %w", err)
49 }
50
51 - var multiHop []types.RelayDescriptor
52 - if len(req.MultiHop) > 0 {
53 - multiHop = append([]types.RelayDescriptor{}, req.MultiHop...)
54 - }
55 -
51 normalizedRequest := types.RegisterChallengeRequest{
52 Identity: normalizedIdentity,
53 Metadata: req.Metadata.Copy(),
54 TTL: req.TTL,
55 UDPEnabled: req.UDPEnabled,
56 TCPEnabled: req.TCPEnabled,
62 - MultiHop: multiHop,
57 + HopToken: strings.TrimSpace(req.HopToken),
58 }
59
60 return &RegisterChallenge{
portal/hop.go deleted
-436
@@ -1,436 +0,0 @@
1 -package portal
2 -
3 -import (
4 - "context"
5 - "errors"
6 - "fmt"
7 - "net"
8 - "strings"
9 - "sync"
10 - "time"
11 -
12 - "github.com/rs/zerolog/log"
13 - "golang.org/x/sync/errgroup"
14 -
15 - "github.com/gosuda/portal-tunnel/v2/portal/overlay"
16 - "github.com/gosuda/portal-tunnel/v2/utils"
17 -)
18 -
19 -var (
20 - errHopRouteConflict = errors.New("hop route conflict")
21 - errHopRouteNotFound = errors.New("hop route not found")
22 -)
23 -
24 -type hopRoute struct {
25 - matchHostname string
26 - matchToken string
27 - forwardOverlayIPv4 string
28 - forwardToken string
29 - identityKey string
30 - expiresAt time.Time
31 -}
32 -
33 -type hopManager struct {
34 - mux *overlay.HopMux
35 - registry *leaseRegistry
36 - proxy *proxy
37 - routesByID map[string]hopRoute
38 - mu sync.RWMutex
39 -}
40 -
41 -func newHopManager(ov *overlay.Overlay, registry *leaseRegistry, relayProxy *proxy) (*hopManager, error) {
42 - mux, err := overlay.NewHopMux(ov)
43 - if err != nil {
44 - return nil, err
45 - }
46 - return &hopManager{
47 - mux: mux,
48 - registry: registry,
49 - proxy: relayProxy,
50 - routesByID: make(map[string]hopRoute),
51 - }, nil
52 -}
53 -
54 -func (h *hopManager) install(record *leaseRecord) error {
55 - if record.isDirect() {
56 - return nil
57 - }
58 - if h == nil || h.mux == nil {
59 - return errFeatureUnavailable
60 - }
61 - if len(record.multiHop) < 2 {
62 - return errors.New("multi-hop lease path is invalid")
63 - }
64 -
65 - rawHopID, err := utils.RandomHex(32)
66 - if err != nil {
67 - return err
68 - }
69 - hopID := "hpr_" + rawHopID
70 - tokens := make([]string, len(record.multiHop)-1)
71 - for i := range tokens {
72 - token, err := utils.RandomHex(32)
73 - if err != nil {
74 - return err
75 - }
76 - tokens[i] = "hpt_" + token
77 - }
78 - record.hopID = hopID
79 -
80 - exitRouteID := fmt.Sprintf("%s_%d", record.hopID, len(record.multiHop)-1)
81 - if err := h.installRoute(exitRouteID, hopRoute{
82 - matchToken: tokens[len(tokens)-1],
83 - identityKey: record.Key(),
84 - expiresAt: record.ExpiresAt,
85 - }, time.Now()); err != nil {
86 - return err
87 - }
88 -
89 - type remoteRoute struct {
90 - overlayIPv4 string
91 - routeID string
92 - }
93 - installedRemoteRoutes := make([]remoteRoute, 0, len(record.multiHop)-1)
94 - rollback := true
95 - defer func() {
96 - if !rollback {
97 - return
98 - }
99 - h.deleteRoute(exitRouteID)
100 - for _, route := range installedRemoteRoutes {
101 - _ = h.mux.Control(context.Background(), route.overlayIPv4, overlay.HopControl{
102 - Action: overlay.HopControlDelete,
103 - Route: overlay.HopRouteSpec{
104 - RouteID: route.routeID,
105 - },
106 - })
107 - }
108 - }()
109 -
110 - ctx, cancel := context.WithTimeout(context.Background(), defaultClaimTimeout)
111 - defer cancel()
112 -
113 - for i := len(record.multiHop) - 2; i >= 0; i-- {
114 - current := record.multiHop[i]
115 - next := record.multiHop[i+1]
116 - routeID := fmt.Sprintf("%s_%d", record.hopID, i)
117 - route := overlay.HopRouteSpec{
118 - RouteID: routeID,
119 - ForwardOverlayIPv4: next.OverlayIPv4,
120 - ForwardToken: tokens[i],
121 - ExpiresAt: record.ExpiresAt,
122 - }
123 - if i == 0 {
124 - route.MatchHostname = record.Hostname
125 - } else {
126 - route.MatchToken = tokens[i-1]
127 - }
128 - if err := h.mux.Control(ctx, current.OverlayIPv4, overlay.HopControl{
129 - Action: overlay.HopControlInstall,
130 - Route: route,
131 - }); err != nil {
132 - return fmt.Errorf("install hop %d route: %w", i, err)
133 - }
134 - installedRemoteRoutes = append(installedRemoteRoutes, remoteRoute{
135 - overlayIPv4: current.OverlayIPv4,
136 - routeID: routeID,
137 - })
138 - }
139 -
140 - rollback = false
141 - return nil
142 -}
143 -
144 -func (h *hopManager) renew(record *leaseRecord) error {
145 - if record.isDirect() {
146 - return nil
147 - }
148 - if h == nil || h.mux == nil {
149 - return errFeatureUnavailable
150 - }
151 - ctx, cancel := context.WithTimeout(context.Background(), defaultClaimTimeout)
152 - defer cancel()
153 -
154 - if len(record.multiHop) < 2 || record.hopID == "" {
155 - return errors.New("multi-hop lease path is invalid")
156 - }
157 - if err := h.renewRoute(fmt.Sprintf("%s_%d", record.hopID, len(record.multiHop)-1), record.ExpiresAt, time.Now()); err != nil {
158 - return err
159 - }
160 - for i := len(record.multiHop) - 2; i >= 0; i-- {
161 - if err := h.mux.Control(ctx, record.multiHop[i].OverlayIPv4, overlay.HopControl{
162 - Action: overlay.HopControlRenew,
163 - Route: overlay.HopRouteSpec{
164 - RouteID: fmt.Sprintf("%s_%d", record.hopID, i),
165 - ExpiresAt: record.ExpiresAt,
166 - },
167 - }); err != nil {
168 - return fmt.Errorf("renew hop %d route: %w", i, err)
169 - }
170 - }
171 - return nil
172 -}
173 -
174 -func (h *hopManager) delete(record *leaseRecord) {
175 - if record == nil || record.isDirect() {
176 - return
177 - }
178 - if h == nil || h.mux == nil {
179 - return
180 - }
181 - h.deleteRoute(fmt.Sprintf("%s_%d", record.hopID, len(record.multiHop)-1))
182 - ctx, cancel := context.WithTimeout(context.Background(), defaultClaimTimeout)
183 - defer cancel()
184 - for i := 0; i < len(record.multiHop)-1; i++ {
185 - _ = h.mux.Control(ctx, record.multiHop[i].OverlayIPv4, overlay.HopControl{
186 - Action: overlay.HopControlDelete,
187 - Route: overlay.HopRouteSpec{
188 - RouteID: fmt.Sprintf("%s_%d", record.hopID, i),
189 - },
190 - })
191 - }
192 -}
193 -
194 -func (h *hopManager) run(ctx context.Context) error {
195 - if h == nil || h.mux == nil {
196 - <-ctx.Done()
197 - return nil
198 - }
199 - group, groupCtx := errgroup.WithContext(ctx)
200 - group.Go(func() error {
201 - return h.mux.Serve(groupCtx)
202 - })
203 - group.Go(func() error {
204 - for {
205 - stream, err := h.mux.Accept(groupCtx)
206 - if err != nil {
207 - if groupCtx.Err() != nil {
208 - return nil
209 - }
210 - return err
211 - }
212 - if stream.Control != nil {
213 - _ = stream.Respond(h.handleControl(*stream.Control))
214 - continue
215 - }
216 - go h.handleStream(groupCtx, stream.Conn, stream.Token)
217 - }
218 - })
219 - return group.Wait()
220 -}
221 -
222 -func (h *hopManager) handleStream(ctx context.Context, conn net.Conn, token string) {
223 - if h == nil || h.registry == nil {
224 - _ = conn.Close()
225 - return
226 - }
227 - route, record, ok := h.routeForToken(token, time.Now())
228 - if !ok {
229 - _ = conn.Close()
230 - return
231 - }
232 - if record != nil {
233 - if record.stream == nil || !h.registry.policy.IsIdentityRoutable(record.Key()) {
234 - _ = conn.Close()
235 - return
236 - }
237 - claimCtx, cancel := context.WithTimeout(ctx, defaultClaimTimeout)
238 - session, err := record.stream.Claim(claimCtx)
239 - cancel()
240 - if err != nil {
241 - _ = conn.Close()
242 - return
243 - }
244 - h.proxy.bridge(conn, session, record.Key(), h.registry.policy.BPSManager())
245 - return
246 - }
247 - if route.forwardOverlayIPv4 == "" || route.forwardToken == "" || h.mux == nil {
248 - _ = conn.Close()
249 - return
250 - }
251 - next, err := h.mux.OpenStream(ctx, route.forwardOverlayIPv4, route.forwardToken)
252 - if err != nil {
253 - _ = conn.Close()
254 - log.Warn().Err(err).Str("next_overlay_ipv4", route.forwardOverlayIPv4).Msg("open next hop stream")
255 - return
256 - }
257 - h.proxy.bridge(conn, next, "", nil)
258 -}
259 -
260 -func (h *hopManager) handleControl(control overlay.HopControl) error {
261 - if h == nil || h.registry == nil {
262 - return errFeatureUnavailable
263 - }
264 - switch strings.TrimSpace(control.Action) {
265 - case overlay.HopControlInstall:
266 - return h.installRoute(control.Route.RouteID, hopRoute{
267 - matchHostname: control.Route.MatchHostname,
268 - matchToken: control.Route.MatchToken,
269 - forwardOverlayIPv4: control.Route.ForwardOverlayIPv4,
270 - forwardToken: control.Route.ForwardToken,
271 - expiresAt: control.Route.ExpiresAt,
272 - }, time.Now())
273 - case overlay.HopControlRenew:
274 - return h.renewRoute(control.Route.RouteID, control.Route.ExpiresAt, time.Now())
275 - case overlay.HopControlDelete:
276 - h.deleteRoute(control.Route.RouteID)
277 - return nil
278 - default:
279 - return errors.New("unknown hop control action")
280 - }
281 -}
282 -
283 -func (h *hopManager) forwardHostname(ctx context.Context, conn net.Conn, hostname string) bool {
284 - if h == nil || h.registry == nil || h.mux == nil {
285 - return false
286 - }
287 - route, ok := h.routeForHostname(hostname, time.Now())
288 - if !ok || route.forwardOverlayIPv4 == "" || route.forwardToken == "" {
289 - return false
290 - }
291 - next, err := h.mux.OpenStream(ctx, route.forwardOverlayIPv4, route.forwardToken)
292 - if err != nil {
293 - log.Warn().Err(err).Str("next_overlay_ipv4", route.forwardOverlayIPv4).Msg("open next hop stream")
294 - return false
295 - }
296 - h.proxy.bridge(conn, next, "", nil)
297 - return true
298 -}
299 -
300 -func (h *hopManager) close() error {
301 - if h == nil || h.mux == nil {
302 - return nil
303 - }
304 - return h.mux.Close()
305 -}
306 -
307 -func (h *hopManager) routeForHostname(hostname string, now time.Time) (hopRoute, bool) {
308 - hostname = utils.NormalizeHostname(hostname)
309 - if h == nil || hostname == "" {
310 - return hopRoute{}, false
311 - }
312 -
313 - h.mu.RLock()
314 - defer h.mu.RUnlock()
315 -
316 - for _, route := range h.routesByID {
317 - if route.matchHostname == hostname && now.Before(route.expiresAt) {
318 - return route, true
319 - }
320 - }
321 - return hopRoute{}, false
322 -}
323 -
324 -func (h *hopManager) routeForToken(token string, now time.Time) (hopRoute, *leaseRecord, bool) {
325 - token = strings.TrimSpace(token)
326 - if h == nil || token == "" {
327 - return hopRoute{}, nil, false
328 - }
329 -
330 - h.mu.RLock()
331 - defer h.mu.RUnlock()
332 -
333 - for _, route := range h.routesByID {
334 - if route.matchToken != token || !now.Before(route.expiresAt) {
335 - continue
336 - }
337 - if route.identityKey == "" {
338 - return route, nil, true
339 - }
340 - record, ok := h.registry.RecordByKey(route.identityKey, now)
341 - if !ok {
342 - return hopRoute{}, nil, false
343 - }
344 - return route, record, true
345 - }
346 - return hopRoute{}, nil, false
347 -}
348 -
349 -func (h *hopManager) installRoute(routeID string, route hopRoute, now time.Time) error {
350 - routeID = strings.TrimSpace(routeID)
351 - route.matchHostname = utils.NormalizeHostname(route.matchHostname)
352 - route.matchToken = strings.TrimSpace(route.matchToken)
353 - route.forwardOverlayIPv4 = strings.TrimSpace(route.forwardOverlayIPv4)
354 - route.forwardToken = strings.TrimSpace(route.forwardToken)
355 - route.identityKey = strings.TrimSpace(route.identityKey)
356 - route.expiresAt = route.expiresAt.UTC()
357 -
358 - switch {
359 - case h == nil:
360 - return errFeatureUnavailable
361 - case routeID == "":
362 - return errors.New("route id is required")
363 - case !route.expiresAt.After(now):
364 - return errors.New("route expiry must be in the future")
365 - case route.matchHostname == "" && route.matchToken == "":
366 - return errors.New("hostname or token matcher is required")
367 - case route.matchHostname != "" && route.matchToken != "":
368 - return errors.New("hostname and token matchers are mutually exclusive")
369 - case route.identityKey == "" && route.forwardOverlayIPv4 == "":
370 - return errors.New("forward overlay ipv4 is required")
371 - case route.identityKey == "" && route.forwardToken == "":
372 - return errors.New("forward token is required")
373 - }
374 - if route.matchHostname != "" {
375 - if record, ok := h.registry.Lookup(route.matchHostname); ok && record != nil && now.Before(record.ExpiresAt) {
376 - return errHostnameConflict
377 - }
378 - }
379 -
380 - h.mu.Lock()
381 - defer h.mu.Unlock()
382 -
383 - if h.routesByID == nil {
384 - h.routesByID = make(map[string]hopRoute)
385 - }
386 - for existingID, existing := range h.routesByID {
387 - if existingID == routeID {
388 - continue
389 - }
390 - if !now.Before(existing.expiresAt) {
391 - delete(h.routesByID, existingID)
392 - continue
393 - }
394 - if route.matchHostname != "" && existing.matchHostname == route.matchHostname {
395 - return errHopRouteConflict
396 - }
397 - if route.matchToken != "" && existing.matchToken == route.matchToken {
398 - return errHopRouteConflict
399 - }
400 - }
401 - h.routesByID[routeID] = route
402 - return nil
403 -}
404 -
405 -func (h *hopManager) renewRoute(routeID string, expiresAt, now time.Time) error {
406 - routeID = strings.TrimSpace(routeID)
407 - expiresAt = expiresAt.UTC()
408 - if routeID == "" {
409 - return errors.New("route id is required")
410 - }
411 - if !expiresAt.After(now) {
412 - return errors.New("route expiry must be in the future")
413 - }
414 -
415 - h.mu.Lock()
416 - defer h.mu.Unlock()
417 -
418 - route, ok := h.routesByID[routeID]
419 - if !ok {
420 - return errHopRouteNotFound
421 - }
422 - route.expiresAt = expiresAt
423 - h.routesByID[routeID] = route
424 - return nil
425 -}
426 -
427 -func (h *hopManager) deleteRoute(routeID string) {
428 - routeID = strings.TrimSpace(routeID)
429 - if h == nil || routeID == "" {
430 - return
431 - }
432 -
433 - h.mu.Lock()
434 - delete(h.routesByID, routeID)
435 - h.mu.Unlock()
436 -}
portal/lease.go
+173 -41
@@ -18,8 +18,9 @@ import (
18 const defaultRegisterChallengeTTL = 2 * time.Minute
19
20 type leaseRegistry struct {
21 - leasesByHostname map[string]string
21 leasesByKey map[string]*leaseRecord
22 + recordsByHostname map[string]*leaseRecord
23 + recordsByHopToken map[string]*leaseRecord
24 registerChallenges map[string]*auth.RegisterChallenge
25 policy *policy.Runtime
26 mu sync.RWMutex
@@ -32,8 +33,9 @@ func newLeaseRegistry(udpEnabled, tcpPortEnabled bool, trustProxyHeaders bool, r
33 }
34
35 return &leaseRegistry{
35 - leasesByHostname: make(map[string]string),
36 leasesByKey: make(map[string]*leaseRecord),
37 + recordsByHostname: make(map[string]*leaseRecord),
38 + recordsByHopToken: make(map[string]*leaseRecord),
39 registerChallenges: make(map[string]*auth.RegisterChallenge),
40 policy: runtime,
41 }, nil
@@ -48,8 +50,9 @@ func (r *leaseRegistry) CloseAll() []*leaseRecord {
50 out = append(out, record)
51 r.policy.ForgetIdentity(record.Key())
52 }
51 - r.leasesByHostname = make(map[string]string)
53 r.leasesByKey = make(map[string]*leaseRecord)
54 + r.recordsByHostname = make(map[string]*leaseRecord)
55 + r.recordsByHopToken = make(map[string]*leaseRecord)
56 r.registerChallenges = make(map[string]*auth.RegisterChallenge)
57 return out
58 }
@@ -63,54 +66,66 @@ func (r *leaseRegistry) Lookup(host string) (*leaseRecord, bool) {
66 r.mu.RLock()
67 defer r.mu.RUnlock()
68
66 - key, ok := r.leasesByHostname[host]
67 - if !ok {
69 + record := r.lookupLocked(host)
70 + return record, record != nil
71 +}
72 +
73 +func (r *leaseRegistry) lookupLocked(host string) *leaseRecord {
74 + record := r.recordsByHostname[host]
75 + if record == nil {
76 parts := strings.Split(host, ".")
77 if len(parts) < 3 {
70 - return nil, false
71 - }
72 - key, ok = r.leasesByHostname["*."+strings.Join(parts[1:], ".")]
73 - if !ok {
74 - return nil, false
78 + return nil
79 }
80 + record = r.recordsByHostname["*."+strings.Join(parts[1:], ".")]
81 }
77 - record, ok := r.leasesByKey[key]
78 - return record, ok && record != nil
82 + return record
83 }
84
81 -func (r *leaseRegistry) Register(record *leaseRecord) (*leaseRecord, error) {
85 +func (r *leaseRegistry) Register(record *leaseRecord) error {
86 if record == nil {
83 - return nil, errors.New("lease record is required")
87 + return errors.New("lease record is required")
88 }
89
90 key := record.Key()
91 if key == "" {
88 - return nil, errors.New("lease identity is required")
92 + return errors.New("lease identity is required")
93 }
94 hostname := utils.NormalizeHostname(record.Hostname)
95 if hostname == "" {
92 - return nil, errors.New("lease hostname is required")
96 + return errors.New("lease hostname is required")
97 }
98 + record.hopToken = strings.TrimSpace(record.hopToken)
99
100 r.mu.Lock()
101
97 - if existingKey, ok := r.leasesByHostname[hostname]; ok && existingKey != key {
98 - r.mu.Unlock()
99 - return nil, errHostnameConflict
102 + now := time.Now()
103 + if record.isDirect() {
104 + if existing := r.recordsByHostname[hostname]; existing != nil && existing.Key() != key && now.Before(existing.ExpiresAt) {
105 + r.mu.Unlock()
106 + return errHostnameConflict
107 + }
108 + }
109 + if record.hopToken != "" {
110 + if existing := r.recordsByHopToken[record.hopToken]; existing != nil && existing.Key() != key && now.Before(existing.ExpiresAt) {
111 + r.mu.Unlock()
112 + return errors.New("hop token conflict")
113 + }
114 }
115
116 var replaced *leaseRecord
117 if existing, ok := r.leasesByKey[key]; ok && existing != nil {
118 replaced = existing
105 - if existing != record && existing.isDirect() {
106 - delete(r.leasesByHostname, utils.NormalizeHostname(existing.Hostname))
107 - }
119 + r.deleteIndexesLocked(existing)
120 r.policy.ForgetIdentity(existing.Key())
121 }
122 record.Hostname = hostname
123 r.leasesByKey[key] = record
124 if record.isDirect() {
113 - r.leasesByHostname[hostname] = key
125 + r.recordsByHostname[hostname] = record
126 + }
127 + if record.hopToken != "" {
128 + r.recordsByHopToken[record.hopToken] = record
129 }
130 r.policy.IPFilter().RegisterIdentityIP(key, record.ClientIP)
131 r.mu.Unlock()
@@ -118,7 +133,7 @@ func (r *leaseRegistry) Register(record *leaseRecord) (*leaseRecord, error) {
133 if replaced != nil && replaced != record {
134 replaced.Close()
135 }
121 - return replaced, nil
136 + return nil
137 }
138
139 func (r *leaseRegistry) Renew(identity types.Identity, ttl time.Duration, clientIP, reportedIP string) (*leaseRecord, error) {
@@ -155,9 +170,7 @@ func (r *leaseRegistry) Unregister(identity types.Identity) (*leaseRecord, error
170 }
171
172 delete(r.leasesByKey, key)
158 - if record.isDirect() {
159 - delete(r.leasesByHostname, utils.NormalizeHostname(record.Hostname))
160 - }
173 + r.deleteIndexesLocked(record)
174 r.policy.ForgetIdentity(key)
175 return record, nil
176 }
@@ -178,8 +191,94 @@ func (r *leaseRegistry) RecordByKey(key string, now time.Time) (*leaseRecord, bo
191 return record, true
192 }
193
194 +func (r *leaseRegistry) RecordByHopToken(token string, now time.Time) (*leaseRecord, bool) {
195 + token = strings.TrimSpace(token)
196 + if token == "" {
197 + return nil, false
198 + }
199 +
200 + r.mu.RLock()
201 + defer r.mu.RUnlock()
202 +
203 + record := r.recordsByHopToken[token]
204 + if record != nil && now.Before(record.ExpiresAt) {
205 + return record, true
206 + }
207 + return nil, false
208 +}
209 +
210 +func (r *leaseRegistry) RegisterHopRoute(route *types.HopRoute, expiresAt, now time.Time) error {
211 + if route == nil {
212 + return errors.New("hop route is required")
213 + }
214 + matchHostname := utils.NormalizeHostname(route.MatchHostname)
215 + matchToken := strings.TrimSpace(route.MatchToken)
216 + overlayIPv4 := strings.TrimSpace(route.ForwardRelay.OverlayIPv4)
217 + forwardToken := strings.TrimSpace(route.ForwardToken)
218 + expiresAt = expiresAt.UTC()
219 +
220 + switch {
221 + case r == nil:
222 + return errFeatureUnavailable
223 + case !expiresAt.After(now):
224 + return errors.New("route expiry must be in the future")
225 + case matchHostname == "" && matchToken == "":
226 + return errors.New("hostname or token matcher is required")
227 + case matchHostname != "" && matchToken != "":
228 + return errors.New("hostname and token matchers are mutually exclusive")
229 + case overlayIPv4 == "":
230 + return errors.New("forward overlay ipv4 is required")
231 + case forwardToken == "":
232 + return errors.New("forward token is required")
233 + }
234 +
235 + r.mu.Lock()
236 + defer r.mu.Unlock()
237 +
238 + record := &leaseRecord{
239 + Hostname: matchHostname,
240 + ExpiresAt: expiresAt,
241 + hopToken: matchToken,
242 + hopNextOverlayIPv4: overlayIPv4,
243 + hopNextToken: forwardToken,
244 + }
245 + if matchHostname != "" {
246 + if existing := r.lookupLocked(matchHostname); existing != nil && !existing.isHopForward() && now.Before(existing.ExpiresAt) {
247 + return errHostnameConflict
248 + }
249 + r.recordsByHostname[matchHostname] = record
250 + return nil
251 + }
252 + if existing := r.recordsByHopToken[matchToken]; existing != nil && !existing.isHopForward() && now.Before(existing.ExpiresAt) {
253 + return errors.New("hop token conflict")
254 + }
255 + r.recordsByHopToken[matchToken] = record
256 + return nil
257 +}
258 +
259 +func (r *leaseRegistry) DeleteHopRoute(route *types.HopRoute) {
260 + if r == nil || route == nil {
261 + return
262 + }
263 + hostname := utils.NormalizeHostname(route.MatchHostname)
264 + token := strings.TrimSpace(route.MatchToken)
265 +
266 + r.mu.Lock()
267 + if hostname != "" {
268 + if record := r.recordsByHostname[hostname]; record != nil && record.isHopForward() {
269 + delete(r.recordsByHostname, hostname)
270 + }
271 + }
272 + if token != "" {
273 + if record := r.recordsByHopToken[token]; record != nil && record.isHopForward() {
274 + delete(r.recordsByHopToken, token)
275 + }
276 + }
277 + r.mu.Unlock()
278 +}
279 +
280 func (r *leaseRegistry) issueRegisterChallenge(req types.RegisterChallengeRequest, domain, uri string) (types.RegisterChallengeResponse, error) {
182 - if len(req.MultiHop) > 0 && (req.UDPEnabled || req.TCPEnabled) {
281 + if strings.TrimSpace(req.HopToken) != "" && (req.UDPEnabled || req.TCPEnabled) {
282 return types.RegisterChallengeResponse{}, errTransportMismatch
283 }
284 if req.UDPEnabled {
@@ -266,9 +365,7 @@ func (r *leaseRegistry) cleanupExpired(now time.Time) []*leaseRecord {
365 if now.After(record.ExpiresAt) {
366 expired = append(expired, record)
367 delete(r.leasesByKey, key)
269 - if record.isDirect() {
270 - delete(r.leasesByHostname, utils.NormalizeHostname(record.Hostname))
271 - }
368 + r.deleteIndexesLocked(record)
369 r.policy.ForgetIdentity(key)
370 }
371 }
@@ -277,6 +374,16 @@ func (r *leaseRegistry) cleanupExpired(now time.Time) []*leaseRecord {
374 delete(r.registerChallenges, challengeID)
375 }
376 }
377 + for hostname, record := range r.recordsByHostname {
378 + if record != nil && !now.Before(record.ExpiresAt) {
379 + delete(r.recordsByHostname, hostname)
380 + }
381 + }
382 + for token, record := range r.recordsByHopToken {
383 + if record != nil && !now.Before(record.ExpiresAt) {
384 + delete(r.recordsByHopToken, token)
385 + }
386 + }
387 return expired
388 }
389
@@ -378,19 +485,44 @@ type leaseRecord struct {
485 UDPEnabled bool
486 TCPEnabled bool
487 Metadata types.LeaseMetadata
381 - multiHop []types.RelayDescriptor
382 - hopID string
383 - datagram *transport.RelayDatagram
384 - udpPorts *transport.PortAllocator
385 - tcpPort *transport.RelayTCPPort
386 - tcpPorts *transport.PortAllocator
387 - stream *transport.RelayStream
388 - startErr error
389 - startOnce sync.Once
488 +
489 + hopToken string
490 + hopNextOverlayIPv4 string
491 + hopNextToken string
492 +
493 + datagram *transport.RelayDatagram
494 + udpPorts *transport.PortAllocator
495 + tcpPort *transport.RelayTCPPort
496 + tcpPorts *transport.PortAllocator
497 + stream *transport.RelayStream
498 + startErr error
499 + startOnce sync.Once
500 }
501
502 func (r *leaseRecord) isDirect() bool {
393 - return len(r.multiHop) == 0
503 + return r == nil || (strings.TrimSpace(r.hopToken) == "" && !r.isHopForward())
504 +}
505 +
506 +func (r *leaseRecord) isHopForward() bool {
507 + return r != nil && (strings.TrimSpace(r.hopNextOverlayIPv4) != "" || strings.TrimSpace(r.hopNextToken) != "")
508 +}
509 +
510 +func (r *leaseRegistry) deleteIndexesLocked(record *leaseRecord) {
511 + if record == nil {
512 + return
513 + }
514 + hostname := utils.NormalizeHostname(record.Hostname)
515 + if hostname != "" {
516 + if r.recordsByHostname[hostname] == record {
517 + delete(r.recordsByHostname, hostname)
518 + }
519 + }
520 + token := strings.TrimSpace(record.hopToken)
521 + if token != "" {
522 + if r.recordsByHopToken[token] == record {
523 + delete(r.recordsByHopToken, token)
524 + }
525 + }
526 }
527
528 func (r *leaseRegistry) AdminSnapshot(record *leaseRecord) types.AdminLease {
portal/lease_test.go
+5 -5
@@ -36,7 +36,7 @@ func TestLeaseRegistryLifecycle(t *testing.T) {
36 stream: transport.NewRelayStream("addr-1", time.Minute, 1),
37 }
38
39 - if _, err := registry.Register(record); err != nil {
39 + if err := registry.Register(record); err != nil {
40 t.Fatalf("Register() error = %v", err)
41 }
42
@@ -85,7 +85,7 @@ func TestLeaseRegistryWildcardAndConflict(t *testing.T) {
85 ExpiresAt: time.Now().Add(30 * time.Second),
86 stream: transport.NewRelayStream("addr-wildcard", time.Minute, 1),
87 }
88 - if _, err := registry.Register(wildcardLease); err != nil {
88 + if err := registry.Register(wildcardLease); err != nil {
89 t.Fatalf("Register(wildcard) error = %v", err)
90 }
91
@@ -105,7 +105,7 @@ func TestLeaseRegistryWildcardAndConflict(t *testing.T) {
105 ExpiresAt: time.Now().Add(30 * time.Second),
106 stream: transport.NewRelayStream("addr-conflict", time.Minute, 1),
107 }
108 - _, err := registry.Register(conflict)
108 + err := registry.Register(conflict)
109 if !errors.Is(err, errHostnameConflict) {
110 t.Fatalf("Register(conflict) error = %v, want hostname conflict", err)
111 }
@@ -129,7 +129,7 @@ func TestLeaseRegistrySnapshotAndRoutableUsePolicy(t *testing.T) {
129 ClientIP: "203.0.113.20",
130 stream: transport.NewRelayStream("addr-policy", time.Minute, 1),
131 }
132 - if _, err := registry.Register(record); err != nil {
132 + if err := registry.Register(record); err != nil {
133 t.Fatalf("Register() error = %v", err)
134 }
135
@@ -169,7 +169,7 @@ func TestLeaseRegistryCleanupExpiredClosesBroker(t *testing.T) {
169 ExpiresAt: time.Now().Add(-time.Second),
170 stream: transport.NewRelayStream("addr-expired", time.Minute, 1),
171 }
172 - if _, err := registry.Register(record); err != nil {
172 + if err := registry.Register(record); err != nil {
173 t.Fatalf("Register() error = %v", err)
174 }
175
portal/overlay/hop_mux.go
+174 -117
@@ -1,6 +1,7 @@
1 package overlay
2
3 import (
4 + "bytes"
5 "context"
6 "encoding/binary"
7 "encoding/json"
@@ -8,6 +9,7 @@ import (
9 "fmt"
10 "io"
11 "net"
12 + "net/http"
13 "strings"
14 "sync"
15 "time"
@@ -17,32 +19,27 @@ import (
19 )
20
21 const (
20 - hopProtocolVersion = 1
21 - hopPrefaceLimit = 4 << 10
22 - hopControlPayloadLimit = 16 << 10
23 - hopIncomingBuffer = 128
24 - defaultPrefaceTimeout = 2 * time.Second
25 - defaultControlTimeout = 10 * time.Second
22 + hopProtocolVersion = 1
23 + hopPrefaceLimit = 4 << 10
24 + hopRegistryFrameLimit = 8 << 20
25 + hopIncomingBuffer = 128
26 + defaultPrefaceTimeout = 2 * time.Second
27 )
28
29 const (
30 hopModeTLSStream = "tls-stream"
30 - hopModeControl = "control"
31 -
32 - HopControlInstall = "install"
33 - HopControlRenew = "renew"
34 - HopControlDelete = "delete"
31 + hopModeRegistry = "registry"
32 )
33
34 type HopMux struct {
35 listener net.Listener
36 overlay *Overlay
37 incoming chan HopStream
38 + registry chan HopRegistryStream
39
40 mu sync.Mutex
43 - sessions map[string]*yamux.Session
44 - active map[*yamux.Session]struct{}
45 - closed bool
41 + outbound map[string]*yamux.Session
42 + done chan struct{}
43 }
44
45 type hopPreface struct {
@@ -51,29 +48,31 @@ type hopPreface struct {
48 Token string `json:"token"`
49 }
50
54 -type HopControl struct {
55 - Action string `json:"action"`
56 - Route HopRouteSpec `json:"route"`
51 +type HopStream struct {
52 + Conn net.Conn
53 + Token string
54 }
55
59 -type HopRouteSpec struct {
60 - RouteID string `json:"route_id"`
61 - MatchHostname string `json:"match_hostname,omitempty"`
62 - MatchToken string `json:"match_token,omitempty"`
63 - ForwardOverlayIPv4 string `json:"forward_overlay_ipv4,omitempty"`
64 - ForwardToken string `json:"forward_token,omitempty"`
65 - ExpiresAt time.Time `json:"expires_at"`
56 +type HopRegistryStream struct {
57 + Path string
58 + Body []byte
59 + conn net.Conn
60 }
61
68 -type hopControlResponse struct {
69 - OK bool `json:"ok"`
70 - Error string `json:"error,omitempty"`
62 +type hopRegistryHTTPResponse struct {
63 + header http.Header
64 + status int
65 + body bytes.Buffer
66 }
67
73 -type HopStream struct {
74 - Conn net.Conn
75 - Token string
76 - Control *HopControl
68 +type hopRegistryRequest struct {
69 + Path string `json:"path"`
70 + Body []byte `json:"body"`
71 +}
72 +
73 +type hopRegistryResponse struct {
74 + Status int `json:"status"`
75 + Body []byte `json:"body"`
76 }
77
78 func NewHopMux(overlay *Overlay) (*HopMux, error) {
@@ -88,8 +87,9 @@ func NewHopMux(overlay *Overlay) (*HopMux, error) {
87 listener: listener,
88 overlay: overlay,
89 incoming: make(chan HopStream, hopIncomingBuffer),
91 - sessions: make(map[string]*yamux.Session),
92 - active: make(map[*yamux.Session]struct{}),
90 + registry: make(chan HopRegistryStream, hopIncomingBuffer),
91 + outbound: make(map[string]*yamux.Session),
92 + done: make(chan struct{}),
93 }, nil
94 }
95
@@ -132,35 +132,43 @@ func (m *HopMux) Accept(ctx context.Context) (HopStream, error) {
132 }
133 }
134
135 +func (m *HopMux) AcceptRegistry(ctx context.Context) (HopRegistryStream, error) {
136 + if m == nil {
137 + <-ctx.Done()
138 + return HopRegistryStream{}, ctx.Err()
139 + }
140 + select {
141 + case stream := <-m.registry:
142 + return stream, nil
143 + case <-ctx.Done():
144 + return HopRegistryStream{}, ctx.Err()
145 + }
146 +}
147 +
148 func (m *HopMux) Close() error {
149 if m == nil {
150 return nil
151 }
152
153 m.mu.Lock()
141 - if m.closed {
154 + if m.done == nil {
155 + m.done = make(chan struct{})
156 + }
157 + select {
158 + case <-m.done:
159 m.mu.Unlock()
160 return nil
161 + default:
162 }
145 - m.closed = true
146 - seen := make(map[*yamux.Session]struct{}, len(m.sessions)+len(m.active))
147 - sessions := make([]*yamux.Session, 0, len(m.sessions)+len(m.active))
148 - for _, session := range m.sessions {
149 - if _, ok := seen[session]; ok {
150 - continue
151 - }
152 - seen[session] = struct{}{}
153 - sessions = append(sessions, session)
154 - }
155 - for session := range m.active {
156 - if _, ok := seen[session]; ok {
163 + close(m.done)
164 + sessions := make([]*yamux.Session, 0, len(m.outbound))
165 + for _, session := range m.outbound {
166 + if session == nil {
167 continue
168 }
159 - seen[session] = struct{}{}
169 sessions = append(sessions, session)
170 }
162 - m.sessions = make(map[string]*yamux.Session)
163 - m.active = make(map[*yamux.Session]struct{})
171 + m.outbound = make(map[string]*yamux.Session)
172 listener := m.listener
173 m.mu.Unlock()
174
@@ -196,45 +204,38 @@ func (m *HopMux) OpenStream(ctx context.Context, overlayIPv4, token string) (net
204 return stream, nil
205 }
206
199 -func (m *HopMux) Control(ctx context.Context, overlayIPv4 string, req HopControl) error {
200 - req.Action = strings.TrimSpace(req.Action)
201 - if req.Action == "" {
202 - return errors.New("hop control action is required")
207 +func (m *HopMux) OpenRegistry(ctx context.Context, overlayIPv4, path string, payload any) (int, []byte, error) {
208 + path = strings.TrimSpace(path)
209 + if path == "" {
210 + return 0, nil, errors.New("registry path is required")
211 + }
212 + body, err := json.Marshal(payload)
213 + if err != nil {
214 + return 0, nil, err
215 }
216
217 stream, err := m.openYamuxStream(ctx, overlayIPv4)
218 if err != nil {
207 - return err
219 + return 0, nil, err
220 }
221 defer stream.Close()
222
211 - deadline := time.Now().Add(defaultControlTimeout)
212 - if ctxDeadline, ok := ctx.Deadline(); ok && ctxDeadline.Before(deadline) {
213 - deadline = ctxDeadline
214 - }
215 - _ = stream.SetDeadline(deadline)
216 -
223 preface := hopPreface{
224 Version: hopProtocolVersion,
219 - Mode: hopModeControl,
225 + Mode: hopModeRegistry,
226 }
227 if err := writeFramedJSON(stream, preface, hopPrefaceLimit); err != nil {
222 - return err
228 + return 0, nil, err
229 }
224 - if err := writeFramedJSON(stream, req, hopControlPayloadLimit); err != nil {
225 - return err
230 + if err := writeFramedJSON(stream, hopRegistryRequest{Path: path, Body: body}, hopRegistryFrameLimit); err != nil {
231 + return 0, nil, err
232 }
227 - var resp hopControlResponse
228 - if err := readFramedJSON(stream, &resp, hopControlPayloadLimit); err != nil {
229 - return err
230 - }
231 - if !resp.OK {
232 - if strings.TrimSpace(resp.Error) == "" {
233 - return errors.New("hop control request failed")
234 - }
235 - return errors.New(resp.Error)
233 +
234 + var resp hopRegistryResponse
235 + if err := readFramedJSON(stream, &resp, hopRegistryFrameLimit); err != nil {
236 + return 0, nil, err
237 }
237 - return nil
238 + return resp.Status, resp.Body, nil
239 }
240
241 func (m *HopMux) openYamuxStream(ctx context.Context, overlayIPv4 string) (*yamux.Stream, error) {
@@ -258,18 +259,19 @@ func (m *HopMux) openYamuxStream(ctx context.Context, overlayIPv4 string) (*yamu
259
260 func (m *HopMux) session(ctx context.Context, overlayIPv4 string) (*yamux.Session, error) {
261 m.mu.Lock()
261 - if m.closed {
262 + select {
263 + case <-m.done:
264 m.mu.Unlock()
265 return nil, net.ErrClosed
266 + default:
267 }
265 - if session := m.sessions[overlayIPv4]; session != nil {
268 + if session := m.outbound[overlayIPv4]; session != nil {
269 if !session.IsClosed() {
270 m.mu.Unlock()
271 return session, nil
272 }
270 - delete(m.active, session)
273 }
272 - delete(m.sessions, overlayIPv4)
274 + delete(m.outbound, overlayIPv4)
275 m.mu.Unlock()
276
277 if m.overlay == nil || m.overlay.stack == nil {
@@ -288,18 +290,19 @@ func (m *HopMux) session(ctx context.Context, overlayIPv4 string) (*yamux.Sessio
290 }
291
292 m.mu.Lock()
291 - if m.closed {
293 + select {
294 + case <-m.done:
295 m.mu.Unlock()
296 _ = session.Close()
297 return nil, net.ErrClosed
298 + default:
299 }
296 - if current := m.sessions[overlayIPv4]; current != nil && !current.IsClosed() {
300 + if current := m.outbound[overlayIPv4]; current != nil && !current.IsClosed() {
301 m.mu.Unlock()
302 _ = session.Close()
303 return current, nil
304 }
301 - m.sessions[overlayIPv4] = session
302 - m.active[session] = struct{}{}
305 + m.outbound[overlayIPv4] = session
306 m.mu.Unlock()
307 return session, nil
308 }
@@ -307,10 +310,9 @@ func (m *HopMux) session(ctx context.Context, overlayIPv4 string) (*yamux.Sessio
310 func (m *HopMux) forgetSession(overlayIPv4 string, session *yamux.Session) {
311 m.mu.Lock()
312 defer m.mu.Unlock()
310 - if m.sessions[overlayIPv4] == session {
311 - delete(m.sessions, overlayIPv4)
313 + if m.outbound[overlayIPv4] == session {
314 + delete(m.outbound, overlayIPv4)
315 }
313 - delete(m.active, session)
316 }
317
318 func (m *HopMux) serveSession(ctx context.Context, conn net.Conn) {
@@ -321,17 +323,27 @@ func (m *HopMux) serveSession(ctx context.Context, conn net.Conn) {
323 return
324 }
325 m.mu.Lock()
324 - if m.closed {
326 + select {
327 + case <-m.done:
328 m.mu.Unlock()
329 _ = session.Close()
330 return
331 + default:
332 }
329 - m.active[session] = struct{}{}
333 + done := m.done
334 m.mu.Unlock()
335 + sessionDone := make(chan struct{})
336 + go func() {
337 + select {
338 + case <-ctx.Done():
339 + _ = session.Close()
340 + case <-done:
341 + _ = session.Close()
342 + case <-sessionDone:
343 + }
344 + }()
345 defer func() {
332 - m.mu.Lock()
333 - delete(m.active, session)
334 - m.mu.Unlock()
346 + close(sessionDone)
347 _ = session.Close()
348 }()
349
@@ -367,31 +379,26 @@ func (m *HopMux) handleStream(ctx context.Context, stream *yamux.Stream) {
379 Conn: stream,
380 Token: preface.Token,
381 })
370 - case hopModeControl:
371 - m.readControlStream(ctx, stream)
382 + case hopModeRegistry:
383 + var req hopRegistryRequest
384 + if err := readFramedJSON(stream, &req, hopRegistryFrameLimit); err != nil {
385 + _ = stream.Close()
386 + return
387 + }
388 + if strings.TrimSpace(req.Path) == "" {
389 + _ = stream.Close()
390 + return
391 + }
392 + m.deliverRegistry(ctx, HopRegistryStream{
393 + Path: req.Path,
394 + Body: req.Body,
395 + conn: stream,
396 + })
397 default:
398 _ = stream.Close()
399 }
400 }
401
377 -func (m *HopMux) readControlStream(ctx context.Context, stream net.Conn) {
378 - _ = stream.SetDeadline(time.Now().Add(defaultControlTimeout))
379 -
380 - var req HopControl
381 - var err error
382 - if err = readFramedJSON(stream, &req, hopControlPayloadLimit); err == nil {
383 - req.Action = strings.TrimSpace(req.Action)
384 - }
385 - if err != nil {
386 - _ = stream.Close()
387 - return
388 - }
389 - m.deliver(ctx, HopStream{
390 - Conn: stream,
391 - Control: &req,
392 - })
393 -}
394 -
402 func (m *HopMux) deliver(ctx context.Context, stream HopStream) {
403 select {
404 case m.incoming <- stream:
@@ -400,16 +407,66 @@ func (m *HopMux) deliver(ctx context.Context, stream HopStream) {
407 }
408 }
409
403 -func (s HopStream) Respond(err error) error {
404 - if s.Conn == nil {
405 - return nil
410 +func (m *HopMux) deliverRegistry(ctx context.Context, stream HopRegistryStream) {
411 + select {
412 + case m.registry <- stream:
413 + case <-ctx.Done():
414 + _ = stream.conn.Close()
415 }
407 - defer s.Conn.Close()
408 - resp := hopControlResponse{OK: err == nil}
416 +}
417 +
418 +func (s HopRegistryStream) Respond(status int, body []byte) error {
419 + if s.conn == nil {
420 + return net.ErrClosed
421 + }
422 + defer s.conn.Close()
423 + return writeFramedJSON(s.conn, hopRegistryResponse{Status: status, Body: body}, hopRegistryFrameLimit)
424 +}
425 +
426 +func (s HopRegistryStream) ServeRegistryHTTP(ctx context.Context, host string, handler http.Handler) error {
427 + if handler == nil {
428 + if s.conn != nil {
429 + _ = s.conn.Close()
430 + }
431 + return errors.New("hop registry handler is required")
432 + }
433 + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://"+strings.TrimSpace(host)+s.Path, bytes.NewReader(s.Body))
434 if err != nil {
410 - resp.Error = err.Error()
435 + if s.conn != nil {
436 + _ = s.conn.Close()
437 + }
438 + return err
439 + }
440 + req.RemoteAddr = "127.0.0.1:0"
441 + req.Header.Set("Content-Type", "application/json")
442 +
443 + resp := &hopRegistryHTTPResponse{header: make(http.Header)}
444 + handler.ServeHTTP(resp, req)
445 + return s.Respond(resp.Status(), resp.body.Bytes())
446 +}
447 +
448 +func (r *hopRegistryHTTPResponse) Header() http.Header {
449 + return r.header
450 +}
451 +
452 +func (r *hopRegistryHTTPResponse) Write(body []byte) (int, error) {
453 + if r.status == 0 {
454 + r.status = http.StatusOK
455 + }
456 + return r.body.Write(body)
457 +}
458 +
459 +func (r *hopRegistryHTTPResponse) WriteHeader(status int) {
460 + if r.status == 0 {
461 + r.status = status
462 + }
463 +}
464 +
465 +func (r *hopRegistryHTTPResponse) Status() int {
466 + if r.status == 0 {
467 + return http.StatusOK
468 }
412 - return writeFramedJSON(s.Conn, resp, hopControlPayloadLimit)
469 + return r.status
470 }
471
472 func hopYamuxConfig() *yamux.Config {
portal/overlay/stack.go
+11 -1
@@ -36,6 +36,7 @@ type stack struct {
36 mu sync.Mutex
37 closed bool
38 peerEndpoints map[string]string
39 + peerConfig string
40 }
41
42 func newStack(cfg Config) (*stack, error) {
@@ -187,11 +188,20 @@ func (s *stack) ApplyPeers(peers []desiredPeer) error {
188 }
189 }
190
190 - if err := s.device.IpcSet(builder.String()); err != nil {
191 + config := builder.String()
192 + s.mu.Lock()
193 + if s.peerConfig == config {
194 + s.mu.Unlock()
195 + return warnErr
196 + }
197 + s.mu.Unlock()
198 +
199 + if err := s.device.IpcSet(config); err != nil {
200 return err
201 }
202 s.mu.Lock()
203 s.peerEndpoints = nextPeerEndpoints
204 + s.peerConfig = config
205 s.mu.Unlock()
206 return warnErr
207 }
portal/server.go
+89 -27
@@ -117,10 +117,10 @@ type Server struct {
117 quicTunnel *quic.Listener
118
119 overlay *overlay.Overlay
120 + hopMux *overlay.HopMux
121 relaySet *discovery.RelaySet
122 announceLimiter *discovery.AnnounceLimiter
123 registry *leaseRegistry
123 - hops *hopManager
124 udpPorts *transport.PortAllocator
125 tcpPorts *transport.PortAllocator
126 }
@@ -175,7 +175,7 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
175 var sniListener net.Listener
176 var apiServer *http.Server
177 var apiCloser io.Closer
178 - var hops *hopManager
178 + var hopMux *overlay.HopMux
179 var ov *overlay.Overlay
180 var quicTunnel *quic.Listener
181 defer func() {
@@ -183,8 +183,8 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
183 return
184 }
185 acmeManager.Stop()
186 - if hops != nil {
187 - _ = hops.close()
186 + if hopMux != nil {
187 + _ = hopMux.Close()
188 }
189 if ov != nil {
190 _ = ov.Shutdown(context.Background())
@@ -225,7 +225,7 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
225 if err != nil {
226 return err
227 }
228 - hops, err = newHopManager(ov, s.registry, &s.proxy)
228 + hopMux, err = overlay.NewHopMux(ov)
229 if err != nil {
230 return err
231 }
@@ -246,7 +246,7 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
246 s.cancel = cancel
247 s.group = group
248 s.overlay = ov
249 - s.hops = hops
249 + s.hopMux = hopMux
250 s.quicTunnel = quicTunnel
251 started = true
252
@@ -259,8 +259,8 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
259 if s.overlay != nil {
260 group.Go(s.overlay.Serve)
261 }
262 - if s.hops != nil {
263 - group.Go(func() error { return s.hops.run(groupCtx) })
262 + if s.hopMux != nil {
263 + group.Go(func() error { return s.runHopMux(groupCtx) })
264 }
265 if s.quicTunnel != nil {
266 group.Go(s.runQUICTunnelListener)
@@ -282,7 +282,7 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
282 Int("max_port", s.cfg.MaxPort).
283 Bool("discovery_enabled", s.cfg.DiscoveryEnabled).
284 Bool("wireguard_enabled", s.overlay != nil).
285 - Bool("multihop_enabled", s.hops != nil).
285 + Bool("multihop_enabled", s.hopMux != nil).
286 Bool("udp_enabled", s.quicTunnel != nil).
287 Bool("tcp_enabled", s.cfg.TCPEnabled)
288 if s.quicTunnel != nil {
@@ -320,7 +320,6 @@ func (s *Server) Shutdown(ctx context.Context) error {
320
321 for _, lease := range s.registry.CloseAll() {
322 if lease != nil {
323 - s.hops.delete(lease)
323 if lease.isDirect() && s.acmeManager != nil {
324 deleteCtx, cancel := context.WithTimeout(ctx, defaultClaimTimeout)
325 if err := s.acmeManager.DeleteENSGaslessHostname(deleteCtx, lease.Hostname); err != nil {
@@ -349,8 +348,8 @@ func (s *Server) Shutdown(ctx context.Context) error {
348 shutdownErr = err
349 }
350 }
352 - if s.hops != nil {
353 - if err := s.hops.close(); err != nil && shutdownErr == nil && !errors.Is(err, net.ErrClosed) {
351 + if s.hopMux != nil {
352 + if err := s.hopMux.Close(); err != nil && shutdownErr == nil && !errors.Is(err, net.ErrClosed) {
353 shutdownErr = err
354 }
355 }
@@ -403,7 +402,7 @@ func (s *Server) LeaseSnapshotByHostname(hostname string) (types.Lease, bool) {
402 }
403
404 record, ok := s.registry.Lookup(hostname)
406 - if !ok || record == nil || time.Now().After(record.ExpiresAt) {
405 + if !ok || record == nil || record.isHopForward() || time.Now().After(record.ExpiresAt) {
406 return types.Lease{}, false
407 }
408 return s.registry.Snapshot(record), true
@@ -484,22 +483,10 @@ func (s *Server) runSNIListener(ctx context.Context) error {
483 }
484
485 record, ok := s.registry.Lookup(serverName)
487 - if !ok || record == nil || time.Now().After(record.ExpiresAt) || record.stream == nil || !s.registry.policy.IsIdentityRoutable(record.Key()) {
488 - if s.hops == nil || !s.hops.forwardHostname(ctx, wrappedConn, serverName) {
489 - _ = wrappedConn.Close()
490 - return
491 - }
492 - return
493 - }
494 -
495 - claimCtx, cancel := context.WithTimeout(ctx, defaultClaimTimeout)
496 - session, err := record.stream.Claim(claimCtx)
497 - cancel()
498 - if err != nil {
486 + if !ok || !s.bridgeLeaseConn(ctx, wrappedConn, record) {
487 _ = wrappedConn.Close()
488 return
489 }
502 - s.proxy.bridge(wrappedConn, session, record.Key(), s.registry.policy.BPSManager())
490 }(conn)
491 case errors.Is(err, net.ErrClosed):
492 return nil
@@ -512,6 +499,82 @@ func (s *Server) runSNIListener(ctx context.Context) error {
499 }
500 }
501
502 +func (s *Server) runHopMux(ctx context.Context) error {
503 + if s == nil || s.hopMux == nil {
504 + <-ctx.Done()
505 + return nil
506 + }
507 +
508 + group, groupCtx := errgroup.WithContext(ctx)
509 + group.Go(func() error { return s.hopMux.Serve(groupCtx) })
510 + group.Go(func() error {
511 + for {
512 + stream, err := s.hopMux.Accept(groupCtx)
513 + if err != nil {
514 + if groupCtx.Err() != nil {
515 + return nil
516 + }
517 + return err
518 + }
519 + go func(stream overlay.HopStream) {
520 + record, ok := s.registry.RecordByHopToken(stream.Token, time.Now())
521 + if !ok || !s.bridgeLeaseConn(groupCtx, stream.Conn, record) {
522 + _ = stream.Conn.Close()
523 + }
524 + }(stream)
525 + }
526 + })
527 + group.Go(func() error {
528 + registryHandler := s.hopRegistryHandler()
529 + for {
530 + stream, err := s.hopMux.AcceptRegistry(groupCtx)
531 + if err != nil {
532 + if groupCtx.Err() != nil {
533 + return nil
534 + }
535 + return err
536 + }
537 + go func(stream overlay.HopRegistryStream) {
538 + if err := stream.ServeRegistryHTTP(groupCtx, s.identity.Name, registryHandler); err != nil {
539 + log.Warn().Err(err).Str("path", stream.Path).Msg("respond hop registry stream")
540 + }
541 + }(stream)
542 + }
543 + })
544 + return group.Wait()
545 +}
546 +
547 +func (s *Server) bridgeLeaseConn(ctx context.Context, conn net.Conn, record *leaseRecord) bool {
548 + if s == nil || s.registry == nil || record == nil || time.Now().After(record.ExpiresAt) {
549 + return false
550 + }
551 + if record.isHopForward() {
552 + overlayIPv4 := strings.TrimSpace(record.hopNextOverlayIPv4)
553 + forwardToken := strings.TrimSpace(record.hopNextToken)
554 + if s.hopMux == nil || overlayIPv4 == "" || forwardToken == "" {
555 + return false
556 + }
557 + next, err := s.hopMux.OpenStream(ctx, overlayIPv4, forwardToken)
558 + if err != nil {
559 + log.Warn().Err(err).Str("next_overlay_ipv4", overlayIPv4).Msg("open next hop stream")
560 + return false
561 + }
562 + s.proxy.bridge(conn, next, "", nil)
563 + return true
564 + }
565 + if record.stream == nil || !s.registry.policy.IsIdentityRoutable(record.Key()) {
566 + return false
567 + }
568 + claimCtx, cancel := context.WithTimeout(ctx, defaultClaimTimeout)
569 + session, err := record.stream.Claim(claimCtx)
570 + cancel()
571 + if err != nil {
572 + return false
573 + }
574 + s.proxy.bridge(conn, session, record.Key(), s.registry.policy.BPSManager())
575 + return true
576 +}
577 +
578 func (s *Server) runLeaseJanitor(ctx context.Context, interval time.Duration) error {
579 if interval <= 0 {
580 return errors.New("janitor interval must be positive")
@@ -526,7 +589,6 @@ func (s *Server) runLeaseJanitor(ctx context.Context, interval time.Duration) er
589 return nil
590 case <-ticker.C:
591 for _, lease := range s.registry.cleanupExpired(time.Now()) {
529 - s.hops.delete(lease)
592 if lease.isDirect() && s.acmeManager != nil {
593 deleteCtx, cancel := context.WithTimeout(context.Background(), defaultClaimTimeout)
594 err := s.acmeManager.DeleteENSGaslessHostname(deleteCtx, lease.Hostname)
sdk/api_client.go
+102 -55
@@ -30,12 +30,18 @@ var errRelayIncompatible = errors.New("relay is incompatible")
30 // API call creates fresh TCP connections. Call this after detecting a system
31 // sleep/wake cycle where pooled connections are almost certainly dead.
32 func (l *listener) resetTransport() {
33 + if l.leaseHTTPClient != nil && l.leaseHTTPClient != l.httpClient {
34 + if transport, ok := l.leaseHTTPClient.Transport.(*http.Transport); ok {
35 + transport.CloseIdleConnections()
36 + }
37 + }
38 if l.httpClient != nil {
39 if transport, ok := l.httpClient.Transport.(*http.Transport); ok {
40 transport.CloseIdleConnections()
41 }
42 }
43 l.httpClient = nil
44 + l.leaseHTTPClient = nil
45 l.tlsConfig = nil
46 }
47
@@ -76,101 +82,142 @@ func (l *listener) initHTTPTransport(ctx context.Context) error {
82 return fmt.Errorf("%w: relay sdk protocol version mismatch: relay=%q client=%q", errRelayIncompatible, protocolVersion, types.SDKVersion)
83 }
84
85 + if l.leaseAPIURL == nil {
86 + l.leaseAPIURL = l.relayURL
87 + }
88 + leaseHTTPClient := httpClient
89 + if l.leaseAPIURL.String() != l.relayURL.String() {
90 + _, leaseHTTPClient, err = utils.NewHTTPTLSClient(bootstrapCtx, l.leaseAPIURL, l.requestTimeout)
91 + if err != nil {
92 + if transport, ok := httpClient.Transport.(*http.Transport); ok {
93 + transport.CloseIdleConnections()
94 + }
95 + return err
96 + }
97 + }
98 +
99 l.httpClient = httpClient
100 + l.leaseHTTPClient = leaseHTTPClient
101 l.tlsConfig = tlsConfig
102 return nil
103 }
104
84 -func (l *listener) registerLease(ctx context.Context, ttl time.Duration, udpEnabled, tcpEnabled bool) (types.RegisterResponse, error) {
85 - multiHop, err := l.multiHopDescriptors()
86 - if err != nil {
87 - return types.RegisterResponse{}, err
105 +func (l *listener) registerLease(ctx context.Context, ttl time.Duration, udpEnabled, tcpEnabled bool) (types.RegisterResponse, *types.HopRoute, error) {
106 + var err error
107 + var exitHopToken string
108 + var publicHostname string
109 + var keylessURL string
110 + var hop *types.HopRoute
111 + if len(l.multiHop) > 0 {
112 + if len(l.multiHop) < 2 {
113 + return types.RegisterResponse{}, nil, errors.New("multi-hop requires at least entry and exit relay urls")
114 + }
115 + if l.relaySet == nil {
116 + return types.RegisterResponse{}, nil, errors.New("multi-hop relay set is unavailable")
117 + }
118 +
119 + states := l.relaySet.AggregateRelays()
120 + descriptors := make(map[string]types.RelayDescriptor, len(states))
121 + for _, state := range states {
122 + desc := state.Descriptor
123 + if desc.APIHTTPSAddr != "" {
124 + descriptors[desc.APIHTTPSAddr] = desc
125 + }
126 + }
127 +
128 + hopPath := make([]types.RelayDescriptor, 0, len(l.multiHop))
129 + for i, relayURL := range l.multiHop {
130 + desc, ok := descriptors[relayURL]
131 + if !ok {
132 + return types.RegisterResponse{}, nil, fmt.Errorf("multi-hop relay %d descriptor was not discovered", i)
133 + }
134 + hopPath = append(hopPath, desc)
135 + }
136 +
137 + publicHostname, err = utils.LeaseHostname(l.identity.Name, utils.PortalRootHost(hopPath[0].APIHTTPSAddr))
138 + if err != nil {
139 + return types.RegisterResponse{}, nil, err
140 + }
141 + keylessURL = hopPath[0].APIHTTPSAddr
142 +
143 + tokens := make([]string, len(hopPath)-1)
144 + for i := range tokens {
145 + token, err := l.identity.DeriveToken(fmt.Sprintf("hop:%s:%d:%s:%s", publicHostname, i, hopPath[i].APIHTTPSAddr, hopPath[i+1].APIHTTPSAddr))
146 + if err != nil {
147 + return types.RegisterResponse{}, nil, err
148 + }
149 + tokens[i] = "hpt_" + token
150 + }
151 + exitHopToken = tokens[len(tokens)-1]
152 +
153 + var next *types.HopRoute
154 + for i := len(hopPath) - 2; i >= 0; i-- {
155 + route := &types.HopRoute{
156 + ForwardRelay: hopPath[i+1],
157 + ForwardToken: tokens[i],
158 + Next: next,
159 + }
160 + if i == 0 {
161 + route.MatchHostname = publicHostname
162 + } else {
163 + route.MatchToken = tokens[i-1]
164 + }
165 + hop = route
166 + next = route
167 + }
168 }
169
170 var challenge types.RegisterChallengeResponse
91 - if err := utils.HTTPDoAPIPath(ctx, l.httpClient, l.relayURL, http.MethodPost, types.PathSDKRegisterChallenge, types.RegisterChallengeRequest{
171 + if err := utils.HTTPDoAPIPath(ctx, l.leaseHTTPClient, l.leaseAPIURL, http.MethodPost, types.PathSDKRegisterChallenge, types.RegisterChallengeRequest{
172 Identity: l.identity,
173 Metadata: l.metadata,
174 TTL: int(ttl / time.Second),
175 UDPEnabled: udpEnabled,
176 TCPEnabled: tcpEnabled,
97 - MultiHop: multiHop,
177 + HopToken: exitHopToken,
178 + Hop: hop,
179 }, nil, &challenge); err != nil {
99 - return types.RegisterResponse{}, err
180 + return types.RegisterResponse{}, nil, err
181 }
182
183 signature, err := utils.SignEthereumPersonalMessage(challenge.SIWEMessage, l.identity.PrivateKey)
184 if err != nil {
104 - return types.RegisterResponse{}, err
185 + return types.RegisterResponse{}, nil, err
186 }
187
188 var resp types.RegisterResponse
108 - if err := utils.HTTPDoAPIPath(ctx, l.httpClient, l.relayURL, http.MethodPost, types.PathSDKRegister, types.RegisterRequest{
189 + if err := utils.HTTPDoAPIPath(ctx, l.leaseHTTPClient, l.leaseAPIURL, http.MethodPost, types.PathSDKRegister, types.RegisterRequest{
190 ChallengeID: challenge.ChallengeID,
191 SIWEMessage: challenge.SIWEMessage,
192 SIWESignature: signature,
193 ReportedIP: utils.ResolvePublicIP(ctx),
194 + Hop: hop,
195 }, nil, &resp); err != nil {
114 - return types.RegisterResponse{}, err
115 - }
116 - return resp, nil
117 -}
118 -
119 -func (l *listener) multiHopDescriptors() ([]types.RelayDescriptor, error) {
120 - if l == nil || len(l.multiHop) == 0 {
121 - return nil, nil
122 - }
123 - if l.relaySet == nil {
124 - return nil, errors.New("multi-hop relay set is unavailable")
125 - }
126 -
127 - states := l.relaySet.AggregateRelays()
128 - byURL := make(map[string]types.RelayDescriptor, len(states))
129 - for _, state := range states {
130 - desc := state.Descriptor
131 - if desc.APIHTTPSAddr != "" {
132 - byURL[desc.APIHTTPSAddr] = desc
133 - }
196 + return types.RegisterResponse{}, nil, err
197 }
135 -
136 - now := time.Now().UTC()
137 - path := make([]types.RelayDescriptor, 0, len(l.multiHop))
138 - seen := make(map[string]struct{}, len(l.multiHop))
139 - for i, relayURL := range l.multiHop {
140 - if _, ok := seen[relayURL]; ok {
141 - return nil, errors.New("multi-hop relay urls must be unique")
142 - }
143 - seen[relayURL] = struct{}{}
144 -
145 - desc, ok := byURL[relayURL]
146 - if !ok {
147 - return nil, fmt.Errorf("multi-hop relay %d descriptor was not discovered", i)
148 - }
149 - if !desc.SupportsOverlayPeer || desc.WireGuardPublicKey == "" || desc.WireGuardEndpoint == "" || desc.OverlayIPv4 == "" {
150 - return nil, fmt.Errorf("multi-hop relay %d does not expose WireGuard overlay metadata", i)
151 - }
152 - if !desc.ExpiresAt.After(now) {
153 - return nil, fmt.Errorf("multi-hop relay %d descriptor is expired", i)
154 - }
155 - path = append(path, desc)
198 + if hop != nil {
199 + resp.Hostname = publicHostname
200 + resp.KeylessURL = keylessURL
201 }
157 - return path, nil
202 + return resp, hop, nil
203 }
204
160 -func (l *listener) renewRegisteredLease(ctx context.Context, ttl time.Duration, accessToken string) (types.RenewResponse, error) {
205 +func (l *listener) renewRegisteredLease(ctx context.Context, ttl time.Duration, accessToken string, hop *types.HopRoute) (types.RenewResponse, error) {
206 var resp types.RenewResponse
162 - if err := utils.HTTPDoAPIPath(ctx, l.httpClient, l.relayURL, http.MethodPost, types.PathSDKRenew, types.RenewRequest{
207 + if err := utils.HTTPDoAPIPath(ctx, l.leaseHTTPClient, l.leaseAPIURL, http.MethodPost, types.PathSDKRenew, types.RenewRequest{
208 AccessToken: accessToken,
209 TTL: int(ttl / time.Second),
210 ReportedIP: utils.ResolvePublicIP(ctx),
211 + Hop: hop,
212 }, nil, &resp); err != nil {
213 return types.RenewResponse{}, err
214 }
215 return resp, nil
216 }
217
172 -func (l *listener) unregisterLease(ctx context.Context, accessToken string) error {
173 - return utils.HTTPDoAPIPath(ctx, l.httpClient, l.relayURL, http.MethodPost, types.PathSDKUnregister, types.UnregisterRequest{
218 +func (l *listener) unregisterLease(ctx context.Context, accessToken string, hop *types.HopRoute) error {
219 + return utils.HTTPDoAPIPath(ctx, l.leaseHTTPClient, l.leaseAPIURL, http.MethodPost, types.PathSDKUnregister, types.UnregisterRequest{
220 AccessToken: accessToken,
221 + Hop: hop,
222 }, nil, nil)
223 }
sdk/expose.go
+5 -5
@@ -76,14 +76,14 @@ func Expose(ctx context.Context, cfg ExposeConfig) (*Exposure, error) {
76 if err != nil {
77 return nil, fmt.Errorf("normalize multi-hop relay urls: %w", err)
78 }
79 - if len(multiHop) > 0 && len(multiHop) < 2 {
80 - return nil, errors.New("multi-hop path requires entry and exit relays")
79 + if len(multiHop) == 1 {
80 + return nil, errors.New("multi-hop requires at least entry and exit relay urls")
81 }
82 if len(multiHop) > 0 && (cfg.UDPEnabled || cfg.TCPEnabled) {
83 return nil, errors.New("multi-hop currently supports only the default SNI TLS stream transport")
84 }
85 - listenerRelayURLs := []string(nil)
86 - relaySetURLs := []string(nil)
85 + var listenerRelayURLs []string
86 + var relaySetURLs []string
87 if len(multiHop) > 0 {
88 listenerRelayURLs = []string{multiHop[len(multiHop)-1]}
89 relaySetURLs = append([]string(nil), multiHop...)
@@ -397,7 +397,7 @@ func (e *Exposure) runDiscoveryLoop(ctx context.Context) {
397 }
398
399 func (e *Exposure) reconcileRelayListeners(failOnError bool) error {
400 - listenerRelayURLs := []string(nil)
400 + var listenerRelayURLs []string
401 var multiHop []string
402 if len(e.multiHop) > 0 {
403 listenerRelayURLs = []string{e.multiHop[len(e.multiHop)-1]}
sdk/listener.go
+32 -11
@@ -52,6 +52,7 @@ type listener struct {
52 closeOnce sync.Once
53
54 relayURL *url.URL
55 + leaseAPIURL *url.URL
56 identity types.Identity
57 metadata types.LeaseMetadata
58 relaySet *discovery.RelaySet
@@ -70,8 +71,9 @@ type listener struct {
71 datagram *transport.ClientDatagram
72 mitmManager *mitmManager
73
73 - httpClient *http.Client
74 - tlsConfig *tls.Config
74 + httpClient *http.Client
75 + leaseHTTPClient *http.Client
76 + tlsConfig *tls.Config
77
78 leaseMu sync.RWMutex
79 lease *listenerLease
@@ -99,11 +101,25 @@ func newListener(ctx context.Context, relayURL string, cfg listenerConfig) (*lis
101 cancel()
102 return nil, fmt.Errorf("parse relay url: %w", err)
103 }
104 + leaseAPIURL := relayurl
105 + if len(cfg.MultiHop) > 0 {
106 + normalizedEntryURL, err := utils.NormalizeRelayURL(cfg.MultiHop[0])
107 + if err != nil {
108 + cancel()
109 + return nil, fmt.Errorf("normalize multi-hop entry relay url: %w", err)
110 + }
111 + leaseAPIURL, err = url.Parse(normalizedEntryURL)
112 + if err != nil {
113 + cancel()
114 + return nil, fmt.Errorf("parse multi-hop entry relay url: %w", err)
115 + }
116 + }
117
118 l := &listener{
119 cancel: cancel,
120 doneCh: listenerCtx.Done(),
121 relayURL: relayurl,
122 + leaseAPIURL: leaseAPIURL,
123 identity: cfg.Identity,
124 metadata: cfg.Metadata,
125 relaySet: cfg.relaySet,
@@ -231,7 +247,7 @@ func (l *listener) Close() error {
247
248 if lease != nil && lease.hostname != "" && l.identity.Key() != "" && lease.accessToken != "" {
249 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
234 - closeErr = errors.Join(closeErr, l.unregisterLease(ctx, lease.accessToken))
250 + closeErr = errors.Join(closeErr, l.unregisterLease(ctx, lease.accessToken, lease.hop))
251 cancel()
252 }
253 if lease != nil && lease.tlsCloser != nil {
@@ -251,6 +267,7 @@ type listenerLease struct {
267 publicURLBase *url.URL
268 tlsConfig *tls.Config
269 tlsCloser io.Closer
270 + hop *types.HopRoute
271 }
272
273 func (l *listener) clearLease(reason string) *listenerLease {
@@ -703,7 +720,7 @@ func (l *listener) renewLease(ctx context.Context) error {
720 }
721
722 requestCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
706 - resp, err := l.renewRegisteredLease(requestCtx, l.leaseTTL, lease.accessToken)
723 + resp, err := l.renewRegisteredLease(requestCtx, l.leaseTTL, lease.accessToken, lease.hop)
724 cancel()
725 if err != nil {
726 if errors.Is(err, &types.APIRequestError{Code: types.APIErrorCodeLeaseNotFound}) {
@@ -734,7 +751,7 @@ func (l *listener) registerAndConfigure(ctx context.Context) error {
751 return err
752 }
753
737 - resp, err := l.registerLease(ctx, l.leaseTTL, l.udpEnabled, l.tcpEnabled)
754 + resp, hop, err := l.registerLease(ctx, l.leaseTTL, l.udpEnabled, l.tcpEnabled)
755 if err != nil {
756 return err
757 }
@@ -744,22 +761,22 @@ func (l *listener) registerAndConfigure(ctx context.Context) error {
761 }
762 registeredIdentity, err := utils.NormalizeIdentity(resp.Identity)
763 if err != nil {
747 - _ = l.unregisterLease(context.Background(), resp.AccessToken)
764 + _ = l.unregisterLease(context.Background(), resp.AccessToken, hop)
765 return err
766 }
767 if registeredIdentity.Key() != l.identity.Key() {
751 - _ = l.unregisterLease(context.Background(), resp.AccessToken)
768 + _ = l.unregisterLease(context.Background(), resp.AccessToken, hop)
769 return errors.New("relay returned mismatched lease identity")
770 }
771 if l.udpEnabled && !resp.UDPEnabled {
755 - _ = l.unregisterLease(context.Background(), resp.AccessToken)
772 + _ = l.unregisterLease(context.Background(), resp.AccessToken, hop)
773 return &types.APIRequestError{
774 Code: types.APIErrorCodeFeatureUnavailable,
775 Message: "relay did not enable required udp support",
776 }
777 }
778 if l.udpEnabled && resp.SNIPort <= 0 {
762 - _ = l.unregisterLease(context.Background(), resp.AccessToken)
779 + _ = l.unregisterLease(context.Background(), resp.AccessToken, hop)
780 return errors.New("relay did not return sni port for udp transport")
781 }
782 keylessURL := strings.TrimSpace(resp.KeylessURL)
@@ -774,12 +791,15 @@ func (l *listener) registerAndConfigure(ctx context.Context) error {
791 }
792 tlsConf, tenantTLSCloser, err := keyless.BuildClientTLSConfig(keylessURL, []string{resp.Hostname})
793 if err != nil {
777 - _ = l.unregisterLease(context.Background(), resp.AccessToken)
794 + _ = l.unregisterLease(context.Background(), resp.AccessToken, hop)
795 + if tenantTLSCloser != nil {
796 + _ = tenantTLSCloser.Close()
797 + }
798 return err
799 }
800
801 if ctx.Err() != nil {
782 - _ = l.unregisterLease(context.Background(), resp.AccessToken)
802 + _ = l.unregisterLease(context.Background(), resp.AccessToken, hop)
803 if tenantTLSCloser != nil {
804 _ = tenantTLSCloser.Close()
805 }
@@ -793,6 +813,7 @@ func (l *listener) registerAndConfigure(ctx context.Context) error {
813 publicURLBase: publicURLBase,
814 tlsConfig: tlsConf,
815 tlsCloser: tenantTLSCloser,
816 + hop: hop,
817 }
818 if l.udpEnabled {
819 next.sniPort = resp.SNIPort
types/api.go
+21 -11
@@ -56,10 +56,11 @@ func (e *APIRequestError) Is(target error) bool {
56 }
57
58 type RegisterRequest struct {
59 - ChallengeID string `json:"challenge_id"`
60 - SIWEMessage string `json:"siwe_message"`
61 - SIWESignature string `json:"siwe_signature"`
62 - ReportedIP string `json:"reported_ip,omitempty"`
59 + ChallengeID string `json:"challenge_id"`
60 + SIWEMessage string `json:"siwe_message"`
61 + SIWESignature string `json:"siwe_signature"`
62 + ReportedIP string `json:"reported_ip,omitempty"`
63 + Hop *HopRoute `json:"hop,omitempty"`
64 }
65
66 type RegisterChallengeRequest struct {
@@ -68,9 +69,8 @@ type RegisterChallengeRequest struct {
69 TTL int `json:"ttl,omitempty"`
70 UDPEnabled bool `json:"udp_enabled,omitempty"`
71 TCPEnabled bool `json:"tcp_enabled,omitempty"`
71 - // MultiHop is the ordered relay path for a multi-hop lease.
72 - // The first hop is public ingress and the last hop is the exit relay receiving registration.
73 - MultiHop []RelayDescriptor `json:"multi_hop,omitempty"`
72 + HopToken string `json:"hop_token,omitempty"`
73 + Hop *HopRoute `json:"hop,omitempty"`
74 }
75
76 type RegisterChallengeResponse struct {
@@ -118,9 +118,10 @@ type QUICControlResponse struct {
118 }
119
120 type RenewRequest struct {
121 - AccessToken string `json:"access_token"`
122 - TTL int `json:"ttl,omitempty"`
123 - ReportedIP string `json:"reported_ip,omitempty"`
121 + AccessToken string `json:"access_token"`
122 + TTL int `json:"ttl,omitempty"`
123 + ReportedIP string `json:"reported_ip,omitempty"`
124 + Hop *HopRoute `json:"hop,omitempty"`
125 }
126
127 type RenewResponse struct {
@@ -129,7 +130,16 @@ type RenewResponse struct {
130 }
131
132 type UnregisterRequest struct {
132 - AccessToken string `json:"access_token"`
133 + AccessToken string `json:"access_token"`
134 + Hop *HopRoute `json:"hop,omitempty"`
135 +}
136 +
137 +type HopRoute struct {
138 + MatchHostname string `json:"match_hostname,omitempty"`
139 + MatchToken string `json:"match_token,omitempty"`
140 + ForwardRelay RelayDescriptor `json:"forward_relay"`
141 + ForwardToken string `json:"forward_token"`
142 + Next *HopRoute `json:"next,omitempty"`
143 }
144
145 type DomainResponse struct {
types/identity.go
+22
@@ -1,7 +1,11 @@
1 package types
2
3 import (
4 + "crypto/hmac"
5 + "crypto/sha256"
6 + "encoding/base64"
7 "encoding/json"
8 + "errors"
9 "strings"
10 "time"
11 )
@@ -57,6 +61,24 @@ func (i Identity) Key() string {
61 return name + IdentityKeySeparator + address
62 }
63
64 +func (i Identity) DeriveToken(nonce string) (string, error) {
65 + privateKey := strings.TrimSpace(i.PrivateKey)
66 + if privateKey == "" {
67 + return "", errors.New("identity private key is required")
68 + }
69 + nonce = strings.TrimSpace(nonce)
70 + if nonce == "" {
71 + return "", errors.New("identity token nonce is required")
72 + }
73 +
74 + mac := hmac.New(sha256.New, []byte(privateKey))
75 + _, _ = mac.Write([]byte("Portal identity token v1\n"))
76 + _, _ = mac.Write([]byte(i.Key()))
77 + _, _ = mac.Write([]byte("\n"))
78 + _, _ = mac.Write([]byte(nonce))
79 + return base64.RawURLEncoding.EncodeToString(mac.Sum(nil)), nil
80 +}
81 +
82 type LeaseMetadata struct {
83 Description string `json:"description,omitempty"`
84 Owner string `json:"owner,omitempty"`
types/types.go
+1 -1
@@ -2,7 +2,7 @@ package types
2
3 const (
4 ReleaseVersion = "v2.1.5"
5 - SDKVersion = "5"
5 + SDKVersion = "6"
6 DiscoveryVersion = "6"
7 PortalRelayRegistryURL = "https://raw.githubusercontent.com/gosuda/portal-tunnel/main/registry.json"
8
utils/api.go
+29 -8
@@ -4,6 +4,7 @@ import (
4 "bytes"
5 "context"
6 "encoding/json"
7 + "errors"
8 "fmt"
9 "io"
10 "net/http"
@@ -38,6 +39,15 @@ func WriteAPIError(w http.ResponseWriter, status int, code, message string) {
39 })
40 }
41
42 +func WriteRawAPIResponse(w http.ResponseWriter, status int, body []byte) {
43 + if status <= 0 {
44 + status = http.StatusBadGateway
45 + }
46 + w.Header().Set("Content-Type", "application/json")
47 + w.WriteHeader(status)
48 + _, _ = w.Write(body)
49 +}
50 +
51 func MethodNotAllowedError() APIErrorResponse {
52 return APIErrorResponse{
53 Status: http.StatusMethodNotAllowed,
@@ -121,21 +131,32 @@ func HTTPDoAPIPath(ctx context.Context, client *http.Client, baseURL *url.URL, m
131 return DecodeAPIRequestError(resp)
132 }
133
134 + respBody, err := io.ReadAll(resp.Body)
135 + if err != nil {
136 + return fmt.Errorf("read response: %w", err)
137 + }
138 + if err := DecodeAPIData(respBody, out); err != nil {
139 + var apiErr *types.APIRequestError
140 + if errors.As(err, &apiErr) && apiErr.StatusCode == 0 {
141 + apiErr.StatusCode = resp.StatusCode
142 + }
143 + return err
144 + }
145 + return nil
146 +}
147 +
148 +func DecodeAPIData(body []byte, out any) error {
149 var envelope types.APIEnvelope[json.RawMessage]
125 - if err := json.NewDecoder(resp.Body).Decode(&envelope); err != nil {
150 + if err := json.Unmarshal(body, &envelope); err != nil {
151 return fmt.Errorf("decode response: %w", err)
152 }
153 if !envelope.OK {
154 if envelope.Error == nil {
130 - return &types.APIRequestError{
131 - StatusCode: resp.StatusCode,
132 - Message: fmt.Sprintf("api request failed with status %d", resp.StatusCode),
133 - }
155 + return &types.APIRequestError{Message: "api response is not ok"}
156 }
157 return &types.APIRequestError{
136 - StatusCode: resp.StatusCode,
137 - Code: envelope.Error.Code,
138 - Message: envelope.Error.Message,
158 + Code: envelope.Error.Code,
159 + Message: envelope.Error.Message,
160 }
161 }
162 if out == nil {
utils/identity.go
+2 -3
@@ -455,10 +455,9 @@ func populateRelayIdentity(identity *types.RelayIdentity, discoveryEnabled bool)
455 }
456
457 if strings.TrimSpace(identity.AdminSecretKey) == "" {
458 - var err error
459 - adminSecretKey, err := RandomHex(16)
458 + adminSecretKey, err := identity.Identity.DeriveToken("admin")
459 if err != nil {
461 - return fmt.Errorf("generate relay admin secret key: %w", err)
460 + return fmt.Errorf("derive relay admin secret key: %w", err)
461 }
462 identity.AdminSecretKey = adminSecretKey
463 }