feat: alert logging

README.md

@@ -91,6 +91,7 @@ Create a .env on the root folder of the project with the following variables:
 13. SMTP_DOMAIN=smtp.xamfra.net
 14. SMTP_USER=[email protected]
 15. DISCORD_LOGIN_NOTIFICATIONS
+16. DISCORD_LOGIN_NOTIFICATIONS_ALERT
 
 ### FRONTEND
 

config.py

@@ -59,6 +59,7 @@ class Config(object):
     DEBUGGING_SEND_EMAILS = True
 
     DISCORD_LOGIN_NOTIFICATIONS = os.getenv("DISCORD_LOGIN_NOTIFICATIONS")
+    DISCORD_LOGIN_NOTIFICATIONS_ALERT = os.getenv("DISCORD_LOGIN_NOTIFICATIONS_ALERT")
 
 
 class DevelopmentConfig(Config):

server/blueprints/auth/config.py

@@ -4,6 +4,7 @@
 from flask import Blueprint, g, request, current_app, session, abort, jsonify
 from server.containers import Container
 from server.services.totp_service import TotpService
+from server.services.audit_service import audit_log_info, audit_log_warn
 import requests
 
 # initialise the blueprint
@@ -14,19 +15,6 @@
 auth_blueprint = Blueprint("auth", __name__)
 
 
-def audit_login(emoji: str, ip: str, username: str, result: str):
-    requests.post(
-        current_app.config["DISCORD_LOGIN_NOTIFICATIONS"],
-        json={"content": f"[{emoji}] [{username}] [{ip}] {result}"},
-    )
-
-
-def get_ip():
-    if request.environ.get("HTTP_X_FORWARDED_FOR") is None:
-        return request.environ["REMOTE_ADDR"]
-    else:
-        return request.environ["HTTP_X_FORWARDED_FOR"]  # if behind a proxy
-
 
 def basic_http_auth_required(f):
     def verify_password(username: str, password: str) -> bool:
@@ -54,18 +42,16 @@ def wrapper(*args, **kwargs):
         auth = request.authorization  # first factor
         totp = request.args.get("totp", None)  # second factor (2FA)
         if not (totp and auth):  # missing fields
-            audit_login("⚠️", get_ip(), "?", "Failed (missing totp or auth)")
+            audit_log_warn("Login failed (missing totp or auth)")
             abort(401)
         if not verify_password(auth.username, auth.password):  # wrong first factor
-            audit_login(
-                "⚠️", get_ip(), auth.username, "Failed (invalid password or username)"
-            )
+            audit_log_warn(f"Login failed (invalid password or username). Email: {auth.username}")
             abort(401)
         if not verify_totp(totp, g.user):  # wrong second factor
-            audit_login("⚠️", get_ip(), auth.username, "Failed (invalid totp)")
+            audit_log_warn(f"Login failed (invalid TOTP). Email: {auth.username}")
             abort(401)
 
-        audit_login("✅", get_ip(), auth.username, "Successful")
+        audit_log_info(f"Login successful. Email: {auth.username}")
         return f(*args, **kwargs)
 
     return wrapper

server/blueprints/emails/resources/bulk_email.py

@@ -13,6 +13,8 @@
 from server.email_notifications.bulk_email import send_bulk_email
 from server.models.student import EnrolmentStatus
 
+from server.services.audit_service import audit_log_info
+
 
 class BulkEmailCollectionRes(Resource):
     @login_required
@@ -77,6 +79,8 @@ def post(self):
                 elif email_preference == "all":
                     emails.extend(student.all_emails)
 
+        audit_log_info(f"Sending {len(emails)} emails")
+
         thread = Thread(
             target=send_bulk_email,
             args=(

server/blueprints/password_reset/resources/password_reset_redeem.py

@@ -6,13 +6,16 @@
 from flask_restful import Resource
 from werkzeug.exceptions import BadRequest, Unauthorized, InternalServerError
 
+
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     from server.blueprints.password_reset.services.password_reset_service import (
         PasswordResetService,
     )
     from server.services.recaptcha_service import RecaptchaService
+
+from server.services.audit_service import audit_log_alert
 from server.containers import Container
 from server.models import User
 
@@ -40,54 +43,66 @@ def post(
     ):
 
         if request.json is None:
+            audit_log_alert("Password redeem failed (no json body)")
             raise BadRequest("no json found")
 
         # validate recaptcha
         recaptcha = request.json.get("recaptcha", None)
         try:
             recaptcha_service.validate(recaptcha)
         except (BadRequest, InternalServerError):
+            audit_log_alert("Password redeem failed (bad recaptcha)")
             raise
 
         try:
             password = request.json["password"]
         except KeyError:
+            audit_log_alert("Password redeem failed (no password)")
             raise BadRequest("no password found in body")
 
         try:
             token = request.json["token"]
         except KeyError:
+            audit_log_alert("Password redeem failed (no token)")
             raise Unauthorized("no token found in body")
 
         if not User.is_strong_enough_password(password):
+            audit_log_alert("Password redeem failed (weak password)")
             raise BadRequestNotStrongPassword()
 
         data = jwt.decode(token, options=DONT_VERIFY)
 
         if "email" not in data:
+            audit_log_alert("Password redeem failed (no email)")
             raise BadRequest("token has no email field in the body")
 
         email = data["email"]
         if email is None:
+            audit_log_alert("Password redeem failed (token has value None for email)")
             raise BadRequest("token has value 'None' for field email")
 
         try:
             user = password_reset_service.get_user_from_email(email)
         except BadRequest:
+            audit_log_alert(f"Password redeem failed (no user with the given email). Email: {email}")
             raise
         except KeyError:
+            audit_log_alert(f"Password redeem failed (no user with the given email). Email: {email}")
             raise BadRequest(f"no user found for email {email!r}")
 
         try:
             password_reset_service.try_decode_token(user.password_hash, token)
         except Unauthorized:
+            audit_log_alert(f"Password redeem failed (invalid token). Email: {email}")
             raise
 
         try:
             password_reset_service.update_user_password(user, password)
         except BadRequest:
+            audit_log_alert(f"Password redeem failed (could not update password). Email: {email}")
             raise
 
         password_reset_service.trigger_event_user_password_reset_redeem(user)
-
+        audit_log_alert(f"Password redeem successful. Email: {email}")
+
         return "password change successful", 200  # todo return json instead

server/blueprints/password_reset/resources/password_reset_request.py

@@ -8,6 +8,7 @@
 )
 from server.containers import Container
 from server.services.recaptcha_service import RecaptchaService
+from server.services.audit_service import audit_log_alert
 
 
 SUCCESS_MESSAGE = "if there is a user for this email, it will receive a password reset token via email"
@@ -22,26 +23,30 @@ def post(
         ],
     ):
         if request.json is None:
+            audit_log_alert("Password request failed (no body)")
             raise BadRequest("no json found")
 
         # validate recaptcha
         recaptcha = request.json.get("recaptcha", None)
         try:
             recaptcha_service.validate(recaptcha)
         except (BadRequest, InternalServerError):
+            audit_log_alert("Password request failed (invalid captcha)")
             raise
 
         email = request.json.get("email", None)
         try:
             user = password_reset_service.get_user_from_email(email)
         except BadRequest:
+            audit_log_alert(f"Password request failed (no email found). Email: {email}")
             raise
         except KeyError:
+            audit_log_alert(f"Password request failed (no user with the given email). Email: {email}")
             # no user found for this email, but don't reveal the result (privacy & security concerns)
             return SUCCESS_MESSAGE
 
         token = password_reset_service.generate_token(user)
 
         password_reset_service.trigger_event_user_password_reset_request(user, token)
-
+        audit_log_alert(f"Password request successful. Email: {email}")
         return SUCCESS_MESSAGE

server/blueprints/pre_enrolment/resources/pre_enrolment.py

@@ -7,6 +7,7 @@
 )
 from server.services.recaptcha_service import RecaptchaService
 from server.containers import Container
+from server.services.audit_service import audit_log_info, audit_log_warn
 
 
 class PreEnrollment(Resource):
@@ -19,24 +20,28 @@ def post(
         ],
     ):
         if request.json is None:
+            audit_log_warn(f"Pre-enrolment failed (no body)")
             raise BadRequest("no json found")
 
         # validate recaptcha passes
         recaptcha = request.json.get("recaptcha", None)
         try:
             recaptcha_service.validate(recaptcha)
         except (BadRequest, InternalServerError):
+            audit_log_warn(f"Pre-enrolment failed (invalid recaptcha)")
             raise
 
         # if recaptcha is valid, then try to parse a new student
         body = request.json.get("body", None)
         try:
             student = pre_enrolment_service.parse_student(body)
         except BadRequest:
+            audit_log_warn(f"Pre-enrolment failed (could not parse student)")
             raise
 
         # save the new student to the db
         pre_enrolment_service.save_student(student)
+        audit_log_info(f"Pre-enrolment successful")
 
         # signal new student added
         pre_enrolment_service.trigger_event_student_pre_enrolled(student)

server/blueprints/user_invites/resources/user_invite_redeem.py

@@ -18,6 +18,9 @@
     from server.blueprints.user_invites.services.user_invites_service import (
         UserInvitesService,
     )
+
+from server.services.audit_service import audit_log_alert
+
 
 
 class UserInviteRedeem(BaseModel):
@@ -40,25 +43,31 @@ def post(
         try:
             body = UserInviteRedeem(**request.json)
         except ValidationError as e:
+            audit_log_alert(f"User invite redeem failed (invalid body)")
             return e.json(), 400
 
         try:
             data = jwt_service.decode(body.token)
         except jwt.ExpiredSignatureError:
+            audit_log_alert(f"User invite redeem failed (token expired)")
             # this is a specific case of the following clause
             return "token expired", 401
         except jwt.InvalidTokenError:
+            audit_log_alert(f"User invite redeem failed (invalid token)")
             return "invalid token", 401
 
         try:
             data = UserInviteJwtBody(**data)
         except ValidationError as e:
+            audit_log_alert(f"User invite redeem failed (invalid token body)")
             return f"Invalid token body. {e.json()}", 400
 
         if not user_invites_service.is_user_email_available(data.user_email):
+            audit_log_alert(f"User invite redeem failed (user already exists)")
             return "user already exists", 400
 
         if not user_invites_service.is_role_id_valid(data.role_id):
+            audit_log_alert(f"User invite redeem failed (invalid role id)")
             return f"invalid role with role_id={data.role_id!r}", 400
 
         user_id = User.generate_new_id()
@@ -77,6 +86,8 @@ def post(
 
         user_invites_service.save_user(user)
 
+        audit_log_alert(f"User invite redeem successful")
+
         return {
             "user_id": user_id,
             "totp": totp_service.generate_url(totp_secret, user.email),

server/blueprints/user_invites/resources/user_invite_request.py

@@ -23,6 +23,9 @@
     from server.services.jwt_service import JwtService
     from ..services.user_invites_service import UserInvitesService
 
+from server.services.audit_service import audit_log_alert
+
+
 USER_INVITE_TOKEN_EXPIRES_IN = datetime.timedelta(days=7)
 
 
@@ -39,20 +42,29 @@ def post(
             Container.user_invites_service
         ],
     ):
-        Require.ensure.create(UserInvitePermission())
+        try:
+            Require.ensure.create(UserInvitePermission())
+        except:
+            audit_log_alert(f"User invite request failed (no authZ)")
+            raise
 
         try:
             body = UserInviteRequest(**request.json)
         except ValidationError as e:
+            audit_log_alert(f"User invite request failed (invalid body)")
             return e.json(), 400
 
         if not user_invites_service.is_role_id_valid(body.role_id):
+            audit_log_alert(f"User invite request failed (invalid role id). For email: {body.user_email}")
             return f"Invalid role_id supplied.", 400
 
         if not user_invites_service.is_user_email_available(body.user_email):
+            audit_log_alert(f"User invite request failed (user already exists). For email: {body.user_email}")
             return "There already exists a user with the provided email.", 400
 
         token = user_invites_service.generate_token(UserInviteJwtBody(**body.dict()))
         user_invites_service.send_invite(body.user_email, token)
+
+        audit_log_alert(f"User invite request successful. For email: {body.user_email}")
 
         return "", 200

server/services/audit_service.py

@@ -0,0 +1,44 @@
+from flask import current_app, request
+from flask_login import current_user
+
+
+def get_ip():
+    try:
+        if request.environ.get("HTTP_X_FORWARDED_FOR") is None:
+            return request.environ["REMOTE_ADDR"]
+        else:
+            return request.environ["HTTP_X_FORWARDED_FOR"]  # if behind a proxy
+    except:
+        return "?"
+
+
+def try_get_email():
+    try:
+        return current_user.email
+    except:
+        return "?"
+
+
+def audit_log(level: str, message: str):
+    try:
+        emoji = "⚠️" if level == "warn" else "🚨" if level == "alert" else "✅"
+        dest = current_app.config["DISCORD_LOGIN_NOTIFICATIONS_ALERT"] if level == "alert" else current_app.config["DISCORD_LOGIN_NOTIFICATIONS"]
+        requests.post(
+            dest, 
+            json={
+            "content": f"{emoji} [email: {try_get_email()}] [ip: {get_ip()}] {message}"
+        })
+    except:
+        print(f"Failed to post audit log")
+
+
+def audit_log_info(message: str):
+    return audit_log("info", message)
+
+
+def audit_log_warn(message: str):
+    return audit_log("warn", message)
+
+
+def audit_log_alert(message: str):
+    return audit_log("alert", message)