feat: send suspension emails and enhanced audit logs for MFA non-compliance

This commit is contained in:
2026-05-29 05:27:49 +00:00
parent 13767d3fa1
commit f869f6c06d
5 changed files with 289 additions and 24 deletions
@@ -28,6 +28,7 @@ from gatehouse_app.services.email_provider import EmailMessage, EmailProviderFac
from gatehouse_app.services.email_templates import (
build_mfa_deadline_reminder_html,
build_mfa_suspension_html,
build_mfa_suspension_admin_html,
)
from gatehouse_app.utils.constants import AuditAction
@@ -209,6 +210,100 @@ class NotificationService:
)
return False
@staticmethod
def send_mfa_suspended_admin_notification(
admin_user: User,
suspended_user: User,
compliance: MfaPolicyCompliance,
org_policy: OrganizationSecurityPolicy,
) -> bool:
"""Notify org admin that a user has been suspended for MFA non-compliance.
Sends an email to organization admins/owners when a member of their
organization has been automatically suspended for failing to meet MFA
compliance requirements.
Args:
admin_user: Admin/owner to notify
suspended_user: The user who was suspended
compliance: Suspended user's compliance record
org_policy: Organization's MFA policy
Returns:
True if notification was sent successfully, False otherwise
"""
try:
org_name = compliance.organization_id
from gatehouse_app.models.organization.organization import Organization
org = Organization.query.get(compliance.organization_id)
if org:
org_name = org.name
from gatehouse_app.utils.constants import MfaPolicyMode
mfa_methods = "Multi-factor authentication"
mode = org_policy.mfa_policy_mode
if mode == MfaPolicyMode.REQUIRE_TOTP:
mfa_methods = "Authenticator app (TOTP)"
elif mode == MfaPolicyMode.REQUIRE_WEBAUTHN:
mfa_methods = "Passkey (WebAuthn)"
elif mode == MfaPolicyMode.REQUIRE_TOTP_OR_WEBAUTHN:
mfa_methods = "Authenticator app (TOTP) OR Passkey (WebAuthn)"
app_url = current_app.config.get("APP_URL", "http://localhost:8080")
members_link = f"{app_url}/organizations/{compliance.organization_id}/members"
deadline_str = compliance.deadline_at.strftime('%Y-%m-%d %H:%M UTC') if compliance.deadline_at else ''
days_overdue = 0
if compliance.deadline_at:
deadline = compliance.deadline_at
if deadline.tzinfo is None:
deadline = deadline.replace(tzinfo=timezone.utc)
from datetime import timezone as dt_tz
now = datetime.now(timezone.utc)
days_overdue = max(0, (now - deadline).days)
subject = f"User Suspended - MFA Non-Compliance in {org_name}"
html_body = build_mfa_suspension_admin_html(
admin_name=admin_user.full_name or admin_user.email,
org_name=org_name,
suspended_user_name=suspended_user.full_name or suspended_user.email,
suspended_user_email=suspended_user.email,
mfa_methods=mfa_methods,
members_link=members_link,
deadline_date=deadline_str,
days_overdue=days_overdue,
)
NotificationService._send_email_async(
to_address=admin_user.email,
subject=subject,
body=f"A user ({suspended_user.email}) in {org_name} has been suspended for MFA non-compliance. Manage members: {members_link}",
html_body=html_body,
)
logger.info(
f"Sent MFA suspension admin notification to {admin_user.email} "
f"regarding suspended user {suspended_user.email}"
)
AuditService.log_action(
action=AuditAction.MFA_SUSPENSION_ADMIN_NOTIFICATION_SENT,
user_id=suspended_user.id,
organization_id=compliance.organization_id,
description=f"Admin {admin_user.email} notified about MFA suspension of user {suspended_user.email}",
metadata={
"admin_user_id": admin_user.id,
"admin_email": admin_user.email,
"suspended_user_email": suspended_user.email,
"org_name": org_name,
},
)
return True
except Exception as e:
logger.exception(
f"Error sending MFA suspension admin notification to {admin_user.email}: {e}"
)
return False
@staticmethod
def _build_deadline_reminder_body(
user: User,