is_anonymous, login_required, has_permission helpers (#114)

* add is_anonymous helper function and login_required, has_permission decorators

* add docs for `is_anonymous`, `login_required` and `has_permission` functions

* permission can be `enum.Enum` object; cover with tests

* isort fix

Author: litwisha

aiohttp_security/__init__.py

@@ -1,13 +1,14 @@
-from .abc import AbstractIdentityPolicy, AbstractAuthorizationPolicy
-from .api import remember, forget, setup, authorized_userid, permits
+from .abc import AbstractAuthorizationPolicy, AbstractIdentityPolicy
+from .api import (authorized_userid, forget, has_permission, is_anonymous,
+                  login_required, permits, remember, setup)
 from .cookies_identity import CookiesIdentityPolicy
 from .session_identity import SessionIdentityPolicy
 
-
 __version__ = '0.1.2'
 
 
 __all__ = ('AbstractIdentityPolicy', 'AbstractAuthorizationPolicy',
            'CookiesIdentityPolicy', 'SessionIdentityPolicy',
            'remember', 'forget', 'authorized_userid',
-           'permits', 'setup')
+           'permits', 'setup', 'is_anonymous',
+           'login_required', 'has_permission')

aiohttp_security/api.py

@@ -1,7 +1,9 @@
 import asyncio
+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'
@@ -62,7 +64,7 @@ def authorized_userid(request):
 
 @asyncio.coroutine
 def permits(request, permission, context=None):
-    assert isinstance(permission, str), permission
+    assert isinstance(permission, (str, enum.Enum)), permission
     assert permission
     identity_policy = request.app.get(IDENTITY_KEY)
     autz_policy = request.app.get(AUTZ_KEY)
@@ -74,6 +76,85 @@ def permits(request, permission, context=None):
     return access
 
 
+@asyncio.coroutine
+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 = yield from 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.
+    """
+    @asyncio.coroutine
+    @wraps(fn)
+    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 = yield from authorized_userid(request)
+        if userid is None:
+            raise web.HTTPUnauthorized
+
+        ret = yield from 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):
+        @asyncio.coroutine
+        @wraps(fn)
+        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 = yield from authorized_userid(request)
+            if userid is None:
+                raise web.HTTPUnauthorized
+
+            allowed = yield from permits(request, permission, context)
+            if not allowed:
+                raise web.HTTPForbidden
+            ret = yield from 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

docs/reference.rst

@@ -78,7 +78,7 @@ Public API functions
 
    :param request: :class:`aiohttp.web.Request` object.
 
-   :param str permission: requested :term:`permission`.
+   :param permission: Requested :term:`permission`. :class:`str` or :class:`enum.Enum` object.
 
    :param context: additional object may be passed into
                    :meth:`AbstractAuthorizationPolicy.permission`
@@ -88,6 +88,33 @@ Public API functions
             ``False`` otherwise.
 
 
+.. coroutinefunction:: is_anonymous(request)
+
+   Checks if user is anonymous user.
+
+   Return ``True`` if user is not remembered in request, otherwise returns ``False``.
+
+   :param request: :class:`aiohttp.web.Request` object.
+
+
+.. decorator:: login_required
+
+   Decorator for handlers that checks if user is authorized.
+
+   Raises :class:`aiohttp.web.HTTPUnauthorized` if user is not authorized.
+
+
+.. decorator:: has_permission(permission)
+
+   Decorator for handlers that checks if user is authorized
+   and has correct permission.
+
+   Raises :class:`aiohttp.web.HTTPUnauthorized` if user is not authorized.
+   Raises :class:`aiohttp.web.HTTPForbidden` if user is authorized but has no access rights.
+
+   :param str permission: requested :term:`permission`.
+
+
 .. function:: setup(app, identity_policy, autz_policy)
 
    Setup :mod:`aiohttp` application with security policies.

tests/test_dict_autz.py

@@ -1,10 +1,11 @@
 import asyncio
+import enum
 
 from aiohttp import web
-from aiohttp_security import (remember,
-                              authorized_userid, permits,
-                              AbstractAuthorizationPolicy)
 from aiohttp_security import setup as _setup
+from aiohttp_security import (AbstractAuthorizationPolicy, authorized_userid,
+                              forget, has_permission, is_anonymous,
+                              login_required, permits, remember)
 from aiohttp_security.cookies_identity import CookiesIdentityPolicy
 
 
@@ -73,7 +74,27 @@ def check(request):
 
 
 @asyncio.coroutine
-def test_permits(loop, test_client):
+def test_permits_enum_permission(loop, test_client):
+    class Permission(enum.Enum):
+        READ = '101'
+        WRITE = '102'
+        UNKNOWN = '103'
+
+    class Autz(AbstractAuthorizationPolicy):
+
+        @asyncio.coroutine
+        def permits(self, identity, permission, context=None):
+            if identity == 'UserID':
+                return permission in {Permission.READ, Permission.WRITE}
+            else:
+                return False
+
+        @asyncio.coroutine
+        def authorized_userid(self, identity):
+            if identity == 'UserID':
+                return 'Andrew'
+            else:
+                return None
 
     @asyncio.coroutine
     def login(request):
@@ -83,11 +104,11 @@ def login(request):
 
     @asyncio.coroutine
     def check(request):
-        ret = yield from permits(request, 'read')
+        ret = yield from permits(request, Permission.READ)
         assert ret
-        ret = yield from permits(request, 'write')
+        ret = yield from permits(request, Permission.WRITE)
         assert ret
-        ret = yield from permits(request, 'unknown')
+        ret = yield from permits(request, Permission.UNKNOWN)
         assert not ret
         return web.Response()
 
@@ -121,3 +142,143 @@ def check(request):
     resp = yield from client.get('/')
     assert 200 == resp.status
     yield from resp.release()
+
+
+@asyncio.coroutine
+def test_is_anonymous(loop, test_client):
+
+    @asyncio.coroutine
+    def index(request):
+        is_anon = yield from is_anonymous(request)
+        if is_anon:
+            return web.HTTPUnauthorized()
+        return web.HTTPOk()
+
+    @asyncio.coroutine
+    def login(request):
+        response = web.HTTPFound(location='/')
+        yield from remember(request, response, 'UserID')
+        return response
+
+    @asyncio.coroutine
+    def logout(request):
+        response = web.HTTPFound(location='/')
+        yield from forget(request, response)
+        return response
+
+    app = web.Application(loop=loop)
+    _setup(app, CookiesIdentityPolicy(), Autz())
+    app.router.add_route('GET', '/', index)
+    app.router.add_route('POST', '/login', login)
+    app.router.add_route('POST', '/logout', logout)
+    client = yield from test_client(app)
+    resp = yield from client.get('/')
+    assert web.HTTPUnauthorized.status_code == resp.status
+
+    yield from client.post('/login')
+    resp = yield from client.get('/')
+    assert web.HTTPOk.status_code == resp.status
+
+    yield from client.post('/logout')
+    resp = yield from client.get('/')
+    assert web.HTTPUnauthorized.status_code == resp.status
+
+
+@asyncio.coroutine
+def test_login_required(loop, test_client):
+    @login_required
+    @asyncio.coroutine
+    def index(request):
+        return web.HTTPOk()
+
+    @asyncio.coroutine
+    def login(request):
+        response = web.HTTPFound(location='/')
+        yield from remember(request, response, 'UserID')
+        return response
+
+    @asyncio.coroutine
+    def logout(request):
+        response = web.HTTPFound(location='/')
+        yield from forget(request, response)
+        return response
+
+    app = web.Application(loop=loop)
+    _setup(app, CookiesIdentityPolicy(), Autz())
+    app.router.add_route('GET', '/', index)
+    app.router.add_route('POST', '/login', login)
+    app.router.add_route('POST', '/logout', logout)
+    client = yield from test_client(app)
+    resp = yield from client.get('/')
+    assert web.HTTPUnauthorized.status_code == resp.status
+
+    yield from client.post('/login')
+    resp = yield from client.get('/')
+    assert web.HTTPOk.status_code == resp.status
+
+    yield from client.post('/logout')
+    resp = yield from client.get('/')
+    assert web.HTTPUnauthorized.status_code == resp.status
+
+
+@asyncio.coroutine
+def test_has_permission(loop, test_client):
+
+    @has_permission('read')
+    @asyncio.coroutine
+    def index_read(request):
+        return web.HTTPOk()
+
+    @has_permission('write')
+    @asyncio.coroutine
+    def index_write(request):
+        return web.HTTPOk()
+
+    @has_permission('forbid')
+    @asyncio.coroutine
+    def index_forbid(request):
+        return web.HTTPOk()
+
+    @asyncio.coroutine
+    def login(request):
+        response = web.HTTPFound(location='/')
+        yield from remember(request, response, 'UserID')
+        return response
+
+    @asyncio.coroutine
+    def logout(request):
+        response = web.HTTPFound(location='/')
+        yield from forget(request, response)
+        return response
+
+    app = web.Application(loop=loop)
+    _setup(app, CookiesIdentityPolicy(), Autz())
+    app.router.add_route('GET', '/permission/read', index_read)
+    app.router.add_route('GET', '/permission/write', index_write)
+    app.router.add_route('GET', '/permission/forbid', index_forbid)
+    app.router.add_route('POST', '/login', login)
+    app.router.add_route('POST', '/logout', logout)
+    client = yield from test_client(app)
+
+    resp = yield from client.get('/permission/read')
+    assert web.HTTPUnauthorized.status_code == resp.status
+    resp = yield from client.get('/permission/write')
+    assert web.HTTPUnauthorized.status_code == resp.status
+    resp = yield from client.get('/permission/forbid')
+    assert web.HTTPUnauthorized.status_code == resp.status
+
+    yield from client.post('/login')
+    resp = yield from client.get('/permission/read')
+    assert web.HTTPOk.status_code == resp.status
+    resp = yield from client.get('/permission/write')
+    assert web.HTTPOk.status_code == resp.status
+    resp = yield from client.get('/permission/forbid')
+    assert web.HTTPForbidden.status_code == resp.status
+
+    yield from client.post('/logout')
+    resp = yield from client.get('/permission/read')
+    assert web.HTTPUnauthorized.status_code == resp.status
+    resp = yield from client.get('/permission/write')
+    assert web.HTTPUnauthorized.status_code == resp.status
+    resp = yield from client.get('/permission/forbid')
+    assert web.HTTPUnauthorized.status_code == resp.status