Flask JSON Authentication
I mentioned earlier in this blog about how Auth is much harder than it should be between Angular and Flask. I decided recently to move this site to Kubernetes so that I could have a professional environment to work on deploy scripts and the like. In the process of trying to tweak stuff, I discovered that a lot of my backend site just didnt work anymore. I mean, sure, it had been a couple years, and I have learned a lot in that time- but it shouldn't have been that bad. Being me, I took this as an opportunity to redo the site from scratch. I had wanted to incorporate my newer approach to api modeling, get rid of the old Jinja based code altogether, and add testing anyway. So I did- and hit the brick wall of Auth all over again.
The libraries that try to solve the problem, like Flask-Security, seem to be... inflexible? I am never one to complain about an open source library, but it was surprising that code I would have thought is at the core of the Flask ecosystem has gotten so little love.
I wound up writing the Auth part of the app built on flask-jwt-extended. This library comes across as a fringe thing, but the code and documentation are solid, although they are pretty high level. Actually applying it to a working app took more understanding then I started with. As a basic point, we handle auth with a JWT web token. When someone hits the login endpoint, if they have a valid password, we send them a cryptographically signed token. All of our other endpoints expect that token to be in the header of requests, and verify that the signature was created with the apps private key. If so, it trusts the permissions encoded in the token.
For this to work we need-
- User and Role models, with password verification + hashing.
- The endpoint to create tokens
- Code for the rest of the app to verify the tokens.
Lets look at some code-
from datetime import datetime
from bcrypt import gensalt, hashpw
from flask import current_app
from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String, Table
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import backref, relationship
from devblog.db import Base
roles_user = Table(
"roles_user",
Base.metadata,
Column("user_id", Integer, ForeignKey("user.id"), primary_key=True),
Column("role_id", Integer, ForeignKey("role.id"), primary_key=True),
)
class Role(Base):
__tablename__ = "role"
id = Column(Integer, primary_key=True)
name = Column(String(length=80), unique=True)
description = Column(String(length=255))
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True)
alias = Column(String(255))
email = Column(String(255))
hashed_password = Column("password", String(255))
active = Column(Boolean(), default=True)
confirmed_at = Column(DateTime(), default=datetime.utcnow)
roles: relationship = relationship(
Role, secondary=roles_user, backref=backref("users", lazy="dynamic")
)
@hybrid_property
def password(self):
return self.hashed_password
@password.setter # type: ignore
def password(self, value):
rounds = current_app.config.get("BCRYPT_ROUNDS") if current_app else 4
if not isinstance(value, bytes):
value = value.encode("utf-8")
self.hashed_password = hashpw(value, gensalt(rounds)).decode("utf-8")
This is a pretty simple SqlAlchemy model setup. We have Users and Roles with a many-to-many relationship between them. The interesting thing here is the `password` property. This is implemented as a `hybrid_property`. This is a sqlalchemy generalization of regular Python properties. It is saying "the sql on this property is handled normally, but the python side of the getter/setter is in this code". So, whenever the code tries to create a password, we run it through a hash algorithm so that it is never stored directly in the database.
from bcrypt import checkpw
from flask import Blueprint, current_app, request
from flask_jwt_extended import create_access_token, get_current_user, jwt_required
from marshmallow import EXCLUDE, Schema, ValidationError, fields, post_load
from pockets.autolog import log
from sqlalchemy.orm import joinedload
from devblog.api_security.models import User
from devblog.responses import ApiError, Unauthorized, enforce_json, make_response
SECURITY_API = Blueprint("security_api", __name__)
@enforce_json
def check_login():
email = request.json.get("email", None)
log.debug(f"Logging in: {email}")
password = request.json.get("password", None)
if not email:
raise ApiError("Missing username")
if not password:
raise ApiError("Missing password")
with current_app.alchemy.session() as session:
user = (
session.query(User)
.options(joinedload(User.roles))
.filter(User.email == email)
.first()
)
log.debug(f"Found user- alias: {user and user.alias}")
if not user or not checkpw(
password.encode("utf-8"), user.password.encode("utf-8")
):
raise Unauthorized("Bad username or password")
session.expunge_all()
return user
@SECURITY_API.route("/api/login", methods=["POST"])
def login():
user = check_login()
access_token = create_access_token(identity=user, fresh=True)
return make_response(access_token=access_token)
class FlaskTokens:
def __init__(self, app: Optional[Flask] = None):
if app:
self.register_app(app)
def register_app(self, app: Flask):
jwt = JWTManager(app)
app.register_blueprint(SECURITY_API)
@jwt.user_claims_loader
def _add_claims_to_access_token(user):
return {"roles": [role.name for role in user.roles]}
@jwt.user_identity_loader
def _user_identity_lookup(user):
return user.email
@jwt.user_loader_callback_loader
def _user_loader_callback(identity):
with current_app.alchemy.session() as session:
user = (
session.query(User)
.filter(User.email == identity)
.options(joinedload(User.roles))
.first()
)
session.expunge_all()
return user
app.flask_tokens = self
We implement the login endpoint here. The first step is to lookup the user record they are trying to log in as from the database so that we can get the password. Then bcrypt provides a useful `checkpw` method that hashes the given plaintext and compares it to a supplied hash text. If that works, we create a jwt for the user. You can see that as part of that we register jwt-flask-extended against our app object. We also define a couple functions to help the extension map our sqlalchemy models into the jwt representation. Seeing that we are actually adding `roles` to the jwt means we can check whether someone is an admin as well as whether they are logged in, without hitting the database.
There is a tricky thing in there- `session.expunge_all()`. SqlAlchemy offers a full fledged API. When a transaction closes, it will mark any objects that are attached to that section. If you try and access a value on one of those objects later, it will try and refresh the value from the current session- otherwise you may be using stale data! However, sometimes we are okay with that- like here. Expunge takes the objects in the session and detaches them so that you can use them outside. Remember though, if you change an expunged object, you would have to use `session.add()` to re-attach it if you want to persist that change.
def admin_required(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
verify_jwt_in_request()
claims = get_jwt_claims()
if "admin" not in claims["roles"]:
raise Forbidden("Not authorized for this endpoint")
return fn(*args, **kwargs)
return wrapper
@app.route("/somepath")
@jwt_required
def some_view():
claims = get_jwt_claims()
if claims and "admin" in claims.get("roles"):
return q.filter(Post.published.is_(False)).all()
current_user = get_jwt_identity()
if current_user:
...
Finally, we want to protect views using this Auth scheme. These are a couple examples. In the first, we set up a decorator that will protect an endpoint based on the roles in the jwt- the user needs the "admin" role to continue. In the second we see how we can pull the current user ID and claims, and use that to make decisions. You can see that we used the `@jwt_required` decorator that is provided by the library here. There is also `@jwt_optional` - the reason for this is that the `get_jwt_claims` and `get_jwt_identity` functions will not work outside of a view decorated with one of the functions! You need to decorator to initialize the state store.
In all cases, we expect the user requests to have a header of the form "Authorization: Bearer ${jwt here}". This scheme requires our users to fetch the token and attach it to later requests intentionally. You could also use Cookies, but I prefer the greater control of this model. This is enough to protect our API endpoints. As one final note, a frontend can use the user identity and claims for client side decisions. We NEVER want to trust client side security, so I didn't bother helping the frontend verify the JWT (by providing the public key). There is a lot in the Flask-jwt-extended docs about refresh tokens as well. I learned that this pattern is only really appropriate for server to server authentication- refresh tokens should never be provided to a client.