{escape(payload.get("message", "Authentication completed."))}
""" return HTMLResponse(content=html, status_code=status_code) def _read_oidc_state(session_state: str) -> AuthOIDCState: session = get_session() try: row = session.execute(select(AuthOIDCState).where(AuthOIDCState.state == session_state)).scalar_one_or_none() if not row: raise HTTPException(status_code=400, detail="Unknown OIDC state") if row.consumed_at: raise HTTPException(status_code=400, detail="OIDC state already consumed") expires = datetime.fromisoformat(row.expires_at.replace("Z", "+00:00")) if expires < _now(): raise HTTPException(status_code=400, detail="OIDC state expired") return row finally: session.close() def _consume_oidc_state(state_value: str) -> None: session = get_session() try: row = session.execute(select(AuthOIDCState).where(AuthOIDCState.state == state_value)).scalar_one_or_none() if not row: return row.consumed_at = utc_now_iso() session.commit() finally: session.close() def _save_oidc_state(state_value: str, nonce: str, code_verifier: str, redirect_uri: str, return_mode: str) -> None: session = get_session() try: now = _now() row = AuthOIDCState( state=state_value, nonce=nonce, code_verifier=code_verifier, redirect_uri=redirect_uri, return_mode=return_mode, created_at=_iso(now), expires_at=_iso(now + timedelta(seconds=_oidc_state_ttl_seconds())), consumed_at=None, ) session.add(row) session.commit() finally: session.close() def _extract_claims_from_compact_jwt(token: str) -> dict[str, Any]: try: parts = token.split(".") if len(parts) != 3: raise ValueError("compact jwt expected") payload_part = parts[1] padding = "=" * (-len(payload_part) % 4) raw = base64.urlsafe_b64decode((payload_part + padding).encode("utf-8")) payload = json.loads(raw.decode("utf-8")) except Exception as exc: # noqa: BLE001 raise HTTPException(status_code=502, detail="OIDC id_token payload invalid") from exc if not isinstance(payload, dict): raise HTTPException(status_code=502, detail="OIDC id_token payload invalid") return payload def _map_role_from_claims(claims: dict[str, Any]) -> Role: role_claim_name = _oidc_role_claim() role_map = _oidc_role_map() source = claims.get(role_claim_name) candidates: list[str] = [] if isinstance(source, list): candidates = [str(item) for item in source] elif isinstance(source, str): candidates = [source] elif source is not None: candidates = [str(source)] for candidate in candidates: mapped = role_map.get(candidate) if mapped: return Role(mapped) raise HTTPException(status_code=403, detail="OIDC role is not mapped") def _upsert_external_identity(claims: dict[str, Any], role: Role) -> None: provider = _oidc_provider() subject = str(claims.get("sub") or "").strip() if not subject: raise HTTPException(status_code=403, detail="OIDC subject missing") username = str( claims.get("preferred_username") or claims.get("username") or claims.get("email") or subject ).strip() now = utc_now_iso() full_name = str(claims.get("name") or "").strip() or None email = str(claims.get("email") or "").strip() or None session = get_session() try: row = session.execute( select(AuthExternalIdentity).where( AuthExternalIdentity.provider == provider, AuthExternalIdentity.external_subject == subject, ) ).scalar_one_or_none() if not row: row = AuthExternalIdentity( provider=provider, external_subject=subject, username=username, email=email, full_name=full_name, role=role.value, linked_user_id=None, created_at=now, updated_at=now, last_login_at=now, ) session.add(row) else: row.username = username row.email = email row.full_name = full_name row.role = role.value row.updated_at = now row.last_login_at = now session.commit() finally: session.close() def _oidc_payload_from_mock(request: Request, state_row: AuthOIDCState) -> dict[str, Any]: if not _oidc_test_mode(): raise HTTPException(status_code=502, detail="OIDC provider exchange requires a live provider") params = request.query_params subject = params.get("mock_sub") or "oidc-user" username = params.get("mock_username") or "oidc.user" email = params.get("mock_email") or "oidc.user@example.com" full_name = params.get("mock_full_name") or "OIDC User" groups = [item.strip() for item in (params.get("mock_groups") or "kc_operator").split(",") if item.strip()] claims = { "sub": subject, "preferred_username": username, "email": email, "name": full_name, _oidc_role_claim(): groups, "iss": _oidc_issuer_url() or "test-issuer", "nonce": state_row.nonce, } return claims def _oidc_payload_from_token_exchange(code: str, state_row: AuthOIDCState) -> dict[str, Any]: metadata = _provider_metadata() token_endpoint = str(metadata.get("token_endpoint") or "").strip() if not token_endpoint: raise HTTPException(status_code=503, detail="OIDC token endpoint unavailable") form = { "grant_type": "authorization_code", "code": code, "redirect_uri": state_row.redirect_uri, "client_id": _oidc_client_id(), "code_verifier": state_row.code_verifier, } if _oidc_client_secret(): form["client_secret"] = _oidc_client_secret() try: response = httpx.post(token_endpoint, data=form, timeout=8) response.raise_for_status() except Exception as exc: # noqa: BLE001 raise HTTPException(status_code=502, detail="OIDC token exchange failed") from exc payload = response.json() if not isinstance(payload, dict): raise HTTPException(status_code=502, detail="OIDC token payload invalid") claims: dict[str, Any] | None = None if isinstance(payload.get("id_token_claims"), dict): claims = payload["id_token_claims"] elif isinstance(payload.get("userinfo"), dict): claims = payload["userinfo"] elif payload.get("id_token"): claims = _extract_claims_from_compact_jwt(str(payload["id_token"])) if not claims: raise HTTPException(status_code=502, detail="OIDC claims unavailable") issuer = str(metadata.get("issuer") or "").strip() if issuer and claims.get("iss") and str(claims.get("iss")) != issuer: raise HTTPException(status_code=403, detail="OIDC issuer mismatch") if claims.get("nonce") and str(claims.get("nonce")) != state_row.nonce: raise HTTPException(status_code=403, detail="OIDC nonce mismatch") return claims def _issue_login_response(user: AuthUser) -> LoginResponse: token = issue_app_token( subject=user.user_id, username=user.username, role=user.role, auth_source="local", full_name=user.full_name, ) return LoginResponse( access_token=token, role=Role(user.role), auth_source="local", full_name=user.full_name, ) def _user_out(user: AuthUser) -> UserOut: return UserOut( user_id=user.user_id, username=user.username, full_name=user.full_name, role=Role(user.role), ) _seed_users() @app.get("/health", response_model=HealthResponse) def health() -> HealthResponse: return HealthResponse(status="ok", service="auth-service") @app.get("/auth/roles") def roles() -> dict: return {"roles": [r.value for r in Role]} @app.get("/auth/oidc/config") def oidc_config() -> dict: return { "enabled": _oidc_enabled(), "provider_label": "Keycloak", "login_path": "/auth/oidc/start?return_mode=popup", "auth_source": "oidc", } @app.get("/auth/oidc/health") def oidc_health() -> dict: if not _oidc_enabled(): return { "status": "disabled", "provider": _oidc_provider(), "issuer": _oidc_issuer_url() or None, "jwks_loaded": False, } metadata = _provider_metadata() jwks_loaded = False try: _provider_jwks() jwks_loaded = True except HTTPException: jwks_loaded = False return { "status": "ok" if jwks_loaded else "degraded", "provider": _oidc_provider(), "issuer": metadata.get("issuer") or _oidc_issuer_url() or None, "jwks_loaded": jwks_loaded, } @app.get("/auth/oidc/start") def oidc_start(return_mode: str = "popup") -> RedirectResponse: if not _oidc_enabled(): raise HTTPException(status_code=503, detail="OIDC is disabled") metadata = _provider_metadata() authorize_url = str(metadata.get("authorization_endpoint") or "").strip() if not authorize_url: raise HTTPException(status_code=503, detail="OIDC authorization endpoint unavailable") state_value = token_urlsafe(24) nonce = token_urlsafe(16) code_verifier = _build_pkce_verifier() redirect_uri = _oidc_redirect_uri() _save_oidc_state(state_value, nonce, code_verifier, redirect_uri, return_mode) params = { "client_id": _oidc_client_id(), "redirect_uri": redirect_uri, "response_type": "code", "scope": _oidc_scopes(), "state": state_value, "nonce": nonce, "code_challenge": _build_pkce_challenge(code_verifier), "code_challenge_method": "S256", } return RedirectResponse(url=f"{authorize_url}?{urlencode(params)}", status_code=302) @app.get("/auth/oidc/callback") def oidc_callback(request: Request, state: str = "", code: str = "", error: str = "") -> HTMLResponse: if not _oidc_enabled(): return _html_bridge({"type": "oidc-error", "message": "Corporate login is disabled"}, status_code=503) if not state: return _html_bridge({"type": "oidc-error", "message": "OIDC state is required"}, status_code=400) try: state_row = _read_oidc_state(state) except HTTPException as exc: return _html_bridge({"type": "oidc-error", "message": exc.detail}, status_code=exc.status_code) if error: _consume_oidc_state(state) return _html_bridge({"type": "oidc-error", "message": f"OIDC provider error: {error}"}, status_code=400) try: if request.query_params.get("mock_sub"): claims = _oidc_payload_from_mock(request, state_row) else: if not code: raise HTTPException(status_code=400, detail="OIDC code is required") claims = _oidc_payload_from_token_exchange(code, state_row) role = _map_role_from_claims(claims) _upsert_external_identity(claims, role) username = str( claims.get("preferred_username") or claims.get("username") or claims.get("email") or claims.get("sub") or "oidc-user" ).strip() full_name = str(claims.get("name") or "").strip() or None email = str(claims.get("email") or "").strip() or None token = issue_app_token( subject=str(claims.get("sub") or ""), username=username, role=role.value, auth_source="oidc", provider=_oidc_provider(), full_name=full_name, email=email, ) _consume_oidc_state(state) return _html_bridge( { "type": "oidc-login", "message": "Corporate login completed", "access_token": token, "role": role.value, "username": username, "full_name": full_name, "auth_source": "oidc", "provider": _oidc_provider(), "email": email, } ) except HTTPException as exc: _consume_oidc_state(state) return _html_bridge({"type": "oidc-error", "message": str(exc.detail)}, status_code=exc.status_code) @app.post("/auth/login", response_model=LoginResponse) def login(payload: LoginRequest) -> LoginResponse: session = get_session() try: user = session.execute( select(AuthUser).where(AuthUser.username == payload.username) ).scalar_one_or_none() if not user or user.password != payload.password: raise HTTPException(status_code=401, detail="Invalid credentials") return _issue_login_response(user) finally: session.close() @app.get("/auth/me") def me(actor: dict = Depends(get_actor)) -> dict: return actor @app.get("/users", response_model=list[UserOut]) def list_users(_: dict = Depends(get_actor)) -> list[UserOut]: actor = _ if actor["role"] not in {Role.ADMIN.value, Role.SUPERVISOR.value}: raise HTTPException(status_code=403, detail="Insufficient role") session = get_session() try: users = session.execute(select(AuthUser).order_by(AuthUser.id.asc())).scalars().all() return [_user_out(u) for u in users] finally: session.close() @app.post("/users", response_model=UserOut) def create_user(payload: UserCreate, actor: dict = Depends(get_actor)) -> UserOut: if actor["role"] != Role.ADMIN.value: raise HTTPException(status_code=403, detail="Insufficient role") session = get_session() try: existing = session.execute( select(AuthUser).where(AuthUser.username == payload.username) ).scalar_one_or_none() if existing: raise HTTPException(status_code=409, detail="Username already exists") user = AuthUser( user_id=new_id("usr"), username=payload.username, password=payload.password, full_name=payload.full_name, role=payload.role.value, created_at=utc_now_iso(), ) session.add(user) session.commit() return _user_out(user) finally: session.close() @app.get("/users/{user_id}", response_model=UserOut) def get_user(user_id: str, actor: dict = Depends(get_actor)) -> UserOut: if actor["role"] not in {Role.ADMIN.value, Role.SUPERVISOR.value}: raise HTTPException(status_code=403, detail="Insufficient role") session = get_session() try: user = session.execute(select(AuthUser).where(AuthUser.user_id == user_id)).scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="User not found") return _user_out(user) finally: session.close() @app.patch("/users/{user_id}", response_model=UserOut) def update_user(user_id: str, payload: UserUpdate, actor: dict = Depends(get_actor)) -> UserOut: if actor["role"] != Role.ADMIN.value: raise HTTPException(status_code=403, detail="Insufficient role") session = get_session() try: user = session.execute(select(AuthUser).where(AuthUser.user_id == user_id)).scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="User not found") if payload.password is not None: user.password = payload.password if payload.full_name is not None: user.full_name = payload.full_name if payload.role is not None: user.role = payload.role.value session.commit() session.refresh(user) return _user_out(user) finally: session.close() @app.delete("/users/{user_id}") def delete_user(user_id: str, actor: dict = Depends(get_actor)) -> dict: if actor["role"] != Role.ADMIN.value: raise HTTPException(status_code=403, detail="Insufficient role") session = get_session() try: user = session.execute(select(AuthUser).where(AuthUser.user_id == user_id)).scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="User not found") if actor.get("user") == user.username: raise HTTPException(status_code=409, detail="Cannot delete current user") session.delete(user) session.commit() return {"user_id": user_id, "deleted": True} finally: session.close()