Privacy Filter in Machine Learning

from presidio_analyzer import AnalyzerEngine, PatternRecognizer, Pattern
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

# Create a custom Aadhaar number recognizer
aadhaar_pattern = Pattern(
    name="aadhaar_pattern",
    regex=r"\b[2-9]\d{3}\s?\d{4}\s?\d{4}\b",
    score=0.85
)

aadhaar_recognizer = PatternRecognizer(
    supported_entity="IN_AADHAAR",
    name="Aadhaar Recognizer",
    patterns=[aadhaar_pattern],
    supported_language="en",
    context=["aadhaar", "uid", "uidai", "aadhar"]
)

# Create a custom PAN number recognizer
pan_pattern = Pattern(
    name="pan_pattern",
    regex=r"\b[A-Z]{5}[0-9]{4}[A-Z]\b",
    score=0.80
)

pan_recognizer = PatternRecognizer(
    supported_entity="IN_PAN",
    name="PAN Recognizer",
    patterns=[pan_pattern],
    supported_language="en",
    context=["pan", "permanent account", "income tax"]
)

# Initialize engines
analyzer = AnalyzerEngine()
analyzer.registry.add_recognizer(aadhaar_recognizer)
analyzer.registry.add_recognizer(pan_recognizer)
anonymizer = AnonymizerEngine()

# Sample text with Indian PII
text = (
    "Customer Rajesh Kumar, Aadhaar 9876 5432 1098, "
    "PAN ABCDE1234F, email [email protected], "
    "mobile +91-98765-43210, lives in Bengaluru."
)

# Detect PII entities
results = analyzer.analyze(
    text=text,
    entities=["PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER",
              "IN_AADHAAR", "IN_PAN", "LOCATION"],
    language="en"
)

print(f"Detected {len(results)} PII entities:")
for r in results:
    print(f"  {r.entity_type}: '{text[r.start:r.end]}' (score={r.score:.2f})")

# Anonymize with type-specific operators
operators = {
    "PERSON": OperatorConfig("replace", {"new_value": "[PERSON]"}),
    "IN_AADHAAR": OperatorConfig("mask", {
        "type": "mask", "masking_char": "X",
        "chars_to_mask": 8, "from_end": False
    }),
    "IN_PAN": OperatorConfig("replace", {"new_value": "[PAN_REDACTED]"}),
    "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "[EMAIL]"}),
    "PHONE_NUMBER": OperatorConfig("replace", {"new_value": "[PHONE]"}),
    "LOCATION": OperatorConfig("replace", {"new_value": "[LOCATION]"}),
}

anonymized = anonymizer.anonymize(
    text=text, analyzer_results=results, operators=operators
)
print(f"\nAnonymized: {anonymized.text}")
# Output: Customer [PERSON], Aadhaar XXXXXXXX1098, [PAN_REDACTED], ...

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
from opacus import PrivacyEngine
from opacus.validators import ModuleValidator

# Define a simple model
model = nn.Sequential(
    nn.Linear(768, 256),
    nn.ReLU(),
    nn.Linear(256, 64),
    nn.ReLU(),
    nn.Linear(64, 10)
)

# Opacus requires models to be compatible (no BatchNorm, etc.)
model = ModuleValidator.fix(model)
errors = ModuleValidator.validate(model, strict=False)
assert len(errors) == 0, f"Model incompatible with DP: {errors}"

# Create dummy data (replace with real data)
X = torch.randn(10000, 768)
y = torch.randint(0, 10, (10000,))
dataset = TensorDataset(X, y)
dataloader = DataLoader(dataset, batch_size=256, shuffle=True)

# Configure optimizer
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)

# Attach PrivacyEngine with DP guarantees
privacy_engine = PrivacyEngine()
model, optimizer, dataloader = privacy_engine.make_private_with_epsilon(
    module=model,
    optimizer=optimizer,
    data_loader=dataloader,
    epochs=10,
    target_epsilon=8.0,       # Privacy budget
    target_delta=1e-5,        # Failure probability
    max_grad_norm=1.0,        # Gradient clipping bound
)

# Training loop with DP
criterion = nn.CrossEntropyLoss()
for epoch in range(10):
    total_loss = 0.0
    for batch_X, batch_y in dataloader:
        optimizer.zero_grad()
        output = model(batch_X)
        loss = criterion(output, batch_y)
        loss.backward()
        optimizer.step()  # DP noise added automatically
        total_loss += loss.item()

    # Check current privacy budget spent
    epsilon = privacy_engine.get_epsilon(delta=1e-5)
    print(f"Epoch {epoch+1}: loss={total_loss/len(dataloader):.4f}, "
          f"epsilon={epsilon:.2f}")

print(f"\nFinal privacy guarantee: (epsilon={epsilon:.2f}, delta=1e-5)-DP")

import pandas as pd
import numpy as np
from typing import Dict, List


def generalize_age(age: int, level: int = 1) -> str:
    """Generalize age into ranges based on level."""
    if level == 0:
        return str(age)
    elif level == 1:
        bucket = (age // 5) * 5
        return f"{bucket}-{bucket + 4}"
    elif level == 2:
        bucket = (age // 10) * 10
        return f"{bucket}-{bucket + 9}"
    else:
        return "*"


def generalize_pincode(pincode: str, level: int = 1) -> str:
    """Generalize Indian PIN code by masking trailing digits."""
    if level == 0:
        return pincode
    elif level == 1:
        return pincode[:4] + "XX"   # e.g., 560001 -> 5600XX
    elif level == 2:
        return pincode[:3] + "XXX"  # e.g., 560001 -> 560XXX
    else:
        return "XXXXXX"


def check_k_anonymity(df: pd.DataFrame,
                       quasi_identifiers: List[str],
                       k: int) -> bool:
    """Check if dataset satisfies k-anonymity."""
    group_sizes = df.groupby(quasi_identifiers).size()
    return group_sizes.min() >= k


def apply_k_anonymity(df: pd.DataFrame,
                       quasi_identifiers: List[str],
                       k: int,
                       generalizers: Dict) -> pd.DataFrame:
    """Apply k-anonymity through iterative generalization."""
    result = df.copy()
    levels = {qi: 0 for qi in quasi_identifiers}

    while not check_k_anonymity(result, quasi_identifiers, k):
        # Find the quasi-identifier with the most unique values
        unique_counts = {
            qi: result[qi].nunique() for qi in quasi_identifiers
        }
        target_qi = max(unique_counts, key=unique_counts.get)

        # Increase generalization level
        levels[target_qi] += 1
        if levels[target_qi] > 3:
            raise ValueError(
                f"Cannot achieve {k}-anonymity without excessive "
                f"information loss on {target_qi}"
            )
        result[target_qi] = result[target_qi].apply(
            lambda x: generalizers[target_qi](x, levels[target_qi])
        )
        print(f"Generalized {target_qi} to level {levels[target_qi]}, "
              f"unique values: {result[target_qi].nunique()}")

    return result


# Example: Patient records from an Indian hospital
df = pd.DataFrame({
    "age": [28, 29, 31, 34, 36, 42, 43, 45, 50, 52],
    "gender": ["M", "F", "M", "F", "M", "F", "M", "F", "M", "F"],
    "pincode": ["560001", "560034", "560002", "560078",
                "400001", "400053", "400012", "110001",
                "110045", "110023"],
    "diagnosis": ["diabetes", "flu", "diabetes", "cancer",
                  "flu", "diabetes", "cancer", "flu",
                  "diabetes", "cancer"]
})

quasi_ids = ["age", "gender", "pincode"]
generalizers = {
    "age": generalize_age,
    "pincode": generalize_pincode,
    "gender": lambda x, l: x if l == 0 else "*"
}

anonymized = apply_k_anonymity(df, quasi_ids, k=2, generalizers=generalizers)
print("\nAnonymized dataset:")
print(anonymized)
print(f"\nK-anonymity satisfied (k=2): "
      f"{check_k_anonymity(anonymized, quasi_ids, 2)}")

import re
import json
import hashlib
from dataclasses import dataclass, field
from typing import List, Optional, Callable
from datetime import datetime


@dataclass
class PIIEntity:
    entity_type: str
    value: str
    start: int
    end: int
    confidence: float
    transformation: str = ""
    transformed_value: str = ""


@dataclass
class AuditRecord:
    timestamp: str
    document_id: str
    entities_detected: int
    entity_types: List[str]
    transformations_applied: List[str]


class PrivacyFilter:
    """Production-grade privacy filter for streaming pipelines."""

    def __init__(self, hmac_key: str = "default-key"):
        self.hmac_key = hmac_key.encode()
        self.patterns = self._compile_patterns()
        self.audit_log: List[AuditRecord] = []

    def _compile_patterns(self) -> dict:
        return {
            "IN_AADHAAR": re.compile(
                r"\b[2-9]\d{3}[\s-]?\d{4}[\s-]?\d{4}\b"
            ),
            "IN_PAN": re.compile(
                r"\b[A-Z]{5}[0-9]{4}[A-Z]\b"
            ),
            "EMAIL": re.compile(
                r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b"
            ),
            "IN_PHONE": re.compile(
                r"(?:\+91[\s-]?)?[6-9]\d{4}[\s-]?\d{5}\b"
            ),
            "CREDIT_CARD": re.compile(
                r"\b(?:4\d{3}|5[1-5]\d{2}|6011)[-\s]?\d{4}"
                r"[-\s]?\d{4}[-\s]?\d{4}\b"
            ),
            "IN_PASSPORT": re.compile(
                r"\b[A-Z][1-9]\d{6}\b"
            ),
        }

    def _pseudonymize(self, value: str, entity_type: str) -> str:
        """Generate a consistent pseudonym using HMAC."""
        h = hashlib.sha256(self.hmac_key + value.encode())
        digest = h.hexdigest()[:12]
        return f"PSEUDO_{entity_type}_{digest}"

    def _mask_aadhaar(self, value: str) -> str:
        """Mask Aadhaar per UIDAI regulations: show only last 4."""
        digits = re.sub(r"[\s-]", "", value)
        return f"XXXX-XXXX-{digits[-4:]}"

    def detect_and_transform(
        self, text: str, doc_id: str = "unknown"
    ) -> tuple:
        """Detect PII and apply transformations."""
        entities: List[PIIEntity] = []

        for entity_type, pattern in self.patterns.items():
            for match in pattern.finditer(text):
                entity = PIIEntity(
                    entity_type=entity_type,
                    value=match.group(),
                    start=match.start(),
                    end=match.end(),
                    confidence=0.90
                )

                # Apply type-specific transformation
                if entity_type == "IN_AADHAAR":
                    entity.transformation = "mask"
                    entity.transformed_value = self._mask_aadhaar(
                        entity.value
                    )
                elif entity_type in ("EMAIL", "CREDIT_CARD"):
                    entity.transformation = "redact"
                    entity.transformed_value = f"[{entity_type}]"
                else:
                    entity.transformation = "pseudonymize"
                    entity.transformed_value = self._pseudonymize(
                        entity.value, entity_type
                    )
                entities.append(entity)

        # Apply transformations (reverse order to preserve indices)
        sanitized = text
        for entity in sorted(entities, key=lambda e: e.start,
                             reverse=True):
            sanitized = (
                sanitized[:entity.start]
                + entity.transformed_value
                + sanitized[entity.end:]
            )

        # Log audit record
        self.audit_log.append(AuditRecord(
            timestamp=datetime.utcnow().isoformat(),
            document_id=doc_id,
            entities_detected=len(entities),
            entity_types=[e.entity_type for e in entities],
            transformations_applied=[e.transformation for e in entities]
        ))

        return sanitized, entities


# Usage
filter_instance = PrivacyFilter(hmac_key="my-secret-key-2026")

record = (
    "Payment from Aadhaar 9876 5432 1098, card 4111-1111-1111-1111, "
    "email [email protected], PAN ABCDE1234F, phone +91-98765-43210"
)

sanitized, entities = filter_instance.detect_and_transform(
    record, doc_id="txn-001"
)

print(f"Original: {record}")
print(f"Sanitized: {sanitized}")
print(f"Entities found: {len(entities)}")
for e in entities:
    print(f"  {e.entity_type}: '{e.value}' -> '{e.transformed_value}' "
          f"({e.transformation})")

# Privacy filter configuration (YAML)
privacy_filter:
  detection:
    engines:
      - type: regex
        entities: [IN_AADHAAR, IN_PAN, EMAIL, CREDIT_CARD, IN_PHONE]
      - type: ner
        model: en_core_web_trf
        entities: [PERSON, LOCATION, ORG, DATE_TIME]
        confidence_threshold: 0.60
      - type: presidio
        language: en
        entities: all
  
  transformation_policies:
    IN_AADHAAR:
      method: mask
      mask_char: "X"
      chars_to_show: 4
      show_from: end  # UIDAI regulation
    PERSON:
      method: pseudonymize
      consistency: document  # Same name -> same pseudonym within doc
    EMAIL:
      method: redact
      replacement: "[EMAIL]"
    CREDIT_CARD:
      method: redact
      replacement: "[CARD]"
    DEFAULT:
      method: redact
      replacement: "[REDACTED]"
  
  differential_privacy:
    enabled: true
    target_epsilon: 8.0
    target_delta: 1e-5
    max_grad_norm: 1.0
    noise_multiplier: auto  # Computed from epsilon/delta/epochs
  
  audit:
    enabled: true
    storage: azure_table_storage
    table_name: pii_audit_log
    retention_days: 365
  
  compliance:
    frameworks: [GDPR, DPDP_ACT]
    breach_notification_hours: 72
    data_retention_policy: minimal