@only_if_auth_required
@not_if_admin_or_readonly_admin
def abort_if_not_public(*args):
    """Abort if the resource is not public and there is no user/admin.

    If auth is required and we are no admin, check if a user is logged in.
    If not abort, since the requested resource is not public.
    """
    if g.current_user is None:
        current_app.logger.debug(
            "Access denied: "
            "Action is not public and user can't be authenticated.")
        abort(401) @only_if_auth_required
@not_if_admin_or_readonly_admin
@only_amiv_token_auth
def add_lookup_filter(auth, resource, request, lookup):
    """Get and add lookup filter for GET, PATCH and DELETE."""
    extra_lookup = auth.create_user_lookup_filter(g.current_user)

    if extra_lookup is None:
        abort(403)  # No lookup at all

    if extra_lookup:
        # Add the additional lookup with an `$and` condition
        # or extend existing `$and`s
        lookup.setdefault('$and', []).append(extra_lookup) class AmivTokenAuth(BasicAuth):
    """Amiv authentication and authorization base class.

    Subclass and overwrite functions if you don't want default behaviour.
    """

    def authorized(self, allowed_roles, resource, method):
        """Authorize Request.

        This is the method Eve will call if the endpoint is not public.

        We use this by setting `g.auth_required` to inform auth hook to abort
        later if user can't be identified.

        Do NOT overwrite this when subclassing `AmivTokenAuth`.
        """
        g.auth_required = True
        return True

    def has_resource_write_permission(self, user_id):
        """Check if the user is alllowed to write to the resource.

        Implement this function for your resource.
        Default behaviour: No user has write permission.

        Args:
            user_id (str): The if of the user

        Returns:
            bool: True if user has permission to write, False otherwise.
        """
        return False

    def has_item_write_permission(self, user_id, item):
        """Check if the user is allowed to modify the item.

        Implement this function for your resource.
        Default behaviour: No user has write permission.

        Args:
            user (str): The id of the user that wants to access the item
            item (dict): The item the user wants to change or delete.
                Attention! If they are any ObjectIds in here, Eve will not have
                converted them yet, so be sure to cast them to str if you want
                to compare them to e.g. g.current_user

        Returns:
            bool: True if user has permission to change the item, False if not.
        """
        return False

    def create_user_lookup_filter(self, user_id):
        """Create a filter for item lookup in GET, PATCH and DELETE.

        Implement this function for your resource.
        Default behaviour: No lookup filter.

        Args:
            user_id (str): The id of the user

        Returns:
            dict: The filter, will be combined with other filters in the hook.
                Return empty dict if no filters should be applied.
                Return None if no lookup should be possible for the user.
        """
        return {} def authenticate(*args):
    """Authenticate user.

    This is not part of the auth class because we want authentication for
    public resources as well.

    Recognized tokens in following formats:
    - Basic auth with token as user and no password
    - "Authorization: <token>"
    - "Authorization: Keyword <token>", Keyword can be anything, e.g. "Bearer"

    After token parsing, look for sessions and user and set the variables:

    - `g.current_token` (str): Token of current request, None if not provided.
    - `g.current_session` (dict): The current session, None if not found.
    - `g.current_user` (str): The id of the currently logged in user, None if
      not found.
    """
    # Get token: First try basicauth username, else just auth header
    token = (getattr(request.authorization, 'username', '') or
             request.headers.get('Authorization', '').strip())
    if (' ' in token):  # Remove keywords, e.g. "Token (...)" or "Bearer (...)"
        token = token.split(' ')[1]
    if not token:  # Set empty token explicitly to "None"
        token = None

    authenticate_token(token) def check_if_admin(resource, *args):
    """Check if the current user is admin for the current resource.

    This is basically the second, resource specific part of authentication,
    separated from `authenticate` only because it needs to be called multiple
    times for the home endpoint.

    Set the variables:

    - `g.resouce_admin` (bool): True if user can see and change anything
    - `g.resource_admin_readonly` (bool): True if user can see anything

    Then notify the auth callback with the current resoure.
    """
    # Set defaults
    g.resource_admin = g.resource_admin_readonly = False

    # Check if root
    if g.get('current_token') == current_app.config['ROOT_PASSWORD']:
        g.resource_admin = True

    # Notify callback
    current_app.after_auth(resource) @only_if_auth_required
@not_if_admin
@only_amiv_token_auth
def check_item_write_permission(auth, resource, item):
    """Check if the user is allowed to PATCH or DELETE the item."""
    user = g.current_user
    if not (user and auth.has_item_write_permission(user, item)):
        current_app.logger.debug(
            "Access denied: "
            "The current user has no permission to write.")
        abort(403) @only_if_auth_required
@not_if_admin
@only_amiv_token_auth
def check_resource_write_permission(auth, resource, *args):
    """Check if the user is allowed to POST to (or DELETE) a resource."""
    user = g.current_user
    if not (user and auth.has_resource_write_permission(user)):
        current_app.logger.debug(
            "Access denied: "
            "The current user has no permission to write.")
        abort(403) def not_if_admin(func):
    """Call function only if `g.resource_admin`."""
    @wraps(func)
    def wrapped(*args):
        if not g.resource_admin:
            func(*args)
    return wrapped def not_if_admin_or_readonly_admin(func):
    """Call function if `g.resource_admin` or `g.resource_admin_readonly`."""
    @wraps(func)
    def wrapped(*args):
        if not (g.resource_admin or g.resource_admin_readonly):
            func(*args)
    return wrapped def only_amiv_token_auth(func):
    """Call function only if auth is a subclass of AmivTokenAuth.

    Add auth as first function argument.
    """
    @wraps(func)
    def wrapped(resource, *args):
        auth = resource_auth(resource)
        if isinstance(auth, AmivTokenAuth):
            func(auth, resource, *args)  # Add auth to function arguments
    return wrapped def only_if_auth_required(func):
    """Call function only if `g.get('auth_required')`."""
    @wraps(func)
    def wrapped(*args):
        if g.get('auth_required'):
            func(*args)
    return wrapped class FakeAuthTest(WebTest):
    """Unittests for the auth functions with fake auth."""

    # Setup and Helper

    def setUp(self):
        """Setup with fake resources that have only auth.

        - AmivTokenAuth subclass for the resource 'fake'.
        - A 'BasicAuth' (from Eve) subclass for 'fake_no_amiv'
        - No auth at all for 'fake_nothing'

        """
        super().setUp()

        self.app.config['DOMAIN']['fake'] = {
            # Its important we use a instance and not the class
            # So we can compare it in some tests
            'authentication': FakeAuth(),
            # some different methods for public and not public
            'resource_methods': ['GET', 'POST', 'DELETE'],
            'public_methods': ['GET', 'POST'],
            'item_methods': ['GET', 'PATCH', 'DELETE'],
            'public_item_methods': ['GET', 'PATCH']
        }

        self.app.config['DOMAIN']['fake_no_amiv'] = {
            'authentication': BasicAuth}

        self.app.config['DOMAIN']['fake_nothing'] = {
            'authentication': None}

    @contextmanager
    def _init_context(self, **g_updates):
        """Create an app context and fill g with values."""
        with self.app.app_context():
            # Defaults - no admins and nothing
            g.current_token = g.current_session = g.current_user = None
            g.resource_admin = g.resource_admin_readonly = False

            # Update g
            for key, value in g_updates.items():
                setattr(g, key, value)

            yield class WebTest(unittest.TestCase, FixtureMixin):
    """Base test class for tests against the full WSGI stack.

    Inspired by eve standard testing class.
    """

    # Test Config overwrites
    test_config = {
        'MONGO_DBNAME': 'test_amivapi',
        'MONGO_USERNAME': 'test_user',
        'MONGO_PASSWORD': 'test_pw',
        'API_MAIL': 'api@test.ch',
        'SMTP_SERVER': '',
        'TESTING': True,
        'DEBUG': True,   # This makes eve's error messages more helpful
        'LDAP_USERNAME': None,  # LDAP test require special treatment
        'LDAP_PASSWORD': None,  # LDAP test require special treatment
        'SENTRY_DSN': None,
        'SENTRY_ENVIRONMENT': None,
        'PASSWORD_CONTEXT': CryptContext(
            schemes=["pbkdf2_sha256"],
            pbkdf2_sha256__default_rounds=10,
            # min_rounds is used to determine if a hash needs to be upgraded
            pbkdf2_sha256__min_rounds=8,
        )
    }

    def setUp(self, **extra_config):
        """Set up the testing client and database connection.

        self.api will be a flask TestClient to make requests
        self.db will be a MongoDB database
        """
        super().setUp()

        # In 3.2, assertItemsEqual was replaced by assertCountEqual
        # Make assertItemsEqual work in tests for py3 as well
        if sys.version_info >= (3, 2):
            self.assertItemsEqual = self.assertCountEqual

        # create eve app and test client
        config = {}
        config.update(self.test_config)
        config.update(extra_config)
        self.app = bootstrap.create_app(**config)
        self.app.response_class = TestResponse
        self.app.test_client_class = TestClient
        self.app.test_mails = []
        self.api = self.app.test_client()

        # Create a separate mongo connection and db reference for tests
        self.connection = MongoClient(host=self.app.config['MONGO_HOST'],
                                      port=self.app.config['MONGO_PORT'])
        self.db = self.connection[self.app.config['MONGO_DBNAME']]
        self.db.authenticate(name=self.app.config['MONGO_USERNAME'],
                             password=self.app.config['MONGO_PASSWORD'],
                             source=self.app.config['MONGO_DBNAME'])

    def tearDown(self):
        """Tear down after testing."""
        # delete testing database
        self.connection.drop_database(self.test_config['MONGO_DBNAME'])
        # close database connection
        self.connection.close()

    # Shortcuts to get a token
    counter = count()

    def get_user_token(self, user_id, created=None):
        """Create session for a user and return a token.

        Args:
            user_id (str): user_id as string.

        Returns:
            str: Token that can be used to authenticate user.
        """
        if created is None:
            created = datetime.now(timezone.utc)

        token = "test_token_" + str(next(self.counter))
        self.db['sessions'].insert_one({u'user': ObjectId(user_id),
                                        u'token': token,
                                        u'_created': created})
        return token

    def get_root_token(self):
        """The root password is the root token.

        Returns:
            str: Token for the root user
        """
        return ROOT_PASSWORD
from base64 import b64encode
from datetime import datetime, timedelta
from bson import ObjectId
from flask import g
from werkzeug.exceptions import Forbidden, Unauthorized
from amivapi.auth import (
    abort_if_not_public,
    add_lookup_filter,
    AmivTokenAuth,
    authenticate,
    check_if_admin,
    check_item_write_permission,
    check_resource_write_permission
)
from amivapi.auth.auth import (
    not_if_admin,
    not_if_admin_or_readonly_admin,
    only_amiv_token_auth,
    only_if_auth_required
)
from amivapi.tests.auth.fake_auth import FakeAuthTest
from amivapi.tests.utils import WebTest
# -*- coding: utf-8 -*-
#
# license: AGPLv3, see LICENSE for details. In addition we strongly encourage
#          you to buy us beer if we meet and you like the software.
"""Tests for auth functions."""





class AmivTokenAuthTest(WebTest):
    """Tests for `AmivTokenAuth` default methods."""

    def test_amiv_token_auth_authorized(self):
        """Test the function `authorized` called by Eve.

        For AmivTokenAuth (and subclasses) it should always return True,
        since everything is handled elsewhere. For this it needs to set
        g.auth_required to True.
        """
        test_auth = AmivTokenAuth()
        with self.app.app_context():
            # No g.auth_required before
            with self.assertRaises(AttributeError):
                g.auth_required
            # Always return true
            self.assertTrue(test_auth.authorized(None, None, None))
            # set g.auth_required
            self.assertTrue(g.auth_required)

    def test_amiv_token_auth_create_user_lookup_filter(self):
        """Test default lookup filter. Should be None."""
        auth = AmivTokenAuth()
        self.assertEqual(auth.create_user_lookup_filter(None), {})

    def test_amiv_token_auth_has_item_write_permission(self):
        """Test default write permission. Should be False."""
        auth = AmivTokenAuth()
        self.assertFalse(auth.has_item_write_permission(None, None))

    def test_amiv_token_auth_has_resource_write_permission(self):
        """Test default write permission. Should be False."""
        auth = AmivTokenAuth()
        self.assertFalse(auth.has_resource_write_permission(None))


class DecoratorTest(FakeAuthTest):
    """Unittests for the decorators that require auth etc."""

    def test_func(self, *args):
        """Create a small function that remembers if its been called."""
        self.was_called = True
        self.call_args = args

    def assertDecorated(self, decorator, context, was_called,
                        func_args=None, call_args=None):
        """Test a decorator.

        Use it to decorate the test function, init context and call the
        decorated function with func_args.
        Then check was_called and call_args (if needed).
        """
        self.was_called = False
        self.call_args = []
        if not func_args:
            func_args = []

        decorated = decorator(self.test_func)
        with self._init_context(**context):
            decorated(*func_args)
            self.assertEqual(was_called, self.was_called)
            if call_args:
                self.assertItemsEqual(call_args, self.call_args)

    def test_auth_required_decorator(self):
        """Call function only if g.get('auth_required')."""

        dec = only_if_auth_required