Guillaume BOURGEOIS
460dfaefba
Email Security API, 11 commands: held-message list/release/reject, message search + info, managed URL create/list (block), block sender, create remediation, URL decode. HMAC-SHA1 signed auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
import json, os, sys, hmac, hashlib, base64, uuid, datetime
|
|
import 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 call(uri, cfg, data_payload=None):
|
|
base_url = str(cfg.get("base_url", "")).rstrip("/")
|
|
app_id = str(cfg.get("app_id", ""))
|
|
app_key = str(cfg.get("app_key", ""))
|
|
access_key = str(cfg.get("access_key", ""))
|
|
secret_key = str(cfg.get("secret_key", ""))
|
|
|
|
request_id = str(uuid.uuid4())
|
|
hdr_date = datetime.datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S UTC")
|
|
to_sign = hdr_date + ":" + request_id + ":" + uri + ":" + app_key
|
|
sig = base64.b64encode(
|
|
hmac.new(base64.b64decode(secret_key), to_sign.encode("utf-8"), hashlib.sha1).digest()
|
|
).decode("utf-8")
|
|
|
|
headers = {
|
|
"Authorization": "MC " + access_key + ":" + sig,
|
|
"x-mc-app-id": app_id,
|
|
"x-mc-date": hdr_date,
|
|
"x-mc-req-id": request_id,
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
}
|
|
body = {"data": [data_payload] if data_payload is not None else []}
|
|
req = urllib.request.Request(base_url + uri, data=json.dumps(body).encode("utf-8"), headers=headers, method="POST")
|
|
with urllib.request.urlopen(req, timeout=60) as r:
|
|
raw = r.read()
|
|
resp = json.loads(raw) if raw else {}
|
|
fails = resp.get("fail") if isinstance(resp, dict) else None
|
|
if fails:
|
|
# Mimecast returns errors in the 'fail' array even on HTTP 200
|
|
raise Exception("Mimecast error: " + json.dumps(fails))
|
|
return resp
|
|
|
|
|
|
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):
|
|
url = inputs.get("url")
|
|
if not url:
|
|
raise Exception("url is required")
|
|
action = inputs.get("action")
|
|
match_type = inputs.get("match_type")
|
|
|
|
payload = {
|
|
"url": url,
|
|
"action": action or "block",
|
|
"matchType": match_type or "explicit",
|
|
}
|
|
comment = inputs.get("comment")
|
|
if comment:
|
|
payload["comment"] = comment
|
|
|
|
return call("/api/ttp/url/create-managed-url", cfg, payload)
|
|
|
|
|
|
_run(main)
|