from __future__ import annotations import logging import time from collections import defaultdict from vuln_scan.core.dependency_graph import DependencyGraph from vuln_scan.core.models import ( Dependencies, Ecosystem, EcosystemPackage, Finding, Findings, FindingStatus, IntroducedVersioned, PolicyDecision, ScanConfig, ScanResult, SecurityVulnerability, ) from vuln_scan.core.policy import BlockingPolicy from vuln_scan.core.registry import EcosystemRegistry from vuln_scan.ecosystems.base import EcosystemHandler from vuln_scan.github.advisory_client import GitHubAdvisoryClient from vuln_scan.github.publisher import StatusPublisher from vuln_scan.github.repo_alerts_client import GithubRepoAlertsClient from vuln_scan.io.default_branch_scanner import DefaultBranchScanner from vuln_scan.io.workspace_scanner import WorkspaceScanner from vuln_scan.reporting.base import Reporter logger = logging.getLogger(__name__) class VulnerabilityScanner: """ Main orchestration class for vulnerability scanning. Responsibilities: - Orchestrate scanning workflow - Match dependencies with alerts - Delegate vulnerability logic to ecosystem handlers - Apply policy - Trigger reporting + publishing """ def __init__( self, *, workspace_scanner: WorkspaceScanner, default_branch_scanner: DefaultBranchScanner, alerts_client: GithubRepoAlertsClient, advisory_client: GitHubAdvisoryClient, registry: EcosystemRegistry, policy: BlockingPolicy, reporter: Reporter, publisher: StatusPublisher, ): self._workspace_scanner = workspace_scanner self._default_branch_scanner = default_branch_scanner self._alerts_client = alerts_client self._advisory_client = advisory_client self._registry = registry self._policy = policy self._reporter = reporter self._publisher = publisher # ----------------------------------------------------- def scan(self, cfg: ScanConfig) -> ScanResult: """ Execute full vulnerability scan. """ self._publisher.set_status("pending") if self._should_skip(cfg): return self._handle_skip() workspace_graph, default_graph = self._scan_graphs() removed, changed_packages, introduced_versioned = self._compute_deltas( workspace_graph, default_graph, ) alerts = self._fetch_alerts(workspace_graph) findings = [] findings.extend( self._evaluate_alerts(workspace_graph, alerts, removed, introduced_versioned) ) findings.extend( self._evaluate_advisories(workspace_graph, changed_packages, introduced_versioned) ) findings = self._filter_and_sort_findings(findings) output = self._report(findings, cfg) self._publish(findings, output, cfg.dependabot_url) return ScanResult(findings=findings) def _should_skip(self, cfg: ScanConfig) -> bool: return bool(cfg.skip_phrase and cfg.skip_phrase in (cfg.pr_comment_text or "")) def _handle_skip(self) -> ScanResult: msg = "Skipping vulnerability scan due to skip phrase." self._publisher.set_status("success") logger.info(msg) return ScanResult(findings=[]) def _scan_graphs(self) -> tuple[DependencyGraph, DependencyGraph]: workspace_graph = self._workspace_scanner.scan(self._registry) default_graph = self._default_branch_scanner.scan( self._registry, manifest_paths=workspace_graph.get_manifest_paths(), ) return workspace_graph, default_graph def _compute_deltas( self, workspace_graph: DependencyGraph, default_graph: DependencyGraph, ) -> tuple[ set[EcosystemPackage], # removed set[EcosystemPackage], # changed_packages (for advisory fetch) IntroducedVersioned, # introduced (version-level, for policy) ]: changed = workspace_graph.changed_versions(default_graph) removed = workspace_graph.removed_packages(default_graph) introduced_versioned = workspace_graph.introduced_versioned(default_graph) introduced_packages = {(eco, pkg) for eco, pkg, _ in introduced_versioned} changed_packages = introduced_packages | set(changed.keys()) return removed, changed_packages, introduced_versioned def _fetch_alerts(self, workspace_graph: DependencyGraph) -> list[SecurityVulnerability]: return self._alerts_client.get_open_alerts( manifest_paths=workspace_graph.get_manifest_paths() ) def _evaluate_alerts( self, workspace_graph: DependencyGraph, alerts: list[SecurityVulnerability], removed: set[EcosystemPackage], introduced_versioned: IntroducedVersioned, ) -> Findings: findings: list[Finding] = [] for alert in alerts: handler = self._registry.for_ecosystem(alert.ecosystem) if not handler: logger.warning("No handler for ecosystem: %s", alert.ecosystem) continue matches = workspace_graph.get_dependencies( alert.ecosystem, alert.package_name, alert.vulnerable_manifest_path, ) if not matches: findings.append(self._handle_no_match(alert, removed)) continue findings.extend(self._evaluate_matches(matches, alert, handler, introduced_versioned)) return findings def _handle_no_match( self, alert: SecurityVulnerability, removed: set[EcosystemPackage], ) -> Finding: pkg_key = (alert.ecosystem, (alert.package_name or "").lower()) is_removed = pkg_key in removed decision = self._policy.evaluate( alert=alert, is_vulnerable=False, has_patch=bool(alert.first_patched_version), is_removed=is_removed, ) return Finding(None, alert, decision) def _evaluate_matches( self, matches: Dependencies, alert: SecurityVulnerability, handler: EcosystemHandler, introduced_versioned: IntroducedVersioned | None = None, ) -> Findings: results = [] introduced_versioned = introduced_versioned or set() deps = self._prefer_lockfiles(matches) for dep in deps: is_vuln, _ = handler.is_vulnerable(dep, alert) is_introduced = ( dep.ecosystem, dep.normalized_name, dep.version, ) in introduced_versioned decision = self._policy.evaluate( alert=alert, is_vulnerable=is_vuln, has_patch=bool(alert.first_patched_version), is_introduced=is_introduced, ) results.append(Finding(dep, alert, decision)) return results def _prefer_lockfiles(self, deps: Dependencies) -> Dependencies: lockfile_deps = [d for d in deps if d.is_lockfile] return lockfile_deps if lockfile_deps else deps def _evaluate_advisories( self, workspace_graph: DependencyGraph, changed_packages: set[EcosystemPackage], introduced_versioned: IntroducedVersioned, ) -> Findings: findings: list[Finding] = [] start_ns = time.perf_counter_ns() logger.info("Evaluating advisories for %d changed packages", len(changed_packages)) if not changed_packages: return findings deps_by_package = self._index_dependencies_by_package(workspace_graph) work_items = self._build_advisory_work_items(changed_packages, deps_by_package) if not work_items: logger.info("Advisories: no eligible packages to fetch advisories for.") return findings advisories_by_package = self._fetch_advisories_for_work_items(work_items) findings = self._evaluate_advisory_work_items( work_items=work_items, advisories_by_package=advisories_by_package, introduced_versioned=introduced_versioned, ) elapsed_ms = (time.perf_counter_ns() - start_ns) / 1_000_000 logger.info( "Done evaluating advisories: %d finding(s), %.2fms elapsed", len(findings), elapsed_ms, ) return findings def _index_dependencies_by_package( self, workspace_graph: DependencyGraph, ) -> dict[tuple[Ecosystem, str], Dependencies]: start_ns = time.perf_counter_ns() deps_by_pkg: dict[tuple[Ecosystem, str], Dependencies] = defaultdict(list) for dep in workspace_graph.dependencies: deps_by_pkg[(dep.ecosystem, dep.normalized_name)].append(dep) elapsed_ms = (time.perf_counter_ns() - start_ns) / 1_000_000 logger.info( "Advisories: indexed %d dependency entries in %.2fms", len(workspace_graph.dependencies), elapsed_ms, ) return deps_by_pkg def _build_advisory_work_items( self, changed_packages: set[EcosystemPackage], deps_by_package: dict[tuple[Ecosystem, str], Dependencies], ) -> list[tuple[Ecosystem, str, EcosystemHandler, Dependencies]]: work_items: list[tuple[Ecosystem, str, EcosystemHandler, Dependencies]] = [] for ecosystem, package_name in sorted(changed_packages): deps = deps_by_package.get((ecosystem, package_name)) if not deps: continue handler = self._registry.for_ecosystem(ecosystem) if not handler: logger.warning("No handler for ecosystem: %s", ecosystem) continue work_items.append((ecosystem, package_name, handler, deps)) return work_items def _fetch_advisories_for_work_items( self, work_items: list[tuple[Ecosystem, str, EcosystemHandler, Dependencies]], ) -> dict[tuple[Ecosystem, str], list[SecurityVulnerability]]: start_ns = time.perf_counter_ns() batch_packages = [ (ecosystem.value, package_name) for ecosystem, package_name, _, _ in work_items ] advisories_by_raw = self._advisory_client.get_advisories_batch(batch_packages) advisories_by_package = { (ecosystem, package_name): advisories_by_raw.get((ecosystem.value, package_name), []) for ecosystem, package_name, _, _ in work_items } elapsed_ms = (time.perf_counter_ns() - start_ns) / 1_000_000 logger.info( "Advisories: batch-fetched for %d packages in %.2fms", len(work_items), elapsed_ms, ) return advisories_by_package def _evaluate_advisory_work_items( self, *, work_items: list[tuple[Ecosystem, str, EcosystemHandler, Dependencies]], advisories_by_package: dict[tuple[Ecosystem, str], list[SecurityVulnerability]], introduced_versioned: IntroducedVersioned, ) -> Findings: start_ns = time.perf_counter_ns() findings: list[Finding] = [] for ecosystem, package_name, handler, deps in work_items: advisories = advisories_by_package.get((ecosystem, package_name), []) if not advisories: continue for advisory in advisories: findings.extend( self._evaluate_matches( matches=deps, alert=advisory, handler=handler, introduced_versioned=introduced_versioned, ) ) elapsed_ms = (time.perf_counter_ns() - start_ns) / 1_000_000 logger.info("Advisories: evaluated matches in %.2fms", elapsed_ms) return findings def _filter_and_sort_findings(self, findings: Findings) -> Findings: # Keep only vulnerable findings vulnerable = [f for f in findings if f.decision.status == FindingStatus.VULNERABLE] # Deduplicate: one finding per (package, version, ghsa_id, manifest) unique = self._deduplicate(vulnerable) # Collapse per (package, version, manifest): # - keep the finding with the highest patch version # - this tells the developer "upgrade to X to fix everything" # - preserve the most restrictive blocking + highest severity collapsed = self._collapse_groups(unique) # Sort: blocking first, severity descending, then package + version return self._sort_findings(collapsed) @staticmethod def _installed_version(f: Finding) -> str: if f.dependency: return f.dependency.version or f.dependency.requirement or "" return "" @staticmethod def _manifest_path(f: Finding) -> str: if f.dependency: return f.dependency.manifest_path or "" return f.alert.vulnerable_manifest_path or "" def _deduplicate(self, findings: list[Finding]) -> list[Finding]: """One finding per (package, version, ghsa_id, manifest).""" seen: set[tuple[str, str, str, str]] = set() unique: list[Finding] = [] for f in findings: key = ( f.package_name, self._installed_version(f), f.alert.ghsa_id, self._manifest_path(f), ) if key not in seen: seen.add(key) unique.append(f) return unique def _collapse_groups(self, findings: list[Finding]) -> list[Finding]: """Collapse per (package, version, manifest): highest patch, worst severity, most restrictive blocking.""" grouped: dict[tuple[str, str, str], list[Finding]] = {} for f in findings: key = ( f.package_name, self._installed_version(f), self._manifest_path(f), ) grouped.setdefault(key, []).append(f) return [self._pick_representative(group) for group in grouped.values()] def _pick_representative(self, group: list[Finding]) -> Finding: best = max(group, key=lambda f: self._parse_patch_version(f.alert.first_patched_version)) worst = max(group, key=lambda f: f.alert.severity.rank) blocking = [f for f in group if f.decision.is_blocking] any_blocking = bool(blocking) needs_elevation = ( best.alert.severity != worst.alert.severity or best.decision.is_blocking != any_blocking ) if not needs_elevation: return best # Decision fields (blocking_date, remaining_days, reason, ...) must all come # from the *same* finding to stay internally consistent. If blocking needs to # flip to True, source them from the finding that is actually blocking rather # than mixing them with "best"'s (not-yet-blocking) fields. decision_source = ( min( blocking, key=lambda f: (f.decision.blocking_date is None, f.decision.blocking_date) ) if any_blocking and not best.decision.is_blocking else best ) return Finding( dependency=best.dependency, alert=SecurityVulnerability( id=best.alert.id, ghsa_id=best.alert.ghsa_id, package_name=best.alert.package_name, ecosystem=best.alert.ecosystem, vulnerable_manifest_path=best.alert.vulnerable_manifest_path, vulnerable_range=best.alert.vulnerable_range, first_patched_version=best.alert.first_patched_version, severity=worst.alert.severity, created_at=best.alert.created_at, source=best.alert.source, ), decision=PolicyDecision( blocking_date=decision_source.decision.blocking_date, remaining_days=decision_source.decision.remaining_days, is_blocking=decision_source.decision.is_blocking, status=best.decision.status, message=decision_source.decision.message, reason=decision_source.decision.reason, ), ) @staticmethod def _sort_findings(findings: list[Finding]) -> Findings: return sorted( findings, key=lambda f: ( not f.decision.is_blocking, -f.alert.severity.rank, f.package_name, (f.dependency.version or f.dependency.requirement or "") if f.dependency else "", ), ) def _parse_patch_version(self, v: str | None) -> tuple[int, ...]: """Parse version string into a comparable tuple.""" if not v: return (0,) try: return tuple(int(x) for x in v.split(".")) except ValueError: return (0,) def _report(self, findings: Findings, cfg: ScanConfig) -> str: return self._reporter.render(findings, cfg.dependabot_url) def _publish(self, findings: Findings, output: str, dependabot_url: str) -> None: if not findings: self._publisher.set_status("success") return has_blocking = any(f.decision.is_blocking for f in findings) self._publisher.set_status("failure" if has_blocking else "success") self._publisher.comment(output, dependabot_url, has_blocking)