feat: Added security logs for email update, forgotten password mail, token emission/refresh/revokation, new consent, consent revokation #177

CHANGES.rst

@@ -1,6 +1,10 @@
 [0.0.56] - Unreleased
 ---------------------
 
+Added
+^^^^^
+- New security events logs :issue:`177`
+
 Changed
 ^^^^^^^
 - Update to HTMX 2.0.3 :pr:`184`

canaille/app/__init__.py

@@ -66,3 +66,7 @@ class classproperty:
 
     def __get__(self, obj, owner):
         return self.f(owner)
+
+
+def generate_security_log(message):
+    return "[SECURITY] " + message

canaille/core/endpoints/account.py

@@ -23,6 +23,7 @@ from werkzeug.datastructures import FileStorage
 from canaille.app import b64_to_obj
 from canaille.app import build_hash
 from canaille.app import default_fields
+from canaille.app import generate_security_log
 from canaille.app import models
 from canaille.app import obj_to_b64
 from canaille.app.flask import current_user
@@ -575,6 +576,17 @@ def profile_edition_remove_email(user, edited_user, email):
     return True
 
 
+def has_email_changed(old_emails):
+    emailstr = "emails-"
+    i = 0
+    new_emails = set()
+    while request.form.get(emailstr + str(i)):
+        new_emails.add(request.form.get(emailstr + str(i)))
+        i += 1
+
+    return set(old_emails) != new_emails
+
+
 @bp.route("/profile/<user:edited_user>", methods=("GET", "POST"))
 @user_needed()
 def profile_edition(user, edited_user):
@@ -583,6 +595,9 @@ def profile_edition(user, edited_user):
     ):
         abort(404)
 
+    is_email_modified = has_email_changed(user.emails)
+
+    request_ip = request.remote_addr or "unknown IP"
     menuitem = "profile" if edited_user.id == user.id else "users"
     emails_readonly = (
         current_app.features.has_email_confirmation and not user.can_manage_users
@@ -611,7 +626,16 @@ def profile_edition(user, edited_user):
             return render_template("profile_edit.html", **render_context)
 
         profile_edition_main_form_validation(user, edited_user, profile_form)
+
+        if is_email_modified:
+            current_app.logger.info(
+                generate_security_log(
+                    f"Updated email for {edited_user.user_name} from {request_ip}"
+                )
+            )
+
         flash(_("Profile updated successfully."), "success")
+
         return redirect(
             url_for("core.account.profile_edition", edited_user=edited_user)
         )
@@ -783,7 +807,9 @@ def profile_settings_edit(editor, edited_user):
             ):
                 Backend.instance.set_user_password(edited_user, form["password1"].data)
                 current_app.logger.info(
-                f'Changed password in settings for {edited_user.user_name} from {request_ip}'
+                    generate_security_log(
+                        f"Changed password in settings for {edited_user.user_name} from {request_ip}"
+                    )
                 )
 
             Backend.instance.save(edited_user)

canaille/core/endpoints/auth.py

@@ -8,6 +8,7 @@ from flask import session
 from flask import url_for
 
 from canaille.app import build_hash
+from canaille.app import generate_security_log
 from canaille.app.flask import current_user
 from canaille.app.flask import login_user
 from canaille.app.flask import logout_user
@@ -96,7 +97,9 @@ def password():
     if not success:
         logout_user()
         current_app.logger.info(
-            f'Failed login attempt for {session["attempt_login"]} from {request_ip}'
+            generate_security_log(
+                f'Failed login attempt for {session["attempt_login"]} from {request_ip}'
+            )
         )
         flash(message or _("Login failed, please check your information"), "error")
         return render_template(
@@ -104,7 +107,9 @@ def password():
         )
 
     current_app.logger.info(
-        f'Succeed login attempt for {session["attempt_login"]} from {request_ip}'
+        generate_security_log(
+            f'Succeed login attempt for {session["attempt_login"]} from {request_ip}'
+        )
     )
     del session["attempt_login"]
     login_user(user)
@@ -121,7 +126,9 @@ def logout():
 
     if user:
         request_ip = request.remote_addr or "unknown IP"
-        current_app.logger.info(f"Logout {user.identifier} from {request_ip}")
+        current_app.logger.info(
+            generate_security_log(f"Logout {user.identifier} from {request_ip}")
+        )
 
         flash(
             _(
@@ -197,8 +204,16 @@ def forgotten():
         )
         return render_template("forgotten-password.html", form=form)
 
-    statuses = [send_password_reset_mail(user, email) for email in user.emails]
-    success = all(statuses)
+    request_ip = request.remote_addr or "unknown IP"
+    success = True
+    for email in user.emails:
+        if not send_password_reset_mail(user, email):
+            success = False
+        current_app.logger.info(
+            generate_security_log(
+                f"Sending a reset password mail to {email} for {user.user_name} from {request_ip}"
+            )
+        )
 
     if success:
         flash(success_message, "success")

canaille/oidc/endpoints/consents.py

@@ -2,10 +2,13 @@ import datetime
 import uuid
 
 from flask import Blueprint
+from flask import current_app
 from flask import flash
 from flask import redirect
+from flask import request
 from flask import url_for
 
+from canaille.app import generate_security_log
 from canaille.app import models
 from canaille.app.flask import user_needed
 from canaille.app.i18n import gettext as _
@@ -77,6 +80,12 @@ def revoke(user, consent):
 
     else:
         consent.revoke()
+        request_ip = request.remote_addr or "unknown IP"
+        current_app.logger.info(
+            generate_security_log(
+                f"Consent revoked for {user.user_name} in client {consent.client.client_name} from {request_ip}"
+            )
+        )
         flash(_("The access has been revoked"), "success")
 
     return redirect(url_for("oidc.consents.consents"))

canaille/oidc/endpoints/oauth.py

@@ -17,6 +17,7 @@ from flask import url_for
 from werkzeug.datastructures import CombinedMultiDict
 
 from canaille import csrf
+from canaille.app import generate_security_log
 from canaille.app import models
 from canaille.app.flask import current_user
 from canaille.app.flask import logout_user
@@ -178,6 +179,12 @@ def authorize_consent(client, user):
                 issue_date=datetime.datetime.now(datetime.timezone.utc),
             )
         Backend.instance.save(consent)
+        request_ip = request.remote_addr or "unknown IP"
+        current_app.logger.info(
+            generate_security_log(
+                f"New consent for {user.user_name} in client {consent.client.client_name} from {request_ip}"
+            )
+        )
 
     response = authorization.create_authorization_response(grant_user=grant_user)
     current_app.logger.debug("authorization endpoint response: %s", response.location)
@@ -187,11 +194,22 @@ def authorize_consent(client, user):
 @bp.route("/token", methods=["POST"])
 @csrf.exempt
 def issue_token():
-    current_app.logger.debug(
-        "token endpoint request: POST: %s", request.form.to_dict(flat=False)
-    )
+    request_params = request.form.to_dict(flat=False)
+    print(request_params["grant_type"])
+    grant_type = request_params["grant_type"][0]
+    current_app.logger.debug("token endpoint request: POST: %s", request_params)
     response = authorization.create_token_response()
     current_app.logger.debug("token endpoint response: %s", response.json)
+
+    if response.json.get("access_token"):
+        access_token = response.json["access_token"]
+        token = Backend.instance.get(models.Token, access_token=access_token)
+        request_ip = request.remote_addr or "unknown IP"
+        current_app.logger.info(
+            generate_security_log(
+                f"Issued {grant_type} token for {token.subject.user_name} in client {token.client.client_name} from {request_ip}"
+            )
+        )
     return response
 
 

canaille/oidc/endpoints/tokens.py

@@ -2,9 +2,11 @@ import datetime
 
 from flask import Blueprint
 from flask import abort
+from flask import current_app
 from flask import flash
 from flask import request
 
+from canaille.app import generate_security_log
 from canaille.app import models
 from canaille.app.flask import permissions_needed
 from canaille.app.flask import render_htmx_template
@@ -42,6 +44,12 @@ def view(user, token):
         elif request.form.get("action") == "revoke":
             token.revokation_date = datetime.datetime.now(datetime.timezone.utc)
             Backend.instance.save(token)
+            request_ip = request.remote_addr or "unknown IP"
+            current_app.logger.info(
+                generate_security_log(
+                    f"Revoked token for {token.subject.user_name} in client {token.client.client_name} by {user.user_name} from {request_ip}"
+                )
+            )
             flash(_("The token has successfully been revoked."), "success")
 
         else:

doc/features.rst

@@ -263,6 +263,18 @@ Logging
 
 Canaille writes :attr:`logs <canaille.core.configuration.CoreSettings.LOGGING>` for every important event happening, to help administrators understand what is going on and debug funky situations.
 
+The following security events are logged with the tag [SECURITY] for easy retrieval :
+
+- Authentication attempt
+- Password update
+- Email update
+- Forgotten password mail sent to user
+- Token emission
+- Token refresh
+- Token revokation
+- New consent given for client application
+- Consent revokation
+
 .. _feature_development:
 
 A tool for your development and tests

tests/core/test_auth.py

@@ -1,6 +1,8 @@
 import datetime
 import logging
 
+from canaille.app import generate_security_log
+
 
 def test_signin_and_out(testclient, user, caplog):
     with testclient.session_transaction() as session:
@@ -25,7 +27,7 @@ def test_signin_and_out(testclient, user, caplog):
     assert (
         "canaille",
         logging.INFO,
-        "Succeed login attempt for user from unknown IP",
+        generate_security_log("Succeed login attempt for user from unknown IP"),
     ) in caplog.record_tuples
     res = res.follow(status=302)
     res = res.follow(status=200)
@@ -44,7 +46,7 @@ def test_signin_and_out(testclient, user, caplog):
     assert (
         "canaille",
         logging.INFO,
-        "Logout user from unknown IP",
+        generate_security_log("Logout user from unknown IP"),
     ) in caplog.record_tuples
     res = res.follow(status=302)
     res = res.follow(status=200)
@@ -81,7 +83,7 @@ def test_signin_wrong_password(testclient, user, caplog):
     assert (
         "canaille",
         logging.INFO,
-        "Failed login attempt for user from unknown IP",
+        generate_security_log("Failed login attempt for user from unknown IP"),
     ) in caplog.record_tuples
 
 

tests/core/test_forgotten_password.py

@@ -1,5 +1,8 @@
+import logging
 from unittest import mock
 
+from canaille.app import generate_security_log
+
 
 def test_password_forgotten_disabled(smtpd, testclient, user):
     testclient.app.config["CANAILLE"]["ENABLE_PASSWORD_RECOVERY"] = False
@@ -11,7 +14,7 @@ def test_password_forgotten_disabled(smtpd, testclient, user):
     res.mustcontain(no="Forgotten password")
 
 
-def test_password_forgotten(smtpd, testclient, user):
+def test_password_forgotten(smtpd, testclient, user, caplog):
     res = testclient.get("/reset", status=200)
 
     res.form["login"] = "user"
@@ -21,12 +24,19 @@ def test_password_forgotten(smtpd, testclient, user):
         "A password reset link has been sent at your email address. You should receive "
         "it within a few minutes.",
     ) in res.flashes
+    assert (
+        "canaille",
+        logging.INFO,
+        generate_security_log(
+            "Sending a reset password mail to john@doe.com for user from unknown IP"
+        ),
+    ) in caplog.record_tuples
     res.mustcontain("Send again")
 
     assert len(smtpd.messages) == 1
 
 
-def test_password_forgotten_multiple_mails(smtpd, testclient, user, backend):
+def test_password_forgotten_multiple_mails(smtpd, testclient, user, backend, caplog):
     user.emails = ["foo@bar.com", "foo@baz.com", "foo@foo.com"]
     backend.save(user)
 
@@ -39,6 +49,14 @@ def test_password_forgotten_multiple_mails(smtpd, testclient, user, backend):
         "A password reset link has been sent at your email address. You should receive "
         "it within a few minutes.",
     ) in res.flashes
+    for email in user.emails:
+        assert (
+            "canaille",
+            logging.INFO,
+            generate_security_log(
+                f"Sending a reset password mail to {email} for user from unknown IP"
+            ),
+        ) in caplog.record_tuples
     res.mustcontain("Send again")
 
     assert len(smtpd.messages) == 3

tests/core/test_profile_edition.py

@@ -1,7 +1,10 @@
+import logging
+
 import pytest
 from flask import g
 from webtest import Upload
 
+from canaille.app import generate_security_log
 from canaille.core.populate import fake_users
 
 
@@ -101,13 +104,7 @@ def test_edition_permission(
     testclient.get("/profile/user", status=200)
 
 
-def test_edition(
-    testclient,
-    logged_user,
-    admin,
-    jpeg_photo,
-    backend,
-):
+def test_edition(testclient, logged_user, admin, jpeg_photo, backend, caplog):
     res = testclient.get("/profile/user", status=200)
     form = res.forms["baseform"]
     form["given_name"] = "given_name"
@@ -131,6 +128,11 @@ def test_edition(
     assert res.flashes == [
         ("success", "Le profil a été mis à jour avec succès.")
     ], res.text
+    assert (
+        "canaille",
+        logging.INFO,
+        generate_security_log("Updated email for user from unknown IP"),
+    ) in caplog.record_tuples
     res = res.follow()
 
     backend.reload(logged_user)

tests/core/test_profile_settings.py

@@ -4,6 +4,7 @@ from unittest import mock
 
 from flask import g
 
+from canaille.app import generate_security_log
 from canaille.app import models
 
 
@@ -141,7 +142,7 @@ def test_password_change(testclient, logged_user, backend, caplog):
     assert (
         "canaille",
         logging.INFO,
-        "Changed password in settings for user from unknown IP",
+        generate_security_log("Changed password in settings for user from unknown IP"),
     ) in caplog.record_tuples
 
     res = res.follow()

tests/oidc/test_authorization_code_flow.py

@@ -1,4 +1,5 @@
 import datetime
+import logging
 from urllib.parse import parse_qs
 from urllib.parse import urlsplit
 
@@ -8,13 +9,14 @@ from authlib.oauth2.rfc7636 import create_s256_code_challenge
 from flask import g
 from werkzeug.security import gen_salt
 
+from canaille.app import generate_security_log
 from canaille.app import models
 
 from . import client_credentials
 
 
 def test_nominal_case(
-    testclient, logged_user, client, keypair, trusted_client, backend
+    testclient, logged_user, client, keypair, trusted_client, backend, caplog
 ):
     assert not backend.query(models.Consent)
 
@@ -28,8 +30,14 @@ def test_nominal_case(
         ),
         status=200,
     )
-
     res = res.form.submit(name="answer", value="accept", status=302)
+    assert (
+        "canaille",
+        logging.INFO,
+        generate_security_log(
+            "New consent for user in client Some client from unknown IP"
+        ),
+    ) in caplog.record_tuples
 
     assert res.location.startswith(client.redirect_uris[0])
     params = parse_qs(urlsplit(res.location).query)
@@ -89,7 +97,13 @@ def test_nominal_case(
     assert claims["sub"] == logged_user.user_name
     assert claims["name"] == logged_user.formatted_name
     assert claims["aud"] == [client.client_id, trusted_client.client_id]
-
+    assert (
+        "canaille",
+        logging.INFO,
+        generate_security_log(
+            "Issued authorization_code token for user in client Some client from unknown IP"
+        ),
+    ) in caplog.record_tuples
     res = testclient.get(
         "/oauth/userinfo",
         headers={"Authorization": f"Bearer {access_token}"},

tests/oidc/test_consent.py

@@ -1,6 +1,8 @@
+import logging
 from urllib.parse import parse_qs
 from urllib.parse import urlsplit
 
+from canaille.app import generate_security_log
 from canaille.app import models
 
 from . import client_credentials
@@ -10,7 +12,7 @@ def test_no_logged_no_access(testclient):
     testclient.get("/consent", status=403)
 
 
-def test_revokation(testclient, client, consent, logged_user, token, backend):
+def test_revokation(testclient, client, consent, logged_user, token, backend, caplog):
     res = testclient.get("/consent", status=200)
     res.mustcontain(client.client_name)
     res.mustcontain("Revoke access")
@@ -20,6 +22,13 @@ def test_revokation(testclient, client, consent, logged_user, token, backend):
 
     res = testclient.get(f"/consent/revoke/{consent.consent_id}", status=302)
     assert ("success", "The access has been revoked") in res.flashes
+    assert (
+        "canaille",
+        logging.INFO,
+        generate_security_log(
+            f"Consent revoked for {logged_user.user_name} in client {client.client_name} from unknown IP"
+        ),
+    ) in caplog.record_tuples
     res = res.follow(status=200)
     res.mustcontain(no="Revoke access")
     res.mustcontain("Restore access")

tests/oidc/test_refresh_token.py

@@ -1,13 +1,15 @@
 import datetime
+import logging
 from urllib.parse import parse_qs
 from urllib.parse import urlsplit
 
+from canaille.app import generate_security_log
 from canaille.app import models
 
 from . import client_credentials
 
 
-def test_refresh_token(testclient, logged_user, client, backend):
+def test_refresh_token(testclient, logged_user, client, backend, caplog):
     assert not backend.query(models.Consent)
 
     res = testclient.get(
@@ -59,7 +61,13 @@ def test_refresh_token(testclient, logged_user, client, backend):
     new_token = backend.get(models.Token, access_token=access_token)
     assert new_token is not None
     assert old_token.access_token != new_token.access_token
-
+    assert (
+        "canaille",
+        logging.INFO,
+        generate_security_log(
+            "Issued refresh_token token for user in client Some client from unknown IP"
+        ),
+    ) in caplog.record_tuples
     backend.reload(old_token)
     assert old_token.revokation_date
 

tests/oidc/test_token_admin.py

@@ -1,7 +1,9 @@
 import datetime
+import logging
 
 from werkzeug.security import gen_salt
 
+from canaille.app import generate_security_log
 from canaille.app import models
 
 
@@ -134,13 +136,20 @@ def test_revoke_bad_request(testclient, token, logged_admin):
     res = res.form.submit(name="action", value="invalid", status=400)
 
 
-def test_revoke_token(testclient, token, logged_admin, backend):
+def test_revoke_token(testclient, token, logged_admin, backend, caplog):
     assert not token.revoked
 
     res = testclient.get(f"/admin/token/{token.token_id}")
     res = res.form.submit(name="action", value="confirm-revoke")
     res = res.form.submit(name="action", value="revoke")
     assert ("success", "The token has successfully been revoked.") in res.flashes
+    assert (
+        "canaille",
+        logging.INFO,
+        generate_security_log(
+            "Revoked token for user in client Some client by admin from unknown IP"
+        ),
+    ) in caplog.record_tuples
 
     backend.reload(token)
     assert token.revoked