Python with Starlette

Cover Page

Back-end Page

To verify Google’s ID Token, first install the Google API python client in your Python’s virtual environment:

server$ uv add google-api-python-client

handlers.py

Change to your chatterd folder and edit handlers.py:

server$ cd ~/reactive/chatterd
server$ vi handlers.py

Add the following libraries to the import block at the top of handlers.py:

from google.auth.transport import requests
from google.oauth2 import id_token
import hashlib, time

Next add the following two new structs:

@dataclass
class AuthChatt:
    chatterID: str
    message: str

@dataclass
class Chatter:
    clientID: str
    idToken: str

Then add the adduser() function:

async def adduser(request):
    try:
        chatter = Chatter(**(await request.json()))
    except Exception as err:
        print(f'{err=}')
        return JSONResponse('Unprocessable entity', status_code=422)

    now = time.time()                  # secs since epoch (1/1/70, 00:00:00 UTC)

    try:
        idinfo = id_token.verify_oauth2_token(chatter.idToken, requests.Request(), chatter.clientID, clock_skew_in_seconds = 50)
    except ValueError as err:
        return JSONResponse('Unauthroized', status_code=401)

    try:
        username = idinfo['name']
    except:
        username = "Profile NA"

    # computer chatterID

The function adduser() first receives a POST request containing a clientID and idToken from the front end. It uses Google’s idtoken package to verify the user’s idToken, passing along the clientID as required by Google. The verification process checks that idToken hasn’t expired and is valid. If the token is invalid or has expired, a 401, “Unauthorized” HTTP error is returned to the front end. If idToken is verified, the user’s name registered with the idToken is returned to the front end.

Next, the function computes a chatterID for the new user. The chatterID is computed as a SHA256 one-way hash of the idToken, a server’s secret, and the current time stamp. The function also assigns a lifetime to the chatterID. The lifetime is set to be no more than the remaining lifetime of the idToken, so always less than the total expected lifetime of the idToken. During the lifetime of a chatterID, the user does not need to check the freshness of their idToken with Google. Replace // compute chatterID with:

    # Compute chatterID
    backendSecret = "ifyougiveamouse"   # or server's private key
    nonce = str(now)
    hashable = chatter.idToken + backendSecret + nonce
    chatterID = hashlib.sha256(hashable.strip().encode('utf-8')).hexdigest()

    lifetime = min(int(idinfo['exp']-now)+1, 300) # secs, up to 1800, idToken lifetime

    # add to database

During testing, setting lifetime to 1 minute allows faster triggering of the various use cases. Longer lifetime leads to less frequent prompting for user to sign in again, but also leaves open a larger window of vulnerability.

The chatterID, the user’s registered name obtained from the idToken, and the chatterID’s lifetime are then entered into the chatters table. At the same time, we take this oppotunity to do some house keeping to remove all expired chatterIDs from the database. Replace // add to database with:

    try:
        async with main.server.pool.connection() as connection:
            async with connection.cursor() as cursor:
                await cursor.execute('DELETE FROM chatters WHERE %s > expiration;', (now, ))

                await cursor.execute('INSERT INTO chatters (chatterid, username, expiration) VALUES '
                                 '(%s, %s, %s);', (chatterID, username, now+lifetime))
        return JSONResponse({'username': username, 'chatterID': chatterID, 'lifetime': lifetime})
    except Exception as err:
        print(f'{err=}')
        return JSONResponse(f'{type(err).__name__}: {str(err)}', status_code = 500)

The registered username, the newly created chatterID and its lifetime are returned to the user as a JSON object.

postauth()

We now add postauth(), which is a modified postchatt(), to your handlers.py. To post a chatt, the front end sends a POST request containing the user’s chatterID and message. The function postauth() retrieves the record matching chatterID from the chatters table. If chatterID is not found in the chatters table, or if the chatterID has expired, it returns a 401, “Unauthorized” HTTP error. Otherwise, the registered username corresponding to chatterID is retrieved from the table. Note: chatterIDs are unique in the chatters table.

async def postauth(request):
    try:
        # loading raw json (not form-encoded)
        chatt = AuthChatt(**(await request.json()))
    except Exception as err:
        print(f'{err=}')
        return JSONResponse('Unprocessable entity', status_code=422)

    try:
        async with main.server.pool.connection() as connection:
            async with connection.cursor() as cursor:
                await cursor.execute('SELECT username, expiration FROM chatters WHERE chatterID = %s;',
                                 (chatt.chatterID,))

                row = await cursor.fetchone()
                now = time.time()
                if row is None or now > row[1]:
                    # return an error if there is no chatter with that ID
                    return JSONResponse('Unauthorized', status_code=401)

                # insert chatt
                

        return JSONResponse({})
    except Exception as err:
        print(f'{err=}')
        return JSONResponse(f'{type(err).__name__}: {str(err)}', status_code = 500)

Insert the chatt into the chatts table under the retrieved username. Replace # insert chatt with:

                await cursor.execute('INSERT INTO chatts (name, message, id) VALUES (%s, %s, gen_random_uuid());',
                                 (row[0], chatt.message))

We will be using the original getchatts() from the chatter lab without modification.

We’re done with handlers.py. Save and exit the file.

main.py

Edit the file main.py:

server$ vi main.py

Find the routes array and add the following new routes for the new API endpoints /adduser and /postauth:

    Route('/postauth/', handlers.postauth, methods=['POST']),
    Route('/adduser/', handlers.adduser, methods=['POST']),

We’re done with main.py. Save and exit the file.

Test run

To test run your server, launch it from the command line:

server$ sudo su
# You are now root, note the command-line prompt changed from '$' or '%' to '#'.
# You can do a lot of harm with all of root's privileges, so be very careful what you do.
server# source .venv/bin/activate
(chattterd) ubuntu@server:/home/ubuntu/reactive/chatterd# granian --host 0.0.0.0 --port 443 --interface asgi --ssl-certificate /home/ubuntu/reactive/chatterd.crt --ssl-keyfile /home/ubuntu/reactive/chatterd.key --access-log --workers-kill-timeout 1 main:server
# Hit ^C to end the test
(chattterd) ubuntu@server:/home/ubuntu/reactive/chatterd# exit
# So that you're no longer root.
server$

The cover back-end spec provides instructions for Testing Signin.

Reference

Prepared by Benjamin Brengman, Ollie Elmgren, Wendan Jiang, Alexander Wu, and Sugih Jamin Last updated March 13th, 2026