""" Enterprise Security & Cryptographic Engine Field-Level Encryption (AES-256-GCM), HMAC Anti-Tamper Signatures, and Token Extraction. """ import os import json import base64 import hmac import hashlib from typing import Dict, Any, Optional, Tuple from cryptography.hazmat.primitives.ciphers.aead import AESGCM from fastapi import HTTPException, status from app.core.config import settings _DEFAULT_SALT = b"sahihislam_staff_secret_salt_v1" _KEY_CACHE: Dict[str, Tuple[bytes, bytes]] = {} def _get_keys() -> Tuple[bytes, bytes]: """Derives 256-bit AES key and HMAC key using PBKDF2 with 100,000 iterations.""" secret = settings.SECRET_KEY if secret in _KEY_CACHE: return _KEY_CACHE[secret] aes_key = hashlib.pbkdf2_hmac("sha256", secret.encode("utf-8"), _DEFAULT_SALT + b"_aes", 100000, dklen=32) hmac_key = hashlib.pbkdf2_hmac("sha256", secret.encode("utf-8"), _DEFAULT_SALT + b"_hmac", 100000, dklen=32) _KEY_CACHE[secret] = (aes_key, hmac_key) return aes_key, hmac_key # Derive default module-level keys for backwards-compatibility with direct imports _AES_KEY, _HMAC_KEY = _get_keys() def parse_bool(val: Any) -> bool: """ Safely converts boolean, numeric, or string representations to bool. Avoids Python's bool("false") == True vulnerability. """ if isinstance(val, bool): return val if isinstance(val, (int, float)): return val != 0 if isinstance(val, str): return val.strip().lower() in ("true", "1", "yes", "t", "active") return bool(val) def encrypt_staff_payload(data: Dict[str, Any]) -> str: """ Encrypts a permission dictionary with AES-256-GCM. Returns ciphertext prefixed with 'v1:' version identifier for rotation readiness. """ try: aes_key, _ = _get_keys() aesgcm = AESGCM(aes_key) nonce = os.urandom(12) raw_json = json.dumps(data, separators=(',', ':'), sort_keys=True).encode("utf-8") ciphertext = aesgcm.encrypt(nonce, raw_json, None) combined = nonce + ciphertext encoded = base64.urlsafe_b64encode(combined).decode("ascii") return f"v1:{encoded}" except Exception as e: raise ValueError(f"Failed to encrypt staff payload: {e}") def decrypt_staff_payload(encrypted_str: str) -> Dict[str, Any]: """ Decrypts an AES-256-GCM ciphertext string back into dictionary. Supports both 'v1:' versioned strings and legacy unversioned payloads. """ try: raw_payload = encrypted_str.strip() if raw_payload.startswith("v1:"): raw_payload = raw_payload[3:] combined = base64.urlsafe_b64decode(raw_payload.encode("ascii")) if len(combined) < 28: raise ValueError("Invalid ciphertext length: minimum 28 bytes required") nonce = combined[:12] ciphertext = combined[12:] aes_key, _ = _get_keys() aesgcm = AESGCM(aes_key) decrypted_bytes = aesgcm.decrypt(nonce, ciphertext, None) return json.loads(decrypted_bytes.decode("utf-8")) except Exception as e: raise ValueError(f"Failed to decrypt staff payload: {e}") def generate_staff_hmac(uid: str, email: str, allowed_tabs: list, is_active: Any) -> str: """ Generates an HMAC-SHA256 signature for a staff permission state. Uses canonical deterministic JSON serialization to eliminate delimiter collision. """ _, hmac_key = _get_keys() sorted_tabs = sorted([str(t).strip().lower() for t in allowed_tabs if t is not None]) payload = { "allowed_tabs": sorted_tabs, "email": email.strip().lower(), "is_active": parse_bool(is_active), "uid": uid.strip(), } canonical_bytes = json.dumps(payload, sort_keys=True, separators=(',', ':')).encode("utf-8") return hmac.new(hmac_key, canonical_bytes, hashlib.sha256).hexdigest() def verify_staff_hmac(uid: str, email: str, allowed_tabs: list, is_active: Any, signature: str) -> bool: """ Verifies that a record's HMAC signature matches its content. Supports canonical JSON signature as well as legacy delimiter joining for backwards compatibility. """ if not signature: return False _, hmac_key = _get_keys() # 1. Primary verification: Canonical JSON HMAC expected = generate_staff_hmac(uid, email, allowed_tabs, is_active) if hmac.compare_digest(expected, signature): return True # 2. Backward compatibility: Legacy delimiter string with safe bool parsing sorted_tabs = sorted([str(t).strip().lower() for t in allowed_tabs if t is not None]) active_bool = parse_bool(is_active) legacy_str = f"{uid.strip()}:{email.strip().lower()}:{','.join(sorted_tabs)}:{active_bool}" legacy_sig = hmac.new(hmac_key, legacy_str.encode("utf-8"), hashlib.sha256).hexdigest() if hmac.compare_digest(legacy_sig, signature): return True # 3. Backward compatibility: Legacy string with raw bool coercion raw_bool_str = f"{uid.strip()}:{email.strip().lower()}:{','.join(sorted_tabs)}:{bool(is_active)}" raw_bool_sig = hmac.new(hmac_key, raw_bool_str.encode("utf-8"), hashlib.sha256).hexdigest() return hmac.compare_digest(raw_bool_sig, signature) def extract_bearer_token(authorization: Optional[str], x_firebase_token: Optional[str]) -> str: """ Extracts raw bearer token string from Authorization header or X-Firebase-Token header. NOTE: This utility ONLY performs header extraction and syntax parsing. Cryptographic verification (signature, issuer, expiration, audience) must be performed downstream by verify_firebase_token() or FirebaseAuth middleware. """ if isinstance(authorization, str) and authorization.lower().startswith("bearer "): token = authorization.split(" ", 1)[1].strip() if token: return token if isinstance(x_firebase_token, str): token = x_firebase_token.strip() if token: return token raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing or invalid Authorization header", headers={"WWW-Authenticate": "Bearer"}, )