feat(malwarebazaar): new MalwareBazaar sample-intel integration

MalwareBazaar (abuse.ch) API v1, 5 commands: sample info by hash, recent samples, tag info, signature info. Auth-Key auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-12 14:58:07 +02:00
parent 82ad3a9369
commit 015131a2d1
6 changed files with 292 additions and 0 deletions
@@ -0,0 +1,46 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://mb-api.abuse.ch/api/v1/"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def query(cfg, fields):
+    data = urllib.parse.urlencode({k: v for k, v in fields.items() if v not in (None, "")}).encode("utf-8")
+    headers = {
+        "Auth-Key": str(cfg.get("auth_key", "")),
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Accept": "application/json",
+    }
+    req = urllib.request.Request(API, data=data, headers=headers, method="POST")
+    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):
+    hash_ = inputs.get("hash")
+    if not hash_:
+        raise Exception("hash is required")
+
+    return query(cfg, {"query": "get_info", "hash": hash_})
+
+
+_run(main)
@@ -0,0 +1,44 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://mb-api.abuse.ch/api/v1/"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def query(cfg, fields):
+    data = urllib.parse.urlencode({k: v for k, v in fields.items() if v not in (None, "")}).encode("utf-8")
+    headers = {
+        "Auth-Key": str(cfg.get("auth_key", "")),
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Accept": "application/json",
+    }
+    req = urllib.request.Request(API, data=data, headers=headers, method="POST")
+    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):
+    selector = inputs.get("selector")
+
+    return query(cfg, {"query": "get_recent", "selector": (selector or "100")})
+
+
+_run(main)
@@ -0,0 +1,47 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://mb-api.abuse.ch/api/v1/"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def query(cfg, fields):
+    data = urllib.parse.urlencode({k: v for k, v in fields.items() if v not in (None, "")}).encode("utf-8")
+    headers = {
+        "Auth-Key": str(cfg.get("auth_key", "")),
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Accept": "application/json",
+    }
+    req = urllib.request.Request(API, data=data, headers=headers, method="POST")
+    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):
+    signature = inputs.get("signature")
+    if not signature:
+        raise Exception("signature is required")
+    limit = inputs.get("limit")
+
+    return query(cfg, {"query": "get_siginfo", "signature": signature, "limit": int(limit or 50)})
+
+
+_run(main)
@@ -0,0 +1,47 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://mb-api.abuse.ch/api/v1/"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def query(cfg, fields):
+    data = urllib.parse.urlencode({k: v for k, v in fields.items() if v not in (None, "")}).encode("utf-8")
+    headers = {
+        "Auth-Key": str(cfg.get("auth_key", "")),
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Accept": "application/json",
+    }
+    req = urllib.request.Request(API, data=data, headers=headers, method="POST")
+    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):
+    tag = inputs.get("tag")
+    if not tag:
+        raise Exception("tag is required")
+    limit = inputs.get("limit")
+
+    return query(cfg, {"query": "get_taginfo", "tag": tag, "limit": int(limit or 50)})
+
+
+_run(main)
@@ -0,0 +1,43 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://mb-api.abuse.ch/api/v1/"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def query(cfg, fields):
+    data = urllib.parse.urlencode({k: v for k, v in fields.items() if v not in (None, "")}).encode("utf-8")
+    headers = {
+        "Auth-Key": str(cfg.get("auth_key", "")),
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Accept": "application/json",
+    }
+    req = urllib.request.Request(API, data=data, headers=headers, method="POST")
+    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):
+    query(cfg, {"query": "get_recent", "selector": "100"})
+    return {"ok": True}
+
+
+_run(main)