orchestrator/agent-team/tests/test_ci_gate_workflow.py

"""Tests for the embedded guard logic in ``ci/agent-team-apply-verify.yml``.

The §3.3.2 trust-boundary guard (diff-integrity hash, the trust-control-surface
denylist, the symlink-escape reject, declared-scope enforcement) lives as an
inline Python heredoc inside the CI workflow, so it cannot be imported directly.
These tests extract that script from the YAML and execute it as the workflow
does — via env vars and a diff file — asserting the exit code for good and
adversarial diffs. This backs the workflow's correctness claim with a real,
runnable suite instead of an "verified during authoring" assertion, and guards
the symlink / non-UTF-8 / header-only-section fixes against regression.

It does NOT enable, provision, or run the workflow itself; it only exercises the
pure-code gate the workflow embeds.
"""

from __future__ import annotations

import hashlib
import os
import subprocess
import sys
from pathlib import Path

import pytest

_WORKFLOW = Path(__file__).resolve().parents[1] / "ci" / "agent-team-apply-verify.yml"


def _extract_guard_script() -> str:
    """Pull the first ``python3 - <<'PY' ... PY`` heredoc (the guard) from the YAML.

    The body is indented to sit under the YAML ``run:`` block; we strip the
    common 10-space lead so it is valid module source.
    """
    lines = _WORKFLOW.read_text(encoding="utf-8").splitlines()
    start = end = None
    for i, line in enumerate(lines):
        if start is None and line.strip() == "python3 - <<'PY'":
            start = i + 1
        elif start is not None and line.strip() == "PY":
            end = i
            break
    assert start is not None and end is not None, "guard heredoc not found"
    body = lines[start:end]
    return "\n".join(ln[10:] if ln.startswith(" " * 10) else ln for ln in body)


@pytest.fixture(scope="module")
def guard_script(tmp_path_factory: pytest.TempPathFactory) -> Path:
    path = tmp_path_factory.mktemp("guard") / "guard.py"
    path.write_text(_extract_guard_script(), encoding="utf-8")
    return path


def _run_guard(
    guard_script: Path,
    tmp_path: Path,
    diff: str | bytes,
    scope: str,
    *,
    bad_hash: bool = False,
) -> int:
    raw = diff.encode("utf-8") if isinstance(diff, str) else diff
    diff_path = tmp_path / "candidate.diff"
    diff_path.write_bytes(raw)
    expected = "deadbeef" if bad_hash else hashlib.sha256(raw).hexdigest()
    env = dict(
        os.environ,
        DIFF_PATH=str(diff_path),
        EXPECTED_DIFF_HASH=expected,
        DECLARED_SCOPE=scope,
    )
    result = subprocess.run(
        [sys.executable, str(guard_script)], env=env, capture_output=True, text=True
    )
    return result.returncode


CLEAN = (
    "diff --git a/src/app.py b/src/app.py\n"
    "--- a/src/app.py\n+++ b/src/app.py\n@@ -1 +1 @@\n-x\n+y\n"
)
SYMLINK = (
    "diff --git a/src/link b/src/link\nnew file mode 120000\n"
    "--- /dev/null\n+++ b/src/link\n@@ -0,0 +1 @@\n+../.github/workflows\n"
)
WORKFLOW_DELETE = (
    "diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\n"
    "deleted file mode 100644\n--- a/.github/workflows/ci.yml\n+++ /dev/null\n"
    "@@ -1 +0,0 @@\n-on: push\n"
)
COPY_TO_DENIED = (
    "diff --git a/src/x.py b/.github/workflows/evil.yml\nsimilarity index 100%\n"
    "copy from src/x.py\ncopy to .github/workflows/evil.yml\n"
)
NON_UTF8 = (
    b"diff --git a/src/app.py b/src/app.py\n--- a/src/app.py\n"
    b"+++ b/src/app.py\n@@ -1 +1 @@\n-x\n+\xff\xfe\n"
)


def test_clean_in_scope_diff_passes(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, CLEAN, "src/**") == 0


def test_hash_mismatch_fails(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, CLEAN, "src/**", bad_hash=True) == 2


def test_workflow_delete_is_rejected(guard_script: Path, tmp_path: Path) -> None:
    # Header-only section (delete) caught via diff --git, not a +++ body line.
    assert (
        _run_guard(guard_script, tmp_path, WORKFLOW_DELETE, "src/**\n.github/**") == 4
    )


def test_copy_into_denied_path_is_rejected(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, COPY_TO_DENIED, "src/**\n.github/**") == 4


def test_symlink_addition_is_rejected(guard_script: Path, tmp_path: Path) -> None:
    # The symlink-escape vector: rejected outright (exit 7).
    assert _run_guard(guard_script, tmp_path, SYMLINK, "src/**") == 7


def test_non_utf8_diff_fails_closed(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, NON_UTF8, "src/**") == 8


def test_out_of_scope_path_is_rejected(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, CLEAN, "other/**") == 6


def test_unscoped_diff_is_rejected(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, CLEAN, "") == 5


def test_escaping_scope_entries_are_dropped(guard_script: Path, tmp_path: Path) -> None:
    # A parent-escaping scope entry must not widen coverage; it is dropped, so a
    # diff under it is treated as unscoped.
    assert _run_guard(guard_script, tmp_path, CLEAN, "../../etc") == 5


# --- security-review regressions: denylist & scope matcher (was fnmatch) ----


def _modify(path: str) -> str:
    return f"diff --git a/{path} b/{path}\n--- a/{path}\n+++ b/{path}\n@@ -1 +1 @@\n-x\n+y\n"


@pytest.mark.parametrize(
    "root_path",
    [
        "template.yaml",
        "main.tf",
        "cdk.json",
        "policy.json",
        "id.pem",
        "signing.key",
        "app-stack.ts",
        "infra_stack.py",
    ],
)
def test_root_level_iac_is_denied(
    guard_script: Path, tmp_path: Path, root_path: str
) -> None:
    # Regression: Python fnmatch '**/' is non-recursive, so root-level IaC/secret
    # files slipped the denylist. The glob->regex matcher must reject them (exit 4).
    assert _run_guard(guard_script, tmp_path, _modify(root_path), ".") == 4


@pytest.mark.parametrize(
    "cased_path", ["Template.YAML", "Main.TF", ".github/Workflows/ci.yml"]
)
def test_denylist_is_case_insensitive(
    guard_script: Path, tmp_path: Path, cased_path: str
) -> None:
    # A case variant of a trust-control filename must not evade the gate.
    assert _run_guard(guard_script, tmp_path, _modify(cased_path), ".") == 4


def test_scope_double_star_cannot_widen_to_whole_tree(
    guard_script: Path, tmp_path: Path
) -> None:
    # Regression: a '**' scope entry made in_scope() true for every path. It must
    # reduce to the empty (root) prefix and be dropped -> unscoped (exit 5).
    assert _run_guard(guard_script, tmp_path, _modify("any/deep/file.py"), "**") == 5


def test_scope_glob_reduces_to_concrete_prefix(
    guard_script: Path, tmp_path: Path
) -> None:
    # A legitimate 'src/**' scope still confines to the src/ prefix: in-scope
    # passes, a sibling path is rejected.
    assert _run_guard(guard_script, tmp_path, _modify("src/app.py"), "src/**") == 0
    assert _run_guard(guard_script, tmp_path, _modify("other/app.py"), "src/**") == 6