From 834f1f26a9bbdcb4ab27c30a4102cd295e2952ee Mon Sep 17 00:00:00 2001 From: Walter Oggioni Date: Sun, 27 Sep 2026 21:10:55 +0000 Subject: [PATCH] Tag traced requests with the session id and OIDC sub --- server/README.md | 6 +- server/packages/tavolo-app/src/tavolo/app.py | 5 + .../tavolo-app/src/tavolo/telemetry.py | 40 +++++- server/tests/test_otel.py | 10 ++ server/tests/test_telemetry.py | 115 ++++++++++++++++++ 5 files changed, 173 insertions(+), 3 deletions(-) diff --git a/server/README.md b/server/README.md index fdfdba4..1fffee3 100644 --- a/server/README.md +++ b/server/README.md @@ -134,7 +134,11 @@ With `OTEL_ENABLED` set, the app exports over OTLP/HTTP (kaya-otel plus database spans for the Postgres (asyncpg, through Tortoise) and Redis clients. Redis command values are redacted by the instrumentation (`GET ? ?`) and asyncpg spans carry the parameterized SQL only — no - parameter values. + parameter values. Each HTTP/WebSocket span is tagged with the request's + session and user, so all traces of a session can be found together: + `session.id` is a truncated SHA-256 of the session id (the cookie is a + credential, so it is never exported verbatim) and `user.id` is the OIDC + `sub`, present once logged in. - **Metrics**: the HTTP server metrics from kaya-otel, plus `tavolo.active_users` (sessions in the session store), `tavolo.ongoing_games` (live matches not finished yet, maintained by the diff --git a/server/packages/tavolo-app/src/tavolo/app.py b/server/packages/tavolo-app/src/tavolo/app.py index eb08d1f..55e8194 100644 --- a/server/packages/tavolo-app/src/tavolo/app.py +++ b/server/packages/tavolo-app/src/tavolo/app.py @@ -52,6 +52,7 @@ from .telemetry import ( CountingInMemorySessionStore, CountingRedisSessionStore, otel_headers, + session_tag_hook, telemetry_from_settings, ) @@ -111,6 +112,10 @@ def otel_mixin_from_settings( "endpoint": settings.otel_exporter_endpoint, "headers": otel_headers(settings), "excluded_paths": settings.otel_excluded_paths, + # Runs after the session mixin's after hook, so the session id has + # been persisted/minted and the OIDC claims are on the session. + "server_response_hook": session_tag_hook, + "websocket_close_hook": session_tag_hook, } if meter_provider is not None: kwargs["meter_provider"] = meter_provider diff --git a/server/packages/tavolo-app/src/tavolo/telemetry.py b/server/packages/tavolo-app/src/tavolo/telemetry.py index 09b8ab9..dbce9a9 100644 --- a/server/packages/tavolo-app/src/tavolo/telemetry.py +++ b/server/packages/tavolo-app/src/tavolo/telemetry.py @@ -12,7 +12,10 @@ the test suite never need them): CPU time/utilization, memory usage and CPython garbage-collection counters; * OTLP log export through a stdlib :class:`logging.Handler` attached to - the root logger, mirroring what the console sees. + the root logger, mirroring what the console sees; +* session/user tagging of request and websocket spans through + :func:`session_tag_hook` (kaya-otel's response/close hooks), so traces + can be correlated with a session (hashed id) and an OIDC user. The store counts are asynchronous while OpenTelemetry instrument callbacks are synchronous, so :class:`TelemetryMixin` polls the stores on @@ -21,10 +24,11 @@ a background task and its callbacks report the cached values. from __future__ import annotations import asyncio +import hashlib import logging from dataclasses import dataclass from logging import getLogger -from typing import TYPE_CHECKING, Any, Dict, Optional, Union +from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Union from kaya.core import KayaApp, KayaMixin from kaya.session import InMemorySessionStore, SessionStore @@ -79,6 +83,38 @@ def _signal_endpoint(base: Optional[str], signal: str) -> Optional[str]: return f"{base}/{signal}" if base else None +def session_tag_hook(span: Any, carrier: Any) -> None: + """Tag an HTTP request/websocket span with the request's session/user. + + Wired into kaya-otel as its ``server_response_hook`` and + ``websocket_close_hook``: those run after the session mixin's after + hook has persisted (and, at login, minted) the session, so the final + session id is visible here. + + The session id is the session cookie, i.e. a credential, so only a + truncated SHA-256 of it is exported; session ids are + ``secrets.token_urlsafe(32)`` (256 bits), which makes the digest + irreversible in practice. The OIDC ``sub`` is a pseudonymous + identifier already persisted in Postgres and served by the API, so it + is exported verbatim as ``user.id`` (OpenTelemetry's current end-user + attribute; ``enduser.id`` is PII-flagged and being superseded). + """ + session = getattr(carrier, "session", None) + if session is None: + return + session_id = getattr(session, "id", None) + if session_id: + span.set_attribute( + "session.id", + hashlib.sha256(session_id.encode()).hexdigest()[:16], + ) + user = session.get("oidc_user") + if isinstance(user, Mapping): + sub = user.get("sub") + if sub: + span.set_attribute("user.id", str(sub)) + + class _NoOtelLogs(logging.Filter): """Drop OpenTelemetry's own log records. diff --git a/server/tests/test_otel.py b/server/tests/test_otel.py index 1037d55..5a42c31 100644 --- a/server/tests/test_otel.py +++ b/server/tests/test_otel.py @@ -16,6 +16,7 @@ from unittest.mock import patch from tavolo.app import otel_mixin_from_settings from tavolo.config import Settings +from tavolo.telemetry import session_tag_hook def _settings(env: dict) -> Settings: @@ -58,8 +59,17 @@ class OTelMixinFromSettingsTests(unittest.TestCase): "endpoint": "http://collector:4318", "headers": {"Authorization": "Bearer t", "X-Tenant": "one"}, "excluded_paths": ("/api/health",), + "server_response_hook": session_tag_hook, + "websocket_close_hook": session_tag_hook, }, mixin.kwargs) + def test_session_tag_hooks_are_wired(self): + with _stub_kaya_otel(): + mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "1"})) + assert isinstance(mixin, _StubOTelMixin) + self.assertIs(session_tag_hook, mixin.kwargs["server_response_hook"]) + self.assertIs(session_tag_hook, mixin.kwargs["websocket_close_hook"]) + def test_defaults_when_only_enabled(self): with _stub_kaya_otel(): mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "on"})) diff --git a/server/tests/test_telemetry.py b/server/tests/test_telemetry.py index 48f63db..8f1e819 100644 --- a/server/tests/test_telemetry.py +++ b/server/tests/test_telemetry.py @@ -5,6 +5,7 @@ so the provider-dependent tests skip when they are not installed. """ from __future__ import annotations +import hashlib import logging import os import sys @@ -19,6 +20,7 @@ from tavolo.telemetry import ( CountingRedisSessionStore, TelemetryMixin, _NoOtelLogs, + session_tag_hook, telemetry_from_settings, ) from tavolo.config import Settings @@ -137,6 +139,119 @@ class NoOtelLogsFilterTest(unittest.TestCase): ) +class _RecordedSpan: + def __init__(self) -> None: + self.attributes: dict = {} + + def set_attribute(self, key: str, value) -> None: + self.attributes[key] = value + + +class SessionTagHookTest(unittest.TestCase): + def test_tags_authenticated_session(self) -> None: + span = _RecordedSpan() + carrier = SimpleNamespace( + session=Session("sid", {"oidc_user": {"sub": "alice"}}) + ) + session_tag_hook(span, carrier) + self.assertEqual( + hashlib.sha256(b"sid").hexdigest()[:16], + span.attributes["session.id"], + ) + self.assertEqual("alice", span.attributes["user.id"]) + + def test_tags_session_without_user(self) -> None: + span = _RecordedSpan() + session_tag_hook(span, SimpleNamespace(session=Session("sid", {}))) + self.assertIn("session.id", span.attributes) + self.assertNotIn("user.id", span.attributes) + + def test_tags_user_when_session_id_not_yet_minted(self) -> None: + span = _RecordedSpan() + carrier = SimpleNamespace( + session=Session(None, {"oidc_user": {"sub": "alice"}}) + ) + session_tag_hook(span, carrier) + self.assertNotIn("session.id", span.attributes) + self.assertEqual("alice", span.attributes["user.id"]) + + def test_missing_session_is_a_noop(self) -> None: + span = _RecordedSpan() + session_tag_hook(span, SimpleNamespace()) + self.assertEqual({}, span.attributes) + + def test_malformed_user_is_ignored(self) -> None: + span = _RecordedSpan() + carrier = SimpleNamespace(session=Session("sid", {"oidc_user": "alice"})) + session_tag_hook(span, carrier) + self.assertIn("session.id", span.attributes) + self.assertNotIn("user.id", span.attributes) + + +@unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed") +class SessionTagIntegrationTest(unittest.TestCase): + """The hook is wired as kaya-otel's response hook, so it sees the + session after the session mixin persisted and minted it.""" + + @async_test + async def test_spans_carry_hashed_session_and_oidc_sub(self) -> None: + from httpx import ASGITransport, AsyncClient + from kaya.core import HttpContext, KayaApp + from kaya.otel import OTelMixin + from kaya.session import InMemorySessionStore, SessionMixin + + span_exporter = InMemorySpanExporter() + tracer_provider = TracerProvider( + resource=Resource.create({"service.name": "test"}) + ) + tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter)) + app = KayaApp( + mixins=[ + OTelMixin( + service_name="test", + tracer_provider=tracer_provider, + meter_provider=MeterProvider( + metric_readers=[InMemoryMetricReader()] + ), + server_response_hook=session_tag_hook, + websocket_close_hook=session_tag_hook, + ), + SessionMixin(InMemorySessionStore()), + ] + ) + + @app.GET("/login") + async def login(ctx: HttpContext) -> None: + session = ctx.session + assert session is not None + session["oidc_user"] = {"sub": "alice"} + await ctx.send_str(200, "ok") + + @app.GET("/me") + async def me(ctx: HttpContext) -> None: + await ctx.send_str(200, "ok") + + transport = ASGITransport(app=app) # type: ignore[arg-type] + async with AsyncClient( + transport=transport, base_url="http://127.0.0.1" + ) as client: + first = await client.get("/login") + second = await client.get("/me") + self.assertEqual(200, first.status_code) + self.assertEqual(200, second.status_code) + # The cookie value is the session id; only its digest is exported. + cookie = first.cookies["session_id"] + digest = hashlib.sha256(cookie.encode()).hexdigest()[:16] + + spans = {span.name: span for span in span_exporter.get_finished_spans()} + self.assertEqual({"GET /login", "GET /me"}, set(spans)) + for name, span in spans.items(): + attributes = span.attributes + assert attributes is not None + self.assertEqual(digest, attributes["session.id"], name) + self.assertEqual("alice", attributes["user.id"], name) + + def _metric_points(reader) -> dict: """Map metric name -> list of (attributes, value) data points.""" data = reader.get_metrics_data()