from __future__ import annotations import ast import json import logging import re from pathlib import Path import toml from packaging.requirements import InvalidRequirement, Requirement from packaging.specifiers import SpecifierSet from packaging.version import Version from vuln_scan.core.models import ( Dependency, Ecosystem, EcosystemParseResult, SecurityVulnerability, ) from vuln_scan.ecosystems.base import EcosystemHandler logger = logging.getLogger(__name__) class PythonEcosystemHandler(EcosystemHandler): """ Ecosystem handler for Python dependencies. """ id = "python" ecosystem = Ecosystem.PIP # -------------------------------------------------------- # File discovery # -------------------------------------------------------- @property def manifest_names(self) -> set[str]: return { "pyproject.toml", "setup.py", } @property def manifest_globs(self) -> set[str]: return { "*requirements*.txt", } @property def lockfile_names(self) -> set[str]: return { "poetry.lock", "Pipfile.lock", "uv.lock", } # -------------------------------------------------------- # Parsing entrypoints # -------------------------------------------------------- def parse_manifest( self, manifest_path: str, full_path: Path, ) -> EcosystemParseResult: content = full_path.read_text( encoding="utf-8", errors="ignore", ) return self.parse_manifest_content( manifest_path, content, ) def parse_manifest_content( self, manifest_path: str, content: str, ) -> EcosystemParseResult: filename = Path(manifest_path).name is_lockfile = self.is_lockfile(filename) if filename.endswith(".txt"): deps = self._parse_requirements_txt_content( content, manifest_path, is_lockfile, ) elif filename == "Pipfile.lock": deps = self._parse_pipfile_lock_content( content, manifest_path, is_lockfile, ) elif filename == "pyproject.toml": deps = self._parse_pyproject_toml_content( content, manifest_path, is_lockfile, ) elif filename in {"poetry.lock", "uv.lock"}: deps = self._parse_poetry_like_lock_content( content, manifest_path, is_lockfile, ) elif filename == "setup.py": deps = self._parse_setup_py_content( content, manifest_path, is_lockfile, ) else: deps = [] return EcosystemParseResult( dependencies=deps, manifest_type="lockfile" if is_lockfile else "manifest", parser_name=self.id, ) # -------------------------------------------------------- # Vulnerability matching # -------------------------------------------------------- def is_vulnerable( self, dependency: Dependency, alert: SecurityVulnerability, ) -> tuple[bool, str]: if not alert.vulnerable_range: return True, "low" try: normalized_range = self._normalize_vuln_range(alert.vulnerable_range) vuln_spec = SpecifierSet(normalized_range) # ---------------------------------------- # 1. Lockfile OR normalized exact version # ---------------------------------------- if dependency.version: is_vuln = Version(dependency.version) in vuln_spec confidence = "high" if dependency.is_lockfile else "medium" return is_vuln, confidence # ---------------------------------------- # 2. Requirement (range overlap) # ---------------------------------------- if dependency.requirement: normalized_requirement = self._normalize_poetry_requirement(dependency.requirement) if not normalized_requirement: # e.g. bare "*" — unbounded, so treat conservatively return True, "low" req_spec = SpecifierSet(normalized_requirement) if self._pip_ranges_overlap(req_spec, vuln_spec): return True, "low" return False, "low" # ---------------------------------------- # 3. Unknown → conservative # ---------------------------------------- return True, "low" except Exception: logger.warning( "Failed to evaluate vulnerability for %s@%s against range '%s', " "assuming vulnerable", dependency.name, dependency.version or dependency.requirement, alert.vulnerable_range, exc_info=True, ) return True, "low" def _pip_ranges_overlap( self, req_spec: SpecifierSet, vuln_spec: SpecifierSet, ) -> bool: """ Determine whether two specifier sets overlap by extracting their effective bounds and checking for interval intersection. Falls back to conservative (True) if bounds cannot be determined. """ try: req_lower, req_lower_inc, req_upper, req_upper_inc = self._extract_bounds(req_spec) vuln_lower, vuln_lower_inc, vuln_upper, vuln_upper_inc = self._extract_bounds(vuln_spec) # Two intervals do NOT overlap when one starts after the other ends. # Must account for inclusive vs exclusive boundaries at equal versions. if req_lower is not None and vuln_upper is not None: if req_lower > vuln_upper: return False if req_lower == vuln_upper and not (req_lower_inc and vuln_upper_inc): return False if vuln_lower is not None and req_upper is not None: if vuln_lower > req_upper: return False if vuln_lower == req_upper and not (vuln_lower_inc and req_upper_inc): return False return True except Exception: logger.debug( "Could not determine overlap for req=%s vuln=%s, assuming overlap", req_spec, vuln_spec, exc_info=True, ) return True def _extract_bounds( self, spec: SpecifierSet, ) -> tuple[Version | None, bool, Version | None, bool]: """ Extract the effective lower and upper bounds from a SpecifierSet. Returns (lower_bound, lower_inclusive, upper_bound, upper_inclusive) where None means unbounded. """ lower: Version | None = None lower_inc: bool = True upper: Version | None = None upper_inc: bool = True for s in spec: op = s.operator ver = Version(s.version) if op in (">=", ">", "~="): if lower is None or ver > lower: lower = ver lower_inc = op in (">=", "~=") elif op in ("<=", "<"): if upper is None or ver < upper: upper = ver upper_inc = op == "<=" elif op == "==": # Exact pin — both bounds are the same version lower = ver lower_inc = True upper = ver upper_inc = True elif op == "!=": # Exclusion doesn't define bounds pass return lower, lower_inc, upper, upper_inc # -------------------------------------------------------- # Poetry constraint normalization (caret / tilde / wildcard) # -------------------------------------------------------- _CARET_RE = re.compile(r"^\^\s*(\d+)(?:\.(\d+))?(?:\.(\d+))?$") _TILDE_RE = re.compile(r"^~\s*(\d+)(?:\.(\d+))?(?:\.(\d+))?$") _WILDCARD_RE = re.compile(r"^(\d+)(?:\.(\d+))?\.[x\*]$", re.IGNORECASE) def _normalize_poetry_requirement( self, requirement: str, ) -> str: """ Translate Poetry-style version constraints (caret ``^``, tilde ``~``, and wildcard ``x``/``*``) into a PEP 440 compatible specifier string that ``packaging.specifiers.SpecifierSet`` can parse. """ parts = [p.strip() for p in requirement.split(",") if p.strip()] normalized_parts: list[str] = [] for part in parts: if part == "*": continue caret_match = self._CARET_RE.match(part) if caret_match: normalized_parts.append(self._expand_caret(caret_match)) continue tilde_match = self._TILDE_RE.match(part) if tilde_match: normalized_parts.append(self._expand_tilde(tilde_match)) continue wildcard_match = self._WILDCARD_RE.match(part) if wildcard_match: normalized_parts.append(self._expand_wildcard(wildcard_match)) continue normalized_parts.append(part) return ",".join(normalized_parts) def _expand_caret(self, match: re.Match[str]) -> str: major, minor, patch = match.group(1), match.group(2), match.group(3) lower = ".".join(p for p in (major, minor, patch) if p is not None) if major != "0": upper = f"{int(major) + 1}.0.0" elif minor is None: upper = "1.0.0" elif minor != "0": upper = f"0.{int(minor) + 1}.0" elif patch is None: upper = "0.1.0" else: upper = f"0.0.{int(patch) + 1}" return f">={lower},<{upper}" def _expand_tilde(self, match: re.Match[str]) -> str: major, minor, patch = match.group(1), match.group(2), match.group(3) lower = ".".join(p for p in (major, minor, patch) if p is not None) upper = f"{int(major) + 1}.0.0" if minor is None else f"{major}.{int(minor) + 1}.0" return f">={lower},<{upper}" def _expand_wildcard(self, match: re.Match[str]) -> str: major, minor = match.group(1), match.group(2) if minor is None: lower = f"{major}.0.0" upper = f"{int(major) + 1}.0.0" else: lower = f"{major}.{minor}.0" upper = f"{major}.{int(minor) + 1}.0" return f">={lower},<{upper}" # -------------------------------------------------------- # Normalization helper # -------------------------------------------------------- def _normalize_specifier( self, spec: str, ) -> tuple[str | None, str | None]: """ Normalize specifier into (version, requirement) """ spec = (spec or "").strip() if not spec: return None, None # == exact if spec.startswith("=="): v = spec.lstrip("=").strip() return (v, None) if v else (None, spec) # bare version (e.g. "1.2.3") if re.fullmatch(r"\d+(\.\d+)*", spec): return spec, None return None, spec # -------------------------------------------------------- # requirements.txt # -------------------------------------------------------- def _parse_requirements_txt_content( self, content: str, manifest_path: str, is_lockfile: bool, ) -> list[Dependency]: deps: list[Dependency] = [] for line in content.splitlines(): s = line.strip() if not s or s.startswith("#"): continue if s.startswith(("-", "--")): continue if " #" in s: s = s.split(" #", 1)[0].strip() try: req = Requirement(s) except InvalidRequirement: logger.debug("Skipping unparseable requirement in %s: %s", manifest_path, s) continue specifier_str = str(req.specifier) or "" version, requirement = self._normalize_specifier(specifier_str) deps.append( Dependency( ecosystem=self.ecosystem, name=req.name, version=version, requirement=requirement, manifest_path=manifest_path, is_lockfile=is_lockfile, ) ) return deps # -------------------------------------------------------- # Pipfile.lock # -------------------------------------------------------- def _parse_pipfile_lock_content( self, content: str, manifest_path: str, is_lockfile: bool, ) -> list[Dependency]: deps: list[Dependency] = [] try: data = json.loads(content) except json.JSONDecodeError: logger.warning("Failed to parse %s as JSON, skipping", manifest_path, exc_info=True) return deps if not isinstance(data, dict): logger.warning("Unexpected Pipfile.lock structure in %s, skipping", manifest_path) return deps for section in ("default", "develop"): sec = data.get(section, {}) if not isinstance(sec, dict): continue for pkg, details in sec.items(): if not isinstance(details, dict): continue ver = str(details.get("version", "")).lstrip("=").strip() or None deps.append( Dependency( ecosystem=self.ecosystem, name=pkg, version=ver, requirement=None, manifest_path=manifest_path, is_lockfile=is_lockfile, ) ) return deps # -------------------------------------------------------- # pyproject.toml # -------------------------------------------------------- def _parse_pyproject_toml_content( self, content: str, manifest_path: str, is_lockfile: bool, ) -> list[Dependency]: deps: list[Dependency] = [] try: data = toml.loads(content) except toml.TomlDecodeError as exc: logger.warning( "Failed to parse pyproject.toml at %s: %s", manifest_path, exc, ) return deps poetry_deps = (((data.get("tool") or {}).get("poetry") or {}).get("dependencies")) or {} if isinstance(poetry_deps, dict): for pkg, v in poetry_deps.items(): if pkg.lower() == "python": continue raw = str(v.get("version", "")) if isinstance(v, dict) else str(v) raw = re.sub( r"[^0-9\.\<\>\=\~\^\*\sx\-,]", "", raw, ) version, requirement = self._normalize_specifier(raw) deps.append( Dependency( ecosystem=self.ecosystem, name=pkg, version=version, requirement=requirement, manifest_path=manifest_path, is_lockfile=is_lockfile, ) ) return deps # -------------------------------------------------------- # poetry.lock / uv.lock # -------------------------------------------------------- def _parse_poetry_like_lock_content( self, content: str, manifest_path: str, is_lockfile: bool, ) -> list[Dependency]: try: data = toml.loads(content) except toml.TomlDecodeError: logger.warning("Failed to parse %s as TOML, skipping", manifest_path) return [] packages = data.get("package", []) if not isinstance(packages, list): return [] deps: list[Dependency] = [] for pkg in packages: if not isinstance(pkg, dict): continue name = pkg.get("name") version = pkg.get("version") if not isinstance(name, str) or not name: continue if not isinstance(version, str) or not version: continue deps.append( Dependency( ecosystem=self.ecosystem, name=name, version=version, requirement=None, manifest_path=manifest_path, is_lockfile=is_lockfile, ) ) return deps # -------------------------------------------------------- # setup.py # -------------------------------------------------------- def _parse_setup_py_content( self, content: str, manifest_path: str, is_lockfile: bool, ) -> list[Dependency]: tree = self._safe_parse_ast(content, manifest_path) if not tree: return [] req_strings = self._extract_setup_requirements(tree) return self._build_dependencies(req_strings, manifest_path, is_lockfile) # -------------------------------------------------------- # AST helpers # -------------------------------------------------------- def _safe_parse_ast(self, content: str, manifest_path: str) -> ast.AST | None: try: return ast.parse(content, filename=manifest_path) except SyntaxError: logger.warning("Failed to parse %s as Python AST", manifest_path, exc_info=True) return None def _extract_setup_requirements(self, tree: ast.AST) -> list[str]: req_strings: list[str] = [] for node in ast.walk(tree): if isinstance(node, ast.Call) and self._is_setup_call(node): req_strings.extend(self._extract_requirements_from_call(node)) return req_strings def _is_setup_call(self, node: ast.AST) -> bool: return isinstance(node, ast.Call) and self._is_name_or_attr(node.func, {"setup"}) def _extract_requirements_from_call(self, node: ast.Call) -> list[str]: results: list[str] = [] for kw in node.keywords: if kw.arg in ("install_requires", "tests_require"): results.extend(self._extract_string_list(kw.value)) elif kw.arg == "extras_require": results.extend(self._extract_extras_require(kw.value)) return results def _build_dependencies( self, req_strings: list[str], manifest_path: str, is_lockfile: bool, ) -> list[Dependency]: deps: list[Dependency] = [] for s in req_strings: dep = self._parse_requirement_string(s, manifest_path, is_lockfile) if dep: deps.append(dep) return deps def _parse_requirement_string( self, s: str, manifest_path: str, is_lockfile: bool, ) -> Dependency | None: s2 = (s or "").strip() if not s2: return None try: req = Requirement(s2) except InvalidRequirement: logger.debug("Skipping unparseable requirement in %s: %s", manifest_path, s2) return None specifier_str = str(req.specifier) or "" version, requirement = self._normalize_specifier(specifier_str) return Dependency( ecosystem=self.ecosystem, name=req.name, version=version, requirement=requirement, manifest_path=manifest_path, is_lockfile=is_lockfile, ) def _is_name_or_attr(self, func: ast.AST, names: set[str]) -> bool: if isinstance(func, ast.Name): return func.id in names if isinstance(func, ast.Attribute): return func.attr in names return False def _extract_string_literal(self, node: ast.AST) -> str | None: if isinstance(node, ast.Constant) and isinstance(node.value, str): return node.value return None def _extract_extras_require(self, node: ast.AST) -> list[str]: if not isinstance(node, ast.Dict): return [] out: list[str] = [] for v in node.values: out.extend(self._extract_string_list(v)) return out def _extract_string_list(self, node: ast.AST) -> list[str]: if isinstance(node, (ast.List, ast.Tuple)): out: list[str] = [] for elt in node.elts: s = self._extract_string_literal(elt) if s is not None: out.append(s) return out s = self._extract_string_literal(node) return [s] if s is not None else []