Zero Trust Architecture: Beyond the Perimeter

# Zero trust identity verification
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta

@dataclass
class AccessContext:
    """Context for access decision."""
    user_id: str
    device_id: str
    application: str
    resource: str
    ip_address: str
    location: str
    time: datetime
    risk_score: float

class ZeroTrustAccessDecision:
    """Make access decisions based on context."""
    
    def __init__(self, policy_engine, risk_engine):
        self.policy = policy_engine
        self.risk = risk_engine
    
    def evaluate(self, context: AccessContext) -> dict:
        """Evaluate access request."""
        
        # Check identity
        identity_verified = self.verify_identity(context.user_id)
        if not identity_verified:
            return {"decision": "deny", "reason": "identity_not_verified"}
        
        # Check device posture
        device_trust = self.check_device_trust(context.device_id)
        if device_trust < 0.7:
            return {"decision": "deny", "reason": "untrusted_device"}
        
        # Calculate risk
        risk = self.risk.calculate(context)
        if risk > 0.8:
            # High risk - require additional verification
            return {
                "decision": "step_up",
                "reason": "high_risk",
                "action": "require_mfa"
            }
        
        # Check policy
        policy_decision = self.policy.evaluate(context)
        
        return {
            "decision": "allow" if policy_decision else "deny",
            "reason": "policy" if policy_decision else "policy_denied",
            "risk_score": risk,
            "device_trust": device_trust
        }
    
    def verify_identity(self, user_id: str) -> bool:
        """Verify user identity."""
        # Check MFA, session validity
        return True
    
    def check_device_trust(self, device_id: str) -> float:
        """Calculate device trust score."""
        # Check compliance, encryption, etc.
        return 0.9

# Policy definition
class AccessPolicy:
    """Define access policies."""
    
    def __init__(self):
        self.rules = []
    
    def add_rule(self, rule: dict):
        """Add access rule."""
        self.rules.append(rule)
    
    def evaluate(self, context: AccessContext) -> bool:
        """Evaluate context against rules."""
        for rule in self.rules:
            if self.matches(rule, context):
                return rule["effect"] == "allow"
        return False
    
    def matches(self, rule: dict, context: AccessContext) -> bool:
        """Check if rule matches context."""
        if "user" in rule and context.user_id not in rule["user"]:
            return False
        if "application" in rule and context.application != rule["application"]:
            return False
        if "time" in rule and not self.in_time_window(context.time, rule["time"]):
            return False
        return True

# Continuous authentication system
class ContinuousAuthenticator:
    """Monitor and re-verify access."""
    
    def __init__(self):
        self.session_tracker = {}
        self.risk_calculator = RiskCalculator()
    
    def start_session(self, user_id: str, device_id: str):
        """Start authenticated session."""
        session_id = self.create_session_id()
        
        self.session_tracker[session_id] = {
            "user_id": user_id,
            "device_id": device_id,
            "start_time": datetime.now(),
            "risk_score": 0.0,
            "last_verified": datetime.now()
        }
        
        return session_id
    
    def verify_continuously(self, session_id: str, action: str):
        """Verify action within session."""
        session = self.session_tracker.get(session_id)
        if not session:
            return False
        
        # Calculate risk based on action
        risk = self.risk_calculator.calculate(session, action)
        
        # Re-verify if needed
        if risk > 0.7 or self.time_since_verification(session) > timedelta(minutes=5):
            if not self.reverify(session):
                return False
        
        session["last_verified"] = datetime.now()
        session["risk_score"] = risk
        
        return True
    
    def time_since_verification(self, session: dict) -> timedelta:
        return datetime.now() - session["last_verified"]

# Kubernetes network policies for microsegmentation
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: payment-service-policy
spec:
  podSelector:
    matchLabels:
      app: payment-service
  policyTypes:
    - Ingress
    - Egress
  ingress:
    - from:
        - podSelector:
            matchLabels:
              app: api-gateway
        - podSelector:
            matchLabels:
              app: order-service
      ports:
        - protocol: TCP
          port: 8080
  egress:
    - to:
        - podSelector:
            matchLabels:
              app: database
      ports:
        - protocol: TCP
          port: 5432
    - to:
        - podSelector:
            matchLabels:
              app: redis
      ports:
        - protocol: TCP
          port: 6379
    - to:
        - namespaceSelector: {}
      ports:
        - protocol: TCP
          port: 443

# Istio authorization policy
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: payment-authz
spec:
  selector:
    matchLabels:
      app: payment-service
  action: ALLOW
  rules:
    - from:
        - source:
            principals:
              - "cluster.local/ns/default/sa/api-gateway"
              - "cluster.local/ns/checkout/sa/order-service"
      to:
        - operation:
            methods: ["GET", "POST"]
            paths: ["/api/v1/*"]
    - from:
        - source:
            principals:
              - "cluster.local/ns/monitoring/sa/prometheus"
      to:
        - operation:
            ports: ["15090", "15020"]

# Device trust evaluation
class DevicePostureEvaluator:
    """Evaluate device trust level."""
    
    def __init__(self):
        self.compliance_checks = {}
    
    async def evaluate(self, device_id: str) -> dict:
        """Evaluate device trust."""
        
        # Collect device signals
        signals = await self.collect_signals(device_id)
        
        # Evaluate compliance
        compliance = self.check_compliance(signals)
        
        # Check security state
        security = self.check_security_state(signals)
        
        # Calculate overall trust
        trust_score = (
            compliance["score"] * 0.5 +
            security["score"] * 0.5
        )
        
        return {
            "device_id": device_id,
            "trust_score": trust_score,
            "compliant": compliance["compliant"],
            "security_state": security["state"],
            "issues": compliance["issues"] + security["issues"]
        }
    
    async def collect_signals(self, device_id: str) -> dict:
        """Collect device signals."""
        return {
            "encryption": await self.get_disk_encryption(device_id),
            "os_version": await self.get_os_version(device_id),
            "patch_level": await self.get_patch_level(device_id),
            "screen_lock": await self.get_screen_lock_status(device_id),
            "mdm_enrolled": await self.check_mdm(device_id),
            "biometric_enabled": await self.check_biometric(device_id)
        }
    
    def check_compliance(self, signals: dict) -> dict:
        """Check compliance requirements."""
        issues = []
        
        if not signals["encryption"]:
            issues.append("Disk not encrypted")
        
        if signals["os_version"] < "14.0":
            issues.append("OS outdated")
        
        if not signals["screen_lock"]:
            issues.append("Screen lock not enabled")
        
        return {
            "score": 1.0 - (len(issues) * 0.2),
            "compliant": len(issues) == 0,
            "issues": issues
        }

# Software Defined Perimeter
class SoftwareDefinedPerimeter:
    """Implement SDP architecture."""
    
    def __init__(self):
        self.controller = SDPController()
        self.gateways = []
    
    def register_gateway(self, gateway: str):
        """Register SDP gateway."""
        self.gateways.append(gateway)
    
    def create_connection(self, user_id: str, device_id: str, resource: str) -> dict:
        """Create single-packet authorized connection."""
        
        # Verify identity
        if not self.verify_identity(user_id):
            raise AccessDenied("Identity verification failed")
        
        # Verify device
        if not self.verify_device(device_id):
            raise AccessDenied("Device not compliant")
        
        # Check authorization
        if not self.check_authorization(user_id, resource):
            raise AccessDenied("Not authorized for resource")
        
        # Get gateway
        gateway = self.select_gateway(resource)
        
        # Create mutual TLS connection
        connection = self.create_mtls_connection(
            user_id, device_id, gateway, resource
        )
        
        return connection
    
    def verify_identity(self, user_id: str) -> bool:
        """Verify user with MFA."""
        # Implementation
        return True
    
    def verify_device(self, device_id: str) -> bool:
        """Verify device compliance."""
        # Implementation
        return True
    
    def check_authorization(self, user_id: str, resource: str) -> bool:
        """Check user can access resource."""
        # Implementation
        return True
    
    def select_gateway(self, resource: str) -> str:
        """Select appropriate gateway."""
        return self.gateways[0]
    
    def create_mtls_connection(self, user_id: str, device_id: str, 
                               gateway: str, resource: str) -> dict:
        """Create mutual TLS connection."""
        # Implementation
        return {"connection_id": "conn_123", "gateway": gateway}

# Zero trust monitoring
class ZeroTrustMonitor:
    """Monitor for threats in zero trust environment."""
    
    def __init__(self):
        self.detectors = []
    
    def add_detector(self, detector):
        """Add threat detector."""
        self.detectors.append(detector)
    
    def analyze(self, event: dict) -> list:
        """Analyze event for threats."""
        threats = []
        
        for detector in self.detectors:
            result = detector.analyze(event)
            if result.detected:
                threats.append({
                    "type": detector.name,
                    "severity": result.severity,
                    "evidence": result.evidence
                })
        
        return threats
    
    def create_alert(self, threats: list):
        """Create alert for detected threats."""
        for threat in threats:
            if threat["severity"] == "critical":
                self.critical_alert(threat)
            elif threat["severity"] == "high":
                self.high_alert(threat)

class AnomalyDetector:
    """Detect anomalous behavior."""
    
    def analyze(self, event: dict) -> DetectionResult:
        """Analyze for anomalies."""
        # Check for unusual access patterns
        # Check for impossible travel
        # Check for privilege escalation
        
        return DetectionResult(detected=False, severity=None, evidence=[])

# Zero trust implementation phases
implementation_roadmap = {
    "phase_1": {
        "name": "Identify",
        "duration": "2-3 months",
        "activities": [
            "Inventory all assets",
            "Identify critical data",
            "Map data flows",
            "Define protection surfaces"
        ]
    },
    "phase_2": {
        "name": "Protect",
        "duration": "3-6 months",
        "activities": [
            "Implement strong authentication",
            "Deploy device trust",
            "Encrypt data in transit",
            "Apply least privilege"
        ]
    },
    "phase_3": {
        "name": "Detect",
        "duration": "2-3 months",
        "activities": [
            "Deploy monitoring",
            "Establish baselines",
            "Implement detection rules",
            "Create alerts"
        ]
    },
    "phase_4": {
        "name": "Respond",
        "duration": "2-3 months",
        "activities": [
            "Create response playbooks",
            "Automate responses",
            "Test procedures",
            "Train teams"
        ]
    }
}

# Prioritize protection
protection_priority = [
    "Crown jewel data",
    "Critical applications",
    "Privileged accounts",
    "Sensitive communications"
]

# Layered security
security_layers = [
    "Identity verification",
    "Device trust",
    "Application security",
    "Network segmentation",
    "Data protection",
    "Monitoring"
]

# Automated response playbooks
response_playbooks = {
    "untrusted_device": [
        "Revoke access",
        "Notify user",
        "Log event",
        "Create ticket"
    ],
    "anomalous_activity": [
        "Step up verification",
        "Limit access",
        "Alert security team",
        "Start investigation"
    ]
}