diff --git a/config.example.yaml b/config.example.yaml index ba0146b50..0c66ec2c1 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -232,8 +232,10 @@ codex: udp-port-min: 0 udp-port-max: 0 # Optional STUN/TURN servers. TURN credentials are never returned by the JSON config API. - # Global and per-auth proxy-url settings apply to call creation and Sideband WebSocket only; - # WebRTC media uses direct ICE/STUN/TURN connectivity and does not traverse that proxy. + # Without a concrete global/per-auth proxy-url, WebRTC uses normal direct ICE/STUN/TURN connectivity. + # With http, https, socks5, or socks5h proxy-url, the OpenAI-facing leg is forced through + # authenticated ICE-TCP over that proxy and never falls back to UDP or a direct connection. + # The Codex Desktop-facing leg remains direct, and configured ICE servers still apply to it. # ice-servers: # - urls: # - "stun:stun.example.com:3478" diff --git a/go.mod b/go.mod index b1544a510..1f5d12fb2 100644 --- a/go.mod +++ b/go.mod @@ -17,8 +17,11 @@ require ( github.com/joho/godotenv v1.5.1 github.com/klauspost/compress v1.17.4 github.com/minio/minio-go/v7 v7.0.66 + github.com/pion/ice/v4 v4.3.0 github.com/pion/interceptor v0.1.45 github.com/pion/rtp v1.10.4 + github.com/pion/sdp/v3 v3.0.19 + github.com/pion/stun/v3 v3.1.6 github.com/pion/webrtc/v4 v4.2.17 github.com/redis/go-redis/v9 v9.19.0 github.com/refraction-networking/utls v1.8.2 @@ -41,15 +44,12 @@ require ( github.com/dlclark/regexp2/v2 v2.5.1 // indirect github.com/pion/datachannel v1.6.2 // indirect github.com/pion/dtls/v3 v3.1.5 // indirect - github.com/pion/ice/v4 v4.3.0 // indirect github.com/pion/logging v0.2.4 // indirect github.com/pion/mdns/v2 v2.1.0 // indirect github.com/pion/randutil v0.1.0 // indirect github.com/pion/rtcp v1.2.17 // indirect github.com/pion/sctp v1.11.0 // indirect - github.com/pion/sdp/v3 v3.0.19 // indirect github.com/pion/srtp/v3 v3.0.12 // indirect - github.com/pion/stun/v3 v3.1.6 // indirect github.com/pion/transport/v4 v4.0.2 // indirect github.com/pion/turn/v5 v5.0.12 // indirect github.com/rogpeppe/go-internal v1.15.0 // indirect diff --git a/internal/client/codex/live/live.go b/internal/client/codex/live/live.go index 77f0c2d48..a691b721c 100644 --- a/internal/client/codex/live/live.go +++ b/internal/client/codex/live/live.go @@ -238,7 +238,7 @@ func (h *Handler) Handle(c *gin.Context) { return } var upstreamOffer string - mediaSession, upstreamOffer, errSDP = mediaRelay.NewSession(ctx, clientOffer) + mediaSession, upstreamOffer, errSDP = mediaRelay.NewSession(ctx, clientOffer, proxyURLForAuth(runtimeConfig, selected)) if errSDP != nil { c.JSON(http.StatusBadGateway, gin.H{"error": errSDP.Error()}) return diff --git a/internal/client/codex/live/live_test.go b/internal/client/codex/live/live_test.go index 93889fedc..40e61f58c 100644 --- a/internal/client/codex/live/live_test.go +++ b/internal/client/codex/live/live_test.go @@ -149,13 +149,15 @@ func (b *trackedResponseBody) Close() error { type fakeMediaRelay struct { clientOffer string + proxyURL string upstreamOffer string session *fakeMediaSession err error } -func (r *fakeMediaRelay) NewSession(_ context.Context, clientOffer string) (mediaRelaySession, string, error) { +func (r *fakeMediaRelay) NewSession(_ context.Context, clientOffer, proxyURL string) (mediaRelaySession, string, error) { r.clientOffer = clientOffer + r.proxyURL = proxyURL return r.session, r.upstreamOffer, r.err } @@ -318,6 +320,20 @@ func TestHandlerRewritesLiveCallAndSchedulesOAuth(t *testing.T) { } } +func TestProxyURLForAuthPrefersCredentialOverride(t *testing.T) { + cfg := &config.Config{} + cfg.ProxyURL = "http://global.example:8080" + if got := proxyURLForAuth(cfg, &auth.Auth{ProxyURL: "socks5://credential.example:1080"}); got != "socks5://credential.example:1080" { + t.Fatalf("effective proxy URL = %q, want credential override", got) + } + if got := proxyURLForAuth(cfg, &auth.Auth{}); got != "http://global.example:8080" { + t.Fatalf("effective proxy URL = %q, want global fallback", got) + } + if got := proxyURLForAuth(cfg, &auth.Auth{ProxyURL: "direct"}); got != "direct" { + t.Fatalf("effective proxy URL = %q, want explicit direct override", got) + } +} + func TestHandlerRelaysWebRTCMediaSDP(t *testing.T) { gin.SetMode(gin.TestMode) @@ -330,6 +346,7 @@ func TestHandlerRelaysWebRTCMediaSDP(t *testing.T) { ID: "codex-oauth", Provider: "codex", Status: auth.StatusActive, + ProxyURL: "socks5://credential-proxy.example:1080", Metadata: map[string]any{"access_token": "oauth-token"}, }) mediaSession := &fakeMediaSession{downstreamSDP: "v=0\r\no=downstream-answer\r\n"} @@ -337,7 +354,9 @@ func TestHandlerRelaysWebRTCMediaSDP(t *testing.T) { upstreamOffer: "v=0\r\no=gateway-offer\r\n", session: mediaSession, } - handler := NewHandler(manager, nil) + runtimeConfig := &config.Config{} + runtimeConfig.ProxyURL = "http://global-proxy.example:8080" + handler := NewHandler(manager, runtimeConfig) handler.mediaRelay = mediaRelay router := gin.New() router.POST("/v1/live", handler.Handle) @@ -355,6 +374,9 @@ func TestHandlerRelaysWebRTCMediaSDP(t *testing.T) { if mediaRelay.clientOffer != "v=0\r\no=desktop-offer\r\n" { t.Fatalf("media client offer = %q", mediaRelay.clientOffer) } + if mediaRelay.proxyURL != "socks5://credential-proxy.example:1080" { + t.Fatalf("media proxy URL = %q, want credential override", mediaRelay.proxyURL) + } var upstreamPayload struct { SDP string `json:"sdp"` } diff --git a/internal/client/codex/live/media.go b/internal/client/codex/live/media.go index 164ab455f..bd57508fd 100644 --- a/internal/client/codex/live/media.go +++ b/internal/client/codex/live/media.go @@ -14,7 +14,9 @@ import ( "github.com/pion/rtp" "github.com/pion/webrtc/v4" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" + "github.com/router-for-me/CLIProxyAPI/v7/sdk/proxyutil" log "github.com/sirupsen/logrus" + "golang.org/x/net/proxy" ) const ( @@ -40,14 +42,15 @@ type mediaRelaySession interface { } type mediaRelayFactory interface { - NewSession(context.Context, string) (mediaRelaySession, string, error) + NewSession(context.Context, string, string) (mediaRelaySession, string, error) } type pionMediaRelay struct { - downstreamAPI *webrtc.API - upstreamAPI *webrtc.API - configuration webrtc.Configuration - limiter *mediaSessionLimiter + downstreamAPI *webrtc.API + upstreamAPI *webrtc.API + proxyUpstreamAPI *webrtc.API + configuration webrtc.Configuration + limiter *mediaSessionLimiter } type mediaSessionLimiter struct { @@ -72,6 +75,12 @@ type pionMediaSession struct { mediaSessionID string callID string releaseSlot func() + + proxyDialer proxy.ContextDialer + proxyScheme string + localOffer string + tunnelsMu sync.Mutex + tunnels []*tcpCandidateTunnel } type dataChannelMessage struct { @@ -118,6 +127,10 @@ func newPionMediaRelayWithLimiter(relayConfig config.CodexLiveMediaRelayConfig, if errAPI != nil { return nil, errAPI } + proxyUpstreamAPI, errAPI := newPionProxyAPI(relayConfig) + if errAPI != nil { + return nil, errAPI + } iceServers := make([]webrtc.ICEServer, 0, len(relayConfig.ICEServers)) for _, server := range relayConfig.ICEServers { urls := make([]string, 0, len(server.URLs)) @@ -136,10 +149,11 @@ func newPionMediaRelayWithLimiter(relayConfig config.CodexLiveMediaRelayConfig, } limiter.setLimit(relayConfig.EffectiveMaxSessions()) return &pionMediaRelay{ - downstreamAPI: downstreamAPI, - upstreamAPI: upstreamAPI, - configuration: webrtc.Configuration{ICEServers: iceServers}, - limiter: limiter, + downstreamAPI: downstreamAPI, + upstreamAPI: upstreamAPI, + proxyUpstreamAPI: proxyUpstreamAPI, + configuration: webrtc.Configuration{ICEServers: iceServers}, + limiter: limiter, }, nil } @@ -177,6 +191,14 @@ func (l *mediaSessionLimiter) release() { } func newPionAPI(relayConfig config.CodexLiveMediaRelayConfig, filterPrivateRemoteIPs bool) (*webrtc.API, error) { + return newPionAPIWithOptions(relayConfig, filterPrivateRemoteIPs, false) +} + +func newPionProxyAPI(relayConfig config.CodexLiveMediaRelayConfig) (*webrtc.API, error) { + return newPionAPIWithOptions(relayConfig, false, true) +} + +func newPionAPIWithOptions(relayConfig config.CodexLiveMediaRelayConfig, filterPrivateRemoteIPs, loopbackOnly bool) (*webrtc.API, error) { mediaEngine := &webrtc.MediaEngine{} if errRegister := mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ RTPCodecCapability: opusCodec, @@ -189,17 +211,31 @@ func newPionAPI(relayConfig config.CodexLiveMediaRelayConfig, filterPrivateRemot return nil, fmt.Errorf("register WebRTC interceptors: %w", errRegister) } settingEngine := webrtc.SettingEngine{} - if relayConfig.UDPPortMin != 0 { - if errPorts := settingEngine.SetEphemeralUDPPortRange(relayConfig.UDPPortMin, relayConfig.UDPPortMax); errPorts != nil { - return nil, fmt.Errorf("configure WebRTC UDP port range: %w", errPorts) + if !loopbackOnly { + if relayConfig.UDPPortMin != 0 { + if errPorts := settingEngine.SetEphemeralUDPPortRange(relayConfig.UDPPortMin, relayConfig.UDPPortMax); errPorts != nil { + return nil, fmt.Errorf("configure WebRTC UDP port range: %w", errPorts) + } + } + if publicIP := strings.TrimSpace(relayConfig.PublicIP); publicIP != "" { + settingEngine.SetNAT1To1IPs([]string{publicIP}, webrtc.ICECandidateTypeHost) } - } - if publicIP := strings.TrimSpace(relayConfig.PublicIP); publicIP != "" { - settingEngine.SetNAT1To1IPs([]string{publicIP}, webrtc.ICECandidateTypeHost) } if filterPrivateRemoteIPs { settingEngine.SetRemoteIPFilter(isPublicRemoteIP) } + if loopbackOnly { + settingEngine.SetNetworkTypes([]webrtc.NetworkType{ + webrtc.NetworkTypeUDP4, + webrtc.NetworkTypeUDP6, + webrtc.NetworkTypeTCP4, + webrtc.NetworkTypeTCP6, + }) + settingEngine.SetIncludeLoopbackCandidate(true) + settingEngine.SetIPFilter(func(ip net.IP) bool { + return ip != nil && ip.IsLoopback() + }) + } return webrtc.NewAPI( webrtc.WithMediaEngine(mediaEngine), webrtc.WithInterceptorRegistry(interceptorRegistry), @@ -212,13 +248,26 @@ func isPublicRemoteIP(ip net.IP) bool { !ip.IsLinkLocalUnicast() && !ip.IsLinkLocalMulticast() && !ip.IsMulticast() } -func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer string) (mediaRelaySession, string, error) { - if r == nil || r.downstreamAPI == nil || r.upstreamAPI == nil || r.limiter == nil { +func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer, proxyURL string) (mediaRelaySession, string, error) { + if r == nil || r.downstreamAPI == nil || r.upstreamAPI == nil || r.proxyUpstreamAPI == nil || r.limiter == nil { return nil, "", errors.New("Codex live media relay unavailable") } if errContext := ctx.Err(); errContext != nil { return nil, "", errContext } + builtProxyDialer, proxyMode, errProxy := proxyutil.BuildDialer(proxyURL) + if errProxy != nil { + return nil, "", fmt.Errorf("configure Codex live remote TCP proxy: %w", errProxy) + } + proxied := proxyMode == proxyutil.ModeProxy + var proxyDialer proxy.ContextDialer + if proxied { + contextDialer, ok := builtProxyDialer.(proxy.ContextDialer) + if !ok { + return nil, "", errors.New("Codex live remote TCP proxy does not support cancellation") + } + proxyDialer = contextDialer + } if !r.limiter.acquire() { return nil, "", errors.New("Codex live media relay capacity exhausted") } @@ -228,7 +277,13 @@ func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer string) (me releaseSlot() return nil, "", fmt.Errorf("create downstream PeerConnection: %w", errDownstream) } - upstream, errUpstream := r.upstreamAPI.NewPeerConnection(r.configuration) + upstreamAPI := r.upstreamAPI + upstreamConfiguration := r.configuration + if proxied { + upstreamAPI = r.proxyUpstreamAPI + upstreamConfiguration.ICEServers = nil + } + upstream, errUpstream := upstreamAPI.NewPeerConnection(upstreamConfiguration) if errUpstream != nil { releaseSlot() if errClose := downstream.Close(); errClose != nil { @@ -243,6 +298,8 @@ func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer string) (me done: make(chan struct{}), mediaSessionID: uuid.NewString(), releaseSlot: releaseSlot, + proxyDialer: proxyDialer, + proxyScheme: proxyScheme(proxyURL), } session.bridge = newDataChannelBridge(session.done, func(err error) { session.fail("data_channel_failed", err) @@ -331,6 +388,7 @@ func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer string) (me _ = session.Close() return nil, "", errors.New("upstream WebRTC offer is empty") } + session.localOffer = localDescription.SDP return session, localDescription.SDP, nil } @@ -338,11 +396,30 @@ func (s *pionMediaSession) AcceptUpstreamAnswer(ctx context.Context, upstreamAns if s == nil || s.upstream == nil || s.downstream == nil { return "", errors.New("Codex live media session unavailable") } + answerToApply := upstreamAnswer + if s.proxyDialer != nil { + rewrittenAnswer, tunnels, errProxy := prepareProxiedUpstreamAnswer(upstreamAnswer, s.localOffer, s.proxyDialer) + if errProxy != nil { + return "", errProxy + } + if !s.installCandidateTunnels(tunnels) { + errClosed := errors.New("Codex live media session closed while configuring TCP proxy") + if errClose := closeCandidateTunnels(tunnels); errClose != nil { + return "", errors.Join(errClosed, fmt.Errorf("close TCP candidate tunnels: %w", errClose)) + } + return "", errClosed + } + answerToApply = rewrittenAnswer + } if errRemote := s.upstream.SetRemoteDescription(webrtc.SessionDescription{ Type: webrtc.SDPTypeAnswer, - SDP: upstreamAnswer, + SDP: answerToApply, }); errRemote != nil { - return "", fmt.Errorf("set upstream WebRTC answer: %w", errRemote) + errSetRemote := fmt.Errorf("set upstream WebRTC answer: %w", errRemote) + if errClose := s.closeCandidateTunnels(); errClose != nil { + return "", errors.Join(errSetRemote, fmt.Errorf("close TCP candidate tunnels: %w", errClose)) + } + return "", errSetRemote } gatherComplete := webrtc.GatheringCompletePromise(s.downstream) answer, errAnswer := s.downstream.CreateAnswer(nil) @@ -364,6 +441,32 @@ func (s *pionMediaSession) AcceptUpstreamAnswer(ctx context.Context, upstreamAns return localDescription.SDP, nil } +func (s *pionMediaSession) installCandidateTunnels(tunnels []*tcpCandidateTunnel) bool { + if s == nil { + return false + } + s.tunnelsMu.Lock() + defer s.tunnelsMu.Unlock() + select { + case <-s.done: + return false + default: + } + s.tunnels = tunnels + return true +} + +func (s *pionMediaSession) closeCandidateTunnels() error { + if s == nil { + return nil + } + s.tunnelsMu.Lock() + tunnels := s.tunnels + s.tunnels = nil + s.tunnelsMu.Unlock() + return closeCandidateTunnels(tunnels) +} + func (s *pionMediaSession) SetCallID(callID string) { if s == nil { return @@ -384,6 +487,10 @@ func (s *pionMediaSession) logFields(peer string) log.Fields { if callID != "" { fields["call_id"] = callID } + if s.proxyDialer != nil && (peer == "remote" || peer == "session") { + fields["remote_transport"] = "tcp" + fields["proxy_scheme"] = s.proxyScheme + } return fields } @@ -421,6 +528,9 @@ func (s *pionMediaSession) CloseWithReason(reason string) error { s.bridge.close() } var closeErrors []error + if errClose := s.closeCandidateTunnels(); errClose != nil { + closeErrors = append(closeErrors, fmt.Errorf("close TCP candidate tunnels: %w", errClose)) + } if errClose := s.closePeerConnection("local", s.downstream); errClose != nil { closeErrors = append(closeErrors, fmt.Errorf("close downstream PeerConnection: %w", errClose)) } diff --git a/internal/client/codex/live/media_test.go b/internal/client/codex/live/media_test.go index a09d723cc..cc9f72850 100644 --- a/internal/client/codex/live/media_test.go +++ b/internal/client/codex/live/media_test.go @@ -3,6 +3,7 @@ package live import ( "context" "net" + "strings" "testing" "time" @@ -14,6 +15,62 @@ import ( logtest "github.com/sirupsen/logrus/hooks/test" ) +func TestPionMediaRelaySelectsRemoteProxyMode(t *testing.T) { + clientAPI := newTestWebRTCAPI(t) + client, errClient := clientAPI.NewPeerConnection(webrtc.Configuration{}) + if errClient != nil { + t.Fatalf("create client PeerConnection: %v", errClient) + } + defer closeTestPeerConnection(t, client) + if _, errChannel := client.CreateDataChannel(realtimeDataChannelLabel, nil); errChannel != nil { + t.Fatalf("create client DataChannel: %v", errChannel) + } + clientOffer := completeOffer(t, client) + relay, errRelay := newPionMediaRelay(config.CodexLiveMediaRelayConfig{ + Enabled: true, + PublicIP: "198.51.100.1", + }) + if errRelay != nil { + t.Fatalf("create media relay: %v", errRelay) + } + + for name, testCase := range map[string]struct { + proxyURL string + proxied bool + }{ + "inherit": {proxyURL: ""}, + "direct": {proxyURL: "direct"}, + "HTTP": {proxyURL: "http://proxy.example:8080", proxied: true}, + "HTTPS": {proxyURL: "https://proxy.example:8443", proxied: true}, + "SOCKS5": {proxyURL: "socks5://proxy.example:1080", proxied: true}, + "SOCKS5H": {proxyURL: "socks5h://proxy.example:1080", proxied: true}, + } { + t.Run(name, func(t *testing.T) { + session, upstreamOffer, errSession := relay.NewSession(context.Background(), clientOffer, testCase.proxyURL) + if errSession != nil { + t.Fatalf("create media session: %v", errSession) + } + pionSession, ok := session.(*pionMediaSession) + if !ok { + t.Fatalf("media session type = %T", session) + } + if got := pionSession.proxyDialer != nil; got != testCase.proxied { + t.Fatalf("proxied = %t, want %t", got, testCase.proxied) + } + if testCase.proxied && !offerCandidatesAreLoopback(t, upstreamOffer) { + t.Fatal("proxied upstream offer exposed a non-loopback candidate") + } + if errClose := session.Close(); errClose != nil { + t.Fatalf("close media session: %v", errClose) + } + }) + } + + if _, _, errSession := relay.NewSession(context.Background(), clientOffer, "invalid-proxy"); errSession == nil { + t.Fatal("expected invalid proxy URL to fail media session creation") + } +} + func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) { logger := log.StandardLogger() previousHooks := logger.ReplaceHooks(make(log.LevelHooks)) @@ -69,7 +126,7 @@ func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) { if errRelay != nil { t.Fatalf("create media relay: %v", errRelay) } - session, relayOffer, errSession := relay.NewSession(context.Background(), clientOffer) + session, relayOffer, errSession := relay.NewSession(context.Background(), clientOffer, "") if errSession != nil { t.Fatalf("create media relay session: %v", errSession) } @@ -83,7 +140,7 @@ func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) { if errRelay != nil { t.Fatalf("reload media relay: %v", errRelay) } - if _, _, errCapacity := reloadedRelay.NewSession(context.Background(), clientOffer); errCapacity == nil { + if _, _, errCapacity := reloadedRelay.NewSession(context.Background(), clientOffer, ""); errCapacity == nil { t.Fatal("reloaded media relay bypassed the shared session capacity") } @@ -165,7 +222,7 @@ func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) { if errClose := session.Close(); errClose != nil { t.Fatalf("close media relay session for logging: %v", errClose) } - replacementSession, _, errReplacement := reloadedRelay.NewSession(context.Background(), clientOffer) + replacementSession, _, errReplacement := reloadedRelay.NewSession(context.Background(), clientOffer, "") if errReplacement != nil { t.Fatalf("shared capacity was not released: %v", errReplacement) } @@ -202,6 +259,27 @@ func TestIsPublicRemoteIP(t *testing.T) { } } +func offerCandidatesAreLoopback(t *testing.T, offer string) bool { + t.Helper() + lines := strings.Split(strings.ReplaceAll(offer, "\r\n", "\n"), "\n") + candidateCount := 0 + for _, line := range lines { + if !strings.HasPrefix(line, "a=candidate:") { + continue + } + candidateCount++ + fields := strings.Fields(strings.TrimPrefix(line, "a=candidate:")) + if len(fields) < 6 { + t.Fatalf("malformed offer candidate: %q", line) + } + address := net.ParseIP(fields[4]) + if address == nil || !address.IsLoopback() { + return false + } + } + return candidateCount > 0 +} + func newTestWebRTCAPI(t *testing.T) *webrtc.API { t.Helper() mediaEngine := &webrtc.MediaEngine{} diff --git a/internal/client/codex/live/sideband.go b/internal/client/codex/live/sideband.go index b27808466..1e2b679a3 100644 --- a/internal/client/codex/live/sideband.go +++ b/internal/client/codex/live/sideband.go @@ -617,10 +617,10 @@ func isNormalWebsocketClose(err error) bool { } func newProxyAwareSidebandDialer(cfg *config.Config, selected *auth.Auth) *websocket.Dialer { - return newSidebandDialer(proxyURLForSideband(cfg, selected)) + return newSidebandDialer(proxyURLForAuth(cfg, selected)) } -func proxyURLForSideband(cfg *config.Config, selected *auth.Auth) string { +func proxyURLForAuth(cfg *config.Config, selected *auth.Auth) string { if selected != nil && strings.TrimSpace(selected.ProxyURL) != "" { return strings.TrimSpace(selected.ProxyURL) } diff --git a/internal/client/codex/live/tcp_proxy.go b/internal/client/codex/live/tcp_proxy.go new file mode 100644 index 000000000..c9d2a0d25 --- /dev/null +++ b/internal/client/codex/live/tcp_proxy.go @@ -0,0 +1,525 @@ +package live + +import ( + "context" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "net/netip" + "strconv" + "strings" + "sync" + + "github.com/pion/ice/v4" + "github.com/pion/sdp/v3" + "github.com/pion/stun/v3" + log "github.com/sirupsen/logrus" + "golang.org/x/net/proxy" +) + +const ( + maxUpstreamICECandidates = 64 + maxProxiedTCPCandidates = 16 + maxUnauthenticatedTCPConns = 4 + maxInitialSTUNFrameSize = 4096 + stunMessageHeaderSize = 20 +) + +var nonRoutableProxyTargetPrefixes = []netip.Prefix{ + netip.MustParsePrefix("0.0.0.0/8"), + netip.MustParsePrefix("10.0.0.0/8"), + netip.MustParsePrefix("100.64.0.0/10"), + netip.MustParsePrefix("127.0.0.0/8"), + netip.MustParsePrefix("169.254.0.0/16"), + netip.MustParsePrefix("172.16.0.0/12"), + netip.MustParsePrefix("192.0.0.0/24"), + netip.MustParsePrefix("192.0.2.0/24"), + netip.MustParsePrefix("192.88.99.0/24"), + netip.MustParsePrefix("192.168.0.0/16"), + netip.MustParsePrefix("198.18.0.0/15"), + netip.MustParsePrefix("198.51.100.0/24"), + netip.MustParsePrefix("203.0.113.0/24"), + netip.MustParsePrefix("224.0.0.0/4"), + netip.MustParsePrefix("240.0.0.0/4"), + netip.MustParsePrefix("::/96"), + netip.MustParsePrefix("::ffff:0:0:0/96"), + netip.MustParsePrefix("64:ff9b::/96"), + netip.MustParsePrefix("64:ff9b:1::/48"), + netip.MustParsePrefix("100::/64"), + netip.MustParsePrefix("2001::/23"), + netip.MustParsePrefix("2001:db8::/32"), + netip.MustParsePrefix("2002::/16"), + netip.MustParsePrefix("3fff::/20"), + netip.MustParsePrefix("5f00::/16"), + netip.MustParsePrefix("fc00::/7"), + netip.MustParsePrefix("fe80::/10"), + netip.MustParsePrefix("fec0::/10"), + netip.MustParsePrefix("ff00::/8"), +} + +type iceCredentials struct { + ufrag string + password string +} + +type tcpCandidateTunnel struct { + listener net.Listener + target netip.AddrPort + dialer proxy.ContextDialer + expectedUser string + remotePassword string + + mu sync.Mutex + closed bool + claimed bool + connections map[net.Conn]struct{} + validationSlots chan struct{} + ctx context.Context + cancel context.CancelFunc +} + +type tcpCandidatePlan struct { + mediaIndex int + attributeIndex int + fields []string + target netip.AddrPort +} + +func prepareProxiedUpstreamAnswer(answer, localOffer string, dialer proxy.ContextDialer) (string, []*tcpCandidateTunnel, error) { + if dialer == nil { + return "", nil, errors.New("Codex live TCP proxy dialer is unavailable") + } + var remoteDescription sdp.SessionDescription + if errUnmarshal := remoteDescription.UnmarshalString(answer); errUnmarshal != nil { + return "", nil, fmt.Errorf("parse upstream WebRTC answer for TCP proxy: %w", errUnmarshal) + } + var localDescription sdp.SessionDescription + if errUnmarshal := localDescription.UnmarshalString(localOffer); errUnmarshal != nil { + return "", nil, fmt.Errorf("parse upstream WebRTC offer for TCP proxy: %w", errUnmarshal) + } + remoteCredentials, errCredentials := bundledICECredentials(&remoteDescription) + if errCredentials != nil { + return "", nil, fmt.Errorf("read upstream WebRTC answer ICE credentials: %w", errCredentials) + } + localCredentials, errCredentials := bundledICECredentials(&localDescription) + if errCredentials != nil { + return "", nil, fmt.Errorf("read upstream WebRTC offer ICE credentials: %w", errCredentials) + } + + plans := make([]tcpCandidatePlan, 0, 4) + candidateCount := 0 + for mediaIndex, media := range remoteDescription.MediaDescriptions { + if media == nil { + continue + } + filtered := make([]sdp.Attribute, 0, len(media.Attributes)) + for attributeIndex := range media.Attributes { + attribute := media.Attributes[attributeIndex] + if !attribute.IsICECandidate() { + filtered = append(filtered, attribute) + continue + } + candidateCount++ + if candidateCount > maxUpstreamICECandidates { + return "", nil, fmt.Errorf("upstream WebRTC answer exceeds the %d candidate limit", maxUpstreamICECandidates) + } + plan, keep, errCandidate := proxiedTCPCandidatePlan(attribute.Value) + if errCandidate != nil { + return "", nil, errCandidate + } + if !keep { + continue + } + if len(plans) >= maxProxiedTCPCandidates { + return "", nil, fmt.Errorf("upstream WebRTC answer exceeds the %d TCP candidate proxy limit", maxProxiedTCPCandidates) + } + plan.mediaIndex = mediaIndex + plan.attributeIndex = len(filtered) + filtered = append(filtered, attribute) + plans = append(plans, plan) + } + media.Attributes = filtered + } + if len(plans) == 0 { + return "", nil, errors.New("upstream WebRTC answer has no supported public TCP passive candidate on port 443") + } + + expectedUser := remoteCredentials.ufrag + ":" + localCredentials.ufrag + tunnels := make([]*tcpCandidateTunnel, 0, len(plans)) + closeTunnels := func() { + for _, tunnel := range tunnels { + if errClose := tunnel.Close(); errClose != nil { + log.WithError(errClose).Debug("codex live TCP proxy: close candidate tunnel after setup error") + } + } + } + for _, plan := range plans { + tunnel, errTunnel := newTCPCandidateTunnel(plan.target, dialer, expectedUser, remoteCredentials.password) + if errTunnel != nil { + closeTunnels() + return "", nil, errTunnel + } + tunnels = append(tunnels, tunnel) + listenerAddress, ok := tunnel.listener.Addr().(*net.TCPAddr) + if !ok || listenerAddress.IP == nil { + closeTunnels() + return "", nil, errors.New("Codex live TCP proxy listener returned an invalid address") + } + fields := append([]string(nil), plan.fields...) + fields[4] = listenerAddress.IP.String() + fields[5] = strconv.Itoa(listenerAddress.Port) + remoteDescription.MediaDescriptions[plan.mediaIndex].Attributes[plan.attributeIndex].Value = strings.Join(fields, " ") + } + + rewritten, errMarshal := remoteDescription.Marshal() + if errMarshal != nil { + closeTunnels() + return "", nil, fmt.Errorf("marshal proxied upstream WebRTC answer: %w", errMarshal) + } + return string(rewritten), tunnels, nil +} + +func proxiedTCPCandidatePlan(rawCandidate string) (tcpCandidatePlan, bool, error) { + trimmed := strings.TrimSpace(rawCandidate) + candidate, errCandidate := ice.UnmarshalCandidate(trimmed) + if errCandidate != nil { + return tcpCandidatePlan{}, false, fmt.Errorf("parse upstream WebRTC candidate: %w", errCandidate) + } + if candidate.NetworkType() != ice.NetworkTypeTCP4 && candidate.NetworkType() != ice.NetworkTypeTCP6 { + return tcpCandidatePlan{}, false, nil + } + if candidate.TCPType() != ice.TCPTypePassive { + return tcpCandidatePlan{}, false, nil + } + if candidate.Component() != uint16(ice.ComponentRTP) || candidate.Type() != ice.CandidateTypeHost { + return tcpCandidatePlan{}, false, nil + } + if candidate.Port() != 443 { + return tcpCandidatePlan{}, false, fmt.Errorf("upstream WebRTC TCP proxy candidate uses disallowed port %d", candidate.Port()) + } + address, errAddress := netip.ParseAddr(candidate.Address()) + if errAddress != nil { + return tcpCandidatePlan{}, false, errors.New("upstream WebRTC TCP proxy candidate address must be an IP") + } + address = address.Unmap() + if !isPublicProxyTarget(address) { + return tcpCandidatePlan{}, false, errors.New("upstream WebRTC TCP proxy candidate address must be globally routable") + } + fields := strings.Fields(trimmed) + if len(fields) < 8 { + return tcpCandidatePlan{}, false, errors.New("upstream WebRTC TCP proxy candidate is malformed") + } + return tcpCandidatePlan{ + fields: fields, + target: netip.AddrPortFrom(address, uint16(candidate.Port())), + }, true, nil +} + +func isPublicProxyTarget(address netip.Addr) bool { + if !address.IsValid() || !address.IsGlobalUnicast() || address.IsUnspecified() || address.IsLoopback() || + address.IsPrivate() || address.IsLinkLocalUnicast() || address.IsLinkLocalMulticast() || address.IsMulticast() { + return false + } + for _, prefix := range nonRoutableProxyTargetPrefixes { + if prefix.Contains(address) { + return false + } + } + return true +} + +func bundledICECredentials(description *sdp.SessionDescription) (iceCredentials, error) { + if description == nil { + return iceCredentials{}, errors.New("SDP is unavailable") + } + sessionUfrag, _ := description.Attribute("ice-ufrag") + sessionPassword, _ := description.Attribute("ice-pwd") + var selected iceCredentials + for _, media := range description.MediaDescriptions { + if media == nil { + continue + } + ufrag := sessionUfrag + if mediaUfrag, ok := media.Attribute("ice-ufrag"); ok { + ufrag = mediaUfrag + } + password := sessionPassword + if mediaPassword, ok := media.Attribute("ice-pwd"); ok { + password = mediaPassword + } + ufrag = strings.TrimSpace(ufrag) + password = strings.TrimSpace(password) + if ufrag == "" && password == "" { + continue + } + if ufrag == "" || password == "" { + return iceCredentials{}, errors.New("SDP contains incomplete ICE credentials") + } + current := iceCredentials{ufrag: ufrag, password: password} + if selected.ufrag == "" { + selected = current + continue + } + if selected != current { + return iceCredentials{}, errors.New("SDP contains inconsistent bundled ICE credentials") + } + } + if selected.ufrag == "" { + selected = iceCredentials{ufrag: strings.TrimSpace(sessionUfrag), password: strings.TrimSpace(sessionPassword)} + } + if selected.ufrag == "" || selected.password == "" { + return iceCredentials{}, errors.New("SDP is missing ICE credentials") + } + return selected, nil +} + +func closeCandidateTunnels(tunnels []*tcpCandidateTunnel) error { + var closeErrors []error + for _, tunnel := range tunnels { + if errClose := tunnel.Close(); errClose != nil { + closeErrors = append(closeErrors, errClose) + } + } + return errors.Join(closeErrors...) +} + +func newTCPCandidateTunnel(target netip.AddrPort, dialer proxy.ContextDialer, expectedUser, remotePassword string) (*tcpCandidateTunnel, error) { + if !isPublicProxyTarget(target.Addr()) || target.Port() != 443 { + return nil, errors.New("Codex live TCP proxy target is not allowed") + } + if dialer == nil || strings.TrimSpace(expectedUser) == "" || strings.TrimSpace(remotePassword) == "" { + return nil, errors.New("Codex live TCP proxy tunnel configuration is incomplete") + } + network := "tcp4" + listenAddress := "127.0.0.1:0" + if target.Addr().Is6() { + network = "tcp6" + listenAddress = "[::1]:0" + } + listener, errListen := net.Listen(network, listenAddress) + if errListen != nil { + return nil, fmt.Errorf("listen for Codex live TCP proxy candidate: %w", errListen) + } + tunnelContext, cancelTunnel := context.WithCancel(context.Background()) + tunnel := &tcpCandidateTunnel{ + listener: listener, + target: target, + dialer: dialer, + expectedUser: expectedUser, + remotePassword: remotePassword, + connections: make(map[net.Conn]struct{}), + validationSlots: make(chan struct{}, maxUnauthenticatedTCPConns), + ctx: tunnelContext, + cancel: cancelTunnel, + } + go tunnel.accept() + return tunnel, nil +} + +func (t *tcpCandidateTunnel) accept() { + for { + connection, errAccept := t.listener.Accept() + if errAccept != nil { + if !errors.Is(errAccept, net.ErrClosed) { + log.WithError(errAccept).Warn("codex live TCP proxy: accept candidate connection failed") + } + return + } + if !t.trackConnection(connection) { + if errClose := connection.Close(); errClose != nil { + log.WithError(errClose).Debug("codex live TCP proxy: close connection after tunnel shutdown") + } + return + } + select { + case t.validationSlots <- struct{}{}: + go func() { + defer func() { <-t.validationSlots }() + t.handleConnection(connection) + }() + default: + t.untrackAndClose(connection) + log.Warn("codex live TCP proxy: rejected excess unauthenticated candidate connection") + } + } +} + +func (t *tcpCandidateTunnel) handleConnection(client net.Conn) { + firstFrame, errValidate := readValidatedICEBindingFrame(client, t.expectedUser, t.remotePassword) + if errValidate != nil { + t.untrackAndClose(client) + log.WithError(errValidate).Warn("codex live TCP proxy: rejected unauthenticated candidate connection") + return + } + if !t.claim() { + t.untrackAndClose(client) + return + } + if errClose := t.listener.Close(); errClose != nil && !errors.Is(errClose, net.ErrClosed) { + log.WithError(errClose).Debug("codex live TCP proxy: close claimed candidate listener") + } + upstream, errDial := t.dialer.DialContext(t.ctx, "tcp", t.target.String()) + if errDial != nil { + t.untrackAndClose(client) + log.WithError(errDial).Warn("codex live TCP proxy: connect fixed upstream candidate failed") + return + } + if !t.trackConnection(upstream) { + if errClose := upstream.Close(); errClose != nil { + log.WithError(errClose).Debug("codex live TCP proxy: close upstream after tunnel shutdown") + } + t.untrackAndClose(client) + return + } + if errWrite := writeAll(upstream, firstFrame); errWrite != nil { + t.untrackAndClose(upstream) + t.untrackAndClose(client) + log.WithError(errWrite).Warn("codex live TCP proxy: forward authenticated ICE frame failed") + return + } + + copyDone := make(chan struct{}, 2) + copyConnection := func(destination, source net.Conn) { + _, _ = io.Copy(destination, source) + copyDone <- struct{}{} + } + go copyConnection(upstream, client) + go copyConnection(client, upstream) + <-copyDone + t.untrackAndClose(upstream) + t.untrackAndClose(client) + <-copyDone +} + +func (t *tcpCandidateTunnel) trackConnection(connection net.Conn) bool { + t.mu.Lock() + defer t.mu.Unlock() + if t.closed { + return false + } + t.connections[connection] = struct{}{} + return true +} + +func (t *tcpCandidateTunnel) untrackAndClose(connection net.Conn) { + if connection == nil { + return + } + t.mu.Lock() + delete(t.connections, connection) + t.mu.Unlock() + if errClose := connection.Close(); errClose != nil && !errors.Is(errClose, net.ErrClosed) { + log.WithError(errClose).Debug("codex live TCP proxy: close tunnel connection") + } +} + +func (t *tcpCandidateTunnel) claim() bool { + t.mu.Lock() + defer t.mu.Unlock() + if t.closed || t.claimed { + return false + } + t.claimed = true + return true +} + +func (t *tcpCandidateTunnel) Close() error { + if t == nil { + return nil + } + t.mu.Lock() + if t.closed { + t.mu.Unlock() + return nil + } + t.closed = true + cancel := t.cancel + connections := make([]net.Conn, 0, len(t.connections)) + for connection := range t.connections { + connections = append(connections, connection) + } + t.connections = make(map[net.Conn]struct{}) + t.mu.Unlock() + if cancel != nil { + cancel() + } + + var closeErrors []error + if t.listener != nil { + if errClose := t.listener.Close(); errClose != nil && !errors.Is(errClose, net.ErrClosed) { + closeErrors = append(closeErrors, errClose) + } + } + for _, connection := range connections { + if errClose := connection.Close(); errClose != nil && !errors.Is(errClose, net.ErrClosed) { + closeErrors = append(closeErrors, errClose) + } + } + return errors.Join(closeErrors...) +} + +func readValidatedICEBindingFrame(connection io.Reader, expectedUser, remotePassword string) ([]byte, error) { + var header [2]byte + if _, errRead := io.ReadFull(connection, header[:]); errRead != nil { + return nil, fmt.Errorf("read ICE-TCP frame header: %w", errRead) + } + frameSize := int(binary.BigEndian.Uint16(header[:])) + if frameSize < stunMessageHeaderSize || frameSize > maxInitialSTUNFrameSize { + return nil, fmt.Errorf("invalid initial ICE-TCP STUN frame size %d", frameSize) + } + payload := make([]byte, frameSize) + if _, errRead := io.ReadFull(connection, payload); errRead != nil { + return nil, fmt.Errorf("read ICE-TCP STUN frame: %w", errRead) + } + message := stun.NewWithOptions(stun.WithStrict(true)) + if errDecode := stun.Decode(payload, message); errDecode != nil { + return nil, fmt.Errorf("decode initial ICE-TCP STUN message: %w", errDecode) + } + if len(payload) != stunMessageHeaderSize+int(message.Length) { + return nil, errors.New("initial ICE-TCP STUN message contains trailing data") + } + if message.Type != stun.BindingRequest { + return nil, fmt.Errorf("initial ICE-TCP STUN message has unexpected type %s", message.Type) + } + var username stun.Username + if errUsername := username.GetFrom(message); errUsername != nil { + return nil, fmt.Errorf("read initial ICE-TCP STUN username: %w", errUsername) + } + if string(username) != expectedUser { + return nil, errors.New("initial ICE-TCP STUN username does not match the media session") + } + if errIntegrity := stun.NewShortTermIntegrity(remotePassword).Check(message); errIntegrity != nil { + return nil, fmt.Errorf("verify initial ICE-TCP STUN integrity: %w", errIntegrity) + } + if errFingerprint := stun.Fingerprint.Check(message); errFingerprint != nil { + return nil, fmt.Errorf("verify initial ICE-TCP STUN fingerprint: %w", errFingerprint) + } + frame := make([]byte, len(header)+len(payload)) + copy(frame, header[:]) + copy(frame[len(header):], payload) + return frame, nil +} + +func writeAll(writer io.Writer, data []byte) error { + for len(data) > 0 { + written, errWrite := writer.Write(data) + if errWrite != nil { + return errWrite + } + if written <= 0 { + return io.ErrShortWrite + } + data = data[written:] + } + return nil +} + +func proxyScheme(rawProxyURL string) string { + trimmed := strings.TrimSpace(rawProxyURL) + if index := strings.Index(trimmed, "://"); index > 0 { + return strings.ToLower(trimmed[:index]) + } + return "proxy" +} diff --git a/internal/client/codex/live/tcp_proxy_test.go b/internal/client/codex/live/tcp_proxy_test.go new file mode 100644 index 000000000..bbc50d819 --- /dev/null +++ b/internal/client/codex/live/tcp_proxy_test.go @@ -0,0 +1,592 @@ +package live + +import ( + "bytes" + "context" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "net/netip" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/pion/sdp/v3" + "github.com/pion/stun/v3" + "github.com/pion/webrtc/v4" + "github.com/router-for-me/CLIProxyAPI/v7/internal/config" +) + +type recordedProxyDial struct { + address string + connection net.Conn +} + +type recordingProxyDialer struct { + mu sync.Mutex + dials chan recordedProxyDial + err error +} + +type blockingContextDialer struct { + started chan struct{} + canceled chan struct{} +} + +func (d *blockingContextDialer) DialContext(ctx context.Context, _, _ string) (net.Conn, error) { + close(d.started) + <-ctx.Done() + close(d.canceled) + return nil, ctx.Err() +} + +func (d *recordingProxyDialer) Dial(network, address string) (net.Conn, error) { + return d.DialContext(context.Background(), network, address) +} + +func (d *recordingProxyDialer) DialContext(ctx context.Context, _ string, address string) (net.Conn, error) { + if errContext := ctx.Err(); errContext != nil { + return nil, errContext + } + d.mu.Lock() + channel := d.dials + errDial := d.err + d.mu.Unlock() + if errDial != nil { + channel <- recordedProxyDial{address: address} + return nil, errDial + } + client, server := net.Pipe() + channel <- recordedProxyDial{address: address, connection: server} + return client, nil +} + +func TestPrepareProxiedUpstreamAnswerRestrictsAndRewritesCandidates(t *testing.T) { + dialer := &recordingProxyDialer{dials: make(chan recordedProxyDial, 1)} + answer := testProxySDP("remote-ufrag", "remote-password", []string{ + "1 1 udp 2130706431 20.42.0.10 3478 typ host", + "2 1 tcp 1671430143 20.42.0.20 443 typ host tcptype passive", + }) + localOffer := testProxySDP("local-ufrag", "local-password", nil) + + rewritten, tunnels, errPrepare := prepareProxiedUpstreamAnswer(answer, localOffer, dialer) + if errPrepare != nil { + t.Fatalf("prepareProxiedUpstreamAnswer returned error: %v", errPrepare) + } + defer func() { + if errClose := closeCandidateTunnels(tunnels); errClose != nil { + t.Errorf("close candidate tunnels: %v", errClose) + } + }() + if len(tunnels) != 1 { + t.Fatalf("tunnel count = %d, want 1", len(tunnels)) + } + if got := tunnels[0].target.String(); got != "20.42.0.20:443" { + t.Fatalf("fixed target = %q, want 20.42.0.20:443", got) + } + if tunnels[0].expectedUser != "remote-ufrag:local-ufrag" { + t.Fatalf("expected STUN username = %q", tunnels[0].expectedUser) + } + + var description sdp.SessionDescription + if errUnmarshal := description.UnmarshalString(rewritten); errUnmarshal != nil { + t.Fatalf("unmarshal rewritten SDP: %v", errUnmarshal) + } + var candidates []string + for _, media := range description.MediaDescriptions { + for _, attribute := range media.Attributes { + if attribute.IsICECandidate() { + candidates = append(candidates, attribute.Value) + } + } + } + if len(candidates) != 1 { + t.Fatalf("rewritten candidate count = %d, want 1: %v", len(candidates), candidates) + } + fields := strings.Fields(candidates[0]) + if len(fields) < 8 || fields[2] != "tcp" || fields[4] != "127.0.0.1" || fields[5] == "443" { + t.Fatalf("rewritten candidate = %q", candidates[0]) + } + if !strings.Contains(candidates[0], "tcptype passive") { + t.Fatalf("rewritten candidate lost passive TCP type: %q", candidates[0]) + } +} + +func TestPrepareProxiedUpstreamAnswerRejectsUnsafeTargets(t *testing.T) { + for name, candidate := range map[string]string{ + "private target": "1 1 tcp 1671430143 10.0.0.1 443 typ host tcptype passive", + "zero network target": "1 1 tcp 1671430143 0.0.0.1 443 typ host tcptype passive", + "carrier NAT target": "1 1 tcp 1671430143 100.64.0.1 443 typ host tcptype passive", + "reserved target": "1 1 tcp 1671430143 203.0.113.10 443 typ host tcptype passive", + "site-local IPv6 target": "1 1 tcp 1671430143 fec0::1 443 typ host tcptype passive", + "wrong port": "1 1 tcp 1671430143 20.42.0.10 8443 typ host tcptype passive", + "relay target": "1 1 tcp 1671430143 20.42.0.10 443 typ relay raddr 192.0.2.1 rport 5000 tcptype passive", + "active target": "1 1 tcp 1671430143 20.42.0.10 443 typ host tcptype active", + } { + t.Run(name, func(t *testing.T) { + dialer := &recordingProxyDialer{dials: make(chan recordedProxyDial, 1)} + _, tunnels, errPrepare := prepareProxiedUpstreamAnswer( + testProxySDP("remote", "remote-password", []string{candidate}), + testProxySDP("local", "local-password", nil), + dialer, + ) + if errPrepare == nil { + _ = closeCandidateTunnels(tunnels) + t.Fatal("expected unsafe candidate to be rejected") + } + }) + } +} + +func TestPrepareProxiedUpstreamAnswerLimitsCandidateCount(t *testing.T) { + candidates := make([]string, 0, maxUpstreamICECandidates+1) + for index := 0; index <= maxUpstreamICECandidates; index++ { + candidates = append(candidates, fmt.Sprintf("%d 1 udp 2130706431 20.42.0.10 3478 typ host", index+1)) + } + _, tunnels, errPrepare := prepareProxiedUpstreamAnswer( + testProxySDP("remote", "remote-password", candidates), + testProxySDP("local", "local-password", nil), + &recordingProxyDialer{dials: make(chan recordedProxyDial, 1)}, + ) + if errPrepare == nil || !strings.Contains(errPrepare.Error(), "candidate limit") { + _ = closeCandidateTunnels(tunnels) + t.Fatalf("error = %v, want candidate limit", errPrepare) + } +} + +func TestReadValidatedICEBindingFrame(t *testing.T) { + validFrame := buildTestICEFrame(t, "remote:local", "remote-password", true) + for name, testCase := range map[string]struct { + frame []byte + expectedUser string + password string + wantError bool + }{ + "valid": { + frame: validFrame, + expectedUser: "remote:local", + password: "remote-password", + }, + "wrong username": { + frame: validFrame, + expectedUser: "local:remote", + password: "remote-password", + wantError: true, + }, + "wrong password": { + frame: validFrame, + expectedUser: "remote:local", + password: "local-password", + wantError: true, + }, + "missing fingerprint": { + frame: buildTestICEFrame(t, "remote:local", "remote-password", false), + expectedUser: "remote:local", + password: "remote-password", + wantError: true, + }, + "undersized": { + frame: []byte{0, 1, 0}, + wantError: true, + }, + } { + t.Run(name, func(t *testing.T) { + validated, errValidate := readValidatedICEBindingFrame( + &fragmentedReader{data: testCase.frame, maximum: 3}, + testCase.expectedUser, + testCase.password, + ) + if testCase.wantError { + if errValidate == nil { + t.Fatal("expected validation error") + } + return + } + if errValidate != nil { + t.Fatalf("readValidatedICEBindingFrame returned error: %v", errValidate) + } + if !bytes.Equal(validated, testCase.frame) { + t.Fatal("validated frame changed") + } + }) + } +} + +func TestTCPCandidateTunnelAuthenticatesBeforeFixedTargetDial(t *testing.T) { + dialer := &recordingProxyDialer{dials: make(chan recordedProxyDial, 1)} + tunnel, errTunnel := newTCPCandidateTunnel( + netip.MustParseAddrPort("20.42.0.20:443"), + dialer, + "remote:local", + "remote-password", + ) + if errTunnel != nil { + t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel) + } + defer func() { _ = tunnel.Close() }() + + client, errDial := net.Dial("tcp", tunnel.listener.Addr().String()) + if errDial != nil { + t.Fatalf("dial candidate listener: %v", errDial) + } + defer func() { _ = client.Close() }() + frame := buildTestICEFrame(t, "remote:local", "remote-password", true) + if errWrite := writeAll(client, frame); errWrite != nil { + t.Fatalf("write authenticated frame: %v", errWrite) + } + + var dial recordedProxyDial + select { + case dial = <-dialer.dials: + case <-time.After(time.Second): + t.Fatal("proxy dial was not attempted after STUN authentication") + } + defer func() { _ = dial.connection.Close() }() + if dial.address != "20.42.0.20:443" { + t.Fatalf("proxy target = %q, want fixed candidate", dial.address) + } + forwarded := make([]byte, len(frame)) + if _, errRead := io.ReadFull(dial.connection, forwarded); errRead != nil { + t.Fatalf("read forwarded STUN frame: %v", errRead) + } + if !bytes.Equal(forwarded, frame) { + t.Fatal("forwarded STUN frame changed") + } + if errWrite := writeAll(dial.connection, []byte("reply")); errWrite != nil { + t.Fatalf("write tunnel reply: %v", errWrite) + } + reply := make([]byte, len("reply")) + if _, errRead := io.ReadFull(client, reply); errRead != nil { + t.Fatalf("read tunnel reply: %v", errRead) + } + if string(reply) != "reply" { + t.Fatalf("tunnel reply = %q", reply) + } +} + +func TestTCPCandidateTunnelCloseCancelsProxyDial(t *testing.T) { + dialer := &blockingContextDialer{ + started: make(chan struct{}), + canceled: make(chan struct{}), + } + tunnel, errTunnel := newTCPCandidateTunnel( + netip.MustParseAddrPort("20.42.0.20:443"), + dialer, + "remote:local", + "remote-password", + ) + if errTunnel != nil { + t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel) + } + client, errDial := net.Dial("tcp", tunnel.listener.Addr().String()) + if errDial != nil { + t.Fatalf("dial candidate listener: %v", errDial) + } + if errWrite := writeAll(client, buildTestICEFrame(t, "remote:local", "remote-password", true)); errWrite != nil { + t.Fatalf("write authenticated frame: %v", errWrite) + } + defer func() { _ = client.Close() }() + select { + case <-dialer.started: + case <-time.After(time.Second): + t.Fatal("proxy dial did not start") + } + if errClose := tunnel.Close(); errClose != nil { + t.Fatalf("close tunnel: %v", errClose) + } + select { + case <-dialer.canceled: + case <-time.After(time.Second): + t.Fatal("tunnel close did not cancel proxy dial") + } +} + +func TestTCPCandidateTunnelProxyFailureDoesNotFallBack(t *testing.T) { + dialer := &recordingProxyDialer{ + dials: make(chan recordedProxyDial, 1), + err: errors.New("proxy blocked"), + } + tunnel, errTunnel := newTCPCandidateTunnel( + netip.MustParseAddrPort("20.42.0.20:443"), + dialer, + "remote:local", + "remote-password", + ) + if errTunnel != nil { + t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel) + } + defer func() { _ = tunnel.Close() }() + client, errDial := net.Dial("tcp", tunnel.listener.Addr().String()) + if errDial != nil { + t.Fatalf("dial candidate listener: %v", errDial) + } + if errWrite := writeAll(client, buildTestICEFrame(t, "remote:local", "remote-password", true)); errWrite != nil { + t.Fatalf("write authenticated frame: %v", errWrite) + } + defer func() { _ = client.Close() }() + select { + case dial := <-dialer.dials: + if dial.address != "20.42.0.20:443" || dial.connection != nil { + t.Fatalf("failed proxy dial = %#v", dial) + } + case <-time.After(time.Second): + t.Fatal("proxy dial was not attempted") + } + if _, errSecondDial := net.Dial("tcp", tunnel.listener.Addr().String()); errSecondDial == nil { + t.Fatal("candidate listener remained available after proxy failure") + } +} + +func TestTCPCandidateTunnelRejectsUnauthenticatedConnectionWithoutDial(t *testing.T) { + dialer := &recordingProxyDialer{dials: make(chan recordedProxyDial, 1)} + tunnel, errTunnel := newTCPCandidateTunnel( + netip.MustParseAddrPort("20.42.0.20:443"), + dialer, + "remote:local", + "remote-password", + ) + if errTunnel != nil { + t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel) + } + defer func() { _ = tunnel.Close() }() + + client, errDial := net.Dial("tcp", tunnel.listener.Addr().String()) + if errDial != nil { + t.Fatalf("dial candidate listener: %v", errDial) + } + if errWrite := writeAll(client, buildTestICEFrame(t, "attacker:local", "remote-password", true)); errWrite != nil { + t.Fatalf("write unauthenticated frame: %v", errWrite) + } + _ = client.Close() + select { + case dial := <-dialer.dials: + _ = dial.connection.Close() + t.Fatalf("unauthenticated connection triggered proxy dial to %q", dial.address) + case <-time.After(100 * time.Millisecond): + } +} + +func TestPionActiveTCPCandidatePassesTunnelAuthentication(t *testing.T) { + localAPI, errAPI := newPionProxyAPI(config.CodexLiveMediaRelayConfig{}) + if errAPI != nil { + t.Fatalf("create local Pion API: %v", errAPI) + } + localPeer, errPeer := localAPI.NewPeerConnection(webrtc.Configuration{}) + if errPeer != nil { + t.Fatalf("create local PeerConnection: %v", errPeer) + } + defer func() { _ = localPeer.Close() }() + if _, errChannel := localPeer.CreateDataChannel(realtimeDataChannelLabel, nil); errChannel != nil { + t.Fatalf("create local DataChannel: %v", errChannel) + } + localGathering := webrtc.GatheringCompletePromise(localPeer) + localOffer, errOffer := localPeer.CreateOffer(nil) + if errOffer != nil { + t.Fatalf("create local offer: %v", errOffer) + } + if errLocal := localPeer.SetLocalDescription(localOffer); errLocal != nil { + t.Fatalf("set local offer: %v", errLocal) + } + <-localGathering + localDescription := localPeer.LocalDescription() + if localDescription == nil { + t.Fatal("local description is nil") + } + + tcpListener, errListen := net.Listen("tcp4", "127.0.0.1:0") + if errListen != nil { + t.Fatalf("listen for remote ICE-TCP: %v", errListen) + } + remoteSettings := webrtc.SettingEngine{} + remoteSettings.SetNetworkTypes([]webrtc.NetworkType{webrtc.NetworkTypeTCP4}) + remoteSettings.SetIncludeLoopbackCandidate(true) + remoteSettings.SetIPFilter(func(ip net.IP) bool { return ip != nil && ip.IsLoopback() }) + tcpMux := webrtc.NewICETCPMux(nil, tcpListener, 8) + remoteSettings.SetICETCPMux(tcpMux) + defer func() { _ = tcpMux.Close() }() + remoteAPI := webrtc.NewAPI(webrtc.WithSettingEngine(remoteSettings)) + remotePeer, errPeer := remoteAPI.NewPeerConnection(webrtc.Configuration{}) + if errPeer != nil { + t.Fatalf("create remote PeerConnection: %v", errPeer) + } + defer func() { _ = remotePeer.Close() }() + if errRemote := remotePeer.SetRemoteDescription(*localDescription); errRemote != nil { + t.Fatalf("set remote offer: %v", errRemote) + } + remoteGathering := webrtc.GatheringCompletePromise(remotePeer) + remoteAnswer, errAnswer := remotePeer.CreateAnswer(nil) + if errAnswer != nil { + t.Fatalf("create remote answer: %v", errAnswer) + } + if errLocal := remotePeer.SetLocalDescription(remoteAnswer); errLocal != nil { + t.Fatalf("set remote answer: %v", errLocal) + } + <-remoteGathering + remoteDescription := remotePeer.LocalDescription() + if remoteDescription == nil { + t.Fatal("remote description is nil") + } + publicAnswer := rewriteTestTCPCandidateTarget(t, remoteDescription.SDP, "20.42.0.20", 443) + + dialer := &recordingProxyDialer{dials: make(chan recordedProxyDial, 1)} + rewrittenAnswer, tunnels, errPrepare := prepareProxiedUpstreamAnswer(publicAnswer, localDescription.SDP, dialer) + if errPrepare != nil { + t.Fatalf("prepare proxied Pion answer: %v", errPrepare) + } + defer func() { _ = closeCandidateTunnels(tunnels) }() + if errRemote := localPeer.SetRemoteDescription(webrtc.SessionDescription{ + Type: webrtc.SDPTypeAnswer, + SDP: rewrittenAnswer, + }); errRemote != nil { + t.Fatalf("set rewritten remote answer: %v", errRemote) + } + + var dial recordedProxyDial + select { + case dial = <-dialer.dials: + case <-time.After(5 * time.Second): + t.Fatal("Pion active ICE-TCP did not reach the authenticated tunnel") + } + defer func() { _ = dial.connection.Close() }() + localCredentials, errCredentials := bundledICECredentialsFromString(localDescription.SDP) + if errCredentials != nil { + t.Fatalf("read local credentials: %v", errCredentials) + } + remoteCredentials, errCredentials := bundledICECredentialsFromString(publicAnswer) + if errCredentials != nil { + t.Fatalf("read remote credentials: %v", errCredentials) + } + if _, errValidate := readValidatedICEBindingFrame( + dial.connection, + remoteCredentials.ufrag+":"+localCredentials.ufrag, + remoteCredentials.password, + ); errValidate != nil { + t.Fatalf("forwarded Pion STUN request failed validation: %v", errValidate) + } +} + +func TestBundledICECredentialsRejectsMixedCredentials(t *testing.T) { + mixed := strings.Replace( + testProxySDP("first", "first-password", nil), + "a=mid:1\r\na=ice-ufrag:first\r\na=ice-pwd:first-password", + "a=mid:1\r\na=ice-ufrag:second\r\na=ice-pwd:second-password", + 1, + ) + var description sdp.SessionDescription + if errUnmarshal := description.UnmarshalString(mixed); errUnmarshal != nil { + t.Fatalf("unmarshal mixed SDP: %v", errUnmarshal) + } + if _, errCredentials := bundledICECredentials(&description); errCredentials == nil { + t.Fatal("expected inconsistent bundled credentials to be rejected") + } +} + +func buildTestICEFrame(t *testing.T, username, password string, fingerprint bool) []byte { + t.Helper() + setters := []stun.Setter{ + stun.BindingRequest, + stun.TransactionID, + stun.NewUsername(username), + stun.NewShortTermIntegrity(password), + } + if fingerprint { + setters = append(setters, stun.Fingerprint) + } + message, errBuild := stun.Build(setters...) + if errBuild != nil { + t.Fatalf("build STUN request: %v", errBuild) + } + if len(message.Raw) > int(^uint16(0)) { + t.Fatal("test STUN request is too large") + } + frame := make([]byte, 2+len(message.Raw)) + binary.BigEndian.PutUint16(frame[:2], uint16(len(message.Raw))) + copy(frame[2:], message.Raw) + return frame +} + +func testProxySDP(ufrag, password string, candidates []string) string { + var builder strings.Builder + _, _ = fmt.Fprintf(&builder, "v=0\r\no=- 1 1 IN IP4 127.0.0.1\r\ns=-\r\nt=0 0\r\na=group:BUNDLE 0 1\r\n") + for _, media := range []struct { + line string + mid string + }{ + {line: "m=audio 9 UDP/TLS/RTP/SAVPF 111", mid: "0"}, + {line: "m=application 9 UDP/DTLS/SCTP webrtc-datachannel", mid: "1"}, + } { + _, _ = fmt.Fprintf(&builder, "%s\r\nc=IN IP4 0.0.0.0\r\na=mid:%s\r\na=ice-ufrag:%s\r\na=ice-pwd:%s\r\n", media.line, media.mid, ufrag, password) + if media.mid == "0" { + for _, candidate := range candidates { + _, _ = fmt.Fprintf(&builder, "a=candidate:%s\r\n", candidate) + } + } + } + return builder.String() +} + +type fragmentedReader struct { + data []byte + maximum int +} + +func (r *fragmentedReader) Read(destination []byte) (int, error) { + if len(r.data) == 0 { + return 0, io.EOF + } + limit := len(destination) + if limit > r.maximum { + limit = r.maximum + } + if limit > len(r.data) { + limit = len(r.data) + } + copy(destination, r.data[:limit]) + r.data = r.data[limit:] + return limit, nil +} + +func rewriteTestTCPCandidateTarget(t *testing.T, rawSDP, address string, port int) string { + t.Helper() + var description sdp.SessionDescription + if errUnmarshal := description.UnmarshalString(rawSDP); errUnmarshal != nil { + t.Fatalf("unmarshal test SDP: %v", errUnmarshal) + } + rewritten := 0 + for _, media := range description.MediaDescriptions { + for index := range media.Attributes { + attribute := &media.Attributes[index] + if !attribute.IsICECandidate() { + continue + } + fields := strings.Fields(attribute.Value) + if len(fields) < 8 || !strings.EqualFold(fields[2], "tcp") || !strings.Contains(attribute.Value, "tcptype passive") { + continue + } + fields[4] = address + fields[5] = strconv.Itoa(port) + attribute.Value = strings.Join(fields, " ") + rewritten++ + } + } + if rewritten == 0 { + t.Fatal("test SDP has no passive TCP candidate") + } + marshaled, errMarshal := description.Marshal() + if errMarshal != nil { + t.Fatalf("marshal test SDP: %v", errMarshal) + } + return string(marshaled) +} + +func bundledICECredentialsFromString(rawSDP string) (iceCredentials, error) { + var description sdp.SessionDescription + if errUnmarshal := description.UnmarshalString(rawSDP); errUnmarshal != nil { + return iceCredentials{}, errUnmarshal + } + return bundledICECredentials(&description) +} diff --git a/sdk/proxyutil/proxy.go b/sdk/proxyutil/proxy.go index 507d5e09e..acead5f0c 100644 --- a/sdk/proxyutil/proxy.go +++ b/sdk/proxyutil/proxy.go @@ -5,6 +5,7 @@ import ( "context" "crypto/tls" "encoding/base64" + "errors" "fmt" "net" "net/http" @@ -156,15 +157,36 @@ type httpConnectDialer struct { } func (d *httpConnectDialer) Dial(network, addr string) (net.Conn, error) { - proxyConn, errDial := d.dialer.Dial(network, proxyDialAddr(d.proxyURL)) + return d.DialContext(context.Background(), network, addr) +} + +func (d *httpConnectDialer) DialContext(ctx context.Context, network, addr string) (net.Conn, error) { + if ctx == nil { + ctx = context.Background() + } + contextDialer, ok := d.dialer.(proxy.ContextDialer) + if !ok { + return nil, errors.New("HTTP proxy base dialer does not support context cancellation") + } + proxyConn, errDial := contextDialer.DialContext(ctx, network, proxyDialAddr(d.proxyURL)) if errDial != nil { return nil, fmt.Errorf("dial HTTP proxy failed: %w", errDial) } conn := proxyConn + cancelDone := make(chan struct{}) + stopCancel := context.AfterFunc(ctx, func() { + _ = proxyConn.Close() + close(cancelDone) + }) + defer func() { + if !stopCancel() { + <-cancelDone + } + }() if d.proxyURL.Scheme == "https" { tlsConn := tls.Client(conn, &tls.Config{ServerName: d.proxyURL.Hostname()}) - if errHandshake := tlsConn.Handshake(); errHandshake != nil { + if errHandshake := tlsConn.HandshakeContext(ctx); errHandshake != nil { if errClose := conn.Close(); errClose != nil { return nil, fmt.Errorf("HTTPS proxy TLS handshake failed: %w; close failed: %v", errHandshake, errClose) } @@ -173,12 +195,12 @@ func (d *httpConnectDialer) Dial(network, addr string) (net.Conn, error) { conn = tlsConn } - req := &http.Request{ + req := (&http.Request{ Method: http.MethodConnect, URL: &url.URL{Host: addr}, Host: addr, Header: make(http.Header), - } + }).WithContext(ctx) if d.proxyURL.User != nil { req.Header.Set("Proxy-Authorization", proxyAuthorization(d.proxyURL.User)) } @@ -207,6 +229,12 @@ func (d *httpConnectDialer) Dial(network, addr string) (net.Conn, error) { return nil, fmt.Errorf("proxy CONNECT returned status %s", resp.Status) } + if errContext := ctx.Err(); errContext != nil { + if errClose := conn.Close(); errClose != nil && !errors.Is(errClose, net.ErrClosed) { + return nil, fmt.Errorf("HTTP proxy context ended: %w; close failed: %v", errContext, errClose) + } + return nil, errContext + } if reader.Buffered() > 0 { return &bufferedConn{Conn: conn, reader: reader}, nil } diff --git a/sdk/proxyutil/proxy_test.go b/sdk/proxyutil/proxy_test.go index 1c957ef7a..5c154c9bf 100644 --- a/sdk/proxyutil/proxy_test.go +++ b/sdk/proxyutil/proxy_test.go @@ -2,6 +2,7 @@ package proxyutil import ( "bufio" + "context" "encoding/base64" "fmt" "io" @@ -269,6 +270,80 @@ func TestBuildDialerHTTPProxyCONNECT(t *testing.T) { } } +func TestBuildDialerHTTPProxyCONNECTCancellation(t *testing.T) { + t.Parallel() + + listener, errListen := net.Listen("tcp", "127.0.0.1:0") + if errListen != nil { + t.Fatalf("net.Listen returned error: %v", errListen) + } + defer func() { _ = listener.Close() }() + requestRead := make(chan struct{}) + serverDone := make(chan error, 1) + go func() { + connection, errAccept := listener.Accept() + if errAccept != nil { + serverDone <- errAccept + return + } + defer func() { _ = connection.Close() }() + if _, errRead := http.ReadRequest(bufio.NewReader(connection)); errRead != nil { + serverDone <- errRead + return + } + close(requestRead) + if errDeadline := connection.SetReadDeadline(time.Now().Add(5 * time.Second)); errDeadline != nil { + serverDone <- errDeadline + return + } + var buffer [1]byte + _, errRead := connection.Read(buffer[:]) + serverDone <- errRead + }() + + dialer, mode, errBuild := BuildDialer("http://" + listener.Addr().String()) + if errBuild != nil || mode != ModeProxy { + t.Fatalf("BuildDialer mode=%d error=%v", mode, errBuild) + } + contextDialer, ok := dialer.(interface { + DialContext(context.Context, string, string) (net.Conn, error) + }) + if !ok { + t.Fatal("HTTP CONNECT dialer does not support context cancellation") + } + ctx, cancel := context.WithCancel(context.Background()) + dialDone := make(chan error, 1) + go func() { + connection, errDial := contextDialer.DialContext(ctx, "tcp", "20.42.0.20:443") + if connection != nil { + _ = connection.Close() + } + dialDone <- errDial + }() + select { + case <-requestRead: + case <-time.After(time.Second): + t.Fatal("proxy did not receive CONNECT request") + } + cancel() + select { + case errDial := <-dialDone: + if errDial == nil { + t.Fatal("canceled CONNECT dial returned nil error") + } + case <-time.After(time.Second): + t.Fatal("canceled CONNECT dial did not return") + } + select { + case errServer := <-serverDone: + if errServer == nil { + t.Fatal("proxy connection stayed open after cancellation") + } + case <-time.After(time.Second): + t.Fatal("proxy connection was not closed after cancellation") + } +} + func TestRedactProxyURL(t *testing.T) { t.Parallel()