Example: flask
The following is an example of using flaat with flask.
# Flaat example with Flask
import logging
from flaat import AuthWorkflow
from flaat.config import AccessLevel
from flaat.flask import Flaat
from flaat.requirements import CheckResult, HasSubIss, IsTrue
from flaat.requirements import get_claim_requirement
from flaat.requirements import get_vo_requirement
from flask import Blueprint, Flask, abort, current_app
from werkzeug import Response
# ------------------------------------------------------------------
# Basic configuration example ---------------------------------------
logging.basicConfig(level="WARNING")
logging.getLogger("flaat").setLevel("DEBUG")
logging.getLogger("werkzeug").setLevel("DEBUG")
# Standard flask blueprint snippet, source:
# https://flask.palletsprojects.com/en/2.1.x/blueprints
frontend = Blueprint("frontend", "frontend")
# Set a list of access levels to use
def is_admin(user_infos):
return user_infos.user_info['email'] in current_app.config['ADMIN_EMAILS']
flaat = Flaat([
AccessLevel("user", HasSubIss()),
AccessLevel("admin", IsTrue(is_admin)),
])
class Config(object):
# Defines the list of Flaat trusted OIDC providers
TRUSTED_OP_LIST = [
"https://aai-demo.egi.eu/oidc",
"https://aai-demo.egi.eu/auth/realms/egi",
"https://aai-dev.egi.eu/oidc",
"https://aai.egi.eu/oidc/",
"https://aai.egi.eu/auth/realms/egi",
"https://accounts.google.com/",
"https://b2access-integration.fz-juelich.de/oauth2",
"https://b2access.eudat.eu/oauth2/",
"https://iam-test.indigo-datacloud.eu/",
"https://iam.deep-hybrid-datacloud.eu/",
"https://iam.extreme-datacloud.eu/",
"https://login-dev.helmholtz.de/oauth2/",
"https://login.elixir-czech.org/oidc/",
"https://login.helmholtz-data-federation.de/oauth2/",
"https://login.helmholtz.de/oauth2/",
"https://oidc.scc.kit.edu/auth/realms/kit/",
"https://orcid.org/",
"https://proxy.demo.eduteams.org",
"https://services.humanbrainproject.eu/oidc/",
"https://unity.eudat-aai.fz-juelich.de/oauth2/",
"https://unity.helmholtz-data-federation.de/oauth2/",
"https://wlcg.cloud.cnaf.infn.it/",
"https://proxy.eduteams.org/",
]
# Additional example configuration:
ADMIN_EMAILS = ["admin@foo.org", "dev@foo.org"]
# ------------------------------------------------------------------
# Production configuration example ----------------------------------
class ProductionConfig(Config):
# In production you might want to reduce the number of OP
TRUSTED_OP_LIST = ["https://aai.egi.eu/oidc/"]
FLAAT_ISS = "https://aai.egi.eu/oidc/"
# Define your request timeout for production
FLAAT_REQUEST_TIMEOUT = 1.2
# Required for using token introspection endpoint:
FLAAT_CLIENT_ID = "oidc-agent"
FLAAT_CLIENT_SECRET = ""
# ------------------------------------------------------------------
# Development configuration example ---------------------------------
class DevelopmentConfig(Config):
# High timeouts might simplify debugging
FLAAT_REQUEST_TIMEOUT = 30
# On development certificate verification might not be needed
FLAAT_VERIFY_TLS = False
FLAAT_VERIFY_JWT = False
# ------------------------------------------------------------------
# Testing configuration example -------------------------------------
class TestingConfig(Config):
# Set TESTING to True to run all Flask plugins on testing mode
TESTING = True
# When testing to run requirements as close as possible to production
FLAAT_REQUEST_TIMEOUT = 1.2
# ------------------------------------------------------------------
# Standard flask Application Factories snippet, source --------------
# https://flask.palletsprojects.com/en/2.1.x/patterns/appfactories
def create_app(config=f"{__name__}.ProductionConfig"):
app = Flask(__name__)
app.config.from_object(config)
# Init application plugins
flaat.init_app(app)
# db.init_app(app)
# mail.init_app(app)
# Register blueprints
app.register_blueprint(frontend)
# app.register_blueprint(admin)
# app.register_blueprint(other)
return app
# ------------------------------------------------------------------
# Routes definition -------------------------------------------------
@frontend.route("/", methods=["GET"])
def root():
text = """This is an example for using flaat with Flask.
The following endpoints are available:
/info General info about the access_token
/info_no_strict General info without token validation
/authenticated Requires a valid user
/authenticated_callback Requires a valid user, uses a custom callback on error
/authorized_level Requires user to fit the specified access level
/authorized_claim Requires user to have one of two claims
/authorized_vo Requires user to have an entitlement
/full_custom Fully custom auth handling
"""
return Response(text, mimetype="text/plain")
# -------------------------------------------------------------------
# Call with user information ----------------------------------------
@frontend.route("/info", methods=["GET"])
@flaat.inject_user_infos() # Fail if no valid authentication is provided
def info_strict_mode(user_infos):
return user_infos.toJSON()
@frontend.route("/info_no_strict", methods=["GET"])
@flaat.inject_user_infos(strict=False) # Pass with invalid authentication
def info(user_infos=None):
return user_infos.toJSON() if user_infos else "No userinfo"
# -------------------------------------------------------------------
# Endpoint which requires of an authenticated user ------------------
@frontend.route("/authenticated", methods=["GET"])
@flaat.is_authenticated()
def authenticated():
return "This worked: there was a valid login"
# -------------------------------------------------------------------
# Instead of giving an error this will return the custom error
# response from `my_on_failure` -------------------------------------
def my_on_failure(exception, user_infos=None):
text = f"""Custom callback 'my_on_failure' invoked:
Error Message: {exception}
User: {user_infos if user_infos else "No Auth"}
"""
abort(401, description=text)
@frontend.route("/authenticated_callback", methods=["GET"])
@flaat.is_authenticated(on_failure=my_on_failure)
def authenticated_callback():
return "This worked: there was a valid login"
# -------------------------------------------------------------------
# Endpoint which requires an access level ---------------------------
@frontend.route("/authorized_level", methods=["GET"])
@flaat.access_level("admin")
def authorized_level():
return "This worked: user has the required rights"
# -------------------------------------------------------------------
# The user needs to satisfy a certain requirement -------------------
email_requirement = get_claim_requirement(
["admin@foo.org", "dev@foo.org"],
claim="email",
match=1,
)
@frontend.route("/authorized_claim", methods=["GET"])
@flaat.requires(email_requirement)
def authorized_claim():
return "This worked: User has the claim"
# -------------------------------------------------------------------
# The user needs belong to a certain virtual organization -----------
vo_requirement = get_vo_requirement(
[
"urn:mace:egi.eu:group:test:foo",
"urn:mace:egi.eu:group:test:bar",
],
"mock_entitlements",
match=2,
)
@frontend.route("/authorized_vo", methods=["GET"])
@flaat.requires(vo_requirement)
def authorized_vo():
return "This worked: user has the required entitlement"
# -------------------------------------------------------------------
# For maximum customization use AuthWorkflow ------------------------
def my_request_check(user_infos, *args, **kwargs):
if len(args) != 1:
return CheckResult(False, "Missing request object")
return CheckResult(True, "The request is allowed")
def my_process_args(user_infos, *args, **kwargs):
"""We can manipulate the view functions arguments here The user is
already authenticated at this point, therefore we have `user_infos`,
therefore we can base our manipulations on the users identity.
"""
kwargs["email"] = user_infos.get("email", "")
return (args, kwargs)
custom = AuthWorkflow(
flaat, # needs the flaat instance
user_requirements=get_claim_requirement("bar", "foo"),
request_requirements=my_request_check,
process_arguments=my_process_args,
on_failure=my_on_failure,
ignore_no_authn=True, # Don't fail if there is no authentication
)
@frontend.route("/full_custom", methods=["GET"])
@custom.decorate_view_func # invoke the workflow here
def full_custom(email=""):
text = f"""This worked: The custom workflow did succeed:
The users email is: {email}
"""
return Response(text, mimetype="text/plain")
# -------------------------------------------------------------------
# Main function -----------------------------------------------------
if __name__ == "__main__":
app = create_app("ProductionConfig")
app.run(host="0.0.0.0", port=8081)