Add Custodia tests

Signed-off-by: Christian Heimes <cheimes@redhat.com>

ipatests/azure/azure_definitions/base-fedora.yml

@@ -14,6 +14,7 @@ vms:
   - container_job: base
     tests:
     - test_cmdline
+    - test_custodia
     - test_install
     - test_ipaclient
     - test_ipalib

ipatests/setup.py

@@ -36,6 +36,7 @@
             "ipatests.pytest_ipa",
             "ipatests.pytest_ipa.integration",
             "ipatests.test_cmdline",
+            "ipatests.test_custodia",
             "ipatests.test_install",
             "ipatests.test_integration",
             "ipatests.test_ipaclient",
@@ -52,6 +53,7 @@
         scripts=['ipa-run-tests', 'ipa-test-config', 'ipa-test-task'],
         package_data={
             'ipatests': ['prci_definitions/*'],
+            'ipatests.test_custodia': ['*.conf', 'empty.conf.d/*.conf'],
             'ipatests.test_install': ['*.update'],
             'ipatests.test_integration': ['scripts/*'],
             'ipatests.test_ipaclient': ['data/*/*/*'],

ipatests/test_custodia/empty.conf

@@ -0,0 +1,2 @@
+[globals]
+makedirs = false

ipatests/test_custodia/empty.conf.d/root.conf

@@ -0,0 +1,2 @@
+[/]
+handler = Root

ipatests/test_custodia/test_authenticators.py

@@ -0,0 +1,166 @@
+# Copyright (C) 2016  Custodia Project Contributors - see LICENSE file
+import configparser
+import grp
+import pwd
+
+from ipaserver.custodia.httpd import authenticators
+
+CONFIG = u"""
+[auth:cred_default]
+
+[auth:cred_int]
+uid = 0
+gid = 0
+
+[auth:cred_root]
+uid = root
+gid = root
+
+[auth:cred_user]
+uid = root
+
+[auth:cred_group]
+gid = root
+
+[auth:cred_other_int]
+uid = ${DEFAULT:other_uid}
+gid = ${DEFAULT:other_gid}
+
+[auth:cred_other_name]
+uid = ${DEFAULT:other_username}
+gid = ${DEFAULT:other_groupname}
+
+[auth:header_default]
+
+[auth:header_other]
+header = GSSAPI
+value =
+
+[auth:header_value]
+header = GSSAPI
+value = admin
+
+[auth:header_values]
+header = GSSAPI
+value = admin user
+
+[auth:header_commaspace]
+header = GSSAPI
+value = admin, user, space user
+
+[auth:header_comma]
+header = GSSAPI
+value = admin,user,other user
+"""
+
+
+class TestAuthenticators:
+    @classmethod
+    def setup_class(cls):
+        # Tests are depending on two existing and distinct users and groups.
+        # We chose 'root' with uid/gid 0 and 'nobody', because both exist on
+        # all relevant platforms. Tests use a mocked request so they run
+        # under any user.
+        cls.user = user = pwd.getpwnam('nobody')
+        cls.group = group = grp.getgrgid(user.pw_gid)
+
+        cls.parser = configparser.ConfigParser(
+            interpolation=configparser.ExtendedInterpolation(),
+            defaults={
+                'other_uid': str(user.pw_uid),
+                'other_username': user.pw_name,
+                'other_gid': str(group.gr_gid),
+                'other_groupname': group.gr_name,
+            }
+        )
+        cls.parser.read_string(CONFIG)
+
+    def assertCredMatch(self, cred, uid, gid):
+        request = {'creds': {'uid': uid, 'gid': gid}, 'client_id': 'tests'}
+        assert cred.handle(request)
+
+    def assertCredMismatch(self, cred, uid, gid):
+        request = {'creds': {'uid': uid, 'gid': gid}, 'client_id': 'tests'}
+        assert not cred.handle(request)
+
+    def assertHeaderMatch(self, header, key, value, client_id):
+        request = {'headers': {key: value}, 'client_id': client_id}
+        assert header.handle(request) is True
+
+    def assertHeaderMismatch(self, header, key, value, client_id):
+        request = {'headers': {key: value}, 'client_id': client_id}
+        assert header.handle(request) is False
+
+    def test_cred(self):
+        parser = self.parser
+        cred = authenticators.SimpleCredsAuth(parser, 'auth:cred_default')
+        assert cred.uid == -1
+        assert cred.gid == -1
+        self.assertCredMismatch(cred, 0, 0)
+
+        cred = authenticators.SimpleCredsAuth(parser, 'auth:cred_int')
+        assert cred.uid == 0
+        assert cred.gid == 0
+        self.assertCredMatch(cred, 0, 0)
+        self.assertCredMatch(cred, 0, self.group.gr_gid)
+        self.assertCredMatch(cred, self.user.pw_uid, 0)
+        self.assertCredMismatch(cred, self.user.pw_uid, self.group.gr_gid)
+
+        cred = authenticators.SimpleCredsAuth(parser, 'auth:cred_root')
+        assert cred.uid == 0
+        assert cred.gid == 0
+
+        cred = authenticators.SimpleCredsAuth(parser, 'auth:cred_user')
+        assert cred.uid == 0
+        assert cred.gid == -1
+        self.assertCredMatch(cred, 0, 0)
+        self.assertCredMismatch(cred, self.user.pw_uid, 0)
+
+        cred = authenticators.SimpleCredsAuth(parser, 'auth:cred_group')
+        assert cred.uid == -1
+        assert cred.gid == 0
+        self.assertCredMatch(cred, 0, 0)
+        self.assertCredMismatch(cred, 0, self.group.gr_gid)
+
+        cred = authenticators.SimpleCredsAuth(parser, 'auth:cred_other_int')
+        assert cred.uid != 0
+        assert cred.uid == self.user.pw_uid
+        assert cred.gid != 0
+        assert cred.gid == self.group.gr_gid
+
+        cred = authenticators.SimpleCredsAuth(parser, 'auth:cred_other_name')
+        assert cred.uid != 0
+        assert cred.uid == self.user.pw_uid
+        assert cred.gid != 0
+        assert cred.gid == self.group.gr_gid
+
+    def test_header(self):
+        parser = self.parser
+        gssapi = 'GSSAPI'
+        hdr = authenticators.SimpleHeaderAuth(parser, 'auth:header_default')
+        assert hdr.header == 'REMOTE_USER'
+        assert hdr.value is None
+        self.assertHeaderMatch(hdr, 'REMOTE_USER', None, 0)
+
+        hdr = authenticators.SimpleHeaderAuth(parser, 'auth:header_other')
+        assert hdr.header == 'GSSAPI'
+        assert hdr.value is None
+        self.assertHeaderMatch(hdr, gssapi, None, 0)
+
+        hdr = authenticators.SimpleHeaderAuth(parser, 'auth:header_value')
+        assert hdr.header == 'GSSAPI'
+        assert hdr.value == {'admin'}
+        self.assertHeaderMatch(hdr, gssapi, 'admin', 0)
+        self.assertHeaderMismatch(hdr, gssapi, 'invalid_rule', 0)
+
+        hdr = authenticators.SimpleHeaderAuth(parser, 'auth:header_values')
+        assert hdr.header == 'GSSAPI'
+        assert hdr.value, {'admin' == 'user'}
+
+        hdr = authenticators.SimpleHeaderAuth(parser,
+                                              'auth:header_commaspace')
+        assert hdr.value, {'admin', 'user' == 'space user'}
+
+        hdr = authenticators.SimpleHeaderAuth(parser,
+                                              'auth:header_comma')
+        assert hdr.value, {'admin', 'user' == 'other user'}

ipatests/test_custodia/test_plugins.py

@@ -0,0 +1,49 @@
+# Copyright (C) 2016  Custodia Project Contributors - see LICENSE file
+import pkg_resources
+import pytest
+
+from ipaserver.custodia.client import CustodiaHTTPClient
+from ipaserver.custodia.plugin import (
+    CSStore, HTTPAuthenticator, HTTPAuthorizer
+)
+
+
+class TestCustodiaPlugins:
+    project_name = 'ipaserver.custodia'
+
+    def get_entry_points(self, group):
+        eps = []
+        for e in pkg_resources.iter_entry_points(group):
+            if e.dist.project_name != self.project_name:
+                # only interested in our own entry points
+                continue
+            eps.append(e)
+        return eps
+
+    def assert_ep(self, ep, basecls):
+        try:
+            # backwards compatibility with old setuptools
+            if hasattr(ep, "resolve"):
+                cls = ep.resolve()
+            else:
+                cls = ep.load(require=False)
+        except Exception as e:  # pylint: disable=broad-except
+            pytest.fail("Failed to load %r: %r" % (ep, e))
+        if not issubclass(cls, basecls):
+            pytest.fail("%r is not a subclass of %r" % (cls, basecls))
+
+    def test_authenticators(self):
+        for ep in self.get_entry_points('custodia.authenticators'):
+            self.assert_ep(ep, HTTPAuthenticator)
+
+    def test_authorizers(self):
+        for ep in self.get_entry_points('custodia.authorizers'):
+            self.assert_ep(ep, HTTPAuthorizer)
+
+    def test_clients(self):
+        for ep in self.get_entry_points('custodia.clients'):
+            self.assert_ep(ep, CustodiaHTTPClient)
+
+    def test_stores(self):
+        for ep in self.get_entry_points('custodia.stores'):
+            self.assert_ep(ep, CSStore)

ipatests/test_custodia/test_server.py

@@ -0,0 +1,99 @@
+# Copyright (C) 2017  Custodia Project Contributors - see LICENSE file
+import os
+import socket
+
+import pytest
+
+from ipaserver.custodia.server.args import parse_args
+from ipaserver.custodia.server.config import parse_config
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+EMPTY_CONF = os.path.join(HERE, 'empty.conf')
+
+# pylint: disable=redefined-outer-name
+
+
+@pytest.fixture()
+def args():
+    return parse_args([EMPTY_CONF])
+
+
+@pytest.fixture()
+def args_instance():
+    return parse_args(['--instance=testing', '--debug', EMPTY_CONF])
+
+
+def test_args(args):
+    assert not args.debug
+    assert args.instance is None
+    assert args.configfile.name == EMPTY_CONF
+
+
+def test_args_instance(args_instance):
+    assert args_instance.debug
+    assert args_instance.instance == 'testing'
+    assert args_instance.configfile.name == EMPTY_CONF
+
+
+def test_parse_config(args):
+    parser, config = parse_config(args)
+
+    assert parser.has_section(u'/')
+    assert parser.get(u'/', u'handler') == u'Root'
+
+    assert config == {
+        'auditlog': u'/var/log/custodia/audit.log',
+        'authenticators': {},
+        'authorizers': {},
+        'confdpattern': EMPTY_CONF + u'.d/*.conf',
+        'configdir': HERE,
+        'configfiles': [
+            EMPTY_CONF,
+            EMPTY_CONF + u'.d/root.conf'
+        ],
+        'consumers': {},
+        'debug': False,
+        'hostname': socket.gethostname(),
+        'instance': u'',
+        'libdir': u'/var/lib/custodia',
+        'logdir': u'/var/log/custodia',
+        'makedirs': False,
+        'rundir': u'/var/run/custodia',
+        'server_url': 'http+unix://%2Fvar%2Frun%2Fcustodia%2Fcustodia.sock/',
+        'socketdir': u'/var/run/custodia',
+        'stores': {},
+        'tls_verify_client': False,
+        'umask': 23
+    }
+
+
+def test_parse_config_instance(args_instance):
+    parser, config = parse_config(args_instance)
+
+    assert parser.has_section(u'/')
+    assert parser.get(u'/', u'handler') == u'Root'
+
+    assert config == {
+        'auditlog': u'/var/log/custodia/testing/audit.log',
+        'authenticators': {},
+        'authorizers': {},
+        'confdpattern': EMPTY_CONF + u'.d/*.conf',
+        'configdir': HERE,
+        'configfiles': [
+            EMPTY_CONF,
+            EMPTY_CONF + u'.d/root.conf'
+        ],
+        'consumers': {},
+        'debug': True,
+        'hostname': socket.gethostname(),
+        'instance': u'testing',
+        'libdir': u'/var/lib/custodia/testing',
+        'logdir': u'/var/log/custodia/testing',
+        'makedirs': False,
+        'rundir': u'/var/run/custodia/testing',
+        'server_url': 'http+unix://%2Fvar%2Frun%2Fcustodia%2Ftesting.sock/',
+        'socketdir': u'/var/run/custodia',
+        'stores': {},
+        'tls_verify_client': False,
+        'umask': 23
+    }