diff --git a/services/hermes/scm-common/scripts/scm_broker.py b/services/hermes/scm-common/scripts/scm_broker.py index 02bd7e23..75d6d508 100644 --- a/services/hermes/scm-common/scripts/scm_broker.py +++ b/services/hermes/scm-common/scripts/scm_broker.py @@ -23,7 +23,7 @@ from gitea_api import ( read_token, ) from gitea_api_policy import _reject_forbidden, _validate_ref, _validate_repo -from scm_broker_io import spool_response +from scm_broker_io import RejectRedirect, response_status as _status, spool_response from scm_broker_server import AbsoluteHeaderDeadlineMixin, BoundedThreadingHTTPServer BROKER_PORT = 9081 @@ -49,22 +49,9 @@ FEATURE_REF_RE = re.compile( ) -class RejectRedirect(urllib.request.HTTPRedirectHandler): - """Reject every upstream redirect before credentials can be forwarded.""" - def redirect_request(self, req, fp, code, msg, headers, newurl): - raise PolicyError("SCM upstream redirects are not allowed") - - UPSTREAM_OPENER = urllib.request.build_opener(RejectRedirect()) -def _status(response: object) -> int | None: - value = getattr(response, "status", None) - if value is None and hasattr(response, "getcode"): - value = response.getcode() # type: ignore[attr-defined] - return value - - def _read_bounded( stream, maximum: int, diff --git a/services/hermes/scm-common/scripts/scm_broker_io.py b/services/hermes/scm-common/scripts/scm_broker_io.py index 427e8517..96df1fe0 100644 --- a/services/hermes/scm-common/scripts/scm_broker_io.py +++ b/services/hermes/scm-common/scripts/scm_broker_io.py @@ -5,11 +5,27 @@ from __future__ import annotations import tempfile import time +import urllib.request from typing import BinaryIO from gitea_api_policy import PolicyError +class RejectRedirect(urllib.request.HTTPRedirectHandler): + """Reject every upstream redirect before credentials can be forwarded.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + raise PolicyError("SCM upstream redirects are not allowed") + + +def response_status(response: object) -> int | None: + """Read an HTTP status from either urllib response interface.""" + value = getattr(response, "status", None) + if value is None and hasattr(response, "getcode"): + value = response.getcode() # type: ignore[attr-defined] + return value + + def spool_response( response, maximum: int, @@ -20,7 +36,9 @@ def spool_response( deadline_seconds: float, ) -> tuple[BinaryIO, int]: """Spool an upstream response to disk while scanning chunk boundaries.""" - spool = tempfile.SpooledTemporaryFile(max_size=memory_limit, dir="/tmp") + spool = tempfile.SpooledTemporaryFile( # noqa: SIM115 - caller owns returned spool + max_size=memory_limit, dir="/tmp" + ) carry = b"" total = 0 deadline = time.monotonic() + deadline_seconds diff --git a/services/hermes/scripts/node_account_hardening.py b/services/hermes/scripts/node_account_hardening.py index 9a1de699..a545af47 100644 --- a/services/hermes/scripts/node_account_hardening.py +++ b/services/hermes/scripts/node_account_hardening.py @@ -11,6 +11,7 @@ import re import shlex import stat import struct +from contextlib import suppress from pathlib import Path from node_account_audit import audit_membership, audit_privilege_policies @@ -54,7 +55,9 @@ ACL_ENTRY = struct.Struct(" tuple[bytes, os.stat_result]: +def _read_regular( + path: Path, maximum: int = MAX_ACCOUNT_FILE +) -> tuple[bytes, os.stat_result]: snapshot = read_regular(path, maximum) return snapshot.value, path.stat(follow_symlinks=False) @@ -162,7 +165,7 @@ def _reconcile_databases() -> None: records = _records(_read_regular(path)[0], fields, name) if expected[name] not in records: raise HardeningError("dedicated Hermes account validation failed") - except Exception: + except Exception as error: for path, snapshot, updated in reversed(written): current = read_regular(path, MAX_ACCOUNT_FILE) if ( @@ -174,7 +177,7 @@ def _reconcile_databases() -> None: ): raise HardeningError( f"concurrent host account change prevents rollback: {path.name}" - ) + ) from error atomic_write(path, snapshot.value, snapshot) raise @@ -192,7 +195,9 @@ def _key_identity(line: bytes) -> tuple[str, bytes] | None: fields = shlex.split(stripped.decode("ascii"), posix=True) except (UnicodeDecodeError, ValueError) as exc: raise HardeningError("authorized key line is malformed") from exc - indexes = [index for index, field in enumerate(fields) if KEY_TYPE_RE.fullmatch(field)] + indexes = [ + index for index, field in enumerate(fields) if KEY_TYPE_RE.fullmatch(field) + ] if len(indexes) != 1 or indexes[0] + 1 >= len(fields): raise HardeningError("authorized key line has ambiguous key material") key_type = fields[indexes[0]] @@ -226,10 +231,8 @@ def _validated_public_key(path: Path) -> tuple[bytes, tuple[str, bytes]]: def _directory(path: Path, *, mode: int, uid: int, gid: int) -> None: - try: + with suppress(FileExistsError): path.mkdir(mode=mode) - except FileExistsError: - pass metadata = path.lstat() if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): raise HardeningError(f"unsafe account directory: {path.name}") diff --git a/testing/quality_contract.json b/testing/quality_contract.json index a919251f..06289d49 100644 --- a/testing/quality_contract.json +++ b/testing/quality_contract.json @@ -20,6 +20,13 @@ "ci/scripts/supply_chain_report.py", "services/mailu/scripts/mailu_sync.py", "services/mailu/scripts/mailu_sync_listener.py", + "services/gitea/scripts/gitea_branch_protection_check.py", + "services/hermes/scm-common/scripts/gitea_api.py", + "services/hermes/scm-common/scripts/gitea_api_policy.py", + "services/hermes/scm-common/scripts/scm_broker.py", + "services/hermes/scm-common/scripts/scm_broker_client.py", + "services/hermes/scm-common/scripts/scm_broker_io.py", + "services/hermes/scm-common/scripts/scm_broker_server.py", "services/hermes/scripts/execution_pool_client.py", "services/hermes/scripts/execution_pool_coordinator.py", "services/hermes/scripts/execution_pool_project.py", @@ -27,6 +34,11 @@ "services/hermes/scripts/execution_pool_scm.py", "services/hermes/scripts/execution_pool_server.py", "services/hermes/scripts/execution_pool_worker.py", + "services/hermes/scripts/hermes_coordinator.py", + "services/hermes/scripts/node_account_audit.py", + "services/hermes/scripts/node_account_hardening.py", + "services/hermes/scripts/node_account_io.py", + "services/hermes/scripts/stage_runtime_access.py", "testing/__init__.py", "testing/quality_contract.py", "testing/quality_docs.py", @@ -50,6 +62,8 @@ "services/comms/scripts/tests", "services/mailu/scripts/mailu_sync.py", "services/mailu/scripts/mailu_sync_listener.py", + "services/gitea/scripts/gitea_branch_protection_check.py", + "services/hermes/scm-common/scripts", "services/hermes/scripts/execution_pool_client.py", "services/hermes/scripts/execution_pool_coordinator.py", "services/hermes/scripts/execution_pool_project.py", @@ -57,6 +71,11 @@ "services/hermes/scripts/execution_pool_scm.py", "services/hermes/scripts/execution_pool_server.py", "services/hermes/scripts/execution_pool_worker.py", + "services/hermes/scripts/hermes_coordinator.py", + "services/hermes/scripts/node_account_audit.py", + "services/hermes/scripts/node_account_hardening.py", + "services/hermes/scripts/node_account_io.py", + "services/hermes/scripts/stage_runtime_access.py", "testing/tests", "testing" ], @@ -72,6 +91,8 @@ "coverage_sources": [ "ci/scripts", "scripts.render.dashboards_render_atlas", + "services/gitea/scripts", + "services/hermes/scm-common/scripts", "services/hermes/scripts", "services/mailu/scripts", "testing" @@ -139,7 +160,14 @@ "ci/tests/**/*.py", "scripts/tests/**/*.py", "services/*/scripts/tests/**/*.py", + "services/gitea/scripts/gitea_branch_protection_check.py", + "services/hermes/scm-common/scripts/*.py", "services/hermes/scripts/execution_pool_*.py", + "services/hermes/scripts/hermes_coordinator.py", + "services/hermes/scripts/node_account_audit.py", + "services/hermes/scripts/node_account_hardening.py", + "services/hermes/scripts/node_account_io.py", + "services/hermes/scripts/stage_runtime_access.py", "services/mailu/scripts/mailu_sync.py", "services/mailu/scripts/mailu_sync_listener.py" ], @@ -183,13 +211,20 @@ }, "coverage": { "minimum_percent": 95.0, - "tracked_files": [ - "ci/scripts/publish_test_metrics.py", - "ci/scripts/publish_test_metrics_quality.py", - "ci/scripts/semgrep_report.py", - "ci/scripts/supply_chain_report.py", - "services/mailu/scripts/mailu_sync.py", - "services/mailu/scripts/mailu_sync_listener.py", + "minimum_branch_percent": 95.0, + "branch_tracked_files": [ + "services/gitea/scripts/gitea_branch_protection_check.py", + "services/hermes/scm-common/scripts/gitea_api.py", + "services/hermes/scm-common/scripts/gitea_api_policy.py", + "services/hermes/scm-common/scripts/scm_broker.py", + "services/hermes/scm-common/scripts/scm_broker_client.py", + "services/hermes/scm-common/scripts/scm_broker_io.py", + "services/hermes/scm-common/scripts/scm_broker_server.py", + "services/hermes/scripts/hermes_coordinator.py", + "services/hermes/scripts/node_account_audit.py", + "services/hermes/scripts/node_account_hardening.py", + "services/hermes/scripts/node_account_io.py", + "services/hermes/scripts/stage_runtime_access.py", "services/hermes/scripts/execution_pool_client.py", "services/hermes/scripts/execution_pool_coordinator.py", "services/hermes/scripts/execution_pool_project.py", @@ -197,6 +232,34 @@ "services/hermes/scripts/execution_pool_scm.py", "services/hermes/scripts/execution_pool_server.py", "services/hermes/scripts/execution_pool_worker.py", + "testing/quality_coverage.py" + ], + "tracked_files": [ + "ci/scripts/publish_test_metrics.py", + "ci/scripts/publish_test_metrics_quality.py", + "ci/scripts/semgrep_report.py", + "ci/scripts/supply_chain_report.py", + "services/mailu/scripts/mailu_sync.py", + "services/mailu/scripts/mailu_sync_listener.py", + "services/gitea/scripts/gitea_branch_protection_check.py", + "services/hermes/scm-common/scripts/gitea_api.py", + "services/hermes/scm-common/scripts/gitea_api_policy.py", + "services/hermes/scm-common/scripts/scm_broker.py", + "services/hermes/scm-common/scripts/scm_broker_client.py", + "services/hermes/scm-common/scripts/scm_broker_io.py", + "services/hermes/scm-common/scripts/scm_broker_server.py", + "services/hermes/scripts/execution_pool_client.py", + "services/hermes/scripts/execution_pool_coordinator.py", + "services/hermes/scripts/execution_pool_project.py", + "services/hermes/scripts/execution_pool_protocol.py", + "services/hermes/scripts/execution_pool_scm.py", + "services/hermes/scripts/execution_pool_server.py", + "services/hermes/scripts/execution_pool_worker.py", + "services/hermes/scripts/hermes_coordinator.py", + "services/hermes/scripts/node_account_audit.py", + "services/hermes/scripts/node_account_hardening.py", + "services/hermes/scripts/node_account_io.py", + "services/hermes/scripts/stage_runtime_access.py", "testing/quality_contract.py", "testing/quality_docs.py", "testing/quality_hygiene.py", diff --git a/testing/quality_coverage.py b/testing/quality_coverage.py index 78d6649f..98cc65d5 100644 --- a/testing/quality_coverage.py +++ b/testing/quality_coverage.py @@ -3,23 +3,31 @@ from __future__ import annotations import xml.etree.ElementTree as ET +from dataclasses import dataclass from pathlib import Path from typing import Any -def _load_percentages(xml_path: Path, root: Path) -> dict[str, float]: - """Load per-file line-rate percentages from a Cobertura XML report.""" +@dataclass(frozen=True) +class CoverageRates: + """Line and branch percentages reported for one source file.""" + + line: float + branch: float | None + + +def _load_percentages(xml_path: Path, root: Path) -> dict[str, CoverageRates]: + """Load per-file line and branch percentages from a Cobertura report.""" tree = ET.parse(xml_path) xml_root = tree.getroot() source_roots = [ - Path(node.text) - for node in xml_root.findall("./sources/source") - if node.text + Path(node.text) for node in xml_root.findall("./sources/source") if node.text ] - percentages: dict[str, float] = {} + percentages: dict[str, CoverageRates] = {} for class_node in xml_root.findall(".//class"): filename = class_node.attrib.get("filename") line_rate = class_node.attrib.get("line-rate") + branch_rate = class_node.attrib.get("branch-rate") if not filename or line_rate is None: continue normalized = filename.replace("\\", "/") @@ -32,7 +40,10 @@ def _load_percentages(xml_path: Path, root: Path) -> dict[str, float]: if candidate.exists(): key = candidate.relative_to(root).as_posix() break - percentages[key] = float(line_rate) * 100.0 + percentages[key] = CoverageRates( + line=float(line_rate) * 100.0, + branch=float(branch_rate) * 100.0 if branch_rate is not None else None, + ) return percentages @@ -47,17 +58,39 @@ def run_check(contract: dict[str, Any], root: Path, xml_path: Path) -> list[str] percentages = _load_percentages(xml_path, root) minimum = float(contract.get("coverage", {}).get("minimum_percent", 95.0)) + coverage_contract = contract.get("coverage", {}) + branch_minimum = coverage_contract.get("minimum_branch_percent") + if branch_minimum is not None: + branch_minimum = float(branch_minimum) + branch_tracked = { + path.replace("\\", "/") + for path in coverage_contract.get( + "branch_tracked_files", + coverage_contract.get("tracked_files", []) + if branch_minimum is not None + else [], + ) + } issues: list[str] = [] for relative_path in contract.get("coverage", {}).get("tracked_files", []): normalized = relative_path.replace("\\", "/") - percent = percentages.get(normalized) - if percent is None: + rates = percentages.get(normalized) + if rates is None: issues.append(f"coverage missing for tracked file: {relative_path}") continue - if percent + 1e-9 < minimum: + if rates.line + 1e-9 < minimum: issues.append( - f"coverage below {minimum:.1f}%: {relative_path} ({percent:.1f}%)" + f"coverage below {minimum:.1f}%: {relative_path} ({rates.line:.1f}%)" + ) + if branch_minimum is None or normalized not in branch_tracked: + continue + if rates.branch is None: + issues.append(f"branch coverage missing for tracked file: {relative_path}") + elif rates.branch + 1e-9 < branch_minimum: + issues.append( + "branch coverage below " + f"{branch_minimum:.1f}%: {relative_path} ({rates.branch:.1f}%)" ) return issues @@ -77,9 +110,9 @@ def compute_workspace_line_coverage( samples: list[float] = [] for relative_path in contract.get("coverage", {}).get("tracked_files", []): normalized = relative_path.replace("\\", "/") - percent = percentages.get(normalized) - if percent is not None: - samples.append(percent) + rates = percentages.get(normalized) + if rates is not None: + samples.append(rates.line) if not samples: return 0.0 return round(sum(samples) / len(samples), 3) diff --git a/testing/tests/test_hermes_coordinator_cassandra.py b/testing/tests/test_hermes_coordinator_cassandra.py index 083a4b60..646db291 100644 --- a/testing/tests/test_hermes_coordinator_cassandra.py +++ b/testing/tests/test_hermes_coordinator_cassandra.py @@ -123,6 +123,9 @@ def test_cassandra_sync_repairs_origin_through_broker_without_token( tmp_path: Path, monkeypatch ): """Repository sync uses the credential-isolated broker remote.""" + monkeypatch.setenv( + "HERMES_GITEA_TOKEN_FILE", str(tmp_path / "explicitly-unavailable-token") + ) workspace = tmp_path / "cassandra" (workspace / ".git").mkdir(parents=True) monkeypatch.setattr(coordinator, "CASSANDRA_BASE_PATH", workspace) diff --git a/testing/tests/test_hermes_coordinator_coverage.py b/testing/tests/test_hermes_coordinator_coverage.py new file mode 100644 index 00000000..fe4f3168 --- /dev/null +++ b/testing/tests/test_hermes_coordinator_coverage.py @@ -0,0 +1,267 @@ +"""Behavioral branch coverage for Hermes project coordination.""" + +from __future__ import annotations + +import json +import subprocess +import sys +import types +from contextlib import nullcontext +from pathlib import Path + +import pytest + +from testing.tests.test_hermes_coordinator_support import coordinator, routing + + +def test_workspace_falls_back_when_unset_or_invalid(tmp_path: Path, monkeypatch): + monkeypatch.delenv("HERMES_CASSANDRA_ACTIVE_WORKTREE", raising=False) + assert coordinator.cassandra_workspace() == coordinator.CASSANDRA_BASE_PATH + monkeypatch.setenv("HERMES_CASSANDRA_ACTIVE_WORKTREE", str(tmp_path / "missing")) + assert coordinator.cassandra_workspace() == coordinator.CASSANDRA_BASE_PATH + + +def test_task_migration_is_noop_for_base_workspace(): + class Board: + @staticmethod + def connect_closing(**_kwargs): + raise AssertionError("base workspace must not access Kanban") + + coordinator._migrate_open_cassandra_tasks(Board, coordinator.CASSANDRA_BASE_PATH) + + +class _Project: + id = "project-id" + + +def _fake_hermes_modules(monkeypatch, *, existing: bool, active: bool): + events = [] + kb = types.ModuleType("hermes_cli.kanban_db") + kb.board_exists = lambda slug: existing + kb.create_board = lambda *args, **kwargs: events.append( + ("create-board", args, kwargs) + ) + kb.set_current_board = lambda slug: events.append(("current", slug)) + kb.connect_closing = lambda **kwargs: nullcontext(object()) + kb.list_tasks = lambda *_a, **_k: [] + + pdb = types.ModuleType("hermes_cli.projects_db") + pdb.connect_closing = lambda: nullcontext(object()) + pdb.get_project = lambda *_a: _Project() if existing else None + pdb.get_active_id = lambda *_a: "active" if active else None + pdb.create_project = ( + lambda *args, **kwargs: events.append(("create-project", kwargs)) or "new-id" + ) + pdb.set_active = lambda *args: events.append(("active", args[-1])) + pdb.update_project = lambda *args, **kwargs: events.append(("update", kwargs)) + pdb.add_folder = lambda *args, **kwargs: events.append(("folder", args[-1], kwargs)) + + package = types.ModuleType("hermes_cli") + package.kanban_db = kb + package.projects_db = pdb + monkeypatch.setitem(sys.modules, "hermes_cli", package) + monkeypatch.setitem(sys.modules, "hermes_cli.kanban_db", kb) + monkeypatch.setitem(sys.modules, "hermes_cli.projects_db", pdb) + return events + + +@pytest.mark.parametrize( + ("existing", "active", "expected"), + [ + (False, False, {"current", "create-project", "active"}), + (False, True, {"current", "create-project"}), + (True, True, {"update", "folder"}), + ], +) +def test_bootstrap_cassandra_creates_or_updates_state( + tmp_path: Path, monkeypatch, existing, active, expected +): + events = _fake_hermes_modules(monkeypatch, existing=existing, active=active) + base = tmp_path / "projects/cassandra" + worktree = tmp_path / "projects/cassandra-worktree" + worktree.mkdir(parents=True) + (worktree / ".git").write_text("gitdir: elsewhere\n", encoding="utf-8") + monkeypatch.setattr(coordinator, "CASSANDRA_BASE_PATH", base) + monkeypatch.setenv("HERMES_CASSANDRA_ACTIVE_WORKTREE", str(worktree)) + coordinator.bootstrap_cassandra(tmp_path / "hermes") + names = {event[0] for event in events} + assert expected <= names + assert "create-board" in names + + +def test_bootstrap_state_ready_path(tmp_path: Path, monkeypatch): + monkeypatch.setattr(coordinator, "bootstrap_cassandra", lambda _root: None) + assert coordinator.bootstrap_cassandra_state(tmp_path) == {"state": "ready"} + + +def test_corruption_type_imports_pinned_exception(monkeypatch): + class Expected(Exception): + pass + + package = types.ModuleType("hermes_cli") + kanban = types.ModuleType("hermes_cli.kanban_db") + kanban.KanbanDbCorruptError = Expected + package.kanban_db = kanban + monkeypatch.setitem(sys.modules, "hermes_cli", package) + monkeypatch.setitem(sys.modules, "hermes_cli.kanban_db", kanban) + assert coordinator._kanban_corruption_type() is Expected + + +def test_repo_sync_reports_missing_git_and_unmanaged_directory( + tmp_path: Path, monkeypatch +): + monkeypatch.setattr(coordinator.shutil, "which", lambda _name: None) + assert coordinator.sync_cassandra_repo({}) == "git-unavailable" + + workspace = tmp_path / "cassandra" + workspace.mkdir() + (workspace / "user-file").write_text("preserve", encoding="utf-8") + monkeypatch.setattr(coordinator, "CASSANDRA_BASE_PATH", workspace) + monkeypatch.setattr(coordinator.shutil, "which", lambda _name: "/usr/bin/git") + assert coordinator.sync_cassandra_repo({}) == "unmanaged-nonempty-directory" + + +@pytest.mark.parametrize( + ("failure", "expected"), + [ + ("get-url-error", "remote-repair-failed"), + ("repair-exit", "remote-repair-failed-9"), + ("repair-error", "remote-repair-failed"), + ("fetch-exit", "sync-failed-8"), + ("fetch-error", "sync-failed"), + ], +) +def test_existing_repo_sync_reports_each_failure( + tmp_path: Path, monkeypatch, failure, expected +): + workspace = tmp_path / "cassandra" + (workspace / ".git").mkdir(parents=True) + monkeypatch.setattr(coordinator, "CASSANDRA_BASE_PATH", workspace) + monkeypatch.setattr(coordinator.shutil, "which", lambda _name: "/usr/bin/git") + + def run(command, **_kwargs): + action = command[-2:] if len(command) >= 2 else [] + if action == ["get-url", "origin"]: + if failure == "get-url-error": + raise OSError("git unavailable") + return subprocess.CompletedProcess(command, 0, stdout="old\n") + if "set-url" in command: + if failure == "repair-error": + raise subprocess.TimeoutExpired(command, 30) + return subprocess.CompletedProcess( + command, 9 if failure == "repair-exit" else 0 + ) + if failure == "fetch-error": + raise OSError("fetch unavailable") + return subprocess.CompletedProcess(command, 8 if failure == "fetch-exit" else 0) + + monkeypatch.setattr(coordinator.subprocess, "run", run) + assert coordinator.sync_cassandra_repo({}) == expected + + +@pytest.mark.parametrize( + ("exit_code", "expected"), [(0, "ready"), (7, "sync-failed-7")] +) +def test_empty_repo_clones_through_broker( + tmp_path: Path, monkeypatch, exit_code, expected +): + workspace = tmp_path / "cassandra" + monkeypatch.setattr(coordinator, "CASSANDRA_BASE_PATH", workspace) + monkeypatch.setattr(coordinator.shutil, "which", lambda _name: "/usr/bin/git") + seen = [] + + def run(command, **kwargs): + seen.append((command, kwargs["env"])) + return subprocess.CompletedProcess(command, exit_code) + + monkeypatch.setattr(coordinator.subprocess, "run", run) + assert ( + coordinator.sync_cassandra_repo( + {"API_SERVER_KEY": "not-forwarded", "SAFE": "yes"} + ) + == expected + ) + assert coordinator.CASSANDRA_REMOTE in seen[0][0] + assert seen[0][1]["SAFE"] == "yes" + assert "API_SERVER_KEY" not in seen[0][1] + + +def _catalog(provider): + return routing.Catalog(provider, ["model"], True, True, "connected") + + +def test_refresh_once_writes_complete_status(tmp_path: Path, monkeypatch): + monkeypatch.setattr(coordinator, "_read_env", lambda _path: {"SAFE": "yes"}) + monkeypatch.setattr(coordinator, "discover_codex_models", lambda: _catalog("codex")) + monkeypatch.setattr( + coordinator, "discover_claude_models", lambda: _catalog("claude") + ) + monkeypatch.setattr( + coordinator, "configure_routes", lambda *_a: {"coordinator": ["auto"]} + ) + monkeypatch.setattr( + coordinator, "bootstrap_cassandra_state", lambda _root: {"state": "ready"} + ) + monkeypatch.setattr(coordinator, "sync_cassandra_repo", lambda _env: "ready") + monkeypatch.setattr( + coordinator, "cassandra_workspace", lambda: tmp_path / "worktree" + ) + written = [] + monkeypatch.setattr( + coordinator, "_atomic_write", lambda path, value: written.append((path, value)) + ) + status = coordinator.refresh_once(tmp_path) + assert status["projects"]["cassandra"]["state"] == "ready" + assert json.loads(written[0][1])["providers"]["codex"]["connected"] is True + + +def test_main_once_success_and_failure(monkeypatch, capsys): + status = {"providers": {"codex": {"state": "ready"}}} + monkeypatch.setattr(sys, "argv", ["coordinator", "--once"]) + monkeypatch.setattr(coordinator, "refresh_once", lambda _root: status) + assert coordinator.main() == 0 + assert "codex=ready" in capsys.readouterr().out + monkeypatch.setattr( + coordinator, + "refresh_once", + lambda _root: (_ for _ in ()).throw(RuntimeError("failed")), + ) + assert coordinator.main() == 1 + assert "RuntimeError" in capsys.readouterr().out + + +@pytest.mark.parametrize(("interval", "expected"), [(1, 300), (400, 400)]) +def test_main_loop_sleeps_with_floor(monkeypatch, interval, expected): + calls = iter(({"providers": {}}, KeyboardInterrupt())) + + def refresh(_root): + value = next(calls) + if isinstance(value, BaseException): + raise value + return value + + sleeps = [] + monkeypatch.setattr(coordinator, "refresh_once", refresh) + monkeypatch.setattr(coordinator.time, "sleep", sleeps.append) + monkeypatch.setattr( + sys, "argv", ["coordinator", "--loop", "--interval", str(interval)] + ) + with pytest.raises(KeyboardInterrupt): + coordinator.main() + assert sleeps == [expected] + + +def test_main_loop_recovers_from_refresh_failure(monkeypatch): + calls = iter((RuntimeError("temporary"), KeyboardInterrupt())) + + def refresh(_root): + error = next(calls) + raise error + + sleeps = [] + monkeypatch.setattr(coordinator, "refresh_once", refresh) + monkeypatch.setattr(coordinator.time, "sleep", sleeps.append) + monkeypatch.setattr(sys, "argv", ["coordinator", "--loop", "--interval", "300"]) + with pytest.raises(KeyboardInterrupt): + coordinator.main() + assert sleeps == [300] diff --git a/testing/tests/test_hermes_gitea_branch_coverage.py b/testing/tests/test_hermes_gitea_branch_coverage.py new file mode 100644 index 00000000..362e8726 --- /dev/null +++ b/testing/tests/test_hermes_gitea_branch_coverage.py @@ -0,0 +1,183 @@ +"""Behavioral branch coverage for the Gitea protection policy checker.""" + +from __future__ import annotations + +import json +import os +import stat +import sys +from pathlib import Path + +import pytest + +from testing.tests.test_hermes_scm_broker_support import ROOT, _load_path + + +def _module(name: str = "branch_protection_coverage"): + return _load_path( + name, + ROOT / "services/gitea/scripts/gitea_branch_protection_check.py", + ) + + +def _rule(module, name: str = "main", **updates): + value = { + "rule_name": name, + "priority": 1, + "created_at": "2026-01-01T00:00:00Z", + **module._required("bstein"), + } + value.update(updates) + return value + + +def test_read_bounded_accepts_regular_file_and_closes_descriptor( + tmp_path: Path, monkeypatch +): + module = _module("branch_read_regular") + source = tmp_path / "rules.json" + source.write_bytes(b"[]") + closed: list[int] = [] + real_close = os.close + + def close(descriptor: int) -> None: + closed.append(descriptor) + real_close(descriptor) + + monkeypatch.setattr(module.os, "close", close) + assert module._read_bounded(source) == b"[]" + assert len(closed) == 1 + + +@pytest.mark.parametrize("kind", ["directory", "oversized", "short"]) +def test_read_bounded_rejects_unsafe_or_changed_input( + tmp_path: Path, monkeypatch, kind +): + module = _module(f"branch_read_{kind}") + source = tmp_path / "rules" + source.write_bytes(b"[]") + real_fstat = module.os.fstat + real_read = module.os.read + + if kind == "directory": + monkeypatch.setattr( + module.os, + "fstat", + lambda descriptor: os.stat_result( + (stat.S_IFDIR | 0o700, 0, 0, 0, 0, 0, 2, 0, 0, 0) + ), + ) + elif kind == "oversized": + monkeypatch.setattr( + module.os, + "fstat", + lambda descriptor: os.stat_result( + (stat.S_IFREG | 0o600, 0, 0, 0, 0, 0, module.MAX_INPUT + 1, 0, 0, 0) + ), + ) + else: + monkeypatch.setattr(module.os, "fstat", real_fstat) + monkeypatch.setattr(module.os, "read", lambda descriptor, maximum: b"[") + with pytest.raises(module.PolicyError): + module._read_bounded(source) + monkeypatch.setattr(module.os, "read", real_read) + + +@pytest.mark.parametrize( + ("pattern", "branch", "expected"), + [ + ("m?in", "main", True), + ("m[ai]in", "main", True), + ("m[!z]in", "main", True), + ("m]ain", "m]ain", True), + ("m}ain", "m}ain", True), + ("m[a-z]in", "main", True), + ("m{ain,aster}", "main", True), + ("m\\ain", "main", True), + ], +) +def test_glob_grammar_accepts_supported_constructs(pattern, branch, expected): + module = _module("branch_glob_supported") + assert module._matches(pattern, branch) is expected + + +@pytest.mark.parametrize( + "pattern", + [ + "m\\", + "m[", + "m[]", + "m[!]", + "m[[a]", + "m[\\a]", + "m[-a]", + "m[a-]", + "m[a-b-c]", + "m[z-a]", + "m[aa-b]", + "m{ain}", + "m{ain,}", + "m{ain,aster", + ], +) +def test_invalid_globs_are_literal_nonmatches(pattern): + module = _module("branch_glob_invalid") + assert module._matches(pattern, "main") is False + + +@pytest.mark.parametrize("pattern", ["", "x" * 256, "máin"]) +def test_rule_names_are_ascii_and_bounded(pattern): + module = _module("branch_name_bounds") + with pytest.raises(module.PolicyError, match="rule name"): + module._matches(pattern, "main") + + +@pytest.mark.parametrize("value", [None, "x" * 65, "invalid", "2026-01-01T00:00:00"]) +def test_created_at_requires_short_timezone_aware_iso(value): + module = _module("branch_created_bounds") + with pytest.raises(module.PolicyError, match="creation time"): + module._created(value) + + +@pytest.mark.parametrize( + "value", + [b"x" * (1024 * 1024 + 1), b"{}", json.dumps([None]).encode()], +) +def test_evaluate_rejects_invalid_top_level_inputs(value): + module = _module("branch_evaluate_top") + with pytest.raises((module.PolicyError, json.JSONDecodeError)): + module.evaluate(value, "main", "bstein") + + +def test_evaluate_rejects_unknown_branch_and_excess_rules(): + module = _module("branch_evaluate_bounds") + with pytest.raises(module.PolicyError, match="input"): + module.evaluate(b"[]", "release", "bstein") + with pytest.raises(module.PolicyError, match="response"): + module.evaluate(json.dumps([{}] * 101).encode(), "main", "bstein") + + +@pytest.mark.parametrize("priority", [True, 0, 1_000_001, "1"]) +def test_evaluate_rejects_ambiguous_priority_types(priority): + module = _module("branch_priority_bounds") + rule = _rule(module, priority=priority) + with pytest.raises(module.PolicyError, match="priority"): + module.evaluate(json.dumps([rule]).encode(), "main", "bstein") + + +def test_main_reports_success_and_failures(tmp_path: Path, monkeypatch, capsys): + module = _module("branch_main_paths") + source = tmp_path / "rules.json" + source.write_text(json.dumps([_rule(module)]), encoding="utf-8") + monkeypatch.setattr(sys, "argv", ["checker", str(source), "main", "bstein"]) + assert module.main() == 0 + assert capsys.readouterr().out.strip() == "PRESENT" + + source.write_text("not-json", encoding="utf-8") + assert module.main() == 1 + assert "branch protection check failed" in capsys.readouterr().err + + monkeypatch.setattr( + module, "_read_bounded", lambda _path: (_ for _ in ()).throw(OSError()) + ) + assert module.main() == 1 diff --git a/testing/tests/test_hermes_gitea_internal_coverage.py b/testing/tests/test_hermes_gitea_internal_coverage.py new file mode 100644 index 00000000..f811827c --- /dev/null +++ b/testing/tests/test_hermes_gitea_internal_coverage.py @@ -0,0 +1,375 @@ +"""Behavioral coverage for internal safe-Gitea client and policy paths.""" + +from __future__ import annotations + +import json +import subprocess +import sys +import types +import urllib.error +import urllib.parse +from email.message import Message +from pathlib import Path + +import pytest + +from testing.tests.test_hermes_gitea_support import CLIENT_PATH, Response, _load +from testing.tests.test_hermes_scm_broker_support import _load_path + + +def _policy(name: str = "gitea_policy_coverage"): + return _load_path(name, CLIENT_PATH.parent / "gitea_api_policy.py") + + +def test_runtime_token_reader_and_fixed_origin(tmp_path: Path, monkeypatch): + module = _load() + token = tmp_path / "token" + token.write_text(" synthetic-runtime-value \n", encoding="utf-8") + assert module.read_token(token) == "synthetic-runtime-value" + token.write_text(" \n", encoding="utf-8") + with pytest.raises(ValueError, match="empty"): + module.read_token(token) + + monkeypatch.setenv("GITEA_BASE_URL", module.CANONICAL_BASE_URL) + assert module.configured_base_url() == module.CANONICAL_BASE_URL + monkeypatch.setenv("GITEA_BASE_URL", "https://example.invalid") + with pytest.raises(module.PolicyError, match="fixed"): + module.configured_base_url() + + +def test_safe_urlopen_delegates_only_to_redirect_rejecting_opener(monkeypatch): + module = _load() + sentinel = object() + monkeypatch.setattr( + module._SAFE_OPENER, + "open", + lambda request, timeout: (request, timeout, sentinel), + ) + request = object() + assert module._safe_urlopen(request, 7) == (request, 7, sentinel) + + +def test_api_target_and_request_body_edge_paths(monkeypatch): + module = _load() + with pytest.raises(module.PolicyError, match="path exceeds"): + module._split_api_path("/api/v1/" + "x" * 510) + with pytest.raises(module.PolicyError, match="request body"): + module.authorize_request( + "GET", "/api/v1/repos/atlas/cassandra", {"unexpected": True} + ) + with pytest.raises(module.PolicyError, match="query"): + module.authorize_request( + "POST", + "/api/v1/repos/atlas/cassandra/pulls?page=1", + {}, + ) + with pytest.raises(module.PolicyError, match="only read"): + module.authorize_request("TRACE", "/api/v1/repos/atlas/cassandra/pulls", None) + with pytest.raises(module.PolicyError, match="accepts only"): + module.authorize_request("POST", "/api/v1/repos/atlas/cassandra/pulls", []) + + +def test_response_status_fallback_and_nested_fail_closed(): + module = _load() + + class GetCodeResponse(Response): + def __init__(self): + super().__init__({"name": "cassandra"}, status=200) + del self.status + + def getcode(self): + return 200 + + assert ( + module.read( + "/api/v1/repos/atlas/cassandra", + token="synthetic", + opener=lambda *_a, **_k: GetCodeResponse(), + ) + == b'{"name": "cassandra"}' + ) + with pytest.raises(module.PolicyError, match="omitted required"): + module._nested({"base": None}, "base", "ref") + + +@pytest.mark.parametrize("body", [b"not-json", b"[]"]) +def test_create_response_requires_json_object(body: bytes): + module = _load() + with pytest.raises(module.PolicyError, match="invalid pull-request metadata"): + module._require_create_response( + body, + repo="cassandra", + base="main", + head="feature/test", + head_sha="a" * 40, + title="WIP: Test", + body="Evidence", + ) + + +def test_write_body_handles_empty_newline_and_missing_newline(monkeypatch): + module = _load() + + class Buffer: + value = bytearray() + + @classmethod + def write(cls, value): + cls.value.extend(value) + + monkeypatch.setattr(module.sys, "stdout", types.SimpleNamespace(buffer=Buffer)) + module._write_body(b"") + module._write_body(b"one\n") + module._write_body(b"two") + assert bytes(Buffer.value) == b"one\ntwo\n" + + +def test_main_executes_broker_read_and_create_paths(monkeypatch): + module = _load() + outputs: list[bytes] = [] + client = types.ModuleType("scm_broker_client") + client.read = lambda path: json.dumps({"path": path}).encode() + client.create_draft = lambda repo, **data: json.dumps( + {"repo": repo, **data}, sort_keys=True + ).encode() + monkeypatch.setitem(sys.modules, "scm_broker_client", client) + monkeypatch.setattr(module, "_write_body", outputs.append) + + assert module.main(["read", "/api/v1/repos/atlas/cassandra"]) == 0 + assert json.loads(outputs.pop()) == {"path": "/api/v1/repos/atlas/cassandra"} + assert ( + module.main( + [ + "create-draft", + "cassandra", + "--base", + "main", + "--head", + "feature/coverage", + "--head-sha", + "a" * 40, + "--title", + "Coverage repair", + "--body", + "Review the focused tests", + ] + ) + == 0 + ) + assert json.loads(outputs.pop())["repo"] == "cassandra" + + +def test_main_handles_http_and_policy_failures(monkeypatch, capsys): + module = _load() + client = types.ModuleType("scm_broker_client") + + def http_failure(_path): + headers = Message() + raise urllib.error.HTTPError("url", 503, "unavailable", headers, None) + + client.read = http_failure + client.create_draft = lambda *_a, **_k: b"{}" + monkeypatch.setitem(sys.modules, "scm_broker_client", client) + assert module.main(["read", "/api/v1/repos/atlas/cassandra"]) == 1 + assert "HTTP 503" in capsys.readouterr().err + client.read = lambda _path: (_ for _ in ()).throw(module.PolicyError("rejected")) + assert module.main(["read", "/not-allowed"]) == 1 + assert "no credential" in capsys.readouterr().err + + +@pytest.mark.parametrize( + ("function", "value"), + [ + ("_validate_repo", "."), + ("_validate_repo", ".."), + ("_validate_sha", "short"), + ("_validate_pr_number", True), + ("_validate_pr_number", 0), + ("_validate_pr_number", 2_147_483_648), + ], +) +def test_policy_scalar_validators_reject_ambiguous_values(function, value): + module = _policy("gitea_policy_scalar") + with pytest.raises(module.PolicyError): + getattr(module, function)(value) + + +def test_ref_bounds_cover_utf8_controls_git_failure_and_success(monkeypatch): + module = _policy("gitea_policy_refs") + with pytest.raises(module.PolicyError, match="valid UTF-8"): + module._validate_ref_bounds("\ud800", "head") + with pytest.raises(module.PolicyError, match="UTF-8"): + module._validate_ref_bounds("🧪" * 64, "head") + with pytest.raises(module.PolicyError, match="safe same"): + module._validate_ref("bad\nref", "head") + monkeypatch.setattr( + module.subprocess, + "run", + lambda *_a, **_k: subprocess.CompletedProcess([], 1), + ) + with pytest.raises(module.PolicyError, match="same-repository"): + module._validate_ref("invalid-ref", "head") + monkeypatch.setattr( + module.subprocess, + "run", + lambda *_a, **_k: subprocess.CompletedProcess([], 0), + ) + assert module._validate_ref("feature/valid", "head") == "feature/valid" + + +@pytest.mark.parametrize( + ("value", "required", "match"), + [ + (None, False, "must be text"), + (" ", True, "must not be empty"), + ("\ud800", False, "valid UTF-8"), + ("x\x00y", False, "safe request limit"), + ], +) +def test_text_validation_rejects_nontext_empty_invalid_and_controls( + value, required, match +): + module = _policy("gitea_policy_text") + with pytest.raises(module.PolicyError, match=match): + module._validate_text(value, "field", 20, 20, required=required) + + +@pytest.mark.parametrize( + ("key", "expected"), + [ + ("password", True), + ("accountKey", True), + ("registry-key", True), + ("clientEmail", True), + ("clientId", True), + ("accessId", True), + ("connectionString", True), + ("dockerConfigJson", True), + ("release_note", False), + ], +) +def test_sensitive_key_semantics_cover_generic_forms(key, expected): + module = _policy("gitea_policy_keys") + assert module._is_sensitive_key(key) is expected + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("", False), + ('"synthetic"', True), + ("{encoded}", True), + ("Bearer value-12345678", True), + ("ghp_" + "a" * 24, True), + ("https://example.invalid/value", True), + ("name@example.invalid", True), + ("${RUNTIME_VALUE}", True), + ("AbCdEf0123456789", True), + ("reject empty values", False), + ("ordinary", True), + ], +) +def test_assignment_value_shape_distinguishes_prose_from_credentials(value, expected): + module = _policy("gitea_policy_assignment_values") + assert module._looks_sensitive_assignment_value(value) is expected + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("single", False), + ("add regression coverage", True), + ("add regression_coverage", False), + ("ordinary engineering prose", False), + ], +) +def test_prose_classifier_covers_length_punctuation_and_lead_words(value, expected): + module = _policy("gitea_policy_prose") + assert module._looks_like_prose(value) is expected + + +def test_structured_json_walk_covers_lists_nonstring_keys_and_limits(): + module = _policy("gitea_policy_json_walk") + assert module._json_value_has_sensitive_assignment({1: "ignored"}) is False + assert module._json_value_has_sensitive_assignment([{"release": "safe"}]) is False + assert ( + module._json_value_has_sensitive_assignment({"client_secret": "value"}) is True + ) + assert ( + module._json_value_has_sensitive_assignment( + {"outer": {"client_secret": "value"}} + ) + is True + ) + assert ( + module._json_value_has_sensitive_assignment({"type": "service-account"}) is True + ) + with pytest.raises(module.PolicyError, match="scan limit"): + module._json_value_has_sensitive_assignment([], depth=33) + with pytest.raises(module.PolicyError, match="scan limit"): + module._json_value_has_sensitive_assignment([], nodes=[2048]) + + +def test_json_key_decoder_and_embedded_document_scanner_cover_failures(): + module = _policy("gitea_policy_json_decoder") + assert module._decode_json_key("client\\u005fsecret") == ("client_secret", True) + decoded, valid = module._decode_json_key("client\\qsecret") + assert valid is False and "client" in decoded + assert list(module._decoded_json_documents("prose only")) == [] + assert list(module._decoded_json_documents("x {bad y [1, 2]")) == [[1, 2]] + with pytest.raises(module.PolicyError, match="scan limit"): + list(module._decoded_json_documents("{" * 33)) + assert ( + module._has_structured_sensitive_assignment('"client\\qsecret": value') is True + ) + assert ( + module._has_structured_sensitive_assignment("type:\n service-account") is True + ) + + +def test_high_entropy_detector_covers_long_mixed_and_low_entropy_values(): + module = _policy("gitea_policy_entropy") + assert module._has_high_entropy_token("A1_" * 100) is True + assert ( + module._has_high_entropy_token( + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789+/_=" + ) + is True + ) + assert module._has_high_entropy_token("a" * 64) is False + assert module._has_high_entropy_token("short prose") is False + + +@pytest.mark.parametrize( + "query", + [ + "bad", + "a=1&b=2&c=3&d=4", + "page=1&page=2", + "=value", + "unexpected=1", + "page=abc", + "page=01", + "page=10001", + "limit=51", + "state=merged", + ], +) +def test_query_validator_covers_each_rejection_class(query): + module = _policy("gitea_policy_query") + target = urllib.parse.urlsplit("/api/v1/repos/atlas/cassandra/pulls?" + query) + with pytest.raises(module.PolicyError): + module._validate_query(target, {"page", "limit", "state"}) + + +def test_draft_title_rejects_prefix_without_content(): + module = _policy("gitea_policy_empty_draft") + with pytest.raises(module.PolicyError, match="after the draft prefix"): + module._draft_title("WIP:") + + +def test_query_validator_rejects_noncanonical_raw_form(): + module = _policy("gitea_policy_raw_query") + target = urllib.parse.SplitResult("", "", "/api/v1/repos/atlas/cassandra", "%", "") + with pytest.raises(module.PolicyError, match="canonical ASCII"): + module._validate_query(target, set()) diff --git a/testing/tests/test_hermes_node_acl_coverage.py b/testing/tests/test_hermes_node_acl_coverage.py new file mode 100644 index 00000000..572fd0d2 --- /dev/null +++ b/testing/tests/test_hermes_node_acl_coverage.py @@ -0,0 +1,206 @@ +"""Behavioral branch coverage for node-storage ACL hardening.""" + +from __future__ import annotations + +import errno +import os +import pytest + +from testing.tests.test_hermes_node_account_support import _fixture, _load + + +def test_acl_decoder_rejects_version_permissions_and_adds_missing_mask(): + module = _load() + valid = [ + (module.ACL_USER_OBJ, 7, module.ACL_UNDEFINED_ID), + (module.ACL_GROUP_OBJ, 5, module.ACL_UNDEFINED_ID), + (module.ACL_OTHER, 5, module.ACL_UNDEFINED_ID), + ] + wrong_version = module.ACL_HEADER.pack(99) + b"".join( + module.ACL_ENTRY.pack(*entry) for entry in valid + ) + with pytest.raises(module.HardeningError, match="version"): + module._decode_acl(wrong_version, 0o755) + bad_permission = module._encode_acl([*valid, (module.ACL_USER, 8, 1200)]) + with pytest.raises(module.HardeningError, match="permission"): + module._decode_acl(bad_permission, 0o755) + updated = module._decode_acl( + module._acl_with_deny(module._encode_acl(valid), 0o755), 0o755 + ) + assert (module.ACL_MASK, 5, module.ACL_UNDEFINED_ID) in updated + assert (module.ACL_USER, 0, module.ACCOUNT_UID) in updated + + +def test_read_acl_handles_absence_and_propagates_other_errors(monkeypatch, tmp_path): + module = _load() + path = tmp_path / "root" + path.mkdir() + + def missing(*_args, **_kwargs): + raise OSError(errno.ENODATA, "missing") + + monkeypatch.setattr(module.os, "getxattr", missing) + assert module._read_acl(path) == b"" + + def denied(*_args, **_kwargs): + raise OSError(errno.EPERM, "denied") + + monkeypatch.setattr(module.os, "getxattr", denied) + with pytest.raises(OSError): + module._read_acl(path) + + +@pytest.mark.parametrize("value", [b"", b"X", b"A", b"Nextra"]) +def test_acl_backup_validator_rejects_malformed_encodings(value): + module = _load() + with pytest.raises(module.HardeningError, match="backup is malformed"): + module._validate_acl_backup(value) + module._validate_acl_backup(b"N") + valid = module._encode_acl( + [ + (module.ACL_USER_OBJ, 7, module.ACL_UNDEFINED_ID), + (module.ACL_GROUP_OBJ, 5, module.ACL_UNDEFINED_ID), + (module.ACL_OTHER, 5, module.ACL_UNDEFINED_ID), + ] + ) + module._validate_acl_backup(b"A" + valid) + + +def test_acl_backup_reuses_valid_existing_copy(tmp_path, monkeypatch): + module = _load() + host_etc = tmp_path / "etc" + root = host_etc / "hermes-node-boundary" + root.mkdir(parents=True) + backup = root / "k3s.acl" + backup.write_bytes(b"N") + monkeypatch.setattr(module, "HOST_ETC", host_etc) + monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid()) + monkeypatch.setattr(module, "HOST_ROOT_GID", os.getgid()) + module._acl_backup_once(tmp_path / "k3s", b"") + assert backup.read_bytes() == b"N" + + +def test_acl_backup_short_write_cleans_temporary(tmp_path, monkeypatch): + module = _load() + host_etc = tmp_path / "etc" + host_etc.mkdir() + monkeypatch.setattr(module, "HOST_ETC", host_etc) + monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid()) + monkeypatch.setattr(module, "HOST_ROOT_GID", os.getgid()) + monkeypatch.setattr(module.os, "write", lambda _fd, _value: 0) + with pytest.raises(module.HardeningError, match="short sensitive"): + module._acl_backup_once(tmp_path / "k3s", b"") + assert not list((host_etc / "hermes-node-boundary").glob(".*.hermes-*")) + + +def test_acl_backup_tolerates_concurrent_first_writer(tmp_path, monkeypatch): + module = _load() + host_etc = tmp_path / "etc" + host_etc.mkdir() + monkeypatch.setattr(module, "HOST_ETC", host_etc) + monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid()) + monkeypatch.setattr(module, "HOST_ROOT_GID", os.getgid()) + real_link = os.link + + def race(source, target, **kwargs): + real_link(source, target, **kwargs) + raise FileExistsError() + + monkeypatch.setattr(module.os, "link", race) + module._acl_backup_once(tmp_path / "k3s", b"") + assert (host_etc / "hermes-node-boundary/k3s.acl").read_bytes() == b"N" + + +def test_sensitive_root_failure_restores_existing_acl(tmp_path, monkeypatch): + module = _load() + root = tmp_path / "k3s" + root.mkdir() + monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid()) + current = module._encode_acl( + [ + (module.ACL_USER_OBJ, 7, module.ACL_UNDEFINED_ID), + (module.ACL_GROUP_OBJ, 5, module.ACL_UNDEFINED_ID), + (module.ACL_MASK, 5, module.ACL_UNDEFINED_ID), + (module.ACL_OTHER, 5, module.ACL_UNDEFINED_ID), + ] + ) + reads = iter((current, current)) + writes = [] + monkeypatch.setattr(module, "_read_acl", lambda _path: next(reads)) + monkeypatch.setattr(module, "_acl_backup_once", lambda *_a, **_k: None) + monkeypatch.setattr( + module.os, "setxattr", lambda *args, **kwargs: writes.append(args[2]) + ) + with pytest.raises(module.HardeningError, match="validation failed"): + module._deny_sensitive_root(root) + assert writes[-1] == current + + +@pytest.mark.parametrize("remove_errno", [errno.ENODATA, errno.EPERM]) +def test_sensitive_root_failure_removes_new_acl_or_propagates_cleanup_error( + tmp_path, monkeypatch, remove_errno +): + module = _load() + root = tmp_path / "k3s" + root.mkdir() + monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid()) + monkeypatch.setattr(module, "_read_acl", lambda _path: b"") + monkeypatch.setattr(module, "_acl_backup_once", lambda *_a, **_k: None) + monkeypatch.setattr( + module.os, + "setxattr", + lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("set failed")), + ) + monkeypatch.setattr( + module.os, + "removexattr", + lambda *_a, **_k: (_ for _ in ()).throw(OSError(remove_errno, "remove")), + ) + expected = OSError if remove_errno == errno.EPERM else RuntimeError + with pytest.raises(expected): + module._deny_sensitive_root(root) + + +def test_database_validation_failure_rolls_back_all_writes(tmp_path, monkeypatch): + module, originals, _key_value, _other, _public = _fixture(tmp_path, monkeypatch) + real_read = module._read_regular + calls = 0 + + def missing_expected(path, maximum=module.MAX_ACCOUNT_FILE): + nonlocal calls + calls += 1 + if calls == 1: + return originals["passwd"].encode(), path.stat(follow_symlinks=False) + return real_read(path, maximum) + + monkeypatch.setattr(module, "_read_regular", missing_expected) + with pytest.raises(module.HardeningError, match="validation failed"): + module._reconcile_databases() + for name, value in originals.items(): + assert (module.HOST_ETC / name).read_text() == value + + +def test_preexisting_target_key_removal_is_verified(tmp_path, monkeypatch): + module, _originals, key, _other, public = _fixture(tmp_path, monkeypatch) + target = module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys" + target.parent.mkdir(parents=True) + target.write_text(key + "\n", encoding="utf-8") + real_write = module.atomic_write + + def ignore_preinstall_target(path, value, metadata): + if path == target and value == b"": + return + real_write(path, value, metadata) + + monkeypatch.setattr(module, "atomic_write", ignore_preinstall_target) + with pytest.raises(module.HardeningError, match="preexisting Hermes authorization"): + module._move_key(public) + + +def test_existing_target_without_key_does_not_need_precleanup(tmp_path, monkeypatch): + module, _originals, _key_value, other, public = _fixture(tmp_path, monkeypatch) + target = module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys" + target.parent.mkdir(parents=True) + target.write_text(other + "\n", encoding="utf-8") + module._move_key(public) + assert target.read_text().endswith("\n") diff --git a/testing/tests/test_hermes_node_audit_coverage.py b/testing/tests/test_hermes_node_audit_coverage.py new file mode 100644 index 00000000..8c011b5b --- /dev/null +++ b/testing/tests/test_hermes_node_audit_coverage.py @@ -0,0 +1,221 @@ +"""Behavioral branch coverage for dedicated node-account privilege audits.""" + +from __future__ import annotations + +import os +import struct +import sys +from types import SimpleNamespace + +import pytest + +from testing.tests.test_hermes_node_account_support import _load + + +def _audit_module(): + hardening = _load() + return sys.modules[hardening.audit_membership.__module__] + + +def test_member_parser_handles_empty_valid_malformed_and_duplicates(): + module = _audit_module() + assert module._members("", "group") == set() + assert module._members("one,two", "group") == {"one", "two"} + for value in ("one,", " one", "one,one"): + with pytest.raises(module.HardeningError): + module._members(value, "group") + + +def test_membership_audit_accepts_unrelated_records_and_rejects_account(): + module = _audit_module() + module.audit_membership( + "hermes-agent", [["disk", "x", "6", "atlas"]], [["disk", "!", "atlas", ""]] + ) + with pytest.raises(module.HardeningError, match="supplementary"): + module.audit_membership( + "hermes-agent", [["disk", "x", "6", "hermes-agent"]], [] + ) + with pytest.raises(module.HardeningError, match="gshadow"): + module.audit_membership("hermes-agent", [], [["disk", "!", "hermes-agent", ""]]) + + +@pytest.mark.parametrize("kind", ["mode", "symlink"]) +def test_policy_files_reject_unsafe_directories(tmp_path, kind): + module = _audit_module() + host_etc = tmp_path / "etc" + host_etc.mkdir() + root = host_etc / "sudoers.d" + if kind == "mode": + root.mkdir(mode=0o777) + root.chmod(0o777) + else: + target = tmp_path / "target" + target.mkdir() + root.symlink_to(target) + with pytest.raises(module.HardeningError, match="unsafe sudo/polkit"): + module._policy_files(host_etc, tmp_path / "share", os.getuid()) + + +def test_policy_file_count_is_bounded(tmp_path): + module = _audit_module() + root = tmp_path / "etc/sudoers.d" + root.mkdir(parents=True) + for index in range(257): + (root / str(index)).write_text("# safe\n", encoding="utf-8") + with pytest.raises(module.HardeningError, match="too many"): + module._policy_files(tmp_path / "etc", tmp_path / "share", os.getuid()) + + +def _acl(*entries): + return struct.pack(" bytes: + command = old + b" " + new + b" " + ref + b"\n" + framed = f"{len(command) + 4:04x}".encode() + command + return framed + (b"0000" if terminator else b"") + + +def test_receive_pack_rejects_nonascii_many_commands_and_missing_terminator( + monkeypatch, +): + broker = _load("scm_broker") + zero = b"0" * 40 + commit = b"1" * 40 + with pytest.raises(broker.PolicyError, match="canonical ASCII"): + broker._validate_receive_pack( + _packet(zero, commit, b"refs/heads/hermes/\xff"), "sentinel" + ) + command = _packet(zero, commit, b"refs/heads/hermes/fix", terminator=False) + with pytest.raises(broker.PolicyError, match="terminator"): + broker._validate_receive_pack(command, "sentinel") + many = command * (broker.MAX_PUSH_COMMANDS + 1) + b"0000" + with pytest.raises(broker.PolicyError, match="too many"): + broker._validate_receive_pack(many, "sentinel") + + +def test_upstream_request_infers_byte_length_and_requires_stream_length(): + broker = _load("scm_broker") + seen = [] + + def opener(request, timeout): + seen.append((request, timeout)) + return Response(b"result", content_type="application/x-git-upload-pack-result") + + assert ( + broker._upstream_git_request( + "/atlas/cassandra.git/git-upload-pack", + method="POST", + body=b"request", + content_type=None, + expected_type="application/x-git-upload-pack-result", + token="sentinel", + opener=opener, + ) + == b"result" + ) + assert seen[0][0].get_header("Content-length") == "7" + with pytest.raises(broker.PolicyError, match="length is missing"): + broker._upstream_git_request( + "/atlas/cassandra.git/git-upload-pack", + method="POST", + body=io.BytesIO(b"request"), + content_type=None, + expected_type="application/x-git-upload-pack-result", + token="sentinel", + opener=opener, + ) + + +def test_upstream_request_rejects_wrong_type_and_supports_getcode_status(): + broker = _load("scm_broker") + + class CodeOnly(Response): + def __init__(self, body, content_type): + super().__init__(body, content_type=content_type) + del self.status + + def getcode(self): + return 200 + + assert ( + broker._upstream_git_request( + "/atlas/cassandra.git/info/refs?service=git-upload-pack", + method="GET", + body=None, + content_type=None, + expected_type="application/x-git-upload-pack-advertisement", + token="sentinel", + opener=lambda *_a, **_k: CodeOnly( + b"advertisement", "application/x-git-upload-pack-advertisement" + ), + ) + == b"advertisement" + ) + with pytest.raises(broker.PolicyError, match="response type"): + broker._upstream_git_request( + "/atlas/cassandra.git/info/refs?service=git-upload-pack", + method="GET", + body=None, + content_type=None, + expected_type="application/x-git-upload-pack-advertisement", + token="sentinel", + opener=lambda *_a, **_k: Response(b"bad", content_type="text/plain"), + ) diff --git a/testing/tests/test_hermes_scm_server_coverage.py b/testing/tests/test_hermes_scm_server_coverage.py new file mode 100644 index 00000000..7a358640 --- /dev/null +++ b/testing/tests/test_hermes_scm_server_coverage.py @@ -0,0 +1,180 @@ +"""Behavioral branch coverage for the SCM broker HTTP server primitives.""" + +from __future__ import annotations + +import io + +import pytest + +from testing.tests.test_hermes_scm_broker_support import _load + + +class _Connection: + def __init__(self): + self.timeouts = [] + + def settimeout(self, value): + self.timeouts.append(value) + + +class _FlushBuffer(io.BytesIO): + def __init__(self): + super().__init__() + self.flushes = 0 + + def flush(self): + self.flushes += 1 + + +def test_absolute_reader_supports_idle_limit_newline_eof_and_attribute_delegation(): + module = _load("scm_broker_server") + stream = io.BytesIO(b"first\nsecond") + reader = module._AbsoluteDeadlineReader(stream, _Connection()) + assert reader.readline(3) == b"fir" + reader.begin(5) + assert reader.readline() == b"st\n" + assert reader.readline() == b"second" + assert reader.readline() == b"" + assert reader.closed is False + reader.end() + assert reader.readline() == b"" + + +def _handler_type(module): + class StubHandler: + request_version = "HTTP/1.1" + command = "GET" + + def setup(self): + self.connection = _Connection() + self.rfile = io.BytesIO(b"") + self.wfile = _FlushBuffer() + + def parse_request(self): + return self.parse_result + + def send_error(self, code, *_args): + self.errors.append(code) + + def do_GET(self): + self.calls.append("GET") + + return type( + "DeadlineHandler", (module.AbsoluteHeaderDeadlineMixin, StubHandler), {} + ) + + +def _handler(module, raw: bytes, *, parse=True): + handler_type = _handler_type(module) + handler = object.__new__(handler_type) + handler.setup() + handler._header_reader._stream = io.BytesIO(raw) + handler.parse_result = parse + handler.errors = [] + handler.calls = [] + return handler + + +def test_header_mixin_setup_and_known_request_dispatch(): + module = _load("scm_broker_server") + handler = _handler(module, b"GET /healthz HTTP/1.1\r\n") + handler.handle_one_request() + assert handler.calls == ["GET"] + assert handler.wfile.flushes == 1 + assert handler._header_reader._deadline is None + + +@pytest.mark.parametrize( + ("raw", "parse", "command", "expected_error", "closed"), + [ + (b"x" * 65537, True, "GET", 414, False), + (b"", True, "GET", None, True), + (b"GET / HTTP/1.1\r\n", False, "GET", None, False), + (b"TRACE / HTTP/1.1\r\n", True, "TRACE", 501, False), + ], +) +def test_header_mixin_rejects_long_empty_unparsed_and_unknown_requests( + raw, parse, command, expected_error, closed +): + module = _load("scm_broker_server") + handler = _handler(module, raw, parse=parse) + handler.command = command + handler.close_connection = False + handler.handle_one_request() + assert handler.errors == ([] if expected_error is None else [expected_error]) + assert handler.close_connection is closed + + +def test_header_mixin_closes_on_absolute_timeout(monkeypatch): + module = _load("scm_broker_server") + handler = _handler(module, b"GET / HTTP/1.1\r\n") + handler.close_connection = False + monkeypatch.setattr( + handler._header_reader, + "readline", + lambda *_a, **_k: (_ for _ in ()).throw(TimeoutError()), + ) + handler.handle_one_request() + assert handler.close_connection is True + + +class _Slots: + def __init__(self, available=True): + self.available = available + self.releases = 0 + + def acquire(self, *, blocking): + assert blocking is False + return self.available + + def release(self): + self.releases += 1 + + +def _server_shell(module, available=True): + server = object.__new__(module.BoundedThreadingHTTPServer) + server._slots = _Slots(available) + server.shutdowns = [] + server.shutdown_request = server.shutdowns.append + return server + + +def test_bounded_server_delegates_accepted_request(monkeypatch): + module = _load("scm_broker_server") + server = _server_shell(module) + calls = [] + monkeypatch.setattr( + module.ThreadingHTTPServer, + "process_request", + lambda self, request, address: calls.append((request, address)), + ) + server.process_request("socket", ("127.0.0.1", 1)) + assert calls == [("socket", ("127.0.0.1", 1))] + assert server._slots.releases == 0 + + +def test_bounded_server_releases_on_dispatch_failure(monkeypatch): + module = _load("scm_broker_server") + server = _server_shell(module) + monkeypatch.setattr( + module.ThreadingHTTPServer, + "process_request", + lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("dispatch")), + ) + with pytest.raises(RuntimeError, match="dispatch"): + server.process_request("socket", ("127.0.0.1", 1)) + assert server._slots.releases == 1 + assert server.shutdowns == ["socket"] + + +def test_bounded_server_thread_always_releases(monkeypatch): + module = _load("scm_broker_server") + server = _server_shell(module) + monkeypatch.setattr( + module.ThreadingHTTPServer, + "process_request_thread", + lambda *_a, **_k: (_ for _ in ()).throw(RuntimeError("handler")), + ) + with pytest.raises(RuntimeError, match="handler"): + server.process_request_thread("socket", ("127.0.0.1", 1)) + assert server._slots.releases == 1 diff --git a/testing/tests/test_quality_coverage_helpers.py b/testing/tests/test_quality_coverage_helpers.py index 5eaa624e..fc22a8ab 100644 --- a/testing/tests/test_quality_coverage_helpers.py +++ b/testing/tests/test_quality_coverage_helpers.py @@ -12,10 +12,15 @@ def test_compute_workspace_line_coverage_handles_missing_xml(tmp_path: Path) -> """Missing coverage XML should produce a zero workspace coverage score.""" contract = {"coverage": {"tracked_files": ["managed.py"]}} - assert compute_workspace_line_coverage(contract, tmp_path, tmp_path / "missing.xml") == 0.0 + assert ( + compute_workspace_line_coverage(contract, tmp_path, tmp_path / "missing.xml") + == 0.0 + ) -def test_compute_workspace_line_coverage_averages_present_tracked_files(tmp_path: Path) -> None: +def test_compute_workspace_line_coverage_averages_present_tracked_files( + tmp_path: Path, +) -> None: """Workspace coverage should average only tracked files that appear in the report.""" coverage_xml = tmp_path / "coverage.xml" @@ -26,8 +31,8 @@ def test_compute_workspace_line_coverage_averages_present_tracked_files(tmp_path - - + + @@ -41,7 +46,9 @@ def test_compute_workspace_line_coverage_averages_present_tracked_files(tmp_path assert compute_workspace_line_coverage(contract, tmp_path, coverage_xml) == 75.0 -def test_run_check_keeps_relative_names_when_source_roots_do_not_match(tmp_path: Path) -> None: +def test_run_check_keeps_relative_names_when_source_roots_do_not_match( + tmp_path: Path, +) -> None: """Relative filenames should remain relative when no declared source root contains them.""" coverage_xml = tmp_path / "coverage.xml" @@ -74,3 +81,73 @@ def test_run_check_keeps_relative_names_when_source_roots_do_not_match(tmp_path: ) assert issues == ["coverage below 95.0%: relative.py (80.0%)"] + + +def test_run_check_enforces_branch_floor_and_requires_branch_data( + tmp_path: Path, +) -> None: + """Tracked modules must report and meet the configured branch floor.""" + + coverage_xml = tmp_path / "coverage.xml" + coverage_xml.write_text( + textwrap.dedent( + """\ + + + + + + + + + + + """ + ), + encoding="utf-8", + ) + + issues = run_check( + { + "coverage": { + "minimum_percent": 95.0, + "minimum_branch_percent": 95.0, + "tracked_files": ["low.py", "missing.py"], + } + }, + tmp_path, + coverage_xml, + ) + + assert issues == [ + "branch coverage below 95.0%: low.py (90.0%)", + "branch coverage missing for tracked file: missing.py", + ] + + +def test_run_check_limits_branch_floor_to_explicit_branch_tracked_files( + tmp_path: Path, +) -> None: + """A branch rollout can cover selected security modules without hiding lines.""" + + coverage_xml = tmp_path / "coverage.xml" + coverage_xml.write_text( + """ + + + """, + encoding="utf-8", + ) + issues = run_check( + { + "coverage": { + "minimum_percent": 95, + "minimum_branch_percent": 95, + "tracked_files": ["new.py", "legacy.py"], + "branch_tracked_files": ["new.py"], + } + }, + tmp_path, + coverage_xml, + ) + assert issues == []