diff --git a/misp_modules/modules/expansion/vulnerability_lookup.py b/misp_modules/modules/expansion/vulnerability_lookup.py index b4821bc4f..6cc38e73f 100644 --- a/misp_modules/modules/expansion/vulnerability_lookup.py +++ b/misp_modules/modules/expansion/vulnerability_lookup.py @@ -39,6 +39,8 @@ "it bounds alias lookup attempts, CPE results and each enrichment section. " "with_linked, with_comments, with_bundles, with_sightings, with_meta, with_epss and with_kev default to true; " "with_timeline defaults to false. Boolean options accept true/false or their string equivalents.\n" + "Unset options (null or blank strings from MISP) use their documented defaults. " + "The request's top-level timeout limits total module execution independently of config.timeout.\n" "CPE lookup uses cpe_source (default cvelistv5) and cpe_page (default 1). Results match vendor and product; " "exact version applicability must be checked. CPE search imports advisories from one bounded page; " "the optional community and priority lookups apply to vulnerability ID input.\n" @@ -80,6 +82,13 @@ def handler(q=False): config = request.get("config") or {} if not isinstance(config, dict): return {"error": "Invalid module configuration."} + # MISP sends all advertised settings, including null/blank values for unset options. + # Keep explicit false and zero; they are meaningful configuration values. + config = { + key: value + for key, value in config.items() + if value is not None and not (isinstance(value, str) and not value.strip()) + } try: timeout = float(config.get("timeout", 30)) max_related = int(config.get("max_related", 50)) diff --git a/tests/test_vulnerability_lookup.py b/tests/test_vulnerability_lookup.py index 5590b6edc..a27e90361 100644 --- a/tests/test_vulnerability_lookup.py +++ b/tests/test_vulnerability_lookup.py @@ -532,6 +532,68 @@ def test_custom_url_timeout_and_identifier_encoding_are_used_for_all_requests(): assert objects(result, "vulnerability") +def test_misp_null_configuration_reaches_the_api_and_keeps_default_enrichment(): + query = { + "module": "vulnerability_lookup", + "attribute": {"type": "vulnerability", "value": CVE, "uuid": UUID}, + "config": dict.fromkeys(vulnerability_lookup.moduleconfig), + "timeout": 10, + } + query["config"]["timeout"] = "80" + with patch("requests.get", side_effect=[response(cve()), response({}, 404), response({}, 404)]) as get: + result = vulnerability_lookup.handler(json.dumps(query)) + assert objects(result, "vulnerability") + assert get.call_count == 3 + main_call = get.call_args_list[0] + assert main_call.kwargs["timeout"] == 80 + assert main_call.kwargs["params"] == { + "with_linked": "true", + "with_comments": "true", + "with_bundles": "true", + "with_sightings": "true", + "with_meta": "true", + } + assert get.call_args_list[-1].kwargs["params"]["per_page"] == 50 + assert not any(call.args[0].endswith("/timeline") for call in get.call_args_list) + + +@pytest.mark.parametrize("unset", [None, "", " "]) +def test_unset_numeric_and_boolean_options_use_defaults(unset): + config = dict.fromkeys(vulnerability_lookup.moduleconfig, unset) + _, get = run(cve(), config=config, extra_responses=[response({}, 404), response({}, 404)]) + assert get.call_args_list[0].kwargs["timeout"] == 30 + assert get.call_args_list[0].kwargs["params"]["with_linked"] == "true" + assert get.call_args_list[-1].kwargs["params"]["per_page"] == 50 + + +@pytest.mark.parametrize("unset", [None, "", " "]) +def test_unset_cpe_settings_use_defaults(unset): + config = dict.fromkeys(vulnerability_lookup.moduleconfig, unset) + result, get = run( + {"cvelistv5": [cve()]}, attribute_type="cpe", value="cpe:2.3:a:example:app:*:*:*:*:*:*:*:*", config=config + ) + assert objects(result, "vulnerability") + assert get.call_args.kwargs["params"] == {"per_page": 50, "page": 1, "source": "cvelistv5"} + + +@pytest.mark.parametrize("disabled", [False, "false", 0, "0"]) +def test_explicit_false_flags_are_not_replaced_by_defaults(disabled): + config = dict.fromkeys(vulnerability_lookup.moduleconfig) + config.update({name: disabled for name in vulnerability_lookup.moduleconfig if name.startswith("with_")}) + _, get = run(cve(), config=config) + assert get.call_count == 1 + assert all(value == "false" for value in get.call_args.kwargs["params"].values()) + + +@pytest.mark.parametrize("zero", [0, "0"]) +def test_explicit_zero_related_limit_is_preserved(zero): + config = dict.fromkeys(vulnerability_lookup.moduleconfig) + config.update({"max_related": zero, "with_epss": False}) + result, get = run(cve(), config=config) + assert objects(result, "vulnerability") + assert get.call_count == 1 + + @pytest.mark.parametrize("query", ["invalid", "[]", "null", "{}", '{"attribute": []}']) def test_invalid_input_is_an_error(query): with patch("requests.get") as get: