Files
ansible-testing/methodologies/zap/scripts/normalize.py
T

111 lines
4.9 KiB
Python

#!/usr/bin/env python3
"""Normalize ZAP baseline alerts and TLS preflight findings with ASVS mappings."""
import argparse
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from jsonschema import Draft202012Validator, FormatChecker
RISK = {"0": "info", "1": "low", "2": "medium", "3": "high", "4": "critical"}
def standards(mapping: dict[str, list[str]], finding_id: str) -> list[dict[str, str]]:
return [
{"framework": "OWASP ASVS", "version": "5.0", "control": control}
for control in mapping.get(finding_id, [])
]
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("zap_json", type=Path)
parser.add_argument("tls_json", type=Path)
parser.add_argument("output", type=Path)
parser.add_argument("--target", required=True)
parser.add_argument("--mapping", type=Path, default=Path("methodologies/zap/asvs-mapping.json"))
parser.add_argument("--schema", type=Path, default=Path("schemas/test-report.schema.json"))
args = parser.parse_args()
zap = json.loads(args.zap_json.read_text(encoding="utf-8"))
tls = json.loads(args.tls_json.read_text(encoding="utf-8"))
mapping = json.loads(args.mapping.read_text(encoding="utf-8"))
results = []
for site in zap.get("site", []):
for alert in site.get("alerts", []):
finding_id = str(alert.get("pluginid", alert.get("alertRef", "ZAP-UNKNOWN")))
instances = alert.get("instances", [])
observed = "; ".join(str(item.get("uri", "")) for item in instances[:5])
results.append(
{
"id": f"ZAP-{finding_id}",
"title": str(alert.get("alert", "ZAP finding")),
"description": str(alert.get("desc", "")),
"status": "failed",
"severity": RISK.get(str(alert.get("riskcode", "0")), "info"),
"category": "Web application security",
"expected": "No ZAP alert",
"observed": observed or str(alert.get("evidence", "")),
"remediation": str(alert.get("solution", "Review and remediate the finding.")),
"standards": standards(mapping, finding_id),
"evidence": [{"type": "text", "name": "ZAP alert", "value": observed or str(alert)}],
}
)
for finding in tls.get("findings", []):
finding_id = str(finding["id"])
results.append(
{
"id": finding_id,
"title": str(finding["title"]),
"description": str(finding.get("description", "")),
"status": "failed",
"severity": str(finding.get("severity", "medium")),
"category": "TLS configuration",
"expected": "Current TLS protocol, cipher, and certificate configuration",
"observed": str(finding.get("evidence", ""))[-2000:],
"remediation": str(finding.get("remediation", "")),
"standards": standards(mapping, finding_id),
"evidence": [{"type": "text", "name": "TLS preflight", "value": str(finding.get("evidence", ""))[-2000:]}],
}
)
counts = dict.fromkeys(("passed", "failed", "errors", "skipped", "review"), 0)
counts["failed"] = len(results)
now = datetime.now(timezone.utc).isoformat()
report = {
"schema_version": "1.0.0",
"run": {
"id": os.environ.get("CI_PIPELINE_ID", now),
"started_at": now,
"source": "gitlab" if os.environ.get("CI") else "local",
"pipeline_url": os.environ.get("CI_PIPELINE_URL", ""),
"commit_sha": os.environ.get("CI_COMMIT_SHA", ""),
},
"project": {
"id": os.environ.get("TEST_PROJECT_ID", "demo-ubuntu-weak"),
"name": os.environ.get("TEST_PROJECT_NAME", "Ubuntu Weak Target Demonstration"),
"environment": os.environ.get("TEST_ENVIRONMENT", "test"),
"customer": os.environ.get("TEST_CUSTOMER", "Internal"),
"location": os.environ.get("TEST_LOCATION", "testserv"),
},
"tool": {"id": "zaproxy", "name": "OWASP ZAP with TLS preflight", "adapter_version": "1.0.0"},
"target": {"id": "demo-web", "type": "web_application", "address": args.target, "groups": ["web_applications"]},
"summary": {"total": len(results), **counts, "score": 0},
"results": results,
"raw_artifacts": [str(args.zap_json), str(args.tls_json)],
}
schema = json.loads(args.schema.read_text(encoding="utf-8"))
Draft202012Validator(schema, format_checker=FormatChecker()).validate(report)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
print(f"Normalized {len(results)} web findings")
return 0
if __name__ == "__main__":
raise SystemExit(main())