Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions misp_modules/modules/expansion/vulnerability_lookup.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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))
Expand Down
62 changes: 62 additions & 0 deletions tests/test_vulnerability_lookup.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down