diff --git a/core/oidc.py b/core/oidc.py index c776d0945..d106b3156 100644 --- a/core/oidc.py +++ b/core/oidc.py @@ -164,11 +164,15 @@ class OidcManager: f"OIDC discovery document missing required key: {key}" ) - # The issuer in the discovery doc SHOULD match the configured issuer - doc_issuer = self._config.get("issuer", "") - if doc_issuer and doc_issuer.rstrip("/") != self.issuer: - logger.warning( - "OIDC issuer mismatch: configured=%r doc=%r", self.issuer, doc_issuer, + # The issuer in the discovery doc MUST match the configured issuer + # (OIDC Discovery §1.1). Failing closed prevents trust-path confusion + # where a misconfigured or malicious discovery document could cause + # id_token validation to accept a different issuer. + doc_issuer = (self._config.get("issuer") or "").rstrip("/") + if doc_issuer and doc_issuer != self.issuer: + raise OidcError( + f"OIDC issuer mismatch: configured {self.issuer!r}, " + f"discovery doc returned {doc_issuer!r}" ) # Pin signing algorithms to those the provider supports. @@ -416,17 +420,22 @@ class OidcManager: ) # Validate audience: aud may be a string or a JSON array. - # When multiple audiences are present, azp MUST be present and match - # the client_id (per OIDC Core 1.0 § 2). + # OIDC Core 1.0 § 2: azp is REQUIRED when aud contains multiple + # values, and MUST equal client_id. We reject multi-audience tokens + # without azp — there is no trusted-additional-audience model. aud = claims.get("aud") if isinstance(aud, list): if self.client_id not in aud: raise OidcError( f"id_token aud mismatch: client_id {self.client_id!r} not in aud {aud!r}" ) - # Multiple audiences — azp MUST identify the authorized party azp = claims.get("azp") - if azp and azp != self.client_id: + if not azp: + raise OidcError( + "id_token has multiple audiences but no azp claim " + "(required by OIDC Core 1.0 § 2)" + ) + if azp != self.client_id: raise OidcError( f"id_token azp mismatch: expected {self.client_id!r}, got {azp!r}" ) diff --git a/routes/auth_routes.py b/routes/auth_routes.py index b9aba33c7..97521f2e2 100644 --- a/routes/auth_routes.py +++ b/routes/auth_routes.py @@ -201,6 +201,8 @@ def setup_auth_routes(auth_manager: AuthManager) -> APIRouter: user = _get_current_user(request) if not user: raise HTTPException(401, "Not authenticated") + if auth_manager.is_oidc_user(user): + raise HTTPException(400, "OIDC users don't have a password — manage credentials through your identity provider") if len(body.new_password) < PASSWORD_MIN_LENGTH: raise HTTPException(400, f"Password must be at least {PASSWORD_MIN_LENGTH} characters") current_token = request.cookies.get(SESSION_COOKIE) @@ -220,6 +222,8 @@ def setup_auth_routes(auth_manager: AuthManager) -> APIRouter: user = _get_current_user(request) if not user: raise HTTPException(401, "Not authenticated") + if auth_manager.is_oidc_user(user): + raise HTTPException(400, "Two-factor authentication is managed by your identity provider for OIDC users") if auth_manager.totp_enabled(user): raise HTTPException(400, "2FA is already enabled") secret = auth_manager.totp_generate_secret(user) @@ -243,6 +247,8 @@ def setup_auth_routes(auth_manager: AuthManager) -> APIRouter: user = _get_current_user(request) if not user: raise HTTPException(401, "Not authenticated") + if auth_manager.is_oidc_user(user): + raise HTTPException(400, "Two-factor authentication is managed by your identity provider for OIDC users") if not auth_manager.totp_confirm_enable(user, body.code): raise HTTPException(400, "Invalid code — try again") backup = auth_manager.users.get(user, {}).get("totp_backup_codes", []) @@ -257,6 +263,8 @@ def setup_auth_routes(auth_manager: AuthManager) -> APIRouter: user = _get_current_user(request) if not user: raise HTTPException(401, "Not authenticated") + if auth_manager.is_oidc_user(user): + raise HTTPException(400, "Two-factor authentication is managed by your identity provider for OIDC users") if not auth_manager.totp_disable(user, body.password): raise HTTPException(400, "Invalid password") return {"ok": True} diff --git a/tests/test_auth_session_revocation.py b/tests/test_auth_session_revocation.py index d6930e5ab..17c0c4bee 100644 --- a/tests/test_auth_session_revocation.py +++ b/tests/test_auth_session_revocation.py @@ -138,6 +138,7 @@ def test_login_route_does_not_set_cookie_when_trusted_session_rejects_stale_user def test_change_password_route_revokes_other_sessions_after_success(monkeypatch): auth = MagicMock() auth.get_username_for_token.return_value = "alice" + auth.is_oidc_user.return_value = False auth.change_password.return_value = True endpoint, ChangePasswordRequest = _change_password_endpoint(auth) monkeypatch.setattr( @@ -157,6 +158,7 @@ def test_change_password_route_revokes_other_sessions_after_success(monkeypatch) def test_change_password_route_wrong_password_does_not_revoke(monkeypatch): auth = MagicMock() auth.get_username_for_token.return_value = "alice" + auth.is_oidc_user.return_value = False auth.change_password.return_value = False endpoint, ChangePasswordRequest = _change_password_endpoint(auth) monkeypatch.setattr( diff --git a/tests/test_oidc_auth.py b/tests/test_oidc_auth.py index eda7ba1f8..3dce9fac5 100644 --- a/tests/test_oidc_auth.py +++ b/tests/test_oidc_auth.py @@ -310,3 +310,105 @@ def test_first_user_bootstrap_suppressed_when_admin_groups_configured( "First OIDC user should NOT be admin when OIDC_ADMIN_GROUPS is set " "and they are not in a group" ) + + +# --------------------------------------------------------------------------- +# Route-level OIDC guards — 2FA and change-password +# --------------------------------------------------------------------------- + +class TestOidcRouteGuards: + """The auth routes reject local 2FA / password mutations for OIDC users. + + OIDC users authenticate through their identity provider; local password + and TOTP controls are not applicable. The frontend already hides these + cards, but the backend must also enforce the policy so a direct API call + cannot create a misleading or stuck 2FA state.""" + + @pytest.fixture + def setup_router(self, tmp_path): + """Create an auth router backed by a temp AuthManager with one OIDC user.""" + from routes.auth_routes import setup_auth_routes + mgr = _make_manager(tmp_path) + mgr.create_user_oidc("alice", sub="abc", issuer="https://idp.example.com") + # Issue a session so the user is "logged in" + token = mgr.create_session_trusted("alice") + router = setup_auth_routes(mgr) + return router, mgr, token + + def _get(self, router, path): + for route in router.routes: + if getattr(route, "path", "") == path: + return route.endpoint + raise AssertionError(f"No route for {path}") + + def _fake_req(self, token): + """Build a fake request with the session cookie set.""" + from types import SimpleNamespace + req = SimpleNamespace() + req.cookies = {"odysseus_session": token} + req.client = SimpleNamespace() + req.client.host = "127.0.0.1" + return req + + def test_change_password_rejected_for_oidc_user(self, setup_router): + router, mgr, token = setup_router + ep = self._get(router, "/api/auth/change-password") + from pydantic import BaseModel + class PW(BaseModel): + current_password: str = "x" + new_password: str = "password123" + import asyncio + from fastapi import HTTPException + with pytest.raises(HTTPException) as exc: + asyncio.run(ep(PW(), self._fake_req(token))) + assert exc.value.status_code == 400 + assert "OIDC" in exc.value.detail + + def test_2fa_setup_rejected_for_oidc_user(self, setup_router): + router, mgr, token = setup_router + ep = self._get(router, "/api/auth/2fa/setup") + import asyncio + from fastapi import HTTPException + with pytest.raises(HTTPException) as exc: + asyncio.run(ep(self._fake_req(token))) + assert exc.value.status_code == 400 + assert "identity provider" in exc.value.detail.lower() + + def test_2fa_confirm_rejected_for_oidc_user(self, setup_router): + router, mgr, token = setup_router + ep = self._get(router, "/api/auth/2fa/confirm") + from pydantic import BaseModel + class TOTP(BaseModel): + code: str = "123456" + import asyncio + from fastapi import HTTPException + with pytest.raises(HTTPException) as exc: + asyncio.run(ep(TOTP(), self._fake_req(token))) + assert exc.value.status_code == 400 + assert "identity provider" in exc.value.detail.lower() + + def test_2fa_disable_rejected_for_oidc_user(self, setup_router): + router, mgr, token = setup_router + ep = self._get(router, "/api/auth/2fa/disable") + from pydantic import BaseModel + class DisableTOTP(BaseModel): + password: str = "x" + import asyncio + from fastapi import HTTPException + with pytest.raises(HTTPException) as exc: + asyncio.run(ep(DisableTOTP(), self._fake_req(token))) + assert exc.value.status_code == 400 + assert "identity provider" in exc.value.detail.lower() + + def test_password_user_still_can_use_2fa(self, setup_router): + """Regression: local password users must still be able to manage 2FA.""" + router, mgr, token = setup_router + # Add a local password user + mgr.create_user("bob", "hunter2") + bob_token = mgr.create_session_trusted("bob") + ep = self._get(router, "/api/auth/2fa/setup") + import asyncio + # Should NOT raise — bob is a password user + result = asyncio.run(ep(self._fake_req(bob_token))) + assert "secret" in result + assert "uri" in result diff --git a/tests/test_oidc_manager.py b/tests/test_oidc_manager.py index e290df4d4..3dff5b8aa 100644 --- a/tests/test_oidc_manager.py +++ b/tests/test_oidc_manager.py @@ -74,8 +74,10 @@ def _make_test_jwks_and_key(): return jwks, private_jwk -def _make_id_token(sub, nonce, issuer=FAKE_ISSUER, aud=FAKE_CLIENT_ID, exp=None): - """Sign a test id_token with the test RSA key.""" +def _make_id_token(sub, nonce, issuer=FAKE_ISSUER, aud=FAKE_CLIENT_ID, exp=None, azp=None): + """Sign a test id_token with the test RSA key. + + When *azp* is provided it is included in the payload.""" from authlib.jose import jwt _, jwk = _make_test_jwks_and_key() @@ -95,6 +97,8 @@ def _make_id_token(sub, nonce, issuer=FAKE_ISSUER, aud=FAKE_CLIENT_ID, exp=None) "name": sub.title(), "preferred_username": sub, } + if azp is not None: + payload["azp"] = azp return jwt.encode(header, payload, jwk).decode() @@ -186,6 +190,22 @@ class TestOidcManagerInit: client_secret=FAKE_CLIENT_SECRET, ) + def test_discovery_issuer_mismatch_fails_closed(self): + """Discovery doc with a different issuer MUST abort (OIDC Discovery §1.1).""" + import core.oidc as mod + + bad_doc = dict(DISCOVERY_DOC) + bad_doc["issuer"] = "https://evil-idp.example.com" + + with patch.object(mod.httpx, "get") as mock_get: + mock_get.return_value = _FakeResponse(200, bad_doc) + with pytest.raises(mod.OidcError, match="issuer mismatch"): + mod.OidcManager( + issuer=FAKE_ISSUER, + client_id=FAKE_CLIENT_ID, + client_secret=FAKE_CLIENT_SECRET, + ) + def test_provider_name_from_hostname(self): jwt_jwks, _ = _make_test_jwks_and_key() import core.oidc as mod @@ -456,7 +476,7 @@ class TestExchangeCode: """aud as a JSON array containing the client_id should pass.""" jwt_jwks, jwk = _make_test_jwks_and_key() nonce = "f" * 64 - id_token = _make_id_token("user123", nonce, aud=[FAKE_CLIENT_ID, "other-client"]) + id_token = _make_id_token("user123", nonce, aud=[FAKE_CLIENT_ID, "other-client"], azp=FAKE_CLIENT_ID) import core.oidc as mod with patch.object(mod.httpx, "get") as mock_get, \ @@ -481,6 +501,47 @@ class TestExchangeCode: claims = mgr.exchange_code("code", state, "https://app.example.com/callback") assert claims["sub"] == "user123" + def test_id_token_aud_array_without_azp_rejected(self): + """Multi-audience token without azp MUST be rejected (OIDC Core §2).""" + from authlib.jose import jwt + jwt_jwks, jwk = _make_test_jwks_and_key() + nonce = "g2" * 32 + # Manually build a multi-audience token without azp + header = {"alg": "RS256", "kid": "test-key-1"} + payload = { + "iss": FAKE_ISSUER, + "sub": "user123", + "aud": [FAKE_CLIENT_ID, "other-client"], + # deliberately omit azp + "exp": int(time.time()) + 3600, + "iat": int(time.time()), + "nonce": nonce, + } + id_token = jwt.encode(header, payload, jwk).decode() + import core.oidc as mod + + with patch.object(mod.httpx, "get") as mock_get, \ + patch.object(mod.httpx, "post") as mock_post: + mock_get.side_effect = [ + _mock_discovery_response(), + _mock_jwks_response(jwt_jwks), + ] + mgr = mod.OidcManager( + issuer=FAKE_ISSUER, + client_id=FAKE_CLIENT_ID, + client_secret=FAKE_CLIENT_SECRET, + ) + + state = mod._encode_state(nonce, "https://app.example.com/callback") + mock_post.return_value = _mock_token_response(id_token) + mock_get.reset_mock() + mock_get.side_effect = [ + _mock_jwks_response(jwt_jwks), + ] + + with pytest.raises(mod.OidcError, match="no azp"): + mgr.exchange_code("code", state, "https://app.example.com/callback") + def test_id_token_aud_array_missing_client_id(self): """aud as a JSON array WITHOUT the client_id should fail.""" jwt_jwks, jwk = _make_test_jwks_and_key()