feat: User IAM audit events (M2-10744) (#2074)

🔗 [Jira Ticket M2-10744](https://mindlogger.atlassian.net/browse/M2-10744)

Changes include:

- `user:create` events
- `user:delete` events
- `user:password:change` events
- `user:password:recovery:initiate` events
- `user:password:recovery:approve` events
- `user:mfa:enable` events
- `user:mfa:disable` (on verify) events
- `user:mfa:disable` (on confirm) events
- `user:mfa:recovery:view` events
- `user:mfa:recovery:download` events
- `user:mfa:recovery:use` (on user login) events
- `user:mfa:recovery:use` (on MFA disable) events
- `user:mfa:recovery:use` (on MFA view) events

Author: divbzero

src/apps/authentication/api/auth.py

@@ -399,7 +399,16 @@ async def verify_mfa_recovery_code(
                     request_info=request_metadata,
                 )
 
-            except RecoveryCodeNotFoundError:
+            except RecoveryCodeNotFoundError as e:
+                await log(
+                    AuditEvent(
+                        event_action=EventAction.USER_MFA_RECOVERY_USE,
+                        user_id=user_id,
+                        user_target_id=user_id,
+                        **http_audit_fields(request, e),
+                    )
+                )
+
                 # No unused codes exist - increment both counters
                 session_count = await mfa_service.increment_failed_totp_attempts(mfa_session_id)
                 global_count = await mfa_service.increment_global_failed_attempts(user_id)
@@ -456,7 +465,16 @@ async def verify_mfa_recovery_code(
 
                 raise
 
-            except RecoveryCodeInvalidError:
+            except RecoveryCodeInvalidError as e:
+                await log(
+                    AuditEvent(
+                        event_action=EventAction.USER_MFA_RECOVERY_USE,
+                        user_id=user_id,
+                        user_target_id=user_id,
+                        **http_audit_fields(request, e),
+                    )
+                )
+
                 # Invalid code - increment both per-session and global counters
                 session_count = await mfa_service.increment_failed_totp_attempts(mfa_session_id)
                 global_count = await mfa_service.increment_global_failed_attempts(user_id)
@@ -569,9 +587,14 @@ async def verify_mfa_recovery_code(
         raise
 
     # Step 7: Return response
-    token = Token(access_token=access_token, refresh_token=refresh_token)
-    public_user = PublicUser.from_user(user)
-
+    await log(
+        AuditEvent(
+            event_action=EventAction.USER_MFA_RECOVERY_USE,
+            user_id=user.id,
+            user_target_id=user.id,
+            **http_audit_fields(request),
+        )
+    )
     await log(
         AuditEvent(
             event_action=EventAction.USER_SESSION_LOGIN,
@@ -580,6 +603,9 @@ async def verify_mfa_recovery_code(
         )
     )
 
+    token = Token(access_token=access_token, refresh_token=refresh_token)
+    public_user = PublicUser.from_user(user)
+
     return Response(
         result=UserLogin(
             token=token,

src/apps/authentication/services/recovery_codes.py

@@ -30,6 +30,7 @@
 __all__ = [
     "generate_random_code",
     "format_recovery_code",
+    "is_recovery_code",
     "hash_recovery_code",
     "verify_recovery_code",
     "encrypt_recovery_code",
@@ -42,20 +43,23 @@
 ]
 
 
+# Alphanumeric recovery codes (uppercase letters and digits)
+ALPHABET = string.ascii_uppercase + string.digits
+
+# Get length from config (10 alphanumeric characters)
+LENGTH = settings.mfa.recovery_code_length
+MID = LENGTH // 2
+
+
 def generate_random_code() -> str:
     """
     Generate a random recovery code.
 
     Returns:
         str: Random alphanumeric code in format XXXXX-XXXXX (e.g., "A3F7K-9B2Q5")
     """
-    # Get length from config (10 alphanumeric characters)
-    length = settings.mfa.recovery_code_length
-
-    # Generate random alphanumeric string (uppercase letters and digits)
     # Using secrets for cryptographically strong random generation
-    alphabet = string.ascii_uppercase + string.digits
-    random_chars = "".join(secrets.choice(alphabet) for _ in range(length))
+    random_chars = "".join(secrets.choice(ALPHABET) for _ in range(LENGTH))
 
     # Format as XXXXX-XXXXX
     return format_recovery_code(random_chars)
@@ -71,12 +75,25 @@ def format_recovery_code(code: str) -> str:
     Returns:
         str: Formatted code with hyphen (e.g., "A3F7K-9B2Q5")
     """
-    if len(code) != settings.mfa.recovery_code_length:
-        raise ValueError(f"Code must be {settings.mfa.recovery_code_length} characters")
+    if len(code) != LENGTH:
+        raise ValueError(f"Code must be {LENGTH} characters")
 
     # Split in middle and add hyphen
-    mid = len(code) // 2
-    return f"{code[:mid]}-{code[mid:]}"
+    return f"{code[:MID]}-{code[MID:]}"
+
+
+def is_recovery_code(code: str) -> bool:
+    """
+    Check if string has the shape of a recovery code (XXXXX-XXXXX).
+
+    Args:
+        code: Recovery code to check (e.g., "A3F7K-9B2Q5")
+
+    Returns:
+        bool: True if code has the right shape
+    """
+    left, _, right = code.partition("-")
+    return len(left) == len(right) == MID and not {*left, *right}.difference(ALPHABET)
 
 
 def hash_recovery_code(code: str) -> str:

src/apps/authentication/tests/test_recovery_code_verification.py

@@ -107,8 +107,8 @@ async def test_valid_recovery_code_returns_tokens(
         mfa_token = login_response.json()["result"]["mfaToken"]
 
         # Step 2: Verify with valid recovery code
-        valid_code = codes[0]
         audit_log = mocker.patch("apps.authentication.api.auth.log")
+        valid_code = codes[0]
         response = await client.post(
             url=self.verify_recovery_url,
             data=dict(
@@ -118,8 +118,15 @@ async def test_valid_recovery_code_returns_tokens(
             ),
         )
 
-        audit_log.assert_awaited_once()
-        event = audit_log.call_args[0][0]
+        assert audit_log.await_count == 2
+
+        event = audit_log.call_args_list[0].args[0]
+        assert event.user_id == user.id
+        assert event.user_target_id == user.id
+        assert event.event_action == EventAction.USER_MFA_RECOVERY_USE
+        assert event.event_outcome == EventOutcome.SUCCESS
+
+        event = audit_log.call_args_list[1].args[0]
         assert event.user_id == user.id
         assert event.event_action == EventAction.USER_SESSION_LOGIN
         assert event.event_outcome == EventOutcome.SUCCESS
@@ -168,8 +175,15 @@ async def test_invalid_recovery_code_returns_error(
             ),
         )
 
-        audit_log.assert_awaited_once()
-        event = audit_log.call_args[0][0]
+        assert audit_log.await_count == 2
+
+        event = audit_log.call_args_list[0].args[0]
+        assert event.user_id == user.id
+        assert event.user_target_id == user.id
+        assert event.event_action == EventAction.USER_MFA_RECOVERY_USE
+        assert event.event_outcome == EventOutcome.FAILURE
+
+        event = audit_log.call_args_list[1].args[0]
         assert event.user_id == user.id
         assert event.event_action == EventAction.USER_SESSION_LOGIN
         assert event.event_outcome == EventOutcome.FAILURE
@@ -378,7 +392,9 @@ async def test_replay_protection_logs_security_warning(
         warning_messages = [record.message for record in caplog.records if record.levelname == "WARNING"]
         assert any("Replay attack detected" in msg for msg in warning_messages)
 
-    async def test_no_recovery_codes_returns_not_found(self, client: TestClient, user_with_mfa_no_codes: User):
+    async def test_no_recovery_codes_returns_not_found(
+        self, client: TestClient, user_with_mfa_no_codes: User, mocker: MockerFixture
+    ):
         """Test that verification fails with 404 when user has no recovery codes."""
         user = user_with_mfa_no_codes
 
@@ -395,6 +411,7 @@ async def test_no_recovery_codes_returns_not_found(self, client: TestClient, use
         mfa_token = login_response.json()["result"]["mfaToken"]
 
         # Try to verify with any code
+        audit_log = mocker.patch("apps.authentication.api.auth.log")
         response = await client.post(
             url=self.verify_recovery_url,
             data=dict(
@@ -404,6 +421,19 @@ async def test_no_recovery_codes_returns_not_found(self, client: TestClient, use
             ),
         )
 
+        assert audit_log.await_count == 2
+
+        event = audit_log.call_args_list[0].args[0]
+        assert event.user_id == user.id
+        assert event.user_target_id == user.id
+        assert event.event_action == EventAction.USER_MFA_RECOVERY_USE
+        assert event.event_outcome == EventOutcome.FAILURE
+
+        event = audit_log.call_args_list[1].args[0]
+        assert event.user_id == user.id
+        assert event.event_action == EventAction.USER_SESSION_LOGIN
+        assert event.event_outcome == EventOutcome.FAILURE
+
         # Assert: Not found error (404)
         assert response.status_code == http.HTTPStatus.NOT_FOUND
 

src/apps/users/api/password.py

@@ -4,10 +4,12 @@
 from fastapi import Body, Depends, Query, Request
 from starlette import status
 
+from apps.audit import AuditEvent, EventAction, http_audit_fields, log
 from apps.authentication.deps import get_current_user
 from apps.authentication.services import AuthenticationService
 from apps.job.service import JobService
 from apps.shared.domain.response import Response
+from apps.shared.exception import BaseError
 from apps.shared.response import EmptyResponse
 from apps.users.cruds.user import UsersCRUD
 from apps.users.domain import (
@@ -30,28 +32,49 @@
 
 
 async def password_update(
+    request: Request,
     user: User = Depends(get_current_user),
     schema: ChangePasswordRequest = Body(...),
     session=Depends(get_session),
 ) -> Response[PublicUser]:
     """General endpoint for update password for signin."""
-    reencryption_in_progress = await JobService(session, user.id).is_job_in_progress("reencrypt_answers")
-    if reencryption_in_progress:
-        raise ReencryptionInProgressError()
-
-    async with atomic(session):
-        AuthenticationService.verify_password(
-            schema.prev_password,
-            user.hashed_password,
+    try:
+        reencryption_in_progress = await JobService(session, user.id).is_job_in_progress("reencrypt_answers")
+        if reencryption_in_progress:
+            raise ReencryptionInProgressError()
+
+        async with atomic(session):
+            AuthenticationService.verify_password(
+                schema.prev_password,
+                user.hashed_password,
+            )
+
+            password_hash: str = AuthenticationService.get_password_hash(schema.password)
+            password = UserChangePassword(hashed_password=password_hash)
+
+            updated_user: User = await UsersCRUD(session).change_password(user, password)
+
+            # Create public representation of the internal user
+            public_user = PublicUser.from_user(updated_user)
+    except BaseError as e:
+        await log(
+            AuditEvent(
+                user_id=user.id,
+                user_target_id=user.id,
+                event_action=EventAction.USER_PASSWORD_CHANGE,
+                **http_audit_fields(request, e),
+            )
         )
-
-        password_hash: str = AuthenticationService.get_password_hash(schema.password)
-        password = UserChangePassword(hashed_password=password_hash)
-
-        updated_user: User = await UsersCRUD(session).change_password(user, password)
-
-        # Create public representation of the internal user
-        public_user = PublicUser.from_user(updated_user)
+        raise
+
+    await log(
+        AuditEvent(
+            user_id=user.id,
+            user_target_id=user.id,
+            event_action=EventAction.USER_PASSWORD_CHANGE,
+            **http_audit_fields(request),
+        )
+    )
 
     email = user.email_encrypted
     retries = settings.task_answer_encryption.max_retries
@@ -69,27 +92,80 @@ async def password_recovery(
     """General endpoint for sending password recovery email
     and stored info in Redis.
     """
-    # Send the password recovery the internal password recovery service
-    async with atomic(session):
-        try:
-            content_source = await get_mindlogger_content_source(request)
-            await PasswordRecoveryService(session).send_password_recovery(schema, content_source, language)
-        except UserNotFound:
-            pass  # mute error in terms of user enumeration vulnerability
+    user_id: uuid.UUID | None = None
+    try:
+        async with atomic(session):
+            try:
+                content_source = await get_mindlogger_content_source(request)
+                public_user = await PasswordRecoveryService(session).send_password_recovery(
+                    schema, content_source, language
+                )
+                user_id = public_user.id
+            except UserNotFound as e:
+                # mute error in terms of user enumeration vulnerability
+                await log(
+                    AuditEvent(
+                        user_id=None,
+                        user_email=schema.email,
+                        event_action=EventAction.USER_PASSWORD_RECOVERY_INITIATE,
+                        **http_audit_fields(request, e),
+                    )
+                )
+                return EmptyResponse(status_code=status.HTTP_201_CREATED)
+    except BaseError as e:
+        await log(
+            AuditEvent(
+                user_id=None,
+                user_email=schema.email,
+                event_action=EventAction.USER_PASSWORD_RECOVERY_INITIATE,
+                **http_audit_fields(request, e),
+            )
+        )
+        raise
+
+    await log(
+        AuditEvent(
+            user_id=user_id,
+            user_target_id=user_id,
+            event_action=EventAction.USER_PASSWORD_RECOVERY_INITIATE,
+            **http_audit_fields(request),
+        )
+    )
 
     return EmptyResponse(status_code=status.HTTP_201_CREATED)
 
 
 async def password_recovery_approve(
+    request: Request,
     schema: PasswordRecoveryApproveRequest = Body(...),
     session=Depends(get_session),
 ) -> Response[PublicUser]:
     """General endpoint to approve the password recovery."""
 
     # Approve the password recovery
     # NOTE: also check if the data exists and tokens are not expired
-    async with atomic(session):
-        public_user: PublicUser = await PasswordRecoveryService(session).approve(schema)
+    try:
+        async with atomic(session):
+            public_user: PublicUser = await PasswordRecoveryService(session).approve(schema)
+    except BaseError as e:
+        await log(
+            AuditEvent(
+                user_id=None,
+                user_email=schema.email,
+                event_action=EventAction.USER_PASSWORD_RECOVERY_APPROVE,
+                **http_audit_fields(request, e),
+            )
+        )
+        raise
+
+    await log(
+        AuditEvent(
+            user_id=public_user.id,
+            user_target_id=public_user.id,
+            event_action=EventAction.USER_PASSWORD_RECOVERY_APPROVE,
+            **http_audit_fields(request),
+        )
+    )
 
     return Response[PublicUser](result=public_user)