diff --git a/integrations/urlhaus/manifest.yaml b/integrations/urlhaus/manifest.yaml new file mode 100644 index 0000000..3e91e1c --- /dev/null +++ b/integrations/urlhaus/manifest.yaml @@ -0,0 +1,63 @@ +id: urlhaus +name: URLhaus +version: 1.0.0 +description: "URLhaus by abuse.ch (API v1) — malicious-URL threat intelligence: look up a URL, host, or payload (hash), and pull recently added malicious URLs. Auth-Key authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: URL/host/payload lookup, recent URLs." +category: enrichment + +# Per-instance configuration. abuse.ch requires an Auth-Key header on API requests. +config_schema: + properties: + auth_key: + type: string + description: "abuse.ch Auth-Key" + x-soar-sensitive: true + required: + - auth_key + +commands: + - id: url_info + name: urlhaus-url-info + description: "Look up a URL in the URLhaus database." + risk: read + inputs_schema: + properties: + url: { type: string, description: "URL to look up" } + required: [url] + outputs_schema: { properties: {} } + - id: host_info + name: urlhaus-host-info + description: "Look up a host (domain or IP) in the URLhaus database." + risk: read + inputs_schema: + properties: + host: { type: string, description: "Domain or IP" } + required: [host] + outputs_schema: { properties: {} } + - id: payload_info + name: urlhaus-payload-info + description: "Look up a malware payload by hash (MD5 or SHA-256)." + risk: read + inputs_schema: + properties: + hash: { type: string, description: "MD5 or SHA-256 hash" } + required: [hash] + outputs_schema: { properties: {} } + - id: recent_urls + name: urlhaus-recent-urls + description: "Get recently added malicious URLs." + risk: read + inputs_schema: + properties: + limit: { type: string, description: "Result set: 'limit/N' amount — use 'recent' feed (returns latest 1000)" } + required: [] + outputs_schema: { properties: {} } + + - id: test_connection + name: urlhaus-test-connection + description: "Verify the Auth-Key (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/urlhaus/scripts/host_info.py b/integrations/urlhaus/scripts/host_info.py new file mode 100644 index 0000000..534b9d4 --- /dev/null +++ b/integrations/urlhaus/scripts/host_info.py @@ -0,0 +1,53 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://urlhaus-api.abuse.ch/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def post_form(path, 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(BASE + path, 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 get(path, cfg): + headers = {"Auth-Key": str(cfg.get("auth_key", "")), "Accept": "application/json"} + req = urllib.request.Request(BASE + path, headers=headers, method="GET") + 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): + host = str(inputs.get("host", "")).strip() + if not host: + raise Exception("host is required") + return post_form("/host/", cfg, {"host": host}) + + +_run(main) diff --git a/integrations/urlhaus/scripts/payload_info.py b/integrations/urlhaus/scripts/payload_info.py new file mode 100644 index 0000000..241fdf2 --- /dev/null +++ b/integrations/urlhaus/scripts/payload_info.py @@ -0,0 +1,54 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://urlhaus-api.abuse.ch/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def post_form(path, 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(BASE + path, 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 get(path, cfg): + headers = {"Auth-Key": str(cfg.get("auth_key", "")), "Accept": "application/json"} + req = urllib.request.Request(BASE + path, headers=headers, method="GET") + 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_ = str(inputs.get("hash", "")).strip() + if not hash_: + raise Exception("hash is required") + field = "md5_hash" if len(hash_) == 32 else "sha256_hash" + return post_form("/payload/", cfg, {field: hash_}) + + +_run(main) diff --git a/integrations/urlhaus/scripts/recent_urls.py b/integrations/urlhaus/scripts/recent_urls.py new file mode 100644 index 0000000..fa51e4f --- /dev/null +++ b/integrations/urlhaus/scripts/recent_urls.py @@ -0,0 +1,50 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://urlhaus-api.abuse.ch/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def post_form(path, 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(BASE + path, 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 get(path, cfg): + headers = {"Auth-Key": str(cfg.get("auth_key", "")), "Accept": "application/json"} + req = urllib.request.Request(BASE + path, headers=headers, method="GET") + 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): + return get("/urls/recent/", cfg) + + +_run(main) diff --git a/integrations/urlhaus/scripts/test_connection.py b/integrations/urlhaus/scripts/test_connection.py new file mode 100644 index 0000000..0985411 --- /dev/null +++ b/integrations/urlhaus/scripts/test_connection.py @@ -0,0 +1,51 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://urlhaus-api.abuse.ch/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def post_form(path, 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(BASE + path, 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 get(path, cfg): + headers = {"Auth-Key": str(cfg.get("auth_key", "")), "Accept": "application/json"} + req = urllib.request.Request(BASE + path, headers=headers, method="GET") + 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): + get("/urls/recent/", cfg) + return {"ok": True} + + +_run(main) diff --git a/integrations/urlhaus/scripts/url_info.py b/integrations/urlhaus/scripts/url_info.py new file mode 100644 index 0000000..06fc913 --- /dev/null +++ b/integrations/urlhaus/scripts/url_info.py @@ -0,0 +1,53 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://urlhaus-api.abuse.ch/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def post_form(path, 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(BASE + path, 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 get(path, cfg): + headers = {"Auth-Key": str(cfg.get("auth_key", "")), "Accept": "application/json"} + req = urllib.request.Request(BASE + path, headers=headers, method="GET") + 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): + url = str(inputs.get("url", "")).strip() + if not url: + raise Exception("url is required") + return post_form("/url/", cfg, {"url": url}) + + +_run(main)