diff --git a/pyproject.toml b/pyproject.toml index 094aaba..2d19791 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -42,6 +42,7 @@ dependencies = [ "pyyaml==6.0.3", "pydantic==2.13.4", "orjson==3.11.9", + "packaging==24.0", ] [project.optional-dependencies] diff --git a/src/ciberwebscan/core/analyzers/cve/models.py b/src/ciberwebscan/core/analyzers/cve/models.py index 9a5e9d4..2e26f1d 100644 --- a/src/ciberwebscan/core/analyzers/cve/models.py +++ b/src/ciberwebscan/core/analyzers/cve/models.py @@ -8,11 +8,16 @@ from __future__ import annotations +import logging from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Any +from packaging.version import InvalidVersion, Version + +logger = logging.getLogger(__name__) + class CVESource(str, Enum): """Supported CVE data sources.""" @@ -94,9 +99,49 @@ class AffectedProduct: cpe: str = "" # CPE 2.3 identifier def matches_version(self, version: str) -> bool: - """Check if a specific version is affected.""" - # Simple version comparison (can be extended with packaging.version) - return bool(self.version_exact and self.version_exact == version) + """Check if a specific version is affected. + + Supports: + - Exact match (version_exact) + - Range match (version_start <= version < version_end) + - Wildcard (no version constraints = affects all versions) + """ + if not version: + return True + + if self.version_exact and self.version_exact == version: + return True + + if self.version_start or self.version_end: + return self._in_range(version) + + return ( + not self.version_exact and not self.version_start and not self.version_end + ) + + def _in_range(self, version: str) -> bool: + """Check if version falls within the affected range.""" + try: + v = Version(version) + except InvalidVersion: + logger.debug("Cannot parse version '%s', assuming affected", version) + return True + + if self.version_start: + try: + if v < Version(self.version_start): + return False + except InvalidVersion: + pass + + if self.version_end: + try: + if v >= Version(self.version_end): + return False + except InvalidVersion: + pass + + return True @dataclass diff --git a/src/ciberwebscan/services/analyze_service.py b/src/ciberwebscan/services/analyze_service.py index 1a3b84f..c41ca2f 100644 --- a/src/ciberwebscan/services/analyze_service.py +++ b/src/ciberwebscan/services/analyze_service.py @@ -624,24 +624,35 @@ def _lookup_cves( technologies: list[TechnologyMatch], options: AnalyzeOptions, ) -> list[CVEResult]: - """Internal CVE lookup.""" + """Internal CVE lookup with version-aware filtering.""" try: all_cves: list[CVEResult] = [] + seen_ids: set[str] = set() for tech in technologies: - # Search CVEs for this technology + version = tech.version or "" + aggregated = self.cve_aggregator.search( product=tech.name, + version=version, limit=options.cve_limit, ) for cve in aggregated.entries: - # Filter by severity if specified + if cve.id in seen_ids: + continue + seen_ids.add(cve.id) + if options.cve_severity and ( str(cve.severity.value).lower() != options.cve_severity.lower() ): continue + if version and not self._cve_affects_version( + cve, tech.name, version + ): + continue + all_cves.append( CVEResult( id=cve.id, @@ -686,3 +697,18 @@ def _lookup_cves( except Exception as e: self.logger.warning(f"CVE lookup failed: {e}") return [] + + def _cve_affects_version(self, cve: Any, product_name: str, version: str) -> bool: + """Check if a CVE entry affects the specific detected version.""" + if not cve.affected_products: + return True + + product_lower = product_name.lower() + for affected in cve.affected_products: + affected_product = affected.product.lower() + if ( + product_lower in affected_product or affected_product in product_lower + ) and affected.matches_version(version): + return True + + return False