Guillaume BOURGEOIS
c406edb5e7
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>
115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
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")
|
|
|
|
|
|
indicator_id = I.get("id")
|
|
name = I.get("name")
|
|
description = I.get("description")
|
|
confidence = I.get("confidence")
|
|
score = I.get("score")
|
|
valid_from = I.get("valid_from")
|
|
valid_until = I.get("valid_until")
|
|
indicator_types = I.get("indicator_types")
|
|
label_id = I.get("label_id")
|
|
marking_id = I.get("marking_id")
|
|
external_references_id = I.get("external_references_id")
|
|
|
|
update_fields = []
|
|
if name:
|
|
update_fields.append({"key": "name", "value": name})
|
|
if description:
|
|
update_fields.append({"key": "description", "value": description})
|
|
if confidence:
|
|
update_fields.append({"key": "confidence", "value": as_int(confidence)})
|
|
if score:
|
|
update_fields.append({"key": "x_opencti_score", "value": as_int(score)})
|
|
if valid_from:
|
|
update_fields.append({"key": "valid_from", "value": valid_from})
|
|
if valid_until:
|
|
update_fields.append({"key": "valid_until", "value": valid_until})
|
|
if indicator_types:
|
|
update_fields.append({"key": "indicator_types", "value": indicator_types.split(",")})
|
|
if label_id:
|
|
update_fields.append({"key": "objectLabel", "value": label_id.split(",")})
|
|
if marking_id:
|
|
update_fields.append({"key": "objectMarking", "value": marking_id.split(",")})
|
|
if external_references_id:
|
|
update_fields.append({"key": "externalReferences", "value": external_references_id.split(",")})
|
|
|
|
mutation = """
|
|
mutation IndicatorEditionOverviewFieldPatchMutation($id: ID!, $input: [EditInput!]!, $commitMessage: String, $references: [String]) {
|
|
indicatorFieldPatch(id: $id, input: $input, commitMessage: $commitMessage, references: $references) {
|
|
id name confidence description valid_from valid_until x_opencti_score indicator_types
|
|
}
|
|
}
|
|
"""
|
|
|
|
variables = {
|
|
"id": indicator_id,
|
|
"input": update_fields,
|
|
"commitMessage": None,
|
|
"references": None,
|
|
}
|
|
|
|
try:
|
|
result = client().query(mutation, variables)
|
|
except Exception as e:
|
|
fail("Can't update indicator in OpenCTI.", detail=str(e))
|
|
|
|
patched = result.get("data", {}).get("indicatorFieldPatch")
|
|
if patched:
|
|
out({
|
|
"id": patched.get("id"),
|
|
"name": patched.get("name"),
|
|
"validFrom": valid_from,
|
|
"validUntil": valid_until,
|
|
"message": "Indicator updated.",
|
|
})
|
|
else:
|
|
fail("Can't update indicator in OpenCTI.")
|