Commit 6d2a9489 authored by Dimitrios Gogos's avatar Dimitrios Gogos
Browse files

feat: add per-session encryption for login session tokens

parent a27aa1c4

pyproject.toml

+1 −0
Changes for pyproject.toml: 1 added line, 0 removed lines.
Original line number Diff line number Diff line
@@ -15,6 +15,7 @@ dependencies = [ 
    "httpx>=0.28.0",

    "sqlalchemy>=2.0.48",

    "asyncpg>=0.31.0",

    "cryptography>=45.0.0",

]



[project.optional-dependencies]

src/portal_service/application/session_crypto.py

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

Encryption for login sessions



The browser's cookie holds a random session id and a random 256-bit key. The

login_sessions row is keyed by SHA-256 of the session id and holds the tokens

AES-GCM-encrypted with that key. The table alone opens nothing, and one cookie

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.domain.errors import InvalidSessionError

from portal_service.domain.models import SessionTokens



_KEY_BYTES = 32

_NONCE_BYTES = 12





@dataclass(frozen=True)

class SessionCredentials:

    """What the session cookie carries. Never stored server-side."""



    session_id: str

    key: bytes = field(repr=False)



    @classmethod

    def generate(cls) -> "SessionCredentials":

        return cls(session_id=secrets.token_urlsafe(32), key=secrets.token_bytes(_KEY_BYTES))



    @property

    def session_id_hash(self) -> str:

        """Primary key of the login_sessions row."""

        return hashlib.sha256(self.session_id.encode()).hexdigest()



    def to_cookie(self) -> str:

        return f"{self.session_id}.{_b64encode(self.key)}"



    @classmethod

    def from_cookie(cls, value: str) -> "SessionCredentials":

        session_id, _, encoded_key = value.partition(".")

        try:

            key = _b64decode(encoded_key)

        except (binascii.Error, ValueError) as exc:

            raise InvalidSessionError("malformed session cookie") from exc

        if not session_id or len(key) != _KEY_BYTES:

            raise InvalidSessionError("malformed session cookie")

        return cls(session_id=session_id, key=key)





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)





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:

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

    return SessionTokens.model_validate_json(plaintext)





def _b64encode(data: bytes) -> str:

    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()





def _b64decode(data: str) -> bytes:

    padded = data + "=" * (-len(data) % 4)

    return base64.b64decode(padded, altchars=b"-_", validate=True)

src/portal_service/domain/errors.py

+5 −0
Changes for src/portal_service/domain/errors.py: 5 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -15,6 +15,11 @@ class InvalidTokenError(AuthenticationError): 
    pass





# The session cookie is malformed or does not open its stored tokens: treat as logged out.

class InvalidSessionError(AuthenticationError):

    pass





# Not an AuthenticationError: the token was never judged, so it is not the caller's fault.

class IdentityProviderUnavailableError(Exception):

    pass

src/portal_service/domain/models.py

+13 −1
Changes for src/portal_service/domain/models.py: 13 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -5,7 +5,7 @@ Core domain models — pure Python, no framework dependencies. 
from enum import StrEnum

from typing import Any



from pydantic import BaseModel

from pydantic import BaseModel, Field





class Role(StrEnum):
@@ -48,3 +48,15 @@ class TokenClaims(BaseModel): 
    def is_human(self) -> bool:

        """True when the token belongs to a human user (not a service account)."""

        return any(r in self.roles for r in (Role.APP_PROVIDER, Role.OPERATOR, Role.PLATFORM_ADMIN))





class SessionTokens(BaseModel):

    """

    Tokens Keycloak issued for one login session. Stored only encrypted, and kept

    out of repr() so they never end up in logs.

    """



    access_token: str = Field(repr=False)

    refresh_token: str | None = Field(default=None, repr=False)

    # Sent back to Keycloak as id_token_hint when logging out.

    id_token: str | None = Field(default=None, repr=False)

tests/test_session_crypto.py

0 → 100644
+77 −0
Changes for tests/test_session_crypto.py: 77 added lines, 0 removed lines.
Original line number Diff line number Diff line 
"""Session encryption: cookie format, token encryption, and failure modes."""



import hashlib



import pytest



from portal_service.application.session_crypto import (

    SessionCredentials,

    decrypt_tokens,

    encrypt_tokens,

)

from portal_service.domain.errors import InvalidSessionError

from portal_service.domain.models import SessionTokens



TOKENS = SessionTokens(access_token="at", refresh_token="rt", id_token="it")





def test_tokens_round_trip() -> None:

    creds = SessionCredentials.generate()

    assert decrypt_tokens(encrypt_tokens(TOKENS, creds), creds) == TOKENS





def test_each_encryption_uses_a_fresh_nonce() -> None:

    creds = SessionCredentials.generate()

    assert encrypt_tokens(TOKENS, creds) != encrypt_tokens(TOKENS, creds)





def test_other_sessions_key_cannot_decrypt() -> None:

    blob = encrypt_tokens(TOKENS, SessionCredentials.generate())

    with pytest.raises(InvalidSessionError):

        decrypt_tokens(blob, SessionCredentials.generate())





def test_tampered_blob_is_rejected() -> None:

    creds = SessionCredentials.generate()

    blob = bytearray(encrypt_tokens(TOKENS, creds))

    blob[-1] ^= 1

    with pytest.raises(InvalidSessionError):

        decrypt_tokens(bytes(blob), creds)





def test_truncated_blob_is_rejected() -> None:

    with pytest.raises(InvalidSessionError):

        decrypt_tokens(b"short", SessionCredentials.generate())





def test_cookie_round_trip() -> None:

    creds = SessionCredentials.generate()

    assert SessionCredentials.from_cookie(creds.to_cookie()) == creds





def test_session_id_hash_is_sha256_of_the_id() -> None:

    creds = SessionCredentials.generate()

    assert creds.session_id_hash == hashlib.sha256(creds.session_id.encode()).hexdigest()





@pytest.mark.parametrize(

    "cookie",

    [

        "",

        "no-dot-at-all",

        ".key-without-id",

        "id.",

        "id.not*base64",

        "id.c2hvcnQ",  # valid base64, but 5 bytes instead of 32

        f"id.{'A' * 43}.extra",

    ],

)

def test_malformed_cookie_is_rejected(cookie: str) -> None:

    with pytest.raises(InvalidSessionError):

        SessionCredentials.from_cookie(cookie)





def test_secrets_stay_out_of_repr() -> None:

    creds = SessionCredentials.generate()

    assert repr(creds.key) not in repr(creds)

    assert repr(TOKENS) == "SessionTokens()"