feat(mimecast): new Mimecast anti-phishing integration

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>
This commit is contained in:
Guillaume BOURGEOIS
2026-07-11 23:46:35 +02:00
parent f826181704
commit 460dfaefba
12 changed files with 912 additions and 0 deletions
+143
View File
@@ -0,0 +1,143 @@
id: mimecast
name: Mimecast
version: 1.0.0
description: "Mimecast (Email Security API) — anti-phishing containment: list/release/reject held messages, search messages and get message info, block URLs (managed URLs), block senders, create remediation incidents, and decode rewritten URLs. HMAC-SHA1 signed authentication; stdlib-only, no extra Python dependencies."
changelog: "1.0.0 — Initial release: held-message list/release/reject, message search + info, managed URL create/list, block sender, create remediation, URL decode."
category: email
# Per-instance configuration. Requests are signed (HMAC-SHA1) with the
# application + access/secret keys.
config_schema:
properties:
base_url:
type: string
description: "Mimecast API base URL (e.g. https://eu-api.mimecast.com)"
app_id:
type: string
description: "Application ID"
app_key:
type: string
description: "Application Key"
x-soar-sensitive: true
access_key:
type: string
description: "Access Key"
x-soar-sensitive: true
secret_key:
type: string
description: "Secret Key (base64)"
x-soar-sensitive: true
required:
- base_url
- app_id
- app_key
- access_key
- secret_key
commands:
- id: list_held_messages
name: mimecast-list-held-messages
description: "List messages currently held for review."
risk: read
inputs_schema:
properties:
admin: { type: boolean, description: "List all held messages (admin view) vs. the account's own (default true)" }
limit: { type: number, description: "Max messages (default 100)" }
required: []
outputs_schema: { properties: {} }
- id: release_held_message
name: mimecast-release-held-message
description: "Release a held message for delivery."
inputs_schema:
properties:
message_id: { type: string, description: "Held message ID" }
required: [message_id]
outputs_schema: { properties: {} }
- id: reject_held_message
name: mimecast-reject-held-message
description: "Reject a held message."
inputs_schema:
properties:
message_id: { type: string, description: "Held message ID" }
reason: { type: string, description: "Optional rejection reason" }
required: [message_id]
outputs_schema: { properties: {} }
- id: search_messages
name: mimecast-search-messages
description: "Search tracked messages by sender, recipient, or subject."
risk: read
inputs_schema:
properties:
from: { type: string, description: "Sender address filter" }
to: { type: string, description: "Recipient address filter" }
subject: { type: string, description: "Subject filter" }
start: { type: string, description: "ISO-8601 start time" }
end: { type: string, description: "ISO-8601 end time" }
required: []
outputs_schema: { properties: {} }
- id: get_message_info
name: mimecast-get-message-info
description: "Get delivery and processing details for a tracked message."
risk: read
inputs_schema:
properties:
message_id: { type: string, description: "Message ID" }
required: [message_id]
outputs_schema: { properties: {} }
- id: create_managed_url
name: mimecast-create-managed-url
description: "Add a URL to the managed URL list (block or permit)."
inputs_schema:
properties:
url: { type: string, description: "URL to manage" }
action: { type: string, description: "block or permit (default block)" }
match_type: { type: string, description: "explicit or domain (default explicit)" }
comment: { type: string, description: "Optional comment" }
required: [url]
outputs_schema: { properties: {} }
- id: list_managed_urls
name: mimecast-list-managed-urls
description: "List managed URLs (block/permit list)."
risk: read
inputs_schema:
properties: {}
required: []
outputs_schema: { properties: {} }
- id: block_sender
name: mimecast-block-sender
description: "Block (or permit) a sender for a recipient."
inputs_schema:
properties:
sender: { type: string, description: "Sender address to block/permit" }
recipient: { type: string, description: "Recipient address the rule applies to" }
action: { type: string, description: "block or permit (default block)" }
required: [sender, recipient]
outputs_schema: { properties: {} }
- id: create_remediation
name: mimecast-create-remediation
description: "Create a remediation incident to pull malicious messages from mailboxes."
inputs_schema:
properties:
message_hash: { type: string, description: "File hash to remediate (provide this or message_id)" }
message_id: { type: string, description: "Message ID / internet message ID to remediate" }
reason: { type: string, description: "Reason for remediation" }
required: []
outputs_schema: { properties: {} }
- id: decode_url
name: mimecast-decode-url
description: "Decode a Mimecast-rewritten URL back to its original form."
risk: read
inputs_schema:
properties:
url: { type: string, description: "Rewritten URL to decode" }
required: [url]
outputs_schema: { properties: {} }
- id: test_connection
name: mimecast-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,75 @@
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):
sender = inputs.get("sender")
if not sender:
raise Exception("sender is required")
recipient = inputs.get("recipient")
if not recipient:
raise Exception("recipient is required")
action = inputs.get("action")
payload = {
"sender": sender,
"to": recipient,
"action": action or "block",
}
return call("/api/managedsender/permit-or-block-sender", cfg, payload)
_run(main)
@@ -0,0 +1,77 @@
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)
@@ -0,0 +1,76 @@
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):
message_hash = inputs.get("message_hash")
message_id = inputs.get("message_id")
if not message_hash and not message_id:
raise Exception("message_hash or message_id is required")
payload = {}
if message_hash:
payload["hash"] = message_hash
if message_id:
payload["messageId"] = message_id
reason = inputs.get("reason")
if reason:
payload["reason"] = reason
return call("/api/ttp/remediation/create", cfg, payload)
_run(main)
@@ -0,0 +1,65 @@
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")
return call("/api/ttp/url/decode-url", cfg, {"url": url})
_run(main)
@@ -0,0 +1,65 @@
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):
message_id = inputs.get("message_id")
if not message_id:
raise Exception("message_id is required")
return call("/api/message-finder/get-message-info", cfg, {"id": message_id})
_run(main)
@@ -0,0 +1,72 @@
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):
admin = inputs.get("admin")
if admin is None:
admin = True
limit = inputs.get("limit")
if limit is None:
limit = 100
payload = {
"admin": bool(admin),
"pagination": {"pageSize": int(limit)},
}
return call("/api/gateway/get-hold-message-list", cfg, payload)
_run(main)
@@ -0,0 +1,62 @@
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):
return call("/api/ttp/url/get-all-managed-urls", cfg, None)
_run(main)
@@ -0,0 +1,69 @@
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):
message_id = inputs.get("message_id")
if not message_id:
raise Exception("message_id is required")
payload = {"id": message_id}
reason = inputs.get("reason")
if reason:
payload["message"] = reason
return call("/api/gateway/hold-reject", cfg, payload)
_run(main)
@@ -0,0 +1,65 @@
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):
message_id = inputs.get("message_id")
if not message_id:
raise Exception("message_id is required")
return call("/api/gateway/hold-release", cfg, {"id": message_id})
_run(main)
@@ -0,0 +1,80 @@
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):
from_addr = inputs.get("from")
to_addr = inputs.get("to")
subject = inputs.get("subject")
start = inputs.get("start")
end = inputs.get("end")
payload = {}
if from_addr:
payload["from"] = from_addr
if to_addr:
payload["to"] = to_addr
if subject:
payload["subject"] = subject
if start:
payload["start"] = start
if end:
payload["end"] = end
return call("/api/message-finder/search", cfg, payload)
_run(main)
@@ -0,0 +1,63 @@
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):
call("/api/account/get-account", cfg, None)
return {"ok": True}
_run(main)