156 lines
5.0 KiB
Python
156 lines
5.0 KiB
Python
import enum
|
|
from aiohttp import web
|
|
from aiohttp_security.abc import (AbstractIdentityPolicy,
|
|
AbstractAuthorizationPolicy)
|
|
from functools import wraps
|
|
|
|
IDENTITY_KEY = 'aiohttp_security_identity_policy'
|
|
AUTZ_KEY = 'aiohttp_security_autz_policy'
|
|
|
|
|
|
async def remember(request, response, identity, **kwargs):
|
|
"""Remember identity into response.
|
|
|
|
The action is performed by identity_policy.remember()
|
|
|
|
Usually the idenity is stored in user cookies homehow but may be
|
|
pushed into custom header also.
|
|
"""
|
|
assert isinstance(identity, str), identity
|
|
assert identity
|
|
identity_policy = request.app.get(IDENTITY_KEY)
|
|
if identity_policy is None:
|
|
text = ("Security subsystem is not initialized, "
|
|
"call aiohttp_security.setup(...) first")
|
|
# in order to see meaningful exception message both: on console
|
|
# output and rendered page we add same message to *reason* and
|
|
# *text* arguments.
|
|
raise web.HTTPInternalServerError(reason=text, text=text)
|
|
await identity_policy.remember(request, response, identity, **kwargs)
|
|
|
|
|
|
async def forget(request, response):
|
|
"""Forget previously remembered identity.
|
|
|
|
Usually it clears cookie or server-side storage to forget user
|
|
session.
|
|
"""
|
|
identity_policy = request.app.get(IDENTITY_KEY)
|
|
if identity_policy is None:
|
|
text = ("Security subsystem is not initialized, "
|
|
"call aiohttp_security.setup(...) first")
|
|
# in order to see meaningful exception message both: on console
|
|
# output and rendered page we add same message to *reason* and
|
|
# *text* arguments.
|
|
raise web.HTTPInternalServerError(reason=text, text=text)
|
|
await identity_policy.forget(request, response)
|
|
|
|
|
|
async def authorized_userid(request):
|
|
identity_policy = request.app.get(IDENTITY_KEY)
|
|
autz_policy = request.app.get(AUTZ_KEY)
|
|
if identity_policy is None or autz_policy is None:
|
|
return None
|
|
identity = await identity_policy.identify(request)
|
|
if identity is None:
|
|
return None # non-registered user has None user_id
|
|
user_id = await autz_policy.authorized_userid(identity)
|
|
return user_id
|
|
|
|
|
|
async def permits(request, permission, context=None):
|
|
assert isinstance(permission, (str, enum.Enum)), permission
|
|
assert permission
|
|
identity_policy = request.app.get(IDENTITY_KEY)
|
|
autz_policy = request.app.get(AUTZ_KEY)
|
|
if identity_policy is None or autz_policy is None:
|
|
return True
|
|
identity = await identity_policy.identify(request)
|
|
# non-registered user still may has some permissions
|
|
access = await autz_policy.permits(identity, permission, context)
|
|
return access
|
|
|
|
|
|
async def is_anonymous(request):
|
|
"""Check if user is anonymous.
|
|
|
|
User is considered anonymous if there is not identity
|
|
in request.
|
|
"""
|
|
identity_policy = request.app.get(IDENTITY_KEY)
|
|
if identity_policy is None:
|
|
return True
|
|
identity = await identity_policy.identify(request)
|
|
if identity is None:
|
|
return True
|
|
return False
|
|
|
|
|
|
def login_required(fn):
|
|
"""Decorator that restrict access only for authorized users.
|
|
|
|
User is considered authorized if authorized_userid
|
|
returns some value.
|
|
"""
|
|
@wraps(fn)
|
|
async def wrapped(*args, **kwargs):
|
|
request = args[-1]
|
|
if not isinstance(request, web.BaseRequest):
|
|
msg = ("Incorrect decorator usage. "
|
|
"Expecting `def handler(request)` "
|
|
"or `def handler(self, request)`.")
|
|
raise RuntimeError(msg)
|
|
|
|
userid = await authorized_userid(request)
|
|
if userid is None:
|
|
raise web.HTTPUnauthorized
|
|
|
|
ret = await fn(*args, **kwargs)
|
|
return ret
|
|
|
|
return wrapped
|
|
|
|
|
|
def has_permission(
|
|
permission,
|
|
context=None,
|
|
):
|
|
"""Decorator that restrict access only for authorized users
|
|
with correct permissions.
|
|
|
|
If user is not authorized - raises HTTPUnauthorized,
|
|
if user is authorized and does not have permission -
|
|
raises HTTPForbidden.
|
|
"""
|
|
def wrapper(fn):
|
|
@wraps(fn)
|
|
async def wrapped(*args, **kwargs):
|
|
request = args[-1]
|
|
if not isinstance(request, web.BaseRequest):
|
|
msg = ("Incorrect decorator usage. "
|
|
"Expecting `def handler(request)` "
|
|
"or `def handler(self, request)`.")
|
|
raise RuntimeError(msg)
|
|
|
|
userid = await authorized_userid(request)
|
|
if userid is None:
|
|
raise web.HTTPUnauthorized
|
|
|
|
allowed = await permits(request, permission, context)
|
|
if not allowed:
|
|
raise web.HTTPForbidden
|
|
ret = await fn(*args, **kwargs)
|
|
return ret
|
|
|
|
return wrapped
|
|
|
|
return wrapper
|
|
|
|
|
|
def setup(app, identity_policy, autz_policy):
|
|
assert isinstance(identity_policy, AbstractIdentityPolicy), identity_policy
|
|
assert isinstance(autz_policy, AbstractAuthorizationPolicy), autz_policy
|
|
|
|
app[IDENTITY_KEY] = identity_policy
|
|
app[AUTZ_KEY] = autz_policy
|