diff --git a/integrations/urlscan/manifest.yaml b/integrations/urlscan/manifest.yaml new file mode 100644 index 0000000..09bb73f --- /dev/null +++ b/integrations/urlscan/manifest.yaml @@ -0,0 +1,56 @@ +id: urlscan +name: urlscan.io +version: 1.0.0 +description: "urlscan.io (API v1) — submit URLs for scanning, retrieve scan results (verdicts, page metadata, screenshot and DOM references) and search historical scans. API-key authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: URL submission, result retrieval and historical search." +category: enrichment + +config_schema: + properties: + api_key: + type: string + description: "urlscan.io API key" + x-soar-sensitive: true + required: + - api_key + +commands: + - id: submit + name: urlscan-submit + description: "Submit a URL for scanning. Returns a scan UUID; retrieve the report with urlscan-get-result once the scan finishes (usually ~10-30s)." + inputs_schema: + properties: + url: { type: string, description: "URL to scan" } + visibility: { type: string, description: "public, unlisted or private (default public)" } + tags: { type: string, description: "Comma-separated tags to attach to the scan" } + required: [url] + outputs_schema: { properties: {} } + - id: get_result + name: urlscan-get-result + description: "Retrieve the result of a completed scan by UUID (404 while the scan is still running)." + risk: read + inputs_schema: + properties: + uuid: { type: string, description: "Scan UUID (from urlscan-submit)" } + required: [uuid] + outputs_schema: { properties: {} } + - id: search + name: urlscan-search + description: "Search historical scans with a query (e.g. domain:example.com, ip:1.2.3.4)." + risk: read + inputs_schema: + properties: + query: { type: string, description: "urlscan search query" } + size: { type: number, description: "Maximum results (default 20, max 10000)" } + search_after: { type: string, description: "Pagination cursor (sort values from the previous page, comma-separated)" } + required: [query] + outputs_schema: { properties: {} } + + - id: test_connection + name: urlscan-test-connection + description: "Verify the API key (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/urlscan/scripts/get_result.py b/integrations/urlscan/scripts/get_result.py new file mode 100644 index 0000000..219d1b4 --- /dev/null +++ b/integrations/urlscan/scripts/get_result.py @@ -0,0 +1,44 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://urlscan.io/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None, body=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Accept": "application/json", "API-Key": str(_cfg().get("api_key") or "")} + 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) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + uuid = inputs.get("uuid") + if not uuid: + raise Exception("uuid is required") + res = request("GET", "/result/%s/" % q(uuid)) + print(json.dumps(res)) + + +try: + main() +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) diff --git a/integrations/urlscan/scripts/search.py b/integrations/urlscan/scripts/search.py new file mode 100644 index 0000000..60232b1 --- /dev/null +++ b/integrations/urlscan/scripts/search.py @@ -0,0 +1,46 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://urlscan.io/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None, body=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Accept": "application/json", "API-Key": str(_cfg().get("api_key") or "")} + 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) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + query = inputs.get("query") + if not query: + raise Exception("query is required") + size = inputs.get("size") + search_after = inputs.get("search_after") + res = request("GET", "/search/", {"q": query, "size": size or 20, "search_after": search_after or None}) + print(json.dumps(res)) + + +try: + main() +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) diff --git a/integrations/urlscan/scripts/submit.py b/integrations/urlscan/scripts/submit.py new file mode 100644 index 0000000..a74ebbc --- /dev/null +++ b/integrations/urlscan/scripts/submit.py @@ -0,0 +1,49 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://urlscan.io/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None, body=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Accept": "application/json", "API-Key": str(_cfg().get("api_key") or "")} + 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) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + url = inputs.get("url") + if not url: + raise Exception("url is required") + visibility = inputs.get("visibility") + tags = inputs.get("tags") + body = {"url": url, "visibility": visibility or "public"} + if tags: + body["tags"] = [t.strip() for t in tags.split(",") if t.strip()] + res = request("POST", "/scan/", body=body) + print(json.dumps(res)) + + +try: + main() +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) diff --git a/integrations/urlscan/scripts/test_connection.py b/integrations/urlscan/scripts/test_connection.py new file mode 100644 index 0000000..8b8e03a --- /dev/null +++ b/integrations/urlscan/scripts/test_connection.py @@ -0,0 +1,42 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://urlscan.io/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None, body=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Accept": "application/json", "API-Key": str(_cfg().get("api_key") or "")} + 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) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + res = request("GET", "/search/", {"q": "*", "size": 1}) + if "results" not in res: + raise Exception("unexpected response") + print(json.dumps({"ok": True})) + + +try: + main() +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)