Commit b8871ba4 authored by George Papathanail's avatar George Papathanail
Browse files

feat(config): add optional callback secret key

parent 19a20fca

src/open_exposure_gateway/core/config.py

+30 −1
Changes for src/open_exposure_gateway/core/config.py: 30 added lines, 1 removed line.
Original line number Diff line number Diff line 
import base64

import binascii

from functools import lru_cache

from uuid import UUID, uuid5



from pydantic import BaseModel, HttpUrl

from pydantic import BaseModel, HttpUrl, SecretStr, field_validator

from pydantic_settings import BaseSettings, SettingsConfigDict



# QoD and Location Retrieval have no CAMARA registration step to resolve a
@@ -60,6 +62,31 @@ class KeycloakSettings(BaseModel): 


class CallbackSettings(BaseModel):

    timeout: float = 10.0

    # base64url of 32 bytes; encrypts client-supplied sink tokens at rest. Unset means

    # OEG rejects those tokens with 501.

    secret_key: SecretStr | None = None



    @field_validator("secret_key")

    @classmethod

    def _secret_key_is_32_bytes(cls, value: SecretStr | None) -> SecretStr | None:

        if value is not None:

            _decode_secret_key(value)

        return value



    @property

    def secret_key_bytes(self) -> bytes | None:

        return None if self.secret_key is None else _decode_secret_key(self.secret_key)





def _decode_secret_key(value: SecretStr) -> bytes:

    raw = value.get_secret_value().strip()

    try:

        key = base64.b64decode(raw + "=" * (-len(raw) % 4), altchars=b"-_", validate=True)

    except (binascii.Error, ValueError):

        raise ValueError("CALLBACK_SETTINGS__SECRET_KEY must be base64url-encoded") from None

    if len(key) != 32:

        raise ValueError("CALLBACK_SETTINGS__SECRET_KEY must decode to exactly 32 bytes")

    return key





class QodSettings(BaseModel):
@@ -99,6 +126,8 @@ class Settings(BaseSettings): 
        case_sensitive=False,

        extra="ignore",

        env_nested_delimiter="__",

        # Keeps rejected secrets (e.g. callback secret_key) out of the startup error log.

        hide_input_in_errors=True,

    )



    app_name: str = "Open Exposure Gateway"

src/open_exposure_gateway/main.py

+9 −0
Changes for src/open_exposure_gateway/main.py: 9 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -305,6 +305,15 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 
            service_specification_id=str(location_retrieval_settings.service_specification_id),

        )



    if settings.callback_settings.secret_key is None:

        logger.warning(

            "callback_secret_key_unset",

            detail=(

                "CALLBACK_SETTINGS__SECRET_KEY is unset; requests carrying a sink "

                "credential will be rejected with 501."

            ),

        )



    try:

        db_engine, session_maker = await build_engine_and_session_maker(

            url=settings.postgresql_settings.url,

tests/unit/test_config.py

+49 −0
Changes for tests/unit/test_config.py: 49 added lines, 0 removed lines.
Original line number Diff line number Diff line 
import base64

import os

from uuid import NAMESPACE_URL, UUID, uuid5



import pytest

from pydantic import ValidationError



from open_exposure_gateway.core.config import (

    DEFAULT_LOCATION_RETRIEVAL_SERVICE_SPECIFICATION_ID,
@@ -126,6 +128,53 @@ class TestLocationRetrievalServiceSpecificationIdIsConfigurable: 
        assert settings.location_retrieval_settings.uses_default_service_specification_id is False





class TestCallbackSecretKey:

    def test_unset_is_none(self, monkeypatch: pytest.MonkeyPatch) -> None:

        monkeypatch.delenv("CALLBACK_SETTINGS__SECRET_KEY", raising=False)

        settings = Settings(_env_file=None)  # type: ignore[call-arg]

        assert settings.callback_settings.secret_key is None

        assert settings.callback_settings.secret_key_bytes is None



    def test_valid_key_decodes_to_32_bytes(self, monkeypatch: pytest.MonkeyPatch) -> None:

        key = bytes(range(32))

        monkeypatch.setenv("CALLBACK_SETTINGS__SECRET_KEY", base64.urlsafe_b64encode(key).decode())

        settings = Settings(_env_file=None)  # type: ignore[call-arg]

        assert settings.callback_settings.secret_key_bytes == key



    def test_valid_key_without_padding_is_accepted(self, monkeypatch: pytest.MonkeyPatch) -> None:

        key = bytes(range(32))

        monkeypatch.setenv(

            "CALLBACK_SETTINGS__SECRET_KEY", base64.urlsafe_b64encode(key).decode().rstrip("=")

        )

        settings = Settings(_env_file=None)  # type: ignore[call-arg]

        assert settings.callback_settings.secret_key_bytes == key



    def test_key_is_masked_in_repr(self, monkeypatch: pytest.MonkeyPatch) -> None:

        encoded = base64.urlsafe_b64encode(bytes(range(32))).decode()

        monkeypatch.setenv("CALLBACK_SETTINGS__SECRET_KEY", encoded)

        settings = Settings(_env_file=None)  # type: ignore[call-arg]

        assert encoded not in repr(settings)



    @pytest.mark.parametrize(

        "bad_key",

        [

            base64.urlsafe_b64encode(bytes(16)).decode(),

            base64.urlsafe_b64encode(bytes(33)).decode(),

            "not!base64url!at!all!not!base64url!at!all!xx",

        ],

        ids=["too-short", "too-long", "not-base64url"],

    )

    def test_bad_key_fails_without_echoing_it(

        self, monkeypatch: pytest.MonkeyPatch, bad_key: str

    ) -> None:

        monkeypatch.setenv("CALLBACK_SETTINGS__SECRET_KEY", bad_key)

        with pytest.raises(ValidationError) as exc_info:

            Settings(_env_file=None)  # type: ignore[call-arg]

        message = str(exc_info.value)

        assert "CALLBACK_SETTINGS__SECRET_KEY" in message

        assert bad_key not in message





class TestGetSettingsCaching:

    def test_get_settings_returns_same_instance(self) -> None:

        assert get_settings() is get_settings()