Harden security audit result contracts
This commit is contained in:
262
tests/test_security_audit_wrapper.py
Normal file
262
tests/test_security_audit_wrapper.py
Normal file
@@ -0,0 +1,262 @@
|
||||
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"
|
||||
|
||||
|
||||
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 _run(
|
||||
self, *arguments: str, **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",
|
||||
"quick",
|
||||
"--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_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)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user