Skip to content

Authentication Component

This component addresses the need to reduce our dependency on country users by removing all authentication logic from the Authenticatable mixin.

Usage

The authentication component provides an api to manage authentication for our front office users (company admin and members). In this component SHOULD include: - an api for other components to create/update/delete an identity - a database for linking a profile with an identity - blueprints that can be bootstrapped in the different apps to handle login/logout/registration flows - all the factories for testing authentication flows

⚠️ This component is global and should not include any country specific logic.

State of the migration

  • [x] All identity management code has been migration
  • [x] Login verification has been migrated
  • [ ] User isn't linked to an identity anymore (or at least the link isn't used as the source of truth)
  • [ ] All token management code has been migrated

How to use the api

from components.authentication.public.api import AuthenticationService
authentication_service = AuthenticationService.create()

from components.authentication.public.inject_authentication_service import inject_authentication_service

@inject_authentication_service
def foo(authentication_service: AuthenticationService, other_arg):
    ...

Backward compatibility

During the migration phase, we need to keep backward compatibility with the existing authentication system. - The RetrocompatibilityRepository is used to read data from the existing authentication system (in the country user models) - The DoubleWriteRepository is used to maintain the consistency between the new authentication system and the existing one.

A few flow examples

Create an identity and assign it a password

from components.authentication.public.api import AuthenticationService

authentication_service = AuthenticationService.create()
# Create the identity in keycloak and safe the link between this identity and a profile in the authentication_identity table
identity_id = authentication_service.create_identity(email=email, language=lang, mfa_required=False, profile_id=profile_id)

# Set a password for this identity
authentication_service.set_identity_credentials(identity_id=identity_id, prehashed_password=prehashed_password)

Get an identity

from components.authentication.public.api import AuthenticationService
from components.authentication.public.entities import AuthIdentity

authentication_service = AuthenticationService.create()
identity: AuthIdentity | None = authentication_service.get_identity_by_profile_id(profile_id=profile_id)
# OR
identity: AuthIdentity | None = authentication_service.get_identity_by_email(email=email)
# OR
identity: AuthIdentity | None = authentication_service.get_keycloak_identity(keycloak_id=keycloak_id)

if identity is not None:
    email = identity.email
    first_name = identity.first_name
    last_name = identity.last_name
    is_mfa_required = identity.mfa_required
    is_mfa_enabled = identity.mfa_enabled

Update an identity MFA status

from components.authentication.public.api import AuthenticationService

authentication_service = AuthenticationService.create()
authentication_service.change_mfa_status(identity_id=identity_id, mfa_enabled=True, mfa_required=False)

Verify an identity credentials

from components.authentication.public.inject_authentication_service import inject_authentication_service

@inject_authentication_service
def verify_credentials(authentication_service: AuthenticationService, user_input):
email = user_input.email
prehashed_password = user_input.prehashed_password

identity: AuthIdentity | None = authentication_service.get_identity_by_email(email=email)
if identity is None:
    raise ValueError("No identity found for this email")
if not authentication_service.check_identity_has_password(identity_id=identity.id):
    raise ValueError("This identity has no password set")
if not authentication_service.check_identity_password(identity_id=identity.id, prehashed_password=prehashed_password):
    raise ValueError("Invalid credentials")
return "Congrats the credentials are valid"

Events

The authentication component follows an event-driven architecture, publishing domain events when authentication-related operations occur. These events enable other components to react to authentication changes without tight coupling.

Published Events

  • IdentityCreatedEvent - Emitted when a new identity is successfully created in Keycloak
  • IdentityEmailChangedEvent - Emitted when an identity's email address is successfully changed
  • PasswordResetEmailSentEvent - Emitted when a password reset email is sent to an identity
  • IdentityMergedEvent - Emitted when two identities are merged due to email conflicts during email changes
  • IdentityEmailCleared - Emitted when an identity's email is replaced with an invalidated placeholder

Event-Driven Workflows

// TODO: @thibaut.caillierez: modify this documentation once we start using a transaction uow Identity Creation: 

identity_id = authentication_service.create_identity(email="user@example.com", language=Lang.ENGLISH)
# → Publishes IdentityCreatedEvent

Email Change with Conflict Resolution: 

authentication_service.change_identity_email(identity_id=identity_id, email="existing@example.com")
# → May publish IdentityEmailChangedEvent OR IdentityMergedEvent depending on email availability

Security & Authentication Details

Password

The authentication component follows strict security practices for credential management:

Pre-hashed Passwords: - All passwords must be pre-hashed on the client side before being sent to the server - The component never handles plaintext passwords - Use check_identity_password() with pre-hashed passwords for verification

Credential Lifecycle: 

# Set initial credentials (identity must exist first)
authentication_service.set_identity_credentials(
    identity_id=identity_id,
    email="user@example.com", 
    prehashed_password=hashed_password,
    is_email_verified=True
)

# Verify credentials during login
has_password = authentication_service.check_identity_has_password(identity_id)
is_valid = authentication_service.check_identity_password(identity_id=identity_id, prehashed_password=hashed_password)

Multi-Factor Authentication (MFA)

The component provides granular MFA control with two independent settings:

  • mfa_enabled - Whether the user has configured MFA (has authenticator app, etc.)
  • mfa_required - Whether MFA is mandatory for this identity (policy enforcement)

MFA States: - Required + Enabled: User must use MFA and has it configured ✅ - Required + Not Enabled: User must configure MFA before login ⚠️ - Not Required + Enabled: User can optionally use MFA - Not Required + Not Enabled: Standard password-only login

# Require MFA for high-privilege users
authentication_service.change_mfa_status(identity_id=admin_identity_id, mfa_required=True)

# User successfully configured MFA
authentication_service.change_mfa_status(identity_id=identity_id, mfa_enabled=True)

CLI Commands

The authentication component provides CLI commands for administrative operations and data management.

Available Commands

Data Migration & Backfill: 

# Backfill authentication_identity table from existing user data across all countries -> It will only work from fr_api, be_api and es_api since they are the only apps with access to all models
flask authentication backfill_users_identity [--dry-run]

# Find keycloak_id conflicts between different country user models
flask authentication find_identity_conlicts

# Check that all users with keycloak_id have corresponding authentication identity
flask authentication check_all_user_have_identity

Identity Data Consistency: 

# Fix profile/identity inconsistencies for a specific profile
flask authentication fix_identity_inconsistencies <profile_id> [--dry-run] [--first-name TEXT] [--last-name TEXT] [--language TEXT]

# Fix inconsistencies for multiple profiles (automatically when possible)
flask authentication fix_many_identities_inconsistencies <profile_id1> <profile_id2> ... [--dry-run]

Administrative Operations: 

# Reset user credentials (removes password and revokes all sessions)
flask authentication reset_users_credentials <keycloak_id1> <keycloak_id2> ... [--dry-run]

Note: All commands support --dry-run flag to preview changes without applying them.

Testing with TestUserWithAuthenticationFactory

For testing authentication flows in your components, use the provided test factory:

from components.authentication.public.tests.test_user_with_authentication_factory import TestUserWithAuthenticationFactory

def test_authentication_flow():
    # Create a complete test user with authentication identity
    test_user = TestUserWithAuthenticationFactory.create(
        email="test@example.com",
        password="hashed_password_123",
        mfa_required=False
    )

    # Access the user profile
    profile = test_user.profile
    assert profile.email == "test@example.com"

    # Access the authentication identity
    auth_identity = test_user.authentication_identity
    assert auth_identity.keycloak_id is not None

    # Use with authentication service
    authentication_service = AuthenticationService.create()
    identity = authentication_service.get_keycloak_identity_by_profile_id(profile.id)
    assert identity is not None
    assert identity.email == "test@example.com"

def test_mfa_enabled_user():
    # Create user with MFA enabled
    test_user = TestUserWithAuthenticationFactory.create(
        email="mfa-user@example.com",
        mfa_required=True,
        mfa_enabled=True
    )

    identity = test_user.authentication_identity
    assert identity.mfa_required is True
    assert identity.mfa_enabled is True

Developer guide

How to use a blueprint in my app

How does the api work

Read data:

  1. Start a unit of work -> it will build instantiate the proper repository (to read the link between profile and identity) and the proper identity provider (eg to reach keycloak's api)
  2. Find the requested data in keycloak or in the authentication_identity table depending on the input
  3. (Optional) Enrich it with data from the authentication_identity table or from keycloak
  4. Map this data to a public entities and return it
  5. Exits the unit of work

Write data:

  1. Start a unit of work -> it will build instantiate the proper repository (to write the link between profile and identity) and the proper identity provider (eg to reach keycloak's api)
  2. Create a command with the input data and add it to the message bus to handle
  3. The command handler mapped to the command will be called and handle the logic
  4. it may call the repository to handle data persistence in our DB
  5. it may call the identity provider to communicate with keycloak
  6. it may add events to the unit of work's events list
  7. If there are still domain events in the message bus to handle, repeat step 3 until it's empty
  8. Exits the unit of work, dispatching events and flushing the session

Component structure

authentication/
├── conftest.py                        # Shared pytest configuration for component tests
├── bootstrap/                         # Component bootstrapping and Flask integration
├── internal/
│   ├── application/                   # Application layer (CQRS pattern)
│   │   ├── command_handlers.py        # Command handlers for business operations
│   │   ├── commands.py               # Command definitions (data structures)
│   │   ├── event_handlers.py         # Event handlers for domain events
│   │   ├── message_bus.py            # Message bus for domain command/event routing
│   │   ├── subscribers.py            # Implemenentation of other components' events handlers
│   ├── commands/                     # CLI commands implementation
│   ├── controllers/                  # Internal API controllers
│   │   └── token.py                  # Token management endpoints
│   ├── domain/                       # Domain layer (business logic)
│   │   ├── entities.py               # Domain entities and value objects
│   │   └── events.py                 # Domain events definitions
│   ├── infrastructure/
│   │   ├── double_write_repository.py  # Dual-write pattern for data migration
│   │   ├── event_dispatcher.py       # Event dispatching implementation
│   │   ├── flask_auth.py             # Flask authentication integration
│   │   ├── identity_provider.py      # Identity provider layer, connecting the component to keycloak
│   │   ├── mfa_auth.py               # Multi-factor authentication service
│   │   ├── repository.py             # Main repository implementation -> writting to authentication.authentication_identity
│   │   ├── retro_compatibility_repository.py  # Backward compatibility data access ⚠️ only for retrocompatibility
│   │   ├── unit_of_work.py           # Unit of work pattern implementations
│   └── models/                  
│       ├── authentication_identity.py  # Authentication identity ORM model to link an identity to a profile
│       └── helpers.py                # Model helper functions
└── public/
    ├── api.py                        # Public API interface
    ├── entities.py                   # Public entities (data contracts)
    ├── feature_flags.py              # Feature flags configuration
    ├── identity_provider.py          # Public identity provider interface ⚠️ only for retrocompatility
    ├── inject_authentication_service.py  # Dependency injection helpers
    ├── blueprints/                   # Flask blueprints
    ├── events/
    │   ├── component_events.py       # Component integration event definitions
    │   └── subscription.py           # Subscription to other components' events
    └── tests/
        └── test_user_with_authentication_factory.py  # Test factory to create a test user with an identity

Reference