@samitouri / QOSAMI-X402Tunnel-1 / commits / 41b7c8f1

feat: implement multi-hop relay support in overlay

Kim committed Apr 16, 2026 at 17:03 UTC 41b7c8f19656ea67d64310f26fd4d566b8f27253
18 files changed +1317 -166
cmd/portal-tunnel/main.go
+4
@@ -34,6 +34,7 @@ func main() {
34
35 type exposeFlags struct {
36 relayCSV string
37 + multiHopCSV string
38 discovery bool
39 banMITM bool
40 identityPath string
@@ -57,6 +58,7 @@ func runExposeCommand(args []string) error {
58 fs := utils.NewFlagSet("expose", printExposeUsage)
59
60 utils.StringFlag(fs, &flags.relayCSV, "relays", "", "Additional Portal relay server API URLs (comma-separated; scheme omitted defaults to https)")
61 + utils.StringFlagEnv(fs, &flags.multiHopCSV, "multi-hop", "", "Ordered multi-hop relay API URLs, comma-separated", "MULTI_HOP")
62 utils.BoolFlag(fs, &flags.discovery, "discovery", true, "Include public registry relays and discover additional relay bootstraps")
63 utils.BoolFlagEnv(fs, &flags.banMITM, "ban-mitm", true, "Ban relay when the MITM self-probe detects TLS termination", "BAN_MITM")
64 utils.StringFlagEnv(fs, &flags.identityPath, "identity-path", "identity.json", "identity json file path", "IDENTITY_PATH")
@@ -110,6 +112,7 @@ func runExposeCommand(args []string) error {
112 UDPAddr: flags.udpAddr,
113 UDPEnabled: flags.udp,
114 TCPEnabled: flags.tcp,
115 + MultiHop: utils.SplitCSV(flags.multiHopCSV),
116 BanMITM: flags.banMITM,
117 MaxActiveRelays: flags.maxActiveRelays,
118 Metadata: types.LeaseMetadata{
@@ -207,6 +210,7 @@ func printExposeUsage(w io.Writer) {
210 "portal expose 3000 --udp --udp-addr 127.0.0.1:5353",
211 "portal expose 3000 --ban-mitm",
212 "portal expose 3000 --relays https://portal.example.com --discovery=false",
213 + "portal expose 3000 --multi-hop https://entry.example.com,https://transit.example.com,https://exit.example.com",
214 },
215 )
216 }
go.mod
+1
@@ -14,6 +14,7 @@ require (
14 github.com/go-jose/go-jose/v4 v4.1.4
15 github.com/go-rod/rod v0.116.2
16 github.com/gosuda/keyless_tls v0.0.1-0.20260304212324-7733f8366abc
17 + github.com/hashicorp/yamux v0.1.2
18 github.com/quic-go/quic-go v0.59.0
19 github.com/rs/zerolog v1.34.0
20 github.com/spruceid/siwe-go v0.2.1
go.sum
+2
@@ -83,6 +83,8 @@ github.com/googleapis/gax-go/v2 v2.17.0 h1:RksgfBpxqff0EZkDWYuz9q/uWsTVz+kf43LsZ
83 github.com/googleapis/gax-go/v2 v2.17.0/go.mod h1:mzaqghpQp4JDh3HvADwrat+6M3MOIDp5YKHhb9PAgDY=
84 github.com/gosuda/keyless_tls v0.0.1-0.20260304212324-7733f8366abc h1:aS9LQ35x6EtrGKCmOWRj6Y9aQ2l5hP8dVva4oxB9VEg=
85 github.com/gosuda/keyless_tls v0.0.1-0.20260304212324-7733f8366abc/go.mod h1:BOhUZgiAAQzxKO3QcC4fCXgd/+lqxgIu1OyIYTqtta8=
86 +github.com/hashicorp/yamux v0.1.2 h1:XtB8kyFOyHXYVFnwT5C3+Bdo8gArse7j2AQ0DA0Uey8=
87 +github.com/hashicorp/yamux v0.1.2/go.mod h1:C+zze2n6e/7wshOZep2A70/aQU6QBRWJO/G6FT1wIns=
88 github.com/holiman/uint256 v1.3.2 h1:a9EgMPSC1AAaj1SZL5zIQD3WbwTuHrMGOerLjGmM/TA=
89 github.com/holiman/uint256 v1.3.2/go.mod h1:EOMSn4q6Nyt9P6efbI3bueV4e1b3dGlUCXeiRV4ng7E=
90 github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
portal/api_server.go
+100 -18
@@ -382,6 +382,35 @@ func (s *Server) handleRegisterChallenge(w http.ResponseWriter, r *http.Request)
382 Path: types.PathSDKRegister,
383 }).String()
384
385 + if len(req.MultiHop) > 0 && (s.hops == nil || s.overlay == nil || s.relaySet == nil) {
386 + utils.WriteAPIError(w, http.StatusServiceUnavailable, types.APIErrorCodeFeatureUnavailable, errFeatureUnavailable.Error())
387 + return
388 + }
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 + }
414 if req.UDPEnabled && (!s.cfg.UDPEnabled || s.group != nil && s.quicTunnel == nil) {
415 utils.WriteAPIError(w, http.StatusServiceUnavailable, types.APIErrorCodeFeatureUnavailable, errFeatureUnavailable.Error())
416 return
@@ -430,6 +459,10 @@ func (s *Server) handleRenew(w http.ResponseWriter, r *http.Request) {
459 writeAPIErrorResponse(w, err)
460 return
461 }
462 + if err := s.hops.renew(record); err != nil {
463 + utils.WriteAPIError(w, http.StatusServiceUnavailable, types.APIErrorCodeFeatureUnavailable, err.Error())
464 + return
465 + }
466 nextAccessToken, _, err := auth.IssueLeaseAccessToken(s.identity.PrivateKey, s.identity.Address, s.cfg.PortalURL, record.Copy(), ttl)
467 if err != nil {
468 utils.WriteAPIError(w, http.StatusInternalServerError, types.APIErrorCodeInternal, err.Error())
@@ -462,18 +495,19 @@ func (s *Server) handleUnregister(w http.ResponseWriter, r *http.Request) {
495 writeAPIErrorResponse(w, err)
496 return
497 }
465 - deleteCtx, cancel := context.WithTimeout(context.Background(), defaultClaimTimeout)
466 - defer cancel()
467 - if err := s.acmeManager.DeleteENSGaslessHostname(deleteCtx, record.Hostname); err != nil {
468 - log.Warn().
469 - Err(err).
470 - Str("hostname", record.Hostname).
471 - Str("address", record.Address).
472 - Msg("delete lease ens gasless txt")
473 - }
474 - if record != nil {
475 - record.Close()
498 + s.hops.delete(record)
499 + if record.isDirect() && s.acmeManager != nil {
500 + deleteCtx, cancel := context.WithTimeout(context.Background(), defaultClaimTimeout)
501 + if err := s.acmeManager.DeleteENSGaslessHostname(deleteCtx, record.Hostname); err != nil {
502 + log.Warn().
503 + Err(err).
504 + Str("hostname", record.Hostname).
505 + Str("address", record.Address).
506 + Msg("delete lease remote state")
507 + }
508 + cancel()
509 }
510 + record.Close()
511
512 utils.WriteAPIData(w, http.StatusOK, map[string]any{})
513 }
@@ -602,9 +636,9 @@ func (s *Server) admitLeaseByToken(token string, requireDatagram bool) (*leaseRe
636 if err != nil {
637 return nil, errUnauthorized
638 }
605 - lease, err := s.registry.Find(claims.Identity)
606 - if err != nil {
607 - return nil, err
639 + lease, ok := s.registry.RecordByKey(claims.Identity.Key(), time.Now())
640 + if !ok {
641 + return nil, errLeaseNotFound
642 }
643 if !s.registry.policy.IsIdentityRoutable(lease.Key()) {
644 return nil, errLeaseRejected
@@ -661,6 +695,16 @@ func (s *Server) registerLease(req types.RegisterChallengeRequest, clientIP, rep
695 }
696 issuedAt := claims.IssuedAt.Time().UTC()
697 expiresAt := claims.Expiry.Time().UTC()
698 + if len(req.MultiHop) > 0 && (s.hops == nil || s.overlay == nil || s.relaySet == nil) {
699 + return types.RegisterResponse{}, errFeatureUnavailable
700 + }
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 + }
708 identityKey := identity.Key()
709 stream := transport.NewRelayStream(identityKey, defaultIdleKeepalive, defaultReadyQueueLimit)
710 record := &leaseRecord{
@@ -674,8 +718,14 @@ func (s *Server) registerLease(req types.RegisterChallengeRequest, clientIP, rep
718 ReportedIP: utils.SanitizeReportedIP(reportedIP),
719 UDPEnabled: req.UDPEnabled,
720 TCPEnabled: req.TCPEnabled,
721 + multiHop: multiHop,
722 stream: stream,
723 }
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 + }
729 if req.UDPEnabled {
730 if s.udpPorts == nil {
731 return types.RegisterResponse{}, errors.New("udp port allocation not available")
@@ -709,17 +759,46 @@ func (s *Server) registerLease(req types.RegisterChallengeRequest, clientIP, rep
759 return types.RegisterResponse{}, err
760 }
761
712 - if err := s.registry.Register(record); err != nil {
762 + replaced, err := s.registry.Register(record)
763 + if err != nil {
764 record.Close()
765 return types.RegisterResponse{}, err
766 }
716 - syncCtx, cancel := context.WithTimeout(context.Background(), defaultClaimTimeout)
717 - defer cancel()
718 - if err := s.acmeManager.SyncENSGaslessHostname(syncCtx, record.Hostname, record.Address); err != nil {
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())
790 record.Close()
791 return types.RegisterResponse{}, err
792 }
793 + if record.isDirect() {
794 + syncCtx, cancel := context.WithTimeout(context.Background(), defaultClaimTimeout)
795 + defer cancel()
796 + if err := s.acmeManager.SyncENSGaslessHostname(syncCtx, record.Hostname, record.Address); err != nil {
797 + _, _ = s.registry.Unregister(record.Copy())
798 + record.Close()
799 + return types.RegisterResponse{}, err
800 + }
801 + }
802
803 resp := types.RegisterResponse{
804 Identity: record.Copy(),
@@ -729,6 +808,9 @@ func (s *Server) registerLease(req types.RegisterChallengeRequest, clientIP, rep
808 UDPEnabled: record.UDPEnabled,
809 TCPEnabled: record.TCPEnabled,
810 }
811 + if !record.isDirect() {
812 + resp.KeylessURL = record.multiHop[0].APIHTTPSAddr
813 + }
814 if record.datagram != nil {
815 resp.SNIPort = s.cfg.SNIPort
816 resp.UDPAddr = fmt.Sprintf("%s:%d", s.identity.Name, record.datagram.UDPPort())
portal/auth/register_challenge.go
+6
@@ -48,12 +48,18 @@ 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 +
56 normalizedRequest := types.RegisterChallengeRequest{
57 Identity: normalizedIdentity,
58 Metadata: req.Metadata.Copy(),
59 TTL: req.TTL,
60 UDPEnabled: req.UDPEnabled,
61 TCPEnabled: req.TCPEnabled,
62 + MultiHop: multiHop,
63 }
64
65 return &RegisterChallenge{
portal/hop.go new
+436
@@ -0,0 +1,436 @@
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
+45 -22
@@ -18,7 +18,7 @@ import (
18 const defaultRegisterChallengeTTL = 2 * time.Minute
19
20 type leaseRegistry struct {
21 - routes map[string]string
21 + leasesByHostname map[string]string
22 leasesByKey map[string]*leaseRecord
23 registerChallenges map[string]*auth.RegisterChallenge
24 policy *policy.Runtime
@@ -32,7 +32,7 @@ func newLeaseRegistry(udpEnabled, tcpPortEnabled bool, trustProxyHeaders bool, r
32 }
33
34 return &leaseRegistry{
35 - routes: make(map[string]string),
35 + leasesByHostname: make(map[string]string),
36 leasesByKey: make(map[string]*leaseRecord),
37 registerChallenges: make(map[string]*auth.RegisterChallenge),
38 policy: runtime,
@@ -48,7 +48,7 @@ func (r *leaseRegistry) CloseAll() []*leaseRecord {
48 out = append(out, record)
49 r.policy.ForgetIdentity(record.Key())
50 }
51 - r.routes = make(map[string]string)
51 + r.leasesByHostname = make(map[string]string)
52 r.leasesByKey = make(map[string]*leaseRecord)
53 r.registerChallenges = make(map[string]*auth.RegisterChallenge)
54 return out
@@ -63,13 +63,13 @@ func (r *leaseRegistry) Lookup(host string) (*leaseRecord, bool) {
63 r.mu.RLock()
64 defer r.mu.RUnlock()
65
66 - key, ok := r.routes[host]
66 + key, ok := r.leasesByHostname[host]
67 if !ok {
68 parts := strings.Split(host, ".")
69 if len(parts) < 3 {
70 return nil, false
71 }
72 - key, ok = r.routes["*."+strings.Join(parts[1:], ".")]
72 + key, ok = r.leasesByHostname["*."+strings.Join(parts[1:], ".")]
73 if !ok {
74 return nil, false
75 }
@@ -78,43 +78,47 @@ func (r *leaseRegistry) Lookup(host string) (*leaseRecord, bool) {
78 return record, ok && record != nil
79 }
80
81 -func (r *leaseRegistry) Register(record *leaseRecord) error {
81 +func (r *leaseRegistry) Register(record *leaseRecord) (*leaseRecord, error) {
82 if record == nil {
83 - return errors.New("lease record is required")
83 + return nil, errors.New("lease record is required")
84 }
85
86 key := record.Key()
87 if key == "" {
88 - return errors.New("lease identity is required")
88 + return nil, errors.New("lease identity is required")
89 }
90 hostname := utils.NormalizeHostname(record.Hostname)
91 if hostname == "" {
92 - return errors.New("lease hostname is required")
92 + return nil, errors.New("lease hostname is required")
93 }
94
95 r.mu.Lock()
96
97 - if existingKey, ok := r.routes[hostname]; ok && existingKey != key {
97 + if existingKey, ok := r.leasesByHostname[hostname]; ok && existingKey != key {
98 r.mu.Unlock()
99 - return errHostnameConflict
99 + return nil, errHostnameConflict
100 }
101
102 var replaced *leaseRecord
103 if existing, ok := r.leasesByKey[key]; ok && existing != nil {
104 replaced = existing
105 - delete(r.routes, utils.NormalizeHostname(existing.Hostname))
105 + if existing != record && existing.isDirect() {
106 + delete(r.leasesByHostname, utils.NormalizeHostname(existing.Hostname))
107 + }
108 r.policy.ForgetIdentity(existing.Key())
109 }
110 record.Hostname = hostname
111 r.leasesByKey[key] = record
110 - r.routes[hostname] = key
112 + if record.isDirect() {
113 + r.leasesByHostname[hostname] = key
114 + }
115 r.policy.IPFilter().RegisterIdentityIP(key, record.ClientIP)
116 r.mu.Unlock()
117
118 if replaced != nil && replaced != record {
119 replaced.Close()
120 }
117 - return nil
121 + return replaced, nil
122 }
123
124 func (r *leaseRegistry) Renew(identity types.Identity, ttl time.Duration, clientIP, reportedIP string) (*leaseRecord, error) {
@@ -127,7 +131,8 @@ func (r *leaseRegistry) Renew(identity types.Identity, ttl time.Duration, client
131 }
132
133 now := time.Now()
130 - record.ExpiresAt = now.Add(ttl)
134 + expiresAt := now.Add(ttl)
135 + record.ExpiresAt = expiresAt
136 record.LastSeenAt = now
137 if strings.TrimSpace(clientIP) != "" {
138 record.ClientIP = clientIP
@@ -150,23 +155,33 @@ func (r *leaseRegistry) Unregister(identity types.Identity) (*leaseRecord, error
155 }
156
157 delete(r.leasesByKey, key)
153 - delete(r.routes, utils.NormalizeHostname(record.Hostname))
158 + if record.isDirect() {
159 + delete(r.leasesByHostname, utils.NormalizeHostname(record.Hostname))
160 + }
161 r.policy.ForgetIdentity(key)
162 return record, nil
163 }
164
158 -func (r *leaseRegistry) Find(identity types.Identity) (*leaseRecord, error) {
165 +func (r *leaseRegistry) RecordByKey(key string, now time.Time) (*leaseRecord, bool) {
166 + key = strings.TrimSpace(key)
167 + if key == "" {
168 + return nil, false
169 + }
170 +
171 r.mu.RLock()
172 defer r.mu.RUnlock()
173
162 - record, ok := r.leasesByKey[identity.Key()]
163 - if !ok || time.Now().After(record.ExpiresAt) {
164 - return nil, errLeaseNotFound
174 + record := r.leasesByKey[key]
175 + if record == nil || now.After(record.ExpiresAt) {
176 + return nil, false
177 }
166 - return record, nil
178 + return record, true
179 }
180
181 func (r *leaseRegistry) issueRegisterChallenge(req types.RegisterChallengeRequest, domain, uri string) (types.RegisterChallengeResponse, error) {
182 + if len(req.MultiHop) > 0 && (req.UDPEnabled || req.TCPEnabled) {
183 + return types.RegisterChallengeResponse{}, errTransportMismatch
184 + }
185 if req.UDPEnabled {
186 if !r.policy.IsUDPEnabled() {
187 return types.RegisterChallengeResponse{}, errUDPDisabled
@@ -251,7 +266,9 @@ func (r *leaseRegistry) cleanupExpired(now time.Time) []*leaseRecord {
266 if now.After(record.ExpiresAt) {
267 expired = append(expired, record)
268 delete(r.leasesByKey, key)
254 - delete(r.routes, utils.NormalizeHostname(record.Hostname))
269 + if record.isDirect() {
270 + delete(r.leasesByHostname, utils.NormalizeHostname(record.Hostname))
271 + }
272 r.policy.ForgetIdentity(key)
273 }
274 }
@@ -361,6 +378,8 @@ type leaseRecord struct {
378 UDPEnabled bool
379 TCPEnabled bool
380 Metadata types.LeaseMetadata
381 + multiHop []types.RelayDescriptor
382 + hopID string
383 datagram *transport.RelayDatagram
384 udpPorts *transport.PortAllocator
385 tcpPort *transport.RelayTCPPort
@@ -370,6 +389,10 @@ type leaseRecord struct {
389 startOnce sync.Once
390 }
391
392 +func (r *leaseRecord) isDirect() bool {
393 + return len(r.multiHop) == 0
394 +}
395 +
396 func (r *leaseRegistry) AdminSnapshot(record *leaseRecord) types.AdminLease {
397 clientIP := record.ClientIP
398 identityKey := record.Key()
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 new
+470
@@ -0,0 +1,470 @@
1 +package overlay
2 +
3 +import (
4 + "context"
5 + "encoding/binary"
6 + "encoding/json"
7 + "errors"
8 + "fmt"
9 + "io"
10 + "net"
11 + "strings"
12 + "sync"
13 + "time"
14 +
15 + "github.com/hashicorp/yamux"
16 + "github.com/rs/zerolog/log"
17 +)
18 +
19 +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
26 +)
27 +
28 +const (
29 + hopModeTLSStream = "tls-stream"
30 + hopModeControl = "control"
31 +
32 + HopControlInstall = "install"
33 + HopControlRenew = "renew"
34 + HopControlDelete = "delete"
35 +)
36 +
37 +type HopMux struct {
38 + listener net.Listener
39 + overlay *Overlay
40 + incoming chan HopStream
41 +
42 + mu sync.Mutex
43 + sessions map[string]*yamux.Session
44 + active map[*yamux.Session]struct{}
45 + closed bool
46 +}
47 +
48 +type hopPreface struct {
49 + Version int `json:"version"`
50 + Mode string `json:"mode"`
51 + Token string `json:"token"`
52 +}
53 +
54 +type HopControl struct {
55 + Action string `json:"action"`
56 + Route HopRouteSpec `json:"route"`
57 +}
58 +
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"`
66 +}
67 +
68 +type hopControlResponse struct {
69 + OK bool `json:"ok"`
70 + Error string `json:"error,omitempty"`
71 +}
72 +
73 +type HopStream struct {
74 + Conn net.Conn
75 + Token string
76 + Control *HopControl
77 +}
78 +
79 +func NewHopMux(overlay *Overlay) (*HopMux, error) {
80 + if overlay == nil || overlay.stack == nil {
81 + return nil, errors.New("overlay is required for multi-hop mux")
82 + }
83 + listener, err := overlay.stack.ListenTCP(DefaultPeerYamuxPort)
84 + if err != nil {
85 + return nil, fmt.Errorf("listen hop yamux: %w", err)
86 + }
87 + return &HopMux{
88 + listener: listener,
89 + overlay: overlay,
90 + incoming: make(chan HopStream, hopIncomingBuffer),
91 + sessions: make(map[string]*yamux.Session),
92 + active: make(map[*yamux.Session]struct{}),
93 + }, nil
94 +}
95 +
96 +func (m *HopMux) Serve(ctx context.Context) error {
97 + if m == nil || m.listener == nil {
98 + <-ctx.Done()
99 + return nil
100 + }
101 + go func() {
102 + <-ctx.Done()
103 + _ = m.Close()
104 + }()
105 +
106 + for {
107 + conn, err := m.listener.Accept()
108 + switch {
109 + case err == nil:
110 + go m.serveSession(ctx, conn)
111 + case errors.Is(err, net.ErrClosed):
112 + return nil
113 + default:
114 + if ctx.Err() != nil {
115 + return nil
116 + }
117 + return fmt.Errorf("accept hop mux connection: %w", err)
118 + }
119 + }
120 +}
121 +
122 +func (m *HopMux) Accept(ctx context.Context) (HopStream, error) {
123 + if m == nil {
124 + <-ctx.Done()
125 + return HopStream{}, ctx.Err()
126 + }
127 + select {
128 + case stream := <-m.incoming:
129 + return stream, nil
130 + case <-ctx.Done():
131 + return HopStream{}, ctx.Err()
132 + }
133 +}
134 +
135 +func (m *HopMux) Close() error {
136 + if m == nil {
137 + return nil
138 + }
139 +
140 + m.mu.Lock()
141 + if m.closed {
142 + m.mu.Unlock()
143 + return nil
144 + }
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 {
157 + continue
158 + }
159 + seen[session] = struct{}{}
160 + sessions = append(sessions, session)
161 + }
162 + m.sessions = make(map[string]*yamux.Session)
163 + m.active = make(map[*yamux.Session]struct{})
164 + listener := m.listener
165 + m.mu.Unlock()
166 +
167 + var closeErr error
168 + if listener != nil {
169 + closeErr = errors.Join(closeErr, listener.Close())
170 + }
171 + for _, session := range sessions {
172 + closeErr = errors.Join(closeErr, session.Close())
173 + }
174 + return closeErr
175 +}
176 +
177 +func (m *HopMux) OpenStream(ctx context.Context, overlayIPv4, token string) (net.Conn, error) {
178 + token = strings.TrimSpace(token)
179 + if token == "" {
180 + return nil, errors.New("next hop token is required")
181 + }
182 +
183 + stream, err := m.openYamuxStream(ctx, overlayIPv4)
184 + if err != nil {
185 + return nil, err
186 + }
187 + preface := hopPreface{
188 + Version: hopProtocolVersion,
189 + Mode: hopModeTLSStream,
190 + Token: token,
191 + }
192 + if err := writeFramedJSON(stream, preface, hopPrefaceLimit); err != nil {
193 + _ = stream.Close()
194 + return nil, err
195 + }
196 + return stream, nil
197 +}
198 +
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")
203 + }
204 +
205 + stream, err := m.openYamuxStream(ctx, overlayIPv4)
206 + if err != nil {
207 + return err
208 + }
209 + defer stream.Close()
210 +
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 +
217 + preface := hopPreface{
218 + Version: hopProtocolVersion,
219 + Mode: hopModeControl,
220 + }
221 + if err := writeFramedJSON(stream, preface, hopPrefaceLimit); err != nil {
222 + return err
223 + }
224 + if err := writeFramedJSON(stream, req, hopControlPayloadLimit); err != nil {
225 + return err
226 + }
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)
236 + }
237 + return nil
238 +}
239 +
240 +func (m *HopMux) openYamuxStream(ctx context.Context, overlayIPv4 string) (*yamux.Stream, error) {
241 + overlayIPv4 = strings.TrimSpace(overlayIPv4)
242 + if overlayIPv4 == "" {
243 + return nil, errors.New("next hop overlay ipv4 is required")
244 + }
245 +
246 + session, err := m.session(ctx, overlayIPv4)
247 + if err != nil {
248 + return nil, err
249 + }
250 + stream, err := session.OpenStream()
251 + if err != nil {
252 + m.forgetSession(overlayIPv4, session)
253 + _ = session.Close()
254 + return nil, err
255 + }
256 + return stream, nil
257 +}
258 +
259 +func (m *HopMux) session(ctx context.Context, overlayIPv4 string) (*yamux.Session, error) {
260 + m.mu.Lock()
261 + if m.closed {
262 + m.mu.Unlock()
263 + return nil, net.ErrClosed
264 + }
265 + if session := m.sessions[overlayIPv4]; session != nil {
266 + if !session.IsClosed() {
267 + m.mu.Unlock()
268 + return session, nil
269 + }
270 + delete(m.active, session)
271 + }
272 + delete(m.sessions, overlayIPv4)
273 + m.mu.Unlock()
274 +
275 + if m.overlay == nil || m.overlay.stack == nil {
276 + return nil, errors.New("overlay is not initialized")
277 + }
278 + addr := net.JoinHostPort(overlayIPv4, fmt.Sprintf("%d", DefaultPeerYamuxPort))
279 + conn, err := m.overlay.stack.DialContext(ctx, "tcp", addr)
280 + if err != nil {
281 + return nil, err
282 + }
283 +
284 + session, err := yamux.Client(conn, hopYamuxConfig())
285 + if err != nil {
286 + _ = conn.Close()
287 + return nil, err
288 + }
289 +
290 + m.mu.Lock()
291 + if m.closed {
292 + m.mu.Unlock()
293 + _ = session.Close()
294 + return nil, net.ErrClosed
295 + }
296 + if current := m.sessions[overlayIPv4]; current != nil && !current.IsClosed() {
297 + m.mu.Unlock()
298 + _ = session.Close()
299 + return current, nil
300 + }
301 + m.sessions[overlayIPv4] = session
302 + m.active[session] = struct{}{}
303 + m.mu.Unlock()
304 + return session, nil
305 +}
306 +
307 +func (m *HopMux) forgetSession(overlayIPv4 string, session *yamux.Session) {
308 + m.mu.Lock()
309 + defer m.mu.Unlock()
310 + if m.sessions[overlayIPv4] == session {
311 + delete(m.sessions, overlayIPv4)
312 + }
313 + delete(m.active, session)
314 +}
315 +
316 +func (m *HopMux) serveSession(ctx context.Context, conn net.Conn) {
317 + session, err := yamux.Server(conn, hopYamuxConfig())
318 + if err != nil {
319 + _ = conn.Close()
320 + log.Warn().Err(err).Msg("create hop yamux session")
321 + return
322 + }
323 + m.mu.Lock()
324 + if m.closed {
325 + m.mu.Unlock()
326 + _ = session.Close()
327 + return
328 + }
329 + m.active[session] = struct{}{}
330 + m.mu.Unlock()
331 + defer func() {
332 + m.mu.Lock()
333 + delete(m.active, session)
334 + m.mu.Unlock()
335 + _ = session.Close()
336 + }()
337 +
338 + for {
339 + stream, err := session.AcceptStream()
340 + if err != nil {
341 + return
342 + }
343 + go m.handleStream(ctx, stream)
344 + }
345 +}
346 +
347 +func (m *HopMux) handleStream(ctx context.Context, stream *yamux.Stream) {
348 + _ = stream.SetReadDeadline(time.Now().Add(defaultPrefaceTimeout))
349 + var preface hopPreface
350 + err := readFramedJSON(stream, &preface, hopPrefaceLimit)
351 + _ = stream.SetReadDeadline(time.Time{})
352 + if err != nil {
353 + _ = stream.Close()
354 + return
355 + }
356 + if preface.Version != hopProtocolVersion {
357 + _ = stream.Close()
358 + return
359 + }
360 + switch preface.Mode {
361 + case hopModeTLSStream:
362 + if strings.TrimSpace(preface.Token) == "" {
363 + _ = stream.Close()
364 + return
365 + }
366 + m.deliver(ctx, HopStream{
367 + Conn: stream,
368 + Token: preface.Token,
369 + })
370 + case hopModeControl:
371 + m.readControlStream(ctx, stream)
372 + default:
373 + _ = stream.Close()
374 + }
375 +}
376 +
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 +
395 +func (m *HopMux) deliver(ctx context.Context, stream HopStream) {
396 + select {
397 + case m.incoming <- stream:
398 + case <-ctx.Done():
399 + _ = stream.Conn.Close()
400 + }
401 +}
402 +
403 +func (s HopStream) Respond(err error) error {
404 + if s.Conn == nil {
405 + return nil
406 + }
407 + defer s.Conn.Close()
408 + resp := hopControlResponse{OK: err == nil}
409 + if err != nil {
410 + resp.Error = err.Error()
411 + }
412 + return writeFramedJSON(s.Conn, resp, hopControlPayloadLimit)
413 +}
414 +
415 +func hopYamuxConfig() *yamux.Config {
416 + cfg := yamux.DefaultConfig()
417 + cfg.Logger = nil
418 + cfg.MaxStreamWindowSize = 16 * 1024 * 1024
419 + cfg.StreamOpenTimeout = 75 * time.Second
420 + cfg.StreamCloseTimeout = 5 * time.Minute
421 + return cfg
422 +}
423 +
424 +func writeFramedJSON(w io.Writer, value any, limit int) error {
425 + payload, err := json.Marshal(value)
426 + if err != nil {
427 + return err
428 + }
429 + if len(payload) == 0 || len(payload) > limit {
430 + return errors.New("frame size is invalid")
431 + }
432 + var size [4]byte
433 + binary.BigEndian.PutUint32(size[:], uint32(len(payload)))
434 + if err := writeAll(w, size[:]); err != nil {
435 + return err
436 + }
437 + return writeAll(w, payload)
438 +}
439 +
440 +func readFramedJSON(r io.Reader, dst any, limit int) error {
441 + var size [4]byte
442 + if _, err := io.ReadFull(r, size[:]); err != nil {
443 + return err
444 + }
445 + n := binary.BigEndian.Uint32(size[:])
446 + if n == 0 || n > uint32(limit) {
447 + return errors.New("frame size is invalid")
448 + }
449 + payload := make([]byte, n)
450 + if _, err := io.ReadFull(r, payload); err != nil {
451 + return err
452 + }
453 + return json.Unmarshal(payload, dst)
454 +}
455 +
456 +func writeAll(w io.Writer, p []byte) error {
457 + for len(p) > 0 {
458 + n, err := w.Write(p)
459 + if n > 0 {
460 + p = p[n:]
461 + }
462 + if err != nil {
463 + return err
464 + }
465 + if n == 0 {
466 + return io.ErrShortWrite
467 + }
468 + }
469 + return nil
470 +}
portal/overlay/stack.go
+13
@@ -22,6 +22,7 @@ const (
22 DefaultMTU = 1420
23 DefaultListenPort = 51820
24 DefaultPeerAPIHTTPPort = 7777
25 + DefaultPeerYamuxPort = 7778
26 DefaultPersistentKeepalive = 25
27 defaultEndpointResolveTTL = 3 * time.Second
28 )
@@ -31,6 +32,7 @@ type stack struct {
32 net *netstack.Net
33 overlayIP netip.Addr
34
35 + applyMu sync.Mutex
36 mu sync.Mutex
37 closed bool
38 peerEndpoints map[string]string
@@ -124,6 +126,14 @@ func (s *stack) ApplyPeers(peers []desiredPeer) error {
126 if s == nil || s.device == nil {
127 return errors.New("wireguard is not initialized")
128 }
129 + s.applyMu.Lock()
130 + defer s.applyMu.Unlock()
131 + s.mu.Lock()
132 + if s.closed {
133 + s.mu.Unlock()
134 + return net.ErrClosed
135 + }
136 + s.mu.Unlock()
137
138 var builder strings.Builder
139 builder.WriteString("replace_peers=true\n")
@@ -232,6 +242,9 @@ func (s *stack) Close() error {
242 return nil
243 }
244
245 + s.applyMu.Lock()
246 + defer s.applyMu.Unlock()
247 +
248 s.mu.Lock()
249 if s.closed {
250 s.mu.Unlock()
portal/server.go
+44 -20
@@ -120,6 +120,7 @@ type Server struct {
120 relaySet *discovery.RelaySet
121 announceLimiter *discovery.AnnounceLimiter
122 registry *leaseRegistry
123 + hops *hopManager
124 udpPorts *transport.PortAllocator
125 tcpPorts *transport.PortAllocator
126 }
@@ -174,15 +175,19 @@ 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
177 - var overlay *overlay.Overlay
178 + var hops *hopManager
179 + var ov *overlay.Overlay
180 var quicTunnel *quic.Listener
181 defer func() {
182 if started {
183 return
184 }
185 acmeManager.Stop()
184 - if overlay != nil {
185 - _ = overlay.Shutdown(context.Background())
186 + if hops != nil {
187 + _ = hops.close()
188 + }
189 + if ov != nil {
190 + _ = ov.Shutdown(context.Background())
191 }
192 if apiServer != nil {
193 _ = apiServer.Close()
@@ -216,7 +221,11 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
221 }
222
223 if s.relaySet != nil && strings.TrimSpace(s.identity.WireGuardPrivateKey) != "" {
219 - overlay, err = s.startOverlay()
224 + ov, err = s.startOverlay()
225 + if err != nil {
226 + return err
227 + }
228 + hops, err = newHopManager(ov, s.registry, &s.proxy)
229 if err != nil {
230 return err
231 }
@@ -236,7 +245,8 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
245 s.acmeManager = acmeManager
246 s.cancel = cancel
247 s.group = group
239 - s.overlay = overlay
248 + s.overlay = ov
249 + s.hops = hops
250 s.quicTunnel = quicTunnel
251 started = true
252
@@ -249,6 +259,9 @@ 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) })
264 + }
265 if s.quicTunnel != nil {
266 group.Go(s.runQUICTunnelListener)
267 }
@@ -269,6 +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).
286 Bool("udp_enabled", s.quicTunnel != nil).
287 Bool("tcp_enabled", s.cfg.TCPEnabled)
288 if s.quicTunnel != nil {
@@ -306,14 +320,15 @@ func (s *Server) Shutdown(ctx context.Context) error {
320
321 for _, lease := range s.registry.CloseAll() {
322 if lease != nil {
309 - if s.acmeManager != nil {
323 + s.hops.delete(lease)
324 + if lease.isDirect() && s.acmeManager != nil {
325 deleteCtx, cancel := context.WithTimeout(ctx, defaultClaimTimeout)
326 if err := s.acmeManager.DeleteENSGaslessHostname(deleteCtx, lease.Hostname); err != nil {
327 log.Warn().
328 Err(err).
329 Str("hostname", lease.Hostname).
330 Str("address", lease.Address).
316 - Msg("delete lease ens gasless txt during shutdown")
331 + Msg("delete lease remote state during shutdown")
332 }
333 cancel()
334 }
@@ -334,6 +349,11 @@ func (s *Server) Shutdown(ctx context.Context) error {
349 shutdownErr = err
350 }
351 }
352 + if s.hops != nil {
353 + if err := s.hops.close(); err != nil && shutdownErr == nil && !errors.Is(err, net.ErrClosed) {
354 + shutdownErr = err
355 + }
356 + }
357 if s.overlay != nil {
358 if err := s.overlay.Shutdown(ctx); err != nil && shutdownErr == nil {
359 shutdownErr = err
@@ -465,19 +485,20 @@ func (s *Server) runSNIListener(ctx context.Context) error {
485
486 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()) {
468 - _ = wrappedConn.Close()
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)
473 - defer cancel()
474 -
496 session, err := record.stream.Claim(claimCtx)
497 + cancel()
498 if err != nil {
499 _ = wrappedConn.Close()
500 return
501 }
480 -
502 s.proxy.bridge(wrappedConn, session, record.Key(), s.registry.policy.BPSManager())
503 }(conn)
504 case errors.Is(err, net.ErrClosed):
@@ -505,15 +526,18 @@ func (s *Server) runLeaseJanitor(ctx context.Context, interval time.Duration) er
526 return nil
527 case <-ticker.C:
528 for _, lease := range s.registry.cleanupExpired(time.Now()) {
508 - deleteCtx, cancel := context.WithTimeout(context.Background(), defaultClaimTimeout)
509 - err := s.acmeManager.DeleteENSGaslessHostname(deleteCtx, lease.Hostname)
510 - cancel()
511 - if err != nil {
512 - log.Warn().
513 - Err(err).
514 - Str("hostname", lease.Hostname).
515 - Str("address", lease.Address).
516 - Msg("delete expired lease ens gasless txt")
529 + s.hops.delete(lease)
530 + if lease.isDirect() && s.acmeManager != nil {
531 + deleteCtx, cancel := context.WithTimeout(context.Background(), defaultClaimTimeout)
532 + err := s.acmeManager.DeleteENSGaslessHostname(deleteCtx, lease.Hostname)
533 + cancel()
534 + if err != nil {
535 + log.Warn().
536 + Err(err).
537 + Str("hostname", lease.Hostname).
538 + Str("address", lease.Address).
539 + Msg("delete expired lease remote state")
540 + }
541 }
542 lease.Close()
543 }
portal/server_test.go
+8 -8
@@ -239,7 +239,7 @@ func TestRegisterLeaseOmitsSNIPortWithoutUDP(t *testing.T) {
239 t.Fatalf("registerLease() error = %v", err)
240 }
241 t.Cleanup(func() {
242 - if record, err := server.registry.Find(resp.Identity); err == nil {
242 + if record, ok := server.registry.RecordByKey(resp.Identity.Key(), time.Now()); ok {
243 record.Close()
244 }
245 })
@@ -341,9 +341,9 @@ func TestRegisterLeaseDerivesFixedHostnameFromName(t *testing.T) {
341 t.Fatalf("registerLease() hostname = %q, want %q", resp.Hostname, wantHostname)
342 }
343
344 - record, err := server.registry.Find(resp.Identity)
345 - if err != nil {
346 - t.Fatalf("registry.Find() error = %v, want registered lease", err)
344 + record, ok := server.registry.RecordByKey(resp.Identity.Key(), time.Now())
345 + if !ok {
346 + t.Fatal("registry.RecordByKey() = false, want registered lease")
347 }
348 snapshot := server.registry.Snapshot(record)
349 if snapshot.Name != "demo-app" {
@@ -380,14 +380,14 @@ func TestRegisterLeaseBuildsUDPEnabledRuntime(t *testing.T) {
380 t.Fatalf("registerLease() error = %v", err)
381 }
382 t.Cleanup(func() {
383 - if record, err := server.registry.Find(resp.Identity); err == nil {
383 + if record, ok := server.registry.RecordByKey(resp.Identity.Key(), time.Now()); ok {
384 record.Close()
385 }
386 })
387
388 - record, err := server.registry.Find(resp.Identity)
389 - if err != nil {
390 - t.Fatalf("registry.Find() error = %v, want registered lease", err)
388 + record, ok := server.registry.RecordByKey(resp.Identity.Key(), time.Now())
389 + if !ok {
390 + t.Fatal("registry.RecordByKey() = false, want registered lease")
391 }
392 if record.stream == nil {
393 t.Fatal("stream = nil, want stream runtime")
sdk/api_client.go
+47
@@ -82,6 +82,11 @@ func (l *listener) initHTTPTransport(ctx context.Context) error {
82 }
83
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
88 + }
89 +
90 var challenge types.RegisterChallengeResponse
91 if err := utils.HTTPDoAPIPath(ctx, l.httpClient, l.relayURL, http.MethodPost, types.PathSDKRegisterChallenge, types.RegisterChallengeRequest{
92 Identity: l.identity,
@@ -89,6 +94,7 @@ func (l *listener) registerLease(ctx context.Context, ttl time.Duration, udpEnab
94 TTL: int(ttl / time.Second),
95 UDPEnabled: udpEnabled,
96 TCPEnabled: tcpEnabled,
97 + MultiHop: multiHop,
98 }, nil, &challenge); err != nil {
99 return types.RegisterResponse{}, err
100 }
@@ -110,6 +116,47 @@ func (l *listener) registerLease(ctx context.Context, ttl time.Duration, udpEnab
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 + }
134 + }
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)
156 + }
157 + return path, nil
158 +}
159 +
160 func (l *listener) renewRegisteredLease(ctx context.Context, ttl time.Duration, accessToken string) (types.RenewResponse, error) {
161 var resp types.RenewResponse
162 if err := utils.HTTPDoAPIPath(ctx, l.httpClient, l.relayURL, http.MethodPost, types.PathSDKRenew, types.RenewRequest{
sdk/expose.go
+67 -30
@@ -31,6 +31,7 @@ type Exposure struct {
31 UDPAddr string
32 udpEnabled bool
33 tcpEnabled bool
34 + multiHop []string
35 banMITM bool
36 maxActiveRelays int
37 metadata types.LeaseMetadata
@@ -47,15 +48,18 @@ type Exposure struct {
48 }
49
50 type ExposeConfig struct {
50 - RelayURLs []string
51 - Discovery bool
52 - IdentityPath string
53 - IdentityJSON string
54 - Name string
55 - TargetAddr string
56 - UDPAddr string
57 - UDPEnabled bool
58 - TCPEnabled bool
51 + RelayURLs []string
52 + Discovery bool
53 + IdentityPath string
54 + IdentityJSON string
55 + Name string
56 + TargetAddr string
57 + UDPAddr string
58 + UDPEnabled bool
59 + TCPEnabled bool
60 + // MultiHop is the caller-selected ordered relay URL path. The first URL is
61 + // the public entry relay and the last URL is the exit relay the SDK registers with.
62 + MultiHop []string
63 BanMITM bool
64 MaxActiveRelays int
65 Metadata types.LeaseMetadata
@@ -68,9 +72,27 @@ func Expose(ctx context.Context, cfg ExposeConfig) (*Exposure, error) {
72 if err != nil {
73 return nil, err
74 }
71 - relayURLs, err := utils.ResolvePortalRelayURLs(ctx, explicitRelayURLs, cfg.Discovery)
75 + multiHop, err := utils.NormalizeRelayURLs(cfg.MultiHop...)
76 if err != nil {
73 - return nil, err
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")
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)
87 + if len(multiHop) > 0 {
88 + listenerRelayURLs = []string{multiHop[len(multiHop)-1]}
89 + relaySetURLs = append([]string(nil), multiHop...)
90 + } else {
91 + listenerRelayURLs, err = utils.ResolvePortalRelayURLs(ctx, explicitRelayURLs, cfg.Discovery)
92 + if err != nil {
93 + return nil, err
94 + }
95 + relaySetURLs = listenerRelayURLs
96 }
97
98 identity, createdIdentity, err := utils.ResolveListenerIdentity(
@@ -104,21 +126,30 @@ func Expose(ctx context.Context, cfg ExposeConfig) (*Exposure, error) {
126 cancel: cancel,
127 done: exposureCtx.Done(),
128 identity: identity,
107 - explicitRelays: append([]string(nil), explicitRelayURLs...),
129 + explicitRelays: explicitRelayURLs,
130 TargetAddr: targetAddr,
131 UDPAddr: udpAddr,
132 udpEnabled: cfg.UDPEnabled,
133 tcpEnabled: cfg.TCPEnabled,
134 + multiHop: multiHop,
135 banMITM: cfg.BanMITM,
136 maxActiveRelays: cfg.MaxActiveRelays,
137 metadata: cfg.Metadata,
115 - accepted: make(chan net.Conn, max(len(relayURLs)*defaultReadyTarget*2, 1)),
116 - datagrams: make(chan types.DatagramFrame, max(len(relayURLs)*32, 1)),
117 - relaySet: discovery.NewRelaySet(relayURLs),
118 - relayListeners: make(map[string]*listener, len(relayURLs)),
138 + accepted: make(chan net.Conn, max(len(listenerRelayURLs)*defaultReadyTarget*2, 1)),
139 + datagrams: make(chan types.DatagramFrame, max(len(listenerRelayURLs)*32, 1)),
140 + relaySet: discovery.NewRelaySet(relaySetURLs),
141 + relayListeners: make(map[string]*listener, len(listenerRelayURLs)),
142 + }
143 +
144 + if len(multiHop) > 0 {
145 + refresher := discovery.NewRefresher(exposure.relaySet, nil)
146 + if err := refresher.Refresh(ctx, nil); err != nil {
147 + _ = exposure.Close()
148 + return nil, fmt.Errorf("discover multi-hop relays: %w", err)
149 + }
150 }
151
121 - if len(relayURLs) > 0 {
152 + if len(listenerRelayURLs) > 0 {
153 if err := exposure.reconcileRelayListeners(true); err != nil {
154 _ = exposure.Close()
155 return nil, err
@@ -366,21 +397,26 @@ func (e *Exposure) runDiscoveryLoop(ctx context.Context) {
397 }
398
399 func (e *Exposure) reconcileRelayListeners(failOnError bool) error {
369 - clientState := discovery.ClientState{
370 - ActiveRelayURLs: e.ActiveRelayURLs(),
371 - ExplicitRelayURLs: append([]string(nil), e.explicitRelays...),
372 - MaxActiveRelays: e.maxActiveRelays,
373 - RequireUDP: e.udpEnabled,
374 - RequireTCP: e.tcpEnabled,
400 + listenerRelayURLs := []string(nil)
401 + var multiHop []string
402 + if len(e.multiHop) > 0 {
403 + listenerRelayURLs = []string{e.multiHop[len(e.multiHop)-1]}
404 + multiHop = append([]string(nil), e.multiHop...)
405 + } else {
406 + listenerRelayURLs = e.relaySet.PriorityRelays(discovery.ClientState{
407 + ActiveRelayURLs: e.ActiveRelayURLs(),
408 + ExplicitRelayURLs: append([]string(nil), e.explicitRelays...),
409 + MaxActiveRelays: e.maxActiveRelays,
410 + RequireUDP: e.udpEnabled,
411 + RequireTCP: e.tcpEnabled,
412 + })
413 }
414
377 - desiredRelayURLs := e.relaySet.PriorityRelays(clientState)
378 -
415 e.listenerMu.Lock()
416 staleRelayListeners := make(map[string]*listener)
417 removedRelayURLs := make([]string, 0)
418 for relayURL, listener := range e.relayListeners {
383 - if slices.Contains(desiredRelayURLs, relayURL) {
419 + if slices.Contains(listenerRelayURLs, relayURL) {
420 continue
421 }
422 staleRelayListeners[relayURL] = listener
@@ -388,8 +424,8 @@ func (e *Exposure) reconcileRelayListeners(failOnError bool) error {
424 delete(e.relayListeners, relayURL)
425 }
426
391 - missingRelayURLs := make([]string, 0, len(desiredRelayURLs))
392 - for _, relayURL := range desiredRelayURLs {
427 + missingRelayURLs := make([]string, 0, len(listenerRelayURLs))
428 + for _, relayURL := range listenerRelayURLs {
429 if _, ok := e.relayListeners[relayURL]; ok {
430 continue
431 }
@@ -411,13 +447,14 @@ func (e *Exposure) reconcileRelayListeners(failOnError bool) error {
447 }
448 for _, relayURL := range missingRelayURLs {
449 retryCount := 10
414 - if slices.Contains(e.explicitRelays, relayURL) {
450 + if len(e.multiHop) > 0 || slices.Contains(e.explicitRelays, relayURL) {
451 retryCount = 0
452 }
453 listener, err := newListener(context.Background(), relayURL, listenerConfig{
454 Identity: e.identity,
455 UDPEnabled: e.udpEnabled,
456 TCPEnabled: e.tcpEnabled,
457 + MultiHop: multiHop,
458 BanMITM: e.banMITM,
459 RetryCount: retryCount,
460 Metadata: e.metadata,
@@ -455,7 +492,7 @@ func (e *Exposure) reconcileRelayListeners(failOnError bool) error {
492 log.Info().
493 Strs("added_relays", addedRelayURLs).
494 Strs("removed_relays", removedRelayURLs).
458 - Strs("desired_relays", desiredRelayURLs).
495 + Strs("listener_relays", listenerRelayURLs).
496 Msg("reconciled relay listeners")
497 }
498 return nil
sdk/listener.go
+45 -31
@@ -40,6 +40,7 @@ type listenerConfig struct {
40 ReadyTarget int
41 RetryCount int
42 RetryWait time.Duration
43 + MultiHop []string
44 relaySet *discovery.RelaySet
45 }
46
@@ -54,6 +55,7 @@ type listener struct {
55 identity types.Identity
56 metadata types.LeaseMetadata
57 relaySet *discovery.RelaySet
58 + multiHop []string
59 udpEnabled bool
60 tcpEnabled bool
61 dialTimeout time.Duration
@@ -105,6 +107,7 @@ func newListener(ctx context.Context, relayURL string, cfg listenerConfig) (*lis
107 identity: cfg.Identity,
108 metadata: cfg.Metadata,
109 relaySet: cfg.relaySet,
110 + multiHop: cfg.MultiHop,
111 udpEnabled: cfg.UDPEnabled,
112 tcpEnabled: cfg.TCPEnabled,
113 dialTimeout: dialTimeout,
@@ -168,7 +171,10 @@ func (l *listener) run(ctx context.Context) {
171 }
172
173 retries = 0
171 - publicURL := l.publicURL()
174 + publicURL := ""
175 + if lease, ok := l.leaseSnapshot(); ok {
176 + publicURL = l.publicURLForLease(lease)
177 + }
178 event := log.Info().Str("address", l.identity.Address)
179 if publicURL != "" {
180 event.Msg("service ready at " + publicURL)
@@ -237,13 +243,14 @@ func (l *listener) Close() error {
243 }
244
245 type listenerLease struct {
240 - hostname string
241 - udpAddr string
242 - accessToken string
243 - expiresAt time.Time
244 - sniPort int
245 - tlsConfig *tls.Config
246 - tlsCloser io.Closer
246 + hostname string
247 + udpAddr string
248 + accessToken string
249 + expiresAt time.Time
250 + sniPort int
251 + publicURLBase *url.URL
252 + tlsConfig *tls.Config
253 + tlsCloser io.Closer
254 }
255
256 func (l *listener) clearLease(reason string) *listenerLease {
@@ -353,33 +360,29 @@ func (l *listener) datagramReady() (string, bool, bool) {
360 return udpAddr, ready, pending
361 }
362
356 -func (l *listener) publicURL() string {
357 - lease, ok := l.leaseSnapshot()
358 - if !ok {
359 - return ""
363 +func (l *listener) publicURLForLease(lease listenerLease) string {
364 + baseURL := lease.publicURLBase
365 + if baseURL == nil {
366 + baseURL = l.relayURL
367 }
361 - return l.publicURLForHostname(lease.hostname)
362 -}
363 -
364 -func (l *listener) publicURLForHostname(hostname string) string {
365 - if l.relayURL == nil {
368 + if baseURL == nil {
369 return ""
370 }
368 - if hostname == "" {
371 + if lease.hostname == "" {
372 return ""
373 }
374
372 - if l.relayURL.Scheme == "" {
373 - return "https://" + hostname
375 + if baseURL.Scheme == "" {
376 + return "https://" + lease.hostname
377 }
378
376 - host := hostname
377 - if port := l.relayURL.Port(); port != "" {
378 - host = net.JoinHostPort(hostname, port)
379 + host := lease.hostname
380 + if port := baseURL.Port(); port != "" {
381 + host = net.JoinHostPort(lease.hostname, port)
382 }
383
384 return (&url.URL{
382 - Scheme: l.relayURL.Scheme,
385 + Scheme: baseURL.Scheme,
386 Host: host,
387 }).String()
388 }
@@ -759,7 +762,17 @@ func (l *listener) registerAndConfigure(ctx context.Context) error {
762 _ = l.unregisterLease(context.Background(), resp.AccessToken)
763 return errors.New("relay did not return sni port for udp transport")
764 }
762 - tlsConf, tenantTLSCloser, err := keyless.BuildClientTLSConfig(l.relayURL.String(), []string{resp.Hostname})
765 + keylessURL := strings.TrimSpace(resp.KeylessURL)
766 + if keylessURL == "" {
767 + keylessURL = l.relayURL.String()
768 + }
769 + publicURLBase := l.relayURL
770 + if normalizedKeylessURL, err := utils.NormalizeRelayURL(keylessURL); err == nil {
771 + if parsedKeylessURL, parseErr := url.Parse(normalizedKeylessURL); parseErr == nil {
772 + publicURLBase = parsedKeylessURL
773 + }
774 + }
775 + tlsConf, tenantTLSCloser, err := keyless.BuildClientTLSConfig(keylessURL, []string{resp.Hostname})
776 if err != nil {
777 _ = l.unregisterLease(context.Background(), resp.AccessToken)
778 return err
@@ -773,12 +786,13 @@ func (l *listener) registerAndConfigure(ctx context.Context) error {
786 return ctx.Err()
787 }
788 next := &listenerLease{
776 - hostname: resp.Hostname,
777 - udpAddr: resp.UDPAddr,
778 - accessToken: resp.AccessToken,
779 - expiresAt: resp.ExpiresAt,
780 - tlsConfig: tlsConf,
781 - tlsCloser: tenantTLSCloser,
789 + hostname: resp.Hostname,
790 + udpAddr: resp.UDPAddr,
791 + accessToken: resp.AccessToken,
792 + expiresAt: resp.ExpiresAt,
793 + publicURLBase: publicURLBase,
794 + tlsConfig: tlsConf,
795 + tlsCloser: tenantTLSCloser,
796 }
797 if l.udpEnabled {
798 next.sniPort = resp.SNIPort
sdk/mitm.go
+1 -1
@@ -86,7 +86,7 @@ func (m *mitmManager) probeTLSPassthrough(ctx context.Context) (MITMProbeReport,
86 return MITMProbeReport{}, errors.New("listener hostname is unavailable")
87 }
88
89 - publicURL := l.publicURLForHostname(lease.hostname)
89 + publicURL := l.publicURLForLease(lease)
90 if publicURL == "" {
91 return MITMProbeReport{}, errors.New("listener is not registered")
92 }
types/api.go
+4
@@ -68,6 +68,9 @@ type RegisterChallengeRequest struct {
68 TTL int `json:"ttl,omitempty"`
69 UDPEnabled bool `json:"udp_enabled,omitempty"`
70 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"`
74 }
75
76 type RegisterChallengeResponse struct {
@@ -81,6 +84,7 @@ type RegisterResponse struct {
84 ExpiresAt time.Time `json:"expires_at"`
85 Hostname string `json:"hostname"`
86 AccessToken string `json:"access_token"`
87 + KeylessURL string `json:"keyless_url,omitempty"`
88 SNIPort int `json:"sni_port,omitempty"`
89 UDPAddr string `json:"udp_addr,omitempty"`
90 UDPEnabled bool `json:"udp_enabled,omitempty"`
types/identity.go
+19 -31
@@ -125,36 +125,6 @@ type RelayDescriptor struct {
125 Signature string `json:"signature,omitempty"`
126 }
127
128 -// canonicalRelayDescriptor mirrors the subset of RelayDescriptor fields that
129 -// participate in the cryptographic signature. Only fields that uniquely
130 -// identify the relay or affect routing are signed; mutable telemetry (Load,
131 -// LoadScore, LastUpdated) and the Signature itself are deliberately excluded
132 -// so that observers may update telemetry without invalidating the signature.
133 -//
134 -// All slice fields are normalized to non-nil to keep encoding deterministic
135 -// (json.Marshal encodes nil slices as `null` and empty slices as `[]`). Time
136 -// fields are encoded as Unix nanoseconds to avoid any RFC3339 round-trip
137 -// ambiguity.
138 -type canonicalRelayDescriptor struct {
139 - Name string `json:"name"`
140 - Address string `json:"address"`
141 - RelayID string `json:"relay_id"`
142 - OwnerAddress string `json:"owner_address"`
143 - Version uint32 `json:"version"`
144 - IssuedAtUnixNano int64 `json:"issued_at_unix_nano"`
145 - ExpiresAtUnixNano int64 `json:"expires_at_unix_nano"`
146 - APIHTTPSAddr string `json:"api_https_addr"`
147 - IngressTLSAddr string `json:"ingress_tls_addr"`
148 - WireGuardPublicKey string `json:"wireguard_public_key"`
149 - WireGuardEndpoint string `json:"wireguard_endpoint"`
150 - OverlayIPv4 string `json:"overlay_ipv4"`
151 - OverlayCIDRs []string `json:"overlay_cidrs"`
152 - Discovery bool `json:"discovery"`
153 - SupportsUDP bool `json:"supports_udp"`
154 - SupportsTCP bool `json:"supports_tcp"`
155 - SupportsOverlayPeer bool `json:"supports_overlay_peer"`
156 -}
157 -
128 // CanonicalBytes returns the deterministic byte representation of a relay
129 // descriptor used for signing and signature verification. Two descriptors
130 // that differ only in mutable telemetry fields produce identical bytes.
@@ -166,7 +136,25 @@ func CanonicalBytes(desc RelayDescriptor) ([]byte, error) {
136 if overlayCIDRs == nil {
137 overlayCIDRs = []string{}
138 }
169 - canonical := canonicalRelayDescriptor{
139 + canonical := struct {
140 + Name string `json:"name"`
141 + Address string `json:"address"`
142 + RelayID string `json:"relay_id"`
143 + OwnerAddress string `json:"owner_address"`
144 + Version uint32 `json:"version"`
145 + IssuedAtUnixNano int64 `json:"issued_at_unix_nano"`
146 + ExpiresAtUnixNano int64 `json:"expires_at_unix_nano"`
147 + APIHTTPSAddr string `json:"api_https_addr"`
148 + IngressTLSAddr string `json:"ingress_tls_addr"`
149 + WireGuardPublicKey string `json:"wireguard_public_key"`
150 + WireGuardEndpoint string `json:"wireguard_endpoint"`
151 + OverlayIPv4 string `json:"overlay_ipv4"`
152 + OverlayCIDRs []string `json:"overlay_cidrs"`
153 + Discovery bool `json:"discovery"`
154 + SupportsUDP bool `json:"supports_udp"`
155 + SupportsTCP bool `json:"supports_tcp"`
156 + SupportsOverlayPeer bool `json:"supports_overlay_peer"`
157 + }{
158 Name: desc.Name,
159 Address: desc.Address,
160 RelayID: desc.RelayID,