Gdpr Compliance

GDPR Compliance

Implement EU General Data Protection Regulation requirements for comprehensive data protection and privacy.

When to Use This Skill

Use this skill when you need to:

• Handle EU personal data
• Implement data subject rights
• Manage consent
• Conduct DPIAs
• Handle data breaches
• Document lawful processing

Trigger phrases: "GDPR compliance", "data protection", "EU privacy", "data subject rights", "DPIA", "consent management", "right to be forgotten"

What This Skill Does

GDPR Principles (Article 5)

Instructions

Step 1: Data Subject Rights Implementation

from enum import Enum
from dataclasses import dataclass
from datetime import datetime, timedelta

class DataSubjectRightType(Enum):
    ACCESS = "access"  # Article 15
    RECTIFICATION = "rectification"  # Article 16
    ERASURE = "erasure"  # Article 17 (Right to be forgotten)
    RESTRICT = "restrict"  # Article 18
    PORTABILITY = "portability"  # Article 20
    OBJECT = "object"  # Article 21

@dataclass
class DataSubjectRequest:
    """GDPR Data Subject Request."""
    request_id: str
    subject_id: str
    right_type: DataSubjectRightType
    submitted_at: datetime
    verified: bool
    status: str
    due_date: datetime

class DataSubjectRightsService:
    """Handle GDPR data subject rights requests."""

    RESPONSE_DEADLINE_DAYS = 30  # Can extend to 90 for complex requests

    def submit_request(
        self,
        subject_email: str,
        right_type: DataSubjectRightType,
        details: dict
    ) -> DataSubjectRequest:
        """Submit new data subject request."""
        request = DataSubjectRequest(
            request_id=f"DSR-{uuid.uuid4().hex[:8]}",
            subject_id=self._identify_subject(subject_email),
            right_type=right_type,
            submitted_at=datetime.utcnow(),
            verified=False,
            status="pending_verification",
            due_date=datetime.utcnow() + timedelta(days=self.RESPONSE_DEADLINE_DAYS)
        )

        # Store request
        self.request_repository.save(request)

        # Send verification email
        self._send_verification(subject_email, request)

        audit_log.info(
            "dsr_submitted",
            request_id=request.request_id,
            right_type=right_type.value
        )

        return request

    def process_access_request(self, request: DataSubjectRequest) -> AccessResponse:
        """Process Article 15 - Right of Access."""
        subject_data = self._collect_all_data(request.subject_id)

        response = AccessResponse(
            request_id=request.request_id,
            data_categories=self._categorize_data(subject_data),
            purposes=self._get_processing_purposes(request.subject_id),
            recipients=self._get_data_recipients(request.subject_id),
            retention_periods=self._get_retention_periods(),
            source=self._get_data_sources(request.subject_id),
            automated_decisions=self._get_automated_decisions(request.subject_id),
            safeguards=self._get_transfer_safeguards(),
            data_export=self._export_data(subject_data, format="json")
        )

        audit_log.info(
            "dsr_access_processed",
            request_id=request.request_id
        )

        return response

    def process_erasure_request(self, request: DataSubjectRequest) -> ErasureResponse:
        """Process Article 17 - Right to Erasure (Right to be Forgotten)."""
        # Check if erasure can be refused
        exemptions = self._check_erasure_exemptions(request.subject_id)
        if exemptions:
            return ErasureResponse(
                request_id=request.request_id,
                status="refused",
                reason=exemptions,
                retained_data=self._get_retained_data(request.subject_id)
            )

        # Perform erasure
        systems_affected = self._identify_affected_systems(request.subject_id)
        erasure_results = []

        for system in systems_affected:
            result = self._erase_from_system(request.subject_id, system)
            erasure_results.append(result)

        # Notify recipients
        recipients = self._get_data_recipients(request.subject_id)
        self._notify_recipients_of_erasure(recipients, request.subject_id)

        audit_log.info(
            "dsr_erasure_processed",
            request_id=request.request_id,
            systems_affected=len(systems_affected)
        )

        return ErasureResponse(
            request_id=request.request_id,
            status="completed",
            systems_affected=systems_affected,
            results=erasure_results
        )

    def process_portability_request(self, request: DataSubjectRequest) -> PortabilityResponse:
        """Process Article 20 - Right to Data Portability."""
        # Only data provided by subject and processed by consent/contract
        portable_data = self._get_portable_data(request.subject_id)

        # Export in structured, machine-readable format
        export = self._export_data(
            portable_data,
            format="json",
            structured=True
        )

        return PortabilityResponse(
            request_id=request.request_id,
            data_format="application/json",
            data_export=export,
            download_link=self._generate_secure_download(export)
        )

Step 2: Consent Management

class ConsentPurpose(Enum):
    MARKETING = "marketing"
    ANALYTICS = "analytics"
    PERSONALIZATION = "personalization"
    THIRD_PARTY_SHARING = "third_party_sharing"
    PROFILING = "profiling"

@dataclass
class Consent:
    """GDPR-compliant consent record."""
    consent_id: str
    subject_id: str
    purpose: ConsentPurpose
    granted: bool
    timestamp: datetime
    method: str  # how consent was collected
    version: str  # privacy policy version
    withdrawal_timestamp: Optional[datetime]

class ConsentManagementService:
    """GDPR Article 7 - Conditions for Consent."""

    def collect_consent(
        self,
        subject_id: str,
        purpose: ConsentPurpose,
        method: str
    ) -> Consent:
        """Record explicit, informed consent."""
        consent = Consent(
            consent_id=f"CONSENT-{uuid.uuid4().hex[:8]}",
            subject_id=subject_id,
            purpose=purpose,
            granted=True,
            timestamp=datetime.utcnow(),
            method=method,  # e.g., "checkbox", "double_opt_in"
            version=self._get_current_policy_version(),
            withdrawal_timestamp=None
        )

        self.consent_repository.save(consent)

        audit_log.info(
            "consent_collected",
            consent_id=consent.consent_id,
            purpose=purpose.value,
            method=method
        )

        return consent

    def withdraw_consent(
        self,
        subject_id: str,
        purpose: ConsentPurpose
    ) -> Consent:
        """Process consent withdrawal."""
        consent = self.consent_repository.find_active(subject_id, purpose)

        if not consent:
            raise ConsentNotFoundError(f"No active consent for {purpose.value}")

        consent.granted = False
        consent.withdrawal_timestamp = datetime.utcnow()
        self.consent_repository.save(consent)

        # Stop processing for this purpose
        self._stop_processing(subject_id, purpose)

        audit_log.info(
            "consent_withdrawn",
            consent_id=consent.consent_id,
            purpose=purpose.value
        )

        return consent

    def check_consent(
        self,
        subject_id: str,
        purpose: ConsentPurpose
    ) -> bool:
        """Check if valid consent exists for purpose."""
        consent = self.consent_repository.find_active(subject_id, purpose)

        if not consent:
            return False

        # Check if policy has been updated since consent
        if consent.version != self._get_current_policy_version():
            # May need to re-consent
            return False

        return consent.granted

Step 3: Data Protection Impact Assessment (DPIA)

class DPIAService:
    """GDPR Article 35 - Data Protection Impact Assessment."""

    def is_dpia_required(self, processing_activity: ProcessingActivity) -> bool:
        """Determine if DPIA is required."""
        # Article 35(3) - DPIA required for:
        triggers = [
            processing_activity.involves_profiling,
            processing_activity.involves_automated_decisions,
            processing_activity.large_scale_special_category,
            processing_activity.systematic_monitoring,
            processing_activity.new_technology,
            processing_activity.cross_border_transfer,
        ]

        return any(triggers)

    def conduct_dpia(
        self,
        processing_activity: ProcessingActivity
    ) -> DPIAReport:
        """Conduct Data Protection Impact Assessment."""
        dpia = DPIAReport(
            dpia_id=f"DPIA-{uuid.uuid4().hex[:8]}",
            processing_activity=processing_activity.name,
            date=datetime.utcnow(),
            sections=[]
        )

        # Section 1: Description of processing
        dpia.sections.append(DPIASection(
            name="Processing Description",
            content={
                "purpose": processing_activity.purpose,
                "data_categories": processing_activity.data_categories,
                "data_subjects": processing_activity.data_subjects,
                "recipients": processing_activity.recipients,
                "retention": processing_activity.retention_period,
                "technology": processing_activity.technology_used
            }
        ))

        # Section 2: Necessity and proportionality
        dpia.sections.append(DPIASection(
            name="Necessity Assessment",
            content={
                "legal_basis": processing_activity.legal_basis,
                "necessity_justification": self._assess_necessity(processing_activity),
                "proportionality": self._assess_proportionality(processing_activity),
                "data_minimization": self._assess_minimization(processing_activity)
            }
        ))

        # Section 3: Risk assessment
        risks = self._identify_risks(processing_activity)
        dpia.sections.append(DPIASection(
            name="Risk Assessment",
            content={
                "risks": [self._assess_risk(r) for r in risks],
                "overall_risk_level": self._calculate_overall_risk(risks)
            }
        ))

        # Section 4: Mitigation measures
        dpia.sections.append(DPIASection(
            name="Mitigation Measures",
            content={
                "measures": self._identify_mitigations(risks),
                "residual_risks": self._calculate_residual_risks(risks)
            }
        ))

        audit_log.info(
            "dpia_completed",
            dpia_id=dpia.dpia_id,
            processing_activity=processing_activity.name
        )

        return dpia

Step 4: Breach Notification

class BreachNotificationService:
    """GDPR Article 33-34 - Breach Notification."""

    AUTHORITY_NOTIFICATION_HOURS = 72  # Within 72 hours

    def report_breach(self, breach_details: dict) -> BreachRecord:
        """Record and assess personal data breach."""
        breach = BreachRecord(
            breach_id=f"BREACH-{uuid.uuid4().hex[:8]}",
            discovered_at=datetime.utcnow(),
            description=breach_details["description"],
            data_categories=breach_details["data_categories"],
            subjects_affected=breach_details["subjects_affected"],
            cause=breach_details["cause"],
            status="under_investigation"
        )

        # Assess risk to individuals
        risk_assessment = self._assess_breach_risk(breach)
        breach.risk_level = risk_assessment.level

        self.breach_repository.save(breach)

        audit_log.critical(
            "data_breach_reported",
            breach_id=breach.breach_id,
            subjects_affected=breach.subjects_affected
        )

        return breach

    def notify_authority(self, breach: BreachRecord) -> AuthorityNotification:
        """Notify supervisory authority within 72 hours."""
        notification = AuthorityNotification(
            breach_id=breach.breach_id,
            submitted_at=datetime.utcnow(),
            content={
                "nature_of_breach": breach.description,
                "categories_of_data": breach.data_categories,
                "approximate_subjects": breach.subjects_affected,
                "dpo_contact": self._get_dpo_contact(),
                "likely_consequences": self._assess_consequences(breach),
                "measures_taken": self._get_remediation_measures(breach),
                "measures_proposed": self._get_proposed_measures(breach)
            }
        )

        # Submit to authority
        self._submit_to_authority(notification)

        audit_log.info(
            "authority_notified",
            breach_id=breach.breach_id,
            notification_time=notification.submitted_at.isoformat()
        )

        return notification

    def notify_subjects(self, breach: BreachRecord) -> SubjectNotification:
        """Notify affected data subjects if high risk."""
        # Article 34 - Only required if high risk to rights and freedoms
        if breach.risk_level != "high":
            return None

        notification = SubjectNotification(
            breach_id=breach.breach_id,
            content={
                "plain_language_description": self._create_plain_description(breach),
                "dpo_contact": self._get_dpo_contact(),
                "likely_consequences": self._describe_consequences(breach),
                "measures_taken": self._describe_remediation(breach),
                "recommended_actions": self._get_subject_recommendations(breach)
            }
        )

        # Send to affected subjects
        self._send_notifications(breach.affected_subjects, notification)

        audit_log.info(
            "subjects_notified",
            breach_id=breach.breach_id,
            subjects_notified=len(breach.affected_subjects)
        )

        return notification

Step 5: Records of Processing Activities

class ProcessingActivityRecord:
    """GDPR Article 30 - Records of Processing Activities."""

    def create_ropa(self) -> ROPA:
        """Create Records of Processing Activities."""
        return ROPA(
            organization=self._get_organization_details(),
            dpo_contact=self._get_dpo_contact(),
            activities=[
                self._document_activity(a) for a in self._get_all_activities()
            ]
        )

    def _document_activity(self, activity: ProcessingActivity) -> dict:
        """Document individual processing activity."""
        return {
            "name": activity.name,
            "controller": self._get_controller_details(),
            "purposes": activity.purposes,
            "legal_basis": activity.legal_basis,
            "data_categories": activity.data_categories,
            "data_subjects": activity.data_subject_categories,
            "recipients": activity.recipients,
            "third_country_transfers": activity.transfers,
            "safeguards": activity.transfer_safeguards,
            "retention_period": activity.retention,
            "security_measures": activity.security_measures,
        }

Documentation Requirements

Required Documents

• Privacy Policy
• Records of Processing Activities (ROPA)
• Data Processing Agreements
• DPIA reports (where required)
• Consent records
• Breach response procedures

Required Records

• Data subject requests and responses
• Consent records
• Processing activity records
• Breach notifications
• DPIA assessments

Quality Checklist

• Lawful basis documented for all processing
• Privacy notices published
• Consent mechanisms implemented
• Data subject rights procedures in place
• DPIA process established
• Breach notification procedures ready
• ROPA maintained
• DPO appointed (if required)
• Training completed

Related Skills

• ccpa-compliance - California privacy compliance
• soc2-compliance - SOC 2 privacy criteria
• security-review - Security controls

Version: 1.0.0 Last Updated: December 2025 Based on: AI Templates compliance_governance/privacy_protection/

Iterative Refinement Strategy

This skill is optimized for an iterative approach:

1. Execute: Perform the core steps defined above.
2. Review: Critically analyze the output (coverage, quality, completeness).
3. Refine: If targets aren't met, repeat the specific implementation steps with improved context.
4. Loop: Continue until the definition of done is satisfied.