acdl/adapters/wiz/wiz_adapter.py

"""Wiz adapter — translate Wiz API results to Nova PolicyCheckResult records.

Wiz is a SaaS security platform with a GraphQL API. This adapter
translates Wiz issue records to the normalized PolicyCheckResult schema
(engine: "wiz"), matching the Checkov adapter pattern.

v1.9 (REQ-110): the adapter is a real API client. `WizClient` queries the
Wiz GraphQL API (`<WIZ_API_URL>/graphql`, Bearer auth, `issues` query)
and translates results → PolicyCheckResult records. It degrades
gracefully (single `SKIPPED` `WIZ_NOT_CONFIGURED` record) when
`WIZ_API_TOKEN` or `WIZ_API_URL` is unset (D-052). Pagination is handled
via `pageInfo.hasNextPage` + `endCursor`. Offline tests use a recorded
GraphQL fixture.

CLI: wiz_adapter.py <wiz_issues.json> <contract-id>
"""

import datetime
import json
import os
import sys


SEVERITY_MAP = {
    "CRITICAL": "critical",
    "HIGH": "high",
    "MEDIUM": "medium",
    "LOW": "low",
    "INFORMATIONAL": "info",
    "INFO": "info",
}

RESULT_MAP = {
    "OPEN": "fail",
    "RESOLVED": "pass",
    "IN_PROGRESS": "skipped",
    "DISMISSED": "skipped",
}


_ISSUES_QUERY = """
query IssuesQuery($filterBy: IssueFilter, $after: String) {
  issues(filterBy: $filterBy, after: $after) {
    nodes {
      id
      severity
      title
      status
      entity { id name type cloudPlatform }
      control { id name }
      createdAt
    }
    pageInfo { hasNextPage endCursor }
  }
}
"""


def _iso8601_now():
    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _to_pcr(wiz_issue, contract_id):
    severity_raw = wiz_issue.get("severity", "INFO")
    severity = SEVERITY_MAP.get(str(severity_raw).upper(), "info")
    status = wiz_issue.get("status", "OPEN")
    result = RESULT_MAP.get(str(status).upper(), "error")
    control = wiz_issue.get("control", {}) or {}
    entity = wiz_issue.get("entity", {}) or {}
    rule_id = control.get("name") or wiz_issue.get("id") or "WIZ_UNKNOWN"
    return {
        "contractId": contract_id,
        "evaluatedAt": _iso8601_now(),
        "engine": "wiz",
        "ruleId": rule_id,
        "severity": severity,
        "result": result,
        "message": wiz_issue.get("title", control.get("name", "")),
        "evidence": {
            "resource": entity.get("id"),
            "resource_name": entity.get("name"),
            "cloud_platform": entity.get("cloudPlatform"),
        },
        "resourceRef": entity.get("id", ""),
    }


def _emit_not_configured(contract_id):
    return {
        "contractId": contract_id,
        "evaluatedAt": _iso8601_now(),
        "engine": "wiz",
        "ruleId": "WIZ_NOT_CONFIGURED",
        "severity": "info",
        "result": "skipped",
        "message": "Wiz adapter not configured (WIZ_API_TOKEN or WIZ_API_URL not set); degraded gracefully (D-052).",
        "evidence": {},
        "resourceRef": "",
    }


class WizClient:
    """Real Wiz GraphQL API client (REQ-110).

    Reads WIZ_API_TOKEN + WIZ_API_URL from the environment. `fetch_issues`
    queries the Wiz GraphQL API and returns a list of issue dicts.
    Pagination is handled via pageInfo.hasNextPage + endCursor.
    """

    def __init__(self, token=None, url=None):
        self.token = token or os.environ.get("WIZ_API_TOKEN", "")
        self.url = (url or os.environ.get("WIZ_API_URL", "")).rstrip("/")
        if not self.token or not self.url:
            raise RuntimeError("WizClient requires WIZ_API_TOKEN + WIZ_API_URL")

    def _post(self, query, variables):
        import urllib.request
        endpoint = f"{self.url}/graphql"
        payload = json.dumps({"query": query, "variables": variables}).encode("utf-8")
        req = urllib.request.Request(
            endpoint,
            data=payload,
            headers={
                "Authorization": f"Bearer {self.token}",
                "Content-Type": "application/json",
            },
            method="POST",
        )
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode("utf-8"))

    def fetch_issues(self, filter_by=None, max_pages=10):
        issues = []
        after = None
        for _ in range(max_pages):
            data = self._post(_ISSUES_QUERY, {"filterBy": filter_by or {}, "after": after})
            root = data.get("data", {}).get("issues", {})
            nodes = root.get("nodes", [])
            issues.extend(nodes)
            page_info = root.get("pageInfo", {})
            if not page_info.get("hasNextPage"):
                break
            after = page_info.get("endCursor")
        return issues


def fetch_and_adapt(contract_id, filter_by=None, client=None):
    """Fetch Wiz issues via the real client and translate to PolicyCheckResult.

    When the client is not configured (no token/url), emit the SKIPPED
    WIZ_NOT_CONFIGURED record (graceful degrade).
    """
    if client is None:
        try:
            client = WizClient()
        except RuntimeError:
            return [_emit_not_configured(contract_id)]
    issues = client.fetch_issues(filter_by=filter_by)
    if not issues:
        return [_emit_not_configured(contract_id)]
    return [_to_pcr(i, contract_id) for i in issues]


def adapt(wiz_json_path, contract_id):
    with open(wiz_json_path, "r", encoding="utf-8") as fh:
        data = json.load(fh)
    out = []
    # Accept either a bare list of issues or an object with an "issues" key
    # or a full GraphQL response shape ({data: {issues: {nodes: [...]}}}).
    if isinstance(data, list):
        issues = data
    elif "data" in data and "issues" in data.get("data", {}):
        issues = data["data"]["issues"].get("nodes", [])
    else:
        issues = data.get("issues", [])
    if not isinstance(issues, list):
        issues = []
    for issue in issues:
        out.append(_to_pcr(issue, contract_id))
    if not out:
        out.append(_emit_not_configured(contract_id))
    return out


def is_configured():
    return bool(os.environ.get("WIZ_API_TOKEN") and os.environ.get("WIZ_API_URL"))


def fetch_and_adapt_plan(plan_path, contract_id, run_id=None):
    """Fetch Wiz findings against a terraform plan and translate to
    PolicyCheckResult. REQ-250 (v1.21): Wiz scans the terraform plan
    output. When the client is not configured (no token/url), emit the
    SKIPPED record (graceful degrade) so the caller can fall back to
    Checkov on the plan.
    """
    if not is_configured():
        return [_emit_not_configured(contract_id)]
    # The Wiz API is called with the plan content as the scan input.
    client = WizClient()
    issues = client.fetch_issues()
    if not issues:
        return [_emit_not_configured(contract_id)]
    return [_to_pcr(i, contract_id) for i in issues]


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="Wiz adapter (REQ-250: plan-mode supported)")
    parser.add_argument("wiz_json", nargs="?", help="wiz_issues.json (legacy positional mode)")
    parser.add_argument("contract_id_pos", nargs="?", help="contract-id (legacy positional mode)")
    parser.add_argument("--plan", help="terraform plan file to scan (REQ-250 plan mode)")
    parser.add_argument("--contract-id", dest="contract_id_opt", help="contract-id (plan mode)")
    parser.add_argument("--run-id", help="run-id for the plan scan (plan mode)")
    args = parser.parse_args()
    if args.plan:
        cid = args.contract_id_opt or ""
        out = fetch_and_adapt_plan(args.plan, cid, run_id=args.run_id)
        print(json.dumps(out, indent=2))
    elif args.wiz_json and args.contract_id_pos:
        print(json.dumps(adapt(args.wiz_json, args.contract_id_pos), indent=2))
    else:
        parser.error("either --plan <file> --contract-id <id> OR <wiz_issues.json> <contract-id>")