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

fix(webui): validate externally influenced event_type at the timeline boundary (#637)

Browse Source 
Review #526 found that event_type reached the serialized timeline payload
without crossing the redaction/validation boundary every other free-text
field on the same event crosses. A synthetic secret-shaped 40-hex canary
was redacted through message but survived verbatim through event_type on
the same control-plane record.

CTH heading path: CTH_TYPES is now the single authority for what a CTH
type may be. canonical_thread_handoff.is_known_cth_type() is that
authority, used by format_cth_body (write), assess_cth_comment (assess),
and now the read path too; parse_cth_comment reports membership as
cth_type_known and stays total. adapt_cth_comments serializes
handoff:<type> only for a declared type and otherwise emits the constant
handoff:unrecognized, so arbitrary, malformed, secret-shaped, or
whitespace-manipulated heading content never becomes an event_type.

Control-plane path: a stored event_type is treated as source data.
_safe_cp_event_type accepts only an ordinary identifier that is not a
bare secret-shaped hex run and that a redaction pass leaves unchanged;
anything else fails closed to the constant unsafe:redacted and marks the
event sensitive. The value is never emitted verbatim, never partially
sanitized, and never rewritten into a different valid-looking type.

Remaining serialized-field audit: event_key ids must be plain numeric
identifiers, the CTH adapter refuses a scope it cannot express, per-source
failure reasons are redacted (they can quote an authenticated fetch error),
and echoed scope/filter values are guarded so reflection is not a bypass.

Legitimate values are preserved: every declared CTH type, the real
producer types (assigned, lease_released, lease_adopted,
dependency_edge_state_change, allocation, pr.opened, lease.renew), and
the existing issue, PR, evidence, and full-SHA references.

Tests seed the canary independently through both event_type paths with a
benign message, so message redaction cannot be why they pass; each
inspects event_type directly, asserts the canary is absent from the
complete serialized payload, and asserts scan_for_secrets finds nothing.

tests/test_webui_timeline.py 64 passed, 15 subtests
pytest -k webui 362 passed, 285 subtests
pytest -k redact 79 passed
tests/test_canonical_thread_handoff.py tests/test_control_plane_db.py 29 passed

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-23 18:45:54 -04:00
co-authored by Claude Opus 4.8
parent 2d95e0fcc6
commit a1e5a4af8c
3 changed files with 558 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_webui_timeline.py
+404 -1
View File
@@ -16,7 +16,13 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
import control_plane_db
from canonical_thread_handoff import format_cth_body
from canonical_thread_handoff import (
CTH_TYPES,
MARKER,
format_cth_body,
is_known_cth_type,
parse_cth_comment,
)
from webui import timeline
from webui.app import create_app
@@ -614,5 +620,402 @@ class TestTimelineApi(unittest.TestCase):
self.assertIsNone(body["error"])
def _seed_event(path: str, *, event_type: str, message: str, created_at: str) -> None:
"""Append one control-plane event with a caller-chosen ``event_type``.
The ``events`` table stores whatever a producer writes, so this seeds the
adapter the way a hostile or buggy producer would.
"""
conn = sqlite3.connect(path)
try:
wid = conn.execute(
"SELECT work_item_id FROM work_items WHERE kind='issue' AND number=637"
).fetchone()[0]
conn.execute(
"INSERT INTO events(work_item_id, event_type, message, created_at) VALUES (?,?,?,?)",
(wid, event_type, message, created_at),
)
conn.commit()
finally:
conn.close()
def _raw_cth_body(heading: str, **fields) -> str:
"""Build a CTH comment with an arbitrary heading.
``format_cth_body`` refuses an undeclared type, which is exactly the write
path already covered. The read path must cope with a body that never went
through it, so this writes the marker and heading directly.
"""
lines = [MARKER, f"## CTH: {heading}", ""]
base = {
"Status": "ready",
"Next owner": "reviewer",
"Current blocker": "none",
"Decision": "d",
"Proof": "none",
"Next action": "review",
"Ready-to-paste prompt": "Review PR #1 now",
}
base.update(fields)
lines.extend(f"{key}: {value}" for key, value in base.items())
return "\n".join(lines)
class TestEventTypeBoundary(unittest.TestCase):
"""F2 residual: event_type must cross the same boundary as every other field.
The canary is seeded through ``event_type`` *only*, with a benign message,
so message redaction cannot be what makes these pass. Each case inspects
``event_type`` explicitly and then the complete serialized payload.
"""
# ---- control-plane stored event_type ---------------------------------- #
def test_cp_event_type_canary_never_serialized(self):
rows = [{
"event_id": 1,
"event_type": SYNTHETIC_SECRET_40_HEX,
"message": "deploy completed", # benign: no redaction happens here
"created_at": "2026-07-23T01:00:00Z",
"kind": "issue",
"number": 637,
}]
events = timeline.adapt_cp_events(rows)
self.assertEqual(len(events), 1)
ev = events[0]
# The field itself, inspected directly.
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, ev.event_type)
self.assertEqual(ev.event_type, timeline.UNSAFE_EVENT_TYPE)
# Message redaction is provably not what saved us: it is untouched.
self.assertEqual(ev.message, "deploy completed")
# And nowhere in the serialized record.
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(ev.to_dict()))
# An unsafe value is a visible fact, not a silent substitution.
self.assertTrue(ev.sensitive)
def test_cp_same_canary_in_message_and_event_type(self):
"""The decisive case: one record, one value, two fields, one verdict."""
rows = [{
"event_id": 2,
"event_type": SYNTHETIC_SECRET_40_HEX,
"message": f"deploy token={SYNTHETIC_SECRET_40_HEX}",
"created_at": "2026-07-23T01:00:00Z",
"kind": "issue",
"number": 637,
}]
ev = timeline.adapt_cp_events(rows)[0]
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, ev.message or "")
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, ev.event_type)
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(ev.to_dict()))
def test_cp_unsafe_event_type_is_not_rewritten_as_a_valid_one(self):
"""A refused value must not be disguised as some other real event type."""
rows = [{
"event_id": 3,
"event_type": SYNTHETIC_SECRET_40_HEX,
"message": "m",
"created_at": "2026-07-23T01:00:00Z",
"kind": "issue",
"number": 637,
}]
ev = timeline.adapt_cp_events(rows)[0]
for legitimate in ("allocation", "pr.opened", "lease.renew", "assigned"):
self.assertNotEqual(ev.event_type, legitimate)
def test_cp_assigned_secret_in_event_type_refused(self):
rows = [{
"event_id": 4,
"event_type": f"token={SYNTHETIC_SECRET_40_HEX}",
"message": "m",
"created_at": "2026-07-23T01:00:00Z",
"kind": "issue",
"number": 637,
}]
ev = timeline.adapt_cp_events(rows)[0]
self.assertEqual(ev.event_type, timeline.UNSAFE_EVENT_TYPE)
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(ev.to_dict()))
def test_cp_legitimate_event_types_preserved(self):
"""Every real producer type in this repo must survive untouched."""
legitimate = [
"allocation", "pr.opened", "lease.renew", "assigned",
"lease_released", "lease_expired", "lease_abandoned",
"lease_adopted", "dependency_edge_state_change",
]
rows = [
{
"event_id": i,
"event_type": name,
"message": "m",
"created_at": "2026-07-23T01:00:00Z",
"kind": "issue",
"number": 637,
}
for i, name in enumerate(legitimate, start=1)
]
events = timeline.adapt_cp_events(rows)
self.assertEqual([e.event_type for e in events], legitimate)
def test_cp_malformed_event_type_shapes_refused(self):
for bad in ("has space", "1leading-digit", "x" * 200, "semi;colon", "new\nline"):
with self.subTest(event_type=bad):
rows = [{
"event_id": 1,
"event_type": bad,
"message": "m",
"created_at": "2026-07-23T01:00:00Z",
"kind": "issue",
"number": 637,
}]
events = timeline.adapt_cp_events(rows)
self.assertEqual(events[0].event_type, timeline.UNSAFE_EVENT_TYPE)
self.assertNotIn(bad, json.dumps(events[0].to_dict()))
# ---- CTH heading event_type ------------------------------------------- #
def test_cth_heading_canary_never_serialized(self):
comments = [{
"id": 11,
"body": _raw_cth_body(SYNTHETIC_SECRET_40_HEX),
"created_at": "2026-07-23T05:00:00Z",
"user": {"login": "jcwalker3"},
}]
events = timeline.adapt_cth_comments(comments, kind="issue", number=637)
self.assertEqual(len(events), 1)
ev = events[0]
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, ev.event_type)
self.assertEqual(ev.event_type, timeline.UNKNOWN_HANDOFF_EVENT_TYPE)
# No message redaction is doing the work here — the message is benign.
self.assertEqual(ev.message, "review")
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(ev.to_dict()))
self.assertTrue(ev.sensitive)
def test_cth_assigned_secret_heading_refused(self):
comments = [{
"id": 12,
"body": _raw_cth_body(f"token={SYNTHETIC_SECRET_40_HEX}"),
"created_at": "2026-07-23T05:00:00Z",
"user": {"login": "jcwalker3"},
}]
ev = timeline.adapt_cth_comments(comments, kind="issue", number=637)[0]
self.assertEqual(ev.event_type, timeline.UNKNOWN_HANDOFF_EVENT_TYPE)
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(ev.to_dict()))
def test_cth_unknown_and_whitespace_headings_normalized(self):
for heading in ("Totally Made Up", "Author Handoff", "author handoff"):
with self.subTest(heading=heading):
comments = [{
"id": 13,
"body": _raw_cth_body(heading),
"created_at": "2026-07-23T05:00:00Z",
"user": {"login": "x"},
}]
events = timeline.adapt_cth_comments(comments, kind="issue", number=637)
self.assertEqual(len(events), 1)
self.assertEqual(events[0].event_type, timeline.UNKNOWN_HANDOFF_EVENT_TYPE)
def test_cth_declared_types_all_preserved(self):
"""Point 5: no legitimate declared type is lost to the new check."""
for cth_type in sorted(CTH_TYPES):
with self.subTest(cth_type=cth_type):
comments = [{
"id": 14,
"body": _raw_cth_body(cth_type),
"created_at": "2026-07-23T05:00:00Z",
"user": {"login": "x"},
}]
ev = timeline.adapt_cth_comments(comments, kind="issue", number=637)[0]
self.assertEqual(ev.event_type, f"handoff:{cth_type}")
self.assertFalse(ev.sensitive)
def test_cth_surrounding_whitespace_still_matches_contract(self):
comments = [{
"id": 15,
"body": _raw_cth_body(" Author Handoff "),
"created_at": "2026-07-23T05:00:00Z",
"user": {"login": "x"},
}]
ev = timeline.adapt_cth_comments(comments, kind="issue", number=637)[0]
self.assertEqual(ev.event_type, "handoff:Author Handoff")
# ---- complete serialized payload, both paths -------------------------- #
def _payload_clean(self, snapshot):
payload = json.dumps(snapshot.to_dict())
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, payload)
from webui import console_redaction
self.assertEqual(console_redaction.scan_for_secrets(snapshot.to_dict()), [])
return snapshot.to_dict()
def test_canary_absent_from_full_payload_via_cp_event_type(self):
import tempfile
with tempfile.TemporaryDirectory() as tmp:
db = os.path.join(tmp, "cp.sqlite3")
_seed_db(db)
_seed_event(
db,
event_type=SYNTHETIC_SECRET_40_HEX,
message="routine deploy",
created_at="2026-07-23T06:00:00Z",
)
snap = timeline.load_timeline(
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
issue=637, db_path=db, comment_source=lambda k, n: [],
)
d = self._payload_clean(snap)
types = [e["event_type"] for e in d["events"]]
self.assertIn(timeline.UNSAFE_EVENT_TYPE, types)
# The legitimate seeded types are still present and unchanged.
self.assertIn("allocation", types)
def test_canary_absent_from_full_payload_via_cth_heading(self):
import tempfile
comments = [{
"id": 21,
"body": _raw_cth_body(SYNTHETIC_SECRET_40_HEX),
"created_at": "2026-07-23T09:00:00Z",
"user": {"login": "jcwalker3"},
}]
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,
)
d = self._payload_clean(snap)
self.assertIn(
timeline.UNKNOWN_HANDOFF_EVENT_TYPE,
[e["event_type"] for e in d["events"]],
)
def test_canary_absent_when_seeded_through_both_paths_at_once(self):
import tempfile
comments = [{
"id": 22,
"body": _raw_cth_body(SYNTHETIC_SECRET_40_HEX),
"created_at": "2026-07-23T09:00:00Z",
"user": {"login": "jcwalker3"},
}]
with tempfile.TemporaryDirectory() as tmp:
db = os.path.join(tmp, "cp.sqlite3")
_seed_db(db)
_seed_event(
db,
event_type=SYNTHETIC_SECRET_40_HEX,
message=f"deploy token={SYNTHETIC_SECRET_40_HEX}",
created_at="2026-07-23T06:00:00Z",
)
snap = timeline.load_timeline(
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
issue=637, db_path=db, comment_source=lambda k, n: comments,
)
self._payload_clean(snap)
class TestSerializedFieldAudit(unittest.TestCase):
"""The remaining externally influenced strings that reach the payload."""
def test_event_key_ids_must_be_plain_identifiers(self):
rows = [
{"event_id": SYNTHETIC_SECRET_40_HEX, "event_type": "allocation",
"message": "m", "created_at": "2026-07-23T01:00:00Z",
"kind": "issue", "number": 637},
{"event_id": 8, "event_type": "allocation", "message": "m",
"created_at": "2026-07-23T01:00:00Z", "kind": "issue", "number": 637},
]
events = timeline.adapt_cp_events(rows)
self.assertEqual([e.event_key for e in events], ["cp:8"])
def test_comment_id_must_be_a_plain_identifier(self):
comments = [{
"id": SYNTHETIC_SECRET_40_HEX,
"body": _raw_cth_body("Author Handoff"),
"created_at": "2026-07-23T05:00:00Z",
"user": {"login": "x"},
}]
self.assertEqual(timeline.adapt_cth_comments(comments, kind="issue", number=637), [])
def test_adapter_refuses_a_scope_it_cannot_express(self):
comments = [{
"id": 1,
"body": _raw_cth_body("Author Handoff"),
"created_at": "2026-07-23T05:00:00Z",
"user": {"login": "x"},
}]
self.assertEqual(
timeline.adapt_cth_comments(comments, kind="issue", number=SYNTHETIC_SECRET_40_HEX),
[],
)
self.assertEqual(
timeline.adapt_cth_comments(comments, kind=SYNTHETIC_SECRET_40_HEX, number=1),
[],
)
def test_source_failure_reason_is_redacted(self):
def boom(kind, number):
raise RuntimeError(f"auth failed with token={SYNTHETIC_SECRET_40_HEX}")
import tempfile
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=boom,
)
d = snap.to_dict()
handoff = [s for s in d["sources"] if s["name"] == "gitea_handoff"][0]
self.assertFalse(handoff["ok"])
self.assertIn("handoff source failed", handoff["reason"])
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(d))
def test_echoed_scope_and_filters_are_guarded(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=SYNTHETIC_SECRET_40_HEX, db_path=db,
comment_source=lambda k, n: [],
).to_dict()
self.assertNotIn(SYNTHETIC_SECRET_40_HEX, json.dumps(d))
# Ordinary scope values are untouched, so the echo stays useful.
self.assertEqual(
d["scope"],
{"remote": "prgs", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools"},
)
self.assertEqual(d["filters"]["issue"], 637)
class TestCthTypeContract(unittest.TestCase):
"""The contract has one authority; the read path consults it."""
def test_declared_types_are_known(self):
for cth_type in CTH_TYPES:
self.assertTrue(is_known_cth_type(cth_type))
def test_undeclared_types_are_not_known(self):
for value in ("", None, "Made Up", SYNTHETIC_SECRET_40_HEX, "author handoff"):
self.assertFalse(is_known_cth_type(value))
def test_parse_reports_contract_membership(self):
known = parse_cth_comment(_raw_cth_body("Author Handoff"))
self.assertTrue(known["cth_type_known"])
self.assertEqual(known["cth_type"], "Author Handoff")
unknown = parse_cth_comment(_raw_cth_body("Not A Real Type"))
# Parsing stays total: the type is still reported, just not endorsed.
self.assertEqual(unknown["cth_type"], "Not A Real Type")
self.assertFalse(unknown["cth_type_known"])
if __name__ == "__main__":
unittest.main()