Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

OOP callbacks #335

Closed
wants to merge 2 commits into from
Closed

OOP callbacks #335

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bd70f8e
JWTManager: Register default callbacks before instantiation.
torotil Jun 15, 2020
c2b1197
Switch to a full OOP approach.
torotil Jun 15, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
240 changes: 121 additions & 119 deletions flask_jwt_extended/default_callbacks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,122 +11,124 @@
from flask_jwt_extended.config import config


def default_user_claims_callback(userdata):
"""
By default, we add no additional claims to the access tokens.

:param userdata: data passed in as the ```identity``` argument to the
```create_access_token``` and ```create_refresh_token```
functions
"""
return {}


def default_jwt_headers_callback(default_headers):
"""
By default header typically consists of two parts: the type of the token,
which is JWT, and the signing algorithm being used, such as HMAC SHA256
or RSA. But we don't set the default header here we set it as empty which
further by default set while encoding the token
:return: default we set None here
"""
return None


def default_user_identity_callback(userdata):
"""
By default, we use the passed in object directly as the jwt identity.
See this for additional info:

:param userdata: data passed in as the ```identity``` argument to the
```create_access_token``` and ```create_refresh_token```
functions
"""
return userdata


def default_expired_token_callback(_expired_jwt_header, _expired_jwt_data):
"""
By default, if an expired token attempts to access a protected endpoint,
we return a generic error message with a 401 status
"""
return jsonify({config.error_msg_key: "Token has expired"}), 401


def default_invalid_token_callback(error_string):
"""
By default, if an invalid token attempts to access a protected endpoint, we
return the error string for why it is not valid with a 422 status code

:param error_string: String indicating why the token is invalid
"""
return jsonify({config.error_msg_key: error_string}), 422


def default_unauthorized_callback(error_string):
"""
By default, if a protected endpoint is accessed without a JWT, we return
the error string indicating why this is unauthorized, with a 401 status code

:param error_string: String indicating why this request is unauthorized
"""
return jsonify({config.error_msg_key: error_string}), 401


def default_needs_fresh_token_callback(jwt_header, jwt_data):
"""
By default, if a non-fresh jwt is used to access a ```fresh_jwt_required```
endpoint, we return a general error message with a 401 status code
"""
return jsonify({config.error_msg_key: "Fresh token required"}), 401


def default_revoked_token_callback():
"""
By default, if a revoked token is used to access a protected endpoint, we
return a general error message with a 401 status code
"""
return jsonify({config.error_msg_key: "Token has been revoked"}), 401


def default_user_lookup_error_callback(_jwt_header, jwt_data):
"""
By default, if a user_lookup callback is defined and the callback
function returns None, we return a general error message with a 401
status code
"""
identity = jwt_data[config.identity_claim_key]
result = {config.error_msg_key: "Error loading the user {}".format(identity)}
return jsonify(result), 401


# TODO: Change this to default_token_verification_callback, pass in header and data.
def default_claims_verification_callback(user_claims):
"""
By default, we do not do any verification of the user claims.
"""
return True


def default_verify_claims_failed_callback(_jwt_header, _jwt_data):
"""
By default, if the user claims verification failed, we return a generic
error message with a 400 status code
"""
return jsonify({config.error_msg_key: "User claims verification failed"}), 400


def default_decode_key_callback(claims, headers):
"""
By default, the decode key specified via the JWT_SECRET_KEY or
JWT_PUBLIC_KEY settings will be used to decode all tokens
"""
return config.decode_key


def default_encode_key_callback(identity):
"""
By default, the encode key specified via the JWT_SECRET_KEY or
JWT_PRIVATE_KEY settings will be used to encode all tokens
"""
return config.encode_key
class DefaultCallbacks:
@staticmethod
def user_claims(userdata):
"""
By default, we add no additional claims to the access tokens.

:param userdata: data passed in as the ```identity``` argument to the
```create_access_token``` and ```create_refresh_token```
functions
"""
return {}

@staticmethod
def jwt_headers(default_headers):
"""
By default header typically consists of two parts: the type of the token,
which is JWT, and the signing algorithm being used, such as HMAC SHA256
or RSA. But we don't set the default header here we set it as empty which
further by default set while encoding the token
:return: default we set None here
"""
return None

@staticmethod
def user_identity(userdata):
"""
By default, we use the passed in object directly as the jwt identity.
See this for additional info:

:param userdata: data passed in as the ```identity``` argument to the
```create_access_token``` and ```create_refresh_token```
functions
"""
return userdata

@staticmethod
def expired_token_response(_expired_jwt_header, _expired_jwt_data):
"""
By default, if an expired token attempts to access a protected endpoint,
we return a generic error message with a 401 status
"""
return jsonify({config.error_msg_key: "Token has expired"}), 401

@staticmethod
def invalid_token_response(error_string):
"""
By default, if an invalid token attempts to access a protected endpoint, we
return the error string for why it is not valid with a 422 status code

:param error_string: String indicating why the token is invalid
"""
return jsonify({config.error_msg_key: error_string}), 422

@staticmethod
def unauthorized_response(error_string):
"""
By default, if a protected endpoint is accessed without a JWT, we return
the error string indicating why this is unauthorized, with a 401 status code

:param error_string: String indicating why this request is unauthorized
"""
return jsonify({config.error_msg_key: error_string}), 401

@staticmethod
def needs_fresh_token_response(jwt_header, jwt_data):
"""
By default, if a non-fresh jwt is used to access a ```fresh_jwt_required```
endpoint, we return a general error message with a 401 status code
"""
return jsonify({config.error_msg_key: "Fresh token required"}), 401

@staticmethod
def revoked_token_response():
"""
By default, if a revoked token is used to access a protected endpoint, we
return a general error message with a 401 status code
"""
return jsonify({config.error_msg_key: "Token has been revoked"}), 401

@staticmethod
def user_lookup_error_response(_jwt_header, jwt_data):
"""
By default, if a user_lookup callback is defined and the callback
function returns None, we return a general error message with a 401
status code
"""
identity = jwt_data[config.identity_claim_key]
result = {config.error_msg_key: "Error loading the user {}".format(identity)}
return jsonify(result), 401

# TODO: Change this to default_token_verification_callback, pass in header and data.
@staticmethod
def verify_claims(user_claims):
"""
By default, we do not do any verification of the user claims.
"""
return True

@staticmethod
def invalid_claims_response(_jwt_header, _jwt_data):
"""
By default, if the user claims verification failed, we return a generic
error message with a 400 status code
"""
return jsonify({config.error_msg_key: "User claims verification failed"}), 400

@staticmethod
def decode_key(claims, headers):
"""
By default, the decode key specified via the JWT_SECRET_KEY or
JWT_PUBLIC_KEY settings will be used to decode all tokens
"""
return config.decode_key

@staticmethod
def encode_key(identity):
"""
By default, the encode key specified via the JWT_SECRET_KEY or
JWT_PRIVATE_KEY settings will be used to encode all tokens
"""
return config.encode_key
Loading