#!/usr/bin/env python3
"""Agent for performing physical intrusion assessment — checklist management and finding documentation."""

import json
import argparse
import csv
from datetime import datetime


ASSESSMENT_CATEGORIES = {
    "perimeter": {
        "name": "Perimeter Security",
        "checks": [
            {"id": "P01", "check": "Perimeter fencing intact and adequate height (>=7ft)", "severity": "HIGH"},
            {"id": "P02", "check": "CCTV cameras covering all entry points", "severity": "HIGH"},
            {"id": "P03", "check": "Adequate exterior lighting at night", "severity": "MEDIUM"},
            {"id": "P04", "check": "Vehicle barriers at building entrances", "severity": "MEDIUM"},
            {"id": "P05", "check": "Signage prohibiting unauthorized access", "severity": "LOW"},
        ],
    },
    "access_control": {
        "name": "Access Control",
        "checks": [
            {"id": "A01", "check": "Badge/card access on all entry doors", "severity": "CRITICAL"},
            {"id": "A02", "check": "Tailgating prevention mechanisms (mantrap, turnstile)", "severity": "HIGH"},
            {"id": "A03", "check": "Visitor sign-in and escort policy enforced", "severity": "HIGH"},
            {"id": "A04", "check": "Badge visible at all times policy", "severity": "MEDIUM"},
            {"id": "A05", "check": "After-hours access controls and logging", "severity": "HIGH"},
            {"id": "A06", "check": "Terminated employee badge deactivation process", "severity": "CRITICAL"},
        ],
    },
    "server_room": {
        "name": "Server Room / Data Center",
        "checks": [
            {"id": "S01", "check": "MFA or biometric access to server room", "severity": "CRITICAL"},
            {"id": "S02", "check": "CCTV monitoring inside server room", "severity": "HIGH"},
            {"id": "S03", "check": "Environmental controls (temp, humidity sensors)", "severity": "MEDIUM"},
            {"id": "S04", "check": "Fire suppression system present", "severity": "HIGH"},
            {"id": "S05", "check": "Access logs reviewed regularly", "severity": "MEDIUM"},
            {"id": "S06", "check": "No unlocked network ports in common areas", "severity": "HIGH"},
        ],
    },
    "social_engineering": {
        "name": "Social Engineering Resistance",
        "checks": [
            {"id": "E01", "check": "Employees challenge unknown visitors", "severity": "HIGH"},
            {"id": "E02", "check": "Clean desk policy enforced", "severity": "MEDIUM"},
            {"id": "E03", "check": "Sensitive documents shredded", "severity": "MEDIUM"},
            {"id": "E04", "check": "USB drives not left unattended", "severity": "HIGH"},
            {"id": "E05", "check": "Dumpster diving countermeasures", "severity": "MEDIUM"},
        ],
    },
}


def generate_checklist(categories=None, output_file=None):
    """Generate physical security assessment checklist."""
    cats = categories or list(ASSESSMENT_CATEGORIES.keys())
    checklist = []
    for cat in cats:
        if cat in ASSESSMENT_CATEGORIES:
            data = ASSESSMENT_CATEGORIES[cat]
            for check in data["checks"]:
                checklist.append({
                    "category": data["name"], "id": check["id"],
                    "check": check["check"], "severity": check["severity"],
                    "status": "NOT_TESTED", "finding": "", "evidence": "",
                })
    if output_file:
        with open(output_file, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=["category", "id", "check", "severity", "status", "finding", "evidence"])
            writer.writeheader()
            writer.writerows(checklist)
    return {"total_checks": len(checklist), "categories": cats, "checklist": checklist, "output": output_file}


def score_assessment(results_csv):
    """Score a completed physical security assessment."""
    with open(results_csv, "r", encoding="utf-8", errors="replace") as f:
        reader = csv.DictReader(f)
        results = list(reader)
    passed = sum(1 for r in results if r.get("status", "").lower() in ("pass", "compliant", "ok"))
    failed = sum(1 for r in results if r.get("status", "").lower() in ("fail", "non-compliant", "nc"))
    total = len(results)
    by_category = {}
    by_severity = {}
    failures = []
    for r in results:
        cat = r.get("category", "unknown")
        sev = r.get("severity", "MEDIUM")
        by_category.setdefault(cat, {"pass": 0, "fail": 0})
        by_severity.setdefault(sev, {"pass": 0, "fail": 0})
        if r.get("status", "").lower() in ("pass", "compliant", "ok"):
            by_category[cat]["pass"] += 1
            by_severity[sev]["pass"] += 1
        elif r.get("status", "").lower() in ("fail", "non-compliant", "nc"):
            by_category[cat]["fail"] += 1
            by_severity[sev]["fail"] += 1
            failures.append({"id": r.get("id"), "check": r.get("check"), "severity": sev,
                             "finding": r.get("finding", "")[:200]})
    return {
        "total_checks": total, "passed": passed, "failed": failed,
        "compliance_pct": round(passed / max(total, 1) * 100, 1),
        "by_category": by_category,
        "by_severity": by_severity,
        "critical_failures": [f for f in failures if f["severity"] == "CRITICAL"],
        "all_failures": failures,
    }


def generate_report(results_csv):
    """Generate executive physical security assessment report."""
    scores = score_assessment(results_csv)
    risk = "CRITICAL" if scores.get("critical_failures") else "HIGH" if scores["failed"] > 5 else "MEDIUM" if scores["failed"] > 0 else "LOW"
    return {
        "generated": datetime.utcnow().isoformat(),
        "overall_risk": risk,
        "compliance_score": scores["compliance_pct"],
        **scores,
        "recommendations": [
            f"CRITICAL: Address {len(scores['critical_failures'])} critical findings immediately"
        ] if scores["critical_failures"] else ["All critical controls passed"],
    }


def main():
    parser = argparse.ArgumentParser(description="Physical Intrusion Assessment Agent")
    sub = parser.add_subparsers(dest="command")
    c = sub.add_parser("checklist", help="Generate assessment checklist")
    c.add_argument("--categories", nargs="*", choices=list(ASSESSMENT_CATEGORIES.keys()))
    c.add_argument("--output", help="Output CSV file")
    s = sub.add_parser("score", help="Score completed assessment")
    s.add_argument("--csv", required=True)
    r = sub.add_parser("report", help="Generate assessment report")
    r.add_argument("--csv", required=True)
    args = parser.parse_args()
    if args.command == "checklist":
        result = generate_checklist(args.categories, args.output)
    elif args.command == "score":
        result = score_assessment(args.csv)
    elif args.command == "report":
        result = generate_report(args.csv)
    else:
        parser.print_help()
        return
    print(json.dumps(result, indent=2, default=str))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Physical Security Assessment Tracker

Tracks physical intrusion testing attempts, documents findings,
and generates assessment reports with remediation recommendations.
"""

import json
import os
from datetime import datetime
from collections import defaultdict
from dataclasses import dataclass, field, asdict


@dataclass
class PhysicalAccessAttempt:
    """A single physical access test attempt."""
    attempt_id: str
    timestamp: str
    location: str
    entry_point: str  # main_entrance, side_door, loading_dock, etc.
    technique: str  # tailgating, badge_clone, lock_pick, social_engineering
    target_area: str  # lobby, server_room, executive_floor, data_center
    successful: bool
    detected: bool
    challenged: bool  # was tester confronted/questioned
    security_notified: bool
    time_inside_minutes: int = 0
    device_deployed: bool = False
    device_type: str = ""
    evidence_collected: str = ""
    notes: str = ""
    severity: str = ""  # critical, high, medium, low


class PhysicalAssessmentTracker:
    """Track physical security assessment results."""

    def __init__(self, assessment_id: str, facility_name: str):
        self.assessment_id = assessment_id
        self.facility_name = facility_name
        self.attempts: list[PhysicalAccessAttempt] = []

    def log_attempt(self, attempt: PhysicalAccessAttempt) -> None:
        """Log an access attempt."""
        # Auto-assign severity
        if not attempt.severity:
            if attempt.target_area in ("server_room", "data_center") and attempt.successful:
                attempt.severity = "critical"
            elif attempt.successful and not attempt.detected:
                attempt.severity = "high"
            elif attempt.successful and attempt.detected:
                attempt.severity = "medium"
            else:
                attempt.severity = "low"
        self.attempts.append(attempt)

    def calculate_metrics(self) -> dict:
        """Calculate assessment metrics."""
        total = len(self.attempts)
        if total == 0:
            return {}

        successful = [a for a in self.attempts if a.successful]
        detected = [a for a in self.attempts if a.detected]
        challenged = [a for a in self.attempts if a.challenged]
        devices = [a for a in self.attempts if a.device_deployed]

        # By technique
        technique_stats = defaultdict(lambda: {"total": 0, "success": 0})
        for a in self.attempts:
            technique_stats[a.technique]["total"] += 1
            if a.successful:
                technique_stats[a.technique]["success"] += 1

        # By area
        area_stats = defaultdict(lambda: {"total": 0, "success": 0})
        for a in self.attempts:
            area_stats[a.target_area]["total"] += 1
            if a.successful:
                area_stats[a.target_area]["success"] += 1

        return {
            "total_attempts": total,
            "successful_entries": len(successful),
            "success_rate": len(successful) / total * 100,
            "detection_rate": len(detected) / total * 100,
            "challenge_rate": len(challenged) / total * 100,
            "devices_deployed": len(devices),
            "technique_breakdown": dict(technique_stats),
            "area_breakdown": dict(area_stats),
            "avg_time_inside": (
                sum(a.time_inside_minutes for a in successful) / len(successful)
                if successful else 0
            ),
        }

    def generate_report(self) -> str:
        """Generate assessment report."""
        metrics = self.calculate_metrics()
        if not metrics:
            return "No attempts logged."

        lines = []
        lines.append("=" * 70)
        lines.append("PHYSICAL SECURITY ASSESSMENT REPORT")
        lines.append(f"Assessment ID: {self.assessment_id}")
        lines.append(f"Facility: {self.facility_name}")
        lines.append(f"Date: {datetime.now().strftime('%Y-%m-%d')}")
        lines.append("=" * 70)

        lines.append(f"\nSUMMARY:")
        lines.append(f"  Total Attempts:       {metrics['total_attempts']}")
        lines.append(f"  Successful Entries:    {metrics['successful_entries']}")
        lines.append(f"  Success Rate:          {metrics['success_rate']:.1f}%")
        lines.append(f"  Detection Rate:        {metrics['detection_rate']:.1f}%")
        lines.append(f"  Challenge Rate:        {metrics['challenge_rate']:.1f}%")
        lines.append(f"  Devices Deployed:      {metrics['devices_deployed']}")
        lines.append(f"  Avg Time Inside:       {metrics['avg_time_inside']:.0f} minutes")

        risk = (
            "CRITICAL" if metrics['success_rate'] > 60
            else "HIGH" if metrics['success_rate'] > 40
            else "MEDIUM" if metrics['success_rate'] > 20
            else "LOW"
        )
        lines.append(f"\n  OVERALL RISK: {risk}")

        lines.append(f"\nTECHNIQUE EFFECTIVENESS:")
        lines.append("-" * 70)
        for tech, stats in metrics["technique_breakdown"].items():
            rate = stats["success"] / stats["total"] * 100 if stats["total"] else 0
            lines.append(f"  {tech:<25} {rate:>5.1f}% ({stats['success']}/{stats['total']})")

        lines.append(f"\nAREA ACCESS:")
        lines.append("-" * 70)
        for area, stats in metrics["area_breakdown"].items():
            rate = stats["success"] / stats["total"] * 100 if stats["total"] else 0
            lines.append(f"  {area:<25} {rate:>5.1f}% ({stats['success']}/{stats['total']})")

        lines.append(f"\nDETAILED FINDINGS:")
        lines.append("-" * 70)
        for a in sorted(self.attempts, key=lambda x: {"critical": 0, "high": 1, "medium": 2, "low": 3}.get(x.severity, 4)):
            status = "SUCCESS" if a.successful else "FAILED"
            lines.append(f"\n  [{a.severity.upper()}] {a.location} - {a.entry_point}")
            lines.append(f"    Technique: {a.technique} | Result: {status}")
            lines.append(f"    Detected: {'Yes' if a.detected else 'No'} | "
                        f"Challenged: {'Yes' if a.challenged else 'No'}")
            if a.device_deployed:
                lines.append(f"    Device Deployed: {a.device_type}")
            if a.notes:
                lines.append(f"    Notes: {a.notes}")

        lines.append(f"\nREMEDIATIONS:")
        lines.append("-" * 70)
        if metrics["success_rate"] > 50:
            lines.append("  [CRITICAL] Physical access controls require immediate review")
        if any(a.technique == "tailgating" and a.successful for a in self.attempts):
            lines.append("  [HIGH] Install mantraps or anti-tailgating turnstiles")
            lines.append("  [HIGH] Implement security awareness training on tailgating")
        if any(a.technique == "badge_clone" and a.successful for a in self.attempts):
            lines.append("  [HIGH] Upgrade to encrypted RFID (SEOS, DESFire EV2)")
        if metrics["challenge_rate"] < 50:
            lines.append("  [HIGH] Improve guard challenge procedures")
        if any(a.device_deployed for a in self.attempts):
            lines.append("  [CRITICAL] Implement physical port security controls")

        return "\n".join(lines)


def main():
    """Demonstrate physical assessment tracking."""
    tracker = PhysicalAssessmentTracker("PHYS-2025-001", "Example Corp HQ")

    attempts = [
        PhysicalAccessAttempt("P001", "2025-02-10T08:45:00", "Main Building",
            "main_entrance", "tailgating", "lobby", True, False, False, False, 45,
            notes="Followed group during morning rush"),
        PhysicalAccessAttempt("P002", "2025-02-10T10:00:00", "Main Building",
            "side_door", "badge_clone", "office_floor", True, False, False, False, 30,
            notes="Cloned badge from elevator reading"),
        PhysicalAccessAttempt("P003", "2025-02-10T14:00:00", "Data Center Wing",
            "secured_door", "badge_clone", "server_room", True, True, True, True, 5,
            device_deployed=True, device_type="LAN Turtle",
            notes="Guard challenged but accepted fake contractor story"),
        PhysicalAccessAttempt("P004", "2025-02-11T07:30:00", "Loading Dock",
            "loading_dock", "social_engineering", "warehouse", True, False, False, False, 20,
            notes="Posed as delivery driver with empty boxes"),
        PhysicalAccessAttempt("P005", "2025-02-11T12:00:00", "Executive Floor",
            "elevator", "tailgating", "executive_office", False, True, True, True, 0,
            notes="Security escort required, access denied"),
    ]

    for attempt in attempts:
        tracker.log_attempt(attempt)

    print(tracker.generate_report())


if __name__ == "__main__":
    main()