94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
"""Retry transient advisory-service failures without accepting incomplete audits."""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
NETWORK_ERRORS = (
|
|
"ReadTimeout",
|
|
"Read timed out",
|
|
"ConnectTimeout",
|
|
"ConnectionError",
|
|
"ConnectionResetError",
|
|
"NameResolutionError",
|
|
"Temporary failure in name resolution",
|
|
"MaxRetryError",
|
|
)
|
|
|
|
|
|
def main():
|
|
cache = Path(tempfile.gettempdir()) / "frame-flow-advisory-cache"
|
|
cache.mkdir(mode=0o700, exist_ok=True)
|
|
timeout = int(os.getenv("CI_AUDIT_TIMEOUT", "60"))
|
|
attempts = int(os.getenv("CI_AUDIT_ATTEMPTS", "3"))
|
|
if not 15 <= timeout <= 120 or not 1 <= attempts <= 3:
|
|
raise SystemExit("Invalid audit timeout or attempt limit")
|
|
print(
|
|
"Advisory host: pypi.org; HTTP cache enabled; socket timeout:",
|
|
timeout,
|
|
flush=True,
|
|
)
|
|
command = [
|
|
sys.executable,
|
|
"-m",
|
|
"pip_audit",
|
|
"-r",
|
|
"requirements.lock",
|
|
"--no-deps",
|
|
"--disable-pip",
|
|
"--vulnerability-service",
|
|
"pypi",
|
|
"--timeout",
|
|
str(timeout),
|
|
"--cache-dir",
|
|
str(cache),
|
|
"--format",
|
|
"json",
|
|
]
|
|
for attempt in range(1, attempts + 1):
|
|
result = subprocess.run(command, capture_output=True, text=True, timeout=300)
|
|
try:
|
|
report = json.loads(result.stdout)
|
|
dependencies = report["dependencies"]
|
|
complete = bool(dependencies) and all(
|
|
isinstance(d.get("vulns"), list) and not d.get("skip_reason")
|
|
for d in dependencies
|
|
)
|
|
except (ValueError, KeyError, TypeError, AttributeError):
|
|
dependencies, complete = [], False
|
|
vulnerabilities = [
|
|
(d["name"], v["id"]) for d in dependencies for v in d.get("vulns", [])
|
|
]
|
|
if vulnerabilities:
|
|
for name, advisory in vulnerabilities:
|
|
print("Vulnerable dependency:", name, advisory, flush=True)
|
|
raise SystemExit("Audit found vulnerabilities; deployment blocked")
|
|
if result.returncode == 0 and complete:
|
|
print(
|
|
"Complete audit passed:",
|
|
len(dependencies),
|
|
"dependencies; no known vulnerabilities",
|
|
flush=True,
|
|
)
|
|
return
|
|
transient = any(marker in result.stderr for marker in NETWORK_ERRORS)
|
|
# Do not print raw network exceptions: proxy URLs can contain credentials.
|
|
print(
|
|
"Audit attempt",
|
|
attempt,
|
|
"failed:",
|
|
"network error" if transient else "incomplete/error result",
|
|
flush=True,
|
|
)
|
|
if not transient or attempt == attempts:
|
|
raise SystemExit("Audit unsuccessful; deployment blocked")
|
|
time.sleep(5 * attempt)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|