Add VirusTotal Sample Downloader expansion module

documentation/expansion/virustotal_downloader.md

@@ -0,0 +1,39 @@
+# VirusTotal Sample Downloader
+
+Download an existing file from VirusTotal by hash.
+
+Requires a VirusTotal API key with file download entitlement. Uses GET /api/v3/files/{hash}/download. Streams at most max_size_mb (default 50, range 1-100) and verifies the downloaded file against the requested hash. Signed HTTPS redirects are followed without forwarding the API key. Returns raw bytes as a MISP attachment in a file object with MD5, SHA-1 and SHA-256 hashes; no sample is executed or submitted.
+
+Input: md5, sha1, sha256. Output: A file object with a binary attachment and verified hashes.
+
+Configuration: apikey, max_size_mb. Required: apikey.
+Requests use a 30 second timeout and verify TLS certificates. Errors distinguish
+missing configuration, invalid input, authentication failures, rate limits and
+invalid responses. No API credentials are included in error messages.
+
+All new enrichment attributes use `to_ids=false`. Source attributes retain their
+UUID and metadata. Returned objects reference the source attribute where possible.
+Empty lookups return the source attribute without fabricated enrichment.
+
+References:
+
+- https://docs.virustotal.com/reference/files-download
+
+Example request:
+
+```json
+{
+  "module": "virustotal_downloader",
+  "attribute": {
+    "type": "sha256",
+    "value": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+    "uuid": "32876e4a-916d-4935-9e51-7b69a76718f8"
+  },
+  "config": {
+    "apikey": "YOUR_API_KEY"
+  }
+}
+```
+
+Validation: `pytest tests/test_virustotal_downloader.py` uses mocked service responses;
+no paid API credentials are needed.

misp_modules/modules/expansion/virustotal_downloader.py

@@ -0,0 +1,156 @@
+import hashlib
+import json
+import re
+from io import BytesIO
+from urllib.parse import urljoin, urlsplit
+
+import requests
+from pymisp import MISPEvent, MISPObject
+
+misperrors = {"error": "Error"}
+mispattributes = {"input": ["md5", "sha1", "sha256"], "format": "misp_standard"}
+moduleconfig = ["apikey", "max_size_mb"]
+moduleinfo = {
+    "version": "1",
+    "author": "MISP Project",
+    "name": "VirusTotal Sample Downloader",
+    "description": "Download an existing file from VirusTotal by hash.",
+    "module-type": ["expansion"],
+    "logo": "",
+    "requirements": ["requests"],
+    "features": (
+        "Requires a VirusTotal API key with file download entitlement. Uses GET /api/v3/files/{hash}/download. Streams"
+        " at most max_size_mb (default 50, range 1-100) and verifies the downloaded file against the requested hash."
+        " Signed HTTPS redirects are followed without forwarding the API key. Returns raw bytes as a MISP attachment in"
+        " a file object with MD5, SHA-1 and SHA-256 hashes; no sample is executed or submitted."
+    ),
+    "references": ["https://docs.virustotal.com/reference/files-download"],
+    "input": "md5, sha1, sha256 MISP attribute.",
+    "output": "A file object with a binary attachment and verified hashes.",
+}
+TIMEOUT = 30
+
+
+class ModuleError(ValueError):
+    """An actionable module error safe to return to the caller."""
+
+
+def _hash(kind, value):
+    if not isinstance(value, str) or not re.fullmatch(
+        "[0-9a-fA-F]{%d}" % {"md5": 32, "sha1": 40, "sha256": 64}[kind], value
+    ):
+        raise ModuleError(f"Invalid {kind} hash.")
+    return value.lower()
+
+
+def _limit(config, key="max_results", default=25, maximum=100):
+    raw = config.get(key)
+    if raw is None or raw == "":
+        return default
+    if isinstance(raw, bool) or not str(raw).isdigit() or (not 1 <= int(raw) <= maximum):
+        raise ModuleError(f"{key} must be an integer between 1 and {maximum}.")
+    return int(raw)
+
+
+def _results(event):
+    data = json.loads(event.to_json())
+    return {"results": {key: data.get(key, []) for key in ("Attribute", "Object")}}
+
+
+def handler(q=False):
+    if q is False:
+        return False
+    try:
+        request = json.loads(q)
+        if not isinstance(request, dict):
+            raise ModuleError("The request must be a JSON object.")
+        attribute = request.get("attribute")
+        if not isinstance(attribute, dict) or not all((attribute.get(k) for k in ("type", "value", "uuid"))):
+            raise ModuleError("An attribute containing type, value and uuid is required.")
+        if attribute["type"] not in mispattributes["input"]:
+            raise ModuleError("Unsupported attribute type.")
+        if not isinstance(attribute["value"], str) or not attribute["value"].strip():
+            raise ModuleError("The attribute value must be a nonempty string.")
+        config = request.get("config") or {}
+        if not isinstance(config, dict):
+            raise ModuleError("The config must be a JSON object.")
+        for key in ["apikey"]:
+            if not isinstance(config.get(key), str) or not config[key].strip():
+                raise ModuleError(f"Missing {key} configuration.")
+        event = MISPEvent()
+        event.add_attribute(**attribute)
+        return _expand(event, attribute, config)
+    except ModuleError as exc:
+        return {"error": str(exc)}
+    except (ValueError, TypeError, KeyError, OverflowError, OSError):
+        return {"error": "Invalid request or service response."}
+
+
+def introspection():
+    return mispattributes
+
+
+def version():
+    return dict(moduleinfo, config=moduleconfig)
+
+
+def _download(value, apikey, maximum):
+    url = f"https://www.virustotal.com/api/v3/files/{value}/download"
+    headers = {"x-apikey": apikey}
+    for attempt in range(4):
+        try:
+            response = requests.get(url, headers=headers, timeout=TIMEOUT, stream=True, allow_redirects=False)
+        except requests.RequestException:
+            raise ModuleError("VirusTotal download failed or timed out.") from None
+        try:
+            if response.status_code in (301, 302, 303, 307, 308):
+                target = urljoin(url, response.headers.get("Location", ""))
+                parsed = urlsplit(target)
+                if (
+                    not response.headers.get("Location")
+                    or parsed.scheme != "https"
+                    or (not parsed.hostname)
+                    or parsed.username
+                    or parsed.password
+                ):
+                    raise ModuleError("VirusTotal returned an invalid download redirect.")
+                url, headers = (target, {})
+                continue
+            if response.status_code == 404:
+                return None
+            if response.status_code != 200:
+                raise ModuleError(f"VirusTotal returned HTTP {response.status_code}.")
+            length = response.headers.get("Content-Length")
+            if length is not None and (not length.isdigit() or int(length) > maximum):
+                raise ModuleError("The download exceeds max_size_mb or has an invalid Content-Length.")
+            content = BytesIO()
+            for chunk in response.iter_content(chunk_size=65536):
+                if content.tell() + len(chunk) > maximum:
+                    raise ModuleError("The download exceeds max_size_mb.")
+                content.write(chunk)
+            return content.getvalue()
+        except requests.RequestException:
+            raise ModuleError("VirusTotal download failed or timed out.") from None
+        finally:
+            response.close()
+    raise ModuleError("VirusTotal returned too many download redirects.")
+
+
+def _expand(event, attribute, config):
+    kind = attribute["type"]
+    value = _hash(kind, attribute["value"].strip())
+    maximum = _limit(config, "max_size_mb", 50, 100) * 1024 * 1024
+    content = _download(value, config["apikey"], maximum)
+    if content is None:
+        return _results(event)
+    if hashlib.new(kind, content).hexdigest() != value:
+        raise ModuleError("The downloaded file does not match the requested hash.")
+    obj = MISPObject("file")
+    obj.comment = "Downloaded from VirusTotal"
+    obj.add_attribute("attachment", value=value + ".bin", data=BytesIO(content), to_ids=False)
+    obj.add_attribute("size-in-bytes", value=len(content), to_ids=False)
+    for algorithm in ("md5", "sha1", "sha256"):
+        obj.add_attribute(algorithm, value=hashlib.new(algorithm, content).hexdigest(), to_ids=False)
+    obj.add_reference(attribute["uuid"], "related-to")
+    event.add_object(obj)
+    return _results(event)

tests/test_virustotal_downloader.py

@@ -0,0 +1,192 @@
+import json
+from unittest.mock import Mock, patch
+
+import pytest
+import requests
+
+from misp_modules.modules.expansion import virustotal_downloader as module
+
+UUID = "32876e4a-916d-4935-9e51-7b69a76718f8"
+SECRET = "test-api-secret"
+
+
+def query(kind="sha256", value="8e13f6c598de092f99affe5c64d3aca48f4b4e0bea6e396bc257c40482674e3a", config=None):
+    return json.dumps(
+        {
+            "attribute": {"type": kind, "value": value, "uuid": UUID, "to_ids": True, "comment": "source metadata"},
+            "config": {"apikey": "test-api-secret"} if config is None else config,
+        }
+    )
+
+
+def response(payload=None, status=200):
+    result = Mock(status_code=status, headers={})
+    result.json.return_value = payload
+    result.iter_content.return_value = iter([b"sample bytes"])
+    return result
+
+
+def handle(payload, **kwargs):
+    with patch.object(module.requests, "get", return_value=response(payload)) as get:
+        result = module.handler(query(**kwargs))
+    return result, get
+
+
+def objects(result, name):
+    return [obj for obj in result["results"]["Object"] if obj["name"] == name]
+
+
+def values(obj):
+    return {item["object_relation"]: item["value"] for item in obj["Attribute"]}
+
+
+def assert_source(result):
+    source = next(item for item in result["results"]["Attribute"] if item["uuid"] == UUID)
+    assert source["to_ids"] is True
+    assert source["comment"] == "source metadata"
+    for obj in result["results"]["Object"]:
+        assert any(ref["referenced_uuid"] == UUID for ref in obj["ObjectReference"])
+        assert all(item["to_ids"] is False for item in obj["Attribute"])
+    assert all(item["to_ids"] is False for item in result["results"]["Attribute"] if item["uuid"] != UUID)
+
+
+def test_discovery_contract():
+    assert module.handler(False) is False
+    assert module.introspection()["format"] == "misp_standard"
+    assert module.version()["config"] == module.moduleconfig
+    assert "expansion" in module.version()["module-type"]
+
+
+@pytest.mark.parametrize(
+    "raw", ["{", "[]", "null", "{}", '{"attribute": []}', '{"attribute":{"type":"ip-dst","value":"1.1.1.1"}}']
+)
+def test_invalid_request_never_calls_service(raw):
+    with patch.object(module.requests, "get") as get:
+        assert "error" in module.handler(raw)
+    get.assert_not_called()
+
+
+def test_unsupported_type_never_calls_service():
+    with patch.object(module.requests, "get") as get:
+        assert "error" in module.handler(query(kind="comment"))
+    get.assert_not_called()
+
+
+@pytest.mark.parametrize("value", ["", "   ", 123, ["example.com"]])
+def test_invalid_values_never_call_service(value):
+    with patch.object(module.requests, "get") as get:
+        assert "error" in module.handler(query(value=value))
+    get.assert_not_called()
+
+
+@pytest.mark.parametrize("status", [401, 403, 429, 500])
+def test_http_failures_are_reported(status):
+    with patch.object(module.requests, "get", return_value=response(status=status)):
+        result = module.handler(query())
+    assert str(status) in result["error"]
+    assert SECRET not in result["error"]
+
+
+@pytest.mark.parametrize("failure", [requests.Timeout, requests.ConnectionError])
+def test_network_failures_do_not_leak_credentials(failure):
+    with patch.object(module.requests, "get", side_effect=failure("url?apikey=" + SECRET)):
+        result = module.handler(query())
+    assert "error" in result
+    assert SECRET not in result["error"]
+
+
+@pytest.mark.parametrize("status", [404])
+def test_empty_service_response_preserves_source(status):
+    with patch.object(module.requests, "get", return_value=response(status=status)):
+        result = module.handler(query())
+    assert_source(result)
+    assert result["results"]["Object"] == []
+
+
+def test_missing_configuration_never_calls_service():
+    with patch.object(module.requests, "get") as get:
+        assert "error" in module.handler(query(config={}))
+    get.assert_not_called()
+
+
+import base64
+import hashlib
+
+
+def downloaded(content=b"sample bytes", status=200, headers=None):
+    reply = response(status=status)
+    reply.headers = headers or {}
+    reply.iter_content.return_value = iter([content])
+    return reply
+
+
+@pytest.mark.parametrize("kind", ["md5", "sha1", "sha256"])
+def test_binary_download_and_all_hashes(kind):
+    content = b"sample bytes"
+    digest = hashlib.new(kind, content).hexdigest()
+    reply = downloaded(content)
+    with patch.object(module.requests, "get", return_value=reply) as get:
+        result = module.handler(query(kind=kind, value=digest))
+    assert_source(result)
+    obj = objects(result, "file")[0]
+    attachment = next(x for x in obj["Attribute"] if x["object_relation"] == "attachment")
+    assert base64.b64decode(attachment["data"]) == content
+    assert values(obj)["sha256"] == hashlib.sha256(content).hexdigest()
+    assert get.call_args.kwargs["stream"] is True
+    assert get.call_args.kwargs["timeout"] == 30
+    reply.close.assert_called_once()
+
+
+def test_signed_redirect_receives_no_api_key():
+    replies = [
+        downloaded(status=302, headers={"Location": "https://download.example/sample?signature=signed"}),
+        downloaded(),
+    ]
+    with patch.object(module.requests, "get", side_effect=replies) as get:
+        result = module.handler(query())
+    assert_source(result)
+    assert get.call_args_list[0].kwargs["headers"] == {"x-apikey": SECRET}
+    assert get.call_args_list[1].kwargs["headers"] == {}
+    assert get.call_args_list[1].args[0] == "https://download.example/sample?signature=signed"
+    assert all(reply.close.call_count == 1 for reply in replies)
+
+
+def test_hash_mismatch_rejects_the_download():
+    with patch.object(module.requests, "get", return_value=downloaded(b"wrong")):
+        result = module.handler(query())
+    assert "does not match" in result["error"]
+
+
+@pytest.mark.parametrize(
+    "headers,body",
+    [({"Content-Length": "1048577"}, b"x"), ({}, b"x" * 1048577)],
+    ids=["declared-size", "streamed-size"],
+)
+def test_download_size_cap_applies_with_and_without_content_length(headers, body):
+    reply = downloaded(body, headers=headers)
+    with patch.object(module.requests, "get", return_value=reply):
+        result = module.handler(query(config={"apikey": SECRET, "max_size_mb": "1"}))
+    assert "max_size_mb" in result["error"]
+    reply.close.assert_called_once()
+
+
+@pytest.mark.parametrize("target", ["http://download.example/sample", "https://user:pass@download.example/sample", ""])
+def test_invalid_redirects_are_rejected(target):
+    with patch.object(module.requests, "get", return_value=downloaded(status=302, headers={"Location": target})) as get:
+        assert "error" in module.handler(query())
+    assert get.call_count == 1
+
+
+def test_redirect_loop_is_bounded():
+    with patch.object(
+        module.requests,
+        "get",
+        return_value=downloaded(status=302, headers={"Location": "https://download.example/loop"}),
+    ) as get:
+        assert "redirects" in module.handler(query())["error"]
+    assert get.call_count == 4
+
+
+def test_empty_http_success_is_reported_as_an_error():
+    with patch.object(module.requests, "get", return_value=downloaded(status=204)):
+        assert "204" in module.handler(query())["error"]