# -*- coding: utf-8 -*- """ Analytics Permissions ==================== Custom permission classes for the Adtlas Analytics module. Provides fine-grained access control for analytics data, reports, and administrative operations. Permission Categories: - Data Access Permissions: Control access to analytics data - Report Permissions: Control report generation and viewing - Administrative Permissions: Control admin operations - API Permissions: Control API endpoint access - Export Permissions: Control data export capabilities - Real-time Permissions: Control real-time data access Key Features: - Role-based access control (RBAC) - Object-level permissions - Time-based access restrictions - IP-based access control - Rate limiting integration - Audit trail logging - Dynamic permission evaluation - Multi-tenant support Permission Classes: - AnalyticsDataPermission: Base analytics data access - ReportPermission: Report generation and viewing - AdminAnalyticsPermission: Administrative operations - ExportPermission: Data export capabilities - RealTimePermission: Real-time data access - CampaignAnalyticsPermission: Campaign-specific analytics - ChannelAnalyticsPermission: Channel-specific analytics - AdvertiserAnalyticsPermission: Advertiser-specific analytics Usage Examples: - Apply to ViewSets for API protection - Use in templates for UI element control - Integrate with middleware for request filtering - Combine with Django's built-in permissions Author: Adtlas Development Team Version: 1.0.0 Last Updated: 2024 """ import logging from datetime import datetime, timedelta from typing import Dict, List, Optional, Any from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission, Group from django.core.cache import cache from django.utils import timezone from django.conf import settings from django.db.models import Q from rest_framework import permissions from rest_framework.request import Request from rest_framework.views import APIView from apps.campaigns.models import Campaign from apps.channels.models import Channel from apps.advertisers.models import Brand from .models import ( SfrAnalytics, BouyguesAnalytics, Impression, VastResponse, PerformanceMetric, AnalyticsReport ) from .utils import get_client_ip, hash_sensitive_data # Configure logging logger = logging.getLogger(__name__) # Get User model User = get_user_model() class BaseAnalyticsPermission(permissions.BasePermission): """ Base permission class for analytics operations. Provides common functionality for all analytics permissions: - User authentication validation - Rate limiting checks - IP-based access control - Time-based restrictions - Audit logging Attributes: required_permissions: List of required Django permissions required_groups: List of required user groups rate_limit_key: Cache key for rate limiting max_requests_per_hour: Maximum requests per hour allowed_ips: List of allowed IP addresses (optional) time_restrictions: Time-based access restrictions """ required_permissions: List[str] = [] required_groups: List[str] = [] rate_limit_key: str = 'analytics_access' max_requests_per_hour: int = 1000 allowed_ips: Optional[List[str]] = None time_restrictions: Optional[Dict[str, Any]] = None def has_permission(self, request: Request, view: APIView) -> bool: """ Check if user has permission to access the view. Args: request: HTTP request object view: API view object Returns: Boolean indicating if permission is granted """ try: # Check authentication if not request.user or not request.user.is_authenticated: logger.warning(f"Unauthenticated access attempt to {view.__class__.__name__}") return False # Check if user is active if not request.user.is_active: logger.warning(f"Inactive user access attempt: {request.user.username}") return False # Check rate limiting if not self._check_rate_limit(request): logger.warning(f"Rate limit exceeded for user: {request.user.username}") return False # Check IP restrictions if not self._check_ip_restrictions(request): logger.warning(f"IP restriction violation: {get_client_ip(request)}") return False # Check time restrictions if not self._check_time_restrictions(request): logger.warning(f"Time restriction violation for user: {request.user.username}") return False # Check Django permissions if not self._check_django_permissions(request.user): logger.warning(f"Django permission check failed for user: {request.user.username}") return False # Check group membership if not self._check_group_membership(request.user): logger.warning(f"Group membership check failed for user: {request.user.username}") return False # Log successful access self._log_access(request, view, granted=True) return True except Exception as e: logger.error(f"Error checking analytics permission: {e}") return False def has_object_permission(self, request: Request, view: APIView, obj: Any) -> bool: """ Check if user has permission to access specific object. Args: request: HTTP request object view: API view object obj: Object being accessed Returns: Boolean indicating if permission is granted """ try: # First check general permission if not self.has_permission(request, view): return False # Check object-specific permissions if not self._check_object_permissions(request.user, obj): logger.warning(f"Object permission check failed for user: {request.user.username}") return False # Log successful object access self._log_object_access(request, view, obj, granted=True) return True except Exception as e: logger.error(f"Error checking object permission: {e}") return False def _check_rate_limit(self, request: Request) -> bool: """ Check rate limiting for user. Args: request: HTTP request object Returns: Boolean indicating if rate limit is not exceeded """ try: cache_key = f"{self.rate_limit_key}_{request.user.id}_{datetime.now().hour}" current_count = cache.get(cache_key, 0) if current_count >= self.max_requests_per_hour: return False # Increment counter cache.set(cache_key, current_count + 1, timeout=3600) # 1 hour return True except Exception as e: logger.error(f"Error checking rate limit: {e}") return True # Allow access on error def _check_ip_restrictions(self, request: Request) -> bool: """ Check IP-based access restrictions. Args: request: HTTP request object Returns: Boolean indicating if IP is allowed """ try: if not self.allowed_ips: return True # No IP restrictions client_ip = get_client_ip(request) return client_ip in self.allowed_ips except Exception as e: logger.error(f"Error checking IP restrictions: {e}") return True # Allow access on error def _check_time_restrictions(self, request: Request) -> bool: """ Check time-based access restrictions. Args: request: HTTP request object Returns: Boolean indicating if current time is allowed """ try: if not self.time_restrictions: return True # No time restrictions current_time = timezone.now() # Check allowed hours if 'allowed_hours' in self.time_restrictions: allowed_hours = self.time_restrictions['allowed_hours'] if current_time.hour not in allowed_hours: return False # Check allowed days if 'allowed_days' in self.time_restrictions: allowed_days = self.time_restrictions['allowed_days'] if current_time.weekday() not in allowed_days: return False return True except Exception as e: logger.error(f"Error checking time restrictions: {e}") return True # Allow access on error def _check_django_permissions(self, user: User) -> bool: """ Check Django permissions. Args: user: User object Returns: Boolean indicating if user has required permissions """ try: if not self.required_permissions: return True # No specific permissions required for permission in self.required_permissions: if not user.has_perm(permission): return False return True except Exception as e: logger.error(f"Error checking Django permissions: {e}") return False def _check_group_membership(self, user: User) -> bool: """ Check user group membership. Args: user: User object Returns: Boolean indicating if user is in required groups """ try: if not self.required_groups: return True # No specific groups required user_groups = user.groups.values_list('name', flat=True) for group in self.required_groups: if group not in user_groups: return False return True except Exception as e: logger.error(f"Error checking group membership: {e}") return False def _check_object_permissions(self, user: User, obj: Any) -> bool: """ Check object-specific permissions. Args: user: User object obj: Object being accessed Returns: Boolean indicating if user can access object """ # Base implementation - override in subclasses return True def _log_access(self, request: Request, view: APIView, granted: bool) -> None: """ Log access attempt. Args: request: HTTP request object view: API view object granted: Whether access was granted """ try: log_data = { 'user_id': request.user.id if request.user.is_authenticated else None, 'username': request.user.username if request.user.is_authenticated else 'anonymous', 'view': view.__class__.__name__, 'method': request.method, 'ip_address': get_client_ip(request), 'user_agent': request.META.get('HTTP_USER_AGENT', ''), 'granted': granted, 'timestamp': timezone.now().isoformat() } # Store in cache for audit trail cache_key = f"access_log_{timezone.now().timestamp()}" cache.set(cache_key, log_data, timeout=86400) # 24 hours if granted: logger.info(f"Access granted: {log_data}") else: logger.warning(f"Access denied: {log_data}") except Exception as e: logger.error(f"Error logging access: {e}") def _log_object_access(self, request: Request, view: APIView, obj: Any, granted: bool) -> None: """ Log object access attempt. Args: request: HTTP request object view: API view object obj: Object being accessed granted: Whether access was granted """ try: log_data = { 'user_id': request.user.id, 'username': request.user.username, 'view': view.__class__.__name__, 'object_type': obj.__class__.__name__, 'object_id': getattr(obj, 'id', None), 'method': request.method, 'granted': granted, 'timestamp': timezone.now().isoformat() } # Store in cache for audit trail cache_key = f"object_access_log_{timezone.now().timestamp()}" cache.set(cache_key, log_data, timeout=86400) # 24 hours if granted: logger.info(f"Object access granted: {log_data}") else: logger.warning(f"Object access denied: {log_data}") except Exception as e: logger.error(f"Error logging object access: {e}") class AnalyticsDataPermission(BaseAnalyticsPermission): """ Permission class for general analytics data access. Controls access to analytics data viewing and basic operations. Requires 'analytics.view_analytics' permission. Features: - Basic analytics data access - Read-only operations - Standard rate limiting - Audit logging """ required_permissions = ['analytics.view_analytics'] required_groups = ['Analytics Users'] rate_limit_key = 'analytics_data_access' max_requests_per_hour = 500 def _check_object_permissions(self, user: User, obj: Any) -> bool: """ Check object-specific permissions for analytics data. Args: user: User object obj: Analytics object being accessed Returns: Boolean indicating if user can access object """ try: # Check if user can access campaign data if hasattr(obj, 'campaign_id') and obj.campaign_id: if not self._can_access_campaign(user, obj.campaign_id): return False # Check if user can access channel data if hasattr(obj, 'channel_id') and obj.channel_id: if not self._can_access_channel(user, obj.channel_id): return False # Check if user can access advertiser data if hasattr(obj, 'advertiser_id') and obj.advertiser_id: if not self._can_access_advertiser(user, obj.advertiser_id): return False return True except Exception as e: logger.error(f"Error checking analytics object permissions: {e}") return False def _can_access_campaign(self, user: User, campaign_id: int) -> bool: """ Check if user can access campaign data. Args: user: User object campaign_id: Campaign ID Returns: Boolean indicating if user can access campaign """ try: # Superusers can access all campaigns if user.is_superuser: return True # Check if user is associated with the campaign campaign = Campaign.objects.get(id=campaign_id) # Check if user is the campaign owner or has access if hasattr(campaign, 'created_by') and campaign.created_by == user: return True # Check if user's advertiser matches campaign advertiser if hasattr(user, 'advertiser') and hasattr(campaign, 'advertiser'): if user.advertiser == campaign.advertiser: return True # Check if user has explicit campaign access if user.has_perm('campaigns.view_campaign', campaign): return True return False except Campaign.DoesNotExist: return False except Exception as e: logger.error(f"Error checking campaign access: {e}") return False def _can_access_channel(self, user: User, channel_id: int) -> bool: """ Check if user can access channel data. Args: user: User object channel_id: Channel ID Returns: Boolean indicating if user can access channel """ try: # Superusers can access all channels if user.is_superuser: return True # Check if user has channel access permission if user.has_perm('channels.view_channel'): return True # Check if user is in channel management group if user.groups.filter(name='Channel Managers').exists(): return True return False except Exception as e: logger.error(f"Error checking channel access: {e}") return False def _can_access_advertiser(self, user: User, advertiser_id: int) -> bool: """ Check if user can access advertiser data. Args: user: User object advertiser_id: Advertiser ID Returns: Boolean indicating if user can access advertiser """ try: # Superusers can access all advertisers if user.is_superuser: return True # Check if user belongs to the advertiser if hasattr(user, 'advertiser') and user.advertiser.id == advertiser_id: return True # Check if user has advertiser access permission if user.has_perm('advertisers.view_advertiser'): return True return False except Exception as e: logger.error(f"Error checking advertiser access: {e}") return False class ReportPermission(BaseAnalyticsPermission): """ Permission class for analytics report operations. Controls access to report generation, viewing, and management. Requires 'analytics.view_report' permission for viewing, 'analytics.add_report' for generation. Features: - Report generation access - Report viewing permissions - Export capabilities - Enhanced rate limiting """ required_permissions = ['analytics.view_analyticsreport'] required_groups = ['Analytics Users'] rate_limit_key = 'analytics_report_access' max_requests_per_hour = 100 # Lower limit for reports def has_permission(self, request: Request, view: APIView) -> bool: """ Check report-specific permissions. Args: request: HTTP request object view: API view object Returns: Boolean indicating if permission is granted """ # Check base permissions first if not super().has_permission(request, view): return False # Check method-specific permissions if request.method == 'POST': # Report generation requires add permission if not request.user.has_perm('analytics.add_analyticsreport'): return False elif request.method in ['PUT', 'PATCH']: # Report modification requires change permission if not request.user.has_perm('analytics.change_analyticsreport'): return False elif request.method == 'DELETE': # Report deletion requires delete permission if not request.user.has_perm('analytics.delete_analyticsreport'): return False return True def _check_object_permissions(self, user: User, obj: AnalyticsReport) -> bool: """ Check object-specific permissions for reports. Args: user: User object obj: AnalyticsReport object being accessed Returns: Boolean indicating if user can access report """ try: # Superusers can access all reports if user.is_superuser: return True # Users can access their own reports if obj.generated_by == user: return True # Check if user can access report's campaign data if obj.filters and 'campaign_id' in obj.filters: campaign_id = obj.filters['campaign_id'] if not self._can_access_campaign(user, campaign_id): return False # Check if user can access report's advertiser data if obj.filters and 'advertiser_id' in obj.filters: advertiser_id = obj.filters['advertiser_id'] if not self._can_access_advertiser(user, advertiser_id): return False # Check if user is in report sharing group if user.groups.filter(name='Report Viewers').exists(): return True return False except Exception as e: logger.error(f"Error checking report object permissions: {e}") return False class AdminAnalyticsPermission(BaseAnalyticsPermission): """ Permission class for analytics administrative operations. Controls access to administrative functions like: - System configuration - User management - Data management - Performance monitoring Features: - Administrative access control - Enhanced security checks - Audit logging - IP restrictions """ required_permissions = [ 'analytics.view_analytics', 'analytics.add_analytics', 'analytics.change_analytics', 'analytics.delete_analytics' ] required_groups = ['Analytics Administrators'] rate_limit_key = 'analytics_admin_access' max_requests_per_hour = 200 def has_permission(self, request: Request, view: APIView) -> bool: """ Check administrative permissions. Args: request: HTTP request object view: API view object Returns: Boolean indicating if permission is granted """ # Check base permissions first if not super().has_permission(request, view): return False # Additional checks for admin operations if not (request.user.is_staff or request.user.is_superuser): return False # Check if user is in admin group if not request.user.groups.filter(name='Analytics Administrators').exists(): if not request.user.is_superuser: return False return True class ExportPermission(BaseAnalyticsPermission): """ Permission class for data export operations. Controls access to data export functionality including: - CSV exports - Excel exports - API data exports - Bulk data downloads Features: - Export-specific rate limiting - Data sensitivity checks - Export audit logging - Size limitations """ required_permissions = ['analytics.export_data'] required_groups = ['Data Exporters'] rate_limit_key = 'analytics_export_access' max_requests_per_hour = 20 # Very limited for exports def has_permission(self, request: Request, view: APIView) -> bool: """ Check export-specific permissions. Args: request: HTTP request object view: API view object Returns: Boolean indicating if permission is granted """ # Check base permissions first if not super().has_permission(request, view): return False # Check export-specific rate limiting if not self._check_export_rate_limit(request): return False # Check export size limitations if not self._check_export_size_limit(request): return False return True def _check_export_rate_limit(self, request: Request) -> bool: """ Check export-specific rate limiting. Args: request: HTTP request object Returns: Boolean indicating if export rate limit is not exceeded """ try: # Daily export limit daily_key = f"export_daily_{request.user.id}_{datetime.now().date()}" daily_count = cache.get(daily_key, 0) if daily_count >= 5: # Max 5 exports per day return False # Increment daily counter cache.set(daily_key, daily_count + 1, timeout=86400) # 24 hours return True except Exception as e: logger.error(f"Error checking export rate limit: {e}") return True # Allow on error def _check_export_size_limit(self, request: Request) -> bool: """ Check export size limitations. Args: request: HTTP request object Returns: Boolean indicating if export size is within limits """ try: # Check if request specifies date range date_from = request.query_params.get('date_from') date_to = request.query_params.get('date_to') if date_from and date_to: from datetime import datetime start_date = datetime.fromisoformat(date_from.replace('Z', '+00:00')) end_date = datetime.fromisoformat(date_to.replace('Z', '+00:00')) # Limit to 90 days of data if (end_date - start_date).days > 90: return False return True except Exception as e: logger.error(f"Error checking export size limit: {e}") return True # Allow on error class RealTimePermission(BaseAnalyticsPermission): """ Permission class for real-time analytics access. Controls access to real-time data streams and live updates. Requires special permissions due to resource intensity. Features: - Real-time data access - WebSocket permissions - Live dashboard access - Streaming data permissions """ required_permissions = ['analytics.view_realtime'] required_groups = ['Real-time Users'] rate_limit_key = 'analytics_realtime_access' max_requests_per_hour = 2000 # Higher limit for real-time def has_permission(self, request: Request, view: APIView) -> bool: """ Check real-time specific permissions. Args: request: HTTP request object view: API view object Returns: Boolean indicating if permission is granted """ # Check base permissions first if not super().has_permission(request, view): return False # Check concurrent connection limit if not self._check_concurrent_connections(request): return False return True def _check_concurrent_connections(self, request: Request) -> bool: """ Check concurrent connection limits for real-time access. Args: request: HTTP request object Returns: Boolean indicating if connection limit is not exceeded """ try: # Check user's concurrent connections connection_key = f"realtime_connections_{request.user.id}" current_connections = cache.get(connection_key, 0) # Limit to 3 concurrent connections per user if current_connections >= 3: return False # Increment connection count (will be decremented on disconnect) cache.set(connection_key, current_connections + 1, timeout=3600) return True except Exception as e: logger.error(f"Error checking concurrent connections: {e}") return True # Allow on error class CampaignAnalyticsPermission(AnalyticsDataPermission): """ Permission class for campaign-specific analytics. Extends AnalyticsDataPermission with campaign-specific checks. Users can only access analytics for campaigns they have access to. Features: - Campaign ownership validation - Advertiser association checks - Campaign status restrictions - Time-based access control """ def _check_object_permissions(self, user: User, obj: Any) -> bool: """ Check campaign-specific object permissions. Args: user: User object obj: Analytics object being accessed Returns: Boolean indicating if user can access object """ try: # Must have campaign_id to proceed if not hasattr(obj, 'campaign_id') or not obj.campaign_id: return False # Check campaign access if not self._can_access_campaign(user, obj.campaign_id): return False # Check campaign status restrictions if not self._check_campaign_status_access(user, obj.campaign_id): return False return True except Exception as e: logger.error(f"Error checking campaign analytics permissions: {e}") return False def _check_campaign_status_access(self, user: User, campaign_id: int) -> bool: """ Check if user can access campaign based on its status. Args: user: User object campaign_id: Campaign ID Returns: Boolean indicating if user can access campaign """ try: campaign = Campaign.objects.get(id=campaign_id) # Superusers can access all campaign statuses if user.is_superuser: return True # Regular users can't access draft campaigns they don't own if campaign.status == 'draft' and campaign.created_by != user: return False # Check if campaign is archived and user has archive access if campaign.status == 'archived': if not user.has_perm('campaigns.view_archived_campaign'): return False return True except Campaign.DoesNotExist: return False except Exception as e: logger.error(f"Error checking campaign status access: {e}") return False class ChannelAnalyticsPermission(AnalyticsDataPermission): """ Permission class for channel-specific analytics. Extends AnalyticsDataPermission with channel-specific checks. Controls access to channel performance data and metrics. Features: - Channel access validation - Geographic restrictions - Channel type permissions - Audience data protection """ def _check_object_permissions(self, user: User, obj: Any) -> bool: """ Check channel-specific object permissions. Args: user: User object obj: Analytics object being accessed Returns: Boolean indicating if user can access object """ try: # Must have channel_id to proceed if not hasattr(obj, 'channel_id') or not obj.channel_id: return False # Check channel access if not self._can_access_channel(user, obj.channel_id): return False # Check geographic restrictions if not self._check_geographic_access(user, obj.channel_id): return False return True except Exception as e: logger.error(f"Error checking channel analytics permissions: {e}") return False def _check_geographic_access(self, user: User, channel_id: int) -> bool: """ Check geographic access restrictions for channel. Args: user: User object channel_id: Channel ID Returns: Boolean indicating if user can access channel geographically """ try: # This would implement geographic restrictions # For now, return True (no restrictions) return True except Exception as e: logger.error(f"Error checking geographic access: {e}") return True class AdvertiserAnalyticsPermission(AnalyticsDataPermission): """ Permission class for advertiser-specific analytics. Extends AnalyticsDataPermission with advertiser-specific checks. Ensures users can only access analytics for their associated advertisers. Features: - Advertiser association validation - Multi-tenant data isolation - Advertiser hierarchy support - Brand-level permissions """ def _check_object_permissions(self, user: User, obj: Any) -> bool: """ Check advertiser-specific object permissions. Args: user: User object obj: Analytics object being accessed Returns: Boolean indicating if user can access object """ try: # Check advertiser access through various relationships if hasattr(obj, 'advertiser_id') and obj.advertiser_id: if not self._can_access_advertiser(user, obj.advertiser_id): return False # Check through campaign relationship elif hasattr(obj, 'campaign_id') and obj.campaign_id: campaign = Campaign.objects.get(id=obj.campaign_id) if hasattr(campaign, 'advertiser_id'): if not self._can_access_advertiser(user, campaign.advertiser_id): return False return True except Exception as e: logger.error(f"Error checking advertiser analytics permissions: {e}") return False # Permission utility functions def check_analytics_permission(user: User, permission_type: str, obj: Any = None) -> bool: """ Utility function to check analytics permissions programmatically. Args: user: User object permission_type: Type of permission to check obj: Optional object for object-level permissions Returns: Boolean indicating if permission is granted """ try: permission_classes = { 'data': AnalyticsDataPermission, 'report': ReportPermission, 'admin': AdminAnalyticsPermission, 'export': ExportPermission, 'realtime': RealTimePermission, 'campaign': CampaignAnalyticsPermission, 'channel': ChannelAnalyticsPermission, 'advertiser': AdvertiserAnalyticsPermission } permission_class = permission_classes.get(permission_type) if not permission_class: return False # Create mock request for permission check from django.http import HttpRequest request = HttpRequest() request.user = user request.method = 'GET' permission = permission_class() if obj: return permission.has_object_permission(request, None, obj) else: return permission.has_permission(request, None) except Exception as e: logger.error(f"Error checking analytics permission: {e}") return False def get_user_analytics_permissions(user: User) -> Dict[str, bool]: """ Get all analytics permissions for a user. Args: user: User object Returns: Dictionary of permission types and their status """ try: permissions = {} permission_types = [ 'data', 'report', 'admin', 'export', 'realtime', 'campaign', 'channel', 'advertiser' ] for perm_type in permission_types: permissions[perm_type] = check_analytics_permission(user, perm_type) return permissions except Exception as e: logger.error(f"Error getting user analytics permissions: {e}") return {} def create_analytics_permissions(): """ Create custom analytics permissions in the database. This function should be called during app initialization to ensure all required permissions exist. """ try: from django.contrib.contenttypes.models import ContentType # Get analytics content type analytics_ct = ContentType.objects.get_for_model(SfrAnalytics) # Define custom permissions custom_permissions = [ ('view_analytics', 'Can view analytics data'), ('export_data', 'Can export analytics data'), ('view_realtime', 'Can view real-time analytics'), ('manage_reports', 'Can manage analytics reports'), ('admin_analytics', 'Can administer analytics system') ] # Create permissions if they don't exist for codename, name in custom_permissions: permission, created = Permission.objects.get_or_create( codename=codename, content_type=analytics_ct, defaults={'name': name} ) if created: logger.info(f"Created analytics permission: {codename}") # Create analytics groups analytics_groups = [ 'Analytics Users', 'Analytics Administrators', 'Data Exporters', 'Real-time Users', 'Report Viewers', 'Channel Managers' ] for group_name in analytics_groups: group, created = Group.objects.get_or_create(name=group_name) if created: logger.info(f"Created analytics group: {group_name}") except Exception as e: logger.error(f"Error creating analytics permissions: {e}") # Export permission classes __all__ = [ 'BaseAnalyticsPermission', 'AnalyticsDataPermission', 'ReportPermission', 'AdminAnalyticsPermission', 'ExportPermission', 'RealTimePermission', 'CampaignAnalyticsPermission', 'ChannelAnalyticsPermission', 'AdvertiserAnalyticsPermission', 'check_analytics_permission', 'get_user_analytics_permissions', 'create_analytics_permissions' ]