| 1 | package sdk |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "net" |
| 8 | "net/http" |
| 9 | "slices" |
| 10 | "strings" |
| 11 | "sync" |
| 12 | "sync/atomic" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/rs/zerolog/log" |
| 16 | |
| 17 | "github.com/gosuda/portal-tunnel/v2/portal/discovery" |
| 18 | "github.com/gosuda/portal-tunnel/v2/portal/identity" |
| 19 | "github.com/gosuda/portal-tunnel/v2/portal/telemetry" |
| 20 | "github.com/gosuda/portal-tunnel/v2/types" |
| 21 | "github.com/gosuda/portal-tunnel/v2/utils" |
| 22 | ) |
| 23 | |
| 24 | // Exposure owns the lifecycle of one or more relay listeners and accepts |
| 25 | // traffic from all of them through one net.Listener. |
| 26 | type Exposure struct { |
| 27 | cancel context.CancelFunc |
| 28 | done <-chan struct{} |
| 29 | |
| 30 | cfg *utils.Snapshot[ExposeConfig] |
| 31 | |
| 32 | accepted chan net.Conn |
| 33 | datagrams chan types.DatagramFrame |
| 34 | |
| 35 | relaySet *discovery.RelaySet |
| 36 | mu sync.RWMutex |
| 37 | relayListeners map[string]*listener |
| 38 | |
| 39 | closeOnce sync.Once |
| 40 | connSeq atomic.Uint64 |
| 41 | } |
| 42 | |
| 43 | type ExposeConfig struct { |
| 44 | RelayURLs []string |
| 45 | Discovery bool |
| 46 | |
| 47 | Identity types.Identity |
| 48 | IdentityPath string |
| 49 | IdentityJSON string |
| 50 | TargetAddr string |
| 51 | UDPAddr string |
| 52 | UDPEnabled bool |
| 53 | TCPEnabled bool |
| 54 | MultiHop []string |
| 55 | MultiHopDepth int |
| 56 | BanMITM bool |
| 57 | MaxActiveRelays int |
| 58 | Metadata types.LeaseMetadata |
| 59 | X402PayTo string |
| 60 | X402Testnet bool |
| 61 | } |
| 62 | |
| 63 | func (cfg ExposeConfig) snapshot() ExposeConfig { |
| 64 | cfg.RelayURLs = utils.CloneSlice(cfg.RelayURLs) |
| 65 | cfg.Identity = cfg.Identity.Copy() |
| 66 | cfg.MultiHop = utils.CloneSlice(cfg.MultiHop) |
| 67 | cfg.Metadata = cfg.Metadata.Copy() |
| 68 | cfg.X402PayTo = strings.TrimSpace(cfg.X402PayTo) |
| 69 | return cfg |
| 70 | } |
| 71 | |
| 72 | // Expose creates relay listeners for the selected relay pool and exposes a |
| 73 | // dynamic listener hub for accepting traffic from all of them. |
| 74 | func Expose(ctx context.Context, cfg ExposeConfig) (*Exposure, error) { |
| 75 | explicitRelayURLs, err := utils.NormalizeRelayURLs(cfg.RelayURLs...) |
| 76 | if err != nil { |
| 77 | return nil, err |
| 78 | } |
| 79 | var multiHop []string |
| 80 | for _, input := range cfg.MultiHop { |
| 81 | relayURL, err := utils.NormalizeRelayURL(input) |
| 82 | if err != nil { |
| 83 | return nil, fmt.Errorf("normalize multi-hop relay url: %w", err) |
| 84 | } |
| 85 | if slices.Contains(multiHop, relayURL) { |
| 86 | return nil, fmt.Errorf("multi-hop relay url repeated: %s", relayURL) |
| 87 | } |
| 88 | multiHop = append(multiHop, relayURL) |
| 89 | } |
| 90 | if len(multiHop) == 1 { |
| 91 | return nil, errors.New("multi-hop requires at least entry and exit relay urls") |
| 92 | } |
| 93 | if cfg.MultiHopDepth < 0 { |
| 94 | return nil, errors.New("multi-hop-depth cannot be negative") |
| 95 | } |
| 96 | if len(multiHop) > 0 && cfg.MultiHopDepth > 1 { |
| 97 | return nil, errors.New("explicit --multi-hop cannot be combined with automatic --multi-hop-depth") |
| 98 | } |
| 99 | if (len(multiHop) > 0 || cfg.MultiHopDepth > 1) && (cfg.UDPEnabled || cfg.TCPEnabled) { |
| 100 | return nil, errors.New("multi-hop currently supports only the default SNI TLS stream transport") |
| 101 | } |
| 102 | x402PayTo := strings.TrimSpace(cfg.X402PayTo) |
| 103 | |
| 104 | var initialRouteCount int |
| 105 | var relaySetURLs []string |
| 106 | if len(multiHop) > 0 { |
| 107 | initialRouteCount = 1 |
| 108 | relaySetURLs = append([]string(nil), multiHop...) |
| 109 | } else if cfg.MultiHopDepth > 1 { |
| 110 | initialRouteCount = 1 |
| 111 | relaySetURLs, err = utils.ResolvePortalRelayURLs(explicitRelayURLs, cfg.Discovery) |
| 112 | if err != nil { |
| 113 | return nil, err |
| 114 | } |
| 115 | } else { |
| 116 | relaySetURLs, err = utils.ResolvePortalRelayURLs(explicitRelayURLs, cfg.Discovery) |
| 117 | if err != nil { |
| 118 | return nil, err |
| 119 | } |
| 120 | initialRouteCount = len(explicitRelayURLs) |
| 121 | } |
| 122 | |
| 123 | listenerIdentity, createdIdentity, err := identity.ResolveListenerIdentity( |
| 124 | cfg.Identity.Copy(), |
| 125 | cfg.TargetAddr, |
| 126 | cfg.IdentityPath, |
| 127 | cfg.IdentityJSON, |
| 128 | ) |
| 129 | if err != nil { |
| 130 | return nil, fmt.Errorf("resolve identity: %w", err) |
| 131 | } |
| 132 | if createdIdentity { |
| 133 | log.Info(). |
| 134 | Str("identity_path", strings.TrimSpace(cfg.IdentityPath)). |
| 135 | Str("address", listenerIdentity.Address). |
| 136 | Msg("generated tunnel identity and saved it to disk") |
| 137 | } |
| 138 | targetAddr, err := utils.NormalizeLoopbackTarget(cfg.TargetAddr) |
| 139 | if err != nil { |
| 140 | return nil, fmt.Errorf("invalid target value %q: %w", cfg.TargetAddr, err) |
| 141 | } |
| 142 | udpAddr := cfg.UDPAddr |
| 143 | if cfg.UDPEnabled { |
| 144 | udpAddr, err = utils.NormalizeLoopbackTarget(utils.StringOrDefault(udpAddr, targetAddr)) |
| 145 | if err != nil { |
| 146 | return nil, fmt.Errorf("invalid --udp-addr value %q: %w", cfg.UDPAddr, err) |
| 147 | } |
| 148 | } |
| 149 | runtimeCfg := cfg.snapshot() |
| 150 | runtimeCfg.RelayURLs = append([]string(nil), explicitRelayURLs...) |
| 151 | runtimeCfg.Identity = listenerIdentity.Copy() |
| 152 | runtimeCfg.TargetAddr = targetAddr |
| 153 | runtimeCfg.UDPAddr = udpAddr |
| 154 | runtimeCfg.MultiHop = append([]string(nil), multiHop...) |
| 155 | runtimeCfg.Metadata = cfg.Metadata.Copy() |
| 156 | runtimeCfg.X402PayTo = x402PayTo |
| 157 | runtimeCfg.X402Testnet = cfg.X402Testnet |
| 158 | |
| 159 | exposureCtx, cancel := context.WithCancel(ctx) |
| 160 | exposure := &Exposure{ |
| 161 | cancel: cancel, |
| 162 | done: exposureCtx.Done(), |
| 163 | cfg: utils.NewSnapshot(runtimeCfg, ExposeConfig.snapshot), |
| 164 | accepted: make(chan net.Conn, max(initialRouteCount*defaultReadyTarget*2, 1)), |
| 165 | datagrams: make(chan types.DatagramFrame, max(initialRouteCount*32, 1)), |
| 166 | relaySet: discovery.NewRelaySet(relaySetURLs), |
| 167 | relayListeners: make(map[string]*listener, initialRouteCount), |
| 168 | } |
| 169 | |
| 170 | if cfg.Discovery || len(multiHop) > 0 || cfg.MultiHopDepth > 1 { |
| 171 | refresher := discovery.NewRefresher(exposure.relaySet, nil) |
| 172 | if err := refresher.Refresh(ctx, nil); err != nil { |
| 173 | _ = exposure.Close() |
| 174 | return nil, fmt.Errorf("discover relays: %w", err) |
| 175 | } |
| 176 | } |
| 177 | |
| 178 | if initialRouteCount > 0 || cfg.Discovery { |
| 179 | if err := exposure.reconcileRelayListeners(true); err != nil { |
| 180 | _ = exposure.Close() |
| 181 | return nil, err |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | if cfg.Discovery || len(multiHop) > 0 || cfg.MultiHopDepth > 1 { |
| 186 | go exposure.runDiscoveryLoop(exposureCtx) |
| 187 | } |
| 188 | |
| 189 | go func() { |
| 190 | <-exposure.done |
| 191 | _ = exposure.Close() |
| 192 | }() |
| 193 | |
| 194 | return exposure, nil |
| 195 | } |
| 196 | |
| 197 | // AddRelay attaches an explicit relay to the running exposure without |
| 198 | // restarting the local tunnel. |
| 199 | func (e *Exposure) AddRelay(relayURL string) error { |
| 200 | relayURL, err := utils.NormalizeRelayURL(relayURL) |
| 201 | if err != nil { |
| 202 | return err |
| 203 | } |
| 204 | if e.closed() { |
| 205 | return net.ErrClosed |
| 206 | } |
| 207 | if e.relaySet == nil { |
| 208 | return errors.New("exposure relay set is not initialized") |
| 209 | } |
| 210 | |
| 211 | e.cfg.UpdateCopy(func(cfg *ExposeConfig) { |
| 212 | if !slices.Contains(cfg.RelayURLs, relayURL) { |
| 213 | cfg.RelayURLs = append(cfg.RelayURLs, relayURL) |
| 214 | } |
| 215 | }) |
| 216 | |
| 217 | e.relaySet.AllowRelayURL(relayURL) |
| 218 | e.relaySet.AddBootstrapRelayURL(relayURL) |
| 219 | return e.reconcileRelayListeners(true) |
| 220 | } |
| 221 | |
| 222 | // RemoveRelay detaches a relay from the running exposure and lets it fall back |
| 223 | // to the discovered candidate pool. |
| 224 | func (e *Exposure) RemoveRelay(relayURL string) error { |
| 225 | relayURL, err := utils.NormalizeRelayURL(relayURL) |
| 226 | if err != nil { |
| 227 | return err |
| 228 | } |
| 229 | if e.closed() { |
| 230 | return net.ErrClosed |
| 231 | } |
| 232 | if e.relaySet == nil { |
| 233 | return errors.New("exposure relay set is not initialized") |
| 234 | } |
| 235 | |
| 236 | if _, ok := e.cfg.UpdateIf(func(cfg ExposeConfig) (ExposeConfig, bool) { |
| 237 | if slices.Contains(cfg.MultiHop, relayURL) { |
| 238 | return cfg, false |
| 239 | } |
| 240 | nextRelays := cfg.RelayURLs[:0] |
| 241 | for _, existing := range cfg.RelayURLs { |
| 242 | if existing != relayURL { |
| 243 | nextRelays = append(nextRelays, existing) |
| 244 | } |
| 245 | } |
| 246 | cfg.RelayURLs = nextRelays |
| 247 | return cfg, true |
| 248 | }); !ok { |
| 249 | return errors.New("relay is part of the multi-hop route; clear multi-hop first") |
| 250 | } |
| 251 | |
| 252 | e.relaySet.DeactivateRelayURL(relayURL) |
| 253 | e.relaySet.RemoveBootstrapRelayURL(relayURL) |
| 254 | return e.reconcileRelayListeners(false) |
| 255 | } |
| 256 | |
| 257 | func (e *Exposure) SetMultiHop(relayURLs []string) error { |
| 258 | multiHop := make([]string, 0, len(relayURLs)) |
| 259 | for _, input := range relayURLs { |
| 260 | relayURL, err := utils.NormalizeRelayURL(input) |
| 261 | if err != nil { |
| 262 | return fmt.Errorf("normalize multi-hop relay url: %w", err) |
| 263 | } |
| 264 | if slices.Contains(multiHop, relayURL) { |
| 265 | return fmt.Errorf("multi-hop relay url repeated: %s", relayURL) |
| 266 | } |
| 267 | multiHop = append(multiHop, relayURL) |
| 268 | } |
| 269 | if len(multiHop) == 1 { |
| 270 | return errors.New("multi-hop requires at least entry and exit relay urls") |
| 271 | } |
| 272 | cfg := e.Config() |
| 273 | if len(multiHop) > 0 && (cfg.UDPEnabled || cfg.TCPEnabled) { |
| 274 | return errors.New("multi-hop currently supports only the default SNI TLS stream transport") |
| 275 | } |
| 276 | if e.closed() { |
| 277 | return net.ErrClosed |
| 278 | } |
| 279 | if e.relaySet == nil { |
| 280 | return errors.New("exposure relay set is not initialized") |
| 281 | } |
| 282 | |
| 283 | for _, relayURL := range multiHop { |
| 284 | e.relaySet.AllowRelayURL(relayURL) |
| 285 | e.relaySet.AddBootstrapRelayURL(relayURL) |
| 286 | } |
| 287 | |
| 288 | e.cfg.UpdateCopy(func(cfg *ExposeConfig) { |
| 289 | cfg.MultiHop = append([]string(nil), multiHop...) |
| 290 | cfg.MultiHopDepth = 0 |
| 291 | }) |
| 292 | return e.reconcileRelayListeners(false) |
| 293 | } |
| 294 | |
| 295 | func (e *Exposure) UpdateMetadata(metadata types.LeaseMetadata) error { |
| 296 | if e.closed() { |
| 297 | return net.ErrClosed |
| 298 | } |
| 299 | |
| 300 | e.cfg.UpdateCopy(func(cfg *ExposeConfig) { |
| 301 | cfg.Metadata = metadata.Copy() |
| 302 | }) |
| 303 | return nil |
| 304 | } |
| 305 | |
| 306 | func (e *Exposure) UpdateMaxActiveRelays(maxActiveRelays int) error { |
| 307 | if maxActiveRelays <= 0 { |
| 308 | return errors.New("max_active_relays must be a positive integer") |
| 309 | } |
| 310 | if e.closed() { |
| 311 | return net.ErrClosed |
| 312 | } |
| 313 | |
| 314 | _, changed := e.cfg.UpdateIf(func(cfg ExposeConfig) (ExposeConfig, bool) { |
| 315 | if cfg.MaxActiveRelays == maxActiveRelays { |
| 316 | return cfg, false |
| 317 | } |
| 318 | cfg.MaxActiveRelays = maxActiveRelays |
| 319 | return cfg, true |
| 320 | }) |
| 321 | if !changed { |
| 322 | return nil |
| 323 | } |
| 324 | return e.reconcileRelayListeners(false) |
| 325 | } |
| 326 | |
| 327 | func (e *Exposure) ActiveRelayURLs() []string { |
| 328 | e.mu.RLock() |
| 329 | defer e.mu.RUnlock() |
| 330 | relayURLs := make([]string, 0, len(e.relayListeners)) |
| 331 | for relayURL := range e.relayListeners { |
| 332 | relayURLs = append(relayURLs, relayURL) |
| 333 | } |
| 334 | slices.Sort(relayURLs) |
| 335 | return relayURLs |
| 336 | } |
| 337 | |
| 338 | func (e *Exposure) closed() bool { |
| 339 | select { |
| 340 | case <-e.done: |
| 341 | return true |
| 342 | default: |
| 343 | return false |
| 344 | } |
| 345 | } |
| 346 | |
| 347 | func (e *Exposure) Addr() net.Addr { |
| 348 | identity := e.Config().Identity |
| 349 | if identity.Address == "" { |
| 350 | return exposureAddr("portal:exposure") |
| 351 | } |
| 352 | return exposureAddr("portal:" + identity.Address) |
| 353 | } |
| 354 | |
| 355 | type exposureAddr string |
| 356 | |
| 357 | func (a exposureAddr) Network() string { return "portal" } |
| 358 | func (a exposureAddr) String() string { return string(a) } |
| 359 | |
| 360 | func (e *Exposure) Config() ExposeConfig { |
| 361 | if e == nil || e.cfg == nil { |
| 362 | return ExposeConfig{} |
| 363 | } |
| 364 | return e.cfg.Load() |
| 365 | } |
| 366 | |
| 367 | func (e *Exposure) Identity() types.Identity { |
| 368 | return e.Config().Identity |
| 369 | } |
| 370 | |
| 371 | func (e *Exposure) Snapshot() types.AgentTunnelStatus { |
| 372 | cfg := e.Config() |
| 373 | e.mu.RLock() |
| 374 | listeners := make([]*listener, 0, len(e.relayListeners)) |
| 375 | for _, listener := range e.relayListeners { |
| 376 | if listener != nil { |
| 377 | listeners = append(listeners, listener) |
| 378 | } |
| 379 | } |
| 380 | e.mu.RUnlock() |
| 381 | |
| 382 | relayByURL := make(map[string]types.AgentRelayStatus, len(listeners)) |
| 383 | for _, listener := range listeners { |
| 384 | relayURL := "" |
| 385 | if listener.relayURL != nil { |
| 386 | relayURL = listener.relayURL.String() |
| 387 | } |
| 388 | explicit := slices.Contains(cfg.RelayURLs, relayURL) |
| 389 | snap := types.AgentRelayStatus{ |
| 390 | RelayURL: relayURL, |
| 391 | Version: listener.releaseVersion, |
| 392 | Explicit: explicit, |
| 393 | Connecting: explicit || len(listener.route.MultiHop()) > 0, |
| 394 | } |
| 395 | if lease, ok := listener.leaseSnapshot(); ok { |
| 396 | snap.PublicURL = listener.publicURLForLease(lease) |
| 397 | snap.Connecting = snap.PublicURL == "" |
| 398 | } |
| 399 | if relayURL != "" { |
| 400 | relayByURL[relayURL] = snap |
| 401 | } |
| 402 | } |
| 403 | if e.relaySet != nil { |
| 404 | for _, state := range e.relaySet.AllRelays() { |
| 405 | relay := state.Descriptor |
| 406 | relayURL := strings.TrimSpace(relay.APIHTTPSAddr) |
| 407 | if relayURL == "" { |
| 408 | continue |
| 409 | } |
| 410 | snap := relayByURL[relayURL] |
| 411 | snap.RelayURL = relayURL |
| 412 | snap.Explicit = slices.Contains(cfg.RelayURLs, relayURL) |
| 413 | snap.Bootstrap = state.Bootstrap |
| 414 | snap.Banned = state.Banned |
| 415 | snap.SupportsOverlay = relay.SupportsOverlay |
| 416 | snap.SupportsUDP = relay.SupportsUDP |
| 417 | snap.SupportsTCP = relay.SupportsTCP |
| 418 | relayByURL[relayURL] = snap |
| 419 | } |
| 420 | } |
| 421 | relays := make([]types.AgentRelayStatus, 0, len(relayByURL)) |
| 422 | for _, snap := range relayByURL { |
| 423 | relays = append(relays, snap) |
| 424 | } |
| 425 | slices.SortFunc(relays, func(a, b types.AgentRelayStatus) int { |
| 426 | aReady := a.PublicURL != "" |
| 427 | bReady := b.PublicURL != "" |
| 428 | if aReady != bReady { |
| 429 | if aReady { |
| 430 | return -1 |
| 431 | } |
| 432 | return 1 |
| 433 | } |
| 434 | if a.Connecting != b.Connecting { |
| 435 | if a.Connecting { |
| 436 | return -1 |
| 437 | } |
| 438 | return 1 |
| 439 | } |
| 440 | return strings.Compare(a.RelayURL, b.RelayURL) |
| 441 | }) |
| 442 | |
| 443 | return types.AgentTunnelStatus{ |
| 444 | Address: cfg.Identity.Address, |
| 445 | TargetAddr: cfg.TargetAddr, |
| 446 | MaxActiveRelays: cfg.MaxActiveRelays, |
| 447 | Metadata: cfg.Metadata, |
| 448 | MultiHop: cfg.MultiHop, |
| 449 | Relays: relays, |
| 450 | } |
| 451 | } |
| 452 | |
| 453 | func (e *Exposure) AcceptDatagram() (types.DatagramFrame, error) { |
| 454 | if !e.Config().UDPEnabled { |
| 455 | return types.DatagramFrame{}, net.ErrClosed |
| 456 | } |
| 457 | |
| 458 | select { |
| 459 | case <-e.done: |
| 460 | return types.DatagramFrame{}, net.ErrClosed |
| 461 | case frame := <-e.datagrams: |
| 462 | return frame, nil |
| 463 | } |
| 464 | } |
| 465 | |
| 466 | func (e *Exposure) SendDatagram(frame types.DatagramFrame) error { |
| 467 | if !e.Config().UDPEnabled { |
| 468 | return net.ErrClosed |
| 469 | } |
| 470 | |
| 471 | e.mu.RLock() |
| 472 | listener := e.relayListeners[frame.RelayURL] |
| 473 | e.mu.RUnlock() |
| 474 | if listener == nil { |
| 475 | return net.ErrClosed |
| 476 | } |
| 477 | return listener.sendDatagram(frame) |
| 478 | } |
| 479 | |
| 480 | func (e *Exposure) WaitDatagramReady(ctx context.Context) ([]string, error) { |
| 481 | if !e.Config().UDPEnabled { |
| 482 | return nil, errors.New("exposure does not have udp enabled") |
| 483 | } |
| 484 | |
| 485 | ticker := time.NewTicker(50 * time.Millisecond) |
| 486 | defer ticker.Stop() |
| 487 | |
| 488 | for { |
| 489 | e.mu.RLock() |
| 490 | addrs := make([]string, 0, len(e.relayListeners)) |
| 491 | seen := make(map[string]struct{}) |
| 492 | resolvedWithoutDatagram := true |
| 493 | for _, listener := range e.relayListeners { |
| 494 | if listener == nil { |
| 495 | continue |
| 496 | } |
| 497 | |
| 498 | udpAddr, ready, pending := listener.datagramReady() |
| 499 | if ready { |
| 500 | if _, ok := seen[udpAddr]; !ok { |
| 501 | seen[udpAddr] = struct{}{} |
| 502 | addrs = append(addrs, udpAddr) |
| 503 | } |
| 504 | } |
| 505 | if pending { |
| 506 | resolvedWithoutDatagram = false |
| 507 | } |
| 508 | } |
| 509 | e.mu.RUnlock() |
| 510 | if len(addrs) > 0 { |
| 511 | return addrs, nil |
| 512 | } |
| 513 | if resolvedWithoutDatagram { |
| 514 | return nil, errors.New("relay did not expose udp") |
| 515 | } |
| 516 | |
| 517 | select { |
| 518 | case <-e.done: |
| 519 | return nil, net.ErrClosed |
| 520 | case <-ctx.Done(): |
| 521 | return nil, ctx.Err() |
| 522 | case <-ticker.C: |
| 523 | } |
| 524 | } |
| 525 | } |
| 526 | |
| 527 | // RunHTTPRoutes serves path-routed HTTP upstreams through the exposure. |
| 528 | func (e *Exposure) RunHTTPRoutes(ctx context.Context, routes []HTTPRouteConfig, localAddr string) error { |
| 529 | cfg := e.Config() |
| 530 | handler, err := NewHTTPRoutes(routes, cfg.X402PayTo, cfg.X402Testnet) |
| 531 | if err != nil { |
| 532 | return err |
| 533 | } |
| 534 | return e.RunHTTP(ctx, handler, localAddr) |
| 535 | } |
| 536 | |
| 537 | func (e *Exposure) RunHTTP(ctx context.Context, handler http.Handler, localAddr string) error { |
| 538 | if handler == nil { |
| 539 | handler = http.NotFoundHandler() |
| 540 | } |
| 541 | |
| 542 | e.mu.RLock() |
| 543 | hasRelayListeners := len(e.relayListeners) > 0 |
| 544 | e.mu.RUnlock() |
| 545 | |
| 546 | if hasRelayListeners { |
| 547 | return RunHTTP(ctx, e, handler, localAddr) |
| 548 | } |
| 549 | return RunHTTP(ctx, nil, handler, localAddr) |
| 550 | } |
| 551 | |
| 552 | type exposureConn struct { |
| 553 | net.Conn |
| 554 | id uint64 |
| 555 | localAddr string |
| 556 | remoteAddr string |
| 557 | closeOnce sync.Once |
| 558 | } |
| 559 | |
| 560 | func (c *exposureConn) Close() error { |
| 561 | var closeErr error |
| 562 | c.closeOnce.Do(func() { |
| 563 | closeErr = c.Conn.Close() |
| 564 | if errors.Is(closeErr, net.ErrClosed) { |
| 565 | closeErr = nil |
| 566 | } |
| 567 | |
| 568 | event := log.Info(). |
| 569 | Uint64("conn_id", c.id). |
| 570 | Str("local_addr", c.localAddr). |
| 571 | Str("remote_addr", c.remoteAddr) |
| 572 | if closeErr != nil { |
| 573 | event = log.Warn(). |
| 574 | Err(closeErr). |
| 575 | Uint64("conn_id", c.id). |
| 576 | Str("local_addr", c.localAddr). |
| 577 | Str("remote_addr", c.remoteAddr) |
| 578 | } |
| 579 | event.Msg("exposure connection closed") |
| 580 | }) |
| 581 | return closeErr |
| 582 | } |
| 583 | |
| 584 | // tunnelCounterConn wraps a net.Conn and calls decr exactly once on the first |
| 585 | // Close invocation to decrement the active_tunnels_per_relay gauge. Subsequent |
| 586 | // Close calls are forwarded to the underlying conn but do not double-decrement. |
| 587 | // Concurrency is guaranteed by sync.Once. |
| 588 | type tunnelCounterConn struct { |
| 589 | net.Conn |
| 590 | once sync.Once |
| 591 | decr func() |
| 592 | } |
| 593 | |
| 594 | func (c *tunnelCounterConn) Close() error { |
| 595 | c.once.Do(c.decr) |
| 596 | return c.Conn.Close() |
| 597 | } |
| 598 | |
| 599 | func (e *Exposure) Accept() (net.Conn, error) { |
| 600 | select { |
| 601 | case <-e.done: |
| 602 | return nil, net.ErrClosed |
| 603 | case conn := <-e.accepted: |
| 604 | if conn == nil { |
| 605 | return nil, net.ErrClosed |
| 606 | } |
| 607 | |
| 608 | connID := e.connSeq.Add(1) |
| 609 | log.Info(). |
| 610 | Uint64("conn_id", connID). |
| 611 | Str("local_addr", conn.LocalAddr().String()). |
| 612 | Str("remote_addr", conn.RemoteAddr().String()). |
| 613 | Msg("exposure connection accepted") |
| 614 | |
| 615 | return &exposureConn{ |
| 616 | Conn: conn, |
| 617 | id: connID, |
| 618 | localAddr: conn.LocalAddr().String(), |
| 619 | remoteAddr: conn.RemoteAddr().String(), |
| 620 | }, nil |
| 621 | } |
| 622 | } |
| 623 | |
| 624 | func (e *Exposure) Close() error { |
| 625 | var closeErr error |
| 626 | e.closeOnce.Do(func() { |
| 627 | if e.cancel != nil { |
| 628 | e.cancel() |
| 629 | } |
| 630 | |
| 631 | e.mu.Lock() |
| 632 | relayListeners := e.relayListeners |
| 633 | e.relayListeners = make(map[string]*listener) |
| 634 | e.mu.Unlock() |
| 635 | |
| 636 | relayURLs := make([]string, 0, len(relayListeners)) |
| 637 | for relayURL, listener := range relayListeners { |
| 638 | relayURLs = append(relayURLs, relayURL) |
| 639 | if listener != nil { |
| 640 | closeErr = errors.Join(closeErr, listener.Close()) |
| 641 | } |
| 642 | } |
| 643 | |
| 644 | event := log.Info(). |
| 645 | Int("relay_count", len(relayListeners)). |
| 646 | Strs("relays", relayURLs) |
| 647 | if closeErr != nil { |
| 648 | event = log.Warn(). |
| 649 | Err(closeErr). |
| 650 | Int("relay_count", len(relayListeners)). |
| 651 | Strs("relays", relayURLs) |
| 652 | } |
| 653 | event.Msg("exposure closed") |
| 654 | }) |
| 655 | return closeErr |
| 656 | } |
| 657 | |
| 658 | func (e *Exposure) runDiscoveryLoop(ctx context.Context) { |
| 659 | refresher := discovery.NewRefresher(e.relaySet, nil) |
| 660 | ticker := time.NewTicker(discovery.DiscoveryPollInterval) |
| 661 | defer ticker.Stop() |
| 662 | |
| 663 | for { |
| 664 | if err := refresher.Refresh(ctx, nil); err != nil { |
| 665 | return |
| 666 | } |
| 667 | if err := e.reconcileRelayListeners(false); err != nil { |
| 668 | return |
| 669 | } |
| 670 | |
| 671 | select { |
| 672 | case <-ctx.Done(): |
| 673 | return |
| 674 | case <-ticker.C: |
| 675 | } |
| 676 | } |
| 677 | } |
| 678 | |
| 679 | func (e *Exposure) reconcileRelayListeners(failOnError bool) error { |
| 680 | if e.relaySet == nil { |
| 681 | return errors.New("relay set is unavailable") |
| 682 | } |
| 683 | cfg := e.Config() |
| 684 | routes, err := e.relaySet.PlanRoutes(append([]string(nil), cfg.MultiHop...), discovery.RouteState{ |
| 685 | ExplicitRelayURLs: append([]string(nil), cfg.RelayURLs...), |
| 686 | MaxActiveRelays: cfg.MaxActiveRelays, |
| 687 | MultiHopDepth: cfg.MultiHopDepth, |
| 688 | RequireUDP: cfg.UDPEnabled, |
| 689 | RequireTCP: cfg.TCPEnabled, |
| 690 | LocalAddress: cfg.Identity.Address, |
| 691 | }) |
| 692 | if err != nil { |
| 693 | return err |
| 694 | } |
| 695 | |
| 696 | routesByRelay := make(map[string]discovery.Route, len(routes)) |
| 697 | for _, route := range routes { |
| 698 | relayURL := route.ListenerRelayURL() |
| 699 | if relayURL == "" { |
| 700 | continue |
| 701 | } |
| 702 | routesByRelay[relayURL] = route |
| 703 | } |
| 704 | |
| 705 | e.mu.Lock() |
| 706 | staleListeners := make(map[string]*listener) |
| 707 | for relayURL, listener := range e.relayListeners { |
| 708 | route, wanted := routesByRelay[relayURL] |
| 709 | if wanted && listener != nil && listener.route.Equal(route) { |
| 710 | continue |
| 711 | } |
| 712 | staleListeners[relayURL] = listener |
| 713 | delete(e.relayListeners, relayURL) |
| 714 | } |
| 715 | missingRoutes := make([]discovery.Route, 0) |
| 716 | for _, route := range routes { |
| 717 | relayURL := route.ListenerRelayURL() |
| 718 | if _, exists := e.relayListeners[relayURL]; exists { |
| 719 | continue |
| 720 | } |
| 721 | missingRoutes = append(missingRoutes, route) |
| 722 | } |
| 723 | e.mu.Unlock() |
| 724 | |
| 725 | addedRelayURLs := make([]string, 0, len(missingRoutes)) |
| 726 | for relayURL, listener := range staleListeners { |
| 727 | if listener == nil { |
| 728 | continue |
| 729 | } |
| 730 | if err := listener.Close(); err != nil && !errors.Is(err, net.ErrClosed) { |
| 731 | log.Warn().Err(err).Str("relay_url", relayURL).Msg("close stale relay listener") |
| 732 | } |
| 733 | } |
| 734 | for _, route := range missingRoutes { |
| 735 | relayURL := route.ListenerRelayURL() |
| 736 | retryCount := 10 |
| 737 | if route.Explicit() || len(route.MultiHop()) > 0 { |
| 738 | retryCount = 0 |
| 739 | } |
| 740 | listener, err := newListener(context.Background(), route, listenerConfig{ |
| 741 | Identity: cfg.Identity.Copy(), |
| 742 | UDPEnabled: cfg.UDPEnabled, |
| 743 | TCPEnabled: cfg.TCPEnabled, |
| 744 | BanMITM: cfg.BanMITM, |
| 745 | Metadata: func() types.LeaseMetadata { |
| 746 | return e.Config().Metadata |
| 747 | }, |
| 748 | RetryCount: retryCount, |
| 749 | relaySet: e.relaySet, |
| 750 | }) |
| 751 | if err != nil { |
| 752 | if failOnError { |
| 753 | return fmt.Errorf("listen %q: %w", relayURL, err) |
| 754 | } |
| 755 | log.Warn().Err(err).Str("relay_url", relayURL).Msg("add relay listener") |
| 756 | continue |
| 757 | } |
| 758 | |
| 759 | select { |
| 760 | case <-e.done: |
| 761 | _ = listener.Close() |
| 762 | continue |
| 763 | default: |
| 764 | } |
| 765 | |
| 766 | e.mu.Lock() |
| 767 | if _, exists := e.relayListeners[relayURL]; exists { |
| 768 | e.mu.Unlock() |
| 769 | _ = listener.Close() |
| 770 | continue |
| 771 | } |
| 772 | e.relayListeners[relayURL] = listener |
| 773 | e.mu.Unlock() |
| 774 | addedRelayURLs = append(addedRelayURLs, relayURL) |
| 775 | |
| 776 | go e.runListenerAcceptLoop(listener) |
| 777 | } |
| 778 | |
| 779 | if len(staleListeners) > 0 || len(addedRelayURLs) > 0 { |
| 780 | removedRelayURLs := make([]string, 0, len(staleListeners)) |
| 781 | for relayURL := range staleListeners { |
| 782 | removedRelayURLs = append(removedRelayURLs, relayURL) |
| 783 | } |
| 784 | if len(removedRelayURLs) > 1 { |
| 785 | slices.Sort(removedRelayURLs) |
| 786 | } |
| 787 | listenerRelayURLs := make([]string, 0, len(routes)) |
| 788 | for _, route := range routes { |
| 789 | listenerRelayURLs = append(listenerRelayURLs, route.ListenerRelayURL()) |
| 790 | } |
| 791 | log.Info(). |
| 792 | Strs("added_relays", addedRelayURLs). |
| 793 | Strs("removed_relays", removedRelayURLs). |
| 794 | Strs("listener_relays", listenerRelayURLs). |
| 795 | Msg("reconciled relay listeners") |
| 796 | } |
| 797 | return nil |
| 798 | } |
| 799 | |
| 800 | func (e *Exposure) runListenerAcceptLoop(listener *listener) { |
| 801 | if listener == nil { |
| 802 | return |
| 803 | } |
| 804 | |
| 805 | relayURL := "" |
| 806 | if listener.relayURL != nil { |
| 807 | relayURL = listener.relayURL.String() |
| 808 | } |
| 809 | if listener.udpEnabled { |
| 810 | go func() { |
| 811 | for { |
| 812 | frame, err := listener.acceptDatagram() |
| 813 | if err != nil { |
| 814 | select { |
| 815 | case <-e.done: |
| 816 | return |
| 817 | default: |
| 818 | } |
| 819 | if errors.Is(err, net.ErrClosed) { |
| 820 | return |
| 821 | } |
| 822 | log.Warn(). |
| 823 | Err(err). |
| 824 | Str("relay_url", relayURL). |
| 825 | Str("address", listener.identity.Address). |
| 826 | Msg("datagram accept failed") |
| 827 | return |
| 828 | } |
| 829 | |
| 830 | select { |
| 831 | case <-e.done: |
| 832 | return |
| 833 | case e.datagrams <- frame: |
| 834 | } |
| 835 | } |
| 836 | }() |
| 837 | } |
| 838 | defer func() { |
| 839 | e.mu.Lock() |
| 840 | if current, ok := e.relayListeners[relayURL]; ok && current == listener { |
| 841 | delete(e.relayListeners, relayURL) |
| 842 | } |
| 843 | e.mu.Unlock() |
| 844 | }() |
| 845 | |
| 846 | for { |
| 847 | conn, err := listener.Accept() |
| 848 | if err != nil { |
| 849 | select { |
| 850 | case <-listener.doneCh: |
| 851 | return |
| 852 | default: |
| 853 | } |
| 854 | if errors.Is(err, net.ErrClosed) { |
| 855 | return |
| 856 | } |
| 857 | log.Warn().Err(err).Str("relay_url", relayURL).Msg("exposure listener accept failed") |
| 858 | return |
| 859 | } |
| 860 | |
| 861 | telemetry.ActiveTunnelsPerRelay.WithLabelValues(relayURL).Inc() |
| 862 | wrappedConn := &tunnelCounterConn{ |
| 863 | Conn: conn, |
| 864 | decr: func() { |
| 865 | telemetry.ActiveTunnelsPerRelay.WithLabelValues(relayURL).Dec() |
| 866 | }, |
| 867 | } |
| 868 | |
| 869 | select { |
| 870 | case <-e.done: |
| 871 | _ = wrappedConn.Close() |
| 872 | return |
| 873 | case e.accepted <- wrappedConn: |
| 874 | } |
| 875 | } |
| 876 | } |