Migrate all resources to json contents;
All checks were successful
/ build (push) Successful in 5s
/ test (push) Successful in 5s

improve token decoding & logging error messages
This commit is contained in:
phil 2025-02-07 16:09:49 +01:00
parent d39adf41ef
commit 3eb6dc3dcf
6 changed files with 77 additions and 87 deletions

View file

@ -4,8 +4,7 @@ import logging
from httpx import AsyncClient
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
from fastapi import FastAPI, HTTPException, Depends, Request, status
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.middleware.cors import CORSMiddleware
# from starlette.middleware.sessions import SessionMiddleware
@ -16,8 +15,8 @@ from .models import User
from .auth_utils import (
get_user_from_token,
UserWithRole,
get_oidc_provider,
get_token,
# get_oidc_provider,
# get_token,
)
from .settings import settings
@ -47,44 +46,46 @@ resource_server.add_middleware(
@resource_server.get("/public")
async def public() -> HTMLResponse:
return HTMLResponse("<h1>Not protected</h1>")
async def public() -> dict:
return {"msg": "Not protected"}
@resource_server.get("/protected")
async def get_protected(user: Annotated[User, Depends(get_user_from_token)]) -> HTMLResponse:
async def get_protected(user: Annotated[User, Depends(get_user_from_token)]):
assert user is not None # Just to keep QA checks happy
return HTMLResponse("<h1>Only authenticated users can see this</h1>")
return {"msg": "Only authenticated users can see this"}
@resource_server.get("/protected-by-foorole")
async def get_protected_by_foorole(
user: Annotated[User, Depends(UserWithRole("foorole"))]
) -> HTMLResponse:
return HTMLResponse("<h1>Only users with foorole can see this</h1>")
user: Annotated[User, Depends(UserWithRole("foorole"))],
):
assert user is not None
return {"msg": "Only users with foorole can see this"}
@resource_server.get("/protected-by-barrole")
async def get_protected_by_barrole(
user: Annotated[User, Depends(UserWithRole("barrole"))]
) -> HTMLResponse:
return HTMLResponse("<h1>Protected by barrole</h1>")
user: Annotated[User, Depends(UserWithRole("barrole"))],
):
assert user is not None
return {"msg": "Protected by barrole"}
@resource_server.get("/protected-by-foorole-and-barrole")
async def get_protected_by_foorole_and_barrole(
user: Annotated[User, Depends(UserWithRole("foorole")), Depends(UserWithRole("barrole"))],
) -> HTMLResponse:
):
assert user is not None # Just to keep QA checks happy
return HTMLResponse("<h1>Only users with foorole and barrole can see this</h1>")
return {"msg": "Only users with foorole and barrole can see this"}
@resource_server.get("/protected-by-foorole-or-barrole")
async def get_protected_by_foorole_or_barrole(
user: Annotated[User, Depends(UserWithRole(["foorole", "barrole"]))]
) -> HTMLResponse:
user: Annotated[User, Depends(UserWithRole(["foorole", "barrole"]))],
):
assert user is not None # Just to keep QA checks happy
return HTMLResponse("<h1>Only users with foorole or barrole can see this</h1>")
return {"msg": "Only users with foorole or barrole can see this"}
# @resource_server.get("/introspect")
@ -118,9 +119,9 @@ async def get_resource_(
# oidc_provider: Annotated[StarletteOAuth2App, Depends(get_oidc_provider)],
# token: Annotated[OAuth2Token, Depends(get_token)],
user: Annotated[User, Depends(get_user_from_token)],
) -> JSONResponse:
):
"""Generic path for testing a resource provided by a provider"""
return JSONResponse(await get_resource(id, user))
return await get_resource(id, user)
async def get_resource(resource_id: str, user: User) -> dict: