feat: send alerts to users without Google TFA

trojstenid/settings.py

@@ -11,6 +11,7 @@
 """
 
 from base64 import b64decode
+from datetime import timedelta
 from pathlib import Path
 
 from environs import Env
@@ -249,6 +250,15 @@
 
 GOOGLE_ADMIN_SERVICE_ACCOUNT = env("GOOGLE_ADMIN_SERVICE_ACCOUNT", default="")
 GOOGLE_ADMIN_SUBJECT = env("GOOGLE_ADMIN_SUBJECT", default="")
+GOOGLE_TFA_MIN_ACCOUNT_AGE = env.timedelta(
+    "GOOGLE_TFA_MIN_ACCOUNT_AGE", default=timedelta(days=5)
+)
+GOOGLE_TFA_MAX_ACCOUNT_AGE = env.timedelta(
+    "GOOGLE_TFA_MAX_ACCOUNT_AGE", default=timedelta(days=14)
+)
+GOOGLE_TFA_ENROLLMENT_PERIOD = env.timedelta(
+    "GOOGLE_TFA_ENROLLMENT_PERIOD", default=timedelta(days=14)
+)
 
 GITHUB_APP_ID = env.int("GITHUB_APP_ID", default=-1)
 GITHUB_APP_PRIVATE_KEY = b64decode(env("GITHUB_APP_PRIVATE_KEY", default="")).decode()

trojstenid/users/google_groups.py → trojstenid/users/google_api.py

@@ -1,15 +1,21 @@
 import json
 import logging
+from datetime import datetime
 
 from allauth.account.models import EmailAddress
+from dateutil.parser import isoparse
 from django.conf import settings
 from django.contrib.auth.models import Group
+from django.utils.timezone import now
 from google.oauth2 import service_account
 from googleapiclient.discovery import build
 
 logger = logging.getLogger(__name__)
 
-SCOPES = ["https://www.googleapis.com/auth/admin.directory.group.readonly"]
+SCOPES = [
+    "https://www.googleapis.com/auth/admin.directory.group.readonly",
+    "https://www.googleapis.com/auth/admin.directory.user.readonly",
+]
 IAM_DOMAIN = "iam.trojsten.sk"
 
 
@@ -103,3 +109,51 @@ def sync_iam_groups() -> None:
     """
     for group in fetch_iam_google_groups():
         sync_group(group)
+
+
+def query_nontfa_users() -> list[tuple[str, datetime, list[str]]]:
+    """
+    Returns all email addresses of users whose account is new and has not enabled 2FA.
+    """
+    min_account_age = getattr(settings, "GOOGLE_TFA_MIN_ACCOUNT_AGE")
+    max_account_age = getattr(settings, "GOOGLE_TFA_MAX_ACCOUNT_AGE")
+    enrollment_period = getattr(settings, "GOOGLE_TFA_ENROLLMENT_PERIOD")
+
+    credentials = _get_credentials()
+    if credentials is None:
+        logger.warning("Google Admin service account not configured")
+        return []
+
+    users = []
+
+    directory = build("admin", "directory_v1", credentials=credentials)
+    request = directory.users().list(
+        customer="my_customer",
+        orderBy="email",
+        query="isEnrolledIn2Sv=false isEnforcedIn2Sv=true",
+    )
+    while request:
+        response = request.execute()
+
+        for user in response.get("users", []):
+            email = user.get("primaryEmail", "")
+            creation_time = (
+                isoparse(user["creationTime"]) if "creationTime" in user else None
+            )
+            if creation_time:
+                account_age = now() - creation_time
+                if min_account_age <= account_age <= max_account_age:
+                    users.append(
+                        (
+                            email,
+                            creation_time + enrollment_period,
+                            [
+                                e.get("address")
+                                for e in user.get("emails", [])
+                                if "address" in e and not e.get("primary", False)
+                            ],
+                        )
+                    )
+
+        request = directory.users().list_next(request, response)
+    return users

trojstenid/users/management/commands/sendtfaalerts.py

@@ -0,0 +1,88 @@
+from django.conf import settings
+from django.core.mail import EmailMultiAlternatives, get_connection
+from django.core.management import BaseCommand
+from django.template.loader import render_to_string
+
+from trojstenid.users.google_api import query_nontfa_users
+
+
+class Command(BaseCommand):
+    help = "Send mail alerts to Google users without TFA enabled"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Run without actually sending emails",
+        )
+
+    def handle(self, *args, **options):
+        dry_run = options["dry_run"]
+        nontfa_users = query_nontfa_users()
+
+        if not nontfa_users:
+            self.stdout.write(self.style.NOTICE("No users found without 2FA enabled."))
+            return
+
+        self.stdout.write(
+            self.style.MIGRATE_HEADING(
+                f"Found {len(nontfa_users)} user(s) requiring 2FA alert.\n"
+            )
+        )
+
+        if dry_run:
+            for email, required_after, _ in nontfa_users:
+                self.stdout.write(
+                    self.style.NOTICE(
+                        f"[DRY-RUN] Would alert {email}, 2FA required after: {required_after}"
+                    )
+                )
+            self.stdout.write(self.style.SUCCESS("\nDry run complete."))
+            return
+
+        sent_count = 0
+        failed_count = 0
+
+        connection = get_connection()
+        try:
+            connection.open()
+
+            for email, required_after, secondary_email in nontfa_users:
+                try:
+                    content = render_to_string(
+                        "account/email/tfa_alert.txt",
+                        {"email": email, "required_after": required_after},
+                    )
+
+                    msg = EmailMultiAlternatives(
+                        subject="Dvojstupňové overenie tvojho Trojsten Google účtu",
+                        body=content,
+                        from_email=settings.DEFAULT_FROM_EMAIL,
+                        to=[email],
+                        cc=secondary_email if secondary_email else None,
+                        connection=connection,
+                    )
+                    msg.send()
+
+                    sent_count += 1
+                    self.stdout.write(
+                        self.style.SUCCESS(
+                            f"Sent to '{email}', 2FA required after: {required_after}"
+                        )
+                    )
+                except Exception as e:
+                    failed_count += 1
+                    self.stderr.write(
+                        self.style.ERROR(f"Failed to send to '{email}', Error: {e}")
+                    )
+
+        except Exception as e:
+            self.stderr.write(self.style.ERROR(f"Error during email delivery: {e}"))
+        finally:
+            connection.close()
+
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"\nCompleted. Sent: {sent_count}, failed: {failed_count}"
+            )
+        )

trojstenid/users/management/commands/syncgooglegroups.py

@@ -2,7 +2,7 @@
 
 from django.core.management.base import BaseCommand
 
-from trojstenid.users.google_groups import sync_iam_groups
+from trojstenid.users.google_api import sync_iam_groups
 
 
 class Command(BaseCommand):

trojstenid/users/tasks.py

@@ -5,7 +5,7 @@
 from django_rq import job
 
 from trojstenid.users.github import sync_github_teams
-from trojstenid.users.google_groups import sync_iam_groups
+from trojstenid.users.google_api import sync_iam_groups
 from trojstenid.users.models import Application, User
 from trojstenid.users.serializers import UserSerializer
 

trojstenid/users/templates/account/email/tfa_alert.txt

@@ -0,0 +1,13 @@
+{% extends "account/email/base_message.txt" %}
+
+{% block content %}Pred pár dňami ti bol vytvorený nový Trojstenový Google účet ({{ email }}).
+
+Tento email ti prichádza preto, že si si v ňom ešte nenastavil dvojstupňové overenie (2FA).
+Do dvoch týždňov od vytvorenie účtu si musíš nastaviť dvojstupňové overenie (pre teba najneskôr {{ required_after|date:"d.m.Y" }} do {{ required_after|time:"H:i" }}), inak stratíš prístup k účtu.
+
+Návod na to, ako to urobiť, nájdeš tu:
+
+https://support.google.com/accounts/answer/185839?hl=sk
+
+Ak to dovtedy nestihneš a nebudeš sa vedieť prihlásiť do účtu, napíš nám na roots@trojsten.sk a vyriešime to.
+{% endblock content %}