diff --git a/.github/actions/contributor-check/tests/test_contributor_check_action.py b/.github/actions/contributor-check/tests/test_contributor_check_action.py index 14a5f1c..fe57bb4 100644 --- a/.github/actions/contributor-check/tests/test_contributor_check_action.py +++ b/.github/actions/contributor-check/tests/test_contributor_check_action.py @@ -103,7 +103,11 @@ def test_genuine_unknown_is_still_returned(tmp_path): @pytest.mark.parametrize("risk,expected", [("LOW", "LOW"), ("MEDIUM", "MEDIUM"), ("HIGH", "HIGH")]) def test_computed_risk_passes_through(tmp_path, risk, expected): - script = _stub(tmp_path, "profile.py", f'print(\'{{"risk": "{risk}"}}\')\n') + exit_code = {"LOW": 0, "MEDIUM": 1, "HIGH": 2}[risk] + script = _stub( + tmp_path, "profile.py", + f'import sys\nprint(\'{{"risk": "{risk}"}}\')\nsys.exit({exit_code})\n', + ) assert action._run_check(script, [], str(tmp_path / "out.json")) == expected diff --git a/.github/actions/contributor-check/tests/test_risk_exit_contract.py b/.github/actions/contributor-check/tests/test_risk_exit_contract.py new file mode 100644 index 0000000..f3673ec --- /dev/null +++ b/.github/actions/contributor-check/tests/test_risk_exit_contract.py @@ -0,0 +1,74 @@ +"""Contract tests for contributor assessment process exit codes.""" +import importlib.util +import json +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +MODULE = Path(__file__).resolve().parents[1] / "vendor" / "agt" / "contributor_check_action.py" +spec = importlib.util.spec_from_file_location("contributor_check_action_under_test", MODULE) +action = importlib.util.module_from_spec(spec) +spec.loader.exec_module(action) + + +class RiskExitContractTests(unittest.TestCase): + def run_case(self, code, value, expected=None): + with tempfile.TemporaryDirectory() as directory: + result = subprocess.CompletedProcess(["python"], code, json.dumps(value), "") + with patch.object(action.subprocess, "run", return_value=result): + if expected is None: + with self.assertRaises(action.CheckExecutionError): + action._run_check("dummy.py", [], str(Path(directory) / "out")) + else: + self.assertEqual( + action._run_check("dummy.py", [], str(Path(directory) / "out")), + expected, + ) + + def test_valid_assessments(self): + for risk, code, normalized in [ + ("NONE", 0, "LOW"), ("LOW", 0, "LOW"), + ("UNKNOWN", 0, "UNKNOWN"), + ("MEDIUM", 1, "MEDIUM"), ("HIGH", 2, "HIGH"), + ]: + with self.subTest(risk=risk): + self.run_case(code, {"risk": risk}, normalized) + + def test_contradictions_and_bad_output_fail_closed(self): + for code, value in [ + (0, {"risk": "HIGH"}), (1, {"risk": "LOW"}), + (2, {"risk": "MEDIUM"}), (0, {"risk": "INVALID"}), + (0, {}), (0, []), (0, {"risk": []}), + (0, {"risk": {}}), (0, {"risk": 3}), (3, {"risk": "HIGH"}), + (127, {"risk": "LOW"}), + ]: + with self.subTest(code=code, value=value): + self.run_case(code, value) + + def test_malformed_and_empty_json_fail_closed(self): + for content in ["", "{", "null", "[1]", "42"]: + with self.subTest(content=content): + with tempfile.TemporaryDirectory() as directory: + result = subprocess.CompletedProcess(["python"], 0, content, "") + with patch.object(action.subprocess, "run", return_value=result): + with self.assertRaises(action.CheckExecutionError): + action._run_check("dummy.py", [], str(Path(directory) / "out")) + + def test_operational_failures_are_not_risk_assessments(self): + for error in [subprocess.TimeoutExpired("test", 120), OSError("missing binary")]: + with self.subTest(error=type(error).__name__): + with tempfile.TemporaryDirectory() as directory: + with patch.object(action.subprocess, "run", side_effect=error): + with self.assertRaises(action.CheckExecutionError): + action._run_check("dummy.py", [], str(Path(directory) / "out")) + + def test_aggregation_preserves_high_and_unknown(self): + self.assertEqual(action._aggregate_risk("LOW", "HIGH"), "HIGH") + self.assertEqual(action._aggregate_risk("LOW", "UNKNOWN"), "UNKNOWN") + self.assertEqual(action._aggregate_risk("MEDIUM", "HIGH"), "HIGH") + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/actions/contributor-check/vendor/agt/contributor_check_action.py b/.github/actions/contributor-check/vendor/agt/contributor_check_action.py index b47db08..7480a5c 100644 --- a/.github/actions/contributor-check/vendor/agt/contributor_check_action.py +++ b/.github/actions/contributor-check/vendor/agt/contributor_check_action.py @@ -99,10 +99,10 @@ def _run_check(script: str, args: list[str], out_path: str) -> str: Path(out_path).write_text(result.stdout) - # A non-zero exit means the check did not run to completion. An import - # error, a missing file or an unhandled exception all land here, and all - # of them used to reach the bare `except` below as UNKNOWN. - if result.returncode != 0: + # Scanners use 0/1/2 for completed LOW-or-NONE/MEDIUM/HIGH + # assessments. A nonzero assessment is not a process failure. + expected_exit = {"NONE": 0, "LOW": 0, "UNKNOWN": 0, "MEDIUM": 1, "HIGH": 2} + if result.returncode not in (0, 1, 2): detail = (result.stderr or "").strip().splitlines() tail = detail[-1] if detail else "no stderr" raise CheckExecutionError( @@ -113,14 +113,19 @@ def _run_check(script: str, args: list[str], out_path: str) -> str: data = json.loads(result.stdout) except ValueError as exc: raise CheckExecutionError( - f"{name} exited 0 but produced no parseable JSON: {exc}" + f"{name} exited {result.returncode} but produced invalid JSON: {exc}" ) from exc - risk = data.get("risk", "UNKNOWN") - # credential_audit returns "NONE" when a user has no merged PRs in the - # target repo (nothing to launder = not applicable). Normalize to "LOW" - # so it aggregates correctly rather than being treated as UNKNOWN (which - # is reserved for checks that errored or could not be determined). + if not isinstance(data, dict): + raise CheckExecutionError(f"{name} produced non-object JSON") + risk = data.get("risk") + if not isinstance(risk, str) or risk not in expected_exit: + raise CheckExecutionError(f"{name} returned missing or unknown risk: {risk!r}") + if result.returncode != expected_exit[risk]: + raise CheckExecutionError( + f"{name} reported risk {risk} but exited {result.returncode}" + ) + # NONE means not applicable, not an unknown or failed scan. return "LOW" if risk == "NONE" else risk