Remove api doc headers

[pyi2ncommon] / src / dial.py

# This Python file uses the following encoding: utf-8

# The software in this package is distributed under the GNU General
# Public License version 2 (with a special exception described below).
#
# A copy of GNU General Public License (GPL) is included in this distribution,
# in the file COPYING.GPL.
#
# As a special exception, if other files instantiate templates or use macros
# or inline functions from this file, or you compile this file and link it
# with other works to produce a work based on this file, this file
# does not by itself cause the resulting work to be covered
# by the GNU General Public License.
#
# However the source code for this file must still be made available
# in accordance with section (3) of the GNU General Public License.
#
# This exception does not invalidate any other reasons why a work based
# on this file might be covered by the GNU General Public License.
#
# Copyright (c) 2016-2018 Intra2net AG <info@intra2net.com>

"""
Dialing, hangup, general provider online/offline state control.

Copyright: 2017 Intra2net AG

This used to be part of the sysmisc utilitiy initially which caused an include
circle with the arnied wrapper.

Contents:
    * dialout
        Generic access to the system’s dialing mode. Allows for requesting manual
        or permanently online dial state.
    * arnied_dial_permanent
        Enter permantly online dialing state.
    * arnied_dial_do
        Enter dial on command state.
    * arnied_dial_hangup
        Terminate uplink unconditionally.

All API methods come with the optional argument ``block`` (bool) to request the
call to block until the system completes the state transition successfully. The
completion timeout is currently 10 seconds (see the definition of
``DIALTOOLS_TIMEOUT`` below).
"""


import re
import io
import time
import logging
log = logging.getLogger('pyi2ncommon.dial')

from . import cnfvar
from . import sysmisc

HAVE_IPADDRESS = True
try:
    import ipaddress
except ImportError:  # guest
    HAVE_IPADDRESS = False

__all__ = (
    "arnied_dial_hangup", "arnied_dial_doc", "arnied_dial_permanent", "dialout", "get_wan_address", "DIALOUT_MODE_PERMANENT", "DIALOUT_MODE_MANUAL", "DIALOUT_MODE_DEFAULT", "DIALOUT_MODE_BY_NAME"
)

DIALOUT_MODE_PERMANENT = 0
DIALOUT_MODE_MANUAL = 1
DIALOUT_MODE_DEFAULT = DIALOUT_MODE_PERMANENT
DIALOUT_MODE_BY_NAME = {"permanent": DIALOUT_MODE_PERMANENT, "manual": DIALOUT_MODE_MANUAL}
DIALOUT_MODE_CNF = {DIALOUT_MODE_PERMANENT: "ONLINE", DIALOUT_MODE_MANUAL: "MANUAL"}

# compiling this regex needs the provider id and is postponed due to
# the peculiar implementation of the connd online condition
NEEDLE_MEMO = "  \[%s\] :(.*connected online fsm<online>.*)"
NEEDLE_OFFLINE = re.compile("connection map:\nend of connection map")

DIALTOOLS_HANGUP_BIN = "/usr/intranator/bin/hangup"

#: binary for manual dialing (dial on command)
DIALTOOLS_DOC_BIN = "/usr/intranator/bin/doc"
DIALTOOLS_TIMEOUT = 10  # s

#: client binary for talking to connd
TELL_CONND_BIN = "/usr/intranator/bin/tell-connd"


def _connd_online(prid="P1"):
    succ, out, _ = sysmisc.run_cmd_with_pipe([TELL_CONND_BIN, "--status"])
    if succ is False:
        return False
    return re.search(NEEDLE_MEMO % prid, out) is not None


def _connd_offline():
    succ, out, _ = sysmisc.run_cmd_with_pipe([TELL_CONND_BIN, "--status"])
    return succ and (NEEDLE_OFFLINE.search(out) is not None)


def arnied_dial_hangup(block=False):
    """
    Take down any currently active provider. This leverages arnied to
    accomplish the disconnect which is apparently more appropriate than
    having connd do it directly.

    :returns:       Whether the ``hangup`` command succeeded.
    :rtype:         int (dial result as above)
    """
    log.debug("requested arnied_dial_hangup%s",
              " (blocking)" if block else "")
    if block is False:
        succ, _, _ = sysmisc.run_cmd_with_pipe([DIALTOOLS_HANGUP_BIN])
        return sysmisc.RUN_RESULT_OK if succ is True else sysmisc.RUN_RESULT_FAIL

    res, err = sysmisc.cmd_block_till([DIALTOOLS_HANGUP_BIN],
                                      DIALTOOLS_TIMEOUT, _connd_offline)
    log.debug("arnied_dial_hangup → (%d, %r)", res, err)
    return res


def arnied_dial_doc(prid="P1", block=False):
    """
    Bring provider up via arnied manual dial (dial on command).

    :param  prid:   Provider id, default *P1*. It is up to the caller to ensure
                    this is a valid provider id.
    :type   prid:   str
    :param block:   block execution until system is online
    :type  block:   bool
    :returns:       Whether the ``doc`` command succeeded.
    :rtype:         int (dial result as above)
    """
    log.debug("requested arnied_dial_doc%s", " (blocking)" if block else "")
    if block is False:
        succ, _, _ = sysmisc.run_cmd_with_pipe([DIALTOOLS_DOC_BIN, prid])
        return sysmisc.RUN_RESULT_OK if succ is True else sysmisc.RUN_RESULT_FAIL
    res, err = sysmisc.cmd_block_till([DIALTOOLS_DOC_BIN, prid],
                                      DIALTOOLS_TIMEOUT, _connd_online,
                                      prid=prid)
    log.debug("arnied_dial_doc → (%d, %r)", res, err)
    return res


def arnied_dial_permanent(prid="P1", block=False):
    """
    Set permanent online state. Since the arnied dial helpers cannot initiate a
    permanent online state, achieve this via arnied.

    :param  prid:   Provider id, default *P1*. It is up to the caller to ensure
                    this is a valid provider id.
    :type   prid:   str
    :param block:   block execution until system is online
    :type  block:   bool
    :returns:       Whether the ``tell-connd`` command succeeded.
    :rtype:         int (dial result as above)

    ..todo:: This function uses old and deprecated methods of cnfvar usage, it
        should therefore be converted and needs unit tests added to verify the
        conversion and perhaps even to test the overall dial functionality.
    """
    log.debug("requested connd_dial_online%s" % " (blocking)" if block else "")

    cnf = cnfvar.CnfList([("dialout_mode", DIALOUT_MODE_CNF[DIALOUT_MODE_PERMANENT]),
                          ("dialout_defaultprovider_ref", "1")])

    def aux():
        store = cnfvar.BinaryCnfStore()
        store.commit(cnf)
        return True, "", None

    if block is False:
        succ, out, _ = aux()
        return sysmisc.RUN_RESULT_OK if succ is True else sysmisc.RUN_RESULT_FAIL

    res, err = sysmisc.cmd_block_till(aux, DIALTOOLS_TIMEOUT, _connd_online,
                                      prid=prid)
    log.debug("arnied_dial_permanent: result (%d, %r)", res, err)
    return res


def dialout(mode=DIALOUT_MODE_DEFAULT, prid="P1", block=True):
    """

    :param  mode:   How to dial (permanent vs. manual).
    :type   mode:   int (``DIALOUT_MODE``) | string
    :param  prid:   Provider id, default *P1*. It is up to the caller to ensure
                    this is a valid provider id.
    :type   prid:   str (constrained by available providers, obviously).
    :param block:   Whether to block until completion of the command.
    :type  block:   bool
    :returns:       Whether the command succeeded.
    :rtype:         int (dial result)
    :raises:        :py:class:`ValueError` if invalid dial mode was selected
    """
    log.info("go online with provider")

    dmode = None
    if isinstance(mode, int) is True:
        dmode = mode
    elif isinstance(mode, str) is True:
        try:
            dmode = DIALOUT_MODE_BY_NAME[mode]
        except Exception:
            log.error("invalid online mode name “%s” requested" % mode)
            pass

    if dmode is None:
        raise ValueError("exiting due to invalid online mode %r" % mode)

    log.debug("go online, mode=%d(%s), id=%r", dmode, mode, prid)

    if dmode == DIALOUT_MODE_PERMANENT:
        return arnied_dial_permanent(prid, block=block)

    if dmode == DIALOUT_MODE_MANUAL:
        return arnied_dial_doc(prid, block=block)

    raise ValueError("invalid dialout mode %r/%r requested" % (mode, dmode))


def get_wan_address(vm=None):
    """
    Retrieve the current WAN IP address of client ``vm`` or localhost.

    :param     vm:  Guest (client) to query; will as local connd if left unspecified.
    :type      vm:  virttest.qemu_vm.VM | None

    :returns:       The IPv4 address. For correctness, it will use the
                    ipaddress module if available. Otherwise it falls back
                    on untyped data.
    :rtype:         None | (ipaddress.IPv4Address | str)
    """
    log.info("query current lease")
    if vm is None:
        succ, connstat, _ = sysmisc.run_cmd_with_pipe([TELL_CONND_BIN, "--status"])
        if succ is False:
            return None
    else:
        connstat = vm.session.cmd_output("%s --status" % TELL_CONND_BIN)
    astr = io.StringIO(connstat)

    while True:
        l = astr.readline()
        if l == "":
            break
        if l.find("connected online fsm<online> IP:") != -1:
            addr = l[l.find("IP:")+3:l.find(" )\n")]  # beurk
            if HAVE_IPADDRESS is True:
                return ipaddress.IPv4Address(str(addr))
            else:
                return addr
    return None