valkyrie, plugins: add vulnera vulnerablity scanner plugin #16

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

Wgoeh merged 32 commits into AllDotPy:master from Einswilli:feat.deps_plugin

Sep 11, 2025

valkyrie/plugins/__init__.py

-Original file line number
+Diff line change
@@ -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,
@@ Expand All / @@ -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:
@@ Expand All / @@ -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,
@@ Expand Down @@

valkyrie/plugins/vulnera/__init__.py

-Original file line number
+Diff line change
@@ -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

-Original file line number
+Diff line change
@@ -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'
+    }

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

valkyrie, plugins: add vulnera vulnerablity scanner plugin #16

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!