Coverage for polar/oauth2/service/oauth2_authorization_code.py: 54%

1import structlog 

2from sqlalchemy import select 

3 

4from polar.config import settings 

5from polar.enums import TokenType 

6from polar.kit.crypto import get_token_hash 

7from polar.kit.services import ResourceServiceReader 

8from polar.logging import Logger 

9from polar.models import OAuth2AuthorizationCode 

10from polar.postgres import AsyncSession 

11 

12log: Logger = structlog.get_logger() 

13 

14 

15class OAuth2AuthorizationCodeService(ResourceServiceReader[OAuth2AuthorizationCode]): 

16 async def revoke_leaked( 

17 self, 

18 session: AsyncSession, 

19 token: str, 

20 token_type: TokenType, 

21 *, 

22 notifier: str, 

23 url: str | None = None, 

24 ) -> bool: 

25 statement = select(OAuth2AuthorizationCode).where( 

26 OAuth2AuthorizationCode.code 

27 == get_token_hash(token, secret=settings.SECRET) 

28 ) 

29 

30 result = await session.execute(statement) 

31 authorization_code = result.unique().scalar_one_or_none() 

32 

33 if authorization_code is None: 

34 return False 

35 

36 authorization_code.set_deleted_at() 

37 session.add(authorization_code) 

38 

39 log.info( 

40 "Revoke leaked authorization code", 

41 id=authorization_code.id, 

42 notifier=notifier, 

43 url=url, 

44 ) 

45 

46 return True 

47 

48 

49oauth2_authorization_code = OAuth2AuthorizationCodeService(OAuth2AuthorizationCode)