From 31e90641a83db0087bcc5be327feba57080f9861 Mon Sep 17 00:00:00 2001 From: Pushpinder Singh Date: Sun, 18 Oct 2026 23:41:03 -0400 Subject: [PATCH] feat: include user identity in activity and stream logs --- .../app/activity_logging_integration_test.go | 120 +++++++++++------- internal/app/auth.go | 27 ++-- .../app/auth_playback_integration_test.go | 2 +- internal/app/logging.go | 17 ++- internal/app/server.go | 3 +- internal/stream/stream.go | 29 +++-- internal/stream/stream_test.go | 13 +- 7 files changed, 138 insertions(+), 73 deletions(-) diff --git a/internal/app/activity_logging_integration_test.go b/internal/app/activity_logging_integration_test.go index 2dc3a49..3e81b50 100644 --- a/internal/app/activity_logging_integration_test.go +++ b/internal/app/activity_logging_integration_test.go @@ -36,52 +36,80 @@ func captureActivityLogs(t *testing.T) func() string { } func TestSignInAndSettingsActivityLogs(t *testing.T) { - logs := captureActivityLogs(t) - p := newAuthTestProvider(t) - p.role = "admin" - a := newAuthTestApp(t, p, nil) - var identity struct{ User AuthUser } - if err := json.Unmarshal([]byte(a.login(t)), &identity); err != nil || identity.User.ID == "" { - t.Fatal("missing signed-in identity", err) - } - for _, expected := range []string{"Sign-in started", "User signed in", "user_id=" + identity.User.ID, "role=admin"} { - if !strings.Contains(logs(), expected) { - t.Fatalf("missing %q: %s", expected, logs()) - } - } - before := logs() - for range 5 { - a.get(t, "/api/auth").Body.Close() - a.get(t, "/api/status").Body.Close() - } - if logs() != before { - t.Fatal("routine auth/status polling produced activity logs", logs()) - } - resp := authPlaybackRequest(t, a, "PATCH", "/api/settings", `{"discovery":{"tmdbReadAccessToken":"private-metadata-token"}}`) - resp.Body.Close() - if resp.StatusCode != http.StatusOK || !strings.Contains(logs(), "Settings updated") || !strings.Contains(logs(), "sections=[discovery]") { - t.Fatal("settings change was not logged", resp.StatusCode, logs()) - } - before = logs() - resp = authPlaybackRequest(t, a, "PATCH", "/api/settings", `{"discovery":{"tmdbReadAccessToken":"private-metadata-token"}}`) - resp.Body.Close() - if logs() != before { - t.Fatal("unchanged settings produced activity logs", logs()) - } - resp = authPlaybackRequest(t, a, "POST", "/auth/logout", "") - resp.Body.Close() - if resp.StatusCode != http.StatusNoContent || !strings.Contains(logs(), "User signed out") { - t.Fatal("sign-out was not logged", resp.StatusCode, logs()) - } - resp = a.get(t, "/auth/callback?code=private-code&state=private-state&error=private-provider-error") - resp.Body.Close() - if !strings.Contains(logs(), "Sign-in failed") || !strings.Contains(logs(), "stage=callback_validation") { - t.Fatal("failed sign-in stage was not logged", logs()) - } - for _, secret := range []string{"private-", "opaque-access-token", "opaque-refresh-token", "test-secret", p.name, p.server.URL} { - if strings.Contains(logs(), secret) { - t.Fatalf("activity logs exposed %q: %s", secret, logs()) - } + for _, tc := range []struct{ label, name, username, wantName string }{ + {"display name", "Alex", "alex-user", "Alex"}, + {"username", " ", "alex-user", "alex-user"}, + {"user ID", "", "", ""}, + } { + t.Run(tc.label, func(t *testing.T) { + logs := captureActivityLogs(t) + p := newAuthTestProvider(t) + p.role = "admin" + p.name = tc.name + p.claims = func(claims map[string]any) { claims["preferred_username"] = tc.username } + a := newAuthTestApp(t, p, nil) + var identity struct{ User AuthUser } + if err := json.Unmarshal([]byte(a.login(t)), &identity); err != nil || identity.User.ID == "" || identity.User.Name != tc.wantName { + t.Fatal("missing signed-in identity", err) + } + for _, expected := range []string{"Sign-in started", "User signed in", "user_id=" + identity.User.ID, "role=admin"} { + if !strings.Contains(logs(), expected) { + t.Fatalf("missing %q: %s", expected, logs()) + } + } + before := logs() + for range 5 { + a.get(t, "/api/auth").Body.Close() + a.get(t, "/api/status").Body.Close() + } + if logs() != before { + t.Fatal("routine auth/status polling produced activity logs", logs()) + } + resp := authPlaybackRequest(t, a, "PATCH", "/api/settings", `{"discovery":{"tmdbReadAccessToken":"private-metadata-token"}}`) + resp.Body.Close() + if resp.StatusCode != http.StatusOK || !strings.Contains(logs(), "Settings updated") || !strings.Contains(logs(), "sections=[discovery]") { + t.Fatal("settings change was not logged", resp.StatusCode, logs()) + } + before = logs() + resp = authPlaybackRequest(t, a, "PATCH", "/api/settings", `{"discovery":{"tmdbReadAccessToken":"private-metadata-token"}}`) + resp.Body.Close() + if logs() != before { + t.Fatal("unchanged settings produced activity logs", logs()) + } + resp = authPlaybackRequest(t, a, "POST", "/auth/logout", "") + resp.Body.Close() + if resp.StatusCode != http.StatusNoContent || !strings.Contains(logs(), "User signed out") { + t.Fatal("sign-out was not logged", resp.StatusCode, logs()) + } + resp = a.get(t, "/auth/callback?code=private-code&state=private-state&error=private-provider-error") + resp.Body.Close() + if !strings.Contains(logs(), "Sign-in failed") || !strings.Contains(logs(), "stage=callback_validation") { + t.Fatal("failed sign-in stage was not logged", logs()) + } + wantUser := tc.wantName + if wantUser == "" { + wantUser = identity.User.ID + } + for _, message := range []string{"User signed in", "Settings updated", "User signed out"} { + found := false + for _, line := range strings.Split(logs(), "\n") { + if strings.Contains(line, `msg="`+message+`"`) { + found = true + if !strings.Contains(line, "user="+wantUser) || !strings.Contains(line, "user_id="+identity.User.ID) { + t.Fatalf("missing user context for %q: %s", message, line) + } + } + } + if !found { + t.Fatalf("missing activity %q", message) + } + } + for _, secret := range []string{"private-", "opaque-access-token", "opaque-refresh-token", "test-secret", p.server.URL} { + if strings.Contains(logs(), secret) { + t.Fatalf("activity logs exposed %q: %s", secret, logs()) + } + } + }) } } diff --git a/internal/app/auth.go b/internal/app/auth.go index 29b0bdd..777dc31 100644 --- a/internal/app/auth.go +++ b/internal/app/auth.go @@ -88,10 +88,10 @@ type authClaims struct { func (c authClaims) valid() bool { return c.Role == "viewer" || c.Role == "admin" } func (c authClaims) name() string { - if c.Name != "" { - return c.Name + if name := strings.TrimSpace(c.Name); name != "" { + return name } - return c.Username + return strings.TrimSpace(c.Username) } func newAuth(cfg AuthConfig, db *sql.DB) (*authManager, error) { @@ -350,18 +350,24 @@ func (a *authManager) callback(w http.ResponseWriter, r *http.Request) { a.sessions.Clear(ctx) a.sessions.Put(ctx, "user_id", userID) a.sessions.Put(ctx, "user_version", version) - slog.Info("User signed in", "user_id", userID, "role", claims.Role) + slog.Info("User signed in", append(userLogAttrs(&AuthUser{ID: userID, Name: claims.name()}), "role", claims.Role)...) http.Redirect(w, r, "/", http.StatusSeeOther) } func (a *authManager) logout(w http.ResponseWriter, r *http.Request) { userID := a.sessions.GetString(r.Context(), "user_id") + var user *AuthUser + if userID != "" { + user = &AuthUser{ID: userID} + // Read the saved display name without refreshing provider credentials. + _ = a.db.QueryRowContext(r.Context(), `SELECT name FROM auth_users WHERE id=?`, userID).Scan(&user.Name) + } if err := a.sessions.Destroy(r.Context()); err != nil { - slog.Error("Sign-out failed", "stage", "session_destroy", "user_id", userID) + slog.Error("Sign-out failed", append(userLogAttrs(user), "stage", "session_destroy")...) fail(w, 500, "auth", "Unable to sign out.") return } if userID != "" { - slog.Info("User signed out", "user_id", userID) + slog.Info("User signed out", userLogAttrs(user)...) } w.WriteHeader(http.StatusNoContent) } @@ -405,7 +411,7 @@ func (a *authManager) user(r *http.Request) *AuthUser { ctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), 3*time.Second) defer cancel() _, _ = a.db.ExecContext(ctx, `UPDATE auth_users SET role='',checked_at=0,session_version=session_version+1 WHERE id=?`, id) - slog.Warn("User access revoked", "user_id", id, "stage", "session_revalidation", "reason", reason) + slog.Warn("User access revoked", append(userLogAttrs(&u), "stage", "session_revalidation", "reason", reason)...) return nil } var token oauth2.Token @@ -445,9 +451,10 @@ func (a *authManager) user(r *http.Request) *AuthUser { if err != nil { return revoke("user_save_failed") } - if u.Role != claims.Role { - slog.Info("User role changed", "user_id", id, "previous_role", u.Role, "role", claims.Role) - } + previousRole := u.Role u.Name, u.Role = claims.name(), claims.Role + if previousRole != u.Role { + slog.Info("User role changed", append(userLogAttrs(&u), "previous_role", previousRole, "role", u.Role)...) + } return &u } diff --git a/internal/app/auth_playback_integration_test.go b/internal/app/auth_playback_integration_test.go index 887f924..7f02170 100644 --- a/internal/app/auth_playback_integration_test.go +++ b/internal/app/auth_playback_integration_test.go @@ -100,7 +100,7 @@ func TestOIDCPlaybackBelongsToItsUser(t *testing.T) { t.Fatal("owner could not stop playback", resp.StatusCode) } output := logs() - if strings.Count(output, "Stream starting") != 1 || !strings.Contains(output, "session_id=test-1") || !strings.Contains(output, "user_id=") { + if strings.Count(output, "Stream starting") != 1 || !strings.Contains(output, "session_id=test-1") || !strings.Contains(output, "user_id=") || !strings.Contains(output, "user=Alex") { t.Fatal("stream start is missing its user/session context", output) } if strings.Contains(output, "secret-command") || strings.Contains(output, "opaque-access-token") || strings.Contains(output, "opaque-refresh-token") { diff --git a/internal/app/logging.go b/internal/app/logging.go index 6170dd9..2f4c2d9 100644 --- a/internal/app/logging.go +++ b/internal/app/logging.go @@ -3,17 +3,26 @@ package app import ( "errors" "net/http" + "strings" "github.com/pushpinderbal/restream/internal/stalker" ) // Log classifications instead of upstream error text: URLs, tokens and response -// bodies can contain credentials. Internal user IDs link activity without names. +// bodies can contain credentials. Keep user IDs alongside readable identities. func activityAttrs(r *http.Request) []any { - if user := authenticatedUser(r); user != nil { - return []any{"user_id", user.ID} + return userLogAttrs(authenticatedUser(r)) +} + +func userLogAttrs(user *AuthUser) []any { + if user == nil { + return []any{"user", "anonymous"} + } + name := strings.TrimSpace(user.Name) + if name == "" { + name = user.ID } - return nil + return []any{"user_id", user.ID, "user", name} } func refreshFailureAttrs(err error) []any { diff --git a/internal/app/server.go b/internal/app/server.go index 59ff10e..840f1ef 100644 --- a/internal/app/server.go +++ b/internal/app/server.go @@ -757,7 +757,8 @@ func (s *Server) start(w http.ResponseWriter, r *http.Request) { fail(w, 400, "invalid", "Invalid start position.") return } - session, err := s.streams.Start(r.Context(), item, body.Start) + ctx := stream.WithLogger(r.Context(), slog.With(activityAttrs(r)...)) + session, err := s.streams.Start(ctx, item, body.Start) if err != nil { if r.Context().Err() == nil { reason := "stream_unavailable" diff --git a/internal/stream/stream.go b/internal/stream/stream.go index 173d6a3..9f3041a 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -69,6 +69,15 @@ type entry struct { gen *generation heartbeat time.Time probe *probeResult + logger *slog.Logger +} + +type loggerContextKey struct{} + +// WithLogger attaches request identity to a stream's lifecycle logs. Start +// captures the logger so background work and later seeks retain the owner. +func WithLogger(ctx context.Context, logger *slog.Logger) context.Context { + return context.WithValue(ctx, loggerContextKey{}, logger) } type Manager struct { @@ -184,7 +193,11 @@ func (m *Manager) Start(ctx context.Context, item model.Item, start float64) (Se } m.next++ id := strconv.FormatUint(uint64(time.Now().UnixNano()), 36) + "-" + strconv.FormatUint(m.next, 36) - e := &entry{item: item, transcodeMode: m.cfg.TranscodeMode, heartbeat: time.Now()} + logger, _ := ctx.Value(loggerContextKey{}).(*slog.Logger) + if logger == nil { + logger = slog.Default() + } + e := &entry{item: item, transcodeMode: m.cfg.TranscodeMode, heartbeat: time.Now(), logger: logger.With("session_id", id)} e.session = Session{ID: id, Duration: validDuration(item.Duration)} m.sessions[id] = e m.launchLocked(e, start, nil) @@ -261,7 +274,7 @@ func (m *Manager) Seek(ctx context.Context, id string, position float64) (Sessio return Session{}, ErrInvalid } old := e.gen - slog.Debug("Stream seeking", "session_id", e.session.ID, "offset", position) + e.logger.Debug("Stream seeking", "offset", position) old.cancel() go m.discard(old) m.launchLocked(e, position, old.done) @@ -275,7 +288,7 @@ func (m *Manager) Stop(id string) error { delete(m.sessions, id) e.gen.cancel() m.stopping++ - slog.Info("Stream stopped", "session_id", id, "reason", "requested", "state", e.session.State) + e.logger.Info("Stream stopped", "reason", "requested", "state", e.session.State) } m.mu.Unlock() if e == nil { @@ -307,7 +320,7 @@ func (m *Manager) Close() error { for id, e := range m.sessions { e.gen.cancel() delete(m.sessions, id) - slog.Info("Stream stopped", "session_id", id, "reason", "server_shutdown", "state", e.session.State) + e.logger.Info("Stream stopped", "reason", "server_shutdown", "state", e.session.State) } } m.mu.Unlock() @@ -339,7 +352,7 @@ func (m *Manager) sweep() { delete(m.sessions, id) e.gen.cancel() m.stopping++ - slog.Info("Stream stopped", "session_id", id, "reason", "idle_timeout", "state", e.session.State) + e.logger.Info("Stream stopped", "reason", "idle_timeout", "state", e.session.State) go m.retire(e.gen) } } @@ -532,7 +545,7 @@ func (m *Manager) run(ctx context.Context, e *entry, g *generation, pos float64, if ready && err == nil && !live { m.update(e, g, func(s *Session) { s.State = "ended" - slog.Info("Stream processing completed", "session_id", s.ID, "elapsed", time.Since(started).Round(time.Millisecond)) + e.logger.Info("Stream processing completed", "elapsed", time.Since(started).Round(time.Millisecond)) }) } else if live && ctx.Err() == nil { reason := exitClass(err, stderr.String()) @@ -567,7 +580,7 @@ func (m *Manager) run(ctx context.Context, e *entry, g *generation, pos float64, func (m *Manager) logReady(e *entry, g *generation, mode string, started time.Time) { m.update(e, g, func(s *Session) { s.State = "ready" - slog.Info("Stream ready", "session_id", s.ID, "processing", mode, "elapsed", time.Since(started).Round(time.Millisecond)) + e.logger.Info("Stream ready", "processing", mode, "elapsed", time.Since(started).Round(time.Millisecond)) }) } @@ -609,7 +622,7 @@ func (m *Manager) logFailure(e *entry, stage, reason string, started time.Time) } else if stage == "ffmpeg_exit" { message = "Stream processing failed" } - slog.Warn(message, "session_id", e.session.ID, "item_id", safeLogID(e.item.ID), "stage", stage, "reason", reason, "elapsed", time.Since(started).Round(time.Millisecond)) + e.logger.Warn(message, "item_id", safeLogID(e.item.ID), "stage", stage, "reason", reason, "elapsed", time.Since(started).Round(time.Millisecond)) } func safeLogID(id string) string { diff --git a/internal/stream/stream_test.go b/internal/stream/stream_test.go index 1081a58..03722ab 100644 --- a/internal/stream/stream_test.go +++ b/internal/stream/stream_test.go @@ -78,7 +78,8 @@ func TestFFmpegSessionLifecycle(t *testing.T) { } defer m.Close() item := model.Item{ID: "v", Kind: "movie", Duration: 60} - s, err := m.Start(context.Background(), item, 0) + owner := WithLogger(context.Background(), slog.With("user_id", "stream-owner", "user", "Alex")) + s, err := m.Start(owner, item, 0) if err != nil { t.Fatal(err) } @@ -163,6 +164,11 @@ func TestFFmpegSessionLifecycle(t *testing.T) { t.Fatalf("missing %q in stream lifecycle logs: %s", expected, output) } } + for _, line := range strings.Split(string(output), "\n") { + if strings.Contains(line, "session_id="+s.ID) && (!strings.Contains(line, "user=Alex") || !strings.Contains(line, "user_id=stream-owner")) { + t.Fatalf("stream lost its owner across lifecycle/seek: %s", line) + } + } if _, err = m.Start(context.Background(), item, 0); err != nil { t.Fatalf("slot not reusable: %v", err) } @@ -318,7 +324,8 @@ func TestResolverDiagnosticDoesNotLogErrorText(t *testing.T) { t.Fatal(err) } defer m.Close() - s, err := m.Start(context.Background(), model.Item{ID: "episode:1:2:3", Kind: "episode"}, 0) + ctx := WithLogger(context.Background(), slog.With("user_id", "stream-owner", "user", "stream-owner")) + s, err := m.Start(ctx, model.Item{ID: "episode:1:2:3", Kind: "episode"}, 0) if err != nil { t.Fatal(err) } @@ -334,7 +341,7 @@ func TestResolverDiagnosticDoesNotLogErrorText(t *testing.T) { t.Fatalf("state: %s", got.State) } logText := logs.String() - if !strings.Contains(logText, "stage=resolve") || !strings.Contains(logText, "reason=error") || strings.Contains(logText, "secret-token") || strings.Contains(logText, "private.invalid") { + if !strings.Contains(logText, "user=stream-owner") || !strings.Contains(logText, "user_id=stream-owner") || !strings.Contains(logText, "stage=resolve") || !strings.Contains(logText, "reason=error") || strings.Contains(logText, "secret-token") || strings.Contains(logText, "private.invalid") { t.Fatalf("unsafe or incomplete resolver diagnostic: %s", logText) } }