From cf058246235ef709552f1ea625644b442ae13f7e Mon Sep 17 00:00:00 2001 From: Tolga SEZER <19777824+projectboot@users.noreply.github.com> Date: Mon, 7 Sep 2026 19:49:11 +0300 Subject: [PATCH 1/4] add: [malwagon] _malwagon_api.py --- .../modules/expansion/_malwagon_api.py | 337 ++++++++++++++++++ 1 file changed, 337 insertions(+) create mode 100644 misp_modules/modules/expansion/_malwagon_api.py diff --git a/misp_modules/modules/expansion/_malwagon_api.py b/misp_modules/modules/expansion/_malwagon_api.py new file mode 100644 index 00000000..df9383b3 --- /dev/null +++ b/misp_modules/modules/expansion/_malwagon_api.py @@ -0,0 +1,337 @@ +"""Shared client and MISP result builder for the Malwagon expansion modules. + +The leading underscore keeps this file out of module discovery: misp_modules +only imports files that do not start with "_", so this is a library, not a +module. Same arrangement as _assemblyline_api.py. + +Two modules use it: malwagon (free, instant hash lookup) and malwagon_submit +(detonation, which costs a submit quota). Both end up holding the same scan +summary, so the parsing and the MISP object building live here once. +""" + +import re +from urllib.parse import quote, urlparse + +import requests +from pymisp import MISPEvent, MISPObject + +DEFAULT_API_URL = "https://malwagon.com/api/v1" + +# Every request carries an explicit timeout. requests has no default one, so a +# hung endpoint would otherwise hold a MISP enrichment worker open forever. +DEFAULT_TIMEOUT = 30 + +USER_AGENT = "malwagon-misp/1.0 (+https://github.com/MISP/misp-modules)" + +# The scan summary is an allowlist on the server side. Mirroring it here means a +# field added later cannot silently become an attribute nobody reviewed. +SCAN_SUMMARY_KEYS = ( + "scan_id", + "status", + "verdict", + "score", + "module", + "mime", + "size", + "tags", + "submitted_at", + "finished_at", +) + +# Map the vendor verdict onto MISP's own vocabulary rather than passing the +# string through raw: misp:threat-level is the platform's severity scale and +# ioc:artifact-state is what the correlation and filtering side reads. +VERDICT_TAGS = { + "clean": ('misp:threat-level="no-risk"', 'ioc:artifact-state="not-malicious"'), + "suspicious": ('misp:threat-level="medium-risk"',), + "malicious": ('misp:threat-level="high-risk"', 'ioc:artifact-state="malicious"'), +} + +SHA256_RE = re.compile(r"^[0-9a-f]{64}$", re.IGNORECASE) + +# Remote strings become MISP attribute values. Cap them so a hostile or merely +# broken response cannot push an unbounded blob into an event. +MAX_VALUE_LENGTH = 512 +MAX_TAGS = 32 + + +class MalwagonError(Exception): + """An error worth showing to the analyst in the MISP interface. + + Every message raised here is written by this file. Nothing derived from the + submitted sample, and nothing derived from the API key, is ever put in one. + """ + + +def is_sha256(value): + return bool(SHA256_RE.match(str(value or "").strip())) + + +def clean_text(value): + """Flatten one remote value into something safe to put in an attribute.""" + text = str(value).replace("\r", " ").replace("\n", " ").strip() + if len(text) > MAX_VALUE_LENGTH: + text = f"{text[:MAX_VALUE_LENGTH]}..." + return text + + +def scan_summary(payload): + """Keep only the documented summary keys, whatever wrapper they arrived in.""" + if not isinstance(payload, dict): + return {} + if isinstance(payload.get("scan"), dict): + payload = payload["scan"] + return {key: payload[key] for key in SCAN_SUMMARY_KEYS if payload.get(key) is not None} + + +def scan_summaries(payload): + """Pull the list of scan summaries out of a hash-lookup response. + + The lookup answers with scans of one digest. Accepting either a bare list or + a list under a container key means a wrapper change does not turn a real + answer into "not found", which is the failure mode that misleads an analyst. + """ + if isinstance(payload, list): + candidates = payload + elif isinstance(payload, dict): + candidates = None + for key in ("scans", "results", "data"): + value = payload.get(key) + # Every list on this API arrives wrapped with its own accounting, + # `{"items": [...], "returned": n, "total": n, "truncated": bool}`, + # so that a truncated answer can never be mistaken for a complete + # one. Unwrap that before looking for a bare list: reading only the + # bare form turned a real answer into "not found", which is exactly + # the failure mode this function exists to avoid. + if isinstance(value, dict) and isinstance(value.get("items"), list): + candidates = value["items"] + break + if isinstance(value, list): + candidates = value + break + if candidates is None: + single = scan_summary(payload) + candidates = [single] if single else [] + else: + candidates = [] + return [summary for summary in (scan_summary(item) for item in candidates) if summary] + + +def is_finished(summary): + """Whether a scan has reached a terminal state. + + Keyed off the two documented facts - a verdict is absent while a scan is + unfinished, and finished_at is set when it is done - rather than off a list + of status strings, which is not part of the published contract. + """ + return bool(summary.get("verdict") or summary.get("finished_at")) + + +class MalwagonClient: + """Minimal Malwagon REST client built on requests.""" + + def __init__(self, api_key, api_url=None, timeout=DEFAULT_TIMEOUT): + if not api_key: + raise MalwagonError("A Malwagon API key is required.") + self.api_key = api_key + self.api_url = self._validate_url(api_url or DEFAULT_API_URL) + self.timeout = timeout + + @staticmethod + def _validate_url(api_url): + url = str(api_url).strip().rstrip("/") + parsed = urlparse(url) + if parsed.scheme not in ("http", "https") or not parsed.netloc: + raise MalwagonError("The configured Malwagon API URL is not a valid http(s) URL.") + return url + + def _headers(self): + # Header only. The key never goes in a URL, a query string or a log line. + return { + "Authorization": f"Bearer {self.api_key}", + "Accept": "application/json", + "User-Agent": USER_AGENT, + } + + def _request(self, method, path, **kwargs): + try: + response = requests.request( + method, + f"{self.api_url}{path}", + headers=self._headers(), + timeout=self.timeout, + # A redirect would re-send the Authorization header to whatever + # host the response names. Refuse instead of following. + allow_redirects=False, + **kwargs, + ) + except requests.exceptions.RequestException: + # The exception text can contain the full request URL; the message + # here is written by us so nothing from the request can leak. + raise MalwagonError("Could not reach the Malwagon API.") + return self._handle(response) + + @staticmethod + def _handle(response): + status = response.status_code + if status in (301, 302, 303, 307, 308): + raise MalwagonError("The Malwagon API answered with a redirect, which is not followed.") + if status == 401: + raise MalwagonError("Malwagon rejected the API key.") + if status == 403: + raise MalwagonError("The Malwagon API key does not carry the scope this request needs.") + if status == 404: + return None + if status == 429: + retry_after = response.headers.get("Retry-After") + # Report and stop. Retrying in a loop is what turns one throttled + # user into a throttled tenant. + if retry_after: + raise MalwagonError(f"Malwagon rate limit reached, retry after {clean_text(retry_after)} seconds.") + raise MalwagonError("Malwagon rate limit reached.") + if status >= 400: + raise MalwagonError(f"The Malwagon API answered with HTTP {status}.") + try: + return response.json() + except ValueError: + raise MalwagonError("The Malwagon API answered with a body that is not JSON.") + + def lookup_hash(self, sha256): + if not is_sha256(sha256): + raise MalwagonError("A Malwagon hash lookup needs a SHA256 digest.") + # Validated against the regex above, so it cannot walk out of the path. + return scan_summaries(self._request("GET", f"/hashes/{sha256.lower()}")) + + def scan_status(self, scan_id): + return scan_summary(self._request("GET", f"/scans/{quote(str(scan_id), safe='')}")) + + def submit_file(self, filename, content, options=None): + data = {key: value for key, value in (options or {}).items() if value is not None} + payload = self._request( + "POST", + "/scans/file", + files={"file": (filename or "sample", content)}, + data=data, + ) + return scan_summary(payload) + + def submit_target(self, module, target, options=None): + body = {"module": module, "target": target} + body.update({key: value for key, value in (options or {}).items() if value is not None}) + return scan_summary(self._request("POST", "/scans", json=body)) + + def report_url(self, scan_id): + """Where a human reads the report for this scan. + + The API endpoint answers 401 without a bearer token, so pointing an + analyst at it hands them a login wall instead of a report. The public + permalink resolves anonymously for a scan that was left public, which + is what a permalink in a MISP object is for. It is only ever attached + to a scan we know is public; a private one has no page to link to. + """ + base = self.api_url.split("/api/", 1)[0] + return f"{base}/s/{quote(str(scan_id), safe='')}" + + +class MalwagonResults: + """Builds the misp_standard result payload out of scan summaries.""" + + def __init__(self, client, sha256=None, attribute_uuid=None): + self.client = client + self.sha256 = sha256.lower() if sha256 and is_sha256(sha256) else None + self.attribute_uuid = attribute_uuid + self.misp_event = MISPEvent() + self.tagged = False + + def add_scan(self, summary): + if not summary: + return + file_object, anchor = self._add_file_object(summary) + report = self._add_report_object(summary) + if report is None: + return + # One verdict tag per enrichment, on the indicator when there is one and + # on the verdict line otherwise, so the tag is never orphaned. + if not self.tagged: + target = anchor if anchor is not None else self._verdict_attribute(report) + if target is not None and self._tag(target, summary.get("verdict")): + self.tagged = True + if file_object is not None: + report.add_reference(file_object.uuid, "analysed-with") + + def _add_file_object(self, summary): + """A file object for the sample, when the scan describes one.""" + if not self.sha256: + return None, None + misp_object = MISPObject("file") + anchor = misp_object.add_attribute("sha256", type="sha256", value=self.sha256) + if summary.get("mime"): + misp_object.add_attribute("mimetype", type="mime-type", value=clean_text(summary["mime"])) + size = summary.get("size") + if isinstance(size, int) and size >= 0: + misp_object.add_attribute("size-in-bytes", type="size-in-bytes", value=size) + if self.attribute_uuid: + misp_object.add_reference(self.attribute_uuid, "related-to") + self.misp_event.add_object(misp_object) + return misp_object, anchor + + def _add_report_object(self, summary): + scan_id = summary.get("scan_id") + if not scan_id: + return None + misp_object = MISPObject("sandbox-report") + # The report is only retrievable with a key, so this is a saas sandbox + # in the object template's vocabulary, not a web one. + misp_object.add_attribute("sandbox-type", type="text", value="saas") + misp_object.add_attribute("saas-sandbox", type="text", value="malwagon") + misp_object.add_attribute("permalink", type="link", value=self.client.report_url(scan_id)) + score = summary.get("score") + if isinstance(score, (int, float)): + misp_object.add_attribute("score", type="text", value=str(score)) + for line in self._result_lines(summary): + misp_object.add_attribute("results", type="text", value=line, disable_correlation=True) + if self.attribute_uuid: + misp_object.add_reference(self.attribute_uuid, "related-to") + self.misp_event.add_object(misp_object) + return misp_object + + @staticmethod + def _verdict_attribute(report): + for attribute in report.attributes: + if attribute.object_relation == "results" and attribute.value.startswith("verdict:"): + return attribute + return None + + @staticmethod + def _result_lines(summary): + lines = [] + if summary.get("verdict"): + # First, so _verdict_attribute finds it and so an analyst reading the + # object sees the conclusion before the metadata. + lines.append(f"verdict: {clean_text(summary['verdict'])}") + for key in ("status", "module", "submitted_at", "finished_at"): + if summary.get(key): + lines.append(f"{key}: {clean_text(summary[key])}") + tags = summary.get("tags") + if isinstance(tags, list) and tags: + # Reported as text on purpose. Turning a remote string into a MISP + # tag would let the service write into the instance's taxonomies. + joined = ", ".join(clean_text(tag) for tag in tags[:MAX_TAGS]) + lines.append(f"tags: {joined}") + return lines + + @staticmethod + def _tag(attribute, verdict): + tags = VERDICT_TAGS.get(str(verdict or "").lower()) + if not tags: + return False + for tag in tags: + attribute.add_tag(tag) + return True + + def get_results(self, empty_message): + event = self.misp_event.to_dict() + results = {key: event[key] for key in ("Attribute", "Object") if event.get(key)} + if not results: + return {"error": empty_message} + return {"results": results} From 2e3709dcc46e7313a67292aaa8a3132224d4c4de Mon Sep 17 00:00:00 2001 From: Tolga SEZER <19777824+projectboot@users.noreply.github.com> Date: Mon, 7 Sep 2026 19:49:13 +0300 Subject: [PATCH 2/4] add: [malwagon] malwagon.py --- misp_modules/modules/expansion/malwagon.py | 105 +++++++++++++++++++++ 1 file changed, 105 insertions(+) create mode 100644 misp_modules/modules/expansion/malwagon.py diff --git a/misp_modules/modules/expansion/malwagon.py b/misp_modules/modules/expansion/malwagon.py new file mode 100644 index 00000000..83a44575 --- /dev/null +++ b/misp_modules/modules/expansion/malwagon.py @@ -0,0 +1,105 @@ +import json + +from . import check_input_attribute, checking_error, standard_error_message +from ._malwagon_api import DEFAULT_TIMEOUT, MalwagonClient, MalwagonError, MalwagonResults, is_sha256 + +mispattributes = { + "input": ["sha256", "filename|sha256"], + "format": "misp_standard", +} +moduleinfo = { + "version": "1", + "author": "Tolga Sezer", + "description": "Look up a SHA256 digest in the Malwagon sandbox and return the scans it already has for it.", + "module-type": ["expansion", "hover"], + "name": "Malwagon Lookup", + "logo": "", + "requirements": ["A Malwagon API key with the read scope."], + "features": ( + "The module takes a sha256 or filename|sha256 attribute and asks Malwagon which analyses it already holds" + " for that digest. The lookup is free and answers immediately, so it is safe to use on hover: no sample" + " leaves the MISP instance and no detonation quota is spent. Only the digest is sent.\n\nEach analysis comes" + " back as a sandbox-report object carrying the verdict, the score and a permalink to the report, referenced" + " from a file object for the sample. The verdict is translated into MISP's own vocabulary - a" + ' misp:threat-level tag, plus ioc:artifact-state where the verdict is conclusive - rather than being passed' + " through as a vendor string.\n\nTo detonate something Malwagon has never seen, use the Malwagon Submit" + " module instead." + ), + "references": [ + "https://malwagon.com", + "https://malwagon.com/docs/api", + "https://www.misp-project.org/taxonomies.html", + ], + "input": "A sha256 or filename|sha256 attribute.", + "output": ( + "A file object for the sample and one sandbox-report object per Malwagon analysis, with the verdict" + " expressed as MISP taxonomy tags." + ), +} +# api_url exists so an operator can point the module at another deployment; it is +# not a way to send the key somewhere else silently, since the value is visible +# in the MISP server settings. +moduleconfig = ["apikey", "api_url", "max_results"] + +DEFAULT_MAX_RESULTS = 10 + + +def _digest(attribute): + value = str(attribute.get("value", "")).strip() + if attribute.get("type") == "filename|sha256" and "|" in value: + value = value.split("|", 1)[1] + return value + + +def _positive_int(value, default): + try: + parsed = int(value) + except (TypeError, ValueError): + return default + return parsed if parsed > 0 else default + + +def handler(q=False): + if q is False: + return False + request = json.loads(q) + + if not request.get("attribute") or not check_input_attribute(request["attribute"], requirements=("type", "value")): + return {"error": f"{standard_error_message}, {checking_error} that is the digest to look up in Malwagon."} + + attribute = request["attribute"] + if attribute["type"] not in mispattributes["input"]: + return {"error": "Unsupported attribute type."} + + digest = _digest(attribute) + if not is_sha256(digest): + return {"error": "Malwagon indexes analyses by SHA256; this attribute does not carry one."} + + config = request.get("config") or {} + if not config.get("apikey"): + return {"error": "A Malwagon API key is required."} + + max_results = _positive_int(config.get("max_results"), DEFAULT_MAX_RESULTS) + + try: + client = MalwagonClient(config["apikey"], config.get("api_url"), timeout=DEFAULT_TIMEOUT) + summaries = client.lookup_hash(digest) + except MalwagonError as error: + return {"error": str(error)} + + if not summaries: + return {"error": "Malwagon has no analysis for this digest."} + + results = MalwagonResults(client, sha256=digest, attribute_uuid=attribute.get("uuid")) + for summary in summaries[:max_results]: + results.add_scan(summary) + return results.get_results("Malwagon returned no usable analysis for this digest.") + + +def introspection(): + return mispattributes + + +def version(): + moduleinfo["config"] = moduleconfig + return moduleinfo From e09183524766857f876cba2a8ede4b03748bb018 Mon Sep 17 00:00:00 2001 From: Tolga SEZER <19777824+projectboot@users.noreply.github.com> Date: Mon, 7 Sep 2026 19:49:14 +0300 Subject: [PATCH 3/4] add: [malwagon] malwagon_submit.py --- .../modules/expansion/malwagon_submit.py | 247 ++++++++++++++++++ 1 file changed, 247 insertions(+) create mode 100644 misp_modules/modules/expansion/malwagon_submit.py diff --git a/misp_modules/modules/expansion/malwagon_submit.py b/misp_modules/modules/expansion/malwagon_submit.py new file mode 100644 index 00000000..02161bee --- /dev/null +++ b/misp_modules/modules/expansion/malwagon_submit.py @@ -0,0 +1,247 @@ +import base64 +import binascii +import hashlib +import io +import json +import re +import time +import zipfile + +from ._malwagon_api import DEFAULT_TIMEOUT, MalwagonClient, MalwagonError, MalwagonResults, is_finished + +mispattributes = { + "input": ["attachment", "malware-sample", "url"], + "format": "misp_standard", +} +moduleinfo = { + "version": "1", + "author": "Tolga Sezer", + "description": "Detonate a sample or a URL in the Malwagon sandbox and return the analysis.", + "module-type": ["expansion"], + "name": "Malwagon Submit", + "logo": "", + "requirements": ["A Malwagon API key with the submit scope."], + "features": ( + "The module takes an attachment, malware-sample or url attribute and detonates it in the Malwagon sandbox." + " A malware-sample arrives zip-encrypted with the password infected and is unpacked before" + " submission.\n\nA file is looked up by its SHA256 first. If Malwagon has already analysed that digest the" + " existing analysis is returned, which is free, instant, and spends no detonation quota; set always_submit" + " to force a fresh detonation instead. The module then polls the scan until it finishes or until" + " poll_timeout seconds have passed, and returns what it has either way, so a long detonation gives back a" + " permalink rather than an error.\n\nThis module submits the sample itself to a third party, so it is" + " deliberately expansion-only and never runs on hover. Submission is blocked when the attribute carries a" + " TLP tag more restrictive than max_tlp, and scans are created private unless private is set to" + " false.\n\nThe sandbox image is chosen by the platform from the sample; on the free and community tiers the" + " analysis virtual machine has no internet access at all, which is structural rather than a quota, so" + " network-dependent samples will look inert there." + ), + "references": [ + "https://malwagon.com", + "https://malwagon.com/docs/api", + "https://www.misp-project.org/taxonomies.html", + ], + "input": "An attachment, malware-sample or url attribute.", + "output": ( + "A file object for the sample and a sandbox-report object for the Malwagon analysis, with the verdict" + " expressed as MISP taxonomy tags." + ), +} +moduleconfig = ["apikey", "api_url", "max_tlp", "private", "internet", "poll_timeout", "always_submit"] + +# The zip password MISP uses for malware-sample attributes. +MALWARE_SAMPLE_PASSWORD = b"infected" + +DEFAULT_POLL_TIMEOUT = 60 +POLL_INTERVAL = 5 +DEFAULT_MAX_TLP = "tlp:amber" + +# Ordered so that "more restrictive than" is a comparison. tlp:clear and the +# older tlp:white are the same level, as are tlp:amber and tlp:amber+strict. +TLP_LEVELS = { + "tlp:clear": 0, + "tlp:white": 0, + "tlp:green": 1, + "tlp:amber": 2, + "tlp:amber+strict": 2, + "tlp:red": 3, +} + +# A filename goes into a multipart header. Keep it to something that cannot +# carry a separator, a newline or a path. +FILENAME_RE = re.compile(r"[^A-Za-z0-9._-]") +MAX_FILENAME_LENGTH = 128 + + +def _safe_filename(filename): + name = FILENAME_RE.sub("_", str(filename or "").strip())[:MAX_FILENAME_LENGTH].lstrip(".") + return name or "sample" + + +def _positive_int(value, default): + try: + parsed = int(value) + except (TypeError, ValueError): + return default + return parsed if parsed > 0 else default + + +def _as_bool(value, default=None): + if value is None or value == "": + return default + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return bool(value) + normalised = str(value).strip().lower() + if normalised in ("true", "1", "yes", "on"): + return True + if normalised in ("false", "0", "no", "off"): + return False + return default + + +def _tlp_blocked(request, config): + """Whether the classification of the input forbids sending it to a third party. + + Only the tags MISP actually passes along with the request can be inspected; + when it passes none, this cannot fire. It is a guard against an analyst + expanding a red-marked attribute by reflex, not a substitute for deciding + whether the module should be enabled at all. + """ + max_tlp = str(config.get("max_tlp") or DEFAULT_MAX_TLP).strip().lower() + ceiling = TLP_LEVELS.get(max_tlp, TLP_LEVELS[DEFAULT_MAX_TLP]) + attribute = request.get("attribute") + tags = attribute.get("Tag") or [] if isinstance(attribute, dict) else [] + for tag in tags: + name = tag.get("name") if isinstance(tag, dict) else tag + level = TLP_LEVELS.get(str(name or "").strip().lower()) + if level is not None and level > ceiling: + return str(name).strip().lower() + return None + + +def _decode_sample(data, is_malware_sample): + """Return the raw bytes of the submitted file. + + Errors raised here are fixed strings on purpose: this function is the only + place that holds decoded sample bytes, and a message that echoed them would + put sample content into the MISP interface and into the module log. + """ + if isinstance(data, bytes): + content = data + else: + try: + content = base64.b64decode(str(data), validate=True) + except (binascii.Error, ValueError): + raise MalwagonError("The attribute data is not valid base64.") + if not is_malware_sample: + return content + try: + with zipfile.ZipFile(io.BytesIO(content)) as archive: + names = archive.namelist() + if not names: + raise MalwagonError("The malware-sample archive is empty.") + return archive.read(names[0], pwd=MALWARE_SAMPLE_PASSWORD) + except MalwagonError: + raise + except Exception: + raise MalwagonError("The malware-sample attribute could not be unpacked.") + + +def _file_options(config): + return { + # Private by default: a sample pulled out of a MISP event is somebody + # else's material, and the opt-out is explicit. + "private": "true" if _as_bool(config.get("private"), True) else "false", + "internet": _internet_option(config), + } + + +def _internet_option(config): + internet = _as_bool(config.get("internet")) + if internet is None: + return None + return "true" if internet else "false" + + +def _poll(client, summary, poll_timeout): + """Poll the scan until it is finished or the budget runs out.""" + scan_id = summary.get("scan_id") + deadline = time.monotonic() + poll_timeout + while not is_finished(summary) and time.monotonic() < deadline: + time.sleep(POLL_INTERVAL) + try: + updated = client.scan_status(scan_id) + except MalwagonError: + # A throttled or unreachable status call is not a reason to throw + # away the submission; return what the submit already gave us. + break + if updated: + summary = updated + return summary + + +def _submitted_file(request): + if "attachment" in request: + return _safe_filename(request.get("attachment")), _decode_sample(request.get("data"), False) + filename = str(request.get("malware-sample") or "").split("|")[0] + return _safe_filename(filename), _decode_sample(request.get("data"), True) + + +def handler(q=False): + if q is False: + return False + request = json.loads(q) + + config = request.get("config") or {} + if not config.get("apikey"): + return {"error": "A Malwagon API key is required."} + + if not any(key in request for key in mispattributes["input"]): + return {"error": "No valid attribute type for this module has been provided."} + + blocked_by = _tlp_blocked(request, config) + if blocked_by: + return {"error": f"This attribute is marked {blocked_by}, which is above the configured max_tlp."} + + poll_timeout = _positive_int(config.get("poll_timeout"), DEFAULT_POLL_TIMEOUT) + always_submit = _as_bool(config.get("always_submit"), False) + + try: + client = MalwagonClient(config["apikey"], config.get("api_url"), timeout=DEFAULT_TIMEOUT) + if "url" in request: + sha256 = None + summary = client.submit_target( + "url", + str(request["url"]), + {"private": _as_bool(config.get("private"), True)}, + ) + else: + filename, content = _submitted_file(request) + sha256 = hashlib.sha256(content).hexdigest() + summary = None + if not always_submit: + # Free and instant where a detonation costs minutes and a quota. + known = client.lookup_hash(sha256) + if known: + summary = known[0] + if summary is None: + summary = client.submit_file(filename, content, _file_options(config)) + if not summary.get("scan_id"): + return {"error": "Malwagon accepted the submission but returned no scan identifier."} + summary = _poll(client, summary, poll_timeout) + except MalwagonError as error: + return {"error": str(error)} + + results = MalwagonResults(client, sha256=sha256, attribute_uuid=request.get("attribute_uuid")) + results.add_scan(summary) + return results.get_results("Malwagon returned no usable analysis for this submission.") + + +def introspection(): + return mispattributes + + +def version(): + moduleinfo["config"] = moduleconfig + return moduleinfo From d743288a31b3ed8c7f5fbe41431ec7023999d6dc Mon Sep 17 00:00:00 2001 From: Tolga SEZER <19777824+projectboot@users.noreply.github.com> Date: Mon, 7 Sep 2026 19:49:16 +0300 Subject: [PATCH 4/4] add: [malwagon] test_malwagon.py --- tests/test_malwagon.py | 517 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 517 insertions(+) create mode 100644 tests/test_malwagon.py diff --git a/tests/test_malwagon.py b/tests/test_malwagon.py new file mode 100644 index 00000000..b0da7bb3 --- /dev/null +++ b/tests/test_malwagon.py @@ -0,0 +1,517 @@ +"""Tests for the Malwagon expansion modules. + +Every HTTP call is mocked, so the suite needs no API key and no network. The +placeholder key below is deliberately not shaped like a real Malwagon token. +""" + +import base64 +import json +import pathlib +from unittest.mock import patch + +import pytest + +from misp_modules.modules.expansion import _malwagon_api, malwagon, malwagon_submit + +API_KEY = "" +UUID = "5b582d80-7a7e-4b6a-9f22-77656e72bb3b" +SHA256 = "275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f" + +# The zip fixture already in this directory: EICAR.com, encrypted with the +# password MISP uses for malware-sample attributes. +INFECTED_ZIP = pathlib.Path(__file__).resolve().parent.joinpath("infected.zip").read_bytes() +EICAR = b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-" + +FINISHED_SCAN = { + "scan_id": "scan-1", + "status": "finished", + "verdict": "malicious", + "score": 92, + "module": "file", + "mime": "application/x-dosexec", + "size": 4096, + "tags": ["trojan", "loader"], + "submitted_at": "2026-09-01T10:00:00Z", + "finished_at": "2026-09-01T10:04:00Z", +} +RUNNING_SCAN = {"scan_id": "scan-1", "status": "running", "module": "file"} + + +class MockResponse: + def __init__(self, payload=None, status_code=200, headers=None): + self.payload = payload + self.status_code = status_code + self.headers = headers or {} + + def json(self): + if self.payload is None: + raise ValueError("no json") + return self.payload + + +class Recorder: + """A requests.request replacement that records calls and replays answers.""" + + def __init__(self, answers): + self.answers = answers + self.calls = [] + + def __call__(self, method, url, **kwargs): + self.calls.append({"method": method, "url": url, **kwargs}) + for fragment, answer in self.answers: + if fragment in url: + if callable(answer): + return answer(len([c for c in self.calls if fragment in c["url"]])) + return answer + return MockResponse({}, status_code=404) + + def urls(self): + return [call["url"] for call in self.calls] + + +def lookup_query(value=SHA256, type_="sha256", config=None): + attribute = {"type": type_, "value": value, "uuid": UUID} + return json.dumps( + { + "module": "malwagon", + "attribute": attribute, + "config": config if config is not None else {"apikey": API_KEY}, + } + ) + + +def submit_query(config=None, **request): + payload = {"module": "malwagon_submit", "config": config if config is not None else {"apikey": API_KEY}} + payload.update(request) + return json.dumps(payload) + + +def objects_by_name(results): + return {obj["name"]: obj for obj in results["results"]["Object"]} + + +def all_tags(results): + tags = [] + for obj in results["results"]["Object"]: + for attribute in obj["Attribute"]: + tags.extend(tag["name"] for tag in attribute.get("Tag", [])) + return tags + + +class FakeClock: + """A clock the polling loop can be driven against without real waiting.""" + + def __init__(self): + self.now = 0.0 + + def monotonic(self): + return self.now + + def sleep(self, seconds): + self.now += seconds + + +@pytest.fixture(autouse=True) +def _fake_clock(): + """The submit module polls; nothing here should actually wait.""" + clock = FakeClock() + with patch.object(malwagon_submit.time, "sleep", clock.sleep): + with patch.object(malwagon_submit.time, "monotonic", clock.monotonic): + yield clock + + +# -------------------------------------------------------------------------- +# Lookup module +# -------------------------------------------------------------------------- + + +def test_lookup_returns_file_and_sandbox_report_objects(): + recorder = Recorder([("/hashes/", MockResponse({"scans": {"items": [FINISHED_SCAN], "returned": 0, "total": 0, "truncated": False}}))]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon.handler(lookup_query()) + + objects = objects_by_name(results) + assert set(objects) == {"file", "sandbox-report"} + file_values = {a["object_relation"]: a["value"] for a in objects["file"]["Attribute"]} + assert file_values["sha256"] == SHA256 + assert file_values["mimetype"] == "application/x-dosexec" + assert file_values["size-in-bytes"] == 4096 + + +def test_lookup_maps_the_verdict_onto_misp_taxonomies(): + """ + The point of the mapping is that a MISP filter on threat level sees the + result. A raw vendor string in a text attribute would be invisible to it. + """ + recorder = Recorder([("/hashes/", MockResponse([FINISHED_SCAN]))]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon.handler(lookup_query()) + + assert 'misp:threat-level="high-risk"' in all_tags(results) + assert 'ioc:artifact-state="malicious"' in all_tags(results) + + +def test_lookup_maps_a_clean_verdict_to_no_risk(): + clean = dict(FINISHED_SCAN, verdict="clean", score=0) + recorder = Recorder([("/hashes/", MockResponse([clean]))]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon.handler(lookup_query()) + + assert 'misp:threat-level="no-risk"' in all_tags(results) + assert 'ioc:artifact-state="not-malicious"' in all_tags(results) + + +def test_remote_tags_are_reported_as_text_not_as_misp_tags(): + """ + A tag string chosen by the remote service must not become a MISP tag: that + would let the service write into the instance's taxonomies. + """ + hostile = dict(FINISHED_SCAN, tags=['tlp:red', 'ioc:artifact-state="not-malicious"']) + recorder = Recorder([("/hashes/", MockResponse([hostile]))]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon.handler(lookup_query()) + + assert "tlp:red" not in all_tags(results) + report = objects_by_name(results)["sandbox-report"] + assert any("tlp:red" in a["value"] for a in report["Attribute"] if a["object_relation"] == "results") + + +def test_lookup_accepts_a_composite_filename_sha256(): + recorder = Recorder([("/hashes/", MockResponse([FINISHED_SCAN]))]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon.handler(lookup_query(f"sample.exe|{SHA256}", "filename|sha256")) + + assert SHA256 in recorder.urls()[0] + assert "results" in results + + +def test_lookup_refuses_a_value_that_is_not_a_sha256(): + """The digest is interpolated into the request path, so it is validated first.""" + with patch.object(_malwagon_api.requests, "request", Recorder([])) as recorder: + results = malwagon.handler(lookup_query("../scans/scan-1", "sha256")) + + assert "error" in results + assert recorder.calls == [] + + +def test_lookup_reports_an_unknown_digest_without_inventing_a_result(): + recorder = Recorder([("/hashes/", MockResponse(None, status_code=404))]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon.handler(lookup_query()) + + assert results == {"error": "Malwagon has no analysis for this digest."} + + +def test_lookup_requires_an_api_key(): + assert "error" in malwagon.handler(lookup_query(config={})) + + +def test_lookup_caps_the_number_of_reports(): + scans = [dict(FINISHED_SCAN, scan_id=f"scan-{i}") for i in range(20)] + recorder = Recorder([("/hashes/", MockResponse({"scans": scans}))]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon.handler(lookup_query(config={"apikey": API_KEY, "max_results": 3})) + + reports = [o for o in results["results"]["Object"] if o["name"] == "sandbox-report"] + assert len(reports) == 3 + + +# -------------------------------------------------------------------------- +# Transport hardening +# -------------------------------------------------------------------------- + + +def test_the_api_key_travels_in_a_header_and_never_in_the_url(): + recorder = Recorder([("/hashes/", MockResponse([FINISHED_SCAN]))]) + with patch.object(_malwagon_api.requests, "request", recorder): + malwagon.handler(lookup_query()) + + call = recorder.calls[0] + assert call["headers"]["Authorization"] == f"Bearer {API_KEY}" + assert API_KEY not in call["url"] + + +def test_every_request_sets_a_timeout_and_refuses_redirects(): + """ + A redirect would re-send the bearer token to a host named by the response, + and requests has no default timeout, so both are set explicitly. + """ + recorder = Recorder([("/hashes/", MockResponse([FINISHED_SCAN]))]) + with patch.object(_malwagon_api.requests, "request", recorder): + malwagon.handler(lookup_query()) + + call = recorder.calls[0] + assert call["allow_redirects"] is False + assert call["timeout"] == _malwagon_api.DEFAULT_TIMEOUT + + +def test_a_redirect_is_reported_rather_than_followed(): + redirect = MockResponse(None, status_code=302, headers={"Location": "https://example.invalid/"}) + recorder = Recorder([("/hashes/", redirect)]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon.handler(lookup_query()) + + assert "redirect" in results["error"] + + +def test_a_rate_limit_surfaces_retry_after_and_does_not_retry(): + throttled = MockResponse(None, status_code=429, headers={"Retry-After": "42"}) + recorder = Recorder([("/hashes/", throttled)]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon.handler(lookup_query()) + + assert "42" in results["error"] + assert len(recorder.calls) == 1 + + +def test_a_missing_scope_is_distinguished_from_a_bad_key(): + recorder = Recorder([("/hashes/", MockResponse(None, status_code=403))]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon.handler(lookup_query()) + + assert "scope" in results["error"] + + +def test_a_non_http_api_url_is_rejected(): + results = malwagon.handler(lookup_query(config={"apikey": API_KEY, "api_url": "file:///etc/passwd"})) + assert "error" in results + + +# -------------------------------------------------------------------------- +# Submit module +# -------------------------------------------------------------------------- + + +def test_a_malware_sample_is_unzipped_with_the_infected_password_before_submission(): + recorder = Recorder( + [ + ("/hashes/", MockResponse(None, status_code=404)), + ("/scans/file", MockResponse({"scan": FINISHED_SCAN}, status_code=202)), + ] + ) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon_submit.handler( + submit_query( + **{ + "malware-sample": "EICAR.com|d41d8cd98f00b204e9800998ecf8427e", + "data": base64.b64encode(INFECTED_ZIP).decode(), + } + ) + ) + + posted = [call for call in recorder.calls if call["method"] == "POST"][0] + assert posted["files"]["file"][1] == EICAR + assert "results" in results + + +def test_an_attachment_is_only_base64_decoded(): + recorder = Recorder( + [ + ("/hashes/", MockResponse(None, status_code=404)), + ("/scans/file", MockResponse({"scan": FINISHED_SCAN}, status_code=202)), + ] + ) + with patch.object(_malwagon_api.requests, "request", recorder): + malwagon_submit.handler( + submit_query(attachment="report.doc", data=base64.b64encode(b"plain bytes").decode()) + ) + + posted = [call for call in recorder.calls if call["method"] == "POST"][0] + assert posted["files"]["file"][1] == b"plain bytes" + + +def test_a_known_digest_is_answered_from_the_lookup_without_spending_submit_quota(): + """ + A lookup is free and instant where a detonation costs minutes and one of a + small number of submits, so the module must not detonate what is already known. + """ + recorder = Recorder([("/hashes/", MockResponse([FINISHED_SCAN]))]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon_submit.handler( + submit_query(attachment="sample.bin", data=base64.b64encode(b"known").decode()) + ) + + assert [call["method"] for call in recorder.calls] == ["GET"] + assert "results" in results + + +def test_always_submit_forces_a_fresh_detonation(): + recorder = Recorder([("/scans/file", MockResponse({"scan": FINISHED_SCAN}, status_code=202))]) + with patch.object(_malwagon_api.requests, "request", recorder): + malwagon_submit.handler( + submit_query( + attachment="sample.bin", + data=base64.b64encode(b"known").decode(), + config={"apikey": API_KEY, "always_submit": "true"}, + ) + ) + + assert [call["method"] for call in recorder.calls] == ["POST"] + + +def test_a_file_submission_is_private_by_default(): + recorder = Recorder( + [ + ("/hashes/", MockResponse(None, status_code=404)), + ("/scans/file", MockResponse({"scan": FINISHED_SCAN}, status_code=202)), + ] + ) + with patch.object(_malwagon_api.requests, "request", recorder): + malwagon_submit.handler(submit_query(attachment="s.bin", data=base64.b64encode(b"x").decode())) + + posted = [call for call in recorder.calls if call["method"] == "POST"][0] + assert posted["data"]["private"] == "true" + # No sandbox image is named: the platform picks one from the sample. + assert "os" not in posted["data"] + + +def test_a_url_is_submitted_as_a_url_module_scan(): + recorder = Recorder([("/scans", MockResponse({"scan": dict(FINISHED_SCAN, module="url")}, status_code=202))]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon_submit.handler(submit_query(url="http://example.invalid/payload")) + + posted = recorder.calls[0] + assert posted["json"] == {"module": "url", "target": "http://example.invalid/payload", "private": True} + assert objects_by_name(results).keys() == {"sandbox-report"} + + +def test_the_scan_is_polled_until_it_finishes(): + def status(call_number): + return MockResponse({"scan": RUNNING_SCAN if call_number < 3 else FINISHED_SCAN}) + + recorder = Recorder( + [ + ("/hashes/", MockResponse(None, status_code=404)), + ("/scans/file", MockResponse({"scan": RUNNING_SCAN}, status_code=202)), + ("/scans/scan-1", status), + ] + ) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon_submit.handler(submit_query(attachment="s.bin", data=base64.b64encode(b"x").decode())) + + assert len([c for c in recorder.calls if "/scans/scan-1" in c["url"]]) == 3 + assert 'misp:threat-level="high-risk"' in all_tags(results) + + +def test_a_scan_still_running_at_the_deadline_returns_the_permalink_instead_of_an_error(): + recorder = Recorder( + [ + ("/hashes/", MockResponse(None, status_code=404)), + ("/scans/file", MockResponse({"scan": RUNNING_SCAN}, status_code=202)), + ("/scans/scan-1", MockResponse({"scan": RUNNING_SCAN})), + ] + ) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon_submit.handler( + submit_query( + attachment="s.bin", + data=base64.b64encode(b"x").decode(), + config={"apikey": API_KEY, "poll_timeout": 1}, + ) + ) + + report = objects_by_name(results)["sandbox-report"] + permalinks = [a["value"] for a in report["Attribute"] if a["object_relation"] == "permalink"] + assert permalinks == ["https://malwagon.com/s/scan-1"] + assert all_tags(results) == [] + + +def test_a_restrictive_tlp_tag_blocks_the_submission(): + recorder = Recorder([]) + query = json.dumps( + { + "module": "malwagon_submit", + "config": {"apikey": API_KEY}, + "attribute": {"type": "url", "value": "http://example.invalid/", "uuid": UUID, "Tag": [{"name": "tlp:red"}]}, + "url": "http://example.invalid/", + } + ) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon_submit.handler(query) + + assert "tlp:red" in results["error"] + assert recorder.calls == [] + + +def test_a_tlp_tag_within_the_ceiling_is_allowed(): + recorder = Recorder([("/scans", MockResponse({"scan": FINISHED_SCAN}, status_code=202))]) + query = json.dumps( + { + "module": "malwagon_submit", + "config": {"apikey": API_KEY}, + "attribute": { + "type": "url", + "value": "http://example.invalid/", + "uuid": UUID, + "Tag": [{"name": "tlp:green"}], + }, + "url": "http://example.invalid/", + } + ) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon_submit.handler(query) + + assert "results" in results + + +def test_an_undecodable_sample_never_echoes_sample_bytes(): + """ + The error goes into the MISP interface and the module log, so it must be a + fixed string and not anything derived from the sample. + """ + payload = base64.b64encode(b"not a zip, but recognisable content").decode() + recorder = Recorder([]) + with patch.object(_malwagon_api.requests, "request", recorder): + results = malwagon_submit.handler( + submit_query(**{"malware-sample": "x.zip|deadbeef", "data": payload}) + ) + + assert results == {"error": "The malware-sample attribute could not be unpacked."} + assert recorder.calls == [] + + +def test_a_filename_cannot_carry_a_path_or_a_separator_into_the_multipart_body(): + recorder = Recorder( + [ + ("/hashes/", MockResponse(None, status_code=404)), + ("/scans/file", MockResponse({"scan": FINISHED_SCAN}, status_code=202)), + ] + ) + with patch.object(_malwagon_api.requests, "request", recorder): + malwagon_submit.handler( + submit_query(attachment='../../etc/pa"sswd\r\nX: y', data=base64.b64encode(b"x").decode()) + ) + + posted = [call for call in recorder.calls if call["method"] == "POST"][0] + name = posted["files"]["file"][0] + assert not set(name) & set('/\\"\r\n') + + +def test_submit_rejects_an_unsupported_request(): + assert "error" in malwagon_submit.handler(submit_query(domain="example.invalid")) + + +def test_submit_requires_an_api_key(): + assert "error" in malwagon_submit.handler(submit_query(url="http://example.invalid/", config={})) + + +# -------------------------------------------------------------------------- +# Module contract +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize("module", (malwagon, malwagon_submit)) +def test_the_module_contract_is_complete(module): + """generate.py exits if any of these is missing, which breaks the docs build.""" + info = module.version() + for field in ("name", "description", "module-type", "author", "version", "logo"): + assert field in info + assert info["config"] == module.moduleconfig + assert module.introspection() == module.mispattributes + assert module.handler(False) is False + + +def test_hover_is_only_offered_by_the_free_lookup(): + """Hover fires on viewing an attribute; a detonation must never be triggered that way.""" + assert "hover" in malwagon.moduleinfo["module-type"] + assert "hover" not in malwagon_submit.moduleinfo["module-type"]