services/common/security.py

"""Security helpers for JWT and AES-256-GCM."""

import base64
import os
from datetime import datetime, timedelta, timezone
from typing import Any, Dict

import jwt
from cryptography.hazmat.primitives.ciphers.aead import AESGCM


def create_jwt(payload: Dict[str, Any], secret: str, expires_minutes: int = 60) -> str:
    """Create JWT token with expiration timestamp."""
    exp = datetime.now(tz=timezone.utc) + timedelta(minutes=expires_minutes)
    data = {**payload, "exp": exp}
    return jwt.encode(data, secret, algorithm="HS256")


def verify_jwt(token: str, secret: str) -> Dict[str, Any]:
    """Validate JWT token and return payload."""
    return jwt.decode(token, secret, algorithms=["HS256"])


def encrypt_aes_gcm(plaintext: bytes, key_b64: str) -> bytes:
    """Encrypt bytes with AES-256-GCM and prepend nonce."""
    key = base64.b64decode(key_b64)
    nonce = os.urandom(12)
    cipher = AESGCM(key)
    ciphertext = cipher.encrypt(nonce, plaintext, None)
    return nonce + ciphertext


def decrypt_aes_gcm(blob: bytes, key_b64: str) -> bytes:
    """Decrypt blob where first 12 bytes are nonce."""
    key = base64.b64decode(key_b64)
    nonce, ciphertext = blob[:12], blob[12:]
    cipher = AESGCM(key)
    return cipher.decrypt(nonce, ciphertext, None)