359 lines
12 KiB
Python
359 lines
12 KiB
Python
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 _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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|