Commit f4e6453d authored by Dimitrios Gogos's avatar Dimitrios Gogos
Browse files

feat: implement AES-GCM encryption and decryption functions with unit tests

parent 0ba11674

src/portal_service/application/crypto.py

0 → 100644
+25 −0
Changes for src/portal_service/application/crypto.py: 25 added lines, 0 removed lines.
Original line number Diff line number Diff line 
"""

AES-GCM encryption, shared by login sessions (a per-cookie key) and API client

registration tokens (one app-wide key, since background work must open them too).

"""



import os



from cryptography.exceptions import InvalidTag

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



_NONCE_BYTES = 12





def encrypt(data: bytes, key: bytes) -> bytes:

    """Returns nonce + ciphertext, with a fresh nonce each call."""

    nonce = os.urandom(_NONCE_BYTES)

    return nonce + AESGCM(key).encrypt(nonce, data, None)





def decrypt(blob: bytes, key: bytes) -> bytes:

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

    try:

        return AESGCM(key).decrypt(nonce, ciphertext, None)

    except InvalidTag as exc:

        raise ValueError("could not decrypt: wrong key or corrupted data") from exc

src/portal_service/application/session_crypto.py

+4 −11
Changes for src/portal_service/application/session_crypto.py: 4 added lines, 11 removed lines.
Original line number Diff line number Diff line
@@ -11,18 +11,14 @@ opens only its own session. There is no app-wide key to manage. 
import base64

import binascii

import hashlib

import os

import secrets

from dataclasses import dataclass, field



from cryptography.exceptions import InvalidTag

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



from portal_service.application.crypto import decrypt, encrypt

from portal_service.domain.errors import InvalidSessionError

from portal_service.domain.models import SessionTokens



_KEY_BYTES = 32

_NONCE_BYTES = 12





@dataclass(frozen=True)
@@ -72,16 +68,13 @@ def hash_id(value: str) -> str: 
def encrypt_tokens(tokens: SessionTokens, credentials: SessionCredentials) -> bytes:

    """Returns nonce + ciphertext. A fresh nonce each call, so re-encrypting after a

    token refresh with the same key is safe."""

    nonce = os.urandom(_NONCE_BYTES)

    plaintext = tokens.model_dump_json().encode()

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

    return encrypt(tokens.model_dump_json().encode(), credentials.key)





def decrypt_tokens(blob: bytes, credentials: SessionCredentials) -> SessionTokens:

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

    try:

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

    except (InvalidTag, ValueError) as exc:

        plaintext = decrypt(blob, credentials.key)

    except ValueError as exc:

        raise InvalidSessionError("session tokens could not be decrypted") from exc

    return SessionTokens.model_validate_json(plaintext)

tests/unit/test_crypto.py

0 → 100644
+35 −0
Changes for tests/unit/test_crypto.py: 35 added lines, 0 removed lines.
Original line number Diff line number Diff line 
"""AES-GCM encryption shared by sessions and registration tokens."""



import secrets



import pytest



from portal_service.application.crypto import decrypt, encrypt



KEY = secrets.token_bytes(32)





def test_round_trip() -> None:

    assert decrypt(encrypt(b"registration-token", KEY), KEY) == b"registration-token"





def test_each_encryption_uses_a_fresh_nonce() -> None:

    assert encrypt(b"same", KEY) != encrypt(b"same", KEY)





def test_wrong_key_is_rejected() -> None:

    with pytest.raises(ValueError, match="could not decrypt"):

        decrypt(encrypt(b"registration-token", KEY), secrets.token_bytes(32))





def test_tampered_data_is_rejected() -> None:

    blob = bytearray(encrypt(b"registration-token", KEY))

    blob[-1] ^= 1



    with pytest.raises(ValueError, match="could not decrypt"):

        decrypt(bytes(blob), KEY)





def test_key_must_be_256_bits() -> None:

    with pytest.raises(ValueError):

        encrypt(b"registration-token", b"too short")