Adding upstream version 2.4.57.upstream/2.4.57

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 613 insertions, 0 deletions

diff --git a/test/modules/md/md_env.py b/test/modules/md/md_env.py
new file mode 100755
index 0000000..e8e36e5
--- /dev/null
+++ b/test/modules/md/md_env.py
@@ -0,0 +1,613 @@
+import copy
+import inspect
+import json
+import logging
+from configparser import ConfigParser, ExtendedInterpolation
+
+import pytest
+import re
+import os
+import shutil
+import subprocess
+import time
+
+from datetime import datetime, timedelta
+from typing import Dict, Optional
+
+from pyhttpd.certs import CertificateSpec
+from .md_cert_util import MDCertUtil
+from pyhttpd.env import HttpdTestSetup, HttpdTestEnv
+from pyhttpd.result import ExecResult
+
+log = logging.getLogger(__name__)
+
+
+class MDTestSetup(HttpdTestSetup):
+
+    def __init__(self, env: 'MDTestEnv'):
+        super().__init__(env=env)
+        self.mdenv = env
+        self.add_modules(["watchdog", "proxy_connect", "md"])
+
+    def make(self):
+        super().make()
+        if "pebble" == self.mdenv.acme_server:
+            self._make_pebble_conf()
+        self.mdenv.clear_store()
+
+    def _make_pebble_conf(self):
+        our_dir = os.path.dirname(inspect.getfile(MDTestSetup))
+        conf_src_dir = os.path.join(our_dir, 'pebble')
+        conf_dest_dir = os.path.join(self.env.gen_dir, 'pebble')
+        if not os.path.exists(conf_dest_dir):
+            os.makedirs(conf_dest_dir)
+        for name in os.listdir(conf_src_dir):
+            src_path = os.path.join(conf_src_dir, name)
+            m = re.match(r'(.+).template', name)
+            if m:
+                self._make_template(src_path, os.path.join(conf_dest_dir, m.group(1)))
+            elif os.path.isfile(src_path):
+                shutil.copy(src_path, os.path.join(conf_dest_dir, name))
+
+
+class MDTestEnv(HttpdTestEnv):
+
+    MD_S_UNKNOWN = 0
+    MD_S_INCOMPLETE = 1
+    MD_S_COMPLETE = 2
+    MD_S_EXPIRED = 3
+    MD_S_ERROR = 4
+
+    EMPTY_JOUT = {'status': 0, 'output': []}
+
+    DOMAIN_SUFFIX = "%d.org" % time.time()
+    LOG_FMT_TIGHT = '%(levelname)s: %(message)s'
+
+    @classmethod
+    def get_acme_server(cls):
+        return os.environ['ACME'] if 'ACME' in os.environ else "pebble"
+
+    @classmethod
+    def has_acme_server(cls):
+        return cls.get_acme_server() != 'none'
+
+    @classmethod
+    def has_acme_eab(cls):
+        return cls.get_acme_server() == 'pebble'
+
+    @classmethod
+    def is_pebble(cls) -> bool:
+        return cls.get_acme_server() == 'pebble'
+
+    @classmethod
+    def lacks_ocsp(cls):
+        return cls.is_pebble()
+
+    @classmethod
+    def has_a2md(cls):
+        d = os.path.dirname(inspect.getfile(HttpdTestEnv))
+        config = ConfigParser(interpolation=ExtendedInterpolation())
+        config.read(os.path.join(d, 'config.ini'))
+        bin_dir = config.get('global', 'bindir')
+        a2md_bin = os.path.join(bin_dir, 'a2md')
+        return os.path.isfile(a2md_bin)
+
+    def __init__(self, pytestconfig=None):
+        super().__init__(pytestconfig=pytestconfig)
+        self.add_httpd_log_modules(["md"])
+        self._acme_server = self.get_acme_server()
+        self._acme_tos = "accepted"
+        self._acme_ca_pemfile = os.path.join(self.gen_dir, "apache/acme-ca.pem")
+        if "pebble" == self._acme_server:
+            self._acme_url = "https://localhost:14000/dir"
+            self._acme_eab_url = "https://localhost:14001/dir"
+        elif "boulder" == self._acme_server:
+            self._acme_url = "http://localhost:4001/directory"
+            self._acme_eab_url = None
+        else:
+            raise Exception(f"unknown ACME server type: {self._acme_server}")
+        self._acme_server_down = False
+        self._acme_server_ok = False
+
+        self._a2md_bin = os.path.join(self.bin_dir, 'a2md')
+        self._default_domain = f"test1.{self.http_tld}"
+        self._tailscale_domain = "test.headless-chicken.ts.net"
+        self._store_dir = "./md"
+        self.set_store_dir_default()
+
+        self.add_cert_specs([
+            CertificateSpec(domains=[f"expired.{self._http_tld}"],
+                            valid_from=timedelta(days=-100),
+                            valid_to=timedelta(days=-10)),
+            CertificateSpec(domains=["localhost"], key_type='rsa2048'),
+            CertificateSpec(domains=[self._tailscale_domain]),
+        ])
+
+    def setup_httpd(self, setup: HttpdTestSetup = None):
+        super().setup_httpd(setup=MDTestSetup(env=self))
+
+    def set_store_dir_default(self):
+        dirpath = "md"
+        if self.httpd_is_at_least("2.5.0"):
+            dirpath = os.path.join("state", dirpath)
+        self.set_store_dir(dirpath)
+
+    def set_store_dir(self, dirpath):
+        self._store_dir = os.path.join(self.server_dir, dirpath)
+        if self.acme_url:
+            self.a2md_stdargs([self.a2md_bin, "-a", self.acme_url,
+                               "-d", self._store_dir,  "-C", self.acme_ca_pemfile, "-j"])
+            self.a2md_rawargs([self.a2md_bin, "-a", self.acme_url,
+                               "-d", self._store_dir,  "-C", self.acme_ca_pemfile])
+
+    def get_apxs_var(self, name: str) -> str:
+        p = subprocess.run([self._apxs, "-q", name], capture_output=True, text=True)
+        if p.returncode != 0:
+            return ""
+        return p.stdout.strip()
+
+    @property
+    def acme_server(self):
+        return self._acme_server
+
+    @property
+    def acme_url(self):
+        return self._acme_url
+
+    @property
+    def acme_tos(self):
+        return self._acme_tos
+
+    @property
+    def a2md_bin(self):
+        return self._a2md_bin
+
+    @property
+    def acme_ca_pemfile(self):
+        return self._acme_ca_pemfile
+
+    @property
+    def store_dir(self):
+        return self._store_dir
+
+    @property
+    def tailscale_domain(self):
+        return self._tailscale_domain
+
+    def get_request_domain(self, request):
+        name = request.node.originalname if request.node.originalname else request.node.name
+        return "%s-%s" % (re.sub(r'[_]', '-', name), MDTestEnv.DOMAIN_SUFFIX)
+
+    def get_method_domain(self, method):
+        return "%s-%s" % (re.sub(r'[_]', '-', method.__name__.lower()), MDTestEnv.DOMAIN_SUFFIX)
+
+    def get_module_domain(self, module):
+        return "%s-%s" % (re.sub(r'[_]', '-', module.__name__.lower()), MDTestEnv.DOMAIN_SUFFIX)
+
+    def get_class_domain(self, c):
+        return "%s-%s" % (re.sub(r'[_]', '-', c.__name__.lower()), MDTestEnv.DOMAIN_SUFFIX)
+
+    # --------- cmd execution ---------
+
+    _a2md_args = []
+    _a2md_args_raw = []
+
+    def a2md_stdargs(self, args):
+        self._a2md_args = [] + args
+
+    def a2md_rawargs(self, args):
+        self._a2md_args_raw = [] + args
+
+    def a2md(self, args, raw=False) -> ExecResult:
+        preargs = self._a2md_args
+        if raw:
+            preargs = self._a2md_args_raw
+        log.debug("running: {0} {1}".format(preargs, args))
+        return self.run(preargs + args)
+
+    def check_acme(self):
+        if self._acme_server_ok:
+            return True
+        if self._acme_server_down:
+            pytest.skip(msg="ACME server not running")
+            return False
+        if self.is_live(self.acme_url, timeout=timedelta(seconds=0.5)):
+            self._acme_server_ok = True
+            return True
+        else:
+            self._acme_server_down = True
+            pytest.fail(msg="ACME server not running", pytrace=False)
+            return False
+
+    def get_ca_pem_file(self, hostname: str) -> Optional[str]:
+        pem_file = super().get_ca_pem_file(hostname)
+        if pem_file is None:
+            pem_file = self.acme_ca_pemfile
+        return pem_file
+
+    # --------- access local store ---------
+
+    def purge_store(self):
+        log.debug("purge store dir: %s" % self._store_dir)
+        assert len(self._store_dir) > 1
+        if os.path.exists(self._store_dir):
+            shutil.rmtree(self._store_dir, ignore_errors=False)
+        os.makedirs(self._store_dir)
+
+    def clear_store(self):
+        log.debug("clear store dir: %s" % self._store_dir)
+        assert len(self._store_dir) > 1
+        if not os.path.exists(self._store_dir):
+            os.makedirs(self._store_dir)
+        for dirpath in ["challenges", "tmp", "archive", "domains", "accounts", "staging", "ocsp"]:
+            shutil.rmtree(os.path.join(self._store_dir, dirpath), ignore_errors=True)
+
+    def clear_ocsp_store(self):
+        assert len(self._store_dir) > 1
+        dirpath = os.path.join(self._store_dir, "ocsp")
+        log.debug("clear ocsp store dir: %s" % dir)
+        if os.path.exists(dirpath):
+            shutil.rmtree(dirpath, ignore_errors=True)
+
+    def authz_save(self, name, content):
+        dirpath = os.path.join(self._store_dir, 'staging', name)
+        os.makedirs(dirpath)
+        open(os.path.join(dirpath, 'authz.json'), "w").write(content)
+
+    def path_store_json(self):
+        return os.path.join(self._store_dir, 'md_store.json')
+
+    def path_account(self, acct):
+        return os.path.join(self._store_dir, 'accounts', acct, 'account.json')
+
+    def path_account_key(self, acct):
+        return os.path.join(self._store_dir, 'accounts', acct, 'account.pem')
+
+    def store_domains(self):
+        return os.path.join(self._store_dir, 'domains')
+
+    def store_archives(self):
+        return os.path.join(self._store_dir, 'archive')
+
+    def store_stagings(self):
+        return os.path.join(self._store_dir, 'staging')
+
+    def store_challenges(self):
+        return os.path.join(self._store_dir, 'challenges')
+
+    def store_domain_file(self, domain, filename):
+        return os.path.join(self.store_domains(), domain, filename)
+
+    def store_archived_file(self, domain, version, filename):
+        return os.path.join(self.store_archives(), "%s.%d" % (domain, version), filename)
+
+    def store_staged_file(self, domain, filename):
+        return os.path.join(self.store_stagings(), domain, filename)
+
+    def path_fallback_cert(self, domain):
+        return os.path.join(self._store_dir, 'domains', domain, 'fallback-pubcert.pem')
+
+    def path_job(self, domain):
+        return os.path.join(self._store_dir, 'staging', domain, 'job.json')
+
+    def replace_store(self, src):
+        shutil.rmtree(self._store_dir, ignore_errors=False)
+        shutil.copytree(src, self._store_dir)
+
+    def list_accounts(self):
+        return os.listdir(os.path.join(self._store_dir, 'accounts'))
+
+    def check_md(self, domain, md=None, state=-1, ca=None, protocol=None, agreement=None, contacts=None):
+        domains = None
+        if isinstance(domain, list):
+            domains = domain
+            domain = domains[0]
+        if md:
+            domain = md
+        path = self.store_domain_file(domain, 'md.json')
+        with open(path) as f:
+            md = json.load(f)
+        assert md
+        if domains:
+            assert md['domains'] == domains
+        if state >= 0:
+            assert md['state'] == state
+        if ca:
+            assert len(md['ca']['urls']) == 1
+            assert md['ca']['urls'][0] == ca
+        if protocol:
+            assert md['ca']['proto'] == protocol
+        if agreement:
+            assert md['ca']['agreement'] == agreement
+        if contacts:
+            assert md['contacts'] == contacts
+
+    def pkey_fname(self, pkeyspec=None):
+        if pkeyspec and not re.match(r'^rsa( ?\d+)?$', pkeyspec.lower()):
+            return "privkey.{0}.pem".format(pkeyspec.lower())
+        return 'privkey.pem'
+
+    def cert_fname(self, pkeyspec=None):
+        if pkeyspec and not re.match(r'^rsa( ?\d+)?$', pkeyspec.lower()):
+            return "pubcert.{0}.pem".format(pkeyspec.lower())
+        return 'pubcert.pem'
+
+    def check_md_complete(self, domain, pkey=None):
+        md = self.get_md_status(domain)
+        assert md
+        assert 'state' in md, "md is unexpected: {0}".format(md)
+        assert md['state'] is MDTestEnv.MD_S_COMPLETE, f"unexpected state: {md['state']}"
+        pkey_file = self.store_domain_file(domain, self.pkey_fname(pkey))
+        cert_file = self.store_domain_file(domain, self.cert_fname(pkey))
+        r = self.run(['ls', os.path.dirname(pkey_file)])
+        if not os.path.isfile(pkey_file):
+            assert False, f"pkey missing: {pkey_file}: {r.stdout}"
+        if not os.path.isfile(cert_file):
+            assert False, f"cert missing: {cert_file}: {r.stdout}"
+        return md
+
+    def check_md_credentials(self, domain):
+        if isinstance(domain, list):
+            domains = domain
+            domain = domains[0]
+        else:
+            domains = [domain]
+        # check private key, validate certificate, etc
+        MDCertUtil.validate_privkey(self.store_domain_file(domain, 'privkey.pem'))
+        cert = MDCertUtil(self.store_domain_file(domain, 'pubcert.pem'))
+        cert.validate_cert_matches_priv_key(self.store_domain_file(domain, 'privkey.pem'))
+        # check SANs and CN
+        assert cert.get_cn() == domain
+        # compare lists twice in opposite directions: SAN may not respect ordering
+        san_list = list(cert.get_san_list())
+        assert len(san_list) == len(domains)
+        assert set(san_list).issubset(domains)
+        assert set(domains).issubset(san_list)
+        # check valid dates interval
+        not_before = cert.get_not_before()
+        not_after = cert.get_not_after()
+        assert not_before < datetime.now(not_before.tzinfo)
+        assert not_after > datetime.now(not_after.tzinfo)
+
+    # --------- check utilities ---------
+
+    def check_json_contains(self, actual, expected):
+        # write all expected key:value bindings to a copy of the actual data ... 
+        # ... assert it stays unchanged 
+        test_json = copy.deepcopy(actual)
+        test_json.update(expected)
+        assert actual == test_json
+
+    def check_file_access(self, path, exp_mask):
+        actual_mask = os.lstat(path).st_mode & 0o777
+        assert oct(actual_mask) == oct(exp_mask)
+
+    def check_dir_empty(self, path):
+        assert os.listdir(path) == []
+
+    def get_http_status(self, domain, path, use_https=True):
+        r = self.get_meta(domain, path, use_https, insecure=True)
+        return r.response['status']
+
+    def get_cert(self, domain, tls=None, ciphers=None):
+        return MDCertUtil.load_server_cert(self._httpd_addr, self.https_port,
+                                           domain, tls=tls, ciphers=ciphers)
+
+    def get_server_cert(self, domain, proto=None, ciphers=None):
+        args = [
+            "openssl", "s_client", "-status",
+            "-connect", "%s:%s" % (self._httpd_addr, self.https_port),
+            "-CAfile", self.acme_ca_pemfile,
+            "-servername", domain,
+            "-showcerts"
+        ]
+        if proto is not None:
+            args.extend(["-{0}".format(proto)])
+        if ciphers is not None:
+            args.extend(["-cipher", ciphers])
+        r = self.run(args)
+        # noinspection PyBroadException
+        try:
+            return MDCertUtil.parse_pem_cert(r.stdout)
+        except:
+            return None
+
+    def verify_cert_key_lenghts(self, domain, pkeys):
+        for p in pkeys:
+            cert = self.get_server_cert(domain, proto="tls1_2", ciphers=p['ciphers'])
+            if 0 == p['keylen']:
+                assert cert is None
+            else:
+                assert cert, "no cert returned for cipher: {0}".format(p['ciphers'])
+                assert cert.get_key_length() == p['keylen'], "key length, expected {0}, got {1}".format(
+                    p['keylen'], cert.get_key_length()
+                )
+
+    def get_meta(self, domain, path, use_https=True, insecure=False):
+        schema = "https" if use_https else "http"
+        port = self.https_port if use_https else self.http_port
+        r = self.curl_get(f"{schema}://{domain}:{port}{path}", insecure=insecure)
+        assert r.exit_code == 0
+        assert r.response
+        assert r.response['header']
+        return r
+
+    def get_content(self, domain, path, use_https=True):
+        schema = "https" if use_https else "http"
+        port = self.https_port if use_https else self.http_port
+        r = self.curl_get(f"{schema}://{domain}:{port}{path}")
+        assert r.exit_code == 0
+        return r.stdout
+
+    def get_json_content(self, domain, path, use_https=True, insecure=False):
+        schema = "https" if use_https else "http"
+        port = self.https_port if use_https else self.http_port
+        url = f"{schema}://{domain}:{port}{path}"
+        r = self.curl_get(url, insecure=insecure)
+        if r.exit_code != 0:
+            log.error(f"curl get on {url} returned {r.exit_code}"
+                      f"\nstdout: {r.stdout}"
+                      f"\nstderr: {r.stderr}")
+        assert r.exit_code == 0, r.stderr
+        return r.json
+
+    def get_certificate_status(self, domain) -> Dict:
+        return self.get_json_content(domain, "/.httpd/certificate-status", insecure=True)
+
+    def get_md_status(self, domain, via_domain=None, use_https=True) -> Dict:
+        if via_domain is None:
+            via_domain = self._default_domain
+        return self.get_json_content(via_domain, f"/md-status/{domain}",
+                                     use_https=use_https)
+
+    def get_server_status(self, query="/", via_domain=None, use_https=True):
+        if via_domain is None:
+            via_domain = self._default_domain
+        return self.get_content(via_domain, "/server-status%s" % query, use_https=use_https)
+
+    def await_completion(self, names, must_renew=False, restart=True, timeout=60,
+                         via_domain=None, use_https=True):
+        try_until = time.time() + timeout
+        renewals = {}
+        names = names.copy()
+        while len(names) > 0:
+            if time.time() >= try_until:
+                return False
+            for name in names:
+                mds = self.get_md_status(name, via_domain=via_domain, use_https=use_https)
+                if mds is None:
+                    log.debug("not managed by md: %s" % name)
+                    return False
+
+                if 'renewal' in mds:
+                    renewal = mds['renewal']
+                    renewals[name] = True
+                    if 'finished' in renewal and renewal['finished'] is True:
+                        if (not must_renew) or (name in renewals):
+                            log.debug(f"domain cert was renewed: {name}")
+                            names.remove(name)
+
+            if len(names) != 0:
+                time.sleep(0.1)
+        if restart:
+            time.sleep(0.1)
+            return self.apache_restart() == 0
+        return True
+
+    def is_renewing(self, name):
+        stat = self.get_certificate_status(name)
+        return 'renewal' in stat
+
+    def await_renewal(self, names, timeout=60):
+        try_until = time.time() + timeout
+        while len(names) > 0:
+            if time.time() >= try_until:
+                return False
+            for name in names:
+                md = self.get_md_status(name)
+                if md is None:
+                    log.debug("not managed by md: %s" % name)
+                    return False
+
+                if 'renewal' in md:
+                    names.remove(name)
+
+            if len(names) != 0:
+                time.sleep(0.1)
+        return True
+
+    def await_error(self, domain, timeout=60, via_domain=None, use_https=True, errors=1):
+        try_until = time.time() + timeout
+        while True:
+            if time.time() >= try_until:
+                return False
+            md = self.get_md_status(domain, via_domain=via_domain, use_https=use_https)
+            if md:
+                if 'state' in md and md['state'] == MDTestEnv.MD_S_ERROR:
+                    return md
+                if 'renewal' in md and 'errors' in md['renewal'] \
+                        and md['renewal']['errors'] >= errors:
+                    return md
+            time.sleep(0.1)
+
+    def await_file(self, fpath, timeout=60):
+        try_until = time.time() + timeout
+        while True:
+            if time.time() >= try_until:
+                return False
+            if os.path.isfile(fpath):
+                return True
+            time.sleep(0.1)
+
+    def check_file_permissions(self, domain):
+        dpath = os.path.join(self.store_dir, 'domains', domain)
+        assert os.path.isdir(dpath)
+        md = json.load(open(os.path.join(dpath, 'md.json')))
+        assert md
+        acct = md['ca']['account']
+        assert acct
+        self.check_file_access(self.path_store_json(), 0o600)
+        # domains
+        self.check_file_access(self.store_domains(), 0o700)
+        self.check_file_access(os.path.join(self.store_domains(), domain), 0o700)
+        self.check_file_access(self.store_domain_file(domain, 'privkey.pem'), 0o600)
+        self.check_file_access(self.store_domain_file(domain, 'pubcert.pem'), 0o600)
+        self.check_file_access(self.store_domain_file(domain, 'md.json'), 0o600)
+        # archive
+        self.check_file_access(self.store_archived_file(domain, 1, 'md.json'), 0o600)
+        # accounts
+        self.check_file_access(os.path.join(self._store_dir, 'accounts'), 0o755)
+        self.check_file_access(os.path.join(self._store_dir, 'accounts', acct), 0o755)
+        self.check_file_access(self.path_account(acct), 0o644)
+        self.check_file_access(self.path_account_key(acct), 0o644)
+        # staging
+        self.check_file_access(self.store_stagings(), 0o755)
+
+    def get_ocsp_status(self, domain, proto=None, cipher=None, ca_file=None):
+        stat = {}
+        args = [
+            "openssl", "s_client", "-status",
+            "-connect", "%s:%s" % (self._httpd_addr, self.https_port),
+            "-CAfile", ca_file if ca_file else self.acme_ca_pemfile,
+            "-servername", domain,
+            "-showcerts"
+        ]
+        if proto is not None:
+            args.extend(["-{0}".format(proto)])
+        if cipher is not None:
+            args.extend(["-cipher", cipher])
+        r = self.run(args, debug_log=False)
+        ocsp_regex = re.compile(r'OCSP response: +([^=\n]+)\n')
+        matches = ocsp_regex.finditer(r.stdout)
+        for m in matches:
+            if m.group(1) != "":
+                stat['ocsp'] = m.group(1)
+        if 'ocsp' not in stat:
+            ocsp_regex = re.compile(r'OCSP Response Status:\s*(.+)')
+            matches = ocsp_regex.finditer(r.stdout)
+            for m in matches:
+                if m.group(1) != "":
+                    stat['ocsp'] = m.group(1)
+        verify_regex = re.compile(r'Verify return code:\s*(.+)')
+        matches = verify_regex.finditer(r.stdout)
+        for m in matches:
+            if m.group(1) != "":
+                stat['verify'] = m.group(1)
+        return stat
+
+    def await_ocsp_status(self, domain, timeout=10, ca_file=None):
+        try_until = time.time() + timeout
+        while True:
+            if time.time() >= try_until:
+                break
+            stat = self.get_ocsp_status(domain, ca_file=ca_file)
+            if 'ocsp' in stat and stat['ocsp'] != "no response sent":
+                return stat
+            time.sleep(0.1)
+        raise TimeoutError(f"ocsp respopnse not available: {domain}")
+
+    def create_self_signed_cert(self, name_list, valid_days, serial=1000, path=None):
+        dirpath = path
+        if not path:
+            dirpath = os.path.join(self.store_domains(), name_list[0])
+        return MDCertUtil.create_self_signed_cert(dirpath, name_list, valid_days, serial)