riposte-marketplace/integrations/opencti/scripts/get_indicators.py

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")


def get_indicators(client, search, created_by, creator, created_after, created_before,
                   valid_until_after, valid_until_before, valid_from_after, valid_from_before,
                   indicator_types, label, limit, last_run_id, get_all):
    filters = {"mode": "and", "filters": [], "filterGroups": []}
    if label:
        filters["filters"].append({"key": "objectLabel", "values": [label], "operator": "eq", "mode": "or"})
    if created_by:
        filters["filters"].append({"key": "createdBy", "values": [created_by], "operator": "eq", "mode": "or"})
    if creator:
        filters["filters"].append({"key": "creator_id", "values": [creator], "operator": "eq"})
    if indicator_types:
        filters["filters"].append({"key": "indicator_types", "values": indicator_types, "operator": "eq", "mode": "or"})
    if created_after:
        filters["filters"].append({"key": "created_at", "values": [created_after], "operator": "gt"})
    if created_before:
        filters["filters"].append({"key": "created_at", "values": [created_before], "operator": "lt"})
    if valid_from_after:
        filters["filters"].append({"key": "valid_from", "values": [valid_from_after], "operator": "gt"})
    if valid_from_before:
        filters["filters"].append({"key": "valid_from", "values": [valid_from_before], "operator": "lt"})
    if valid_until_after:
        filters["filters"].append({"key": "valid_until", "values": [valid_until_after], "operator": "gt"})
    if valid_until_before:
        filters["filters"].append({"key": "valid_until", "values": [valid_until_before], "operator": "lt"})
    return client.indicator.list(
        after=last_run_id, first=limit, withPagination=True, getAll=get_all, filters=filters, search=search
    )


search = I.get("search")
created_by = I.get("created_by")
creator = I.get("creator")
created_after = I.get("created_after")
created_before = I.get("created_before")
valid_until_after = I.get("valid_until_after")
valid_until_before = I.get("valid_until_before")
valid_from_after = I.get("valid_from_after")
valid_from_before = I.get("valid_from_before")
indicator_types = as_list(I.get("indicator_types"))
label = I.get("label_id")
limit = as_int(I.get("limit", 50), 50)
last_run_id = I.get("last_run_id")
get_all = as_bool(I.get("all_results", "false"))

try:
    raw = get_indicators(
        client(), search, created_by, creator, created_after, created_before,
        valid_until_after, valid_until_before, valid_from_after, valid_from_before,
        indicator_types, label, limit, last_run_id, get_all,
    )
except Exception as e:
    fail("Can't list indicators from OpenCTI.", detail=str(e))

last_run = None if get_all else raw.get("pagination", {}).get("endCursor")
indicators_list = raw if get_all else raw.get("entities", [])

normalized = []
for ind in indicators_list:
    normalized.append({
        "id": ind.get("id"),
        "name": ind.get("name"),
        "description": ind.get("description"),
        "pattern": ind.get("pattern"),
        "validFrom": ind.get("valid_from"),
        "validUntil": ind.get("valid_until"),
        "score": ind.get("x_opencti_score"),
        "confidence": ind.get("confidence"),
        "createdBy": ind.get("createdBy").get("name") if ind.get("createdBy") else "",
        "creators": [c.get("name") for c in ind.get("creators", [])],
        "labels": [l.get("value") for l in ind.get("objectLabel", [])],
        "indicatorTypes": ind.get("indicator_types"),
        "created": ind.get("created"),
        "updatedAt": ind.get("updated_at"),
    })

out({"indicators": normalized, "lastRunID": last_run})