Coverage for polar/oauth2/dependencies.py: 83%

1from collections.abc import Generator 

2 

3from fastapi import Depends, Request 

4from fastapi.security import OpenIdConnect 

5from fastapi.security.utils import get_authorization_scheme_param 

6 

7from polar.auth.scope import SCOPES_SUPPORTED 

8from polar.exceptions import Unauthorized 

9from polar.kit.db.postgres import SyncSessionMaker 

10from polar.models import OAuth2Token 

11from polar.postgres import AsyncSession, get_db_session 

12 

13from .authorization_server import AuthorizationServer 

14from .exceptions import InvalidTokenError 

15from .service.oauth2_token import oauth2_token as oauth2_token_service 

16 

17openid_scheme = OpenIdConnect( 

18 scheme_name="oidc", 

19 openIdConnectUrl="/.well-known/openid-configuration", 

20 auto_error=False, 

21) 

22 

23 

24async def get_optional_token( 

25 authorization: str = Depends(openid_scheme), 

26 session: AsyncSession = Depends(get_db_session), 

27) -> tuple[OAuth2Token | None, bool]: 

28 scheme, access_token = get_authorization_scheme_param(authorization) 

29 if not authorization or scheme.lower() != "bearer": 29 ↛ 32line 29 didn't jump to line 32 because the condition on line 29 was always true

30 return None, False 

31 

32 token = await oauth2_token_service.get_by_access_token(session, access_token) 

33 return token, True 

34 

35 

36async def get_token( 

37 credentials: tuple[OAuth2Token | None, bool] = Depends(get_optional_token), 

38) -> OAuth2Token: 

39 token, authorization_set = credentials 

40 if token is None: 40 ↛ 44line 40 didn't jump to line 44 because the condition on line 40 was always true

41 if authorization_set: 41 ↛ 42line 41 didn't jump to line 42 because the condition on line 41 was never true

42 raise InvalidTokenError() 

43 raise Unauthorized() 

44 return token 

45 

46 

47def get_authorization_server( 

48 request: Request, 

49) -> Generator[AuthorizationServer, None, None]: 

50 sync_sessionmaker: SyncSessionMaker = request.state.sync_sessionmaker 

51 with sync_sessionmaker() as session: 

52 authorization_server = AuthorizationServer.build( 

53 session, scopes_supported=SCOPES_SUPPORTED 

54 ) 

55 try: 

56 yield authorization_server 

57 except: 

58 session.rollback() 

59 raise 

60 else: 

61 session.commit()