titan-iac/services/quality/zap-baseline-configmap.yaml

# services/quality/zap-baseline-configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: zap-baseline-config
  namespace: quality
data:
  targets.txt: |
    https://bstein.dev
    https://chat.ai.bstein.dev
    https://sso.bstein.dev
    https://auth.bstein.dev
    https://scm.bstein.dev
    https://ci.bstein.dev
    https://registry.bstein.dev
    https://quality.bstein.dev
    https://secret.bstein.dev
    https://vault.bstein.dev
    https://logs.bstein.dev
    https://metrics.bstein.dev
    https://stream.bstein.dev
    https://pegasus.bstein.dev
    https://tasks.bstein.dev
    https://cloud.bstein.dev
    https://notes.bstein.dev
    https://budget.bstein.dev
    https://money.bstein.dev
    https://health.bstein.dev
    https://agent.bstein.dev
    https://cassandra.bstein.dev
    https://veles.bstein.dev
    https://matrix.live.bstein.dev
    https://live.bstein.dev
    https://wolf.bstein.dev
    https://mail.bstein.dev
    https://recovery.bstein.dev
    https://backup.bstein.dev
  run_zap_baseline.sh: |
    #!/usr/bin/env bash
    set -euo pipefail

    target_file="${ZAP_TARGET_FILE:-/zap/config/targets.txt}"
    pushgateway="${PUSHGATEWAY_URL:-http://platform-quality-gateway.monitoring.svc.cluster.local:9091}"
    job_name="${QUALITY_GATE_JOB_NAME:-platform-security-zap}"
    spider_minutes="${ZAP_SPIDER_MINUTES:-1}"
    max_scan_minutes="${ZAP_MAX_SCAN_MINUTES:-5}"
    work_dir="${ZAP_WORK_DIR:-/zap/wrk}"
    metrics_path="${work_dir}/zap-baseline.prom"
    reports_dir="${work_dir}/reports"

    mkdir -p "${reports_dir}"
    : > "${metrics_path}"

    python3 - <<'PY' > "${metrics_path}"
    print("# TYPE platform_zap_baseline_target_up gauge")
    print("# TYPE platform_zap_baseline_scan_status gauge")
    print("# TYPE platform_zap_baseline_alerts_total gauge")
    print("# TYPE platform_zap_baseline_target_health_percent gauge")
    print("# TYPE platform_zap_baseline_last_run_timestamp_seconds gauge")
    PY

    while IFS= read -r raw_target || [ -n "${raw_target}" ]; do
      target="$(printf '%s' "${raw_target}" | sed 's/#.*//' | xargs)"
      [ -n "${target}" ] || continue

      host="$(python3 - "${target}" <<'PY'
    import sys
    from urllib.parse import urlparse

    print(urlparse(sys.argv[1]).hostname or "unknown")
    PY
      )"
      safe_host="$(printf '%s' "${host}" | tr -c 'A-Za-z0-9_.-' '_')"
      json_report="${reports_dir}/${safe_host}.json"
      html_report="${reports_dir}/${safe_host}.html"
      markdown_report="${reports_dir}/${safe_host}.md"
      xml_report="${reports_dir}/${safe_host}.xml"

      set +e
      zap-baseline.py \
        -t "${target}" \
        -m "${spider_minutes}" \
        -T "${max_scan_minutes}" \
        -I \
        -J "${json_report}" \
        -r "${html_report}" \
        -w "${markdown_report}" \
        -x "${xml_report}"
      zap_rc=$?
      set -e

      python3 - "${target}" "${host}" "${json_report}" "${zap_rc}" >> "${metrics_path}" <<'PY'
    import json
    import sys
    import time
    from pathlib import Path

    target, host, report_path, zap_rc_raw = sys.argv[1:5]
    try:
        zap_rc = int(zap_rc_raw)
    except ValueError:
        zap_rc = 3

    def esc(value: str) -> str:
        return value.replace("\\", "\\\\").replace("\n", "\\n").replace('"', '\\"')

    def labels(extra: dict[str, str]) -> str:
        base = {"target": target, "host": host}
        base.update(extra)
        return "{" + ",".join(f'{key}="{esc(value)}"' for key, value in base.items()) + "}"

    risk_counts = {"high": 0, "medium": 0, "low": 0, "informational": 0, "unknown": 0}
    completed = Path(report_path).exists()
    if completed:
        try:
            payload = json.loads(Path(report_path).read_text(encoding="utf-8"))
        except json.JSONDecodeError:
            completed = False
            payload = {}
        for site in payload.get("site", []) if isinstance(payload, dict) else []:
            if not isinstance(site, dict):
                continue
            alerts = site.get("alerts", [])
            if not isinstance(alerts, list):
                continue
            for alert in alerts:
                if not isinstance(alert, dict):
                    continue
                risk_code = str(alert.get("riskcode") or "").strip()
                risk_name = str(alert.get("risk") or alert.get("riskdesc") or "unknown").split()[0].lower()
                risk = {
                    "3": "high",
                    "2": "medium",
                    "1": "low",
                    "0": "informational",
                }.get(risk_code, risk_name)
                if risk not in risk_counts:
                    risk = "unknown"
                instances = alert.get("instances")
                count = len(instances) if isinstance(instances, list) and instances else 1
                risk_counts[risk] += count

    if not completed or zap_rc not in {0, 1, 2}:
        status = "error"
    elif risk_counts["high"] > 0:
        status = "fail"
    elif risk_counts["medium"] > 0 or risk_counts["low"] > 0:
        status = "warn"
    else:
        status = "ok"

    health = 0 if status in {"error", "fail"} else 85 if status == "warn" else 100
    print(f"platform_zap_baseline_target_up{labels({})} {1 if completed else 0}")
    for candidate in ["ok", "warn", "fail", "error"]:
        print(f'platform_zap_baseline_scan_status{labels({"status": candidate})} {1 if status == candidate else 0}')
    for risk, count in risk_counts.items():
        print(f'platform_zap_baseline_alerts_total{labels({"risk": risk})} {count}')
    print(f"platform_zap_baseline_target_health_percent{labels({})} {health}")
    print(f"platform_zap_baseline_last_run_timestamp_seconds{labels({})} {int(time.time())}")
    PY
    done < "${target_file}"

    python3 - "${pushgateway}" "${job_name}" "${metrics_path}" <<'PY'
    import sys
    import urllib.request
    from pathlib import Path

    pushgateway, job_name, metrics_path = sys.argv[1:4]
    payload = Path(metrics_path).read_bytes()
    url = f"{pushgateway.rstrip('/')}/metrics/job/{job_name}"
    request = urllib.request.Request(
        url,
        data=payload,
        method="PUT",
        headers={"Content-Type": "text/plain"},
    )
    with urllib.request.urlopen(request, timeout=15) as response:
        if response.status >= 400:
            raise SystemExit(f"pushgateway returned HTTP {response.status}")
    PY