Aadhaar (UIDAI).

# deduplication/abis_orchestrator.py

"""
ABIS (Automated Biometric Identification System) Orchestration

Aadhaar uses three independent ABIS vendors for deduplication.
This provides redundancy, accuracy, and vendor neutrality.
"""

from dataclasses import dataclass
from typing import List, Optional, Tuple
from enum import Enum
import asyncio


class DeduplicationResult(Enum):
    UNIQUE = "unique"           # No duplicates found
    DUPLICATE = "duplicate"     # Duplicate found
    MANUAL_REVIEW = "review"    # Uncertain, needs human review


@dataclass
class ABISMatch:
    candidate_aadhaar: str
    fingerprint_score: float      # 0-100
    iris_score: float             # 0-100
    fusion_score: float           # Combined score
    confidence: str               # HIGH, MEDIUM, LOW


@dataclass
class ABISResponse:
    vendor_id: str
    result: DeduplicationResult
    matches: List[ABISMatch]
    processing_time_ms: int


class ABISOrchestrator:
    """
    Orchestrates deduplication across three ABIS vendors.
    
    This design's multi-ABIS approach (Aadhaar itself shares the load
    across vendors and cross-checks suspected duplicates):
    1. Send enrollment to all three vendors in parallel
    2. Each vendor searches against their copy of the database
    3. Consensus determines final result
    """
    
    def __init__(
        self,
        abis_clients: List,  # Three ABIS vendor clients
        demographic_filter,
        manual_review_queue
    ):
        self.abis_clients = abis_clients
        self.demo_filter = demographic_filter
        self.review_queue = manual_review_queue
        
        # Thresholds for decision
        self.duplicate_threshold = 80  # Fusion score > 80 = duplicate
        self.unique_threshold = 30     # Fusion score < 30 = unique
    
    async def deduplicate(
        self,
        enrollment_packet: dict
    ) -> Tuple[DeduplicationResult, Optional[str]]:
        """
        Main deduplication flow.
        
        Returns:
            (result, duplicate_aadhaar if found)
        """
        # Step 1: Demographic blocking to reduce search space
        candidate_pool = await self.demo_filter.get_candidates(
            gender=enrollment_packet['gender'],
            dob=enrollment_packet['dob'],
            state=enrollment_packet['state'],
            name_phonetic=enrollment_packet['name_phonetic']
        )
        
        # Log the reduction achieved
        reduction_ratio = 1_400_000_000 / len(candidate_pool)
        print(f"Demographic blocking: {len(candidate_pool):,} candidates")
        print(f"Search space reduced by {reduction_ratio:,.0f}x")
        
        # Step 2: Send to all three ABIS in parallel
        abis_tasks = [
            client.search(
                fingerprints=enrollment_packet['fingerprints'],
                irises=enrollment_packet['irises'],
                candidate_pool=candidate_pool
            )
            for client in self.abis_clients
        ]
        
        responses: List[ABISResponse] = await asyncio.gather(*abis_tasks)
        
        # Step 3: Consensus decision
        return self._consensus_decision(responses)
    
    def _consensus_decision(
        self,
        responses: List[ABISResponse]
    ) -> Tuple[DeduplicationResult, Optional[str]]:
        """
        Three-way consensus logic.
        
        This design requires 2/3 agreement for automated decision.
        Mixed results go to manual adjudication.
        """
        unique_count = sum(
            1 for r in responses if r.result == DeduplicationResult.UNIQUE
        )
        duplicate_count = sum(
            1 for r in responses if r.result == DeduplicationResult.DUPLICATE
        )
        
        # Case 1: All three agree UNIQUE
        if unique_count == 3:
            return (DeduplicationResult.UNIQUE, None)
        
        # Case 2: All three agree DUPLICATE
        if duplicate_count == 3:
            # Find the matching Aadhaar (should be same across all)
            duplicate_aadhaar = self._find_common_match(responses)
            return (DeduplicationResult.DUPLICATE, duplicate_aadhaar)
        
        # Case 3: 2/3 agree UNIQUE
        if unique_count >= 2:
            return (DeduplicationResult.UNIQUE, None)
        
        # Case 4: 2/3 agree DUPLICATE
        if duplicate_count >= 2:
            duplicate_aadhaar = self._find_common_match(responses)
            return (DeduplicationResult.DUPLICATE, duplicate_aadhaar)
        
        # Case 5: No consensus — manual review
        # Queue for human adjudicator
        return (DeduplicationResult.MANUAL_REVIEW, None)
    
    def _find_common_match(
        self,
        responses: List[ABISResponse]
    ) -> Optional[str]:
        """
        Find the Aadhaar number that multiple ABIS agree is a duplicate.
        """
        # Collect all top matches
        match_counts = {}
        for response in responses:
            if response.matches:
                top_match = response.matches[0]
                aadhaar = top_match.candidate_aadhaar
                match_counts[aadhaar] = match_counts.get(aadhaar, 0) + 1
        
        # Return the one with most agreement
        if match_counts:
            return max(match_counts, key=match_counts.get)
        return None


class DemographicBlockingFilter:
    """
    Pre-filters the biometric search space using demographics.
    
    This is crucial for making billion-scale deduplication tractable.
    """
    
    def __init__(self, database):
        self.db = database
    
    async def get_candidates(
        self,
        gender: str,
        dob: str,
        state: str,
        name_phonetic: str
    ) -> List[str]:
        """
        Get candidate Aadhaar numbers that match demographic criteria.
        
        Reduces 1.4 billion to ~20,000-50,000 candidates.
        """
        # Parse DOB to get age range
        birth_year = int(dob[:4])
        age_range_start = birth_year - 5
        age_range_end = birth_year + 5
        
        # Query with demographic filters
        candidates = await self.db.query("""
            SELECT aadhaar_number 
            FROM residents
            WHERE gender = ?
              AND birth_year BETWEEN ? AND ?
              AND state_code = ?
              AND name_phonetic_hash = ?
        """, [gender, age_range_start, age_range_end, 
              state, self._phonetic_hash(name_phonetic)])
        
        return [row['aadhaar_number'] for row in candidates]
    
    def _phonetic_hash(self, name: str) -> str:
        """
        Generate phonetic hash for name matching.
        
        Handles spelling variations:
        "Rahul" and "Rahool" → same hash
        "Priya" and "Priyaa" → same hash
        """
        # Simplified phonetic algorithm (a real system would use a Soundex-style one)
        # Remove vowels, normalize consonants
        consonants = ''.join(c for c in name.upper() if c not in 'AEIOU')
        return consonants[:4].ljust(4, '0')

# authentication/auth_service.py

"""
Aadhaar Authentication Service

Processes 90+ million authentications per day with <200ms latency.
Returns only Yes/No — never exposes biometric or demographic data.
"""

from dataclasses import dataclass
from typing import Optional
from enum import Enum
from datetime import datetime, timedelta
import asyncio


class AuthMode(Enum):
    DEMOGRAPHIC = "demo"           # Name/address matching
    OTP = "otp"                    # One-time password
    FINGERPRINT = "fmr"            # Fingerprint biometric
    IRIS = "iir"                   # Iris biometric
    FACE = "fid"                   # Face biometric
    MULTI_FACTOR = "mf"            # Combination


@dataclass
class AuthRequest:
    aadhaar_number: str            # Or VID (Virtual ID)
    auth_mode: AuthMode
    pid_block: bytes               # Encrypted biometric/OTP
    aua_code: str                  # Authentication User Agency
    timestamp: datetime
    transaction_id: str
    consent: bool                  # Must be True


@dataclass
class AuthResponse:
    transaction_id: str
    status: str                    # "y" (yes) or "n" (no)
    error_code: Optional[str]      # If status is "n"
    auth_token: Optional[str]      # Unique token for this auth
    timestamp: datetime
    # NOTE: No PII is ever returned!


class AadhaarAuthService:
    """
    Core authentication service running in CIDR.
    
    Security principles:
    1. All requests encrypted with 2048-bit PKI
    2. Biometric data decrypted only inside CIDR
    3. Response is only Yes/No (no data leakage)
    4. All transactions logged for audit
    """
    
    def __init__(
        self,
        biometric_db,
        biometric_matcher,
        otp_service,
        hsm_client,          # Hardware Security Module
        audit_logger
    ):
        self.db = biometric_db
        self.matcher = biometric_matcher
        self.otp = otp_service
        self.hsm = hsm_client
        self.audit = audit_logger
        
        # Performance tuning
        self.template_cache = {}  # LRU cache for hot Aadhaars
        self.cache_ttl = 300      # 5 minutes
    
    async def authenticate(
        self,
        request: AuthRequest
    ) -> AuthResponse:
        """
        Main authentication flow.
        
        Must complete in <200ms.
        """
        start_time = datetime.utcnow()
        
        try:
            # Step 1: Validate request format
            self._validate_request(request)
            
            # Step 2: Decrypt PID block using HSM
            # Only CIDR's HSM can decrypt
            pid_data = await self.hsm.decrypt_pid(request.pid_block)
            
            # Step 3: Resolve Aadhaar number (handle VID)
            aadhaar = await self._resolve_aadhaar(request.aadhaar_number)
            
            # Step 4: Load resident's template (with caching)
            resident_template = await self._load_template(aadhaar)
            
            # Step 5: Perform matching based on auth mode
            match_result = await self._perform_match(
                mode=request.auth_mode,
                pid_data=pid_data,
                stored_template=resident_template
            )
            
            # Step 6: Generate response
            response = AuthResponse(
                transaction_id=request.transaction_id,
                status="y" if match_result.success else "n",
                error_code=match_result.error_code,
                auth_token=self._generate_token(aadhaar, request.aua_code),
                timestamp=datetime.utcnow()
            )
            
            # Step 7: Audit logging (async, don't wait)
            asyncio.create_task(self.audit.log(
                aadhaar=aadhaar,
                aua_code=request.aua_code,
                auth_mode=request.auth_mode,
                result=response.status,
                latency_ms=(datetime.utcnow() - start_time).total_seconds() * 1000
            ))
            
            return response
            
        except ValidationError as e:
            return self._error_response(request, str(e))
        except Exception as e:
            # Never expose internal errors
            return self._error_response(request, "INTERNAL_ERROR")
    
    async def _load_template(self, aadhaar: str) -> dict:
        """
        Load biometric template with caching.
        
        Hot Aadhaars (frequently authenticated) are cached.
        """
        # Check cache first
        if aadhaar in self.template_cache:
            cached = self.template_cache[aadhaar]
            if cached['expires'] > datetime.utcnow():
                return cached['template']
        
        # Cache miss — load from database
        template = await self.db.get_template(aadhaar)
        
        # Cache for hot Aadhaars
        self.template_cache[aadhaar] = {
            'template': template,
            'expires': datetime.utcnow() + timedelta(seconds=self.cache_ttl)
        }
        
        return template
    
    async def _perform_match(
        self,
        mode: AuthMode,
        pid_data: dict,
        stored_template: dict
    ) -> MatchResult:
        """
        Perform biometric/demographic/OTP matching.
        """
        if mode == AuthMode.FINGERPRINT:
            # 1:1 fingerprint matching
            return await self.matcher.match_fingerprint(
                captured=pid_data['fingerprint'],
                stored=stored_template['fingerprints'],
                finger_position=pid_data.get('position', 'ANY')
            )
        
        elif mode == AuthMode.IRIS:
            # 1:1 iris matching
            return await self.matcher.match_iris(
                captured=pid_data['iris'],
                stored=stored_template['irises']
            )
        
        elif mode == AuthMode.FACE:
            # AI-powered face matching
            return await self.matcher.match_face(
                captured=pid_data['face_image'],
                stored=stored_template['photo'],
                liveness_check=pid_data.get('liveness_data')
            )
        
        elif mode == AuthMode.OTP:
            # Verify OTP sent to registered mobile
            return await self.otp.verify(
                aadhaar=stored_template['aadhaar'],
                submitted_otp=pid_data['otp']
            )
        
        elif mode == AuthMode.DEMOGRAPHIC:
            # Fuzzy matching on name/address
            return self._demographic_match(
                submitted=pid_data['demographics'],
                stored=stored_template['demographics']
            )
    
    def _demographic_match(self, submitted: dict, stored: dict) -> MatchResult:
        """
        Fuzzy matching for name and address.
        
        Handles variations like:
        - "Raj Kumar" vs "Rajkumar"  
        - "Bangalore" vs "Bengaluru"
        - Hindi transliterations
        """
        name_score = self._fuzzy_name_match(
            submitted.get('name', ''),
            stored['name']
        )
        
        address_score = self._fuzzy_address_match(
            submitted.get('address', ''),
            stored['address']
        )
        
        # Thresholds from UIDAI guidelines
        if name_score >= submitted.get('match_threshold', 100):
            return MatchResult(success=True)
        
        return MatchResult(success=False, error_code="DEMO_MISMATCH")

# security/encryption_service.py

"""
Aadhaar Encryption Service

All biometric data is encrypted at the point of capture.
Only CIDR can decrypt using HSM-protected private keys.
"""

from dataclasses import dataclass
from typing import Tuple
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
import hmac
import hashlib
import os


class PIDBlockEncryption:
    """
    PID (Personal Identity Data) Block encryption.
    
    The PID block contains biometric/OTP data captured at
    the authentication point. It's encrypted before transmission.
    
    Encryption scheme (illustrative; UIDAI's API spec wraps the
    session key with RSA/ECB/PKCS1Padding, this sketch uses OAEP):
    1. Generate random session key (AES-256)
    2. Encrypt biometric data with session key
    3. Encrypt session key with UIDAI's public key (RSA-2048)
    4. Add HMAC for integrity
    """
    
    def __init__(self, uidai_public_key: bytes):
        self.uidai_public_key = serialization.load_pem_public_key(
            uidai_public_key
        )
    
    def encrypt_pid(
        self,
        biometric_data: bytes,
        timestamp: str,
        device_id: str
    ) -> Tuple[bytes, bytes]:
        """
        Encrypt PID block for transmission to CIDR.
        
        Returns:
            (encrypted_data, encrypted_session_key)
        """
        # Generate random session key
        session_key = os.urandom(32)  # 256 bits
        iv = os.urandom(12)  # 96 bits for GCM
        
        # Create PID plaintext with metadata
        pid_plaintext = self._create_pid_xml(
            biometric_data=biometric_data,
            timestamp=timestamp,
            device_id=device_id
        )
        
        # Encrypt with AES-256-GCM
        cipher = Cipher(algorithms.AES(session_key), modes.GCM(iv))
        encryptor = cipher.encryptor()
        ciphertext = encryptor.update(pid_plaintext) + encryptor.finalize()
        
        # Encrypt session key with UIDAI's RSA public key
        encrypted_session_key = self.uidai_public_key.encrypt(
            session_key + iv,  # Include IV
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hashes.SHA256()),
                algorithm=hashes.SHA256(),
                label=None
            )
        )
        
        # Add HMAC for integrity
        hmac_value = hmac.new(
            session_key,
            ciphertext,
            hashlib.sha256
        ).digest()
        
        return (ciphertext + encryptor.tag + hmac_value, 
                encrypted_session_key)


class HSMKeyManager:
    """
    Hardware Security Module interface.
    
    All cryptographic keys are stored and used within HSM.
    Keys never leave the HSM in plaintext.
    
    UIDAI requires FIPS 140-2 Level 3 (or higher) HSMs for partner
    agencies' keys; assume the same class here.
    """
    
    def __init__(self, hsm_connection):
        self.hsm = hsm_connection
    
    async def decrypt_pid(self, encrypted_pid: bytes, 
                          encrypted_key: bytes) -> dict:
        """
        Decrypt PID block inside HSM.
        
        The private key never leaves the HSM.
        Decryption happens entirely within the HSM.
        """
        # Send to HSM for decryption
        decrypted = await self.hsm.decrypt(
            data=encrypted_pid,
            encrypted_key=encrypted_key,
            key_id="UIDAI_AUTH_PRIVATE_KEY",
            algorithm="RSA-OAEP-256"
        )
        
        return self._parse_pid_xml(decrypted)
    
    async def sign_response(self, response_data: bytes) -> bytes:
        """
        Sign authentication response using HSM.
        
        All responses are digitally signed so AUAs can
        verify they came from authentic CIDR.
        """
        signature = await self.hsm.sign(
            data=response_data,
            key_id="UIDAI_SIGNING_KEY",
            algorithm="RSA-SHA256"
        )
        
        return signature


class AadhaarDataVault:
    """
    Secure storage for Aadhaar numbers at AUA/KUA.
    
    UIDAI mandates that any entity storing Aadhaar numbers
    must use an "Aadhaar Data Vault" with:
    - AES-256 encryption
    - Keys in HSM
    - Access logging
    - No plaintext storage
    """
    
    def __init__(self, hsm_client, database):
        self.hsm = hsm_client
        self.db = database
    
    async def store_aadhaar(
        self,
        reference_id: str,  # Your internal customer ID
        aadhaar_number: str
    ):
        """
        Store Aadhaar number securely.
        """
        # Generate reference key in HSM
        encrypted_aadhaar = await self.hsm.encrypt(
            data=aadhaar_number.encode(),
            key_id="AADHAAR_VAULT_KEY"
        )
        
        # Store only encrypted value
        await self.db.insert({
            'reference_id': reference_id,
            'encrypted_aadhaar': encrypted_aadhaar,
            'created_at': datetime.utcnow()
        })
    
    async def get_aadhaar(self, reference_id: str) -> str:
        """
        Retrieve and decrypt Aadhaar number.
        """
        record = await self.db.get(reference_id)
        
        # Decrypt in HSM
        aadhaar = await self.hsm.decrypt(
            data=record['encrypted_aadhaar'],
            key_id="AADHAAR_VAULT_KEY"
        )
        
        # Log access
        await self._log_access(reference_id)
        
        return aadhaar.decode()

# authentication/face_auth.py

"""
UIDAI Face Authentication Service

AI/ML-powered face matching developed in-house by UIDAI.
Uses liveness detection to prevent spoofing.
"""

from dataclasses import dataclass
from typing import Tuple
import numpy as np


@dataclass
class FaceAuthRequest:
    aadhaar_or_vid: str
    face_image: bytes              # Captured selfie
    liveness_data: dict            # Blink detection, head movement
    device_info: dict              # Camera specs, OS version


@dataclass
class FaceAuthResult:
    match: bool
    confidence: float              # 0-100
    liveness_verified: bool
    error_code: str = None


class FaceAuthenticationService:
    """
    Face authentication with liveness detection.
    
    Architecture:
    1. Liveness detection (prevent photo/video attacks)
    2. Face detection and alignment
    3. Feature extraction (deep learning)
    4. 1:1 matching against stored photo
    """
    
    def __init__(
        self,
        face_detector,
        liveness_model,
        face_encoder,
        match_threshold: float = 0.85
    ):
        self.detector = face_detector
        self.liveness = liveness_model
        self.encoder = face_encoder
        self.threshold = match_threshold
    
    async def authenticate(
        self,
        request: FaceAuthRequest,
        stored_photo: bytes
    ) -> FaceAuthResult:
        """
        Perform face authentication.
        
        Steps:
        1. Verify liveness (not a photo/video)
        2. Detect and align faces
        3. Extract embeddings
        4. Compare embeddings
        """
        # Step 1: Liveness verification
        liveness_result = await self._verify_liveness(
            image=request.face_image,
            liveness_data=request.liveness_data
        )
        
        if not liveness_result.is_live:
            return FaceAuthResult(
                match=False,
                confidence=0,
                liveness_verified=False,
                error_code="LIVENESS_FAILED"
            )
        
        # Step 2: Face detection and quality check
        captured_face = await self._detect_and_align(request.face_image)
        stored_face = await self._detect_and_align(stored_photo)
        
        if captured_face is None:
            return FaceAuthResult(
                match=False,
                confidence=0,
                liveness_verified=True,
                error_code="FACE_NOT_DETECTED"
            )
        
        # Step 3: Extract face embeddings
        captured_embedding = await self.encoder.encode(captured_face)
        stored_embedding = await self.encoder.encode(stored_face)
        
        # Step 4: Compare embeddings
        similarity = self._cosine_similarity(
            captured_embedding,
            stored_embedding
        )
        
        # Convert to percentage
        confidence = (similarity + 1) / 2 * 100
        
        return FaceAuthResult(
            match=confidence >= self.threshold * 100,
            confidence=confidence,
            liveness_verified=True
        )
    
    async def _verify_liveness(
        self,
        image: bytes,
        liveness_data: dict
    ) -> LivenessResult:
        """
        Verify the face is from a live person.
        
        Liveness checks:
        1. Texture analysis (detect printed photos)
        2. Depth estimation (detect flat screens)
        3. Eye blink detection (detect videos)
        4. Random challenge-response (head movement)
        """
        # Challenge: User was asked to blink/move head
        challenge_type = liveness_data.get('challenge_type')
        frames = liveness_data.get('frames', [])
        
        if challenge_type == 'BLINK':
            # Detect eye blink across frames
            return await self.liveness.detect_blink(frames)
        
        elif challenge_type == 'HEAD_TURN':
            # Detect head movement
            return await self.liveness.detect_head_movement(frames)
        
        elif challenge_type == 'PASSIVE':
            # Passive liveness (texture + depth analysis)
            return await self.liveness.passive_check(image)
        
        return LivenessResult(is_live=False)
    
    def _cosine_similarity(
        self,
        embedding1: np.ndarray,
        embedding2: np.ndarray
    ) -> float:
        """
        Cosine similarity between face embeddings.
        
        Range: -1 to 1 (higher = more similar)
        """
        dot_product = np.dot(embedding1, embedding2)
        norm1 = np.linalg.norm(embedding1)
        norm2 = np.linalg.norm(embedding2)
        
        return dot_product / (norm1 * norm2)


class LivenessDetector:
    """
    Deep learning model for liveness detection.
    
    Trained to distinguish:
    - Real faces
    - Printed photos
    - Screen displays (replay attacks)
    - 3D masks
    """
    
    def __init__(self, model_path: str):
        self.model = self._load_model(model_path)
    
    async def passive_check(self, image: bytes) -> LivenessResult:
        """
        Passive liveness without user action.
        
        Analyzes:
        - Moiré patterns (screen artifacts)
        - Color distribution
        - Texture frequency
        - Specular reflection
        """
        features = self._extract_liveness_features(image)
        
        # Model predicts: real (1) vs spoof (0)
        prediction = self.model.predict(features)
        
        return LivenessResult(
            is_live=prediction > 0.7,
            confidence=prediction,
            method="PASSIVE"
        )
    
    async def detect_blink(self, frames: list) -> LivenessResult:
        """
        Detect eye blink across video frames.
        
        A real person blinks; a photo doesn't.
        """
        eye_aspect_ratios = []
        
        for frame in frames:
            # Detect eyes
            eyes = self._detect_eyes(frame)
            if eyes:
                ear = self._eye_aspect_ratio(eyes)
                eye_aspect_ratios.append(ear)
        
        # Blink = dip in eye aspect ratio
        if len(eye_aspect_ratios) > 5:
            min_ear = min(eye_aspect_ratios)
            max_ear = max(eye_aspect_ratios)
            
            # Significant dip indicates blink
            if (max_ear - min_ear) > 0.15:
                return LivenessResult(is_live=True, method="BLINK")
        
        return LivenessResult(is_live=False, method="BLINK")