Loading src/portal_service/application/pkce.py 0 → 100644 +43 −0 Changes for src/portal_service/application/pkce.py: 43 added lines, 0 removed lines. Original line number Diff line number Diff line """ One-time values for the OpenID Connect login redirect. PKCE (RFC 7636) stops a stolen authorization code from being exchanged by anyone else: the login redirect carries the SHA-256 of a secret, and the code exchange must present the secret itself. state — tied to the browser that started the login, to reject a callback it did not ask for (CSRF). nonce — carried into the ID token, to reject one replayed from another login. Pure functions, no I/O. """ import base64 import hashlib import secrets # RFC 7636 §4.1 allows 43-128 characters; 32 random bytes encode to 43. _VERIFIER_BYTES = 32 def new_code_verifier() -> str: return _b64encode(secrets.token_bytes(_VERIFIER_BYTES)) def code_challenge(code_verifier: str) -> str: """The S256 transformation of RFC 7636 §4.2.""" digest = hashlib.sha256(code_verifier.encode("ascii")).digest() return _b64encode(digest) def new_state() -> str: return secrets.token_urlsafe(_VERIFIER_BYTES) def new_nonce() -> str: return secrets.token_urlsafe(_VERIFIER_BYTES) def _b64encode(data: bytes) -> str: # RFC 7636 uses base64url without padding. return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") src/portal_service/config.py +14 −8 Changes for src/portal_service/config.py: 14 added lines, 8 removed lines. Original line number Diff line number Diff line """ Application configuration. All settings are loaded from environment variables (or a .env file). Nested models use double-underscore as delimiter, e.g. OIDC__ISSUER. """ from enum import StrEnum from functools import lru_cache Loading @@ -26,7 +19,14 @@ class OIDCSettings(BaseModel): client_id: str client_secret: str redirect_uri: str # Where the callback sends the browser after a successful login (the frontend # origin). Empty means the portal's own root. post_login_url: str = "" timeout: float = 10.0 jwks_uri: str = "" authorization_endpoint: str = "" token_endpoint: str = "" end_session_endpoint: str = "" admin_base_url: str = "" realm: str = "" Loading @@ -34,6 +34,12 @@ class OIDCSettings(BaseModel): def derive_keycloak_endpoints(self) -> "OIDCSettings": if not self.jwks_uri: self.jwks_uri = f"{self.issuer}/protocol/openid-connect/certs" if not self.authorization_endpoint: self.authorization_endpoint = f"{self.issuer}/protocol/openid-connect/auth" if not self.token_endpoint: self.token_endpoint = f"{self.issuer}/protocol/openid-connect/token" if not self.end_session_endpoint: self.end_session_endpoint = f"{self.issuer}/protocol/openid-connect/logout" if "/realms/" in self.issuer: base, _, realm = self.issuer.partition("/realms/") if not self.admin_base_url: Loading Loading @@ -64,7 +70,7 @@ class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", env_nested_delimiter="__") app_name: str = "OOP Portal Backend" app_version: str = "2.0.0" app_version: str app_description: str = "Portal backend API for the Open Operator Platform" auth: AuthSettings = AuthSettings() Loading tests/conftest.py +6 −3 Changes for tests/conftest.py: 6 added lines, 3 removed lines. Original line number Diff line number Diff line Loading @@ -33,13 +33,16 @@ _BASE_ENV = { "OIDC__CLIENT_ID": CLIENT_ID, "OIDC__CLIENT_SECRET": "test-client-secret", "OIDC__REDIRECT_URI": "http://localhost:8000/v1/auth/callback", # TODO: no matching field on OIDCSettings yet, so pydantic silently drops this. # Add `post_login_url` when the /v1/auth/callback route lands — it is where the # callback redirects the browser after a successful login (the frontend origin). "OIDC__POST_LOGIN_URL": "http://localhost:3000", } @pytest.fixture(autouse=True) def app_version_env(monkeypatch: pytest.MonkeyPatch) -> None: """APP_VERSION has no default: helm supplies it in the cluster, tests here.""" monkeypatch.setenv("APP_VERSION", "test") def _apply_env(monkeypatch: pytest.MonkeyPatch, **overrides: str) -> None: for key, value in {**_BASE_ENV, **overrides}.items(): monkeypatch.setenv(key, value) Loading tests/test_config.py +12 −0 Changes for tests/test_config.py: 12 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -40,10 +40,20 @@ def test_explicit_overrides_are_respected() -> None: jwks_uri="https://idp.example.com/keys", admin_base_url="http://keycloak:8080", realm="custom", token_endpoint="https://idp.example.com/token", ) assert oidc.jwks_uri == "https://idp.example.com/keys" assert oidc.admin_base_url == "http://keycloak:8080" assert oidc.realm == "custom" assert oidc.token_endpoint == "https://idp.example.com/token" def test_login_endpoints_derived_from_issuer() -> None: oidc = _oidc() base = "http://localhost:30083/realms/oop/protocol/openid-connect" assert oidc.authorization_endpoint == f"{base}/auth" assert oidc.token_endpoint == f"{base}/token" assert oidc.end_session_endpoint == f"{base}/logout" def test_non_keycloak_issuer_leaves_admin_fields_empty() -> None: Loading @@ -55,6 +65,7 @@ def test_non_keycloak_issuer_leaves_admin_fields_empty() -> None: def test_jwt_mode_requires_oidc(monkeypatch: pytest.MonkeyPatch) -> None: with pytest.raises(ValueError, match="OIDC"): Settings( app_version="test", auth=AuthSettings(security_mode=SecurityMode.JWT), oidc=None, session=SessionSettings(secret_key="s"), Loading @@ -63,6 +74,7 @@ def test_jwt_mode_requires_oidc(monkeypatch: pytest.MonkeyPatch) -> None: def test_none_mode_allows_missing_oidc() -> None: s = Settings( app_version="test", auth=AuthSettings(security_mode=SecurityMode.NONE), oidc=None, session=SessionSettings(secret_key="s"), Loading tests/test_pkce.py 0 → 100644 +51 −0 Changes for tests/test_pkce.py: 51 added lines, 0 removed lines. Original line number Diff line number Diff line """PKCE, state and nonce: checked against RFC 7636's own example and its rules.""" import base64 import hashlib import string from portal_service.application.pkce import ( code_challenge, new_code_verifier, new_nonce, new_state, ) # RFC 7636 Appendix B: this verifier must produce this challenge. RFC_VERIFIER = "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk" RFC_CHALLENGE = "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM" # RFC 7636 §4.1: verifiers use the unreserved characters only. UNRESERVED = set(string.ascii_letters + string.digits + "-._~") def test_matches_the_rfc_7636_example() -> None: assert code_challenge(RFC_VERIFIER) == RFC_CHALLENGE def test_challenge_is_sha256_in_unpadded_base64url() -> None: verifier = new_code_verifier() expected = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=") challenge = code_challenge(verifier) assert challenge == expected.decode() assert "=" not in challenge assert len(challenge) == 43 def test_verifier_length_is_within_the_rfc_range() -> None: assert 43 <= len(new_code_verifier()) <= 128 def test_verifier_uses_only_unreserved_characters() -> None: assert set(new_code_verifier()) <= UNRESERVED def test_each_login_gets_fresh_values() -> None: assert new_code_verifier() != new_code_verifier() assert new_state() != new_state() assert new_nonce() != new_nonce() def test_state_and_nonce_are_long_enough_to_be_unguessable() -> None: assert len(new_state()) >= 43 assert len(new_nonce()) >= 43 Loading
src/portal_service/application/pkce.py 0 → 100644 +43 −0 Changes for src/portal_service/application/pkce.py: 43 added lines, 0 removed lines. Original line number Diff line number Diff line """ One-time values for the OpenID Connect login redirect. PKCE (RFC 7636) stops a stolen authorization code from being exchanged by anyone else: the login redirect carries the SHA-256 of a secret, and the code exchange must present the secret itself. state — tied to the browser that started the login, to reject a callback it did not ask for (CSRF). nonce — carried into the ID token, to reject one replayed from another login. Pure functions, no I/O. """ import base64 import hashlib import secrets # RFC 7636 §4.1 allows 43-128 characters; 32 random bytes encode to 43. _VERIFIER_BYTES = 32 def new_code_verifier() -> str: return _b64encode(secrets.token_bytes(_VERIFIER_BYTES)) def code_challenge(code_verifier: str) -> str: """The S256 transformation of RFC 7636 §4.2.""" digest = hashlib.sha256(code_verifier.encode("ascii")).digest() return _b64encode(digest) def new_state() -> str: return secrets.token_urlsafe(_VERIFIER_BYTES) def new_nonce() -> str: return secrets.token_urlsafe(_VERIFIER_BYTES) def _b64encode(data: bytes) -> str: # RFC 7636 uses base64url without padding. return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
src/portal_service/config.py +14 −8 Changes for src/portal_service/config.py: 14 added lines, 8 removed lines. Original line number Diff line number Diff line """ Application configuration. All settings are loaded from environment variables (or a .env file). Nested models use double-underscore as delimiter, e.g. OIDC__ISSUER. """ from enum import StrEnum from functools import lru_cache Loading @@ -26,7 +19,14 @@ class OIDCSettings(BaseModel): client_id: str client_secret: str redirect_uri: str # Where the callback sends the browser after a successful login (the frontend # origin). Empty means the portal's own root. post_login_url: str = "" timeout: float = 10.0 jwks_uri: str = "" authorization_endpoint: str = "" token_endpoint: str = "" end_session_endpoint: str = "" admin_base_url: str = "" realm: str = "" Loading @@ -34,6 +34,12 @@ class OIDCSettings(BaseModel): def derive_keycloak_endpoints(self) -> "OIDCSettings": if not self.jwks_uri: self.jwks_uri = f"{self.issuer}/protocol/openid-connect/certs" if not self.authorization_endpoint: self.authorization_endpoint = f"{self.issuer}/protocol/openid-connect/auth" if not self.token_endpoint: self.token_endpoint = f"{self.issuer}/protocol/openid-connect/token" if not self.end_session_endpoint: self.end_session_endpoint = f"{self.issuer}/protocol/openid-connect/logout" if "/realms/" in self.issuer: base, _, realm = self.issuer.partition("/realms/") if not self.admin_base_url: Loading Loading @@ -64,7 +70,7 @@ class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", env_nested_delimiter="__") app_name: str = "OOP Portal Backend" app_version: str = "2.0.0" app_version: str app_description: str = "Portal backend API for the Open Operator Platform" auth: AuthSettings = AuthSettings() Loading
tests/conftest.py +6 −3 Changes for tests/conftest.py: 6 added lines, 3 removed lines. Original line number Diff line number Diff line Loading @@ -33,13 +33,16 @@ _BASE_ENV = { "OIDC__CLIENT_ID": CLIENT_ID, "OIDC__CLIENT_SECRET": "test-client-secret", "OIDC__REDIRECT_URI": "http://localhost:8000/v1/auth/callback", # TODO: no matching field on OIDCSettings yet, so pydantic silently drops this. # Add `post_login_url` when the /v1/auth/callback route lands — it is where the # callback redirects the browser after a successful login (the frontend origin). "OIDC__POST_LOGIN_URL": "http://localhost:3000", } @pytest.fixture(autouse=True) def app_version_env(monkeypatch: pytest.MonkeyPatch) -> None: """APP_VERSION has no default: helm supplies it in the cluster, tests here.""" monkeypatch.setenv("APP_VERSION", "test") def _apply_env(monkeypatch: pytest.MonkeyPatch, **overrides: str) -> None: for key, value in {**_BASE_ENV, **overrides}.items(): monkeypatch.setenv(key, value) Loading
tests/test_config.py +12 −0 Changes for tests/test_config.py: 12 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -40,10 +40,20 @@ def test_explicit_overrides_are_respected() -> None: jwks_uri="https://idp.example.com/keys", admin_base_url="http://keycloak:8080", realm="custom", token_endpoint="https://idp.example.com/token", ) assert oidc.jwks_uri == "https://idp.example.com/keys" assert oidc.admin_base_url == "http://keycloak:8080" assert oidc.realm == "custom" assert oidc.token_endpoint == "https://idp.example.com/token" def test_login_endpoints_derived_from_issuer() -> None: oidc = _oidc() base = "http://localhost:30083/realms/oop/protocol/openid-connect" assert oidc.authorization_endpoint == f"{base}/auth" assert oidc.token_endpoint == f"{base}/token" assert oidc.end_session_endpoint == f"{base}/logout" def test_non_keycloak_issuer_leaves_admin_fields_empty() -> None: Loading @@ -55,6 +65,7 @@ def test_non_keycloak_issuer_leaves_admin_fields_empty() -> None: def test_jwt_mode_requires_oidc(monkeypatch: pytest.MonkeyPatch) -> None: with pytest.raises(ValueError, match="OIDC"): Settings( app_version="test", auth=AuthSettings(security_mode=SecurityMode.JWT), oidc=None, session=SessionSettings(secret_key="s"), Loading @@ -63,6 +74,7 @@ def test_jwt_mode_requires_oidc(monkeypatch: pytest.MonkeyPatch) -> None: def test_none_mode_allows_missing_oidc() -> None: s = Settings( app_version="test", auth=AuthSettings(security_mode=SecurityMode.NONE), oidc=None, session=SessionSettings(secret_key="s"), Loading
tests/test_pkce.py 0 → 100644 +51 −0 Changes for tests/test_pkce.py: 51 added lines, 0 removed lines. Original line number Diff line number Diff line """PKCE, state and nonce: checked against RFC 7636's own example and its rules.""" import base64 import hashlib import string from portal_service.application.pkce import ( code_challenge, new_code_verifier, new_nonce, new_state, ) # RFC 7636 Appendix B: this verifier must produce this challenge. RFC_VERIFIER = "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk" RFC_CHALLENGE = "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM" # RFC 7636 §4.1: verifiers use the unreserved characters only. UNRESERVED = set(string.ascii_letters + string.digits + "-._~") def test_matches_the_rfc_7636_example() -> None: assert code_challenge(RFC_VERIFIER) == RFC_CHALLENGE def test_challenge_is_sha256_in_unpadded_base64url() -> None: verifier = new_code_verifier() expected = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=") challenge = code_challenge(verifier) assert challenge == expected.decode() assert "=" not in challenge assert len(challenge) == 43 def test_verifier_length_is_within_the_rfc_range() -> None: assert 43 <= len(new_code_verifier()) <= 128 def test_verifier_uses_only_unreserved_characters() -> None: assert set(new_code_verifier()) <= UNRESERVED def test_each_login_gets_fresh_values() -> None: assert new_code_verifier() != new_code_verifier() assert new_state() != new_state() assert new_nonce() != new_nonce() def test_state_and_nonce_are_long_enough_to_be_unguessable() -> None: assert len(new_state()) >= 43 assert len(new_nonce()) >= 43