Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ dependencies = [
"pyyaml==6.0.3",
"pydantic==2.13.4",
"orjson==3.11.9",
"packaging==24.0",
]

[project.optional-dependencies]
Expand Down
51 changes: 48 additions & 3 deletions src/ciberwebscan/core/analyzers/cve/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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
Expand Down
32 changes: 29 additions & 3 deletions src/ciberwebscan/services/analyze_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Loading