Coverage for polar/oauth2/endpoints/oauth2.py: 75%
83 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 12:42 +0000
1from collections.abc import Sequence
2from typing import Literal, cast
4from fastapi import Depends, Form, HTTPException, Request, Response
5from fastapi.openapi.constants import REF_TEMPLATE
7from polar.auth.dependencies import WebUserOrAnonymous, WebUserRead, WebUserWrite
8from polar.auth.models import is_user
9from polar.kit.pagination import ListResource, PaginationParamsQuery
10from polar.models import OAuth2Token, Organization
11from polar.openapi import APITag
12from polar.organization.repository import OrganizationRepository
13from polar.postgres import AsyncSession, get_db_session
14from polar.routing import APIRouter
16from ..authorization_server import (
17 AuthorizationServer,
18 ClientConfigurationEndpoint,
19 ClientRegistrationEndpoint,
20 IntrospectionEndpoint,
21 RevocationEndpoint,
22)
23from ..dependencies import get_authorization_server, get_token
24from ..grants import AuthorizationCodeGrant
25from ..schemas import (
26 AuthorizeResponse,
27 IntrospectTokenResponse,
28 OAuth2Client,
29 OAuth2ClientConfiguration,
30 OAuth2ClientConfigurationUpdate,
31 RevokeTokenResponse,
32 TokenResponse,
33 authorize_response_adapter,
34)
35from ..schemas import (
36 UserInfo as UserInfoSchema,
37)
38from ..service.oauth2_client import oauth2_client as oauth2_client_service
39from ..sub_type import SubType
40from ..userinfo import UserInfo, generate_user_info
42router = APIRouter(prefix="/oauth2", tags=["oauth2"])
45@router.get(
46 "/",
47 summary="List Clients",
48 tags=["clients", APITag.private],
49 response_model=ListResource[OAuth2Client],
50)
51async def list(
52 auth_subject: WebUserRead,
53 pagination: PaginationParamsQuery,
54 session: AsyncSession = Depends(get_db_session),
55) -> ListResource[OAuth2Client]:
56 """List OAuth2 clients."""
57 results, count = await oauth2_client_service.list(
58 session, auth_subject, pagination=pagination
59 )
60 return ListResource.from_paginated_results(
61 [OAuth2Client.model_validate(result) for result in results], count, pagination
62 )
65@router.post(
66 "/register",
67 summary="Create Client",
68 tags=["clients", APITag.public],
69 name="oauth2:create_client",
70)
71async def create(
72 client_configuration: OAuth2ClientConfiguration,
73 request: Request,
74 auth_subject: WebUserOrAnonymous,
75 authorization_server: AuthorizationServer = Depends(get_authorization_server),
76) -> Response:
77 """Create an OAuth2 client."""
78 request.state.user = auth_subject.subject if is_user(auth_subject) else None
79 request.state.parsed_data = client_configuration.model_dump(
80 mode="json", exclude_none=True
81 )
82 return authorization_server.create_endpoint_response(
83 ClientRegistrationEndpoint.ENDPOINT_NAME, request
84 )
87@router.get(
88 "/register/{client_id}",
89 tags=["clients", APITag.public],
90 summary="Get Client",
91 name="oauth2:get_client",
92)
93async def get(
94 client_id: str,
95 request: Request,
96 auth_subject: WebUserOrAnonymous,
97 authorization_server: AuthorizationServer = Depends(get_authorization_server),
98) -> Response:
99 """Get an OAuth2 client by Client ID."""
100 request.state.user = auth_subject.subject if is_user(auth_subject) else None
101 return authorization_server.create_endpoint_response(
102 ClientConfigurationEndpoint.ENDPOINT_NAME, request
103 )
106@router.put(
107 "/register/{client_id}",
108 tags=["clients", APITag.public],
109 summary="Update Client",
110 name="oauth2:update_client",
111)
112async def update(
113 client_id: str,
114 client_configuration: OAuth2ClientConfigurationUpdate,
115 request: Request,
116 auth_subject: WebUserOrAnonymous,
117 authorization_server: AuthorizationServer = Depends(get_authorization_server),
118) -> Response:
119 """Update an OAuth2 client."""
120 request.state.user = auth_subject.subject if is_user(auth_subject) else None
121 request.state.parsed_data = client_configuration.model_dump(
122 mode="json", exclude_none=True
123 )
124 return authorization_server.create_endpoint_response(
125 ClientConfigurationEndpoint.ENDPOINT_NAME, request
126 )
129@router.delete(
130 "/register/{client_id}",
131 tags=["clients", APITag.public],
132 summary="Delete Client",
133 name="oauth2:delete_client",
134)
135async def delete(
136 client_id: str,
137 request: Request,
138 auth_subject: WebUserOrAnonymous,
139 authorization_server: AuthorizationServer = Depends(get_authorization_server),
140) -> Response:
141 """Delete an OAuth2 client."""
142 request.state.user = auth_subject.subject if is_user(auth_subject) else None
143 return authorization_server.create_endpoint_response(
144 ClientConfigurationEndpoint.ENDPOINT_NAME, request
145 )
148@router.get("/authorize", tags=[APITag.public])
149async def authorize(
150 request: Request,
151 auth_subject: WebUserOrAnonymous,
152 authorization_server: AuthorizationServer = Depends(get_authorization_server),
153 session: AsyncSession = Depends(get_db_session),
154) -> AuthorizeResponse:
155 user = auth_subject.subject if is_user(auth_subject) else None
156 await request.form()
157 grant: AuthorizationCodeGrant = authorization_server.get_consent_grant(
158 request=request, end_user=user
159 )
161 if grant.prompt == "login":
162 raise HTTPException(status_code=401)
163 elif grant.prompt == "none":
164 return authorization_server.create_authorization_response(
165 request=request, grant_user=user, save_consent=False
166 )
168 organizations: Sequence[Organization] | None = None
169 if grant.sub_type == SubType.organization:
170 assert is_user(auth_subject)
171 organization_repository = OrganizationRepository.from_session(session)
172 organizations = await organization_repository.get_all_by_user(
173 auth_subject.subject.id
174 )
176 payload = grant.request.payload
177 assert payload is not None
179 return authorize_response_adapter.validate_python(
180 {
181 "client": grant.client,
182 "scopes": payload.scope,
183 "sub_type": grant.sub_type,
184 "sub": grant.sub,
185 "organizations": organizations,
186 }
187 )
190@router.post("/consent", tags=[APITag.private])
191async def consent(
192 request: Request,
193 auth_subject: WebUserWrite,
194 action: Literal["allow", "deny"] = Form(...),
195 authorization_server: AuthorizationServer = Depends(get_authorization_server),
196) -> Response:
197 await request.form()
198 grant_user = auth_subject.subject if action == "allow" else None
199 return authorization_server.create_authorization_response(
200 request=request, grant_user=grant_user, save_consent=True
201 )
204@router.post(
205 "/token",
206 summary="Request Token",
207 name="oauth2:request_token",
208 operation_id="oauth2:request_token",
209 tags=[APITag.public],
210 openapi_extra={
211 "requestBody": {
212 "required": True,
213 "content": {
214 "application/x-www-form-urlencoded": {
215 "schema": {
216 "oneOf": [
217 {
218 "$ref": REF_TEMPLATE.format(
219 model="AuthorizationCodeTokenRequest"
220 )
221 },
222 {"$ref": REF_TEMPLATE.format(model="RefreshTokenRequest")},
223 {"$ref": REF_TEMPLATE.format(model="WebTokenRequest")},
224 ]
225 }
226 }
227 },
228 },
229 },
230 response_model=TokenResponse,
231)
232async def token(
233 request: Request,
234 authorization_server: AuthorizationServer = Depends(get_authorization_server),
235) -> Response:
236 """Request an access token using a valid grant."""
237 await request.form()
238 return authorization_server.create_token_response(request)
241@router.post(
242 "/revoke",
243 summary="Revoke Token",
244 name="oauth2:revoke_token",
245 operation_id="oauth2:revoke_token",
246 tags=[APITag.public],
247 openapi_extra={
248 "requestBody": {
249 "required": True,
250 "content": {
251 "application/x-www-form-urlencoded": {
252 "schema": {"$ref": REF_TEMPLATE.format(model="RevokeTokenRequest")}
253 }
254 },
255 },
256 },
257 response_model=RevokeTokenResponse,
258)
259async def revoke(
260 request: Request,
261 authorization_server: AuthorizationServer = Depends(get_authorization_server),
262) -> Response:
263 """Revoke an access token or a refresh token."""
264 await request.form()
265 return authorization_server.create_endpoint_response(
266 RevocationEndpoint.ENDPOINT_NAME, request
267 )
270@router.post(
271 "/introspect",
272 summary="Introspect Token",
273 name="oauth2:introspect_token",
274 operation_id="oauth2:introspect_token",
275 tags=[APITag.public],
276 openapi_extra={
277 "requestBody": {
278 "required": True,
279 "content": {
280 "application/x-www-form-urlencoded": {
281 "schema": {
282 "$ref": REF_TEMPLATE.format(model="IntrospectTokenRequest")
283 }
284 }
285 },
286 },
287 },
288 response_model=IntrospectTokenResponse,
289)
290async def introspect(
291 request: Request,
292 authorization_server: AuthorizationServer = Depends(get_authorization_server),
293) -> Response:
294 """Get information about an access token."""
295 await request.form()
296 return authorization_server.create_endpoint_response(
297 IntrospectionEndpoint.ENDPOINT_NAME, request
298 )
301@router.get(
302 "/userinfo",
303 summary="Get User Info",
304 name="oauth2:userinfo",
305 operation_id="oauth2:userinfo",
306 response_model=UserInfoSchema,
307 response_model_exclude_unset=True,
308 tags=[APITag.public],
309 openapi_extra={"x-speakeasy-name-override": "userinfo"},
310)
311async def userinfo_get(token: OAuth2Token = Depends(get_token)) -> UserInfo:
312 """Get information about the authenticated user."""
313 return generate_user_info(token.get_sub_type_value(), cast(str, token.scope))
316# Repeat the /userinfo endpoint to support POST requests
317# But don't include it in the OpenAPI schema
318@router.post(
319 "/userinfo",
320 summary="Get User Info",
321 response_model=UserInfoSchema,
322 response_model_exclude_unset=True,
323 include_in_schema=False,
324)
325async def userinfo_post(token: OAuth2Token = Depends(get_token)) -> UserInfo:
326 """Get information about the authenticated user."""
327 return generate_user_info(token.get_sub_type_value(), cast(str, token.scope))