Vulnerability Services

class VulnerabilityService(ABC):
    """
    Represents an abstract vulnerability service.
    
    Individual concrete vulnerability services are expected to subclass
    VulnerabilityService and implement it in their terms.
    """
    
    @abstractmethod
    def query(self, spec: Dependency) -> tuple[Dependency, list[VulnerabilityResult]]:
        """
        Query the service for vulnerabilities in the given dependency.
        
        Parameters:
        - spec: Dependency, the dependency to query for vulnerabilities
        
        Returns:
        Tuple of (dependency, list of vulnerability results)
        """
    
    def query_all(
        self, specs: Iterator[Dependency]
    ) -> Iterator[tuple[Dependency, list[VulnerabilityResult]]]:
        """
        Query the vulnerability service for information on multiple dependencies.
        
        VulnerabilityService implementations can override this with a more optimized
        implementation if they support batched or bulk requests.
        
        Parameters:
        - specs: Iterator[Dependency], iterator of dependencies to query
        
        Returns:
        Iterator of (dependency, vulnerabilities) tuples
        """

class PyPIService(VulnerabilityService):
    """
    An implementation of VulnerabilityService that uses PyPI to provide Python
    package vulnerability information.
    """
    
    def __init__(self, cache_dir: Path | None = None, timeout: int | None = None):
        """
        Create a new PyPIService.
        
        Parameters:
        - cache_dir: Path | None, optional cache directory for PyPI API requests
        - timeout: int | None, timeout in seconds for network requests
        """

    def query(self, spec: Dependency) -> tuple[Dependency, list[VulnerabilityResult]]:
        """
        Query PyPI for vulnerabilities in the given package.
        
        Parameters:
        - spec: Dependency, the package to query
        
        Returns:
        Tuple of (dependency, vulnerabilities)
        """

class OsvService(VulnerabilityService):
    """
    An implementation of VulnerabilityService that uses OSV to provide vulnerability
    information.
    """
    
    def __init__(self, cache_dir: Path | None = None, timeout: int | None = None):
        """
        Create a new OsvService.
        
        Parameters:
        - cache_dir: Path | None, optional cache directory for OSV API requests
        - timeout: int | None, timeout in seconds for network requests
        """

    def query(self, spec: Dependency) -> tuple[Dependency, list[VulnerabilityResult]]:
        """
        Query OSV for vulnerabilities in the given package.
        
        Parameters:
        - spec: Dependency, the package to query
        
        Returns:
        Tuple of (dependency, vulnerabilities)
        """

class ServiceError(Exception):
    """
    Base exception for vulnerability service errors.
    """

class ConnectionError(ServiceError):
    """
    Raised when a vulnerability service fails to connect or times out.
    """

from pip_audit._service import PyPIService
from pip_audit._service.interface import ResolvedDependency
from packaging.version import Version

# Create service
service = PyPIService()

# Query for vulnerabilities
dependency = ResolvedDependency(name="requests", version=Version("2.25.0"))
dep, vulnerabilities = service.query(dependency)

if vulnerabilities:
    print(f"Found {len(vulnerabilities)} vulnerabilities in {dep.name}")
    for vuln in vulnerabilities:
        print(f"  {vuln.id}: {vuln.description}")
        if vuln.fix_versions:
            print(f"    Fix versions: {[str(v) for v in vuln.fix_versions]}")

from pip_audit._service import OsvService
from pip_audit._service.interface import ResolvedDependency
from packaging.version import Version

# Create service with custom timeout
service = OsvService(timeout=30)

# Query for vulnerabilities
dependency = ResolvedDependency(name="django", version=Version("3.0.0"))
dep, vulnerabilities = service.query(dependency)

for vuln in vulnerabilities:
    print(f"Vulnerability: {vuln.id}")
    print(f"Description: {vuln.description}")
    print(f"Aliases: {vuln.aliases}")

from pip_audit._service import PyPIService, OsvService
from pip_audit._service.interface import ResolvedDependency
from packaging.version import Version

# Compare results from different services
dependency = ResolvedDependency(name="flask", version=Version("1.0.0"))

pypi_service = PyPIService()
osv_service = OsvService()

_, pypi_vulns = pypi_service.query(dependency)
_, osv_vulns = osv_service.query(dependency)

print(f"PyPI found {len(pypi_vulns)} vulnerabilities")
print(f"OSV found {len(osv_vulns)} vulnerabilities")

# Combine results
all_vulns = pypi_vulns + osv_vulns
unique_ids = set(vuln.id for vuln in all_vulns)
print(f"Total unique vulnerabilities: {len(unique_ids)}")