sanic-jwt

This example shows how to integrate sanic-jwt with sanic-boom, using a layered middleware, to determine that from a specific route onwards, every user will have to be authenticated - so you won’t need to declare anything else in your endpoints to protect them.

It also figures the usage of a component that is solely required by another component, as a simple example.

from sanic.response import text
from sanic_jwt import Initialize, exceptions

from sanic_boom import Component, ComponentCache, SanicBoom

# --------------------------------------------------------------------------- #
# sanic-jwt related code
# --------------------------------------------------------------------------- #


class User(object):
    def __init__(self, id, username, password):
        self.user_id = id
        self.username = username
        self.password = password

    def __str__(self):
        return "User(id='{}')".format(self.id)

    def to_dict(self):
        return {"user_id": self.user_id, "username": self.username}


users = [
    User(1, "user1", "abcxyz"),
    User(2, "user2", "abcxyz"),
    User(3, "user3", "abcxyz"),
    User(4, "user4", "abcxyz"),
]

username_table = {u.username: u for u in users}
userid_table = {u.user_id: u for u in users}


async def authenticate(request, *args, **kwargs):
    username = request.json.get("username", None)
    password = request.json.get("password", None)

    if not username or not password:
        raise exceptions.AuthenticationFailed("Missing username or password.")

    user = username_table.get(username, None)
    if user is None:
        raise exceptions.AuthenticationFailed("User not found.")

    if password != user.password:
        raise exceptions.AuthenticationFailed("Password is incorrect.")

    return user


# --------------------------------------------------------------------------- #
# sanic-boom related code
# --------------------------------------------------------------------------- #


class AuthComponent(Component):  # for shorthand
    def resolve(self, param) -> bool:
        return param.name == "auth"

    async def get(self, request, param):
        return request.app.auth


class JWTComponent(Component):
    def resolve(self, param) -> bool:
        return param.name == "jwt_user_id"

    async def get(self, request, param, auth):  # component inter-dependency
        is_valid, status, reasons = auth._check_authentication(
            request, None, None
        )
        if not is_valid:
            raise exceptions.Unauthorized(reasons, status_code=status)
        return auth.extract_user_id(request)

    def get_cache_lifecycle(self):
        return ComponentCache.REQUEST


app = SanicBoom(__name__)
sanicjwt = Initialize(app, authenticate=authenticate)

# adding components
app.add_component(AuthComponent)
app.add_component(JWTComponent)


@app.middleware(uri="/restricted")
async def restricted_middleware(jwt_user_id):
    pass  # this is really it!


@app.get("/restricted/foo")
async def restricted_handler(jwt_user_id):
    return text(jwt_user_id)


@app.get("/restricted/bar")
async def another_restricted_handler():
    return text("this is restricted!")


@app.get("/")
async def unrestricted_handler():
    return text("OK")


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, workers=1)

Testing

Calling the root endpoint:

$ curl -v http://127.0.0.1:8000/
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8000 (#0)
> GET / HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/7.61.1
> Accept: */*
>
< HTTP/1.1 200 OK
< Connection: keep-alive
< Keep-Alive: 5
< Content-Length: 2
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host 127.0.0.1 left intact
OK

So far so good, this was the expected result. Now, let’s try to access a restricted endpoint (by the code, any endpoint starting with /restricted/ will have authentication required), without a token:

$ curl -v http://127.0.0.1:8000/restricted/foo
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8000 (#0)
> GET /restricted/foo HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/7.61.1
> Accept: */*
>
< HTTP/1.1 401 Unauthorized
< Connection: keep-alive
< Keep-Alive: 5
< Content-Length: 76
< Content-Type: application/json
<
* Connection #0 to host 127.0.0.1 left intact
{"reasons":["Authorization header not present."],"exception":"Unauthorized"}

But, but … Is that black magic? Actually, no. This is really straightforward. Now, let’s finally authenticate a user:

$ curl -v http://127.0.0.1:8000/auth -d '{"username":"user1","password":"abcxyz"}'
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8000 (#0)
> POST /auth HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/7.61.1
> Accept: */*
> Content-Length: 40
> Content-Type: application/x-www-form-urlencoded
>
* upload completely sent off: 40 out of 40 bytes
< HTTP/1.1 200 OK
< Connection: keep-alive
< Keep-Alive: 5
< Content-Length: 140
< Content-Type: application/json
<
* Connection #0 to host 127.0.0.1 left intact
{"access_token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE1Mzk4OTUxODh9.FF2zld_RM01nhkFLVPIa6SRg6PZkGCCW6rFjrpTkc0o"}

Great, we have an access_token! Let’s try to access our restricted endpoint again:

$ curl -v http://127.0.0.1:8000/restricted/foo -H "Authorization: Bearer <token>"
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8000 (#0)
> GET /restricted/foo HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/7.61.1
> Accept: */*
> Authorization: Bearer <token>
>
< HTTP/1.1 200 OK
< Connection: keep-alive
< Keep-Alive: 5
< Content-Length: 1
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host 127.0.0.1 left intact
1

And our return is 1, as is the user_id parameter for user1. You can try to get a token for each user (user2, user3 and user4) and execute this last endpoint. The result should be the number of the user.

And what about the layered middleware? You just need to implement one argument in a middleware and all endpoints starting with it will run it, and in this example, will require an authenticated user. Another example? Sure!

$ curl -v http://127.0.0.1:8000/restricted/bar -H "Authorization: Bearer <token>"
*   Trying 127.0.0.1...
* TCP_NODELAY set
* Connected to 127.0.0.1 (127.0.0.1) port 8000 (#0)
> GET /restricted/bar HTTP/1.1
> Host: 127.0.0.1:8000
> User-Agent: curl/7.61.1
> Accept: */*
> Authorization: Bearer <token>
>
< HTTP/1.1 200 OK
< Connection: keep-alive
< Keep-Alive: 5
< Content-Length: 19
< Content-Type: text/plain; charset=utf-8
<
* Connection #0 to host 127.0.0.1 left intact
this is restricted!

Well, a lot of boilerplate code has just vanished ¯\_(ツ)_/¯