Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 98 Pull Requests 11 Actions Packages Projects Releases Wiki Activity

fix(webui): make timeline session filter truthful and redact evidence refs (#637)

Browse Source 
Remediates the two blocking findings in review #522 on PR #849.

F1 — the session filter dimension was dead end to end. No source could
produce an event carrying a session identifier, so filter_events dropped
every event whenever session was supplied and the API answered with a
green, empty page. An empty result reads to an operator as "no such
session activity", which is a stronger and false claim.

Each source now declares which filter dimensions its records can actually
carry. The control-plane events table is (event_id, work_item_id,
event_type, message, created_at) and records no session, so that source
declares the session dimension unsupported rather than pretending to
answer it; the dead read of a non-existent session_id column is removed.
A CTH handoff comment declares its own Session field, so the handoff
adapter populates session_id from that declared field — authoritative
source data, never inferred from an actor, work item, or message text.

When no source that ran can carry a requested dimension, load_timeline
refuses with ok=false and a structured error naming the unsupported
filters and the per-source reason, and the route answers 422. A source
that can answer the dimension and simply matched nothing still returns
200 with an honest empty page. Ordering, pagination, and the issue/PR
filters are unchanged.

F2 — evidence_refs bypassed redaction and could emit a credential
verbatim. proof and decision were passed to _extract_evidence_refs before
redaction, and the SHA pattern matched any 7-40 character lowercase hex
run, which is exactly the shape of a Gitea access token.

Redaction now runs first and every derived value is taken from the
redacted text. A commit reference is recognised only where the source
text declares one (commit, head, base, sha, ...), so an undeclared hex
run is never lifted out of prose into a structured field; this also drops
the ordinary-word noise the reviewer noted. Every reference is then
independently revalidated against an allowed shape and a second redaction
pass immediately before serialization, failing closed by dropping
anything unproven and flagging the event sensitive. Actor is redacted for
the same reason, and a secret-shaped session value is dropped rather than
emitted. Legitimate issue, PR, short-SHA and full 40-character SHA
references stay usable.

Tests: 16 added. Session filtering is now driven through the CTH adapter
and the composed load_timeline/API path rather than a hand-built
WorkflowEvent, covering a match, an honest empty result, pagination and
ordering under the filter, the 422 refusal, and the per-source support
declaration. Redaction coverage asserts a synthetic 40-character hex
value (not a real credential) appears nowhere in the complete serialized
payload including evidence_refs, that the independent revalidation drops
unproven references, and that legitimate references still resolve.

pytest tests/test_webui_timeline.py: 42 passed (was 26).
pytest -k webui: 340 passed, 270 subtests passed (was 324).
Full suite: 4607 passed, 12 failed, 6 skipped, 684 subtests passed — the
same 12 failures as the master baseline, none under webui/.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-23 16:45:43 -04:00
co-authored by Claude Opus 4.8
parent a20975688d
commit 8ba1c5b87c
4 changed files with 568 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_webui_timeline.py
+254
View File
@@ -4,6 +4,7 @@ Covers the acceptance criteria: versioned schema, adaptation of control-plane
events and Gitea handoff comments, filter by issue/PR/session, redaction of
secret-like payloads, and stable pagination.
"""
import json
import os
import sqlite3
import sys
@@ -302,6 +303,243 @@ class TestLoadTimeline(unittest.TestCase):
self.assertIn("network down", handoff["reason"])
# A fabricated 40-character lowercase hex value with the shape of a Gitea
# personal access token. Never a real credential — its only job is to prove it
# cannot reach any part of a serialized timeline payload.
SYNTHETIC_SECRET_40_HEX = "a3f9c17be44d2058e6b17c9d0f5321ab77c4e9d1"
def _cth_comment(comment_id, *, created_at, session=None, decision="d", proof="none", **kw):
"""Build one real CTH comment record, optionally declaring a session."""
extra = {"Session": session} if session is not None else None
body = format_cth_body(
cth_type=kw.pop("cth_type", "Author Handoff"),
status=kw.pop("status", "ready"),
next_owner=kw.pop("next_owner", "reviewer"),
decision=decision,
proof=proof,
next_action=kw.pop("next_action", "review"),
ready_to_paste_prompt=kw.pop("ready_to_paste_prompt", "Review PR #1 now"),
extra_fields=extra,
)
return {
"id": comment_id,
"body": body,
"created_at": created_at,
"user": {"login": kw.pop("login", "jcwalker3")},
}
class TestSessionFilterThroughAdapter(unittest.TestCase):
"""F1: the session dimension must be real, or explicitly refused.
These drive the filter through the CTH adapter and the composed
``load_timeline``/API path, never through a hand-built ``WorkflowEvent``.
"""
def _source(self, comments):
return lambda kind, number: list(comments)
def test_adapter_populates_declared_session(self):
events = timeline.adapt_cth_comments(
[_cth_comment(1, created_at="2026-07-23T05:00:00Z", session="sess-alpha")],
kind="issue",
number=637,
)
self.assertEqual(len(events), 1)
self.assertEqual(events[0].session_id, "sess-alpha")
def test_session_filter_matches_through_adapter(self):
import tempfile
comments = [
_cth_comment(1, created_at="2026-07-23T09:00:00Z", session="sess-alpha"),
_cth_comment(2, created_at="2026-07-23T08:00:00Z", session="sess-alpha"),
_cth_comment(3, created_at="2026-07-23T10:00:00Z", session="sess-beta"),
]
with tempfile.TemporaryDirectory() as tmp:
db = os.path.join(tmp, "cp.sqlite3")
_seed_db(db)
snap = timeline.load_timeline(
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
issue=637, session="sess-alpha", db_path=db,
comment_source=self._source(comments),
)
d = snap.to_dict()
self.assertTrue(d["ok"])
self.assertIsNone(d["error"])
keys = [e["event_key"] for e in d["events"]]
# Only the two sess-alpha events, still in ascending timestamp order.
self.assertEqual(keys, ["cth:issue:637:2", "cth:issue:637:1"])
self.assertTrue(all(e["session_id"] == "sess-alpha" for e in d["events"]))
self.assertEqual(d["pagination"]["total"], 2)
def test_session_filter_paginates_and_keeps_order(self):
import tempfile
comments = [
_cth_comment(i, created_at=f"2026-07-23T0{i}:00:00Z", session="sess-alpha")
for i in range(1, 4)
]
with tempfile.TemporaryDirectory() as tmp:
db = os.path.join(tmp, "cp.sqlite3")
_seed_db(db)
kwargs = dict(
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
issue=637, session="sess-alpha", db_path=db,
comment_source=self._source(comments),
)
page1 = timeline.load_timeline(limit=2, offset=0, **kwargs).to_dict()
page2 = timeline.load_timeline(limit=2, offset=2, **kwargs).to_dict()
self.assertEqual(
[e["event_key"] for e in page1["events"]],
["cth:issue:637:1", "cth:issue:637:2"],
)
self.assertEqual(page1["pagination"]["next_offset"], 2)
self.assertEqual([e["event_key"] for e in page2["events"]], ["cth:issue:637:3"])
self.assertIsNone(page2["pagination"]["next_offset"])
def test_unknown_session_is_honestly_empty_when_supported(self):
"""A source that *can* answer the dimension may legitimately match nothing."""
import tempfile
comments = [_cth_comment(1, created_at="2026-07-23T09:00:00Z", session="sess-alpha")]
with tempfile.TemporaryDirectory() as tmp:
db = os.path.join(tmp, "cp.sqlite3")
_seed_db(db)
d = timeline.load_timeline(
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
issue=637, session="sess-nope", db_path=db,
comment_source=self._source(comments),
).to_dict()
self.assertTrue(d["ok"])
self.assertEqual(d["events"], [])
def test_session_filter_refused_when_no_source_can_answer(self):
"""The F1 defect: an empty-and-healthy page for an unanswerable filter."""
import tempfile
with tempfile.TemporaryDirectory() as tmp:
db = os.path.join(tmp, "cp.sqlite3")
_seed_db(db)
d = timeline.load_timeline(
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
issue=637, session="sess-alpha", db_path=db, comment_source=None,
).to_dict()
self.assertFalse(d["ok"])
self.assertEqual(d["error"]["code"], "filter_not_supported")
self.assertEqual(d["error"]["unsupported_filters"], ["session"])
self.assertEqual(d["events"], [])
self.assertEqual(d["pagination"]["total"], 0)
# The refusal states which source could not answer, and why.
explained = {r["source"] for r in d["error"]["sources"]}
self.assertEqual(explained, {"control_plane", "gitea_handoff"})
def test_control_plane_declares_session_unsupported(self):
import tempfile
with tempfile.TemporaryDirectory() as tmp:
db = os.path.join(tmp, "cp.sqlite3")
_seed_db(db)
d = timeline.load_timeline(
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
issue=637, session="sess-alpha", db_path=db,
comment_source=self._source([]),
).to_dict()
cp = [s for s in d["sources"] if s["name"] == "control_plane"][0]
handoff = [s for s in d["sources"] if s["name"] == "gitea_handoff"][0]
self.assertNotIn("session", cp["supported_filters"])
self.assertEqual(cp["unsupported_filters"], ["session"])
self.assertIn("session", handoff["supported_filters"])
self.assertEqual(handoff["unsupported_filters"], [])
def test_secret_shaped_session_value_is_dropped(self):
events = timeline.adapt_cth_comments(
[_cth_comment(1, created_at="2026-07-23T05:00:00Z", session=SYNTHETIC_SECRET_40_HEX)],
kind="issue",
number=637,
)
self.assertIsNone(events[0].session_id)
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(events[0].to_dict()))
class TestEvidenceRefRedaction(unittest.TestCase):
"""F2: evidence_refs must not be a hole in the redaction boundary."""
def _refs_for(self, *, proof="none", decision="d"):
events = timeline.adapt_cth_comments(
[_cth_comment(1, created_at="2026-07-23T05:00:00Z", proof=proof, decision=decision)],
kind="issue",
number=637,
)
self.assertEqual(len(events), 1)
return events[0]
def test_assigned_secret_never_reaches_evidence_refs(self):
ev = self._refs_for(proof=f"authenticated with token={SYNTHETIC_SECRET_40_HEX}")
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, ev.evidence_refs)
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(ev.to_dict()))
def test_bare_secret_shaped_value_never_reaches_evidence_refs(self):
# An undeclared hex run in proof text is not evidence of anything, and
# proof itself is never serialized — so the value has no way out.
ev = self._refs_for(proof=f"proof {SYNTHETIC_SECRET_40_HEX}", decision="rotate the credential")
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, ev.evidence_refs)
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(ev.to_dict()))
def test_secret_absent_from_complete_serialized_payload(self):
import tempfile
comments = [
_cth_comment(
1,
created_at="2026-07-23T09:00:00Z",
proof=f"lease token={SYNTHETIC_SECRET_40_HEX} and bare {SYNTHETIC_SECRET_40_HEX}",
decision=f"rotate api_key={SYNTHETIC_SECRET_40_HEX}",
)
]
with tempfile.TemporaryDirectory() as tmp:
db = os.path.join(tmp, "cp.sqlite3")
_seed_db(db)
snap = timeline.load_timeline(
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
issue=637, db_path=db, comment_source=lambda k, n: comments,
)
payload = json.dumps(snap.to_dict())
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, payload)
# And the surface is clean by the redaction policy's own detectors.
from webui import console_redaction
self.assertEqual(console_redaction.scan_for_secrets(snap.to_dict()), [])
def test_legitimate_references_still_usable(self):
head = "4f3a464a1c455b6ecaaae9b6eef496c8f8ed451a"
ev = self._refs_for(
proof=f"closes #637, PR #849, commit abc1234, at head {head}",
decision="none",
)
for token in ("#637", "#849", "abc1234", head):
self.assertIn(token, ev.evidence_refs)
def test_undeclared_hex_words_are_not_references(self):
ev = self._refs_for(proof="the record was defaced and the facade decayed")
self.assertEqual(ev.evidence_refs, ())
def test_refs_revalidated_independently_before_serialization(self):
"""Extraction is not trusted: the validator drops anything unproven."""
safe, dropped = timeline._validated_evidence_refs(
["#637", "abc1234", "not-a-ref", f"token={SYNTHETIC_SECRET_40_HEX}", ""]
)
self.assertEqual(safe, ("#637", "abc1234"))
self.assertTrue(dropped)
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(list(safe)))
def test_clean_event_is_not_marked_sensitive(self):
ev = self._refs_for(proof="closes #637")
self.assertEqual(ev.evidence_refs, ("#637",))
self.assertFalse(ev.sensitive)
class TestTimelineApi(unittest.TestCase):
def setUp(self):
self._prev_db = os.environ.get(control_plane_db.DB_PATH_ENV)
@@ -359,6 +597,22 @@ class TestTimelineApi(unittest.TestCase):
resp = self.client.get("/api/v1/timeline?issue=637")
self.assertNotIn("ghs_ABCDEF1234567890abcdef", resp.text)
def test_api_refuses_unanswerable_session_filter(self):
"""No handoff source is configured offline, so nothing can carry a session."""
resp = self.client.get("/api/v1/timeline?issue=637&session=sess-alpha")
self.assertEqual(resp.status_code, 422)
body = resp.json()
self.assertFalse(body["ok"])
self.assertEqual(body["error"]["code"], "filter_not_supported")
self.assertEqual(body["error"]["unsupported_filters"], ["session"])
self.assertEqual(body["events"], [])
self.assertEqual(body["pagination"]["total"], 0)
def test_api_unfiltered_read_stays_ok(self):
body = self.client.get("/api/v1/timeline?issue=637").json()
self.assertTrue(body["ok"])
self.assertIsNone(body["error"])
if __name__ == "__main__":
unittest.main()