feat: implement auth, projects, and frontend foundation
This commit is contained in:
@@ -0,0 +1,77 @@
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import httpx
|
||||
from jose import jwt # type: ignore[import-untyped]
|
||||
|
||||
from src.config import Settings
|
||||
|
||||
|
||||
def build_login_redirect_url(
|
||||
*,
|
||||
settings: Settings,
|
||||
redirect_uri: str,
|
||||
state: str,
|
||||
nonce: str,
|
||||
) -> str:
|
||||
query = urlencode(
|
||||
{
|
||||
"response_type": "code",
|
||||
"client_id": settings.authentik_client_id,
|
||||
"redirect_uri": redirect_uri,
|
||||
"scope": "openid profile email",
|
||||
"state": state,
|
||||
"nonce": nonce,
|
||||
}
|
||||
)
|
||||
return f"{settings.authentik_authorize_url}?{query}"
|
||||
|
||||
|
||||
async def exchange_code_for_tokens(
|
||||
*,
|
||||
settings: Settings,
|
||||
code: str,
|
||||
redirect_uri: str,
|
||||
client: httpx.AsyncClient,
|
||||
) -> dict[str, str]:
|
||||
response = await client.post(
|
||||
settings.authentik_token_url,
|
||||
data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"redirect_uri": redirect_uri,
|
||||
"client_id": settings.authentik_client_id,
|
||||
"client_secret": settings.authentik_client_secret,
|
||||
},
|
||||
)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
return {
|
||||
"access_token": payload["access_token"],
|
||||
"refresh_token": payload["refresh_token"],
|
||||
}
|
||||
|
||||
|
||||
async def fetch_jwks(*, settings: Settings, client: httpx.AsyncClient) -> dict[str, list[dict[str, str]]]:
|
||||
response = await client.get(settings.authentik_jwks_url)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
return {"keys": payload["keys"]}
|
||||
|
||||
|
||||
def verify_provider_access_token(
|
||||
*,
|
||||
settings: Settings,
|
||||
token: str,
|
||||
jwks: dict[str, list[dict[str, str]]],
|
||||
) -> dict[str, str | int]:
|
||||
unverified_header = jwt.get_unverified_header(token)
|
||||
key_id = unverified_header["kid"]
|
||||
jwk_key = next(key for key in jwks["keys"] if key.get("kid") == key_id)
|
||||
claims = jwt.decode(
|
||||
token,
|
||||
jwk_key,
|
||||
algorithms=[jwk_key.get("alg", "HS256")],
|
||||
audience=settings.authentik_audience,
|
||||
issuer=settings.authentik_issuer,
|
||||
)
|
||||
return dict(claims)
|
||||
Reference in New Issue
Block a user