diff --git a/documentation/logos/extuno.png b/documentation/logos/extuno.png new file mode 100644 index 00000000..54156f6b Binary files /dev/null and b/documentation/logos/extuno.png differ diff --git a/misp_modules/modules/expansion/extuno.py b/misp_modules/modules/expansion/extuno.py new file mode 100644 index 00000000..8d7819dc --- /dev/null +++ b/misp_modules/modules/expansion/extuno.py @@ -0,0 +1,336 @@ +import json +from urllib.parse import urlparse + +import requests +from pymisp import MISPEvent + +from . import check_input_attribute, standard_error_message + +misperrors = {"error": "Error"} +mispattributes = { + "input": ["text", "url", "link", "other"], + "output": ["text", "link"], + "format": "misp_standard", +} +moduleinfo = { + "version": "1.0", + "author": "extuno.com Tolga SEZER", + "description": ( + "Query Extuno for the reputation of a browser extension, an IDE extension or a package," + " identified by its store id or its listing URL." + ), + "module-type": ["expansion", "hover"], + "name": "Extuno Lookup", + "logo": "extuno.png", + "requirements": ["An Extuno API key"], + "features": ( + "The module takes a store listing URL, or a `store:id` pair in a text attribute, and reports" + " what Extuno knows about that artifact: whether it appears in the known-malicious catalog," + " with the threat type and the advisory it came from, and whether Extuno's own static and" + " sandbox analysis reached a verdict on it.\n\nIt covers browser extensions (Chrome, Firefox," + " Edge), IDE extensions (VS Code, JetBrains, Eclipse), Discord client mods and the npm, PyPI," + " Packagist, Maven, WordPress and Open VSX registries. Browser and IDE extensions are the" + " reason the module exists: they are not covered by OSV or any other public advisory" + " database, so an extension id found on a workstation cannot be checked against one.\n\n" + "Only the identifier being looked up leaves the MISP instance. The lookup is read-only and" + " never submits the artifact for analysis." + ), + "references": ["https://extuno.com", "https://extuno.com/pip-scanner"], + "input": "A store listing URL, or a text attribute holding `store:id` (e.g. `chrome:cjpalhd...`).", + "output": "A text attribute carrying the verdict, and a link to the advisory or store listing.", +} +# The API key is mandatory: the lookup runs against a tenant account. api_url is exposed only so an +# operator can point the module at a self-hosted deployment. +moduleconfig = ["api_key", "api_url"] + +DEFAULT_API_URL = "https://extuno.com" +USER_AGENT = "extuno-misp/1.0 (+https://github.com/MISP/misp-modules)" +TIMEOUT = 20 + +# Store identifiers accepted by the API, and the URL shapes they are published under. The path index +# is where the identifier sits after the marker segment; a negative index counts from the end, which +# is what a trailing-slash-tolerant listing URL needs. +STORES = ( + "chrome", "firefox", "edge", "vscode", "jetbrains", "eclipse", "discord", + "npm", "pypi", "composer", "maven", "wordpress", "openvsx", +) + +# A Chrome or Edge extension id is exactly 32 characters drawn from a-p. Nothing else in a MISP +# event looks like that, so a bare value of that shape is worth resolving without a store prefix. +_WEBSTORE_ID_LENGTH = 32 +_WEBSTORE_ID_ALPHABET = set("abcdefghijklmnop") + + +def _host_is(host, domain): + """Whether the host is that domain or a subdomain of it. + + str.endswith() is the wrong test: "evil-pypi.org".endswith("pypi.org") is true, which would let a + lookalike delivery URL inherit the verdict of the legitimate package it is imitating. That URL is + exactly the kind of indicator a MISP event holds, so the boundary has to be a dot. + """ + return host == domain or host.endswith("." + domain) + + +def _webstore_id(value): + """A Chrome or Edge extension id, or None. The format is exactly 32 characters from a-p.""" + candidate = (value or "").strip().lower() + if len(candidate) == _WEBSTORE_ID_LENGTH and set(candidate) <= _WEBSTORE_ID_ALPHABET: + return candidate + return None + + +class UnknownArtifact(Exception): + """The attribute does not name an artifact this module can look up. + + Distinct from an empty answer on purpose: telling an analyst "not known malicious" for a value + we never managed to parse would be a verdict we did not actually obtain. + """ + + +def _from_url(value): + """Resolve a store listing URL to a (store, identifier) pair, or None.""" + parsed = urlparse(value) + host = (parsed.hostname or "").lower() + parts = [p for p in parsed.path.split("/") if p] + + def after(marker, offset=1): + if marker in parts: + index = parts.index(marker) + offset + if index < len(parts): + return parts[index] + return None + + def webstore(store): + # .../detail//, and older listings omit the slug. Taking whichever segment is + # present would turn a URL truncated at the slug into a lookup for the slug, so the id is + # validated rather than assumed. + return next(((store, found) for found in + (_webstore_id(after("detail", 2)), _webstore_id(after("detail", 1))) if found), + None) + + if host in ("chromewebstore.google.com", "chrome.google.com"): + return webstore("chrome") + if _host_is(host, "microsoftedge.microsoft.com"): + return webstore("edge") + if _host_is(host, "addons.mozilla.org"): + slug = after("addon") + return ("firefox", slug) if slug else None + if _host_is(host, "marketplace.visualstudio.com"): + # itemName=. is the only stable identifier here. + for pair in (parsed.query or "").split("&"): + if pair.startswith("itemName="): + item = pair[len("itemName="):] + return ("vscode", item) if item else None + return None + if _host_is(host, "open-vsx.org"): + publisher, name = after("extension", 1), after("extension", 2) + return ("openvsx", f"{publisher}.{name}") if publisher and name else None + if _host_is(host, "plugins.jetbrains.com"): + # /plugin/-. A URL carrying only the slug names a plugin this module + # cannot address, so it is rejected rather than looked up under the slug. + plugin = (after("plugin") or "").split("-", 1)[0] + return ("jetbrains", plugin) if plugin.isdigit() else None + if _host_is(host, "marketplace.eclipse.org"): + slug = after("content") + return ("eclipse", slug) if slug else None + if _host_is(host, "npmjs.com"): + name = after("package") + if name and name.startswith("@"): + # A scope on its own is not a package. + scoped = after("package", 2) + return ("npm", f"{name}/{scoped}") if scoped else None + return ("npm", name) if name else None + if _host_is(host, "pypi.org"): + name = after("project") + return ("pypi", name) if name else None + if _host_is(host, "packagist.org"): + vendor, package = after("packages", 1), after("packages", 2) + return ("composer", f"{vendor}/{package}") if vendor and package else None + if _host_is(host, "wordpress.org"): + slug = after("plugins") + return ("wordpress", slug) if slug else None + return None + + +def _resolve(attribute): + """Work out which store and identifier the attribute names.""" + value = str(attribute.get("value", "")).strip() + if not value: + raise UnknownArtifact("the attribute value is empty") + + if value.lower().startswith(("http://", "https://")): + resolved = _from_url(value) + if resolved: + return resolved + raise UnknownArtifact(f"{value} is not a store listing URL Extuno recognises") + + # An explicit store prefix is unambiguous, so it wins over any shape heuristic. + if ":" in value: + store, _, identifier = value.partition(":") + store = store.strip().lower() + if store in STORES and identifier.strip(): + return store, identifier.strip() + + bare = _webstore_id(value) + if bare: + # Chrome and Edge share this id format, so both are asked and the first hit answers. + return "chrome", bare + + raise UnknownArtifact( + "prefix the value with its store (e.g. chrome:cjpalhdlnbpafiamejdnhcphjbkeiagm) " + "or use the store listing URL" + ) + + +def _lookup(api_url, api_key, store, identifier): + try: + response = requests.get( + f"{api_url}/v1/lookup", + params={"store": store, "id": identifier}, + headers={"X-Api-Key": api_key, "User-Agent": USER_AGENT, "Accept": "application/json"}, + timeout=TIMEOUT, + ) + if response.status_code in (401, 403): + return {"__error__": "Extuno rejected the API key."} + if response.status_code == 429: + return {"__error__": "Extuno rate limit reached; try again shortly."} + response.raise_for_status() + body = response.json() + # A 200 carrying a JSON array or a bare string is not an answer from this API. Returning it + # would put a stack trace where the module contract expects an error dict. + return body if isinstance(body, dict) else None + except (requests.exceptions.RequestException, ValueError): + return None + + +class ExtunoParser: + def __init__(self, api_url, api_key): + self.api_url = api_url + self.api_key = api_key + self.misp_event = MISPEvent() + self.found = False + + def _add(self, **kwargs): + self.misp_event.add_attribute(**kwargs) + self.found = True + + def _query(self, store, identifier): + """One lookup. Returns (payload, failure); exactly one of them is set.""" + result = _lookup(self.api_url, self.api_key, store, identifier) + if result is None: + return None, "Extuno could not be reached." + if result.get("__error__"): + return None, result["__error__"] + return result, None + + def parse(self, store, identifier): + result, failure = self._query(store, identifier) + if failure: + return failure + + # A Chrome id and an Edge id are the same 32 characters, so a miss on one is not an answer. + # The verdict is normalised the same way here as it is below: a response with no verdict, or + # a null one, is a miss and must reach the fallback like an explicit "unknown" does. + if store == "chrome" and (result.get("verdict") or "unknown") == "unknown": + edge, edge_failure = self._query("edge", identifier) + # A rejected key or a rate limit on the second call is a failed lookup, not a clean + # answer. Reporting "not in the malicious catalog" for a question that was never + # answered is the one outcome a reputation module must never produce. + if edge_failure: + return edge_failure + if (edge.get("verdict") or "unknown") != "unknown": + store, result = "edge", edge + + verdict = result.get("verdict") or "unknown" + # A field of the wrong type is treated as absent rather than trusted: an operator can point + # api_url at another deployment, and a proxy can rewrite a body. + catalog = result.get("catalog") if isinstance(result.get("catalog"), dict) else {} + scan = result.get("scan") if isinstance(result.get("scan"), dict) else {} + + if result.get("known_malicious"): + threat = catalog.get("threat_type") or "malicious" + summary = f"Extuno: {store}:{identifier} is listed as malicious ({threat})" + if catalog.get("still_active") is False: + summary += ", removed from the store" + self._add(type="text", value=summary, comment="Extuno: known-malicious catalog", + disable_correlation=True) + if catalog.get("reason"): + self._add(type="text", value=f"Extuno: {catalog['reason']}", + comment="Extuno: catalog detail", disable_correlation=True) + if catalog.get("source_url"): + self._add(type="link", value=catalog["source_url"], + comment="Extuno: advisory this listing came from", disable_correlation=True) + elif verdict != "unknown": + summary = f"Extuno analysed {store}:{identifier} and reached the verdict {verdict}" + if result.get("risk_score") is not None: + summary += f" (risk {result['risk_score']}/100)" + self._add(type="text", value=summary, comment="Extuno: analysis verdict", + disable_correlation=True) + else: + # Reported rather than returned as an error: "we looked and it is not on record" is a + # useful answer, and is not the same as the lookup having failed. + self._add( + type="text", + value=f"Extuno has no record of {store}:{identifier}: not in the malicious catalog " + "and not analysed.", + comment="Extuno: no record", disable_correlation=True) + return None + + findings = scan.get("top_findings") + findings = findings if isinstance(findings, list) else [] + for title in [f["title"] for f in findings + if isinstance(f, dict) and f.get("title")][:5]: + self._add(type="text", value=f"Extuno finding: {title}", + comment="Extuno: evidence from analysis", disable_correlation=True) + return None + + def get_results(self): + if not self.found: + return {"error": "No Extuno results for this attribute."} + event = json.loads(self.misp_event.to_json()) + results = {key: event[key] for key in ("Attribute", "Object") if event.get(key)} + if not results: + return {"error": "No Extuno results for this attribute."} + return {"results": results} + + +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"]): + return {"error": f"{standard_error_message}, which should contain at least a type, a value and an UUID."} + + attribute = request["attribute"] + if attribute.get("type") not in mispattributes["input"]: + return {"error": "Unsupported attribute type."} + + config = request.get("config") or {} + api_key = str(config.get("api_key") or "").strip() + if not api_key: + return {"error": "An Extuno API key is required; set api_key in the module configuration."} + api_url = str(config.get("api_url") or DEFAULT_API_URL).rstrip("/") + + try: + store, identifier = _resolve(attribute) + except UnknownArtifact as error: + return {"error": f"Extuno cannot look this up: {error}"} + + parser = ExtunoParser(api_url, api_key) + # The identifier is passed raw: requests encodes it once as a query parameter. Encoding it here + # as well would put a literal "%3A" on the wire for a Maven coordinate and look up an artifact + # that cannot exist. + failure = parser.parse(store, identifier) + if failure: + return {"error": failure} + return parser.get_results() + + +def introspection(): + return mispattributes + + +def version(): + moduleinfo["config"] = moduleconfig + return moduleinfo diff --git a/tests/test_extuno.py b/tests/test_extuno.py new file mode 100644 index 00000000..c10a97e2 --- /dev/null +++ b/tests/test_extuno.py @@ -0,0 +1,237 @@ +import json +from unittest.mock import MagicMock, patch + +from misp_modules.modules.expansion import extuno + +UUID = "5b582d80-7a7e-4b6a-9f22-77656e72bb3b" + +MALICIOUS = { + "store": "chrome", + "ext_id": "bidgllfieacmghieipmhgabodmljimfh", + "verdict": "malicious", + "known_malicious": True, + "catalog": { + "threat_type": "Bundling Unwanted Software", + "reason": "Listed by a public advisory.", + "source_url": "https://example.com/report/", + "still_active": False, + }, + "scan": None, +} +UNKNOWN = {"store": "chrome", "verdict": "unknown", "known_malicious": False, "catalog": None, "scan": None} +ANALYSED = { + "store": "chrome", + "verdict": "review", + "known_malicious": False, + "risk_score": 72, + "catalog": None, + "scan": {"top_findings": [{"title": "Cookie access combined with broad host reach"}]}, +} + + +class MockResponse: + def __init__(self, payload, status_code=200): + self.payload = payload + self.status_code = status_code + + def json(self): + return self.payload + + def raise_for_status(self): + if self.status_code >= 400: + raise extuno.requests.exceptions.HTTPError(response=self) + + +def _query(value, type_="text", config=None): + attribute = {"type": type_, "value": value, "uuid": UUID} + config = {"api_key": "extk_test"} if config is None else config + return json.dumps({"module": "extuno", "attribute": attribute, "config": config}) + + +def _answer(*payloads): + """Return a requests.get replacement serving the given payloads in order.""" + queue = list(payloads) + + def _get(url, params=None, headers=None, timeout=None): + return MockResponse(queue.pop(0) if queue else UNKNOWN) + + return _get + + +def test_a_catalog_listing_reports_the_threat_and_the_advisory_it_came_from(): + with patch.object(extuno.requests, "get", _answer(MALICIOUS)): + results = extuno.handler(_query("chrome:bidgllfieacmghieipmhgabodmljimfh")) + + values = [a["value"] for a in results["results"]["Attribute"]] + assert any("listed as malicious" in v and "Bundling Unwanted Software" in v for v in values) + assert "https://example.com/report/" in values + # An extension pulled from the store is no longer installable from it, which changes what an + # analyst does next, so it is stated rather than left out. + assert any("removed from the store" in v for v in values) + + +def test_an_analysis_verdict_carries_its_evidence(): + with patch.object(extuno.requests, "get", _answer(ANALYSED)): + results = extuno.handler(_query("chrome:kbnfbcpkiaganjpcanopcgeoehkleeck")) + + values = [a["value"] for a in results["results"]["Attribute"]] + assert any("verdict review" in v and "72/100" in v for v in values) + assert any("Cookie access combined with broad host reach" in v for v in values) + + +def test_a_bare_webstore_id_is_also_looked_up_against_edge(): + """Chrome and Edge extension ids share one format, so a miss on Chrome is not an answer.""" + edge_hit = dict(MALICIOUS, store="edge") + with patch.object(extuno.requests, "get", _answer(UNKNOWN, edge_hit)): + results = extuno.handler(_query("bidgllfieacmghieipmhgabodmljimfh")) + + assert any("listed as malicious" in a["value"] for a in results["results"]["Attribute"]) + + +def test_no_record_is_reported_as_a_result_rather_than_an_error(): + """"We looked and it is not on record" is an answer; an error would read as "we did not look".""" + with patch.object(extuno.requests, "get", _answer(UNKNOWN)): + results = extuno.handler(_query("chrome:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")) + + assert "error" not in results + assert any("no record" in a["value"] for a in results["results"]["Attribute"]) + + +def test_a_value_that_cannot_be_resolved_is_never_reported_as_clean(): + """The dangerous failure for a reputation module: a parse failure read as a clean verdict.""" + results = extuno.handler(_query("some free text")) + assert "error" in results + assert "results" not in results + + +def test_a_missing_api_key_is_reported_before_any_request_is_made(): + """No key means no lookup: the module must not query on the caller's behalf without one.""" + request = MagicMock(return_value=MockResponse(MALICIOUS)) + with patch.object(extuno.requests, "get", request): + results = extuno.handler(_query("chrome:x", config={})) + assert results["error"].startswith("An Extuno API key is required") + request.assert_not_called() + + +def test_a_rejected_key_is_distinguished_from_an_absent_result(): + with patch.object(extuno.requests, "get", lambda *a, **k: MockResponse({}, 401)): + results = extuno.handler(_query("chrome:bidgllfieacmghieipmhgabodmljimfh")) + assert results["error"] == "Extuno rejected the API key." + + +def test_store_listing_urls_resolve_to_the_right_store_and_identifier(): + cases = { + "https://chromewebstore.google.com/detail/ublock-origin/cjpalhdlnbpafiamejdnhcphjbkeiagm": ( + "chrome", "cjpalhdlnbpafiamejdnhcphjbkeiagm"), + "https://addons.mozilla.org/en-US/firefox/addon/ublock-origin/": ("firefox", "ublock-origin"), + "https://marketplace.visualstudio.com/items?itemName=esbenp.prettier-vscode": ( + "vscode", "esbenp.prettier-vscode"), + "https://open-vsx.org/extension/vscodevim/vim": ("openvsx", "vscodevim.vim"), + "https://plugins.jetbrains.com/plugin/7495-intellijruby": ("jetbrains", "7495"), + "https://marketplace.eclipse.org/content/checkstyle-plug": ("eclipse", "checkstyle-plug"), + "https://www.npmjs.com/package/@types/node": ("npm", "@types/node"), + "https://pypi.org/project/requests/": ("pypi", "requests"), + "https://packagist.org/packages/monolog/monolog": ("composer", "monolog/monolog"), + "https://wordpress.org/plugins/classic-editor/": ("wordpress", "classic-editor"), + } + for url, expected in cases.items(): + assert extuno._from_url(url) == expected, url + assert extuno._from_url("https://example.com/whatever") is None + + +def test_a_failure_on_the_edge_fallback_is_not_reported_as_a_clean_answer(): + """The second lookup must propagate its failure like the first one does. + + A rejected key or a rate limit on the fallback used to be accepted as the answer, which turned + a question that was never answered into "not in the malicious catalog". The fallback doubles the + request rate for every bare id, so a rate limit there is the expected case, not the rare one. + """ + calls = {"n": 0} + + def _get(url, params=None, headers=None, timeout=None): + calls["n"] += 1 + return MockResponse(UNKNOWN) if calls["n"] == 1 else MockResponse({}, 401) + + with patch.object(extuno.requests, "get", _get): + results = extuno.handler(_query("bidgllfieacmghieipmhgabodmljimfh")) + + assert results["error"] == "Extuno rejected the API key." + assert "results" not in results + + +def test_a_lookalike_host_does_not_inherit_the_real_package_verdict(): + """A typosquat delivery URL is exactly the indicator a MISP event holds. + + Suffix matching accepted evil-pypi.org as pypi.org, so the lookalike was enriched with the + legitimate package's verdict. + """ + for url in ("https://evil-pypi.org/project/requests/", + "https://notnpmjs.com/package/left-pad", + "https://myaddons.mozilla.org/en-US/firefox/addon/ublock-origin/", + "https://fakepackagist.org/packages/monolog/monolog"): + assert extuno._from_url(url) is None, url + + # The genuine hosts, and subdomains of them, still resolve. + assert extuno._from_url("https://pypi.org/project/requests/") == ("pypi", "requests") + assert extuno._from_url("https://www.npmjs.com/package/left-pad") == ("npm", "left-pad") + + +def test_the_identifier_reaches_the_api_encoded_exactly_once(): + """requests encodes a query parameter; encoding it here as well queried a different artifact. + + A Maven coordinate is `group:artifact`, so the colon is intrinsic and the double encoding put a + literal %3A on the wire, guaranteeing a miss reported as "no record". + """ + seen = {} + + def _get(url, params=None, headers=None, timeout=None): + seen.update(params or {}) + return MockResponse(UNKNOWN) + + with patch.object(extuno.requests, "get", _get): + extuno.handler(_query("maven:org.apache.commons:commons-lang3")) + + assert seen["id"] == "org.apache.commons:commons-lang3" + + +def test_a_listing_url_without_an_identifier_is_rejected_rather_than_guessed(): + """Taking whichever path segment is present turned a truncated URL into a lookup for the slug, + which then answered "no record" for an extension that exists.""" + assert extuno._from_url("https://chromewebstore.google.com/detail/ublock-origin") is None + assert extuno._from_url("https://plugins.jetbrains.com/plugin/intellijruby") is None + assert extuno._from_url("https://www.npmjs.com/package/@types") is None + + assert extuno._from_url( + "https://chromewebstore.google.com/detail/ublock-origin/cjpalhdlnbpafiamejdnhcphjbkeiagm" + ) == ("chrome", "cjpalhdlnbpafiamejdnhcphjbkeiagm") + assert extuno._from_url("https://plugins.jetbrains.com/plugin/7495-intellijruby") == ( + "jetbrains", "7495") + + +def test_a_response_of_the_wrong_shape_returns_an_error_rather_than_raising(): + """api_url is operator configurable and a proxy can rewrite a body, so a 200 carrying something + other than the expected object must not escape as a stack trace.""" + for payload in ([], "text", 7, {"verdict": "review", "scan": "not a dict"}, + {"verdict": "review", "scan": {"top_findings": ["not a dict"]}}, + {"verdict": "malicious", "known_malicious": True, "catalog": []}): + with patch.object(extuno.requests, "get", _answer(payload)): + results = extuno.handler(_query("chrome:bidgllfieacmghieipmhgabodmljimfh")) + assert isinstance(results, dict) + assert "error" in results or "results" in results + + +def test_a_response_with_no_verdict_still_reaches_the_edge_fallback(): + """The fallback tested the raw field while the parser normalised it, so a missing or null + verdict skipped the fallback and a known-malicious Edge extension was reported as no record.""" + for first in ({"known_malicious": False}, {"verdict": None}): + stores = [] + + def _get(url, params=None, headers=None, timeout=None): + stores.append((params or {}).get("store")) + return MockResponse(first if len(stores) == 1 else dict(MALICIOUS, store="edge")) + + with patch.object(extuno.requests, "get", _get): + results = extuno.handler(_query("bidgllfieacmghieipmhgabodmljimfh")) + + assert stores == ["chrome", "edge"] + assert any("listed as malicious" in a["value"] for a in results["results"]["Attribute"])