Audit logging in Python

I'd like to make note of sensitive changes in my app.
I'd like to be able to query these changes from day one.
I'd like to leave architecting a long term solution for later.
I'd like these records to persist.
I'd like to create alarms around some changes that get sent to a business (not development) team.

💡 Idea!

Add a sub-logger to python's logging module.
Add a rule in my infrastructure that recognizes the sub-logger logs and directs them to a log group with different persistence rules.
Format the sub-logger logs to be JSON for easier querying and processing.

By making this a part of the logging infrastructure, the sub-logger is always handy and easily retrievable. AWS CloudWatch has support for queries, so we have a strong search ability out of the box. In theory, an update to the sub-logger could have the records be directed to a more appropriate storage option, such as DynamoDB.

import logging
from typing import cast

import ecs_logging


class AuditLogger(logging.Logger):
    """A logger class with extra functions and customization for use in tracking sensitive changes in the app."""

    def __init__(self, name, level=logging.INFO):
        super().__init__(name, level)
        self._setup()

    def _setup(self):
        handler = logging.StreamHandler()
        handler.setFormatter(ecs_logging.StdlibFormatter())
        self.addHandler(handler)

    def record_updated(self, msg: str, record_type: str, before: dict, after: dict, *args, **kwargs):
        """A general use audit function for when a model has been changed."""
        updated = {'record_type': record_type, 'before': {}, 'after': {}}

        # Find keys with different values
        for key in before.keys() & after.keys():
            if before[key] != after[key]:
                updated['before'][key] = before[key]
                updated['after'][key] = after[key]

        if 'extra' not in kwargs:
            kwargs['extra'] = {}
        kwargs['extra']['changes'] = updated

        self.info(msg, *args, **kwargs)


class MyLogger(logging.Logger):
    """Custom app logger with a special sub-logger for auditing."""

    def __init__(self, name, level=logging.NOTSET, add_starlette_context: bool = False):
        super().__init__(name, level)
        self.audit = AuditLogger(f"{name}.audit", logging.INFO)

    @classmethod
    def get(cls, name: str, add_starlette_context: bool = False):
        """Use in place of logging.getLogger() so typing systems recognize AuditLogger.audit"""
        return cast(cls, logging.getLogger(name))

import logging

from audit_logger import MyLogger

logging.setLoggerClass(MyLogger)

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = MyLogger.get(__name__)

if __name__ == '__main__':
    logger.info('This is an info log from the regular logger, nothing has changed about it\'s functionality')
    logger.audit.info(
        'Simple text audit logs can be made by using the regular level functions',
        stack_info=True,
        extra={"Extra info": "can be passed via extra"}
    )

    user_before = {'id': 1, 'phone': '+1234567890'}
    user_after = {'id': 1, 'phone': '+19876543210'}
    logger.audit.record_updated(
        "Edit made to MFA information",
        record_type='user',
        before=user_before,
        after=user_after
    )

2024-08-20 10:18:53,049 - __main__ - INFO - This is an info log from the regular logger, nothing has changed about it's functionality
{"@timestamp":"2024-08-20T14:18:53.049Z","log.level":"info","message":"Simple text audit logs can be made by using the regular level functions","Extra info":"can be passed via extra","ecs.version":"1.6.0","error":{"stack_trace":"Stack (most recent call last):\n  File \"C:\\Users\\avroh\\PycharmProjects\\audit_logger\\test.py\", line 16, in \n    logger.audit.info("},"log":{"logger":"__main__.audit","origin":{"file":{"line":16,"name":"test.py"},"function":""},"original":"Simple text audit logs can be made by using the regular level functions"},"process":{"name":"MainProcess","pid":18768,"thread":{"id":11020,"name":"MainThread"}}}
{"@timestamp":"2024-08-20T14:18:53.050Z","log.level":"info","message":"Edit made to MFA information","changes":{"after":{"phone":"+19876543210"},"before":{"phone":"+1234567890"},"record_type":"user"},"ecs.version":"1.6.0","log":{"logger":"__main__.audit","origin":{"file":{"line":47,"name":"__init__.py"},"function":"record_updated"},"original":"Edit made to MFA information"},"process":{"name":"MainProcess","pid":18768,"thread":{"id":11020,"name":"MainThread"}}}

Comments

Add your comment!