feat(callbacks): add AES-GCM helper for callback secrets

"""

Encryption for client-supplied callback secrets (sinkCredential, notificationAuthToken)

Follows the portal's session_crypto pattern, but with one app-wide key from settings:

OEG decrypts in the background when delivering a notification, with no client request

in flight to carry a per-session key.

"""

import os

from cryptography.exceptions import InvalidTag

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

from open_exposure_gateway.domain.errors import CallbackSecretError

_NONCE_BYTES = 12

def encrypt_secret(plaintext: str, key: bytes) -> bytes:

    """Returns nonce + ciphertext. A fresh nonce each call, so one key can encrypt

    every secret safely."""

    nonce = os.urandom(_NONCE_BYTES)

    return nonce + AESGCM(key).encrypt(nonce, plaintext.encode(), None)

def decrypt_secret(blob: bytes, key: bytes) -> str:

    nonce, ciphertext = blob[:_NONCE_BYTES], blob[_NONCE_BYTES:]

    try:

        plaintext = AESGCM(key).decrypt(nonce, ciphertext, None)

    except (InvalidTag, ValueError) as exc:

        raise CallbackSecretError("callback secret could not be decrypted") from exc

    return plaintext.decode()

class CallbackSecretError(Exception):

    """A stored callback secret could not be decrypted. Carries no secret material."""

import secrets

import pytest

from open_exposure_gateway.application.callback_secrets import decrypt_secret, encrypt_secret

from open_exposure_gateway.domain.errors import CallbackSecretError

TOKEN = "client-issued-bearer-token-0123456789"

@pytest.fixture

def key() -> bytes:

    return secrets.token_bytes(32)

def test_round_trip(key: bytes) -> None:

    assert decrypt_secret(encrypt_secret(TOKEN, key), key) == TOKEN

def test_ciphertext_does_not_contain_the_plaintext(key: bytes) -> None:

    assert TOKEN.encode() not in encrypt_secret(TOKEN, key)

def test_same_input_gives_different_ciphertexts(key: bytes) -> None:

    assert encrypt_secret(TOKEN, key) != encrypt_secret(TOKEN, key)

def test_wrong_key_raises_without_secret_material(key: bytes) -> None:

    blob = encrypt_secret(TOKEN, key)

    with pytest.raises(CallbackSecretError) as exc_info:

        decrypt_secret(blob, secrets.token_bytes(32))

    assert TOKEN not in str(exc_info.value)

def test_tampered_ciphertext_raises(key: bytes) -> None:

    blob = bytearray(encrypt_secret(TOKEN, key))

    blob[-1] ^= 0x01

    with pytest.raises(CallbackSecretError):

        decrypt_secret(bytes(blob), key)

def test_truncated_blob_raises(key: bytes) -> None:

    with pytest.raises(CallbackSecretError):

        decrypt_secret(b"short", key)