| 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, Any |
| |
| from pyhttpd.certs import CertificateSpec, Credentials, HttpdTestCA |
| 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): |
| # Pebble, in v2.5.0 no longer supported HS256 for EAB, which |
| # is the only thing mod_md supports. Issue opened at pebble: |
| # https://github.com/letsencrypt/pebble/issues/455 |
| # is fixed in v2.6.0 |
| 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_ini = os.getenv("PYHTTPD_CONFIG", os.path.join(d, 'config.ini')) |
| config = ConfigParser(interpolation=ExtendedInterpolation()) |
| config.read(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 not self._acme_server_down: |
| if self.is_live(self.acme_url, timeout=timedelta(seconds=0.5)): |
| self._acme_server_ok = True |
| return True |
| self._acme_server_down = True |
| # Both the first test to notice and every one after it report the |
| # same way: a configured-but-unreachable ACME server is a missing |
| # dependency, so skip, or fail under --strict-optional. Previously |
| # the first failed and the rest skipped, and both calls passed the |
| # 'msg' keyword that pytest removed in 8.0, raising TypeError. |
| self.require(False, "ACME server not running", |
| dep="ACME test server", |
| detected=f"ACME={self.get_acme_server()}, " |
| f"no response from {self.acme_url}") |
| 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}" |
| 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.add_privkey(self.store_domain_file(domain, 'privkey.pem')) |
| # No longer check CN, it may not be set or is not trusted anyway |
| # assert cert.get_cn() == domain, f'CN: expected "{domain}", got {cert.get_cn()}' |
| # check SANs |
| # 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), f'{san_list} not subset of {domains}' |
| assert set(domains).issubset(san_list), f'{domains} not subset of {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, spec: CertificateSpec, |
| valid_from: timedelta = timedelta(days=-1), |
| valid_to: timedelta = timedelta(days=89), |
| serial: Optional[int] = None) -> Credentials: |
| key_type = spec.key_type if spec.key_type else 'rsa4096' |
| return HttpdTestCA.create_credentials(spec=spec, issuer=None, |
| key_type=key_type, |
| valid_from=valid_from, |
| valid_to=valid_to, |
| serial=serial) |
