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 ¯\_(ツ)_/¯