/mandos/release

#!/usr/bin/python3 -bbI

# -*- coding: utf-8; lexical-binding: t -*-

#

# Mandos Control - Control or query the Mandos server

#

# Copyright © 2008-2022 Teddy Hogeborn

# Copyright © 2008-2022 Björn Påhlsson

#

# This file is part of Mandos.

#

# Mandos is free software: you can redistribute it and/or modify it

# under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

#     Mandos is distributed in the hope that it will be useful, but

#     WITHOUT ANY WARRANTY; without even the implied warranty of

#     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

#     GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with Mandos.  If not, see <http://www.gnu.org/licenses/>.

#

# Contact the authors at <mandos@recompile.se>.

#

from __future__ import (division, absolute_import, print_function,

                        unicode_literals)

try:

    from future_builtins import *

except ImportError:

    pass

import sys

import unittest

import argparse

import logging

import os

import locale

import datetime

import re

import collections

import json

import io

import tempfile

import contextlib

if sys.version_info.major == 2:

    __metaclass__ = type

    str = unicode

    input = raw_input

class gi:

    """Dummy gi module, for the tests"""

    class repository:

        class GLib:

            class Error(Exception):

                pass

dbussy = None

ravel = None

dbus_python = None

pydbus = None

try:

    import dbussy

    import ravel

except ImportError:

    try:

        import pydbus

        import gi

    except ImportError:

        import dbus as dbus_python

# Show warnings by default

if not sys.warnoptions:

    import warnings

    warnings.simplefilter("default")

log = logging.getLogger(os.path.basename(sys.argv[0]))

logging.basicConfig(level="INFO", # Show info level messages

                    format="%(message)s") # Show basic log messages

logging.captureWarnings(True)   # Show warnings via the logging system

if sys.version_info.major == 2:

    import StringIO

    io.StringIO = StringIO.StringIO

locale.setlocale(locale.LC_ALL, "")

version = "1.8.14"

def main():

    parser = argparse.ArgumentParser()

    add_command_line_options(parser)

    options = parser.parse_args()

    check_option_syntax(parser, options)

    clientnames = options.client

    if options.debug:

        logging.getLogger("").setLevel(logging.DEBUG)

    if dbussy is not None and ravel is not None:

        bus = dbussy_adapter.CachingBus(dbussy, ravel)

    elif pydbus is not None:

        bus = pydbus_adapter.CachingBus(pydbus)

    else:

        bus = dbus_python_adapter.CachingBus(dbus_python)

    try:

        all_clients = bus.get_clients_and_properties()

    except dbus.ConnectFailed as e:

        log.critical("Could not connect to Mandos server: %s", e)

        sys.exit(1)

    except dbus.Error as e:

        log.critical(

            "Failed to access Mandos server through D-Bus:\n%s", e)

        sys.exit(1)

    # Compile dict of (clientpath: properties) to process

    if not clientnames:

        clients = all_clients

    else:

        clients = {}

        for name in clientnames:

            for objpath, properties in all_clients.items():

                if properties["Name"] == name:

                    clients[objpath] = properties

                    break

            else:

                log.critical("Client not found on server: %r", name)

                sys.exit(1)

    commands = commands_from_options(options)

    for command in commands:

        command.run(clients, bus)

def add_command_line_options(parser):

    parser.add_argument("--version", action="version",

                        version="%(prog)s {}".format(version),

                        help="show version number and exit")

    parser.add_argument("-a", "--all", action="store_true",

                        help="Select all clients")

    parser.add_argument("-v", "--verbose", action="store_true",

                        help="Print all fields")

    parser.add_argument("-j", "--dump-json", dest="commands",

                        action="append_const", default=[],

                        const=command.DumpJSON(),

                        help="Dump client data in JSON format")

    enable_disable = parser.add_mutually_exclusive_group()

    enable_disable.add_argument("-e", "--enable", dest="commands",

                                action="append_const", default=[],

                                const=command.Enable(),

                                help="Enable client")

    enable_disable.add_argument("-d", "--disable", dest="commands",

                                action="append_const", default=[],

                                const=command.Disable(),

                                help="disable client")

    parser.add_argument("-b", "--bump-timeout", dest="commands",

                        action="append_const", default=[],

                        const=command.BumpTimeout(),

                        help="Bump timeout for client")

    start_stop_checker = parser.add_mutually_exclusive_group()

    start_stop_checker.add_argument("--start-checker",

                                    dest="commands",

                                    action="append_const", default=[],

                                    const=command.StartChecker(),

                                    help="Start checker for client")

    start_stop_checker.add_argument("--stop-checker", dest="commands",

                                    action="append_const", default=[],

                                    const=command.StopChecker(),

                                    help="Stop checker for client")

    parser.add_argument("-V", "--is-enabled", dest="commands",

                        action="append_const", default=[],

                        const=command.IsEnabled(),

                        help="Check if client is enabled")

    parser.add_argument("-r", "--remove", dest="commands",

                        action="append_const", default=[],

                        const=command.Remove(),

                        help="Remove client")

    parser.add_argument("-c", "--checker", dest="commands",

                        action="append", default=[],

                        metavar="COMMAND", type=command.SetChecker,

                        help="Set checker command for client")

    parser.add_argument(

        "-t", "--timeout", dest="commands", action="append",

        default=[], metavar="TIME",

        type=command.SetTimeout.argparse(string_to_delta),

        help="Set timeout for client")

    parser.add_argument(

        "--extended-timeout", dest="commands", action="append",

        default=[], metavar="TIME",

        type=command.SetExtendedTimeout.argparse(string_to_delta),

        help="Set extended timeout for client")

    parser.add_argument(

        "-i", "--interval", dest="commands", action="append",

        default=[], metavar="TIME",

        type=command.SetInterval.argparse(string_to_delta),

        help="Set checker interval for client")

    approve_deny_default = parser.add_mutually_exclusive_group()

    approve_deny_default.add_argument(

        "--approve-by-default", dest="commands",

        action="append_const", default=[],

        const=command.ApproveByDefault(),

        help="Set client to be approved by default")

    approve_deny_default.add_argument(

        "--deny-by-default", dest="commands",

        action="append_const", default=[],

        const=command.DenyByDefault(),

        help="Set client to be denied by default")

    parser.add_argument(

        "--approval-delay", dest="commands", action="append",

        default=[], metavar="TIME",

        type=command.SetApprovalDelay.argparse(string_to_delta),

        help="Set delay before client approve/deny")

    parser.add_argument(

        "--approval-duration", dest="commands", action="append",

        default=[], metavar="TIME",

        type=command.SetApprovalDuration.argparse(string_to_delta),

        help="Set duration of one client approval")

    parser.add_argument("-H", "--host", dest="commands",

                        action="append", default=[], metavar="STRING",

                        type=command.SetHost,

                        help="Set host for client")

    parser.add_argument(

        "-s", "--secret", dest="commands", action="append",

        default=[], metavar="FILENAME",

        type=command.SetSecret.argparse(argparse.FileType(mode="rb")),

        help="Set password blob (file) for client")

    approve_deny = parser.add_mutually_exclusive_group()

    approve_deny.add_argument(

        "-A", "--approve", dest="commands", action="append_const",

        default=[], const=command.Approve(),

        help="Approve any current client request")

    approve_deny.add_argument("-D", "--deny", dest="commands",

                              action="append_const", default=[],

                              const=command.Deny(),

                              help="Deny any current client request")

    parser.add_argument("--debug", action="store_true",

                        help="Debug mode (show D-Bus commands)")

    parser.add_argument("--check", action="store_true",

                        help="Run self-test")

    parser.add_argument("client", nargs="*", help="Client name")

def string_to_delta(interval):

    """Parse a string and return a datetime.timedelta"""

    try:

        return rfc3339_duration_to_delta(interval)

    except ValueError as e:

        log.warning("%s - Parsing as pre-1.6.1 interval instead",

                    " ".join(e.args))

    return parse_pre_1_6_1_interval(interval)

def rfc3339_duration_to_delta(duration):

    """Parse an RFC 3339 "duration" and return a datetime.timedelta

    >>> rfc3339_duration_to_delta("P7D") == datetime.timedelta(7)

    True

    >>> rfc3339_duration_to_delta("PT60S") == datetime.timedelta(0, 60)

    True

    >>> rfc3339_duration_to_delta("PT60M") == datetime.timedelta(hours=1)

    True

    >>> # 60 months

    >>> rfc3339_duration_to_delta("P60M") == datetime.timedelta(1680)

    True

    >>> rfc3339_duration_to_delta("PT24H") == datetime.timedelta(1)

    True

    >>> rfc3339_duration_to_delta("P1W") == datetime.timedelta(7)

    True

    >>> rfc3339_duration_to_delta("PT5M30S") == datetime.timedelta(0, 330)

    True

    >>> rfc3339_duration_to_delta("P1DT3M20S") == datetime.timedelta(1, 200)

    True

    >>> # Can not be empty:

    >>> rfc3339_duration_to_delta("")

    Traceback (most recent call last):

    ...

    ValueError: Invalid RFC 3339 duration: ""

    >>> # Must start with "P":

    >>> rfc3339_duration_to_delta("1D")

    Traceback (most recent call last):

    ...

    ValueError: Invalid RFC 3339 duration: "1D"

    >>> # Must use correct order

    >>> rfc3339_duration_to_delta("PT1S2M")

    Traceback (most recent call last):

    ...

    ValueError: Invalid RFC 3339 duration: "PT1S2M"

    >>> # Time needs time marker

    >>> rfc3339_duration_to_delta("P1H2S")

    Traceback (most recent call last):

    ...

    ValueError: Invalid RFC 3339 duration: "P1H2S"

    >>> # Weeks can not be combined with anything else

    >>> rfc3339_duration_to_delta("P1D2W")

    Traceback (most recent call last):

    ...

    ValueError: Invalid RFC 3339 duration: "P1D2W"

    >>> rfc3339_duration_to_delta("P2W2H")

    Traceback (most recent call last):

    ...

    ValueError: Invalid RFC 3339 duration: "P2W2H"

    """

    # Parsing an RFC 3339 duration with regular expressions is not

    # possible - there would have to be multiple places for the same

    # values, like seconds.  The current code, while more esoteric, is

    # cleaner without depending on a parsing library.  If Python had a

    # built-in library for parsing we would use it, but we'd like to

    # avoid excessive use of external libraries.

    # New type for defining tokens, syntax, and semantics all-in-one

    Token = collections.namedtuple("Token", (

        "regexp",  # To match token; if "value" is not None, must have

                   # a "group" containing digits

        "value",   # datetime.timedelta or None

        "followers"))           # Tokens valid after this token

    # RFC 3339 "duration" tokens, syntax, and semantics; taken from

    # the "duration" ABNF definition in RFC 3339, Appendix A.

    token_end = Token(re.compile(r"$"), None, frozenset())

    token_second = Token(re.compile(r"(\d+)S"),

                         datetime.timedelta(seconds=1),

                         frozenset((token_end, )))

    token_minute = Token(re.compile(r"(\d+)M"),

                         datetime.timedelta(minutes=1),

                         frozenset((token_second, token_end)))

    token_hour = Token(re.compile(r"(\d+)H"),

                       datetime.timedelta(hours=1),

                       frozenset((token_minute, token_end)))

    token_time = Token(re.compile(r"T"),

                       None,

                       frozenset((token_hour, token_minute,

                                  token_second)))

    token_day = Token(re.compile(r"(\d+)D"),

                      datetime.timedelta(days=1),

                      frozenset((token_time, token_end)))

    token_month = Token(re.compile(r"(\d+)M"),

                        datetime.timedelta(weeks=4),

                        frozenset((token_day, token_end)))

    token_year = Token(re.compile(r"(\d+)Y"),

                       datetime.timedelta(weeks=52),

                       frozenset((token_month, token_end)))

    token_week = Token(re.compile(r"(\d+)W"),

                       datetime.timedelta(weeks=1),

                       frozenset((token_end, )))

    token_duration = Token(re.compile(r"P"), None,

                           frozenset((token_year, token_month,

                                      token_day, token_time,

                                      token_week)))

    # Define starting values:

    # Value so far

    value = datetime.timedelta()

    found_token = None

    # Following valid tokens

    followers = frozenset((token_duration, ))

    # String left to parse

    s = duration

    # Loop until end token is found

    while found_token is not token_end:

        # Search for any currently valid tokens

        for token in followers:

            match = token.regexp.match(s)

            if match is not None:

                # Token found

                if token.value is not None:

                    # Value found, parse digits

                    factor = int(match.group(1), 10)

                    # Add to value so far

                    value += factor * token.value

                # Strip token from string

                s = token.regexp.sub("", s, 1)

                # Go to found token

                found_token = token

                # Set valid next tokens

                followers = found_token.followers

                break

        else:

            # No currently valid tokens were found

            raise ValueError("Invalid RFC 3339 duration: \"{}\""

                             .format(duration))

    # End token found

    return value

def parse_pre_1_6_1_interval(interval):

    """Parse an interval string as documented by Mandos before 1.6.1,

    and return a datetime.timedelta

    >>> parse_pre_1_6_1_interval("7d") == datetime.timedelta(days=7)

    True

    >>> parse_pre_1_6_1_interval("60s") == datetime.timedelta(0, 60)

    True

    >>> parse_pre_1_6_1_interval("60m") == datetime.timedelta(hours=1)

    True

    >>> parse_pre_1_6_1_interval("24h") == datetime.timedelta(days=1)

    True

    >>> parse_pre_1_6_1_interval("1w") == datetime.timedelta(days=7)

    True

    >>> parse_pre_1_6_1_interval("5m 30s") == datetime.timedelta(0, 330)

    True

    >>> parse_pre_1_6_1_interval("") == datetime.timedelta(0)

    True

    >>> # Ignore unknown characters, allow any order and repetitions

    >>> parse_pre_1_6_1_interval("2dxy7zz11y3m5m") == datetime.timedelta(2, 480, 18000)

    True

    """

    value = datetime.timedelta(0)

    regexp = re.compile(r"(\d+)([dsmhw]?)")

    for num, suffix in regexp.findall(interval):

        if suffix == "d":

            value += datetime.timedelta(int(num))

        elif suffix == "s":

            value += datetime.timedelta(0, int(num))

        elif suffix == "m":

            value += datetime.timedelta(0, 0, 0, 0, int(num))

        elif suffix == "h":

            value += datetime.timedelta(0, 0, 0, 0, 0, int(num))

        elif suffix == "w":

            value += datetime.timedelta(0, 0, 0, 0, 0, 0, int(num))

        elif suffix == "":

            value += datetime.timedelta(0, 0, 0, int(num))

    return value

def check_option_syntax(parser, options):

    """Apply additional restrictions on options, not expressible in

argparse"""

    def has_commands(options, commands=None):

        if commands is None:

            commands = (command.Enable,

                        command.Disable,

                        command.BumpTimeout,

                        command.StartChecker,

                        command.StopChecker,

                        command.IsEnabled,

                        command.Remove,

                        command.SetChecker,

                        command.SetTimeout,

                        command.SetExtendedTimeout,

                        command.SetInterval,

                        command.ApproveByDefault,

                        command.DenyByDefault,

                        command.SetApprovalDelay,

                        command.SetApprovalDuration,

                        command.SetHost,

                        command.SetSecret,

                        command.Approve,

                        command.Deny)

        return any(isinstance(cmd, commands)

                   for cmd in options.commands)

    if has_commands(options) and not (options.client or options.all):

        parser.error("Options require clients names or --all.")

    if options.verbose and has_commands(options):

        parser.error("--verbose can only be used alone.")

    if (has_commands(options, (command.DumpJSON,))

        and (options.verbose or len(options.commands) > 1)):

        parser.error("--dump-json can only be used alone.")

    if options.all and not has_commands(options):

        parser.error("--all requires an action.")

    if (has_commands(options, (command.IsEnabled,))

        and len(options.client) > 1):

        parser.error("--is-enabled requires exactly one client")

    if (len(options.commands) > 1

        and has_commands(options, (command.Remove,))

        and not has_commands(options, (command.Deny,))):

        parser.error("--remove can only be combined with --deny")

class dbus:

    class SystemBus:

        object_manager_iface = "org.freedesktop.DBus.ObjectManager"

        def get_managed_objects(self, busname, objectpath):

            return self.call_method("GetManagedObjects", busname,

                                    objectpath,

                                    self.object_manager_iface)

        properties_iface = "org.freedesktop.DBus.Properties"

        def set_property(self, busname, objectpath, interface, key,

                         value):

            self.call_method("Set", busname, objectpath,

                             self.properties_iface, interface, key,

                             value)

        def call_method(self, methodname, busname, objectpath,

                        interface, *args):

            raise NotImplementedError()

    class MandosBus(SystemBus):

        busname_domain = "se.recompile"

        busname = busname_domain + ".Mandos"

        server_path = "/"

        server_interface = busname_domain + ".Mandos"

        client_interface = busname_domain + ".Mandos.Client"

        del busname_domain

        def get_clients_and_properties(self):

            managed_objects = self.get_managed_objects(

                self.busname, self.server_path)

            return {objpath: properties[self.client_interface]

                    for objpath, properties in managed_objects.items()

                    if self.client_interface in properties}

        def set_client_property(self, objectpath, key, value):

            return self.set_property(self.busname, objectpath,

                                     self.client_interface, key,

                                     value)

        def call_client_method(self, objectpath, method, *args):

            return self.call_method(method, self.busname, objectpath,

                                    self.client_interface, *args)

        def call_server_method(self, method, *args):

            return self.call_method(method, self.busname,

                                    self.server_path,

                                    self.server_interface, *args)

    class Error(Exception):

        pass

    class ConnectFailed(Error):

        pass

class dbus_python_adapter:

    class SystemBus(dbus.MandosBus):

        """Use dbus-python"""

        def __init__(self, module=dbus_python):

            self.dbus_python = module

            self.bus = self.dbus_python.SystemBus()

        @contextlib.contextmanager

        def convert_exception(self, exception_class=dbus.Error):

            try:

                yield

            except self.dbus_python.exceptions.DBusException as e:

                # This does what "raise from" would do

                exc = exception_class(*e.args)

                exc.__cause__ = e

                raise exc

        def call_method(self, methodname, busname, objectpath,

                        interface, *args):

            proxy_object = self.get_object(busname, objectpath)

            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,

                      interface, methodname,

                      ", ".join(repr(a) for a in args))

            method = getattr(proxy_object, methodname)

            with self.convert_exception():

                with dbus_python_adapter.SilenceLogger(

                        "dbus.proxies"):

                    value = method(*args, dbus_interface=interface)

            return self.type_filter(value)

        def get_object(self, busname, objectpath):

            log.debug("D-Bus: Connect to: (busname=%r, path=%r)",

                      busname, objectpath)

            with self.convert_exception(dbus.ConnectFailed):

                return self.bus.get_object(busname, objectpath)

        def type_filter(self, value):

            """Convert the most bothersome types to Python types"""

            if isinstance(value, self.dbus_python.Boolean):

                return bool(value)

            if isinstance(value, self.dbus_python.ObjectPath):

                return str(value)

            # Also recurse into dictionaries

            if isinstance(value, self.dbus_python.Dictionary):

                return {self.type_filter(key):

                        self.type_filter(subval)

                        for key, subval in value.items()}

            return value

        def set_client_property(self, objectpath, key, value):

            if key == "Secret":

                if not isinstance(value, bytes):

                    value = value.encode("utf-8")

                value = self.dbus_python.ByteArray(value)

            return self.set_property(self.busname, objectpath,

                                     self.client_interface, key,

                                     value)

    class SilenceLogger:

        "Simple context manager to silence a particular logger"

        def __init__(self, loggername):

            self.logger = logging.getLogger(loggername)

        def __enter__(self):

            self.logger.addFilter(self.nullfilter)

        class NullFilter(logging.Filter):

            def filter(self, record):

                return False

        nullfilter = NullFilter()

        def __exit__(self, exc_type, exc_val, exc_tb):

            self.logger.removeFilter(self.nullfilter)

    class CachingBus(SystemBus):

        """A caching layer for dbus_python_adapter.SystemBus"""

        def __init__(self, *args, **kwargs):

            self.object_cache = {}

            super(dbus_python_adapter.CachingBus,

                  self).__init__(*args, **kwargs)

        def get_object(self, busname, objectpath):

            try:

                return self.object_cache[(busname, objectpath)]

            except KeyError:

                new_object = super(

                    dbus_python_adapter.CachingBus,

                    self).get_object(busname, objectpath)

                self.object_cache[(busname, objectpath)]  = new_object

                return new_object

class pydbus_adapter:

    class SystemBus(dbus.MandosBus):

        def __init__(self, module=pydbus):

            self.pydbus = module

            self.bus = self.pydbus.SystemBus()

        @contextlib.contextmanager

        def convert_exception(self, exception_class=dbus.Error):

            try:

                yield

            except gi.repository.GLib.Error as e:

                # This does what "raise from" would do

                exc = exception_class(*e.args)

                exc.__cause__ = e

                raise exc

        def call_method(self, methodname, busname, objectpath,

                        interface, *args):

            proxy_object = self.get(busname, objectpath)

            log.debug("D-Bus: %s:%s:%s.%s(%s)", busname, objectpath,

                      interface, methodname,