Skip to content

application.commands.handlers.logout_user_handler

src.application.commands.handlers.logout_user_handler

Logout User handler for User Authentication.

Flow: 1. Emit UserLogoutAttempted event 2. Find refresh token by verification 3. Revoke refresh token (or revoke by session if token found) 4. Emit UserLogoutSucceeded event 5. Return Success(message)

Note: JWT access tokens cannot be revoked (they expire naturally in 15 minutes). This handler only revokes the refresh token to prevent new access tokens.

Architecture: - Application layer ONLY imports from domain layer (entities, protocols, events) - NO infrastructure imports (repositories are injected via protocols)

Classes

LogoutError

Logout error reasons.

Source code in src/application/commands/handlers/logout_user_handler.py 
41
42
43
44
45
class LogoutError:
    """Logout error reasons."""

    TOKEN_NOT_FOUND = "token_not_found"
    TOKEN_ALREADY_REVOKED = "token_already_revoked"

LogoutResponse dataclass

Response data for successful logout.

Source code in src/application/commands/handlers/logout_user_handler.py 
48
49
50
51
52
@dataclass
class LogoutResponse:
    """Response data for successful logout."""

    message: str = "Successfully logged out."

LogoutUserHandler

Handler for logout user command.

Revokes the refresh token to prevent new access tokens from being issued. The current access token remains valid until it expires (15 minutes).

Follows hexagonal architecture: - Application layer (this handler) - Domain layer (protocols) - Infrastructure layer (repositories, services via dependency injection)

Source code in src/application/commands/handlers/logout_user_handler.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class LogoutUserHandler:
    """Handler for logout user command.

    Revokes the refresh token to prevent new access tokens from being issued.
    The current access token remains valid until it expires (15 minutes).

    Follows hexagonal architecture:
    - Application layer (this handler)
    - Domain layer (protocols)
    - Infrastructure layer (repositories, services via dependency injection)
    """

    def __init__(
        self,
        refresh_token_repo: RefreshTokenRepository,
        refresh_token_service: RefreshTokenServiceProtocol,
        event_bus: EventBusProtocol,
    ) -> None:
        """Initialize logout handler with dependencies.

        Args:
            refresh_token_repo: Refresh token repository for revocation.
            refresh_token_service: Service for token verification.
            event_bus: Event bus for publishing domain events.
        """
        self._refresh_token_repo = refresh_token_repo
        self._refresh_token_service = refresh_token_service
        self._event_bus = event_bus

    async def handle(
        self, cmd: LogoutUser, request: "Request | None" = None
    ) -> Result[LogoutResponse, str]:
        """Handle logout user command.

        Args:
            cmd: LogoutUser command with user_id and refresh_token.
            request: Optional FastAPI Request for IP/user agent tracking (PCI-DSS 10.2.7).

        Returns:
            Success(LogoutResponse) on successful logout.
            Failure(error_message) on failure.

        Side Effects:
            - Publishes UserLogoutAttempted event (always).
            - Publishes UserLogoutSucceeded/Failed event.
            - Revokes refresh token in database.
        """
        # Extract request metadata for audit trail (PCI-DSS 10.2.7)
        metadata: dict[str, str] = {}
        if request and request.client:
            metadata["ip_address"] = request.client.host
            metadata["user_agent"] = request.headers.get("user-agent", "Unknown")
        # Step 1: Emit ATTEMPTED event
        await self._event_bus.publish(
            UserLogoutAttempted(
                event_id=uuid7(),
                occurred_at=datetime.now(UTC),
                user_id=cmd.user_id,
            ),
            metadata=metadata,
        )

        # Step 2: Find refresh token by verification
        token_data = await self._refresh_token_repo.find_by_token_verification(
            cmd.refresh_token,
            self._refresh_token_service.verify_token,
        )

        session_id: UUID | None = None

        if token_data is None:
            # Token not found - could be already revoked or invalid
            # For security, we still return success but log internally
            await self._event_bus.publish(
                UserLogoutFailed(
                    event_id=uuid7(),
                    occurred_at=datetime.now(UTC),
                    user_id=cmd.user_id,
                    reason=LogoutError.TOKEN_NOT_FOUND,
                ),
                metadata=metadata,
            )
            # Return success anyway to prevent information leakage
            # User experience: they wanted to logout, we say they're logged out
            return Success(value=LogoutResponse())

        # Check if token belongs to this user (security check)
        if token_data.user_id != cmd.user_id:
            # Token doesn't belong to user - security issue
            await self._event_bus.publish(
                UserLogoutFailed(
                    event_id=uuid7(),
                    occurred_at=datetime.now(UTC),
                    user_id=cmd.user_id,
                    reason="token_user_mismatch",
                ),
                metadata=metadata,
            )
            # Return success to prevent information leakage
            return Success(value=LogoutResponse())

        # Check if already revoked
        if token_data.revoked_at is not None:
            await self._event_bus.publish(
                UserLogoutFailed(
                    event_id=uuid7(),
                    occurred_at=datetime.now(UTC),
                    user_id=cmd.user_id,
                    reason=LogoutError.TOKEN_ALREADY_REVOKED,
                ),
                metadata=metadata,
            )
            # Return success - user wanted to logout, token is already revoked
            return Success(value=LogoutResponse())

        session_id = token_data.session_id

        # Step 3: Revoke the session (revokes all tokens for this session)
        await self._refresh_token_repo.revoke_by_session(session_id)

        # Step 4: Emit SUCCEEDED event
        await self._event_bus.publish(
            UserLogoutSucceeded(
                event_id=uuid7(),
                occurred_at=datetime.now(UTC),
                user_id=cmd.user_id,
                session_id=session_id,
            ),
            metadata=metadata,
        )

        # Step 5: Return Success
        return Success(value=LogoutResponse())
Functions
__init__ 
__init__(
    refresh_token_repo: RefreshTokenRepository,
    refresh_token_service: RefreshTokenServiceProtocol,
    event_bus: EventBusProtocol,
) -> None

Parameters:

Name Type Description Default
refresh_token_repo RefreshTokenRepository

Refresh token repository for revocation.

 required
refresh_token_service RefreshTokenServiceProtocol

Service for token verification.

 required
event_bus EventBusProtocol

Event bus for publishing domain events.

 required
Source code in src/application/commands/handlers/logout_user_handler.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def __init__(
    self,
    refresh_token_repo: RefreshTokenRepository,
    refresh_token_service: RefreshTokenServiceProtocol,
    event_bus: EventBusProtocol,
) -> None:
    """Initialize logout handler with dependencies.

    Args:
        refresh_token_repo: Refresh token repository for revocation.
        refresh_token_service: Service for token verification.
        event_bus: Event bus for publishing domain events.
    """
    self._refresh_token_repo = refresh_token_repo
    self._refresh_token_service = refresh_token_service
    self._event_bus = event_bus
handle async 
handle(
    cmd: LogoutUser, request: Request | None = None
) -> Result[LogoutResponse, str]

Handle logout user command.

Parameters:

Name Type Description Default
cmd LogoutUser

LogoutUser command with user_id and refresh_token.

 required
request Request | None

Optional FastAPI Request for IP/user agent tracking (PCI-DSS 10.2.7).

 None

Returns:

Type Description
Result[LogoutResponse, str]

Success(LogoutResponse) on successful logout.
Result[LogoutResponse, str]

Failure(error_message) on failure.
Side Effects
  • Publishes UserLogoutAttempted event (always).
  • Publishes UserLogoutSucceeded/Failed event.
  • Revokes refresh token in database.
Source code in src/application/commands/handlers/logout_user_handler.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
async def handle(
    self, cmd: LogoutUser, request: "Request | None" = None
) -> Result[LogoutResponse, str]:
    """Handle logout user command.

    Args:
        cmd: LogoutUser command with user_id and refresh_token.
        request: Optional FastAPI Request for IP/user agent tracking (PCI-DSS 10.2.7).

    Returns:
        Success(LogoutResponse) on successful logout.
        Failure(error_message) on failure.

    Side Effects:
        - Publishes UserLogoutAttempted event (always).
        - Publishes UserLogoutSucceeded/Failed event.
        - Revokes refresh token in database.
    """
    # Extract request metadata for audit trail (PCI-DSS 10.2.7)
    metadata: dict[str, str] = {}
    if request and request.client:
        metadata["ip_address"] = request.client.host
        metadata["user_agent"] = request.headers.get("user-agent", "Unknown")
    # Step 1: Emit ATTEMPTED event
    await self._event_bus.publish(
        UserLogoutAttempted(
            event_id=uuid7(),
            occurred_at=datetime.now(UTC),
            user_id=cmd.user_id,
        ),
        metadata=metadata,
    )

    # Step 2: Find refresh token by verification
    token_data = await self._refresh_token_repo.find_by_token_verification(
        cmd.refresh_token,
        self._refresh_token_service.verify_token,
    )

    session_id: UUID | None = None

    if token_data is None:
        # Token not found - could be already revoked or invalid
        # For security, we still return success but log internally
        await self._event_bus.publish(
            UserLogoutFailed(
                event_id=uuid7(),
                occurred_at=datetime.now(UTC),
                user_id=cmd.user_id,
                reason=LogoutError.TOKEN_NOT_FOUND,
            ),
            metadata=metadata,
        )
        # Return success anyway to prevent information leakage
        # User experience: they wanted to logout, we say they're logged out
        return Success(value=LogoutResponse())

    # Check if token belongs to this user (security check)
    if token_data.user_id != cmd.user_id:
        # Token doesn't belong to user - security issue
        await self._event_bus.publish(
            UserLogoutFailed(
                event_id=uuid7(),
                occurred_at=datetime.now(UTC),
                user_id=cmd.user_id,
                reason="token_user_mismatch",
            ),
            metadata=metadata,
        )
        # Return success to prevent information leakage
        return Success(value=LogoutResponse())

    # Check if already revoked
    if token_data.revoked_at is not None:
        await self._event_bus.publish(
            UserLogoutFailed(
                event_id=uuid7(),
                occurred_at=datetime.now(UTC),
                user_id=cmd.user_id,
                reason=LogoutError.TOKEN_ALREADY_REVOKED,
            ),
            metadata=metadata,
        )
        # Return success - user wanted to logout, token is already revoked
        return Success(value=LogoutResponse())

    session_id = token_data.session_id

    # Step 3: Revoke the session (revokes all tokens for this session)
    await self._refresh_token_repo.revoke_by_session(session_id)

    # Step 4: Emit SUCCEEDED event
    await self._event_bus.publish(
        UserLogoutSucceeded(
            event_id=uuid7(),
            occurred_at=datetime.now(UTC),
            user_id=cmd.user_id,
            session_id=session_id,
        ),
        metadata=metadata,
    )

    # Step 5: Return Success
    return Success(value=LogoutResponse())