Chapter 56: Privacy Engineering & Data Protection¶

Overview¶

Privacy engineering is the systematic discipline of embedding privacy protections into systems, processes, and architectures from inception rather than bolting them on as compliance afterthoughts. Where traditional security operations focus on confidentiality, integrity, and availability of systems, privacy engineering concerns itself with a fundamentally different question: How do we process personal data in ways that respect the rights, expectations, and autonomy of the individuals that data represents — while still achieving legitimate business and security objectives? This question is not academic. The regulatory landscape has shifted irreversibly. The EU's General Data Protection Regulation (GDPR) imposed fines exceeding EUR 4.3 billion in its first five years. The California Consumer Privacy Act (CCPA) and its successor California Privacy Rights Act (CPRA) created new categories of consumer rights that require technical implementation, not merely legal acknowledgment. Brazil's LGPD, South Korea's PIPA, India's DPDPA, and dozens of other frameworks have created a global patchwork of privacy obligations that every organization processing personal data must navigate.

Yet most security operations teams treat privacy as someone else's problem — a legal concern, a compliance checkbox, a DPO's headache. This is a catastrophic mistake. Privacy incidents are security incidents. A misconfigured S3 bucket exposing customer PII is simultaneously a security vulnerability and a privacy breach requiring regulatory notification within 72 hours under GDPR. A SOC analyst who queries a SIEM for user behavioral analytics is simultaneously performing security monitoring and processing personal data under a lawful basis that must be documented. The SOC's detection queries, log retention policies, endpoint telemetry collection, and incident response procedures all have privacy implications that, if ignored, create regulatory exposure far exceeding the cost of any single security incident.

This chapter bridges the gap between privacy theory and security operations practice. We begin with Privacy by Design — the foundational framework that should inform every system architecture decision. We operationalize major regulatory frameworks (GDPR, CCPA/CPRA, LGPD, PIPA) into technical controls that security teams can implement and verify. We cover LINDDUN, the privacy-specific threat modeling methodology that complements STRIDE and PASTA. We explore Privacy-Enhancing Technologies (PETs) that enable data utility without data exposure. We build automated pipelines for data discovery, classification, consent management, and data subject rights fulfillment. And we integrate all of this into the SOC — showing how privacy monitoring, breach assessment, and notification workflows operate alongside traditional security operations. Every section connects to detection engineering, incident response, and the operational realities of running a security program that respects privacy as a first-class requirement.

The organizations that will thrive in the next decade are those that treat privacy not as a constraint on security operations but as a force multiplier. Privacy-aware security architectures collect less data, retain it for shorter periods, apply stronger access controls, and maintain better audit trails — all of which reduce attack surface, limit blast radius, and improve incident response times. Privacy engineering is not the enemy of security. It is security done right.

Learning Objectives¶

By the end of this chapter, students SHALL be able to:

1. Apply Hoepman's 8 privacy design strategies (MINIMIZE, HIDE, SEPARATE, AGGREGATE, INFORM, CONTROL, ENFORCE, DEMONSTRATE) to system architecture decisions, mapping each strategy to concrete technical controls (Application)
2. Operationalize GDPR requirements (Articles 25, 30, 32, 35) into implementable technical and organizational measures within security operations workflows (Synthesis)
3. Implement CCPA/CPRA consumer rights (access, deletion, opt-out, correction) through automated data subject request pipelines with identity verification and erasure cascades (Application)
4. Conduct Data Protection Impact Assessments (DPIAs) using structured methodologies integrated with threat modeling outputs, producing risk assessment matrices with quantified residual risk (Analysis)
5. Execute LINDDUN privacy threat modeling against data flow diagrams, identifying linkability, identifiability, non-repudiation, detectability, disclosure, unawareness, and non-compliance threats (Analysis)
6. Evaluate Privacy-Enhancing Technologies (differential privacy, homomorphic encryption, secure multi-party computation, federated learning, k-anonymity) for fitness against specific use cases, balancing utility loss against privacy guarantees (Evaluation)
7. Design automated PII discovery and data classification pipelines using regex, NLP, and entropy-based detection integrated with DLP controls (Synthesis)
8. Build consent management architectures that support granular purpose-based consent, withdrawal, and preference propagation across distributed systems (Synthesis)
9. Create privacy monitoring dashboards with KPIs covering breach detection, purpose limitation violations, retention compliance, and DSR fulfillment SLAs (Synthesis)
10. Integrate privacy breach assessment and regulatory notification workflows into SOC incident response procedures, including 72-hour GDPR and 45-day CCPA timelines (Application)

Prerequisites¶

• Completion of Chapter 7: Data Loss Prevention — DLP architectures, data classification, content inspection engines
• Completion of Chapter 12: Security Governance & Compliance — governance frameworks, compliance program management
• Familiarity with Chapter 13: Risk Management — risk assessment methodologies, risk treatment options, risk register management
• Familiarity with Chapter 36: Regulations & Compliance — regulatory landscape, compliance mapping, audit preparation
• Familiarity with Chapter 20: Cloud Security Fundamentals — cloud data storage, IAM, encryption at rest/in transit
• Familiarity with Chapter 55: Threat Modeling Operations — STRIDE, PASTA, threat modeling processes
• Working knowledge of database systems, API design, and data pipeline architectures

MITRE ATT&CK Privacy-Relevant Technique Mapping¶

1. Privacy by Design — Hoepman's 8 Strategies¶

1.1 The Foundation: Privacy by Design as Engineering Discipline¶

Privacy by Design (PbD) was originally articulated by Ann Cavoukian as seven foundational principles. Jaap-Henk Hoepman translated these principles into eight concrete design strategies that engineers can directly implement. Unlike Cavoukian's principles, which operate at a philosophical level ("proactive not reactive," "privacy as the default"), Hoepman's strategies are actionable: they tell you what to build, not just what to believe. GDPR Article 25 codified Privacy by Design and Privacy by Default as legal requirements, transforming Hoepman's strategies from best practices into regulatory obligations.

1.2 The Eight Strategies¶

Strategy 1: MINIMIZE¶

Principle: Limit the processing of personal data to the minimal amount necessary for the stated purpose.

Data minimization is not simply "collect less data." It requires systematic analysis of every data element against every processing purpose, eliminating any element that is not strictly necessary. This applies to collection, storage, access, and retention at every stage of the data lifecycle.

Technical Controls:

• Schema-level enforcement: database schemas that reject unnecessary fields
• API input validation: endpoints that strip or reject non-required PII fields
• Log sanitization: automated redaction of PII from application and infrastructure logs
• Query result filtering: database views that expose only purpose-relevant columns
• Retention automation: TTL-based deletion of data beyond its retention period

# PII Minimization Middleware — strips unnecessary fields before storage
# Synthetic example — all data is fictional

from functools import wraps
import re
from datetime import datetime

REQUIRED_FIELDS = {
    "user_registration": {"email", "username", "password_hash"},
    "order_processing": {"order_id", "shipping_address", "payment_token"},
    "support_ticket": {"ticket_id", "issue_description", "contact_email"},
}

SENSITIVE_PATTERNS = {
    "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
    "credit_card": re.compile(r"\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b"),
    "phone": re.compile(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b"),
}

def minimize_data(purpose: str):
    """Decorator that strips non-required fields based on processing purpose."""
    def decorator(func):
        @wraps(func)
        def wrapper(data: dict, *args, **kwargs):
            required = REQUIRED_FIELDS.get(purpose, set())
            if not required:
                raise ValueError(f"Unknown processing purpose: {purpose}")

            # Strip non-required fields
            minimized = {k: v for k, v in data.items() if k in required}
            stripped_fields = set(data.keys()) - required

            if stripped_fields:
                print(f"[MINIMIZE] Purpose '{purpose}': stripped fields "
                      f"{stripped_fields} at {datetime.utcnow().isoformat()}")

            return func(minimized, *args, **kwargs)
        return wrapper
    return decorator

@minimize_data(purpose="user_registration")
def register_user(data: dict) -> dict:
    """Register user with only required fields."""
    # Only email, username, password_hash reach this function
    # Fields like phone_number, date_of_birth, ssn are stripped
    print(f"[REGISTER] Processing with fields: {set(data.keys())}")
    return {"status": "registered", "fields_processed": list(data.keys())}

# Test with over-collected data
test_data = {
    "email": "testuser@example.com",
    "username": "testuser",
    "password_hash": "REDACTED",
    "phone_number": "555-0100",       # Not required — stripped
    "date_of_birth": "1990-01-01",    # Not required — stripped
    "ssn": "000-00-0000",             # Not required — stripped
    "favorite_color": "blue",         # Not required — stripped
}

result = register_user(test_data)
# Output: [MINIMIZE] Purpose 'user_registration': stripped fields
#         {'phone_number', 'date_of_birth', 'ssn', 'favorite_color'}

Strategy 2: HIDE¶

Principle: Protect personal data by making it unlinkable or unobservable to unauthorized parties.

HIDE encompasses encryption (at rest and in transit), pseudonymization, anonymization, and access controls that prevent unauthorized observation of personal data. The goal is to ensure that even when data must be stored, it is not accessible in plaintext to anyone who does not have a legitimate, documented need.

Technical Controls:

• Encryption at rest: AES-256 for databases, file systems, and backups
• Encryption in transit: TLS 1.3 for all data flows
• Pseudonymization: replacing direct identifiers with tokens via a separation-controlled mapping table
• Anonymization: irreversible transformation that prevents re-identification
• Column-level encryption: encrypting specific PII columns rather than entire databases
• Tokenization: replacing sensitive values with non-reversible tokens for analytics

# Pseudonymization Engine — synthetic example
import hashlib
import secrets
import json
from typing import Optional

class PseudonymizationEngine:
    """
    Replaces direct identifiers with pseudonyms.
    Mapping table stored separately with strict access controls.
    """

    def __init__(self, salt: Optional[str] = None):
        self.salt = salt or secrets.token_hex(32)
        self._mapping: dict[str, str] = {}
        self._reverse: dict[str, str] = {}

    def pseudonymize(self, identifier: str, category: str = "default") -> str:
        """Generate a deterministic pseudonym for a given identifier."""
        key = f"{category}:{identifier}"
        if key in self._mapping:
            return self._mapping[key]

        # HMAC-based pseudonym generation
        pseudonym = hashlib.sha256(
            f"{self.salt}:{key}".encode()
        ).hexdigest()[:16]

        token = f"PSE-{category.upper()}-{pseudonym}"
        self._mapping[key] = token
        self._reverse[token] = identifier
        return token

    def re_identify(self, token: str) -> Optional[str]:
        """Re-identify only with mapping table access (separate authorization)."""
        return self._reverse.get(token)

    def export_mapping(self) -> str:
        """Export mapping for secure storage — NEVER store with pseudonymized data."""
        return json.dumps(self._reverse, indent=2)

# Usage example
engine = PseudonymizationEngine()

# Original record
record = {
    "name": "Test User",
    "email": "testuser@example.com",
    "ip_address": "192.0.2.45",
    "department": "Engineering",  # Not an identifier — no pseudonymization needed
}

# Pseudonymize direct identifiers
pseudonymized = {
    "name": engine.pseudonymize(record["name"], "name"),
    "email": engine.pseudonymize(record["email"], "email"),
    "ip_address": engine.pseudonymize(record["ip_address"], "ip"),
    "department": record["department"],  # Retained as-is
}

print(pseudonymized)
# {'name': 'PSE-NAME-a3f8c1...', 'email': 'PSE-EMAIL-7b2d4e...',
#  'ip_address': 'PSE-IP-9c1f3a...', 'department': 'Engineering'}

Strategy 3: SEPARATE¶

Principle: Process personal data in a distributed manner, across separate compartments, to prevent correlation and reduce blast radius.

Separation means that different categories of personal data are stored and processed in isolated systems so that a breach of one system does not expose the complete profile of a data subject. This maps directly to the security principle of compartmentalization but applies it specifically to privacy concerns.

Technical Controls:

• Separate databases for different data categories (identity, financial, health, behavioral)
• Microservice-level data ownership: each service owns only its data domain
• Purpose-bound data stores: analytics data physically separated from operational data
• Cross-system identifier federation without shared PII stores
• Network segmentation between PII-processing and non-PII systems

graph LR
    subgraph "Identity Service"
        A[("User Profile DB<br/>name, email")]
    end
    subgraph "Payment Service"
        B[("Payment DB<br/>tokens only")]
    end
    subgraph "Analytics Service"
        C[("Analytics DB<br/>pseudonymized")]
    end
    subgraph "Health Service"
        D[("Health DB<br/>encrypted, separate keys")]
    end

    E[API Gateway] --> A
    E --> B
    E --> C
    E --> D

    A -.->|"user_id only"| B
    A -.->|"pseudonym_token"| C
    A -.->|"encrypted_ref"| D

    style A fill:#e74c3c,color:#fff
    style B fill:#f39c12,color:#fff
    style C fill:#2ecc71,color:#fff
    style D fill:#9b59b6,color:#fff

Strategy 4: AGGREGATE¶

Principle: Process personal data at the highest level of aggregation possible, with the least possible detail.

Aggregation limits privacy risk by processing groups rather than individuals. Instead of analyzing individual user behavior, aggregate to cohorts. Instead of retaining individual transaction records indefinitely, summarize into statistical aggregates and delete the originals.

Technical Controls:

• Statistical aggregation: replacing individual records with group statistics
• Generalization: reducing precision (full date of birth → age range; exact location → city-level)
• Binning: grouping continuous values into discrete ranges
• Differential privacy noise injection (covered in detail in Section 6)
• Aggregate-only analytics views

Strategy 5: INFORM¶

Principle: Inform data subjects about the processing of their personal data in a timely and transparent manner.

Transparency is not merely a privacy notice posted once and forgotten. It requires dynamic, contextual information delivery at the moment of collection, at the moment of purpose change, and continuously throughout the data lifecycle.

Technical Controls:

• Just-in-time privacy notices at data collection points
• Machine-readable privacy policies (P3P successor formats, schema.org DataPrivacy)
• Data processing activity logs accessible to data subjects
• Purpose-of-collection metadata attached to every data element
• Transparency dashboards showing what data is held and why

Strategy 6: CONTROL¶

Principle: Provide data subjects with mechanisms to control the processing of their personal data.

Control means operationalizable consent and preference management — not a blanket "I agree" checkbox, but granular, purpose-specific controls that data subjects can modify at any time, with those modifications propagated across all processing systems.

Technical Controls:

• Granular consent management platforms (CMPs)
• Per-purpose consent flags stored with data
• Consent withdrawal propagation across microservices
• Data subject access portals with self-service controls
• Preference centers with purpose-level granularity

Strategy 7: ENFORCE¶

Principle: Commit to processing personal data in a privacy-compliant way and enforce this through technical mechanisms.

Enforcement means that privacy policies are not merely documented but are technically enforced — systems physically prevent non-compliant processing, rather than relying on humans to follow procedures.

Technical Controls:

• Policy-as-code: privacy policies encoded in OPA/Rego, enforced at API gateways
• Purpose limitation enforcement via attribute-based access control (ABAC)
• Automated retention enforcement: TTL-based deletion with audit trails
• DLP rules preventing PII in unauthorized channels
• Privacy-aware CI/CD gates: blocking deployments that introduce new PII processing without DPIA

# Purpose Limitation Enforcement — OPA-style policy (synthetic)

PROCESSING_PURPOSES = {
    "marketing": {
        "allowed_fields": {"email", "first_name", "consent_marketing"},
        "requires_consent": True,
        "consent_field": "consent_marketing",
        "retention_days": 365,
    },
    "fraud_detection": {
        "allowed_fields": {"transaction_id", "amount", "ip_address", "device_fingerprint"},
        "requires_consent": False,  # Legitimate interest basis
        "lawful_basis": "legitimate_interest",
        "retention_days": 180,
    },
    "service_delivery": {
        "allowed_fields": {"user_id", "email", "shipping_address", "order_id"},
        "requires_consent": False,  # Contractual necessity
        "lawful_basis": "contract",
        "retention_days": 730,
    },
}

def enforce_purpose_limitation(data: dict, purpose: str, user_consent: dict) -> dict:
    """
    Enforce purpose limitation: only allow access to fields
    permitted for the stated processing purpose.
    """
    policy = PROCESSING_PURPOSES.get(purpose)
    if not policy:
        raise PermissionError(f"Unknown processing purpose: {purpose}")

    # Check consent if required
    if policy.get("requires_consent"):
        consent_field = policy["consent_field"]
        if not user_consent.get(consent_field):
            raise PermissionError(
                f"Processing for purpose '{purpose}' requires consent "
                f"'{consent_field}' which has not been granted"
            )

    # Filter to allowed fields only
    allowed = policy["allowed_fields"]
    filtered = {k: v for k, v in data.items() if k in allowed}
    blocked = set(data.keys()) - allowed

    if blocked:
        print(f"[ENFORCE] Purpose '{purpose}': blocked access to {blocked}")

    return filtered

# Example: marketing team tries to access transaction data
user_data = {
    "email": "testuser@example.com",
    "first_name": "Test",
    "consent_marketing": True,
    "transaction_id": "TXN-00001",   # Not allowed for marketing
    "ssn": "000-00-0000",            # Not allowed for ANY purpose shown
}

consent = {"consent_marketing": True}
result = enforce_purpose_limitation(user_data, "marketing", consent)
# Output: [ENFORCE] Purpose 'marketing': blocked access to {'transaction_id', 'ssn'}
# result = {'email': 'testuser@example.com', 'first_name': 'Test',
#           'consent_marketing': True}

Strategy 8: DEMONSTRATE¶

Principle: Demonstrate compliance with privacy policies and applicable regulations through documentation, audit trails, and accountability mechanisms.

Accountability is GDPR's most operationally demanding principle. Organizations must not merely comply — they must be able to prove they comply at any time. This requires comprehensive audit trails, processing activity records, DPIA documentation, and evidence of technical measures.

Technical Controls:

• Immutable audit logs for all PII access and processing events
• Automated Records of Processing Activities (ROPA) generation
• DPIA document management with version control
• Privacy control effectiveness testing and evidence collection
• Consent receipt archival with tamper-evident storage

2. GDPR Operationalization¶

2.1 Article 25: Data Protection by Design and by Default¶

Article 25 requires that data protection principles are implemented through "appropriate technical and organisational measures" both at the time of design and at the time of processing. This is not a suggestion — it is a legally binding requirement with potential fines of up to 4% of global annual turnover or EUR 20 million, whichever is higher.

Operationalization Checklist:

2.2 Article 30: Records of Processing Activities (ROPA)¶

Every controller and processor must maintain records of processing activities. This is not a one-time documentation exercise — it must be continuously maintained and available for supervisory authority inspection at any time.

# Automated ROPA Generator — Synthetic Example
import json
from datetime import datetime, timedelta
from typing import Optional

class ROPAEntry:
    """Single processing activity record per GDPR Article 30."""

    def __init__(
        self,
        activity_name: str,
        purpose: str,
        lawful_basis: str,
        data_categories: list[str],
        data_subjects: list[str],
        recipients: list[str],
        retention_period: str,
        technical_measures: list[str],
        transfer_safeguards: Optional[str] = None,
    ):
        self.activity_name = activity_name
        self.purpose = purpose
        self.lawful_basis = lawful_basis
        self.data_categories = data_categories
        self.data_subjects = data_subjects
        self.recipients = recipients
        self.retention_period = retention_period
        self.technical_measures = technical_measures
        self.transfer_safeguards = transfer_safeguards
        self.created = datetime.utcnow().isoformat()
        self.last_reviewed = self.created

    def to_dict(self) -> dict:
        return {
            "activity_name": self.activity_name,
            "purpose": self.purpose,
            "lawful_basis": self.lawful_basis,
            "data_categories": self.data_categories,
            "data_subjects": self.data_subjects,
            "recipients": self.recipients,
            "retention_period": self.retention_period,
            "technical_measures": self.technical_measures,
            "transfer_safeguards": self.transfer_safeguards,
            "created": self.created,
            "last_reviewed": self.last_reviewed,
        }

class ROPARegistry:
    """Central registry of all processing activities."""

    def __init__(self, controller_name: str, dpo_contact: str):
        self.controller_name = controller_name
        self.dpo_contact = dpo_contact
        self.entries: list[ROPAEntry] = []

    def add_activity(self, entry: ROPAEntry) -> None:
        self.entries.append(entry)
        print(f"[ROPA] Added activity: {entry.activity_name}")

    def find_stale(self, days: int = 180) -> list[str]:
        """Find entries not reviewed within the specified period."""
        cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat()
        return [
            e.activity_name for e in self.entries
            if e.last_reviewed < cutoff
        ]

    def export(self) -> str:
        return json.dumps({
            "controller": self.controller_name,
            "dpo_contact": self.dpo_contact,
            "generated": datetime.utcnow().isoformat(),
            "activities": [e.to_dict() for e in self.entries],
        }, indent=2)

# Build ROPA
ropa = ROPARegistry(
    controller_name="SynthCorp International",
    dpo_contact="dpo@synthcorp.example.com"
)

ropa.add_activity(ROPAEntry(
    activity_name="Employee Onboarding",
    purpose="Employment contract fulfillment and legal obligations",
    lawful_basis="Contract (Art. 6(1)(b)) + Legal Obligation (Art. 6(1)(c))",
    data_categories=["name", "address", "national_id", "bank_details", "emergency_contact"],
    data_subjects=["employees"],
    recipients=["HR department", "payroll processor (PayCorp.example.com)"],
    retention_period="Duration of employment + 7 years (tax obligation)",
    technical_measures=["AES-256 encryption at rest", "RBAC access control",
                        "audit logging", "pseudonymization of national_id"],
))

ropa.add_activity(ROPAEntry(
    activity_name="Security Monitoring (SIEM)",
    purpose="Detection of security threats and incident response",
    lawful_basis="Legitimate Interest (Art. 6(1)(f))",
    data_categories=["IP addresses", "user agent strings", "authentication events",
                      "network flow data", "endpoint telemetry"],
    data_subjects=["employees", "contractors", "website visitors"],
    recipients=["SOC team", "incident responders", "MSSP (SecOps.example.com)"],
    retention_period="90 days (hot) + 365 days (cold archive)",
    technical_measures=["pseudonymization of user identifiers in analytics",
                        "role-based SIEM access", "query audit logging",
                        "automated PII redaction in log pipelines"],
))

2.3 Article 32: Security of Processing¶

Article 32 requires "appropriate technical and organisational measures to ensure a level of security appropriate to the risk." This directly connects privacy obligations to security controls — your security program is part of your GDPR compliance program.

Required Measures (Article 32(1)):

1. Pseudonymization and encryption of personal data
2. Confidentiality, integrity, availability, and resilience of processing systems
3. Ability to restore access to personal data in a timely manner after an incident
4. Regular testing and evaluation of technical and organizational measures

2.4 Article 35: Data Protection Impact Assessments (DPIAs)¶

DPIAs are mandatory when processing is "likely to result in a high risk to the rights and freedoms of natural persons." Article 35(3) specifies three situations where DPIAs are always required:

1. Systematic and extensive evaluation of personal aspects (profiling)
2. Large-scale processing of special category data (health, biometrics, etc.)
3. Systematic monitoring of a publicly accessible area

DPIA Triggers in Security Operations:

• Deploying UEBA (User and Entity Behavior Analytics) — profiling trigger
• Implementing DLP with content inspection — systematic monitoring trigger
• Endpoint detection with user activity monitoring — profiling trigger
• Deploying video analytics for physical security — public area monitoring trigger
• Correlating HR data with security events — special category data trigger

See Section 4 for the complete DPIA methodology.

2.5 Lawful Basis Selection for Security Operations¶

3. CCPA/CPRA Implementation¶

3.1 Consumer Rights Under CCPA/CPRA¶

The California Consumer Privacy Act (CCPA), as amended by the California Privacy Rights Act (CPRA), grants California consumers the following rights:

3.2 CPRA New Obligations¶

The CPRA (effective January 1, 2023) introduced several significant expansions:

Sensitive Personal Information (SPI):

• Social Security numbers, driver's license numbers, state ID numbers
• Financial account information (with credentials)
• Precise geolocation
• Racial/ethnic origin, religious beliefs, union membership
• Contents of mail, email, or text messages (unless directed to the business)
• Genetic data, biometric data, health data
• Sex life or sexual orientation data

Automated Decision-Making:

• Consumers have the right to opt out of automated decision-making technology
• Businesses must provide meaningful information about the logic involved
• Access to results of automated decisions is required

3.3 Technical Implementation: Opt-Out Mechanisms¶

# CCPA/CPRA Opt-Out Signal Processing — Synthetic Example
import json
from datetime import datetime
from enum import Enum

class OptOutType(Enum):
    SALE = "do_not_sell"
    SHARING = "do_not_share"
    SENSITIVE_PI = "limit_sensitive_pi"
    AUTOMATED_DECISIONS = "opt_out_automated"
    TARGETED_ADS = "opt_out_targeted_ads"

class ConsentSignalProcessor:
    """
    Processes opt-out signals from multiple sources:
    - User preference center
    - Global Privacy Control (GPC) browser signal
    - "Do Not Sell" link
    - Authorized agent requests
    """

    def __init__(self):
        self._preferences: dict[str, dict] = {}

    def process_gpc_signal(self, user_id: str, gpc_header: str) -> dict:
        """
        Process Global Privacy Control signal (Sec-GPC: 1).
        Under CPRA, GPC signal MUST be treated as valid opt-out.
        """
        if gpc_header == "1":
            # GPC = 1 is a valid opt-out of sale AND sharing
            self._set_preference(user_id, OptOutType.SALE, True, "GPC")
            self._set_preference(user_id, OptOutType.SHARING, True, "GPC")
            return {
                "user_id": user_id,
                "gpc_honored": True,
                "opted_out": ["sale", "sharing"],
                "timestamp": datetime.utcnow().isoformat(),
            }
        return {"user_id": user_id, "gpc_honored": False}

    def _set_preference(
        self, user_id: str, opt_type: OptOutType, value: bool, source: str
    ) -> None:
        if user_id not in self._preferences:
            self._preferences[user_id] = {}
        self._preferences[user_id][opt_type.value] = {
            "opted_out": value,
            "source": source,
            "timestamp": datetime.utcnow().isoformat(),
        }

    def check_allowed(self, user_id: str, processing_type: str) -> bool:
        """Check if a specific processing type is allowed for a user."""
        prefs = self._preferences.get(user_id, {})
        opt_out_entry = prefs.get(processing_type, {})
        return not opt_out_entry.get("opted_out", False)

# Example usage
processor = ConsentSignalProcessor()

# Simulate GPC header from browser
result = processor.process_gpc_signal("USR-12345", gpc_header="1")
print(json.dumps(result, indent=2))

# Check if sale is allowed
can_sell = processor.check_allowed("USR-12345", "do_not_sell")
print(f"Can sell data: {can_sell}")  # False — user opted out via GPC

3.4 Regulatory Comparison: GDPR vs CCPA vs LGPD vs PIPA¶

4. Data Protection Impact Assessments (DPIAs)¶

4.1 When a DPIA Is Required¶

Under GDPR Article 35, a DPIA is mandatory when processing is likely to result in a "high risk" to data subjects. The Article 29 Working Party (now EDPB) identified nine criteria — processing that meets two or more of these criteria generally requires a DPIA:

1. Evaluation or scoring (profiling, prediction)
2. Automated decision-making with legal or significant effects
3. Systematic monitoring of data subjects
4. Sensitive data or highly personal data (special categories, financial, location)
5. Large-scale processing (number of subjects, data volume, geographic scope)
6. Matching or combining datasets from different sources
7. Vulnerable data subjects (employees, children, patients, elderly)
8. Innovative use of new technology (AI/ML, biometrics, IoT)
9. Processing that prevents rights exercise (access denial, service blocking)

4.2 DPIA Methodology¶

flowchart TD
    A[Identify Need for DPIA] --> B{Meets 2+ WP29<br/>Criteria?}
    B -->|Yes| C[Describe Processing]
    B -->|No| B2[Document Decision<br/>Not to Conduct DPIA]
    C --> D[Assess Necessity &<br/>Proportionality]
    D --> E[Identify Risks to<br/>Data Subjects]
    E --> F[Assess Risk<br/>Likelihood x Impact]
    F --> G{Residual Risk<br/>Acceptable?}
    G -->|Yes| H[Document & Implement<br/>Measures]
    G -->|No| I[Identify Additional<br/>Mitigation Measures]
    I --> F
    H --> J[DPO Review &<br/>Sign-off]
    J --> K{DPO Approves?}
    K -->|Yes| L[Proceed with<br/>Processing]
    K -->|No| M[Revise Processing<br/>or Consult DPA]
    L --> N[Ongoing Monitoring<br/>& Review]
    N --> O{Material Change<br/>in Processing?}
    O -->|Yes| C
    O -->|No| N

    style A fill:#3498db,color:#fff
    style F fill:#e74c3c,color:#fff
    style H fill:#2ecc71,color:#fff
    style L fill:#2ecc71,color:#fff
    style M fill:#e74c3c,color:#fff

4.3 Risk Assessment Matrix¶

Risk Rating Thresholds:

• Critical (20-25): Processing must not proceed without supervisory authority consultation (Art. 36)
• High (12-19): Significant additional measures required; DPO must approve
• Medium (5-11): Additional measures recommended; document risk acceptance
• Low (1-4): Standard controls sufficient; document in DPIA

4.4 DPIA Template: UEBA Deployment¶

4.5 Integrating DPIAs with Threat Modeling¶

DPIAs and threat models address complementary risks: threat models focus on attacks against systems, while DPIAs focus on harms to data subjects. Combining them produces a comprehensive risk picture.

Integration Points:

1. Data flow diagrams from threat models serve as inputs to DPIAs
2. LINDDUN threat modeling (Section 5) directly feeds DPIA risk identification
3. STRIDE threats against PII-processing components map to DPIA impact scenarios
4. ATT&CK techniques (T1530, T1005, T1567) map to DPIA breach scenarios
5. Threat model mitigations become DPIA "measures to address risk"

For threat modeling methodology details, see Chapter 55: Threat Modeling Operations.

5. LINDDUN Privacy Threat Modeling¶

5.1 Overview¶

LINDDUN is a privacy-specific threat modeling framework developed at KU Leuven. While STRIDE identifies security threats (Spoofing, Tampering, Repudiation, Information Disclosure, Denial of Service, Elevation of Privilege), LINDDUN identifies privacy threats. The two frameworks are complementary and should be applied together for systems processing personal data.

5.2 The Seven LINDDUN Threat Categories¶

5.3 LINDDUN Methodology Process¶

flowchart LR
    A[1. Define DFD] --> B[2. Map LINDDUN<br/>Threats to DFD]
    B --> C[3. Identify Threat<br/>Scenarios]
    C --> D[4. Prioritize<br/>Threats]
    D --> E[5. Select Privacy<br/>Patterns]
    E --> F[6. Map Patterns<br/>to Controls]
    F --> G[7. Validate &<br/>Document]

    style A fill:#3498db,color:#fff
    style D fill:#e74c3c,color:#fff
    style F fill:#2ecc71,color:#fff

5.4 LINDDUN Applied: SOC Telemetry Pipeline¶

Consider a typical SOC telemetry pipeline that collects endpoint data, processes it in a SIEM, and generates alerts for analyst review.

graph TB
    subgraph "Data Sources"
        EP[Endpoint Agent<br/>192.168.10.0/24]
        FW[Firewall Logs<br/>10.0.1.1]
        AD[Active Directory<br/>10.0.1.10]
        WP[Web Proxy<br/>10.0.1.20]
    end

    subgraph "Processing"
        COL[Log Collector<br/>10.0.2.5]
        SIEM[SIEM Platform<br/>10.0.2.10]
        UEBA[UEBA Engine<br/>10.0.2.15]
    end

    subgraph "Output"
        DASH[Analyst Dashboard]
        ALERT[Alert Queue]
        RPT[Reports]
    end

    EP --> COL
    FW --> COL
    AD --> COL
    WP --> COL
    COL --> SIEM
    SIEM --> UEBA
    SIEM --> DASH
    SIEM --> ALERT
    UEBA --> ALERT
    SIEM --> RPT

    style EP fill:#e74c3c,color:#fff
    style SIEM fill:#3498db,color:#fff
    style UEBA fill:#f39c12,color:#fff

LINDDUN Threat Analysis of SOC Pipeline:

5.5 LINDDUN-to-Privacy-Pattern Mapping¶

6. Privacy-Enhancing Technologies (PETs)¶

6.1 Differential Privacy¶

Differential privacy provides a mathematical guarantee that the output of a computation does not reveal whether any individual's data was included in the input dataset. It achieves this by adding calibrated noise to query results.

Formal Definition: A randomized algorithm M gives epsilon-differential privacy if for all datasets D1 and D2 differing on at most one element, and for all subsets S of outputs:

Pr[M(D1) in S] <= exp(epsilon) * Pr[M(D2) in S]

The privacy budget (epsilon) controls the privacy-utility tradeoff:

• epsilon < 1: Strong privacy, higher noise, lower utility
• epsilon = 1-3: Moderate privacy, balanced for most use cases
• epsilon > 10: Weak privacy, minimal noise, near-exact results

# Differential Privacy — Laplace Mechanism (Synthetic Example)
import numpy as np
from typing import Callable

class DifferentialPrivacy:
    """
    Implements the Laplace mechanism for epsilon-differential privacy.
    Adds calibrated noise to numeric query results.
    """

    def __init__(self, epsilon: float = 1.0):
        self.epsilon = epsilon
        self._privacy_budget_spent = 0.0
        self._query_count = 0

    def laplace_mechanism(
        self, true_value: float, sensitivity: float
    ) -> float:
        """
        Add Laplace noise calibrated to sensitivity/epsilon.

        Args:
            true_value: The exact query result
            sensitivity: Maximum change from one individual's data
        Returns:
            Noisy result satisfying epsilon-DP
        """
        scale = sensitivity / self.epsilon
        noise = np.random.laplace(0, scale)
        self._privacy_budget_spent += self.epsilon
        self._query_count += 1
        return true_value + noise

    def private_count(self, data: list, predicate: Callable) -> float:
        """Count elements matching predicate with DP noise (sensitivity=1)."""
        true_count = sum(1 for x in data if predicate(x))
        return self.laplace_mechanism(true_count, sensitivity=1.0)

    def private_mean(self, values: list[float], lower: float, upper: float) -> float:
        """Compute mean with DP noise. Values must be bounded."""
        n = len(values)
        clipped = [max(lower, min(upper, v)) for v in values]
        true_sum = sum(clipped)
        sensitivity = (upper - lower) / n
        noisy_sum = self.laplace_mechanism(true_sum, sensitivity=upper - lower)
        return noisy_sum / n

    @property
    def budget_remaining(self) -> str:
        return (f"Queries: {self._query_count}, "
                f"Total epsilon spent: {self._privacy_budget_spent:.2f}")

# Example: Privacy-preserving analytics
dp = DifferentialPrivacy(epsilon=1.0)

# Synthetic dataset: employee login hours (24-hour format)
login_hours = [8.5, 9.0, 8.0, 10.5, 7.5, 9.5, 8.0, 11.0, 9.0, 8.5,
               22.0, 23.5, 9.0, 8.5, 10.0, 7.0, 9.5, 8.0, 9.0, 8.5]

# Q1: How many employees log in before 9 AM?
early_count = dp.private_count(login_hours, lambda h: h < 9.0)
print(f"Employees logging in before 9 AM: {early_count:.1f}")
# True answer: 9; DP answer: ~9 +/- noise

# Q2: What is the average login hour?
avg_hour = dp.private_mean(login_hours, lower=0.0, upper=24.0)
print(f"Average login hour: {avg_hour:.1f}")
# True answer: 10.2; DP answer: ~10.2 +/- noise

print(dp.budget_remaining)
# Queries: 2, Total epsilon spent: 2.00

6.2 Homomorphic Encryption¶

Homomorphic encryption (HE) allows computation on encrypted data without decrypting it. The result, when decrypted, matches what would have been produced by performing the same computation on the plaintext.

Types:

SOC Application: A cloud MSSP can run detection queries on your encrypted logs without ever seeing the plaintext log data. The encrypted results are returned to you for decryption, preserving both security monitoring capability and data confidentiality.

6.3 Secure Multi-Party Computation (SMPC)¶

SMPC enables multiple parties to jointly compute a function over their private inputs without revealing those inputs to each other. Each party learns only the output and nothing about others' inputs beyond what can be inferred from the output itself.

Security Operations Use Case: Multiple organizations want to identify shared indicators of compromise (IOCs) without revealing their internal security telemetry to each other.

# Simplified SMPC: Private Set Intersection for IOC Sharing
# (Conceptual — real SMPC uses oblivious transfer / garbled circuits)
import hashlib
import secrets

class PrivateSetIntersection:
    """
    Simplified PSI protocol for IOC sharing between organizations.
    Each organization hashes their IOCs with a shared secret,
    then compares hashes to find common IOCs without revealing unique ones.

    NOTE: This is a simplified illustration. Production SMPC uses
    cryptographic protocols (OT, garbled circuits, secret sharing).
    """

    def __init__(self):
        self.shared_salt = secrets.token_hex(32)

    def _hash_elements(self, elements: set[str]) -> dict[str, str]:
        """Hash elements with shared salt."""
        return {
            hashlib.sha256(f"{self.shared_salt}:{e}".encode()).hexdigest(): e
            for e in elements
        }

    def find_intersection(
        self, org_a_iocs: set[str], org_b_iocs: set[str]
    ) -> set[str]:
        """Find common IOCs without revealing unique ones."""
        hashes_a = self._hash_elements(org_a_iocs)
        hashes_b = self._hash_elements(org_b_iocs)

        common_hashes = set(hashes_a.keys()) & set(hashes_b.keys())
        return {hashes_a[h] for h in common_hashes}

# Example: Two SOCs comparing IOCs
psi = PrivateSetIntersection()

soc_alpha_iocs = {
    "192.0.2.100",       # RFC 5737 — synthetic
    "198.51.100.50",     # RFC 5737 — synthetic
    "malware.example.com",
    "c2-server.example.com",
    "203.0.113.77",      # RFC 5737 — synthetic
}

soc_beta_iocs = {
    "192.0.2.100",       # Shared IOC
    "malware.example.com",  # Shared IOC
    "10.0.5.200",
    "dropper.example.com",
    "198.51.100.99",
}

shared = psi.find_intersection(soc_alpha_iocs, soc_beta_iocs)
print(f"Shared IOCs: {shared}")
# Output: {'192.0.2.100', 'malware.example.com'}
# Neither SOC learns about the other's unique IOCs

6.4 Federated Learning¶

Federated learning trains machine learning models across decentralized data sources without centralizing the raw data. Each participant trains a local model on their data and shares only model updates (gradients), not the data itself.

Privacy Benefits:

• Raw data never leaves the originating organization
• Only model gradients are shared (though gradient leakage attacks exist — see below)
• Reduces data aggregation risk and cross-border transfer issues

SOC Application: Multiple organizations collaboratively train a malware detection model without sharing their proprietary threat intelligence or endpoint telemetry.

6.5 k-Anonymity, l-Diversity, and t-Closeness¶

These are syntactic privacy models that transform datasets to prevent re-identification:

k-Anonymity: Every record in a dataset must be indistinguishable from at least k-1 other records with respect to quasi-identifiers (attributes that could enable re-identification when combined).

l-Diversity: Each equivalence class (group of k-identical records) must contain at least l "well-represented" values for the sensitive attribute. This prevents homogeneity attacks where all records in a k-anonymous group share the same sensitive value.

t-Closeness: The distribution of sensitive attributes within each equivalence class must be within distance t of the distribution in the overall dataset. This prevents skewness attacks where the distribution within a group reveals information.

# k-Anonymity Verification Script — Synthetic Example
import pandas as pd
from collections import Counter

def check_k_anonymity(
    df: pd.DataFrame,
    quasi_identifiers: list[str],
    k: int
) -> dict:
    """
    Verify k-anonymity of a dataset.

    Returns dict with status, minimum group size, and violating groups.
    """
    groups = df.groupby(quasi_identifiers).size().reset_index(name="count")

    min_group = groups["count"].min()
    violations = groups[groups["count"] < k]

    return {
        "k_target": k,
        "k_achieved": int(min_group),
        "is_k_anonymous": min_group >= k,
        "total_groups": len(groups),
        "violating_groups": len(violations),
        "violation_details": violations.to_dict("records") if len(violations) > 0 else [],
    }

def check_l_diversity(
    df: pd.DataFrame,
    quasi_identifiers: list[str],
    sensitive_attr: str,
    l: int
) -> dict:
    """Verify l-diversity: each equivalence class has >= l distinct sensitive values."""
    groups = df.groupby(quasi_identifiers)[sensitive_attr].nunique().reset_index(
        name="distinct_sensitive"
    )

    min_diversity = groups["distinct_sensitive"].min()
    violations = groups[groups["distinct_sensitive"] < l]

    return {
        "l_target": l,
        "l_achieved": int(min_diversity),
        "is_l_diverse": min_diversity >= l,
        "violating_groups": len(violations),
    }

# Synthetic patient dataset
data = pd.DataFrame({
    "age_range": ["20-30", "20-30", "20-30", "30-40", "30-40", "30-40",
                  "40-50", "40-50", "40-50"],
    "zip_prefix": ["479**", "479**", "479**", "480**", "480**", "480**",
                   "481**", "481**", "481**"],
    "condition": ["Flu", "Cold", "Allergy", "Flu", "Cold", "Diabetes",
                  "Cold", "Cold", "Cold"],  # Last group lacks diversity
})

qi = ["age_range", "zip_prefix"]
k_result = check_k_anonymity(data, qi, k=3)
l_result = check_l_diversity(data, qi, "condition", l=2)

print(f"k-Anonymity (k=3): {'PASS' if k_result['is_k_anonymous'] else 'FAIL'}")
print(f"l-Diversity (l=2): {'PASS' if l_result['is_l_diverse'] else 'FAIL'}")
# k=3: PASS (all groups have 3 records)
# l=2: FAIL (40-50/481** group has only 1 distinct condition: "Cold")

6.6 Synthetic Data Generation¶

Synthetic data is artificially generated data that preserves the statistical properties of the original dataset without containing any real personal data. It is increasingly used for testing, development, analytics, and ML model training where real data poses privacy risks.

Approaches:

• Statistical models: Generate data matching original distributions (means, variances, correlations)
• Generative Adversarial Networks (GANs): Train a generator to produce realistic synthetic records
• Variational Autoencoders (VAEs): Learn latent representations and generate new samples
• Rule-based generation: Apply domain rules to produce structurally valid but fictional data

6.7 PET Selection Matrix¶

7. Data Discovery & Classification¶

7.1 Automated PII Discovery¶

Effective privacy engineering requires knowing where personal data exists across all systems. Manual data inventories are incomplete by definition — you cannot protect what you do not know about. Automated PII discovery combines pattern matching, NLP, and entropy analysis to continuously scan data stores for personal information.

# Automated PII Discovery Scanner — Synthetic Example
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

class PIICategory(Enum):
    EMAIL = "email"
    PHONE = "phone"
    SSN = "social_security_number"
    CREDIT_CARD = "credit_card"
    IP_ADDRESS = "ip_address"
    DATE_OF_BIRTH = "date_of_birth"
    NAME = "person_name"
    ADDRESS = "postal_address"
    PASSPORT = "passport_number"
    IBAN = "iban"

@dataclass
class PIIFinding:
    category: PIICategory
    location: str
    column_or_field: str
    sample: str  # Redacted sample
    confidence: float
    count: int

class PIIScanner:
    """
    Scans data sources for PII using regex patterns and heuristics.
    All patterns are for detection only — never exfiltrate discovered PII.
    """

    PATTERNS = {
        PIICategory.EMAIL: re.compile(
            r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"
        ),
        PIICategory.PHONE: re.compile(
            r"\b(?:\+1[-.]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"
        ),
        PIICategory.SSN: re.compile(
            r"\b\d{3}-\d{2}-\d{4}\b"
        ),
        PIICategory.CREDIT_CARD: re.compile(
            r"\b(?:4\d{3}|5[1-5]\d{2}|3[47]\d{2}|6011)[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b"
        ),
        PIICategory.IP_ADDRESS: re.compile(
            r"\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b"
        ),
        PIICategory.DATE_OF_BIRTH: re.compile(
            r"\b(?:0[1-9]|1[0-2])[/-](?:0[1-9]|[12]\d|3[01])[/-](?:19|20)\d{2}\b"
        ),
        PIICategory.IBAN: re.compile(
            r"\b[A-Z]{2}\d{2}[A-Z0-9]{4}\d{7}(?:[A-Z0-9]?\d{0,16})\b"
        ),
    }

    COLUMN_NAME_INDICATORS = {
        PIICategory.EMAIL: {"email", "e_mail", "email_address", "mail"},
        PIICategory.PHONE: {"phone", "telephone", "mobile", "cell", "fax"},
        PIICategory.SSN: {"ssn", "social_security", "sin", "national_id", "tax_id"},
        PIICategory.NAME: {"name", "first_name", "last_name", "full_name",
                          "fname", "lname", "surname", "given_name"},
        PIICategory.ADDRESS: {"address", "street", "city", "zip", "postal",
                             "zip_code", "postal_code"},
        PIICategory.DATE_OF_BIRTH: {"dob", "birth_date", "date_of_birth", "birthday"},
    }

    def scan_text(self, text: str, source: str) -> list[PIIFinding]:
        """Scan text content for PII patterns."""
        findings = []
        for category, pattern in self.PATTERNS.items():
            matches = pattern.findall(text)
            if matches:
                # Redact sample for reporting
                sample = self._redact(matches[0], category)
                findings.append(PIIFinding(
                    category=category,
                    location=source,
                    column_or_field="text_content",
                    sample=sample,
                    confidence=0.85,
                    count=len(matches),
                ))
        return findings

    def scan_column_names(self, columns: list[str], source: str) -> list[PIIFinding]:
        """Scan database/CSV column names for PII indicators."""
        findings = []
        for col in columns:
            col_lower = col.lower().strip()
            for category, indicators in self.COLUMN_NAME_INDICATORS.items():
                if col_lower in indicators or any(ind in col_lower for ind in indicators):
                    findings.append(PIIFinding(
                        category=category,
                        location=source,
                        column_or_field=col,
                        sample="[column name match]",
                        confidence=0.70,
                        count=0,
                    ))
        return findings

    def _redact(self, value: str, category: PIICategory) -> str:
        """Redact PII for safe reporting."""
        if category == PIICategory.EMAIL:
            parts = value.split("@")
            return f"{parts[0][:2]}***@{parts[1]}" if len(parts) == 2 else "***"
        elif category == PIICategory.SSN:
            return "***-**-" + value[-4:]
        elif category == PIICategory.CREDIT_CARD:
            return "****-****-****-" + value[-4:]
        return value[:3] + "***"

# Example scan
scanner = PIIScanner()

# Scan a log file (synthetic content)
log_content = """
2026-04-10 10:23:45 INFO User testuser@example.com logged in from 192.0.2.45
2026-04-10 10:24:12 INFO Payment processed for card 4111-1111-1111-1111
2026-04-10 10:25:00 WARN Failed login for user admin@example.com from 198.51.100.33
2026-04-10 10:26:30 INFO SSN verification: 000-00-0000 matched record
"""

findings = scanner.scan_text(log_content, source="app-server.example.com:/var/log/app.log")
for f in findings:
    print(f"[{f.category.value}] Found {f.count} instance(s) in {f.location} "
          f"(confidence: {f.confidence:.0%}) — sample: {f.sample}")

# Scan database schema
db_columns = ["user_id", "email_address", "full_name", "phone_number",
              "date_of_birth", "account_balance", "last_login"]
schema_findings = scanner.scan_column_names(
    db_columns, source="db.example.com/users_table"
)
for f in schema_findings:
    print(f"[{f.category.value}] Column '{f.column_or_field}' in {f.location} "
          f"likely contains PII (confidence: {f.confidence:.0%})")

7.2 Data Classification Schema¶

7.3 Data Flow Mapping¶

graph TB
    subgraph "Collection Points"
        WEB[Web Forms<br/>portal.example.com]
        API[REST API<br/>api.example.com]
        MOB[Mobile App]
        IOT[IoT Sensors]
    end

    subgraph "Processing Layer"
        GW[API Gateway<br/>10.0.1.5]
        APP[Application Server<br/>10.0.2.10]
        ML[ML Pipeline<br/>10.0.2.20]
    end

    subgraph "Storage Layer"
        DB[(Primary DB<br/>Encrypted)]
        DW[(Data Warehouse<br/>Pseudonymized)]
        DL[(Data Lake<br/>Classified)]
        BK[(Backup<br/>Encrypted)]
    end

    subgraph "Output"
        RPT[Reports<br/>Aggregated]
        DASH[Dashboards<br/>Role-based]
        EXT[Third-Party<br/>Contractual]
    end

    WEB --> GW
    API --> GW
    MOB --> GW
    IOT --> GW
    GW --> APP
    APP --> DB
    APP --> ML
    DB --> DW
    DB --> BK
    DW --> DL
    DW --> RPT
    DW --> DASH
    APP --> EXT

    style DB fill:#e74c3c,color:#fff
    style DW fill:#f39c12,color:#fff
    style DL fill:#3498db,color:#fff

7.4 DLP Integration for Privacy¶

Data Loss Prevention (DLP) systems serve double duty: they prevent data exfiltration (security) and enforce data handling policies (privacy). Effective integration requires:

1. Content inspection rules aligned with data classification schema
2. Policy actions that enforce privacy controls (block, encrypt, quarantine, audit)
3. Endpoint DLP preventing PII in unauthorized locations (personal cloud storage, USB drives)
4. Network DLP inspecting outbound traffic for PII patterns
5. Cloud DLP scanning SaaS applications for unauthorized PII storage

For comprehensive DLP architecture and implementation, see Chapter 7: Data Loss Prevention.

8. Consent & Preference Management¶

8.1 Consent Management Platforms (CMPs)¶

A CMP manages the lifecycle of user consent: collection, storage, retrieval, modification, withdrawal, and evidence preservation. Under GDPR, consent must be freely given, specific, informed, and unambiguous. Under CCPA/CPRA, the model is opt-out rather than opt-in, but preference management is equally critical.

CMP Architecture Requirements:

8.2 Consent Receipt Specification¶

{
  "version": "1.1.0",
  "jurisdiction": "EU",
  "consentTimestamp": "2026-04-10T14:23:00Z",
  "collectionMethod": "web_form",
  "consentReceiptID": "CR-2026-04-10-7f3a9b2c",
  "publicKey": "-----BEGIN PUBLIC KEY-----\nREDACTED\n-----END PUBLIC KEY-----",
  "language": "en",
  "piiPrincipalId": "USR-PSE-a3f8c1d2",
  "piiControllers": [
    {
      "piiController": "SynthCorp International",
      "contact": "privacy@synthcorp.example.com",
      "address": "123 Example Street, Example City",
      "phone": "+1-555-0100"
    }
  ],
  "policyUrl": "https://synthcorp.example.com/privacy-policy",
  "services": [
    {
      "service": "Marketing Communications",
      "purposes": [
        {
          "purpose": "Email marketing about product updates",
          "purposeCategory": ["marketing"],
          "consentType": "EXPLICIT",
          "piiCategory": ["email_address", "first_name"],
          "primaryPurpose": true,
          "termination": "withdrawal or account deletion",
          "thirdPartyDisclosure": false,
          "thirdPartyName": null
        }
      ]
    },
    {
      "service": "Analytics",
      "purposes": [
        {
          "purpose": "Website usage analytics for service improvement",
          "purposeCategory": ["analytics"],
          "consentType": "EXPLICIT",
          "piiCategory": ["pseudonymized_browsing_data"],
          "primaryPurpose": false,
          "termination": "withdrawal or 90-day data deletion cycle",
          "thirdPartyDisclosure": true,
          "thirdPartyName": "AnalyticsCorp (analytics.example.com)"
        }
      ]
    }
  ],
  "sensitive": false,
  "spiCat": null
}

8.3 IAB Transparency & Consent Framework (TCF)¶

The IAB TCF provides a standardized mechanism for publishers and ad-tech vendors to collect and propagate consent for digital advertising. TCF 2.2 (current version) supports:

• Purpose-based consent (10 purposes including personalized ads, ad measurement, content personalization)
• Vendor-level consent (specific consent for each ad-tech vendor)
• Legitimate interest declarations with right to object
• Publisher restrictions overriding vendor declarations

8.4 Google Consent Mode¶

Google Consent Mode adjusts the behavior of Google Analytics and Google Ads tags based on user consent status:

9. Data Subject Rights Automation¶

9.1 DSR Workflow Architecture¶

flowchart TD
    A[DSR Request<br/>Received] --> B[Identity<br/>Verification]
    B --> C{Identity<br/>Verified?}
    C -->|No| D[Request Additional<br/>Verification]
    D --> B
    C -->|Yes| E[Classify<br/>Request Type]
    E --> F{Request Type}
    F -->|Access| G[Data Retrieval<br/>Pipeline]
    F -->|Deletion| H[Erasure<br/>Cascade]
    F -->|Correction| I[Data Update<br/>Pipeline]
    F -->|Portability| J[Export<br/>Pipeline]
    F -->|Opt-Out| K[Preference<br/>Update]
    G --> L[Compile<br/>Response]
    H --> L
    I --> L
    J --> L
    K --> L
    L --> M[Quality<br/>Review]
    M --> N[Deliver to<br/>Data Subject]
    N --> O[Archive Receipt<br/>& Evidence]

    style A fill:#3498db,color:#fff
    style H fill:#e74c3c,color:#fff
    style N fill:#2ecc71,color:#fff

9.2 Identity Verification¶

Before fulfilling any DSR, you must verify the requestor's identity. Fulfilling a fraudulent DSR is itself a privacy violation — disclosing personal data to an unauthorized party.

Verification Methods by Risk Level:

9.3 Erasure Cascade Implementation¶

The right to erasure (GDPR Art. 17, CCPA Sec. 1798.105) requires deletion of personal data across all systems where it is stored — not just the primary database. An erasure cascade must propagate deletion to:

• Primary databases
• Data warehouses and analytics stores
• Backup systems (with documented timeline)
• Log aggregation platforms (SIEM, log management)
• Third-party processors (contractual obligation)
• CDN caches
• Search engine indexes (Art. 17(2) — notify search engines)
• ML training datasets (retrain or unlearn)

# Erasure Cascade Orchestrator — Synthetic Example
import json
from datetime import datetime
from enum import Enum
from typing import Callable, Optional

class ErasureStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    EXEMPT = "exempt"  # Legal hold, retention requirement

class ErasureTarget:
    def __init__(
        self, name: str, system_type: str,
        delete_func: Callable[[str], bool],
        sla_hours: int = 72,
        exemption_check: Optional[Callable[[str], Optional[str]]] = None,
    ):
        self.name = name
        self.system_type = system_type
        self.delete_func = delete_func
        self.sla_hours = sla_hours
        self.exemption_check = exemption_check

class ErasureCascade:
    """Orchestrates deletion across all data stores."""

    def __init__(self):
        self.targets: list[ErasureTarget] = []
        self.audit_log: list[dict] = []

    def register_target(self, target: ErasureTarget) -> None:
        self.targets.append(target)

    def execute(self, user_id: str, request_id: str) -> dict:
        """Execute erasure cascade for a user across all registered targets."""
        results = {}
        start_time = datetime.utcnow()

        for target in self.targets:
            # Check for exemptions (legal hold, regulatory retention)
            if target.exemption_check:
                exemption = target.exemption_check(user_id)
                if exemption:
                    results[target.name] = {
                        "status": ErasureStatus.EXEMPT.value,
                        "reason": exemption,
                        "timestamp": datetime.utcnow().isoformat(),
                    }
                    self._audit(request_id, target.name, "EXEMPT", exemption)
                    continue

            # Execute deletion
            try:
                success = target.delete_func(user_id)
                status = ErasureStatus.COMPLETED if success else ErasureStatus.FAILED
                results[target.name] = {
                    "status": status.value,
                    "timestamp": datetime.utcnow().isoformat(),
                }
                self._audit(request_id, target.name, status.value)
            except Exception as e:
                results[target.name] = {
                    "status": ErasureStatus.FAILED.value,
                    "error": str(e),
                    "timestamp": datetime.utcnow().isoformat(),
                }
                self._audit(request_id, target.name, "FAILED", str(e))

        return {
            "request_id": request_id,
            "user_id": user_id,
            "started": start_time.isoformat(),
            "completed": datetime.utcnow().isoformat(),
            "results": results,
            "fully_erased": all(
                r["status"] in ("completed", "exempt")
                for r in results.values()
            ),
        }

    def _audit(self, request_id: str, target: str, status: str,
               detail: str = "") -> None:
        self.audit_log.append({
            "request_id": request_id,
            "target": target,
            "status": status,
            "detail": detail,
            "timestamp": datetime.utcnow().isoformat(),
        })

# Register erasure targets (synthetic)
cascade = ErasureCascade()

cascade.register_target(ErasureTarget(
    name="primary_db",
    system_type="PostgreSQL",
    delete_func=lambda uid: True,  # Simulated success
    sla_hours=24,
))

cascade.register_target(ErasureTarget(
    name="data_warehouse",
    system_type="BigQuery",
    delete_func=lambda uid: True,
    sla_hours=48,
))

cascade.register_target(ErasureTarget(
    name="siem_logs",
    system_type="Sentinel",
    delete_func=lambda uid: True,
    sla_hours=72,
    exemption_check=lambda uid: (
        "Legal hold LH-2026-003 active" if uid == "USR-HELD" else None
    ),
))

cascade.register_target(ErasureTarget(
    name="backup_system",
    system_type="Azure Backup",
    delete_func=lambda uid: True,
    sla_hours=720,  # 30 days for backup rotation
))

cascade.register_target(ErasureTarget(
    name="third_party_analytics",
    system_type="analytics.example.com API",
    delete_func=lambda uid: True,
    sla_hours=168,  # 7 days per processor agreement
))

# Execute erasure
result = cascade.execute("USR-12345", "DSR-2026-04-10-001")
print(json.dumps(result, indent=2))

9.4 Portability Formats¶

GDPR Article 20 requires data portability in a "structured, commonly used and machine-readable format." Common formats include:

9.5 DSR Fulfillment SLAs¶

10. Privacy Monitoring & Metrics¶

10.1 Privacy KPIs¶

10.2 KQL Detection Queries for Privacy Violations¶

// Unauthorized Bulk PII Access — Detects large-scale data access patterns
// that may indicate unauthorized data harvesting (T1119, T1005)
let pii_tables = dynamic(["customers", "employees", "patients", "users"]);
let bulk_threshold = 1000;
DatabaseAccessLogs
| where TimeGenerated > ago(1h)
| where DatabaseName has_any (pii_tables)
| where QueryType in ("SELECT", "EXPORT", "COPY")
| summarize
    TotalRows = sum(RowsReturned),
    QueryCount = count(),
    DistinctTables = dcount(TableName),
    Queries = make_set(QueryText, 10)
    by UserPrincipalName, SourceIP, bin(TimeGenerated, 5m)
| where TotalRows > bulk_threshold
| extend AlertSeverity = case(
    TotalRows > 100000, "Critical",
    TotalRows > 10000, "High",
    TotalRows > 1000, "Medium",
    "Low"
)
| project TimeGenerated, UserPrincipalName, SourceIP,
    TotalRows, QueryCount, DistinctTables, AlertSeverity, Queries

// Consent Bypass Detection — Identifies data processing without valid consent
// Correlates processing events with consent management system
let consent_valid = materialize(
    ConsentManagementLogs
    | where TimeGenerated > ago(30d)
    | where ConsentStatus == "active"
    | summarize arg_max(TimeGenerated, *) by UserId, ProcessingPurpose
    | project UserId, ProcessingPurpose, ConsentGranted = TimeGenerated
);
DataProcessingEvents
| where TimeGenerated > ago(1h)
| join kind=leftanti consent_valid
    on $left.SubjectId == $right.UserId,
       $left.Purpose == $right.ProcessingPurpose
| where Purpose != "security_monitoring"  // Legitimate interest exemption
| where Purpose != "legal_obligation"     // Legal obligation exemption
| project TimeGenerated, SubjectId, Purpose, ProcessingSystem,
    DataCategories, ProcessedBy
| extend Alert = "Data processed without valid consent record"