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:
authenticate
handler) is REQUIREDInstance¶
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, app=app, 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 auser_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:
dict
, orto_dict
method, orNone
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. retrieve_user_secret
*¶
Purpose: It is a handler to retrieve or generate a unique secret for a user. All JWTs are encoded using a secret, and this allows for you to provide a unique secret for encoding per user. You must also initialize with user_secret_enabled
.
Example:
async def retrieve_user_secret(user_id):
return f"my-super-safe-dynamically-generated-secret|{user_id}"
Initialize(
app,
authenticate=lambda: True,
retrieve_user_secret=retrieve_user_secret)
Warning
* This parameter is not required. However, if you decide to enable user secrets (by setting user_secret_enabled=True
in your configurations) then the application will raise a UserSecretNotImplemented
exception if you forget to implement this.
6. 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)
7. 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)
8. 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)
9. 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:
Authentication
- for more advanced usage, see source code, or ask a questionConfiguration
- see Configuration for more informationResponses
- see Endpoints and Responses for more information
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.