feat(opencti): new OpenCTI threat-intelligence integration

Add a marketplace integration for the OpenCTI platform (GraphQL API,
compatible with OpenCTI 5.x/6.x), built on the Python pycti client and
run from a remote engine.

26 commands: observables (list/create/delete/field update/add/remove),
indicators (list/create/update/field add/remove, types), incidents
(list/create/delete, types) with an OCSF ingestion mapper, relationships
(list/create/delete), and reference data (organizations, labels, marking
definitions, external references).

- Ingestion: get_incidents to an OCSF finding mapper + an OpenCTI Incident type.
- Auth: user API key (Bearer) via pycti; requires pip install pycti on
  the engine host.
- Scripts are self-contained (INTEGRATION_SECRETS/INTEGRATION_INPUTS in,
  JSON out) following the established marketplace pattern.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
2026-06-29 23:24:57 +02:00
parent 0263619e30
commit c406edb5e7
29 changed files with 2353 additions and 0 deletions
@@ -0,0 +1,63 @@
import json, os, sys
S = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
I = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
def out(value):
print(json.dumps(value, default=str))
def fail(message, **extra):
payload = {"error": message}
payload.update(extra)
print(json.dumps(payload, default=str))
sys.exit(1)
def as_bool(v):
return v if isinstance(v, bool) else str(v).lower() in ("1", "true", "yes")
def as_int(v, default=None):
try:
return int(v)
except (TypeError, ValueError):
return default
def as_list(v):
if isinstance(v, list):
return v
if v in (None, ""):
return []
return [x.strip() for x in str(v).split(",") if x.strip()]
try:
from pycti import OpenCTIApiClient
except ImportError as e:
fail("The 'pycti' Python library is required for the OpenCTI integration. "
"Install it on the execution host (engine): pip install pycti", detail=str(e))
def client():
base = str(S.get("base_url", "")).strip().rstrip("/")
api_key = S.get("api_key") or (S.get("credentials") or {}).get("password")
return OpenCTIApiClient(base, api_key, ssl_verify=not as_bool(S.get("insecure")), log_level="error")
KEY_TO_CTI_NAME = {"description": "x_opencti_description", "score": "x_opencti_score"}
field = I.get("field")
if field not in KEY_TO_CTI_NAME:
fail("Invalid field was provided.")
key = KEY_TO_CTI_NAME[field]
try:
result = client().stix_cyber_observable.update_field(id=I.get("id"), key=key, value=I.get("value"))
except Exception as e:
fail("Can't update observable field in OpenCTI.", detail=str(e))
out({"id": result.get("id"), "message": "Observable updated."})