Audit Action Message Bus Handler Guide (Beta)

Overview

The Audit Action Message Bus Handler is a specialized extension type in the Q2 SDK that allows you to process audit events asynchronously through the message bus system. This handler automatically receives audit records when specific audit actions occur in the Q2 platform and allows you to implement custom business logic in response to these events.

Key Features

  • Asynchronous Processing: Handle audit events without blocking the main application flow

  • Selective Subscription: Subscribe only to specific audit actions you care about

  • Structured Data: Receive rich, structured audit record data

  • Type Safety: Built-in dataclass support for audit records

  • Automatic Filtering: Framework automatically filters subscribed vs unsubscribed actions

Architecture

The audit action message bus handler follows this flow:

  1. Q2 Platform generates an audit event (e.g., user login, transaction)

  2. Message Bus receives the audit record and routes it to subscribed handlers

  3. Your Handler receives the structured audit record and processes it

  4. Custom Logic executes your business-specific code

Core Components

AuditRecord Dataclass

The AuditRecord dataclass represents a structured audit event with the following fields:

Required Fields:

*Note: All fields except audit_action(Audit action name) are defaulted to some default value

customer_key: str        # Customer identifier
audit_uuid: str         # Unique audit event ID
audit_action: str       # Type of action (LOGIN, LOGOUT, TRANSFER, etc.)
session_id: str         # User session identifier
version: str            # Application version
git_rev: str           # Git revision
protobuf_type: str     # Record type
envstack: str          # Environment stack
app_source: str        # Source application
cryptokey: str         # Encryption key

Optional Fields:

audit_id: Optional[int]               # Numeric audit ID
global_id: Optional[str]              # Global identifier
start_timestamp: Optional[datetime]   # Event start time
timestamp: Optional[datetime]         # Event timestamp
end_timestamp: Optional[datetime]     # Event end time
action_id: Optional[int]              # Action identifier
audit_description: Optional[str]      # Human-readable description
workstation: Optional[str]            # Workstation identifier
legacy_details: Optional[LegacyDetails]      # Legacy audit details
execution_context: Optional[ExecutionContext] # Execution context
is_reservation: Optional[bool]        # Reservation flag
hmac: Optional[bytes]                 # HMAC signature
zone_id: Optional[int]                # Zone identifier
entity_changes: list[EntityChange]    # List of entity changes
end_of_frame: bool                    # End of frame flag

Creating an Audit Action Message Bus Extension

Step 1: Generate the Extension

Use the Q2 CLI to create a new audit action message bus extension:

q2 create_extension

New Extension Name: MyAuditActionMsgBus
What type of extension are you creating?

    1) Online (default)
    2) SSO (Third Party Integration)
    3) Ardent (API)
    4) Q2Console (Backoffice)
    5) Central (Legacy Backoffice)
    6) Adapter
    7) Audit Action <---------
    8) Custom Health Check
    9) Message Bus
    10) Caliper API Custom Endpoint
    11) Q2Config Extensions (Client Side Rendered)
    12) Base Extension

Please make a selection and press Return [1]: 7
Audit Action
Select Audit Action type to generate

    1) Audit Action (Message Bus style (default))  <---------
    2) Audit Action (Legacy Db Registration)

Please make a selection and press Return: 1
Audit Action (Message Bus style (default))

This has registered as an endpoint. If you do NOT want this endpoint exposed, remove it from the INSTALLED_EXTENSIONS in /Users/hfaraaz/q2dev/repos/extension_testing/configuration/settings.py

Provide audit actions you would like to handle (comma-separated if multiple). Use tab completion to verify options: "LOGIN", "LOGOUT", "TRANSFER"

writing rendered jinja template to MyAuditActionMsgBus/extension.py
writing rendered jinja template to MyAuditActionMsgBus/urls.py

Step 2: Implement Your Handler

The generated extension will create a handler class that looks like this:

from q2_sdk.core.http_handlers.audit_action_msg_bus_handler import (
    AuditActionMsgBusHandler,
    AuditRecord
)

class MyAuditProcessorHandler(AuditActionMsgBusHandler):
    # Specify which audit actions this handler subscribes to
    SUBSCRIBED_AUDIT_ACTIONS = ["LOGIN", "LOGOUT", "TRANSFER"]
    async def handle_audit_record(self, audit_record: AuditRecord) -> AuditRecord:
        """
        This is the hook where you can put your handling logic.

        message.text contains the message body posted from the bus.
        """
        self.logger.info(
            f"received subscribed audit action :- {audit_record.audit_action} on {self.extension_name}"
        )

Handler Base Class

The AuditActionMsgBusHandler base class provides:

Core Attributes

MESSAGE_TYPE = ["q2msg.shared.audit_record_pb2.AuditRecord"]
SUBSCRIBED_AUDIT_ACTIONS = []  # Selected Audit actions at the time of creation will be populated here

Core Methods

handle(message: Message) -> Any

Framework entry point. Converts message to AuditRecord and routes to your handler.

handle_audit_record(audit_record: AuditRecord) -> Any

Override this method - Your custom business logic goes here.

Filtering Behavior

The handler automatically filters audit actions:

  • Subscribed Actions: Actions in SUBSCRIBED_AUDIT_ACTIONS are processed

  • Unsubscribed Actions: Other actions are logged as warnings and ignored

# This handler only processes LOGIN and LOGOUT events
SUBSCRIBED_AUDIT_ACTIONS = ["LOGIN", "LOGOUT"]

Troubleshooting

Common Issues

Handler not receiving messages: - Verify SUBSCRIBED_AUDIT_ACTIONS contains the correct action names - Check message bus connectivity - Review deployment logs

Data parsing errors: - Verify audit record field access - Check for None values in optional fields - Review audit record structure changes

Testing

We are providing a cli utility for testing locally which can be invoked by

q2 send_msg_to_bus audit_record  --audit-action "sdk_OAuthRefreshToken" # To send an AuditRecord message

** Note :- Most of the fields are dummy defaulted except the audit_action

q2 send_msg_to_bus string_message "hello world" "Lowercase" # To send an normal string message

To test you new audit_action_msg_bus extension locally, start your extension by running q2 run command and in another terminal window run the above command to send a test message to your extension.

API Reference

AuditActionMsgBusHandler Extension