forensics_authorization.py

"""
Jenkins authorization and visibility mapping.

Provides:
- ACL parsing from config.xml
- Credential visibility detection (anonymous, authenticated, admin-only)
- Permission matrix analysis
- Security configuration assessment
"""

import xml.etree.ElementTree as ET
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Dict, List, Optional, Set


class VisibilityLevel(Enum):
    """Credential visibility levels."""
    ANONYMOUS = "Anonymous (Public)"
    AUTHENTICATED = "Authenticated Users"
    SPECIFIC_USERS = "Specific Users"
    ADMIN_ONLY = "Admin Only"
    SYSTEM_ONLY = "System Only"
    UNKNOWN = "Unknown"


@dataclass
class PermissionEntry:
    """Represents a single permission entry."""
    permission: str
    principal: str
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            "permission": self.permission,
            "principal": self.principal,
        }


@dataclass
class CredentialVisibility:
    """Visibility analysis for a credential."""
    credential_id: str
    credential_scope: str
    visibility_level: VisibilityLevel
    allowed_users: Set[str]
    allowed_groups: Set[str]
    is_global: bool
    is_anonymous_readable: bool
    risk_level: str
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            "credential_id": self.credential_id,
            "scope": self.credential_scope,
            "visibility": self.visibility_level.value,
            "allowed_users": list(self.allowed_users),
            "allowed_groups": list(self.allowed_groups),
            "is_global": self.is_global,
            "anonymous_readable": self.is_anonymous_readable,
            "risk_level": self.risk_level,
        }


@dataclass
class SecurityConfiguration:
    """Overall security configuration analysis."""
    authorization_strategy: str
    security_realm: str
    anonymous_read_enabled: bool
    signup_enabled: bool
    permissions: List[PermissionEntry]
    admin_users: Set[str]
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            "authorization_strategy": self.authorization_strategy,
            "security_realm": self.security_realm,
            "anonymous_read": self.anonymous_read_enabled,
            "signup_enabled": self.signup_enabled,
            "admin_users": list(self.admin_users),
            "permission_count": len(self.permissions),
        }


class AuthorizationMapper:
    """Authorization and visibility mapping engine."""
    
    def __init__(self, jenkins_home: Path):
        """
        Initialize authorization mapper.
        
        Args:
            jenkins_home: Path to Jenkins home directory
        """
        self.jenkins_home = Path(jenkins_home)
        self.config_file = self.jenkins_home / "config.xml"
        self.security_config: Optional[SecurityConfiguration] = None
    
    def load_security_configuration(self) -> Optional[SecurityConfiguration]:
        """
        Load security configuration from config.xml.
        
        Returns:
            Security configuration if successful
        """
        if not self.config_file.exists():
            return None
        
        try:
            tree = ET.parse(self.config_file)
            root = tree.getroot()
            
            # Get authorization strategy
            auth_strategy_elem = root.find('.//authorizationStrategy')
            auth_strategy = "Unknown"
            if auth_strategy_elem is not None:
                auth_strategy = auth_strategy_elem.get('class', 'Unknown')
            
            # Get security realm
            security_realm_elem = root.find('.//securityRealm')
            security_realm = "Unknown"
            if security_realm_elem is not None:
                security_realm = security_realm_elem.get('class', 'Unknown')
            
            # Check for anonymous read access
            anonymous_read = False
            for perm in root.findall('.//permission'):
                if perm.text and 'anonymous' in perm.text.lower() and 'read' in perm.text.lower():
                    anonymous_read = True
                    break
            
            # Check if signup is enabled
            signup_enabled = False
            allow_signup = root.find('.//allowsSignup')
            if allow_signup is not None and allow_signup.text == 'true':
                signup_enabled = True
            
            # Extract permissions
            permissions = []
            admin_users = set()
            
            # Parse ProjectMatrixAuthorizationStrategy permissions
            for perm_elem in root.findall('.//permission'):
                if perm_elem.text:
                    perm_text = perm_elem.text
                    # Format: permission:principal
                    if ':' in perm_text:
                        permission, principal = perm_text.split(':', 1)
                        permissions.append(PermissionEntry(permission, principal))
                        
                        # Check for admin permissions
                        if 'Administer' in permission or 'Admin' in permission:
                            admin_users.add(principal)
            
            self.security_config = SecurityConfiguration(
                authorization_strategy=auth_strategy,
                security_realm=security_realm,
                anonymous_read_enabled=anonymous_read,
                signup_enabled=signup_enabled,
                permissions=permissions,
                admin_users=admin_users,
            )
            
            return self.security_config
        
        except (ET.ParseError, IOError):
            return None
    
    def determine_credential_visibility(
        self, 
        credential_id: str,
        credential_scope: str = "GLOBAL",
        credential_domain: Optional[str] = None
    ) -> CredentialVisibility:
        """
        Determine visibility level for a credential.
        
        Args:
            credential_id: Credential ID
            credential_scope: Credential scope (GLOBAL, SYSTEM, etc.)
            credential_domain: Optional domain restriction
            
        Returns:
            Visibility analysis
        """
        # Load security config if not already loaded
        if not self.security_config:
            self.load_security_configuration()
        
        allowed_users = set()
        allowed_groups = set()
        visibility_level = VisibilityLevel.UNKNOWN
        is_anonymous_readable = False
        
        # If no security config, assume worst case (anonymous)
        if not self.security_config:
            visibility_level = VisibilityLevel.ANONYMOUS
            is_anonymous_readable = True
        else:
            # Check if anonymous read is enabled globally
            if self.security_config.anonymous_read_enabled:
                is_anonymous_readable = True
                visibility_level = VisibilityLevel.ANONYMOUS
            else:
                # Check scope
                if credential_scope == "SYSTEM":
                    visibility_level = VisibilityLevel.SYSTEM_ONLY
                elif credential_scope == "GLOBAL":
                    # Global credentials with authentication required
                    if self.security_config.permissions:
                        # Extract users with credential read permissions
                        for perm in self.security_config.permissions:
                            if 'Credentials' in perm.permission or 'Read' in perm.permission:
                                if perm.principal != 'anonymous':
                                    allowed_users.add(perm.principal)
                        
                        if allowed_users:
                            # Check if it's admin-only or specific users
                            if allowed_users == self.security_config.admin_users:
                                visibility_level = VisibilityLevel.ADMIN_ONLY
                            else:
                                visibility_level = VisibilityLevel.SPECIFIC_USERS
                        else:
                            # Default authenticated users
                            visibility_level = VisibilityLevel.AUTHENTICATED
                    else:
                        # No specific permissions defined
                        visibility_level = VisibilityLevel.AUTHENTICATED
                else:
                    # Project/folder scoped credentials
                    visibility_level = VisibilityLevel.SPECIFIC_USERS
        
        # Determine risk level
        risk_level = self._assess_risk_level(visibility_level, is_anonymous_readable)
        
        return CredentialVisibility(
            credential_id=credential_id,
            credential_scope=credential_scope,
            visibility_level=visibility_level,
            allowed_users=allowed_users,
            allowed_groups=allowed_groups,
            is_global=credential_scope == "GLOBAL",
            is_anonymous_readable=is_anonymous_readable,
            risk_level=risk_level,
        )
    
    def _assess_risk_level(
        self, 
        visibility: VisibilityLevel, 
        anonymous_readable: bool
    ) -> str:
        """
        Assess risk level based on visibility.
        
        Args:
            visibility: Visibility level
            anonymous_readable: Whether anonymous can read
            
        Returns:
            Risk level string
        """
        if anonymous_readable or visibility == VisibilityLevel.ANONYMOUS:
            return "CRITICAL"
        elif visibility == VisibilityLevel.AUTHENTICATED:
            return "HIGH"
        elif visibility == VisibilityLevel.SPECIFIC_USERS:
            return "MEDIUM"
        elif visibility == VisibilityLevel.ADMIN_ONLY:
            return "LOW"
        elif visibility == VisibilityLevel.SYSTEM_ONLY:
            return "MINIMAL"
        else:
            return "UNKNOWN"
    
    def parse_project_matrix_authorization(self) -> List[PermissionEntry]:
        """
        Parse ProjectMatrixAuthorizationStrategy configuration.
        
        Returns:
            List of permission entries
        """
        if not self.config_file.exists():
            return []
        
        permissions = []
        
        try:
            tree = ET.parse(self.config_file)
            root = tree.getroot()
            
            # Find authorization strategy section
            auth_strategy = root.find('.//authorizationStrategy')
            if auth_strategy is None:
                return permissions
            
            # Extract all permission elements
            for perm_elem in auth_strategy.findall('.//permission'):
                if perm_elem.text:
                    perm_text = perm_elem.text.strip()
                    if ':' in perm_text:
                        permission, principal = perm_text.split(':', 1)
                        permissions.append(PermissionEntry(
                            permission=permission.strip(),
                            principal=principal.strip()
                        ))
        
        except (ET.ParseError, IOError):
            pass
        
        return permissions
    
    def find_global_credential_visibility(self) -> List[CredentialVisibility]:
        """
        Analyze visibility of all global credentials.
        
        Returns:
            List of credential visibility analyses
        """
        visibilities = []
        
        credentials_file = self.jenkins_home / "credentials.xml"
        if not credentials_file.exists():
            return visibilities
        
        try:
            tree = ET.parse(credentials_file)
            root = tree.getroot()
            
            # Find all credential entries
            for domain_cred in root.findall('.//domainCredentials'):
                domain_elem = domain_cred.find('.//domain')
                domain_name = None
                if domain_elem is not None:
                    domain_name = domain_elem.findtext('.//name', '_')
                
                # Find all credentials in this domain
                for cred in domain_cred.findall('.//credentials/*'):
                    cred_id = cred.findtext('.//id')
                    cred_scope = cred.findtext('.//scope', 'GLOBAL')
                    
                    if cred_id:
                        visibility = self.determine_credential_visibility(
                            credential_id=cred_id,
                            credential_scope=cred_scope,
                            credential_domain=domain_name
                        )
                        visibilities.append(visibility)
        
        except (ET.ParseError, IOError):
            pass
        
        return visibilities
    
    def generate_visibility_report(
        self, 
        visibilities: List[CredentialVisibility]
    ) -> Dict:
        """
        Generate visibility summary report.
        
        Args:
            visibilities: List of credential visibilities
            
        Returns:
            Summary report
        """
        total = len(visibilities)
        anonymous_count = sum(1 for v in visibilities if v.is_anonymous_readable)
        global_count = sum(1 for v in visibilities if v.is_global)
        critical_risk = sum(1 for v in visibilities if v.risk_level == "CRITICAL")
        high_risk = sum(1 for v in visibilities if v.risk_level == "HIGH")
        
        visibility_breakdown = {}
        for v in visibilities:
            level = v.visibility_level.value
            visibility_breakdown[level] = visibility_breakdown.get(level, 0) + 1
        
        return {
            "total_credentials": total,
            "anonymous_readable": anonymous_count,
            "global_scope": global_count,
            "critical_risk": critical_risk,
            "high_risk": high_risk,
            "visibility_breakdown": visibility_breakdown,
            "anonymous_percentage": round(anonymous_count / total * 100, 1) if total > 0 else 0,
            "critical_percentage": round(critical_risk / total * 100, 1) if total > 0 else 0,
        }