import base64 import hashlib import hmac import json import time from cryptography.hazmat.primitives.ciphers.aead import AESGCM from app.config import settings def get_aes_key() -> bytes: """Derive a 32-byte AES key from the configured encryption key.""" return hashlib.sha256(settings.ENCRYPTION_KEY.encode()).digest() def encrypt_temp_token(record_id: str, expires_in: int = 3600) -> str: """Create an encrypted token containing record_id and expiry timestamp.""" payload = json.dumps({"rid": record_id, "exp": int(time.time()) + expires_in}) aesgcm = AESGCM(get_aes_key()) nonce = AESGCM.generate_key(bit_length=96) ciphertext = aesgcm.encrypt(nonce, payload.encode(), None) token_bytes = nonce + ciphertext return base64.urlsafe_b64encode(token_bytes).decode() def decrypt_temp_token(token: str) -> str | None: """Decrypt token and return record_id if valid and not expired.""" try: token_bytes = base64.urlsafe_b64decode(token) nonce = token_bytes[:12] ciphertext = token_bytes[12:] aesgcm = AESGCM(get_aes_key()) payload = aesgcm.decrypt(nonce, ciphertext, None) data = json.loads(payload) if data["exp"] < time.time(): return None return data["rid"] except Exception: return None def encrypt_text(plaintext: str) -> str: """使用 AES-256-GCM 加密字符串,返回 base64 编码的密文。""" import os aesgcm = AESGCM(get_aes_key()) nonce = os.urandom(12) # 96-bit nonce for GCM ciphertext = aesgcm.encrypt(nonce, plaintext.encode(), None) return base64.urlsafe_b64encode(nonce + ciphertext).decode() def decrypt_text(token: str) -> str | None: """解密 AES-256-GCM 加密的字符串。失败返回 None。""" try: token_bytes = base64.urlsafe_b64decode(token) nonce = token_bytes[:12] ciphertext = token_bytes[12:] aesgcm = AESGCM(get_aes_key()) return aesgcm.decrypt(nonce, ciphertext, None).decode() except Exception: return None def hmac_sign(data: str) -> str: """Create HMAC-SHA256 signature.""" return hmac.new( settings.SECRET_KEY.encode(), data.encode(), hashlib.sha256 ).hexdigest() def hmac_verify(data: str, signature: str) -> bool: """Verify HMAC-SHA256 signature.""" expected = hmac_sign(data) return hmac.compare_digest(expected, signature)