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>
98 lines
3.1 KiB
Python
98 lines
3.1 KiB
Python
import json, os, sys
|
|
import io
|
|
|
|
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")
|
|
|
|
|
|
TYPES_TO_OPENCTI = {
|
|
"account": "User-Account", "domain": "Domain-Name", "email": "Email-Addr",
|
|
"file-md5": "StixFile", "file-sha1": "StixFile", "file-sha256": "StixFile", "file": "StixFile",
|
|
"host": "X-OpenCTI-Hostname", "ip": "IPv4-Addr", "ipv6": "IPv6-Addr",
|
|
"registry key": "Windows-Registry-Key", "url": "Url",
|
|
}
|
|
FILE_TYPES = {
|
|
"file-md5": "file.hashes.md5", "file-sha1": "file.hashes.sha-1", "file-sha256": "file.hashes.sha-256",
|
|
}
|
|
|
|
|
|
observable_type = I.get("type")
|
|
data = {"type": TYPES_TO_OPENCTI.get(observable_type.lower(), observable_type), "value": I.get("value")}
|
|
if observable_type == "Registry Key":
|
|
data["key"] = I.get("value")
|
|
if observable_type == "Account":
|
|
data["account_login"] = I.get("value")
|
|
|
|
simple_observable_key = None
|
|
simple_observable_value = None
|
|
if "file" in observable_type.lower():
|
|
simple_observable_key = FILE_TYPES.get(observable_type.lower(), observable_type)
|
|
simple_observable_value = I.get("value")
|
|
|
|
try:
|
|
sys.stdout = io.StringIO()
|
|
try:
|
|
result = client().stix_cyber_observable.create(
|
|
simple_observable_key=simple_observable_key,
|
|
simple_observable_value=simple_observable_value,
|
|
type=observable_type,
|
|
createdBy=I.get("created_by"),
|
|
objectMarking=I.get("marking_id"),
|
|
objectLabel=I.get("label_id"),
|
|
externalReferences=I.get("external_references_id"),
|
|
simple_observable_description=I.get("description"),
|
|
x_opencti_score=as_int(I.get("score", "50"), 50),
|
|
observableData=data,
|
|
createIndicator=as_bool(I.get("create_indicator", "false")),
|
|
)
|
|
finally:
|
|
sys.stdout = sys.__stdout__
|
|
except Exception as e:
|
|
sys.stdout = sys.__stdout__
|
|
fail("Can't create observable in OpenCTI.", detail=str(e))
|
|
|
|
out({"id": result.get("id"), "value": I.get("value"), "type": observable_type})
|