From ecc301d95df5b9b189538347aaf8583766aa23dc Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sun, 12 Jul 2026 00:15:10 +0200 Subject: [PATCH] feat(threatconnect): new ThreatConnect threat-intel integration ThreatConnect API v3, 9 commands: list/get/create/delete indicators, list/get/ create groups, add indicator tag. HMAC-SHA256 signed auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) --- integrations/threatconnect/manifest.yaml | 113 ++++++++++++++++++ .../scripts/add_indicator_tag.py | 56 +++++++++ .../threatconnect/scripts/create_group.py | 57 +++++++++ .../threatconnect/scripts/create_indicator.py | 63 ++++++++++ .../threatconnect/scripts/delete_indicator.py | 55 +++++++++ .../threatconnect/scripts/get_group.py | 52 ++++++++ .../threatconnect/scripts/get_indicator.py | 52 ++++++++ .../threatconnect/scripts/list_groups.py | 53 ++++++++ .../threatconnect/scripts/list_indicators.py | 53 ++++++++ .../threatconnect/scripts/test_connection.py | 49 ++++++++ 10 files changed, 603 insertions(+) create mode 100644 integrations/threatconnect/manifest.yaml create mode 100644 integrations/threatconnect/scripts/add_indicator_tag.py create mode 100644 integrations/threatconnect/scripts/create_group.py create mode 100644 integrations/threatconnect/scripts/create_indicator.py create mode 100644 integrations/threatconnect/scripts/delete_indicator.py create mode 100644 integrations/threatconnect/scripts/get_group.py create mode 100644 integrations/threatconnect/scripts/get_indicator.py create mode 100644 integrations/threatconnect/scripts/list_groups.py create mode 100644 integrations/threatconnect/scripts/list_indicators.py create mode 100644 integrations/threatconnect/scripts/test_connection.py diff --git a/integrations/threatconnect/manifest.yaml b/integrations/threatconnect/manifest.yaml new file mode 100644 index 0000000..f0195d7 --- /dev/null +++ b/integrations/threatconnect/manifest.yaml @@ -0,0 +1,113 @@ +id: threatconnect +name: ThreatConnect +version: 1.0.0 +description: "ThreatConnect (API v3) — threat intelligence: query and read indicators and groups (TQL), create and delete indicators, create groups, and tag indicators. HMAC-SHA256 signed authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: list/get/create/delete indicators, list/get/create groups, add indicator tag." +category: threat_intel + +# Per-instance configuration. Requests are signed with HMAC-SHA256 using the +# access ID + secret key (header 'Authorization: TC :'). +config_schema: + properties: + base_url: + type: string + description: "ThreatConnect base URL (e.g. https://app.threatconnect.com)" + access_id: + type: string + description: "API access ID" + api_secret_key: + type: string + description: "API secret key" + x-soar-sensitive: true + required: + - base_url + - access_id + - api_secret_key + +commands: + - id: list_indicators + name: tc-list-indicators + description: "Query indicators using a TQL expression." + risk: read + inputs_schema: + properties: + tql: { type: string, description: "ThreatConnect Query Language expression (e.g. summary contains \"1.2.3.4\")" } + limit: { type: number, description: "Max indicators (default 50)" } + required: [] + outputs_schema: { properties: {} } + - id: get_indicator + name: tc-get-indicator + description: "Get a single indicator by ID." + risk: read + inputs_schema: + properties: + indicator_id: { type: string, description: "Indicator ID" } + required: [indicator_id] + outputs_schema: { properties: {} } + - id: create_indicator + name: tc-create-indicator + description: "Create an indicator (e.g. Address, Host, File, URL, EmailAddress)." + inputs_schema: + properties: + type: { type: string, description: "Indicator type: Address, Host, File, URL, or EmailAddress" } + summary: { type: string, description: "Indicator value (IP, domain, hash, URL, email)" } + owner_name: { type: string, description: "Owner (organization) name" } + rating: { type: number, description: "Threat rating 0-5 (optional)" } + confidence: { type: number, description: "Confidence 0-100 (optional)" } + required: [type, summary] + outputs_schema: { properties: {} } + - id: delete_indicator + name: tc-delete-indicator + description: "Delete an indicator by ID." + inputs_schema: + properties: + indicator_id: { type: string, description: "Indicator ID" } + required: [indicator_id] + outputs_schema: { properties: {} } + - id: list_groups + name: tc-list-groups + description: "Query groups (incidents, threats, campaigns, ...) using a TQL expression." + risk: read + inputs_schema: + properties: + tql: { type: string, description: "TQL expression" } + limit: { type: number, description: "Max groups (default 50)" } + required: [] + outputs_schema: { properties: {} } + - id: get_group + name: tc-get-group + description: "Get a single group by ID." + risk: read + inputs_schema: + properties: + group_id: { type: string, description: "Group ID" } + required: [group_id] + outputs_schema: { properties: {} } + - id: create_group + name: tc-create-group + description: "Create a group (Incident, Threat, Campaign, Adversary, ...)." + inputs_schema: + properties: + type: { type: string, description: "Group type: Incident, Threat, Campaign, Adversary, Document, ..." } + name: { type: string, description: "Group name" } + owner_name: { type: string, description: "Owner (organization) name" } + required: [type, name] + outputs_schema: { properties: {} } + - id: add_indicator_tag + name: tc-add-indicator-tag + description: "Add a tag to an indicator." + inputs_schema: + properties: + indicator_id: { type: string, description: "Indicator ID" } + tag: { type: string, description: "Tag name" } + required: [indicator_id, tag] + outputs_schema: { properties: {} } + + - id: test_connection + name: tc-test-connection + description: "Verify connectivity and the signed credentials (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/threatconnect/scripts/add_indicator_tag.py b/integrations/threatconnect/scripts/add_indicator_tag.py new file mode 100644 index 0000000..f119ad6 --- /dev/null +++ b/integrations/threatconnect/scripts/add_indicator_tag.py @@ -0,0 +1,56 @@ +import json, os, sys, time, hmac, hashlib, base64 +import urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, api_path, cfg, body=None): + # api_path MUST include the leading /api/v3/... and any ?query string. + base = str(cfg.get("base_url", "")).rstrip("/") + url = base + api_path + timestamp = str(int(time.time())) + to_sign = api_path + ":" + method.upper() + ":" + timestamp + secret = str(cfg.get("api_secret_key", "")).encode("utf-8") + signature = base64.b64encode(hmac.new(secret, to_sign.encode("utf-8"), hashlib.sha256).digest()).decode("utf-8") + authorization = "TC " + str(cfg.get("access_id", "")) + ":" + signature + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": authorization, "Timestamp": timestamp, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + indicator_id = inputs.get("indicator_id") + tag = inputs.get("tag") + if not indicator_id: + raise Exception("indicator_id is required") + if not tag: + raise Exception("tag is required") + q = lambda v: urllib.parse.quote(str(v), safe="") + path = "/api/v3/indicators/" + q(indicator_id) + body = {"tags": {"data": [{"name": tag}], "mode": "append"}} + return request("PUT", path, cfg, body) + + +_run(main) diff --git a/integrations/threatconnect/scripts/create_group.py b/integrations/threatconnect/scripts/create_group.py new file mode 100644 index 0000000..f7386ab --- /dev/null +++ b/integrations/threatconnect/scripts/create_group.py @@ -0,0 +1,57 @@ +import json, os, sys, time, hmac, hashlib, base64 +import urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, api_path, cfg, body=None): + # api_path MUST include the leading /api/v3/... and any ?query string. + base = str(cfg.get("base_url", "")).rstrip("/") + url = base + api_path + timestamp = str(int(time.time())) + to_sign = api_path + ":" + method.upper() + ":" + timestamp + secret = str(cfg.get("api_secret_key", "")).encode("utf-8") + signature = base64.b64encode(hmac.new(secret, to_sign.encode("utf-8"), hashlib.sha256).digest()).decode("utf-8") + authorization = "TC " + str(cfg.get("access_id", "")) + ":" + signature + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": authorization, "Timestamp": timestamp, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + type_ = inputs.get("type") + name = inputs.get("name") + if not type_: + raise Exception("type is required") + if not name: + raise Exception("name is required") + owner_name = inputs.get("owner_name") + body = {"type": type_, "name": name} + if owner_name: + body["ownerName"] = owner_name + return request("POST", "/api/v3/groups", cfg, body) + + +_run(main) diff --git a/integrations/threatconnect/scripts/create_indicator.py b/integrations/threatconnect/scripts/create_indicator.py new file mode 100644 index 0000000..4eb67bb --- /dev/null +++ b/integrations/threatconnect/scripts/create_indicator.py @@ -0,0 +1,63 @@ +import json, os, sys, time, hmac, hashlib, base64 +import urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, api_path, cfg, body=None): + # api_path MUST include the leading /api/v3/... and any ?query string. + base = str(cfg.get("base_url", "")).rstrip("/") + url = base + api_path + timestamp = str(int(time.time())) + to_sign = api_path + ":" + method.upper() + ":" + timestamp + secret = str(cfg.get("api_secret_key", "")).encode("utf-8") + signature = base64.b64encode(hmac.new(secret, to_sign.encode("utf-8"), hashlib.sha256).digest()).decode("utf-8") + authorization = "TC " + str(cfg.get("access_id", "")) + ":" + signature + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": authorization, "Timestamp": timestamp, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + type_ = inputs.get("type") + summary = inputs.get("summary") + if not type_: + raise Exception("type is required") + if not summary: + raise Exception("summary is required") + owner_name = inputs.get("owner_name") + rating = inputs.get("rating") + confidence = inputs.get("confidence") + body = {"type": type_, "summary": summary} + if owner_name: + body["ownerName"] = owner_name + if rating not in (None, ""): + body["rating"] = int(rating) + if confidence not in (None, ""): + body["confidence"] = int(confidence) + return request("POST", "/api/v3/indicators", cfg, body) + + +_run(main) diff --git a/integrations/threatconnect/scripts/delete_indicator.py b/integrations/threatconnect/scripts/delete_indicator.py new file mode 100644 index 0000000..961feaf --- /dev/null +++ b/integrations/threatconnect/scripts/delete_indicator.py @@ -0,0 +1,55 @@ +import json, os, sys, time, hmac, hashlib, base64 +import urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, api_path, cfg, body=None): + # api_path MUST include the leading /api/v3/... and any ?query string. + base = str(cfg.get("base_url", "")).rstrip("/") + url = base + api_path + timestamp = str(int(time.time())) + to_sign = api_path + ":" + method.upper() + ":" + timestamp + secret = str(cfg.get("api_secret_key", "")).encode("utf-8") + signature = base64.b64encode(hmac.new(secret, to_sign.encode("utf-8"), hashlib.sha256).digest()).decode("utf-8") + authorization = "TC " + str(cfg.get("access_id", "")) + ":" + signature + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": authorization, "Timestamp": timestamp, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + indicator_id = inputs.get("indicator_id") + if not indicator_id: + raise Exception("indicator_id is required") + q = lambda v: urllib.parse.quote(str(v), safe="") + path = "/api/v3/indicators/" + q(indicator_id) + result = request("DELETE", path, cfg) + if not result: + return {"ok": True, "deleted": indicator_id} + return result + + +_run(main) diff --git a/integrations/threatconnect/scripts/get_group.py b/integrations/threatconnect/scripts/get_group.py new file mode 100644 index 0000000..bb28677 --- /dev/null +++ b/integrations/threatconnect/scripts/get_group.py @@ -0,0 +1,52 @@ +import json, os, sys, time, hmac, hashlib, base64 +import urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, api_path, cfg, body=None): + # api_path MUST include the leading /api/v3/... and any ?query string. + base = str(cfg.get("base_url", "")).rstrip("/") + url = base + api_path + timestamp = str(int(time.time())) + to_sign = api_path + ":" + method.upper() + ":" + timestamp + secret = str(cfg.get("api_secret_key", "")).encode("utf-8") + signature = base64.b64encode(hmac.new(secret, to_sign.encode("utf-8"), hashlib.sha256).digest()).decode("utf-8") + authorization = "TC " + str(cfg.get("access_id", "")) + ":" + signature + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": authorization, "Timestamp": timestamp, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + group_id = inputs.get("group_id") + if not group_id: + raise Exception("group_id is required") + q = lambda v: urllib.parse.quote(str(v), safe="") + path = "/api/v3/groups/" + q(group_id) + return request("GET", path, cfg) + + +_run(main) diff --git a/integrations/threatconnect/scripts/get_indicator.py b/integrations/threatconnect/scripts/get_indicator.py new file mode 100644 index 0000000..97e74d3 --- /dev/null +++ b/integrations/threatconnect/scripts/get_indicator.py @@ -0,0 +1,52 @@ +import json, os, sys, time, hmac, hashlib, base64 +import urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, api_path, cfg, body=None): + # api_path MUST include the leading /api/v3/... and any ?query string. + base = str(cfg.get("base_url", "")).rstrip("/") + url = base + api_path + timestamp = str(int(time.time())) + to_sign = api_path + ":" + method.upper() + ":" + timestamp + secret = str(cfg.get("api_secret_key", "")).encode("utf-8") + signature = base64.b64encode(hmac.new(secret, to_sign.encode("utf-8"), hashlib.sha256).digest()).decode("utf-8") + authorization = "TC " + str(cfg.get("access_id", "")) + ":" + signature + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": authorization, "Timestamp": timestamp, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + indicator_id = inputs.get("indicator_id") + if not indicator_id: + raise Exception("indicator_id is required") + q = lambda v: urllib.parse.quote(str(v), safe="") + path = "/api/v3/indicators/" + q(indicator_id) + return request("GET", path, cfg) + + +_run(main) diff --git a/integrations/threatconnect/scripts/list_groups.py b/integrations/threatconnect/scripts/list_groups.py new file mode 100644 index 0000000..ed91558 --- /dev/null +++ b/integrations/threatconnect/scripts/list_groups.py @@ -0,0 +1,53 @@ +import json, os, sys, time, hmac, hashlib, base64 +import urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, api_path, cfg, body=None): + # api_path MUST include the leading /api/v3/... and any ?query string. + base = str(cfg.get("base_url", "")).rstrip("/") + url = base + api_path + timestamp = str(int(time.time())) + to_sign = api_path + ":" + method.upper() + ":" + timestamp + secret = str(cfg.get("api_secret_key", "")).encode("utf-8") + signature = base64.b64encode(hmac.new(secret, to_sign.encode("utf-8"), hashlib.sha256).digest()).decode("utf-8") + authorization = "TC " + str(cfg.get("access_id", "")) + ":" + signature + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": authorization, "Timestamp": timestamp, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + tql = inputs.get("tql") + limit = inputs.get("limit") + params = {"resultLimit": int(limit or 50)} + if tql: + params["tql"] = tql + path = "/api/v3/groups?" + urllib.parse.urlencode(params) + return request("GET", path, cfg) + + +_run(main) diff --git a/integrations/threatconnect/scripts/list_indicators.py b/integrations/threatconnect/scripts/list_indicators.py new file mode 100644 index 0000000..8339ba5 --- /dev/null +++ b/integrations/threatconnect/scripts/list_indicators.py @@ -0,0 +1,53 @@ +import json, os, sys, time, hmac, hashlib, base64 +import urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, api_path, cfg, body=None): + # api_path MUST include the leading /api/v3/... and any ?query string. + base = str(cfg.get("base_url", "")).rstrip("/") + url = base + api_path + timestamp = str(int(time.time())) + to_sign = api_path + ":" + method.upper() + ":" + timestamp + secret = str(cfg.get("api_secret_key", "")).encode("utf-8") + signature = base64.b64encode(hmac.new(secret, to_sign.encode("utf-8"), hashlib.sha256).digest()).decode("utf-8") + authorization = "TC " + str(cfg.get("access_id", "")) + ":" + signature + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": authorization, "Timestamp": timestamp, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + tql = inputs.get("tql") + limit = inputs.get("limit") + params = {"resultLimit": int(limit or 50)} + if tql: + params["tql"] = tql + path = "/api/v3/indicators?" + urllib.parse.urlencode(params) + return request("GET", path, cfg) + + +_run(main) diff --git a/integrations/threatconnect/scripts/test_connection.py b/integrations/threatconnect/scripts/test_connection.py new file mode 100644 index 0000000..34326c3 --- /dev/null +++ b/integrations/threatconnect/scripts/test_connection.py @@ -0,0 +1,49 @@ +import json, os, sys, time, hmac, hashlib, base64 +import urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, api_path, cfg, body=None): + # api_path MUST include the leading /api/v3/... and any ?query string. + base = str(cfg.get("base_url", "")).rstrip("/") + url = base + api_path + timestamp = str(int(time.time())) + to_sign = api_path + ":" + method.upper() + ":" + timestamp + secret = str(cfg.get("api_secret_key", "")).encode("utf-8") + signature = base64.b64encode(hmac.new(secret, to_sign.encode("utf-8"), hashlib.sha256).digest()).decode("utf-8") + authorization = "TC " + str(cfg.get("access_id", "")) + ":" + signature + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": authorization, "Timestamp": timestamp, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + path = "/api/v3/indicators?" + urllib.parse.urlencode({"resultLimit": 1}) + request("GET", path, cfg) + return {"ok": True} + + +_run(main)