Flask OAuth 2.0 Server

This section is not a step by step guide on how to create an OAuth 2.0 server in Flask. Instead, we will learn how the Flask implementation works, and some technical details in an OAuth 2.0 provider.

If you need a quick example, here are the official tutorial guide and examples on GitHub:

  1. Example of OAuth 2.0 server
  2. Example of OpenID Connect server (not ready)

At the very beginning, we need to have some basic understanding of the OAuth 2.0 specification. Read The OAuth 2.0 Authorization Framework at first.

Important

If you are developing on your localhost, remember to set the environment variable:

export AUTHLIB_INSECURE_TRANSPORT=true

Looking for OAuth 2 client? Check out Flask OAuth Client.

Authorization Server

The Authorization Server provides several endpoints for authorization, issuing tokens, refreshing tokens and revoking tokens. When the resource owner (user) grants the authorization, this server will issue an access token to the client.

Before creating the authorization server, we need to understand several concepts:

Resource Owner

Resource Owner is the user who is using your service. A resource owner can log in your website with username/email and password, or other methods.

A resource owner SHOULD implement get_user_id() method, lets take SQLAlchemy models for example:

class User(Model):
    id = Column(Integer, primary_key=True)
    # other columns

    def get_user_id(self):
        return self.id

Client

A client is an application making protected resource requests on behalf of the resource owner and with its authorization. It contains at least three information:

  • Client Identifier, usually called client_id
  • Client Password, usually called client_secret
  • Client Token Endpoint Authentication Method

Authlib has provided a mixin for SQLAlchemy, define the client with this mixin:

from authlib.flask.oauth2.sqla import OAuth2ClientMixin

class Client(Model, OAuth2ClientMixin):
    id = Column(Integer, primary_key=True)
    user_id = Column(
        Integer, ForeignKey('user.id', ondelete='CASCADE')
    )
    user = relationship('User')

A client is registered by a user (developer) on your website. If you decide to implement all the missing methods by yourself, get a deep inside with ClientMixin API reference.

Token

Note

Only Bearer Token is supported by now. MAC Token is still under drafts, it will be available when it goes into RFC.

Tokens are used to access the users’ resources. A token is issued with a valid duration, limited scopes and etc. It contains at least:

  • access_token: a token to authorize the http requests.
  • refresh_token: (optional) a token to exchange a new access token
  • client_id: this token is issued to which client
  • expires_at: when will this token expired
  • scope: a limited scope of resources that this token can access

With the SQLAlchemy mixin provided by Authlib:

from authlib.flask.oauth2.sqla import OAuth2TokenMixin

class Token(db.Model, OAuth2TokenMixin):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(
        db.Integer, db.ForeignKey('user.id', ondelete='CASCADE')
    )
    user = db.relationship('User')

A token is associated with a resource owner. There is no certain name for it, here we call it user, but it can be anything else.

Server

Authlib provides a ready to use AuthorizationServer which has built-in tools to handle requests and responses:

from authlib.flask.oauth2 import AuthorizationServer

def query_client(client_id):
    return Client.query.filter_by(client_id=client_id).first()

def save_token(token, request):
    if request.user:
        user_id = request.user.get_user_id()
    else:
        # client_credentials grant_type
        user_id = request.client.user_id
        # or, depending on how you treat client_credentials
        user_id = None
    item = Token(
        client_id=request.client.client_id,
        user_id=user_id,
        **token
    )
    db.session.add(item)
    db.session.commit()

# or with the helper
from authlib.flask.oauth2.sqla import (
    create_query_client_func,
    create_save_token_func
)
query_client = create_query_client_func(db.session, Client)
save_token = create_save_token_func(db.session, Token)

server = AuthorizationServer(
    app, query_client=query_client, save_token=save_token
)

It can also be initialized lazily with init_app:

server = AuthorizationServer()
server.init_app(app, query_client=query_client, save_token=save_token)

It works well without configuration. However, it can be configured with these settings:

OAUTH2_TOKEN_EXPIRES_IN A dict to define expires_in for each grant
OAUTH2_ACCESS_TOKEN_GENERATOR A function or string of module path for importing a function to generate access_token
OAUTH2_REFRESH_TOKEN_GENERATOR A function or string of module path for importing a function to generate refresh_token. It can also be True/False
OAUTH2_ERROR_URIS A list of tuple for (error, error_uri)

Hint

Here is an example of OAUTH2_TOKEN_EXPIRES_IN:

OAUTH2_TOKEN_EXPIRES_IN = {
    'authorization_code': 864000,
    'implicit': 3600,
    'password': 864000,
    'client_credentials': 864000
}

Here is an example of OAUTH2_ACCESS_TOKEN_GENERATOR:

def gen_access_token(client, grant_type, user, scope):
    return create_some_random_string()

OAUTH2_REFRESH_TOKEN_GENERATOR accepts the same parameters.

Now define an endpoint for authorization. This endpoint is used by authorization_code and implicit grants:

from flask import request, render_template
from your_project.auth import current_user

@app.route('/oauth/authorize', methods=['GET', 'POST'])
def authorize():
    # Login is required since we need to know the current resource owner.
    # It can be done with a redirection to the login page, or a login
    # form on this authorization page.
    if request.method == 'GET':
        grant = server.validate_consent_request(end_user=current_user)
        return render_template(
            'authorize.html',
            grant=grant,
            user=current_user,
        )
    confirmed = request.form['confirm']
    if confirmed:
        # granted by resource owner
        return server.create_authorization_response(current_user)
    # denied by resource owner
    return server.create_authorization_response(None)

This is a simple demo, the real case should be more complex. There is a demo in authlib/playground, get a real taste with Authlib Playground.

The token endpoint is much easier:

@app.route('/oauth/token', methods=['POST'])
def issue_token():
    return server.create_token_response()

However, the routes will not work properly. We need to register supported grants for them.

Register Grants

There are four grant types defined by RFC6749, you can also create your own extended grant. Register the supported grant types to the authorization server.

Authorization Code Grant

Authorization Code Grant is a very common grant type, it is supported by almost every OAuth 2 providers. It uses an authorization code to exchange access token. In this case, we need a place to store the authorization code. It can be kept in a database or a cache like redis. Here is a SQLAlchemy mixin for AuthorizationCode:

from authlib.flask.oauth2.sqla import OAuth2AuthorizationCodeMixin

class AuthorizationCode(db.Model, OAuth2AuthorizationCodeMixin):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(
        db.Integer, db.ForeignKey('user.id', ondelete='CASCADE')
    )
    user = db.relationship('User')

Implement this grant by subclass AuthorizationCodeGrant:

from authlib.specs.rfc6749 import grants
from authlib.common.security import generate_token

class AuthorizationCodeGrant(grants.AuthorizationCodeGrant):
    def create_authorization_code(self, client, grant_user, request):
        # you can use other method to generate this code
        code = generate_token(48)
        item = AuthorizationCode(
            code=code,
            client_id=client.client_id,
            redirect_uri=request.redirect_uri,
            scope=request.scope,
            user_id=grant_user.get_user_id(),
        )
        db.session.add(item)
        db.session.commit()
        return code

    def parse_authorization_code(self, code, client):
        item = AuthorizationCode.query.filter_by(
            code=code, client_id=client.client_id).first()
        if item and not item.is_expired():
            return item

    def delete_authorization_code(self, authorization_code):
        db.session.delete(authorization_code)
        db.session.commit()

    def authenticate_user(self, authorization_code):
        return User.query.get(authorization_code.user_id)

# register it to grant endpoint
server.register_grant(AuthorizationCodeGrant)

Note

AuthorizationCodeGrant is the most complex grant.

Default allowed Token Endpoint Auth Methods are:

  1. client_secret_basic
  2. client_secret_post
  3. none

You can change it in the subclass, e.g. remove the none authentication method:

class AuthorizationCodeGrant(grants.AuthorizationCodeGrant):
    TOKEN_ENDPOINT_AUTH_METHODS = ['client_secret_basic', 'client_secret_post']

Implicit Grant

The implicit grant type is usually used in a browser, when resource owner granted the access, access token is issued in the redirect URI, there is no missing implementation, which means it can be easily registered with:

from authlib.specs.rfc6749 import grants

# register it to grant endpoint
server.register_grant(grants.ImplicitGrant)

Implicit Grant is used by public client which has no client_secret. Only allowed Token Endpoint Auth Methods: none.

Resource Owner Password Credentials Grant

Resource owner uses his username and password to exchange an access token, this grant type should be used only when the client is trustworthy, implement it with a subclass of ResourceOwnerPasswordCredentialsGrant:

from authlib.specs.rfc6749 import grants

class PasswordGrant(grants.ResourceOwnerPasswordCredentialsGrant):
    def authenticate_user(self, username, password):
        user = User.query.filter_by(username=username).first()
        if user.check_password(password):
            return user

# register it to grant endpoint
server.register_grant(PasswordGrant)

Default allowed Token Endpoint Auth Methods: client_secret_basic. You can add more in the subclass:

class PasswordGrant(grants.ResourceOwnerPasswordCredentialsGrant):
    TOKEN_ENDPOINT_AUTH_METHODS = [
        'client_secret_basic', 'client_secret_post'
    ]

Client Credentials Grant

Client credentials grant type can access public resources and MAYBE the client’s creator’s resources, depending on how you issue tokens to this grant type. It can be easily registered with:

from authlib.specs.rfc6749 import grants

# register it to grant endpoint
server.register_grant(grants.ClientCredentialsGrant)

Default allowed Token Endpoint Auth Methods: client_secret_basic. You can add more in the subclass:

class ClientCredentialsGrant(grants.ClientCredentialsGrant):
    TOKEN_ENDPOINT_AUTH_METHODS = [
        'client_secret_basic', 'client_secret_post'
    ]

Refresh Token

Many OAuth 2 providers haven’t implemented refresh token endpoint. Authlib provides it as a grant type, implement it with a subclass of RefreshTokenGrant:

from authlib.specs.rfc6749 import grants

class RefreshTokenGrant(grants.RefreshTokenGrant):
    def authenticate_refresh_token(self, refresh_token):
        item = Token.query.filter_by(refresh_token=refresh_token).first()
        # define is_refresh_token_expired by yourself
        if item and not item.is_refresh_token_expired():
            return item

    def authenticate_user(self, credential):
        return User.query.get(credential.user_id)

# register it to grant endpoint
server.register_grant(RefreshTokenGrant)

Default allowed Token Endpoint Auth Methods: client_secret_basic. You can add more in the subclass:

class RefreshTokenGrant(grants.RefreshTokenGrant):
    TOKEN_ENDPOINT_AUTH_METHODS = [
        'client_secret_basic', 'client_secret_post'
    ]

Other Token Endpoints

Flask OAuth 2.0 authorization server has a method to register other token endpoints: authorization_server.register_endpoint. Find the available endpoints:

Protect Resources

Protect users resources, so that only the authorized clients with the authorized access token can access the given scope resources.

A resource server can be a different server other than the authorization server. Here is the way to protect your users’ resources:

from flask import jsonify
from authlib.flask.oauth2 import ResourceProtector, current_token
from authlib.specs.rfc6750 import BearerTokenValidator

class MyBearerTokenValidator(BearerTokenValidator):
    def authenticate_token(self, token_string):
        return Token.query.filter_by(access_token=token_string).first()

    def request_invalid(self, request):
        return False

    def token_revoked(self, token):
        return token.revoked

# only bearer token is supported currently
ResourceProtector.register_token_validator(MyBearerTokenValidator())

# you can also create BearerTokenValidator with shortcut
from authlib.flask.oauth2.sqla import create_bearer_token_validator

BearerTokenValidator = create_bearer_token_validator(db.session, Token)
ResourceProtector.register_token_validator(BearerTokenValidator())

require_oauth = ResourceProtector()

@app.route('/user')
@require_oauth('profile')
def user_profile():
    user = current_token.user
    return jsonify(user)

If the resource is not protected by a scope, use None:

@app.route('/user')
@require_oauth()
def user_profile():
    user = current_token.user
    return jsonify(user)

# or with None

@app.route('/user')
@require_oauth(None)
def user_profile():
    user = current_token.user
    return jsonify(user)

The current_token is a proxy to the Token model you have defined above. Since there is a user relationship on the Token model, we can access this user with current_token.user.

If decorator is not your favorite, there is a with statement for you:

@app.route('/user')
def user_profile():
    with require_oauth.acquire('profile') as token:
        user = token.user
        return jsonify(user)

MethodView & Flask-Restful

You can also use the require_oauth decorator in flask.views.MethodView and flask_restful.Resource:

from flask.views import MethodView

class UserAPI(MethodView):
    decorators = [require_oauth('profile')]


from flask_restful import Resource

class UserAPI(Resource):
    method_decorators = [require_oauth('profile')]

Register Error URIs

To create a better developer experience for debugging, it is suggested that you creating some documentation for errors. Here is a list of built-in Errors.

You can design a documentation page with a description of each error. For instance, there is a web page for invalid_client:

https://developer.your-company.com/errors#invalid-client

In this case, you can register the error URI with OAUTH2_ERROR_URIS configuration:

OAUTH2_ERROR_URIS = [
   ('invalid_client', 'https://developer.your-company.com/errors#invalid-client'),
   # other error URIs
]

If there is no OAUTH2_ERROR_URIS, the error response will not contain any error_uri data.

I18N on Errors

It is also possible to add i18n support to the error_description. The feature has been implemented in version 0.8, but there are still work to do.

Custom Grant Types

It is also possible to create your own grant types. In Authlib, a Grant supports two endpoints:

  1. Authorization Endpoint: which can handle requests with response_type.
  2. Token Endpoint: which is the endpoint to issue tokens.

Creating a custom grant type with BaseGrant:

from authlib.specs.rfc6749 import grants


class MyCustomGrant(grants.BaseGrant):
    AUTHORIZATION_ENDPOINT = False  # if you want to support it
    TOKEN_ENDPOINT = True  # if you want to support it
    GRANT_TYPE = 'custom-grant-type-name'

    def validate_authorization_request(self):
        # only needed if AUTHORIZATION_ENDPOINT = True

    def create_authorization_response(self, grant_user):
        # only needed if AUTHORIZATION_ENDPOINT = True

    def validate_token_request(self):
        # only needed if TOKEN_ENDPOINT = True

    def create_token_response(self):
        # only needed if TOKEN_ENDPOINT = True

For a better understanding, you can read the source code of the built-in grant types. And there are extended grant types defined by other specs:

  1. Using JWTs as Authorization Grants