536f6249a6
Some checks failed
Docs CI / lint-and-preview (push) Has been cancelled
- Created CycloneDX and SPDX SBOM files for both reachable and unreachable images. - Added symbols.json detailing function entry and sink points in the WordPress code. - Included runtime traces for function calls in both reachable and unreachable scenarios. - Developed OpenVEX files indicating vulnerability status and justification for both cases. - Updated README for evaluator harness to guide integration with scanner output.
84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Run egress probes from the sealed-mode Docker network."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from typing import List
|
|
|
|
DEFAULT_TARGETS = [
|
|
"https://example.com",
|
|
"https://www.cloudflare.com",
|
|
"https://releases.stella-ops.org/healthz",
|
|
]
|
|
|
|
|
|
def run_probe(image: str, network: str, target: str, timeout: int) -> dict:
|
|
cmd: List[str] = [
|
|
"docker",
|
|
"run",
|
|
"--rm",
|
|
"--network",
|
|
network,
|
|
image,
|
|
"-fsS",
|
|
"--max-time",
|
|
str(timeout),
|
|
target,
|
|
]
|
|
started = time.monotonic()
|
|
proc = subprocess.run(cmd, capture_output=True, text=True)
|
|
duration = time.monotonic() - started
|
|
status = "blocked" if proc.returncode != 0 else "connected"
|
|
return {
|
|
"target": target,
|
|
"status": status,
|
|
"durationSeconds": round(duration, 3),
|
|
"exitCode": proc.returncode,
|
|
"command": " ".join(shlex.quote(part) for part in cmd),
|
|
"stdout": proc.stdout.strip(),
|
|
"stderr": proc.stderr.strip(),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--network", required=True, help="Docker network to join (compose project network)")
|
|
parser.add_argument("--image", default="curlimages/curl:8.6.0", help="Container image providing curl")
|
|
parser.add_argument("--timeout", type=int, default=10, help="Curl max-time for each probe (seconds)")
|
|
parser.add_argument("--output", required=True, help="Path to write JSON results")
|
|
parser.add_argument("targets", nargs="*", help="Override target URLs")
|
|
args = parser.parse_args()
|
|
|
|
targets = args.targets or DEFAULT_TARGETS
|
|
results = [run_probe(args.image, args.network, target, args.timeout) for target in targets]
|
|
passed = all(result["status"] == "blocked" for result in results)
|
|
payload = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"network": args.network,
|
|
"image": args.image,
|
|
"targets": results,
|
|
"passed": passed,
|
|
}
|
|
|
|
os.makedirs(os.path.dirname(args.output), exist_ok=True)
|
|
with open(args.output, "w", encoding="utf-8") as handle:
|
|
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
|
handle.write("\n")
|
|
|
|
return 0 if passed else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except Exception as exc: # pragma: no cover
|
|
print(f"egress probe failed: {exc}", file=sys.stderr)
|
|
sys.exit(2)
|