102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
|
from enum import Enum
|
||
|
|
from flask_principal import Identity, identity_changed
|
||
|
|
|
||
|
|
from app import app
|
||
|
|
from auth.auth_context import (set_authenticated_user, set_validated_token, set_grant_context,
|
||
|
|
set_validated_oauth_token)
|
||
|
|
from auth.scopes import scopes_from_scope_string
|
||
|
|
from auth.permissions import QuayDeferredPermissionUser
|
||
|
|
|
||
|
|
|
||
|
|
class AuthKind(Enum):
|
||
|
|
cookie = 'cookie'
|
||
|
|
basic = 'basic'
|
||
|
|
oauth = 'oauth'
|
||
|
|
signed_grant = 'signed_grant'
|
||
|
|
|
||
|
|
|
||
|
|
class ValidateResult(object):
|
||
|
|
""" A result of validating auth in one form or another. """
|
||
|
|
def __init__(self, kind, missing=False, user=None, token=None, oauthtoken=None,
|
||
|
|
robot=None, signed_data=None, error_message=None):
|
||
|
|
self.kind = kind
|
||
|
|
self.missing = missing
|
||
|
|
self.user = user
|
||
|
|
self.robot = robot
|
||
|
|
self.token = token
|
||
|
|
self.oauthtoken = oauthtoken
|
||
|
|
self.signed_data = signed_data
|
||
|
|
self.error_message = error_message
|
||
|
|
|
||
|
|
def tuple(self):
|
||
|
|
return (self.kind, self.missing, self.user, self.token, self.oauthtoken, self.robot,
|
||
|
|
self.signed_data, self.error_message)
|
||
|
|
|
||
|
|
def __eq__(self, other):
|
||
|
|
return self.tuple() == other.tuple()
|
||
|
|
|
||
|
|
def apply_to_context(self):
|
||
|
|
""" Applies this auth result to the auth context and Flask-Principal. """
|
||
|
|
# Set the various pieces of the auth context.
|
||
|
|
if self.oauthtoken:
|
||
|
|
set_authenticated_user(self.authed_user)
|
||
|
|
set_validated_oauth_token(self.oauthtoken)
|
||
|
|
elif self.authed_user:
|
||
|
|
set_authenticated_user(self.authed_user)
|
||
|
|
elif self.token:
|
||
|
|
set_validated_token(self.token)
|
||
|
|
elif self.signed_data:
|
||
|
|
if self.signed_data['user_context']:
|
||
|
|
set_grant_context({
|
||
|
|
'user': self.signed_data['user_context'],
|
||
|
|
'kind': 'user',
|
||
|
|
})
|
||
|
|
|
||
|
|
# Set the identity for Flask-Principal.
|
||
|
|
if self.identity:
|
||
|
|
identity_changed.send(app, identity=self.identity)
|
||
|
|
|
||
|
|
@property
|
||
|
|
def authed_user(self):
|
||
|
|
""" Returns the authenticated user, whether directly, or via an OAuth token. """
|
||
|
|
if not self.auth_valid:
|
||
|
|
return None
|
||
|
|
|
||
|
|
if self.oauthtoken:
|
||
|
|
return self.oauthtoken.authorized_user
|
||
|
|
|
||
|
|
return self.user if self.user else self.robot
|
||
|
|
|
||
|
|
@property
|
||
|
|
def identity(self):
|
||
|
|
""" Returns the identity for the auth result. """
|
||
|
|
if not self.auth_valid:
|
||
|
|
return None
|
||
|
|
|
||
|
|
if self.oauthtoken:
|
||
|
|
scope_set = scopes_from_scope_string(self.oauthtoken.scope)
|
||
|
|
return QuayDeferredPermissionUser.for_user(self.oauthtoken.authorized_user, scope_set)
|
||
|
|
|
||
|
|
if self.authed_user:
|
||
|
|
return QuayDeferredPermissionUser.for_user(self.authed_user)
|
||
|
|
|
||
|
|
if self.token:
|
||
|
|
return Identity(self.token.code, 'token')
|
||
|
|
|
||
|
|
if self.signed_data:
|
||
|
|
identity = Identity(None, 'signed_grant')
|
||
|
|
identity.provides.update(self.signed_data['grants'])
|
||
|
|
return identity
|
||
|
|
|
||
|
|
return None
|
||
|
|
|
||
|
|
@property
|
||
|
|
def has_user(self):
|
||
|
|
""" Returns whether a user (not a robot) was authenticated successfully. """
|
||
|
|
return bool(self.user)
|
||
|
|
|
||
|
|
@property
|
||
|
|
def auth_valid(self):
|
||
|
|
""" Returns whether authentication successfully occurred. """
|
||
|
|
return self.user or self.token or self.oauthtoken or self.robot or self.signed_data
|