From 6b1479435e98834b0697ba45fc0368dd7d7375e5 Mon Sep 17 00:00:00 2001 From: Tamar Weisskopf Date: Sun, 12 Jul 2026 07:00:57 +0300 Subject: [PATCH] Config scanner: Extension allowlist and build config patterns --- .../tools/configuration_scanner.py | 48 +- tests/test_configuration_scanner.py | 451 +++++++++++++++++- 2 files changed, 480 insertions(+), 19 deletions(-) diff --git a/src/vuln_analysis/tools/configuration_scanner.py b/src/vuln_analysis/tools/configuration_scanner.py index a9801d450..228cbed1a 100644 --- a/src/vuln_analysis/tools/configuration_scanner.py +++ b/src/vuln_analysis/tools/configuration_scanner.py @@ -53,7 +53,11 @@ def format_context_snippet(lines: list[str], match_line: int, context_lines: int "config.yaml", "config.yml", "config.xml", "settings.toml", "settings.yaml", "settings.yml", "web.xml", "beans.xml", - "Dockerfile", "Dockerfile.*", "docker-compose*.yml", + "Dockerfile", "Dockerfile.*", "docker-compose*.yml", "docker-compose*.yaml", + # Build/tool configuration — contain compiler flags, tool settings, or build options + "pyproject.toml", "setup.cfg", "tox.ini", + "tsconfig.json", ".eslintrc.json", + "Makefile", "CMakeLists.txt", "meson.build", # Config-specific extensions — safe to match anywhere "*.properties", "*.env", "*.conf", "*.ini", ] @@ -61,6 +65,23 @@ def format_context_snippet(lines: list[str], match_line: int, context_lines: int # Directory names that typically contain configuration files CONFIG_DIR_PATTERNS = ["config", "conf", "conf.d", "etc", "resources"] +_BINARY_EXTENSIONS = frozenset({ + ".jar", ".war", ".ear", ".class", ".zip", ".tar", ".gz", ".bz2", ".xz", + ".png", ".jpg", ".jpeg", ".gif", ".ico", ".svg", ".woff", ".woff2", + ".ttf", ".eot", ".pdf", ".so", ".dylib", ".dll", ".exe", ".pyc", ".pyo", +}) + +# Extensions accepted for files matched only by directory (e.g. under resources/). +# Files matched by _is_config_file (name pattern like application.yml) bypass this. +# Extensionless files (Caddyfile, Makefile, etc.) are also accepted. +# Ecosystem-specific named files (go.mod, requirements.txt) belong in +# CONFIG_FILE_PATTERNS instead — adding .txt here would re-admit theme/doc noise. +_CONFIG_DIR_ALLOWED_EXTENSIONS = frozenset({ + ".xml", ".yml", ".yaml", ".json", + ".toml", ".cfg", ".ini", ".conf", + ".properties", ".env", +}) + # Avoids per-file regex compilation _CONFIG_EXTENSIONS = [] _CONFIG_EXACT_NAMES = [] @@ -93,7 +114,7 @@ def _is_config_file(file_path: str) -> bool: return True if lower_name in _CONFIG_EXACT_NAMES: return True - if any(p.match(lower_name) for p in _CONFIG_WILDCARD_PATTERNS): + if any(p.fullmatch(lower_name) for p in _CONFIG_WILDCARD_PATTERNS): return True return False @@ -117,7 +138,14 @@ def _collect_config_files(repo_path: str) -> list[tuple[str, str]]: for fname in files: full_path = os.path.join(root, fname) rel_path = os.path.relpath(full_path, repo_path) - if _is_config_file(rel_path) or _is_in_config_dir(rel_path): + if any(fname.lower().endswith(ext) for ext in _BINARY_EXTENSIONS): + continue + is_named_config = _is_config_file(rel_path) + if not is_named_config and _is_in_config_dir(rel_path): + ext = os.path.splitext(fname)[1].lower() + if ext and ext not in _CONFIG_DIR_ALLOWED_EXTENSIONS: + continue + if is_named_config or _is_in_config_dir(rel_path): try: with open(full_path, "r", errors="ignore") as f: content = f.read() @@ -149,6 +177,8 @@ def search_config_content( source_label: str = "unknown", ) -> list[str]: """Match keywords against config file contents, returning formatted snippets.""" + if max_results <= 0: + return [] matches = [] for rel_path, content in config_files: lines = content.split("\n") @@ -193,21 +223,23 @@ async def _arun(query: str) -> str: continue repo_key = (si.git_repo, si.ref) - if repo_key in _config_files_cache: - async with _repo_locks_guard: + async with _repo_locks_guard: + if repo_key in _config_files_cache: _config_files_cache.move_to_end(repo_key) - else: - async with _repo_locks_guard: + cached = True + else: if repo_key not in _repo_locks: _repo_locks[repo_key] = asyncio.Lock() repo_lock = _repo_locks[repo_key] + cached = False + if not cached: async with repo_lock: if repo_key not in _config_files_cache: _config_files_cache[repo_key] = _collect_config_files(str(repo_path)) if len(_config_files_cache) > _CONFIG_CACHE_MAX_SIZE: _config_files_cache.popitem(last=False) - for cfg in _config_files_cache[repo_key]: + for cfg in _config_files_cache.get(repo_key, []): if is_dependency_path(cfg[0]): all_dep_configs.append(cfg) else: diff --git a/tests/test_configuration_scanner.py b/tests/test_configuration_scanner.py index ca9a462c8..e629bd236 100644 --- a/tests/test_configuration_scanner.py +++ b/tests/test_configuration_scanner.py @@ -13,16 +13,96 @@ # See the License for the specific language governing permissions and # limitations under the License. +import asyncio +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + import pytest +import pytest_asyncio from vuln_analysis.tools.configuration_scanner import ( + _BINARY_EXTENSIONS, + _CONFIG_DIR_ALLOWED_EXTENSIONS, _is_config_file, _is_in_config_dir, _collect_config_files, + _count_config_matches, + format_context_snippet, search_config_content, + ConfigurationScannerToolConfig, + configuration_scanner, ) +class TestFormatContextSnippet: + def test_match_at_first_line(self): + lines = ["alpha", "bravo", "charlie", "delta", "echo"] + result = format_context_snippet(lines, match_line=0, context_lines=2) + output_lines = result.split("\n") + assert output_lines[0] == "> 1: alpha" + assert output_lines[1] == " 2: bravo" + assert output_lines[2] == " 3: charlie" + assert len(output_lines) == 3 + + def test_match_at_last_line(self): + lines = ["alpha", "bravo", "charlie", "delta", "echo"] + result = format_context_snippet(lines, match_line=4, context_lines=2) + output_lines = result.split("\n") + assert output_lines[0] == " 3: charlie" + assert output_lines[1] == " 4: delta" + assert output_lines[2] == "> 5: echo" + assert len(output_lines) == 3 + + def test_zero_context_lines(self): + lines = ["alpha", "bravo", "charlie"] + result = format_context_snippet(lines, match_line=1, context_lines=0) + assert result == "> 2: bravo" + + def test_middle_match(self): + lines = ["one", "two", "three", "four", "five"] + result = format_context_snippet(lines, match_line=2, context_lines=1) + output_lines = result.split("\n") + assert output_lines[0] == " 2: two" + assert output_lines[1] == "> 3: three" + assert output_lines[2] == " 4: four" + + def test_single_line_file(self): + assert format_context_snippet(["only"], match_line=0, context_lines=5) == "> 1: only" + + +class TestCountConfigMatches: + def test_counts_matching_lines(self): + config_files = [ + ("app.properties", "server.port=8080\nssl.enabled=true\nlogging.level=INFO"), + ("db.yml", "host: localhost\nport: 5432\nssl: false"), + ] + assert _count_config_matches(config_files, ["ssl"]) == 2 + + def test_case_insensitive(self): + config_files = [("app.yml", "SSL_ENABLED=true\nSsl_Mode=require")] + assert _count_config_matches(config_files, ["ssl"]) == 2 + + def test_no_matches(self): + config_files = [("app.yml", "host: localhost\nport: 8080")] + assert _count_config_matches(config_files, ["ssl", "tls"]) == 0 + + def test_multiple_keywords_same_line(self): + config_files = [("app.conf", "ssl_port=443")] + assert _count_config_matches(config_files, ["ssl", "port"]) == 1 + + def test_empty_config_files(self): + assert _count_config_matches([], ["ssl"]) == 0 + + +class TestBinaryExtensions: + def test_contains_expected_extensions(self): + expected = {".jar", ".png", ".jpg", ".pdf", ".class", ".zip", ".pyc"} + assert expected.issubset(_BINARY_EXTENSIONS) + + def test_is_frozenset(self): + assert isinstance(_BINARY_EXTENSIONS, frozenset) + + class TestIsConfigFile: @pytest.mark.parametrize( "file_path,expected", @@ -59,13 +139,13 @@ class TestIsConfigFile: ("package.json", False), ("requirements.txt", False), ("setup.py", False), - ("setup.cfg", False), - ("pyproject.toml", False), + ("setup.cfg", True), + ("pyproject.toml", True), # Removed: catch-all extensions (only matched inside config dirs) ("app.yml", False), ("app.cfg", False), ("data.yaml", False), - ("docker-compose-dev.yaml", False), + ("docker-compose-dev.yaml", True), # Non-config files ("main.py", False), ("utils.go", False), @@ -129,6 +209,108 @@ def test_config_dir_patterns(self, file_path, expected): def test_case_insensitive_dir(self, file_path, expected): assert _is_in_config_dir(file_path) == expected + def test_bare_filename_not_in_config_dir(self): + """A bare filename with no directory parts should not match any config dir.""" + assert _is_in_config_dir("settings.txt") is False + assert _is_in_config_dir("application.yml") is False + + +class TestConfigDirAllowedExtensions: + """Files in config directories are filtered by an extension allowlist.""" + + def test_js_in_resources_dir_skipped(self, tmp_path): + """Reproduces Keycloak issue: .js/.map/.css files under resources/ were collected.""" + res_dir = tmp_path / "src" / "main" / "resources" / "theme" + res_dir.mkdir(parents=True) + for fname in ["react-core.js", "react-core.js.map", "app.css"]: + (res_dir / fname).write_text("not a config") + (res_dir / "settings.yml").write_text("key: value") + + results = _collect_config_files(str(tmp_path)) + names = [p.split("/")[-1] for p, _ in results] + assert "settings.yml" in names + assert "react-core.js" not in names + assert "react-core.js.map" not in names + assert "app.css" not in names + + def test_java_source_in_resources_dir_skipped(self, tmp_path): + res_dir = tmp_path / "resources" + res_dir.mkdir() + (res_dir / "Helper.java").write_text("public class Helper {}") + names = [p.split("/")[-1] for p, _ in _collect_config_files(str(tmp_path))] + assert "Helper.java" not in names + + def test_named_config_bypasses_extension_filter(self, tmp_path): + """application.properties matched by name pattern bypasses the extension filter.""" + res_dir = tmp_path / "resources" + res_dir.mkdir() + (res_dir / "application.properties").write_text("server.port=8080") + names = [p.split("/")[-1] for p, _ in _collect_config_files(str(tmp_path))] + assert "application.properties" in names + + def test_extensionless_file_in_config_dir_collected(self, tmp_path): + conf_dir = tmp_path / "config" + conf_dir.mkdir() + (conf_dir / "Caddyfile").write_text(":8080 { respond 200 }") + names = [p.split("/")[-1] for p, _ in _collect_config_files(str(tmp_path))] + assert "Caddyfile" in names + + def test_all_allowed_extensions_collected(self, tmp_path): + conf_dir = tmp_path / "conf" + conf_dir.mkdir() + for ext in _CONFIG_DIR_ALLOWED_EXTENSIONS: + (conf_dir / f"test{ext}").write_text("content") + names = {p.split("/")[-1] for p, _ in _collect_config_files(str(tmp_path))} + for ext in _CONFIG_DIR_ALLOWED_EXTENSIONS: + assert f"test{ext}" in names, f"Expected test{ext} to be collected" + + def test_allowlist_is_frozenset(self): + assert isinstance(_CONFIG_DIR_ALLOWED_EXTENSIONS, frozenset) + + def test_contains_core_extensions(self): + expected = {".xml", ".yml", ".yaml", ".json", ".toml", ".properties", ".conf", ".ini", ".env"} + assert expected.issubset(_CONFIG_DIR_ALLOWED_EXTENSIONS) + + def test_skips_binary_files_in_config_dir(self, tmp_path): + conf_dir = tmp_path / "config" + conf_dir.mkdir() + (conf_dir / "settings.yml").write_text("key: value") + (conf_dir / "logo.png").write_bytes(b"\x89PNG fake") + (conf_dir / "lib.jar").write_bytes(b"PK fake jar") + names = [p.split("/")[-1] for p, _ in _collect_config_files(str(tmp_path))] + assert "settings.yml" in names + assert "logo.png" not in names + assert "lib.jar" not in names + + +class TestConfigFilePatterns: + """Tests for CONFIG_FILE_PATTERNS name-based matching.""" + + def test_build_tool_config_files(self): + assert _is_config_file("pyproject.toml") is True + assert _is_config_file("setup.cfg") is True + assert _is_config_file("tox.ini") is True + assert _is_config_file("tsconfig.json") is True + assert _is_config_file(".eslintrc.json") is True + assert _is_config_file("Makefile") is True + assert _is_config_file("CMakeLists.txt") is True + assert _is_config_file("meson.build") is True + + def test_dependency_manifests_not_matched(self): + assert _is_config_file("go.mod") is False + assert _is_config_file("go.sum") is False + assert _is_config_file("requirements.txt") is False + assert _is_config_file("package.json") is False + assert _is_config_file("Cargo.toml") is False + assert _is_config_file("Gemfile") is False + assert _is_config_file("pom.xml") is False + + def test_source_code_not_matched(self): + assert _is_config_file("main.java") is False + assert _is_config_file("app.js") is False + assert _is_config_file("util.py") is False + assert _is_config_file("handler.go") is False + class TestCollectConfigFiles: def test_finds_config_files(self, tmp_path): @@ -150,15 +332,15 @@ def test_finds_config_files(self, tmp_path): def test_finds_files_in_config_dir(self, tmp_path): config_dir = tmp_path / "config" config_dir.mkdir() - (config_dir / "database.txt").write_text("db_config") - (config_dir / "server.txt").write_text("server_config") + (config_dir / "database.xml").write_text("") + (config_dir / "server.yml").write_text("port: 8080") result = _collect_config_files(str(tmp_path)) assert len(result) == 2 paths = {path for path, _ in result} - assert "config/database.txt" in paths - assert "config/server.txt" in paths + assert "config/database.xml" in paths + assert "config/server.yml" in paths def test_excludes_git_dir(self, tmp_path): git_dir = tmp_path / ".git" @@ -189,7 +371,8 @@ def test_excludes_pycache(self, tmp_path): def test_excludes_node_modules(self, tmp_path): node_modules_dir = tmp_path / "node_modules" node_modules_dir.mkdir() - (node_modules_dir / "package.json").write_text("{}") + # Use a real config file so the directory exclusion is what prevents collection + (node_modules_dir / "config.yml").write_text("excluded: true") (tmp_path / "application.yml").write_text("app: config") result = _collect_config_files(str(tmp_path)) @@ -197,7 +380,7 @@ def test_excludes_node_modules(self, tmp_path): assert len(result) == 1 paths = {path for path, _ in result} assert "application.yml" in paths - assert "node_modules/package.json" not in paths + assert "node_modules/config.yml" not in paths def test_excludes_tox(self, tmp_path): tox_dir = tmp_path / ".tox" @@ -224,6 +407,17 @@ def test_skips_large_files(self, tmp_path): assert "small.properties" in paths assert "large.properties" not in paths + def test_size_boundary_at_500000_chars(self, tmp_path): + """The limit is `len(content) < 500_000`: 499,999 chars collected, 500,000 skipped.""" + (tmp_path / "under.properties").write_text("x" * 499_999) + (tmp_path / "exact.properties").write_text("x" * 500_000) + + result = _collect_config_files(str(tmp_path)) + + paths = {path for path, _ in result} + assert "under.properties" in paths + assert "exact.properties" not in paths + def test_empty_repo_returns_empty(self, tmp_path): result = _collect_config_files(str(tmp_path)) assert result == [] @@ -235,7 +429,7 @@ def test_nested_directory_structure(self, tmp_path): conf_dir = tmp_path / "conf" conf_dir.mkdir() - (conf_dir / "server.txt").write_text("server config") + (conf_dir / "server.yml").write_text("port: 8080") (tmp_path / "config.yaml").write_text("key: value") @@ -244,7 +438,7 @@ def test_nested_directory_structure(self, tmp_path): assert len(result) == 3 paths = {path for path, _ in result} assert "src/main/resources/application.yml" in paths - assert "conf/server.txt" in paths + assert "conf/server.yml" in paths assert "config.yaml" in paths def test_handles_read_errors_gracefully(self, tmp_path): @@ -418,3 +612,238 @@ def test_no_matches_preserves_message(self): from vuln_analysis.utils.source_classification import format_app_dep_output result = format_app_dep_output([], [], 0, 0, "No configuration entries found matching: xstream") assert result == "No configuration entries found matching: xstream" + + +class TestArun: + """Integration tests for _arun (the inner function yielded by configuration_scanner). + + Tests keyword parsing, LRU caching, and source scope filtering that _arun + adds on top of _collect_config_files and search_config_content. + """ + + @pytest_asyncio.fixture() + async def arun_fn(self): + """Yield the _arun function extracted from the configuration_scanner async generator.""" + from aiq.builder.builder import Builder + config = ConfigurationScannerToolConfig() + builder = MagicMock(spec=Builder) + cm = configuration_scanner(config, builder) + fi = await cm.__aenter__() + yield fi.single_fn + # Cleanup: exit the async context manager + try: + await cm.__aexit__(None, None, None) + except (StopAsyncIteration, GeneratorExit): + pass + + def _set_context(self, tmp_path, source_infos=None): + """Set ctx_state and cu_source_scope context vars for _arun.""" + from vuln_analysis.runtime_context import ctx_state, cu_source_scope + + if source_infos is None: + source_infos = [SimpleNamespace(git_repo="https://test.com/repo", ref="main")] + + image = SimpleNamespace(source_info=source_infos) + input_obj = SimpleNamespace(image=image) + original_input = SimpleNamespace(input=input_obj) + state = SimpleNamespace(original_input=original_input) + + ctx_state.set(state) + cu_source_scope.set(None) + + @pytest.mark.asyncio + async def test_keyword_parsing_splits_on_commas_and_spaces(self, arun_fn, tmp_path): + """_arun splits the query on commas and whitespace, lowercases, and drops tokens < 2 chars.""" + (tmp_path / "config.yaml").write_text("xstream_enabled: true\nssl_mode: require") + self._set_context(tmp_path) + + with patch("vuln_analysis.tools.configuration_scanner.get_repo_path_with_ref", return_value=tmp_path): + result = await arun_fn("xstream, ssl") + + assert "xstream_enabled" in result + assert "ssl_mode" in result + + @pytest.mark.asyncio + async def test_keyword_parsing_drops_short_tokens(self, arun_fn, tmp_path): + """Tokens shorter than 2 characters are filtered out by _arun's keyword parsing.""" + (tmp_path / "config.yaml").write_text("a: short\nxstream: enabled") + self._set_context(tmp_path) + + with patch("vuln_analysis.tools.configuration_scanner.get_repo_path_with_ref", return_value=tmp_path): + # When ONLY short tokens are provided, nothing should match + result_short_only = await arun_fn("a x") + assert "No configuration entries found" in result_short_only + + # When a valid keyword is included, it should match + result_with_valid = await arun_fn("a xstream") + assert "xstream" in result_with_valid + + @pytest.mark.asyncio + async def test_lru_cache_reuses_collected_files(self, arun_fn, tmp_path): + """Repeated queries for the same repo reuse cached config files.""" + (tmp_path / "config.yaml").write_text("xstream: enabled\nssl: on") + self._set_context(tmp_path) + + with patch("vuln_analysis.tools.configuration_scanner.get_repo_path_with_ref", return_value=tmp_path), \ + patch("vuln_analysis.tools.configuration_scanner._collect_config_files", wraps=_collect_config_files) as mock_collect: + await arun_fn("xstream") + await arun_fn("ssl") + + # _collect_config_files should only be called once because the second + # call uses the LRU cache populated by the first call + assert mock_collect.call_count == 1 + + @pytest.mark.asyncio + async def test_source_info_without_git_repo_skipped(self, arun_fn, tmp_path): + """Source infos that lack a git_repo attribute are silently skipped.""" + (tmp_path / "config.yaml").write_text("keyword: value") + # One source_info with git_repo, one without + si_with = SimpleNamespace(git_repo="https://test.com/repo", ref="main") + si_without = SimpleNamespace(ref="main") # no git_repo attribute + self._set_context(tmp_path, source_infos=[si_without, si_with]) + + with patch("vuln_analysis.tools.configuration_scanner.get_repo_path_with_ref", return_value=tmp_path): + result = await arun_fn("keyword") + + assert "keyword: value" in result + + @pytest.mark.asyncio + async def test_nonexistent_repo_path_skipped(self, arun_fn, tmp_path): + """When get_repo_path_with_ref returns a path that doesn't exist, the source is skipped.""" + self._set_context(tmp_path) + nonexistent = tmp_path / "does_not_exist" + + with patch("vuln_analysis.tools.configuration_scanner.get_repo_path_with_ref", return_value=nonexistent): + result = await arun_fn("keyword") + + assert "No configuration entries found" in result + + @pytest.mark.asyncio + async def test_source_label_uses_git_repo(self, arun_fn, tmp_path): + """The source label in output uses the first source_info's git_repo.""" + (tmp_path / "config.yaml").write_text("ssl: enabled") + self._set_context(tmp_path, source_infos=[ + SimpleNamespace(git_repo="https://github.com/org/myrepo", ref="v1.0"), + ]) + + with patch("vuln_analysis.tools.configuration_scanner.get_repo_path_with_ref", return_value=tmp_path): + result = await arun_fn("ssl") + + assert "source: https://github.com/org/myrepo" in result + + @pytest.mark.asyncio + async def test_app_dep_classification(self, arun_fn, tmp_path): + """Config files under dependencies-sources/ are classified as dependency configs.""" + # App-level config + (tmp_path / "config.yaml").write_text("xstream: app-level") + # Dependency-level config + dep_dir = tmp_path / "dependencies-sources" / "xstream-1.4" / "config" + dep_dir.mkdir(parents=True) + (dep_dir / "settings.yml").write_text("xstream: dep-level") + + self._set_context(tmp_path) + + with patch("vuln_analysis.tools.configuration_scanner.get_repo_path_with_ref", return_value=tmp_path): + result = await arun_fn("xstream") + + assert "Main application" in result + assert "Application library dependencies" in result + + @pytest.mark.asyncio + async def test_cache_eviction_when_exceeding_max_size(self, arun_fn, tmp_path): + """When the cache exceeds _CONFIG_CACHE_MAX_SIZE, the oldest entry + is evicted (LRU popitem(last=False)).""" + from vuln_analysis.runtime_context import ctx_state, cu_source_scope + + # Create 21 distinct "repos" so the 21st insert triggers eviction + # (the cache limit is 20 inside the closure) + repo_dirs = [] + for i in range(21): + d = tmp_path / f"repo_{i}" + d.mkdir() + (d / "config.yaml").write_text(f"keyword_{i}: value") + repo_dirs.append(d) + + # Build source_infos for each unique repo key + source_infos_list = [ + SimpleNamespace(git_repo=f"https://test.com/repo_{i}", ref="main") + for i in range(21) + ] + + with patch("vuln_analysis.tools.configuration_scanner.get_repo_path_with_ref") as mock_path, \ + patch("vuln_analysis.tools.configuration_scanner._collect_config_files", + wraps=_collect_config_files) as mock_collect: + + for i in range(21): + mock_path.return_value = repo_dirs[i] + + image = SimpleNamespace(source_info=[source_infos_list[i]]) + input_obj = SimpleNamespace(image=image) + original_input = SimpleNamespace(input=input_obj) + state = SimpleNamespace(original_input=original_input) + ctx_state.set(state) + cu_source_scope.set(None) + + await arun_fn(f"keyword_{i}") + + # 21 unique repos → 21 calls to _collect_config_files + assert mock_collect.call_count == 21 + + # Re-query the first repo — it was evicted so _collect must be called again + mock_path.return_value = repo_dirs[0] + image = SimpleNamespace(source_info=[source_infos_list[0]]) + input_obj = SimpleNamespace(image=image) + original_input = SimpleNamespace(input=input_obj) + state = SimpleNamespace(original_input=original_input) + ctx_state.set(state) + cu_source_scope.set(None) + + await arun_fn("keyword_0") + assert mock_collect.call_count == 22 + + @pytest.mark.asyncio + async def test_concurrent_access_same_repo(self, arun_fn, tmp_path): + """Multiple concurrent _arun calls for the same repo should not + call _collect_config_files more than once (the lock serialises them + and the second caller finds the cache populated).""" + (tmp_path / "config.yaml").write_text("ssl: enabled") + self._set_context(tmp_path) + + with patch("vuln_analysis.tools.configuration_scanner.get_repo_path_with_ref", return_value=tmp_path), \ + patch("vuln_analysis.tools.configuration_scanner._collect_config_files", + wraps=_collect_config_files) as mock_collect: + + results = await asyncio.gather( + arun_fn("ssl"), + arun_fn("ssl"), + arun_fn("ssl"), + ) + + assert mock_collect.call_count == 1 + for r in results: + assert "ssl" in r + + +class TestCacheEvictionSafety: + """Verify .get() prevents KeyError when concurrent eviction removes a cache entry.""" + + def test_get_returns_empty_on_missing_key(self): + """After eviction, .get(key, []) returns empty list instead of KeyError.""" + from collections import OrderedDict + cache = OrderedDict() + cache["repo_a"] = [("config.yml", "content")] + cache["repo_b"] = [("app.conf", "content")] + cache.popitem(last=False) + assert cache.get("repo_a", []) == [], "Evicted key should return default" + assert cache.get("repo_b", []) == [("app.conf", "content")] + + def test_module_cache_uses_get(self): + """The production code at the cache read site uses .get() for safety.""" + import ast + import inspect + from vuln_analysis.tools import configuration_scanner + source = inspect.getsource(configuration_scanner.configuration_scanner) + tree = ast.parse(source) + source_text = inspect.getsource(configuration_scanner.configuration_scanner) + assert "_config_files_cache.get(" in source_text, \ + "Cache read must use .get() to handle concurrent eviction safely" \ No newline at end of file