diff --git a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py index 3d9f8382..f1e435fc 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py +++ b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py @@ -486,6 +486,79 @@ def _extract_packages_from_pyproject(content: str) -> list[tuple[str, str | None return results +_LOCKFILE_PACKAGE_BLOCK_RE = re.compile( + r"(?ms)^\s*\[\[package\]\]\s*$.*?(?=^\s*\[\[package\]\]\s*$|\Z)" +) + + +def _normalize_package_name(name: str) -> str: + """Normalize package names the same way OSV/fallback coverage does.""" + return name.lower().replace("_", "-") + + +def _is_python_lockfile(file_path: str) -> bool: + lower_path = file_path.lower() + return "uv.lock" in lower_path or "poetry.lock" in lower_path + + +def _extract_packages_from_toml_lock(content: str) -> list[tuple[str, str | None, int]]: + """Extract exact package versions from TOML lockfiles such as uv.lock and poetry.lock.""" + try: + data = tomllib.loads(content) + except tomllib.TOMLDecodeError: + return [] + packages = data.get("package") + if not isinstance(packages, list): + return [] + blocks = list(_LOCKFILE_PACKAGE_BLOCK_RE.finditer(content)) + results: list[tuple[str, str | None, int]] = [] + for package, block in zip(packages, blocks, strict=False): + if not isinstance(package, dict): + continue + name = package.get("name") + version = package.get("version") + if not isinstance(name, str) or not name.strip(): + continue + version_value = version.strip() if isinstance(version, str) and version.strip() else None + name_match = re.search(r"(?m)^\s*name\s*=", block.group(0)) + idx = block.start() + name_match.start() if name_match else block.start() + line_num = get_line_number(content, idx) + results.append((name, version_value, line_num)) + return results + + +def _apply_locked_versions( + packages: list[tuple[str, str | None, int]], + locked_versions: dict[str, str] | None, +) -> list[tuple[str, str | None, int]]: + """Prefer lockfile versions for manifest dependencies without exact versions.""" + if not locked_versions: + return packages + resolved: list[tuple[str, str | None, int]] = [] + for name, version, line_num in packages: + locked_version = locked_versions.get(_normalize_package_name(name)) + resolved.append((name, version or locked_version, line_num)) + return resolved + + +def _collect_locked_versions( + file_cache: dict[str, str], + components: list[str], +) -> dict[str, str]: + """Build package -> exact version map from Python lockfiles in the project.""" + locked_versions: dict[str, str] = {} + for path in components: + if not _is_python_lockfile(path): + continue + content = file_cache.get(path) + if not content: + continue + for name, version, _line_num in _extract_packages_from_toml_lock(content): + if version: + locked_versions[_normalize_package_name(name)] = version + return locked_versions + + def _version_lt(v1: str, v2: str) -> bool: """Simple version comparison: True if v1 < v2 (numeric tuple comparison).""" @@ -754,14 +827,17 @@ def _sc4_from_fallback( def _analyze_dependencies( content: str, file_path: str, + locked_versions: dict[str, str] | None = None, ) -> list[AnalyzerFinding]: """Run SC4/SC5/SC6 checks on dependency files.""" findings: list[AnalyzerFinding] = [] tag = [PatternCategory.SUPPLY_CHAIN.value] lower_path = file_path.lower() - is_python_dep = any( - n in lower_path for n in ["requirements", "pyproject.toml", "setup.py", "pipfile"] + is_lockfile = _is_python_lockfile(lower_path) + is_python_dep = ( + any(n in lower_path for n in ["requirements", "pyproject.toml", "setup.py", "pipfile"]) + or is_lockfile ) is_npm_dep = "package.json" in lower_path @@ -771,8 +847,12 @@ def _analyze_dependencies( if is_python_dep: if "pyproject.toml" in lower_path: packages = _extract_packages_from_pyproject(content) + elif is_lockfile: + packages = _extract_packages_from_toml_lock(content) else: packages = _extract_packages_from_requirements(content) + if not is_lockfile: + packages = _apply_locked_versions(packages, locked_versions) ecosystem = ECOSYSTEM_PYPI fallback_db = _FALLBACK_VULNERABLE_PYPI popular = _POPULAR_PYPI @@ -959,18 +1039,27 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse: # SC4–SC6: dependency-level analysis on dependency files components: list[str] = state.get("components") or [] file_cache: dict[str, str] = state.get("file_cache") or {} + locked_versions = _collect_locked_versions(file_cache, components) for path in components: lower_path = path.lower() is_dep_file = any( n in lower_path - for n in ["requirements", "package.json", "pyproject.toml", "setup.py", "pipfile"] + for n in [ + "requirements", + "package.json", + "pyproject.toml", + "setup.py", + "pipfile", + "uv.lock", + "poetry.lock", + ] ) if not is_dep_file: continue content = file_cache.get(path) if not content: continue - dep_findings = _analyze_dependencies(content, path) + dep_findings = _analyze_dependencies(content, path, locked_versions) findings.extend(analyzer_finding_to_finding(af) for af in dep_findings) # TR1–TR3: trigger analysis from manifest diff --git a/tests/nodes/analyzers/test_static_patterns_supply_chain_lockfiles.py b/tests/nodes/analyzers/test_static_patterns_supply_chain_lockfiles.py new file mode 100644 index 00000000..36b242c3 --- /dev/null +++ b/tests/nodes/analyzers/test_static_patterns_supply_chain_lockfiles.py @@ -0,0 +1,113 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from skillspector.nodes.analyzers import static_patterns_supply_chain as supply_chain + + +def _capture_osv_packages(monkeypatch): + seen = {} + + def fake_query_batch(packages, ecosystem): + seen["packages"] = packages + seen["ecosystem"] = ecosystem + return [[] for _ in packages] + + monkeypatch.setattr(supply_chain, "query_batch", fake_query_batch) + return seen + + +def test_uv_lock_versions_are_passed_to_osv(monkeypatch): + seen = _capture_osv_packages(monkeypatch) + content = """ +version = 1 +[[package]] +name = "mlx" +version = "0.31.2" +[[package]] +name = "requests" +version = "2.31.0" +""" + supply_chain._analyze_dependencies(content, "uv.lock") + assert seen["ecosystem"] == supply_chain.ECOSYSTEM_PYPI + assert ("mlx", "0.31.2") in seen["packages"] + assert ("requests", "2.31.0") in seen["packages"] + + +def test_poetry_lock_versions_are_passed_to_osv(monkeypatch): + seen = _capture_osv_packages(monkeypatch) + content = """ +[[package]] +name = "jinja2" +version = "3.1.6" +description = "A fast template engine." +""" + supply_chain._analyze_dependencies(content, "poetry.lock") + assert seen["ecosystem"] == supply_chain.ECOSYSTEM_PYPI + assert ("jinja2", "3.1.6") in seen["packages"] + + +def test_pyproject_unpinned_dependency_uses_locked_version_for_osv(monkeypatch): + seen = _capture_osv_packages(monkeypatch) + content = """ +[project] +dependencies = [ + "mlx", +] +""" + supply_chain._analyze_dependencies(content, "pyproject.toml", {"mlx": "0.31.2"}) + assert ("mlx", "0.31.2") in seen["packages"] + + +def test_requirements_unpinned_dependency_uses_locked_version_for_osv(monkeypatch): + seen = _capture_osv_packages(monkeypatch) + content = """ +fastmcp +""" + supply_chain._analyze_dependencies(content, "requirements.txt", {"fastmcp": "3.3.1"}) + assert ("fastmcp", "3.3.1") in seen["packages"] + + +def test_toml_lock_parser_anchors_line_numbers_to_package_blocks(): + content = """ +[[package]] +name = "root" +version = "1.0.0" +dependencies = [ + { name = "requests" }, +] +[[package]] +name = "requests" +version = "2.31.0" +""" + packages = supply_chain._extract_packages_from_toml_lock(content) + line_by_name = {name: line_num for name, _version, line_num in packages} + assert content.splitlines()[line_by_name["requests"] - 1].strip() == 'name = "requests"' + + +def test_toml_lock_parser_returns_empty_for_malformed_toml(): + content = """ +[[package] +name = "broken" +""" + assert supply_chain._extract_packages_from_toml_lock(content) == [] + + +def test_toml_lock_parser_keeps_package_without_version(): + content = """ +[[package]] +name = "local-package" +""" + packages = supply_chain._extract_packages_from_toml_lock(content) + assert packages[0][:2] == ("local-package", None)