Firebase Auth Custom Claims for Access Control and Firestore Security Rules

Assign Firebase Auth Custom Claims in Google Cloud Functions (Python)

Create Google Cloud Functions (Python) to assign is_admin custom claims to user.

import datetimeimport loggingfrom flask import abortfrom firebase_admin import authlog = logging.getLogger(__name__)def update_user_role(request):    uid = request.args.get('uid')    if not uid:        abort(400)    if request.args.get('is_admin', type=int, default=0):        is_admin = True    else:        is_admin = False    data = {        'is_admin': is_admin    }    log.info(f"{uid}={data}")    auth.set_custom_user_claims(uid, data)    return 'OK'

Call https://us-central1-PROJECT_ID.cloudfunctions.net/update_user_role?uid=USER_ID&is_admin=1.

NOTE: Claims payload must also not be larger then 1000 characters when serialized into a JSON string.

Propagate custom claims to the client

After new claims are modified on a user via the Admin SDK, they are propagated to an authenticated user on the client side via the ID token in the following ways:

A user signs in or re-authenticates after the custom claims are modified. The ID token issued as a result will contain the latest claims.

An existing user session gets its ID token refreshed after an older token expires.

An ID token is force refreshed by calling currentUser.getIdToken(true).

Assuming a user already signin to your Android app, then you decide to assign or revoke is_admin user claims.

The ID Token will expire after 1 hour and I assume FirebaseAuth.IdTokenListener shall be called, where we could retrieve and update the latest user claims.
If user claims update is time critical (need to respond immediately), what we can do it perform an update on Firebase Realtime Database or Firestore. The client (Android) could listen and receive realtime changes/updates on these databases and perform ID token refresh via FirebaseUser.getIdToken().

NOTE: Example on how to update via firebase realtime database.

This following code update firestore upon set user claims.

import datetimeimport firebase_adminfrom firebase_admin import authfrom firebase_admin import credentialsfrom google.cloud import firestorePROJECT_ID = ...uid = ...data = {...}# cred = credentials.Certificate('PROJECT_ID-adminsdk.json')cred = credentials.ApplicationDefault()default_app = firebase_admin.initialize_app(cred, {  'projectId': PROJECT_ID})auth.set_custom_user_claims(uid, data)# update firestore to trigger client updatedb = firestore.Client()doc_ref = db.collection('user_metadata').document(uid)doc_ref.update({'token_refresh_time': datetime.datetime.now()})

NOTE: Since this cloud functions is used to assign is_admin access control to any users, you might want to secure the cloud function from unauthorized access.

Retrieve User Claims in Android (Kotlin)

NOTE: Refer Firebase Authentication on Android (Kotlin) for basic setup.

Listen to FirebaseAuth.AuthStateListener and call FirebaseUser.getIdToken()

FirebaseAuth.getInstance().addAuthStateListener { auth ->    Timber.d("FirebaseAuth.getInstance().addAuthStateListener")    val user = auth.currentUser    if (user != null) {        user.getIdToken(false).addOnSuccessListener { result ->            val isAdmin = result.claims["is_admin"] as? Boolean            Timber.d("isAdmin=$isAdmin, timestamp=${result.issuedAtTimestamp}")        }    }}

Listen to

Listen to token refresh after 1 hour

FirebaseAuth.getInstance().addAuthStateListener { auth ->    auth.addIdTokenListener(IdTokenListener {        Timber.d("IdTokenListener fired")        FirebaseAuth.getInstance().currentUser?.getIdToken(false)?.addOnSuccessListener { result ->            val isAdmin = result.claims["is_admin"] as? Boolean            Timber.d("isAdmin=$isAdmin, timestamp=${result.issuedAtTimestamp}")        }    })}

NOTE: I have yet to test this.

Listen to Firestore Updates

Listen to firestore updates fired from cloud functions (if you implement them).

private var userTokenRegistration: ListenerRegistration? = null

FirebaseAuth.getInstance().addAuthStateListener { auth ->    val user = auth.currentUser    userTokenRegistration?.remove()    if (user != null) {        user.getIdToken(false).addOnSuccessListener { result ->            val isAdmin = result.claims["is_admin"] as? Boolean            Timber.d("isAdmin=$isAdmin, timestamp=${result.issuedAtTimestamp}")        }        val db = FirebaseFirestore.getInstance()        val docRef = db.collection("user_metadata").document(user.uid)        userTokenRegistration = docRef.addSnapshotListener { doc, e ->            val timestamp = doc?.getTimestamp("token_refresh_time")            Timber.d("Token refresh snapshot received: ${App.currentUser?.timestamp} < ${timestamp?.seconds}")            if (App.currentUser?.timestamp ?: 0 < timestamp?.seconds ?: 0) {                Timber.d("Refresh required")                user.getIdToken(true).addOnSuccessListener { result ->                    val isAdmin = result.claims["is_admin"] as? Boolean                    Timber.d("isAdmin=$isAdmin, timestamp=${result.issuedAtTimestamp}")                }            }        }    }}

Firestore Security Rules

match /private_documents/{doc_id} {  allow read: if request.auth != null;  allow write: if request.auth.token.is_admin == true;}

References: