lywsvip/CLIProxyAPI
1
0
Fork 0
mirror of https://github.com/router-for-me/CLIProxyAPI.git synced 2026-09-03 06:35:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(live): add TCP proxy for Codex Live WebRTC relay

Browse Source 
- Implemented a TCP proxy for WebRTC candidate tunneling in Codex Live, supporting passive TCP candidates on port 443.
- Restricted tunneling to globally routable public IPs and added safeguards for rejecting unsafe/private targets.
- Added robust validation of STUN BindingRequest frames before forwarding to upstream candidates.
- Includes extensive unit tests for proxying behavior, candidate validation, and tunnel edge cases.
This commit is contained in:
Luis Pater
2026-07-25 07:12:28 +08:00
parent ebc7446929
commit 49be36aef6
11 changed files with 1469 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
config.example.yaml
View File

@@ -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"

6
go.mod
View File

@@ -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

2
internal/client/codex/live/live.go
View File

@@ -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

26
internal/client/codex/live/live_test.go
View File

@@ -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"`
}

150
internal/client/codex/live/media.go
View File

@@ -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))
}

84
internal/client/codex/live/media_test.go
View File

@@ -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{}

4
internal/client/codex/live/sideband.go
View File

@@ -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)
}

525
internal/client/codex/live/tcp_proxy.go Normal file
View File

@@ -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"
}

592
internal/client/codex/live/tcp_proxy_test.go Normal file
View File

@@ -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)
}

36
sdk/proxyutil/proxy.go
View File

@@ -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
}

75
sdk/proxyutil/proxy_test.go
View File

@@ -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()