""" Cryptographic helpers — password hashing, Fernet encryption, TOTP, JWT tokens. """ import base64 import hashlib import hmac import secrets import struct import time from datetime import datetime, timedelta import jwt as pyjwt from cryptography.fernet import Fernet as _Fernet from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC as _PBKDF2HMAC from cryptography.hazmat.primitives import hashes as _hashes from config import APP_SECRET, JWT_ALG, JWT_EXP_H # ── Password Hashing (PBKDF2-SHA256, 100k iterations) ─────────────────────── def _hash_pw(pw): salt = secrets.token_hex(16) h = hashlib.pbkdf2_hmac("sha256", pw.encode(), salt.encode(), 100_000) return f"{salt}:{h.hex()}" def _verify_pw(pw, stored): salt, h = stored.split(":") return hmac.compare_digest( hashlib.pbkdf2_hmac("sha256", pw.encode(), salt.encode(), 100_000).hex(), h ) # ── TOTP (RFC 6238, Google Authenticator compatible) ───────────────────────── def _totp_secret(): return base64.b32encode(secrets.token_bytes(20)).decode() def _totp_verify(secret, code, window=1): key = base64.b32decode(secret) now = int(time.time()) // 30 for off in range(-window, window + 1): msg = struct.pack(">Q", now + off) h = hmac.new(key, msg, hashlib.sha1).digest() o = h[-1] & 0x0F c = str((struct.unpack(">I", h[o:o+4])[0] & 0x7FFFFFFF) % 1_000_000).zfill(6) if hmac.compare_digest(c, code): return True return False def _totp_uri(secret, user): return f"otpauth://totp/AI-Agent:{user}?secret={secret}&issuer=AI-Agent" # ── JWT Token ──────────────────────────────────────────────────────────────── def _make_token(uid, role, sid): return pyjwt.encode( {"sub": uid, "role": role, "sid": sid, "exp": datetime.utcnow() + timedelta(hours=JWT_EXP_H), "iat": datetime.utcnow()}, APP_SECRET, algorithm=JWT_ALG ) # ── Fernet Encryption (AES-128-CBC + HMAC-SHA256) ─────────────────────────── _FERNET_PREFIX = "fernet:" def _derive_fernet_key(secret: str) -> bytes: kdf = _PBKDF2HMAC( algorithm=_hashes.SHA256(), length=32, salt=b"oci-cis-agent-fernet-v1", iterations=100_000 ) return base64.urlsafe_b64encode(kdf.derive(secret.encode())) _fernet = _Fernet(_derive_fernet_key(APP_SECRET)) def _enc(v): if not v: return v return _FERNET_PREFIX + _fernet.encrypt(v.encode()).decode() def _dec(v): if not v: return v if v.startswith(_FERNET_PREFIX): return _fernet.decrypt(v[len(_FERNET_PREFIX):].encode()).decode() return base64.b64decode(v.encode()).decode() def _mask(v, show=6): if not v or len(v) <= show: return "\u2022" * 8 return "\u2022" * min(len(v) - show, 20) + v[-show:] def _safe_dec(v): if not v: return v try: return _dec(v) except Exception: return v