import contextlib import grp import importlib.resources from multiprocessing import Process import os import pwd import shutil import subprocess from subprocess import DEVNULL import sys import time from unittest.mock import patch, Mock import flask import gssapi import ldap3 import pytest import requests import socket from zope import component from .utils import gssapi_token_ctx, ccache_cleanup # noqa: F401 from ceo_common.interfaces import IConfig, IKerberosService, ILDAPService, \ IFileService, IMailmanService, IHTTPClient, IUWLDAPService, IMailService, \ IDatabaseService from ceo_common.model import Config, HTTPClient from ceod.api import create_app from ceod.db import MySQLService, PostgreSQLService from ceod.model import KerberosService, LDAPService, FileService, User, \ MailmanService, Group, UWLDAPService, UWLDAPRecord, MailService import ceod.utils as utils from .MockSMTPServer import MockSMTPServer from .MockMailmanServer import MockMailmanServer from .conftest_ceod_api import client # noqa: F401 from .conftest_ceo import cli_setup # noqa: F401 @pytest.fixture(scope='session', autouse=True) def _drone_hostname_mock(): # Drone doesn't appear to set the hostname of the container. # Mock it instead. if 'DRONE_STEP_NAME' in os.environ: hostname = os.environ['DRONE_STEP_NAME'] fqdn = hostname + '.csclub.internal' socket.gethostname = Mock(return_value=hostname) socket.getfqdn = Mock(return_value=fqdn) @pytest.fixture(scope='session') def cfg(_drone_hostname_mock): with importlib.resources.path('tests', 'ceod_test_local.ini') as p: config_file = p.__fspath__() _cfg = Config(config_file) component.getGlobalSiteManager().registerUtility(_cfg, IConfig) return _cfg def delete_test_princs(krb_srv): proc = subprocess.run([ 'kadmin', '-k', '-p', krb_srv.admin_principal, 'listprincs', 'test_*', ], text=True, capture_output=True, check=True) princs = [line.strip() for line in proc.stdout.splitlines()] for princ in princs: krb_srv.delprinc(princ) @pytest.fixture(scope='session') def krb_srv(cfg): # TODO: create temporary Kerberos database using kdb5_util. # We need to be root to read the keytab assert os.geteuid() == 0 # this dance again... ugh if socket.gethostname() == cfg.get('ceod_admin_host'): principal = 'ceod/admin' else: principal = 'ceod/' + socket.getfqdn() krb = KerberosService(principal) component.getGlobalSiteManager().registerUtility(krb, IKerberosService) delete_test_princs(krb) yield krb delete_test_princs(krb) def delete_subtree(conn: ldap3.Connection, base_dn: str): try: conn.search(base_dn, '(objectClass=*)', search_scope=ldap3.LEVEL) for entry in conn.entries: conn.delete(entry.entry_dn) conn.delete(base_dn) except ldap3.core.exceptions.LDAPNoSuchObjectResult: pass @pytest.fixture def g_admin_ctx(app): """ Store the credentials for ceod/admin in flask.g, and override KBR5CCNAME. This context manager should be used any time LDAP is modified via the LDAPService, and we are not in a request context. This should NOT be active when CeodTestClient is making a request, since that will override the values in flask.g. """ @contextlib.contextmanager def wrapper(): with gssapi_token_ctx('ceod/admin') as token, app.app_context(): try: flask.g.auth_user = 'ceod/admin' flask.g.client_token = token yield finally: flask.g.pop('client_token') flask.g.pop('auth_user') return wrapper @pytest.fixture def g_syscom(app): """ Store the credentials for ctdalek in flask.g and override KRB5CCNAME. Use this fixture if you need syscom credentials for an HTTP request to a different process. """ with gssapi_token_ctx('ctdalek') as token, app.app_context(): try: flask.g.sasl_user = 'ctdalek' flask.g.client_token = token yield finally: flask.g.pop('client_token') flask.g.pop('sasl_user') @pytest.fixture(scope='session') def ldap_conn(cfg) -> ldap3.Connection: # Assume that the same server URL is being used for the CSC # and UWLDAP during the tests. cfg = component.getUtility(IConfig) server_url = cfg.get('ldap_server_url') # sanity check assert server_url == cfg.get('uwldap_server_url') with gssapi_token_ctx('ceod/admin') as token: creds = gssapi.Credentials(token=token) conn = ldap3.Connection( server_url, auto_bind=True, raise_exceptions=True, authentication=ldap3.SASL, sasl_mechanism=ldap3.KERBEROS, sasl_credentials=(None, None, creds)) return conn @pytest.fixture(scope='session') def ldap_srv_session(cfg, krb_srv, ldap_conn): conn = ldap_conn users_base = cfg.get('ldap_users_base') groups_base = cfg.get('ldap_groups_base') sudo_base = cfg.get('ldap_sudo_base') for base_dn in [users_base, groups_base, sudo_base]: delete_subtree(conn, base_dn) conn.add(base_dn, 'organizationalUnit') _ldap_srv = LDAPService() component.getGlobalSiteManager().registerUtility(_ldap_srv, ILDAPService) yield _ldap_srv for base_dn in [users_base, groups_base, sudo_base]: delete_subtree(conn, base_dn) @pytest.fixture def ldap_srv(ldap_srv_session, g_admin_ctx): # This is an ugly hack to get around the fact that function-scoped # fixtures (g_admin_ctx) can't be used from session-scoped fixtures # (ldap_srv_session). with g_admin_ctx(): yield ldap_srv_session @pytest.fixture(scope='session') def file_srv(cfg): _file_srv = FileService() component.getGlobalSiteManager().registerUtility(_file_srv, IFileService) members_home = cfg.get('members_home') clubs_home = cfg.get('clubs_home') shutil.rmtree(members_home, ignore_errors=True) shutil.rmtree(clubs_home, ignore_errors=True) yield _file_srv shutil.rmtree(members_home, ignore_errors=True) shutil.rmtree(clubs_home, ignore_errors=True) @pytest.fixture(scope='session') def http_client(cfg): _client = HTTPClient() component.getGlobalSiteManager().registerUtility(_client, IHTTPClient) return _client @pytest.fixture(scope='session') def mock_mailman_server(): server = MockMailmanServer() server.start() yield server server.stop() @pytest.fixture(scope='session') def mailman_srv(mock_mailman_server, cfg, http_client): # TODO: test the RemoteMailmanService as well mailman = MailmanService() component.getGlobalSiteManager().registerUtility(mailman, IMailmanService) return mailman @pytest.fixture(scope='session') def uwldap_srv(cfg, ldap_conn): conn = ldap_conn base_dn = cfg.get('uwldap_base') delete_subtree(conn, base_dn) conn.add(base_dn, 'organizationalUnit') _uwldap_srv = UWLDAPService() component.getGlobalSiteManager().registerUtility(_uwldap_srv, IUWLDAPService) yield _uwldap_srv delete_subtree(conn, base_dn) @pytest.fixture(scope='session') def mock_mail_server(): mock_server = MockSMTPServer() mock_server.start() yield mock_server mock_server.stop() @pytest.fixture(scope='session') def mail_srv(cfg, mock_mail_server): _mail_srv = MailService() component.getGlobalSiteManager().registerUtility(_mail_srv, IMailService) return _mail_srv @pytest.fixture(scope='session') def mysql_srv(cfg): mysql_srv = MySQLService() component.getGlobalSiteManager().registerUtility(mysql_srv, IDatabaseService, 'mysql') return mysql_srv @pytest.fixture(scope='session') def postgresql_srv(cfg): psql_srv = PostgreSQLService() component.getGlobalSiteManager().registerUtility(psql_srv, IDatabaseService, 'postgresql') return psql_srv @pytest.fixture(autouse=True, scope='session') def app( cfg, krb_srv, ldap_srv_session, file_srv, mailman_srv, uwldap_srv, mail_srv, mysql_srv, postgresql_srv, ): app = create_app({'TESTING': True}) return app @pytest.fixture(scope='session') def mocks_for_create_user(): with patch.object(utils, 'gen_password') as gen_password_mock, \ patch.object(pwd, 'getpwuid') as getpwuid_mock, \ patch.object(grp, 'getgrgid') as getgrgid_mock: gen_password_mock.return_value = 'krb5' # Normally, if getpwuid or getgrgid do *not* raise a KeyError, # then LDAPService will skip that UID. Therefore, by raising a # KeyError, we are making sure that the UID will *not* be skipped. getpwuid_mock.side_effect = KeyError() getgrgid_mock.side_effect = KeyError() yield @pytest.fixture def simple_user(): return User( uid='test_jdoe', cn='John Doe', program='Math', terms=['s2021'], ) @pytest.fixture def simple_club(): return User( uid='test_club1', cn='Club One', is_club=True, ) @pytest.fixture def ldap_user(simple_user, g_admin_ctx): with g_admin_ctx(): simple_user.add_to_ldap() yield simple_user with g_admin_ctx(): simple_user.remove_from_ldap() @pytest.fixture def krb_user(simple_user): # We don't want to use add_to_kerberos() here because that expires the # user's password, which we don't want for testing subprocess.run( ['kadmin', '-k', '-p', 'ceod/admin', 'addprinc', '-pw', 'krb5', simple_user.uid], stdout=DEVNULL, check=True) yield simple_user simple_user.remove_from_kerberos() @pytest.fixture def simple_group(): return Group( cn='group1', gid_number=21000, user_cn='Group One', ) @pytest.fixture def ldap_group(simple_group, g_admin_ctx): with g_admin_ctx(): simple_group.add_to_ldap() yield simple_group with g_admin_ctx(): simple_group.remove_from_ldap() @pytest.fixture def uwldap_user(cfg, uwldap_srv, ldap_conn): conn = ldap_conn base_dn = cfg.get('uwldap_base') user = UWLDAPRecord( uid='test_jdoe', mail_local_addresses=['test_jdoe@uwaterloo.internal'], program='Math', cn='John Doe', sn='Doe', given_name='John', ) dn = f'uid={user.uid},{base_dn}' conn.add( dn, [ 'inetLocalMailRecipient', 'inetOrgPerson', 'organizationalPerson', 'person', ], { 'mailLocalAddress': user.mail_local_addresses, 'ou': user.program, 'cn': user.cn, 'sn': user.sn, 'givenName': user.given_name, }, ) yield user conn.delete(dn) @pytest.fixture(scope='module') def app_process(cfg, app, http_client): port = cfg.get('ceod_port') hostname = socket.gethostname() def server_start(): sys.stdout = open('/dev/null', 'w') sys.stderr = sys.stdout app.run(debug=False, host='0.0.0.0', port=port) proc = Process(target=server_start) proc.start() try: for i in range(5): try: http_client.get(hostname, '/ping') except requests.exceptions.ConnectionError: time.sleep(1) continue break assert i != 5, 'Timed out' yield finally: proc.terminate() proc.join()