CCPA Technical Compliance: Implementation Guide

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Set
from enum import Enum
from datetime import datetime, timedelta
from abc import ABC, abstractmethod
import json
import hashlib
 
class CCPARight(Enum):
    KNOW = "right_to_know"
    DELETE = "right_to_delete"
    OPT_OUT = "right_to_opt_out"
    NON_DISCRIMINATION = "right_to_non_discrimination"
    CORRECT = "right_to_correct"
    LIMIT_USE = "right_to_limit_use"
 
class RequestStatus(Enum):
    RECEIVED = "received"
    VERIFYING = "verifying"
    VERIFIED = "verified"
    PROCESSING = "processing"
    COMPLETED = "completed"
    DENIED = "denied"
 
class PersonalInfoCategory(Enum):
    IDENTIFIERS = "identifiers"
    COMMERCIAL_INFO = "commercial_information"
    BIOMETRIC = "biometric_information"
    INTERNET_ACTIVITY = "internet_activity"
    GEOLOCATION = "geolocation_data"
    AUDIO_VISUAL = "audio_visual_information"
    PROFESSIONAL_INFO = "professional_information"
    EDUCATION_INFO = "education_information"
    INFERENCES = "inferences"
    SENSITIVE_PI = "sensitive_personal_information"
 
@dataclass
class ConsumerRequest:
    request_id: str
    consumer_id: str
    right_type: CCPARight
    submitted_at: datetime
    deadline: datetime
    status: RequestStatus
    verification_method: str
    verified_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    response_sent_at: Optional[datetime] = None
    categories_requested: List[PersonalInfoCategory] = field(default_factory=list)
    denial_reason: Optional[str] = None
    notes: List[str] = field(default_factory=list)
 
@dataclass
class DataInventoryEntry:
    category: PersonalInfoCategory
    source: str
    business_purpose: str
    third_parties_shared: List[str]
    retention_period_days: int
    sold_or_shared: bool
 
class CCPAComplianceManager:
    def __init__(self):
        self.requests: Dict[str, ConsumerRequest] = {}
        self.data_inventory: Dict[PersonalInfoCategory, List[DataInventoryEntry]] = {}
        self.opted_out_consumers: Set[str] = set()
        self.deadline_days = 45
        self.extension_days = 45
        self._initialize_data_inventory()
 
    def _initialize_data_inventory(self):
        """Initialize data inventory categories."""
        self.data_inventory = {
            PersonalInfoCategory.IDENTIFIERS: [
                DataInventoryEntry(
                    category=PersonalInfoCategory.IDENTIFIERS,
                    source="direct_collection",
                    business_purpose="Account creation and management",
                    third_parties_shared=["Payment processor", "Email service"],
                    retention_period_days=1095,
                    sold_or_shared=False
                )
            ],
            PersonalInfoCategory.COMMERCIAL_INFO: [
                DataInventoryEntry(
                    category=PersonalInfoCategory.COMMERCIAL_INFO,
                    source="transactions",
                    business_purpose="Order fulfillment and customer service",
                    third_parties_shared=["Shipping provider", "Analytics"],
                    retention_period_days=2555,
                    sold_or_shared=False
                )
            ],
            PersonalInfoCategory.INTERNET_ACTIVITY: [
                DataInventoryEntry(
                    category=PersonalInfoCategory.INTERNET_ACTIVITY,
                    source="automatic_collection",
                    business_purpose="Website optimization and security",
                    third_parties_shared=["Analytics provider", "CDN"],
                    retention_period_days=365,
                    sold_or_shared=True
                )
            ]
        }
 
    def submit_request(
        self,
        consumer_id: str,
        right_type: CCPARight,
        categories: List[PersonalInfoCategory] = None,
        verification_method: str = "email"
    ) -> ConsumerRequest:
        """Submit a new consumer request."""
        request_id = self._generate_request_id()
 
        request = ConsumerRequest(
            request_id=request_id,
            consumer_id=consumer_id,
            right_type=right_type,
            submitted_at=datetime.utcnow(),
            deadline=datetime.utcnow() + timedelta(days=self.deadline_days),
            status=RequestStatus.RECEIVED,
            verification_method=verification_method,
            categories_requested=categories or list(PersonalInfoCategory)
        )
 
        self.requests[request_id] = request
        self._initiate_verification(request)
 
        return request
 
    def _initiate_verification(self, request: ConsumerRequest):
        """Initiate consumer verification process."""
        request.status = RequestStatus.VERIFYING
 
        if request.verification_method == "email":
            self._send_verification_email(request)
        elif request.verification_method == "account":
            self._verify_via_account(request)
 
    def verify_consumer(self, request_id: str, verification_code: str) -> bool:
        """Verify consumer identity."""
        request = self.requests.get(request_id)
        if not request:
            return False
 
        if self._validate_verification_code(request, verification_code):
            request.status = RequestStatus.VERIFIED
            request.verified_at = datetime.utcnow()
            self._process_request(request)
            return True
 
        return False
 
    def _process_request(self, request: ConsumerRequest):
        """Process verified request based on type."""
        request.status = RequestStatus.PROCESSING
 
        handlers = {
            CCPARight.KNOW: self._handle_know_request,
            CCPARight.DELETE: self._handle_delete_request,
            CCPARight.OPT_OUT: self._handle_opt_out_request,
            CCPARight.CORRECT: self._handle_correct_request,
            CCPARight.LIMIT_USE: self._handle_limit_use_request
        }
 
        handler = handlers.get(request.right_type)
        if handler:
            handler(request)
 
    def _handle_know_request(self, request: ConsumerRequest):
        """Handle Right to Know request."""
        report = {
            'request_id': request.request_id,
            'consumer_id': request.consumer_id,
            'generated_at': datetime.utcnow().isoformat(),
            'categories_collected': [],
            'specific_pieces': {},
            'sources': [],
            'business_purposes': [],
            'third_parties': [],
            'selling_disclosure': {}
        }
 
        for category in request.categories_requested:
            entries = self.data_inventory.get(category, [])
            for entry in entries:
                if category not in [c['category'] for c in report['categories_collected']]:
                    report['categories_collected'].append({
                        'category': category.value,
                        'description': self._get_category_description(category)
                    })
 
                report['sources'].append(entry.source)
                report['business_purposes'].append(entry.business_purpose)
                report['third_parties'].extend(entry.third_parties_shared)
 
                if entry.sold_or_shared:
                    report['selling_disclosure'][category.value] = {
                        'sold_or_shared': True,
                        'recipients': entry.third_parties_shared
                    }
 
        report['sources'] = list(set(report['sources']))
        report['business_purposes'] = list(set(report['business_purposes']))
        report['third_parties'] = list(set(report['third_parties']))
 
        self._send_know_response(request, report)
        request.status = RequestStatus.COMPLETED
        request.completed_at = datetime.utcnow()
 
    def _handle_delete_request(self, request: ConsumerRequest):
        """Handle Right to Delete request."""
        deletion_results = {
            'deleted': [],
            'retained': [],
            'errors': []
        }
 
        for category in request.categories_requested:
            can_delete, reason = self._can_delete_category(category, request.consumer_id)
 
            if can_delete:
                try:
                    self._delete_consumer_data(request.consumer_id, category)
                    deletion_results['deleted'].append(category.value)
                except Exception as e:
                    deletion_results['errors'].append({
                        'category': category.value,
                        'error': str(e)
                    })
            else:
                deletion_results['retained'].append({
                    'category': category.value,
                    'reason': reason
                })
 
        self._notify_service_providers_deletion(request.consumer_id)
 
        self._send_delete_response(request, deletion_results)
        request.status = RequestStatus.COMPLETED
        request.completed_at = datetime.utcnow()
 
    def _can_delete_category(self, category: PersonalInfoCategory, consumer_id: str) -> tuple:
        """Check if data category can be deleted."""
        exceptions = {
            'complete_transaction': "Required to complete transaction",
            'security': "Required for security purposes",
            'legal_obligation': "Required by law to retain",
            'internal_use': "Used solely for internal purposes"
        }
 
        if category == PersonalInfoCategory.COMMERCIAL_INFO:
            if self._has_pending_transactions(consumer_id):
                return (False, exceptions['complete_transaction'])
 
        if category == PersonalInfoCategory.IDENTIFIERS:
            if self._has_legal_hold(consumer_id):
                return (False, exceptions['legal_obligation'])
 
        return (True, None)
 
    def _handle_opt_out_request(self, request: ConsumerRequest):
        """Handle Right to Opt-Out of Sale request."""
        self.opted_out_consumers.add(request.consumer_id)
 
        self._update_advertising_preferences(request.consumer_id, opted_out=True)
        self._notify_data_partners_opt_out(request.consumer_id)
        self._set_gpc_preference(request.consumer_id, True)
 
        self._send_opt_out_confirmation(request)
        request.status = RequestStatus.COMPLETED
        request.completed_at = datetime.utcnow()
 
    def _handle_correct_request(self, request: ConsumerRequest):
        """Handle Right to Correct request."""
        # Implementation for data correction
        pass
 
    def _handle_limit_use_request(self, request: ConsumerRequest):
        """Handle Right to Limit Use of Sensitive PI."""
        # Implementation for limiting sensitive PI use
        pass
 
    def check_opt_out_status(self, consumer_id: str) -> bool:
        """Check if consumer has opted out of sale."""
        return consumer_id in self.opted_out_consumers
 
    def _get_category_description(self, category: PersonalInfoCategory) -> str:
        descriptions = {
            PersonalInfoCategory.IDENTIFIERS: "Real name, alias, postal address, email, IP address, account name",
            PersonalInfoCategory.COMMERCIAL_INFO: "Products purchased, obtained, or considered; purchasing histories",
            PersonalInfoCategory.INTERNET_ACTIVITY: "Browsing history, search history, interaction with websites or apps",
            PersonalInfoCategory.GEOLOCATION: "Physical location or movements",
            PersonalInfoCategory.PROFESSIONAL_INFO: "Employment-related information",
            PersonalInfoCategory.SENSITIVE_PI: "Government IDs, financial accounts, precise geolocation, racial/ethnic origin"
        }
        return descriptions.get(category, category.value)
 
    def _generate_request_id(self) -> str:
        return f"CCPA-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}-{hashlib.sha256(str(datetime.utcnow().timestamp()).encode()).hexdigest()[:8]}"
 
    # Placeholder methods
    def _send_verification_email(self, request): pass
    def _verify_via_account(self, request): pass
    def _validate_verification_code(self, request, code): return True
    def _send_know_response(self, request, report): pass
    def _send_delete_response(self, request, results): pass
    def _send_opt_out_confirmation(self, request): pass
    def _delete_consumer_data(self, consumer_id, category): pass
    def _notify_service_providers_deletion(self, consumer_id): pass
    def _has_pending_transactions(self, consumer_id): return False
    def _has_legal_hold(self, consumer_id): return False
    def _update_advertising_preferences(self, consumer_id, opted_out): pass
    def _notify_data_partners_opt_out(self, consumer_id): pass
    def _set_gpc_preference(self, consumer_id, value): pass

class GPCHandler:
    """Handle Global Privacy Control signals."""
 
    def __init__(self, ccpa_manager: CCPAComplianceManager):
        self.ccpa_manager = ccpa_manager
 
    def check_gpc_signal(self, request_headers: Dict[str, str]) -> bool:
        """Check for GPC signal in request headers."""
        gpc_header = request_headers.get('Sec-GPC', '')
        return gpc_header == '1'
 
    def process_gpc_request(self, request_headers: Dict, user_context: Dict) -> Dict:
        """Process GPC signal and apply opt-out."""
        if not self.check_gpc_signal(request_headers):
            return {'gpc_detected': False}
 
        consumer_id = user_context.get('consumer_id')
        if not consumer_id:
            return {
                'gpc_detected': True,
                'action': 'anonymous_opt_out',
                'cookies_blocked': True
            }
 
        if not self.ccpa_manager.check_opt_out_status(consumer_id):
            self.ccpa_manager.submit_request(
                consumer_id=consumer_id,
                right_type=CCPARight.OPT_OUT,
                verification_method="automatic_gpc"
            )
 
        return {
            'gpc_detected': True,
            'action': 'opt_out_applied',
            'consumer_id': consumer_id
        }
 
    def get_gpc_javascript(self) -> str:
        """Generate JavaScript for GPC detection."""
        return """
        (function() {
            // Check for GPC signal
            const gpcEnabled = navigator.globalPrivacyControl;
 
            if (gpcEnabled) {
                // Disable tracking cookies
                document.cookie = '_tracking=; expires=Thu, 01 Jan 1970 00:00:00 UTC; path=/;';
 
                // Notify backend of GPC
                fetch('/api/privacy/gpc', {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/json' },
                    body: JSON.stringify({ gpc: true })
                });
 
                // Disable third-party scripts
                window.disableTracking = true;
            }
        })();
        """

from flask import Flask, request, jsonify, render_template
 
app = Flask(__name__)
ccpa_manager = CCPAComplianceManager()
gpc_handler = GPCHandler(ccpa_manager)
 
@app.route('/do-not-sell', methods=['GET'])
def do_not_sell_page():
    """Render Do Not Sell My Personal Information page."""
    gpc_status = gpc_handler.check_gpc_signal(dict(request.headers))
 
    return render_template('do-not-sell.html', {
        'gpc_detected': gpc_status,
        'categories_sold': get_categories_sold(),
        'third_parties': get_third_party_recipients()
    })
 
@app.route('/api/privacy/opt-out', methods=['POST'])
def opt_out_api():
    """API endpoint for opt-out requests."""
    data = request.json
 
    consumer_id = data.get('consumer_id') or request.cookies.get('consumer_id')
 
    if not consumer_id:
        consumer_id = generate_anonymous_id()
        set_anonymous_opt_out(consumer_id)
 
    ccpa_request = ccpa_manager.submit_request(
        consumer_id=consumer_id,
        right_type=CCPARight.OPT_OUT,
        verification_method="web_form"
    )
 
    response = jsonify({
        'success': True,
        'request_id': ccpa_request.request_id,
        'status': ccpa_request.status.value
    })
 
    response.set_cookie('ccpa_opt_out', '1', max_age=31536000, secure=True, httponly=True)
 
    return response
 
@app.route('/api/privacy/gpc', methods=['POST'])
def gpc_api():
    """Handle GPC signal notification from client."""
    user_context = {
        'consumer_id': request.cookies.get('consumer_id'),
        'session_id': request.cookies.get('session_id')
    }
 
    result = gpc_handler.process_gpc_request(dict(request.headers), user_context)
 
    return jsonify(result)
 
@app.route('/api/privacy/request', methods=['POST'])
def privacy_request_api():
    """Submit CCPA privacy request."""
    data = request.json
 
    right_type = CCPARight(data.get('right_type', 'right_to_know'))
 
    categories = [
        PersonalInfoCategory(cat) for cat in data.get('categories', [])
    ] or None
 
    ccpa_request = ccpa_manager.submit_request(
        consumer_id=data['email'],
        right_type=right_type,
        categories=categories,
        verification_method=data.get('verification_method', 'email')
    )
 
    return jsonify({
        'request_id': ccpa_request.request_id,
        'status': ccpa_request.status.value,
        'deadline': ccpa_request.deadline.isoformat(),
        'verification_required': True
    })
 
@app.route('/api/privacy/verify', methods=['POST'])
def verify_request():
    """Verify consumer identity for request."""
    data = request.json
 
    success = ccpa_manager.verify_consumer(
        request_id=data['request_id'],
        verification_code=data['verification_code']
    )
 
    if success:
        ccpa_request = ccpa_manager.requests[data['request_id']]
        return jsonify({
            'verified': True,
            'status': ccpa_request.status.value
        })
 
    return jsonify({'verified': False, 'error': 'Invalid verification code'}), 400
 
def get_categories_sold():
    """Get list of PI categories that are sold/shared."""
    return [
        {
            'category': PersonalInfoCategory.INTERNET_ACTIVITY.value,
            'description': 'Browsing and search history',
            'recipients': ['Advertising partners', 'Analytics providers']
        }
    ]
 
def get_third_party_recipients():
    """Get list of third parties receiving PI."""
    return [
        {'name': 'Advertising Network A', 'purpose': 'Targeted advertising'},
        {'name': 'Analytics Provider B', 'purpose': 'Website analytics'}
    ]
 
def generate_anonymous_id():
    import uuid
    return f"anon_{uuid.uuid4().hex}"
 
def set_anonymous_opt_out(consumer_id):
    ccpa_manager.opted_out_consumers.add(consumer_id)

class PrivacyNoticeGenerator:
    """Generate CCPA-compliant privacy notices."""
 
    def __init__(self, ccpa_manager: CCPAComplianceManager):
        self.ccpa_manager = ccpa_manager
 
    def generate_privacy_policy_section(self) -> Dict:
        """Generate CCPA section for privacy policy."""
        return {
            'title': 'California Privacy Rights (CCPA)',
            'sections': [
                {
                    'heading': 'Categories of Personal Information Collected',
                    'content': self._get_categories_collected()
                },
                {
                    'heading': 'Sources of Personal Information',
                    'content': self._get_sources()
                },
                {
                    'heading': 'Business Purposes for Collection',
                    'content': self._get_business_purposes()
                },
                {
                    'heading': 'Categories Sold or Shared',
                    'content': self._get_categories_sold()
                },
                {
                    'heading': 'Your Rights Under CCPA',
                    'content': self._get_consumer_rights()
                },
                {
                    'heading': 'How to Exercise Your Rights',
                    'content': self._get_exercise_rights()
                },
                {
                    'heading': 'Non-Discrimination',
                    'content': "We will not discriminate against you for exercising your CCPA rights."
                }
            ]
        }
 
    def _get_categories_collected(self) -> List[Dict]:
        categories = []
        for category, entries in self.ccpa_manager.data_inventory.items():
            if entries:
                categories.append({
                    'category': category.value,
                    'description': self.ccpa_manager._get_category_description(category),
                    'collected': True
                })
        return categories
 
    def _get_sources(self) -> List[str]:
        sources = set()
        for entries in self.ccpa_manager.data_inventory.values():
            for entry in entries:
                sources.add(entry.source)
        return list(sources)
 
    def _get_business_purposes(self) -> List[str]:
        purposes = set()
        for entries in self.ccpa_manager.data_inventory.values():
            for entry in entries:
                purposes.add(entry.business_purpose)
        return list(purposes)
 
    def _get_categories_sold(self) -> List[Dict]:
        sold = []
        for category, entries in self.ccpa_manager.data_inventory.items():
            for entry in entries:
                if entry.sold_or_shared:
                    sold.append({
                        'category': category.value,
                        'recipients': entry.third_parties_shared
                    })
        return sold
 
    def _get_consumer_rights(self) -> List[Dict]:
        return [
            {'right': 'Right to Know', 'description': 'Request disclosure of personal information collected'},
            {'right': 'Right to Delete', 'description': 'Request deletion of personal information'},
            {'right': 'Right to Opt-Out', 'description': 'Opt-out of sale or sharing of personal information'},
            {'right': 'Right to Correct', 'description': 'Request correction of inaccurate information'},
            {'right': 'Right to Limit Use', 'description': 'Limit use of sensitive personal information'}
        ]
 
    def _get_exercise_rights(self) -> str:
        return """
        You may exercise your rights by:
        - Visiting our "Do Not Sell My Personal Information" page
        - Emailing privacy@company.com
        - Calling our toll-free number: 1-800-XXX-XXXX
        - Using our online privacy request form
 
        We will verify your identity before processing your request.
        """