emit logs for authentication failures and successes

Author: markafarrell

ansible_base/authentication/authenticator_plugins/local.py

@@ -52,6 +52,15 @@ def authenticate(self, request, username=None, password=None, **kwargs):
         if new_username != username:
             return None
 
+        auth_log_headers = (
+            f"HTTP_USER_AGENT: {request.META['HTTP_USER_AGENT'] if 'HTTP_USER_AGENT' in request.META else 'UNKNOWN'} "
+            f"HTTP_X_FORWARDED_FOR: {request.META['HTTP_X_FORWARDED_FOR'] if 'HTTP_X_FORWARDED_FOR' in request.META else 'UNKNOWN'} "
+            f"REMOTE_ADDR: {request.META['REMOTE_ADDR'] if 'REMOTE_ADDR' in request.META else 'UNKNOWN'} "
+            f"REMOTE_HOST: {request.META['REMOTE_HOST'] if 'REMOTE_HOST' in request.META else 'UNKNOWN'}"
+        )
+
+        logger.warning(f"Login attempt for user: {username} {auth_log_headers}")
+
         user = super().authenticate(request, username, password, **kwargs)
 
         # This auth class doesn't create any new local users, but we still need to make sure
@@ -69,5 +78,8 @@ def authenticate(self, request, username=None, password=None, **kwargs):
                     "is_superuser": user.is_superuser,
                 },
             )
+            logger.warning(f"Successful login for user: {username} {auth_log_headers}")
+        else:
+            logger.warning(f"Failed login for user: {username} {auth_log_headers}")
 
         return update_user_claims(user, self.database_instance, [])

ansible_base/authentication/authenticator_plugins/oidc.py

@@ -221,6 +221,29 @@ class AuthenticatorPlugin(SocialAuthMixin, OpenIdConnectAuth, AbstractAuthentica
     def groups_claim(self):
         return self.setting('GROUPS_CLAIM')
 
+    def authenticate(self, *args, **kwargs):
+        request = args[0]
+
+        auth_log_headers = (
+            f"HTTP_USER_AGENT: {request.META['HTTP_USER_AGENT'] if 'HTTP_USER_AGENT' in request.META else 'UNKNOWN'} "
+            f"HTTP_X_FORWARDED_FOR: {request.META['HTTP_X_FORWARDED_FOR'] if 'HTTP_X_FORWARDED_FOR' in request.META else 'UNKNOWN'} "
+            f"REMOTE_ADDR: {request.META['REMOTE_ADDR'] if 'REMOTE_ADDR' in request.META else 'UNKNOWN'} "
+            f"REMOTE_HOST: {request.META['REMOTE_HOST'] if 'REMOTE_HOST' in request.META else 'UNKNOWN'}"
+        )
+
+        if "backend" in kwargs and kwargs["backend"].name == self.name:
+            logger.warning(f"Login attempt for {auth_log_headers}")
+
+        user = super().authenticate(*args, **kwargs)
+
+        if "backend" in kwargs and kwargs["backend"].name == self.name:
+            if user:
+                logger.warning(f"Successful login for {user} {auth_log_headers}")
+            else:
+                logger.warning(f"Failed login {auth_log_headers}")
+
+        return user
+
     def extra_data(self, user, backend, response, *args, **kwargs):
         for perm in ["is_superuser", get_setting('ANSIBLE_BASE_SOCIAL_AUDITOR_FLAG')]:
             if perm in response: