Coverage for server / services / common / auth0_management.py: 86%

1""" 

2Auth0 Management API Client (Teacher-Student API) 

3 

4Provides async methods for creating, updating, and deleting users in Auth0 

5via the Management API. Used when students are created via the local fallback 

6path (when Express bulk-upload is unavailable) so they also exist in Auth0 

7and can log in. 

8 

9Uses M2M (client_credentials) tokens with caching. 

10""" 

11 

12import logging 

13import os 

14import time 

15from typing import Optional 

16 

17import httpx 

18from dotenv import load_dotenv 

19 

20load_dotenv() 

21 

22logger = logging.getLogger(__name__) 

23 

24# Auth0 Management API configuration 

25AUTH0_DOMAIN = os.getenv("AUTH0_DOMAIN", "") 

26AUTH0_MGMT_CLIENT_ID = os.getenv("AUTH0_MGMT_CLIENT_ID", "") 

27AUTH0_MGMT_CLIENT_SECRET = os.getenv("AUTH0_MGMT_CLIENT_SECRET", "") 

28AUTH0_MGMT_AUDIENCE = os.getenv( 

29 "AUTH0_MGMT_AUDIENCE", 

30 f"https://{AUTH0_DOMAIN}/api/v2/" if AUTH0_DOMAIN else "", 

31) 

32 

33# Role ID mappings 

34AUTH0_ROLE_IDS = { 

35 "admin": os.getenv("AUTH0_ROLE_ID_ADMIN", ""), 

36 "staff": os.getenv("AUTH0_ROLE_ID_STAFF", ""), 

37 "teacher": os.getenv("AUTH0_ROLE_ID_TEACHER", ""), 

38 "student": os.getenv("AUTH0_ROLE_ID_STUDENT", ""), 

39} 

40 

41 

42class Auth0ManagementService: 

43 """Auth0 Management API client for user lifecycle operations.""" 

44 

45 def __init__(self): 

46 self._cached_token: Optional[str] = None 

47 self._token_expires_at: float = 0 

48 self._last_call_time: float = 0 

49 

50 def _is_configured(self) -> bool: 

51 """Check if Auth0 Management API credentials are configured.""" 

52 return bool(AUTH0_DOMAIN and AUTH0_MGMT_CLIENT_ID and AUTH0_MGMT_CLIENT_SECRET) 

53 

54 async def _rate_limit(self): 

55 """Enforce minimum 100ms between API calls to avoid Auth0 rate limits.""" 

56 now = time.monotonic() 

57 elapsed = now - self._last_call_time 

58 if elapsed < 0.1: 

59 import asyncio 

60 await asyncio.sleep(0.1 - elapsed) 

61 self._last_call_time = time.monotonic() 

62 

63 async def get_management_token(self) -> str: 

64 """ 

65 Get a cached M2M token for the Auth0 Management API. 

66 

67 Token is cached for 58 minutes (Auth0 tokens expire after 24h by default, 

68 but we refresh conservatively). 

69 """ 

70 now = time.time() 

71 if self._cached_token and now < self._token_expires_at: 

72 return self._cached_token 

73 

74 async with httpx.AsyncClient(timeout=10.0) as client: 

75 response = await client.post( 

76 f"https://{AUTH0_DOMAIN}/oauth/token", 

77 json={ 

78 "client_id": AUTH0_MGMT_CLIENT_ID, 

79 "client_secret": AUTH0_MGMT_CLIENT_SECRET, 

80 "audience": AUTH0_MGMT_AUDIENCE, 

81 "grant_type": "client_credentials", 

82 }, 

83 ) 

84 response.raise_for_status() 

85 data = response.json() 

86 

87 self._cached_token = data["access_token"] 

88 # Cache for 58 minutes (or use expires_in if shorter) 

89 expires_in = min(data.get("expires_in", 3480), 3480) 

90 self._token_expires_at = now + expires_in 

91 return self._cached_token 

92 

93 async def create_user( 

94 self, 

95 email: str, 

96 password: str, 

97 role: str, 

98 given_name: str = "", 

99 family_name: str = "", 

100 ) -> dict: 

101 """ 

102 Create a user in Auth0 and assign their role. 

103 

104 Args: 

105 email: User email address. 

106 password: Initial password for the user. 

107 role: Role to assign (teacher, student). 

108 given_name: User's first name. 

109 family_name: User's last name. 

110 

111 Returns: 

112 dict with user_id, email, and other Auth0 user fields. 

113 

114 Raises: 

115 httpx.HTTPStatusError: On Auth0 API errors. 

116 """ 

117 if not self._is_configured(): 

118 raise RuntimeError("Auth0 Management API not configured. Set AUTH0_DOMAIN, AUTH0_MGMT_CLIENT_ID, AUTH0_MGMT_CLIENT_SECRET.") 

119 

120 token = await self.get_management_token() 

121 headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} 

122 

123 # Create the user 

124 await self._rate_limit() 

125 async with httpx.AsyncClient(timeout=10.0) as client: 

126 create_response = await client.post( 

127 f"https://{AUTH0_DOMAIN}/api/v2/users", 

128 headers=headers, 

129 json={ 

130 "email": email, 

131 "password": password, 

132 "connection": "Username-Password-Authentication", 

133 "given_name": given_name or "User", 

134 "family_name": family_name or "User", 

135 "name": f"{given_name} {family_name}".strip() or email, 

136 "app_metadata": {"role": role}, 

137 }, 

138 ) 

139 if create_response.status_code == 429: 

140 logger.warning("Auth0 rate limit hit during user creation") 

141 raise httpx.HTTPStatusError( 

142 "Rate limited", request=create_response.request, response=create_response 

143 ) 

144 create_response.raise_for_status() 

145 user_data = create_response.json() 

146 

147 auth0_user_id = user_data["user_id"] 

148 

149 # Assign role if role ID is configured 

150 role_id = AUTH0_ROLE_IDS.get(role) 

151 if role_id: 

152 await self._rate_limit() 

153 async with httpx.AsyncClient(timeout=10.0) as client: 

154 role_response = await client.post( 

155 f"https://{AUTH0_DOMAIN}/api/v2/users/{auth0_user_id}/roles", 

156 headers=headers, 

157 json={"roles": [role_id]}, 

158 ) 

159 if role_response.status_code != 204: 

160 logger.warning( 

161 f"Failed to assign role '{role}' to Auth0 user {auth0_user_id}: " 

162 f"{role_response.status_code} {role_response.text}" 

163 ) 

164 

165 logger.info(f"Created Auth0 user {auth0_user_id} with role '{role}'") 

166 return user_data 

167 

168 async def update_user(self, auth0_user_id: str, **fields) -> dict: 

169 """ 

170 Update a user's profile in Auth0. 

171 

172 Accepts keyword arguments matching Auth0 user fields: 

173 email, given_name, family_name, name, blocked, etc. 

174 

175 Returns the updated user object from Auth0. 

176 """ 

177 if not self._is_configured(): 

178 raise RuntimeError("Auth0 Management API not configured.") 

179 

180 token = await self.get_management_token() 

181 headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} 

182 

183 await self._rate_limit() 

184 async with httpx.AsyncClient(timeout=10.0) as client: 

185 response = await client.patch( 

186 f"https://{AUTH0_DOMAIN}/api/v2/users/{auth0_user_id}", 

187 headers=headers, 

188 json=fields, 

189 ) 

190 response.raise_for_status() 

191 return response.json() 

192 

193 async def delete_user(self, auth0_user_id: str) -> None: 

194 """Delete a user from Auth0.""" 

195 if not self._is_configured(): 

196 raise RuntimeError("Auth0 Management API not configured.") 

197 

198 token = await self.get_management_token() 

199 headers = {"Authorization": f"Bearer {token}"} 

200 

201 await self._rate_limit() 

202 async with httpx.AsyncClient(timeout=10.0) as client: 

203 response = await client.delete( 

204 f"https://{AUTH0_DOMAIN}/api/v2/users/{auth0_user_id}", 

205 headers=headers, 

206 ) 

207 response.raise_for_status() 

208 

209 logger.info(f"Deleted Auth0 user {auth0_user_id}") 

210 

211 async def generate_password_change_ticket( 

212 self, 

213 auth0_user_id: str, 

214 result_url: str, 

215 ttl_sec: int = 432000, # 5 days 

216 mark_email_as_verified: bool = False, 

217 ) -> dict: 

218 """ 

219 Generate an Auth0 password-change ticket for a user. 

220 

221 The returned ticket URL sends the user to Auth0's hosted page where 

222 they can set a new password without this backend ever touching the 

223 plaintext. This is the Auth0-recommended pattern and mirrors the 

224 invitation-email flow in the bulk-upload service. 

225 

226 Args: 

227 auth0_user_id: The Auth0 user id (e.g., "auth0|abc123"). 

228 result_url: Where Auth0 redirects after success (typically /login). 

229 ttl_sec: Ticket validity in seconds. Default 5 days. 

230 mark_email_as_verified: If True, Auth0 flips email_verified when 

231 the ticket is *generated*. Default False. 

232 

233 Returns: 

234 dict with keys ``ticket`` (the opaque URL the user must open) 

235 and ``expires_at`` (ISO-8601 timestamp, computed locally). 

236 """ 

237 if not self._is_configured(): 

238 raise RuntimeError("Auth0 Management API not configured.") 

239 

240 token = await self.get_management_token() 

241 headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} 

242 

243 payload = { 

244 "user_id": auth0_user_id, 

245 "result_url": result_url, 

246 "ttl_sec": ttl_sec, 

247 "mark_email_as_verified": mark_email_as_verified, 

248 } 

249 

250 await self._rate_limit() 

251 async with httpx.AsyncClient(timeout=10.0) as client: 

252 response = await client.post( 

253 f"https://{AUTH0_DOMAIN}/api/v2/tickets/password-change", 

254 headers=headers, 

255 json=payload, 

256 ) 

257 response.raise_for_status() 

258 data = response.json() 

259 

260 from datetime import datetime, timedelta, timezone 

261 expires_at = (datetime.now(timezone.utc) + timedelta(seconds=ttl_sec)).isoformat() 

262 

263 logger.info( 

264 f"Generated password-change ticket for Auth0 user {auth0_user_id} " 

265 f"(expires {expires_at})" 

266 ) 

267 return {"ticket": data.get("ticket"), "expires_at": expires_at} 

268 

269 

270# Module-level singleton 

271auth0_management = Auth0ManagementService()