Initialization

Sanic JWT operates under the hood by creating a Blueprint, and attaching a few routes to your application. This is accomplished with the Initialize class.

from sanic_jwt import Initialize
from sanic import Sanic

async def authenticate(request):
    return dict(user_id='some_id')

app = Sanic()
Initialize(app, authenticate=authenticate)

Concept

Sanic JWT is a user authentication system that does not require the developer to settle on any single user management system. This part is left up to the developer. Therefore, you (as the developer) are left with the responsibility of telling Sanic JWT how to tie into your user management system.

The Initialize class

This is the gateway worker into Sanic JWT. When initialized, it allows you to pass run time configurations to it, and gives you a window into customizing how the module will work for you. There are five main parts to it when initializing:

1. the instance of your Sanic app or a Sanic blueprint
REQUIRED
2. handler methods
one of which (the authenticate handler) is REQUIRED
3. any runtime configurations you want to make
4. custom view classes for adding your own authentication endpoints
5. component overrides

Instance

Most web applications need authentication. With Sanic JWT, all you do is create your Sanic app, and then tell Sanic JWT.

from sanic_jwt import Initialize
from sanic import Sanic

app = Sanic()
Initialize(app, authenticate=lambda: True)

You can now go ahead and protect any route (whether on a blueprint or not).

from sanic_jwt import protected
from sanic.response import json

...

@app.route("/")
@protected()
async def test(request):
    return json({ "protected": True })

What if we ONLY want the authentication on some subset of our web application? Say, a Blueprint. Not a problem. Just initialize on the blueprint instance and continue as normal.

from sanic_jwt import Initialize
from sanic import Sanic, Blueprint

app = Sanic()
bp = Blueprint('my_blueprint')
Initialize(bp, authenticate=lambda: True)
app.blueprint(bp)

Warning

If you are initializing on a blueprint, be careful of the ordering of app.blueprint() and Initialize. Putting them in the wrong order will cause the authentication endpoints to not properly attach.

Note

If you decide to initialize more than one instance of Sanic JWT (on multiple blueprints, for example), then an access token generated by one will be acceptable on ALL your instances unless they have different a secret. You can learn more about how to set that in Configuration.

Under the hood, Sanic JWT creates its own Blueprint for holding all of the Endpoints and Responses. If you decide to use your own blueprint (and by all means, feel free to do so!), just know that Sanic JWT will not create its own. When this happens, Sanic JWT instead will attach to the blueprint that you passed to it.

This is a very powerful tool that allows you to really gain some granularity in your applications’ authentication systems.

async def authenticate(request, *args, **kwargs):
    return get_my_user()

app = Sanic()
bp1 = Blueprint('my_blueprint_1')
bp2 = Blueprint('my_blueprint_2')

Initialize(app, authenticate=authenticate)
Initialize(bp1, authenticate=authenticate, access_token_name='mytoken')
Initialize(bp2, authenticate=authenticate, access_token_name='yourtoken')

In the above example, I now have three independent instances of Sanic JWT running side by side. Each is isolated to its own environment, and can have its own set of Configuration.

Handlers

There is a group of methods that Sanic JWT uses to hook into your application code. This is how it is able to live alongside your application and seemlessly plug in.

Each handler can be either a method or an awaitable. You decide.

# This works
async def authenticate(request, *args, **kwargs):
    ...

# And so does this
def authenticate(request, *args, **kwargs):
    ...

1. authenticate - Required

Purpose: Just like Django’s authenticate method, this is responsible for taking a given request and deciding whether or not there is a valid user to be authenticated. If yes, it MUST return:

  • a dict with a user_id key, or
  • an instance with an id and to_dict property.

By default, it looks for the id on the user_id property of a user instance. However, you can change that to another property.

If your user should not be authenticated, then you should raise an exception, preferably AuthenticationFailed. Please do not just return None. If you do, you will likely get a 500 error.

Example:

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 = await User.get(username=username)
    if user is None:
        raise exceptions.AuthenticationFailed("User not found.")

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

    return user

Initialize(app, authenticate)

2. store_refresh_token *

Purpose: It is a handler to persist a refresh token to disk. See refresh tokens for more information. Sanic JWT create the refresh token, but you get to decide how it is stored.

Example:

async def store_refresh_token(user_id, refresh_token, *args, **kwargs):
    key = 'refresh_token_{user_id}'.format(user_id=user_id)
    await aredis.set(key, refresh_token)

Initialize(
    app,
    authenticate=lambda: True,
    store_refresh_token=store_refresh_token)

Warning

* This parameter is not required. However, if you decide to enable refresh tokens (by setting refresh_token_enabled=True in your configurations) then the application will raise a RefreshTokenNotImplemented exception if you forget to implement this.

3. retrieve_refresh_token *

Purpose: It is a handler to retrieve refresh token from disk. See refresh tokens for more information. Sanic JWT created the refresh token. You stored it. Now Sanic JWT wants it back, it is your job to retrieve it.

Example:

async def retrieve_refresh_token(user_id, *args, **kwargs):
    key = 'refresh_token_{user_id}'.format(user_id=user_id)
    return await aredis.get(key)

Initialize(
    app,
    authenticate=lambda: True,
    retrieve_refresh_token=retrieve_refresh_token)

Warning

* This parameter is not required. However, if you decide to enable refresh tokens (by setting refresh_token_enabled=True in your configurations) then the application will raise a RefreshTokenNotImplemented exception if you forget to implement this.

4. retrieve_user

Purpose: It is a handler to retrieve a user object from your application. It is used to return the user object in the /auth/me endpoint, and also the @inject_user decorator that you will learn about later.

It should return:

- a dict, or
- an instance of some object with a to_dict method, or
- None

As we said before, you are deciding on the user management system. Sanic JWT is acting as the gatekeeper. But, inherently tied in are a number of use cases where it would be convenient to get your user object. This is how you do it.

Example:

class User:
    ...

    def to_dict(self):
        properties = ['user_id', 'username', 'email', 'verified']
        return {prop: getattr(self, prop, None) for prop in properties}

async def retrieve_user(request, payload, *args, **kwargs):
    if payload:
        user_id = payload.get('user_id', None)
        user = await User.get(user_id=user_id)
        return user
    else:
        return None

Initialize(
    app,
    authenticate=lambda: True,
    retrieve_user=retrieve_user)

You should now have an endpoint at /auth/me that will return a serialized form of your currently authenticated user.

{
    "me": {
        "user_id": "4",
        "username": "joe",
        "email": "joe@joemail.com",
        "verified": true
    }
}

5. add_scopes_to_payload *

Purpose: It is a handler to add scopes to an access token. See Scopes for more information.

Scoping is a long discussion by itself. In short, it is a highly powerful tool to help with providing permissioning to your application. It is your job to add these scopes (if you want them) to the JWT. Then, you can specifiy which scopes are required on specific endpoints.

For now, all you need to do is return a list of one or more strings.

Example:

async def add_scopes_to_payload(user):
    return await user.get_scopes()

Initialize(
    app,
    authenticate=lambda: True,
    add_scopes_to_payload=add_scopes_to_payload)

6. override_scope_validator *

Purpose: It is a handler to override the default scope validation. See Scopes for more information.

This could be useful if you decide to bake some additional logic into your scopes. At its most simplified level, Sanic JWT looks at scopes and compares fruit:apples to fruit:apples. What if sometimes fruit:oranges should be accepted? You have the ability to code that override and make your own decision.

Note

Above, we said “Each of them can be either a method or an awaitable. You decide.” What we forgot to mention was that override_scope_validator needs to be a regular callable and not an awaitable.

No async programming here. Sorry for the confusion.

Example:

def my_scope_override(is_valid,
    required,
    user_scopes,
    require_all_actions,
    *args,
    **kwargs):
    return False

Initialize(
    app,
    authenticate=lambda: True,
    override_scope_validator=my_scope_override)

7. destructure_scopes

Purpose: It is a handler that allows you to manipulate and handle the scopes before they are validated.

Sometimes, you may find the need to manipulate the scopes before they are validated against the protected resource. In this case, feel free to make changes:

Example:

async def my_destructure_scopes(scopes, *args, **kwargs):
    return scopes.replace("|", ":")

@app.route("/protected/nonstandardscopes")
@scoped("foo|bar")
def scoped_sync_route(request):
    return json({"nonstandardscopes": True})

Initialize(
    app,
    authenticate=lambda: True,
    destructure_scopes=my_destructure_scopes)

8. extend_payload

Purpose: It is a handler to allow the developer to modify the payload by adding additional claims to it before it is bundled up and packaged inside a JWT.

One of the most powerful concepts of the JWT is that you are able to pass data (aka claims) inside its payload for use by a client application, and reuse when that JWT is being returned for verification. It is simply a method that takes the existing payload and returns it (with your brilliant modifications, of course)

Example:

async def my_extender(payload, user):
    username = user.to_dict().get("username")
    payload.update({"username": username})
    return payload

Initialize(
    app,
    authenticate=lambda: True,
    extend_payload=my_extender)

Runtime Configuration

There are several ways to configure the settings for Sanic JWT. One of the easiest is to simply pass the configurations as keyword objects on Initialize.

Initialize(
    app,
    access_token_name='mytoken',
    cookie_access_token_name='mytoken',
    cookie_set=True,
    user_id='id',
    claim_iat=True,
    cookie_domain='example.com',)

Additional Views

Sometimes you may need to add some endpoints to the authentication system. When this need arises, create a class based view, and map it as a tuple with the path and handler.

As an example, perhaps you would like to create a “passwordless” login. You could create a form that sends a POST with a user’s email address to a MagicLoginHandler. That handler sends out an email with a link to your /auth endpoint that makes sure the link came from the email.

from sanic_jwt import BaseEndpoint

class MagicLoginHandler(BaseEndpoint):
    async def options(self, request):
        return response.text('', status=204)

    async def post(self, request):
        helper = MyCustomUserAuthHelper(app, request)
        token = helper.get_make_me_a_magic_token()
        helper.send_magic_token_to_user_email()

        # Persist the token
        key = f'magic-token-{token}'
        await app.redis.set(key, helper.user.uuid)

        response = {
            'magic-token': token
        }
        return json(response)

def check_magic_token(request):
    token = request.json.get('magic_token', '')
    key = f'magic-token-{token}'

    retrieval = await request.app.redis.get(key)
    if retrieval is None:
        raise Exception('Token expired or invalid')
    retrieval = str(retrieval)

    user = User.get(uuid=retrieval)

    return user

Initialize(
    app,
    authenticate=check_magic_token,
    class_views=[
        # The path will be relative to the url prefix (which defaults to /auth)
        ('/magic-login', MagicLoginHandler)
    ])

Note

Your class based views will probably also need to handle preflight requests, so do not forget to add an options response.

async def options(self, request):
    return response.text('', status=204)

Component Overrides

There are three components that are used under the hood that you can subclass and control:

Simply import, modify, and attach.

from sanic_jwt import Authentication, Configuration, Responses, Initialize

class MyAuthentication(Authentication):
    pass

class MyConfiguration(Configuration):
    pass

class MyResponses(Responses):
    pass

Initialize(
    app,
    authentication_class=MyAuthentication,
    configuration_class=MyConfiguration,
    responses_class=MyResponses,)

The initialize method

The old method for initializing Sanic JWT was to do so with the initialize method. It still works, and is in fact now just a wrapper for the Initialize class. However, it is recommended that you use the class because it is more explicit that you are declaring a new instance. And, even though there are no plans (as of June 2018) to depracate this, some day it likely will be.