admin / Syanpse-Vanguard

"""AES-256-GCM credential vault — Python port of Synapse Sonar's lib/crypto.ts.

Secrets (integration configs, AI API keys) are stored as an EncryptedBlob JSON
{iv, authTag, ciphertext} (all base64). The plaintext never leaves the server
once stored; the UI only ever sees mask_secret()."""
from __future__ import annotations

import base64
import json
from dataclasses import dataclass

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

from sv.config import settings


@dataclass(frozen=True)
class EncryptedBlob:
    iv: str
    auth_tag: str
    ciphertext: str

    def to_json(self) -> str:
        return json.dumps({"iv": self.iv, "authTag": self.auth_tag,
                           "ciphertext": self.ciphertext})

    @classmethod
    def from_json(cls, raw: str) -> "EncryptedBlob":
        d = json.loads(raw)
        return cls(iv=d["iv"], auth_tag=d["authTag"], ciphertext=d["ciphertext"])


def _key() -> bytes:
    hex_key = settings.vault_key
    if len(hex_key) != 64:
        raise ValueError("SV_VAULT_KEY must be 64 hex characters (32 bytes).")
    return bytes.fromhex(hex_key)


def encrypt_json(payload: dict) -> EncryptedBlob:
    """AESGCM in the `cryptography` lib appends the 16-byte tag to the ciphertext;
    we split it back out to match Sonar's {iv, authTag, ciphertext} shape."""
    iv = _rand(12)  # 96-bit nonce recommended for GCM
    aes = AESGCM(_key())
    ct_and_tag = aes.encrypt(iv, json.dumps(payload).encode(), None)
    ciphertext, tag = ct_and_tag[:-16], ct_and_tag[-16:]
    return EncryptedBlob(
        iv=_b64(iv), auth_tag=_b64(tag), ciphertext=_b64(ciphertext)
    )


def decrypt_json(blob: EncryptedBlob) -> dict:
    aes = AESGCM(_key())
    ct_and_tag = _unb64(blob.ciphertext) + _unb64(blob.auth_tag)
    plaintext = aes.decrypt(_unb64(blob.iv), ct_and_tag, None)
    return json.loads(plaintext.decode())


def mask_secret(value: str) -> str:
    """Render a secret for the UI as ••••••••<last4>, matching Sonar."""
    if not value:
        return ""
    return "•" * 8 + value[-4:]


def _rand(n: int) -> bytes:
    import os
    return os.urandom(n)


def _b64(b: bytes) -> str:
    return base64.b64encode(b).decode()


def _unb64(s: str) -> bytes:
    return base64.b64decode(s)