Security & Access Boundaries

import logging
import json
from datetime import datetime
from typing import Dict, Optional, List, Any
from dataclasses import dataclass, field
from enum import Enum

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)

class AccessLevel(Enum):
    READ_ONLY = "read"
    EDIT = "edit"
    ADMIN = "admin"
    AUDIT = "audit"

class SensitivityLevel(Enum):
    STANDARD = "standard"
    CONFIDENTIAL = "confidential"
    RESTRICTED = "restricted"

@dataclass
class LeaseContext:
    lease_id: str
    tenant_id: str
    portfolio_id: str
    classification_tags: List[str]
    sensitivity_level: SensitivityLevel
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class UserContext:
    user_id: str
    role: AccessLevel
    assigned_tenant_ids: List[str]
    assigned_portfolio_ids: List[str]

class AccessBoundaryError(Exception):
    """Custom exception for access boundary violations."""
    pass

class AccessBoundaryValidator:
    def __init__(self, role_permissions: Dict[AccessLevel, List[str]]):
        self.role_permissions = role_permissions
        self._audit_log: List[Dict[str, Any]] = []

    def validate_access(self, user: UserContext, lease: LeaseContext) -> bool:
        """
        Validates whether a user's role and assigned scopes grant access to a specific lease.
        Raises AccessBoundaryError on violation.
        """
        if lease.sensitivity_level == SensitivityLevel.RESTRICTED and user.role not in (AccessLevel.ADMIN, AccessLevel.AUDIT):
            self._log_event(user, lease, "DENIED", "Restricted lease requires ADMIN or AUDIT role.")
            raise AccessBoundaryError("Insufficient role for restricted lease data.")

        if lease.tenant_id not in user.assigned_tenant_ids:
            self._log_event(user, lease, "DENIED", "Tenant boundary violation.")
            raise AccessBoundaryError("User lacks tenant-level access.")

        if lease.portfolio_id not in user.assigned_portfolio_ids:
            self._log_event(user, lease, "DENIED", "Portfolio boundary violation.")
            raise AccessBoundaryError("User lacks portfolio-level access.")

        self._log_event(user, lease, "GRANTED", "Access validated successfully.")
        return True

    def apply_field_redaction(self, lease_metadata: Dict[str, Any], user_role: AccessLevel) -> Dict[str, Any]:
        """
        Redacts sensitive fields based on user role.
        Non-admin/audit roles cannot view raw financial or PII fields.
        """
        restricted_fields = {"rent_amount", "security_deposit", "tenant_ssn", "guarantor_info"}
        redacted = lease_metadata.copy()

        if user_role not in (AccessLevel.ADMIN, AccessLevel.AUDIT):
            for field_name in restricted_fields:
                if field_name in redacted:
                    redacted[field_name] = "[REDACTED]"
                    logging.debug(f"Redacted field: {field_name} for role {user_role.value}")

        return redacted

    def _log_event(self, user: UserContext, lease: LeaseContext, status: str, reason: str) -> None:
        audit_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user.user_id,
            "user_role": user.role.value,
            "lease_id": lease.lease_id,
            "tenant_id": lease.tenant_id,
            "status": status,
            "reason": reason
        }
        self._audit_log.append(audit_entry)
        logging.info(f"AUDIT | {status} | {user.user_id} -> {lease.lease_id} | {reason}")

    def export_audit_log(self) -> str:
        return json.dumps(self._audit_log, indent=2)


# --- Production Usage Example ---
if __name__ == "__main__":
    # Define role-to-permission mapping (can be loaded from config/DB)
    ROLE_PERMISSIONS = {
        AccessLevel.READ_ONLY: ["read_lease", "view_clauses"],
        AccessLevel.EDIT: ["read_lease", "edit_lease", "update_clauses"],
        AccessLevel.ADMIN: ["*"],
        AccessLevel.AUDIT: ["read_lease", "view_clauses", "view_audit_logs"]
    }

    validator = AccessBoundaryValidator(ROLE_PERMISSIONS)

    # Simulated lease context
    sample_lease = LeaseContext(
        lease_id="LS-2024-8891",
        tenant_id="TEN-442",
        portfolio_id="PF-01",
        classification_tags=["retail", "triple_net", "renewal_pending"],
        sensitivity_level=SensitivityLevel.CONFIDENTIAL,
        metadata={
            "rent_amount": 125000.00,
            "tenant_ssn": "XXX-XX-1234",
            "commencement_date": "2024-01-01",
            "guarantor_info": "Parent Corp LLC"
        }
    )

    # Simulated user context
    ops_manager = UserContext(
        user_id="USR-991",
        role=AccessLevel.EDIT,
        assigned_tenant_ids=["TEN-442", "TEN-501"],
        assigned_portfolio_ids=["PF-01", "PF-02"]
    )

    try:
        # 1. Validate boundary
        validator.validate_access(ops_manager, sample_lease)

        # 2. Apply redaction before downstream processing
        safe_metadata = validator.apply_field_redaction(sample_lease.metadata, ops_manager.role)
        print(f"Processed Metadata: {json.dumps(safe_metadata, indent=2)}")

    except AccessBoundaryError as e:
        logging.error(f"Pipeline halted: {e}")
    finally:
        print("\nAudit Log Export:")
        print(validator.export_audit_log())