mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 11:49:28 +00:00
111 lines
4.8 KiB
Python
111 lines
4.8 KiB
Python
"""Offline packaged CLI/registry and disposable Git hook validation, not detector qualification."""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from openultrasast.model.scan import ScanBudget
|
|
from openultrasast.push.eligibility import _DEFAULT, load_registry
|
|
from openultrasast.push.runner import _provenance
|
|
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--out", type=Path, required=True)
|
|
parser.add_argument("--ops", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
out = args.out
|
|
out.mkdir(exist_ok=False)
|
|
|
|
|
|
def run(argv, **kwargs):
|
|
return subprocess.run(argv, capture_output=True, timeout=20, **kwargs)
|
|
|
|
|
|
help_result = run(["ousast", "pre-push", "--help"])
|
|
assert help_result.returncode == 0, help_result.stderr
|
|
(out / "help.txt").write_bytes(help_result.stdout)
|
|
current = _provenance(out, ScanBudget(max_model_calls=0, max_regions=500, order_by_evidence=True))
|
|
registry = load_registry(current=current)
|
|
assert _DEFAULT.is_file() and _DEFAULT.stat().st_size > 0
|
|
assert registry.status == "NO-GO" and not registry.capabilities, registry
|
|
results = []
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
tmp = Path(tmp)
|
|
for prior_exit in (0, 7):
|
|
root = tmp / str(prior_exit)
|
|
root.mkdir()
|
|
|
|
def git(*args, root=root):
|
|
result = run(["git", "-C", str(root), *args])
|
|
assert result.returncode == 0, result.stderr
|
|
return result.stdout.decode().strip()
|
|
|
|
git("init")
|
|
git("config", "user.name", "Packaged smoke")
|
|
git("config", "user.email", "[email protected]")
|
|
(root / "api.js").write_text("function endpoint(req) { return eval(req.body.value); }\n")
|
|
assert len((root / "api.js").read_bytes()) > 0
|
|
git("add", "api.js")
|
|
git("commit", "-m", "Authored interface smoke, not a security control")
|
|
head = git("rev-parse", "HEAD")
|
|
remote = root / "remote.git"
|
|
git("init", "--bare", str(remote))
|
|
git("remote", "add", "origin", str(remote))
|
|
git("config", "core.hooksPath", ".custom-hooks")
|
|
installer = run([str(args.ops / "install-pre-push"), str(root)])
|
|
assert installer.returncode == 0, installer.stderr
|
|
hook = root / ".custom-hooks/pre-push"
|
|
before_hook = hook.read_bytes()
|
|
refused = run([str(args.ops / "install-pre-push"), str(root)])
|
|
assert refused.returncode != 0 and hook.read_bytes() == before_hook
|
|
prior = hook.with_name("pre-push.before-ousast")
|
|
prior.write_text('#!/bin/sh\nprintf "%s\\n" "$@" > prior-args\ncat > prior-stdin\nexit ' + str(prior_exit) + "\n")
|
|
prior.chmod(0o755)
|
|
before = ((root / ".git/index").read_bytes(), (root / "api.js").read_bytes(), head)
|
|
artifacts = out / str(prior_exit)
|
|
env = {**os.environ, "OUSAST_ARTIFACT_DIR": str(artifacts), "OUSAST_PUSH_DEADLINE": "0.1"}
|
|
result = run(["git", "-C", str(root), "push", "origin", "HEAD:refs/heads/smoke"], env=env)
|
|
assert (result.returncode == 0) == (prior_exit == 0), result.stderr
|
|
assert (root / "prior-stdin").read_text() == f"HEAD {head} refs/heads/smoke {'0' * 40}\n"
|
|
assert (root / "prior-args").read_text().splitlines() == ["origin", str(remote)]
|
|
assert before == ((root / ".git/index").read_bytes(), (root / "api.js").read_bytes(), git("rev-parse", "HEAD"))
|
|
assert git("config", "core.hooksPath") == ".custom-hooks"
|
|
saved = list(artifacts.glob("*.json"))
|
|
assert len(saved) == 1
|
|
artifact = json.loads(saved[0].read_bytes())
|
|
assert not artifact["admission"]["defects"]
|
|
assert artifact["result"]["coverage_status"] != "complete_within_scope"
|
|
original = prior.read_bytes()
|
|
hook.unlink()
|
|
prior.rename(hook)
|
|
assert hook.read_bytes() == original
|
|
after = run(["git", "-C", str(root), "push", "origin", "HEAD:refs/heads/restored"], env=env)
|
|
assert (after.returncode == 0) == (prior_exit == 0)
|
|
results.append(
|
|
dict(
|
|
prior_exit=prior_exit,
|
|
push_exit=result.returncode,
|
|
removal_push_exit=after.returncode,
|
|
exact_stdin=True,
|
|
exact_remote_args=True,
|
|
live_state_preserved=True,
|
|
overwrite_refused=True,
|
|
coverage=artifact["result"]["coverage_status"],
|
|
source_bytes=len(before[1]),
|
|
terminal=(result.stdout + result.stderr).decode(),
|
|
)
|
|
)
|
|
receipt = dict(
|
|
uid=os.getuid(),
|
|
provenance=current,
|
|
registry_status=registry.status,
|
|
enabled=0,
|
|
no_model_configuration=True,
|
|
network="disabled by docker invocation",
|
|
cases=results,
|
|
)
|
|
(out / "receipt.json").write_text(json.dumps(receipt, indent=2) + "\n")
|
|
print(json.dumps(receipt))
|