Par Luciano Balmaceda
Ce tutoriel montre comment ajouter une couche d’autorisation à une API Python créée avec Flask. Nous vous recommandons de vous connecter pour suivre ce démarrage rapide avec des exemples configurés pour votre compte.
Dans la section APIs du Auth0 Dashboard, cliquez sur Create API. Indiquez un nom et un identifiant pour votre API, par exemple https://quickstarts/api. Vous utiliserez cet identifiant comme audience plus tard, lors de la configuration de la vérification du jeton d’accès. Conservez RS256 comme Signing Algorithm.
Par défaut, votre API utilise RS256 comme algorithme pour signer les jetons. Puisque RS256 utilise une paire de clés privée/publique, il vérifie les jetons à l’aide de la clé publique de votre compte Auth0. La clé publique est au format JSON Web Key Set (JWKS) et est accessible ici.
Définir les autorisations
messages si les utilisateurs ont le niveau d’accès gestionnaire, et un accès en écriture à cette ressource s’ils ont le niveau d’accès administrateur.
Vous pouvez définir les autorisations dans la vue Permissions de la section APIs de l’Auth0 Dashboard.
Cet exemple utilise la portée read:messages.
- Comment vérifier la présence d’un JSON Web Token (JWT) dans l’en-tête
Authorization d’une requête HTTP entrante.
- Comment vérifier si le jeton est valide en utilisant l’ensemble de clés JSON Web (JWKS) de votre compte Auth0. Pour en savoir plus sur la validation des jetons d’accès, consultez Valider les jetons d’accès.
Valider les jetons d’accès
Installer les dépendances
requirements.txt :
# /requirements.txt
flask==2.3.3
python-dotenv
pyjwt
flask-cors
six
Créer une application Flask
server.py et initialisez l’application Flask. Configurez le domaine, l’audience et la gestion des erreurs.
Créez le décorateur de validation de JWT
# /server.py
# Format error response and append status code
def get_token_auth_header():
"""Obtains the Access Token from the Authorization Header
"""
auth = request.headers.get("Authorization", None)
if not auth:
raise AuthError({"code": "authorization_header_missing",
"description":
"Authorization header is expected"}, 401)
parts = auth.split()
if parts[0].lower() != "bearer":
raise AuthError({"code": "invalid_header",
"description":
"Authorization header must start with"
" Bearer"}, 401)
elif len(parts) == 1:
raise AuthError({"code": "invalid_header",
"description": "Token not found"}, 401)
elif len(parts) > 2:
raise AuthError({"code": "invalid_header",
"description":
"Authorization header must be"
" Bearer token"}, 401)
token = parts[1]
return token
def requires_auth(f):
"""Determines if the Access Token is valid
"""
@wraps(f)
def decorated(*args, **kwargs):
token = get_token_auth_header()
jsonurl = urlopen("https://"+AUTH0_DOMAIN+"/.well-known/jwks.json")
jwks = json.loads(jsonurl.read())
unverified_header = jwt.get_unverified_header(token)
public_key = None
for key in jwks["keys"]:
if key["kid"] == unverified_header["kid"]:
public_key = jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(jwk))
if public_key:
try:
payload = jwt.decode(
token,
public_key,
algorithms=ALGORITHMS,
audience=API_AUDIENCE,
issuer="https://"+AUTH0_DOMAIN+"/"
)
except jwt.ExpiredSignatureError:
raise AuthError({"code": "token_expired",
"description": "token is expired"}, 401)
except jwt.InvalidAudienceError:
raise AuthError({"code": "invalid_audience",
"description":
"audience incorrecte,"
" veuillez vérifier l'audience"}, 401)
except jwt.InvalidIssuerError
raise AuthError({"code": "invalid_issuer",
"description":
"incorrect issuer,"
" please check the issuer"}, 401)
except Exception:
raise AuthError({"code": "invalid_header",
"description":
"Unable to parse authentication"
" token."}, 401)
_request_ctx_stack.top.current_user = payload
return f(*args, **kwargs)
raise AuthError({"code": "invalid_header",
"description": "Unable to find appropriate key"}, 401)
return decorated
# /server.py
def requires_scope(required_scope):
"""Détermine si la portée requise est présente dans le jeton d'accès
Args:
required_scope (str): La portée requise pour accéder à la ressource
"""
token = get_token_auth_header()
unverified_claims = jwt.decode(token, options={"verify_signature": False})
if unverified_claims.get("scope"):
token_scopes = unverified_claims["scope"].split()
for token_scope in token_scopes:
if token_scope == required_scope:
return True
return False
Protéger les points de terminaison d’API
GET /api/public : disponible pour les requêtes non authentifiéesGET /api/private : disponible pour les requêtes authentifiées contenant un jeton d’accès sans scopes supplémentairesGET /api/private-scoped : disponible pour les requêtes authentifiées contenant un jeton d’accès pour lequel la portée read:messages a été accordée
Vous pouvez utiliser les décorateurs et les fonctions définis ci-dessus dans les points de terminaison correspondants.
# API des contrôleurs
# Ceci ne nécessite pas d'authentification
@APP.route("/api/public")
@cross_origin(headers=["Content-Type", "Authorization"])
def public():
response = "Hello from a public endpoint! You don't need to be authenticated to see this."
return jsonify(message=response)
# Ceci nécessite une authentification
@APP.route("/api/private")
@cross_origin(headers=["Content-Type", "Authorization"])
@requires_auth
def private():
response = "Hello from a private endpoint! You need to be authenticated to see this."
return jsonify(message=response)
# Ceci nécessite une autorisation
@APP.route("/api/private-scoped")
@cross_origin(headers=["Content-Type", "Authorization"])
@requires_auth
def private_scoped():
if requires_scope("read:messages"):
response = "Hello from a private endpoint! You need to be authenticated and have a scope of read:messages to see this."
return jsonify(message=response)
raise AuthError({
"code": "Unauthorized",
"description": "You don't have access to this resource"
}, 403)
