Authentication Events Reference

This document provides reference schemas, payloads, and code examples for the authentication domain events emitted by Framework M on the EventBusProtocol.

---

1. Event Schemas & Payloads

Whenever a user logs in, logs out, or has their identifiers modified, the system publishes domain events to the event bus.

UserLoggedIn Event

• Topic: user.logged_in
• Emitted When: A user successfully authenticates via password, OTP, magic-link, or federated OIDC.
• Payload Schema (Python):

from dataclasses import dataclass
from datetime import datetime

@dataclass(frozen=True)
class UserLoggedIn:
    user_id: str          # UUID of the authenticated user
    email: str | None     # User's email (if available)
    ip_address: str       # Client IP address
    user_agent: str       # Client User-Agent header string
    auth_method: str      # password, otp, magic_link, oidc
    tenant_id: str | None # Active tenant ID (if multi-tenant)
    timestamp: datetime   # Time of event execution

UserLoggedOut Event

• Topic: user.logged_out
• Emitted When: A user manually calls /auth/logout or when a session is invalidated via OIDC frontchannel logout.
• Payload Schema (Python):

@dataclass(frozen=True)
class UserLoggedOut:
    user_id: str          # UUID of the logged out user
    session_id: str       # Invalidated session UUID
    ip_address: str       # Client IP address
    tenant_id: str | None # Active tenant ID (if multi-tenant)
    timestamp: datetime   # Time of event execution

---

2. Registering Event Listeners

To listen to these events for audit logging, security monitoring, or session tracking, write a subscriber class and register it with the event bus.

Example Listener Implementation

from framework_m_core.di import inject
from framework_m_core.events import UserLoggedIn

class AuthAuditSubscriber:
    """Subscriber that logs logins to the security audit database."""
    
    async def handle_login(self, event: UserLoggedIn) -> None:
        # Custom logging logic
        print(
            f"User {event.user_id} logged in from IP {event.ip_address} "
            f"using {event.auth_method} at {event.timestamp}"
        )

Registering the Subscriber during App Startup

Bind your listener to the EventBus during initialization:

from framework_m_core.interfaces.bootstrap import BootstrapProtocol

class RegisterAuthListeners(BootstrapProtocol):
    order = 50 # runs after event bus is ready
    
    @inject
    def run(self, container: Container) -> None:
        event_bus = container.event_bus()
        subscriber = AuthAuditSubscriber()
        
        # Subscribe handler to the login topic
        event_bus.subscribe("user.logged_in", subscriber.handle_login)