from __future__ import annotations import json import os from pathlib import Path import shutil import subprocess import tempfile import textwrap import unittest META_ROOT = Path(__file__).resolve().parents[1] AUDIT_SCRIPT = META_ROOT / "tools" / "checks" / "check-security-audit.sh" CONTAINER_RUNNER = META_ROOT / "tools" / "checks" / "security-audit" / "run.sh" class SecurityAuditWrapperTests(unittest.TestCase): def setUp(self) -> None: self._temporary_directory = tempfile.TemporaryDirectory() self.temporary_root = Path(self._temporary_directory.name) self.checkout = self.temporary_root / "checkout" checks = self.checkout / "tools" / "checks" checks.mkdir(parents=True) shutil.copy2(AUDIT_SCRIPT, checks / AUDIT_SCRIPT.name) (self.checkout / "README.md").write_text("audit fixture\n", encoding="utf-8") subprocess.run(["git", "init", "-q", str(self.checkout)], check=True) subprocess.run( [ "git", "-C", str(self.checkout), "config", "user.email", "audit@example.invalid", ], check=True, ) subprocess.run( ["git", "-C", str(self.checkout), "config", "user.name", "Audit Test"], check=True, ) subprocess.run(["git", "-C", str(self.checkout), "add", "."], check=True) subprocess.run( ["git", "-C", str(self.checkout), "commit", "-qm", "audit fixture"], check=True, ) self.stub_bin = self.temporary_root / "bin" self.stub_bin.mkdir() self._write_stub( "semgrep", r""" if [[ " $* " == *" --version "* ]]; then echo 'semgrep stub 1.0' exit 0 fi if [[ " $* " != *" --error "* ]]; then echo 'missing --error' >&2 exit 3 fi output='' while [[ $# -gt 0 ]]; do if [[ "$1" == '--output' ]]; then output="$2" shift 2 else shift fi done case "${STUB_SEMGREP_REPORT:-valid}" in valid) printf '%s\n' '{"version":"2.1.0","runs":[]}' > "$output" ;; malformed) printf '%s\n' '{' > "$output" ;; invalid-sarif) printf '%s\n' '{}' > "$output" ;; missing) ;; esac if [[ -n "${STUB_MUTATE_PATH:-}" ]]; then printf '%s\n' 'changed during audit' >> "$STUB_MUTATE_PATH" fi exit "${STUB_SEMGREP_EXIT:-0}" """, ) self._write_stub( "gitleaks", r""" if [[ "${1:-}" == '--version' || "${1:-}" == 'version' ]]; then echo 'gitleaks stub 1.0' exit 0 fi if [[ "${1:-}" == 'git' && "${2:-}" == '--help' ]]; then exit 0 fi output='' while [[ $# -gt 0 ]]; do if [[ "$1" == '--report-path' ]]; then output="$2" shift 2 else shift fi done mkdir -p "$(dirname "$output")" printf '%s\n' '[]' > "$output" exit "${STUB_GITLEAKS_EXIT:-0}" """, ) for tool in ("bandit", "ruff"): self._write_stub( tool, f""" if [[ " $* " == *" --version "* ]]; then echo '{tool} stub 1.0' fi exit 0 """, ) def tearDown(self) -> None: self._temporary_directory.cleanup() def _write_stub(self, name: str, body: str) -> None: path = self.stub_bin / name path.write_text( "#!/usr/bin/env bash\nset -euo pipefail\n" + textwrap.dedent(body).lstrip(), encoding="utf-8", ) path.chmod(0o755) def _install_full_mode_stubs(self) -> None: self._write_stub( "trivy", r""" if [[ " $* " == *" --version "* ]]; then echo 'trivy stub 1.0' exit 0 fi output='' while [[ $# -gt 0 ]]; do if [[ "$1" == '--output' ]]; then output="$2" shift 2 else shift fi done printf '%s\n' '{"version":"2.1.0","runs":[]}' > "$output" """, ) self._write_stub( "osv-scanner", r""" if [[ " $* " == *" --version "* ]]; then echo 'osv-scanner stub 1.0' exit 0 fi if [[ " $* " == *" scan --help "* ]]; then exit 0 fi output='' while [[ $# -gt 0 ]]; do if [[ "$1" == '--output-file' ]]; then output="$2" shift 2 else shift fi done printf '{"results": %s}\n' "${STUB_OSV_RESULTS:-null}" > "$output" """, ) self._write_stub( "jscpd", r""" if [[ " $* " == *" --version "* ]]; then echo 'jscpd stub 1.0' exit 0 fi output='' while [[ $# -gt 0 ]]; do if [[ "$1" == '--output' ]]; then output="$2" shift 2 else shift fi done mkdir -p "$output" printf '%s\n' '{"duplicates":[],"statistics":{}}' > "$output/jscpd-report.json" """, ) for tool in ("pip-audit", "npm", "radon", "xenon"): self._write_stub( tool, f""" if [[ " $* " == *" --version "* ]]; then echo '{tool} stub 1.0' exit 0 fi exit 0 """, ) def _run( self, *arguments: str, mode: str = "quick", **environment: str ) -> tuple[subprocess.CompletedProcess[str], Path]: reports = self.temporary_root / "reports" env = os.environ.copy() env.update( { "PATH": f"{self.stub_bin}:{env['PATH']}", "SECURITY_AUDIT_REQUIRE_TOOLS": "1", "SECURITY_AUDIT_TOOLBOX_FINGERPRINT": 'stub-"safe', **environment, } ) result = subprocess.run( [ "bash", str(self.checkout / "tools" / "checks" / AUDIT_SCRIPT.name), "--mode", mode, "--scope", "current", "--reports-dir", str(reports), *arguments, ], cwd=self.checkout, env=env, check=False, capture_output=True, text=True, ) return result, reports @staticmethod def _manifest(reports: Path) -> dict[str, object]: return json.loads((reports / "manifest.json").read_text(encoding="utf-8")) def test_report_only_findings_are_nonfatal_and_strict_findings_fail(self) -> None: result, reports = self._run("--report-only", STUB_SEMGREP_EXIT="1") self.assertEqual(0, result.returncode, result.stderr) manifest = self._manifest(reports) self.assertEqual(0, manifest["overall_status"]) self.assertEqual(1, manifest["finding_status"]) self.assertEqual(0, manifest["execution_error_status"]) self.assertIn( "Semgrep SAST\t1\tfindings", (reports / "step-status.tsv").read_text() ) strict_result, strict_reports = self._run("--strict", STUB_SEMGREP_EXIT="1") self.assertEqual(1, strict_result.returncode) self.assertEqual(1, self._manifest(strict_reports)["overall_status"]) def test_scanner_execution_error_is_fatal_in_report_only_mode(self) -> None: result, reports = self._run("--report-only", STUB_SEMGREP_EXIT="2") self.assertEqual(1, result.returncode) manifest = self._manifest(reports) self.assertEqual(1, manifest["overall_status"]) self.assertEqual(1, manifest["execution_error_status"]) self.assertIn( "Semgrep SAST\t2\texecution-error", (reports / "step-status.tsv").read_text(), ) def test_malformed_or_missing_machine_report_is_fatal(self) -> None: for report_kind in ("malformed", "invalid-sarif", "missing"): with self.subTest(report_kind=report_kind): result, reports = self._run( "--report-only", STUB_SEMGREP_REPORT=report_kind, ) self.assertEqual(1, result.returncode) manifest = self._manifest(reports) self.assertEqual(1, manifest["execution_error_status"]) if report_kind == "missing": self.assertIn("semgrep.sarif", manifest["missing_reports"]) self.assertIn( "Validate machine-readable audit reports\t2\texecution-error", (reports / "step-status.tsv").read_text(), ) def test_osv_null_results_are_valid_but_wrong_shapes_fail(self) -> None: self._install_full_mode_stubs() result, reports = self._run("--report-only", mode="full") self.assertEqual(0, result.returncode, result.stderr) osv_report = json.loads( (reports / "osv-scanner-checkout.json").read_text(encoding="utf-8") ) self.assertIsNone(osv_report["results"]) invalid_result, invalid_reports = self._run( "--report-only", mode="full", STUB_OSV_RESULTS="{}", ) self.assertEqual(1, invalid_result.returncode) self.assertIn( "invalid OSV-Scanner report", invalid_result.stderr, ) self.assertEqual(1, self._manifest(invalid_reports)["execution_error_status"]) def test_workspace_mutation_invalidates_the_audit(self) -> None: result, reports = self._run( "--report-only", STUB_MUTATE_PATH=str(self.checkout / "README.md"), ) self.assertEqual(1, result.returncode) manifest = self._manifest(reports) self.assertFalse(manifest["workspace_unchanged"]) self.assertEqual(1, manifest["overall_status"]) self.assertNotEqual( (reports / "workspace-state-start.tsv").read_text(), (reports / "workspace-state-end.tsv").read_text(), ) def test_manifest_and_checksums_cover_only_current_run_reports(self) -> None: reports = self.temporary_root / "reports" reports.mkdir() (reports / "stale-malformed.json").write_text("{", encoding="utf-8") result, reports = self._run("--report-only") self.assertEqual(0, result.returncode, result.stderr) manifest = self._manifest(reports) self.assertEqual('stub-"safe', manifest["toolbox_fingerprint"]) self.assertNotIn("stale-malformed.json", manifest["reports"]) self.assertEqual([], manifest["missing_reports"]) checksums = (reports / "report-sha256.txt").read_text(encoding="utf-8") self.assertNotIn("stale-malformed.json", checksums) verification = subprocess.run( ["sha256sum", "--check", "report-sha256.txt"], cwd=reports, check=False, capture_output=True, text=True, ) self.assertEqual(0, verification.returncode, verification.stderr) checksummed_paths = { line.split(" ", 1)[1] for line in checksums.splitlines() if " " in line } self.assertEqual(set(manifest["reports"]), checksummed_paths) class SecurityAuditContainerRunnerTests(unittest.TestCase): def setUp(self) -> None: self._temporary_directory = tempfile.TemporaryDirectory( prefix="govoplan-audit-runner-" ) root = Path(self._temporary_directory.name) self.stub_bin = root / "bin" self.stub_bin.mkdir() self.docker_log = root / "docker.jsonl" docker_stub = self.stub_bin / "docker" docker_stub.write_text( textwrap.dedent( """\ #!/usr/bin/env python3 import json import os from pathlib import Path import sys arguments = sys.argv[1:] with Path(os.environ["DOCKER_STUB_LOG"]).open( "a", encoding="utf-8" ) as handle: handle.write(json.dumps(arguments) + "\\n") if arguments and arguments[0] == "version": print("26.1.0") if arguments[:2] == ["container", "inspect"]: print(os.environ["DOCKER_STUB_MOUNTS"]) """ ), encoding="utf-8", ) docker_stub.chmod(0o755) self.environment = os.environ.copy() self.environment.update( { "DOCKER_STUB_LOG": str(self.docker_log), "DOCKER_STUB_MOUNTS": "[]", "PATH": f"{self.stub_bin}:{self.environment['PATH']}", } ) def tearDown(self) -> None: self._temporary_directory.cleanup() def _run( self, *, scope: str, actions_mounts: object | None = None, ) -> tuple[subprocess.CompletedProcess[str], list[list[str]]]: self.docker_log.write_text("", encoding="utf-8") environment = self.environment.copy() if actions_mounts is None: environment.pop("GITEA_ACTIONS", None) else: environment["GITEA_ACTIONS"] = "true" environment["DOCKER_STUB_MOUNTS"] = json.dumps(actions_mounts) result = subprocess.run( [ "bash", str(CONTAINER_RUNNER), "--mode", "quick", "--scope", scope, ], cwd=META_ROOT, env=environment, check=False, capture_output=True, text=True, ) commands = [ json.loads(line) for line in self.docker_log.read_text(encoding="utf-8").splitlines() ] return result, commands def test_gitea_job_shares_only_its_workspace_mount(self) -> None: mounts = [ { "Type": "bind", "Source": "/var/run/docker.sock", "Destination": "/var/run/docker.sock", "RW": True, }, { "Type": "volume", "Name": "govoplan-actions-workspace", "Destination": str(META_ROOT.parent), "RW": True, }, { "Type": "volume", "Name": "govoplan-actions-environment", "Destination": "/var/run/act", "RW": True, }, ] container_id = subprocess.run( ["hostname"], check=True, capture_output=True, text=True, ).stdout.strip() for scope in ("current", "govoplan"): with self.subTest(scope=scope): result, commands = self._run(scope=scope, actions_mounts=mounts) self.assertEqual(0, result.returncode, result.stderr) run_command = next( command for command in commands if command[0] == "run" ) inspect_command = next( command for command in commands if command[:2] == ["container", "inspect"] ) self.assertEqual(container_id, inspect_command[-1]) mount_index = run_command.index("--mount") self.assertEqual( ( "type=volume,source=govoplan-actions-workspace," f"target={META_ROOT.parent}" ), run_command[mount_index + 1], ) self.assertNotIn("--volumes-from", run_command) self.assertNotIn("-v", run_command) self.assertNotIn("/var/run/docker.sock", " ".join(run_command)) self.assertNotIn("/var/run/act", " ".join(run_command)) repository_roots = [ value for value in run_command if value.startswith("GOVOPLAN_REPOS_ROOT=") ] expected_roots = ( [f"GOVOPLAN_REPOS_ROOT={META_ROOT.parent}"] if scope == "govoplan" else [] ) self.assertEqual(expected_roots, repository_roots) def test_non_actions_runner_keeps_the_scoped_bind_mount(self) -> None: result, commands = self._run(scope="govoplan") self.assertEqual(0, result.returncode, result.stderr) run_command = next(command for command in commands if command[0] == "run") bind_index = run_command.index("-v") self.assertEqual( f"{META_ROOT.parent}:/workspace", run_command[bind_index + 1], ) self.assertNotIn("--mount", run_command) if __name__ == "__main__": unittest.main()