verify_firestore_jwt.py

verify_firestore_jwt.py

FIREBASE_JWKS_PUBLIC_KEYS = 'https://www.googleapis.com/robot/v1/metadata/x509/[email protected]'

keys_cache = {'keys': None, 'last_fetched': 0.0}


async def fetch_public_keys() -> dict[str, str]:
    """Fetch Firebase public keys from URL, with caching for 1 hour."""

    # Check if the cache is older than 1 hour (3600 seconds)
    if time.time() - keys_cache['last_fetched'] > 3600 or keys_cache['keys'] is None:
        # Fetch new public keys from the JWKS URL
        with request.urlopen(FIREBASE_JWKS_PUBLIC_KEYS) as response:
            keys_cache['keys'] = json.loads(response.read())
            keys_cache['last_fetched'] = time.time()  # Update last fetched time

    return cast(dict, keys_cache['keys'])


async def find_public_key(firebase_token: str) -> str:
    public_keys = await fetch_public_keys()

    # Decode the JWT header to extract the kid (Key ID)
    headers = jwt.get_unverified_header(firebase_token)
    kid = headers['kid']

    # Get the correct public key using the 'kid'
    if kid not in public_keys:
        raise Exception('Public key not found')

    return public_keys[kid]


async def decode_firebase_token(firebase_jwt):
    public_key = find_public_key(firebase_jwt)

    # Now verify and decode the JWT
    try:
        decoded_token = jwt.decode(
            firebase_jwt,
            public_key,
            algorithms=['RS256'],
            audience='your-firebase-project-id',  # This should match your Firebase project ID
            issuer='https://securetoken.google.com/your-firebase-project-id',  # Replace with your Firebase project ID
        )
        return decoded_token
    except InvalidTokenError as e:
        raise Exception(f'Token verification failed: {e}')