feat(darktrace): new Darktrace NDR integration
Darktrace Threat Visualizer API, 7 commands: get/acknowledge model breaches, get devices, get/activate Antigena actions (autonomous-response containment). HMAC-SHA1 signed auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -0,0 +1,97 @@
|
||||
id: darktrace
|
||||
name: Darktrace
|
||||
version: 1.0.0
|
||||
description: "Darktrace (Threat Visualizer API) — network detection and autonomous response: read model breaches, acknowledge a breach, look up devices, and read/activate Antigena network actions (containment). HMAC-SHA1 signed authentication; stdlib-only, no extra Python dependencies."
|
||||
changelog: "1.0.0 — Initial release: get/acknowledge model breaches, get devices, get/activate Antigena actions."
|
||||
category: ndr
|
||||
|
||||
# Per-instance configuration. Requests are signed (HMAC-SHA1) with the public +
|
||||
# private API tokens (headers DTAPI-Token / DTAPI-Date / DTAPI-Signature).
|
||||
config_schema:
|
||||
properties:
|
||||
base_url:
|
||||
type: string
|
||||
description: "Darktrace master URL (e.g. https://darktrace.example.com)"
|
||||
public_token:
|
||||
type: string
|
||||
description: "Public API token"
|
||||
private_token:
|
||||
type: string
|
||||
description: "Private API token"
|
||||
x-soar-sensitive: true
|
||||
insecure:
|
||||
type: boolean
|
||||
description: "Trust any TLS certificate (not secure)"
|
||||
default: false
|
||||
required:
|
||||
- base_url
|
||||
- public_token
|
||||
- private_token
|
||||
|
||||
commands:
|
||||
- id: get_model_breaches
|
||||
name: darktrace-get-model-breaches
|
||||
description: "List model breaches (optionally since a time)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
min_score: { type: number, description: "Minimum breach score 0-1 (optional)" }
|
||||
from_time: { type: string, description: "Start time YYYY-MM-DD HH:MM:SS (optional)" }
|
||||
count: { type: number, description: "Max breaches (default 50)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_model_breach
|
||||
name: darktrace-get-model-breach
|
||||
description: "Get a single model breach by its PBID."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
pbid: { type: string, description: "Policy breach ID (PBID)" }
|
||||
required: [pbid]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: acknowledge_breach
|
||||
name: darktrace-acknowledge-breach
|
||||
description: "Acknowledge a model breach."
|
||||
inputs_schema:
|
||||
properties:
|
||||
pbid: { type: string, description: "Policy breach ID (PBID)" }
|
||||
required: [pbid]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_devices
|
||||
name: darktrace-get-devices
|
||||
description: "Look up devices (by IP or device ID)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
did: { type: string, description: "Device ID (optional)" }
|
||||
ip: { type: string, description: "IP address (optional)" }
|
||||
count: { type: number, description: "Max devices (default 50)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_antigena_actions
|
||||
name: darktrace-get-antigena-actions
|
||||
description: "List Antigena (autonomous response) actions."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
include_cleared: { type: boolean, description: "Include cleared actions (default false)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: activate_antigena
|
||||
name: darktrace-activate-antigena
|
||||
description: "Activate an Antigena action (containment — enforce autonomous response)."
|
||||
inputs_schema:
|
||||
properties:
|
||||
codeid: { type: string, description: "Antigena action code ID" }
|
||||
duration: { type: number, description: "Duration in seconds (optional)" }
|
||||
required: [codeid]
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: test_connection
|
||||
name: darktrace-test-connection
|
||||
description: "Verify connectivity and the signed credentials (used by the Test button)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
@@ -0,0 +1,81 @@
|
||||
import json, os, sys, hmac, hashlib, ssl, datetime
|
||||
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 _ctx(cfg):
|
||||
if cfg.get("insecure"):
|
||||
c = ssl.create_default_context()
|
||||
c.check_hostname = False
|
||||
c.verify_mode = ssl.CERT_NONE
|
||||
return c
|
||||
return None
|
||||
|
||||
|
||||
def _sign(request_path, cfg, date):
|
||||
public_token = str(cfg.get("public_token", ""))
|
||||
private_token = str(cfg.get("private_token", "")).encode("utf-8")
|
||||
signed = request_path + "\n" + public_token + "\n" + date
|
||||
return hmac.new(private_token, signed.encode("utf-8"), hashlib.sha1).hexdigest()
|
||||
|
||||
|
||||
def request(method, path, cfg, params=None, body=None):
|
||||
request_path = path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
request_path += "?" + urllib.parse.urlencode(clean)
|
||||
date = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")
|
||||
sig = _sign(request_path, cfg, date)
|
||||
url = str(cfg.get("base_url", "")).rstrip("/") + request_path
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {
|
||||
"DTAPI-Token": str(cfg.get("public_token", "")),
|
||||
"DTAPI-Date": date,
|
||||
"DTAPI-Signature": sig,
|
||||
"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, context=_ctx(cfg)) 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)
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
pbid = inputs.get("pbid")
|
||||
if pbid in (None, ""):
|
||||
raise Exception("pbid is required")
|
||||
|
||||
result = request(
|
||||
"POST",
|
||||
"/modelbreaches/" + q(pbid) + "/acknowledge",
|
||||
cfg,
|
||||
params={"acknowledge": "true"},
|
||||
)
|
||||
return result if result else {"ok": True}
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,78 @@
|
||||
import json, os, sys, hmac, hashlib, ssl, datetime
|
||||
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 _ctx(cfg):
|
||||
if cfg.get("insecure"):
|
||||
c = ssl.create_default_context()
|
||||
c.check_hostname = False
|
||||
c.verify_mode = ssl.CERT_NONE
|
||||
return c
|
||||
return None
|
||||
|
||||
|
||||
def _sign(request_path, cfg, date):
|
||||
public_token = str(cfg.get("public_token", ""))
|
||||
private_token = str(cfg.get("private_token", "")).encode("utf-8")
|
||||
signed = request_path + "\n" + public_token + "\n" + date
|
||||
return hmac.new(private_token, signed.encode("utf-8"), hashlib.sha1).hexdigest()
|
||||
|
||||
|
||||
def request(method, path, cfg, params=None, body=None):
|
||||
request_path = path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
request_path += "?" + urllib.parse.urlencode(clean)
|
||||
date = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")
|
||||
sig = _sign(request_path, cfg, date)
|
||||
url = str(cfg.get("base_url", "")).rstrip("/") + request_path
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {
|
||||
"DTAPI-Token": str(cfg.get("public_token", "")),
|
||||
"DTAPI-Date": date,
|
||||
"DTAPI-Signature": sig,
|
||||
"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, context=_ctx(cfg)) 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):
|
||||
codeid = inputs.get("codeid")
|
||||
duration = inputs.get("duration")
|
||||
if codeid in (None, ""):
|
||||
raise Exception("codeid is required")
|
||||
|
||||
params = {"codeid": codeid, "action": "activate"}
|
||||
if duration not in (None, ""):
|
||||
params["duration"] = int(duration)
|
||||
|
||||
result = request("POST", "/antigena/action", cfg, params=params)
|
||||
return result if result else {"ok": True}
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,80 @@
|
||||
import json, os, sys, hmac, hashlib, ssl, datetime
|
||||
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 _ctx(cfg):
|
||||
if cfg.get("insecure"):
|
||||
c = ssl.create_default_context()
|
||||
c.check_hostname = False
|
||||
c.verify_mode = ssl.CERT_NONE
|
||||
return c
|
||||
return None
|
||||
|
||||
|
||||
def _sign(request_path, cfg, date):
|
||||
public_token = str(cfg.get("public_token", ""))
|
||||
private_token = str(cfg.get("private_token", "")).encode("utf-8")
|
||||
signed = request_path + "\n" + public_token + "\n" + date
|
||||
return hmac.new(private_token, signed.encode("utf-8"), hashlib.sha1).hexdigest()
|
||||
|
||||
|
||||
def request(method, path, cfg, params=None, body=None):
|
||||
request_path = path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
request_path += "?" + urllib.parse.urlencode(clean)
|
||||
date = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")
|
||||
sig = _sign(request_path, cfg, date)
|
||||
url = str(cfg.get("base_url", "")).rstrip("/") + request_path
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {
|
||||
"DTAPI-Token": str(cfg.get("public_token", "")),
|
||||
"DTAPI-Date": date,
|
||||
"DTAPI-Signature": sig,
|
||||
"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, context=_ctx(cfg)) 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 _as_bool(v):
|
||||
if isinstance(v, bool):
|
||||
return v
|
||||
if isinstance(v, str):
|
||||
return v.strip().lower() in ("1", "true", "yes", "on")
|
||||
return bool(v)
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
include_cleared = _as_bool(inputs.get("include_cleared", False))
|
||||
|
||||
params = {"includecleared": "true"} if include_cleared else {}
|
||||
|
||||
return request("GET", "/antigena", cfg, params=params)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,78 @@
|
||||
import json, os, sys, hmac, hashlib, ssl, datetime
|
||||
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 _ctx(cfg):
|
||||
if cfg.get("insecure"):
|
||||
c = ssl.create_default_context()
|
||||
c.check_hostname = False
|
||||
c.verify_mode = ssl.CERT_NONE
|
||||
return c
|
||||
return None
|
||||
|
||||
|
||||
def _sign(request_path, cfg, date):
|
||||
public_token = str(cfg.get("public_token", ""))
|
||||
private_token = str(cfg.get("private_token", "")).encode("utf-8")
|
||||
signed = request_path + "\n" + public_token + "\n" + date
|
||||
return hmac.new(private_token, signed.encode("utf-8"), hashlib.sha1).hexdigest()
|
||||
|
||||
|
||||
def request(method, path, cfg, params=None, body=None):
|
||||
request_path = path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
request_path += "?" + urllib.parse.urlencode(clean)
|
||||
date = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")
|
||||
sig = _sign(request_path, cfg, date)
|
||||
url = str(cfg.get("base_url", "")).rstrip("/") + request_path
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {
|
||||
"DTAPI-Token": str(cfg.get("public_token", "")),
|
||||
"DTAPI-Date": date,
|
||||
"DTAPI-Signature": sig,
|
||||
"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, context=_ctx(cfg)) 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):
|
||||
did = inputs.get("did")
|
||||
ip = inputs.get("ip")
|
||||
count = inputs.get("count")
|
||||
|
||||
params = {"count": int(count or 50)}
|
||||
if did not in (None, ""):
|
||||
params["did"] = did
|
||||
if ip not in (None, ""):
|
||||
params["ip"] = ip
|
||||
|
||||
return request("GET", "/devices", cfg, params=params)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,72 @@
|
||||
import json, os, sys, hmac, hashlib, ssl, datetime
|
||||
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 _ctx(cfg):
|
||||
if cfg.get("insecure"):
|
||||
c = ssl.create_default_context()
|
||||
c.check_hostname = False
|
||||
c.verify_mode = ssl.CERT_NONE
|
||||
return c
|
||||
return None
|
||||
|
||||
|
||||
def _sign(request_path, cfg, date):
|
||||
public_token = str(cfg.get("public_token", ""))
|
||||
private_token = str(cfg.get("private_token", "")).encode("utf-8")
|
||||
signed = request_path + "\n" + public_token + "\n" + date
|
||||
return hmac.new(private_token, signed.encode("utf-8"), hashlib.sha1).hexdigest()
|
||||
|
||||
|
||||
def request(method, path, cfg, params=None, body=None):
|
||||
request_path = path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
request_path += "?" + urllib.parse.urlencode(clean)
|
||||
date = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")
|
||||
sig = _sign(request_path, cfg, date)
|
||||
url = str(cfg.get("base_url", "")).rstrip("/") + request_path
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {
|
||||
"DTAPI-Token": str(cfg.get("public_token", "")),
|
||||
"DTAPI-Date": date,
|
||||
"DTAPI-Signature": sig,
|
||||
"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, context=_ctx(cfg)) 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):
|
||||
pbid = inputs.get("pbid")
|
||||
if pbid in (None, ""):
|
||||
raise Exception("pbid is required")
|
||||
|
||||
return request("GET", "/modelbreaches", cfg, params={"pbid": pbid})
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,78 @@
|
||||
import json, os, sys, hmac, hashlib, ssl, datetime
|
||||
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 _ctx(cfg):
|
||||
if cfg.get("insecure"):
|
||||
c = ssl.create_default_context()
|
||||
c.check_hostname = False
|
||||
c.verify_mode = ssl.CERT_NONE
|
||||
return c
|
||||
return None
|
||||
|
||||
|
||||
def _sign(request_path, cfg, date):
|
||||
public_token = str(cfg.get("public_token", ""))
|
||||
private_token = str(cfg.get("private_token", "")).encode("utf-8")
|
||||
signed = request_path + "\n" + public_token + "\n" + date
|
||||
return hmac.new(private_token, signed.encode("utf-8"), hashlib.sha1).hexdigest()
|
||||
|
||||
|
||||
def request(method, path, cfg, params=None, body=None):
|
||||
request_path = path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
request_path += "?" + urllib.parse.urlencode(clean)
|
||||
date = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")
|
||||
sig = _sign(request_path, cfg, date)
|
||||
url = str(cfg.get("base_url", "")).rstrip("/") + request_path
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {
|
||||
"DTAPI-Token": str(cfg.get("public_token", "")),
|
||||
"DTAPI-Date": date,
|
||||
"DTAPI-Signature": sig,
|
||||
"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, context=_ctx(cfg)) 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):
|
||||
count = inputs.get("count")
|
||||
min_score = inputs.get("min_score")
|
||||
from_time = inputs.get("from_time")
|
||||
|
||||
params = {"count": int(count or 50)}
|
||||
if min_score not in (None, ""):
|
||||
params["minscore"] = float(min_score)
|
||||
if from_time not in (None, ""):
|
||||
params["from"] = from_time
|
||||
|
||||
return request("GET", "/modelbreaches", cfg, params=params)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,69 @@
|
||||
import json, os, sys, hmac, hashlib, ssl, datetime
|
||||
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 _ctx(cfg):
|
||||
if cfg.get("insecure"):
|
||||
c = ssl.create_default_context()
|
||||
c.check_hostname = False
|
||||
c.verify_mode = ssl.CERT_NONE
|
||||
return c
|
||||
return None
|
||||
|
||||
|
||||
def _sign(request_path, cfg, date):
|
||||
public_token = str(cfg.get("public_token", ""))
|
||||
private_token = str(cfg.get("private_token", "")).encode("utf-8")
|
||||
signed = request_path + "\n" + public_token + "\n" + date
|
||||
return hmac.new(private_token, signed.encode("utf-8"), hashlib.sha1).hexdigest()
|
||||
|
||||
|
||||
def request(method, path, cfg, params=None, body=None):
|
||||
request_path = path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
request_path += "?" + urllib.parse.urlencode(clean)
|
||||
date = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")
|
||||
sig = _sign(request_path, cfg, date)
|
||||
url = str(cfg.get("base_url", "")).rstrip("/") + request_path
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {
|
||||
"DTAPI-Token": str(cfg.get("public_token", "")),
|
||||
"DTAPI-Date": date,
|
||||
"DTAPI-Signature": sig,
|
||||
"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, context=_ctx(cfg)) 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):
|
||||
request("GET", "/status", cfg)
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
_run(main)
|
||||
Reference in New Issue
Block a user