Token management

How PayPulse Cloud stores, refreshes, and validates OAuth 2.0 tokens.

OAuth token management in PayPulse Cloud is handled by oauth_utils.py and secretsmanager_utils.py in the common Lambda layer. The design keeps token handling centralized so that every Lambda function that needs Gmail access follows the same refresh-and-validate path.

Token storage pattern

Tokens are stored per user in AWS Secrets Manager using the key pattern:

gmail/user/{user_id}

Where {user_id} is the internal PayPulse user ID (a UUID prefixed with user_). Each secret is a JSON object containing the following fields:

Field	Type	Description
access_token	string	Short-lived OAuth access token (1-hour lifetime)
refresh_token	string	Long-lived token used to acquire new access tokens
expires_at	string	ISO 8601 UTC timestamp of access token expiry
expires_in	integer	Token lifetime in seconds at time of issue
scope	string	Space-separated list of granted OAuth scopes
token_type	string	Always "Bearer"
created_at	string	ISO 8601 UTC timestamp of when the secret was written
google_user_id	string	Google’s unique identifier for the connected account
google_email	string	Google email address of the connected account
google_name	string	Display name from the Google account
google_verified_email	boolean	Whether Google has verified the email address

The prepare_oauth_secret_data function in oauth_utils.py builds this structure before storage:

oauth_utils.py

def prepare_oauth_secret_data(
    access_token: str,
    refresh_token: Optional[str],
    expires_in: int,
    scope: str,
    google_user_info: Dict[str, str] = None
) -> Dict[str, Any]:
    import time
    from datetime import datetime, timedelta

    # Calculate expiration timestamp
    expires_at = datetime.utcnow() + timedelta(seconds=expires_in)

    oauth_data = {
        "access_token": access_token,
        "expires_at": expires_at.isoformat(),
        "expires_in": expires_in,
        "scope": scope,
        "token_type": "Bearer",
        "created_at": datetime.utcnow().isoformat()
    }

    if refresh_token:
        oauth_data.update({"refresh_token": refresh_token})

    # Add Google user information if provided
    if google_user_info:
        oauth_data.update(google_user_info)

    return oauth_data

Token expiry check

Before a token is used, is_token_expired compares the stored expires_at timestamp against the current UTC time:

oauth_utils.py

def is_token_expired(oauth_data: Dict[str, Any]) -> bool:
    if 'expires_at' not in oauth_data:
        return False

    from datetime import datetime
    expires_at = datetime.fromisoformat(oauth_data['expires_at'])
    return datetime.utcnow() >= expires_at

If expires_at is absent (for example, in older records), the function returns False and lets the token attempt proceed.

Automatic token refresh

Token refresh happens inside create_gmail_service in gmail_api_utils.py. The function checks whether the token has expired or will expire within the next 5 minutes, and refreshes proactively if so:

gmail_api_utils.py

# Check if token is expired or will expire soon, and refresh proactively
should_refresh = False

# Check expiration using the stored expires_at timestamp
if expires_at:
    try:
        expiry_time = dateutil.parser.isoparse(expires_at).replace(tzinfo=None)
        current_time = datetime.utcnow()
        time_until_expiry = expiry_time - current_time

        # Refresh if expired or expiring within 5 minutes
        should_refresh = time_until_expiry < timedelta(minutes=5)
        logging.info(f"Token expires in {time_until_expiry.total_seconds():.0f} seconds")

    except Exception as e:
        logging.warning(f"Could not parse expires_at '{expires_at}': {e}. Checking credentials.expired")
        should_refresh = credentials.expired
else:
    # Fallback to credentials.expired if no expires_at provided
    should_refresh = credentials.expired

if should_refresh:
    logging.info("Access token expired or expiring soon, refreshing...")
    try:
        import google.auth.transport.requests
        request = google.auth.transport.requests.Request()
        credentials.refresh(request)

        # Update tokens in Secrets Manager
        new_expires_in = int((credentials.expiry - datetime.utcnow()).total_seconds()) if credentials.expiry else 3600
        update_oauth_tokens(
            user_id=user_id,
            access_token=credentials.token,
            refresh_token=credentials.refresh_token or refresh_token,
            expires_in=new_expires_in,
            region=region
        )
        logging.info("Access token refreshed and updated in Secrets Manager")

    except RefreshError as e:
        error_str = str(e)
        # Check if this is an expired/revoked refresh token
        if 'invalid_grant' in error_str and ('expired' in error_str or 'revoked' in error_str):
            logging.warning(f"Refresh token expired/revoked for user {user_id}. Clearing tokens.")
            from utils.secretsmanager_utils import delete_oauth_tokens
            try:
                delete_oauth_tokens(user_id, region)
            except Exception as delete_error:
                logging.error(f"Failed to clear expired tokens: {delete_error}")

            raise RefreshTokenExpiredError(
                "Refresh token has expired or been revoked. Please re-connect your Gmail account."
            ) from e
        else:
            raise OAuthValidationError(f"Token refresh failed: {str(e)}") from e

After a successful refresh, update_oauth_tokens writes the new access_token and updated expires_at back to Secrets Manager while preserving the existing scope and Google user info.

Account switch detection

When the iOS app sends new tokens, the backend compares the google_user_id in the incoming token against the one already stored for the user:

oauth_utils.py

def validate_google_account_consistency(
    user_id: str,
    new_google_user_id: str,
    new_google_email: str,
    region: str
) -> Dict[str, Any]:
    try:
        from utils.secretsmanager_utils import get_oauth_tokens

        try:
            existing_oauth_data = get_oauth_tokens(user_id, region)
            existing_google_user_id = existing_oauth_data.get('google_user_id')
            existing_google_email = existing_oauth_data.get('google_email')

            if existing_google_user_id and existing_google_user_id != new_google_user_id:
                return {
                    'is_account_switch': True,
                    'existing_email': existing_google_email,
                    'new_email': new_google_email,
                    'message': f"Switching from {existing_google_email} to {new_google_email}"
                }
            else:
                return {
                    'is_account_switch': False,
                    'existing_email': existing_google_email,
                    'new_email': new_google_email,
                    'message': "Same Google account or first connection"
                }

        except Exception:
            # No existing OAuth data found - first connection
            return {
                'is_account_switch': False,
                'existing_email': None,
                'new_email': new_google_email,
                'message': "First Gmail connection"
            }

    except Exception as e:
        raise OAuthValidationError(
            f"Error validating Google account consistency: {str(e)}"
        ) from e

If is_account_switch is True, the new tokens still overwrite the old ones — the check is informational. The store-tokens endpoint includes account_switch in its response so the iOS app can display a warning to the user.

Token validation flow

1. Retrieve tokens from Secrets Manager

   Lambda calls get_oauth_tokens(user_id, region), which reads the gmail/user/{user_id} secret. If expires_at is in the past, the function adds is_expired: True to the returned dict.

2. Create Gmail service (refresh if needed)

   create_gmail_service is called with the stored tokens. It checks whether the token expires within 5 minutes and refreshes proactively using the refresh token.

3. Update Secrets Manager after refresh

   If a refresh occurred, update_oauth_tokens writes the new access token and expiry timestamp back to Secrets Manager, preserving the existing Google user info and scope.

4. Use the access token

   The Gmail API service object is returned and used to search for or download emails. The service holds valid credentials for the duration of the Lambda invocation.

If a refresh token has expired or been revoked by the user, create_gmail_service raises a RefreshTokenExpiredError and deletes the stale tokens from Secrets Manager. The user must reconnect their Gmail account from the iOS app.