517 lines
18 KiB
Python
517 lines
18 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"
|
|
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()
|