flip_api.utils.cognito_helpers

Attributes

`_EMAIL_VALIDATOR`
`_DEFAULT_PORTS`
`_MFA_STATE_TTL_SECONDS`
`_mfa_state_cache`
`_mfa_state_cache_lock`

Functions

`_cognito_client`(→ Any)	Module-level cached cognito-idp client. boto3 clients are thread-safe
`_origin_from_url`(→ str \| None)	Return `scheme://host[:port]` for `url`, omitting ports that match the scheme default.
`get_cors_allowed_origins`(→ list[str])	Derive the CORS allowlist from the Cognito user pool client's CallbackURLs.
`get_pool_id`(→ str)	Extract the user pool ID from the request context.
`get_user_pool_id`(→ str)
`get_cognito_users`(...)	Get users from Cognito user pool.
`_safe_email_for_cognito_filter`(→ str)	Validate that `email` is safe to interpolate into a Cognito ListUsers
`_safe_uuid_for_cognito_filter`(→ str)	Coerce `user_id` to its canonical string UUID form, raising 400 if
`get_user_by_email_or_id`(...)	Get a user from Cognito by email or ID.
`get_username`(→ str)	Get a username from Cognito by user ID.
`update_user`(→ flip_api.domain.schemas.users.Disabled)	Enable or disable a user in Cognito.
`delete_cognito_user`(→ None)	Delete a user from Cognito.
`reset_user_mfa`(→ None)	Disable a user's TOTP MFA preference and invalidate their sessions.
`_invalidate_mfa_cache`(→ None)	Drop the cached MFA-enabled state for a user (used after admin reset).
`is_mfa_enabled`(→ bool)	Check whether a user has TOTP MFA active in Cognito.
`revoke_token`(→ None)	Revoke a refresh token in Cognito.
`get_user_role_data`(...)	Get user role data with pagination and filtering.
`get_all_roles`(→ list[uuid.UUID])	Get all role IDs from the database.
`validate_roles`(→ None)	Validate that all user roles exist in the database.
`create_cognito_user`(→ uuid.UUID)	Create a new user in Cognito.
`filter_enabled_users`(→ list[uuid.UUID])	Filter out disabled users from a list of user IDs.

Module Contents

flip_api.utils.cognito_helpers._EMAIL_VALIDATOR: pydantic.TypeAdapter[str]

flip_api.utils.cognito_helpers._cognito_client() → Any: Module-level cached cognito-idp client. boto3 clients are thread-safe and expensive to construct; sharing one avoids paying endpoint-resolution cost on every authenticated request.

flip_api.utils.cognito_helpers._DEFAULT_PORTS

flip_api.utils.cognito_helpers._origin_from_url(url: str) → str | None

Return scheme://host[:port] for url, omitting ports that match the scheme default.

Browsers strip default ports from the Origin header (RFC 6454), so an allowlist entry like https://localhost:443 would never match an actual request — normalize before use. Returns None for URLs without a usable scheme/host.

flip_api.utils.cognito_helpers.get_cors_allowed_origins() → list[str]

Derive the CORS allowlist from the Cognito user pool client’s CallbackURLs.

The same Cognito app client that authenticates UI logins already enumerates the trusted UI origins per environment (see deploy/providers/AWS/services.tf). Reusing it as the CORS allowlist keeps “where users can sign in” and “where the UI may call this API” in lockstep, without a separate env var.

Returns:: Unique normalized origins (scheme://host[:port]) suitable for CORSMiddleware(allow_origins=...).
Return type:: list[str]

flip_api.utils.cognito_helpers.get_pool_id(request: fastapi.Request) → str

Extract the user pool ID from the request context.

Parameters:: request (Request) – FastAPI request object
Returns:: The user pool ID extracted from the request context
Return type:: str
Raises:: HTTPException – If the user pool ID is not found

flip_api.utils.cognito_helpers.get_user_pool_id(request: fastapi.Request) → str

flip_api.utils.cognito_helpers.get_cognito_users(params: dict[str, Any] | None = None) → list[flip_api.domain.schemas.users.CognitoUser]

Get users from Cognito user pool.

Parameters:: params (dict[str, Any] | None) – Additional parameters to pass to the ListUsers API call.
Returns:: List of CognitoUser objects.
Return type:: list[CognitoUser]
Raises:: HTTPException – If there is an error fetching users from Cognito or if the user pool ID is not found.

flip_api.utils.cognito_helpers._safe_email_for_cognito_filter(email: str) → str

Validate that email is safe to interpolate into a Cognito ListUsers Filter expression.

Cognito’s filter syntax delimits values with double quotes; an unescaped " (or backslash) in the value can break out of the quoted context and inject additional clauses. EmailStr already forbids the characters that would let an attacker do this, but rejecting them explicitly here keeps this helper safe to call from any future caller that forgets the upstream validation.

flip_api.utils.cognito_helpers._safe_uuid_for_cognito_filter(user_id: str | uuid.UUID) → str

Coerce user_id to its canonical string UUID form, raising 400 if it isn’t a valid UUID.

A valid UUID’s string form is hex+hyphen only, so once normalised it can be interpolated into Cognito’s filter syntax without escaping.

flip_api.utils.cognito_helpers.get_user_by_email_or_id(user_pool_id: str, email: str | None = None, user_id: uuid.UUID | None = None) → flip_api.domain.schemas.users.CognitoUser

Get a user from Cognito by email or ID.

Parameters:

user_pool_id (str) – Cognito user pool ID
email (str | None) – User email (optional)
user_id (UUID | None) – User ID (optional)

Returns:

The user matching the email or ID.

Return type:

CognitoUser

Raises:

HTTPException – If neither email nor user_id is provided, or if the supplied identifier fails format validation.

flip_api.utils.cognito_helpers.get_username(user_id: str, user_pool_id: str) → str

Get a username from Cognito by user ID.

Parameters:

user_id (str) – User ID (sub in Cognito)
user_pool_id (str) – Cognito user pool ID

Returns:

The username (email) associated with the user ID.

Return type:

str

Raises:

HTTPException – 400 if user_id isn’t a valid UUID, 404 if no matching user, 500 on Cognito errors.

flip_api.utils.cognito_helpers.update_user(username: str, user_pool_id: str, disabled: bool) → flip_api.domain.schemas.users.Disabled

Enable or disable a user in Cognito.

Parameters:

username (str) – Username (email)
user_pool_id (str) – Cognito user pool ID
disabled (bool) – Whether to disable the user

Returns:

An object indicating the disabled status of the user after the update.

Return type:

Disabled

Raises:

HTTPException – If the request cannot be processed.

flip_api.utils.cognito_helpers.delete_cognito_user(username: str, user_pool_id: str) → None

Delete a user from Cognito.

Parameters:

username (str) – Username (email)
user_pool_id (str) – Cognito user pool ID

Returns:

None

Raises:

HTTPException – If there is an error deleting the user from Cognito.

flip_api.utils.cognito_helpers.reset_user_mfa(username: str, user_pool_id: str) → None

Disable a user’s TOTP MFA preference and invalidate their sessions.

Cognito has no admin API to delete a verified TOTP secret, so clearing the preference is the only server-side handle; the app-layer MFA gate (verify_token + router guard) then funnels the user through post-auth enrolment, which mints a fresh secret. A global sign-out revokes any active refresh tokens so a pre-reset session cannot keep operating.

Parameters:

username (str) – Username (email)
user_pool_id (str) – Cognito user pool ID

Returns:

None

Raises:

HTTPException – If resetting MFA or signing the user out fails

flip_api.utils.cognito_helpers._MFA_STATE_TTL_SECONDS = 60.0

flip_api.utils.cognito_helpers._mfa_state_cache: dict[tuple[str, str], tuple[bool, float]]

flip_api.utils.cognito_helpers._mfa_state_cache_lock

flip_api.utils.cognito_helpers._invalidate_mfa_cache(username: str, user_pool_id: str) → None: Drop the cached MFA-enabled state for a user (used after admin reset).

flip_api.utils.cognito_helpers.is_mfa_enabled(username: str, user_pool_id: str) → bool

Check whether a user has TOTP MFA active in Cognito.

A user is considered MFA-active if SOFTWARE_TOKEN_MFA is present in their UserMFASettingList — Cognito only adds that entry after the user has both verified a software token and had their preference set with Enabled=True.

Results are cached for a short TTL because verify_token calls this on every authenticated request; see _MFA_STATE_TTL_SECONDS.

Parameters:

username (str) – Username (email)
user_pool_id (str) – Cognito user pool ID

Returns:

True if TOTP MFA is enabled for the user, False otherwise.

Return type:

bool

Raises:

HTTPException – If the Cognito lookup fails.

flip_api.utils.cognito_helpers.revoke_token(refresh_token: str, client_id: str) → None

Revoke a refresh token in Cognito.

Parameters:

refresh_token (str) – Refresh token to revoke
client_id (str) – Cognito app client ID

Returns:

None

Raises:

HTTPException – If token revocation fails

flip_api.utils.cognito_helpers.get_user_role_data(paging_info: flip_api.utils.paging_utils.PagingInfo, users: list[flip_api.domain.schemas.users.CognitoUser], session: sqlmodel.Session) → list[flip_api.domain.schemas.users.IUser]

Get user role data with pagination and filtering.

Parameters:

paging_info (PagingInfo) – Pagination and filtering information.
users (list[CognitoUser]) – List of Cognito users.
session (Session) – Database session.

Returns:

List of IUser objects with roles.

Return type:

list[IUser]

flip_api.utils.cognito_helpers.get_all_roles(db: sqlmodel.Session) → list[uuid.UUID]

Get all role IDs from the database.

Parameters:: db (Session) – Database session
Returns:: List of role IDs
Return type:: list[UUID]

flip_api.utils.cognito_helpers.validate_roles(user_roles: list[uuid.UUID], roles_from_db: list[uuid.UUID]) → None

Validate that all user roles exist in the database.

Parameters:

user_roles (list[UUID]) – List of role IDs to validate
roles_from_db (list[UUID]) – List of valid role IDs from the database

Returns:

None

Raises:

HTTPException – If any role is invalid

flip_api.utils.cognito_helpers.create_cognito_user(email: str, user_pool_id: str) → uuid.UUID

Create a new user in Cognito.

Parameters:

email (str) – User email
user_pool_id (str) – Cognito user pool ID

Returns:

The ID of the created user

Return type:

UUID

Raises:

HTTPException – If user creation fails

flip_api.utils.cognito_helpers.filter_enabled_users(user_pool_id: str, users: list[uuid.UUID]) → list[uuid.UUID]

Filter out disabled users from a list of user IDs.

Parameters:

user_pool_id (str) – Cognito user pool ID
users (list[UUID]) – List of user IDs to filter

Returns:

List of enabled user IDs

Return type:

list[UUID]