11"""
22Authentication-related CRUD operations.
3+
4+ Covers:
5+ - email+password validation
6+ - JWT access and refresh token creation and validation
7+ - various user validation helpers (e.g. is user account valid).
8+ - Altcha captcha generation and verification
9+
10+ NOTE: For Altcha captcha generation and verification, we are using the V1 API,
11+ consistent with the Django implementation https://github.com/aboutcode-org/django-altcha/
312"""
413
5- from datetime import datetime , timezone
14+ from datetime import datetime , timedelta , timezone
615
716import structlog
17+ from altcha import (
18+ ChallengeOptionsV1 , # pyright: ignore[reportPrivateImportUsage]
19+ create_challenge_v1 , # pyright: ignore[reportPrivateImportUsage]
20+ verify_solution_v1 , # pyright: ignore[reportPrivateImportUsage]
21+ )
822from fastapi import Response
923from sqlalchemy .ext .asyncio import AsyncSession
1024
25+ from divbase_api .api_config import api_settings
1126from divbase_api .crud .revoked_tokens import token_is_revoked
1227from divbase_api .crud .users import get_user_by_email , get_user_by_id , get_user_by_id_or_raise
1328from divbase_api .exceptions import AuthenticationError
1732
1833logger = structlog .get_logger (__name__ )
1934
35+ # Altcha (Captcha for e.g. registration + password reset) is disabled in test environments
36+ ALTCHA_ENABLED = api_settings .general .environment != "test"
37+
2038
2139async def authenticate_user (db : AsyncSession , email : str , password : str ) -> UserDB :
2240 """
@@ -38,8 +56,9 @@ async def authenticate_user(db: AsyncSession, email: str, password: str) -> User
3856 raise AuthenticationError (message = generic_error_msg )
3957
4058 if not user .email_verified :
59+ verify_url = f"{ api_settings .general .frontend_base_url } /resend-email-verification"
4160 raise AuthenticationError (
42- message = "Email address not verified, check your inbox or visit the DivBase website to resend a new verification email."
61+ message = f "Email address not verified, check your inbox or visit { verify_url } to get a new verification email."
4362 )
4463
4564 return user
@@ -130,3 +149,39 @@ def delete_auth_cookies(response: Response) -> Response:
130149 response .delete_cookie (TokenType .ACCESS .value )
131150 response .delete_cookie (TokenType .REFRESH .value )
132151 return response
152+
153+
154+ def generate_altcha_challenge () -> dict :
155+ """Generate a new Altcha captcha challenge."""
156+ expires = datetime .now (tz = timezone .utc ) + timedelta (minutes = 1 )
157+ options = ChallengeOptionsV1 (
158+ algorithm = "SHA-256" ,
159+ hmac_key = api_settings .general .altcha_hmac_secret .get_secret_value (),
160+ expires = expires ,
161+ )
162+ challenge = create_challenge_v1 (options )
163+ return challenge .to_dict ()
164+
165+
166+ def verify_altcha_solution (altcha : str | None , email : str ) -> bool :
167+ """
168+ Verify the Altcha captcha solution. Returns True for a valid solution.
169+ Validation skipped if in test environment (to avoid having to deal with Altcha in e2e playwright tests).
170+ """
171+ if not ALTCHA_ENABLED :
172+ logger .info ("Skipping Altcha verification as in test mode." )
173+ return True
174+
175+ if not altcha :
176+ logger .warning ("Altcha verification failed: no solution provided." )
177+ return False
178+
179+ verified , error = verify_solution_v1 (
180+ payload = altcha ,
181+ hmac_key = api_settings .general .altcha_hmac_secret .get_secret_value (),
182+ check_expires = True ,
183+ )
184+ if not verified :
185+ logger .warning (f"Altcha verification failed for email { email } with error message: { error } " )
186+ return False
187+ return True
0 commit comments