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") try: result = client().incident.create( name=I.get("name"), incident_type=I.get("incident_type"), confidence=as_int(I.get("confidence", 50), 50), severity=I.get("severity"), description=I.get("description"), source=I.get("source"), objective=I.get("objective"), createdBy=I.get("created_by"), first_seen=I.get("first_seen"), last_seen=I.get("last_seen"), objectLabel=I.get("label_id"), objectMarking=I.get("marking_id"), externalReferences=I.get("external_references_id"), ) except Exception as e: fail("Can't create incident in OpenCTI.", detail=str(e)) if result.get("id"): out({"id": result.get("id"), "message": f"Incident created. id: {result.get('id')}"}) else: fail("Incident creation failed.")