Improve ebauth log parsing, and parse stepup-authentication logs also

roles/rsyslog/tasks/main.yml

@@ -1,9 +1,10 @@
-- name: Install rsyslog
+- name: Install rsyslog and python modules
   ansible.builtin.package:
     name:
       - rsyslog
       - rsyslog-gnutls
       - rsyslog-relp
+      - python3-dateutil
     state: present
   notify:
     - "restart rsyslog"

roles/rsyslog/tasks/process_auth_logs.yml

@@ -39,7 +39,7 @@
     state: present
   when: ansible_os_family == "Debian"
 
-- name: Create a python script that parses log_logins per environment
+- name: Create a python script that parses eb log_logins per environment
   ansible.builtin.template:
     src: parse_ebauth_to_mysql.py.j2
     dest: /usr/local/sbin/parse_ebauth_to_mysql_{{ item.name }}.py
@@ -49,7 +49,17 @@
   with_items: "{{ rsyslog_environments }}"
   when: item.db_loglogins_name is defined
 
-- name: Put log_logins logrotate scripts
+- name: Create a python script that parses stepup log_logins per environment
+  ansible.builtin.template:
+    src: parse_stepupauth_to_mysql.py.j2
+    dest: /usr/local/sbin/parse_stepupauth_to_mysql_{{ item.name }}.py
+    mode: 0740
+    owner: root
+    group: root
+  with_items: "{{ rsyslog_environments }}"
+  when: item.db_loglogins_name is defined
+
+- name: Put log_logins logrotate scripts for eb
   ansible.builtin.template:
     src: logrotate_ebauth.j2
     dest: /etc/logrotate.d/logrotate_ebauth_{{ item.name }}
@@ -59,6 +69,16 @@
   with_items: "{{ rsyslog_environments }}"
   when: item.db_loglogins_name is defined
 
+- name: Put log_logins logrotate scripts for stepup
+  ansible.builtin.template:
+    src: logrotate_stepupauth.j2
+    dest: /etc/logrotate.d/logrotate_stepupauth_{{ item.name }}
+    mode: 0644
+    owner: root
+    group: root
+  with_items: "{{ rsyslog_environments }}"
+  when: item.db_loglogins_name is defined
+
 - name: Create logdirectory for log_logins cleanup script
   ansible.builtin.file:
     path: "{{ rsyslog_dir }}/apps/{{ item.name }}/loglogins_cleanup/"

roles/rsyslog/templates/logrotate_stepupauth.j2

@@ -0,0 +1,16 @@
+{{ rsyslog_dir }}/log_logins/{{ item.name }}/stepup-authentication.log
+{
+    missingok
+    daily
+    rotate 180
+    sharedscripts
+    dateext
+    dateyesterday
+    compress
+    delaycompress
+    create 0640 root {{ rsyslog_read_group }}
+    postrotate
+      /usr/local/sbin/parse_stepupauth_to_mysql_{{ item.name }}.py > /dev/null
+      systemctl kill -s HUP rsyslog.service
+    endscript
+}

roles/rsyslog/templates/parse_ebauth_to_mysql.py.j2

@@ -21,11 +21,17 @@ cursor = db.cursor()
 
 def update_lastseen(user_id, date):
     query = """
-    REPLACE INTO last_login (userid, lastseen)
+    INSERT INTO last_login (userid, lastseen)
     VALUES (%s, %s)
+    ON DUPLICATE KEY UPDATE
+    lastseen = GREATEST(lastseen, VALUES(lastseen))
     """
-    cursor.execute(query, (user_id, date))
-    db.commit()
+    try:
+        cursor.execute(query, (user_id, date))
+        db.commit()
+    except Exception as e:
+        db.rollback()
+        print(f"Error updating last_login for user {user_id}: {e}")
 
 def load_in_mysql(a,b,c,d,e,f,g,h):
     sql = """insert into log_logins(idpentityid,spentityid,loginstamp,userid,keyid,sessionid,requestid,trustedproxyentityid) values(%s,%s,%s,%s,%s,%s,%s,%s)"""
@@ -73,4 +79,3 @@ for filename in os.listdir(workdir):
 
 cursor.close()
 db.close()
-

roles/rsyslog/templates/parse_stepupauth_to_mysql.py.j2

@@ -0,0 +1,152 @@
+#!/usr/bin/python3
+# This script parses rotated stepup-authentication.log files produced by engineblock.
+# It filters for successful logins (authentication_result:OK) and inserts the data
+# into the log_logins and last_login MySQL tables.
+# This script is intended to be run separately during logrotate.
+
+import os
+import sys
+import json
+import MySQLdb
+from dateutil.parser import parse
+
+# Configuration variables (to be injected by Ansible/Jinja2)
+mysql_host="{{ item.db_loglogins_host }}"
+mysql_user="{{ item.db_loglogins_user }}"
+mysql_password="{{ item.db_loglogins_password }}"
+mysql_db="{{ item.db_loglogins_name }}"
+workdir="{{ rsyslog_dir }}/log_logins/{{ item.name}}/"
+
+# Establish database connection
+try:
+    db = MySQLdb.connect(mysql_host,mysql_user,mysql_password,mysql_db )
+    cursor = db.cursor()
+except Exception as e:
+    print(f"Error connecting to MySQL: {e}")
+    sys.exit(1)
+
+# --- Database Functions ---
+
+def update_lastseen(user_id, date):
+    """
+    Updates the last_login table.
+    Uses GREATEST() to ensure only newer dates overwrite the existing 'lastseen' value.
+    """
+    query = """
+    INSERT INTO last_login (userid, lastseen)
+    VALUES (%s, %s)
+    ON DUPLICATE KEY UPDATE
+    lastseen = GREATEST(lastseen, VALUES(lastseen))
+    """
+    try:
+        cursor.execute(query, (user_id, date))
+        db.commit()
+    except Exception as e:
+        db.rollback()
+        print(f"Error updating last_login for user {user_id}: {e}")
+
+def load_stepup_in_mysql(idp, sp, loginstamp, userid, requestid):
+    """
+    Inserts Step-up login data into the log_logins table.
+    Fills keyid, sessionid, and trustedproxyentityid with NULL.
+    """
+    # Columns in log_logins: idpentityid, spentityid, loginstamp, userid, keyid, sessionid, requestid, trustedproxyentityid
+
+    keyid = None
+    sessionid = None
+    trustedproxyentityid = None
+
+    sql = """
+    INSERT INTO log_logins(idpentityid, spentityid, loginstamp, userid, keyid, sessionid, requestid, trustedproxyentityid)
+    VALUES(%s, %s, %s, %s, %s, %s, %s, %s)
+    """
+    try:
+        cursor.execute(sql, (idp, sp, loginstamp, userid, keyid, sessionid, requestid, trustedproxyentityid))
+        db.commit()
+    except Exception as e:
+        db.rollback()
+        print(f"Error inserting stepup data: {e}")
+        # Print the data that failed insertion
+        print((idp, sp, loginstamp, userid, keyid, sessionid, requestid, trustedproxyentityid))
+
+# --- Parsing Function ---
+
+def parse_stepup_lines(a):
+    """
+    Opens the stepup log file, parses each line, filters for successful logins,
+    and loads the data into MySQL.
+    """
+    input_file = open((a), 'r')
+    for line in input_file:
+        try:
+            # Assumes JSON data starts after the first ']:'
+            jsonline = line.split(']:',2)[1]
+            data = json.loads(jsonline)
+        except:
+            continue
+
+        # 1. Filtering condition: Only parse logs having authentication_result:OK
+        # Only successful authentications are logged, so this check is not
+        # necessary. There is currently a bug in the Stepup-Gateway where
+        # FAILED is logged, even though the result is OK, making this check
+        # do the wrong thing now.
+        #
+        #if data.get("context").("authentication_result") != "OK":
+        #    continue
+
+        # 2. Extract required fields
+        context = data.get("context")
+
+        if not isinstance(context, dict):
+            print("Skipping line: context is missing or invalid")
+            continue
+
+        user_id = context.get("identity_id")
+        timestamp = context.get("datetime")
+        request_id = context.get("request_id")
+        sp_entity_id = context.get("requesting_sp")
+        idp_entity_id = context.get("authenticating_idp")
+
+        # Basic data validation
+        if not user_id or not timestamp:
+            print(
+                    "Skipping line: validation failed "
+                    f"(user_id={user_id!r}, timestamp={timestamp!r}, request_id={request_id!r})"
+            )
+            continue
+
+        try:
+            # 3. Format date and time for MySQL
+            loginstamp = parse(timestamp).strftime("%Y-%m-%d %H:%M:%S")
+            last_login_date = parse(timestamp).strftime("%Y-%m-%d")
+        except:
+            print(
+                    "Skipping line: timestamp parsing failed "
+                    f"(timestamp={timestamp!r}, user_id={user_id!r}, error={e})"
+            )
+            continue
+
+        # 4. Insert into MySQL
+        load_stepup_in_mysql(idp_entity_id, sp_entity_id, loginstamp, user_id, request_id)
+
+        # 5. Update last login date
+        update_lastseen(user_id, last_login_date)
+
+
+# --- Main Execution ---
+
+## Loop over the files and parse them one by one
+for filename in os.listdir(workdir):
+    filetoparse=(os.path.join(workdir, filename))
+
+    # Check for Stepup files, ignore compressed files
+    if os.path.isfile(filetoparse) and filename.startswith("stepup-authentication.log-") and not filename.endswith(".gz"):
+        print(f"Parsing stepup log file: {filename}")
+        parse_stepup_lines(filetoparse)
+    else:
+        continue
+
+# Close database connection
+cursor.close()
+db.close()
+print("Stepup log parsing complete.")