Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 81 Pull Requests 13 Actions Packages Projects Releases Wiki Activity
Files
9b8e315b49db41e81843a5779cc6e9707fb50e59
Gitea-Tools/tests/test_webui_sessions_view.py
T
sysadminandClaude Opus 4.8 1ca2b50406 fix(webui): authority-aware ownership + redacted contamination text (#641) 
Addresses the two blockers from the PR #898 review at a81db754.

B1 - degraded ownership inventory was rendered as affirmative absence.
_build_session_rows read the sessions/leases/locks sections without
consulting their status, so a session row emitted lease_ids=() and
worktree_paths=() whether the session genuinely held nothing or the
lease store simply could not be read. The renderer printed both as
"none" and "unbound", contradicting the ownership_authority_complete
invariant documented on InventorySnapshot.

SessionRow now carries lease_authority and worktree_authority. A
worktree binding is correlated through lease work numbers, so it is
unproven when either the leases or the locks section fails to read --
this covers the narrow variant where locks hold real worktree paths but
a degraded leases section leaves work_numbers empty. The renderer emits
"unknown (inventory <status>)" with an authority-unproven badge instead
of none/unbound, the card names the unreadable sections, and an empty
session list from an unreadable sessions section no longer reads as
"no sessions recorded". snapshot_to_dict exports
ownership_authority_complete, ownership_section_status, and per-row
lease_authority / worktree_authority so /api/sessions consumers can
distinguish the two cases.

B2 - contamination payload strings bypassed redaction.
_inspect_contamination copied command_summary, session_id, role and
reason_class out of the marker payload with only str(), while every
inventory-sourced field on the same page arrives through
webui.inventory.scrub(). The write-time redactor
stable_branch_push_guard.redact_command is a narrow denylist that leaves
absolute $HOME paths, -H 'X-Api-Key: <value>', --password <value>, and
PRIVATE_KEY=<value> intact, and this is the first web surface to render
command_summary at all.

Adds webui.inventory.scrub_text(), which collapses $HOME and redacts
credential-shaped tokens and URL userinfo anywhere inside a string rather
than only at its start, and routes the marker payload through it. scrub()
and every existing caller are untouched. The command_summary field is
kept: it is the #630 evidence naming which daemon was killed. The module
docstring claiming absolute paths were already collapsed is corrected.

Also: removes the locks_by_session_hint dead loop and its discard (N1),
adds the missing trailing newline to webui/runtime_views.py (N4), drops
an unused dataclasses.field import, and documents both honesty rules in
docs/webui-local-dev.md.

Tests: tests/test_webui_sessions_view.py grows from 12 to 26 cases,
covering degraded and unavailable ownership sections in both the HTML and
JSON paths, the locks-readable/leases-degraded variant, a guard against
over-correcting clean inventory into "unknown", the previously untested
expired-lease flag, HTML escaping of hostile values in clean and degraded
renders, and each secret class the write-time denylist misses. The
STATUS_UNAVAILABLE import that was present but unused is now exercised.

Validation, from the issue worktree with venv/bin/python (Python 3.14.5,
pytest 9.1.1):

  pytest tests/test_webui_sessions_view.py tests/test_webui_*.py -q
    -> 512 passed, 376 subtests (was 498 / 372; +14 new tests)
  pytest tests/test_issue_854_semantic_container_exclusion.py -q
    -> 13 passed, 8 subtests
  13-file runtime/health/inventory/restart set
    -> 1 failed, 223 passed; the single failure is
       test_runtime_clarity.py::TestRuntimeClarity::
       test_activate_profile_succeeds_when_enabled, the identical test and
       assertion the reviewer recorded on master at 7af40fb5, so it is
       baseline-equivalent and not introduced here.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-25 01:25:28 -04:00

740 lines
27 KiB
Python
Raw Blame History

"""Tests for the Runtime and session view (Phase 1, #641).
Covers clean and stale session rendering, contamination marker surfacing,
worktree binding display, sanctioned recovery links (no pkill), nav/live
status, and the JSON API export.
Also pins the two invariants a reviewer found violated at head a81db754:
degraded ownership sections must render as *unknown* rather than as an
affirmative "none"/"unbound", and contamination payload text must be redacted
at the display boundary rather than trusted from the write-time denylist.
"""
from __future__ import annotations
import os
import sys
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from tests.webui_testclient import TestClient
from webui.app import create_app
from webui.inventory import (
AUTHORITY_CONTROL_PLANE_DB,
AUTHORITY_FILESYSTEM,
InventorySection,
InventorySnapshot,
STATUS_DEGRADED,
STATUS_OK,
STATUS_UNAVAILABLE,
)
from webui.nav import NAV_GROUPS, STUB_PAGES, iter_nav_items
from webui.runtime_health import FileHash, RuntimeSnapshot
from webui.session_loader import (
ContaminationMarker,
SessionRow,
SessionViewSnapshot,
_build_session_rows,
_inspect_contamination,
load_session_view_snapshot,
snapshot_to_dict,
)
from webui.session_views import render_sessions_page
def _runtime(
*,
stale: str | None = None,
profile: str = "prgs-author",
role: str = "author",
) -> RuntimeSnapshot:
return RuntimeSnapshot(
project_id="gitea-tools",
repo_root="/tmp/repo",
remote="prgs",
host="gitea.prgs.cc",
profile_name=profile,
role_kind=role,
config_model="v2-contexts",
profile_mode="dynamic-profile",
profile_source="config file profile",
authenticated_username="jcwalker3",
identity_error=None,
repo_sha="a" * 40,
remote_master_sha="a" * 40,
commits_behind_master=0,
stale_runtime_warning=stale,
shell_health={"shell_use_allowed": True, "consecutive_spawn_failures": 0},
workflow_hashes=(
FileHash(label="SKILL.md", path="skills/llm-project-workflow/SKILL.md", sha256="abc"),
),
schema_hashes=(),
restart_guidance="docs/mcp-namespace-eof-recovery.md",
fetch_error=None,
)
def _inventory(
*,
sessions: tuple[dict, ...] = (),
leases: tuple[dict, ...] = (),
locks: tuple[dict, ...] = (),
worktrees: tuple[dict, ...] = (),
namespaces: tuple[dict, ...] = (),
statuses: dict[str, str] | None = None,
) -> InventorySnapshot:
"""Build a snapshot; ``statuses`` degrades named sections (default all ok)."""
status_of = statuses or {}
def _status(name: str) -> str:
return status_of.get(name, STATUS_OK)
sections = (
InventorySection(
name="sessions",
authority=AUTHORITY_CONTROL_PLANE_DB,
status=_status("sessions"),
items=sessions,
),
InventorySection(
name="leases",
authority=AUTHORITY_CONTROL_PLANE_DB,
status=_status("leases"),
items=leases,
),
InventorySection(
name="locks",
authority=AUTHORITY_FILESYSTEM,
status=_status("locks"),
items=locks,
),
InventorySection(
name="worktrees",
authority=AUTHORITY_FILESYSTEM,
status=_status("worktrees"),
items=worktrees,
),
InventorySection(
name="namespaces",
authority=AUTHORITY_FILESYSTEM,
status=_status("namespaces"),
items=namespaces
or (
{
"profile_name": "prgs-author",
"role": "author",
"mcp_namespace": "gitea-author",
"capability_summary": {
"can_author": True,
"can_review": False,
"can_merge": False,
},
"active": True,
},
),
reason="only the profile serving this web process is observable",
),
)
index = {section.name: section for section in sections}
return InventorySnapshot(
generated_at="2026-07-25T00:00:00+00:00",
sections=sections,
collisions=(),
correlations=(),
scan_ms=1.0,
_section_index=index,
)
def _clean_session() -> dict:
return {
"session_id": "prgs-author-111-clean",
"role": "author",
"profile": "prgs-author",
"namespace": "gitea-author",
"pid": 1111,
"pid_alive": True,
"status": "active",
"started_at": "2026-07-25T00:00:00Z",
"last_heartbeat_at": "2026-07-25T01:00:00Z",
}
def _stale_session() -> dict:
return {
"session_id": "prgs-author-222-stale",
"role": "author",
"profile": "prgs-author",
"namespace": "gitea-author",
"pid": 2222,
"pid_alive": False,
"status": "active",
"started_at": "2026-07-24T00:00:00Z",
"last_heartbeat_at": "2026-07-24T01:00:00Z",
}
class TestBuildSessionRows(unittest.TestCase):
def test_clean_session_has_no_stale_or_contamination_flags(self):
inventory = _inventory(
sessions=(_clean_session(),),
leases=(
{
"lease_id": "lease-clean",
"session_id": "prgs-author-111-clean",
"status": "active",
"expired": False,
"work_kind": "issue",
"work_number": 641,
},
),
locks=(
{
"issue_number": 641,
"branch_name": "feat/issue-641-runtime-session-view",
"worktree_path": "~/Development/Gitea-Tools/branches/feat-issue-641",
"live": True,
},
),
)
rows = _build_session_rows(inventory, contamination=())
self.assertEqual(len(rows), 1)
row = rows[0]
self.assertEqual(row.session_id, "prgs-author-111-clean")
self.assertEqual(row.role, "author")
self.assertEqual(row.namespace, "gitea-author")
self.assertEqual(row.pid_alive, True)
self.assertEqual(row.lease_ids, ("lease-clean",))
self.assertEqual(row.work_refs, ("issue#641",))
self.assertTrue(row.worktree_paths)
self.assertEqual(row.stale_flags, ())
self.assertEqual(row.contamination_flags, ())
def test_stale_session_flags_dead_pid(self):
inventory = _inventory(sessions=(_stale_session(),))
rows = _build_session_rows(inventory, contamination=())
self.assertEqual(rows[0].stale_flags, ("pid-dead",))
def test_contamination_marker_binds_to_session(self):
inventory = _inventory(sessions=(_clean_session(),))
marker = ContaminationMarker(
kind="runtime_recovery_contamination",
on_disk=True,
has_payload=True,
summary="manual daemon kill",
reason_class="manual_daemon_kill",
session_id="prgs-author-111-clean",
role="author",
command_summary="pkill -f mcp_server.py",
cleared=False,
)
rows = _build_session_rows(inventory, contamination=(marker,))
self.assertIn("runtime_recovery_contamination", rows[0].contamination_flags)
def test_process_wide_contamination_surfaces_on_all_sessions(self):
inventory = _inventory(sessions=(_clean_session(), _stale_session()))
marker = ContaminationMarker(
kind="stable_branch_contamination",
on_disk=True,
has_payload=True,
summary="direct master push attempt",
reason_class="stable_branch_push",
session_id=None,
cleared=False,
)
rows = _build_session_rows(inventory, contamination=(marker,))
self.assertEqual(len(rows), 2)
for row in rows:
self.assertTrue(
any("stable_branch_contamination" in f for f in row.contamination_flags)
)
class TestRenderSessionsPage(unittest.TestCase):
def _snapshot(
self,
*,
sessions: tuple[dict, ...],
contamination: tuple[ContaminationMarker, ...] = (),
stale_runtime: str | None = None,
) -> SessionViewSnapshot:
inventory = _inventory(
sessions=sessions,
leases=(
{
"lease_id": "lease-1",
"session_id": sessions[0]["session_id"] if sessions else "",
"status": "active",
"expired": False,
"work_kind": "issue",
"work_number": 641,
},
)
if sessions
else (),
locks=(
{
"issue_number": 641,
"worktree_path": "branches/feat-issue-641",
},
)
if sessions
else (),
worktrees=(
{
"rel_path": "branches/feat-issue-641",
"branch": "feat/issue-641-runtime-session-view",
"classification": "active_issue_work",
"registered_worktree": True,
"dirty": False,
},
),
)
rows = _build_session_rows(inventory, contamination)
return SessionViewSnapshot(
runtime=_runtime(stale=stale_runtime),
inventory=inventory,
sessions=rows,
contamination_markers=contamination,
)
def test_clean_session_render(self):
html = render_sessions_page(self._snapshot(sessions=(_clean_session(),)))
self.assertIn("Runtime and sessions", html)
self.assertIn("prgs-author-111-clean", html)
self.assertIn("gitea-author", html)
self.assertIn("branches/feat-issue-641", html)
self.assertIn("Sanctioned recovery", html)
self.assertIn("docs/mcp-namespace-eof-recovery.md", html)
# Recovery section must name reconnect and forbid manual kill.
recovery_idx = html.lower().find("sanctioned recovery")
self.assertGreaterEqual(recovery_idx, 0)
recovery = html[recovery_idx:].lower()
self.assertIn("reconnect", recovery)
self.assertIn("contamination", recovery)
self.assertIn("not recovery", recovery)
self.assertNotIn("run pkill", recovery)
self.assertNotIn("killall", recovery)
def test_stale_session_render(self):
html = render_sessions_page(self._snapshot(sessions=(_stale_session(),)))
self.assertIn("prgs-author-222-stale", html)
self.assertIn("pid-dead", html)
self.assertIn("badge-stale", html)
def test_contamination_render_is_not_silent(self):
marker = ContaminationMarker(
kind="runtime_recovery_contamination",
on_disk=True,
has_payload=True,
summary="manual kill",
reason_class="manual_daemon_kill",
session_id="prgs-author-111-clean",
command_summary="pkill -f mcp_server.py",
cleared=False,
)
html = render_sessions_page(
self._snapshot(sessions=(_clean_session(),), contamination=(marker,))
)
self.assertIn("Contamination markers", html)
self.assertIn("runtime_recovery_contamination", html)
self.assertIn("ACTIVE", html)
self.assertIn("badge-blocked", html)
def test_stale_runtime_banner(self):
html = render_sessions_page(
self._snapshot(
sessions=(_clean_session(),),
stale_runtime="server behind master by 3 commits",
)
)
self.assertIn("Stale runtime", html)
self.assertIn("server behind master", html)
class TestSessionLoaderComposition(unittest.TestCase):
def test_load_with_injected_sources(self):
inventory = _inventory(sessions=(_clean_session(), _stale_session()))
snap = load_session_view_snapshot(
load_runtime=lambda: _runtime(),
load_inventory=lambda: inventory,
inspect_contamination=lambda **_k: {
"on_disk": False,
"has_payload": False,
"summary": "absent",
},
load_contamination_payload=lambda **_k: None,
)
self.assertEqual(len(snap.sessions), 2)
self.assertEqual(snap.stale_session_count, 1)
self.assertEqual(snap.contaminated_session_count, 0)
data = snapshot_to_dict(snap)
self.assertEqual(data["view"], "runtime-sessions")
self.assertEqual(data["issue"], 641)
self.assertTrue(data["read_only"])
self.assertEqual(data["session_counts"]["total"], 2)
self.assertEqual(data["session_counts"]["stale"], 1)
self.assertIn("recovery_docs", data)
class TestSessionsRoutes(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
inventory = _inventory(
sessions=(_clean_session(), _stale_session()),
leases=(
{
"lease_id": "lease-x",
"session_id": "prgs-author-111-clean",
"status": "active",
"expired": False,
"work_kind": "issue",
"work_number": 641,
},
),
locks=(
{
"issue_number": 641,
"worktree_path": "branches/feat-issue-641",
},
),
worktrees=(
{
"rel_path": "branches/feat-issue-641",
"branch": "feat/issue-641-runtime-session-view",
"classification": "active_issue_work",
"registered_worktree": True,
"dirty": False,
},
),
)
rows = _build_session_rows(inventory, contamination=())
self.snapshot = SessionViewSnapshot(
runtime=_runtime(stale="stale for test"),
inventory=inventory,
sessions=rows,
contamination_markers=(),
)
self._patch = mock.patch(
"webui.app.load_session_view_snapshot",
return_value=self.snapshot,
)
self._patch.start()
def tearDown(self):
self._patch.stop()
def test_sessions_page_live(self):
response = self.client.get("/sessions")
self.assertEqual(response.status_code, 200)
self.assertIn("Runtime and sessions", response.text)
self.assertIn("prgs-author-111-clean", response.text)
self.assertIn("prgs-author-222-stale", response.text)
self.assertIn("pid-dead", response.text)
self.assertIn("Sanctioned recovery", response.text)
self.assertNotIn("Phase 1 shell placeholder", response.text)
self.assertNotIn("child issue of #425", response.text.lower())
def test_api_sessions_json(self):
for path in ("/api/sessions", "/api/v1/sessions"):
response = self.client.get(path)
self.assertEqual(response.status_code, 200, path)
data = response.json()
self.assertEqual(data["view"], "runtime-sessions")
self.assertEqual(data["session_counts"]["total"], 2)
self.assertEqual(data["session_counts"]["stale"], 1)
self.assertTrue(data["read_only"])
self.assertEqual(data["mutations"], [])
def test_nav_marks_sessions_live(self):
sessions_items = [
item for item in iter_nav_items() if item.href == "/sessions"
]
self.assertEqual(len(sessions_items), 1)
self.assertEqual(sessions_items[0].status, "live")
self.assertNotIn("/sessions", STUB_PAGES)
# Home page should not mark Sessions as stub.
home = self.client.get("/")
self.assertEqual(home.status_code, 200)
self.assertIn('href="/sessions"', home.text)
# Stub marker only appears next to remaining stub destinations.
self.assertNotIn(
'href="/sessions">Sessions</a> <span class="muted">(stub)</span>',
home.text,
)
class TestDegradedOwnershipAuthority(unittest.TestCase):
"""B1: a section that could not be read must never render as absence."""
def _snapshot(self, inventory: InventorySnapshot) -> SessionViewSnapshot:
return SessionViewSnapshot(
runtime=_runtime(),
inventory=inventory,
sessions=_build_session_rows(inventory, contamination=()),
contamination_markers=(),
)
def test_unavailable_leases_mark_row_authority_unproven(self):
inventory = _inventory(
sessions=(_clean_session(),),
statuses={"leases": STATUS_UNAVAILABLE},
)
row = _build_session_rows(inventory, contamination=())[0]
self.assertEqual(row.lease_ids, ())
self.assertEqual(row.lease_authority, STATUS_UNAVAILABLE)
# Worktree binding is correlated through lease work numbers, so it
# inherits the unreadable lease section.
self.assertEqual(row.worktree_authority, STATUS_UNAVAILABLE)
self.assertFalse(row.ownership_authority_complete)
def test_readable_locks_are_not_reported_unbound_when_leases_degrade(self):
# The narrow variant: locks hold a real worktree_path and read cleanly,
# but the lease section that supplies the correlating work number does
# not. The row must say unknown, not "unbound".
inventory = _inventory(
sessions=(_clean_session(),),
locks=(
{
"issue_number": 641,
"worktree_path": "branches/feat-issue-641",
},
),
statuses={"leases": STATUS_DEGRADED},
)
row = _build_session_rows(inventory, contamination=())[0]
self.assertEqual(row.worktree_paths, ())
self.assertEqual(row.worktree_authority, STATUS_DEGRADED)
self.assertFalse(row.ownership_authority_complete)
def test_degraded_render_says_unknown_not_none_or_unbound(self):
inventory = _inventory(
sessions=(_clean_session(),),
statuses={"leases": STATUS_UNAVAILABLE, "locks": STATUS_UNAVAILABLE},
)
html = render_sessions_page(self._snapshot(inventory))
self.assertIn("unknown (inventory unavailable)", html)
self.assertIn("authority unproven", html)
self.assertIn("Ownership authority incomplete", html)
# The affirmative-absence strings must be gone from the row entirely.
self.assertNotIn(">none<", html)
self.assertNotIn(">unbound<", html)
def test_clean_inventory_still_renders_affirmative_absence(self):
# Guards against over-correcting B1 into "everything is unknown".
inventory = _inventory(sessions=(_clean_session(),))
html = render_sessions_page(self._snapshot(inventory))
self.assertIn(">none<", html)
self.assertIn(">unbound<", html)
# The column legend mentions "unknown (inventory …)" as static copy, so
# assert on the per-row marker and the concrete statuses instead.
self.assertNotIn("authority unproven", html)
self.assertNotIn("unknown (inventory unavailable)", html)
self.assertNotIn("unknown (inventory degraded)", html)
self.assertNotIn("Ownership authority incomplete", html)
def test_json_export_carries_snapshot_and_per_row_authority(self):
inventory = _inventory(
sessions=(_clean_session(),),
statuses={"locks": STATUS_UNAVAILABLE},
)
data = snapshot_to_dict(self._snapshot(inventory))
self.assertFalse(data["ownership_authority_complete"])
self.assertEqual(
data["ownership_section_status"]["locks"], STATUS_UNAVAILABLE
)
self.assertEqual(data["ownership_section_status"]["leases"], STATUS_OK)
self.assertIn("unknown, not unowned", data["ownership_note"])
row = data["sessions"][0]
self.assertTrue(row["lease_authority_complete"])
self.assertFalse(row["worktree_authority_complete"])
self.assertEqual(row["worktree_authority"], STATUS_UNAVAILABLE)
self.assertFalse(row["ownership_authority_complete"])
self.assertIn("unknown, not unowned", row["ownership_note"])
def test_json_export_is_affirmative_when_every_source_reads(self):
inventory = _inventory(sessions=(_clean_session(),))
data = snapshot_to_dict(self._snapshot(inventory))
self.assertTrue(data["ownership_authority_complete"])
self.assertTrue(data["sessions"][0]["ownership_authority_complete"])
def test_missing_session_list_is_not_reported_as_no_sessions(self):
inventory = _inventory(statuses={"sessions": STATUS_UNAVAILABLE})
html = render_sessions_page(self._snapshot(inventory))
self.assertIn("could not be read", html)
self.assertIn("not evidence that no sessions exist", html)
def test_expired_lease_flags_row_as_stale(self):
inventory = _inventory(
sessions=(_clean_session(),),
leases=(
{
"lease_id": "lease-expired-1",
"session_id": "prgs-author-111-clean",
"status": "active",
"expired": True,
"work_kind": "issue",
"work_number": 641,
},
),
)
row = _build_session_rows(inventory, contamination=())[0]
self.assertIn("lease-expired", row.stale_flags)
self.assertIn("active-lease-past-expiry", row.stale_flags)
class TestContaminationRedaction(unittest.TestCase):
"""B2: marker payload text is redacted at the display boundary."""
def _marker(self, payload: dict) -> ContaminationMarker:
return _inspect_contamination(
"runtime_recovery_contamination",
remote="prgs",
inspect=lambda **_k: {
"on_disk": True,
"has_payload": True,
"summary": "",
},
load=lambda **_k: payload,
)
def test_home_paths_are_collapsed(self):
home = os.path.expanduser("~")
marker = self._marker(
{"command_summary": f"pkill -f {home}/Development/Gitea-Tools/x.py"}
)
self.assertNotIn(home, marker.command_summary)
self.assertIn("~/Development/Gitea-Tools/x.py", marker.command_summary)
def test_secrets_missed_by_the_write_time_denylist_are_redacted(self):
# Each of these was verified in review to survive
# stable_branch_push_guard.redact_command untouched.
cases = (
("curl -H 'X-Api-Key: SUPERSECRET123' https://example.invalid", "SUPERSECRET123"),
("cmd --password hunter2 origin master", "hunter2"),
("PRIVATE_KEY=abc123 python deploy.py", "abc123"),
("fetch https://user:[email protected]/x.git", "user:pw"),
)
for raw, secret in cases:
with self.subTest(raw=raw):
marker = self._marker({"command_summary": raw})
self.assertNotIn(secret, marker.command_summary)
self.assertIn("[redacted]", marker.command_summary)
def test_command_summary_is_redacted_not_removed(self):
# It is legitimate #630 evidence: the operator must still see which
# daemon was killed.
marker = self._marker(
{
"command_summary": "pkill -f gitea_mcp_server.py",
"reason_class": "manual_daemon_kill",
"session_id": "prgs-author-111-clean",
"role": "author",
}
)
self.assertIn("pkill -f gitea_mcp_server.py", marker.command_summary)
self.assertEqual(marker.reason_class, "manual_daemon_kill")
self.assertEqual(marker.session_id, "prgs-author-111-clean")
self.assertEqual(marker.role, "author")
def test_rendered_page_exposes_no_home_path_from_a_marker(self):
home = os.path.expanduser("~")
marker = self._marker(
{
"command_summary": f"pkill -f {home}/Development/Gitea-Tools/x.py",
"reason_class": "manual_daemon_kill",
}
)
inventory = _inventory(sessions=(_clean_session(),))
html = render_sessions_page(
SessionViewSnapshot(
runtime=_runtime(),
inventory=inventory,
sessions=_build_session_rows(inventory, (marker,)),
contamination_markers=(marker,),
)
)
self.assertIn("Contamination markers", html)
self.assertNotIn(home, html)
class TestSessionsPageEscaping(unittest.TestCase):
"""Hostile values from every rendered source stay inert (N2)."""
HOSTILE = '<script>alert("xss")</script>'
def test_hostile_session_and_marker_values_are_escaped(self):
session = dict(_clean_session())
session["session_id"] = f"sid-{self.HOSTILE}"
session["role"] = self.HOSTILE
session["profile"] = self.HOSTILE
session["namespace"] = self.HOSTILE
session["status"] = self.HOSTILE
inventory = _inventory(
sessions=(session,),
leases=(
{
"lease_id": self.HOSTILE,
"session_id": session["session_id"],
"status": "active",
"expired": False,
"work_kind": self.HOSTILE,
"work_number": 641,
},
),
locks=(
{
"issue_number": 641,
"worktree_path": self.HOSTILE,
},
),
)
marker = ContaminationMarker(
kind="runtime_recovery_contamination",
on_disk=True,
has_payload=True,
summary=self.HOSTILE,
reason_class=self.HOSTILE,
session_id=session["session_id"],
role=self.HOSTILE,
command_summary=self.HOSTILE,
cleared=False,
)
html = render_sessions_page(
SessionViewSnapshot(
runtime=_runtime(),
inventory=inventory,
sessions=_build_session_rows(inventory, (marker,)),
contamination_markers=(marker,),
)
)
self.assertNotIn("<script>", html)
self.assertNotIn('alert("xss")', html)
self.assertIn("&lt;script&gt;", html)
def test_hostile_values_in_a_degraded_render_are_escaped(self):
inventory = _inventory(
sessions=(dict(_clean_session(), session_id=f"sid-{self.HOSTILE}"),),
statuses={"leases": STATUS_UNAVAILABLE, "locks": STATUS_DEGRADED},
)
html = render_sessions_page(
SessionViewSnapshot(
runtime=_runtime(),
inventory=inventory,
sessions=_build_session_rows(inventory, contamination=()),
contamination_markers=(),
)
)
self.assertNotIn("<script>", html)
self.assertIn("&lt;script&gt;", html)
self.assertIn("unknown (inventory", html)
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink