From f8dffa0522c628c27970148319a50f25f0ffebdd Mon Sep 17 00:00:00 2001 From: Luis Pater Date: Sat, 25 Jul 2026 07:51:43 +0800 Subject: [PATCH] feat(logging): log Codex media forwarding start with detailed fields - Implemented logging for Codex remote media forwarding start events, including detailed connection and credential metadata. - Added `formatLogFieldValue` for quoting specific log fields and ensured newline safety in log output. - Enhanced unit tests to validate log content, escaping, and field inclusion. --- internal/client/codex/live/live.go | 44 +++++-- internal/client/codex/live/live_test.go | 49 +++++++- internal/client/codex/live/media.go | 65 ++++++++-- internal/client/codex/live/media_test.go | 119 ++++++++++++++++++- internal/client/codex/live/tcp_proxy.go | 37 ++++-- internal/client/codex/live/tcp_proxy_test.go | 59 +++++++++ internal/logging/global_logger.go | 26 +++- internal/logging/global_logger_test.go | 39 ++++++ 8 files changed, 401 insertions(+), 37 deletions(-) diff --git a/internal/client/codex/live/live.go b/internal/client/codex/live/live.go index a691b721c..3a699ca27 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" + "path/filepath" "reflect" "strings" "sync" @@ -222,7 +223,8 @@ func (h *Handler) Handle(c *gin.Context) { ctx = attemptCtx defer releaseAttempt() } - logging.SetGinCPATraceID(c, selected.EnsureIndex()) + selectedIndex := selected.EnsureIndex() + logging.SetGinCPATraceID(c, selectedIndex) if selection != nil { defer func() { if selection.Active() && !selection.Retained() { @@ -238,7 +240,11 @@ func (h *Handler) Handle(c *gin.Context) { return } var upstreamOffer string - mediaSession, upstreamOffer, errSDP = mediaRelay.NewSession(ctx, clientOffer, proxyURLForAuth(runtimeConfig, selected)) + mediaSession, upstreamOffer, errSDP = mediaRelay.NewSession(ctx, clientOffer, mediaSessionRoute{ + proxyURL: proxyURLForAuth(runtimeConfig, selected), + credential: mediaCredentialName(selected, selectedIndex), + authIndex: selectedIndex, + }) if errSDP != nil { c.JSON(http.StatusBadGateway, gin.H{"error": errSDP.Error()}) return @@ -334,6 +340,17 @@ func (h *Handler) Handle(c *gin.Context) { helps.AppendAPIResponseChunk(ctx, runtimeConfig, responseBody) responseBodyToWrite := responseBody success := resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices + callID := "" + if success { + callID = callIDFromLocation(resp.Header.Get("Location")) + if callID == "" && mediaSession != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": "Codex live response is missing a valid call ID"}) + return + } + if mediaSession != nil { + mediaSession.SetCallID(callID) + } + } if success && mediaSession != nil { upstreamAnswer, errSDP := callResponseSDP(responseBody, resp.Header.Get("Content-Type")) if errSDP != nil { @@ -351,15 +368,7 @@ func (h *Handler) Handle(c *gin.Context) { var storedSession liveSession sessionStored := false if success && h.sessions != nil { - callID := callIDFromLocation(resp.Header.Get("Location")) - if callID == "" && mediaSession != nil { - c.JSON(http.StatusBadGateway, gin.H{"error": "Codex live response is missing a valid call ID"}) - return - } if callID != "" { - if mediaSession != nil { - mediaSession.SetCallID(callID) - } session := liveSession{authID: selected.ID, model: model, media: mediaSession} if selection != nil { if mediaSession != nil { @@ -404,6 +413,21 @@ func (h *Handler) Handle(c *gin.Context) { } } +func mediaCredentialName(selected *auth.Auth, authIndex string) string { + if selected == nil { + return strings.TrimSpace(authIndex) + } + if label := strings.TrimSpace(selected.Label); label != "" { + return label + } + if fileName := strings.TrimSpace(selected.FileName); fileName != "" { + if baseName := strings.TrimSpace(filepath.Base(fileName)); baseName != "" && baseName != "." { + return baseName + } + } + return strings.TrimSpace(authIndex) +} + func (h *Handler) selectOAuth(ctx context.Context, model string, opts coreexecutor.Options) (*auth.HomeDispatchSelection, *auth.Auth, error) { var selection *auth.HomeDispatchSelection var selected *auth.Auth diff --git a/internal/client/codex/live/live_test.go b/internal/client/codex/live/live_test.go index 40e61f58c..a4f54b2e4 100644 --- a/internal/client/codex/live/live_test.go +++ b/internal/client/codex/live/live_test.go @@ -149,20 +149,21 @@ func (b *trackedResponseBody) Close() error { type fakeMediaRelay struct { clientOffer string - proxyURL string + route mediaSessionRoute upstreamOffer string session *fakeMediaSession err error } -func (r *fakeMediaRelay) NewSession(_ context.Context, clientOffer, proxyURL string) (mediaRelaySession, string, error) { +func (r *fakeMediaRelay) NewSession(_ context.Context, clientOffer string, route mediaSessionRoute) (mediaRelaySession, string, error) { r.clientOffer = clientOffer - r.proxyURL = proxyURL + r.route = route return r.session, r.upstreamOffer, r.err } type fakeMediaSession struct { upstreamAnswer string + callIDAtAccept string downstreamSDP string closeHandler func(string) callID string @@ -173,6 +174,7 @@ type fakeMediaSession struct { func (s *fakeMediaSession) AcceptUpstreamAnswer(_ context.Context, answer string) (string, error) { s.upstreamAnswer = answer + s.callIDAtAccept = s.callID return s.downstreamSDP, s.err } @@ -320,6 +322,36 @@ func TestHandlerRewritesLiveCallAndSchedulesOAuth(t *testing.T) { } } +func TestMediaCredentialNameUsesSafeIdentity(t *testing.T) { + for name, testCase := range map[string]struct { + selected *auth.Auth + index string + want string + }{ + "label": { + selected: &auth.Auth{Label: "Voice credential", FileName: "/auths/codex-user.json", ID: "secret-id"}, + index: "auth-index", + want: "Voice credential", + }, + "file basename": { + selected: &auth.Auth{FileName: "/auths/codex-user.json", ID: "secret-id"}, + index: "auth-index", + want: "codex-user.json", + }, + "opaque index": { + selected: &auth.Auth{ID: "secret-id"}, + index: "auth-index", + want: "auth-index", + }, + } { + t.Run(name, func(t *testing.T) { + if got := mediaCredentialName(testCase.selected, testCase.index); got != testCase.want { + t.Fatalf("mediaCredentialName() = %q, want %q", got, testCase.want) + } + }) + } +} + func TestProxyURLForAuthPrefersCredentialOverride(t *testing.T) { cfg := &config.Config{} cfg.ProxyURL = "http://global.example:8080" @@ -346,6 +378,7 @@ func TestHandlerRelaysWebRTCMediaSDP(t *testing.T) { ID: "codex-oauth", Provider: "codex", Status: auth.StatusActive, + Label: "Voice credential", ProxyURL: "socks5://credential-proxy.example:1080", Metadata: map[string]any{"access_token": "oauth-token"}, }) @@ -374,8 +407,11 @@ 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) + if mediaRelay.route.proxyURL != "socks5://credential-proxy.example:1080" { + t.Fatalf("media proxy URL = %q, want credential override", mediaRelay.route.proxyURL) + } + if mediaRelay.route.credential != "Voice credential" || mediaRelay.route.authIndex == "" { + t.Fatalf("media credential route = %#v", mediaRelay.route) } var upstreamPayload struct { SDP string `json:"sdp"` @@ -392,6 +428,9 @@ func TestHandlerRelaysWebRTCMediaSDP(t *testing.T) { if mediaSession.callID != "call-123" { t.Fatalf("media call ID = %q, want call-123", mediaSession.callID) } + if mediaSession.callIDAtAccept != "call-123" { + t.Fatalf("media call ID at answer acceptance = %q, want call-123", mediaSession.callIDAtAccept) + } if got := recorder.Body.String(); got != mediaSession.downstreamSDP { t.Fatalf("downstream SDP = %q, want %q", got, mediaSession.downstreamSDP) } diff --git a/internal/client/codex/live/media.go b/internal/client/codex/live/media.go index bd57508fd..fac9d1d47 100644 --- a/internal/client/codex/live/media.go +++ b/internal/client/codex/live/media.go @@ -42,7 +42,13 @@ type mediaRelaySession interface { } type mediaRelayFactory interface { - NewSession(context.Context, string, string) (mediaRelaySession, string, error) + NewSession(context.Context, string, mediaSessionRoute) (mediaRelaySession, string, error) +} + +type mediaSessionRoute struct { + proxyURL string + credential string + authIndex string } type pionMediaRelay struct { @@ -76,11 +82,14 @@ type pionMediaSession struct { callID string releaseSlot func() - proxyDialer proxy.ContextDialer - proxyScheme string - localOffer string - tunnelsMu sync.Mutex - tunnels []*tcpCandidateTunnel + proxyDialer proxy.ContextDialer + proxyScheme string + credential string + authIndex string + forwardingLogOnce sync.Once + localOffer string + tunnelsMu sync.Mutex + tunnels []*tcpCandidateTunnel } type dataChannelMessage struct { @@ -248,14 +257,14 @@ func isPublicRemoteIP(ip net.IP) bool { !ip.IsLinkLocalUnicast() && !ip.IsLinkLocalMulticast() && !ip.IsMulticast() } -func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer, proxyURL string) (mediaRelaySession, string, error) { +func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer string, route mediaSessionRoute) (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) + builtProxyDialer, proxyMode, errProxy := proxyutil.BuildDialer(route.proxyURL) if errProxy != nil { return nil, "", fmt.Errorf("configure Codex live remote TCP proxy: %w", errProxy) } @@ -299,7 +308,9 @@ func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer, proxyURL s mediaSessionID: uuid.NewString(), releaseSlot: releaseSlot, proxyDialer: proxyDialer, - proxyScheme: proxyScheme(proxyURL), + proxyScheme: proxyScheme(route.proxyURL), + credential: strings.TrimSpace(route.credential), + authIndex: strings.TrimSpace(route.authIndex), } session.bridge = newDataChannelBridge(session.done, func(err error) { session.fail("data_channel_failed", err) @@ -402,6 +413,9 @@ func (s *pionMediaSession) AcceptUpstreamAnswer(ctx context.Context, upstreamAns if errProxy != nil { return "", errProxy } + for _, tunnel := range tunnels { + tunnel.setForwardingStartedHandler(s.logForwardingStarted) + } if !s.installCandidateTunnels(tunnels) { errClosed := errors.New("Codex live media session closed while configuring TCP proxy") if errClose := closeCandidateTunnels(tunnels); errClose != nil { @@ -494,6 +508,36 @@ func (s *pionMediaSession) logFields(peer string) log.Fields { return fields } +func (s *pionMediaSession) forwardingLogFields() log.Fields { + fields := s.logFields("remote") + if s.authIndex != "" { + fields["auth_index"] = s.authIndex + } + if s.credential != "" { + fields["credential"] = s.credential + } + if s.proxyDialer != nil { + fields["connection"] = "via " + s.proxyScheme + " proxy" + fields["remote_transport"] = "tcp" + } else { + fields["connection"] = "direct" + fields["remote_transport"] = "ice" + } + if s.upstream != nil { + fields["state"] = s.upstream.ConnectionState().String() + } + return fields +} + +func (s *pionMediaSession) logForwardingStarted() { + if s == nil { + return + } + s.forwardingLogOnce.Do(func() { + log.WithFields(s.forwardingLogFields()).Info("codex live remote media forwarding started") + }) +} + func (s *pionMediaSession) SetCloseHandler(handler func(string)) { if s == nil { return @@ -576,6 +620,9 @@ func (s *pionMediaSession) installStateHandlers() { log.WithFields(fields).Info("codex live WebRTC peer connecting") case webrtc.PeerConnectionStateConnected: log.WithFields(fields).Info("codex live WebRTC peer connected") + if peer == "remote" { + s.logForwardingStarted() + } case webrtc.PeerConnectionStateDisconnected: log.WithFields(fields).Warn("codex live WebRTC peer disconnected") case webrtc.PeerConnectionStateFailed: diff --git a/internal/client/codex/live/media_test.go b/internal/client/codex/live/media_test.go index cc9f72850..0a50335f5 100644 --- a/internal/client/codex/live/media_test.go +++ b/internal/client/codex/live/media_test.go @@ -2,6 +2,7 @@ package live import ( "context" + "fmt" "net" "strings" "testing" @@ -46,7 +47,7 @@ func TestPionMediaRelaySelectsRemoteProxyMode(t *testing.T) { "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) + session, upstreamOffer, errSession := relay.NewSession(context.Background(), clientOffer, mediaSessionRoute{proxyURL: testCase.proxyURL}) if errSession != nil { t.Fatalf("create media session: %v", errSession) } @@ -66,11 +67,79 @@ func TestPionMediaRelaySelectsRemoteProxyMode(t *testing.T) { }) } - if _, _, errSession := relay.NewSession(context.Background(), clientOffer, "invalid-proxy"); errSession == nil { + if _, _, errSession := relay.NewSession(context.Background(), clientOffer, mediaSessionRoute{proxyURL: "invalid-proxy"}); errSession == nil { t.Fatal("expected invalid proxy URL to fail media session creation") } } +func TestMediaForwardingStartedLogRedactsProxyCredentials(t *testing.T) { + logger := log.StandardLogger() + previousHooks := logger.ReplaceHooks(make(log.LevelHooks)) + hook := logtest.NewLocal(logger) + defer logger.ReplaceHooks(previousHooks) + + for name, testCase := range map[string]struct { + proxyURL string + connection string + credential string + }{ + "direct": { + connection: "direct", + credential: "Voice credential", + }, + "HTTP": { + proxyURL: "http://user:secret@proxy.example:8080", + connection: "via http proxy", + credential: "Voice credential", + }, + "SOCKS5 without label": { + proxyURL: "socks5://user:secret@proxy.example:1080", + connection: "via socks5 proxy", + credential: "auth-index", + }, + } { + t.Run(name, func(t *testing.T) { + session := &pionMediaSession{ + mediaSessionID: "media-session-" + name, + proxyScheme: proxyScheme(testCase.proxyURL), + credential: testCase.credential, + authIndex: "auth-index", + } + if testCase.proxyURL != "" { + session.proxyDialer = &recordingProxyDialer{dials: make(chan recordedProxyDial, 1)} + } + earlyFields := session.logFields("session") + for _, field := range []string{"auth_id", "auth_label", "auth_index", "credential", "connection"} { + if _, exists := earlyFields[field]; exists { + t.Fatalf("session log exposed forwarding-only field %q before forwarding started: %#v", field, earlyFields) + } + } + session.logForwardingStarted() + session.logForwardingStarted() + + matching := 0 + for _, entry := range hook.AllEntries() { + if entry.Message != "codex live remote media forwarding started" || entry.Data["media_session_id"] != session.mediaSessionID { + continue + } + matching++ + if entry.Data["connection"] != testCase.connection || entry.Data["credential"] != testCase.credential { + t.Fatalf("forwarding fields = %#v", entry.Data) + } + serialized := fmt.Sprint(entry.Data) + for _, secret := range []string{"user", "secret", "proxy.example"} { + if strings.Contains(serialized, secret) { + t.Fatalf("forwarding log leaked %q: %s", secret, serialized) + } + } + } + if matching != 1 { + t.Fatalf("forwarding log count = %d, want 1", matching) + } + }) + } +} + func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) { logger := log.StandardLogger() previousHooks := logger.ReplaceHooks(make(log.LevelHooks)) @@ -126,7 +195,10 @@ 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, mediaSessionRoute{ + credential: "Voice credential", + authIndex: "auth-index", + }) if errSession != nil { t.Fatalf("create media relay session: %v", errSession) } @@ -140,7 +212,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, mediaSessionRoute{}); errCapacity == nil { t.Fatal("reloaded media relay bypassed the shared session capacity") } @@ -222,7 +294,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, mediaSessionRoute{}) if errReplacement != nil { t.Fatalf("shared capacity was not released: %v", errReplacement) } @@ -233,6 +305,8 @@ func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) { assertPeerLog(t, hook, "codex live WebRTC peer connected", peer, "call-log-test") assertPeerLog(t, hook, "codex live WebRTC peer closed", peer, "call-log-test") } + assertForwardingLog(t, hook, "direct", "Voice credential", "auth-index", "connected") + assertForwardingAfterRemoteConnected(t, hook) assertSessionLog(t, hook, "codex live WebRTC media session closed", "closed", "call-log-test") } @@ -397,6 +471,41 @@ func sendTestRTP(t *testing.T, track *webrtc.TrackLocalStaticRTP, payload []byte t.Fatal("RTP packet was not relayed") } +func assertForwardingAfterRemoteConnected(t *testing.T, hook *logtest.Hook) { + t.Helper() + connectedIndex := -1 + forwardingIndex := -1 + for index, entry := range hook.AllEntries() { + if entry.Message == "codex live WebRTC peer connected" && entry.Data["peer"] == "remote" && connectedIndex == -1 { + connectedIndex = index + } + if entry.Message == "codex live remote media forwarding started" && forwardingIndex == -1 { + forwardingIndex = index + } + } + if connectedIndex == -1 || forwardingIndex <= connectedIndex { + t.Fatalf("remote connected index=%d, forwarding index=%d", connectedIndex, forwardingIndex) + } +} + +func assertForwardingLog(t *testing.T, hook *logtest.Hook, connection, credential, authIndex, state string) { + t.Helper() + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + for _, entry := range hook.AllEntries() { + if entry.Message == "codex live remote media forwarding started" && + entry.Data["connection"] == connection && + entry.Data["credential"] == credential && + entry.Data["auth_index"] == authIndex && + entry.Data["state"] == state { + return + } + } + time.Sleep(time.Millisecond) + } + t.Fatalf("missing forwarding log for connection %q and credential %q", connection, credential) +} + func assertSessionLog(t *testing.T, hook *logtest.Hook, message, reason, callID string) { t.Helper() deadline := time.Now().Add(time.Second) diff --git a/internal/client/codex/live/tcp_proxy.go b/internal/client/codex/live/tcp_proxy.go index c9d2a0d25..ddf379f23 100644 --- a/internal/client/codex/live/tcp_proxy.go +++ b/internal/client/codex/live/tcp_proxy.go @@ -71,13 +71,14 @@ type tcpCandidateTunnel struct { 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 + mu sync.Mutex + closed bool + claimed bool + connections map[net.Conn]struct{} + validationSlots chan struct{} + onForwardingStarted func() + ctx context.Context + cancel context.CancelFunc } type tcpCandidatePlan struct { @@ -379,6 +380,7 @@ func (t *tcpCandidateTunnel) handleConnection(client net.Conn) { log.WithError(errWrite).Warn("codex live TCP proxy: forward authenticated ICE frame failed") return } + t.notifyForwardingStarted() copyDone := make(chan struct{}, 2) copyConnection := func(destination, source net.Conn) { @@ -393,6 +395,27 @@ func (t *tcpCandidateTunnel) handleConnection(client net.Conn) { <-copyDone } +func (t *tcpCandidateTunnel) setForwardingStartedHandler(handler func()) { + if t == nil { + return + } + t.mu.Lock() + t.onForwardingStarted = handler + t.mu.Unlock() +} + +func (t *tcpCandidateTunnel) notifyForwardingStarted() { + if t == nil { + return + } + t.mu.Lock() + handler := t.onForwardingStarted + t.mu.Unlock() + if handler != nil { + handler() + } +} + func (t *tcpCandidateTunnel) trackConnection(connection net.Conn) bool { t.mu.Lock() defer t.mu.Unlock() diff --git a/internal/client/codex/live/tcp_proxy_test.go b/internal/client/codex/live/tcp_proxy_test.go index bbc50d819..dd6723f91 100644 --- a/internal/client/codex/live/tcp_proxy_test.go +++ b/internal/client/codex/live/tcp_proxy_test.go @@ -37,6 +37,14 @@ type blockingContextDialer struct { canceled chan struct{} } +type closedUpstreamDialer struct{} + +func (*closedUpstreamDialer) DialContext(context.Context, string, string) (net.Conn, error) { + client, server := net.Pipe() + _ = server.Close() + return client, nil +} + func (d *blockingContextDialer) DialContext(ctx context.Context, _, _ string) (net.Conn, error) { close(d.started) <-ctx.Done() @@ -228,6 +236,10 @@ func TestTCPCandidateTunnelAuthenticatesBeforeFixedTargetDial(t *testing.T) { t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel) } defer func() { _ = tunnel.Close() }() + forwardingStarted := make(chan struct{}, 1) + tunnel.setForwardingStartedHandler(func() { + forwardingStarted <- struct{}{} + }) client, errDial := net.Dial("tcp", tunnel.listener.Addr().String()) if errDial != nil { @@ -256,6 +268,11 @@ func TestTCPCandidateTunnelAuthenticatesBeforeFixedTargetDial(t *testing.T) { if !bytes.Equal(forwarded, frame) { t.Fatal("forwarded STUN frame changed") } + select { + case <-forwardingStarted: + case <-time.After(time.Second): + t.Fatal("forwarding start handler was not called") + } if errWrite := writeAll(dial.connection, []byte("reply")); errWrite != nil { t.Fatalf("write tunnel reply: %v", errWrite) } @@ -295,6 +312,8 @@ func TestTCPCandidateTunnelCloseCancelsProxyDial(t *testing.T) { case <-time.After(time.Second): t.Fatal("proxy dial did not start") } + forwardingStarted := make(chan struct{}, 1) + tunnel.setForwardingStartedHandler(func() { forwardingStarted <- struct{}{} }) if errClose := tunnel.Close(); errClose != nil { t.Fatalf("close tunnel: %v", errClose) } @@ -303,6 +322,7 @@ func TestTCPCandidateTunnelCloseCancelsProxyDial(t *testing.T) { case <-time.After(time.Second): t.Fatal("tunnel close did not cancel proxy dial") } + assertNoForwardingStart(t, forwardingStarted) } func TestTCPCandidateTunnelProxyFailureDoesNotFallBack(t *testing.T) { @@ -320,6 +340,8 @@ func TestTCPCandidateTunnelProxyFailureDoesNotFallBack(t *testing.T) { t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel) } defer func() { _ = tunnel.Close() }() + forwardingStarted := make(chan struct{}, 1) + tunnel.setForwardingStartedHandler(func() { forwardingStarted <- struct{}{} }) client, errDial := net.Dial("tcp", tunnel.listener.Addr().String()) if errDial != nil { t.Fatalf("dial candidate listener: %v", errDial) @@ -339,6 +361,31 @@ func TestTCPCandidateTunnelProxyFailureDoesNotFallBack(t *testing.T) { if _, errSecondDial := net.Dial("tcp", tunnel.listener.Addr().String()); errSecondDial == nil { t.Fatal("candidate listener remained available after proxy failure") } + assertNoForwardingStart(t, forwardingStarted) +} + +func TestTCPCandidateTunnelWriteFailureDoesNotLogForwardingStart(t *testing.T) { + tunnel, errTunnel := newTCPCandidateTunnel( + netip.MustParseAddrPort("20.42.0.20:443"), + &closedUpstreamDialer{}, + "remote:local", + "remote-password", + ) + if errTunnel != nil { + t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel) + } + defer func() { _ = tunnel.Close() }() + forwardingStarted := make(chan struct{}, 1) + tunnel.setForwardingStartedHandler(func() { forwardingStarted <- struct{}{} }) + 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) + } + _ = client.Close() + assertNoForwardingStart(t, forwardingStarted) } func TestTCPCandidateTunnelRejectsUnauthenticatedConnectionWithoutDial(t *testing.T) { @@ -353,6 +400,8 @@ func TestTCPCandidateTunnelRejectsUnauthenticatedConnectionWithoutDial(t *testin t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel) } defer func() { _ = tunnel.Close() }() + forwardingStarted := make(chan struct{}, 1) + tunnel.setForwardingStartedHandler(func() { forwardingStarted <- struct{}{} }) client, errDial := net.Dial("tcp", tunnel.listener.Addr().String()) if errDial != nil { @@ -368,6 +417,16 @@ func TestTCPCandidateTunnelRejectsUnauthenticatedConnectionWithoutDial(t *testin t.Fatalf("unauthenticated connection triggered proxy dial to %q", dial.address) case <-time.After(100 * time.Millisecond): } + assertNoForwardingStart(t, forwardingStarted) +} + +func assertNoForwardingStart(t *testing.T, started <-chan struct{}) { + t.Helper() + select { + case <-started: + t.Fatal("forwarding start handler was called for an unestablished tunnel") + case <-time.After(100 * time.Millisecond): + } } func TestPionActiveTCPCandidatePassesTunnelAuthentication(t *testing.T) { diff --git a/internal/logging/global_logger.go b/internal/logging/global_logger.go index 9d6fffcb3..b27585c4b 100644 --- a/internal/logging/global_logger.go +++ b/internal/logging/global_logger.go @@ -6,6 +6,7 @@ import ( "io" "os" "path/filepath" + "strconv" "strings" "sync" @@ -35,10 +36,33 @@ var logFieldOrder = []string{ "plugin_id", "plugin_name", "source_id", "version", "active_version", "retired_version", "overwritten", "mode", "budget", "level", "original_mode", "original_value", "min", "max", "clamped_to", "error", + "credential", "connection", "proxy_scheme", "remote_transport", + "media_session_id", "call_id", "peer", "state", "reason", +} + +var quotedLogFields = map[string]struct{}{ + "credential": {}, + "connection": {}, + "proxy_scheme": {}, + "remote_transport": {}, + "media_session_id": {}, + "call_id": {}, + "peer": {}, + "state": {}, + "reason": {}, } var pluginPathFieldOrder = []string{"path", "active_path", "retired_path"} +func formatLogFieldValue(key string, value any) string { + if _, quoted := quotedLogFields[key]; quoted { + if stringValue, ok := value.(string); ok { + return strconv.Quote(stringValue) + } + } + return fmt.Sprint(value) +} + // Format renders a single log entry with custom formatting. func (m *LogFormatter) Format(entry *log.Entry) ([]byte, error) { var buffer *bytes.Buffer @@ -68,7 +92,7 @@ func (m *LogFormatter) Format(entry *log.Entry) ([]byte, error) { var fields []string for _, k := range logFieldOrder { if v, ok := entry.Data[k]; ok { - fields = append(fields, fmt.Sprintf("%s=%v", k, v)) + fields = append(fields, fmt.Sprintf("%s=%s", k, formatLogFieldValue(k, v))) } } if pluginID, ok := entry.Data["plugin_id"]; ok && strings.TrimSpace(fmt.Sprint(pluginID)) != "" { diff --git a/internal/logging/global_logger_test.go b/internal/logging/global_logger_test.go index 417a4e65f..884a95673 100644 --- a/internal/logging/global_logger_test.go +++ b/internal/logging/global_logger_test.go @@ -26,6 +26,45 @@ func TestLogFormatterPrintsVersionField(t *testing.T) { } } +func TestLogFormatterPrintsMediaForwardingFields(t *testing.T) { + entry := log.NewEntry(log.New()) + entry.Time = time.Date(2026, 7, 25, 7, 36, 4, 0, time.Local) + entry.Level = log.InfoLevel + entry.Message = "codex live remote media forwarding started" + entry.Data["credential"] = "Voice credential\nsecondary" + entry.Data["connection"] = "via socks5 proxy" + entry.Data["proxy_scheme"] = "socks5" + entry.Data["remote_transport"] = "tcp" + entry.Data["media_session_id"] = "media-session-id" + entry.Data["call_id"] = "call-id" + entry.Data["peer"] = "remote" + entry.Data["state"] = "connected" + + formatted, errFormat := (&LogFormatter{}).Format(entry) + if errFormat != nil { + t.Fatalf("Format() error = %v", errFormat) + } + + line := string(formatted) + for _, want := range []string{ + `credential="Voice credential\nsecondary"`, + `connection="via socks5 proxy"`, + `proxy_scheme="socks5"`, + `remote_transport="tcp"`, + `media_session_id="media-session-id"`, + `call_id="call-id"`, + `peer="remote"`, + `state="connected"`, + } { + if !strings.Contains(line, want) { + t.Fatalf("formatted line %q missing %s", line, want) + } + } + if strings.Count(line, "\n") != 1 { + t.Fatalf("formatted line contains an unescaped newline: %q", line) + } +} + func TestLogFormatterPrintsPluginFields(t *testing.T) { entry := log.NewEntry(log.New()) entry.Time = time.Date(2026, 6, 25, 20, 10, 0, 0, time.Local)