From ebc74469292ccafeef9b549347d5398170faaadf Mon Sep 17 00:00:00 2001 From: Luis Pater Date: Sat, 25 Jul 2026 06:09:50 +0800 Subject: [PATCH] feat(diff): enhance config diff and relay updates for Codex Live Media - Added comprehensive diffing for Codex live media relay settings, including support for public IP, UDP port ranges, and ICE server changes. - Introduced `displayOptionalValue` utility to handle optional values in diff outputs. - Improved test coverage for config change detection, ensuring no sensitive information leakage. - Replaced `allow-private-remote-ips` with the new `disable-private-remote-ips` property, adding YAML backward compatibility. - Updated Codex live handler to differentiate media relay configuration changes and runtime updates. --- config.example.yaml | 8 +- internal/client/codex/live/live.go | 116 ++++++++++--- internal/client/codex/live/live_test.go | 53 +++++- internal/client/codex/live/media.go | 188 +++++++++++++++++----- internal/client/codex/live/media_test.go | 74 ++++++++- internal/client/codex/live/sideband.go | 21 +-- internal/config/codex_live.go | 43 +++++ internal/config/codex_live_test.go | 35 +++- internal/config/config.go | 16 +- internal/watcher/config_reload.go | 4 +- internal/watcher/diff/config_diff.go | 49 +++++- internal/watcher/diff/config_diff_test.go | 78 ++++++++- internal/watcher/diff/openai_compat.go | 2 +- 13 files changed, 578 insertions(+), 109 deletions(-) diff --git a/config.example.yaml b/config.example.yaml index 58eaa03e5..ba0146b50 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -223,15 +223,17 @@ codex: enabled: false # Maximum concurrent media sessions. Zero uses the default of 32. max-sessions: 32 - # Allow downstream SDP candidates that target private, loopback, link-local, or unspecified IPs. - # Enable only when Codex Desktop reaches CPA over a trusted local network. - allow-private-remote-ips: false + # Reject downstream SDP candidates that target private, loopback, link-local, or unspecified IPs. + # Keep false for local or trusted-network Codex Desktop connections. + disable-private-remote-ips: false # Public IPv4 or IPv6 address advertised when CPA is behind 1:1 NAT. public-ip: "" # Optional UDP allocation range. Both values must be set together and provide at least two ports per session. 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. # ice-servers: # - urls: # - "stun:stun.example.com:3478" diff --git a/internal/client/codex/live/live.go b/internal/client/codex/live/live.go index ac2949258..77f0c2d48 100644 --- a/internal/client/codex/live/live.go +++ b/internal/client/codex/live/live.go @@ -11,6 +11,7 @@ import ( "mime" "mime/multipart" "net/http" + "reflect" "strings" "sync" @@ -40,13 +41,16 @@ var liveProtocolHeaders = []string{ // Handler forwards Codex live session requests through the shared auth scheduler. type Handler struct { - authManager *auth.Manager - cfg *config.Config - sessions *sessionStore - sidebandAPIBaseURL string - mediaRelayMu sync.RWMutex - mediaRelay mediaRelayFactory - mediaRelayErr error + authManager *auth.Manager + cfg *config.Config + sessions *sessionStore + sidebandAPIBaseURL string + mediaRelayMu sync.RWMutex + mediaRelay mediaRelayFactory + mediaRelayErr error + mediaRelayConfig config.CodexLiveMediaRelayConfig + mediaRelayConfigured bool + mediaLimiter *mediaSessionLimiter } // NewHandler creates a Codex live session handler. @@ -57,7 +61,9 @@ func NewHandler(authManager *auth.Manager, cfg *config.Config) *Handler { sessions: newSessionStore(), sidebandAPIBaseURL: defaultSidebandAPIBaseURL, } - _ = handler.UpdateConfig(cfg) + if errUpdate := handler.UpdateConfig(cfg); errUpdate != nil { + log.WithError(errUpdate).Error("failed to configure Codex Live media relay") + } return handler } @@ -66,18 +72,81 @@ func (h *Handler) UpdateConfig(cfg *config.Config) error { if h == nil { return nil } - var relay mediaRelayFactory - var relayErr error - if cfg != nil && cfg.Codex.LiveMediaRelay.Enabled { - relay, relayErr = newPionMediaRelay(cfg.Codex.LiveMediaRelay) + var relayConfig config.CodexLiveMediaRelayConfig + if cfg != nil { + relayConfig = cfg.Codex.LiveMediaRelay } h.mediaRelayMu.Lock() + previousConfig := h.mediaRelayConfig + previouslyConfigured := h.mediaRelayConfigured + h.cfg = cfg + if previouslyConfigured && reflect.DeepEqual(previousConfig, relayConfig) { + currentErr := h.mediaRelayErr + h.mediaRelayMu.Unlock() + return currentErr + } + if h.mediaLimiter == nil { + h.mediaLimiter = &mediaSessionLimiter{} + } + var relay mediaRelayFactory + var relayErr error + if relayConfig.Enabled { + relay, relayErr = newPionMediaRelayWithLimiter(relayConfig, h.mediaLimiter) + } h.mediaRelay = relay h.mediaRelayErr = relayErr + h.mediaRelayConfig = relayConfig + h.mediaRelayConfigured = true h.mediaRelayMu.Unlock() + + if relayErr == nil && (previouslyConfigured || relayConfig.Enabled) { + message := "codex live media relay configured" + if previouslyConfigured { + message = "codex live media relay configuration reloaded; changes apply to new sessions" + } + log.WithFields(liveMediaConfigLogFields(relayConfig)).Info(message) + } return relayErr } +func liveMediaConfigLogFields(relayConfig config.CodexLiveMediaRelayConfig) log.Fields { + publicIP := strings.TrimSpace(relayConfig.PublicIP) + if publicIP == "" { + publicIP = "auto" + } + return log.Fields{ + "enabled": relayConfig.Enabled, + "max_sessions": relayConfig.EffectiveMaxSessions(), + "disable_private_remote_ips": relayConfig.DisablePrivateRemoteIPs, + "public_ip": publicIP, + "udp_port_min": relayConfig.UDPPortMin, + "udp_port_max": relayConfig.UDPPortMax, + "ice_server_count": len(relayConfig.ICEServers), + } +} + +func (h *Handler) currentRuntime() (*config.Config, mediaRelayFactory, error) { + if h == nil { + return nil, nil, nil + } + h.mediaRelayMu.RLock() + cfg := h.cfg + relay := h.mediaRelay + relayErr := h.mediaRelayErr + h.mediaRelayMu.RUnlock() + return cfg, relay, relayErr +} + +func (h *Handler) currentConfig() *config.Config { + if h == nil { + return nil + } + h.mediaRelayMu.RLock() + cfg := h.cfg + h.mediaRelayMu.RUnlock() + return cfg +} + func (h *Handler) currentMediaRelay() (mediaRelayFactory, error) { if h == nil { return nil, nil @@ -117,7 +186,7 @@ func (h *Handler) Handle(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"error": errPayload.Error()}) return } - mediaRelay, mediaRelayErr := h.currentMediaRelay() + runtimeConfig, mediaRelay, mediaRelayErr := h.currentRuntime() if mediaRelayErr != nil { c.JSON(http.StatusServiceUnavailable, gin.H{"error": mediaRelayErr.Error()}) return @@ -176,7 +245,7 @@ func (h *Handler) Handle(c *gin.Context) { } defer func() { if !mediaRetained { - if errClose := mediaSession.Close(); errClose != nil { + if errClose := mediaSession.CloseWithReason("request_not_retained"); errClose != nil { log.WithError(errClose).Debug("codex live media: close unretained session") } } @@ -201,7 +270,7 @@ func (h *Handler) Handle(c *gin.Context) { } authType, authValue := selected.AccountInfo() - helps.RecordAPIRequest(ctx, h.cfg, helps.UpstreamRequestLog{ + helps.RecordAPIRequest(ctx, runtimeConfig, helps.UpstreamRequestLog{ URL: upstreamCallURL, Method: http.MethodPost, Headers: headersForLogging(req.Header), @@ -225,7 +294,7 @@ func (h *Handler) Handle(c *gin.Context) { if selection != nil { selection.End("request_failed") } - helps.RecordAPIResponseError(ctx, h.cfg, errRequest) + helps.RecordAPIResponseError(ctx, runtimeConfig, errRequest) c.JSON(http.StatusBadGateway, gin.H{"error": errRequest.Error()}) return } @@ -251,10 +320,10 @@ func (h *Handler) Handle(c *gin.Context) { } responseHeaders := callResponseHeaders(resp.Header) - helps.RecordAPIResponseMetadata(ctx, h.cfg, resp.StatusCode, responseHeaders) + helps.RecordAPIResponseMetadata(ctx, runtimeConfig, resp.StatusCode, responseHeaders) responseBody, errResponse := readLimitedBody(resp.Body) if errResponse != nil { - helps.RecordAPIResponseError(ctx, h.cfg, errResponse) + helps.RecordAPIResponseError(ctx, runtimeConfig, errResponse) message := "Failed to read Codex live response" if errors.Is(errResponse, errBodyTooLarge) { message = "Codex live response body too large" @@ -262,7 +331,7 @@ func (h *Handler) Handle(c *gin.Context) { c.JSON(http.StatusBadGateway, gin.H{"error": message}) return } - helps.AppendAPIResponseChunk(ctx, h.cfg, responseBody) + helps.AppendAPIResponseChunk(ctx, runtimeConfig, responseBody) responseBodyToWrite := responseBody success := resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices if success && mediaSession != nil { @@ -288,10 +357,15 @@ func (h *Handler) Handle(c *gin.Context) { return } if callID != "" { + if mediaSession != nil { + mediaSession.SetCallID(callID) + } session := liveSession{authID: selected.ID, model: model, media: mediaSession} if selection != nil { if mediaSession != nil { - if errBind := selection.Bind(mediaSession.Close); errBind != nil { + if errBind := selection.Bind(func() error { + return mediaSession.CloseWithReason("home_selection_closed") + }); errBind != nil { selection.End("media_bind_failed") c.JSON(http.StatusServiceUnavailable, gin.H{"error": errBind.Error()}) return @@ -325,7 +399,7 @@ func (h *Handler) Handle(c *gin.Context) { if sessionStored { h.sessions.complete(storedSession, "response_write_failed") } - helps.RecordAPIResponseError(ctx, h.cfg, errWrite) + helps.RecordAPIResponseError(ctx, runtimeConfig, errWrite) log.WithError(errWrite).Warn("codex live: write response body failed") } } diff --git a/internal/client/codex/live/live_test.go b/internal/client/codex/live/live_test.go index 931cb8eff..93889fedc 100644 --- a/internal/client/codex/live/live_test.go +++ b/internal/client/codex/live/live_test.go @@ -163,6 +163,8 @@ type fakeMediaSession struct { upstreamAnswer string downstreamSDP string closeHandler func(string) + callID string + closeReason string closed atomic.Bool err error } @@ -172,11 +174,20 @@ func (s *fakeMediaSession) AcceptUpstreamAnswer(_ context.Context, answer string return s.downstreamSDP, s.err } +func (s *fakeMediaSession) SetCallID(callID string) { + s.callID = callID +} + func (s *fakeMediaSession) SetCloseHandler(handler func(string)) { s.closeHandler = handler } func (s *fakeMediaSession) Close() error { + return s.CloseWithReason("closed") +} + +func (s *fakeMediaSession) CloseWithReason(reason string) error { + s.closeReason = reason s.closed.Store(true) return nil } @@ -356,6 +367,9 @@ func TestHandlerRelaysWebRTCMediaSDP(t *testing.T) { if mediaSession.upstreamAnswer != "v=0\r\no=upstream-answer\r\n" { t.Fatalf("accepted upstream answer = %q", mediaSession.upstreamAnswer) } + if mediaSession.callID != "call-123" { + t.Fatalf("media call ID = %q, want call-123", mediaSession.callID) + } if got := recorder.Body.String(); got != mediaSession.downstreamSDP { t.Fatalf("downstream SDP = %q, want %q", got, mediaSession.downstreamSDP) } @@ -432,6 +446,9 @@ func TestHandlerClosesUnretainedMediaSession(t *testing.T) { if !mediaSession.closed.Load() { t.Fatal("failed request retained its media session") } + if mediaSession.closeReason != "request_not_retained" { + t.Fatalf("media close reason = %q, want request_not_retained", mediaSession.closeReason) + } if _, ok := handler.sessions.peek("call-123"); ok { t.Fatal("failed request stored its media session") } @@ -503,6 +520,9 @@ func TestHandlerClosesMediaWhenResponseWriteFails(t *testing.T) { if !mediaSession.closed.Load() { t.Fatal("response write failure retained its media session") } + if mediaSession.closeReason != "response_write_failed" { + t.Fatalf("media close reason = %q, want response_write_failed", mediaSession.closeReason) + } if _, ok := handler.sessions.peek("call-123"); ok { t.Fatal("response write failure retained a stored session") } @@ -762,15 +782,38 @@ func TestHandlerUpdatesMediaRelayConfig(t *testing.T) { t.Fatalf("initial media relay = %#v, error = %v", relay, errRelay) } enabled := &config.Config{Codex: config.CodexConfig{LiveMediaRelay: config.CodexLiveMediaRelayConfig{ - Enabled: true, - MaxSessions: 1, - AllowPrivateRemoteIPs: true, + Enabled: true, + MaxSessions: 1, + DisablePrivateRemoteIPs: false, }}} if errUpdate := handler.UpdateConfig(enabled); errUpdate != nil { t.Fatalf("enable media relay: %v", errUpdate) } - if relay, errRelay := handler.currentMediaRelay(); relay == nil || errRelay != nil { - t.Fatalf("enabled media relay = %#v, error = %v", relay, errRelay) + enabledRelay, errRelay := handler.currentMediaRelay() + if enabledRelay == nil || errRelay != nil { + t.Fatalf("enabled media relay = %#v, error = %v", enabledRelay, errRelay) + } + unchanged := *enabled + unchanged.Debug = true + unchanged.ProxyURL = "http://new-proxy.example" + if errUpdate := handler.UpdateConfig(&unchanged); errUpdate != nil { + t.Fatalf("apply unrelated config change: %v", errUpdate) + } + unchangedRelay, errRelay := handler.currentMediaRelay() + if unchangedRelay != enabledRelay || errRelay != nil { + t.Fatalf("unrelated config change rebuilt media relay: before=%#v after=%#v error=%v", enabledRelay, unchangedRelay, errRelay) + } + if current := handler.currentConfig(); current == nil || current.ProxyURL != "http://new-proxy.example" { + t.Fatalf("runtime config was not updated: %#v", current) + } + changed := *enabled + changed.Codex.LiveMediaRelay.MaxSessions = 2 + if errUpdate := handler.UpdateConfig(&changed); errUpdate != nil { + t.Fatalf("reload media relay: %v", errUpdate) + } + changedRelay, errRelay := handler.currentMediaRelay() + if changedRelay == nil || changedRelay == enabledRelay || errRelay != nil { + t.Fatalf("changed media relay = %#v, previous=%#v error=%v", changedRelay, enabledRelay, errRelay) } if errUpdate := handler.UpdateConfig(&config.Config{}); errUpdate != nil { t.Fatalf("disable media relay: %v", errUpdate) diff --git a/internal/client/codex/live/media.go b/internal/client/codex/live/media.go index ba7b77ad1..164ab455f 100644 --- a/internal/client/codex/live/media.go +++ b/internal/client/codex/live/media.go @@ -9,6 +9,7 @@ import ( "strings" "sync" + "github.com/google/uuid" "github.com/pion/interceptor" "github.com/pion/rtp" "github.com/pion/webrtc/v4" @@ -32,8 +33,10 @@ var opusCodec = webrtc.RTPCodecCapability{ type mediaRelaySession interface { AcceptUpstreamAnswer(context.Context, string) (string, error) + SetCallID(string) SetCloseHandler(func(string)) Close() error + CloseWithReason(string) error } type mediaRelayFactory interface { @@ -44,7 +47,13 @@ type pionMediaRelay struct { downstreamAPI *webrtc.API upstreamAPI *webrtc.API configuration webrtc.Configuration - slots chan struct{} + limiter *mediaSessionLimiter +} + +type mediaSessionLimiter struct { + mu sync.Mutex + limit int + active int } type pionMediaSession struct { @@ -52,15 +61,17 @@ type pionMediaSession struct { upstream *webrtc.PeerConnection bridge *dataChannelBridge - done chan struct{} - closeOnce sync.Once - closeErr error - failureOnce sync.Once - handlerMu sync.Mutex - onClose func(string) - failureReason string - handlerCalled bool - releaseSlot func() + done chan struct{} + closeOnce sync.Once + closeErr error + failureOnce sync.Once + handlerMu sync.Mutex + onClose func(string) + failureReason string + handlerCalled bool + mediaSessionID string + callID string + releaseSlot func() } type dataChannelMessage struct { @@ -92,10 +103,14 @@ type dataChannelBridge struct { } func newPionMediaRelay(relayConfig config.CodexLiveMediaRelayConfig) (*pionMediaRelay, error) { + return newPionMediaRelayWithLimiter(relayConfig, &mediaSessionLimiter{}) +} + +func newPionMediaRelayWithLimiter(relayConfig config.CodexLiveMediaRelayConfig, limiter *mediaSessionLimiter) (*pionMediaRelay, error) { if errValidate := relayConfig.Validate(); errValidate != nil { return nil, errValidate } - downstreamAPI, errAPI := newPionAPI(relayConfig, !relayConfig.AllowPrivateRemoteIPs) + downstreamAPI, errAPI := newPionAPI(relayConfig, relayConfig.DisablePrivateRemoteIPs) if errAPI != nil { return nil, errAPI } @@ -116,14 +131,51 @@ func newPionMediaRelay(relayConfig config.CodexLiveMediaRelayConfig) (*pionMedia CredentialType: webrtc.ICECredentialTypePassword, }) } + if limiter == nil { + limiter = &mediaSessionLimiter{} + } + limiter.setLimit(relayConfig.EffectiveMaxSessions()) return &pionMediaRelay{ downstreamAPI: downstreamAPI, upstreamAPI: upstreamAPI, configuration: webrtc.Configuration{ICEServers: iceServers}, - slots: make(chan struct{}, relayConfig.EffectiveMaxSessions()), + limiter: limiter, }, nil } +func (l *mediaSessionLimiter) setLimit(limit int) { + if l == nil { + return + } + l.mu.Lock() + l.limit = limit + l.mu.Unlock() +} + +func (l *mediaSessionLimiter) acquire() bool { + if l == nil { + return false + } + l.mu.Lock() + defer l.mu.Unlock() + if l.limit <= 0 || l.active >= l.limit { + return false + } + l.active++ + return true +} + +func (l *mediaSessionLimiter) release() { + if l == nil { + return + } + l.mu.Lock() + if l.active > 0 { + l.active-- + } + l.mu.Unlock() +} + func newPionAPI(relayConfig config.CodexLiveMediaRelayConfig, filterPrivateRemoteIPs bool) (*webrtc.API, error) { mediaEngine := &webrtc.MediaEngine{} if errRegister := mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{ @@ -161,17 +213,16 @@ func isPublicRemoteIP(ip net.IP) bool { } func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer string) (mediaRelaySession, string, error) { - if r == nil || r.downstreamAPI == nil || r.upstreamAPI == nil { + if r == nil || r.downstreamAPI == nil || r.upstreamAPI == nil || r.limiter == nil { return nil, "", errors.New("Codex live media relay unavailable") } - select { - case r.slots <- struct{}{}: - case <-ctx.Done(): - return nil, "", ctx.Err() - default: + if errContext := ctx.Err(); errContext != nil { + return nil, "", errContext + } + if !r.limiter.acquire() { return nil, "", errors.New("Codex live media relay capacity exhausted") } - releaseSlot := func() { <-r.slots } + releaseSlot := r.limiter.release downstream, errDownstream := r.downstreamAPI.NewPeerConnection(r.configuration) if errDownstream != nil { releaseSlot() @@ -187,15 +238,17 @@ func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer string) (me } session := &pionMediaSession{ - downstream: downstream, - upstream: upstream, - done: make(chan struct{}), - releaseSlot: releaseSlot, + downstream: downstream, + upstream: upstream, + done: make(chan struct{}), + mediaSessionID: uuid.NewString(), + releaseSlot: releaseSlot, } session.bridge = newDataChannelBridge(session.done, func(err error) { session.fail("data_channel_failed", err) }) session.installStateHandlers() + log.WithFields(session.logFields("session")).Info("codex live WebRTC media session created") if errRemote := downstream.SetRemoteDescription(webrtc.SessionDescription{ Type: webrtc.SDPTypeOffer, @@ -311,6 +364,29 @@ func (s *pionMediaSession) AcceptUpstreamAnswer(ctx context.Context, upstreamAns return localDescription.SDP, nil } +func (s *pionMediaSession) SetCallID(callID string) { + if s == nil { + return + } + s.handlerMu.Lock() + s.callID = strings.TrimSpace(callID) + s.handlerMu.Unlock() +} + +func (s *pionMediaSession) logFields(peer string) log.Fields { + fields := log.Fields{ + "media_session_id": s.mediaSessionID, + "peer": peer, + } + s.handlerMu.Lock() + callID := s.callID + s.handlerMu.Unlock() + if callID != "" { + fields["call_id"] = callID + } + return fields +} + func (s *pionMediaSession) SetCloseHandler(handler func(string)) { if s == nil { return @@ -329,59 +405,95 @@ func (s *pionMediaSession) SetCloseHandler(handler func(string)) { } func (s *pionMediaSession) Close() error { + return s.CloseWithReason("closed") +} + +func (s *pionMediaSession) CloseWithReason(reason string) error { if s == nil { return nil } s.closeOnce.Do(func() { + fields := s.logFields("session") + fields["reason"] = reason + log.WithFields(fields).Info("codex live WebRTC media session closing") close(s.done) if s.bridge != nil { s.bridge.close() } var closeErrors []error - if s.downstream != nil { - if errClose := s.downstream.Close(); errClose != nil { - closeErrors = append(closeErrors, fmt.Errorf("close downstream PeerConnection: %w", errClose)) - } + if errClose := s.closePeerConnection("local", s.downstream); errClose != nil { + closeErrors = append(closeErrors, fmt.Errorf("close downstream PeerConnection: %w", errClose)) } - if s.upstream != nil { - if errClose := s.upstream.Close(); errClose != nil { - closeErrors = append(closeErrors, fmt.Errorf("close upstream PeerConnection: %w", errClose)) - } + if errClose := s.closePeerConnection("remote", s.upstream); errClose != nil { + closeErrors = append(closeErrors, fmt.Errorf("close upstream PeerConnection: %w", errClose)) } if s.releaseSlot != nil { s.releaseSlot() } s.closeErr = errors.Join(closeErrors...) + if s.closeErr != nil { + log.WithFields(fields).WithError(s.closeErr).Warn("codex live WebRTC media session closed with errors") + } else { + log.WithFields(fields).Info("codex live WebRTC media session closed") + } }) return s.closeErr } +func (s *pionMediaSession) closePeerConnection(peer string, connection *webrtc.PeerConnection) error { + if connection == nil { + return nil + } + fields := s.logFields(peer) + fields["state_before"] = connection.ConnectionState().String() + errClose := connection.Close() + fields["state_after"] = connection.ConnectionState().String() + if errClose != nil { + log.WithFields(fields).WithError(errClose).Warn("codex live WebRTC peer close failed") + return errClose + } + log.WithFields(fields).Info("codex live WebRTC peer closed") + return nil +} + func (s *pionMediaSession) installStateHandlers() { - handle := func(leg string) func(webrtc.PeerConnectionState) { + handle := func(peer, reasonPrefix string) func(webrtc.PeerConnectionState) { return func(state webrtc.PeerConnectionState) { + fields := s.logFields(peer) + fields["state"] = state.String() switch state { + case webrtc.PeerConnectionStateConnecting: + log.WithFields(fields).Info("codex live WebRTC peer connecting") + case webrtc.PeerConnectionStateConnected: + log.WithFields(fields).Info("codex live WebRTC peer connected") + case webrtc.PeerConnectionStateDisconnected: + log.WithFields(fields).Warn("codex live WebRTC peer disconnected") case webrtc.PeerConnectionStateFailed: - s.fail(leg+"_failed", fmt.Errorf("%s PeerConnection failed", leg)) + log.WithFields(fields).Warn("codex live WebRTC peer failed") + s.fail(reasonPrefix+"_failed", fmt.Errorf("%s PeerConnection failed", reasonPrefix)) case webrtc.PeerConnectionStateClosed: select { case <-s.done: return default: - s.fail(leg+"_closed", fmt.Errorf("%s PeerConnection closed", leg)) + log.WithFields(fields).Info("codex live WebRTC peer closed by remote") + s.fail(reasonPrefix+"_closed", fmt.Errorf("%s PeerConnection closed", reasonPrefix)) } + default: + log.WithFields(fields).Debug("codex live WebRTC peer state changed") } } } - s.downstream.OnConnectionStateChange(handle("downstream")) - s.upstream.OnConnectionStateChange(handle("upstream")) + s.downstream.OnConnectionStateChange(handle("local", "downstream")) + s.upstream.OnConnectionStateChange(handle("remote", "upstream")) } func (s *pionMediaSession) fail(reason string, err error) { s.failureOnce.Do(func() { if err != nil { - log.WithError(err).Debug("codex live media relay closed") + log.WithFields(s.logFields("session")).WithField("reason", reason).WithError(err).Warn("codex live WebRTC media session failed") } - if errClose := s.Close(); errClose != nil { + if errClose := s.CloseWithReason(reason); errClose != nil { log.WithError(errClose).Debug("codex live media: close failed session") } s.handlerMu.Lock() diff --git a/internal/client/codex/live/media_test.go b/internal/client/codex/live/media_test.go index a972500f0..a09d723cc 100644 --- a/internal/client/codex/live/media_test.go +++ b/internal/client/codex/live/media_test.go @@ -10,9 +10,20 @@ import ( "github.com/pion/rtp" "github.com/pion/webrtc/v4" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" + log "github.com/sirupsen/logrus" + logtest "github.com/sirupsen/logrus/hooks/test" ) func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) { + logger := log.StandardLogger() + previousHooks := logger.ReplaceHooks(make(log.LevelHooks)) + previousLevel := logger.GetLevel() + logger.SetLevel(log.DebugLevel) + hook := logtest.NewLocal(logger) + defer func() { + logger.ReplaceHooks(previousHooks) + logger.SetLevel(previousLevel) + }() clientAPI := newTestWebRTCAPI(t) client, errClient := clientAPI.NewPeerConnection(webrtc.Configuration{}) if errClient != nil { @@ -49,11 +60,12 @@ func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) { }) clientOffer := completeOffer(t, client) - relay, errRelay := newPionMediaRelay(config.CodexLiveMediaRelayConfig{ - Enabled: true, - MaxSessions: 1, - AllowPrivateRemoteIPs: true, - }) + relayConfig := config.CodexLiveMediaRelayConfig{ + Enabled: true, + MaxSessions: 1, + DisablePrivateRemoteIPs: false, + } + relay, errRelay := newPionMediaRelay(relayConfig) if errRelay != nil { t.Fatalf("create media relay: %v", errRelay) } @@ -61,13 +73,18 @@ func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) { if errSession != nil { t.Fatalf("create media relay session: %v", errSession) } + session.SetCallID("call-log-test") defer func() { if errClose := session.Close(); errClose != nil { t.Errorf("close media relay session: %v", errClose) } }() - if _, _, errCapacity := relay.NewSession(context.Background(), clientOffer); errCapacity == nil { - t.Fatal("media relay accepted a session beyond its configured capacity") + reloadedRelay, errRelay := newPionMediaRelayWithLimiter(relayConfig, relay.limiter) + if errRelay != nil { + t.Fatalf("reload media relay: %v", errRelay) + } + if _, _, errCapacity := reloadedRelay.NewSession(context.Background(), clientOffer); errCapacity == nil { + t.Fatal("reloaded media relay bypassed the shared session capacity") } upstreamAPI := newTestWebRTCAPI(t) @@ -145,6 +162,21 @@ func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) { sendTestRTP(t, clientAudio, clientPayload, upstreamAudioMessages) upstreamPayload := []byte{0xf8, 0xfe, 0xfd} sendTestRTP(t, upstreamAudio, upstreamPayload, clientAudioMessages) + if errClose := session.Close(); errClose != nil { + t.Fatalf("close media relay session for logging: %v", errClose) + } + replacementSession, _, errReplacement := reloadedRelay.NewSession(context.Background(), clientOffer) + if errReplacement != nil { + t.Fatalf("shared capacity was not released: %v", errReplacement) + } + if errClose := replacementSession.CloseWithReason("test_complete"); errClose != nil { + t.Fatalf("close replacement media session: %v", errClose) + } + for _, peer := range []string{"local", "remote"} { + assertPeerLog(t, hook, "codex live WebRTC peer connected", peer, "call-log-test") + assertPeerLog(t, hook, "codex live WebRTC peer closed", peer, "call-log-test") + } + assertSessionLog(t, hook, "codex live WebRTC media session closed", "closed", "call-log-test") } func TestIsPublicRemoteIP(t *testing.T) { @@ -287,6 +319,34 @@ func sendTestRTP(t *testing.T, track *webrtc.TrackLocalStaticRTP, payload []byte t.Fatal("RTP packet was not relayed") } +func assertSessionLog(t *testing.T, hook *logtest.Hook, message, reason, callID string) { + t.Helper() + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + for _, entry := range hook.AllEntries() { + if entry.Message == message && entry.Data["reason"] == reason && entry.Data["call_id"] == callID { + return + } + } + time.Sleep(time.Millisecond) + } + t.Fatalf("missing session log message %q for reason %q and call %q", message, reason, callID) +} + +func assertPeerLog(t *testing.T, hook *logtest.Hook, message, peer, callID string) { + t.Helper() + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + for _, entry := range hook.AllEntries() { + if entry.Message == message && entry.Data["peer"] == peer && entry.Data["call_id"] == callID { + return + } + } + time.Sleep(time.Millisecond) + } + t.Fatalf("missing log message %q for peer %q and call %q", message, peer, callID) +} + func closeTestPeerConnection(t *testing.T, connection *webrtc.PeerConnection) { t.Helper() if errClose := connection.Close(); errClose != nil { diff --git a/internal/client/codex/live/sideband.go b/internal/client/codex/live/sideband.go index f16ecb924..b27808466 100644 --- a/internal/client/codex/live/sideband.go +++ b/internal/client/codex/live/sideband.go @@ -113,7 +113,7 @@ func (s *sessionStore) put(callID string, session liveSession) liveSession { previous.session.resources.close() } if previous.session.media != nil && previous.session.media != session.media { - if errClose := previous.session.media.Close(); errClose != nil { + if errClose := previous.session.media.CloseWithReason("session_replaced"); errClose != nil { log.WithError(errClose).Debug("codex live media: close replaced session") } } @@ -237,7 +237,7 @@ func endLiveSession(session liveSession, reason string) { session.resources.close() } if session.media != nil { - if errClose := session.media.Close(); errClose != nil { + if errClose := session.media.CloseWithReason(reason); errClose != nil { log.WithError(errClose).Debug("codex live media: close stored session") } } @@ -305,6 +305,7 @@ func (h *Handler) HandleSideband(c *gin.Context) { c.JSON(http.StatusServiceUnavailable, gin.H{"error": "Codex live sideband unavailable"}) return } + runtimeConfig := h.currentConfig() if !websocket.IsWebSocketUpgrade(c.Request) { c.JSON(http.StatusUpgradeRequired, gin.H{"error": "WebSocket upgrade required"}) return @@ -392,7 +393,7 @@ func (h *Handler) HandleSideband(c *gin.Context) { } authType, authValue := selected.AccountInfo() - helps.RecordAPIWebsocketRequest(ctx, h.cfg, helps.UpstreamRequestLog{ + helps.RecordAPIWebsocketRequest(ctx, runtimeConfig, helps.UpstreamRequestLog{ URL: upstreamURL, Method: "WEBSOCKET", Headers: headersForLogging(req.Header), @@ -403,15 +404,15 @@ func (h *Handler) HandleSideband(c *gin.Context) { AuthValue: authValue, }) - dialer := newProxyAwareSidebandDialer(h.cfg, selected) + dialer := newProxyAwareSidebandDialer(runtimeConfig, selected) dialer.Subprotocols = websocket.Subprotocols(c.Request) upstream, handshakeResponse, errDial := dialer.DialContext(ctx, upstreamURL, req.Header) if errDial != nil { - handleSidebandDialError(c, ctx, h, handshakeResponse, errDial) + handleSidebandDialError(c, ctx, runtimeConfig, handshakeResponse, errDial) return } if handshakeResponse != nil { - helps.RecordAPIWebsocketHandshake(ctx, h.cfg, handshakeResponse.StatusCode, callResponseHeaders(handshakeResponse.Header)) + helps.RecordAPIWebsocketHandshake(ctx, runtimeConfig, handshakeResponse.StatusCode, callResponseHeaders(handshakeResponse.Header)) if handshakeResponse.Body != nil { if errClose := handshakeResponse.Body.Close(); errClose != nil { log.Errorf("codex live sideband: close handshake response body error: %v", errClose) @@ -454,7 +455,7 @@ func (h *Handler) HandleSideband(c *gin.Context) { consumeSession = true if errRelay := relayWebsockets(downstream, upstream); errRelay != nil && !isNormalWebsocketClose(errRelay) { - helps.RecordAPIWebsocketError(ctx, h.cfg, "relay", errRelay) + helps.RecordAPIWebsocketError(ctx, runtimeConfig, "relay", errRelay) log.WithError(errRelay).Debug("codex live sideband relay closed") } } @@ -524,20 +525,20 @@ func callIDFromLocation(location string) string { return callID } -func handleSidebandDialError(c *gin.Context, ctx context.Context, h *Handler, response *http.Response, errDial error) { +func handleSidebandDialError(c *gin.Context, ctx context.Context, cfg *config.Config, response *http.Response, errDial error) { status := http.StatusBadGateway if response != nil { if response.StatusCode > 0 { status = response.StatusCode } - helps.RecordAPIWebsocketHandshake(ctx, h.cfg, response.StatusCode, callResponseHeaders(response.Header)) + helps.RecordAPIWebsocketHandshake(ctx, cfg, response.StatusCode, callResponseHeaders(response.Header)) if response.Body != nil { if errClose := response.Body.Close(); errClose != nil { log.Errorf("codex live sideband: close rejected handshake body error: %v", errClose) } } } - helps.RecordAPIWebsocketError(ctx, h.cfg, "dial", errDial) + helps.RecordAPIWebsocketError(ctx, cfg, "dial", errDial) c.JSON(status, gin.H{"error": "Codex live sideband upstream unavailable"}) } diff --git a/internal/config/codex_live.go b/internal/config/codex_live.go index 611ed5ed3..5fbc54e66 100644 --- a/internal/config/codex_live.go +++ b/internal/config/codex_live.go @@ -6,11 +6,54 @@ import ( "net" "net/url" "strings" + + log "github.com/sirupsen/logrus" + "gopkg.in/yaml.v3" ) // DefaultCodexLiveMediaMaxSessions is the default in-process media session limit. const DefaultCodexLiveMediaMaxSessions = 32 +// UnmarshalYAML supports the deprecated allow-private-remote-ips setting while +// preserving the default behavior of allowing private downstream candidates. +func (c *CodexLiveMediaRelayConfig) UnmarshalYAML(value *yaml.Node) error { + type plain CodexLiveMediaRelayConfig + var decoded plain + if errDecode := value.Decode(&decoded); errDecode != nil { + return errDecode + } + var allowPrivate *bool + var disablePrivate *bool + if value.Kind == yaml.MappingNode { + for index := 0; index+1 < len(value.Content); index += 2 { + key := value.Content[index].Value + switch key { + case "allow-private-remote-ips": + var setting bool + if errDecode := value.Content[index+1].Decode(&setting); errDecode != nil { + return fmt.Errorf("decode codex.live-media-relay.allow-private-remote-ips: %w", errDecode) + } + allowPrivate = &setting + case "disable-private-remote-ips": + var setting bool + if errDecode := value.Content[index+1].Decode(&setting); errDecode != nil { + return fmt.Errorf("decode codex.live-media-relay.disable-private-remote-ips: %w", errDecode) + } + disablePrivate = &setting + } + } + } + if allowPrivate != nil && disablePrivate != nil { + return errors.New("codex.live-media-relay cannot set both allow-private-remote-ips and disable-private-remote-ips") + } + if allowPrivate != nil { + decoded.DisablePrivateRemoteIPs = !*allowPrivate + log.Warn("codex.live-media-relay.allow-private-remote-ips is deprecated; use disable-private-remote-ips with the inverse value") + } + *c = CodexLiveMediaRelayConfig(decoded) + return nil +} + // EffectiveMaxSessions returns the configured media session limit. func (c CodexLiveMediaRelayConfig) EffectiveMaxSessions() int { if c.MaxSessions > 0 { diff --git a/internal/config/codex_live_test.go b/internal/config/codex_live_test.go index d9f93e13b..583a9e40b 100644 --- a/internal/config/codex_live_test.go +++ b/internal/config/codex_live_test.go @@ -14,7 +14,7 @@ func TestCodexLiveMediaRelayConfigParsesAndValidates(t *testing.T) { live-media-relay: enabled: true max-sessions: 64 - allow-private-remote-ips: true + disable-private-remote-ips: true public-ip: "203.0.113.10" udp-port-min: 40000 udp-port-max: 40150 @@ -28,7 +28,7 @@ func TestCodexLiveMediaRelayConfigParsesAndValidates(t *testing.T) { t.Fatalf("unmarshal Codex Live media relay config: %v", errUnmarshal) } relay := cfg.Codex.LiveMediaRelay - if !relay.Enabled || relay.MaxSessions != 64 || !relay.AllowPrivateRemoteIPs || relay.PublicIP != "203.0.113.10" { + if !relay.Enabled || relay.MaxSessions != 64 || !relay.DisablePrivateRemoteIPs || relay.PublicIP != "203.0.113.10" { t.Fatalf("parsed media relay = %#v", relay) } if relay.UDPPortMin != 40000 || relay.UDPPortMax != 40150 { @@ -44,8 +44,35 @@ func TestCodexLiveMediaRelayConfigParsesAndValidates(t *testing.T) { if errMarshal != nil { t.Fatalf("marshal media relay config: %v", errMarshal) } - if strings.Contains(string(encoded), "relay-secret") || strings.Contains(string(encoded), "credential") { - t.Fatalf("JSON media relay config leaked TURN credential: %s", encoded) + for _, sensitive := range []string{"relay-secret", "credential", "relay-user", "username"} { + if strings.Contains(string(encoded), sensitive) { + t.Fatalf("JSON media relay config leaked TURN field %q: %s", sensitive, encoded) + } + } +} + +func TestCodexLiveMediaRelayConfigMigratesLegacyPrivateIPSetting(t *testing.T) { + for name, raw := range map[string]string{ + "legacy allow true": "allow-private-remote-ips: true\n", + "legacy allow false": "allow-private-remote-ips: false\n", + "new default": "enabled: true\n", + } { + t.Run(name, func(t *testing.T) { + var relay CodexLiveMediaRelayConfig + if errUnmarshal := yaml.Unmarshal([]byte(raw), &relay); errUnmarshal != nil { + t.Fatalf("unmarshal media relay config: %v", errUnmarshal) + } + wantDisabled := name == "legacy allow false" + if relay.DisablePrivateRemoteIPs != wantDisabled { + t.Fatalf("disable-private-remote-ips = %t, want %t", relay.DisablePrivateRemoteIPs, wantDisabled) + } + }) + } + + var relay CodexLiveMediaRelayConfig + errUnmarshal := yaml.Unmarshal([]byte("allow-private-remote-ips: true\ndisable-private-remote-ips: false\n"), &relay) + if errUnmarshal == nil { + t.Fatal("accepted conflicting private IP settings") } } diff --git a/internal/config/config.go b/internal/config/config.go index c0e27b312..b4353f1fe 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -299,19 +299,19 @@ type CodexConfig struct { // CodexLiveMediaRelayConfig configures the in-process Codex Live WebRTC gateway. type CodexLiveMediaRelayConfig struct { - Enabled bool `yaml:"enabled" json:"enabled"` - MaxSessions int `yaml:"max-sessions" json:"max-sessions"` - AllowPrivateRemoteIPs bool `yaml:"allow-private-remote-ips" json:"allow-private-remote-ips"` - PublicIP string `yaml:"public-ip" json:"public-ip"` - UDPPortMin uint16 `yaml:"udp-port-min" json:"udp-port-min"` - UDPPortMax uint16 `yaml:"udp-port-max" json:"udp-port-max"` - ICEServers []CodexLiveICEServer `yaml:"ice-servers" json:"ice-servers"` + Enabled bool `yaml:"enabled" json:"enabled"` + MaxSessions int `yaml:"max-sessions" json:"max-sessions"` + DisablePrivateRemoteIPs bool `yaml:"disable-private-remote-ips" json:"disable-private-remote-ips"` + PublicIP string `yaml:"public-ip" json:"public-ip"` + UDPPortMin uint16 `yaml:"udp-port-min" json:"udp-port-min"` + UDPPortMax uint16 `yaml:"udp-port-max" json:"udp-port-max"` + ICEServers []CodexLiveICEServer `yaml:"ice-servers" json:"ice-servers"` } // CodexLiveICEServer configures a STUN or TURN server for the media relay. type CodexLiveICEServer struct { URLs []string `yaml:"urls" json:"urls"` - Username string `yaml:"username" json:"username"` + Username string `yaml:"username" json:"-"` Credential string `yaml:"credential" json:"-"` } diff --git a/internal/watcher/config_reload.go b/internal/watcher/config_reload.go index 92c386492..68b5916da 100644 --- a/internal/watcher/config_reload.go +++ b/internal/watcher/config_reload.go @@ -125,9 +125,9 @@ func (w *Watcher) reloadConfig() bool { if oldConfig != nil { details := diff.BuildConfigChangeDetails(oldConfig, newConfig) if len(details) > 0 { - log.Debugf("config changes detected:") + log.Info("config changes detected:") for _, d := range details { - log.Debugf(" %s", d) + log.Infof(" %s", d) } } else { log.Debugf("no material config field changes detected") diff --git a/internal/watcher/diff/config_diff.go b/internal/watcher/diff/config_diff.go index e554f36b3..820fe6009 100644 --- a/internal/watcher/diff/config_diff.go +++ b/internal/watcher/diff/config_diff.go @@ -108,6 +108,29 @@ func BuildConfigChangeDetails(oldCfg, newCfg *config.Config) []string { if oldCfg.Codex.OptimizeMultiAgentV2 != newCfg.Codex.OptimizeMultiAgentV2 { changes = append(changes, fmt.Sprintf("codex.optimize-multi-agent-v2: %t -> %t", oldCfg.Codex.OptimizeMultiAgentV2, newCfg.Codex.OptimizeMultiAgentV2)) } + oldLiveRelay := oldCfg.Codex.LiveMediaRelay + newLiveRelay := newCfg.Codex.LiveMediaRelay + if oldLiveRelay.Enabled != newLiveRelay.Enabled { + changes = append(changes, fmt.Sprintf("codex.live-media-relay.enabled: %t -> %t", oldLiveRelay.Enabled, newLiveRelay.Enabled)) + } + if oldLiveRelay.MaxSessions != newLiveRelay.MaxSessions { + changes = append(changes, fmt.Sprintf("codex.live-media-relay.max-sessions: %d -> %d", oldLiveRelay.MaxSessions, newLiveRelay.MaxSessions)) + } + if oldLiveRelay.DisablePrivateRemoteIPs != newLiveRelay.DisablePrivateRemoteIPs { + changes = append(changes, fmt.Sprintf("codex.live-media-relay.disable-private-remote-ips: %t -> %t", oldLiveRelay.DisablePrivateRemoteIPs, newLiveRelay.DisablePrivateRemoteIPs)) + } + if strings.TrimSpace(oldLiveRelay.PublicIP) != strings.TrimSpace(newLiveRelay.PublicIP) { + changes = append(changes, fmt.Sprintf("codex.live-media-relay.public-ip: %s -> %s", displayOptionalValue(oldLiveRelay.PublicIP), displayOptionalValue(newLiveRelay.PublicIP))) + } + if oldLiveRelay.UDPPortMin != newLiveRelay.UDPPortMin { + changes = append(changes, fmt.Sprintf("codex.live-media-relay.udp-port-min: %d -> %d", oldLiveRelay.UDPPortMin, newLiveRelay.UDPPortMin)) + } + if oldLiveRelay.UDPPortMax != newLiveRelay.UDPPortMax { + changes = append(changes, fmt.Sprintf("codex.live-media-relay.udp-port-max: %d -> %d", oldLiveRelay.UDPPortMax, newLiveRelay.UDPPortMax)) + } + if !reflect.DeepEqual(oldLiveRelay.ICEServers, newLiveRelay.ICEServers) { + changes = append(changes, fmt.Sprintf("codex.live-media-relay.ice-servers: updated (%d -> %d entries, credentials redacted)", len(oldLiveRelay.ICEServers), len(newLiveRelay.ICEServers))) + } if oldCfg.Routing.Strategy != newCfg.Routing.Strategy { changes = append(changes, fmt.Sprintf("routing.strategy: %s -> %s", oldCfg.Routing.Strategy, newCfg.Routing.Strategy)) @@ -129,7 +152,7 @@ func BuildConfigChangeDetails(oldCfg, newCfg *config.Config) []string { o := oldCfg.GeminiKey[i] n := newCfg.GeminiKey[i] if strings.TrimSpace(o.BaseURL) != strings.TrimSpace(n.BaseURL) { - changes = append(changes, fmt.Sprintf("gemini[%d].base-url: %s -> %s", i, strings.TrimSpace(o.BaseURL), strings.TrimSpace(n.BaseURL))) + changes = append(changes, fmt.Sprintf("gemini[%d].base-url: %s -> %s", i, formatURL(o.BaseURL), formatURL(n.BaseURL))) } if strings.TrimSpace(o.ProxyURL) != strings.TrimSpace(n.ProxyURL) { changes = append(changes, fmt.Sprintf("gemini[%d].proxy-url: %s -> %s", i, formatProxyURL(o.ProxyURL), formatProxyURL(n.ProxyURL))) @@ -162,7 +185,7 @@ func BuildConfigChangeDetails(oldCfg, newCfg *config.Config) []string { o := oldCfg.InteractionsKey[i] n := newCfg.InteractionsKey[i] if strings.TrimSpace(o.BaseURL) != strings.TrimSpace(n.BaseURL) { - changes = append(changes, fmt.Sprintf("interactions[%d].base-url: %s -> %s", i, strings.TrimSpace(o.BaseURL), strings.TrimSpace(n.BaseURL))) + changes = append(changes, fmt.Sprintf("interactions[%d].base-url: %s -> %s", i, formatURL(o.BaseURL), formatURL(n.BaseURL))) } if strings.TrimSpace(o.ProxyURL) != strings.TrimSpace(n.ProxyURL) { changes = append(changes, fmt.Sprintf("interactions[%d].proxy-url: %s -> %s", i, formatProxyURL(o.ProxyURL), formatProxyURL(n.ProxyURL))) @@ -197,7 +220,7 @@ func BuildConfigChangeDetails(oldCfg, newCfg *config.Config) []string { o := oldCfg.ClaudeKey[i] n := newCfg.ClaudeKey[i] if strings.TrimSpace(o.BaseURL) != strings.TrimSpace(n.BaseURL) { - changes = append(changes, fmt.Sprintf("claude[%d].base-url: %s -> %s", i, strings.TrimSpace(o.BaseURL), strings.TrimSpace(n.BaseURL))) + changes = append(changes, fmt.Sprintf("claude[%d].base-url: %s -> %s", i, formatURL(o.BaseURL), formatURL(n.BaseURL))) } if strings.TrimSpace(o.ProxyURL) != strings.TrimSpace(n.ProxyURL) { changes = append(changes, fmt.Sprintf("claude[%d].proxy-url: %s -> %s", i, formatProxyURL(o.ProxyURL), formatProxyURL(n.ProxyURL))) @@ -246,7 +269,7 @@ func BuildConfigChangeDetails(oldCfg, newCfg *config.Config) []string { o := oldCfg.CodexKey[i] n := newCfg.CodexKey[i] if strings.TrimSpace(o.BaseURL) != strings.TrimSpace(n.BaseURL) { - changes = append(changes, fmt.Sprintf("codex[%d].base-url: %s -> %s", i, strings.TrimSpace(o.BaseURL), strings.TrimSpace(n.BaseURL))) + changes = append(changes, fmt.Sprintf("codex[%d].base-url: %s -> %s", i, formatURL(o.BaseURL), formatURL(n.BaseURL))) } if strings.TrimSpace(o.ProxyURL) != strings.TrimSpace(n.ProxyURL) { changes = append(changes, fmt.Sprintf("codex[%d].proxy-url: %s -> %s", i, formatProxyURL(o.ProxyURL), formatProxyURL(n.ProxyURL))) @@ -284,7 +307,7 @@ func BuildConfigChangeDetails(oldCfg, newCfg *config.Config) []string { o := oldCfg.XAIKey[i] n := newCfg.XAIKey[i] if strings.TrimSpace(o.BaseURL) != strings.TrimSpace(n.BaseURL) { - changes = append(changes, fmt.Sprintf("xai[%d].base-url: %s -> %s", i, strings.TrimSpace(o.BaseURL), strings.TrimSpace(n.BaseURL))) + changes = append(changes, fmt.Sprintf("xai[%d].base-url: %s -> %s", i, formatURL(o.BaseURL), formatURL(n.BaseURL))) } if strings.TrimSpace(o.ProxyURL) != strings.TrimSpace(n.ProxyURL) { changes = append(changes, fmt.Sprintf("xai[%d].proxy-url: %s -> %s", i, formatProxyURL(o.ProxyURL), formatProxyURL(n.ProxyURL))) @@ -340,7 +363,7 @@ func BuildConfigChangeDetails(oldCfg, newCfg *config.Config) []string { oldPanelRepo := strings.TrimSpace(oldCfg.RemoteManagement.PanelGitHubRepository) newPanelRepo := strings.TrimSpace(newCfg.RemoteManagement.PanelGitHubRepository) if oldPanelRepo != newPanelRepo { - changes = append(changes, fmt.Sprintf("remote-management.panel-github-repository: %s -> %s", oldPanelRepo, newPanelRepo)) + changes = append(changes, fmt.Sprintf("remote-management.panel-github-repository: %s -> %s", formatURL(oldPanelRepo), formatURL(newPanelRepo))) } if oldCfg.RemoteManagement.SecretKey != newCfg.RemoteManagement.SecretKey { switch { @@ -369,7 +392,7 @@ func BuildConfigChangeDetails(oldCfg, newCfg *config.Config) []string { o := oldCfg.VertexCompatAPIKey[i] n := newCfg.VertexCompatAPIKey[i] if strings.TrimSpace(o.BaseURL) != strings.TrimSpace(n.BaseURL) { - changes = append(changes, fmt.Sprintf("vertex[%d].base-url: %s -> %s", i, strings.TrimSpace(o.BaseURL), strings.TrimSpace(n.BaseURL))) + changes = append(changes, fmt.Sprintf("vertex[%d].base-url: %s -> %s", i, formatURL(o.BaseURL), formatURL(n.BaseURL))) } if strings.TrimSpace(o.ProxyURL) != strings.TrimSpace(n.ProxyURL) { changes = append(changes, fmt.Sprintf("vertex[%d].proxy-url: %s -> %s", i, formatProxyURL(o.ProxyURL), formatProxyURL(n.ProxyURL))) @@ -442,7 +465,19 @@ func equalStringMap(a, b map[string]string) bool { return true } +func displayOptionalValue(raw string) string { + trimmed := strings.TrimSpace(raw) + if trimmed == "" { + return "" + } + return trimmed +} + func formatProxyURL(raw string) string { + return formatURL(raw) +} + +func formatURL(raw string) string { trimmed := strings.TrimSpace(raw) if trimmed == "" { return "" diff --git a/internal/watcher/diff/config_diff_test.go b/internal/watcher/diff/config_diff_test.go index 936a3eb04..f0e6aa21d 100644 --- a/internal/watcher/diff/config_diff_test.go +++ b/internal/watcher/diff/config_diff_test.go @@ -1,6 +1,7 @@ package diff import ( + "strings" "testing" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" @@ -93,6 +94,46 @@ func TestBuildConfigChangeDetails_NoChanges(t *testing.T) { } } +func TestBuildConfigChangeDetails_CodexLiveMediaRelay(t *testing.T) { + oldCfg := &config.Config{Codex: config.CodexConfig{LiveMediaRelay: config.CodexLiveMediaRelayConfig{ + Enabled: false, + MaxSessions: 16, + ICEServers: []config.CodexLiveICEServer{{ + URLs: []string{"turn:old.example.com"}, + Username: "old-user", + Credential: "old-secret", + }}, + }}} + newCfg := &config.Config{Codex: config.CodexConfig{LiveMediaRelay: config.CodexLiveMediaRelayConfig{ + Enabled: true, + MaxSessions: 32, + DisablePrivateRemoteIPs: true, + PublicIP: "203.0.113.10", + UDPPortMin: 40000, + UDPPortMax: 40063, + ICEServers: []config.CodexLiveICEServer{{ + URLs: []string{"turn:new.example.com"}, + Username: "new-user", + Credential: "new-secret", + }}, + }}} + + details := BuildConfigChangeDetails(oldCfg, newCfg) + expectContains(t, details, "codex.live-media-relay.enabled: false -> true") + expectContains(t, details, "codex.live-media-relay.max-sessions: 16 -> 32") + expectContains(t, details, "codex.live-media-relay.disable-private-remote-ips: false -> true") + expectContains(t, details, "codex.live-media-relay.public-ip: -> 203.0.113.10") + expectContains(t, details, "codex.live-media-relay.udp-port-min: 0 -> 40000") + expectContains(t, details, "codex.live-media-relay.udp-port-max: 0 -> 40063") + expectContains(t, details, "codex.live-media-relay.ice-servers: updated (1 -> 1 entries, credentials redacted)") + joined := strings.Join(details, "\n") + for _, secret := range []string{"old-secret", "new-secret", "old-user", "new-user"} { + if strings.Contains(joined, secret) { + t.Fatalf("config change details leaked %q: %s", secret, joined) + } + } +} + func TestBuildConfigChangeDetails_GeminiVertexHeaders(t *testing.T) { oldCfg := &config.Config{ GeminiKey: []config.GeminiKey{ @@ -180,7 +221,7 @@ func TestBuildConfigChangeDetails_XAIKeys(t *testing.T) { }}} changes := BuildConfigChangeDetails(oldCfg, newCfg) - expectContains(t, changes, "xai[0].base-url: https://old.example.com/v1 -> https://new.example.com/v1") + expectContains(t, changes, "xai[0].base-url: https://old.example.com -> https://new.example.com") expectContains(t, changes, "xai[0].proxy-url: http://old-proxy -> http://new-proxy") expectContains(t, changes, "xai[0].prefix: old -> new") expectContains(t, changes, "xai[0].priority: 1 -> 2") @@ -240,6 +281,37 @@ func TestBuildConfigChangeDetails_SecretsAndCounts(t *testing.T) { expectContains(t, details, "remote-management.secret-key: created") } +func TestBuildConfigChangeDetails_RedactsEndpointURLs(t *testing.T) { + oldCfg := &config.Config{ + GeminiKey: []config.GeminiKey{{BaseURL: "https://old-user:old-pass@old.example/v1?token=old-token"}}, + RemoteManagement: config.RemoteManagement{ + PanelGitHubRepository: "https://old-user:old-pass@old-panel.example/private?token=old-token", + }, + OpenAICompatibility: []config.OpenAICompatibility{{ + BaseURL: "https://old-user:old-pass@old-compat.example/v1?token=old-token", + }}, + } + newCfg := &config.Config{ + GeminiKey: []config.GeminiKey{{BaseURL: "https://new-user:new-pass@new.example/v1?token=new-token"}}, + RemoteManagement: config.RemoteManagement{ + PanelGitHubRepository: "https://new-user:new-pass@new-panel.example/private?token=new-token", + }, + OpenAICompatibility: []config.OpenAICompatibility{{ + BaseURL: "https://new-user:new-pass@new-compat.example/v1?token=new-token", + }}, + } + + details := BuildConfigChangeDetails(oldCfg, newCfg) + expectContains(t, details, "gemini[0].base-url: https://old.example -> https://new.example") + expectContains(t, details, "remote-management.panel-github-repository: https://old-panel.example -> https://new-panel.example") + joined := strings.Join(details, "\n") + for _, sensitive := range []string{"old-user", "new-user", "old-pass", "new-pass", "old-token", "new-token", "/private", "/v1"} { + if strings.Contains(joined, sensitive) { + t.Fatalf("config change details leaked %q: %s", sensitive, joined) + } + } +} + func TestBuildConfigChangeDetails_FlagsAndKeys(t *testing.T) { oldCfg := &config.Config{ Port: 1000, @@ -328,7 +400,7 @@ func TestBuildConfigChangeDetails_FlagsAndKeys(t *testing.T) { expectContains(t, details, "codex-api-key count: 1 -> 2") expectContains(t, details, "remote-management.disable-control-panel: false -> true") expectContains(t, details, "remote-management.disable-auto-update-panel: false -> true") - expectContains(t, details, "remote-management.panel-github-repository: old/repo -> new/repo") + expectContains(t, details, "remote-management.panel-github-repository: old -> new") expectContains(t, details, "remote-management.secret-key: deleted") } @@ -482,7 +554,7 @@ func TestBuildConfigChangeDetails_AllBranches(t *testing.T) { expectContains(t, changes, "remote-management.allow-remote: false -> true") expectContains(t, changes, "remote-management.disable-control-panel: false -> true") expectContains(t, changes, "remote-management.disable-auto-update-panel: false -> true") - expectContains(t, changes, "remote-management.panel-github-repository: old/repo -> new/repo") + expectContains(t, changes, "remote-management.panel-github-repository: old -> new") expectContains(t, changes, "remote-management.secret-key: deleted") expectContains(t, changes, "openai-compatibility:") } diff --git a/internal/watcher/diff/openai_compat.go b/internal/watcher/diff/openai_compat.go index acdf39f92..25889dcef 100644 --- a/internal/watcher/diff/openai_compat.go +++ b/internal/watcher/diff/openai_compat.go @@ -114,7 +114,7 @@ func openAICompatKey(entry config.OpenAICompatibility, index int) (string, strin } base := strings.TrimSpace(entry.BaseURL) if base != "" { - return "base:" + base, base + return "base:" + base, formatURL(base) } for _, model := range entry.Models { alias := strings.TrimSpace(model.Alias)