apache/test/modules/arch/linux/env.py

import inspect
import logging
import os
import re
import shutil
import signal
import socket
import subprocess
import threading
import time
from typing import Callable, Dict, List, Optional

from pyhttpd.env import HttpdTestEnv, HttpdTestSetup

log = logging.getLogger(__name__)


class SystemdTestSetup(HttpdTestSetup):

    def __init__(self, env: 'HttpdTestEnv'):
        super().__init__(env=env)
        self.add_source_dir(os.path.dirname(inspect.getfile(SystemdTestSetup)))
        # mod_systemd is only built with --enable-systemd, so it must not be
        # a hard requirement; the tests skip when it is absent.
        self.add_optional_modules(["systemd"])


class SystemdTestEnv(HttpdTestEnv):

    def __init__(self, pytestconfig=None):
        super().__init__(pytestconfig=pytestconfig)
        self.add_httpd_log_modules(["core"])
        self._notify = None

    def setup_httpd(self, setup: HttpdTestSetup = None):
        super().setup_httpd(setup=SystemdTestSetup(env=self))

    @property
    def systemd_is_dso(self) -> bool:
        return os.path.isfile(os.path.join(self.libexec_dir, 'mod_systemd.so'))

    @property
    def has_systemd_module(self) -> bool:
        """Whether mod_systemd is available, however it was built:
        --enable-systemd links it statically, --enable-systemd=shared
        builds the DSO."""
        if self.systemd_is_dso:
            return True
        p = subprocess.run([self.httpd_bin, '-l'], capture_output=True,
                           text=True)
        return re.search(r'^\s+mod_systemd\.c$', p.stdout, re.M) is not None

    @property
    def notify(self) -> 'NotifyListener':
        """The stand-in notification socket httpd reports to."""
        return self._notify

    def start_notify_listener(self) -> 'NotifyListener':
        """Bind the notification socket and point httpd's $NOTIFY_SOCKET at it.

        This must happen before the server is first started, since libsystemd
        reads $NOTIFY_SOCKET from the environment of the httpd process.
        """
        assert self._notify is None
        self._notify = NotifyListener(
            os.path.join(self.server_dir, 'systemd-notify.sock'))
        self.set_httpd_env('NOTIFY_SOCKET', self._notify.path)
        return self._notify

    def stop_notify_listener(self):
        if self._notify is not None:
            self._notify.close()
            self._notify = None

    def server_env(self) -> Dict[str, str]:
        """The environment httpd is started with, as apachectl gets it."""
        return self._clean_path_env()

    @property
    def httpd_bin(self) -> str:
        return os.path.join(self.bin_dir, 'httpd')


class NotifyListener:
    """A stand-in for the systemd notification socket.

    sd_notify(3) does nothing more than send a datagram to the AF_UNIX
    socket named by $NOTIFY_SOCKET, so an unconnected datagram socket is
    enough to observe everything mod_systemd reports, with no systemd
    instance and no privileges involved.

    Datagrams are drained by a background thread so that the periodic
    notifications from the monitor hook cannot fill the socket buffer
    while a test is doing something else.
    """

    def __init__(self, path: str):
        # sockaddr_un is limited to 108 bytes; well within reach for a
        # source tree in a home directory, but check rather than fail
        # obscurely inside bind().
        assert len(path) < 100, f"notification socket path too long: {path}"
        if os.path.exists(path):
            os.unlink(path)
        self.path = path
        self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        self._sock.bind(path)
        self._sock.settimeout(0.1)
        self._lock = threading.Lock()
        self._messages: List[Dict[str, str]] = []
        self._stop = threading.Event()
        self._thread = threading.Thread(target=self._drain, daemon=True)
        self._thread.start()

    def _drain(self):
        while not self._stop.is_set():
            try:
                data = self._sock.recv(8192)
            except socket.timeout:
                continue
            except OSError:
                break
            try:
                msg = self.parse(data)
            except Exception as ex:
                # Never let one unreadable datagram stop the listener: the
                # tests would then see silence rather than a failure.
                log.warning(f"undecodable notification {data!r}: {ex}")
                continue
            log.debug(f"notify: {msg}")
            with self._lock:
                self._messages.append(msg)

    def close(self):
        self._stop.set()
        self._thread.join(timeout=2)
        self._sock.close()
        if os.path.exists(self.path):
            os.unlink(self.path)

    @staticmethod
    def parse(data: bytes) -> Dict[str, str]:
        """Split one notification datagram into its NAME=VALUE assignments.

        The last line carries no trailing newline in the MAINPID
        notification, and a value may itself contain '='.
        """
        msg = {}
        for line in data.decode(errors='replace').split('\n'):
            name, sep, value = line.partition('=')
            if sep:
                msg[name] = value
        return msg

    @property
    def messages(self) -> List[Dict[str, str]]:
        with self._lock:
            return list(self._messages)

    def clear(self):
        with self._lock:
            self._messages.clear()

    def wait_for(self, match: Callable[[Dict[str, str]], bool],
                 timeout: float = 5.0) -> Optional[Dict[str, str]]:
        """Return the first message satisfying `match`, waiting for it to
        arrive if it has not already.  Returns None on timeout."""
        end = time.time() + timeout
        seen = 0
        while True:
            with self._lock:
                pending = self._messages[seen:]
                seen = len(self._messages)
            for msg in pending:
                if match(msg):
                    return msg
            if time.time() >= end:
                return None
            time.sleep(0.05)

    def wait_for_key(self, key: str, timeout: float = 5.0) \
            -> Optional[Dict[str, str]]:
        return self.wait_for(lambda m: key in m, timeout=timeout)

    def wait_for_status(self, pattern: str, timeout: float = 5.0) \
            -> Optional[Dict[str, str]]:
        rx = re.compile(pattern)
        return self.wait_for(lambda m: 'STATUS' in m and rx.search(m['STATUS']),
                             timeout=timeout)

    def statuses(self) -> List[str]:
        return [m['STATUS'] for m in self.messages if 'STATUS' in m]


# The STATUS= line the monitor hook reports, from systemd_monitor() in
# modules/arch/unix/mod_systemd.c.  Idle and busy are the percentages
# ap_get_sload() computes, and are -1 when there are no workers at all.
MONITOR_STATUS = re.compile(
    r'^Total requests: (?P<requests>\d+);\s*'
    r'Idle/Busy workers (?P<idle>-?\d+)%/(?P<busy>-?\d+)%;\s*'
    r'Requests/sec: (?P<rate>\S+);\s*'
    r'Bytes served/sec: (?P<bps>.*)B/sec$')

# ap_run_monitor() is called every INTERVAL_OF_WRITABLE_PROBES (10) turns of
# the ~1s parent loop in ap_wait_or_timeout(), so a status update is up to
# roughly 10 seconds away.  Waiting for one costs that; waiting to be sure
# none is coming costs the whole timeout, so keep it to a small multiple.
MONITOR_TIMEOUT = 25.0

# Showing that no report is coming costs the whole wait, so it only has to
# comfortably outlast one turn of that cycle.
NO_MONITOR_TIMEOUT = 15.0


# The keep-alive notification is sent from the same monitor hook, so waiting
# for one costs the same as waiting for a status report.
WATCHDOG_TIMEOUT = MONITOR_TIMEOUT

# The shortest WatchdogSec mod_systemd will accept without complaining, which
# is twice the monitor interval: the recommended keep-alive period is half the
# watchdog timeout, and half of anything shorter than this is out of reach of a
# hook which runs every ten seconds.  Keep in step with mod_systemd.c.
SYSTEMD_MONITOR_INTERVAL = 10
MIN_WATCHDOG_SEC = 2 * SYSTEMD_MONITOR_INTERVAL


def is_watchdog_ping(msg: Dict[str, str]) -> bool:
    """Whether a notification carries the keep-alive ping.

    mod_systemd is free to send WATCHDOG=1 in a datagram of its own or
    alongside whatever else it is reporting, so match on the assignment
    rather than on the message being only that.
    """
    return msg.get('WATCHDOG') == '1'


# A configuration for a server run directly rather than through apachectl,
# sharing the server root, module list and error log with the rest of the
# suite but with its own pid file and port.
STANDALONE_CONF = """
ServerRoot "${server_dir}"
DefaultRuntimeDir logs
PidFile "${pidfile}"
Include "conf/${modules_conf}"
ServerName standalone.test
ErrorLog "logs/error_log"
LogLevel ${loglevel}
DocumentRoot "${server_dir}/htdocs"
<Directory "${server_dir}/htdocs">
    Require all granted
</Directory>
<IfModule mod_ssl.c>
    SSLSessionCache "shmcb:ssl_gcache_data(32000)"
</IfModule>
${extra}
Listen ${port}
"""


def write_server_conf(env: SystemdTestEnv, name: str, port: int,
                      modules_conf: str = 'modules.conf',
                      extra: str = '') -> str:
    """Write a standalone configuration and return its path."""
    path = os.path.join(env.server_conf_dir, f'{name}.conf')
    with open(path, 'w') as fd:
        fd.write(STANDALONE_CONF
                 .replace('${server_dir}', env.server_dir)
                 .replace('${pidfile}',
                          os.path.join(env.server_logs_dir, f'{name}.pid'))
                 .replace('${loglevel}', 'debug' if env.verbosity else 'warn')
                 .replace('${modules_conf}', modules_conf)
                 .replace('${extra}', extra)
                 .replace('${port}', str(port)))
    return path


def http_responds(port: int, timeout: float = 2.0) -> bool:
    """One HTTP request, without curl, so that a listening socket which
    nothing is serving cannot be mistaken for a running server: under socket
    activation the listener exists before httpd does."""
    try:
        with socket.create_connection(('127.0.0.1', port), 1.0) as c:
            c.settimeout(timeout)
            c.sendall(b'GET / HTTP/1.0\r\nHost: standalone.test\r\n\r\n')
            return c.recv(64).startswith(b'HTTP/1.')
    except OSError:
        return False


def error_log_size(env: SystemdTestEnv) -> int:
    """Where the shared error log ends now, so that a later read can take
    only what one operation wrote."""
    try:
        return os.path.getsize(env.httpd_error_log.path)
    except OSError:
        return 0


def error_log_since(env: SystemdTestEnv, pos: int) -> str:
    """The error log written since error_log_size() returned pos."""
    try:
        with open(env.httpd_error_log.path, errors='replace') as fd:
            fd.seek(pos)
            return fd.read()
    except OSError:
        return ''


class ActivatedServer:
    """An httpd handed a listening socket the way a service manager does.

    The protocol is only $LISTEN_FDS descriptors starting at 3, and
    $LISTEN_PID naming the process they were meant for, so the test opens
    the socket and speaks it directly.  systemd-socket-activate would do
    the same, but its --now option is too recent to rely on, and this
    needs no systemd tooling at all.

    apachectl cannot pass descriptors, so httpd is run directly, in the
    foreground: LISTEN_PID has to be the process which calls
    sd_listen_fds(), and a daemonised parent would not be it.
    """

    def __init__(self, env: SystemdTestEnv, port: int, name: str = 'activate',
                 extra: str = '', listen_port: int = None,
                 modules_conf: str = 'modules.conf'):
        self.env = env
        self.port = port
        # The port systemd-socket-activate binds, which is the same as the
        # configured one unless a test wants them to disagree.
        self.listen_port = port if listen_port is None else listen_port
        self.name = name
        self.pid_file = os.path.join(env.server_logs_dir, f'{name}.pid')
        self.proc = None
        self.stdout = None
        self.stderr = None
        self.conf_file = write_server_conf(env, name, port,
                                           modules_conf=modules_conf,
                                           extra=extra)

    @staticmethod
    def modules_conf_without(env: SystemdTestEnv, module: str) -> str:
        """Write a copy of the generated module list with one module left
        out, to check what happens when it is not loaded."""
        name = f'modules-no-{module}.conf'
        src = os.path.join(env.server_conf_dir, 'modules.conf')
        rx = re.compile(rf'^\s*LoadModule\s+{module}_module\b')
        with open(src) as fd:
            lines = [l for l in fd if not rx.match(l)]
        with open(os.path.join(env.server_conf_dir, name), 'w') as fd:
            fd.writelines(lines)
        return name

    def args(self, fd: int) -> List[str]:
        # The shell moves the inherited socket to descriptor 3 and names
        # itself in LISTEN_PID before exec'ing httpd in its place, which is
        # the one thing this cannot do from the parent: the pid has to be
        # the one which will call sd_listen_fds().  bash rather than sh
        # because dash parses only one digit in a redirection, and the
        # socket lands well above descriptor 9.
        return [
            'bash', '-c',
            f'exec 3<&{fd}; export LISTEN_FDS=1 LISTEN_FDNAMES=activate '
            'LISTEN_PID=$$; exec "$0" "$@"',
            self.env.httpd_bin, '-DFOREGROUND',
            '-d', self.env.server_dir, '-f', self.conf_file,
        ]

    def start(self) -> 'ActivatedServer':
        lsock = socket.create_server(('', self.listen_port),
                                     family=socket.AF_INET6,
                                     dualstack_ipv6=True, backlog=128)
        try:
            # A new session so that the whole group can be signalled on the
            # way out.
            self.proc = subprocess.Popen(
                self.args(lsock.fileno()), env=self.env.server_env(),
                start_new_session=True, pass_fds=(lsock.fileno(),),
                stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        finally:
            # The child holds it now; keeping a copy here would leave the
            # port bound after the server is gone.
            lsock.close()
        return self

    def _reap(self):
        """Collect the output of a server which has exited."""
        if self.stderr is None and self.proc.poll() is not None:
            self.stdout, self.stderr = self.proc.communicate()

    def wait_exit(self, timeout: float = 10.0) -> int:
        """Wait for a server which is expected to fail to start."""
        self.stdout, self.stderr = self.proc.communicate(timeout=timeout)
        return self.proc.returncode

    def is_live(self, timeout: float = 10.0) -> bool:
        end = time.time() + timeout
        while True:
            if self.proc.poll() is not None:
                # Collect its diagnostics, so that a test reporting the
                # server did not come up can say why.
                self._reap()
                return False
            if http_responds(self.port):
                return True
            if time.time() >= end:
                return False
            time.sleep(0.2)

    def is_running(self, settle: float = 2.0) -> bool:
        """Whether the server is still up once it has had time to fail.

        A restart which cannot find its sockets takes a few milliseconds to
        bring the parent down, and its old children go on serving after it,
        so neither an immediate check nor a request proves anything.
        """
        end = time.time() + settle
        while time.time() < end:
            if self.proc.poll() is not None:
                return False
            time.sleep(0.1)
        return True

    def reload(self) -> int:
        """Ask the running server to restart gracefully."""
        r = self.env.run([self.env.httpd_bin, '-d', self.env.server_dir,
                          '-f', self.conf_file, '-k', 'graceful'],
                         env=self.env.server_env())
        return r.exit_code

    def _signal_group(self, sig: int) -> bool:
        """Signal every process still in the server's group, reporting
        whether any remained.  start_new_session() made the process we
        launched the group leader, so its pid is the group id whether or
        not it is still alive."""
        try:
            os.killpg(self.proc.pid, sig)
            return True
        except OSError:
            return False

    def stop(self):
        if self.proc is None:
            return
        self._signal_group(signal.SIGTERM)
        if self.proc.poll() is None:
            try:
                self.stdout, self.stderr = self.proc.communicate(timeout=10)
            except subprocess.TimeoutExpired:
                self._signal_group(signal.SIGKILL)
                self.stdout, self.stderr = self.proc.communicate()
        # A parent which died during a failed restart leaves its children
        # behind, still holding the listening socket and still answering.
        # They have to go too, or the next test finds the port taken.
        end = time.time() + 5
        while self._signal_group(0):
            if time.time() >= end:
                self._signal_group(signal.SIGKILL)
                break
            time.sleep(0.1)

    def __enter__(self) -> 'ActivatedServer':
        return self.start()

    def __exit__(self, *args):
        self.stop()


# Type=notify-reload and ReloadSignal= both arrived in systemd 253.  An
# unrecognised Type= does not degrade to anything, it stops the unit loading
# at all, so there is nothing to fall back to and the tests are skipped.
NOTIFY_RELOAD_VERSION = 253


def systemd_version() -> int:
    """The version of the systemd on this host, or 0 if it cannot be asked.
    "systemctl --version" opens with "systemd 259 (259.8-1.fc44)"."""
    try:
        p = subprocess.run(['systemctl', '--version'], capture_output=True,
                           text=True, timeout=15)
    except (OSError, subprocess.TimeoutExpired):
        return 0
    m = re.match(r'systemd (\d+)', p.stdout)
    return int(m.group(1)) if m else 0


class TransientService:
    """httpd run as a real transient systemd unit, with systemd-run.

    This is the only harness here which exercises the notification protocol
    against systemd itself rather than a stand-in socket: systemd provides
    NOTIFY_SOCKET, holds the service in "activating" until READY=1 arrives,
    tracks MAINPID, and shows the reported STATUS= as the unit's status
    text.  It needs a per-user service manager, which a login session has
    but a bare CI container does not.

    service_type selects what is exercised: "notify" for startup and
    shutdown, "notify-reload" for the reload protocol on top of them.
    """

    def __init__(self, env: SystemdTestEnv, port: int,
                 name: str = None, extra: str = '',
                 service_type: str = 'notify', exec_reload: bool = True,
                 properties: List[str] = None):
        self.env = env
        self.port = port
        self.unit = name or f'httpd-test-{os.getpid()}'
        self.service_type = service_type
        self.exec_reload = exec_reload
        self.conf_file = write_server_conf(env, 'transient', port, extra=extra)
        self.pid_file = os.path.join(env.server_logs_dir, 'transient.pid')
        # Extra --property arguments for the unit, such as WatchdogSec= or
        # ReloadSignal=.
        self.properties = list(properties or [])

    def read_pid(self) -> Optional[int]:
        try:
            with open(self.pid_file) as fd:
                return int(fd.read().strip())
        except (OSError, ValueError):
            return None

    @staticmethod
    def is_available() -> bool:
        """Whether this user has a systemd manager to run services under."""
        if shutil.which('systemd-run') is None:
            return False
        if not os.environ.get('XDG_RUNTIME_DIR'):
            return False
        # Talking to the manager at all is the test: without a login
        # session, or with lingering off, there is none to talk to.
        try:
            return subprocess.run(['systemctl', '--user', 'show', '-p',
                                   'Version'], capture_output=True,
                                  timeout=15).returncode == 0
        except (OSError, subprocess.TimeoutExpired):
            return False

    def systemctl(self, *args,
                  timeout: float = 60.0) -> subprocess.CompletedProcess:
        return subprocess.run(['systemctl', '--user', *args],
                              capture_output=True, text=True, timeout=timeout)

    def show(self, prop: str) -> str:
        r = self.systemctl('show', '-p', prop, '--value', f'{self.unit}.service')
        return r.stdout.strip()

    def start(self, timeout: float = 20.0) -> subprocess.CompletedProcess:
        httpd = self.env.httpd_bin
        props = [
            f'--service-type={self.service_type}',
            '--property=KillMode=mixed',
            # A reload which is never reported finished holds the job open
            # until this elapses, so keep it to the same bound as the start.
            f'--property=TimeoutStartSec={int(timeout)}',
        ]
        if self.exec_reload:
            props.append(f'--property=ExecReload={httpd} '
                         f'-d {self.env.server_dir} '
                         f'-f {self.conf_file} -k graceful')
        props += [f'--property={p}' for p in self.properties]
        r = subprocess.run([
            'systemd-run', '--user', '--collect', '--quiet',
            '--unit', self.unit, *props,
            httpd, '-DFOREGROUND',
            '-d', self.env.server_dir, '-f', self.conf_file,
        ], capture_output=True, text=True, timeout=timeout)
        return r

    def reload(self, timeout: float = 60.0) -> subprocess.CompletedProcess:
        """Ask the manager to reload the unit.  Under Type=notify-reload
        this returns once the server has reported the reload finished;
        under Type=notify, as soon as ExecReload= has exited."""
        return self.systemctl('reload', f'{self.unit}.service',
                              timeout=timeout)

    def rewrite_conf(self, extra: str = ''):
        """Replace the configuration the unit reads on its next reload.
        The path does not change, so ExecStart= and ExecReload= still name
        it."""
        self.conf_file = write_server_conf(self.env, 'transient', self.port,
                                           extra=extra)

    def wait_active(self, timeout: float = 20.0) -> bool:
        end = time.time() + timeout
        while time.time() < end:
            if self.show('ActiveState') == 'active':
                return True
            time.sleep(0.2)
        return False

    def stop(self):
        self.systemctl('stop', f'{self.unit}.service')
        self.systemctl('reset-failed', f'{self.unit}.service')

    def __enter__(self) -> 'TransientService':
        return self

    def __exit__(self, *args):
        self.stop()


class ForegroundServer(ActivatedServer):
    """httpd run directly in the foreground, with extra environment.

    The watchdog protocol is keyed to a process id: sd_watchdog_enabled(3)
    ignores $WATCHDOG_USEC unless $WATCHDOG_PID is unset or names the
    process reading it.  apachectl cannot be used to set that, since the
    variable has to name the parent httpd and the pid is not known until it
    exists, so the value is assigned in a shell which then exec's httpd in
    its own place -- the same trick ActivatedServer uses for $LISTEN_PID.

    Assignments are shell words, so "$$" in a value is the pid httpd will
    have.
    """

    def __init__(self, env: SystemdTestEnv, port: int,
                 name: str = 'foreground', extra: str = '',
                 setenv: Dict[str, str] = None,
                 modules_conf: str = 'modules.conf'):
        super().__init__(env, port, name=name, extra=extra,
                         modules_conf=modules_conf)
        self.setenv = dict(setenv or {})

    def args(self, fd: int = None) -> List[str]:
        assigns = ' '.join(f'{k}={v}' for k, v in self.setenv.items())
        return [
            'bash', '-c',
            f'export {assigns}; exec "$0" "$@"' if assigns else 'exec "$0" "$@"',
            self.env.httpd_bin, '-DFOREGROUND',
            '-d', self.env.server_dir, '-f', self.conf_file,
        ]

    def start(self) -> 'ForegroundServer':
        # A new session, so the whole group can be signalled on the way out
        # even when a failed restart leaves children behind.
        self.proc = subprocess.Popen(
            self.args(), env=self.env.server_env(), start_new_session=True,
            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return self