Server Reference

This part of the documentation covers the interface of Authlib Server.

Flask OAuth 1 Server

class authlib.flask.oauth1.AuthorizationServer(app=None, query_client=None, token_generator=None)

Flask implementation of authlib.rfc5849.AuthorizationServer. Initialize it with Flask app instance, client model class and cache:

server = AuthorizationServer(app=app, query_client=query_client)
# or initialize lazily
server = AuthorizationServer()
server.init_app(app, query_client=query_client)
Parameters:
  • app – A Flask app instance
  • query_client – A function to get client by client_id. The client model class MUST implement the methods described by ClientMixin.
  • token_generator – A function to generate token
create_authorization_verifier(request)

Create and bind oauth_verifier to temporary credential. It could be re-implemented in this way:

def create_authorization_verifier(self, request):
    verifier = generate_token(36)

    temporary_credential = request.credential
    user_id = request.user.get_user_id()

    temporary_credential.user_id = user_id
    temporary_credential.oauth_verifier = verifier
    # if the credential has a save method
    temporary_credential.save()

    # remember to return the verifier
    return verifier
Parameters:request – OAuth1Request instance
Returns:A string of oauth_verifier
create_temporary_credential(request)

Generate and save a temporary credential into database or cache. A temporary credential is used for exchanging token credential. This method should be re-implemented:

def create_temporary_credential(self, request):
    oauth_token = generate_token(36)
    oauth_token_secret = generate_token(48)
    temporary_credential = TemporaryCredential(
        oauth_token=oauth_token,
        oauth_token_secret=oauth_token_secret,
        client_id=request.client_id,
        redirect_uri=request.redirect_uri,
    )
    # if the credential has a save method
    temporary_credential.save()
    return temporary_credential
Parameters:request – OAuth1Request instance
Returns:TemporaryCredential instance
create_token_credential(request)

Create and save token credential into database. This method would be re-implemented like this:

def create_token_credential(self, request):
    oauth_token = generate_token(36)
    oauth_token_secret = generate_token(48)
    temporary_credential = request.credential

    token_credential = TokenCredential(
        oauth_token=oauth_token,
        oauth_token_secret=oauth_token_secret,
        client_id=temporary_credential.get_client_id(),
        user_id=temporary_credential.get_user_id()
    )
    # if the credential has a save method
    token_credential.save()
    return token_credential
Parameters:request – OAuth1Request instance
Returns:TokenCredential instance
delete_temporary_credential(request)

Delete temporary credential from database or cache. For instance, if temporary credential is saved in cache:

def delete_temporary_credential(self, request):
    key = 'a-key-prefix:{}'.format(request.token)
    cache.delete(key)
Parameters:request – OAuth1Request instance
exists_nonce(nonce, request)

The nonce value MUST be unique across all requests with the same timestamp, client credentials, and token combinations.

Parameters:
  • nonce – A string value of oauth_nonce
  • request – OAuth1Request instance
Returns:

Boolean

get_client_by_id(client_id)

Get client instance with the given client_id.

Parameters:client_id – A string of client_id
Returns:Client instance
get_temporary_credential(request)

Get the temporary credential from database or cache. A temporary credential should share the same methods as described in models of TemporaryCredentialMixin:

def get_temporary_credential(self, request):
    key = 'a-key-prefix:{}'.format(request.token)
    data = cache.get(key)
    # TemporaryCredential shares methods from TemporaryCredentialMixin
    return TemporaryCredential(data)
Parameters:request – OAuth1Request instance
Returns:TemporaryCredential instance
class authlib.flask.oauth1.ResourceProtector(app=None, query_client=None, query_token=None, exists_nonce=None)

A protecting method for resource servers. Initialize a resource protector with the query_token method:

from authlib.flask.oauth1 import ResourceProtector, current_credential
from authlib.flask.oauth1.cache import create_exists_nonce_func
from authlib.flask.oauth1.sqla import (
    create_query_client_func,
    create_query_token_func,
)
from your_project.models import Token, User, cache

# you need to define a ``cache`` instance yourself

require_oauth= ResourceProtector(
    app,
    query_client=create_query_client_func(db.session, OAuth1Client),
    query_token=create_query_token_func(db.session, OAuth1Token),
    exists_nonce=create_exists_nonce_func(cache)
)
# or initialize it lazily
require_oauth = ResourceProtector()
require_oauth.init_app(
    app,
    query_client=create_query_client_func(db.session, OAuth1Client),
    query_token=create_query_token_func(db.session, OAuth1Token),
    exists_nonce=create_exists_nonce_func(cache)
)
get_client_by_id(client_id)

Get client instance with the given client_id.

Parameters:client_id – A string of client_id
Returns:Client instance
exists_nonce(nonce, request)

The nonce value MUST be unique across all requests with the same timestamp, client credentials, and token combinations.

Parameters:
  • nonce – A string value of oauth_nonce
  • request – OAuth1Request instance
Returns:

Boolean

authlib.flask.oauth1.current_credential

Routes protected by ResourceProtector can access current credential with this variable.

Cache Helper Functions

authlib.flask.oauth1.cache.create_exists_nonce_func(cache, key_prefix='nonce:', expires=300)

Create an exists_nonce function that can be used in hooks and resource protector.

Parameters:
  • cache – Cache instance
  • key_prefix – key prefix for temporary credential
  • expires – Expire time for nonce
authlib.flask.oauth1.cache.register_nonce_hooks(authorization_server, cache, key_prefix='nonce:', expires=300)

Register nonce related hooks to authorization server.

Parameters:
  • authorization_server – AuthorizationServer instance
  • cache – Cache instance
  • key_prefix – key prefix for temporary credential
  • expires – Expire time for nonce
authlib.flask.oauth1.cache.register_temporary_credential_hooks(authorization_server, cache, key_prefix='temporary_credential:')

Register temporary credential related hooks to authorization server.

Parameters:
  • authorization_server – AuthorizationServer instance
  • cache – Cache instance
  • key_prefix – key prefix for temporary credential

SQLAlchemy Help Functions

Flask OAuth 2 Server

class authlib.flask.oauth2.AuthorizationServer(app=None, query_client=None, save_token=None, **config)

Flask implementation of authlib.rfc6749.AuthorizationServer. Initialize it with query_client, save_token methods and Flask app instance:

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:
        user_id = None
    client = request.client
    tok = Token(
        client_id=client.client_id,
        user_id=user.get_user_id(),
        **token
    )
    db.session.add(tok)
    db.session.commit()

server = AuthorizationServer(app, query_client, save_token)
# or initialize lazily
server = AuthorizationServer()
server.init_app(app, query_client, save_token)
create_authorization_response(request=None, grant_user=None)

Create the HTTP response for authorization. If resource owner granted the authorization, pass the resource owner as the user parameter, otherwise None:

@app.route('/authorize', methods=['POST'])
def confirm_authorize():
    if request.form['confirm'] == 'ok':
        grant_user = current_user
    else:
        grant_user = None
    return server.create_authorization_response(grant_user=grant_user)
create_bearer_token_generator(app)

Create a generator function for generating token value. This method will create a Bearer Token generator with authlib.specs.rfc6750.BearerToken. By default, it will not generate refresh_token, which can be turn on by configuration OAUTH2_REFRESH_TOKEN_GENERATOR=True.

create_expires_generator(app)

Create a generator function for generating expires_in value. Developers can re-implement this method with a subclass if other means required. The default expires_in value is defined by grant_type, different grant_type has different value. It can be configured with: OAUTH2_EXPIRES_{{grant_type|upper}}.

create_token_response(request=None)

Create the HTTP response for token endpoint. It is ready to use, as simple as:

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

Register a grant class into the endpoint registry. Developers can implement the grants in authlib.specs.rfc6749.grants and register with this method:

class AuthorizationCodeGrant(grants.AuthorizationCodeGrant):
    def authenticate_user(self, credential):
        # ...

authorization_server.register_grant(AuthorizationCodeGrant)
Parameters:grant_cls – a grant class.

Validate current HTTP request for authorization page. This page is designed for resource owner to grant or deny the authorization:

@app.route('/authorize', methods=['GET'])
def authorize():
    try:
        grant = server.validate_consent_request(end_user=current_user)
        return render_template(
            'authorize.html',
            grant=grant,
            user=current_user
        )
    except OAuth2Error as error:
        return render_template(
            'error.html',
            error=error
        )
class authlib.flask.oauth2.ResourceProtector

A protecting method for resource servers. Creating a require_oauth decorator easily with ResourceProtector:

from authlib.flask.oauth2 import ResourceProtector

require_oauth = ResourceProtector()

# add bearer token validator
from authlib.specs.rfc6750 import BearerTokenValidator
from project.models import Token

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 False

ResourceProtector.register_token_validator(MyBearerTokenValidator())

# protect resource with require_oauth

@app.route('/user')
@require_oauth('profile')
def user_profile():
    user = User.query.get(current_token.user_id)
    return jsonify(user.to_dict())
authlib.flask.oauth2.current_token

Routes protected by ResourceProtector can access current token with this variable:

from authlib.flask.oauth2 import current_token

@require_oauth()
@app.route('/user_id')
def user_id():
    # current token instance of the OAuth Token model
    return current_token.user_id

Cache Helper Functions

authlib.flask.oauth2.cache.register_cache_authorization_code(cache, authorization_server, authenticate_user)

Use cache for authorization code grant endpoint.

Parameters:
  • cache – Cache instance.
  • authorization_server – AuthorizationServer instance.
  • authenticate_user – A function to authenticate user.

SQLAlchemy Helper Functions