From aac05f919b7bd566e029bb30479617eab4dc2f0a Mon Sep 17 00:00:00 2001 From: Alexandre Dulaunoy Date: Thu, 29 Oct 2026 18:08:53 +0100 Subject: [PATCH 1/2] feat: [fortisandbox] add hash ratings and sample submissions --- documentation/expansion/fortisandbox.md | 11 + .../modules/expansion/fortisandbox.py | 270 ++++++++++++++++++ tests/test_fortisandbox.py | 113 ++++++++ 3 files changed, 394 insertions(+) create mode 100644 documentation/expansion/fortisandbox.md create mode 100644 misp_modules/modules/expansion/fortisandbox.py create mode 100644 tests/test_fortisandbox.py diff --git a/documentation/expansion/fortisandbox.md b/documentation/expansion/fortisandbox.md new file mode 100644 index 00000000..a94baaf4 --- /dev/null +++ b/documentation/expansion/fortisandbox.md @@ -0,0 +1,11 @@ +# FortiSandbox Enrichment + +Authenticate to a FortiSandbox appliance and retrieve an existing rating by MD5, SHA-1 or SHA-256. Set api_url to the appliance HTTPS origin, without /jsonrpc. For supplied files, calculate SHA-256 and look up the rating first. submit_samples defaults to false; true permits submission when the rating is unknown. max_size_mb defaults to 25 (1–100). A submission receipt is returned without polling. Provider ratings remain in their original vocabulary. + +Input: md5, sha1, sha256, attachment, malware-sample. + +Configuration: api_url, username, password, submit_samples, max_size_mb, max_results. Required: api_url, username, password. + +max_results defaults to 25 and accepts 1–100. New attributes use `to_ids=false`; the source UUID, detection flag and metadata are preserved. Objects reference the source attribute. Requests verify TLS, reject redirects and use connection/read timeouts of 5/30 seconds. API errors do not echo credentials. No automatic submission retries or background polling occur. + +Reference: https://www.fortinet.com/products/sandbox/fortisandbox diff --git a/misp_modules/modules/expansion/fortisandbox.py b/misp_modules/modules/expansion/fortisandbox.py new file mode 100644 index 00000000..8e5476e6 --- /dev/null +++ b/misp_modules/modules/expansion/fortisandbox.py @@ -0,0 +1,270 @@ +import base64 +import hashlib +import io +import ipaddress +import json +import re +import zipfile +from urllib.parse import urlsplit +from uuid import UUID +import requests +from pymisp import MISPEvent, MISPObject +from pymisp.exceptions import PyMISPError + + +class ModuleError(ValueError): + """A diagnostic safe to show to the analyst.""" + + +def _integer(config, key, default, maximum=100): + value = config.get(key) + if value is None or value == "": + return default + if isinstance(value, bool) or not str(value).isdigit() or (not 1 <= int(value) <= maximum): + raise ModuleError(f"{key} must be an integer between 1 and {maximum}.") + return int(value) + + +def _enabled(config, key, default=False): + value = config.get(key) + if value is None or value == "": + return default + if isinstance(value, bool): + return value + if str(value).lower() not in ("true", "false", "1", "0"): + raise ModuleError(f"{key} must be true or false.") + return str(value).lower() in ("true", "1") + + +def _url(value): + if not isinstance(value, str) or any(ord(c) <= 32 or ord(c) == 127 for c in value): + raise ModuleError("Invalid HTTPS URL.") + parsed = urlsplit(value) + if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.fragment: + raise ModuleError("An HTTPS URL without credentials or fragment is required.") + parsed.port + return value.rstrip("/") + + +def _origin(config): + value = _url(config.get("api_url", "")) + if urlsplit(value).query: + raise ModuleError("api_url cannot contain a query string.") + return value + + +def _normalise(kind, value): + if not isinstance(value, str) or not value.strip() or any(ord(c) < 32 for c in value): + raise ModuleError("A nonempty string value is required.") + value = value.strip() + if kind in ("md5", "sha1", "sha256", "x509-fingerprint-sha256"): + length = {"md5": 32, "sha1": 40}.get(kind, 64) + if not re.fullmatch("[a-fA-F0-9]{%d}" % length, value): + raise ModuleError("Invalid hash.") + return value.lower() + if kind in ("ip-src", "ip-dst"): + return str(ipaddress.ip_address(value)) + if kind in ("domain", "hostname"): + value = value.rstrip(".").encode("idna").decode("ascii").lower() + if len(value) > 253 or not all( + re.fullmatch("[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", p) for p in value.split(".") + ): + raise ModuleError("Invalid DNS name.") + if kind == "vulnerability": + value = value.upper() + if not re.fullmatch("CVE-[0-9]{4}-[0-9]{4,}", value): + raise ModuleError("A CVE identifier is required.") + return value + + +def _call(method, url, **kwargs): + try: + response = requests.request(method, url, timeout=(5, 30), verify=True, allow_redirects=False, **kwargs) + except requests.RequestException: + raise ModuleError( + "The service request failed or timed out; check the service before repeating a submission." + ) from None + if response.status_code in (204, 404) and method == "GET": + return None + if response.status_code not in (200, 201, 202): + raise ModuleError(f"The service returned HTTP {response.status_code}.") + try: + return response.json() + except ValueError: + raise ModuleError("The service returned invalid JSON.") from None + + +def _object(event, attribute, name): + obj = MISPObject(name) + obj.add_reference(attribute["uuid"], "describes") + event.add_object(obj) + return obj + + +def _annotation(event, attribute, title, data): + obj = _object(event, attribute, "annotation") + obj.comment = title + obj.add_attribute("text", value=json.dumps(data, sort_keys=True), to_ids=False, disable_correlation=True) + return obj + + +def _sandbox(event, attribute, payload, link=None): + obj = _object(event, attribute, "sandbox-report") + obj.add_attribute("sandbox-type", value=moduleinfo["name"], to_ids=False) + obj.add_attribute("results", value=json.dumps(payload, sort_keys=True), to_ids=False, disable_correlation=True) + if link: + obj.add_attribute("permalink", value=link, to_ids=False) + return obj + + +def _binary(attribute, config): + limit = _integer(config, "max_size_mb", 25, 100) * 1024 * 1024 + encoded = attribute.get("data") + if not isinstance(encoded, str) or len(encoded) > 2 * limit: + raise ModuleError("Missing or oversized base64 sample data.") + content = base64.b64decode(encoded, validate=True) + filename = attribute["value"].split("|", 1)[0].replace("\\", "/").rsplit("/", 1)[-1] + if attribute["type"] == "malware-sample": + with zipfile.ZipFile(io.BytesIO(content)) as archive: + entries = [entry for entry in archive.infolist() if not entry.is_dir()] + if len(entries) != 1 or entries[0].file_size > limit: + raise ModuleError("A malware-sample archive must contain one file within max_size_mb.") + content = archive.read(entries[0], pwd=b"infected") + if not filename or not content or len(content) > limit: + raise ModuleError("The sample is empty or exceeds max_size_mb.") + return (filename, content, hashlib.sha256(content).hexdigest()) + + +def handler(q=False): + if q is False: + return False + try: + request = json.loads(q) + if not isinstance(request, dict) or not isinstance(request.get("attribute"), dict): + raise ModuleError("An attribute containing type, value and uuid is required.") + attribute = request["attribute"] + UUID(attribute["uuid"]) + if attribute["type"] not in mispattributes["input"]: + raise ModuleError("Unsupported attribute type.") + value = _normalise(attribute["type"], attribute["value"]) + config = request.get("config") or {} + if not isinstance(config, dict): + raise ModuleError("The config must be an object.") + for key in REQUIRED: + if not isinstance(config.get(key), str) or not config[key].strip(): + raise ModuleError(f"Missing {key} configuration.") + limit = _integer(config, "max_results", 25) + event = MISPEvent() + event.add_attribute(**attribute) + _expand(event, attribute, value, config, limit) + result = json.loads(event.to_json()) + return {"results": {key: result.get(key, []) for key in ("Attribute", "Object")}} + except ModuleError as exc: + return {"error": str(exc)} + except ( + ValueError, + TypeError, + KeyError, + AttributeError, + IndexError, + OSError, + RuntimeError, + zipfile.BadZipFile, + PyMISPError, + ): + return {"error": "Invalid request or service response."} + + +def introspection(): + return mispattributes + + +def version(): + return dict(moduleinfo, config=moduleconfig) + + +moduleinfo = { + "version": "1.0", + "author": "MISP Project", + "name": "FortiSandbox Enrichment", + "description": ( + "Authenticate to a FortiSandbox appliance and retrieve an existing rating by MD5, SHA-1 or SHA-256. Set api_url" + " to the appliance HTTPS origin, without /jsonrpc. For supplied files, calculate SHA-256 and look up the rating" + " first. submit_samples defaults to false; true permits submission when the rating is unknown. max_size_mb" + " defaults to 25 (1–100). A submission receipt is returned without polling. Provider ratings remain in their" + " original vocabulary." + ), + "module-type": ["expansion"], + "data-source": "external", + "logo": "", + "requirements": ["requests", "pymisp"], + "features": ( + "Authenticate to a FortiSandbox appliance and retrieve an existing rating by MD5, SHA-1 or SHA-256. Set api_url" + " to the appliance HTTPS origin, without /jsonrpc. For supplied files, calculate SHA-256 and look up the rating" + " first. submit_samples defaults to false; true permits submission when the rating is unknown. max_size_mb" + " defaults to 25 (1–100). A submission receipt is returned without polling. Provider ratings remain in their" + " original vocabulary." + ), + "references": ["https://www.fortinet.com/products/sandbox/fortisandbox"], + "input": "md5, sha1, sha256, attachment, malware-sample", + "output": "MISP objects linked to the original attribute.", +} +mispattributes = {"input": ["md5", "sha1", "sha256", "attachment", "malware-sample"], "format": "misp_standard"} +moduleconfig = ["api_url", "username", "password", "submit_samples", "max_size_mb", "max_results"] +REQUIRED = ["api_url", "username", "password"] + + +def _rpc(base, method, path, session=None, **params): + body = {"method": method, "params": [{"url": path, **params}], "id": 1, "ver": "2.1"} + if session: + body["session"] = session + data = _call("POST", base + "/jsonrpc", json=body) + if not isinstance(data, dict): + raise ModuleError("FortiSandbox returned invalid JSON-RPC data.") + result = data.get("result", {}) + if isinstance(result, list): + result = result[0] if result else {} + if not isinstance(result, dict) or result.get("status", {}).get("code", 0) != 0: + raise ModuleError("FortiSandbox rejected the JSON-RPC operation.") + return (data, result) + + +def _expand(event, attribute, value, config, limit): + base = _origin(config) + if urlsplit(base).path not in ("", "/"): + raise ModuleError("api_url must be a FortiSandbox HTTPS origin without a path.") + login, _ = _rpc(base, "exec", "/sys/login/user", data=[{"user": config["username"], "passwd": config["password"]}]) + session = login.get("session") + if not isinstance(session, str) or not session: + raise ModuleError("FortiSandbox authentication failed.") + try: + kind = attribute["type"] + sample = None + if kind in ("attachment", "malware-sample"): + sample = _binary(attribute, config) + kind, value = ("sha256", sample[2]) + _, report = _rpc(base, "get", "/scan/result/filerating", session, checksum=value, ctype=kind) + payload = report.get("data", report) + if isinstance(payload, list): + payload = payload[0] if payload else {} + rating = payload.get("rating") if isinstance(payload, dict) else None + if rating is not None and str(rating).lower() not in ("unknown", ""): + _sandbox(event, attribute, payload) + elif sample and _enabled(config, "submit_samples"): + _, receipt = _rpc( + base, + "set", + "/alert/ondemand/submit-file", + session, + type="file", + filename=sample[0], + file=base64.b64encode(sample[1]).decode(), + ) + _annotation(event, attribute, "FortiSandbox submission receipt", receipt) + else: + _annotation(event, attribute, "FortiSandbox lookup", {"hash": value, "rating": "unknown"}) + finally: + try: + _rpc(base, "exec", "/sys/logout", session) + except ModuleError: + pass diff --git a/tests/test_fortisandbox.py b/tests/test_fortisandbox.py new file mode 100644 index 00000000..6a2a8808 --- /dev/null +++ b/tests/test_fortisandbox.py @@ -0,0 +1,113 @@ +import json +from unittest.mock import Mock, patch + +import pytest +import requests + +from misp_modules.modules.expansion import fortisandbox as module + +UUID = "32876e4a-916d-4935-9e51-7b69a76718f8" +SECRET = "test-service-secret" + + +def query(kind=None, value="example.com", config=None, **attribute_fields): + attribute = { + "type": kind or module.mispattributes["input"][0], + "value": value, + "uuid": UUID, + "to_ids": True, + "comment": "source context", + **attribute_fields, + } + return json.dumps( + {"attribute": attribute, "config": config if config is not None else {k: SECRET for k in module.REQUIRED}} + ) + + +def reply(payload=None, status=200): + return Mock(status_code=status, headers={}, json=Mock(return_value=payload)) + + +def assert_source(result): + assert "results" in result, result + source = next(a for a in result["results"]["Attribute"] if a["uuid"] == UUID) + assert source["to_ids"] is True + assert source["comment"] == "source context" + for obj in result["results"]["Object"]: + assert any(r["referenced_uuid"] == UUID for r in obj["ObjectReference"]) + assert all(a["to_ids"] is False for a in obj["Attribute"]) + + +def objects(result, name): + return [obj for obj in result["results"]["Object"] if obj["name"] == name] + + +def values(obj, relation): + return [a["value"] for a in obj["Attribute"] if a["object_relation"] == relation] + + +def test_discovery(): + assert module.handler(False) is False + assert module.introspection()["format"] == "misp_standard" + assert module.version()["config"] == module.moduleconfig + + +@pytest.mark.parametrize("raw", ["{", "[]", "null", "{}", '{"attribute": []}']) +def test_invalid_envelope(raw): + with patch.object(module.requests, "request") as request: + assert "error" in module.handler(raw) + request.assert_not_called() + + +@pytest.mark.parametrize("status", [301, 401, 403, 429, 503]) +def test_http_errors_are_sanitized(status): + with patch.object(module.requests, "request", return_value=reply({"error": SECRET}, status)): + with pytest.raises(module.ModuleError) as failure: + module._call("GET", "https://service.example/report") + assert str(status) in str(failure.value) + assert SECRET not in str(failure.value) + + +def test_timeout_is_bounded_and_sanitized(): + with patch.object(module.requests, "request", side_effect=requests.Timeout(SECRET)) as request: + with pytest.raises(module.ModuleError) as failure: + module._call("POST", "https://service.example/submit") + assert SECRET not in str(failure.value) + assert request.call_count == 1 + assert request.call_args.kwargs["verify"] is True + assert request.call_args.kwargs["allow_redirects"] is False + assert request.call_args.kwargs["timeout"] == (5, 30) + + +@pytest.mark.parametrize("raw", [0, -1, True, "bad", 101]) +def test_invalid_result_limit(raw): + with pytest.raises(module.ModuleError): + module._integer({"max_results": raw}, "max_results", 25) + + +def test_rating_query_uses_correct_checksum_type_and_logs_out(): + responses = [ + reply({"session": "session-id", "result": {"status": {"code": 0}}}), + reply({"result": {"status": {"code": 0}, "data": {"rating": "Clean"}}}), + reply({"result": {"status": {"code": 0}}}), + ] + config = {"api_url": "https://sandbox.example", "username": "analyst", "password": SECRET} + with patch.object(module.requests, "request", side_effect=responses) as request: + result = module.handler(query(kind="sha256", value="a" * 64, config=config)) + assert_source(result) + assert request.call_args_list[1].kwargs["json"]["params"][0]["ctype"] == "sha256" + assert request.call_args_list[-1].kwargs["json"]["params"][0]["url"] == "/sys/logout" + assert objects(result, "sandbox-report") + + +def test_failed_rating_still_closes_session_and_hides_password(): + responses = [ + reply({"session": "session-id"}), + reply({"result": {"status": {"code": -1, "message": SECRET}}}), + reply({"result": {}}), + ] + config = {"api_url": "https://sandbox.example", "username": "analyst", "password": SECRET} + with patch.object(module.requests, "request", side_effect=responses) as request: + result = module.handler(query(kind="md5", value="a" * 32, config=config)) + assert "error" in result and SECRET not in str(result) + assert request.call_count == 3 From e3f8a371cf43ba771039f5f47fb0ba36db92c042 Mon Sep 17 00:00:00 2001 From: Alexandre Dulaunoy Date: Thu, 29 Oct 2026 23:24:35 +0100 Subject: [PATCH 2/2] fix: [fortisandbox] address review findings --- documentation/expansion/fortisandbox.md | 6 +- .../modules/expansion/fortisandbox.py | 32 +++++++---- tests/test_fortisandbox.py | 57 ++++++++++++++++++- 3 files changed, 81 insertions(+), 14 deletions(-) diff --git a/documentation/expansion/fortisandbox.md b/documentation/expansion/fortisandbox.md index a94baaf4..2b862a5b 100644 --- a/documentation/expansion/fortisandbox.md +++ b/documentation/expansion/fortisandbox.md @@ -1,11 +1,13 @@ # FortiSandbox Enrichment -Authenticate to a FortiSandbox appliance and retrieve an existing rating by MD5, SHA-1 or SHA-256. Set api_url to the appliance HTTPS origin, without /jsonrpc. For supplied files, calculate SHA-256 and look up the rating first. submit_samples defaults to false; true permits submission when the rating is unknown. max_size_mb defaults to 25 (1–100). A submission receipt is returned without polling. Provider ratings remain in their original vocabulary. +Authenticate to a FortiSandbox appliance and retrieve an existing rating by SHA-1 or SHA-256. Set api_url to the appliance HTTPS origin, without /jsonrpc. For supplied files, calculate SHA-256 and look up the rating first. submit_samples defaults to false; true permits submission when the rating is unknown. max_size_mb defaults to 25 (1–100). A submission receipt is returned without polling. Provider ratings remain in their original vocabulary. -Input: md5, sha1, sha256, attachment, malware-sample. +Input: sha1, sha256, attachment, malware-sample. Configuration: api_url, username, password, submit_samples, max_size_mb, max_results. Required: api_url, username, password. max_results defaults to 25 and accepts 1–100. New attributes use `to_ids=false`; the source UUID, detection flag and metadata are preserved. Objects reference the source attribute. Requests verify TLS, reject redirects and use connection/read timeouts of 5/30 seconds. API errors do not echo credentials. No automatic submission retries or background polling occur. Reference: https://www.fortinet.com/products/sandbox/fortisandbox + +MD5-only lookup is not supported by the appliance single-file rating endpoint. diff --git a/misp_modules/modules/expansion/fortisandbox.py b/misp_modules/modules/expansion/fortisandbox.py index 8e5476e6..f025a9fb 100644 --- a/misp_modules/modules/expansion/fortisandbox.py +++ b/misp_modules/modules/expansion/fortisandbox.py @@ -110,7 +110,8 @@ def _annotation(event, attribute, title, data): def _sandbox(event, attribute, payload, link=None): obj = _object(event, attribute, "sandbox-report") - obj.add_attribute("sandbox-type", value=moduleinfo["name"], to_ids=False) + obj.add_attribute("sandbox-type", value="on-premise", to_ids=False) + obj.add_attribute("on-premise-sandbox", value="fortisandbox", to_ids=False) obj.add_attribute("results", value=json.dumps(payload, sort_keys=True), to_ids=False, disable_correlation=True) if link: obj.add_attribute("permalink", value=link, to_ids=False) @@ -123,16 +124,24 @@ def _binary(attribute, config): if not isinstance(encoded, str) or len(encoded) > 2 * limit: raise ModuleError("Missing or oversized base64 sample data.") content = base64.b64decode(encoded, validate=True) - filename = attribute["value"].split("|", 1)[0].replace("\\", "/").rsplit("/", 1)[-1] + filename = attribute["value"] if attribute["type"] == "malware-sample": + filename, separator, digest = filename.rpartition("|") + if not separator or not re.fullmatch(r"[a-fA-F0-9]{32}", digest): + raise ModuleError("A malware-sample value must contain filename|md5.") with zipfile.ZipFile(io.BytesIO(content)) as archive: - entries = [entry for entry in archive.infolist() if not entry.is_dir()] + entries = [ + e for e in archive.infolist() if not e.is_dir() and e.filename != digest.lower() + ".filename.txt" + ] if len(entries) != 1 or entries[0].file_size > limit: - raise ModuleError("A malware-sample archive must contain one file within max_size_mb.") + raise ModuleError("A malware-sample archive must contain one payload within max_size_mb.") content = archive.read(entries[0], pwd=b"infected") + if hashlib.md5(content).hexdigest() != digest.lower(): + raise ModuleError("The unpacked sample does not match the declared MD5.") + filename = filename.replace("\\", "/").rsplit("/", 1)[-1] if not filename or not content or len(content) > limit: raise ModuleError("The sample is empty or exceeds max_size_mb.") - return (filename, content, hashlib.sha256(content).hexdigest()) + return filename, content, hashlib.sha256(content).hexdigest() def handler(q=False): @@ -188,7 +197,7 @@ def version(): "author": "MISP Project", "name": "FortiSandbox Enrichment", "description": ( - "Authenticate to a FortiSandbox appliance and retrieve an existing rating by MD5, SHA-1 or SHA-256. Set api_url" + "Authenticate to a FortiSandbox appliance and retrieve an existing rating by SHA-1 or SHA-256. Set api_url" " to the appliance HTTPS origin, without /jsonrpc. For supplied files, calculate SHA-256 and look up the rating" " first. submit_samples defaults to false; true permits submission when the rating is unknown. max_size_mb" " defaults to 25 (1–100). A submission receipt is returned without polling. Provider ratings remain in their" @@ -199,17 +208,17 @@ def version(): "logo": "", "requirements": ["requests", "pymisp"], "features": ( - "Authenticate to a FortiSandbox appliance and retrieve an existing rating by MD5, SHA-1 or SHA-256. Set api_url" + "Authenticate to a FortiSandbox appliance and retrieve an existing rating by SHA-1 or SHA-256. Set api_url" " to the appliance HTTPS origin, without /jsonrpc. For supplied files, calculate SHA-256 and look up the rating" " first. submit_samples defaults to false; true permits submission when the rating is unknown. max_size_mb" " defaults to 25 (1–100). A submission receipt is returned without polling. Provider ratings remain in their" " original vocabulary." ), "references": ["https://www.fortinet.com/products/sandbox/fortisandbox"], - "input": "md5, sha1, sha256, attachment, malware-sample", + "input": "sha1, sha256, attachment, malware-sample", "output": "MISP objects linked to the original attribute.", } -mispattributes = {"input": ["md5", "sha1", "sha256", "attachment", "malware-sample"], "format": "misp_standard"} +mispattributes = {"input": ["sha1", "sha256", "attachment", "malware-sample"], "format": "misp_standard"} moduleconfig = ["api_url", "username", "password", "submit_samples", "max_size_mb", "max_results"] REQUIRED = ["api_url", "username", "password"] @@ -248,7 +257,8 @@ def _expand(event, attribute, value, config, limit): if isinstance(payload, list): payload = payload[0] if payload else {} rating = payload.get("rating") if isinstance(payload, dict) else None - if rating is not None and str(rating).lower() not in ("unknown", ""): + ratings = rating if isinstance(rating, list) else [rating] + if any(r is not None and str(r).lower() not in ("unknown", "") for r in ratings): _sandbox(event, attribute, payload) elif sample and _enabled(config, "submit_samples"): _, receipt = _rpc( @@ -257,7 +267,7 @@ def _expand(event, attribute, value, config, limit): "/alert/ondemand/submit-file", session, type="file", - filename=sample[0], + filename=base64.b64encode(sample[0].encode("utf-8")).decode(), file=base64.b64encode(sample[1]).decode(), ) _annotation(event, attribute, "FortiSandbox submission receipt", receipt) diff --git a/tests/test_fortisandbox.py b/tests/test_fortisandbox.py index 6a2a8808..1b96ccfe 100644 --- a/tests/test_fortisandbox.py +++ b/tests/test_fortisandbox.py @@ -108,6 +108,61 @@ def test_failed_rating_still_closes_session_and_hides_password(): ] config = {"api_url": "https://sandbox.example", "username": "analyst", "password": SECRET} with patch.object(module.requests, "request", side_effect=responses) as request: - result = module.handler(query(kind="md5", value="a" * 32, config=config)) + result = module.handler(query(kind="sha1", value="a" * 40, config=config)) assert "error" in result and SECRET not in str(result) assert request.call_count == 3 + + +def test_current_sample_zip_metadata_digest_and_pipe_filename(): + import base64, hashlib, io, zipfile + + payload = b"actual sample" + digest = hashlib.md5(payload).hexdigest() + output = io.BytesIO() + with zipfile.ZipFile(output, "w") as archive: + archive.writestr(digest, payload) + archive.writestr(digest + ".filename.txt", "sample|name.exe") + attribute = { + "type": "malware-sample", + "value": "sample|name.exe|" + digest, + "data": base64.b64encode(output.getvalue()).decode(), + } + assert module._binary(attribute, {})[:2] == ("sample|name.exe", payload) + attribute["value"] = "sample|name.exe|" + "0" * 32 + with pytest.raises(module.ModuleError): + module._binary(attribute, {}) + assert ( + module._binary({"type": "attachment", "value": "sample|name.exe", "data": "c2FtcGxl"}, {})[0] + == "sample|name.exe" + ) + + +def test_sandbox_classification_matches_template_contract(): + from pymisp import MISPEvent + + event = MISPEvent() + attribute = {"uuid": UUID} + obj = module._sandbox(event, attribute, {"score": 0}) + fields = {a.object_relation: a.value for a in obj.attributes} + assert fields["sandbox-type"] == "on-premise" + assert fields["on-premise-sandbox"] == "fortisandbox" + assert set(obj._definition.get("required", [])) <= set(fields) + assert set(obj._definition["requiredOneOf"]) & set(fields) + + +def test_unknown_rating_array_submits_encoded_filename(): + import base64 + + replies = [ + reply({"session": "session-id"}), + reply({"result": {"data": {"rating": ["Unknown"]}}}), + reply({"result": {"data": {"sid": "1"}}}), + reply({"result": {}}), + ] + config = {"api_url": "https://sandbox.example", "username": "analyst", "password": SECRET, "submit_samples": True} + with patch.object(module.requests, "request", side_effect=replies) as request: + result = module.handler(query(kind="attachment", value="sample.bin", data="c2FtcGxl", config=config)) + assert_source(result) + params = request.call_args_list[2].kwargs["json"]["params"][0] + assert params["filename"] == base64.b64encode(b"sample.bin").decode() + assert not objects(result, "sandbox-report")