feat(anyrun): new ANY.RUN sandbox integration

7 commands: file/URL detonation, report + verdict retrieval, analysis
history, user limits, delete task. API-key auth, stdlib-only.

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

integrations/anyrun/manifest.yaml

@@ -0,0 +1,83 @@
+id: anyrun
+name: ANY.RUN
+version: 1.0.0
+description: "ANY.RUN (API v1) — interactive malware sandbox: detonate files and URLs on Windows/Linux, poll the analysis report and verdict, list analysis history, read user limits and delete tasks. API-key authentication; stdlib-only, no extra Python dependencies."
+changelog: "1.0.0 — Initial release: file/URL detonation, report and verdict retrieval, analysis history, user limits and task deletion."
+category: enrichment
+
+# The API key is sent as 'Authorization: API-Key <key>' on every request.
+config_schema:
+  properties:
+    api_key:
+      type: string
+      description: "ANY.RUN API key"
+      x-soar-sensitive: true
+  required:
+    - api_key
+
+commands:
+  - id: detonate_file
+    name: anyrun-detonate-file
+    description: "Detonate a file (base64) in the ANY.RUN sandbox. Returns a task_id; poll with anyrun-get-report."
+    inputs_schema:
+      properties:
+        file_name: { type: string, description: "File name" }
+        content_base64: { type: string, description: "File content, base64-encoded" }
+        os: { type: string, description: "Sandbox OS: windows or linux (default windows)" }
+        env_bitness: { type: number, description: "Windows bitness: 32 or 64 (default 64)" }
+      required: [file_name, content_base64]
+    outputs_schema: { properties: {} }
+  - id: detonate_url
+    name: anyrun-detonate-url
+    description: "Detonate a URL in the ANY.RUN sandbox. Returns a task_id."
+    inputs_schema:
+      properties:
+        url: { type: string, description: "URL to detonate" }
+        os: { type: string, description: "Sandbox OS: windows or linux (default windows)" }
+        env_bitness: { type: number, description: "Windows bitness: 32 or 64 (default 64)" }
+      required: [url]
+    outputs_schema: { properties: {} }
+  - id: get_report
+    name: anyrun-get-report
+    description: "Get the full analysis report for a task (includes the verdict once the analysis completes)."
+    risk: read
+    inputs_schema:
+      properties:
+        task_id: { type: string, description: "Task ID (from a detonate command)" }
+      required: [task_id]
+    outputs_schema: { properties: {} }
+  - id: get_history
+    name: anyrun-get-history
+    description: "List the analysis history for the account."
+    risk: read
+    inputs_schema:
+      properties:
+        limit: { type: number, description: "Maximum records (default 25)" }
+        skip: { type: number, description: "Records to skip (pagination)" }
+      required: []
+    outputs_schema: { properties: {} }
+  - id: get_user_limits
+    name: anyrun-get-user-limits
+    description: "Read the account's API usage limits."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }
+  - id: delete_task
+    name: anyrun-delete-task
+    description: "Delete an analysis task by ID."
+    inputs_schema:
+      properties:
+        task_id: { type: string, description: "Task ID" }
+      required: [task_id]
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: anyrun-test-connection
+    description: "Verify the API key (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/anyrun/scripts/delete_task.py

@@ -0,0 +1,55 @@
+import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.any.run/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _headers(extra=None):
+    h = {"Accept": "application/json", "Authorization": "API-Key " + str(_cfg().get("api_key") or "")}
+    if extra:
+        h.update(extra)
+    return h
+
+
+def request(method, path, params=None, form=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 = None
+    headers = _headers()
+    if form is not None:
+        data = urllib.parse.urlencode({k: str(v) for k, v in form.items() if v not in (None, "")}).encode("utf-8")
+        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=90) 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", "{}"))
+    task_id = inputs.get("task_id")
+    if not task_id:
+        raise Exception("task_id is required")
+
+    result = request("DELETE", "/analysis/" + q(task_id))
+    if not result:
+        result = {"ok": True, "task_id": task_id}
+    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/anyrun/scripts/detonate_file.py

@@ -0,0 +1,81 @@
+import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
+import uuid
+
+API = "https://api.any.run/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _headers(extra=None):
+    h = {"Accept": "application/json", "Authorization": "API-Key " + str(_cfg().get("api_key") or "")}
+    if extra:
+        h.update(extra)
+    return h
+
+
+def request(method, path, params=None, form=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 = None
+    headers = _headers()
+    if form is not None:
+        data = urllib.parse.urlencode({k: str(v) for k, v in form.items() if v not in (None, "")}).encode("utf-8")
+        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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def multipart(fields, file_field, file_name, file_bytes):
+    boundary = "----riposte" + uuid.uuid4().hex
+    parts = []
+    for name, value in fields.items():
+        parts.append(("--" + boundary + "\r\n"
+                      + 'Content-Disposition: form-data; name="' + name + '"\r\n\r\n'
+                      + str(value) + "\r\n").encode("utf-8"))
+    parts.append(("--" + boundary + "\r\n"
+                  + 'Content-Disposition: form-data; name="' + file_field + '"; filename="' + file_name + '"\r\n'
+                  + "Content-Type: application/octet-stream\r\n\r\n").encode("utf-8"))
+    parts.append(file_bytes)
+    parts.append(("\r\n--" + boundary + "--\r\n").encode("utf-8"))
+    return boundary, b"".join(parts)
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    file_name = inputs.get("file_name")
+    if not file_name:
+        raise Exception("file_name is required")
+    content_base64 = inputs.get("content_base64")
+    if not content_base64:
+        raise Exception("content_base64 is required")
+    env_os = inputs.get("os") or "windows"
+    env_bitness = inputs.get("env_bitness") or 64
+
+    fields = {"obj_type": "file", "env_os": env_os, "env_bitness": str(env_bitness)}
+    boundary, body = multipart(fields, "file", file_name, base64.b64decode(content_base64))
+    req = urllib.request.Request(
+        API + "/analysis",
+        data=body,
+        headers=_headers({"Content-Type": "multipart/form-data; boundary=" + boundary}),
+        method="POST",
+    )
+    with urllib.request.urlopen(req, timeout=120) as r:
+        raw = r.read()
+        result = json.loads(raw) if raw else {}
+    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/anyrun/scripts/detonate_url.py

@@ -0,0 +1,52 @@
+import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.any.run/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _headers(extra=None):
+    h = {"Accept": "application/json", "Authorization": "API-Key " + str(_cfg().get("api_key") or "")}
+    if extra:
+        h.update(extra)
+    return h
+
+
+def request(method, path, params=None, form=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 = None
+    headers = _headers()
+    if form is not None:
+        data = urllib.parse.urlencode({k: str(v) for k, v in form.items() if v not in (None, "")}).encode("utf-8")
+        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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    url = inputs.get("url")
+    if not url:
+        raise Exception("url is required")
+    env_os = inputs.get("os") or "windows"
+    env_bitness = inputs.get("env_bitness") or 64
+
+    result = request("POST", "/analysis", form={"obj_type": "url", "obj_url": url, "env_os": env_os, "env_bitness": env_bitness})
+    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/anyrun/scripts/get_history.py

@@ -0,0 +1,49 @@
+import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.any.run/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _headers(extra=None):
+    h = {"Accept": "application/json", "Authorization": "API-Key " + str(_cfg().get("api_key") or "")}
+    if extra:
+        h.update(extra)
+    return h
+
+
+def request(method, path, params=None, form=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 = None
+    headers = _headers()
+    if form is not None:
+        data = urllib.parse.urlencode({k: str(v) for k, v in form.items() if v not in (None, "")}).encode("utf-8")
+        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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    limit = inputs.get("limit") or 25
+    skip = inputs.get("skip")
+
+    result = request("GET", "/analysis", params={"limit": limit, "skip": skip})
+    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/anyrun/scripts/get_report.py

@@ -0,0 +1,53 @@
+import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.any.run/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _headers(extra=None):
+    h = {"Accept": "application/json", "Authorization": "API-Key " + str(_cfg().get("api_key") or "")}
+    if extra:
+        h.update(extra)
+    return h
+
+
+def request(method, path, params=None, form=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 = None
+    headers = _headers()
+    if form is not None:
+        data = urllib.parse.urlencode({k: str(v) for k, v in form.items() if v not in (None, "")}).encode("utf-8")
+        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=90) 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", "{}"))
+    task_id = inputs.get("task_id")
+    if not task_id:
+        raise Exception("task_id is required")
+
+    result = request("GET", "/analysis/" + q(task_id))
+    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/anyrun/scripts/get_user_limits.py

@@ -0,0 +1,45 @@
+import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.any.run/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _headers(extra=None):
+    h = {"Accept": "application/json", "Authorization": "API-Key " + str(_cfg().get("api_key") or "")}
+    if extra:
+        h.update(extra)
+    return h
+
+
+def request(method, path, params=None, form=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 = None
+    headers = _headers()
+    if form is not None:
+        data = urllib.parse.urlencode({k: str(v) for k, v in form.items() if v not in (None, "")}).encode("utf-8")
+        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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def main():
+    result = request("GET", "/user/limits")
+    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/anyrun/scripts/test_connection.py

@@ -0,0 +1,47 @@
+import base64, json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.any.run/v1"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _headers(extra=None):
+    h = {"Accept": "application/json", "Authorization": "API-Key " + str(_cfg().get("api_key") or "")}
+    if extra:
+        h.update(extra)
+    return h
+
+
+def request(method, path, params=None, form=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 = None
+    headers = _headers()
+    if form is not None:
+        data = urllib.parse.urlencode({k: str(v) for k, v in form.items() if v not in (None, "")}).encode("utf-8")
+        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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def main():
+    result = request("GET", "/user/limits")
+    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)