Adam Moussa
c4bea7270b
GPT-4.1 cross-family review (APPROVE, no critical/high) raised two MEDIUMs on the rollback tooling; addressed both: - require_keys: each restore_* asserts its required baseline keys up front and refuses a PARTIAL (silently-weaker) restore. environment accepts ids OR logins (equivalent); a missing protection.full now REFUSES the enforce_admins-only degrade unless P3_ROLLBACK_ALLOW_PARTIAL=1 is set (loud DEGRADED warning). - out-of-band ACK: an --apply that needs a MANUAL App neutralise (no APP JWT, or action=out-of-band) refuses unless P3_ROLLBACK_OOB_ACK=1 — so the App is never left un-neutralised without a conscious operator sign-off; with the ack the other surfaces still restore. Documented both env vars in usage. +3 tests (required-key refuse, partial-protection ack, oob ack). Suite: 1362 passed, ruff clean.
846 lines
30 KiB
Python
846 lines
30 KiB
Python
"""Tests for ``scripts/p3_rollback.sh`` — the P3 privileged-surface rollback.
|
|
|
|
The script restores EVERY privileged P3 surface (the apply/verify workflow flip,
|
|
the ``agent-apply`` environment, the GitHub App perms/installation, and branch
|
|
protection) from a recorded baseline, and asserts post-restore == baseline. The
|
|
apply/verify workflow is ALREADY LIVE (flipped + provisioned 2026-06-22), so the
|
|
rollback targets the LIVE state.
|
|
|
|
These tests exercise the script with NO real ``gh``/``git`` calls:
|
|
|
|
* The ``--dry-run`` default must print a PLAN and perform NO mutations. We assert
|
|
the plan output covers every surface (every destructive call is described but
|
|
not executed).
|
|
* Argument parsing: a missing surface, an unknown flag, and a missing value each
|
|
fail closed (exit 2).
|
|
* Fail-closed posture: a baseline whose ``include_administrators`` is not ``true``
|
|
is refused; a missing baseline file is refused.
|
|
* ``--apply`` is verified against a PATH-shimmed ``gh``/``git`` that only RECORDS
|
|
its argv into a log file (never touches a network or a repo), so we can assert
|
|
the exact destructive calls the script would make — with zero real side effects.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "p3_rollback.sh"
|
|
|
|
|
|
def test_script_exists_and_is_executable() -> None:
|
|
assert _SCRIPT.is_file(), f"missing rollback script: {_SCRIPT}"
|
|
mode = _SCRIPT.stat().st_mode
|
|
assert mode & stat.S_IXUSR, "p3_rollback.sh must be executable"
|
|
|
|
|
|
def test_script_has_bash_shebang() -> None:
|
|
first = _SCRIPT.read_text(encoding="utf-8").splitlines()[0]
|
|
assert first.startswith("#!") and "bash" in first
|
|
|
|
|
|
def test_script_passes_bash_syntax_check() -> None:
|
|
res = subprocess.run(["bash", "-n", str(_SCRIPT)], capture_output=True, text=True)
|
|
assert res.returncode == 0, res.stderr
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Fixtures: a recorded baseline + a PATH shim for gh/git
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
_BASELINE = {
|
|
"repo": "Sea-Haven-Industries/orchestrator",
|
|
"default_branch": "main",
|
|
"workflow_path": ".github/workflows/agent-team-apply-verify.yml",
|
|
"workflow_baseline_sha": "0123abc",
|
|
"environment": {
|
|
"name": "agent-apply",
|
|
# Reviewers recorded as NUMERIC user ids (restore is exact + assertable).
|
|
"required_reviewer_ids": [1234567],
|
|
"required_reviewers": ["amoussa1229"],
|
|
"deployment_branch_policy": "protected",
|
|
},
|
|
"app": {
|
|
"slug": "agent-apply",
|
|
"installation_id": 424242,
|
|
# The only programmatic neutralise is uninstall (App JWT). There is no
|
|
# permission-reduction REST endpoint.
|
|
"action": "uninstall",
|
|
},
|
|
"protection": {
|
|
"branch": "main",
|
|
"include_administrators": True,
|
|
# The FULL protection payload recorded pre-flip (the exact body restored).
|
|
"full": {
|
|
"enforce_admins": {"enabled": True},
|
|
"required_status_checks": {
|
|
"strict": True,
|
|
"contexts": ["guard", "build-test"],
|
|
},
|
|
"required_pull_request_reviews": {
|
|
"required_approving_review_count": 1,
|
|
"dismiss_stale_reviews": True,
|
|
"require_code_owner_reviews": False,
|
|
},
|
|
"required_linear_history": {"enabled": True},
|
|
"allow_force_pushes": {"enabled": False},
|
|
"allow_deletions": {"enabled": False},
|
|
},
|
|
"required_status_checks": ["guard", "build-test"],
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def baseline(tmp_path: Path) -> Path:
|
|
p = tmp_path / "p3-baseline.json"
|
|
p.write_text(json.dumps(_BASELINE), encoding="utf-8")
|
|
return p
|
|
|
|
|
|
@pytest.fixture
|
|
def shim_bin(tmp_path: Path) -> tuple[Path, Path]:
|
|
"""A bin dir with stub ``gh`` and ``git`` that only record their argv.
|
|
|
|
Returns ``(bin_dir, calls_log)``. The script, run with this dir prepended to
|
|
PATH, makes ZERO real gh/git calls — every invocation appends a line to
|
|
``calls_log`` and exits 0. Where the script reads command output (the
|
|
post-restore asserts), the stubs emit the baseline value so the assert holds.
|
|
"""
|
|
bin_dir = tmp_path / "bin"
|
|
bin_dir.mkdir()
|
|
calls_log = tmp_path / "calls.log"
|
|
|
|
# gh stub: record argv; emit canned output for the read-only post-restore
|
|
# asserts so --apply asserts pass. The script passes a server-side --jq to gh
|
|
# (the real gh applies it); the stub must therefore emit the ALREADY-jq'd
|
|
# value the script expects:
|
|
# * env GET with the reviewer-ids --jq -> "1234567" (space-joined ids)
|
|
# * enforce_admins GET -> "true"
|
|
# * protection GET (no enforce_admins) -> the full protection JSON, which
|
|
# the script then pipes through its own normalize_protection. We emit the
|
|
# same shape the baseline records so the normalized compare holds.
|
|
gh = bin_dir / "gh"
|
|
_protection_json = json.dumps(_BASELINE["protection"]["full"])
|
|
gh.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
f'printf "gh %s\\n" "$*" >> "{calls_log}"\n'
|
|
'argv="$*"\n'
|
|
'for a in "$@"; do\n'
|
|
' case "$a" in\n'
|
|
" */enforce_admins) echo 'true'; exit 0 ;;\n"
|
|
" esac\n"
|
|
"done\n"
|
|
"# protection GET (full object) -> emit the baseline full protection JSON.\n"
|
|
'case "$argv" in\n'
|
|
" *branches/*/protection*)\n"
|
|
f" cat <<'JSON'\n{_protection_json}\nJSON\n"
|
|
" exit 0 ;;\n"
|
|
" *users/*)\n"
|
|
" # login->id resolution (gh api users/{login} --jq .id).\n"
|
|
" echo '1234567'; exit 0 ;;\n"
|
|
" *environments/*)\n"
|
|
" # reviewer-ids --jq result (space-joined) for the post-restore assert.\n"
|
|
" echo '1234567'; exit 0 ;;\n"
|
|
"esac\n"
|
|
"exit 0\n",
|
|
encoding="utf-8",
|
|
)
|
|
git = bin_dir / "git"
|
|
git.write_text(
|
|
f'#!/usr/bin/env bash\nprintf "git %s\\n" "$*" >> "{calls_log}"\nexit 0\n',
|
|
encoding="utf-8",
|
|
)
|
|
for f in (gh, git):
|
|
f.chmod(f.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
|
|
return bin_dir, calls_log
|
|
|
|
|
|
def _run(
|
|
*args: str,
|
|
baseline: Path,
|
|
shim: tuple[Path, Path] | None = None,
|
|
extra_env: dict[str, str] | None = None,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
env = dict(os.environ)
|
|
if shim is not None:
|
|
bin_dir, _ = shim
|
|
env["PATH"] = f"{bin_dir}{os.pathsep}{env['PATH']}"
|
|
if extra_env:
|
|
env.update(extra_env)
|
|
return subprocess.run(
|
|
["bash", str(_SCRIPT), *args, "--baseline", str(baseline)],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
# cwd kept default (no git repo needed: gh/git are shimmed).
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Dry-run plan output (the default; no shim required — nothing executes)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_default_is_dry_run_and_prints_plan_not_apply(baseline: Path) -> None:
|
|
res = _run("all", "--flip-pr", "7", baseline=baseline)
|
|
assert res.returncode == 0, res.stderr
|
|
out = res.stdout
|
|
assert "--dry-run (plan only" in out
|
|
assert "[PLAN]" in out
|
|
# No APPLY lines must appear in dry-run.
|
|
assert "[APPLY]" not in out
|
|
|
|
|
|
def test_dry_run_covers_all_four_surfaces(baseline: Path) -> None:
|
|
res = _run("all", "--flip-pr", "7", baseline=baseline)
|
|
out = res.stdout
|
|
assert "Restore the apply/verify workflow flip" in out
|
|
assert "Restore the agent-apply environment" in out
|
|
assert "Neutralise the GitHub App" in out
|
|
assert "Restore the FULL branch protection baseline" in out
|
|
|
|
|
|
def test_dry_run_workflow_premerge_closes_pr_and_deletes_branch(
|
|
baseline: Path,
|
|
) -> None:
|
|
res = _run(
|
|
"workflow",
|
|
"--flip-pr",
|
|
"7",
|
|
"--flip-branch",
|
|
"agent-team/apply/t1",
|
|
baseline=baseline,
|
|
)
|
|
out = res.stdout
|
|
assert "gh pr close 7" in out
|
|
assert "--delete-branch" in out
|
|
assert "git/refs/heads/agent-team/apply/t1" in out
|
|
# LIVE: the run-name/permissions edits get reverted to the baseline SHA.
|
|
assert "git checkout 0123abc --" in out
|
|
|
|
|
|
def test_dry_run_workflow_postmerge_reverts_commit_and_reruns_ci(
|
|
baseline: Path,
|
|
) -> None:
|
|
res = _run(
|
|
"workflow",
|
|
"--merged",
|
|
"--flip-commit",
|
|
"cafef00d",
|
|
baseline=baseline,
|
|
)
|
|
out = res.stdout
|
|
assert "git revert --no-edit cafef00d" in out
|
|
assert "git push origin HEAD" in out
|
|
assert "gh workflow run" in out
|
|
|
|
|
|
def test_dry_run_app_uninstall_requires_app_jwt_not_operator_gh(
|
|
baseline: Path,
|
|
) -> None:
|
|
"""The corrected model: there is NO permission-reduction endpoint, and the
|
|
installation token is NOT revoked by operator gh. The only programmatic
|
|
neutralise is uninstall, which needs an App JWT."""
|
|
res = _run("app", baseline=baseline)
|
|
out = res.stdout
|
|
# The fictional permission-reduction endpoint must NOT appear.
|
|
assert "/permissions" not in out
|
|
assert "PATCH" not in out
|
|
# The token is NOT revoked by operator gh.
|
|
assert "DELETE installation/token" not in out
|
|
assert "cannot be revoked by operator gh" in out
|
|
# Uninstall is planned and clearly flagged as needing an App JWT.
|
|
assert "UNINSTALL App 'agent-apply' installation 424242" in out
|
|
assert "APP JWT" in out
|
|
assert "app/installations/424242" in out
|
|
|
|
|
|
def test_dry_run_app_out_of_band_path(tmp_path: Path) -> None:
|
|
data = json.loads(json.dumps(_BASELINE))
|
|
data["app"]["action"] = "out-of-band"
|
|
p = tmp_path / "b.json"
|
|
p.write_text(json.dumps(data), encoding="utf-8")
|
|
res = _run("app", baseline=p)
|
|
out = res.stdout
|
|
assert "OUT-OF-BAND App neutralise" in out
|
|
assert "no REST endpoint reduces App permissions" in out
|
|
# Still no fictional permission API.
|
|
assert "/permissions" not in out
|
|
|
|
|
|
def test_apply_app_uninstall_without_jwt_fails_closed(
|
|
baseline: Path, shim_bin: tuple[Path, Path]
|
|
) -> None:
|
|
"""--apply uninstall with no AGENT_APPLY_APP_JWT must refuse — operator gh
|
|
cannot perform DELETE /app/installations/{id} (it needs an App JWT)."""
|
|
env = dict(os.environ)
|
|
env.pop("AGENT_APPLY_APP_JWT", None)
|
|
bin_dir, _ = shim_bin
|
|
env["PATH"] = f"{bin_dir}{os.pathsep}{env['PATH']}"
|
|
res = subprocess.run(
|
|
["bash", str(_SCRIPT), "app", "--apply", "--baseline", str(baseline)],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
assert res.returncode != 0
|
|
assert "needs an APP JWT" in res.stderr
|
|
|
|
|
|
def test_apply_app_uninstall_with_jwt_invokes_delete(
|
|
baseline: Path, shim_bin: tuple[Path, Path]
|
|
) -> None:
|
|
"""With an App JWT present, --apply uninstall issues the DELETE to the
|
|
(shimmed) gh with a Bearer Authorization header."""
|
|
env = dict(os.environ)
|
|
env["AGENT_APPLY_APP_JWT"] = "jwt-token-abc"
|
|
bin_dir, calls_log = shim_bin
|
|
env["PATH"] = f"{bin_dir}{os.pathsep}{env['PATH']}"
|
|
res = subprocess.run(
|
|
["bash", str(_SCRIPT), "app", "--apply", "--baseline", str(baseline)],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
assert res.returncode == 0, res.stderr + res.stdout
|
|
calls = calls_log.read_text(encoding="utf-8")
|
|
assert "DELETE app/installations/424242" in calls
|
|
assert "Authorization: Bearer jwt-token-abc" in calls
|
|
|
|
|
|
def test_dry_run_protection_restores_full_baseline(baseline: Path) -> None:
|
|
res = _run("protection", baseline=baseline)
|
|
out = res.stdout
|
|
# The FULL protection object is restored (not just enforce_admins).
|
|
assert "PUT full branch protection on main from baseline protection.full" in out
|
|
# And the post-restore asserts both enforce_admins and the full object.
|
|
assert "assert enforce_admins.enabled == true" in out
|
|
assert "assert LIVE protection == baseline protection.full" in out
|
|
|
|
|
|
def test_protection_without_full_refused_unless_partial_acked(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
# MEDIUM-1: a baseline missing protection.full must NOT silently restore only
|
|
# enforce_admins (a weaker posture). Without the explicit ack it fails hard.
|
|
data = json.loads(json.dumps(_BASELINE))
|
|
del data["protection"]["full"]
|
|
p = tmp_path / "nofull.json"
|
|
p.write_text(json.dumps(data), encoding="utf-8")
|
|
|
|
refused = _run("protection", baseline=p)
|
|
assert refused.returncode != 0
|
|
assert "no protection.full" in (refused.stdout + refused.stderr)
|
|
assert "P3_ROLLBACK_ALLOW_PARTIAL" in (refused.stdout + refused.stderr)
|
|
|
|
# With the explicit ack the degraded enforce_admins-only restore proceeds, but
|
|
# LOUDLY warns it is partial.
|
|
acked = _run("protection", baseline=p, extra_env={"P3_ROLLBACK_ALLOW_PARTIAL": "1"})
|
|
out = acked.stdout + acked.stderr
|
|
assert "DEGRADED protection restore" in out
|
|
assert "branches/main/protection/enforce_admins" in acked.stdout
|
|
|
|
|
|
def test_dry_run_incident_path_full_sequence(baseline: Path) -> None:
|
|
res = _run(
|
|
"incident",
|
|
"--flip-pr",
|
|
"13",
|
|
"--flip-branch",
|
|
"agent-team/apply/t9",
|
|
baseline=baseline,
|
|
)
|
|
assert res.returncode == 0, res.stderr
|
|
out = res.stdout
|
|
# a. neutralise App b. revert draft PR/branch c. audit Checks d. restore e. note
|
|
assert "Neutralise the GitHub App" in out
|
|
# The corrected model: NOT an operator-gh token revoke.
|
|
assert "DELETE installation/token" not in out
|
|
assert "UNINSTALL App 'agent-apply' installation 424242" in out
|
|
assert "gh pr close 13" in out
|
|
assert "git/refs/heads/agent-team/apply/t9" in out
|
|
assert "Audit the Checks trail" in out
|
|
assert "gh run list" in out
|
|
assert "Restore the agent-apply environment" in out
|
|
assert "Restore the FULL branch protection baseline" in out
|
|
assert "incident note" in out
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Argument parsing — fail closed
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_missing_surface_exits_2(baseline: Path) -> None:
|
|
res = _run(baseline=baseline)
|
|
assert res.returncode == 2
|
|
assert "a surface is required" in res.stderr
|
|
|
|
|
|
def test_unknown_flag_exits_2(baseline: Path) -> None:
|
|
res = _run("workflow", "--bogus", baseline=baseline)
|
|
assert res.returncode == 2
|
|
assert "unknown argument: --bogus" in res.stderr
|
|
|
|
|
|
def test_two_surfaces_is_rejected(baseline: Path) -> None:
|
|
res = _run("workflow", "protection", baseline=baseline)
|
|
assert res.returncode == 2
|
|
assert "surface already set" in res.stderr
|
|
|
|
|
|
def test_flag_missing_value_exits_2(baseline: Path) -> None:
|
|
# --flip-pr with no following value (the trailing --baseline is consumed as
|
|
# the value, but then --baseline has no value -> still a parse error path).
|
|
res = subprocess.run(
|
|
["bash", str(_SCRIPT), "workflow", "--flip-pr"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert res.returncode == 2
|
|
|
|
|
|
def test_help_exits_0_and_lists_surfaces() -> None:
|
|
res = subprocess.run(
|
|
["bash", str(_SCRIPT), "--help"], capture_output=True, text=True
|
|
)
|
|
assert res.returncode == 0
|
|
for surface in ("workflow", "environment", "app", "protection", "incident"):
|
|
assert surface in res.stdout
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Fail-closed posture
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_missing_baseline_file_is_refused(tmp_path: Path) -> None:
|
|
missing = tmp_path / "nope.json"
|
|
res = _run("protection", baseline=missing)
|
|
assert res.returncode != 0
|
|
assert "baseline file not found" in res.stderr
|
|
|
|
|
|
def test_protection_baseline_without_admins_on_is_refused(tmp_path: Path) -> None:
|
|
data = json.loads(json.dumps(_BASELINE))
|
|
data["protection"]["include_administrators"] = False
|
|
p = tmp_path / "weak.json"
|
|
p.write_text(json.dumps(data), encoding="utf-8")
|
|
res = _run("protection", baseline=p)
|
|
assert res.returncode != 0
|
|
assert "include_administrators is not true" in res.stderr
|
|
|
|
|
|
def test_repo_mismatch_is_refused(baseline: Path) -> None:
|
|
res = _run("protection", "--repo", "evil/other", baseline=baseline)
|
|
assert res.returncode != 0
|
|
assert "repo mismatch" in res.stderr
|
|
|
|
|
|
def test_workflow_premerge_requires_flip_pr(baseline: Path) -> None:
|
|
res = _run("workflow", baseline=baseline)
|
|
assert res.returncode != 0
|
|
assert "pre-merge path needs --flip-pr" in res.stderr
|
|
|
|
|
|
def test_workflow_postmerge_requires_flip_commit(baseline: Path) -> None:
|
|
res = _run("workflow", "--merged", baseline=baseline)
|
|
assert res.returncode != 0
|
|
assert "post-merge path needs --flip-commit" in res.stderr
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# --apply against a PATH-shimmed gh/git (records argv, no real side effects)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_apply_protection_invokes_gh_and_asserts(
|
|
baseline: Path, shim_bin: tuple[Path, Path]
|
|
) -> None:
|
|
_, calls_log = shim_bin
|
|
res = _run("protection", "--apply", baseline=baseline, shim=shim_bin)
|
|
assert res.returncode == 0, res.stderr + res.stdout
|
|
assert "[APPLY]" in res.stdout
|
|
# The post-restore assert ran and held (shim emits 'true').
|
|
assert "[OK]" in res.stdout
|
|
calls = calls_log.read_text(encoding="utf-8")
|
|
# The FULL protection object was PUT to the (shimmed) gh, then asserted.
|
|
assert (
|
|
"PUT repos/Sea-Haven-Industries/orchestrator/branches/main/protection" in calls
|
|
)
|
|
assert "branches/main/protection/enforce_admins" in calls
|
|
|
|
|
|
def test_apply_environment_puts_reviewer_ids_and_asserts(
|
|
baseline: Path, shim_bin: tuple[Path, Path]
|
|
) -> None:
|
|
_, calls_log = shim_bin
|
|
res = _run("environment", "--apply", baseline=baseline, shim=shim_bin)
|
|
assert res.returncode == 0, res.stderr + res.stdout
|
|
calls = calls_log.read_text(encoding="utf-8")
|
|
assert "environments/agent-apply" in calls
|
|
# Reviewers are sent as proper typed JSON fields, by NUMERIC id.
|
|
assert "reviewers[][type]=User" in calls
|
|
assert "reviewers[][id]=1234567" in calls
|
|
# The post-restore assert compared live reviewer ids to the baseline and held.
|
|
assert "[OK]" in res.stdout
|
|
|
|
|
|
def test_apply_workflow_premerge_records_close_and_branch_delete(
|
|
baseline: Path, shim_bin: tuple[Path, Path]
|
|
) -> None:
|
|
_, calls_log = shim_bin
|
|
res = _run(
|
|
"workflow",
|
|
"--apply",
|
|
"--flip-pr",
|
|
"7",
|
|
"--flip-branch",
|
|
"agent-team/apply/t1",
|
|
baseline=baseline,
|
|
shim=shim_bin,
|
|
)
|
|
assert res.returncode == 0, res.stderr + res.stdout
|
|
calls = calls_log.read_text(encoding="utf-8")
|
|
assert "pr close 7" in calls
|
|
assert "git/refs/heads/agent-team/apply/t1" in calls
|
|
assert "checkout 0123abc" in calls
|
|
|
|
|
|
def test_dry_run_environment_resolves_login_to_id_when_no_ids_recorded(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""A baseline that recorded only logins (no required_reviewer_ids) plans a
|
|
login->id resolution via 'gh api users/{login} --jq .id'."""
|
|
data = json.loads(json.dumps(_BASELINE))
|
|
del data["environment"]["required_reviewer_ids"]
|
|
p = tmp_path / "logins.json"
|
|
p.write_text(json.dumps(data), encoding="utf-8")
|
|
res = _run("environment", baseline=p)
|
|
out = res.stdout
|
|
assert "resolve reviewer login 'amoussa1229'" in out
|
|
assert "gh api users/amoussa1229 --jq .id" in out
|
|
|
|
|
|
def test_apply_environment_resolves_login_to_id(tmp_path: Path) -> None:
|
|
"""--apply with a login-only baseline resolves the login to a numeric id via
|
|
the shimmed 'gh api users/{login}' and sends it as a typed reviewer field."""
|
|
# Build a login-only baseline.
|
|
data = json.loads(json.dumps(_BASELINE))
|
|
del data["environment"]["required_reviewer_ids"]
|
|
bpath = tmp_path / "logins.json"
|
|
bpath.write_text(json.dumps(data), encoding="utf-8")
|
|
|
|
# Build a shim bin in this tmp_path.
|
|
bin_dir = tmp_path / "bin"
|
|
bin_dir.mkdir()
|
|
calls_log = tmp_path / "calls.log"
|
|
gh = bin_dir / "gh"
|
|
gh.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
f'printf "gh %s\\n" "$*" >> "{calls_log}"\n'
|
|
'argv="$*"\n'
|
|
'for a in "$@"; do\n'
|
|
' case "$a" in\n'
|
|
" */enforce_admins) echo 'true'; exit 0 ;;\n"
|
|
" esac\n"
|
|
"done\n"
|
|
'case "$argv" in\n'
|
|
" *users/*) echo '7654321'; exit 0 ;;\n"
|
|
" *environments/*) echo '7654321'; exit 0 ;;\n"
|
|
"esac\n"
|
|
"exit 0\n",
|
|
encoding="utf-8",
|
|
)
|
|
gh.chmod(gh.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
|
|
|
|
env = dict(os.environ)
|
|
env["PATH"] = f"{bin_dir}{os.pathsep}{env['PATH']}"
|
|
res = subprocess.run(
|
|
["bash", str(_SCRIPT), "environment", "--apply", "--baseline", str(bpath)],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
assert res.returncode == 0, res.stderr + res.stdout
|
|
calls = calls_log.read_text(encoding="utf-8")
|
|
assert "api users/amoussa1229" in calls
|
|
assert "reviewers[][id]=7654321" in calls
|
|
assert "[OK]" in res.stdout
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# --record-baseline mode
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_record_baseline_rejects_a_surface(tmp_path: Path) -> None:
|
|
out = tmp_path / "p3-baseline.json"
|
|
res = subprocess.run(
|
|
[
|
|
"bash",
|
|
str(_SCRIPT),
|
|
"--record-baseline",
|
|
"protection",
|
|
"--baseline",
|
|
str(out),
|
|
"--repo",
|
|
"Sea-Haven-Industries/orchestrator",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert res.returncode == 2
|
|
assert "does not take a surface" in res.stderr
|
|
|
|
|
|
def test_record_baseline_dry_run_prints_plan(tmp_path: Path) -> None:
|
|
out = tmp_path / "p3-baseline.json"
|
|
res = subprocess.run(
|
|
[
|
|
"bash",
|
|
str(_SCRIPT),
|
|
"--record-baseline",
|
|
"--baseline",
|
|
str(out),
|
|
"--repo",
|
|
"Sea-Haven-Industries/orchestrator",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert res.returncode == 0, res.stderr
|
|
assert "RECORD BASELINE" in res.stdout
|
|
assert "[PLAN]" in res.stdout
|
|
# Dry-run captures nothing.
|
|
assert not out.exists()
|
|
|
|
|
|
def test_record_baseline_requires_repo(tmp_path: Path) -> None:
|
|
out = tmp_path / "p3-baseline.json"
|
|
# Clear the env-derived repo default so no repo is resolvable.
|
|
env = dict(os.environ)
|
|
env.pop("AGENT_TEAM_REPO_OWNER", None)
|
|
env.pop("AGENT_TEAM_REPO_NAME", None)
|
|
res = subprocess.run(
|
|
["bash", str(_SCRIPT), "--record-baseline", "--baseline", str(out)],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
assert res.returncode != 0
|
|
assert "no target repo" in res.stderr
|
|
|
|
|
|
def test_record_baseline_apply_writes_baseline_from_live_state(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""--record-baseline --apply captures the live workflow SHA, env reviewer
|
|
ids, full branch protection and App installation id into the baseline JSON,
|
|
creating the directory if absent. The recorded file is then a valid restore
|
|
target whose protection.include_administrators is True."""
|
|
out = tmp_path / ".security-review" / "p3-baseline.json" # dir absent on purpose
|
|
|
|
bin_dir = tmp_path / "bin"
|
|
bin_dir.mkdir()
|
|
calls_log = tmp_path / "calls.log"
|
|
protection_json = json.dumps(_BASELINE["protection"]["full"])
|
|
gh = bin_dir / "gh"
|
|
gh.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
f'printf "gh %s\\n" "$*" >> "{calls_log}"\n'
|
|
'argv="$*"\n'
|
|
'case "$argv" in\n'
|
|
" *contents/*) echo 'deadbeefsha'; exit 0 ;;\n"
|
|
" *branches/*/protection*)\n"
|
|
f" cat <<'JSON'\n{protection_json}\nJSON\n"
|
|
" exit 0 ;;\n"
|
|
" *environments/*) echo '[1234567]'; exit 0 ;;\n"
|
|
"esac\n"
|
|
"exit 0\n",
|
|
encoding="utf-8",
|
|
)
|
|
gh.chmod(gh.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
|
|
|
|
env = dict(os.environ)
|
|
env["PATH"] = f"{bin_dir}{os.pathsep}{env['PATH']}"
|
|
res = subprocess.run(
|
|
[
|
|
"bash",
|
|
str(_SCRIPT),
|
|
"--record-baseline",
|
|
"--apply",
|
|
"--baseline",
|
|
str(out),
|
|
"--repo",
|
|
"Sea-Haven-Industries/orchestrator",
|
|
"--app-installation-id",
|
|
"424242",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
assert res.returncode == 0, res.stderr + res.stdout
|
|
assert out.exists(), "baseline file (and its dir) must be created"
|
|
recorded = json.loads(out.read_text(encoding="utf-8"))
|
|
assert recorded["repo"] == "Sea-Haven-Industries/orchestrator"
|
|
assert recorded["workflow_baseline_sha"] == "deadbeefsha"
|
|
assert recorded["environment"]["required_reviewer_ids"] == [1234567]
|
|
assert recorded["app"]["installation_id"] == 424242
|
|
assert recorded["app"]["action"] == "uninstall"
|
|
assert recorded["protection"]["include_administrators"] is True
|
|
assert recorded["protection"]["full"]["enforce_admins"]["enabled"] is True
|
|
|
|
|
|
def test_recorded_baseline_is_a_valid_restore_target(tmp_path: Path) -> None:
|
|
"""A baseline produced by --record-baseline --apply can be fed straight back
|
|
into a restore (dry-run) without error — closing the record->restore loop."""
|
|
out = tmp_path / ".security-review" / "p3-baseline.json"
|
|
bin_dir = tmp_path / "bin"
|
|
bin_dir.mkdir()
|
|
calls_log = tmp_path / "calls.log"
|
|
protection_json = json.dumps(_BASELINE["protection"]["full"])
|
|
gh = bin_dir / "gh"
|
|
gh.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
f'printf "gh %s\\n" "$*" >> "{calls_log}"\n'
|
|
'argv="$*"\n'
|
|
'case "$argv" in\n'
|
|
" *contents/*) echo 'deadbeefsha'; exit 0 ;;\n"
|
|
" *branches/*/protection*)\n"
|
|
f" cat <<'JSON'\n{protection_json}\nJSON\n"
|
|
" exit 0 ;;\n"
|
|
" *environments/*) echo '[1234567]'; exit 0 ;;\n"
|
|
"esac\n"
|
|
"exit 0\n",
|
|
encoding="utf-8",
|
|
)
|
|
gh.chmod(gh.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
|
|
env = dict(os.environ)
|
|
env["PATH"] = f"{bin_dir}{os.pathsep}{env['PATH']}"
|
|
rec = subprocess.run(
|
|
[
|
|
"bash",
|
|
str(_SCRIPT),
|
|
"--record-baseline",
|
|
"--apply",
|
|
"--baseline",
|
|
str(out),
|
|
"--repo",
|
|
"Sea-Haven-Industries/orchestrator",
|
|
"--app-installation-id",
|
|
"424242",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
assert rec.returncode == 0, rec.stderr + rec.stdout
|
|
|
|
# Now restore (dry-run) from the recorded baseline.
|
|
res = subprocess.run(
|
|
["bash", str(_SCRIPT), "protection", "--baseline", str(out)],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert res.returncode == 0, res.stderr + res.stdout
|
|
assert "PUT full branch protection" in res.stdout
|
|
|
|
|
|
def test_apply_protection_DETECTS_divergent_live_restore(
|
|
baseline: Path, tmp_path: Path
|
|
) -> None:
|
|
"""Regression for the normalize_protection heredoc bug (vacuous assert).
|
|
|
|
Previously ``normalize_protection`` ran ``python3 - <<'PY'`` and read
|
|
``json.load(sys.stdin)`` — but the heredoc IS stdin, so it parsed the program
|
|
text, raised, and the bare except swallowed it to "" for EVERY input. Both
|
|
live and baseline normalized to "" and the post-restore assert was always
|
|
"" == "" -> a broken protection restore reported SUCCESS. The fix reads the
|
|
JSON from argv[1]. This test feeds a LIVE protection that DIFFERS from the
|
|
baseline and asserts the script now FAILS (non-zero) instead of falsely
|
|
passing.
|
|
"""
|
|
import copy
|
|
|
|
divergent = copy.deepcopy(_BASELINE["protection"]["full"])
|
|
# Flip a projected field so the normalized live != normalized baseline.
|
|
divergent["allow_force_pushes"] = {"enabled": True}
|
|
|
|
bin_dir = tmp_path / "divbin"
|
|
bin_dir.mkdir()
|
|
calls_log = tmp_path / "divcalls.log"
|
|
div_json = json.dumps(divergent)
|
|
gh = bin_dir / "gh"
|
|
gh.write_text(
|
|
"#!/usr/bin/env bash\n"
|
|
f'printf "gh %s\\n" "$*" >> "{calls_log}"\n'
|
|
'argv="$*"\n'
|
|
'for a in "$@"; do\n'
|
|
' case "$a" in\n'
|
|
" */enforce_admins) echo 'true'; exit 0 ;;\n"
|
|
" esac\n"
|
|
"done\n"
|
|
'case "$argv" in\n'
|
|
" *branches/*/protection*)\n"
|
|
f" cat <<'JSON'\n{div_json}\nJSON\n"
|
|
" exit 0 ;;\n"
|
|
" *users/*) echo '1234567'; exit 0 ;;\n"
|
|
" *environments/*) echo '1234567'; exit 0 ;;\n"
|
|
"esac\n"
|
|
"exit 0\n",
|
|
encoding="utf-8",
|
|
)
|
|
git = bin_dir / "git"
|
|
git.write_text(
|
|
f'#!/usr/bin/env bash\nprintf "git %s\\n" "$*" >> "{calls_log}"\nexit 0\n',
|
|
encoding="utf-8",
|
|
)
|
|
for f in (gh, git):
|
|
f.chmod(f.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
|
|
|
|
res = _run("protection", "--apply", baseline=baseline, shim=(bin_dir, calls_log))
|
|
assert res.returncode != 0, (
|
|
"divergent live protection must FAIL the post-restore assert, not pass:\n"
|
|
+ res.stdout
|
|
+ res.stderr
|
|
)
|
|
assert "protection.full" in (res.stdout + res.stderr)
|
|
|
|
|
|
def test_missing_required_key_refuses_partial_restore(tmp_path: Path) -> None:
|
|
# MEDIUM-1: a baseline missing a required key for a surface aborts that
|
|
# surface up front rather than half-restoring it.
|
|
data = json.loads(json.dumps(_BASELINE))
|
|
del data["workflow_baseline_sha"]
|
|
p = tmp_path / "no_sha.json"
|
|
p.write_text(json.dumps(data), encoding="utf-8")
|
|
res = _run("workflow", baseline=p)
|
|
assert res.returncode != 0
|
|
out = res.stdout + res.stderr
|
|
assert "missing required key" in out
|
|
assert "workflow_baseline_sha" in out
|
|
|
|
|
|
def test_app_uninstall_without_jwt_requires_oob_ack(tmp_path: Path) -> None:
|
|
# MEDIUM-2: an --apply that needs a MANUAL App neutralise (no APP JWT) must
|
|
# not silently skip it — it refuses unless the operator acknowledges.
|
|
p = tmp_path / "bl.json"
|
|
p.write_text(json.dumps(_BASELINE), encoding="utf-8")
|
|
|
|
# No JWT, no ack -> refuse.
|
|
refused = _run("app", "--apply", baseline=p)
|
|
assert refused.returncode != 0
|
|
assert "P3_ROLLBACK_OOB_ACK" in (refused.stdout + refused.stderr)
|
|
|
|
# No JWT, but acked -> proceeds (App neutralise is operator-owed, loudly warned).
|
|
acked = _run("app", "--apply", baseline=p, extra_env={"P3_ROLLBACK_OOB_ACK": "1"})
|
|
assert acked.returncode == 0, acked.stdout + acked.stderr
|
|
assert "OUT-OF-BAND ACK accepted" in (acked.stdout + acked.stderr)
|