psdiff

#!/usr/bin/env python
# psdiff (part of ossobv/vcutil) // wdoekes/2016-2020,2021,2023,2025-2026
#   // Public Domain
#
# Generic (coarse) monitoring of daemon processes. Use in conjunction
# with a monitoring suite like Zabbix.
#
import argparse
import collections
import difflib
import os
import re
import subprocess
import sys
import textwrap
import time
import warnings


DBNAME = '/var/lib/psdiff.db'


def human_key(s):
    """Sort key that handles numbers naturally. Works for IPs, ports, etc."""
    # On Python3, tuple([]) is slightly faster than [] or tuple() when passing
    # the result to sort, which we do.
    return tuple([
        int(token) if token.isdigit() else token
        for token in human_key.re.split(s)
        if token])  # against leading and trailing empty separators
human_key.re = re.compile(r'(\d+)')  # noqa


class Process(object):
    """
    A single process as found in the psdiff.db: an indented cmdline with
    properties.

    The line:

        '  INIT  {user=root, listen=(tcp:0.0.0.0:22,tcp:[::]:22)}'

    is parsed by from_line() into:

        Process(indent=1, cmdline='INIT', user='root',
                listen=('tcp:0.0.0.0:22', 'tcp:[::]:22'))

    and to_line() turns it back into the exact same line. Properties we
    do not know about are kept -- in order -- so a db written by a newer
    psdiff does not lose information when it is read by an older one.
    """
    # 'cmdline  {user=root, listen=(a,b)}': the cmdline is greedy, so we
    # split on the last '  {', not on one that is part of the cmdline.
    re_line = re.compile(r'^( *)(.*)  \{(.*)\}$')
    # ', ' separates properties, but only when a 'key=' follows: values
    # (like a listen list) may contain a comma too.
    re_prop = re.compile(r', (?=[a-z_]+=)')
    # Properties that hold zero or more values instead of exactly one.
    tuple_props = ('listen',)

    @classmethod
    def from_line(cls, line):
        match = cls.re_line.match(line)
        if not match:
            raise ValueError('unparsable psdiff.db line: {!r}'.format(line))
        indent, cmdline, propstr = match.groups()

        props = collections.OrderedDict(
            cls.parse_prop(prop) for prop in cls.re_prop.split(propstr))
        try:
            user = props.pop('user')
        except KeyError:
            raise ValueError(
                'psdiff.db line without user: {!r}'.format(line))

        return cls(len(indent) // 2, cmdline, user, props=props)

    @classmethod
    def parse_prop(cls, prop):
        "Parse 'user=root' or 'listen=(a,b)' into a (key, value) tuple."
        key, sep, value = prop.partition('=')
        if not sep:
            raise ValueError('unparsable psdiff.db property: {!r}'.format(
                prop))
        if value.startswith('(') and value.endswith(')'):
            value = tuple(i for i in value[1:-1].split(',') if i)
        elif key in cls.tuple_props:
            value = (value,)
        return key, value

    @staticmethod
    def format_value(value):
        "The inverse of parse_prop() value handling."
        if not isinstance(value, tuple):
            return value
        if len(value) == 1:
            return value[0]  # a single value is written without parens
        return '({})'.format(','.join(value))

    def __init__(self, indent, cmdline, user, listen=(), props=None):
        self.indent = indent
        self.cmdline = cmdline
        self.user = user
        # Everything after 'user=', in the order we found it. Storing
        # listen in here -- instead of next to it -- keeps the order of
        # any unknown properties intact when we write the line back.
        self.props = collections.OrderedDict(props or ())
        if listen:
            self.listen = listen

    @property
    def listen(self):
        return self.props.get('listen', ())

    @listen.setter
    def listen(self, value):
        if value:
            self.props['listen'] = tuple(value)
        else:
            self.props.pop('listen', None)

    def without_listen(self):
        "Return a copy without listen info, for --no-net on a --net db."
        return type(self)(
            self.indent, self.cmdline, self.user, props=[
                (key, value) for key, value in self.props.items()
                if key != 'listen'])

    def same_process(self, other):
        "Whether this is the same process, listen info excluded."
        if (self.indent != other.indent or self.cmdline != other.cmdline or
                self.user != other.user):
            return False
        keys = set(self.props) | set(other.props)
        keys.discard('listen')
        return all(
            self.props.get(key) == other.props.get(key) for key in keys)

    def to_line(self):
        "Return the single db line for this process."
        return self.render(expand_listen=False)[0]

    def to_diff_lines(self):
        """
        Return (indent, head, items, tail) for multiline diff output.

        Without listen info, items and tail are empty and head holds the
        entire line (sans indent). With listen info, every address gets
        its own line, so that a diff between two of these shows only the
        addresses that actually changed:

            /usr/sbin/named  {user=bind, listen=(
                tcp:0.0.0.0:53,
                udp:0.0.0.0:53,
            )}
        """
        indent = self.indent * u'  '
        head, items, tail = self.render(expand_listen=True)
        return indent, head[len(indent):], items, tail

    def render(self, expand_listen=False):
        """
        Return (head, items, tail); the single source of db line layout.

        Unless expand_listen is set -- and there is listen info -- items
        and tail are empty and head is the complete line.
        """
        expand_listen = bool(expand_listen and self.listen)
        head = [self.indent * u'  ', self.cmdline, u'  {user=', self.user]
        items = ()
        tail = []

        current = head
        for key, value in self.props.items():
            if key == 'listen' and expand_listen:
                current.append(u', listen=(')
                items = self.listen
                current = tail  # any next properties go after the ')'
                current.append(u')')
            else:
                current.append(u', {}={}'.format(key, self.format_value(
                    value)))
        current.append(u'}')

        return u''.join(head), items, u''.join(tail)

    def sort_key(self):
        return (self.indent, self.cmdline, self.user,
                tuple(self.props.items()))

    def __hash__(self):
        return hash(self.sort_key())

    def __eq__(self, other):
        return (
            self.indent == other.indent and
            self.cmdline == other.cmdline and
            self.user == other.user and
            self.props == other.props)

    def __ne__(self, other):  # py2
        return not self.__eq__(other)

    def __lt__(self, other):
        return self.sort_key() < other.sort_key()

    def __repr__(self):
        extra = ''.join(
            ', {}={!r}'.format(key, value)
            for key, value in self.props.items())
        return 'Process(indent={!r}, cmdline={!r}, user={!r}{})'.format(
            self.indent, self.cmdline, self.user, extra)

    def __str__(self):
        return self.to_line()


class LiveProcess(object):
    """
    A process as found in the ps(1) output: part of a pid/ppid tree.
    """
    split = re.compile(r'\s+')

    @classmethod
    def from_line(cls, line, root):
        args = cls.split.split(line, 3)

        user = args[0]
        pid = int(args[1])
        ppid = int(args[2])
        exe = args[3][0:8].rstrip()
        assert args[3][8] == ' '
        cmdline = args[3][9:]

        return cls(ppid, pid, user, exe, cmdline, root=root)

    def __init__(self, parent, pid, user, exe, cmdline, root=None):
        self.parent = parent
        self.pid = pid
        self.user = user
        self.exe = exe
        self.cmdline = cmdline
        self.root = root or self
        self.include = True

        if not root:
            self.process_map = {}

        self.root.process_map[pid] = self
        self.children = set()

        # Enriched info
        self._listens = set()

    def skip(self):
        self.include = False

        # Move the _listens to the parent.
        if self._listens:
            obj = self
            while obj:
                obj = obj.parent
                if obj.include:
                    break
                assert not obj._listens, (self, obj)
            obj.add_listens(self._listens)
            self._listens = frozenset()  # make it empty/unusable

    def has_parent(self, include_self=False,
                   cmdline__startswith=None, pid=None):
        obj = self
        if not include_self:
            obj = obj.parent

        while obj:
            if (cmdline__startswith is not None and
                    obj.cmdline.startswith(cmdline__startswith)):
                return True
            if pid is not None and obj.pid == pid:
                return True
            obj = obj.parent

        return False

    def fix_links(self):
        if self.parent is not None:
            # Convert ppid to parent.
            self.parent = self.root.process_map[self.parent]
            # Add us as child of the parent.
            self.parent.children.add(self)

    def get_process(self, pid):
        if not pid:
            return None
        return self.root.process_map[pid]

    def to_process(self, indent=0):
        "Return the db representation (a Process) of this process."
        return Process(
            indent, self.cmdline.rstrip(), self.user, listen=self._listens)

    def to_string(self, indent=0):
        return self.to_process(indent).to_line()

    def sort(self):
        # Sort the children and convert the set into a list.
        for child in self.children:
            child.sort()
        self.children = tuple(sorted(self.children))
        # Sort the listen addresses the same way we write them out, so
        # that comparison and output agree on the order.
        self._listens = tuple(sorted(self._listens, key=human_key))

    def add_listens(self, listens):
        self._listens |= set(listens)

    def __hash__(self):
        # Needs to be reimplemented because Python3 drops the
        # auto-generated one when __eq__ is defined.
        return id(self)

    def __eq__(self, other):
        # Not only identity comparison yields same, otherwise
        # we end up comparing children as unequal and then
        # sorting of parents fails because non-id siblings
        # aren't considered.
        return (
            self.include == other.include and
            self.cmdline == other.cmdline and
            self.user == other.user and
            self._listens == other._listens and
            self.children == other.children)

    def __lt__(self, other):
        if self.include != other.include:
            return (self.include > other.include)  # first True

        # Lazy comparison.
        if self.cmdline != other.cmdline:
            return (self.cmdline < other.cmdline)
        if self.user != other.user:
            return (self.user < other.user)
        if self._listens != other._listens:
            return (self._listens < other._listens)
        assert isinstance(self.children, tuple), self.children
        assert isinstance(other.children, tuple), other.children
        return (self.children < other.children)

    def __repr__(self):
        return '<LiveProcess({}, user={}, id={})>'.format(
            self.cmdline, self.user, id(self))

    def __str__(self):
        return self.to_string()


class ProcessFormatter(object):
    def __init__(self, root):
        self.root = root

        # Add self.adjust hook to alter process traits before sort.
        self.visit(self.adjust)
        # Skip processes.
        self.visit((lambda process: (
            None if self.include(process) else process.skip())))
        # Sort processes.
        self.visit((lambda process: process.sort()))

    def visit(self, callable_):
        "Visit all processes with callable."
        for process in self.root.process_map.values():
            callable_(process)

    def to_strings(self, process, indent=0):
        "Return a list of stringified children with indentation."
        ret = []
        if process.include:
            ret.append(self.to_string(process, indent))
        for child in process.children:  # has been sorted already
            ret.extend(self.to_strings(child, indent + 1))
        return ret

    def to_processes(self):
        """
        Return the listing as a list of Process items.

        This goes through to_strings() -- and thus through the
        to_string() hook that psdiff.conf files may override -- so that
        what we compare can never drift from what we write to the db.
        """
        processes = []
        for line in self.to_strings(self.root):
            try:
                processes.append(Process.from_line(line))
            except ValueError as e:
                raise ValueError(
                    '{} (from {}.to_string(); a psdiff.conf hook must '
                    'return a valid db line)'.format(
                        e, type(self).__name__))
        return processes

    def __str__(self):
        return u'\n'.join(self.to_strings(self.root)) + '\n'

    def adjust(self, process):
        """
        The possibility to adjust cmdline and other process traits.

        This is called before sort, so you'll want to use this to alter
        cmdline.
        """
        pass

    def include(self, process):
        "The possibility to exclude processes from the listing."
        return True

    def to_string(self, process, indent=0):
        "The old hook to alter cmdline appearance."
        return process.to_string(indent)


class FilteredProcessFormatter(ProcessFormatter):
    def __init__(self, *args, **kwargs):
        self._include_once = set()
        super(FilteredProcessFormatter, self).__init__(*args, **kwargs)

    def adjust(self, process):
        super(FilteredProcessFormatter, self).adjust(process)

        if process.cmdline.startswith((
                'astcanary',  # astcanary /var/run/asterisk/... <pid>
                '/usr/sbin/amavisd-new ')):
            # These processes have fluctuating arguments. Drop them.
            process.cmdline = process.cmdline.split(' ', 1)[0]

        elif process.cmdline.startswith(
                'sshd: /usr/sbin/sshd -D [listener] '):
            parts = process.cmdline.split(' ', 6)
            if len(parts) == 7 and parts[4].isdigit() and parts[5] == 'of':
                parts[4] = '0'
                process.cmdline = ' '.join(parts)

        elif process.cmdline.startswith((
                '/usr/sbin/zabbix_proxy: ',
                '/usr/sbin/zabbix_server: ')):
            # zabbix_proxy and zabbix_server add " [info]" which changes.
            # Drop it.
            process.cmdline = process.cmdline.split(' [', 1)[0]

        elif process.cmdline.startswith((
                '/usr/bin/containerd-shim-runc-v2 ',
                'containerd-shim ',
                'docker-containerd-shim ')):
            # Docker 18+ instances have fluctuating arguments:
            #   /usr/bin/containerd-shim-runc-v2 -namespace moby -id <ID> ...
            #   containerd-shim ... -workdir /var/...containerd/<ID> ...
            args = process.cmdline.split()
            if '-id' in args:
                # -id [ID]
                pos = args.index('-id') + 1
                if pos < len(args):
                    args[pos] = '<ID>'
            if '-workdir' in args:
                # -workdir [PATH]
                # /var/lib/containerd/io.containerd.runtime.v1.linux/moby/<ID>
                pos = args.index('-workdir') + 1
                if pos < len(args):
                    args[pos] = args[pos].rsplit('/', 1)[0] + '/<ID>'
            if len(args) == 4 and args[3] == 'docker-runc':
                # Docker 17- instances have fluctuating arguments:
                #   docker-containerd-shim <ID> /var/...containerd/<ID> \
                #     docker-runc
                args[1] = '<ID>'
                args[2] = args[2].rsplit('/', 1)[0] + '/<ID>'

            process.cmdline = ' '.join(args)

    def include(self, process):
        # Ignore kernel threads.
        if process.has_parent(include_self=True, pid=2):
            return False

        # Systemd renames itself after an update. We can't rename it
        # back to /sbin/init because it may have been called differently
        # (/sbin/init splash or whatever) in the first place.
        elif process.pid == 1:
            # /sbin/init [splash]
            # /lib/systemd/systemd --system --deserialize 19
            process.cmdline = 'INIT'

        # Children of these commands are generally not daemons, skip
        # them:
        elif process.has_parent(include_self=True, cmdline__startswith=(
                'CRON', 'SCREEN', '-tmux',
                '/USR/SBIN/CRON',  # older cron
                '/usr/sbin/CRON',  # newer cron
                # Is a daemon, but spawns children of init for extra work.
                '/usr/bin/python /usr/bin/salt-minion',
                # Comes and goes.
                '/usr/libexec/fwupd/fwupd')):
            return False

        # We want to monitor these daemons, but not their
        # (grand)children, as they come and go:
        elif process.has_parent(include_self=False, cmdline__startswith=(
                '/usr/sbin/dovecot',
                '/usr/sbin/gocollect',              # gc: sysv/systemd
                'gocollect',                        # gc: ubuntu (upstart)
                '/usr/lib/postfix/master',          # pf: debian/ubuntu
                '/usr/lib/postfix/sbin/master',     # pf: ubuntu16.04+
                '/usr/libexec/postfix/master',      # pf: redhat
                '/usr/lib/postgresql/',
                '/usr/sbin/sshd ',                  # sshd: sysv
                'sshd: /usr/sbin/sshd ',            # sshd: modern
                '/lib/systemd/systemd-udevd',       # udev: pre-usrmerge path
                '/usr/lib/systemd/systemd-udevd',   # udev: modern
                '/usr/sbin/vsftpd',
                '/usr/sbin/zabbix_agent2',
                '/usr/sbin/zabbix_agentd')):
            return False

        # These children may come and go, but we expect at least one:
        # - multiprocess apache creates at least N processes but may add/remove
        #   based on demand
        elif process.cmdline.startswith((
                '/usr/sbin/apache2 ',       # debian/ubuntu
                '/usr/sbin/httpd ',         # redhat
                'php-fpm: ')):
            key = (process.parent.pid, process.user, process.cmdline)
            if key in self._include_once:
                return False
            else:
                self._include_once.add(key)

        # These ones may come and go. Don't care if they exist.
        # - uuidd gets spawned at will
        # - zfs get can be slow; it's used by various processes (like kubelet)
        elif ((process.user == 'uuidd' and
                process.cmdline == '/usr/sbin/uuidd --socket-activation') or
                (process.user == 'root' and
                 process.cmdline.startswith(('zfs get ', 'zfs list')))):
            return False

        # Special case since Ubuntu noble (systemd 255), we expect this one in
        # combination with sd-pam, but no other children.
        elif process.cmdline in (
                '/lib/systemd/systemd --user',
                '/usr/lib/systemd/systemd --user') and all(
                    child.exe == '(sd-pam)' and not child.children
                    for child in process.children):
            return False

        # (sd-pam) comes and goes. Generally as child of systemd --user.
        elif process.cmdline == process.exe == '(sd-pam)':
            return False

        return super(FilteredProcessFormatter, self).include(process)


def diff_items(alines, blines):
    """
    Diff two sequences of hashable items (Process items, or the listen
    addresses of a single Process).

    Returns a list of (direction, item) tuples where direction is
    -1 (removed), 0 (unchanged/context) or 1 (added):

        # alines: ['aa','bb','cc']
        # blines: ['bb','dd','ff']
        [(-1, 'aa'), (0, 'bb'), (-1, 'cc'), (1, 'dd'), (1, 'ff')]

    We use SequenceMatcher.get_opcodes() directly (the same primitive
    difflib.unified_diff and difflib.ndiff build on) rather than ndiff
    itself: ndiff takes exponential time on some machines for psdiff
    dumps of ~200 lines, which isn't acceptable.
    See: https://bugs.python.org/issue6931

    Unlike unified_diff, we don't truncate to N lines of context around
    each change: callers that don't care (the top-level whole-process
    diff) drop the context items themselves, and callers that diff the
    listen addresses of a process need every address as context, not
    just the ones close to a change.
    """
    changes = []
    for tag, i1, i2, j1, j2 in (
            difflib.SequenceMatcher(None, alines, blines).get_opcodes()):
        if tag == 'equal':
            changes.extend((0, item) for item in alines[i1:i2])
        else:
            if tag in ('delete', 'replace'):
                changes.extend((-1, item) for item in alines[i1:i2])
            if tag in ('insert', 'replace'):
                changes.extend((1, item) for item in blines[j1:j2])
    return changes


def parse_db(data):
    """
    Parse psdiff.db contents into a list of Process items.
    """
    processes = []
    for lineno, line in enumerate(data.split(u'\n'), 1):
        if not line:
            continue  # the trailing LF, mostly
        try:
            processes.append(Process.from_line(line))
        except ValueError as e:
            raise ValueError('{} (line {})'.format(e, lineno))
    return processes


def format_db(processes):
    """
    The inverse of parse_db(): the psdiff.db contents for these
    Process items.
    """
    return u''.join(
        u'{}\n'.format(process.to_line()) for process in processes)


def group_process_changes(changes):
    """
    Post-process diff_items() output: pair up adjacent removed/added
    processes that are the same process -- only their listen addresses
    differ -- so they can be diffed by listen address instead of as two
    unrelated processes.

    Yields ('single', direction, process) for a process shown on its
    own, or ('pair', old_process, new_process) for a same-process pair.
    """
    i, n = 0, len(changes)
    while i < n:
        direction, process = changes[i]
        if direction >= 0:
            yield ('single', direction, process)
            i += 1
            continue

        removed = []
        while i < n and changes[i][0] < 0:
            removed.append(changes[i][1])
            i += 1
        added = []
        while i < n and changes[i][0] > 0:
            added.append(changes[i][1])
            i += 1

        unmatched_added = list(added)
        for old in removed:
            # Only pair when there is listen info to diff; without it
            # the pair would have nothing to show.
            match = next(
                (new for new in unmatched_added
                 if (old.listen or new.listen) and old.same_process(new)),
                None)
            if match is None:
                yield ('single', -1, old)
            else:
                unmatched_added.remove(match)
                yield ('pair', old, match)
        for new in unmatched_added:
            yield ('single', 1, new)


def ps_faxu(with_network=False):
    cmd = ['ps', 'ax', '-o', 'user,pid,ppid,fname,args']
    try:
        output = subprocess.check_output
    except AttributeError:
        # Blegh. Python 2.6. (You did already `pip install argparse`, yes?)
        proc = subprocess.Popen(cmd, bufsize=-1, stdout=subprocess.PIPE)
        output = proc.communicate()[0]
        proc.wait()
    else:
        output = subprocess.check_output(cmd, bufsize=-1)

    # Get socket data as quickly as possible.
    sockets_by_pid = None
    if with_network:
        try:
            # Get listen addresses per pid in a dictionary:
            # {1234: ['tcp:0.0.0.0:80', 'tcp:0.0.0.0:443']}
            sockets_by_pid = netstat()
        except (subprocess.CalledProcessError, OSError):
            # If there is no 'ss' (super-netstat) or we have too few perms
            # to use it fruitfully, accept it.
            # - CalledProcessError = ss returned non-zero
            # - FileNotFoundError(OSError) = there is no ss
            # - PermissionDeniedError(OSError) = we're not allowed to call ss
            warnings.warn(
                '--net info requested but failed (no perms? no ss(1)?)')

    output = output.decode('ascii', 'replace')

    root = LiveProcess(None, 0, 'root', 'root', 'root')
    for i, line in enumerate(output.split('\n')):
        if i == 0 or not line:
            pass
        else:
            LiveProcess.from_line(line, root)

    # Update processes with proper links. This must be done last because
    # the process output is unordered and we may not have the parent
    # process info yet earlier.
    for process in root.process_map.values():
        process.fix_links()

    # We have pids and netstat has pids. Amend the processes with socket info.
    if sockets_by_pid is not None:
        process_pids = set(root.process_map.keys())
        socket_pids = set(sockets_by_pid.keys())
        both_pids = process_pids & socket_pids
        for pid in both_pids:
            root.process_map[pid].add_listens(sockets_by_pid[pid])
        for pid in (socket_pids - both_pids):
            root.process_map[0].add_listens(sockets_by_pid[pid])

    return root


def get_ephemeral_ports():
    # '32768<TAB>60999<LF>'
    with open('/proc/sys/net/ipv4/ip_local_port_range', 'r') as fp:
        ports = tuple(int(port) for port in fp.read().strip().split(None, 1))
    assert 0 <= ports[0] < ports[1] < 65536, ports
    return ports


def netstat():
    # -H no-header
    # -n numeric
    # -l listen
    # -p with-process
    cmd = ['ss', '-Hnlp']
    proc = subprocess.Popen(
        cmd, bufsize=-1, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        env={'LC_ALL': 'C'})
    output, error = proc.communicate()
    if proc.wait() != 0:
        raise ValueError('{cmd} returned {ret}: {err}'.format(
            cmd=cmd, ret=proc.returncode, err=error))

    error = error.decode('ascii', 'replace')
    # On Ubuntu/Noble with kernel 7.0, we get EINVAL on netlink query of --dccp
    # sockets. Since we're not specifying the exact list, we cannot exclude
    # it either. Silence the error instead.
    if set(error.splitlines()) != set(['RTNETLINK answers: Invalid argument']):
        print(error, end='', file=sys.stderr)

    output = output.decode('ascii', 'replace')

    # Fetch local port range.
    ephemeral_min, ephemeral_max = get_ephemeral_ports()

    output = output.strip().split('\n')
    lines = [line.rstrip().split(None, 6) for line in output]

    # Netid  State  Recv-Q  Send-Q  LocalAddr:Port  PeerAddr:Port  Proc
    proc_listens = collections.defaultdict(list)
    for line in lines:
        transport = line[0]

        # Ignore/skip unix sockets and netlink.
        if transport in ('u_dgr', 'u_seq', 'u_str', 'nl'):
            continue

        addr, port = line[4].rsplit(':', 1)  # 0.0.0.0:12345 | [::]:12345
        is_ephemeral = None  # maybe

        if len(line) < 7:
            # Some are handled by the kernel (module). Pretend we saw it
            # as kernel. (It will get appended to the INIT process.)
            if transport == 'udp' and addr in ('0.0.0.0', '[::]') and port in (
                    '51871', '6081',    # cilium/wg + geneve/wg
                    '8472',             # vxlan
                    '500', '4500',      # ipsec
                    ):
                line.append('users:(("KERNEL",pid=1,fd=-1))')
                is_ephemeral = False
            else:
                # We need root to get the process info. Bail immediately.
                raise OSError(1, 'Operation not permitted')  # EPERM

        if transport == 'p_raw' and addr == '*' and port.startswith((
                'enx', 'fwln', 'fwpr', 'lxc', 'tap')):
            # Skip these.
            continue

        if (is_ephemeral is None and transport in ('mptcp', 'tcp', 'udp')
                and ephemeral_min <= int(port) <= ephemeral_max):
            is_ephemeral = True

        if is_ephemeral and transport in ('mptcp', 'udp'):
            port = '0'  # likely the port was assigned this way using bind()

        bindaddr = '{}:{}:{}'.format(transport, addr, port)
        users = line[6]
        assert users.startswith('users:('), line
        assert users.endswith(')'), line
        idx = 7
        state = 0
        while True:
            # ("zabbix_agentd",pid=1744582,fd=6)
            if state == 0:
                assert users[idx] == '('
                state = 1
            elif state == 1:
                assert users[idx] == '"'
                state = 2
            elif state == 2:
                if users[idx] == '"':
                    state = 3
            elif state == 3:
                assert users[idx] == ','
                nidx = users.index(')', idx + 1)
                pidinfo = users[idx + 1:nidx].split(',', 1)[0]
                assert pidinfo.startswith('pid='), users
                pid = int(pidinfo[4:])
                proc_listens[pid].append(bindaddr)
                state = 4
                idx = nidx
            elif state == 4:
                if users[idx] == ')':
                    state = 5
                    break
                elif users[idx] == ',':
                    state = 0
            idx += 1

    new_listens = {}
    for key, value in proc_listens.items():
        new_listens[key] = value

    # {fd:   [addr1, addr2]}
    # {1047: ['icmp6:*:58', 'tcp:127.0.0.1:2601']}
    return new_listens


def eval_psdiff_d(dirname):
    sources = [
        os.path.join(dirname, f) for f in os.listdir(dirname)
        if not f.startswith('.') and f.endswith('.py')]

    sources.sort()
    mixins = []
    for source_file in sources:
        try:
            with open(source_file, 'r') as fh:
                source = fh.read()
            # You really don't need FilteredProcessFormatter or
            # ProcessFormatter or eval_psdiff_conf or any of that.
            # If you're going to do really fancy stuff, use custom
            # psdiff.conf and go from there.
            io = {}
            exec(source, io)
            mixins.append(io['ProcessFormatterMixin'])  # one Mixin per file
        except Exception:
            import traceback
            raise ValueError(
                'exec error reading {!r}\n\n    {}'.format(
                    source_file,
                    '\n    '.join(traceback.format_exc().split('\n'))))

    return {'LocalProcessFormatterMixins': mixins}


def eval_psdiff_conf(filename):
    with open(filename, 'r') as fh:
        source = fh.read()
    # Ooohh.. eval/exec. Supply FilteredProcessFormatter and
    # ProcessFormatter so they can be used as superclass.
    # And pass eval_psdiff_d so you can use a custom
    # psdiff.conf and _also_ load psdiff.d files.
    io = {
        'FilteredProcessFormatter': FilteredProcessFormatter,
        'ProcessFormatter': ProcessFormatter,
        'eval_psdiff_conf': eval_psdiff_conf,
        'eval_psdiff_d': eval_psdiff_d,
    }
    exec(source, io)
    return {
        'LocalFilteredProcessFormatter': io['LocalFilteredProcessFormatter']}


def get_formatter_class():
    for path in ('/usr/local/etc/psdiff.conf', '/etc/psdiff.conf'):
        # First check, and then open without exception handling. That way we
        # see if anything is wrong with permissions and such.
        if os.path.exists(path):
            return eval_psdiff_conf(path)['LocalFilteredProcessFormatter']

    # No psdiff.conf? Check for psdiff.d.
    for path in ('/usr/local/etc/psdiff.d', '/etc/psdiff.d'):
        if os.path.exists(path):
            # Mere existence of the path is enough for us to use that: if you
            # create an empty /usr/local/etc/psdiff.d, then /etc/psdiff.d will
            # NOT be used.
            # Don't forget the 'object' in ProcessFormatterMixin(object) for
            # python2.
            mixins = eval_psdiff_d(path)['LocalProcessFormatterMixins']
            class_ = type(
                'LocalFilteredProcessFormatter',
                tuple(mixins + [FilteredProcessFormatter]),
                {})
            return class_

    # Nothing found? Return the plain unaltered version.
    return FilteredProcessFormatter


def get_new_processes(formatter_class, with_network=False):
    root = ps_faxu(with_network=with_network)
    formatter = formatter_class(root)
    return formatter.to_processes()


def show_diff(changes, missing=True, extra=True):
    has_diff = False
    for item in group_process_changes(changes):
        if item[0] == 'single':
            _, direction, process = item
            if direction < 0 and missing:
                show_diff_process('-', process)
                has_diff = True
            elif direction > 0 and extra:
                show_diff_process('+', process)
                has_diff = True
        else:
            _, old, new = item
            if show_diff_processes(old, new, missing, extra):
                has_diff = True
    return has_diff


def print_line(line):
    # NOTE: We never print() with u'' below, because in py2 it would
    # "guess" the encoding of the recipient (tty) instead of choosing
    # utf-8.
    if isinstance('', bytes):  # py2
        line = line.encode('utf-8', 'replace')
    print(line)


def show_diff_process(prefix, process):
    """
    Print a process, prefixed with '-', '+' or ' '.

    A process with listen info gets a line per listen address:

        -/usr/sbin/named  {user=bind, listen=(
        -    tcp:0.0.0.0:53,
        -    udp:0.0.0.0:53,
        -)}

    so that show_diff_processes() can mark only the addresses that
    changed.
    """
    indent, head, items, tail = process.to_diff_lines()
    print_line(u'{}{}{}'.format(prefix, indent, head))
    for item in items:
        print_line(u'{}{}    {},'.format(prefix, indent, item))
    if items:
        print_line(u'{}{}{}'.format(prefix, indent, tail))


def show_diff_processes(old, new, missing, extra):
    """
    Diff two same-process items -- same cmdline/user, different listen
    addresses -- by listen address.

    The process itself and the closing ')}' are shown as context; only
    the changed addresses get a '-' or '+' prefix.
    """
    shown = [
        (direction, item)
        for direction, item in diff_items(old.listen, new.listen) if (
            direction == 0
            or (direction < 0 and missing)
            or (direction > 0 and extra))]
    if not any(direction for direction, item in shown):
        return False

    # One of the two may have no listen info at all; take the framing
    # from the one that has (they are the same process otherwise).
    indent, head, _, tail = (old if old.listen else new).to_diff_lines()

    print_line(u' {}{}'.format(indent, head))
    prefixes = {0: ' ', -1: '-', 1: '+'}
    for direction, item in shown:
        print_line(u'{}{}    {},'.format(prefixes[direction], indent, item))
    print_line(u' {}{}'.format(indent, tail))
    return True


def manual_parent(processes, parent_cmdline):
    """
    Return the (index, process) that manual add/remove works below.
    """
    for index, process in enumerate(processes):
        if process.cmdline == parent_cmdline:
            return index, process
    raise ValueError('cannot find {!r} in the db'.format(parent_cmdline))


def manual_subtree(processes, parent_index):
    """
    Return the indexes of the direct children of processes[parent_index]
    and the index right after its last (grand)child.
    """
    parent_indent = processes[parent_index].indent
    children = []
    end = len(processes)
    for index in range(parent_index + 1, len(processes)):
        indent = processes[index].indent
        if indent <= parent_indent:
            end = index
            break
        if indent == parent_indent + 1:
            children.append(index)
    return children, end


def manual_add(processes, cmdline, user, parent_cmdline='INIT'):
    """
    Return a copy of processes with a process added as a child of
    parent_cmdline, in the right sort position.
    """
    parent_index, parent = manual_parent(processes, parent_cmdline)
    new_process = Process(parent.indent + 1, cmdline, user)
    children, end = manual_subtree(processes, parent_index)

    position = end  # after the last (grand)child, unless we find a spot
    for index in children:
        if processes[index].same_process(new_process):
            raise ValueError('already in the db: {}'.format(new_process))
        if position == end and new_process < processes[index]:
            position = index  # before this sibling, and its children

    return processes[0:position] + [new_process] + processes[position:]


def manual_remove(processes, cmdline, user, parent_cmdline='INIT'):
    """
    Return a copy of processes without the child of parent_cmdline that
    has this cmdline and user -- and without its children.

    Listen info is ignored, so a process that started listening after it
    was added can be removed as well.
    """
    parent_index, parent = manual_parent(processes, parent_cmdline)
    old_process = Process(parent.indent + 1, cmdline, user)
    children = manual_subtree(processes, parent_index)[0]

    removed = set()
    for index in children:
        if processes[index].same_process(old_process):
            removed.update(range(index, manual_subtree(processes, index)[1]))

    return [
        process for index, process in enumerate(processes)
        if index not in removed]


def db_has_network(processes):
    """
    Whether the db was written with network info in it.

    Useful for backwards compatibility: only show with_network output if
    the db had it.
    """
    return any(process.listen for process in processes)


def write(output, has_old=False):
    if isinstance('', bytes):  # py2
        output = output.encode('utf-8', 'replace')

    new_name = '{}.new'.format(DBNAME.rsplit('.', 1)[0])
    old_name = '{}.old'.format(DBNAME.rsplit('.', 1)[0])
    with open(new_name, 'w') as fh:
        fh.write(output)
    if has_old:
        os.rename(DBNAME, old_name)
    os.rename(new_name, DBNAME)


def get_argv():
    """
    Fix up argv.
    """
    argv = sys.argv[:]

    # show-missing or show-extra: convert to show --missing|--extra
    if 'show-missing' in argv[1:]:
        warnings.warn(
            'Replacing show-missing with show --missing', DeprecationWarning)
        idx = argv.index('show-missing', 1)
        if argv[idx] == 'show-missing':
            argv[idx:idx + 1] = ['show', '--missing']
    elif 'show-extra' in argv[1:]:
        warnings.warn(
            'Replacing show-extra with show --extra', DeprecationWarning)
        idx = argv.index('show-extra', 1)
        if argv[idx] == 'show-extra':
            argv[idx:idx + 1] = ['show', '--extra']

    # No args? Needed on python2 where ArgumentParser does not proceed after
    # not finding an action.
    if all(arg.startswith('-') for arg in argv[1:]):
        argv.append('show')

    return argv


def main():
    parser = argparse.ArgumentParser(
        prog='psdiff',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=textwrap.dedent('''\
            Monitor differences between the list of expected running processes
            and the actual running processes.
            '''),
        epilog=textwrap.dedent('''\
            Expected usage
            --------------

            - set up server with various processes;
            - run `psdiff write' to store a dump in /var/lib/psdiff.db;
            - have zabbix (or your favorite monitoring tool) call
              `psdiff show --missing' and `psdiff show --extra';
            - have the monitoring tool show errors if there is output for any
              of the commands.

            This is just a STARTING POINT, it is NOT a replacement for DETAILED
            process monitoring. You will still want to add daemon-specific
            monitoring through other means.


            Adjustments through psdiff.conf
            -------------------------------

            On startup, an attempt is made to import /usr/local/etc/psdiff.conf
            or /etc/psdiff.conf (a python file) where it looks for a class
            called `LocalFilteredProcessFormatter', which will be used as
            formatter class instead of the builtin FilteredProcessFormatter.

            For example:

                class LocalFilteredProcessFormatter(
                        FilteredProcessFormatter):
                    def adjust(self, process):
                        super(LocalFilteredProcessFormatter, self).adjust(
                            process)

                        # haproxy(1) sometimes adds " -sf PIDLIST" at the tail
                        if process.cmdline.startswith('/usr/sbin/haproxy'):
                            process.cmdline = (
                                process.cmdline.split(' -sf ', 1)[0])

                        # Java processes get unordered arguments...
                        if process.cmdline.startswith((
                                '/usr/bin/java', 'java')):
                            args = process.cmdline.split(' ')
                            process.cmdline = ' '.join(
                                [args[0]] + sorted(args[1:]))

                    def include(self, process):
                        # atop(1) has fluctuating arguments. I don't care
                        # whether it runs.
                        if process.cmdline.startswith('/usr/bin/atop '):
                            return False

                        # Don't monitor children of LXD/LXC containers.
                        if process.has_parent(
                                #include_self=True,  # hide the monitor too?
                                cmdline__startswith='[lxc monitor] '):
                            return False

                        return (
                            super(LocalFilteredProcessFormatter, self)
                            .include(process))

                # vim: set syn=python:


            Adjustments through psdiff.d
            ----------------------------

            If psdiff.conf is not found, but a psdiff.d directory is, then all
            .py files will be loaded for ProcessFilteredMixin classes and used
            to create a big LocalFilteredProcessFormatter class through
            multiple inheritance.

            For example /etc/psdiff.d/kvm.py:

                class ProcessFormatterMixin(object):
                    def adjust(self, process):
                        super(ProcessFormatterMixin, self).adjust(process)

                        # For kvm, show only "/usr/bin/kvm -id X -name Y".
                        if process.cmdline.startswith('/usr/bin/kvm '):
                            p = process.cmdline.split()
                            new = [p[0]]
                            [new.extend(p[i:i+2])
                             for i, j in enumerate(p)
                             if j.startswith(('-id', '-name'))]
                            process.cmdline = ' '.join(new)

            Method resolution order (MRO) is alphabetical, with the
            FilteredProcessFormatter base last.


            Zabbix example
            --------------

                UserParameter=psdiff.missing,psdiff show --missing --retry 2>&1
                UserParameter=psdiff.extra,psdiff show --extra --retry 2>&1

            With triggers like this:

                {Template Role Daemons:psdiff.missing.strlen()}<>0 or
                {Template Role Daemons:psdiff.missing.nodata(30m)}=1
            '''))

    if hasattr(argparse, 'BooleanOptionalAction'):
        parser.add_argument(
            '--net', action=argparse.BooleanOptionalAction,
            dest='with_network', help='with network info')
    else:
        parser.add_argument(
            '--net', action='store_true', dest='with_network',
            help='with network info')
        parser.add_argument(
            '--no-net', action='store_false', dest='with_network',
            help='without network info')

    actions = parser.add_subparsers(dest='action', help='subcommand help')
    show_action = actions.add_parser('show', help='show current diff')
    show_action.add_argument(
        '--missing', action='store_true', help='show missing (only)')
    show_action.add_argument(
        '--extra', action='store_true', help='show extra (only)')
    show_action.add_argument(
        '--retry', action='store_true',
        help=("retry up to 2 seconds; avoids false positives caused "
              "by restarts and short lived children"))
    dump_action = actions.add_parser('dump', help='show current state')
    dump_action.add_argument(
        '--verbose', '-v', action='store_true',
        help='show diff format instead of db format')
    write_action = actions.add_parser('write', help='save current state')
    write_action.add_argument(
        '--verbose', '-v', action='store_true',
        help='show which changes are written')
    manual_action = actions.add_parser('manual', help='do manual writes')
    manual_action.add_argument(
        'manual', choices=('add', 'remove'), help='manually add/remove')
    manual_action.add_argument('user', help='process owner')
    manual_action.add_argument('cmdline', help='process cmdline')

    args = parser.parse_args(args=get_argv()[1:])

    # If no action was chosen, default to 'show' and (unfortunately) add
    # the subparser specific arguments.
    if not args.action:
        args.action = 'show'
        args.missing = args.extra = args.retry = False

    # First load up config.
    formatter_class = get_formatter_class()

    # Then load up old db.
    try:
        with open(DBNAME, 'r') as fh:
            old_output = fh.read()
        if isinstance('', bytes):  # py2
            old_output = old_output.decode('utf-8', 'replace')
    except IOError as e:
        if e.errno != 2:  # no such file
            raise
        old_output = u''

    # Python2 compatibility.
    if not hasattr(argparse, 'BooleanOptionalAction'):
        if '--net' not in sys.argv[1:] and '--no-net' not in sys.argv[1:]:
            # Otherwise it will be set to the negation of the first defined
            # argument.
            args.with_network = None

    process(args, old_output, formatter_class)


def process(args, old_output, formatter_class):
    old_processes = parse_db(old_output)

    if args.action == 'dump':
        if args.with_network is None and db_has_network(old_processes):
            args.with_network = True

        new_processes = get_new_processes(
            formatter_class, with_network=args.with_network)
        if args.verbose:
            for new_process in new_processes:
                show_diff_process('', new_process)
        else:
            print_line(format_db(new_processes)[0:-1])  # without trailing LF

    elif args.action == 'write':
        if args.with_network is None and db_has_network(old_processes):
            args.with_network = True

        new_processes = get_new_processes(
            formatter_class, with_network=args.with_network)
        new_output = format_db(new_processes)

        if old_output != new_output:
            write(new_output, has_old=bool(old_output))
            print('Wrote to {}'.format(DBNAME))
            if args.verbose:
                show_diff(diff_items(old_processes, new_processes))
        else:
            print('No changes to {}'.format(DBNAME))

    elif args.action == 'manual':
        if not old_processes:
            raise ValueError('add/remove only works for existing db')

        # For now, only accept children of init.
        if args.manual == 'add':
            new_processes = manual_add(
                old_processes, args.cmdline, args.user)
        else:
            assert args.manual == 'remove', args
            new_processes = manual_remove(
                old_processes, args.cmdline, args.user)

        if old_processes == new_processes:
            raise ValueError('no change!')
        write(format_db(new_processes), has_old=True)
        print('Wrote to {}'.format(DBNAME))
        show_diff(diff_items(old_processes, new_processes))

    else:
        assert args.action.startswith('show'), args

        # Check old db format.
        if db_has_network(old_processes):
            if args.with_network is None:
                if os.getuid() == 0:
                    # DB format has network, use it.
                    args.with_network = True
                else:
                    # DB format has network, but we cannot use it.
                    warnings.warn(
                        'Not listing --net info without root powers')

            if not args.with_network:
                # Look at the db without the network info.
                old_processes = [
                    old_process.without_listen()
                    for old_process in old_processes]

        # If neither missing nor extra is set, then we want to see both.
        if not args.missing and not args.extra:
            args.missing = args.extra = True

        # If args.retry, then try fetching a changeset for 5 times
        # before concluding that something really has changed.
        for sleeptime in (0.1, 0.3, 0.6, 1.0, 0):
            new_processes = get_new_processes(
                formatter_class, with_network=args.with_network)

            # Quick optimization.
            if old_processes == new_processes:
                changes = ()
                break

            changes = diff_items(old_processes, new_processes)

            # There are changes. If we're not retrying, start showing
            # the results.
            if not args.retry or not sleeptime:
                break

            # Do the changes apply to us?
            if args.missing and not any(
                    direction < 0 for direction, item in changes):
                break
            if args.extra and not any(
                    direction > 0 for direction, item in changes):
                break

            # Sleep a while.
            time.sleep(sleeptime)

        # Show changes, if any.
        has_diff = show_diff(changes, missing=args.missing, extra=args.extra)
        status = (0, 1)[has_diff]

        sys.exit(status)  # possibly non-zero exit


if os.environ.get('RUNTESTS', '') not in ('', '0', 'n'):
    from unittest import TestCase, main as unittest_main

    class ProcessTestCase(TestCase):
        def test_parse_plain(self):
            process = Process.from_line('root  {user=root}')
            self.assertEqual(process.indent, 0)
            self.assertEqual(process.cmdline, 'root')
            self.assertEqual(process.user, 'root')
            self.assertEqual(process.listen, ())

        def test_parse_indent(self):
            process = Process.from_line('      /usr/bin/x  {user=nobody}')
            self.assertEqual(process.indent, 3)
            self.assertEqual(process.cmdline, '/usr/bin/x')

        def test_parse_single_listen(self):
            process = Process.from_line(
                '  x  {user=root, listen=tcp:127.0.0.1:34189}')
            self.assertEqual(process.listen, ('tcp:127.0.0.1:34189',))

        def test_parse_multiple_listen(self):
            process = Process.from_line(
                '  INIT  {user=root, listen=(tcp:0.0.0.0:22,tcp:[::]:22)}')
            self.assertEqual(
                process.listen, ('tcp:0.0.0.0:22', 'tcp:[::]:22'))

        def test_parse_unknown_property(self):
            # We don't know 'cgroup', but we don't lose it either.
            line = 'x  {user=root, listen=(a,b), cgroup=/system.slice}'
            process = Process.from_line(line)
            self.assertEqual(process.props['cgroup'], '/system.slice')
            self.assertEqual(process.to_line(), line)

        def test_parse_cmdline_with_braces(self):
            # The cmdline may contain anything, including our separators.
            line = "    awk '{print $1}  {user=x}'  {user=nobody}"
            process = Process.from_line(line)
            self.assertEqual(process.cmdline, "awk '{print $1}  {user=x}'")
            self.assertEqual(process.user, 'nobody')
            self.assertEqual(process.to_line(), line)

        def test_parse_bad_line(self):
            self.assertRaises(ValueError, Process.from_line, 'x')
            self.assertRaises(ValueError, Process.from_line, 'x  {pid=1}')

        def test_roundtrip(self):
            for line in (
                    'root  {user=root}',
                    '  INIT  {user=root, listen=(tcp:0.0.0.0:22,tcp:[::]:22)}',
                    '    /usr/bin/containerd  '
                    '{user=root, listen=tcp:1.2.3.4:5}',
                    '    /usr/lib/systemd/systemd-networkd  '
                    '{user=systemd+, listen=(p_raw:[35020]:eno0,'
                    'udp:217.21.199.20%eno0:68)}',
                    '      x  {user=u, listen=(a,b), zz=1}'):
                process = Process.from_line(line)
                self.assertEqual(process.to_line(), line)
                self.assertEqual(Process.from_line(process.to_line()), process)

        def test_equality(self):
            a = Process.from_line('x  {user=root, listen=(a,b)}')
            b = Process.from_line('x  {user=root, listen=(a,b)}')
            c = Process.from_line('x  {user=root, listen=a}')
            self.assertEqual(a, b)
            self.assertEqual(hash(a), hash(b))
            self.assertNotEqual(a, c)
            self.assertNotEqual(hash(a), hash(c))

        def test_same_process(self):
            a = Process.from_line('x  {user=root, listen=(a,b)}')
            b = Process.from_line('x  {user=root, listen=a}')
            c = Process.from_line('x  {user=nobody, listen=(a,b)}')
            d = Process.from_line('  x  {user=root, listen=(a,b)}')
            e = Process.from_line('x  {user=root}')
            self.assertTrue(a.same_process(b))   # only listen differs
            self.assertFalse(a.same_process(c))  # user differs
            self.assertFalse(a.same_process(d))  # indent differs
            self.assertTrue(a.same_process(e))   # listen gone

        def test_same_process_unknown_property(self):
            a = Process.from_line('x  {user=root, listen=a, zz=1}')
            b = Process.from_line('x  {user=root, listen=b, zz=1}')
            c = Process.from_line('x  {user=root, listen=a, zz=2}')
            d = Process.from_line('x  {user=root, listen=a}')
            self.assertTrue(a.same_process(b))
            self.assertFalse(a.same_process(c))
            self.assertFalse(a.same_process(d))

        def test_without_listen(self):
            process = Process.from_line('  x  {user=root, listen=(a,b), z=1}')
            self.assertEqual(
                process.without_listen().to_line(), '  x  {user=root, z=1}')
            self.assertEqual(  # unchanged
                process.to_line(), '  x  {user=root, listen=(a,b), z=1}')

        def test_listen_setter(self):
            process = Process.from_line('x  {user=root}')
            process.listen = ['b', 'a']
            self.assertEqual(process.to_line(), 'x  {user=root, listen=(b,a)}')
            process.listen = ()
            self.assertEqual(process.to_line(), 'x  {user=root}')

        def test_to_diff_lines(self):
            self.assertEqual(
                Process.from_line('  x  {user=root}').to_diff_lines(),
                ('  ', 'x  {user=root}', (), ''))
            self.assertEqual(
                Process.from_line('  x  {user=root, listen=a}')
                .to_diff_lines(),
                ('  ', 'x  {user=root, listen=(', ('a',), ')}'))
            self.assertEqual(
                Process.from_line('x  {user=root, listen=(a,b), z=1}')
                .to_diff_lines(),
                ('', 'x  {user=root, listen=(', ('a', 'b'), '), z=1}'))

        def test_sorting(self):
            lines = [
                'b  {user=root}', 'a  {user=zabbix}', 'a  {user=root}',
                '  a  {user=root}']
            self.assertEqual(
                [i.to_line() for i in sorted(
                    Process.from_line(line) for line in lines)],
                ['a  {user=root}', 'a  {user=zabbix}', 'b  {user=root}',
                 '  a  {user=root}'])


    class DiffTestCase(TestCase):
        def parse(self, *lines):
            return parse_db(u''.join('{}\n'.format(i) for i in lines))

        def group(self, old, new):
            return list(group_process_changes(diff_items(old, new)))

        def test_parse_db(self):
            data = (
                'root  {user=root}\n'
                '  INIT  {user=root, listen=(tcp:0.0.0.0:22,tcp:[::]:22)}\n'
                '    x  {user=nobody}\n')
            processes = parse_db(data)
            self.assertEqual(len(processes), 3)
            self.assertEqual([i.indent for i in processes], [0, 1, 2])
            self.assertEqual(format_db(processes), data)

        def test_parse_db_bad_line(self):
            self.assertRaises(ValueError, parse_db, 'root  {user=root}\nx\n')

        def test_manual_add(self):
            processes = self.parse(
                'root  {user=root}',
                '  INIT  {user=root}',
                '    a  {user=root}',
                '      a-child  {user=root}',
                '    c  {user=root}',
                '  other  {user=root}',
                '    b  {user=root}')
            self.assertEqual(
                [i.to_line() for i in manual_add(processes, 'b', 'root')],
                ['root  {user=root}',
                 '  INIT  {user=root}',
                 '    a  {user=root}',
                 '      a-child  {user=root}',
                 '    b  {user=root}',   # between a (+child) and c
                 '    c  {user=root}',
                 '  other  {user=root}',
                 '    b  {user=root}'])  # not a sibling, not our business

        def test_manual_add_last(self):
            processes = self.parse(
                'root  {user=root}',
                '  INIT  {user=root}',
                '    a  {user=root}',
                '  other  {user=root}')
            self.assertEqual(
                [i.to_line() for i in manual_add(processes, 'z', 'nobody')],
                ['root  {user=root}',
                 '  INIT  {user=root}',
                 '    a  {user=root}',
                 '    z  {user=nobody}',   # last child of INIT
                 '  other  {user=root}'])

        def test_manual_add_without_children(self):
            processes = self.parse('root  {user=root}', '  INIT  {user=root}')
            self.assertEqual(
                [i.to_line() for i in manual_add(processes, 'z', 'root')],
                ['root  {user=root}',
                 '  INIT  {user=root}',
                 '    z  {user=root}'])

        def test_manual_add_twice(self):
            processes = self.parse(
                'root  {user=root}',
                '  INIT  {user=root}',
                '    a  {user=root}')
            self.assertRaises(ValueError, manual_add, processes, 'a', 'root')

        def test_manual_add_no_parent(self):
            self.assertRaises(
                ValueError, manual_add, self.parse('root  {user=root}'),
                'z', 'root')

        def test_manual_remove(self):
            processes = self.parse(
                'root  {user=root}',
                '  INIT  {user=root}',
                '    a  {user=root, listen=x}',  # listen is ignored
                '      a-child  {user=root}',    # goes along with a
                '    b  {user=root}',
                '  other  {user=root}',
                '    a  {user=root}')            # not a sibling
            self.assertEqual(
                [i.to_line() for i in manual_remove(processes, 'a', 'root')],
                ['root  {user=root}',
                 '  INIT  {user=root}',
                 '    b  {user=root}',
                 '  other  {user=root}',
                 '    a  {user=root}'])
            self.assertEqual(  # wrong user, no match
                [i.to_line() for i in manual_remove(processes, 'a', 'x')],
                [i.to_line() for i in processes])

        def test_db_has_network(self):
            self.assertTrue(db_has_network(self.parse(
                'x  {user=root}', 'y  {user=root, listen=a}')))
            self.assertFalse(db_has_network(self.parse(
                'x  {user=root}', 'y  {user=root}')))
            self.assertFalse(db_has_network([]))
            # Not fooled by a cmdline that mentions listen.
            self.assertFalse(db_has_network(self.parse(
                'x --opt=", listen=(a,b)}"  {user=root}')))

        def test_diff_items(self):
            self.assertEqual(
                diff_items(['aa', 'bb', 'cc'], ['bb', 'dd', 'ff']),
                [(-1, 'aa'), (0, 'bb'), (-1, 'cc'), (1, 'dd'), (1, 'ff')])

        def test_no_changes(self):
            old = self.parse('x  {user=root}')
            self.assertEqual(
                [i[0:2] for i in self.group(old, old)], [('single', 0)])

        def test_pair_listen_change(self):
            old = self.parse('x  {user=root, listen=(a,b)}')
            new = self.parse('x  {user=root, listen=(a,c)}')
            grouped = self.group(old, new)
            self.assertEqual(len(grouped), 1)
            self.assertEqual(grouped[0][0], 'pair')
            self.assertEqual(grouped[0][1], old[0])
            self.assertEqual(grouped[0][2], new[0])

        def test_pair_single_to_multiple_listen(self):
            # A process going from one to two listen addresses is the
            # same process; the db writes one address without parens,
            # which used to keep these two from being paired up.
            old = self.parse('x  {user=root, listen=a}')
            new = self.parse('x  {user=root, listen=(a,b)}')
            self.assertEqual(
                [i[0] for i in self.group(old, new)], ['pair'])

        def test_pair_none_to_listen(self):
            old = self.parse('x  {user=root}')
            new = self.parse('x  {user=root, listen=a}')
            self.assertEqual(
                [i[0] for i in self.group(old, new)], ['pair'])

        def test_no_pair_for_other_process(self):
            old = self.parse('x  {user=root, listen=a}')
            new = self.parse('x  {user=nobody, listen=a}')
            self.assertEqual(
                [i[0:2] for i in self.group(old, new)],
                [('single', -1), ('single', 1)])

        def test_no_pair_without_listen(self):
            # Nothing to diff by address; show them as separate lines.
            old = self.parse('x  {user=root}')
            new = self.parse('y  {user=root}')
            self.assertEqual(
                [i[0:2] for i in self.group(old, new)],
                [('single', -1), ('single', 1)])

        def test_pairs_the_right_processes(self):
            old = self.parse(
                'a  {user=root, listen=(1,2)}',
                'b  {user=root, listen=(3,4)}')
            new = self.parse(
                'a  {user=root, listen=(1,9)}',
                'b  {user=root, listen=(3,8)}')
            grouped = self.group(old, new)
            self.assertEqual([i[0] for i in grouped], ['pair', 'pair'])
            self.assertEqual(
                [(i[1].cmdline, i[2].cmdline) for i in grouped],
                [('a', 'a'), ('b', 'b')])


def formatwarning(message, category, filename, lineno, line=None):
    """
    Override default Warning layout, from:

        /PATH/TO/argv0:326: UserWarning:
            [Errno 2] No such file or directory: '/0.d/05.d'
          warnings.warn(str(e))

    To:

        argv0:330: UserWarning:
            [Errno 2] No such file or directory: '/0.d/05.d'
    """
    return '{basename}:{lineno}: {category}: {message}\n'.format(
        basename=os.path.basename(filename), lineno=lineno,
        category=category.__name__, message=message)
warnings.formatwarning = formatwarning  # noqa


if __name__ == '__main__':
    if os.environ.get('RUNTESTS', '') not in ('', '0', 'n'):
        unittest_main(argv=sys.argv[0:1])
        raise RuntimeError('(unreachable code)')

    if not sys.stderr.isatty():
        warnings.filterwarnings(
            'ignore', 'Not listing --net info without root powers')
        warnings.simplefilter('ignore', DeprecationWarning)

    main()

# vim: set ts=8 sw=4 sts=4 et ai: