Documentation for TokenAuth and implementing endpoints for login, register, reset

[arunabhdas]

Hi @dantownsend

Have managed to get the admin section working for my app here

https://github.com/arunabhdas/unicorn-fullstack/blob/main/unicorn-fullstack-fastapi-piccolo/piccolo/README.md

Am able to get the admin up and running with your help

Since I need to use this piccolo API as a backend for my mobile app, I need to implement the /login and /register endpoints in a way that supports token auth.

I tried to look at the docs here

However, I am not able to figure out what my /login and /register endpoint implementation should look like if I would like to authenticate using the BaseUser (PiccoloUser) model class.

Could you help @dantownsend ?

Please note that my app.py currently looks as below

app = FastAPI(
    routes=[
        Route("/", HomeEndpoint),
        Mount(
            "/admin/",
            create_admin(
                tables=APP_CONFIG.table_classes + [BaseUser],
                # Required when running under HTTPS:
                # allowed_hosts=['my_site.com']
            ),
        ),
        Mount("/static/", StaticFiles(directory="static")),
    ],
)

###############################################################################

###############################################################################


###############################################################################
# Rather than defining the FastAPI endpoints by hand, we can use
# `FastAPIWrapper`, which can save us a lot of time.

FastAPIWrapper(
    "/projects",
    fastapi_app=app,
    piccolo_crud=PiccoloCRUD(Project, read_only=False),
    fastapi_kwargs=FastAPIKwargs(
        all_routes={"tags": ["Project"]},
    ),
)

FastAPIWrapper(
    "/users",
    fastapi_app=app,
    piccolo_crud=PiccoloCRUD(BaseUser, read_only=False),
    fastapi_kwargs=FastAPIKwargs(
        all_routes={"tags": ["BaseUser"]},
    ),
)


###############################################################################


###############################################################################
# Connection pool.

@app.on_event("startup")
async def open_database_connection_pool():
    try:
        engine = engine_finder()
        await engine.start_connection_pool()
    except Exception:
        print("Unable to connect to the database")


@app.on_event("shutdown")
async def close_database_connection_pool():
    try:
        engine = engine_finder()
        await engine.close_connection_pool()
    except Exception:
        print("Unable to connect to the database")

And I would like to add TokenAuth to support my mobile app login and registration.

Thanks in advance for your help 🙇

[dantownsend]

The docs aren't great for TokenAuth, but this is the endpoint you need for logging in and providing a token:

https://github.com/piccolo-orm/piccolo_api/blob/master/piccolo_api/token_auth/endpoints.py

You'll have to mount that endpoint to a public route in FastAPI.

To create the tables, add piccolo_api.token_auth.piccolo_app to installed apps in piccolo_conf.py then run the migrations using piccolo migrations forwards token_auth.

[arunabhdas]

Would you happen to have some example code for how to do add that ?

Will it be mounted in app.py?

It would be good if you had some sample code for login and register

Thanks again @dantownsend 🙇‍♂️

[dantownsend]

This is for session auth, but is pretty similar:

https://piccolo-api.readthedocs.io/en/latest/session_auth/example.html

[arunabhdas]

Thanks. Let me try that.

Also, if I have a route like /projects

FastAPIWrapper(
    "/projects",
    fastapi_app=app,
    piccolo_crud=PiccoloCRUD(Project, read_only=False),
    fastapi_kwargs=FastAPIKwargs(
        all_routes={"tags": ["Project"]},
    ),
)

How can I make that protected by requiring a token like access token? Will the access token be returned in the /login call?

Is there a decorator way to protect the /projects route?

[dantownsend]

You need a private FastAPI app, which is wrapped in token auth middleware, and then mount that to a public FastAPI app.

Make sure FastAPIWrapper.fastapi_app targets the private FastAPI app, and then it'll be wrapped in the token auth middleware.

It's something roughly like this:

from starlette.middleware.authentication import AuthenticationMiddleware

from piccolo_api.token_auth.middleware import (
    TokenAuthBackend,
    PiccoloTokenAuthProvider,
)

private_app = FastAPI()

FastAPIWrapper("/projects", fastapi_app=private_app, ...)

protected_app = AuthenticationMiddleware(
    private_app,
    backend=PiccoloTokenAuthProvider(),
)

public_app = FastAPI()
public_app.mount('/login', ...) # Mount the token login
public_app.mount('/private', protected_app)

[arunabhdas]

Was trying to implement the TokenAuth using the steps you mentioned above

This is what I have in app.py so far

import typing as t

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from piccolo_admin.endpoints import create_admin
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.routing import Mount, Route

from piccolo_api.crud.serializers import create_pydantic_model
from piccolo_api.csrf.middleware import CSRFMiddleware
from piccolo_api.openapi.endpoints import swagger_ui
from piccolo_api.token_auth.endpoints import token_login, token_logout
from piccolo_api.session_auth.middleware import SessionsAuthBackend
from piccolo_api.shared.auth.junction import AuthenticationBackendJunction
from starlette.middleware.authentication import AuthenticationMiddleware

from piccolo_api.token_auth.middleware import (
    TokenAuthBackend,
    SecretTokenAuthProvider,
    PiccoloTokenAuthProvider,
)

from piccolo_api.crud.serializers import create_pydantic_model
from piccolo_api.fastapi.endpoints import (
    FastAPIWrapper,
    PiccoloCRUD,
    FastAPIKwargs,
)
from piccolo.engine import engine_finder
from starlette.routing import Route, Mount
from starlette.staticfiles import StaticFiles

from home.endpoints import HomeEndpoint
from home.piccolo_app import APP_CONFIG
from home.tables import Project
from piccolo.apps.user.tables import BaseUser

app = FastAPI(
    routes=[
        Route("/", HomeEndpoint),
        Mount(
            "/admin/",
            create_admin(
                tables=APP_CONFIG.table_classes + [BaseUser],
                # Required when running under HTTPS:
                # allowed_hosts=['my_site.com']
            ),
        ),
        # Session Auth login:
        Mount(
            "/login/",
            token_login(redirect_to="/private/docs/"),
        ),
        Mount("/static/", StaticFiles(directory="static")),
    ],
    docs_url=None,
    redoc_url=None,
)

###############################################################################
# The private app with SessionAuth protected endpoints
private_app = FastAPI(
    routes=[
        Route("/logout/", token_logout(redirect_to="/")),
        # We use a custom Swagger docs endpoint instead of the default FastAPI
        # one, because this one supports CSRF middleware:
        Mount("/docs/", swagger_ui(schema_url="/private/openapi.json")),
    ],
    middleware=[
        Middleware(
            AuthenticationMiddleware,
            backend=PiccoloTokenAuthProvider(),
        ),
        # CSRF middleware provides additional protection for older browsers, as
        # we're using cookies.
        Middleware(CSRFMiddleware, allow_form_param=True),
    ],
    # We disable the default Swagger docs, as we have a custom one (see above).
    docs_url=None,
    redoc_url=None,
)


# Mount our private app within the main app.
app.mount("/private/", private_app)
###############################################################################


###############################################################################
# Rather than defining the FastAPI endpoints by hand, we can use
# `FastAPIWrapper`, which can save us a lot of time.

FastAPIWrapper(
    "/projects",
    fastapi_app=app,
    piccolo_crud=PiccoloCRUD(Project, read_only=False),
    fastapi_kwargs=FastAPIKwargs(
        all_routes={"tags": ["Project"]},
    ),
)

FastAPIWrapper(
    "/users",
    fastapi_app=app,
    piccolo_crud=PiccoloCRUD(BaseUser, read_only=False),
    fastapi_kwargs=FastAPIKwargs(
        all_routes={"tags": ["BaseUser"]},
    ),
)


###############################################################################


###############################################################################
# Connection pool.

@app.on_event("startup")
async def open_database_connection_pool():
    try:
        engine = engine_finder()
        await engine.start_connection_pool()
    except Exception:
        print("Unable to connect to the database")


@app.on_event("shutdown")
async def close_database_connection_pool():
    try:
        engine = engine_finder()
        await engine.close_connection_pool()
    except Exception:
        print("Unable to connect to the database")

This is based on the Session Auth code you provided and on

https://github.com/piccolo-orm/piccolo_api/blob/master/piccolo_api/token_auth/endpoints.py

However, it seems, that the token_auth/endpoints middleware does not have token_login and token_logout methods

Any ideas on what is wrong with the above code?

[dantownsend]

You don't need this:

        # CSRF middleware provides additional protection for older browsers, as
        # we're using cookies.
        Middleware(CSRFMiddleware, allow_form_param=True),

The FastAPIWrapper should be targeting private_app.

There's no logout endpoint, as it doesn't really make sense with token auth. This is used for things like mobile apps where you don't usually logout.

I wouldn't expose BaseUser with FastAPIWrapper - that should be private, and not accessible via an API, as it's used for authentication.

[arunabhdas]

Updated the code as below -

import typing as t

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from piccolo_admin.endpoints import create_admin
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.routing import Mount, Route

from piccolo_api.crud.serializers import create_pydantic_model
from piccolo_api.openapi.endpoints import swagger_ui
from piccolo_api.token_auth.endpoints import login
from piccolo_api.session_auth.middleware import SessionsAuthBackend
from piccolo_api.shared.auth.junction import AuthenticationBackendJunction
from starlette.middleware.authentication import AuthenticationMiddleware

from piccolo_api.token_auth.middleware import (
    TokenAuthBackend,
    SecretTokenAuthProvider,
    PiccoloTokenAuthProvider,
)

from piccolo_api.crud.serializers import create_pydantic_model
from piccolo_api.fastapi.endpoints import (
    FastAPIWrapper,
    PiccoloCRUD,
    FastAPIKwargs,
)
from piccolo.engine import engine_finder
from starlette.routing import Route, Mount
from starlette.staticfiles import StaticFiles

from home.endpoints import HomeEndpoint
from home.piccolo_app import APP_CONFIG
from home.tables import Project
from piccolo.apps.user.tables import BaseUser

public_app = FastAPI(
    routes=[
        Route("/", HomeEndpoint),
        Mount(
            "/admin/",
            create_admin(
                tables=APP_CONFIG.table_classes + [BaseUser],
                # Required when running under HTTPS:
                # allowed_hosts=['my_site.com']
            ),
        ),
        # Session Auth login:
        Mount(
            "/login/",
            login(redirect_to="/private/docs/"),
        ),
        Mount("/static/", StaticFiles(directory="static")),
    ],
    docs_url=None,
    redoc_url=None,
)

###############################################################################
private_app = FastAPI()

FastAPIWrapper(
    "/projects",
    fastapi_app=private_app,
    piccolo_crud=PiccoloCRUD(Project, read_only=False),
    fastapi_kwargs=FastAPIKwargs(
        all_routes={"tags": ["Project"]},
    ),
)

protected_app = AuthenticationMiddleware(
    private_app,
    backend=PiccoloTokenAuthProvider(),
)


public_app.mount('/login', login) # Mount the token login
public_app.mount('/private', protected_app)
###############################################################################


###############################################################################
FastAPIWrapper(
    "/users",
    fastapi_app=private_app,
    piccolo_crud=PiccoloCRUD(BaseUser, read_only=False),
    fastapi_kwargs=FastAPIKwargs(
        all_routes={"tags": ["BaseUser"]},
    ),
)


###############################################################################


###############################################################################
# Connection pool.

@app.on_event("startup")
async def open_database_connection_pool():
    try:
        engine = engine_finder()
        await engine.start_connection_pool()
    except Exception:
        print("Unable to connect to the database")


@app.on_event("shutdown")
async def close_database_connection_pool():
    try:
        engine = engine_finder()
        await engine.close_connection_pool()
    except Exception:
        print("Unable to connect to the database")

However, I'm still getting the error -

  File "<PATH>/piccolo/app.py", line 12, in <module>
    from piccolo_api.token_auth.endpoints import login
ImportError: cannot import name 'login' from 'piccolo_api.token_auth.endpoints' (<PATH>/piccolo/myvenv/lib/python3.10/site-packages/piccolo_api/token_auth/endpoints.py)

There doesn't seem to be a login function in
https://github.com/piccolo-orm/piccolo_api/blob/master/piccolo_api/token_auth/endpoints.py

Regarding BaseUser, I just need it accessible via the private app (protected by token)

[dantownsend]

This is the login function: https://github.com/piccolo-orm/piccolo_api/blob/193a73ed00eb33324181741b04e4f71c509afd0a/piccolo_api/token_auth/endpoints.py#L44

[sinisaos]

I apologize for intruding into this discussion, but I will try to help. @arunabhdas I think you are mixing SessionAuth and TokenAuth. Piccolo has a simple token authentication that is only suitable for mobile clients (no token refresh etc.). Updated @dantownsend example

from piccolo_api.token_auth.endpoints import TokenAuthLoginEndpoint
from piccolo_api.token_auth.middleware import (
    TokenAuthBackend,
    PiccoloTokenAuthProvider,
)
from piccolo_api.token_auth.tables import TokenAuth

public_app = FastAPI(
    routes=[
        Mount(
            "/admin/",
            create_admin(
                tables=[Manager, TokenAuth]
            ),
        ),
        Route("/login/", TokenAuthLoginEndpoint),
    ],
)

private_app = FastAPI()

protected_app = AuthenticationMiddleware(
    private_app,
    backend=TokenAuthBackend(PiccoloTokenAuthProvider()),
)

FastAPIWrapper(
    "/projects/",
    fastapi_app=private_app,
    piccolo_crud=PiccoloCRUD(Manager, read_only=False),
    fastapi_kwargs=FastAPIKwargs(
        all_routes={"tags": ["Project"]},
    ),
)

public_app.mount("/private", protected_app)

After that, you have to create a token for the user in the TokenAuth table (easiest in Piccolo Admin). Then you can log in a user with that token like this (I used curl but you can use any client)

curl -X POST -H "Content-Type: application/json" -d '{"username": "piccolo", "password": "piccolo123"}' http://localhost:8000/login/ 

and result is {"token":"your-token-from-TokenAuth-table-from-db"}
After that, you can access protected endpoints passing that token in the authorization header of request like this

curl -H "Authorization: Bearer your-token-from-TokenAuth-table-from-db" http://localhost:8000/private/projects/

Hope that helps.

@dantownsend We should change docs for PiccoloTokenAuthProvider to this

app = AuthenticationMiddleware(
    my_asgi_app,
    backend=TokenAuthBackend(PiccoloTokenAuthProvider()), 
)

because without TokenAuthBackend raises AttributeError: 'PiccoloTokenAuthProvider' object has no attribute 'authenticate'

[arunabhdas]

Thanks @sinisaos for your help on this thread.

So I did what you suggested and modified app.py as below

import typing as t

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from piccolo_admin.endpoints import create_admin
from starlette.middleware import Middleware
from starlette.routing import Mount, Route

from piccolo_api.openapi.endpoints import swagger_ui
from piccolo_api.token_auth.endpoints import TokenAuthLoginEndpoint
from piccolo_api.session_auth.middleware import SessionsAuthBackend
from piccolo_api.shared.auth.junction import AuthenticationBackendJunction
from starlette.middleware.authentication import AuthenticationMiddleware

from piccolo_api.token_auth.middleware import (
    TokenAuthBackend,
    PiccoloTokenAuthProvider,
)
from piccolo_api.token_auth.tables import TokenAuth

from piccolo_api.crud.serializers import create_pydantic_model
from piccolo_api.fastapi.endpoints import (
    FastAPIWrapper,
    PiccoloCRUD,
    FastAPIKwargs,
)
from piccolo.engine import engine_finder
from starlette.routing import Route, Mount
from starlette.staticfiles import StaticFiles

from home.endpoints import HomeEndpoint
from home.piccolo_app import APP_CONFIG
from home.tables import Project
from piccolo.apps.user.tables import BaseUser

public_app = FastAPI(
    routes=[
        Route("/", HomeEndpoint),
        Mount(
            "/admin/",
            create_admin(
                tables=APP_CONFIG.table_classes + [BaseUser, TokenAuth],
                # Required when running under HTTPS:
                # allowed_hosts=['my_site.com']
            ),
        ),
        # Session Auth login:
        Route("/login/", TokenAuthLoginEndpoint),
        Mount("/static/", StaticFiles(directory="static")),
    ],
)

###############################################################################
private_app = FastAPI()


protected_app = AuthenticationMiddleware(
    private_app,
    backend=TokenAuthBackend(PiccoloTokenAuthProvider()),
)

FastAPIWrapper(
    "/projects",
    fastapi_app=private_app,
    piccolo_crud=PiccoloCRUD(Project, read_only=False),
    fastapi_kwargs=FastAPIKwargs(
        all_routes={"tags": ["Project"]},
    ),
)

public_app.mount('/private', protected_app)

###############################################################################
FastAPIWrapper(
    "/users",
    fastapi_app=private_app,
    piccolo_crud=PiccoloCRUD(BaseUser, read_only=False),
    fastapi_kwargs=FastAPIKwargs(
        all_routes={"tags": ["BaseUser"]},
    ),
)


###############################################################################


###############################################################################
# Connection pool.

@public_app.on_event("startup")
async def open_database_connection_pool():
    try:
        engine = engine_finder()
        await engine.start_connection_pool()
    except Exception:
        print("Unable to connect to the database")


@public_app.on_event("shutdown")
async def close_database_connection_pool():
    try:
        engine = engine_finder()
        await engine.close_connection_pool()
    except Exception:
        print("Unable to connect to the database")

Modified main.py to run public_app as below

if __name__ == "__main__":
    import uvicorn

    uvicorn.run("app:public_app", reload=True)

Now when I run using python main.py I get the TokenAuth table show up in my admin as below

However, when I try to add a token for user admin, I get the below error -

And the error in the logs is

    statement = await self._protocol.prepare(
  File "asyncpg/protocol/protocol.pyx", line 168, in prepare
asyncpg.exceptions.UndefinedTableError: relation "token_auth" does not exist

So I'm wondering, do I have to create a relation between BaseUser and TokenAuth and do a migration?

Not sure what I'm doing wrong.

I checked the migrations also

==> piccolo migrations check
Listing migrations ...
APP NAME             | MIGRATION_ID               | DESCRIPTION                              | RAN
home                 | 2023-03-20T05:53:23:759108 |                                          | True
home                 | 2023-03-20T06:14:09:303206 |                                          | True
home                 | 2023-03-21T10:12:08:736195 |                                          | True
session_auth         | 2019-11-12T20:47:17        | -                                        | True
user                 | 2019-11-14T21:52:21        | -                                        | True
user                 | 2020-06-11T21:38:55        | -                                        | True
user                 | 2021-04-30T16:14:15        | -                                        | True

Updated source with the above changes here
https://github.com/arunabhdas/unicorn-fullstack/tree/main/unicorn-fullstack-fastapi-piccolo/piccolo

Your help and inputs are greatly appreciated @dantownsend and @sinisaos 🙏

[arunabhdas]

Just a quick update on this thread and thanks to @sinisaos and @dantownsend for your help and patience.

It seems I had to do the additional step after declaring the private and protected routes -

To create the tables, add piccolo_api.token_auth.piccolo_app to installed apps in piccolo_conf.py then run the migrations using piccolo migrations forwards token_auth.

After doing the above step, things work correctly as @sinisaos described !!

However, it seems my API docs endpoint at /docs has gone away -

How can I bring back the FastAPI doc at http://localhost:8000/docs?

Have a few more questions on this same thread -

• It seems that the token auth that is coming out of the box is like an API_KEY (so that it can be used by any client that presents the Bearer token). It would be good to have the ability access and refresh tokens on a per-user and per-endpoint level

• What is the CURL / POST request equivalent of the token add that one does from admin URL http://localhost:8000/admin/#/token_auth/add ? The reason this is important is that the mobile app would have to do this if one needed to restrict tokens to a per-user level?

• What if I need to implement Authorization on a per-user basis, i.e. have the user present JWT access token for the /private/projects endpoint?

• If I needed a /logout functionality endpoint for my mobile app, would I just NULL out the token auth value for the user?

• How would /register functionality work in this scenario for the mobile app?

• Also, would there be a way to implement reset_password and forgot_password functionality in this Token Auth paradigm

• It would be nice if there was JWT documentation also, since refresh token is important for mobile apps that need to logout users after inactivity (for more security, etc)

Thanks again for all the support 🙏

[sinisaos]

@arunabhdas Sorry, but you're mixing up session and token authentication again. As far as I know you can't send a GET request with an authorization header directly from the browser. You can access all FastAPIWrapper endpoints using curl like this

curl -H "Authorization: Bearer your-token-from-TokenAuth-table-from-db" http://localhost:8000/private/projects/
curl -H "Authorization: Bearer your-token-from-TokenAuth-table-from-db" http://localhost:8000/private/projects/1/ # single record 
curl -H "Authorization: Bearer your-token-from-TokenAuth-table-from-db" http://localhost:8000/private/projects/count/ # count records
etc.

Maybe you can take a look at this example how to secure certain endpoints with OAuth. There are other examples in my repo of how to use Piccolo with a separate frontend application. Maybe you can find something useful.

[dantownsend]

You can send a header, but you have to install a browser extension. This is what I usually use for Chrome:

https://chrome.google.com/webstore/detail/modheader-modify-http-hea/idgpnmonknjnojddfkpgkljpfnnfcklj

[arunabhdas]

I was using Postman to test those endpoints and pass the Bearer token.

Was still hoping for more detailed docs for JWT auth if you have time since my use-case is to have refresh and access tokens for the mobile user (that can be made to expire after logout / timeout)

[sinisaos]

@arunabhdas The implementation of Piccolo API JWT auth is similar to TokenAuth. Here is an example

from datetime import timedelta
from piccolo_api.jwt_auth.endpoints import jwt_login
from piccolo_api.jwt_auth.middleware import JWTMiddleware,  JWTBlacklist

public_app = FastAPI(
    routes=[
        Mount(
            "/admin/",
            create_admin(
                tables=[Manager]
            ),
        ),
        Route(path="/login/", endpoint=jwt_login(
                secret="mysecret123",
                expiry=timedelta(minutes=10), # default is 1 day
            ),
    ],
)


BLACKLISTED_TOKENS = []

class MyBlacklist(JWTBlacklist):
    async def in_blacklist(self, token: str) -> bool:
        return token in BLACKLISTED_TOKENS

private_app = FastAPI()

protected_app = JWTMiddleware(
    private_app,
    auth_table=BaseUser,
    secret="mysecret123",
)

FastAPIWrapper(
    "/projects/",
    fastapi_app=private_app,
    piccolo_crud=PiccoloCRUD(Manager, read_only=False),
    fastapi_kwargs=FastAPIKwargs(
        all_routes={"tags": ["Managers"]},
    ),
)

public_app.mount("/private", protected_app)

@private_app.get("/logout/")
async def logout(request: Request) -> None:
    BLACKLISTED_TOKENS.append(request.headers.get("authorization")[7:])

Usage is same as TokenAuth. As for the refresh token Piccolo API has no endpoint for that. I don't know anything about mobile applications but there must be a way to check JWT token expiration like in web frontend applications. So you can try to make a request to login endpoint and get a new token before previous JWT expires.
To logout, create your own logout endpoint and blacklist the token in that endpoint, after which the JWT token will no longer be valid and the user will have to log in again. I have updated the example above.

[arunabhdas]

Thanks @sinisaos 🙏 Appreciate all the help 💯

[sinisaos]

@arunabhdas No problem. Glad I can help. I think it should work for login and logout endpoint and you just need to check the JWT token expiration in the mobile app and for example 5 seconds before the JWT token expires call the login endpoint again to get a new token for the user and with that keep your user logged in.

[dantownsend]

The docs for JWT and Token Auth have now been updated.