Spaces:
Running
Running
File size: 3,773 Bytes
9633e56 b6b72a4 9633e56 b6b72a4 9633e56 808f060 b6b72a4 9633e56 6f3b0c2 9633e56 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """Prove model handling without inference runtime evidence is not killed."""
from __future__ import annotations
import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from smoke_content_ai import assets
ROOT = Path(__file__).resolve().parents[1]
DETECTIONS = Path("/var/lib/lumi-eggcracker/detections")
MODES = ("read", "hash", "map", "copy", "read", "hash", "map", "copy", "read", "map")
def main() -> int:
if os.geteuid() != 0:
raise SystemExit("content benign matrix must run as root")
parser = argparse.ArgumentParser()
parser.add_argument("--assets-manifest", required=True, type=Path)
parser.add_argument("--user", required=True)
parser.add_argument("--repetitions", type=int, default=300)
parser.add_argument("--output", required=True, type=Path)
args = parser.parse_args()
if (
args.repetitions != 300
or args.output.exists()
or args.output.is_symlink()
or not args.output.parent.is_dir()
):
raise SystemExit("output must be new and repetitions must equal 300")
result: dict[str, object] = {"classes": {}, "kills": 0, "result": "FAIL"}
try:
_runner, model, _manifest = assets(args.assets_manifest)
source_fixture = ROOT / "tests" / "fixtures" / "benign_model_handler.py"
with tempfile.TemporaryDirectory(prefix="lumi-content-benign-", dir="/tmp") as raw:
root = Path(raw)
os.chmod(root, 0o777)
fixture = root / "benign_model_handler.py"
shutil.copyfile(source_fixture, fixture)
os.chmod(fixture, 0o644)
for index in range(args.repetitions):
mode = MODES[index % len(MODES)]
destination = root / f"copy-{index}"
before = set(DETECTIONS.glob("*.json"))
try:
process = subprocess.run(
[
"/usr/sbin/runuser",
"-u",
args.user,
"--",
sys.executable,
str(fixture),
str(model),
mode,
str(destination),
],
capture_output=True,
text=True,
check=False,
timeout=30 if mode == "copy" else 10,
)
finally:
destination.unlink(missing_ok=True)
if process.returncode:
raise RuntimeError(
process.stderr.strip() or "benign model handler was interrupted"
)
new = set(DETECTIONS.glob("*.json")) - before
if new:
result["kills"] = int(result["kills"]) + len(new)
raise RuntimeError("benign model handler received an autonomous detection")
classes = result["classes"]
assert isinstance(classes, dict)
classes[mode] = int(classes.get(mode, 0)) + 1
result["result"] = "PASS"
args.output.write_text(json.dumps(result, sort_keys=True) + "\n", encoding="utf-8")
return 0
except (OSError, RuntimeError, subprocess.SubprocessError, json.JSONDecodeError) as error:
result["error"] = str(error)
if not args.output.exists():
args.output.write_text(json.dumps(result, sort_keys=True) + "\n", encoding="utf-8")
raise SystemExit(f"content benign matrix failed: {error}") from error
if __name__ == "__main__":
raise SystemExit(main())
|