valkyrie,plugins: add vulnera vulnerablity scanner plugin

Author: Einswilli

valkyrie/plugins/__init__.py

@@ -1,7 +1,8 @@
 """
 """
 from pathlib import Path
-from typing import List, Set, Dict, Any
+from typing import List, Set, Dict, Any, Optional
+import logging
 
 from valkyrie.core.types import (
     RuleMetadata, SecurityFinding, ScanRule,
@@ -15,8 +16,13 @@
 class BaseSecurityRule(ScanRule):
     """Base implementation for security rules"""
 
-    def __init__(self, metadata: RuleMetadata):
+    def __init__(
+        self, 
+        metadata: RuleMetadata,
+        logger: Optional[logging.Logger] = None
+    ):
         self._metadata = metadata
+        self.logger = logger or logging.getLogger(__name__)
 
     @property
     def metadata(self) -> RuleMetadata:
@@ -42,9 +48,13 @@ async def scan(
 class PluginManager:
     """Manages scanner plugins and their lifecycle"""
 
-    def __init__(self):
+    def __init__(
+        self,
+        logger: Optional[logging.Logger] = None
+    ):
         self.plugins: Dict[str, ScannerPlugin] = {}
         self.enabled_plugins: Set[str] = set()
+        self.logger = logger or logging.getLogger(__name__)
 
     async def register_plugin(
         self, 

valkyrie/plugins/vulnera/__init__.py

@@ -0,0 +1,172 @@
+import hashlib
+from pathlib import Path
+from typing import Dict, List, Any
+
+from valkyrie.plugins import BaseSecurityRule
+from valkyrie.core.types import (
+    ScanRule, ScannerPlugin, RuleMetadata, SecurityFinding, 
+    FileLocation, SeverityLevel, FindingCategory,
+)
+
+from .conf import VulnerabilityInfo
+from .parser import parse_dependencies, is_supported
+
+
+####
+##      VULNERABILITY RULE
+#####
+class DependencyVulnerabilityRule(BaseSecurityRule):
+    """Rule for detecting vulnerable dependencies"""
+    
+    def __init__(
+        self, 
+        vulnerability_db: Dict[str, List[VulnerabilityInfo]]
+    ):
+        metadata = RuleMetadata(
+            id = "deps-001",
+            name = "Dependency Vulnerability Scanner",
+            description = "Scans dependencies for known vulnerabilities",
+            category = FindingCategory.DEPENDENCIES,
+            severity = SeverityLevel.HIGH,
+            author = "Valkyrie Core Team",
+            tags = {"dependencies", "vulnerabilities", "sbom"}
+        )
+        super().__init__(metadata)
+        self.vulnerability_db = vulnerability_db
+    
+    def is_applicable(self, file_path: Path) -> bool:
+        """Check if file is a supported dependency manifest"""
+
+        return is_supported(file_path)
+    
+    async def scan(
+        self, 
+        file_path: Path, 
+        content: str
+    ) -> List[SecurityFinding]:
+        """Scan dependency file for vulnerabilities."""
+
+        findings = []
+        
+        try:
+            dependencies = await self._parse_dependencies(file_path)
+            
+            for dep_name, version in dependencies.items():
+                if dep_name in self.vulnerability_db:
+                    vulnerabilities = self.vulnerability_db[dep_name]
+                    
+                    for vuln in vulnerabilities:
+                        if self._is_version_affected(version, vuln.affected_versions):
+
+                            # Then add it to findings
+                            finding = SecurityFinding(
+                                id = hashlib.md5(f"{file_path}:{dep_name}:{vuln.cve_id}".encode()).hexdigest(),
+                                title = f"Vulnerable dependency: {dep_name}",
+                                description = f"Dependency {dep_name}@{version} has vulnerability {vuln.cve_id}: {vuln.description}",
+                                severity = vuln.severity,
+                                category = self.metadata.category,
+                                location = FileLocation(file_path=file_path, line_number=1),
+                                rule_id = self.metadata.id,
+                                confidence = 0.9,
+                                metadata = {
+                                    "dependency": dep_name,
+                                    "version": version,
+                                    "cve_id": vuln.cve_id,
+                                    "fixed_versions": vuln.fixed_versions,
+                                    "references": vuln.references
+                                },
+                                remediation_advice = (
+                                    f"Update {dep_name} to version "
+                                    f"{', '.join(vuln.fixed_versions) if vuln.fixed_versions else 'latest'}"
+                                )
+                            )
+                            findings.append(finding)
+        
+        except Exception as e:
+            # Log parsing error but don't fail the scan
+            self.logger.warning(
+                f'Error scanning file {file_path}: {e}'
+            )
+        
+        return findings
+    
+    async def _parse_dependencies(self, file_path: Path) -> Dict[str, str]:
+        """Parse dependencies from file content"""
+
+        dependencies = {}
+        
+        # Parse the dependency file
+        for dep in parse_dependencies(file_path=file_path):
+            dependencies[dep.name] = dep.version
+        
+        return dependencies
+    
+    def _is_version_affected(
+        self, 
+        version: str, 
+        affected_versions: List[str]
+    ) -> bool:
+        """Check if version is affected by vulnerability"""
+        # Simplified version comparison for now
+        # I'll use a proper semver library in next push
+        return version in affected_versions
+
+
+####
+##      VULNERABILITY PLUGIN
+#####
+class DependenciesPlugin(ScannerPlugin):
+    """Plugin for dependency vulnerability scanning"""
+    
+    def __init__(self):
+        self.vulnerability_db: Dict[str, List[VulnerabilityInfo]] = {}
+    
+    @property
+    def name(self) -> str:
+        return "vulnera"
+    
+    @property
+    def version(self) -> str:
+        return "1.0.0"
+    
+    async def initialize(self, config: Dict[str, Any]) -> None:
+        """Initialize plugin and load vulnerability database"""
+        await self._load_vulnerability_db()
+    
+    async def _load_vulnerability_db(self) -> None:
+        """Load vulnerability database from external sources"""
+        
+        # Normally we need to call an external service 
+        # Or load a local vulnerabilities dbm
+        # but i'm usinng a mock database and
+        # i'll fix that in next push
+        self.vulnerability_db = {
+            "lodash": [
+                VulnerabilityInfo(
+                    cve_id="CVE-2021-23337",
+                    severity=SeverityLevel.HIGH,
+                    description="Prototype pollution in lodash",
+                    affected_versions=["4.17.20"],
+                    fixed_versions=["4.17.21"],
+                    references=["https://nvd.nist.gov/vuln/detail/CVE-2021-23337"]
+                )
+            ],
+            "requests": [
+                VulnerabilityInfo(
+                    cve_id="CVE-2023-32681",
+                    severity=SeverityLevel.MEDIUM,
+                    description="Certificate verification bypass in requests",
+                    affected_versions=["2.30.0", "2.29.0"],
+                    fixed_versions=["2.31.0"],
+                    references=["https://nvd.nist.gov/vuln/detail/CVE-2023-32681"]
+                )
+            ]
+        }
+    
+    async def get_rules(self) -> List[ScanRule]:
+        """Return dependency scanning rules"""
+        return [DependencyVulnerabilityRule(self.vulnerability_db)]
+    
+    async def cleanup(self) -> None:
+        """Cleanup plugin resources"""
+        pass

valkyrie/plugins/vulnera/conf.py

@@ -0,0 +1,60 @@
+from typing import List, Optional
+from dataclasses import dataclass, field
+from valkyrie.core.types import (
+    SeverityLevel
+)
+
+
+####
+##      VULNERABILITY MODEL
+#####
+@dataclass
+class VulnerabilityInfo:
+    """Information about a vulnerability"""
+    
+    cve_id: str
+    severity: SeverityLevel
+    description: str
+    affected_versions: List[str]
+    fixed_versions: List[str] = field(default_factory=list)
+    references: List[str] = field(default_factory=list)
+
+
+####
+##      DEPENDENCY MODEL
+#####
+@dataclass
+class Dependency:
+    """Project dependency rpresentation model"""
+
+    name: str
+    version: Optional[str] = None
+    dev: bool = False
+    source: Optional[str] = None 
+    
+    def __str__(self):
+        version_str = f"@{self.version}" if self.version else ""
+        dev_str = " (dev)" if self.dev else ""
+        return f"{self.name}{version_str}{dev_str}"
+
+
+####    DEPENDENCIES
+DEPS_FIES = {
+    # Node.js
+    'package.json', 'package-lock.json', 'yarn.lock',
+
+    # Python
+    'requirements.txt', 'Pipfile', 'Pipfile.lock', 'poetry.lock',
+
+    # Java
+    'pom.xml', 'gradle.build',
+
+    # Rust
+    'Cargo.toml', 'Cargo.lock',
+
+    # Go
+    'go.mod', 'go.sum',
+
+    # PHP
+    'composer.json', 'composer.lock'
+}