riposte-marketplace/integrations/opencti/scripts/get_observables.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")


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",
}
OPENCTI_TO_TYPE = {
    "User-Account": "Account", "Domain-Name": "Domain", "Email-Addr": "Email", "StixFile": "File",
    "X-OpenCTI-Hostname": "Host", "IPv4-Addr": "IP", "IPv6-Addr": "IPv6",
    "Windows-Registry-Key": "Registry Key", "Url": "URL",
}


def build_observable_list(observable_list):
    result = []
    if "ALL" in observable_list:
        result = ["User-Account", "Domain-Name", "Email-Addr", "StixFile", "X-OpenCTI-Hostname",
                  "IPv4-Addr", "IPv6-Addr", "Windows-Registry-Key", "Url"]
    else:
        result = [TYPES_TO_OPENCTI.get(o.lower(), o) for o in observable_list]
    return result


def get_observables(client, observable_types, score=None, limit=500, last_run_id=None, search="", get_all=False):
    observable_type = build_observable_list(observable_types)
    filters = {
        "mode": "and",
        "filters": [{"key": "entity_type", "values": observable_type, "operator": "eq", "mode": "or"}],
        "filterGroups": [],
    }
    if score:
        filters["filters"].append({"key": "x_opencti_score", "values": score, "operator": "eq", "mode": "or"})
    return client.stix_cyber_observable.list(
        after=last_run_id, first=limit, withPagination=True, getAll=get_all, filters=filters, search=search
    )


observable_types = as_list(I.get("observable_types", "ALL")) or ["ALL"]
last_run_id = I.get("last_run_id")
limit = as_int(I.get("limit", 50), 50)
start = as_int(I.get("score_start", 0), 0)
end = as_int(I.get("score_end", 100), 100)
score = I.get("score")
search = I.get("search", "")
get_all = as_bool(I.get("all_results", "false"))

scores = None
if score:
    if str(score).lower() == "unknown":
        scores = [None]
    elif str(score).isdigit():
        scores = [score]
    else:
        fail("Invalid score was provided.")
elif start or end:
    scores = [str(n) for n in range(start, end + 1)]

try:
    raw = get_observables(
        client(),
        observable_types,
        score=scores,
        limit=limit,
        last_run_id=last_run_id,
        search=search,
        get_all=get_all,
    )
except Exception as e:
    fail("Can't list observables from OpenCTI.", detail=str(e))

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

normalized = []
for o in observables_list:
    normalized.append({
        "type": OPENCTI_TO_TYPE.get(o["entity_type"], o["entity_type"]),
        "value": o.get("observable_value"),
        "id": o.get("id"),
        "createdBy": o.get("createdBy").get("id") if o.get("createdBy") else None,
        "score": o.get("x_opencti_score"),
        "description": o.get("x_opencti_description"),
        "labels": [l.get("value") for l in o.get("objectLabel", [])],
        "marking": [m.get("definition") for m in o.get("objectMarking", [])],
        "externalReferences": o.get("externalReferences"),
    })

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