Fix CVE version-range matching: numeric comparison + Excluding bounds

query_cves() (the primary/default correlation path once the SQLite CVE
snapshot is installed) compared version_start/version_end with plain
SQL >=/<=, which is a lexicographic string comparison, not numeric.
"2.4.9" sorts after "2.4.10", so ranges spanning a digit-width boundary
silently produced both false negatives (real vulnerabilities missed)
and false positives (patched versions flagged). Version filtering now
happens in Python via cve_matcher.version_in_range, which does real
semver comparison.

Both the SQLite ingestion path (_extract_cpe_matches_from_node) and the
JSON-fallback path (cve_matcher.extract_cve_info) also only read NVD's
versionStartIncluding/versionEndIncluding fields, ignoring
versionStartExcluding/versionEndExcluding entirely. Since NVD commonly
expresses "fixed in version X" as an Excluding bound, this silently
dropped the upper bound for many CVEs (every later version stayed
"vulnerable" forever) and mistreated Excluding boundaries as inclusive
elsewhere. Both extraction paths now capture and honor exclusivity.

version_in_range's fallback for versions packaging.version can't parse
(e.g. Debian/Ubuntu-suffixed versions like "2.4.41-1ubuntu1", which are
extremely common in real banners) used to unconditionally `return True`
for any bounded range, which is the opposite of what this module exists
to prevent. It now recovers the leading numeric version when possible
and otherwise declines to claim a match rather than flooding results.

cve_matcher.py had zero test coverage before this change.

Author: Ubuntu

bitprobe/scanner/cve_db_manager.py

@@ -13,7 +13,6 @@
 from contextlib import closing
 from datetime import datetime, timedelta, timezone
 from typing import List, Dict, Optional, Any
-from packaging import version
 from pathlib import Path
 from scanner.paths import CVE_DB_PATH, CVE_META_PATH, migrate_legacy_cve_database
 from scanner.update_lock import bitsentry_update_lock
@@ -1067,13 +1066,36 @@ def _extract_cpe_matches_from_node(node: Dict) -> List[Dict]:
         product = parts[4] if len(parts) > 4 else ''
         version_str = parts[5] if len(parts) > 5 else '*'
 
+        # Including and Excluding are mutually exclusive per NVD's schema;
+        # track which applied so an Excluding bound (that version is
+        # already patched) isn't treated as inclusive downstream.
+        version_start = match.get('versionStartIncluding')
+        version_start_including = 1
+        if version_start is None:
+            version_start = match.get('versionStartExcluding')
+            if version_start is not None:
+                version_start_including = 0
+
+        version_end = match.get('versionEndIncluding')
+        version_end_including = 1
+        if version_end is None:
+            version_end = match.get('versionEndExcluding')
+            if version_end is not None:
+                version_end_including = 0
+
+        if version_start is None and version_end is None and version_str != '*':
+            # Exact-version CPE entry (e.g. "...:1.18.0:*:*:..."): treat as
+            # an inclusive single-version range.
+            version_start = version_str
+            version_end = version_str
+
         products.append({
             'vendor': vendor.lower(),
             'product': product.lower(),
-            'version_start': match.get('versionStartIncluding', version_str if version_str != '*' else None),
-            'version_end': match.get('versionEndIncluding', version_str if version_str != '*' else None),
-            'version_start_including': 1 if match.get('versionStartIncluding') else 0,
-            'version_end_including': 1 if match.get('versionEndIncluding') else 0,
+            'version_start': version_start,
+            'version_end': version_end,
+            'version_start_including': version_start_including,
+            'version_end_including': version_end_including,
         })
 
     for child in node.get('children', []):
@@ -1235,6 +1257,12 @@ def query_cves(
     Uses PRODUCT_ALIASES from cve_matcher to resolve detected technology names
     to their known CPE product identifiers, avoiding false positives from
     substring matching (e.g., 'astro' no longer matches 'astrocam').
+
+    Version filtering is done in Python (not SQL): version_start/version_end
+    are dotted version strings, and comparing them with SQL's ">="/"<="
+    does a lexicographic string comparison, not a numeric one (e.g. the
+    string "2.4.9" sorts after "2.4.10"), which silently produces both
+    false positives and false negatives. See scanner.cve_matcher.version_in_range.
     
     Args:
         product: Product name (e.g., "nginx", "wordpress")
@@ -1247,7 +1275,7 @@ def query_cves(
     if not os.path.exists(CVE_DB_PATH):
         raise FileNotFoundError("CVE database not found. Run 'bitprobe update-cve-db' first.")
 
-    from scanner.cve_matcher import _get_cpe_names, _get_expected_vendor
+    from scanner.cve_matcher import _get_cpe_names, _get_expected_vendor, version_in_range
 
     cpe_names = _get_cpe_names(product)
     if not cpe_names:
@@ -1265,7 +1293,9 @@ def query_cves(
         query = f"""
             SELECT DISTINCT 
                 c.cve_id, c.description, c.severity, 
-                c.cvss_score, c."references", c.published_date
+                c.cvss_score, c."references", c.published_date,
+                p.version_start, p.version_end,
+                p.version_start_including, p.version_end_including
             FROM cve_entries c
             JOIN cve_products p ON c.cve_id = p.cve_id
             WHERE p.product IN ({placeholders})
@@ -1279,34 +1309,37 @@ def query_cves(
             query += " AND p.vendor = ?"
             params.append(expected_vendor)
 
-        # Version matching if provided
-        if version:
-            query += """
-                AND (
-                    (p.version_start IS NULL OR ? >= p.version_start)
-                    AND (p.version_end IS NULL OR ? <= p.version_end)
-                )
-            """
-            params.extend([version, version])
-        
         query += " ORDER BY c.cvss_score DESC NULLS LAST"
 
         cursor.execute(query, params)
         rows = cursor.fetchall()
 
-        cves = []
+        cves = {}
         for row in rows:
-            cve = {
-                'cve_id': row['cve_id'],
+            cve_id = row['cve_id']
+            if cve_id in cves:
+                # Already matched via a different product/version row for
+                # this CVE; each row is an independent vulnerable
+                # configuration, so one match is enough.
+                continue
+            if version and not version_in_range(
+                version,
+                row['version_start'],
+                row['version_end'],
+                min_inclusive=bool(row['version_start_including']),
+                max_inclusive=bool(row['version_end_including']),
+            ):
+                continue
+            cves[cve_id] = {
+                'cve_id': cve_id,
                 'description': row['description'],
                 'severity': row['severity'],
                 'cvss_score': row['cvss_score'],
                 'published_date': row['published_date'],
                 'references': json.loads(row['references'] or '[]')
             }
-            cves.append(cve)
 
-        return cves
+        return list(cves.values())
 
     finally:
         conn.close()

bitprobe/scanner/cve_matcher.py

@@ -126,38 +126,78 @@ def parse_version_range(cpe: Dict) -> Tuple[Optional[str], Optional[str]]:
     return (version, version)
 
 
-def version_in_range(detected: str, min_ver: Optional[str], max_ver: Optional[str]) -> bool:
-    """Check if detected version falls within the range."""
+def _coerce_version(value: str):
+    """
+    Parse a version string leniently.
+
+    Real-world banner/package versions are frequently not valid PEP 440
+    (e.g. Debian/Ubuntu suffixes like "2.4.41-1ubuntu1" or
+    "5.7.31-0ubuntu0.18.04.1"), which packaging.version.parse rejects
+    outright. Fall back to the leading dotted-numeric prefix so those
+    versions can still be compared; return None only if no numeric
+    version can be recovered at all.
+    """
+    try:
+        return pkg_version.parse(value)
+    except Exception:
+        pass
+    match = re.match(r"[0-9]+(?:\.[0-9]+)*", value)
+    if not match:
+        return None
+    try:
+        return pkg_version.parse(match.group(0))
+    except Exception:
+        return None
+
+
+def version_in_range(
+    detected: str,
+    min_ver: Optional[str],
+    max_ver: Optional[str],
+    min_inclusive: bool = True,
+    max_inclusive: bool = True,
+) -> bool:
+    """Check if detected version falls within [min_ver, max_ver].
+
+    min_inclusive/max_inclusive control whether each bound is inclusive
+    (versionStartIncluding/versionEndIncluding in NVD terms) or exclusive
+    (versionStartExcluding/versionEndExcluding).
+    """
     if not detected:
         # No version detected - can't determine vulnerability
         # Only match if CVE affects all versions (no version constraints)
         return min_ver is None and max_ver is None
-    
-    try:
-        detected_v = pkg_version.parse(detected)
-        
-        if min_ver and max_ver:
-            # Specific version or range
-            if min_ver == max_ver:
-                return detected_v == pkg_version.parse(min_ver)
-            return pkg_version.parse(min_ver) <= detected_v <= pkg_version.parse(max_ver)
-        
-        elif min_ver:
-            return detected_v >= pkg_version.parse(min_ver)
-        
-        elif max_ver:
-            return detected_v <= pkg_version.parse(max_ver)
-        
-        else:
-            # No version constraints - any version matches
-            return True
-            
-    except Exception:
-        # Fallback to string comparison
-        if min_ver and max_ver and min_ver == max_ver:
-            return detected == min_ver
+
+    if min_ver is None and max_ver is None:
         return True
 
+    detected_v = _coerce_version(detected)
+    if detected_v is None:
+        # Can't parse the detected version at all - don't claim a match
+        # against a bounded range; that would flood results with false
+        # positives for every technology we can't version-compare.
+        return False
+
+    if min_ver is not None:
+        min_v = _coerce_version(min_ver)
+        if min_v is not None:
+            if min_inclusive:
+                if detected_v < min_v:
+                    return False
+            elif detected_v <= min_v:
+                return False
+
+    if max_ver is not None:
+        max_v = _coerce_version(max_ver)
+        if max_v is not None:
+            if max_inclusive:
+                if detected_v > max_v:
+                    return False
+            elif detected_v >= max_v:
+                return False
+
+    return True
+
 
 def extract_cve_info(cve_entry: Dict) -> List[Dict]:
     """
@@ -180,21 +220,35 @@ def extract_cve_info(cve_entry: Dict) -> List[Dict]:
                 if not cpe:
                     continue
 
-                # Check for version range in versionEndExcluding/versionEndIncluding
-                version_start = match.get("versionStartIncluding")
-                version_end = match.get("versionEndExcluding") or match.get("versionEndIncluding")
-                
-                if version_start or version_end:
-                    min_ver = version_start
-                    max_ver = version_end
-                else:
+                # Check for version range in versionStart/EndIncluding/Excluding.
+                # Including and Excluding are mutually exclusive per NVD's
+                # schema; track which one applied so callers can honor the
+                # correct boundary (an Excluding bound means that version
+                # itself is already patched).
+                min_ver = match.get("versionStartIncluding")
+                min_inclusive = True
+                if min_ver is None:
+                    min_ver = match.get("versionStartExcluding")
+                    min_inclusive = False
+
+                max_ver = match.get("versionEndIncluding")
+                max_inclusive = True
+                if max_ver is None:
+                    max_ver = match.get("versionEndExcluding")
+                    max_inclusive = False
+
+                if min_ver is None and max_ver is None:
                     min_ver, max_ver = parse_version_range(cpe)
+                    min_inclusive = True
+                    max_inclusive = True
 
                 products.append({
                     "vendor": cpe.get("vendor", ""),
                     "product": cpe.get("product", ""),
                     "min_version": min_ver,
                     "max_version": max_ver,
+                    "min_inclusive": min_inclusive,
+                    "max_inclusive": max_inclusive,
                     "version": cpe.get("version"),
                 })
 
@@ -211,7 +265,13 @@ def match_technology_to_cve(tech_name: str, tech_version: Optional[str], cve_ent
     for product in affected_products:
         if product_names_match(tech_name, product["product"]):
             # Product matches, check version
-            if version_in_range(tech_version, product["min_version"], product["max_version"]):
+            if version_in_range(
+                tech_version,
+                product["min_version"],
+                product["max_version"],
+                min_inclusive=product.get("min_inclusive", True),
+                max_inclusive=product.get("max_inclusive", True),
+            ):
                 return {
                     "matched_product": product["product"],
                     "detected_version": tech_version,