🔒 Security Troubleshooting¶

# debug/permission_debugger.py
async def debug_user_permissions(user_id: str) -> Dict[str, Any]:
    """Debuggt Benutzer-Berechtigungen."""

    user = await user_service.get_user(user_id)
    if not user:
        return {"error": "User not found"}

    # Rollen abrufen
    user_roles = await rbac_service.get_user_roles(user_id)

    # Berechtigungen für jede Rolle
    role_permissions = {}
    for role in user_roles:
        permissions = await rbac_service.get_role_permissions(role.name)
        role_permissions[role.name] = [p.name for p in permissions]

    # Effektive Berechtigungen
    effective_permissions = await rbac_service.get_user_permissions(user_id)

    return {
        "user_id": user_id,
        "username": user.username,
        "roles": [r.name for r in user_roles],
        "role_permissions": role_permissions,
        "effective_permissions": [p.name for p in effective_permissions],
        "permission_count": len(effective_permissions)
    }

# Permission-Check mit Debugging
async def check_permission_with_debug(user_id: str, permission: str) -> Dict[str, Any]:
    """Prüft Berechtigung mit Debug-Informationen."""

    has_permission = await rbac_service.check_permission(user_id, permission)

    debug_info = await debug_user_permissions(user_id)
    debug_info.update({
        "requested_permission": permission,
        "has_permission": has_permission,
        "permission_source": None
    })

    # Quelle der Berechtigung finden
    if has_permission:
        for role, permissions in debug_info["role_permissions"].items():
            if permission in permissions:
                debug_info["permission_source"] = role
                break

    return debug_info

# security/event_detector.py
from dataclasses import dataclass
from typing import List, Dict, Any
from enum import Enum

class SecurityEventType(Enum):
    FAILED_LOGIN = "failed_login"
    SUSPICIOUS_ACTIVITY = "suspicious_activity"
    PRIVILEGE_ESCALATION = "privilege_escalation"
    DATA_ACCESS_VIOLATION = "data_access_violation"
    RATE_LIMIT_EXCEEDED = "rate_limit_exceeded"

@dataclass
class SecurityEvent:
    event_type: SecurityEventType
    user_id: Optional[str]
    ip_address: str
    user_agent: str
    timestamp: datetime
    details: Dict[str, Any]
    severity: str  # low, medium, high, critical

class SecurityEventDetector:
    """Erkennt Security-Events."""

    def __init__(self):
        self.failed_login_attempts: Dict[str, List[datetime]] = {}
        self.suspicious_ips: Set[str] = set()

    async def detect_failed_login_pattern(
        self,
        user_id: str,
        ip_address: str,
        user_agent: str
    ) -> Optional[SecurityEvent]:
        """Erkennt Failed-Login-Patterns."""

        now = datetime.utcnow()
        key = f"{user_id}:{ip_address}"

        # Failed-Login-Attempts tracken
        if key not in self.failed_login_attempts:
            self.failed_login_attempts[key] = []

        self.failed_login_attempts[key].append(now)

        # Alte Attempts entfernen (älter als 1 Stunde)
        cutoff = now - timedelta(hours=1)
        self.failed_login_attempts[key] = [
            attempt for attempt in self.failed_login_attempts[key]
            if attempt > cutoff
        ]

        # Threshold prüfen
        if len(self.failed_login_attempts[key]) >= 5:
            return SecurityEvent(
                event_type=SecurityEventType.FAILED_LOGIN,
                user_id=user_id,
                ip_address=ip_address,
                user_agent=user_agent,
                timestamp=now,
                details={
                    "failed_attempts": len(self.failed_login_attempts[key]),
                    "time_window": "1 hour"
                },
                severity="high"
            )

        return None

    async def detect_privilege_escalation(
        self,
        user_id: str,
        old_permissions: Set[str],
        new_permissions: Set[str],
        ip_address: str
    ) -> Optional[SecurityEvent]:
        """Erkennt Privilege-Escalation."""

        added_permissions = new_permissions - old_permissions

        # Kritische Berechtigungen prüfen
        critical_permissions = {
            "system:admin", "users:delete", "agents:delete",
            "system:config", "security:manage"
        }

        critical_added = added_permissions & critical_permissions

        if critical_added:
            return SecurityEvent(
                event_type=SecurityEventType.PRIVILEGE_ESCALATION,
                user_id=user_id,
                ip_address=ip_address,
                user_agent="",
                timestamp=datetime.utcnow(),
                details={
                    "added_permissions": list(added_permissions),
                    "critical_permissions": list(critical_added)
                },
                severity="critical"
            )

        return None

    async def detect_suspicious_activity(
        self,
        user_id: str,
        action: str,
        resource: str,
        ip_address: str,
        user_agent: str
    ) -> Optional[SecurityEvent]:
        """Erkennt verdächtige Aktivitäten."""

        # Ungewöhnliche IP-Adressen
        user_ips = await self.get_user_ip_history(user_id)
        if ip_address not in user_ips and len(user_ips) > 0:
            # Geolocation-Check
            user_country = await self.get_ip_country(user_ips[0])
            current_country = await self.get_ip_country(ip_address)

            if user_country != current_country:
                return SecurityEvent(
                    event_type=SecurityEventType.SUSPICIOUS_ACTIVITY,
                    user_id=user_id,
                    ip_address=ip_address,
                    user_agent=user_agent,
                    timestamp=datetime.utcnow(),
                    details={
                        "reason": "unusual_location",
                        "user_country": user_country,
                        "current_country": current_country,
                        "action": action,
                        "resource": resource
                    },
                    severity="medium"
                )

        # Ungewöhnliche Zeiten
        now = datetime.utcnow()
        if now.hour < 6 or now.hour > 22:  # Außerhalb Geschäftszeiten
            return SecurityEvent(
                event_type=SecurityEventType.SUSPICIOUS_ACTIVITY,
                user_id=user_id,
                ip_address=ip_address,
                user_agent=user_agent,
                timestamp=now,
                details={
                    "reason": "unusual_time",
                    "hour": now.hour,
                    "action": action,
                    "resource": resource
                },
                severity="low"
            )

        return None

# Security-Event-Handler
class SecurityEventHandler:
    """Behandelt Security-Events."""

    async def handle_security_event(self, event: SecurityEvent):
        """Behandelt Security-Event."""

        # Event protokollieren
        await self.log_security_event(event)

        # Automatische Reaktionen
        if event.severity == "critical":
            await self.handle_critical_event(event)
        elif event.severity == "high":
            await self.handle_high_severity_event(event)

        # Benachrichtigungen senden
        await self.send_security_notification(event)

    async def handle_critical_event(self, event: SecurityEvent):
        """Behandelt kritische Security-Events."""

        if event.event_type == SecurityEventType.PRIVILEGE_ESCALATION:
            # Benutzer temporär sperren
            await self.temporarily_lock_user(event.user_id, duration_minutes=30)

            # Admin-Benachrichtigung
            await self.notify_security_team(event)

        elif event.event_type == SecurityEventType.DATA_ACCESS_VIOLATION:
            # Zugriff protokollieren und blockieren
            await self.block_ip_address(event.ip_address, duration_hours=24)

    async def handle_high_severity_event(self, event: SecurityEvent):
        """Behandelt High-Severity-Events."""

        if event.event_type == SecurityEventType.FAILED_LOGIN:
            # IP-Adresse temporär blockieren
            await self.rate_limit_ip(event.ip_address, duration_minutes=15)

            # Benutzer über verdächtige Aktivität informieren
            await self.notify_user_suspicious_activity(event.user_id, event)

# security/compliance_monitor.py
class ComplianceMonitor:
    """Überwacht Security-Compliance."""

    async def check_gdpr_compliance(self) -> Dict[str, bool]:
        """Prüft GDPR-Compliance."""

        checks = {
            "data_encryption": await self.check_data_encryption(),
            "access_logging": await self.check_access_logging(),
            "data_retention": await self.check_data_retention_policy(),
            "user_consent": await self.check_user_consent_tracking(),
            "data_portability": await self.check_data_portability_support(),
            "right_to_erasure": await self.check_erasure_capability()
        }

        return checks

    async def check_security_headers(self) -> Dict[str, bool]:
        """Prüft Security-Headers."""

        # Test-Request an API
        async with aiohttp.ClientSession() as session:
            async with session.get("http://localhost:8000/health") as response:
                headers = response.headers

                return {
                    "strict_transport_security": "Strict-Transport-Security" in headers,
                    "content_security_policy": "Content-Security-Policy" in headers,
                    "x_frame_options": "X-Frame-Options" in headers,
                    "x_content_type_options": "X-Content-Type-Options" in headers,
                    "x_xss_protection": "X-XSS-Protection" in headers,
                    "referrer_policy": "Referrer-Policy" in headers
                }

    async def check_password_policy_compliance(self) -> Dict[str, Any]:
        """Prüft Password-Policy-Compliance."""

        # Schwache Passwörter in Database prüfen
        weak_passwords = await self.find_weak_passwords()

        # Password-Policy-Einstellungen prüfen
        policy = await self.get_password_policy()

        return {
            "min_length_enforced": policy.min_length >= 8,
            "complexity_enforced": policy.require_special_chars and policy.require_numbers,
            "weak_passwords_count": len(weak_passwords),
            "password_expiry_enabled": policy.max_age_days is not None,
            "password_history_enabled": policy.history_count > 0
        }