added authorization and authentification

development.ini

@@ -19,6 +19,8 @@ sqlalchemy.url = sqlite:///%(here)s/ordr.sqlite
 
 retry.attempts = 3
 
+auth.secret = 'change me!'
+
 # passlib settings
 # setup the context to support only argon2 for the moment
 passlib.schemes = argon2 bcrypt

ordr/models/account.py

@@ -84,7 +84,7 @@ class User(Base):
         return 'user:{}'.format(self.id)
 
     @property
-    def all_principals(self):
+    def principals(self):
         ''' returns all principal identifiers for the user including roles '''
         principals = [self.principal, self.role.principal]
         if self.role is Role.PURCHASER:

ordr/security.py

@@ -1,10 +1,54 @@
+''' User Authentication and Authorization '''
+
 from passlib.context import CryptContext
+from pyramid.authentication import AuthTktAuthenticationPolicy
+from pyramid.authorization import ACLAuthorizationPolicy
+from pyramid.security import Authenticated, Everyone
 from pyramid.settings import aslist
 
+from ordr.models.account import User
+
+#:  passlib context for hashing passwords
+password_context = CryptContext(schemes=['argon2'])
+
+
+class AuthenticationPolicy(AuthTktAuthenticationPolicy):
+    ''' How to authenticate users '''
+
+    def authenticated_userid(self, request):
+        ''' returns the id of an authenticated user
+
+        heavy lifting done in get_user() attached to request
+        '''
+        user = request.user
+        if user is not None:
+            return user.id
 
-password_context = CryptContext()
+    def effective_principals(self, request):
+        ''' returns a list of principals for the user '''
+        principals = [Everyone]
+        user = request.user
+        if user is not None:
+            principals.append(Authenticated)
+            principals.extend(user.principals)
+        return principals
 
 
+def get_user(request):
+    ''' retrieves the user object by the unauthenticated user id
+
+    :param pyramid.request.Request request:
+        the current request object
+    :rtype: :class:`ordr.models.account.User` or None
+    '''
+    user_id = request.unauthenticated_userid
+    if user_id is not None:
+        user = request.dbsession.query(User).filter_by(id=user_id).first()
+        if user and user.is_active:
+            return user
+    return None
+     
+
 def crypt_context_settings_to_string(settings, prefix='passlib.'):
     ''' returns a passlib context setting as a INI-formatted content
     
@@ -38,3 +82,14 @@ def includeme(config):
     # configure the passlib context manager for hashing user passwords
     config_str = crypt_context_settings_to_string(settings, prefix='passlib.')
     password_context.load(config_str)
+
+    # config for authentication and authorization
+    authn_policy = AuthenticationPolicy(
+        settings.get('auth.secret', ''),
+        hashalg='sha512',
+        )
+    config.set_authentication_policy(authn_policy)
+    config.set_authorization_policy(ACLAuthorizationPolicy())
+
+    # attach the get_user function returned by get_user_closure()
+    config.add_request_method(get_user, 'user', reify=True)

ordr/tests/__init__.py

@@ -7,7 +7,18 @@ from pyramid import testing
 APP_SETTINGS = {
     'sqlalchemy.url': 'sqlite:///:memory:',
     }
+    
+EXAMPLE_USER_DATA = {
+    'UNVALIDATED': (1, 'Graham', 'Chapman'),
+    'NEW': (2, 'John', 'Cleese'),
+    'USER': (3, 'Terry', 'Gilliam'),
+    'PURCHASER': (4, 'Eric', 'Idle'),
+    'ADMIN': (5, 'Terry', 'Jones'),
+    'INACTIVE': (6, 'Michael', 'Palin'),
+    }
+
 
+# fixtures
 
 @pytest.fixture(scope='session')
 def app_config():
@@ -18,7 +29,7 @@ def app_config():
         yield config
 
 
-@pytest.fixture(scope='session')
+@pytest.fixture(scope='function')
 def dbsession(app_config):
     ''' fixture for testing with database connection '''
     from ordr.models.meta import Base
@@ -38,3 +49,21 @@ def dbsession(app_config):
 
     transaction.abort()
     Base.metadata.drop_all(engine)
+
+
+# helpers
+
+def get_example_user(role):
+    ''' get the user model for one well known user '''
+    from ordr.models import User
+    id_, first_name, last_name = EXAMPLE_USER_DATA[role.name]
+    user = User(
+        id=id_,
+        username=first_name + last_name,
+        first_name = first_name,
+        last_name = last_name,
+        email = last_name.lower() + '@example.com',
+        role=role
+        )
+    user.set_password(first_name)
+    return user

ordr/tests/models/account.py

@@ -41,12 +41,12 @@ def test_user_principal(id_):
         ('INACTIVE', ['role:inactive']),
         ]
     )
-def test_user_all_principals(name, principals):
+def test_user_principals(name, principals):
     from ordr.models.account import User, Role
     user = User(id=1, role=Role[name])
     expected = ['user:1']
     expected.extend(principals)
-    assert expected == user.all_principals
+    assert expected == user.principals
     
     
 @pytest.mark.parametrize(

ordr/tests/security.py

@@ -1,3 +1,9 @@
+import pytest
+
+from pyramid.testing import DummyRequest
+
+from . import app_config, dbsession, get_example_user
+
 
 def test_crypt_context_to_settings():
     from ordr.security import crypt_context_settings_to_string
@@ -15,3 +21,90 @@ def test_crypt_context_to_settings():
         'depreceated = do, not, adjust, this, list',
         }
     assert set(result.split('\n')) == expected_lines
+
+
+def test_authentication_policy_authenticated_user_id_no_user():
+    from ordr.security import AuthenticationPolicy
+    ap = AuthenticationPolicy('')
+    request = DummyRequest(user=None)
+    assert ap.authenticated_userid(request) is None
+    
+
+def test_authentication_policy_authenticated_user_id_with_user():
+    from ordr.security import AuthenticationPolicy
+    from ordr.models import User
+    ap = AuthenticationPolicy('')
+    request = DummyRequest(user=User(id=123))
+    assert ap.authenticated_userid(request) == 123
+
+
+def test_authentication_policy_effective_principals_no_user():
+    from ordr.security import AuthenticationPolicy
+    from pyramid.security import Everyone 
+    request = DummyRequest(user=None)
+    ap = AuthenticationPolicy('')
+    result = ap.effective_principals(request)
+    assert result == [Everyone]
+    
+    
+def test_authentication_policy_effective_principals_no_user():
+    from ordr.security import AuthenticationPolicy
+    from ordr.models import User, Role
+    from pyramid.security import Authenticated, Everyone 
+    ap = AuthenticationPolicy('')
+    user = User(id=123, role=Role.PURCHASER)
+    request = DummyRequest(user=user)
+    result = ap.effective_principals(request)
+    expected = [
+        Everyone, 
+        Authenticated, 
+        'user:123', 
+        'role:purchaser', 
+        'role:user'
+        ]
+    assert result == expected
+
+
+@pytest.mark.parametrize(
+    'uauid,role_name',  [
+        (3, 'USER'),
+        (4, 'PURCHASER'),
+        (5, 'ADMIN'),
+        ]
+    )
+def test_get_user_returns_user(dbsession, uauid, role_name):
+    from ordr.security import get_user
+    from ordr.models import User, Role
+    # this is a dirty hack, but DummyRequest does not accept setting an
+    # unauthenticated_userid
+    from pyramid.testing import DummyResource
+    request = DummyResource(unauthenticated_userid=uauid, dbsession=dbsession)
+    user_role = Role[role_name]
+    user = get_example_user(user_role)
+    dbsession.add(user)
+    dbsession.flush()
+    assert get_user(request) == user
+
+
+
+@pytest.mark.parametrize(
+    'uauid,role_name',  [
+        (1, 'UNVALIDATED'),
+        (2, 'NEW'),
+        (6, 'INACTIVE'),
+        (2, 'USER'),
+        (None, 'USER'),
+        ]
+    )
+def test_get_user_returns_none(dbsession, uauid, role_name):
+    from ordr.security import get_user
+    from ordr.models import User, Role
+    # this is a dirty hack, but DummyRequest does not accept setting an
+    # unauthenticated_userid
+    from pyramid.testing import DummyResource
+    request = DummyResource(unauthenticated_userid=uauid, dbsession=dbsession)
+    user_role = Role[role_name]
+    user = get_example_user(user_role)
+    dbsession.add(user)
+    dbsession.flush()
+    assert get_user(request) is None