import typing from flask import current_app from atlas_um import consts from atlas_um.helpers import either from atlas_um import pgdb from atlas_um.extensions import redis from atlas_um.helpers.services import BaseLogicService from atlas_um.helpers.tokens import get_uid_str from atlas_um.tokens.bearer_tokens import DNABearerToken from redis.exceptions import RedisError class CreateBearerTokenService(BaseLogicService): """ Creates new bearer token. Decouples the creation logic and the tokens classes. """ commit = False def process( self, dna_account: pgdb.DNAAccount, resource_group: typing.Optional[pgdb.ResourceGroup] = None, ttl: int = None, ) -> either.Either: dna_bearer_token = DNABearerToken(dna_account, resource_group, ttl) return either.Right(dna_bearer_token.encode()) class CreateRefreshTokenService(BaseLogicService): """ Creates new refresh token. Stores new refresh token in Redis under 3 keys: - session key (hard session limit TTL) - tokens family key for history (rolling active session limit TTL) - swappable refresh token key (rolling active session limit TTL) Session key-values logic: := Family keys-values logic: := Tokens keys-values logic: ::= Session keys are being used for generic max session time limit, determining by their TTL, and to keep resource group id the session relates to. Tokens keys are swappable and should be deleted after reissue, Family keys are persistent to track malicious reuse. Family itself is needed for separate invalidation of tokens chains on different devices. This is something like sessions. Session TTL is time when session will expire anyway and there is no way to expand it by refreshing the token. It is a hard limit. Active session TTL is time frame when it is possible to refresh token and it will be extended on each refresh, until the hard limit in previously mentioned session TTL. Example use case: user would be logged out after 12 hours of inactivity, but even if the activity is regular, the logout anyway should happen after 30 days, even it hte user refresh the token each 10 hours. """ def process( self, dna_account: pgdb.DNAAccount, resource_group: typing.Optional[pgdb.ResourceGroup] = None, ) -> either.Either: token = get_uid_str() session_key = f"{consts.DNA_REFRESH_TOKEN_SESSION_KEY_PREFIX}:{token}" family_key = f"{consts.DNA_REFRESH_TOKEN_FAMILY_KEY_PREFIX}:{token}" token_key = f"{consts.DNA_REFRESH_TOKEN_KEY_PREFIX}:{token}:{token}" session_ttl = ( resource_group.session_ttl if resource_group else current_app.config["DEFAULT_REFRESH_TOKEN_TTL"] ) active_session_ttl = ( resource_group.active_session_ttl if resource_group else current_app.config["DEFAULT_REFRESH_TOKEN_TTL"] ) pipeline = redis.pipeline(transaction=True) pipeline.multi() pipeline.setex( session_key, session_ttl, resource_group.id if resource_group else consts.DNA_REFRESH_TOKEN_SESSION_EMPTY_VALUE, ) pipeline.setex( family_key, active_session_ttl, token, ) pipeline.setex( token_key, active_session_ttl, dna_account.id, ) pipeline.execute() return either.Right(token) class RotateTokensService(BaseLogicService): """ Rotates the existing refresh and bearer tokens. Checks if the refresh token exists, if account is active, if session exists (the keys exists, otherwise it expired), swaps refresh token for new, issues new bearer token. Stores new refresh token in Redis under 3 keys: - session key (hard session limit TTL) - tokens family key for history (rolling active session limit TTL) - swappable refresh token key (rolling active session limit TTL) If the refresh token does not exist, then check if we have it in history. If yes, then the refresh chain is broken and possible this token was stolen and already refreshed, so we need to cleanup tokens by family and force the user to login with the actual credentials. Session TTL is time when session will expire anyway and there is no way to expand it by refreshing the token. It is a hard limit. Active session TTL is time frame when it is possible to refresh token and it will be extended on each refresh, until the hard limit in previously mentioned session TTL. Example use case: user would be logged out after 12 hours of inactivity, but even if the activity is regular, the logout anyway should happen after 30 days, even it hte user refresh the token each 10 hours. :returns (new_refresh_token, new_dna_bearer_token) """ def process( self, refresh_token: str, resource_group_hint_name: typing.Optional[str] = None, ) -> either.Either: family = self._get_family(refresh_token) ( resource_group, active_session_ttl, ) = self._get_resource_group_and_active_session_ttl(family) if not family or not active_session_ttl: return either.Left("No existing session") dna_account = self._get_dna_account(refresh_token, family) if dna_account.status != pgdb.DNAAccountStatuses.ACTIVE: self._invalidate_family(family) return either.Left("Invalid token") # for services which deployed on the same domain with atlas # we get empty resource group, so get this rg from token data resource_group_hint = resource_group if resource_group_hint_name and ( not resource_group_hint or resource_group_hint.namespace_url == current_app.config["RELATED_CLAIMS_NAMESPACE"] ): resource_group_hint = ( pgdb.ResourceGroup.query.active() .filter( pgdb.ResourceGroup.namespace_url == resource_group_hint_name ) .first() ) pgdb.DNAAccountActivity.update_last_activity( dna_account, resource_group_hint, ) new_refresh_token = self._rotate_refresh_token( refresh_token, family, active_session_ttl ) if ( resource_group and resource_group.namespace_url == current_app.config.get("RELATED_CLAIMS_NAMESPACE") ): resource_group = None new_dna_bearer_token = DNABearerToken( dna_account, resource_group ).encode() return either.Right( (new_refresh_token, new_dna_bearer_token, dna_account) ) def _get_resource_group_and_active_session_ttl( self, family: str ) -> typing.Tuple: session_key = f"{consts.DNA_REFRESH_TOKEN_SESSION_KEY_PREFIX}:{family}" resource_group_id = redis.get(session_key) # generic cases when there was no resource group on initial login if resource_group_id == consts.DNA_REFRESH_TOKEN_SESSION_EMPTY_VALUE: return None, current_app.config["DEFAULT_REFRESH_TOKEN_TTL"] if resource_group_id: resource_group = ( pgdb.ResourceGroup.query.active() .filter(pgdb.ResourceGroup.id == resource_group_id) .first() ) else: resource_group = None active_session_ttl = ( resource_group.active_session_ttl if resource_group else None ) return resource_group, active_session_ttl def _get_family(self, refresh_token: str) -> typing.Optional[str]: family_key = ( f"{consts.DNA_REFRESH_TOKEN_FAMILY_KEY_PREFIX}:{refresh_token}" ) return redis.get(family_key) def _get_dna_account( self, refresh_token: str, family: str ) -> pgdb.DNAAccount: token_key = ( f"{consts.DNA_REFRESH_TOKEN_KEY_PREFIX}:{family}:{refresh_token}" ) dna_account_id = redis.get(token_key) if not dna_account_id: return pgdb.NullDNAAccount() dna_account = ( pgdb.DNAAccount.query.get(dna_account_id) or pgdb.NullDNAAccount() ) return dna_account def _invalidate_family(self, family: str): keys = redis.keys(f"{consts.DNA_REFRESH_TOKEN_KEY_PREFIX}:{family}:*") try: redis.delete(*keys) except RedisError: pass def _rotate_refresh_token( self, refresh_token: str, family: str, active_session_ttl: int ) -> str: token_key = ( f"{consts.DNA_REFRESH_TOKEN_KEY_PREFIX}:{family}:{refresh_token}" ) new_refresh_token = get_uid_str() new_token_key = ( f"{consts.DNA_REFRESH_TOKEN_KEY_PREFIX}:{family}:" f"{new_refresh_token}" ) new_family_key = ( f"{consts.DNA_REFRESH_TOKEN_FAMILY_KEY_PREFIX}:{new_refresh_token}" ) pipeline = redis.pipeline(transaction=True) pipeline.multi() pipeline.rename(token_key, new_token_key) # using active_session_ttl to prolongate the session by it # each time on token refresh # the max session time is limited by generic session ttl # using for session key pipeline.expire(new_token_key, active_session_ttl) pipeline.setex( new_family_key, active_session_ttl, family, ) pipeline.execute() return new_refresh_token class InvalidateRefreshToken(BaseLogicService): """ Deletes refresh token in Redis. Prevents further tokens rotation for tokens family. """ def process(self, refresh_token: str) -> either.Either: family = self._get_family(refresh_token) if family: keys = redis.keys( f"{consts.DNA_REFRESH_TOKEN_KEY_PREFIX}:{family}:*" ) try: redis.delete(*keys) except RedisError: pass return either.Right(None) def _get_family(self, refresh_token: str) -> typing.Optional[str]: family_key = ( f"{consts.DNA_REFRESH_TOKEN_FAMILY_KEY_PREFIX}:{refresh_token}" ) return redis.get(family_key)