Spaces:
Running on Zero
Running on Zero
feat(backend): centralized environment configuration, Razorpay standard web checkout, Google Play keyless auth & security hardening
bbd07c7 | """ | |
| 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"}, | |
| ) | |