diff --git a/control_plane_db.py b/control_plane_db.py index 5da9098..d7ad05b 100644 --- a/control_plane_db.py +++ b/control_plane_db.py @@ -577,6 +577,11 @@ class ControlPlaneDB: conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_role_model ON usage_events(role, model);") conn.execute("CREATE INDEX IF NOT EXISTS idx_usage_events_stage ON usage_events(stage);") + # #651 retention: cap growth so unauthenticated or high-volume ingest + # cannot DoS the control-plane DB (PR #876 F3). Applied after every write. + USAGE_EVENTS_MAX_ROWS = 10_000 + USAGE_EVENTS_RETENTION_DAYS = 90 + def record_usage_event( self, *, @@ -649,7 +654,35 @@ class ControlPlaneDB: ts, ), ) - return cursor.lastrowid + usage_id = cursor.lastrowid + self._enforce_usage_events_retention(conn) + return usage_id + + def _enforce_usage_events_retention(self, conn: sqlite3.Connection) -> None: + """Drop aged and excess usage_events rows (PR #876 F3).""" + # Age-based: ISO-8601 UTC timestamps compare lexicographically. + cutoff = ( + datetime.now(timezone.utc) + - timedelta(days=int(self.USAGE_EVENTS_RETENTION_DAYS)) + ).strftime("%Y-%m-%dT%H:%M:%SZ") + conn.execute( + "DELETE FROM usage_events WHERE created_at < ?", + (cutoff,), + ) + # Count-based: keep the newest USAGE_EVENTS_MAX_ROWS by usage_id. + max_rows = int(self.USAGE_EVENTS_MAX_ROWS) + if max_rows > 0: + conn.execute( + """ + DELETE FROM usage_events + WHERE usage_id NOT IN ( + SELECT usage_id FROM usage_events + ORDER BY usage_id DESC + LIMIT ? + ) + """, + (max_rows,), + ) def query_usage_events( self, diff --git a/docs/observability/analytics-instrumentation.md b/docs/observability/analytics-instrumentation.md index 3994045..55d7057 100644 --- a/docs/observability/analytics-instrumentation.md +++ b/docs/observability/analytics-instrumentation.md @@ -89,11 +89,17 @@ record_usage( ) ``` -### HTTP Ingestion API +### HTTP Ingestion API (authorized write) + +`POST /api/v1/analytics/usage` is a **gated write**. It runs through +`console_authz` action `record_analytics_usage` (operator+, Phase 2 execution). +Unauthenticated or phase-inactive requests receive **403** and do not write. +Prefer in-process `record_usage` for MCP / session instrumentation. ```http POST /api/v1/analytics/usage HTTP/1.1 Content-Type: application/json +# Requires authenticated principal with record_analytics_usage execution enabled { "remote": "dadeschools", @@ -114,6 +120,18 @@ Content-Type: application/json } ``` +### Retention + +`usage_events` is retained with hard caps applied on every write: + +| Limit | Default | +|---|---| +| Max rows | 10,000 (`ControlPlaneDB.USAGE_EVENTS_MAX_ROWS`) | +| Max age | 90 days (`ControlPlaneDB.USAGE_EVENTS_RETENTION_DAYS`) | + +Older rows (by `created_at`) and excess oldest rows (by `usage_id`) are deleted +after each insert so unbounded growth / DoS-by-volume cannot fill the DB. + --- ## 6. Querying Analytics API diff --git a/docs/webui-authz-audit.md b/docs/webui-authz-audit.md index 8776429..290c3d9 100644 --- a/docs/webui-authz-audit.md +++ b/docs/webui-authz-audit.md @@ -91,6 +91,7 @@ already define, and a regression test asserts each mapping matches. | `close_pr` | controller | privileged | `gitea.pr.close` | Yes | No | No | 3 | | `merge_pr` | controller | privileged | `gitea.pr.merge` | Yes | **Yes** | **Yes** | 3 | | `delete_branch` | admin | destructive | `gitea.branch.delete` | Yes | **Yes** | **Yes** | 3 | +| `record_analytics_usage` | operator | gated_write | `runtime.record_analytics_usage` | Yes | No | No | 2 | **Dual control** means the acting principal may not be the sole authority: a second distinct principal must confirm. **Break-glass** means the action is diff --git a/task_capability_map.py b/task_capability_map.py index bf991e4..c5911c0 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -495,6 +495,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.issue.comment", "role": "author", }, + + # #651 console analytics ingest — control-plane DB write, not a Gitea API + # call. Authority comes from console RBAC (operator+) plus phase gating; + # permission string is a non-Gitea runtime capability so no Gitea profile + # can satisfy it by accident. + "record_analytics_usage": { + "permission": "runtime.record_analytics_usage", + "role": "author", + }, } diff --git a/tests/test_webui_analytics.py b/tests/test_webui_analytics.py index c71322b..bf1a419 100644 --- a/tests/test_webui_analytics.py +++ b/tests/test_webui_analytics.py @@ -165,7 +165,8 @@ class AnalyticsWebUITest(unittest.TestCase): self.assertEqual(data["total_events"], 1) self.assertIn("gemini-3.6-flash", data["by_model"]) - def test_analytics_ingest_endpoint(self) -> None: + def test_analytics_ingest_unauthorized_denied(self) -> None: + """F2: unauthenticated POST must not write the control-plane DB.""" payload = { "remote": "dadeschools", "org": "Scaled-Tech-Consulting", @@ -180,16 +181,71 @@ class AnalyticsWebUITest(unittest.TestCase): "metadata": "Review note token=secret456", } response = self.client.post("/api/v1/analytics/usage", json=payload) - self.assertEqual(response.status_code, 201) + self.assertEqual(response.status_code, 403) res_json = response.json() - self.assertTrue(res_json["ok"]) - self.assertGreater(res_json["usage_id"], 0) + self.assertFalse(res_json.get("ok", True)) + self.assertEqual(res_json.get("error"), "unauthorized") + authorization = res_json.get("authorization") or {} + self.assertFalse(authorization.get("allowed")) + self.assertFalse(authorization.get("execution_enabled")) - # Check that it appears in GET /api/v1/analytics + # No new row written res2 = self.client.get("/api/v1/analytics") self.assertEqual(res2.status_code, 200) - data2 = res2.json() - self.assertEqual(data2["total_events"], 2) + self.assertEqual(res2.json()["total_events"], 1) + + def test_html_escapes_script_bearing_model_role_stage(self) -> None: + """F1: stored XSS — dynamic model/role/stage must render escaped.""" + xss = '' + record_usage( + db_path=self.db_path, + remote="dadeschools", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + role=xss, + model=xss, + stage=xss, + issue_number=999, + status="success", + ) + response = self.client.get("/analytics") + self.assertEqual(response.status_code, 200) + # Raw tag must not appear; escaped form must. + self.assertNotIn("", response.text) + self.assertIn("<script>alert(1)</script>", response.text) + + def test_load_analytics_coerces_none_scope(self) -> None: + """F4: None remote/org/repo become empty strings, never None.""" + snapshot = load_analytics(db_path=self.db_path, remote=None, org=None, repo=None) + self.assertIsInstance(snapshot.remote, str) + self.assertIsInstance(snapshot.org, str) + self.assertIsInstance(snapshot.repo, str) + self.assertEqual(snapshot.remote, "") + self.assertEqual(snapshot.org, "") + self.assertEqual(snapshot.repo, "") + + def test_usage_events_retention_max_rows(self) -> None: + """F3: record_usage_event enforces USAGE_EVENTS_MAX_ROWS.""" + db = control_plane_db.ControlPlaneDB(db_path=self.db_path) + original_max = db.USAGE_EVENTS_MAX_ROWS + try: + db.USAGE_EVENTS_MAX_ROWS = 3 + for i in range(5): + db.record_usage_event( + remote="dadeschools", + org="org", + repo="repo", + role="author", + model=f"model-{i}", + stage="test", + ) + rows = db.query_usage_events(limit=100) + self.assertLessEqual(len(rows), 3) + # Newest three retained + models = {r["model"] for r in rows} + self.assertEqual(models, {"model-2", "model-3", "model-4"}) + finally: + db.USAGE_EVENTS_MAX_ROWS = original_max if __name__ == "__main__": diff --git a/webui/analytics_loader.py b/webui/analytics_loader.py index d6ef3d3..15a6138 100644 --- a/webui/analytics_loader.py +++ b/webui/analytics_loader.py @@ -306,6 +306,11 @@ def load_analytics( model_filter = (model or "").strip() or None stage_filter = (stage or "").strip() or None + # F4: coerce optional scope filters to str so AnalyticsSnapshot never holds None. + scope_remote = (remote or "").strip() + scope_org = (org or "").strip() + scope_repo = (repo or "").strip() + try: db = control_plane_db.ControlPlaneDB(db_path=db_path) rows = db.query_usage_events( @@ -326,9 +331,9 @@ def load_analytics( ok=False, reason=f"control_plane_db_unavailable: {exc}", schema_version=ANALYTICS_SCHEMA_VERSION, - remote=remote, - org=org, - repo=repo, + remote=scope_remote, + org=scope_org, + repo=scope_repo, total_events=0, overall_summary=empty_summary, by_project={}, @@ -346,24 +351,24 @@ def load_analytics( UsageEvent( usage_id=r["usage_id"], session_id=r.get("session_id"), - remote=r.get("remote", remote), - org=r.get("org", org), - repo=r.get("repo", repo), + remote=r.get("remote") or scope_remote, + org=r.get("org") or scope_org, + repo=r.get("repo") or scope_repo, project_id=r.get("project_id"), - role=r.get("role", "unknown"), - model=r.get("model", "unknown"), + role=r.get("role") or "unknown", + model=r.get("model") or "unknown", issue_number=r.get("issue_number"), pr_number=r.get("pr_number"), - stage=r.get("stage", "unknown"), + stage=r.get("stage") or "unknown", input_tokens=r.get("input_tokens"), output_tokens=r.get("output_tokens"), total_tokens=r.get("total_tokens"), estimated_cost_usd=r.get("estimated_cost_usd"), latency_ms=r.get("latency_ms"), duration_ms=r.get("duration_ms"), - status=r.get("status", "success"), + status=r.get("status") or "success", metadata=meta, - created_at=r.get("created_at", ""), + created_at=r.get("created_at") or "", ) ) @@ -410,9 +415,9 @@ def load_analytics( ok=True, reason="ok", schema_version=ANALYTICS_SCHEMA_VERSION, - remote=remote, - org=org, - repo=repo, + remote=scope_remote, + org=scope_org, + repo=scope_repo, total_events=len(parsed_events), overall_summary=overall_summary, by_project=by_project, diff --git a/webui/analytics_views.py b/webui/analytics_views.py index b2583ca..506aa96 100644 --- a/webui/analytics_views.py +++ b/webui/analytics_views.py @@ -2,52 +2,59 @@ from __future__ import annotations +import html + from webui.analytics_loader import AnalyticsSnapshot, GroupMetrics, UsageEvent from webui.layout import render_page +def _escape(text: object) -> str: + """HTML-escape dynamic analytics fields (mirrors audit_views / project_views).""" + return html.escape(str(text), quote=True) + + def _render_badge(text: str, badge_type: str = "muted") -> str: - return f'{text}' + return f'{_escape(text)}' def _render_group_table(title: str, groups: dict[str, GroupMetrics], key_header: str = "Group") -> str: if not groups: return ( - f"

{title}

" + f"

{_escape(title)}

" '

No telemetry events recorded for this dimension.

' ) rows = [] for key, g in sorted(groups.items(), key=lambda x: x[1].total_events, reverse=True): cost_cell = ( - f'{g.display_cost}' + f'{_escape(g.display_cost)}' if g.events_with_cost > 0 else _render_badge("Unknown") ) tokens_cell = ( - g.display_tokens + _escape(g.display_tokens) if g.events_with_tokens > 0 else _render_badge("Unknown") ) lat_p50 = ( - g.display_latency_p50 + _escape(g.display_latency_p50) if g.events_with_latency > 0 else _render_badge("Unknown") ) lat_p90 = ( - g.display_latency_p90 + _escape(g.display_latency_p90) if g.events_with_latency > 0 else _render_badge("Unknown") ) dur_avg = ( - g.display_duration_avg + _escape(g.display_duration_avg) if g.events_with_duration > 0 else _render_badge("Unknown") ) rows.append( "" - f"{key}" + f"{_escape(key)}" f"{g.total_events}" f"{tokens_cell}" f"{cost_cell}" @@ -59,12 +66,12 @@ def _render_group_table(title: str, groups: dict[str, GroupMetrics], key_header: rows_html = "".join(rows) return f""" -

{title}

+

{_escape(title)}

- + @@ -85,26 +92,49 @@ def _render_events_table(events: tuple[UsageEvent, ...]) -> str: if not events: return ( "

Recent Usage & Instrumentation Events

" - '

No individual telemetry events recorded yet. Opt-in instrumentation via session logging or POST /api/v1/analytics/usage.

' + '

No individual telemetry events recorded yet. Opt-in instrumentation via session logging or authorized POST /api/v1/analytics/usage.

' ) rows = [] for e in list(events)[-50:]: # Display latest 50 - work_item = f"issue #{e.issue_number}" if e.issue_number else (f"pr #{e.pr_number}" if e.pr_number else "unlinked") - tokens = f"{e.total_tokens:,}" if e.total_tokens is not None else _render_badge("Unknown") - cost = f"${e.estimated_cost_usd:.4f}" if e.estimated_cost_usd is not None else _render_badge("Unknown") - latency = f"{e.latency_ms} ms" if e.latency_ms is not None else _render_badge("Unknown") - duration = f"{e.duration_ms} ms" if e.duration_ms is not None else _render_badge("Unknown") - status_badge = _render_badge(e.status, "success" if e.status == "success" else "danger") + if e.issue_number is not None: + work_item = f"issue #{e.issue_number}" + elif e.pr_number is not None: + work_item = f"pr #{e.pr_number}" + else: + work_item = "unlinked" + tokens = ( + _escape(f"{e.total_tokens:,}") + if e.total_tokens is not None + else _render_badge("Unknown") + ) + cost = ( + _escape(f"${e.estimated_cost_usd:.4f}") + if e.estimated_cost_usd is not None + else _render_badge("Unknown") + ) + latency = ( + _escape(f"{e.latency_ms} ms") + if e.latency_ms is not None + else _render_badge("Unknown") + ) + duration = ( + _escape(f"{e.duration_ms} ms") + if e.duration_ms is not None + else _render_badge("Unknown") + ) + status_badge = _render_badge( + e.status, "success" if e.status == "success" else "danger" + ) rows.append( "" f"" - f"" - f"" - f"" - f"" - f"" + f"" + f"" + f"" + f"" + f"" f"" f"" f"" @@ -145,15 +175,32 @@ def render_analytics_page(snapshot: AnalyticsSnapshot) -> str: """Render the main Model Usage & Performance Analytics console page.""" summary = snapshot.overall_summary - kpi_tokens = summary.display_tokens if summary.events_with_tokens > 0 else _render_badge("Unknown") - kpi_cost = summary.display_cost if summary.events_with_cost > 0 else _render_badge("Unknown") - kpi_lat_p50 = summary.display_latency_p50 if summary.events_with_latency > 0 else _render_badge("Unknown") - kpi_dur_avg = summary.display_duration_avg if summary.events_with_duration > 0 else _render_badge("Unknown") + kpi_tokens = ( + _escape(summary.display_tokens) + if summary.events_with_tokens > 0 + else _render_badge("Unknown") + ) + kpi_cost = ( + _escape(summary.display_cost) + if summary.events_with_cost > 0 + else _render_badge("Unknown") + ) + kpi_lat_p50 = ( + _escape(summary.display_latency_p50) + if summary.events_with_latency > 0 + else _render_badge("Unknown") + ) + kpi_dur_avg = ( + _escape(summary.display_duration_avg) + if summary.events_with_duration > 0 + else _render_badge("Unknown") + ) status_notice = "" if not snapshot.ok: status_notice = ( - f'
Degraded Data Source: {snapshot.reason}
' + f'
Degraded Data Source: ' + f'{_escape(snapshot.reason)}
' ) body_html = f""" diff --git a/webui/app.py b/webui/app.py index d0615f1..f7648eb 100644 --- a/webui/app.py +++ b/webui/app.py @@ -607,14 +607,53 @@ async def api_v1_analytics(request: Request) -> JSONResponse: async def api_v1_analytics_ingest(request: Request) -> JSONResponse: - """Optional session instrumentation ingestion endpoint (#651).""" + """Optional session instrumentation ingestion endpoint (#651). + + Fail-closed write: every request is authorized through console_authz + (``record_analytics_usage``) before any control-plane DB mutation. Phase 1 + keeps ``execution_enabled=False`` and denies unauthenticated callers, so + this route cannot be used as an unauthenticated write or XSS injection + vector (PR #876 F2). + """ try: body = await request.json() except Exception: - return JSONResponse({"error": "invalid_json", "detail": "body must be valid JSON"}, status_code=400) - + body = {} if not isinstance(body, dict): - return JSONResponse({"error": "invalid_payload", "detail": "payload must be a JSON object"}, status_code=400) + body = {} + + principal = resolve_principal(headers=dict(request.headers)) + decision = authorize( + "record_analytics_usage", principal, for_execution=True + ) + allowed = bool(decision.allowed and decision.execution_enabled) + console_audit.record_event( + action_id="record_analytics_usage", + result=( + console_audit.RESULT_ALLOWED + if allowed + else console_audit.RESULT_DENIED + ), + decision=decision, + principal=principal, + target=_audit_target("record_analytics_usage", body), + request_id=_request_id(), + detail=decision.detail, + ) + authorization = decision.to_dict() + if not allowed: + return JSONResponse( + { + "ok": False, + "error": "unauthorized", + "detail": ( + "POST /api/v1/analytics/usage requires an authenticated " + "principal with record_analytics_usage execution enabled" + ), + "authorization": authorization, + }, + status_code=403, + ) usage_id = record_usage( session_id=body.get("session_id"), @@ -636,7 +675,10 @@ async def api_v1_analytics_ingest(request: Request) -> JSONResponse: status=body.get("status", "success"), metadata=body.get("metadata"), ) - return JSONResponse({"ok": True, "usage_id": usage_id}, status_code=201) + return JSONResponse( + {"ok": True, "usage_id": usage_id, "authorization": authorization}, + status_code=201, + ) async def method_not_allowed(request: Request, _exc: Exception) -> Response: diff --git a/webui/console_authz.py b/webui/console_authz.py index 2e64b2c..e64d5f8 100644 --- a/webui/console_authz.py +++ b/webui/console_authz.py @@ -236,6 +236,19 @@ _ACTION_SPECS: tuple[ConsoleAction, ...] = ( phase=3, summary="Remove a remote feature branch.", ), + # #651 analytics ingest: local control-plane write, not a Gitea mutation. + # Phase 2 gated write so Phase 1 (ACTIVE_PHASE=1) fails closed on execution. + ConsoleAction( + action_id="record_analytics_usage", + task_key="record_analytics_usage", + action_class=CLASS_WRITE, + minimum_role=OPERATOR, + requires_confirmation=True, + dual_control=False, + break_glass=False, + phase=2, + summary="Ingest a model-usage / latency analytics event into the control-plane DB.", + ), ) ACTIONS: dict[str, ConsoleAction] = {a.action_id: a for a in _ACTION_SPECS}
{key_header}{_escape(key_header)} Events Total Tokens Est. Cost
#{e.usage_id}{e.created_at}{e.role}{e.model}{e.stage}{work_item}{_escape(e.created_at)}{_escape(e.role)}{_escape(e.model)}{_escape(e.stage)}{_escape(work_item)}{tokens}{cost}{latency}