from __future__ import annotations from dataclasses import dataclass, field from vuln_scan.core.models import ( Dependency, Ecosystem, EcosystemPackage, ManifestPackage, ManifestPackageVersion, ) @dataclass class DependencyGraph: """ Represents a collection of dependencies discovered in a workspace or repository branch. Tracks: - All dependencies - Occurrences by ecosystem/manifest/package - Dependencies grouped by manifest """ dependencies: list[Dependency] = field(default_factory=list) occurrences: dict[ tuple[Ecosystem, str, str], list[Dependency], ] = field(default_factory=dict) dependencies_by_manifest: dict[ str, list[Dependency], ] = field(default_factory=dict) # ------------------------------------------------------------------ # Core mutation methods # ------------------------------------------------------------------ def add_dependency(self, manifest_path: str, dep: Dependency) -> None: """Add a dependency occurrence to the graph.""" self.dependencies.append(dep) self.dependencies_by_manifest.setdefault( manifest_path, [], ).append(dep) pkg_key = dep.normalized_name occ_key = ( dep.ecosystem, manifest_path, pkg_key, ) self.occurrences.setdefault( occ_key, [], ).append(dep) # ------------------------------------------------------------------ # Query helpers # ------------------------------------------------------------------ def get_dependencies( self, ecosystem: Ecosystem, package_name: str, manifest_path: str, ) -> list[Dependency]: """Return all dependency occurrences for a given package within a specific manifest.""" pkg_key = package_name.lower() occ_key = (ecosystem, manifest_path, pkg_key) return self.occurrences.get(occ_key, []) def get_version( self, ecosystem: Ecosystem, package_name: str, manifest_path: str, ) -> str | None: """Return the first detected version for a package in a manifest.""" deps = self.get_dependencies(ecosystem, package_name, manifest_path) if not deps: return None return deps[0].version def get_versions( self, ecosystem: Ecosystem, package_name: str, manifest_path: str, ) -> list[str]: """Return all detected versions for a package in a manifest.""" deps = self.get_dependencies(ecosystem, package_name, manifest_path) return [d.version for d in deps if d.version] def get_manifests(self) -> set[str]: """Return all manifests discovered in the graph.""" return set(self.dependencies_by_manifest.keys()) def get_dependencies_for_manifest( self, manifest_path: str, ) -> list[Dependency]: """Return dependencies declared in a specific manifest.""" return self.dependencies_by_manifest.get(manifest_path, []) # ------------------------------------------------------------------ # Package-level helpers # ------------------------------------------------------------------ def packages(self) -> set[EcosystemPackage]: """Return all unique (ecosystem, package) pairs.""" return {(dep.ecosystem, dep.normalized_name) for dep in self.dependencies} def package_versions(self) -> dict[EcosystemPackage, set[str]]: """Map each package to all its detected versions.""" result: dict[EcosystemPackage, set[str]] = {} for dep in self.dependencies: key = (dep.ecosystem, dep.normalized_name) if dep.version: result.setdefault(key, set()).add(dep.version) return result # ------------------------------------------------------------------ # Graph comparison helpers # ------------------------------------------------------------------ def to_manifest_package_version_map(self) -> dict[ManifestPackageVersion, str | None]: """ Map each (ecosystem, package, manifest, version) to its best detected version. Tracks every installed version separately so npm-style multi-version lockfiles are handled correctly. Priority (highest to lowest): 1. Lockfile dependency with an exact version 2. Lockfile dependency with a requirement 3. Non-lockfile dependency with an exact version 4. Non-lockfile dependency with a requirement """ result: dict[ManifestPackageVersion, tuple[int, str | None]] = {} for dep in self.dependencies: resolved = dep.version or dep.requirement key: ManifestPackageVersion = ( dep.ecosystem, dep.normalized_name, dep.manifest_path, dep.version, ) score = (2 if dep.is_lockfile else 0) + (1 if dep.version else 0) existing = result.get(key) if existing is None or score > existing[0]: result[key] = (score, resolved) return {k: v for k, (_, v) in result.items()} def to_manifest_package_map(self) -> dict[ManifestPackage, str | None]: """ Map each (ecosystem, package, manifest) to its best detected version. For single-version ecosystems (Python) this works as before. For multi-version ecosystems (npm) use to_manifest_package_version_map() instead. """ result: dict[ManifestPackage, tuple[int, str | None]] = {} for dep in self.dependencies: key = (dep.ecosystem, dep.normalized_name, dep.manifest_path) version = dep.version or dep.requirement score = (2 if dep.is_lockfile else 0) + (1 if dep.version else 0) existing = result.get(key) if existing is None or score > existing[0]: result[key] = (score, version) return {k: v for k, (_, v) in result.items()} def introduced_packages( self, other: DependencyGraph, ) -> set[EcosystemPackage]: """ Packages/versions present in this graph but not in the other. Compared per-manifest and per-version to handle multi-version lockfiles. """ current = self.to_manifest_package_version_map() base = other.to_manifest_package_version_map() introduced_keys = set(current.keys()) - set(base.keys()) return {(eco, pkg) for eco, pkg, _, _ in introduced_keys} def introduced_versioned( self, other: DependencyGraph, ) -> set[tuple[Ecosystem, str, str | None]]: """ Packages+versions present in this graph but not in the other. Returns (ecosystem, normalized_name, version) triples so callers can distinguish a newly added version of an existing package from a brand-new package entirely. """ current = self.to_manifest_package_version_map() base = other.to_manifest_package_version_map() introduced_keys = set(current.keys()) - set(base.keys()) # (ecosystem, pkg, manifest, version) → drop manifest, keep version return {(eco, pkg, version) for eco, pkg, _, version in introduced_keys} def removed_packages( self, other: DependencyGraph, ) -> set[EcosystemPackage]: """ Packages/versions present in the other graph but not in this one. """ current = self.to_manifest_package_version_map() base = other.to_manifest_package_version_map() removed_keys = set(base.keys()) - set(current.keys()) return {(eco, pkg) for eco, pkg, _, _ in removed_keys} def changed_versions( self, other: DependencyGraph, ) -> dict[EcosystemPackage, tuple[str | None, str | None]]: """ Detect packages whose versions differ between graphs. Compared per-manifest and per-version for deterministic results. For multi-version packages, version additions/removals are reported via introduced_packages() / removed_packages() instead. """ current = self.to_manifest_package_map() base = other.to_manifest_package_map() changed: dict[EcosystemPackage, tuple[str | None, str | None]] = {} for key in current.keys() & base.keys(): if current[key] != base[key]: eco, pkg, _ = key if (eco, pkg) not in changed: changed[(eco, pkg)] = (base[key], current[key]) return changed # ------------------------------------------------------------------ # Stats helpers # ------------------------------------------------------------------ def dependency_count(self) -> int: return len(self.dependencies) def manifest_count(self) -> int: return len(self.dependencies_by_manifest) def package_count(self) -> int: return len(self.packages()) def get_manifest_paths(self) -> set[str]: return set(self.dependencies_by_manifest.keys())