Coverage for polar/organization_access_token/service.py: 83%

1import uuid 

2from collections.abc import Sequence 

3from typing import Any 

4from uuid import UUID 

5 

6import structlog 

7from sqlalchemy import UnaryExpression, asc, desc 

8 

9from polar.auth.models import AuthSubject 

10from polar.config import settings 

11from polar.email.react import render_email_template 

12from polar.email.schemas import ( 

13 OrganizationAccessTokenLeakedEmail, 

14 OrganizationAccessTokenLeakedProps, 

15) 

16from polar.email.sender import enqueue_email 

17from polar.enums import TokenType 

18from polar.integrations.loops.service import loops as loops_service 

19from polar.kit.crypto import generate_token_hash_pair, get_token_hash 

20from polar.kit.pagination import PaginationParams 

21from polar.kit.sorting import Sorting 

22from polar.kit.utils import utc_now 

23from polar.logging import Logger 

24from polar.models import OrganizationAccessToken, User 

25from polar.organization.resolver import get_payload_organization 

26from polar.postgres import AsyncSession 

27from polar.user_organization.service import ( 

28 user_organization as user_organization_service, 

29) 

30 

31from .repository import OrganizationAccessTokenRepository 

32from .schemas import OrganizationAccessTokenCreate, OrganizationAccessTokenUpdate 

33from .sorting import OrganizationAccessTokenSortProperty 

34 

35log: Logger = structlog.get_logger() 

36 

37TOKEN_PREFIX = "polar_oat_" 

38 

39 

40class OrganizationAccessTokenService: 

41 async def list( 

42 self, 

43 session: AsyncSession, 

44 auth_subject: AuthSubject[User], 

45 *, 

46 organization_id: Sequence[uuid.UUID] | None = None, 

47 pagination: PaginationParams, 

48 sorting: list[Sorting[OrganizationAccessTokenSortProperty]] = [ 

49 (OrganizationAccessTokenSortProperty.created_at, False) 

50 ], 

51 ) -> tuple[Sequence[OrganizationAccessToken], int]: 

52 repository = OrganizationAccessTokenRepository.from_session(session) 

53 statement = repository.get_readable_statement(auth_subject) 

54 

55 if organization_id is not None: 55 ↛ 60line 55 didn't jump to line 60 because the condition on line 55 was always true

56 statement = statement.where( 

57 OrganizationAccessToken.organization_id.in_(organization_id) 

58 ) 

59 

60 order_by_clauses: list[UnaryExpression[Any]] = [] 

61 for criterion, is_desc in sorting: 

62 clause_function = desc if is_desc else asc 

63 if criterion == OrganizationAccessTokenSortProperty.created_at: 

64 order_by_clauses.append( 

65 clause_function(OrganizationAccessToken.created_at) 

66 ) 

67 elif criterion == OrganizationAccessTokenSortProperty.comment: 

68 order_by_clauses.append( 

69 clause_function(OrganizationAccessToken.comment) 

70 ) 

71 elif criterion == OrganizationAccessTokenSortProperty.last_used_at: 

72 order_by_clauses.append( 

73 clause_function(OrganizationAccessToken.last_used_at) 

74 ) 

75 elif criterion == OrganizationAccessTokenSortProperty.organization_id: 75 ↛ 61line 75 didn't jump to line 61 because the condition on line 75 was always true

76 order_by_clauses.append( 

77 clause_function(OrganizationAccessToken.organization_id) 

78 ) 

79 statement = statement.order_by(*order_by_clauses) 

80 

81 return await repository.paginate( 

82 statement, limit=pagination.limit, page=pagination.page 

83 ) 

84 

85 async def get( 

86 self, session: AsyncSession, auth_subject: AuthSubject[User], id: UUID 

87 ) -> OrganizationAccessToken | None: 

88 repository = OrganizationAccessTokenRepository.from_session(session) 

89 statement = repository.get_readable_statement(auth_subject).where( 

90 OrganizationAccessToken.id == id, 

91 OrganizationAccessToken.deleted_at.is_(None), 

92 ) 

93 return await repository.get_one_or_none(statement) 

94 

95 async def get_by_token( 

96 self, session: AsyncSession, token: str, *, expired: bool = False 

97 ) -> OrganizationAccessToken | None: 

98 token_hash = get_token_hash(token, secret=settings.SECRET) 

99 repository = OrganizationAccessTokenRepository.from_session(session) 

100 return await repository.get_by_token_hash(token_hash, expired=expired) 

101 

102 async def create( 

103 self, 

104 session: AsyncSession, 

105 auth_subject: AuthSubject[User], 

106 create_schema: OrganizationAccessTokenCreate, 

107 ) -> tuple[OrganizationAccessToken, str]: 

108 organization = await get_payload_organization( 

109 session, auth_subject, create_schema 

110 ) 

111 token, token_hash = generate_token_hash_pair( 

112 secret=settings.SECRET, prefix=TOKEN_PREFIX 

113 ) 

114 organization_access_token = OrganizationAccessToken( 

115 **create_schema.model_dump( 

116 exclude={"scopes", "expires_in", "organization_id"} 

117 ), 

118 organization=organization, 

119 token=token_hash, 

120 expires_at=utc_now() + create_schema.expires_in 

121 if create_schema.expires_in 

122 else None, 

123 scope=" ".join(create_schema.scopes), 

124 ) 

125 repository = OrganizationAccessTokenRepository.from_session(session) 

126 organization_access_token = await repository.create( 

127 organization_access_token, flush=True 

128 ) 

129 

130 user = auth_subject.subject 

131 await loops_service.user_created_personal_access_token(session, user) 

132 

133 return organization_access_token, token 

134 

135 async def update( 

136 self, 

137 session: AsyncSession, 

138 organization_access_token: OrganizationAccessToken, 

139 update_schema: OrganizationAccessTokenUpdate, 

140 ) -> OrganizationAccessToken: 

141 repository = OrganizationAccessTokenRepository.from_session(session) 

142 

143 update_dict = update_schema.model_dump(exclude={"scopes"}, exclude_unset=True) 

144 if update_schema.scopes is not None: 

145 update_dict["scope"] = " ".join(update_schema.scopes) 

146 

147 return await repository.update( 

148 organization_access_token, update_dict=update_dict 

149 ) 

150 

151 async def delete( 

152 self, session: AsyncSession, organization_access_token: OrganizationAccessToken 

153 ) -> None: 

154 repository = OrganizationAccessTokenRepository.from_session(session) 

155 await repository.soft_delete(organization_access_token) 

156 

157 async def revoke_leaked( 

158 self, 

159 session: AsyncSession, 

160 token: str, 

161 token_type: TokenType, 

162 *, 

163 notifier: str, 

164 url: str | None = None, 

165 ) -> bool: 

166 organization_access_token = await self.get_by_token(session, token) 

167 

168 if organization_access_token is None: 

169 return False 

170 

171 repository = OrganizationAccessTokenRepository.from_session(session) 

172 await repository.soft_delete(organization_access_token) 

173 

174 organization_members = await user_organization_service.list_by_org( 

175 session, organization_access_token.organization_id 

176 ) 

177 for organization_member in organization_members: 

178 email = organization_member.user.email 

179 body = render_email_template( 

180 OrganizationAccessTokenLeakedEmail( 

181 props=OrganizationAccessTokenLeakedProps( 

182 email=email, 

183 organization_access_token=organization_access_token.comment, 

184 notifier=notifier, 

185 url=url or "", 

186 ) 

187 ) 

188 ) 

189 enqueue_email( 

190 to_email_addr=email, 

191 subject="Security Notice - Your Polar Organization Access Token has been leaked", 

192 html_content=body, 

193 ) 

194 

195 log.info( 

196 "Revoke leaked organization access token", 

197 id=organization_access_token.id, 

198 notifier=notifier, 

199 url=url, 

200 ) 

201 

202 return True 

203 

204 

205organization_access_token = OrganizationAccessTokenService()