feat(urlscan): new urlscan.io enrichment integration

4 commands: submit URL scan, retrieve result, search historical scans.
API-key auth, stdlib-only.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

integrations/abuseipdb/manifest.yaml

@@ -0,0 +1,67 @@
+id: abuseipdb
+name: AbuseIPDB
+version: 1.0.0
+description: "AbuseIPDB (API v2) — check the abuse reputation of an IP, report abusive IPs, pull the blacklist and check a CIDR block. API-key authentication; stdlib-only, no extra Python dependencies."
+changelog: "1.0.0 — Initial release: IP reputation check, report, blacklist retrieval and CIDR-block check."
+category: enrichment
+
+config_schema:
+  properties:
+    api_key:
+      type: string
+      description: "AbuseIPDB API key"
+      x-soar-sensitive: true
+  required:
+    - api_key
+
+commands:
+  - id: check_ip
+    name: abuseipdb-check-ip
+    description: "Check the abuse-confidence reputation of an IP address."
+    risk: read
+    inputs_schema:
+      properties:
+        ip: { type: string, description: "IP address to check" }
+        max_age_days: { type: number, description: "Only consider reports within this many days (default 30, max 365)" }
+        verbose: { type: boolean, description: "Include the detailed report list" }
+      required: [ip]
+    outputs_schema: { properties: {} }
+  - id: report_ip
+    name: abuseipdb-report-ip
+    description: "Report an abusive IP address to AbuseIPDB."
+    inputs_schema:
+      properties:
+        ip: { type: string, description: "IP address to report" }
+        categories: { type: string, description: "Comma-separated AbuseIPDB category IDs (e.g. 18,22)" }
+        comment: { type: string, description: "Description of the abusive activity (avoid sensitive data)" }
+      required: [ip, categories]
+    outputs_schema: { properties: {} }
+  - id: get_blacklist
+    name: abuseipdb-get-blacklist
+    description: "Retrieve the AbuseIPDB blacklist of the most-reported IPs."
+    risk: read
+    inputs_schema:
+      properties:
+        confidence_minimum: { type: number, description: "Minimum abuse-confidence score (default 100)" }
+        limit: { type: number, description: "Maximum entries (default 100)" }
+      required: []
+    outputs_schema: { properties: {} }
+  - id: check_block
+    name: abuseipdb-check-block
+    description: "Check the reports for every address in a CIDR block (max /24 on the free tier)."
+    risk: read
+    inputs_schema:
+      properties:
+        network: { type: string, description: "CIDR network, e.g. 192.0.2.0/24" }
+        max_age_days: { type: number, description: "Only consider reports within this many days (default 30)" }
+      required: [network]
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: abuseipdb-test-connection
+    description: "Verify the API key (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/abuseipdb/scripts/check_block.py

@@ -0,0 +1,45 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.abuseipdb.com/api/v2"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None, body=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    data = urllib.parse.urlencode(body).encode("utf-8") if body is not None else None
+    headers = {"Accept": "application/json", "Key": str(_cfg().get("api_key") or "")}
+    if data is not None:
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+    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 {}
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    network = inputs.get("network")
+    if not network:
+        raise Exception("network is required")
+    max_age_days = inputs.get("max_age_days")
+
+    params = {"network": network, "maxAgeInDays": max_age_days or 30}
+
+    result = request("GET", "/check-block", params=params)
+    print(json.dumps(result))
+
+
+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)

integrations/abuseipdb/scripts/check_ip.py

@@ -0,0 +1,48 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.abuseipdb.com/api/v2"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None, body=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    data = urllib.parse.urlencode(body).encode("utf-8") if body is not None else None
+    headers = {"Accept": "application/json", "Key": str(_cfg().get("api_key") or "")}
+    if data is not None:
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+    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 {}
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    ip = inputs.get("ip")
+    if not ip:
+        raise Exception("ip is required")
+    max_age_days = inputs.get("max_age_days")
+    verbose = inputs.get("verbose")
+
+    params = {"ipAddress": ip, "maxAgeInDays": max_age_days or 30}
+    if verbose:
+        params["verbose"] = ""
+
+    result = request("GET", "/check", params=params)
+    print(json.dumps(result))
+
+
+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)

integrations/abuseipdb/scripts/get_blacklist.py

@@ -0,0 +1,46 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.abuseipdb.com/api/v2"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None, body=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    data = urllib.parse.urlencode(body).encode("utf-8") if body is not None else None
+    headers = {"Accept": "application/json", "Key": str(_cfg().get("api_key") or "")}
+    if data is not None:
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+    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 {}
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    confidence_minimum = inputs.get("confidence_minimum")
+    limit = inputs.get("limit")
+
+    params = {
+        "confidenceMinimum": confidence_minimum or 100,
+        "limit": limit or 100,
+    }
+
+    result = request("GET", "/blacklist", params=params)
+    print(json.dumps(result))
+
+
+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)

integrations/abuseipdb/scripts/report_ip.py

@@ -0,0 +1,50 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.abuseipdb.com/api/v2"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None, body=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    data = urllib.parse.urlencode(body).encode("utf-8") if body is not None else None
+    headers = {"Accept": "application/json", "Key": str(_cfg().get("api_key") or "")}
+    if data is not None:
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+    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 {}
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    ip = inputs.get("ip")
+    if not ip:
+        raise Exception("ip is required")
+    categories = inputs.get("categories")
+    if not categories:
+        raise Exception("categories is required")
+    comment = inputs.get("comment")
+
+    body = {"ip": ip, "categories": categories}
+    if comment:
+        body["comment"] = comment
+
+    result = request("POST", "/report", body=body)
+    print(json.dumps(result))
+
+
+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)

integrations/abuseipdb/scripts/test_connection.py

@@ -0,0 +1,40 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.abuseipdb.com/api/v2"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None, body=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    data = urllib.parse.urlencode(body).encode("utf-8") if body is not None else None
+    headers = {"Accept": "application/json", "Key": str(_cfg().get("api_key") or "")}
+    if data is not None:
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+    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 {}
+
+
+def main():
+    params = {"ipAddress": "8.8.8.8", "maxAgeInDays": 1}
+    result = request("GET", "/check", params=params)
+    if "data" not in result:
+        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)

integrations/alienvault-otx/manifest.yaml

@@ -0,0 +1,101 @@
+id: alienvault_otx
+name: AlienVault OTX
+version: 1.0.0
+description: "AlienVault OTX (Open Threat Exchange, API v1) — reputation and threat context for IPs, domains, URLs and file hashes, pulse details and search, and passive DNS / related-URL pivots. API-key authentication; stdlib-only, no extra Python dependencies."
+changelog: "1.0.0 — Initial release: IP/domain/URL/file reputation, pulse details and search, passive DNS and related URLs."
+category: enrichment
+
+config_schema:
+  properties:
+    api_key:
+      type: string
+      description: "AlienVault OTX API key (from your OTX account settings)"
+      x-soar-sensitive: true
+  required:
+    - api_key
+
+commands:
+  - id: ip_reputation
+    name: alienvault-otx-ip
+    description: "Threat context for an IP address (IPv4 or IPv6)."
+    risk: read
+    inputs_schema:
+      properties:
+        ip: { type: string, description: "IP address" }
+      required: [ip]
+    outputs_schema: { properties: {} }
+  - id: domain_reputation
+    name: alienvault-otx-domain
+    description: "Threat context for a domain."
+    risk: read
+    inputs_schema:
+      properties:
+        domain: { type: string, description: "Domain name" }
+      required: [domain]
+    outputs_schema: { properties: {} }
+  - id: url_reputation
+    name: alienvault-otx-url
+    description: "Threat context for a URL."
+    risk: read
+    inputs_schema:
+      properties:
+        url: { type: string, description: "URL" }
+      required: [url]
+    outputs_schema: { properties: {} }
+  - id: file_reputation
+    name: alienvault-otx-file
+    description: "Threat context for a file hash (MD5, SHA1 or SHA256)."
+    risk: read
+    inputs_schema:
+      properties:
+        file: { type: string, description: "File hash" }
+      required: [file]
+    outputs_schema: { properties: {} }
+  - id: get_pulse
+    name: alienvault-otx-get-pulse
+    description: "Get the details of a pulse (threat report) by ID."
+    risk: read
+    inputs_schema:
+      properties:
+        pulse_id: { type: string, description: "Pulse ID" }
+      required: [pulse_id]
+    outputs_schema: { properties: {} }
+  - id: search_pulses
+    name: alienvault-otx-search-pulses
+    description: "Search pulses by keyword."
+    risk: read
+    inputs_schema:
+      properties:
+        query: { type: string, description: "Search string" }
+        limit: { type: number, description: "Maximum pulses (default 20)" }
+      required: [query]
+    outputs_schema: { properties: {} }
+  - id: passive_dns
+    name: alienvault-otx-passive-dns
+    description: "Passive DNS records for an IP or domain indicator."
+    risk: read
+    inputs_schema:
+      properties:
+        indicator: { type: string, description: "IP or domain" }
+        indicator_type: { type: string, description: "IPv4, IPv6 or domain (default auto-detected)" }
+      required: [indicator]
+    outputs_schema: { properties: {} }
+  - id: related_urls
+    name: alienvault-otx-related-urls
+    description: "URLs associated with an IP or domain indicator."
+    risk: read
+    inputs_schema:
+      properties:
+        indicator: { type: string, description: "IP or domain" }
+        indicator_type: { type: string, description: "IPv4, IPv6 or domain (default auto-detected)" }
+      required: [indicator]
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: alienvault-otx-test-connection
+    description: "Verify the API key (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/alienvault-otx/scripts/domain_reputation.py

@@ -0,0 +1,41 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://otx.alienvault.com/api/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=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)
+    headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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", "{}"))
+    domain = inputs.get("domain")
+    if not domain:
+        raise Exception("domain is required")
+    res = request("GET", "/indicators/domain/%s/general" % q(domain))
+    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)

integrations/alienvault-otx/scripts/file_reputation.py

@@ -0,0 +1,41 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://otx.alienvault.com/api/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=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)
+    headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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", "{}"))
+    file_hash = inputs.get("file")
+    if not file_hash:
+        raise Exception("file is required")
+    res = request("GET", "/indicators/file/%s/general" % q(file_hash))
+    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)

integrations/alienvault-otx/scripts/get_pulse.py

@@ -0,0 +1,41 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://otx.alienvault.com/api/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=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)
+    headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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", "{}"))
+    pulse_id = inputs.get("pulse_id")
+    if not pulse_id:
+        raise Exception("pulse_id is required")
+    res = request("GET", "/pulses/%s" % q(pulse_id))
+    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)

integrations/alienvault-otx/scripts/ip_reputation.py

@@ -0,0 +1,43 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+import re
+
+API = "https://otx.alienvault.com/api/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=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)
+    headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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", "{}"))
+    ip = inputs.get("ip")
+    if not ip:
+        raise Exception("ip is required")
+    section = "IPv6" if ":" in ip else "IPv4"
+    res = request("GET", "/indicators/%s/%s/general" % (section, q(ip)))
+    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)

integrations/alienvault-otx/scripts/passive_dns.py

@@ -0,0 +1,44 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+import re
+
+API = "https://otx.alienvault.com/api/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=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)
+    headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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", "{}"))
+    indicator = inputs.get("indicator")
+    if not indicator:
+        raise Exception("indicator is required")
+    indicator_type = inputs.get("indicator_type")
+    itype = indicator_type or ("IPv6" if ":" in indicator else ("IPv4" if re.match(r"^\d+\.\d+\.\d+\.\d+$", indicator) else "domain"))
+    res = request("GET", "/indicators/%s/%s/passive_dns" % (itype, q(indicator)))
+    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)

integrations/alienvault-otx/scripts/related_urls.py

@@ -0,0 +1,44 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+import re
+
+API = "https://otx.alienvault.com/api/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=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)
+    headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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", "{}"))
+    indicator = inputs.get("indicator")
+    if not indicator:
+        raise Exception("indicator is required")
+    indicator_type = inputs.get("indicator_type")
+    itype = indicator_type or ("IPv6" if ":" in indicator else ("IPv4" if re.match(r"^\d+\.\d+\.\d+\.\d+$", indicator) else "domain"))
+    res = request("GET", "/indicators/%s/%s/url_list" % (itype, q(indicator)), {"limit": 100})
+    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)

integrations/alienvault-otx/scripts/search_pulses.py

@@ -0,0 +1,42 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://otx.alienvault.com/api/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=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)
+    headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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")
+    limit = inputs.get("limit")
+    res = request("GET", "/search/pulses", {"q": query, "limit": limit or 20})
+    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)

integrations/alienvault-otx/scripts/test_connection.py

@@ -0,0 +1,39 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://otx.alienvault.com/api/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=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)
+    headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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", "/user/me")
+    if "username" not in res:
+        raise Exception("unexpected response")
+    print(json.dumps({"ok": True, "user": res.get("username")}))
+
+
+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)

integrations/alienvault-otx/scripts/url_reputation.py

@@ -0,0 +1,41 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://otx.alienvault.com/api/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=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)
+    headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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")
+    res = request("GET", "/indicators/url/%s/general" % q(url))
+    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)

integrations/greynoise/manifest.yaml

@@ -0,0 +1,84 @@
+id: greynoise
+name: GreyNoise
+version: 1.0.0
+description: "GreyNoise (API v2/v3) — tell internet background-noise scanners from targeted activity: IP context and quick classification, RIOT common-business-service lookup, GNQL query and stats, and IP scan timeline. API-key authentication; stdlib-only, no extra Python dependencies."
+changelog: "1.0.0 — Initial release: IP context, quick check, RIOT, GNQL query and stats, and IP timeline."
+category: enrichment
+
+config_schema:
+  properties:
+    api_key:
+      type: string
+      description: "GreyNoise API key"
+      x-soar-sensitive: true
+  required:
+    - api_key
+
+commands:
+  - id: ip_context
+    name: greynoise-ip-context
+    description: "Full GreyNoise context for an IP (classification, actor, tags, metadata)."
+    risk: read
+    inputs_schema:
+      properties:
+        ip: { type: string, description: "IP address" }
+      required: [ip]
+    outputs_schema: { properties: {} }
+  - id: ip_quick_check
+    name: greynoise-ip-quick-check
+    description: "Quick noise/RIOT classification for an IP (lightweight)."
+    risk: read
+    inputs_schema:
+      properties:
+        ip: { type: string, description: "IP address" }
+      required: [ip]
+    outputs_schema: { properties: {} }
+  - id: riot
+    name: greynoise-riot
+    description: "RIOT lookup — is the IP a known common business service (e.g. Google DNS, Office365)?"
+    risk: read
+    inputs_schema:
+      properties:
+        ip: { type: string, description: "IP address" }
+      required: [ip]
+    outputs_schema: { properties: {} }
+  - id: query
+    name: greynoise-query
+    description: "Run a GNQL (GreyNoise Query Language) query and return matching IPs."
+    risk: read
+    inputs_schema:
+      properties:
+        query: { type: string, description: "GNQL query string" }
+        size: { type: number, description: "Maximum results (default 50)" }
+        scroll: { type: string, description: "Scroll token for the next page" }
+      required: [query]
+    outputs_schema: { properties: {} }
+  - id: stats
+    name: greynoise-stats
+    description: "Aggregate statistics for a GNQL query (counts by classification, org, tag, …)."
+    risk: read
+    inputs_schema:
+      properties:
+        query: { type: string, description: "GNQL query string" }
+        count: { type: number, description: "Number of top aggregations to return" }
+      required: [query]
+    outputs_schema: { properties: {} }
+  - id: timeline
+    name: greynoise-timeline
+    description: "Daily scan-activity timeline for an IP."
+    risk: read
+    inputs_schema:
+      properties:
+        ip: { type: string, description: "IP address" }
+        days: { type: number, description: "Number of days of history (default 30)" }
+      required: [ip]
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: greynoise-test-connection
+    description: "Verify the API key (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/greynoise/scripts/ip_context.py

@@ -0,0 +1,42 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.greynoise.io"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    headers = {"Accept": "application/json", "key": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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", "{}"))
+    ip = inputs.get("ip")
+    if not ip:
+        raise Exception("ip is required")
+
+    result = request("GET", "/v2/noise/context/" + q(ip))
+    print(json.dumps(result))
+
+
+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)

integrations/greynoise/scripts/ip_quick_check.py

@@ -0,0 +1,42 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.greynoise.io"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    headers = {"Accept": "application/json", "key": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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", "{}"))
+    ip = inputs.get("ip")
+    if not ip:
+        raise Exception("ip is required")
+
+    result = request("GET", "/v2/noise/quick/" + q(ip))
+    print(json.dumps(result))
+
+
+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)

integrations/greynoise/scripts/query.py

@@ -0,0 +1,46 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.greynoise.io"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    headers = {"Accept": "application/json", "key": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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")
+    scroll = inputs.get("scroll")
+
+    params = {"query": query, "size": size or 50, "scroll": scroll or None}
+
+    result = request("GET", "/v2/experimental/gnql", params=params)
+    print(json.dumps(result))
+
+
+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)

integrations/greynoise/scripts/riot.py

@@ -0,0 +1,42 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.greynoise.io"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    headers = {"Accept": "application/json", "key": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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", "{}"))
+    ip = inputs.get("ip")
+    if not ip:
+        raise Exception("ip is required")
+
+    result = request("GET", "/v2/riot/" + q(ip))
+    print(json.dumps(result))
+
+
+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)

integrations/greynoise/scripts/stats.py

@@ -0,0 +1,45 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.greynoise.io"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    headers = {"Accept": "application/json", "key": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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")
+    count = inputs.get("count")
+
+    params = {"query": query, "count": count or None}
+
+    result = request("GET", "/v2/experimental/gnql/stats", params=params)
+    print(json.dumps(result))
+
+
+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)

integrations/greynoise/scripts/test_connection.py

@@ -0,0 +1,36 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.greynoise.io"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    headers = {"Accept": "application/json", "key": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def main():
+    result = request("GET", "/v2/noise/quick/8.8.8.8")
+    if not isinstance(result, dict):
+        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)

integrations/greynoise/scripts/timeline.py

@@ -0,0 +1,45 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.greynoise.io"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    headers = {"Accept": "application/json", "key": str(_cfg().get("api_key") or "")}
+    req = urllib.request.Request(url, 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", "{}"))
+    ip = inputs.get("ip")
+    if not ip:
+        raise Exception("ip is required")
+    days = inputs.get("days")
+
+    params = {"days": days or 30}
+
+    result = request("GET", "/v3/noise/ips/" + q(ip) + "/timeline", params=params)
+    print(json.dumps(result))
+
+
+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)

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: {} }

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)

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)

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)

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)