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

init discovery

rabbitprincess committed Mar 20, 2026 at 22:26 UTC 1dbaf20c2b96223e8f389317645c1c548cd6c270

25 files changed +1652 -429

cmd/demo-udp/main.go

+3 -2

@@ -108,7 +108,7 @@ func runDemoUDP() error {

108

109 func runUDPEchoLoop(ctx context.Context, exposure *sdk.Exposure, logger zerolog.Logger) {

110 for {

111 - frame, _, _, _, reply, err := exposure.AcceptDatagram()

111 + frame, err := exposure.AcceptDatagram()

112 if err != nil {

113 if ctx.Err() != nil || errors.Is(err, net.ErrClosed) {

114 return

@@ -121,7 +121,8 @@ func runUDPEchoLoop(ctx context.Context, exposure *sdk.Exposure, logger zerolog.

121 if len(payload) == 0 {

122 payload = []byte("pong")

123 }

124 - if err := reply(payload); err != nil && ctx.Err() == nil && !errors.Is(err, net.ErrClosed) {

124 + frame.Payload = payload

125 + if err := exposure.SendDatagram(frame); err != nil && ctx.Err() == nil && !errors.Is(err, net.ErrClosed) {

126 logger.Warn().Err(err).Uint32("flow_id", frame.FlowID).Msg("demo udp reply failed")

127 return

128 }

cmd/portal-tunnel/config.go

+3 -2

@@ -16,8 +16,9 @@ const (

16 )

18 type cliConfig struct {

19 - ClientID string `json:"client_id,omitempty"`

20 - Relays []string `json:"relays,omitempty"`

19 + ClientID string `json:"client_id,omitempty"`

20 + OwnerPrivateKey string `json:"owner_private_key,omitempty"`

21 + Relays []string `json:"relays,omitempty"`

22 }

24 func loadCLIConfig() (cliConfig, string, error) {

cmd/portal-tunnel/main.go

+55 -24

@@ -69,16 +69,18 @@ func runExposeCommand(args []string) error {

69 fs.SetOutput(io.Discard)

71 var (

72 - relayCSV string

73 - target string

74 - udpAddr string

75 - name string

76 - desc string

77 - tags string

78 - thumbnail string

79 - owner string

80 - hide bool

81 - udp bool

72 + relayCSV string

73 + target string

74 + udpAddr string

75 + name string

76 + desc string

77 + tags string

78 + thumbnail string

79 + owner string

80 + privateKey string

81 + discoveryEnabled bool

82 + hide bool

83 + udp bool

84 )

85 fs.StringVar(&relayCSV, "relays", "", "Additional Portal relay server API URLs (comma-separated; scheme omitted defaults to https)")

86 fs.BoolVar(&defaultRelays, "default-relays", defaultRelays, "Include public registry relays")

@@ -87,6 +89,8 @@ func runExposeCommand(args []string) error {

89 fs.StringVar(&tags, "tags", "", "Service tags metadata (comma-separated)")

90 fs.StringVar(&thumbnail, "thumbnail", "", "Service thumbnail URL metadata")

91 fs.StringVar(&owner, "owner", "", "Service owner metadata")

92 + fs.StringVar(&privateKey, "private-key", "", "Owner private key used to derive a discovery address")

93 + fs.BoolVar(&discoveryEnabled, "discovery", false, "Advertise known relay URLs and discover additional relay bootstraps")

94 fs.BoolVar(&hide, "hide", false, "Hide service from discovery")

96 fs.BoolVar(&udp, "udp", utils.ParseBoolEnv("UDP_ENABLED", false), "Enable public UDP relay in addition to the default TCP relay")

@@ -172,21 +176,53 @@ func runExposeCommand(args []string) error {

176 return errors.New("no relay URLs configured; run the installer first or pass --relays")

177 }

178

179 + previousOwnerPrivateKey := cfg.OwnerPrivateKey

180 + if strings.TrimSpace(privateKey) != "" {

181 + cfg.OwnerPrivateKey = privateKey

182 + }

183 + var ownerPrivateKey *string

184 + if discoveryEnabled || strings.TrimSpace(cfg.OwnerPrivateKey) != "" {

185 + ownerPrivateKey = &cfg.OwnerPrivateKey

186 + }

187 +

188 + exposure, err := sdk.ExposeWithConfig(ctx, sdk.ExposeConfig{

189 + RelayURLs: relayURLs,

190 + Name: name,

191 + UDPEnabled: udp,

192 + Discovery: discoveryEnabled,

193 + Metadata: types.LeaseMetadata{

194 + Description: desc,

195 + Tags: utils.SplitCSV(tags),

196 + Owner: owner,

197 + Thumbnail: thumbnail,

198 + Hide: hide,

199 + },

200 + OwnerPrivateKey: ownerPrivateKey,

201 + })

202 + if err != nil {

203 + return fmt.Errorf("service %s: failed to start relays: %w", name, err)

204 + }

205 + if exposure == nil {

206 + return errors.New("no relay URLs provided")

207 + }

208 + if identity := exposure.OwnerIdentity(); identity.PrivateKey != "" {

209 + cfg.OwnerPrivateKey = identity.PrivateKey

210 + }

211 + if cfg.OwnerPrivateKey != previousOwnerPrivateKey {

212 + if saveErr := saveCLIConfig(cfgPath, cfg); saveErr != nil {

213 + _ = exposure.Close()

214 + return fmt.Errorf("persist portal config: %w", saveErr)

215 + }

216 + }

217 +

218 return runTunnel(

219 ctx,

220 stop,

178 - relayURLs,

221 + exposure,

222 target,

223 udpAddr,

224 name,

225 udp,

183 - types.LeaseMetadata{

184 - Description: desc,

185 - Tags: utils.SplitCSV(tags),

186 - Owner: owner,

187 - Thumbnail: thumbnail,

188 - Hide: hide,

189 - },

226 )

227 }

228

@@ -242,18 +278,13 @@ func runListCommand(args []string) error {

278 func runTunnel(

279 ctx context.Context,

280 stop func(),

245 - relayURLs []string,

281 + exposure *sdk.Exposure,

282 tcpTarget string,

283 udpTarget string,

284 name string,

285 udpEnabled bool,

250 - metadata types.LeaseMetadata,

286 ) error {

287 logger := log.With().Str("component", "portal").Logger()

253 - exposure, err := sdk.Expose(ctx, relayURLs, name, udpEnabled, metadata)

254 - if err != nil {

255 - return fmt.Errorf("service %s: failed to start relays: %w", name, err)

256 - }

288 if exposure == nil {

289 return errors.New("no relay URLs provided")

290 }

cmd/portal-tunnel/relays.go

+33 -23

@@ -13,6 +13,7 @@ import (

13 "github.com/rs/zerolog/log"

15 "github.com/gosuda/portal/v2/sdk"

16 + "github.com/gosuda/portal/v2/types"

17 "github.com/gosuda/portal/v2/utils"

18 )

@@ -185,7 +186,7 @@ func proxyExposureDatagrams(ctx context.Context, exposure *sdk.Exposure, localAd

186 type flowEntry struct {

187 conn *net.UDPConn

188 lastSeen time.Time

188 - reply func([]byte) error

189 + frame types.DatagramFrame

190 }

191

192 var mu sync.Mutex

@@ -212,11 +213,16 @@ func proxyExposureDatagrams(ctx context.Context, exposure *sdk.Exposure, localAd

213 }

214 }()

215

215 - getOrCreateFlow := func(key flowKey, reply func([]byte) error) (*net.UDPConn, error) {

216 + getOrCreateFlow := func(frame types.DatagramFrame) (*net.UDPConn, error) {

217 + key := flowKey{

218 + flowID: frame.FlowID,

219 + leaseID: frame.LeaseID,

220 + relayURL: frame.RelayURL,

221 + }

222 +

223 mu.Lock()

224 if f, ok := flows[key]; ok {

225 f.lastSeen = time.Now()

219 - f.reply = reply

226 mu.Unlock()

227 return f.conn, nil

228 }

@@ -232,10 +238,18 @@ func proxyExposureDatagrams(ctx context.Context, exposure *sdk.Exposure, localAd

238 mu.Unlock()

239 _ = localConn.Close()

240 f.lastSeen = time.Now()

235 - f.reply = reply

241 return f.conn, nil

242 }

238 - flows[key] = &flowEntry{conn: localConn, lastSeen: time.Now(), reply: reply}

243 + flows[key] = &flowEntry{

244 + conn: localConn,

245 + lastSeen: time.Now(),

246 + frame: types.DatagramFrame{

247 + FlowID: frame.FlowID,

248 + LeaseID: frame.LeaseID,

249 + RelayURL: frame.RelayURL,

250 + UDPAddr: frame.UDPAddr,

251 + },

252 + }

253 mu.Unlock()

254

255 go func() {

@@ -260,13 +274,14 @@ func proxyExposureDatagrams(ctx context.Context, exposure *sdk.Exposure, localAd

274 if entry != nil {

275 entry.lastSeen = time.Now()

276 }

263 - replyFn := func([]byte) error { return net.ErrClosed }

264 - if entry != nil && entry.reply != nil {

265 - replyFn = entry.reply

277 + replyFrame := types.DatagramFrame{}

278 + if entry != nil {

279 + replyFrame = entry.frame

280 + replyFrame.Payload = append([]byte(nil), buf[:n]...)

281 }

282 mu.Unlock()

283

269 - if sendErr := replyFn(buf[:n]); sendErr != nil {

284 + if sendErr := exposure.SendDatagram(replyFrame); sendErr != nil {

285 logger.Debug().

286 Err(sendErr).

287 Uint32("flow_id", key.flowID).

@@ -283,7 +298,7 @@ func proxyExposureDatagrams(ctx context.Context, exposure *sdk.Exposure, localAd

298

299 logger.Info().Str("target", targetAddr).Msg("udp proxy loop started, waiting for datagrams")

300 for {

286 - frame, leaseID, relayURL, udpAddr, reply, err := exposure.AcceptDatagram()

301 + frame, err := exposure.AcceptDatagram()

302 if err != nil {

303 if ctx.Err() != nil {

304 return ctx.Err()

@@ -297,24 +312,19 @@ func proxyExposureDatagrams(ctx context.Context, exposure *sdk.Exposure, localAd

312 logger.Debug().

313 Uint32("flow_id", frame.FlowID).

314 Int("bytes", len(frame.Payload)).

300 - Str("lease_id", leaseID).

301 - Str("relay_url", relayURL).

302 - Str("udp_addr", udpAddr).

315 + Str("lease_id", frame.LeaseID).

316 + Str("relay_url", frame.RelayURL).

317 + Str("udp_addr", frame.UDPAddr).

318 Str("target", targetAddr).

319 Msg("datagram received from relay, forwarding to local")

320

306 - key := flowKey{

307 - flowID: frame.FlowID,

308 - leaseID: leaseID,

309 - relayURL: relayURL,

310 - }

311 - localConn, err := getOrCreateFlow(key, reply)

321 + localConn, err := getOrCreateFlow(frame)

322 if err != nil {

323 logger.Warn().

324 Err(err).

325 Uint32("flow_id", frame.FlowID).

316 - Str("lease_id", leaseID).

317 - Str("relay_url", relayURL).

326 + Str("lease_id", frame.LeaseID).

327 + Str("relay_url", frame.RelayURL).

328 Msg("dial local udp failed")

329 continue

330 }

@@ -323,8 +333,8 @@ func proxyExposureDatagrams(ctx context.Context, exposure *sdk.Exposure, localAd

333 logger.Warn().

334 Err(err).

335 Uint32("flow_id", frame.FlowID).

326 - Str("lease_id", leaseID).

327 - Str("relay_url", relayURL).

336 + Str("lease_id", frame.LeaseID).

337 + Str("relay_url", frame.RelayURL).

338 Msg("write to local udp failed")

339 }

340 }

cmd/relay-server/main.go

+63 -1

@@ -2,6 +2,7 @@ package main

3 import (

4 "context"

5 + "errors"

6 "flag"

7 "fmt"

8 "os"

@@ -30,10 +31,13 @@ const (

32 type relayServerConfig struct {

33 PortalURL string

34 + OwnerPrivateKey string

35 + Bootstraps string

36 APIPort int

37 SNIPort int

38 UDPPortCount int

39 AdminSecretKey string

40 + DiscoveryEnabled bool

41 TrustProxyHeaders bool

42 TrustedProxyCIDRs string

43 KeylessDir string

@@ -59,7 +63,10 @@ func main() {

63 apiPort := parsePortNumber(os.Getenv("API_PORT"), defaultAPIPort)

64 sniPort := parsePortNumber(os.Getenv("SNI_PORT"), defaultSNIPort)

65 udpPortCount := parseNonNegativeInt(os.Getenv("UDP_PORT_COUNT"), defaultUDPPortCount)

66 + ownerPrivateKey := trimmedEnv("OWNER_PRIVATE_KEY")

67 + bootstraps := trimmedEnv("BOOTSTRAPS")

68 adminSecretKey := trimmedEnv("ADMIN_SECRET_KEY")

69 + discoveryEnabled := utils.ParseBoolEnv("DISCOVERY_ENABLED", false)

70 trustProxyHeaders := utils.ParseBoolEnv("TRUST_PROXY_HEADERS", false)

71 trustedProxyCIDRs := trimmedEnv("TRUSTED_PROXY_CIDRS")

72 keylessDir := trimmedEnv("KEYLESS_DIR")

@@ -86,7 +93,10 @@ func main() {

93 flag.IntVar(&cfg.SNIPort, "sni-port", sniPort, "TCP SNI router port number (env: SNI_PORT)")

94 flag.IntVar(&cfg.UDPPortCount, "udp-port-count", udpPortCount, "Number of UDP ports to allocate for leases, starting at port 50000 (0=disabled) (env: UDP_PORT_COUNT)")

96 + flag.StringVar(&cfg.OwnerPrivateKey, "owner-private-key", ownerPrivateKey, "relay owner private key used to derive a discovery address (env: OWNER_PRIVATE_KEY)")

97 + flag.StringVar(&cfg.Bootstraps, "bootstraps", bootstraps, "additional bootstrap relay API URLs used for discovery expansion (env: BOOTSTRAPS)")

98 flag.StringVar(&cfg.AdminSecretKey, "admin-secret-key", adminSecretKey, "admin auth secret (env: ADMIN_SECRET_KEY)")

99 + flag.BoolVar(&cfg.DiscoveryEnabled, "discovery", discoveryEnabled, "serve relay discovery endpoints and poll discovery peers (env: DISCOVERY_ENABLED)")

100 flag.BoolVar(&cfg.TrustProxyHeaders, "trust-proxy-headers", trustProxyHeaders, "trust X-Forwarded-* and X-Real-IP headers from trusted proxies (env: TRUST_PROXY_HEADERS)")

101 flag.StringVar(&cfg.TrustedProxyCIDRs, "trusted-proxy-cidrs", trustedProxyCIDRs, "trusted proxy CIDR allowlist for forwarded headers, comma-separated; defaults to private/loopback proxy ranges when trust-proxy-headers is enabled (env: TRUSTED_PROXY_CIDRS)")

102

@@ -103,6 +113,7 @@ func main() {

113 logger.Info().

114 Str("release_version", types.ReleaseVersion).

115 Str("portal_url", cfg.PortalURL).

116 + Bool("discovery_enabled", cfg.DiscoveryEnabled).

117 Bool("udp_enabled", cfg.UDPPortCount > 0).

118 Msg("configured relay server")

119

@@ -124,8 +135,21 @@ func runServer(cfg relayServerConfig) error {

135 if err != nil {

136 return fmt.Errorf("parse trusted proxy cidrs: %w", err)

137 }

138 + bootstraps, err := utils.NormalizeRelayURLs(utils.SplitCSV(cfg.Bootstraps))

139 + if err != nil {

140 + return fmt.Errorf("normalize bootstraps: %w", err)

141 + }

142 + if strings.TrimSpace(cfg.OwnerPrivateKey) == "" {

143 + cfg.OwnerPrivateKey, err = loadOwnerPrivateKey(cfg.KeylessDir)

144 + if err != nil {

145 + return fmt.Errorf("load relay owner private key: %w", err)

146 + }

147 + }

148 + previousOwnerPrivateKey := cfg.OwnerPrivateKey

149 server, err := portal.NewServer(portal.ServerConfig{

128 - PortalURL: cfg.PortalURL,

150 + PortalURL: cfg.PortalURL,

151 + OwnerPrivateKey: cfg.OwnerPrivateKey,

152 + Bootstraps: bootstraps,

153 ACME: acme.Config{

154 KeyDir: cfg.KeylessDir,

155 DNSProvider: cfg.ACMEDNSProvider,

@@ -140,11 +164,20 @@ func runServer(cfg relayServerConfig) error {

164 SNIListenAddr: sniListenAddr,

165 TrustedProxyCIDRs: trustedProxyCIDRs,

166 TrustProxyHeaders: cfg.TrustProxyHeaders,

167 + DiscoveryEnabled: cfg.DiscoveryEnabled,

168 UDPPortCount: cfg.UDPPortCount,

169 })

170 if err != nil {

171 return fmt.Errorf("create relay server: %w", err)

172 }

173 + if identity := server.OwnerIdentity(); identity.PrivateKey != "" {

174 + cfg.OwnerPrivateKey = identity.PrivateKey

175 + }

176 + if cfg.OwnerPrivateKey != previousOwnerPrivateKey {

177 + if err := saveOwnerPrivateKey(cfg.KeylessDir, cfg.OwnerPrivateKey); err != nil {

178 + return fmt.Errorf("persist relay owner private key: %w", err)

179 + }

180 + }

181

182 frontend, err := NewFrontend(cfg.PortalURL, server, cfg.AdminSecretKey, trustedProxyCIDRs, cfg.TrustProxyHeaders)

183 if err != nil {

@@ -160,6 +193,7 @@ func runServer(cfg relayServerConfig) error {

193 Str("sni_addr", server.SNIAddr()).

194 Str("root_host", rootHost).

195 Str("acme_dns_provider", cfg.ACMEDNSProvider).

196 + Bool("discovery_enabled", server.DiscoveryEnabled()).

197 Bool("udp_enabled", cfg.UDPPortCount > 0).

198 Bool("acme_enabled", !strings.HasSuffix(rootHost, "localhost") && rootHost != "127.0.0.1" && rootHost != "::1")

199 if quicAddr := server.QUICTunnelAddr(); quicAddr != "" {

@@ -197,3 +231,31 @@ func parseNonNegativeInt(raw string, fallback int) int {

231 }

232 return v

233 }

234 +

235 +func ownerPrivateKeyPath(keylessDir string) string {

236 + return filepath.Join(strings.TrimSpace(keylessDir), "owner_private_key.hex")

237 +}

238 +

239 +func loadOwnerPrivateKey(keylessDir string) (string, error) {

240 + keyPath := ownerPrivateKeyPath(keylessDir)

241 + data, err := os.ReadFile(keyPath)

242 + if err != nil {

243 + if errors.Is(err, os.ErrNotExist) {

244 + return "", nil

245 + }

246 + return "", err

247 + }

248 + return strings.TrimSpace(string(data)), nil

249 +}

250 +

251 +func saveOwnerPrivateKey(keylessDir, privateKey string) error {

252 + privateKey = strings.TrimSpace(privateKey)

253 + if privateKey == "" {

254 + return nil

255 + }

256 + keyPath := ownerPrivateKeyPath(keylessDir)

257 + if err := os.MkdirAll(filepath.Dir(keyPath), 0o700); err != nil {

258 + return err

259 + }

260 + return os.WriteFile(keyPath, []byte(privateKey+"\n"), 0o600)

261 +}

docs/architecture.md

+5 -5

@@ -129,8 +129,8 @@ That distinction matters because `/sdk/connect` stops being ordinary HTTP once h

129 - `Listener` embeds a `datagram.Session` for QUIC datagram transport (no separate UDP listener type)

130 - `Listener.AcceptDatagram()` / `SendDatagram()`: read/write datagram frames via the session

131 - `Listener.WaitDatagramReady()`: blocks until relay publishes `udp_addr` and `quic_addr`

132 -- `ExposureDatagram`: wraps a `DatagramFrame` with relay context (FlowID, LeaseID, RelayURL, UDPAddr) and a `Reply()` callback for bidirectional flow

133 -- `Exposure.AcceptDatagram()`: receives datagrams from all backing relay listeners

132 +- `Exposure.AcceptDatagram()`: receives datagrams from all backing relay listeners with relay context populated on `DatagramFrame`

133 +- `Exposure.SendDatagram()`: sends a datagram frame back through the owning relay listener

134 - `Exposure.WaitDatagramReady()`: blocks until at least one relay's datagram plane is ready

135

136 ### Tunnel (`cmd/portal-tunnel`)

@@ -144,7 +144,7 @@ That distinction matters because `/sdk/connect` stops being ordinary HTTP once h

144 - `--udp` flag (bool, default `false`): enables UDP relay in addition to TCP

145 - `--udp-addr` flag (string): local UDP target address (`host:port` or port only); required when `--udp` is enabled

146 - `runUDPBestEffort`: waits for datagram readiness, then calls `proxyExposureDatagrams`

147 -- `proxyExposureDatagrams` (`relays.go`): per-flow UDP sockets to local target with idle cleanup; uses `ExposureDatagram.Reply()` for return path

147 +- `proxyExposureDatagrams` (`relays.go`): per-flow UDP sockets to local target with idle cleanup; uses `Exposure.SendDatagram()` for the return path

148 - Best-effort UDP — failures logged but do not terminate the TCP tunnel

149

150 ## Transport Model

@@ -171,7 +171,7 @@ Result: the relay decides routing, but tenant TLS termination still happens at t

171 5. Authentication: SDK sends `{lease_id, reverse_token}` JSON on the first QUIC stream; relay validates and calls `FlowMux.Register(conn)`.

172 6. External UDP client sends a packet to `udp_addr` → `Relay.readLoop` → `FlowMux.TouchFlow` (assigns flow ID) → `FlowMux.SendDatagram` → QUIC DATAGRAM frame.

173 7. SDK-side `Session.receiveLoop` decodes frame → `Listener.AcceptDatagram()` → `Exposure.AcceptDatagram()` → `proxyExposureDatagrams` → local UDP target.

174 -8. Return path: local response → per-flow read goroutine → `ExposureDatagram.Reply()` → `Session.Send` → QUIC DATAGRAM → `FlowMux.runDispatchLoop` → reply callback → `conn.WriteToUDP` to original client.

174 +8. Return path: local response → per-flow read goroutine → `Exposure.SendDatagram()` → `Session.Send` → QUIC DATAGRAM → `FlowMux.runDispatchLoop` → reply callback → `conn.WriteToUDP` to original client.

175

176 ```text

177 Client --UDP--> [:50000+ Relay] --DATAGRAM--> [FlowMux/Session] --QUIC--> [Session/Listener] --UDP--> Local Service

@@ -273,7 +273,7 @@ Cross-package public contract lives in:

273 - shared `/sdk/*`, admin, health, install, and signer paths

274 - `types/transport.go`

275 - `LeaseCapabilities` (Stream/Datagram booleans)

276 - - `DatagramFrame` wire format

276 + - `DatagramFrame` wire frame plus SDK relay context

277 - `EncodeDatagram` / `DecodeDatagram`

278 - Transport constants: `TransportTCP`, `TransportUDP`, `TransportBoth`

279

go.mod

+2 -1

@@ -7,10 +7,12 @@ require (

7 github.com/aws/aws-sdk-go-v2/config v1.32.8

8 github.com/aws/aws-sdk-go-v2/credentials v1.19.8

9 github.com/aws/aws-sdk-go-v2/service/route53 v1.62.1

10 + github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1

11 github.com/go-acme/lego/v4 v4.32.0

12 github.com/gosuda/keyless_tls v0.0.1-0.20260304212324-7733f8366abc

13 github.com/quic-go/quic-go v0.59.0

14 github.com/rs/zerolog v1.34.0

15 + golang.org/x/crypto v0.48.0

16 golang.org/x/net v0.50.0

17 golang.org/x/sync v0.19.0

18 )

@@ -32,7 +34,6 @@ require (

34 github.com/mattn/go-colorable v0.1.13 // indirect

35 github.com/mattn/go-isatty v0.0.20 // indirect

36 github.com/miekg/dns v1.1.72 // indirect

35 - golang.org/x/crypto v0.48.0 // indirect

37 golang.org/x/mod v0.32.0 // indirect

38 golang.org/x/sys v0.41.0 // indirect

39 golang.org/x/text v0.34.0 // indirect

go.sum

@@ -33,6 +33,9 @@ github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F9

33 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=

34 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=

35 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

36 +github.com/decred/dcrd/crypto/blake256 v1.0.0/go.mod h1:sQl2p6Y26YV+ZOcSTP6thNdn47hh8kt6rqSlvmrXFAc=

37 +github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 h1:YLtO71vCjJRCBcrPMtQ9nqBsqpA1m5sE92cU+pd5Mcc=

38 +github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1/go.mod h1:hyedUtir6IdtD/7lIxGeCxkaw7y45JueMRL4DIyJDKs=

39 github.com/go-acme/lego/v4 v4.32.0 h1:z7Ss7aa1noabhKj+DBzhNCO2SM96xhE3b0ucVW3x8Tc=

40 github.com/go-acme/lego/v4 v4.32.0/go.mod h1:lI2fZNdgeM/ymf9xQ9YKbgZm6MeDuf91UrohMQE4DhI=

41 github.com/go-jose/go-jose/v4 v4.1.3 h1:CVLmWDhDVRa6Mi/IgCgaopNosCaHz7zrMeF9MlZRkrs=

portal/api_server.go

+100 -9

@@ -15,6 +15,7 @@ import (

15 "github.com/quic-go/quic-go"

16 "github.com/rs/zerolog/log"

18 + "github.com/gosuda/portal/v2/portal/discovery"

19 "github.com/gosuda/portal/v2/portal/keyless"

20 "github.com/gosuda/portal/v2/portal/policy"

21 "github.com/gosuda/portal/v2/portal/transport"

@@ -74,6 +75,12 @@ func (s *Server) apiHandler(base *http.ServeMux, keylessSignerHandler http.Handl

75 s.handleUnregister(w, r)

76 case types.PathSDKConnect:

77 s.handleConnect(w, r)

78 + case types.PathDiscovery:

79 + if !s.DiscoveryEnabled() {

80 + base.ServeHTTP(w, r)

81 + return

82 + }

83 + s.discovery.ServeHTTP(w, r)

84 case types.PathV1Sign:

85 if keylessSignerHandler == nil {

86 http.NotFound(w, r)

@@ -97,6 +104,57 @@ func (s *Server) handleHealthz(w http.ResponseWriter, _ *http.Request) {

104 utils.WriteAPIData(w, http.StatusOK, map[string]any{"status": "ok"})

105 }

106

107 +func (s *Server) discover(_ context.Context, req types.DiscoverRequest) (types.DiscoverResponse, error) {

108 + resp := types.DiscoverResponse{

109 + OwnerAddress: s.ownerIdentity.Address,

110 + }

111 + if req.RootHost != "" && req.RootHost != s.rootHost {

112 + return resp, nil

113 + }

114 + if req.Name == "" {

115 + return resp, nil

116 + }

117 +

118 + hostname, err := utils.LeaseHostname(req.Name, s.rootHost)

119 + if err != nil {

120 + return types.DiscoverResponse{}, err

121 + }

122 +

123 + now := time.Now()

124 + var lease types.Lease

125 + ok := false

126 +

127 + s.registry.mu.RLock()

128 + if leaseID, found := s.registry.routes.LookupExact(hostname); found {

129 + record, found := s.registry.leaseByID[leaseID]

130 + if found && record != nil && !now.After(record.ExpiresAt) && !record.Metadata.Hide && s.registry.policy.IsLeaseRoutable(record.ID) {

131 + lease = record.Lease

132 + lease.Bootstraps = append([]string(nil), lease.Bootstraps...)

133 + lease.Metadata = lease.Metadata.Copy()

134 + ok = true

135 + }

136 + }

137 + s.registry.mu.RUnlock()

138 +

139 + if !ok {

140 + return resp, nil

141 + }

142 +

143 + ownerAddress := lease.OwnerAddress

144 + if strings.TrimSpace(ownerAddress) == "" {

145 + ownerAddress = s.ownerIdentity.Address

146 + }

147 +

148 + return types.DiscoverResponse{

149 + Found: true,

150 + Name: lease.Name,

151 + Hostname: lease.Hostname,

152 + ExpiresAt: lease.ExpiresAt,

153 + OwnerAddress: ownerAddress,

154 + Bootstraps: lease.Bootstraps,

155 + }, nil

156 +}

157 +

158 func (s *Server) handleDomain(w http.ResponseWriter, r *http.Request) {

159 if r.Method != http.MethodGet {

160 utils.WriteAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")

@@ -393,6 +451,17 @@ func (s *Server) registerLease(req types.RegisterRequest, clientIP string) (type

451 if req.TTL > 0 {

452 ttl = time.Duration(req.TTL) * time.Second

453 }

454 + bootstraps, err := utils.NormalizeRelayURLs(req.Bootstraps)

455 + if err != nil {

456 + return types.RegisterResponse{}, fmt.Errorf("normalize bootstraps: %w", err)

457 + }

458 + ownerAddress := strings.TrimSpace(req.OwnerAddress)

459 + if ownerAddress != "" {

460 + ownerAddress, err = discovery.NormalizeEVMAddress(ownerAddress)

461 + if err != nil {

462 + return types.RegisterResponse{}, fmt.Errorf("normalize owner address: %w", err)

463 + }

464 + }

465

466 if err := s.requireDatagramPlane(req.UDPEnabled); err != nil {

467 return types.RegisterResponse{}, err

@@ -412,15 +481,17 @@ func (s *Server) registerLease(req types.RegisterRequest, clientIP string) (type

481 expiresAt := now.Add(ttl)

482 record := &leaseRecord{

483 Lease: types.Lease{

415 - ID: leaseID,

416 - Name: name,

417 - Hostname: hostname,

418 - Metadata: req.Metadata,

419 - ExpiresAt: expiresAt,

420 - FirstSeenAt: now,

421 - LastSeenAt: now,

422 - ClientIP: clientIP,

423 - UDPEnabled: req.UDPEnabled,

484 + ID: leaseID,

485 + Name: name,

486 + Hostname: hostname,

487 + Bootstraps: append([]string(nil), bootstraps...),

488 + Metadata: req.Metadata,

489 + OwnerAddress: ownerAddress,

490 + ExpiresAt: expiresAt,

491 + FirstSeenAt: now,

492 + LastSeenAt: now,

493 + ClientIP: clientIP,

494 + UDPEnabled: req.UDPEnabled,

495 },

496 ReverseToken: req.ReverseToken,

497 stream: transport.NewRelayStream(leaseID, s.cfg.IdleKeepaliveInterval, s.cfg.ReadyQueueLimit),

@@ -446,12 +517,32 @@ func (s *Server) registerLease(req types.RegisterRequest, clientIP string) (type

517 record.Close()

518 return types.RegisterResponse{}, err

519 }

520 + if s.discovery != nil {

521 + if err := s.discovery.MergeBootstraps(bootstraps); err != nil {

522 + record.Close()

523 + _, _ = s.registry.Unregister(record.ID, record.ReverseToken)

524 + return types.RegisterResponse{}, err

525 + }

526 + }

527 +

528 + responseBootstraps := append([]string(nil), s.cfg.Bootstraps...)

529 + if s.discovery != nil {

530 + responseBootstraps = s.discovery.Bootstraps()

531 + } else {

532 + responseBootstraps, err = utils.NormalizeRelayURLs(append(responseBootstraps, record.Bootstraps...))

533 + if err != nil {

534 + record.Close()

535 + _, _ = s.registry.Unregister(record.ID, record.ReverseToken)

536 + return types.RegisterResponse{}, err

537 + }

538 + }

539

540 resp := types.RegisterResponse{

541 LeaseID: leaseID,

542 Hostname: hostname,

543 Metadata: record.Metadata,

544 ExpiresAt: expiresAt,

545 + Bootstraps: responseBootstraps,

546 UDPEnabled: record.UDPEnabled,

547 }

548 resp.ConnectURL = strings.TrimRight(s.cfg.PortalURL, "/") + types.PathSDKConnect

portal/discovery/discovery.go new

+378

@@ -0,0 +1,378 @@

1 +package discovery

2 +

3 +import (

4 + "context"

5 + "crypto/tls"

6 + "encoding/json"

7 + "errors"

8 + "fmt"

9 + "net/http"

10 + "net/url"

11 + "strings"

12 + "sync"

13 + "time"

14 +

15 + "github.com/rs/zerolog/log"

16 +

17 + "github.com/gosuda/portal/v2/portal/keyless"

18 + "github.com/gosuda/portal/v2/types"

19 + "github.com/gosuda/portal/v2/utils"

20 +)

21 +

22 +const (

23 + defaultRequestTimeout = 15 * time.Second

24 + defaultMaxPeers = 32

25 + defaultPollInterval = 30 * time.Second

26 +)

27 +

28 +type Resolver interface {

29 + Discover(context.Context, types.DiscoverRequest) (types.DiscoverResponse, error)

30 +}

31 +

32 +type ResolverFunc func(context.Context, types.DiscoverRequest) (types.DiscoverResponse, error)

33 +

34 +func (fn ResolverFunc) Discover(ctx context.Context, req types.DiscoverRequest) (types.DiscoverResponse, error) {

35 + return fn(ctx, req)

36 +}

37 +

38 +type Config struct {

39 + SelfURLs []string

40 + Bootstraps []string

41 + RootCAPEM []byte

42 + RequestTimeout time.Duration

43 + MaxPeers int

44 + OnBootstraps func([]string)

45 +}

46 +

47 +type Service struct {

48 + resolver Resolver

49 + selfURLs []string

50 + rootCAPEM []byte

51 + requestTimeout time.Duration

52 + maxPeers int

53 + onBootstraps func([]string)

54 +

55 + mu sync.RWMutex

56 + bootstraps []string

57 +}

58 +

59 +func New(cfg Config, resolver Resolver) (*Service, error) {

60 + service := &Service{

61 + resolver: resolver,

62 + rootCAPEM: append([]byte(nil), cfg.RootCAPEM...),

63 + requestTimeout: utils.DurationOrDefault(cfg.RequestTimeout, defaultRequestTimeout),

64 + maxPeers: utils.IntOrDefault(cfg.MaxPeers, defaultMaxPeers),

65 + onBootstraps: cfg.OnBootstraps,

66 + }

67 + if err := service.SetSelfURLs(cfg.SelfURLs); err != nil {

68 + return nil, err

69 + }

70 + if err := service.MergeBootstraps(cfg.Bootstraps); err != nil {

71 + return nil, err

72 + }

73 + return service, nil

74 +}

75 +

76 +func (s *Service) Bootstraps() []string {

77 + if s == nil {

78 + return nil

79 + }

80 +

81 + s.mu.RLock()

82 + defer s.mu.RUnlock()

83 + return append([]string(nil), s.bootstraps...)

84 +}

85 +

86 +func (s *Service) MergeBootstraps(inputs []string) error {

87 + if s == nil || len(inputs) == 0 {

88 + return nil

89 + }

90 +

91 + normalized, err := utils.NormalizeRelayURLs(inputs)

92 + if err != nil {

93 + return fmt.Errorf("normalize bootstraps: %w", err)

94 + }

95 +

96 + s.mu.Lock()

97 + combined, err := utils.NormalizeRelayURLs(append(append([]string(nil), s.bootstraps...), normalized...))

98 + if err != nil {

99 + s.mu.Unlock()

100 + return fmt.Errorf("normalize bootstraps: %w", err)

101 + }

102 + next := utils.ExcludeURLs(combined, s.selfURLs)

103 + changed := strings.Join(s.bootstraps, "\x00") != strings.Join(next, "\x00")

104 + callback := s.onBootstraps

105 + if changed {

106 + s.bootstraps = next

107 + }

108 + bootstraps := append([]string(nil), s.bootstraps...)

109 + s.mu.Unlock()

110 +

111 + if changed && callback != nil {

112 + callback(bootstraps)

113 + }

114 + return nil

115 +}

116 +

117 +func (s *Service) SetSelfURLs(inputs []string) error {

118 + if s == nil {

119 + return nil

120 + }

121 +

122 + selfURLs, err := utils.NormalizeRelayURLs(inputs)

123 + if err != nil {

124 + return fmt.Errorf("normalize self urls: %w", err)

125 + }

126 +

127 + s.mu.Lock()

128 + current := strings.Join(s.bootstraps, "\x00")

129 + s.selfURLs = selfURLs

130 + s.bootstraps = utils.ExcludeURLs(s.bootstraps, s.selfURLs)

131 + changed := current != strings.Join(s.bootstraps, "\x00")

132 + callback := s.onBootstraps

133 + bootstraps := append([]string(nil), s.bootstraps...)

134 + s.mu.Unlock()

135 +

136 + if changed && callback != nil {

137 + callback(bootstraps)

138 + }

139 + return nil

140 +}

141 +

142 +func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) {

143 + if r.Method != http.MethodGet {

144 + utils.WriteAPIError(w, http.StatusMethodNotAllowed, types.APIErrorCodeMethodNotAllowed, "method not allowed")

145 + return

146 + }

147 +

148 + req, err := normalizeRequest(types.DiscoverRequest{

149 + RootHost: r.URL.Query().Get("root_host"),

150 + Name: r.URL.Query().Get("name"),

151 + })

152 + if err != nil {

153 + utils.WriteAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidRequest, err.Error())

154 + return

155 + }

156 +

157 + bootstraps, err := s.responseBootstraps(nil)

158 + if err != nil {

159 + utils.WriteAPIError(w, http.StatusInternalServerError, types.APIErrorCodeInternal, err.Error())

160 + return

161 + }

162 + resp := types.DiscoverResponse{

163 + Found: false,

164 + Bootstraps: bootstraps,

165 + }

166 + if s == nil || s.resolver == nil {

167 + utils.WriteAPIData(w, http.StatusOK, resp)

168 + return

169 + }

170 +

171 + localResp, err := s.resolver.Discover(r.Context(), req)

172 + if err != nil {

173 + utils.WriteAPIError(w, http.StatusInternalServerError, types.APIErrorCodeInternal, err.Error())

174 + return

175 + }

176 +

177 + bootstraps, err = s.responseBootstraps(localResp.Bootstraps)

178 + if err != nil {

179 + utils.WriteAPIError(w, http.StatusInternalServerError, types.APIErrorCodeInternal, err.Error())

180 + return

181 + }

182 + localResp.Bootstraps = bootstraps

183 + utils.WriteAPIData(w, http.StatusOK, localResp)

184 +}

185 +

186 +func (s *Service) Poll(ctx context.Context, req types.DiscoverRequest) (types.DiscoverResponse, error) {

187 + if ctx == nil {

188 + ctx = context.Background()

189 + }

190 +

191 + req, err := normalizeRequest(req)

192 + if err != nil {

193 + return types.DiscoverResponse{}, err

194 + }

195 +

196 + discovered := s.Bootstraps()

197 + if len(discovered) == 0 {

198 + return types.DiscoverResponse{}, errors.New("at least one bootstrap is required")

199 + }

200 +

201 + queue := append([]string(nil), discovered...)

202 + seen := make(map[string]struct{}, len(queue))

203 + var lastErr error

204 + contacted := false

205 +

206 + for len(queue) > 0 && len(seen) < s.maxPeers {

207 + bootstrap := queue[0]

208 + queue = queue[1:]

209 + if _, ok := seen[bootstrap]; ok {

210 + continue

211 + }

212 + seen[bootstrap] = struct{}{}

213 +

214 + resp, err := discoverPeer(ctx, bootstrap, req, s.rootCAPEM, s.requestTimeout)

215 + if err != nil {

216 + lastErr = err

217 + continue

218 + }

219 + contacted = true

220 +

221 + if err := s.MergeBootstraps(resp.Bootstraps); err != nil {

222 + return types.DiscoverResponse{}, err

223 + }

224 + discovered = s.Bootstraps()

225 + for _, nextBootstrap := range discovered {

226 + if _, ok := seen[nextBootstrap]; !ok {

227 + queue = append(queue, nextBootstrap)

228 + }

229 + }

230 +

231 + if resp.Found {

232 + resp.Bootstraps = discovered

233 + return resp, nil

234 + }

235 + }

236 +

237 + if !contacted && lastErr != nil {

238 + return types.DiscoverResponse{}, lastErr

239 + }

240 +

241 + return types.DiscoverResponse{

242 + Found: false,

243 + Bootstraps: discovered,

244 + }, nil

245 +}

246 +

247 +func (s *Service) RunPollLoop(ctx context.Context, interval time.Duration, req types.DiscoverRequest) error {

248 + if s == nil {

249 + return nil

250 + }

251 +

252 + interval = utils.DurationOrDefault(interval, defaultPollInterval)

253 + lastPollErr := ""

254 + for {

255 + if len(s.Bootstraps()) > 0 {

256 + if _, err := s.Poll(ctx, req); err != nil {

257 + if ctx.Err() != nil {

258 + return ctx.Err()

259 + }

260 + errText := err.Error()

261 + if errText != lastPollErr {

262 + log.Warn().

263 + Err(err).

264 + Int("bootstrap_count", len(s.Bootstraps())).

265 + Str("root_host", req.RootHost).

266 + Str("name", req.Name).

267 + Msg("discovery poll failed")

268 + lastPollErr = errText

269 + }

270 + } else if lastPollErr != "" {

271 + log.Info().

272 + Int("bootstrap_count", len(s.Bootstraps())).

273 + Str("root_host", req.RootHost).

274 + Str("name", req.Name).

275 + Msg("discovery poll recovered")

276 + lastPollErr = ""

277 + }

278 + }

279 + if !utils.SleepOrDone(ctx, interval) {

280 + return nil

281 + }

282 + }

283 +}

284 +

285 +func normalizeRequest(req types.DiscoverRequest) (types.DiscoverRequest, error) {

286 + req.RootHost = utils.NormalizeHostname(req.RootHost)

287 + req.Name = strings.TrimSpace(req.Name)

288 + if req.Name == "" {

289 + return req, nil

290 + }

291 + if req.RootHost == "" {

292 + return types.DiscoverRequest{}, errors.New("root host is required when name is set")

293 + }

294 + name, err := utils.NormalizeDNSLabel(req.Name)

295 + if err != nil {

296 + return types.DiscoverRequest{}, err

297 + }

298 + req.Name = name

299 + return req, nil

300 +}

301 +

302 +func (s *Service) responseBootstraps(extra []string) ([]string, error) {

303 + base := []string(nil)

304 + if s != nil {

305 + s.mu.RLock()

306 + base = append(base, s.selfURLs...)

307 + base = append(base, s.bootstraps...)

308 + s.mu.RUnlock()

309 + }

310 + bootstraps, err := utils.NormalizeRelayURLs(append(base, extra...))

311 + if err != nil {

312 + return nil, fmt.Errorf("normalize bootstraps: %w", err)

313 + }

314 + return bootstraps, nil

315 +}

316 +

317 +func discoverPeer(ctx context.Context, bootstrap string, req types.DiscoverRequest, rootCAPEM []byte, requestTimeout time.Duration) (types.DiscoverResponse, error) {

318 + baseURL, err := url.Parse(bootstrap)

319 + if err != nil {

320 + return types.DiscoverResponse{}, fmt.Errorf("parse bootstrap url: %w", err)

321 + }

322 +

323 + rootCAs, err := keyless.RelayRootCAs(ctx, bootstrap, baseURL.Hostname(), rootCAPEM)

324 + if err != nil {

325 + return types.DiscoverResponse{}, err

326 + }

327 +

328 + httpClient := &http.Client{

329 + Transport: &http.Transport{

330 + TLSClientConfig: &tls.Config{

331 + MinVersion: tls.VersionTLS12,

332 + ServerName: baseURL.Hostname(),

333 + RootCAs: rootCAs,

334 + },

335 + ForceAttemptHTTP2: false,

336 + },

337 + Timeout: utils.DurationOrDefault(requestTimeout, defaultRequestTimeout),

338 + }

339 + defer httpClient.CloseIdleConnections()

340 +

341 + query := url.Values{}

342 + if req.RootHost != "" {

343 + query.Set("root_host", req.RootHost)

344 + }

345 + if req.Name != "" {

346 + query.Set("name", req.Name)

347 + }

348 +

349 + ref := &url.URL{Path: types.PathDiscovery, RawQuery: query.Encode()}

350 + httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, baseURL.ResolveReference(ref).String(), nil)

351 + if err != nil {

352 + return types.DiscoverResponse{}, err

353 + }

354 +

355 + httpResp, err := httpClient.Do(httpReq)

356 + if err != nil {

357 + return types.DiscoverResponse{}, err

358 + }

359 + defer httpResp.Body.Close()

360 +

361 + if httpResp.StatusCode < http.StatusOK || httpResp.StatusCode >= http.StatusMultipleChoices {

362 + return types.DiscoverResponse{}, utils.DecodeAPIRequestError(httpResp)

363 + }

364 +

365 + envelope, err := utils.DecodeAPIEnvelope[json.RawMessage](httpResp.Body)

366 + if err != nil {

367 + return types.DiscoverResponse{}, fmt.Errorf("decode response: %w", err)

368 + }

369 + if !envelope.OK {

370 + return types.DiscoverResponse{}, utils.NewAPIRequestError(httpResp.StatusCode, envelope.Error)

371 + }

372 +

373 + var resp types.DiscoverResponse

374 + if err := json.Unmarshal(envelope.Data, &resp); err != nil {

375 + return types.DiscoverResponse{}, err

376 + }

377 + return resp, nil

378 +}

portal/discovery/identity.go new

+127

@@ -0,0 +1,127 @@

1 +package discovery

2 +

3 +import (

4 + "encoding/hex"

5 + "errors"

6 + "fmt"

7 + "strings"

8 +

9 + "github.com/decred/dcrd/dcrec/secp256k1/v4"

10 + "golang.org/x/crypto/sha3"

11 +)

12 +

13 +type Identity struct {

14 + Generated bool `json:"generated,omitempty"`

15 + Address string `json:"address"`

16 + PrivateKey string `json:"private_key"`

17 +}

18 +

19 +func NormalizeEVMAddress(raw string) (string, error) {

20 + trimmed := strings.TrimSpace(raw)

21 + if trimmed == "" {

22 + return "", errors.New("address is required")

23 + }

24 + if !strings.HasPrefix(strings.ToLower(trimmed), "0x") {

25 + return "", errors.New("address must start with 0x")

26 + }

27 +

28 + hexPart := trimmed[2:]

29 + if len(hexPart) != 40 {

30 + return "", errors.New("address must be 20 bytes")

31 + }

32 + if _, err := hex.DecodeString(hexPart); err != nil {

33 + return "", errors.New("address must be hex encoded")

34 + }

35 +

36 + lowerHex := strings.ToLower(hexPart)

37 + hasher := sha3.NewLegacyKeccak256()

38 + _, _ = hasher.Write([]byte(lowerHex))

39 + hash := hasher.Sum(nil)

40 +

41 + var builder strings.Builder

42 + builder.Grow(len(lowerHex))

43 + for idx, ch := range lowerHex {

44 + if ch >= '0' && ch <= '9' {

45 + builder.WriteRune(ch)

46 + continue

47 + }

48 +

49 + nibble := hash[idx/2]

50 + if idx%2 == 0 {

51 + nibble >>= 4

52 + } else {

53 + nibble &= 0x0f

54 + }

55 + if nibble > 7 {

56 + builder.WriteRune(ch - ('a' - 'A'))

57 + continue

58 + }

59 + builder.WriteRune(ch)

60 + }

61 +

62 + checksummed := builder.String()

63 + if hexPart != lowerHex && hexPart != strings.ToUpper(hexPart) && hexPart != checksummed {

64 + return "", errors.New("address checksum is invalid")

65 + }

66 + return "0x" + checksummed, nil

67 +}

68 +

69 +func ResolveIdentity(rawPrivateKey string) (Identity, error) {

70 + privateKeyHex := strings.TrimSpace(rawPrivateKey)

71 + generated := false

72 + if privateKeyHex == "" {

73 + privateKey, err := secp256k1.GeneratePrivateKey()

74 + if err != nil {

75 + return Identity{}, fmt.Errorf("generate secp256k1 private key: %w", err)

76 + }

77 + privateKeyHex = hex.EncodeToString(privateKey.Serialize())

78 + generated = true

79 + }

80 + if strings.HasPrefix(strings.ToLower(privateKeyHex), "0x") {

81 + privateKeyHex = privateKeyHex[2:]

82 + }

83 +

84 + decoded, err := hex.DecodeString(privateKeyHex)

85 + if err != nil {

86 + return Identity{}, errors.New("secp256k1 private key must be hex encoded")

87 + }

88 + if len(decoded) != secp256k1.PrivKeyBytesLen {

89 + return Identity{}, fmt.Errorf("secp256k1 private key must be %d bytes", secp256k1.PrivKeyBytesLen)

90 + }

91 +

92 + isZero := true

93 + for _, b := range decoded {

94 + if b != 0 {

95 + isZero = false

96 + break

97 + }

98 + }

99 + if isZero {

100 + return Identity{}, errors.New("secp256k1 private key must not be zero")

101 + }

102 +

103 + privateKey := secp256k1.PrivKeyFromBytes(decoded)

104 + if privateKey == nil {

105 + return Identity{}, errors.New("invalid secp256k1 private key")

106 + }

107 +

108 + uncompressed := privateKey.PubKey().SerializeUncompressed()

109 + if len(uncompressed) != 65 || uncompressed[0] != 0x04 {

110 + return Identity{}, errors.New("invalid uncompressed secp256k1 public key")

111 + }

112 +

113 + hasher := sha3.NewLegacyKeccak256()

114 + _, _ = hasher.Write(uncompressed[1:])

115 + hash := hasher.Sum(nil)

116 +

117 + address, err := NormalizeEVMAddress("0x" + hex.EncodeToString(hash[len(hash)-20:]))

118 + if err != nil {

119 + return Identity{}, err

120 + }

121 +

122 + return Identity{

123 + Generated: generated,

124 + Address: address,

125 + PrivateKey: privateKeyHex,

126 + }, nil

127 +}

portal/keyless/tls.go

+17

@@ -1,15 +1,32 @@

1 package keyless

3 import (

4 + "context"

5 "crypto/tls"

6 + "crypto/x509"

7 "errors"

8 "fmt"

9 "io"

10 "net/http"

12 keylesstls "github.com/gosuda/keyless_tls/keyless"

13 +

14 + "github.com/gosuda/portal/v2/utils"

15 )

17 +func RelayRootCAs(ctx context.Context, endpoint, serverName string, rootCAPEM []byte) (*x509.CertPool, error) {

18 + resolvedRootCAPEM := append([]byte(nil), rootCAPEM...)

19 + if len(resolvedRootCAPEM) == 0 && utils.IsLocalRelayHost(serverName) {

20 + _, fetchedRootCAPEM, err := ResolveMaterials(ctx, endpoint, serverName)

21 + if err != nil {

22 + return nil, fmt.Errorf("bootstrap localhost relay trust: %w", err)

23 + }

24 + resolvedRootCAPEM = fetchedRootCAPEM

25 + }

26 +

27 + return utils.CertPoolFromPEM(resolvedRootCAPEM)

28 +}

29 +

30 type TLSMaterialConfig struct {

31 Keyless *RemoteSignerConfig

32 CertPEM []byte

portal/lease.go

@@ -231,6 +231,7 @@ func (r *leaseRegistry) Snapshot(record *leaseRecord) types.Lease {

231 }

232

233 snapshot := record.Lease

234 + snapshot.Bootstraps = append([]string(nil), snapshot.Bootstraps...)

235 snapshot.Metadata = snapshot.Metadata.Copy()

236 clientIP := record.ClientIP

237 snapshot.BPS = r.policy.BPSManager().LeaseBPS(record.ID)

portal/server.go

+69 -17

@@ -8,6 +8,7 @@ import (

8 "io"

9 "net"

10 "net/http"

11 + "strings"

12 "sync"

13 "time"

@@ -17,6 +18,7 @@ import (

18 "golang.org/x/sync/errgroup"

20 "github.com/gosuda/portal/v2/portal/acme"

21 + "github.com/gosuda/portal/v2/portal/discovery"

22 "github.com/gosuda/portal/v2/portal/keyless"

23 "github.com/gosuda/portal/v2/portal/policy"

24 "github.com/gosuda/portal/v2/portal/transport"

@@ -41,6 +43,8 @@ const (

44 type ServerConfig struct {

45 PortalURL string

46 + OwnerPrivateKey string

47 + Bootstraps []string

48 ACME acme.Config

49 APIListenAddr string

50 SNIListenAddr string

@@ -52,23 +56,26 @@ type ServerConfig struct {

56 ReadyQueueLimit int

57 ClientHelloTimeout time.Duration

58 TrustProxyHeaders bool

59 + DiscoveryEnabled bool

60 UDPPortCount int

61 }

63 type Server struct {

59 - sniListener net.Listener

60 - apiListener net.Listener

61 - apiServer *http.Server

62 - apiTLSClose io.Closer

63 - acmeManager *acme.Manager

64 - quicTunnel *quic.Listener

65 - cancel context.CancelFunc

66 - group *errgroup.Group

67 - registry *leaseRegistry

68 - ports *transport.PortAllocator

69 - cfg ServerConfig

70 - rootHost string

71 - shutdownOnce sync.Once

64 + sniListener net.Listener

65 + apiListener net.Listener

66 + apiServer *http.Server

67 + apiTLSClose io.Closer

68 + acmeManager *acme.Manager

69 + quicTunnel *quic.Listener

70 + cancel context.CancelFunc

71 + group *errgroup.Group

72 + discovery *discovery.Service

73 + registry *leaseRegistry

74 + ports *transport.PortAllocator

75 + ownerIdentity discovery.Identity

76 + cfg ServerConfig

77 + rootHost string

78 + shutdownOnce sync.Once

79 }

81 func NewServer(cfg ServerConfig) (*Server, error) {

@@ -90,6 +97,13 @@ func NewServer(cfg ServerConfig) (*Server, error) {

97 if cfg.QUICListenAddr == "" {

98 cfg.QUICListenAddr = cfg.SNIListenAddr

99 }

100 + if len(cfg.Bootstraps) > 0 {

101 + bootstraps, err := utils.NormalizeRelayURLs(cfg.Bootstraps)

102 + if err != nil {

103 + return nil, err

104 + }

105 + cfg.Bootstraps = bootstraps

106 + }

107

108 portMin, portMax := 0, 0

109 if cfg.UDPPortCount > 0 {

@@ -97,14 +111,25 @@ func NewServer(cfg ServerConfig) (*Server, error) {

111 portMax = defaultUDPPortBase + cfg.UDPPortCount - 1

112 }

113

114 + ownerIdentity := discovery.Identity{}

115 + if cfg.DiscoveryEnabled || strings.TrimSpace(cfg.OwnerPrivateKey) != "" {

116 + identity, err := discovery.ResolveIdentity(cfg.OwnerPrivateKey)

117 + if err != nil {

118 + return nil, fmt.Errorf("resolve owner identity: %w", err)

119 + }

120 + ownerIdentity = identity

121 + cfg.OwnerPrivateKey = identity.PrivateKey

122 + }

123 +

124 registry := newLeaseRegistry(policy.NewRuntime())

125 ports := transport.NewPortAllocator(portMin, portMax, 5*time.Minute)

126

127 s := &Server{

104 - cfg: cfg,

105 - rootHost: rootHost,

106 - registry: registry,

107 - ports: ports,

128 + cfg: cfg,

129 + rootHost: rootHost,

130 + registry: registry,

131 + ports: ports,

132 + ownerIdentity: ownerIdentity,

133 }

134

135 // Tear down all lease resources when leases expire via TTL janitor.

@@ -112,6 +137,17 @@ func NewServer(cfg ServerConfig) (*Server, error) {

137 s.closeLease(record)

138 }

139

140 + if cfg.DiscoveryEnabled {

141 + service, err := discovery.New(discovery.Config{

142 + SelfURLs: []string{cfg.PortalURL},

143 + Bootstraps: cfg.Bootstraps,

144 + }, discovery.ResolverFunc(s.discover))

145 + if err != nil {

146 + return nil, err

147 + }

148 + s.discovery = service

149 + }

150 +

151 return s, nil

152 }

153

@@ -162,6 +198,11 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {

198 group.Go(s.runAPIServer)

199 group.Go(func() error { return s.runSNIListener(groupCtx) })

200 group.Go(func() error { return s.registry.RunJanitor(groupCtx, 5*time.Second) })

201 + if s.discovery != nil {

202 + group.Go(func() error {

203 + return s.discovery.RunPollLoop(groupCtx, 0, types.DiscoverRequest{RootHost: s.rootHost})

204 + })

205 + }

206 group.Go(func() error { return s.watchContext(groupCtx) })

207 s.acmeManager.Start(serverCtx)

208

@@ -243,6 +284,17 @@ func (s *Server) QUICTunnelAddr() string {

284 return s.quicTunnel.Addr().String()

285 }

286

287 +func (s *Server) DiscoveryEnabled() bool {

288 + return s != nil && s.cfg.DiscoveryEnabled && s.discovery != nil

289 +}

290 +

291 +func (s *Server) OwnerIdentity() discovery.Identity {

292 + if s == nil {

293 + return discovery.Identity{}

294 + }

295 + return s.ownerIdentity

296 +}

297 +

298 func (s *Server) LeaseSnapshots() []types.Lease {

299 s.registry.mu.RLock()

300 defer s.registry.mu.RUnlock()

portal/server_test.go

+131

@@ -9,6 +9,7 @@ import (

9 "testing"

11 "github.com/gosuda/portal/v2/portal/acme"

12 + "github.com/gosuda/portal/v2/portal/discovery"

13 "github.com/gosuda/portal/v2/types"

14 "github.com/gosuda/portal/v2/utils"

15 )

@@ -199,3 +200,133 @@ func TestRegisterLeaseBuildsUDPEnabledRuntime(t *testing.T) {

200 t.Fatal("RegisterResponse.UDPAddr = empty, want public udp address")

201 }

202 }

203 +

204 +func TestServerStartServesOptionalDiscoveryRoutes(t *testing.T) {

205 + t.Parallel()

206 +

207 + ownerPrivateKey := strings.Repeat("11", 32)

208 + ownerIdentity, err := discovery.ResolveIdentity(ownerPrivateKey)

209 + if err != nil {

210 + t.Fatalf("ResolveIdentity() error = %v", err)

211 + }

212 +

213 + server, err := NewServer(ServerConfig{

214 + PortalURL: "https://localhost:4017",

215 + OwnerPrivateKey: ownerPrivateKey,

216 + Bootstraps: []string{"https://bootstrap.example.com"},

217 + ACME: acme.Config{KeyDir: t.TempDir()},

218 + APIListenAddr: "127.0.0.1:0",

219 + SNIListenAddr: "127.0.0.1:0",

220 + DiscoveryEnabled: true,

221 + })

222 + if err != nil {

223 + t.Fatalf("NewServer() error = %v", err)

224 + }

225 + if _, err := server.registerLease(types.RegisterRequest{

226 + Name: "demo",

227 + ReverseToken: "tok_demo",

228 + Bootstraps: []string{"https://relay-a.example.com", "https://bootstrap.example.com"},

229 + }, "203.0.113.10"); err != nil {

230 + t.Fatalf("registerLease() error = %v", err)

231 + }

232 +

233 + ctx, cancel := context.WithCancel(context.Background())

234 + defer cancel()

235 +

236 + if err := server.Start(ctx, nil); err != nil {

237 + t.Fatalf("Start() error = %v", err)

238 + }

239 +

240 + client := &http.Client{

241 + Transport: &http.Transport{

242 + TLSClientConfig: &tls.Config{InsecureSkipVerify: true},

243 + },

244 + }

245 + t.Cleanup(func() {

246 + client.CloseIdleConnections()

247 + cancel()

248 + if err := server.Wait(); err != nil {

249 + t.Fatalf("Wait() error = %v", err)

250 + }

251 + })

252 +

253 + resp, err := client.Get("https://" + utils.HostPortOrLoopback(server.APIAddr()) + types.PathDiscovery + "?root_host=localhost&name=demo")

254 + if err != nil {

255 + t.Fatalf("GET discovery resolve error = %v", err)

256 + }

257 + defer resp.Body.Close()

258 +

259 + if resp.StatusCode != http.StatusOK {

260 + t.Fatalf("GET discovery resolve status = %d, want %d", resp.StatusCode, http.StatusOK)

261 + }

262 +

263 + var envelope types.APIEnvelope[types.DiscoverResponse]

264 + if err := json.NewDecoder(resp.Body).Decode(&envelope); err != nil {

265 + t.Fatalf("decode discovery resolve response: %v", err)

266 + }

267 + if !envelope.OK {

268 + t.Fatalf("discovery resolve envelope = %+v, want ok", envelope)

269 + }

270 + if !envelope.Data.Found {

271 + t.Fatalf("resolve found = %v, want true", envelope.Data.Found)

272 + }

273 + if envelope.Data.OwnerAddress != ownerIdentity.Address {

274 + t.Fatalf("resolve owner address = %q, want relay owner address", envelope.Data.OwnerAddress)

275 + }

276 + if envelope.Data.Hostname != "demo.localhost" {

277 + t.Fatalf("resolve hostname = %q, want %q", envelope.Data.Hostname, "demo.localhost")

278 + }

279 + if len(envelope.Data.Bootstraps) != 3 || envelope.Data.Bootstraps[0] != "https://localhost:4017" || envelope.Data.Bootstraps[1] != "https://bootstrap.example.com" || envelope.Data.Bootstraps[2] != "https://relay-a.example.com" {

280 + t.Fatalf("resolve bootstraps = %v, want [%q %q %q]", envelope.Data.Bootstraps, "https://localhost:4017", "https://bootstrap.example.com", "https://relay-a.example.com")

281 + }

282 + if !server.DiscoveryEnabled() {

283 + t.Fatal("DiscoveryEnabled() = false, want true")

284 + }

285 +}

286 +

287 +func TestServerStartHidesDiscoveryRoutesWhenDisabled(t *testing.T) {

288 + t.Parallel()

289 +

290 + server, err := NewServer(ServerConfig{

291 + PortalURL: "https://localhost:4017",

292 + ACME: acme.Config{KeyDir: t.TempDir()},

293 + APIListenAddr: "127.0.0.1:0",

294 + SNIListenAddr: "127.0.0.1:0",

295 + })

296 + if err != nil {

297 + t.Fatalf("NewServer() error = %v", err)

298 + }

299 +

300 + ctx, cancel := context.WithCancel(context.Background())

301 + defer cancel()

302 +

303 + if err := server.Start(ctx, nil); err != nil {

304 + t.Fatalf("Start() error = %v", err)

305 + }

306 +

307 + client := &http.Client{

308 + Transport: &http.Transport{

309 + TLSClientConfig: &tls.Config{InsecureSkipVerify: true},

310 + },

311 + }

312 + t.Cleanup(func() {

313 + client.CloseIdleConnections()

314 + cancel()

315 + if err := server.Wait(); err != nil {

316 + t.Fatalf("Wait() error = %v", err)

317 + }

318 + })

319 +

320 + resp, err := client.Get("https://" + utils.HostPortOrLoopback(server.APIAddr()) + types.PathDiscovery + "?root_host=localhost&name=demo")

321 + if err != nil {

322 + t.Fatalf("GET discovery resolve error = %v", err)

323 + }

324 + defer resp.Body.Close()

325 +

326 + if resp.StatusCode != http.StatusNotFound {

327 + t.Fatalf("GET discovery resolve status = %d, want %d", resp.StatusCode, http.StatusNotFound)

328 + }

329 + if server.DiscoveryEnabled() {

330 + t.Fatal("DiscoveryEnabled() = true, want false without configured discovery service")

331 + }

332 +}

sdk/api_client.go

+30 -23

@@ -17,6 +17,7 @@ import (

18 "github.com/quic-go/quic-go"

20 + "github.com/gosuda/portal/v2/portal/discovery"

21 "github.com/gosuda/portal/v2/portal/keyless"

22 "github.com/gosuda/portal/v2/types"

23 "github.com/gosuda/portal/v2/utils"

@@ -44,7 +45,9 @@ type apiClient struct {

45 rootCAPEM []byte

46 name string

47 reverseToken string

48 + discovery bool

49 metadata types.LeaseMetadata

50 + ownerAddress string

51 }

53 func newApiClient(relayURL string, cfg ListenerConfig) (*apiClient, error) {

@@ -70,17 +73,25 @@ func newApiClient(relayURL string, cfg ListenerConfig) (*apiClient, error) {

74 dialTimeout := utils.DurationOrDefault(cfg.DialTimeout, defaultDialTimeout)

75 requestTimeout := utils.DurationOrDefault(cfg.RequestTimeout, defaultRequestTimeout)

76 + ownerAddress := strings.TrimSpace(cfg.OwnerAddress)

77 + if ownerAddress != "" {

78 + ownerAddress, err = discovery.NormalizeEVMAddress(ownerAddress)

79 + if err != nil {

80 + return nil, fmt.Errorf("normalize owner address: %w", err)

81 + }

82 + }

74 - api := &apiClient{

84 + return &apiClient{

85 baseURL: baseURL,

86 dialTimeout: dialTimeout,

87 requestTimeout: requestTimeout,

88 rootCAPEM: append([]byte(nil), cfg.RootCAPEM...),

89 name: name,

90 reverseToken: reverseToken,

91 + discovery: cfg.Discovery,

92 metadata: cfg.Metadata.Copy(),

82 - }

83 - return api, nil

93 + ownerAddress: ownerAddress,

94 + }, nil

95 }

97 func (a *apiClient) close() {

@@ -92,13 +103,18 @@ func (a *apiClient) close() {

103 }

104 }

105

95 -func (a *apiClient) registerLease(ctx context.Context, ttl time.Duration, udpEnabled bool) (types.RegisterResponse, error) {

106 +func (a *apiClient) registerLease(ctx context.Context, ttl time.Duration, udpEnabled bool, bootstraps []string) (types.RegisterResponse, error) {

107 var resp types.RegisterResponse

97 - if err := a.doJSONWithClient(ctx, a.httpClient, http.MethodPost, types.PathSDKRegister, types.RegisterRequest{

108 + if !a.discovery {

109 + bootstraps = nil

110 + }

111 + if err := a.doJSON(ctx, http.MethodPost, types.PathSDKRegister, types.RegisterRequest{

112 Name: a.name,

113 Metadata: a.metadata.Copy(),

114 + OwnerAddress: a.ownerAddress,

115 ReverseToken: a.reverseToken,

116 TTL: int(ttl / time.Second),

117 + Bootstraps: bootstraps,

118 UDPEnabled: udpEnabled,

119 }, &resp); err != nil {

120 return types.RegisterResponse{}, err

@@ -111,23 +127,10 @@ func (a *apiClient) ensureReady(ctx context.Context) error {

127 return nil

128 }

129

114 - rootCAPEM := append([]byte(nil), a.rootCAPEM...)

115 - if len(rootCAPEM) == 0 && utils.IsLocalRelayHost(a.baseURL.Hostname()) {

116 - bootstrapParent := ctx

117 - if bootstrapParent == nil {

118 - bootstrapParent = context.Background()

119 - }

120 - bootstrapCtx, cancel := context.WithTimeout(bootstrapParent, defaultDialTimeout+defaultHandshakeTimeout)

121 - defer cancel()

122 -

123 - _, resolvedCAPEM, err := keyless.ResolveMaterials(bootstrapCtx, a.baseURL.String(), a.baseURL.Hostname())

124 - if err != nil {

125 - return fmt.Errorf("bootstrap localhost relay trust: %w", err)

126 - }

127 - rootCAPEM = resolvedCAPEM

128 - }

130 + bootstrapCtx, cancel := context.WithTimeout(ctx, defaultDialTimeout+defaultHandshakeTimeout)

131 + defer cancel()

132

130 - rootCAs, err := utils.CertPoolFromPEM(rootCAPEM)

133 + rootCAs, err := keyless.RelayRootCAs(bootstrapCtx, a.baseURL.String(), a.baseURL.Hostname(), a.rootCAPEM)

134 if err != nil {

135 return err

136 }

@@ -177,7 +180,7 @@ func (a *apiClient) ensureCompatible(ctx context.Context, httpClient *http.Clien

180 }

181

182 func (a *apiClient) renewLease(ctx context.Context, leaseID string, ttl time.Duration) error {

180 - return a.doJSONWithClient(ctx, a.httpClient, http.MethodPost, types.PathSDKRenew, types.RenewRequest{

183 + return a.doJSON(ctx, http.MethodPost, types.PathSDKRenew, types.RenewRequest{

184 LeaseID: leaseID,

185 ReverseToken: a.reverseToken,

186 TTL: int(ttl / time.Second),

@@ -185,7 +188,7 @@ func (a *apiClient) renewLease(ctx context.Context, leaseID string, ttl time.Dur

188 }

189

190 func (a *apiClient) unregisterLease(ctx context.Context, leaseID string) error {

188 - return a.doJSONWithClient(ctx, a.httpClient, http.MethodPost, types.PathSDKUnregister, types.UnregisterRequest{

191 + return a.doJSON(ctx, http.MethodPost, types.PathSDKUnregister, types.UnregisterRequest{

192 LeaseID: leaseID,

193 ReverseToken: a.reverseToken,

194 }, nil)

@@ -239,6 +242,10 @@ func (a *apiClient) openReverseSession(ctx context.Context, leaseID string) (net

242 return wrapBufferedConn(conn, reader), nil

243 }

244

245 +func (a *apiClient) doJSON(ctx context.Context, method, path string, payload any, out any) error {

246 + return a.doJSONWithClient(ctx, a.httpClient, method, path, payload, out)

247 +}

248 +

249 func (a *apiClient) doJSONWithClient(ctx context.Context, httpClient *http.Client, method, path string, payload any, out any) error {

250 if httpClient == nil {

251 return errors.New("api client is not ready")

sdk/expose.go

+347 -288

@@ -13,6 +13,7 @@ import (

14 "github.com/rs/zerolog/log"

16 + "github.com/gosuda/portal/v2/portal/discovery"

17 "github.com/gosuda/portal/v2/types"

18 "github.com/gosuda/portal/v2/utils"

19 )

@@ -20,157 +21,206 @@ import (

21 // Exposure owns the lifecycle of one or more relay listeners and accepts

22 // traffic from all of them through one net.Listener.

23 type Exposure struct {

23 - udpEnabled bool

24 - listener net.Listener

25 - listeners []*Listener

26 - datagrams chan exposureDatagram

27 - done chan struct{}

24 + ctx context.Context

25 + cancel context.CancelFunc

26 +

27 + name string

28 + reverseToken string

29 + udpEnabled bool

30 + identity discovery.Identity

31 + metadata types.LeaseMetadata

32 + ownerAddress string

33 + rootCAPEM []byte

34 + discoveryEnabled bool

35 + discovery *discovery.Service

36 +

37 + accepted chan net.Conn

38 + datagrams chan types.DatagramFrame

39 + done chan struct{}

40 +

41 + mu sync.RWMutex

42 + relayURLs []string

43 + listeners map[string]*Listener

44 + starting map[string]struct{}

46 closeOnce sync.Once

47 connSeq atomic.Uint64

48 }

33 -type exposureDatagram struct {

34 - FlowID uint32

35 - LeaseID string

36 - Payload []byte

37 - RelayURL string

38 - UDPAddr string

39 -

40 - reply func([]byte) error

50 +type ExposeConfig struct {

51 + RelayURLs []string

52 + Name string

53 + ReverseToken string

54 + UDPEnabled bool

55 + Discovery bool

56 + Metadata types.LeaseMetadata

57 + OwnerAddress string

58 + OwnerPrivateKey *string

59 + RootCAPEM []byte

60 }

43 -func (e *Exposure) SupportsStream() bool {

44 - return e != nil

62 +// Expose creates relay listeners for each normalized relay URL and exposes a

63 +// dynamic listener hub for accepting traffic from all of them.

64 +func Expose(ctx context.Context, relayURLs []string, name string, udpEnabled bool, metadata types.LeaseMetadata) (*Exposure, error) {

65 + return ExposeWithConfig(ctx, ExposeConfig{

66 + RelayURLs: relayURLs,

67 + Name: name,

68 + UDPEnabled: udpEnabled,

69 + Metadata: metadata,

70 + })

71 }

47 -// Expose creates relay listeners for each normalized relay URL and exposes a

48 -// merged listener for accepting traffic from all of them. Empty relay input

49 -// returns nil, nil so callers can fall back to local-only serving.

50 -func Expose(ctx context.Context, relayUrls []string, name string, udpEnabled bool, metadata types.LeaseMetadata) (*Exposure, error) {

51 - relayURLs, err := utils.NormalizeRelayURLs(relayUrls)

73 +func ExposeWithConfig(ctx context.Context, cfg ExposeConfig) (*Exposure, error) {

74 + relayURLs, err := utils.NormalizeRelayURLs(cfg.RelayURLs)

75 if err != nil {

76 return nil, err

77 }

78 if len(relayURLs) == 0 {

79 return nil, nil

80 }

58 - listeners := make([]*Listener, 0, len(relayURLs))

59 - cleanup := func() error {

60 - var closeErr error

61 - for _, listener := range listeners {

62 - if listener != nil {

63 - closeErr = errors.Join(closeErr, listener.Close())

64 - }

65 - }

66 - return closeErr

81 + if ctx == nil {

82 + ctx = context.Background()

83 }

69 - for _, relayURL := range relayURLs {

70 - listener, err := NewListener(ctx, relayURL, ListenerConfig{

71 - Name: name,

72 - UDPEnabled: udpEnabled,

73 - Metadata: metadata,

74 - })

85 + ownerAddress := strings.TrimSpace(cfg.OwnerAddress)

86 + identity := discovery.Identity{}

87 + if cfg.OwnerPrivateKey != nil {

88 + identity, err = discovery.ResolveIdentity(*cfg.OwnerPrivateKey)

89 if err != nil {

76 - return nil, errors.Join(fmt.Errorf("listen %q: %w", relayURL, err), cleanup())

90 + return nil, fmt.Errorf("resolve owner identity: %w", err)

91 }

78 -

79 - listeners = append(listeners, listener)

92 + ownerAddress = identity.Address

93 }

82 - mergedListeners := make([]net.Listener, 0, len(listeners))

83 - for _, listener := range listeners {

84 - mergedListeners = append(mergedListeners, listener)

85 - }

86 - merged, err := mergeListeners(mergedListeners...)

87 - if err != nil {

88 - return nil, errors.Join(fmt.Errorf("merge listeners: %w", err), cleanup())

95 + exposureCtx, cancel := context.WithCancel(ctx)

96 + exposure := &Exposure{

97 + ctx: exposureCtx,

98 + cancel: cancel,

99 + name: cfg.Name,

100 + reverseToken: cfg.ReverseToken,

101 + udpEnabled: cfg.UDPEnabled,

102 + identity: identity,

103 + metadata: cfg.Metadata.Copy(),

104 + ownerAddress: ownerAddress,

105 + rootCAPEM: append([]byte(nil), cfg.RootCAPEM...),

106 + discoveryEnabled: cfg.Discovery,

107 + accepted: make(chan net.Conn, max(len(relayURLs)*defaultReadyTarget*2, 1)),

108 + datagrams: make(chan types.DatagramFrame, max(len(relayURLs)*32, 1)),

109 + done: make(chan struct{}),

110 + listeners: make(map[string]*Listener, len(relayURLs)),

111 + starting: make(map[string]struct{}, len(relayURLs)),

112 + }

113 +

114 + if exposure.discoveryEnabled {

115 + service, err := discovery.New(discovery.Config{

116 + RootCAPEM: exposure.rootCAPEM,

117 + OnBootstraps: func(relays []string) {

118 + if err := exposure.applyRelayURLs(relays, false); err != nil {

119 + log.Warn().Err(err).Strs("relays", relays).Msg("apply discovered relay urls")

120 + }

121 + },

122 + }, nil)

123 + if err != nil {

124 + cancel()

125 + return nil, err

126 + }

127 + exposure.discovery = service

128 }

129

91 - exposure := &Exposure{

92 - udpEnabled: udpEnabled,

93 - listener: merged,

94 - listeners: listeners,

95 - datagrams: make(chan exposureDatagram, max(len(listeners)*32, 1)),

96 - done: make(chan struct{}),

130 + if exposure.discovery != nil {

131 + if err := exposure.discovery.MergeBootstraps(relayURLs); err != nil {

132 + _ = exposure.Close()

133 + return nil, err

134 + }

135 + if err := exposure.applyRelayURLs(exposure.discovery.Bootstraps(), true); err != nil {

136 + _ = exposure.Close()

137 + return nil, err

138 + }

139 + } else if err := exposure.applyRelayURLs(relayURLs, true); err != nil {

140 + _ = exposure.Close()

141 + return nil, err

142 }

98 - go exposure.monitorStartupCounts(ctx)

99 - if exposure.UDPEnabled() {

100 - exposure.attachDatagramPlanes(ctx)

143 +

144 + go exposure.monitorStartupCounts(exposureCtx)

145 + if exposure.discovery != nil {

146 + go func() {

147 + _ = exposure.discovery.RunPollLoop(exposureCtx, 0, types.DiscoverRequest{})

148 + }()

149 }

150

151 log.Info().

152 Str("release_version", types.ReleaseVersion).

105 - Int("relay_count", len(exposure.listeners)).

153 + Int("relay_count", len(exposure.RelayURLs())).

154 Strs("relays", exposure.RelayURLs()).

107 - Msgf("exposure relay started")

155 + Msg("exposure relay started")

156

157 return exposure, nil

158 }

159

112 -// RelayURLs returns the normalized relay URLs backing the exposure.

160 func (e *Exposure) RelayURLs() []string {

114 - if e == nil || len(e.listeners) == 0 {

161 + if e == nil {

162 return nil

163 }

164

118 - out := make([]string, 0, len(e.listeners))

119 - for _, listener := range e.listeners {

120 - if listener == nil {

121 - continue

122 - }

123 - out = append(out, listener.relayURL)

165 + e.mu.RLock()

166 + defer e.mu.RUnlock()

167 +

168 + if len(e.relayURLs) == 0 {

169 + return nil

170 }

125 - return out

171 +

172 + return append([]string(nil), e.relayURLs...)

173 +}

174 +

175 +func (e *Exposure) OwnerIdentity() discovery.Identity {

176 + if e == nil {

177 + return discovery.Identity{}

178 + }

179 + return e.identity

180 }

181

182 func (e *Exposure) Accept() (net.Conn, error) {

129 - if e == nil || e.listener == nil {

183 + if e == nil {

184 return nil, net.ErrClosed

185 }

186

133 - conn, err := e.listener.Accept()

134 - if err != nil {

135 - if !errors.Is(err, net.ErrClosed) {

136 - log.Warn().

137 - Err(err).

138 - Str("local_addr", utils.AddrString(e.listener.Addr())).

139 - Msg("exposure accept failed")

187 + select {

188 + case <-e.done:

189 + return nil, net.ErrClosed

190 + case conn := <-e.accepted:

191 + if conn == nil {

192 + return nil, net.ErrClosed

193 }

141 - return nil, err

142 - }

194

144 - connID := e.connSeq.Add(1)

145 - log.Info().

146 - Uint64("conn_id", connID).

147 - Str("local_addr", utils.AddrString(conn.LocalAddr())).

148 - Str("remote_addr", utils.AddrString(conn.RemoteAddr())).

149 - Msg("exposure connection accepted")

195 + connID := e.connSeq.Add(1)

196 + log.Info().

197 + Uint64("conn_id", connID).

198 + Str("local_addr", utils.AddrString(conn.LocalAddr())).

199 + Str("remote_addr", utils.AddrString(conn.RemoteAddr())).

200 + Msg("exposure connection accepted")

201

151 - return &exposureConn{

152 - Conn: conn,

153 - id: connID,

154 - localAddr: utils.AddrString(conn.LocalAddr()),

155 - remoteAddr: utils.AddrString(conn.RemoteAddr()),

156 - }, nil

202 + return &exposureConn{

203 + Conn: conn,

204 + id: connID,

205 + localAddr: utils.AddrString(conn.LocalAddr()),

206 + remoteAddr: utils.AddrString(conn.RemoteAddr()),

207 + }, nil

208 + }

209 }

210

211 func (e *Exposure) Addr() net.Addr {

160 - if e == nil || e.listener == nil {

161 - return listenerAddr("portal:exposure")

162 - }

163 - return e.listener.Addr()

212 + return listenerAddr("portal:exposure")

213 }

214

215 func (e *Exposure) PublicURLs() []string {

167 - if e == nil || len(e.listeners) == 0 {

216 + listeners := e.listenersOrdered()

217 + if len(listeners) == 0 {

218 return nil

219 }

220

171 - out := make([]string, 0, len(e.listeners))

221 + out := make([]string, 0, len(listeners))

222 seen := make(map[string]struct{})

173 - for _, listener := range e.listeners {

223 + for _, listener := range listeners {

224 if listener == nil {

225 continue

226 }

@@ -191,10 +241,11 @@ func (e *Exposure) PublicURLs() []string {

241 }

242

243 func (e *Exposure) RunHTTP(ctx context.Context, handler http.Handler, localAddr string) error {

194 - if e == nil || e.listener == nil {

195 - return RunHTTP(ctx, nil, handler, localAddr)

244 + var relayListener net.Listener

245 + if e != nil {

246 + relayListener = e

247 }

197 - return RunHTTP(ctx, e, handler, localAddr)

248 + return RunHTTP(ctx, relayListener, handler, localAddr)

249 }

250

251 func RunHTTP(ctx context.Context, relayListener net.Listener, handler http.Handler, localAddr string) error {

@@ -298,145 +349,192 @@ func RunHTTP(ctx context.Context, relayListener net.Listener, handler http.Handl

349 return errors.Join(serveErr, shutdownErr)

350 }

351

301 -func mergeListeners(listeners ...net.Listener) (net.Listener, error) {

302 - if len(listeners) == 0 {

303 - return nil, errors.New("at least one listener is required")

352 +func (e *Exposure) Close() error {

353 + if e == nil {

354 + return nil

355 }

356

306 - merged := &mergedListener{

307 - listeners: make([]net.Listener, 0, len(listeners)),

308 - accepted: make(chan net.Conn),

309 - closed: make(chan struct{}),

310 - }

311 - for i, listener := range listeners {

312 - if listener == nil {

313 - return nil, fmt.Errorf("listener %d is nil", i)

357 + var closeErr error

358 + e.closeOnce.Do(func() {

359 + if e.cancel != nil {

360 + e.cancel()

361 + }

362 + if e.done != nil {

363 + close(e.done)

364 + }

365 +

366 + listeners := e.listenersOrdered()

367 + for _, listener := range listeners {

368 + if listener != nil {

369 + closeErr = errors.Join(closeErr, listener.Close())

370 + }

371 + }

372 +

373 + event := log.Info().

374 + Int("relay_count", len(listeners)).

375 + Strs("relays", e.RelayURLs())

376 + if closeErr != nil {

377 + event = log.Warn().

378 + Err(closeErr).

379 + Int("relay_count", len(listeners)).

380 + Strs("relays", e.RelayURLs())

381 }

315 - merged.listeners = append(merged.listeners, listener)

382 + event.Msg("exposure closed")

383 + })

384 + return closeErr

385 +}

386 +

387 +func (e *Exposure) applyRelayURLs(relayURLs []string, failOnError bool) error {

388 + if e == nil || len(relayURLs) == 0 {

389 + return nil

390 }

391

318 - merged.active = len(merged.listeners)

319 - for _, listener := range merged.listeners {

320 - source := listener

321 - go merged.runAcceptLoop(source)

392 + snapshot := append([]string(nil), relayURLs...)

393 +

394 + e.mu.Lock()

395 + if strings.Join(e.relayURLs, "\x00") != strings.Join(snapshot, "\x00") {

396 + e.relayURLs = snapshot

397 }

323 - return merged, nil

398 + e.mu.Unlock()

399 + return e.syncListeners(failOnError)

400 }

401

326 -type mergedListener struct {

327 - listeners []net.Listener

328 - accepted chan net.Conn

329 - closed chan struct{}

402 +func (e *Exposure) syncListeners(failOnError bool) error {

403 + if e == nil {

404 + return nil

405 + }

406

331 - closeOnce sync.Once

332 - mu sync.Mutex

333 - active int

334 - terminalErr error

407 + missing := e.reserveMissingRelayURLs()

408 + for _, relayURL := range missing {

409 + listener, err := e.newListener(relayURL)

410 + if err != nil {

411 + e.mu.Lock()

412 + delete(e.starting, relayURL)

413 + e.mu.Unlock()

414 + if failOnError {

415 + return fmt.Errorf("listen %q: %w", relayURL, err)

416 + }

417 + log.Warn().Err(err).Str("relay_url", relayURL).Msg("add relay listener")

418 + continue

419 + }

420 + e.installListener(relayURL, listener)

421 + }

422 + return nil

423 }

424

337 -func (l *mergedListener) Accept() (net.Conn, error) {

338 - conn, ok := <-l.accepted

339 - if ok {

340 - select {

341 - case <-l.closed:

342 - _ = conn.Close()

343 - return nil, l.terminalErrorOr(net.ErrClosed)

344 - default:

425 +func (e *Exposure) reserveMissingRelayURLs() []string {

426 + e.mu.Lock()

427 + defer e.mu.Unlock()

428 +

429 + missing := make([]string, 0)

430 + for _, relayURL := range e.relayURLs {

431 + if _, ok := e.listeners[relayURL]; ok {

432 + continue

433 + }

434 + if _, ok := e.starting[relayURL]; ok {

435 + continue

436 }

346 - return conn, nil

437 + e.starting[relayURL] = struct{}{}

438 + missing = append(missing, relayURL)

439 }

440 + return missing

441 +}

442

349 - return nil, l.terminalErrorOr(net.ErrClosed)

443 +func (e *Exposure) newListener(relayURL string) (*Listener, error) {

444 + cfg := ListenerConfig{

445 + Name: e.name,

446 + ReverseToken: e.reverseToken,

447 + UDPEnabled: e.udpEnabled,

448 + Discovery: e.discoveryEnabled,

449 + OwnerAddress: e.ownerAddress,

450 + Metadata: e.metadata.Copy(),

451 + RootCAPEM: append([]byte(nil), e.rootCAPEM...),

452 + bootstrapService: e.discovery,

453 + }

454 + return NewListener(e.ctx, relayURL, cfg)

455 }

456

352 -func (l *mergedListener) Close() error {

353 - var closeErr error

354 - l.closeOnce.Do(func() {

355 - close(l.closed)

356 - for _, listener := range l.listeners {

357 - err := listener.Close()

358 - if errors.Is(err, net.ErrClosed) {

359 - err = nil

360 - }

361 - closeErr = errors.Join(closeErr, err)

457 +func (e *Exposure) installListener(relayURL string, listener *Listener) {

458 + if e == nil || listener == nil {

459 + return

460 + }

461 +

462 + shouldClose := false

463 + e.mu.Lock()

464 + delete(e.starting, relayURL)

465 + select {

466 + case <-e.done:

467 + shouldClose = true

468 + default:

469 + if _, exists := e.listeners[relayURL]; exists {

470 + shouldClose = true

471 + } else {

472 + e.listeners[relayURL] = listener

473 }

363 - l.recordTerminalError(closeErr)

364 - })

365 - return closeErr

366 -}

474 + }

475 + e.mu.Unlock()

476

368 -func (l *mergedListener) Addr() net.Addr {

369 - if len(l.listeners) == 1 {

370 - return l.listeners[0].Addr()

477 + if shouldClose {

478 + _ = listener.Close()

479 + return

480 }

481

373 - parts := make([]string, 0, len(l.listeners))

374 - for _, listener := range l.listeners {

375 - parts = append(parts, listener.Addr().String())

482 + log.Info().Str("relay_url", relayURL).Msg("relay added to exposure")

483 + go e.runListenerAcceptLoop(listener)

484 + if e.udpEnabled {

485 + go e.attachDatagramPlane(e.ctx, listener)

486 }

377 - return listenerAddr("merged:" + strings.Join(parts, ","))

487 }

488

380 -func (l *mergedListener) runAcceptLoop(listener net.Listener) {

381 - for {

382 - conn, err := listener.Accept()

383 - if err != nil {

384 - if !errors.Is(err, net.ErrClosed) {

385 - l.recordTerminalError(fmt.Errorf("accept %s: %w", listener.Addr().String(), err))

386 - }

387 - l.finishWorker()

388 - return

389 - }

489 +func (e *Exposure) listenersOrdered() []*Listener {

490 + if e == nil {

491 + return nil

492 + }

493

391 - select {

392 - case <-l.closed:

393 - _ = conn.Close()

394 - l.finishWorker()

395 - return

396 - default:

397 - }

494 + e.mu.RLock()

495 + defer e.mu.RUnlock()

496

399 - select {

400 - case l.accepted <- conn:

401 - case <-l.closed:

402 - _ = conn.Close()

403 - l.finishWorker()

404 - return

497 + out := make([]*Listener, 0, len(e.listeners))

498 + for _, relayURL := range e.relayURLs {

499 + if listener, ok := e.listeners[relayURL]; ok {

500 + out = append(out, listener)

501 }

502 }

503 + return out

504 }

505

409 -func (l *mergedListener) finishWorker() {

410 - l.mu.Lock()

411 - l.active--

412 - last := l.active == 0

413 - if last && l.terminalErr == nil {

414 - l.terminalErr = net.ErrClosed

506 +func (e *Exposure) listenerForRelayURL(relayURL string) *Listener {

507 + if e == nil {

508 + return nil

509 }

416 - l.mu.Unlock()

510

418 - if last {

419 - close(l.accepted)

420 - }

511 + e.mu.RLock()

512 + defer e.mu.RUnlock()

513 + return e.listeners[relayURL]

514 }

515

423 -func (l *mergedListener) recordTerminalError(err error) {

424 - if err == nil {

516 +func (e *Exposure) runListenerAcceptLoop(listener *Listener) {

517 + if e == nil || listener == nil {

518 return

519 }

520

428 - l.mu.Lock()

429 - l.terminalErr = errors.Join(l.terminalErr, err)

430 - l.mu.Unlock()

431 -}

521 + for {

522 + conn, err := listener.Accept()

523 + if err != nil {

524 + if listener.done() || errors.Is(err, net.ErrClosed) {

525 + return

526 + }

527 + log.Warn().Err(err).Str("relay_url", listener.relayURL).Msg("exposure listener accept failed")

528 + return

529 + }

530

433 -func (l *mergedListener) terminalErrorOr(fallback error) error {

434 - l.mu.Lock()

435 - defer l.mu.Unlock()

436 - if l.terminalErr == nil {

437 - return fallback

531 + select {

532 + case <-e.done:

533 + _ = conn.Close()

534 + return

535 + case e.accepted <- conn:

536 + }

537 }

439 - return l.terminalErr

538 }

539

540 type exposureConn struct {

@@ -471,40 +569,6 @@ func (c *exposureConn) Close() error {

569 return closeErr

570 }

571

474 -// Close closes the merged listener and all underlying relay listeners.

475 -func (e *Exposure) Close() error {

476 - if e == nil {

477 - return nil

478 - }

479 -

480 - var closeErr error

481 - e.closeOnce.Do(func() {

482 - if e.done != nil {

483 - close(e.done)

484 - }

485 - if e.listener != nil {

486 - closeErr = errors.Join(closeErr, e.listener.Close())

487 - }

488 - for _, listener := range e.listeners {

489 - if listener != nil {

490 - closeErr = errors.Join(closeErr, listener.Close())

491 - }

492 - }

493 -

494 - event := log.Info().

495 - Int("relay_count", len(e.listeners)).

496 - Strs("relays", e.RelayURLs())

497 - if closeErr != nil {

498 - event = log.Warn().

499 - Err(closeErr).

500 - Int("relay_count", len(e.listeners)).

501 - Strs("relays", e.RelayURLs())

502 - }

503 - event.Msg("exposure closed")

504 - })

505 - return closeErr

506 -}

507 -

572 func (e *Exposure) SupportsDatagram() bool {

573 return e != nil && e.udpEnabled

574 }

@@ -513,41 +577,52 @@ func (e *Exposure) UDPEnabled() bool {

577 return e != nil && e.udpEnabled

578 }

579

516 -func (e *Exposure) AcceptDatagram() (types.DatagramFrame, string, string, string, func([]byte) error, error) {

580 +func (e *Exposure) AcceptDatagram() (types.DatagramFrame, error) {

581 if e == nil || !e.SupportsDatagram() {

518 - return types.DatagramFrame{}, "", "", "", nil, net.ErrClosed

582 + return types.DatagramFrame{}, net.ErrClosed

583 }

584

585 select {

586 case <-e.done:

523 - return types.DatagramFrame{}, "", "", "", nil, net.ErrClosed

524 - case dg := <-e.datagrams:

525 - reply := dg.reply

526 - if reply == nil {

527 - reply = func([]byte) error { return errors.New("reply path is unavailable") }

528 - }

529 - return types.DatagramFrame{

530 - FlowID: dg.FlowID,

531 - Payload: dg.Payload,

532 - }, dg.LeaseID, dg.RelayURL, dg.UDPAddr, reply, nil

587 + return types.DatagramFrame{}, net.ErrClosed

588 + case frame := <-e.datagrams:

589 + return frame, nil

590 }

591 }

592

593 +func (e *Exposure) SendDatagram(frame types.DatagramFrame) error {

594 + if e == nil || !e.SupportsDatagram() {

595 + return net.ErrClosed

596 + }

597 +

598 + relayURL := strings.TrimSpace(frame.RelayURL)

599 + if relayURL == "" {

600 + return errors.New("relay url is required")

601 + }

602 +

603 + listener := e.listenerForRelayURL(relayURL)

604 + if listener == nil {

605 + return net.ErrClosed

606 + }

607 + if leaseID := strings.TrimSpace(frame.LeaseID); leaseID != "" && leaseID != listener.LeaseID() {

608 + return errors.New("datagram frame targets stale lease")

609 + }

610 + return listener.SendDatagram(frame.FlowID, frame.Payload)

611 +}

612 +

613 func (e *Exposure) UDPAddrs() []string {

537 - if e == nil || len(e.listeners) == 0 || !e.SupportsDatagram() {

614 + listeners := e.listenersOrdered()

615 + if len(listeners) == 0 || !e.SupportsDatagram() {

616 return nil

617 }

618

541 - out := make([]string, 0, len(e.listeners))

619 + out := make([]string, 0, len(listeners))

620 seen := make(map[string]struct{})

543 - for _, listener := range e.listeners {

621 + for _, listener := range listeners {

622 if listener == nil {

623 continue

624 }

547 -

548 - listener.mu.Lock()

549 - udpAddr := listener.udpAddr

550 - listener.mu.Unlock()

625 + udpAddr := listener.UDPAddr()

626 if udpAddr == "" {

627 continue

628 }

@@ -590,13 +665,14 @@ func (e *Exposure) WaitDatagramReady(ctx context.Context) ([]string, error) {

665 }

666

667 func (e *Exposure) readyUDPAddrs() []string {

593 - if e == nil || len(e.listeners) == 0 || !e.SupportsDatagram() {

668 + listeners := e.listenersOrdered()

669 + if len(listeners) == 0 || !e.SupportsDatagram() {

670 return nil

671 }

672

597 - out := make([]string, 0, len(e.listeners))

673 + out := make([]string, 0, len(listeners))

674 seen := make(map[string]struct{})

599 - for _, listener := range e.listeners {

675 + for _, listener := range listeners {

676 if listener == nil || !listener.datagramConnected() {

677 continue

678 }

@@ -618,12 +694,13 @@ func (e *Exposure) readyUDPAddrs() []string {

694 }

695

696 func (e *Exposure) allDatagramNegotiationsResolvedWithoutDatagram() bool {

621 - if e == nil || len(e.listeners) == 0 {

697 + listeners := e.listenersOrdered()

698 + if len(listeners) == 0 {

699 return true

700 }

701

702 resolved := 0

626 - for _, listener := range e.listeners {

703 + for _, listener := range listeners {

704 if listener == nil {

705 resolved++

706 continue

@@ -642,17 +719,7 @@ func (e *Exposure) allDatagramNegotiationsResolvedWithoutDatagram() bool {

719 resolved++

720 }

721

645 - return resolved == len(e.listeners)

646 -}

647 -

648 -func (e *Exposure) attachDatagramPlanes(ctx context.Context) {

649 - for _, listener := range e.listeners {

650 - if listener == nil {

651 - continue

652 - }

653 -

654 - go e.attachDatagramPlane(ctx, listener)

655 - }

722 + return resolved == len(listeners)

723 }

724

725 func (e *Exposure) attachDatagramPlane(ctx context.Context, listener *Listener) {

@@ -679,7 +746,7 @@ func (e *Exposure) attachDatagramPlane(ctx context.Context, listener *Listener)

746

747 func (e *Exposure) forwardDatagrams(relayURL string, listener *Listener) {

748 for {

682 - dg, err := listener.AcceptDatagram()

749 + frame, err := listener.AcceptDatagram()

750 if err != nil {

751 if e.closed() || errors.Is(err, net.ErrClosed) {

752 return

@@ -692,22 +759,15 @@ func (e *Exposure) forwardDatagrams(relayURL string, listener *Listener) {

759 return

760 }

761

695 - flowID := dg.FlowID

696 - reply := func(payload []byte) error {

697 - return listener.SendDatagram(flowID, payload)

698 - }

762 + frame.Payload = append([]byte(nil), frame.Payload...)

763 + frame.LeaseID = listener.LeaseID()

764 + frame.RelayURL = relayURL

765 + frame.UDPAddr = listener.UDPAddr()

766

767 select {

768 case <-e.done:

769 return

703 - case e.datagrams <- exposureDatagram{

704 - FlowID: flowID,

705 - LeaseID: listener.LeaseID(),

706 - Payload: append([]byte(nil), dg.Payload...),

707 - RelayURL: relayURL,

708 - UDPAddr: listener.UDPAddr(),

709 - reply: reply,

710 - }:

770 + case e.datagrams <- frame:

771 }

772 }

773 }

@@ -732,27 +792,26 @@ func (e *Exposure) monitorStartupCounts(ctx context.Context) {

792

793 ticker := time.NewTicker(time.Second)

794 defer ticker.Stop()

735 - prevStatuses := make(map[string]listenerStatus, len(e.listeners))

795 + prevStatuses := make(map[string]listenerStatus)

796 firstRun := true

797

798 for {

799 readyCount, inactiveCount := 0, 0

800 activated := make([]string, 0)

801 deactivated := make([]string, 0)

742 - for _, listener := range e.listeners {

743 - status := listenerStatusInactive

744 - if listener != nil {

745 - status = listener.StartupStatus()

802 +

803 + for _, listener := range e.listenersOrdered() {

804 + if listener == nil {

805 + continue

806 }

807 +

808 + status := listener.StartupStatus()

809 if status == listenerStatusReady {

810 readyCount++

811 } else {

812 inactiveCount++

813 }

814

753 - if listener == nil {

754 - continue

755 - }

815 if prev, ok := prevStatuses[listener.relayURL]; ok && prev != status {

816 if status == listenerStatusReady {

817 activated = append(activated, listener.relayURL)

sdk/listener.go

+24 -7

@@ -14,6 +14,7 @@ import (

14 "github.com/quic-go/quic-go"

15 "github.com/rs/zerolog/log"

17 + "github.com/gosuda/portal/v2/portal/discovery"

18 "github.com/gosuda/portal/v2/portal/keyless"

19 "github.com/gosuda/portal/v2/portal/transport"

20 "github.com/gosuda/portal/v2/types"

@@ -24,6 +25,8 @@ type ListenerConfig struct {

25 Name string

26 ReverseToken string

27 UDPEnabled bool

28 + Discovery bool

29 + OwnerAddress string

30 Metadata types.LeaseMetadata

31 RootCAPEM []byte

32 DialTimeout time.Duration

@@ -34,6 +37,8 @@ type ListenerConfig struct {

37 ReadyTarget int

38 RetryCount int

39 RetryWait time.Duration

40 +

41 + bootstrapService *discovery.Service

42 }

44 type listenerStatus string

@@ -55,6 +60,7 @@ type Listener struct {

60 cancel context.CancelFunc

61 api *apiClient

62 relayURL string

63 + bootstrapSvc *discovery.Service

64 startupStatus listenerStatus

65 leaseID string

66 hostname string

@@ -64,7 +70,7 @@ type Listener struct {

70 stream *transport.ClientStream

71 datagram *transport.ClientDatagram

67 - registered chan struct{} // closed after first successful registration

73 + registered chan struct{}

74 closeOnce sync.Once

75 registerOnce sync.Once

76 mu sync.Mutex

@@ -73,6 +79,10 @@ type Listener struct {

79 // NewListener creates one relay listener and its dedicated relay transport for one relay URL.

80 // Only local config validation fails immediately; relay startup runs in the background until ready.

81 func NewListener(ctx context.Context, relayURL string, cfg ListenerConfig) (*Listener, error) {

82 + if ctx == nil {

83 + ctx = context.Background()

84 + }

85 +

86 listenerCtx, cancel := context.WithCancel(ctx)

87 readyTarget := utils.IntOrDefault(cfg.ReadyTarget, defaultReadyTarget)

88 leaseTTL := utils.DurationOrDefault(cfg.LeaseTTL, defaultLeaseTTL)

@@ -92,6 +102,7 @@ func NewListener(ctx context.Context, relayURL string, cfg ListenerConfig) (*Lis

102 api: api,

103 registered: make(chan struct{}),

104 relayURL: api.baseURL.String(),

105 + bootstrapSvc: cfg.bootstrapService,

106 startupStatus: listenerStatusInactive,

107 readyTarget: readyTarget,

108 retryCount: cfg.RetryCount,

@@ -99,6 +110,7 @@ func NewListener(ctx context.Context, relayURL string, cfg ListenerConfig) (*Lis

110 leaseTTL: leaseTTL,

111 renewBefore: renewBefore,

112 udpEnabled: cfg.UDPEnabled,

113 + metadata: cfg.Metadata.Copy(),

114 }

115 l.stream = transport.NewClientStream(readyTarget, handshakeTimeout)

116 if cfg.UDPEnabled {

@@ -442,7 +454,11 @@ func (l *Listener) registerAndConfigure(ctx context.Context) error {

454 return err

455 }

456

445 - resp, err := l.api.registerLease(ctx, l.leaseTTL, l.udpEnabled)

457 + bootstraps := []string(nil)

458 + if l.bootstrapSvc != nil {

459 + bootstraps = l.bootstrapSvc.Bootstraps()

460 + }

461 + resp, err := l.api.registerLease(ctx, l.leaseTTL, l.udpEnabled, bootstraps)

462 if err != nil {

463 return err

464 }

@@ -454,11 +470,7 @@ func (l *Listener) registerAndConfigure(ctx context.Context) error {

470 }

471 }

472

457 - var (

458 - tlsConf *tls.Config

459 - tlsCloser io.Closer

460 - )

461 - tlsConf, tlsCloser, err = keyless.BuildClientTLSConfig(l.api.baseURL.String(), []string{resp.Hostname})

473 + tlsConf, tlsCloser, err := keyless.BuildClientTLSConfig(l.api.baseURL.String(), []string{resp.Hostname})

474 if err != nil {

475 _ = l.api.unregisterLease(context.Background(), resp.LeaseID)

476 return err

@@ -497,6 +509,11 @@ func (l *Listener) registerAndConfigure(ctx context.Context) error {

509 if datagram != nil {

510 datagram.Clear("lease updated")

511 }

512 + if l.bootstrapSvc != nil && len(resp.Bootstraps) > 0 {

513 + if err := l.bootstrapSvc.MergeBootstraps(resp.Bootstraps); err != nil {

514 + log.Warn().Err(err).Strs("bootstraps", resp.Bootstraps).Msg("learn bootstraps from register response")

515 + }

516 + }

517 l.registerOnce.Do(func() { close(l.registered) })

518 return nil

519 }

sdk/sdk_test.go

+170

@@ -6,10 +6,12 @@ import (

6 "errors"

7 "net/http"

8 "net/http/httptest"

9 + "strings"

10 "sync/atomic"

11 "testing"

12 "time"

14 + "github.com/gosuda/portal/v2/portal/discovery"

15 "github.com/gosuda/portal/v2/types"

16 )

@@ -360,6 +362,174 @@ func TestExposeNoRelayInputs(t *testing.T) {

362 }

363 }

364

365 +func TestExposeRegistersKnownRelayURLs(t *testing.T) {

366 + registerReqCh := make(chan types.RegisterRequest, 2)

367 + newRelayServer := func() *httptest.Server {

368 + return httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

369 + switch r.URL.Path {

370 + case types.PathSDKDomain:

371 + writeSDKTestEnvelope(w, http.StatusOK, types.APIEnvelope[types.DomainResponse]{

372 + OK: true,

373 + Data: types.DomainResponse{

374 + Version: types.SDKProtocolVersion,

375 + },

376 + })

377 + case types.PathSDKRegister:

378 + var registerReq types.RegisterRequest

379 + if err := json.NewDecoder(r.Body).Decode(&registerReq); err != nil {

380 + t.Fatalf("decode register request: %v", err)

381 + }

382 + select {

383 + case registerReqCh <- registerReq:

384 + default:

385 + }

386 + writeSDKTestEnvelope(w, http.StatusCreated, types.APIEnvelope[types.RegisterResponse]{

387 + OK: true,

388 + Data: types.RegisterResponse{

389 + LeaseID: "lease-1",

390 + Hostname: "127.0.0.1",

391 + },

392 + })

393 + case types.PathSDKConnect:

394 + writeSDKTestEnvelope(w, http.StatusForbidden, types.APIEnvelope[any]{

395 + OK: false,

396 + Error: &types.APIError{Code: types.APIErrorCodeUnauthorized, Message: "not used in test"},

397 + })

398 + case types.PathSDKRenew:

399 + writeSDKTestEnvelope(w, http.StatusOK, types.APIEnvelope[types.RenewResponse]{

400 + OK: true,

401 + Data: types.RenewResponse{LeaseID: "lease-1"},

402 + })

403 + case types.PathSDKUnregister:

404 + writeSDKTestEnvelope(w, http.StatusOK, types.APIEnvelope[any]{OK: true})

405 + default:

406 + http.NotFound(w, r)

407 + }

408 + }))

409 + }

410 +

411 + relayA := newRelayServer()

412 + defer relayA.Close()

413 + relayB := newRelayServer()

414 + defer relayB.Close()

415 +

416 + exposure, err := ExposeWithConfig(context.Background(), ExposeConfig{

417 + RelayURLs: []string{relayA.URL, relayB.URL},

418 + Discovery: true,

419 + Name: "demo",

420 + OwnerAddress: "0x52908400098527886E0F7030069857D2E4169EE7",

421 + })

422 + if err != nil {

423 + t.Fatalf("ExposeWithConfig() error = %v", err)

424 + }

425 + defer exposure.Close()

426 +

427 + var requests []types.RegisterRequest

428 + waitForSDKTest(t, func() bool {

429 + for len(requests) < 2 {

430 + select {

431 + case req := <-registerReqCh:

432 + requests = append(requests, req)

433 + default:

434 + return false

435 + }

436 + }

437 + return true

438 + })

439 +

440 + for _, req := range requests {

441 + if req.OwnerAddress != "0x52908400098527886E0F7030069857D2E4169EE7" {

442 + t.Fatalf("register request OwnerAddress = %q, want configured owner address", req.OwnerAddress)

443 + }

444 + if len(req.Bootstraps) != 2 || req.Bootstraps[0] != relayA.URL || req.Bootstraps[1] != relayB.URL {

445 + t.Fatalf("register request Bootstraps = %v, want [%q %q]", req.Bootstraps, relayA.URL, relayB.URL)

446 + }

447 + }

448 +}

449 +

450 +func TestExposeResolvesOwnerPrivateKey(t *testing.T) {

451 + ownerPrivateKey := strings.Repeat("11", 32)

452 + identity, err := discovery.ResolveIdentity(ownerPrivateKey)

453 + if err != nil {

454 + t.Fatalf("ResolveIdentity() error = %v", err)

455 + }

456 +

457 + registerReqCh := make(chan types.RegisterRequest, 1)

458 + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

459 + switch r.URL.Path {

460 + case types.PathSDKDomain:

461 + writeSDKTestEnvelope(w, http.StatusOK, types.APIEnvelope[types.DomainResponse]{

462 + OK: true,

463 + Data: types.DomainResponse{

464 + Version: types.SDKProtocolVersion,

465 + },

466 + })

467 + case types.PathSDKRegister:

468 + var registerReq types.RegisterRequest

469 + if err := json.NewDecoder(r.Body).Decode(&registerReq); err != nil {

470 + t.Fatalf("decode register request: %v", err)

471 + }

472 + select {

473 + case registerReqCh <- registerReq:

474 + default:

475 + }

476 + writeSDKTestEnvelope(w, http.StatusCreated, types.APIEnvelope[types.RegisterResponse]{

477 + OK: true,

478 + Data: types.RegisterResponse{

479 + LeaseID: "lease-1",

480 + Hostname: "127.0.0.1",

481 + },

482 + })

483 + case types.PathSDKConnect:

484 + writeSDKTestEnvelope(w, http.StatusForbidden, types.APIEnvelope[any]{

485 + OK: false,

486 + Error: &types.APIError{Code: types.APIErrorCodeUnauthorized, Message: "not used in test"},

487 + })

488 + case types.PathSDKRenew:

489 + writeSDKTestEnvelope(w, http.StatusOK, types.APIEnvelope[types.RenewResponse]{

490 + OK: true,

491 + Data: types.RenewResponse{LeaseID: "lease-1"},

492 + })

493 + case types.PathSDKUnregister:

494 + writeSDKTestEnvelope(w, http.StatusOK, types.APIEnvelope[any]{OK: true})

495 + default:

496 + http.NotFound(w, r)

497 + }

498 + }))

499 + defer server.Close()

500 +

501 + exposure, err := ExposeWithConfig(context.Background(), ExposeConfig{

502 + RelayURLs: []string{server.URL},

503 + Name: "demo",

504 + OwnerPrivateKey: &ownerPrivateKey,

505 + })

506 + if err != nil {

507 + t.Fatalf("ExposeWithConfig() error = %v", err)

508 + }

509 + defer exposure.Close()

510 +

511 + var registerReq types.RegisterRequest

512 + waitForSDKTest(t, func() bool {

513 + select {

514 + case registerReq = <-registerReqCh:

515 + return true

516 + default:

517 + return false

518 + }

519 + })

520 +

521 + if registerReq.OwnerAddress != identity.Address {

522 + t.Fatalf("register request OwnerAddress = %q, want %q", registerReq.OwnerAddress, identity.Address)

523 + }

524 + resolvedIdentity := exposure.OwnerIdentity()

525 + if resolvedIdentity.Address != identity.Address {

526 + t.Fatalf("OwnerIdentity().Address = %q, want %q", resolvedIdentity.Address, identity.Address)

527 + }

528 + if resolvedIdentity.PrivateKey != ownerPrivateKey {

529 + t.Fatalf("OwnerIdentity().PrivateKey = %q, want configured private key", resolvedIdentity.PrivateKey)

530 + }

531 +}

532 +

533 func TestNewListenerUDPEnabledKeepsStreamAndDatagram(t *testing.T) {

534 var connectCount atomic.Int32

535 server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

types/api.go

+17

@@ -57,7 +57,9 @@ type RegisterRequest struct {

57 Name string `json:"name"`

58 ReverseToken string `json:"reverse_token"`

59 Metadata LeaseMetadata `json:"metadata"`

60 + OwnerAddress string `json:"owner_address,omitempty"`

61 TTL int `json:"ttl,omitempty"`

62 + Bootstraps []string `json:"bootstraps,omitempty"`

63 UDPEnabled bool `json:"udp_enabled,omitempty"`

64 }

@@ -67,10 +69,25 @@ type RegisterResponse struct {

69 ConnectURL string `json:"connect_url"`

70 Hostname string `json:"hostname"`

71 Metadata LeaseMetadata `json:"metadata"`

72 + Bootstraps []string `json:"bootstraps,omitempty"`

73 UDPAddr string `json:"udp_addr,omitempty"`

74 UDPEnabled bool `json:"udp_enabled,omitempty"`

75 }

77 +type DiscoverRequest struct {

78 + RootHost string `json:"root_host"`

79 + Name string `json:"name"`

80 +}

81 +

82 +type DiscoverResponse struct {

83 + Found bool `json:"found"`

84 + Name string `json:"name,omitempty"`

85 + Hostname string `json:"hostname,omitempty"`

86 + ExpiresAt time.Time `json:"expires_at,omitempty"`

87 + OwnerAddress string `json:"owner_address,omitempty"`

88 + Bootstraps []string `json:"bootstraps,omitempty"`

89 +}

90 +

91 type QUICControlMessage struct {

92 LeaseID string `json:"lease_id"`

93 ReverseToken string `json:"reverse_token"`

types/error.go

@@ -13,6 +13,7 @@ const (

13 APIErrorCodeInvalidLeaseID = "invalid_lease_id"

14 APIErrorCodeInvalidMode = "invalid_mode"

15 APIErrorCodeInvalidRequest = "invalid_request"

16 + APIErrorCodeInternal = "internal"

17 APIErrorCodeIPBanned = "ip_banned"

18 APIErrorCodeLeaseNotFound = "lease_not_found"

19 APIErrorCodeLeaseRejected = "lease_rejected"

types/lease.go

+18 -16

@@ -21,20 +21,22 @@ func (m LeaseMetadata) Copy() LeaseMetadata {

21 }

23 type Lease struct {

24 - ExpiresAt time.Time

25 - FirstSeenAt time.Time

26 - LastSeenAt time.Time

27 - ID string

28 - Name string

29 - BPS int64

30 - ClientIP string

31 - Hostname string

32 - UDPEnabled bool

33 - Metadata LeaseMetadata

34 - Ready int

35 - UDPPort int

36 - IsApproved bool

37 - IsBanned bool

38 - IsDenied bool

39 - IsIPBanned bool

24 + ExpiresAt time.Time

25 + FirstSeenAt time.Time

26 + LastSeenAt time.Time

27 + ID string

28 + Name string

29 + BPS int64

30 + ClientIP string

31 + Hostname string

32 + Bootstraps []string

33 + UDPEnabled bool

34 + Metadata LeaseMetadata

35 + OwnerAddress string

36 + Ready int

37 + UDPPort int

38 + IsApproved bool

39 + IsBanned bool

40 + IsDenied bool

41 + IsIPBanned bool

42 }

types/paths.go

@@ -28,4 +28,5 @@ const (

28 PathSDKRenew = "/sdk/renew"

29 PathSDKUnregister = "/sdk/unregister"

30 PathSDKConnect = "/sdk/connect"

31 + PathDiscovery = "/discovery"

32 )

types/transport.go

+8 -4

@@ -9,11 +9,15 @@ import (

9 // contain a valid flow ID varint.

10 var ErrDatagramTooSmall = errors.New("datagram too small to decode")

12 -// DatagramFrame is the wire format for QUIC DATAGRAM payloads.

13 -// Layout: [flowID varint][payload bytes]

12 +// DatagramFrame carries one relayed datagram.

13 +// Wire encoding uses only FlowID and Payload with layout:

14 +// [flowID varint][payload bytes]

15 type DatagramFrame struct {

15 - FlowID uint32

16 - Payload []byte

16 + FlowID uint32

17 + Payload []byte

18 + LeaseID string

19 + RelayURL string

20 + UDPAddr string

21 }

23 // EncodeDatagram serialises a flow-framed datagram for transmission.

utils/utils.go

+46 -7

@@ -132,7 +132,6 @@ func NormalizeHostname(host string) string {

132

133 func NormalizeRelayURLs(inputs []string) ([]string, error) {

134 out := make([]string, 0, len(inputs))

135 - seen := make(map[string]struct{}, len(inputs))

135

136 for _, input := range inputs {

137 for _, part := range SplitCSV(input) {

@@ -140,18 +139,58 @@ func NormalizeRelayURLs(inputs []string) ([]string, error) {

139 if err != nil {

140 return nil, err

141 }

143 - if _, ok := seen[normalized]; ok {

144 - continue

145 - }

146 - seen[normalized] = struct{}{}

142 out = append(out, normalized)

143 }

144 }

145

146 + return UniqueURLs(out), nil

147 +}

148 +

149 +func UniqueURLs(inputs []string) []string {

150 + if len(inputs) == 0 {

151 + return nil

152 + }

153 +

154 + out := make([]string, 0, len(inputs))

155 + seen := make(map[string]struct{}, len(inputs))

156 + for _, input := range inputs {

157 + input = strings.TrimSpace(input)

158 + if input == "" {

159 + continue

160 + }

161 + if _, ok := seen[input]; ok {

162 + continue

163 + }

164 + seen[input] = struct{}{}

165 + out = append(out, input)

166 + }

167 if len(out) == 0 {

152 - return nil, nil

168 + return nil

169 + }

170 + return out

171 +}

172 +

173 +func ExcludeURLs(inputs []string, excluded []string) []string {

174 + if len(inputs) == 0 {

175 + return nil

176 + }

177 + if len(excluded) == 0 {

178 + return UniqueURLs(inputs)

179 + }

180 +

181 + skip := make(map[string]struct{}, len(excluded))

182 + for _, input := range UniqueURLs(excluded) {

183 + skip[input] = struct{}{}

184 + }

185 +

186 + filtered := make([]string, 0, len(inputs))

187 + for _, input := range inputs {

188 + if _, ok := skip[input]; ok {

189 + continue

190 + }

191 + filtered = append(filtered, input)

192 }

154 - return out, nil

193 + return UniqueURLs(filtered)

194 }

195

196 func LeaseHostname(name, rootHost string) (string, error) {