This repository has been archived on 2026-08-04. You can view files and clone it, but cannot push or open issues or pull requests.
orchestrator/agent-team/tests/test_no_write_token.py
Adam Moussa cb84629c2b feat(agent-team): P3 Phases A/B/E — safety tooling, wiring, docs
Phase A (safety):
- scripts/p3_rollback.sh (+test): restore all privileged P3 surfaces from a
  recorded baseline; --dry-run default, --apply gated. Correct App-uninstall
  (App JWT) model; per-task env-reviewer restore by numeric id; real
  protection post-restore assert (normalize reads argv, fails loud, divergent
  state exits non-zero — regression-tested). KNOWN-LIMITATIONS header flags the
  branch-protection GET->PUT transform + live-validation for the C1 gate.
- scripts/assert_no_write_token.py (+test): box/CI audit that no write token
  (incl. ghu_/ghr_ prefixes + App PEM) lives on the box.
- draft_pr_monitor.py (+test): runaway (>3/15min) + stale (7d) draft-PR sweep,
  wired into tick() and bound a read-only provider in serve.

Phase B (wiring): systemd EnvironmentFile P3 vars + verification; new-draft-PR
lifecycle notice.

Phase E (docs): P3-LIVE-FLIP-PLAN/README/ci-README reflect CI-live-since-6/22 +
box-integration; runbook consolidated (rollback Incident 7 + box-env wiring);
removed a stray duplicate runbook.

Suite: 1360 passed, ruff clean. Branch only; not merged/deployed.
REMAINING HUMAN GATES: C1 /sh-security-review + GPT-4.1 cross-review on the
enabled workflow + rollback script; D box deploy + smoke + merge.
2026-06-23 19:52:04 -04:00

327 lines
11 KiB
Python

"""Unit tests for ``scripts/assert_no_write_token.py`` (P3 no-write-token audit).
The audit asserts the box invariant from ``docs/P3-PHASE0-DESIGN.md`` and
``ci/README.md``: the ``pull-requests: write`` GitHub App credential
(``AGENT_APPLY_APP_ID`` / ``AGENT_APPLY_APP_PRIVATE_KEY``) lives ONLY as an
Actions secret, and the always-on box holds **no standing write token**.
These tests never touch real box paths: they grep temp files and a patched
``os.environ``. ``scripts/`` is not a package, so (matching the ``run-team.py``
idiom) the module is loaded from its file path via :mod:`importlib`.
"""
from __future__ import annotations
import importlib.util
import os
from pathlib import Path
from types import ModuleType
import pytest
_SCRIPT_PATH = (
Path(__file__).resolve().parents[1] / "scripts" / "assert_no_write_token.py"
)
# A sample PEM private key header for each algorithm the design calls out. We
# assert the regex covers RSA / EC / OPENSSH (and a bare PKCS#8 block).
_PEM_BODY = "\nMIIB...redacted...\n"
_RSA_KEY = (
"-----BEGIN RSA PRIVATE KEY-----" + _PEM_BODY + "-----END RSA PRIVATE KEY-----"
)
_EC_KEY = "-----BEGIN EC PRIVATE KEY-----" + _PEM_BODY + "-----END EC PRIVATE KEY-----"
_OPENSSH_KEY = (
"-----BEGIN OPENSSH PRIVATE KEY-----"
+ _PEM_BODY
+ "-----END OPENSSH PRIVATE KEY-----"
)
_PKCS8_KEY = "-----BEGIN PRIVATE KEY-----" + _PEM_BODY + "-----END PRIVATE KEY-----"
def _load_script() -> ModuleType:
"""Load the (non-package) audit script from its file path."""
spec = importlib.util.spec_from_file_location("assert_no_write_token", _SCRIPT_PATH)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
@pytest.fixture(scope="module")
def mod() -> ModuleType:
"""The loaded audit module (loaded once per test module)."""
return _load_script()
# --------------------------------------------------------------------------- #
# scan_environ
# --------------------------------------------------------------------------- #
def test_clean_environ_has_no_findings(mod: ModuleType) -> None:
env = {"PATH": "/usr/bin", "AGENT_TEAM_REPO_OWNER": "Sea-Haven-Industries"}
assert mod.scan_environ(env) == []
def test_forbidden_app_id_env_var_is_flagged(mod: ModuleType) -> None:
findings = mod.scan_environ({"AGENT_APPLY_APP_ID": "123456"})
assert len(findings) == 1
assert "AGENT_APPLY_APP_ID" in findings[0]
def test_forbidden_app_private_key_env_var_is_flagged(mod: ModuleType) -> None:
findings = mod.scan_environ({"AGENT_APPLY_APP_PRIVATE_KEY": _RSA_KEY})
assert len(findings) == 1
assert "AGENT_APPLY_APP_PRIVATE_KEY" in findings[0]
def test_write_token_shaped_name_is_flagged(mod: ModuleType) -> None:
for name in (
"GITHUB_APP_TOKEN",
"GH_PAT_TOKEN",
"GITHUB_WRITE_TOKEN",
"GH_APPLY_KEY",
):
findings = mod.scan_environ({name: "whatever"})
assert findings, f"{name} should be flagged by the write-name heuristic"
def test_write_capable_token_value_is_flagged_regardless_of_name(
mod: ModuleType,
) -> None:
# A benign-looking var name but a write-capable PAT value.
findings = mod.scan_environ({"SOME_VAR": "ghp_" + "a" * 36})
assert len(findings) == 1
assert "SOME_VAR" in findings[0]
findings = mod.scan_environ({"OTHER": "github_pat_" + "b" * 30})
assert len(findings) == 1
findings = mod.scan_environ({"INSTALL": "ghs_" + "c" * 36})
assert len(findings) == 1
# user-to-server (ghu_) and refresh (ghr_) tokens are also write-risk.
findings = mod.scan_environ({"U2S": "ghu_" + "d" * 36})
assert len(findings) == 1
findings = mod.scan_environ({"REFRESH": "ghr_" + "e" * 36})
assert len(findings) == 1
def test_read_only_tokens_are_not_flagged(mod: ModuleType) -> None:
# Known read-only / non-GitHub tokens must not trip the audit.
env = {
"AGENT_TEAM_CI_READ_TOKEN": "ghp_"
+ "r" * 36, # read-only by contract, allowlisted
"AGENT_TEAM_API_TOKEN": "secret-bearer",
"SLACK_APP_TOKEN": "xapp-1-abc",
"SLACK_BOT_TOKEN": "xoxb-abc",
"CLAUDE_CODE_OAUTH_TOKEN": "oauth-abc",
"ANTHROPIC_API_KEY": "sk-ant-abc",
}
assert mod.scan_environ(env) == []
def test_plain_github_token_fallback_is_not_flagged_by_name(mod: ModuleType) -> None:
# The read-only GITHUB_TOKEN runtime fallback should not match the *name*
# heuristic (it carries no GH_*_WRITE/APP/PAT shape).
assert mod.scan_environ({"GITHUB_TOKEN": "a-non-write-shaped-value"}) == []
# --------------------------------------------------------------------------- #
# scan_config
# --------------------------------------------------------------------------- #
def test_scan_config_none_and_empty(mod: ModuleType) -> None:
assert mod.scan_config(None) == []
assert mod.scan_config({}) == []
def test_scan_config_flags_forbidden_key(mod: ModuleType) -> None:
findings = mod.scan_config({"AGENT_APPLY_APP_PRIVATE_KEY": _EC_KEY})
assert len(findings) == 1
assert "AGENT_APPLY_APP_PRIVATE_KEY" in findings[0]
def test_scan_config_flags_write_value(mod: ModuleType) -> None:
findings = mod.scan_config({"token": "ghp_" + "z" * 36})
assert len(findings) == 1
def test_scan_config_ignores_non_string_values(mod: ModuleType) -> None:
assert mod.scan_config({"timeout": 15, "enabled": True}) == []
# --------------------------------------------------------------------------- #
# scan_env_file
# --------------------------------------------------------------------------- #
def test_missing_file_is_not_a_finding(mod: ModuleType, tmp_path: Path) -> None:
assert mod.scan_env_file(tmp_path / "nope.env") == []
def test_clean_file_is_not_a_finding(mod: ModuleType, tmp_path: Path) -> None:
f = tmp_path / "secrev.env"
f.write_text("CLAUDE_CODE_OAUTH_TOKEN=abc\nAGENT_TEAM_CI_READ_TOKEN=def\n")
assert mod.scan_env_file(f) == []
@pytest.mark.parametrize("key", [_RSA_KEY, _EC_KEY, _OPENSSH_KEY, _PKCS8_KEY])
def test_private_key_header_is_flagged_for_each_algo(
mod: ModuleType, tmp_path: Path, key: str
) -> None:
f = tmp_path / "orchestrator.env"
f.write_text("HARMLESS=1\n" + key + "\n")
findings = mod.scan_env_file(f)
assert any("PRIVATE KEY" in fn for fn in findings)
def test_app_credential_name_in_file_is_flagged(
mod: ModuleType, tmp_path: Path
) -> None:
f = tmp_path / "secrev.env"
f.write_text("AGENT_APPLY_APP_ID=987654\n")
findings = mod.scan_env_file(f)
assert any("AGENT_APPLY_APP_ID" in fn for fn in findings)
def test_configured_app_id_value_in_file_is_flagged(
mod: ModuleType, tmp_path: Path
) -> None:
f = tmp_path / "secrev.env"
# The literal id value leaking under any var name.
f.write_text("SOMETHING=installed-as-987654-here\n")
findings = mod.scan_env_file(f, app_id="987654")
assert any("App id value" in fn for fn in findings)
def test_unreadable_present_file_is_a_finding(
mod: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
f = tmp_path / "secrev.env"
f.write_text("ok\n")
def _boom(*_a: object, **_k: object) -> str:
raise OSError("permission denied")
monkeypatch.setattr(Path, "read_text", _boom)
findings = mod.scan_env_file(f)
assert len(findings) == 1
assert "could not read" in findings[0]
# --------------------------------------------------------------------------- #
# audit() orchestration + file resolution
# --------------------------------------------------------------------------- #
def test_audit_clean_box_returns_empty(mod: ModuleType, tmp_path: Path) -> None:
clean = tmp_path / "secrev.env"
clean.write_text("CLAUDE_CODE_OAUTH_TOKEN=abc\n")
findings = mod.audit(
environ={"PATH": "/usr/bin"},
config={"timeout": 15},
env_files=[str(clean)],
)
assert findings == []
def test_audit_aggregates_findings_across_scanners(
mod: ModuleType, tmp_path: Path
) -> None:
leaky = tmp_path / "orchestrator.env"
leaky.write_text(_OPENSSH_KEY + "\n")
findings = mod.audit(
environ={"AGENT_APPLY_APP_ID": "42", "GH_PAT_TOKEN": "x"},
config={"token": "ghp_" + "q" * 36},
env_files=[str(leaky)],
)
# env (2) + config (1) + file (1)
assert len(findings) >= 4
def test_audit_uses_configured_app_id_from_environ(
mod: ModuleType, tmp_path: Path
) -> None:
f = tmp_path / "secrev.env"
f.write_text("LEAK=value-555-leaked\n")
findings = mod.audit(
environ={"AGENT_APPLY_APP_ID": "555"},
env_files=[str(f)],
)
# AGENT_APPLY_APP_ID in environ is itself a finding, and "555" in the file
# is a second.
assert any("AGENT_APPLY_APP_ID" in fn for fn in findings)
assert any("App id value" in fn for fn in findings)
def test_env_file_resolution_prefers_cli(mod: ModuleType, tmp_path: Path) -> None:
a = tmp_path / "a.env"
paths = mod._resolve_env_files([str(a)], {})
assert paths == [a]
def test_env_file_resolution_uses_env_var(mod: ModuleType, tmp_path: Path) -> None:
a = tmp_path / "a.env"
b = tmp_path / "b.env"
env = {"ASSERT_NO_WRITE_TOKEN_ENV_FILES": os.pathsep.join([str(a), str(b)])}
paths = mod._resolve_env_files(None, env)
assert paths == [a, b]
def test_env_file_resolution_defaults_and_expands_home(mod: ModuleType) -> None:
paths = mod._resolve_env_files(None, {})
# Defaults to ~/secrev.env and ~/orchestrator/.env, expanded (no literal ~).
assert len(paths) == 2
assert all("~" not in str(p) for p in paths)
assert paths[0].name == "secrev.env"
# --------------------------------------------------------------------------- #
# main() exit codes + messaging
# --------------------------------------------------------------------------- #
def test_main_clean_exits_zero(
mod: ModuleType,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture,
) -> None:
clean = tmp_path / "secrev.env"
clean.write_text("CLAUDE_CODE_OAUTH_TOKEN=abc\n")
# Patch the live environ so the real process env can't leak into the scan.
monkeypatch.setattr(os, "environ", {"PATH": "/usr/bin"})
rc = mod.main(["--env-file", str(clean)])
assert rc == 0
out = capsys.readouterr().out
assert "OK" in out
assert "no standing write token" in out
def test_main_finding_exits_nonzero(
mod: ModuleType,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture,
) -> None:
leaky = tmp_path / "secrev.env"
leaky.write_text(_RSA_KEY + "\n")
monkeypatch.setattr(os, "environ", {"PATH": "/usr/bin"})
rc = mod.main(["--env-file", str(leaky)])
assert rc == 1
err = capsys.readouterr().err
assert "FAIL" in err
assert "Actions secret" in err
def test_main_flags_write_token_in_live_environ(
mod: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
clean = tmp_path / "secrev.env"
clean.write_text("ok\n")
monkeypatch.setattr(os, "environ", {"AGENT_APPLY_APP_PRIVATE_KEY": _RSA_KEY})
rc = mod.main(["--env-file", str(clean)])
assert rc == 1