hub

#!/usr/bin/python3
"""General command-line interface to the Virgin Media Hub3

"""
import argparse
import json
import os

import netaddr
import snmp
import utils
import virginmedia

SUBCOMMANDS = []

def subcommand(func):
    """A function decorator for subcommands"""
    SUBCOMMANDS.append(func)
    return func

# pylint: disable=R0903
class Args:
    """A function decorator that adds arguments to a subcommand.

    It should be passed an array of (tuple, dict) entries - these will
    be used (eventually) call ArgumentParser.add_argument(args, kwargs)

    """
    def __init__(self, args):
        self._args = args

    def __call__(self, func):
        func.args = self._args
        return func

def nologin(func):
    """A function decorator that flags the subcommand as not needing login"""
    func.needs_login = False
    return func

def nohub(func):
    """A function decorator that flags the subcommand as not needing a hub to talk to"""
    func.needs_hub = False
    return func

def dump_properties(hub, props):
    """Print the listed properties on stdout, nicely formatted"""
    for prop in props:
        res = getattr(hub, prop)
        if isinstance(res, snmp.Table):
            print(prop, ":")
            print(utils.format_table(res))
        else:
            print(prop, ":", getattr(hub, prop))

@subcommand
def info(hub, _args):
    """Show General information about the hub."""
    dump_properties(hub,
                    [
                        "modelname",
                        "family",
                        "hardware_version",
                        "serial_number",
                        "bootcode_version",
                        "firmware_version",
                        "name",
                        "current_time_status",
                        "current_time",
                        "uptime",
                        "first_install_wizard_completed",
                        "wan_ip_prov_mode",
                        "wan_current_ipaddr_ipv4",
                        "wan_current_ipaddr_ipv6",
                        "dns_servers"
                    ])

@subcommand
@Args([
    ("outfile", dict(help="File name to write backup to - defaults to 'router.data'",
                     type=argparse.FileType(mode='wb'),
                     default='router.data',
                     nargs='?'))])
def backup(hub, args):
    """Make a backup of the router configuration.

    The resulting file is a binary hub-proprietary file, essentially
    only useful for restoring into the hub - the internal structure is
    not (yet) known.

    """
    args.outfile.write(hub.backup())
    args.outfile.close()

@subcommand
def wan_status(hub, _args):
    """Show Wide Area Network settings."""
    dump_properties(hub,
                    [
                        "wan_if_macaddr",
                        "wan_mtu_size",

                        "wan_current_ipaddr_ipv4",
                        "wan_current_netmask",
                        "wan_current_gw_ipv4",
                        "wan_dhcp_duration_ipv4",

                        "wan_current_ipaddr_ipv6",
                        "wan_dhcp_duration_ipv6",
                        "wan_current_gw_ipv6",

                        "wan_dhcp_server_ip",

                        "wan_conn_hostname",
                        "wan_conn_domainname",
                        "dns_servers"
                        ])

@subcommand
@Args([
    (("--long", "-l"), dict(help="Show the 'long' format of the table",
                            action="store_true"))
    ])
def wan_networks(hub, args):
    """Prints the current external IP addresses of the hub."""
    networks = utils.filter_table(hub.wan_networks,
                                  lambda k, v: v.addr_type in [snmp.IPVersion.IPv4,
                                                               snmp.IPVersion.IPv6])

    if args.long:
        print(utils.format_by_row(networks))
    else:
        networks = utils.unselect_columns(networks,
                                          ['gw_ip_type',
                                           'netmask'])
        print(utils.format_table(networks))

@subcommand
@Args([
    (("--long", "-l"), dict(help="Show the 'long' format of the table",
                            action="store_true"))
    ])
def lan_networks(hub, args):
    """Information about the local LAN networks.

    The router can actually handle more than one network.  A single
    network can span multiple interfaces.

    """
    networks = hub.lan_networks
    if args.long:
        print(utils.format_by_row(networks))
    else:
        networks = utils.unselect_columns(networks,
                                          ['subnet_mask_type',
                                           'gw_ip_type',
                                           'dhcp_start_ip_type',
                                           'dhcp_end_ip_type',
                                           'env_control'])
        print(utils.format_table(networks))

@subcommand
def wifi_status(hub, _args):
    """Show WIFI Status."""
    dump_properties(hub,
                    [
                        "wifi_24ghz_essid",
                        "wifi_24ghz_password",
                        "wifi_5ghz_essid",
                        "wifi_5ghz_password",
                        "lan_gateway",
                        "lan_subnetmask",
                        "lan_dhcp_enabled",
                        "lan_dhcpv4_range_start",
                        "lan_dhcpv4_range_end",
                        "lan_dhcpv4_leasetime",
                    ])

@subcommand
@Args([
    (("--long", "-l"), dict(help="Show the 'long' format of the table",
                            action="store_true"))
    ])
def wifi_clients(hub, args):
    """List WIFI clients."""
    client_list = hub.wifi_clients

    print("There are {0:d} distinct WIFI clients (across {1:d} IP addresses):"
          .format(len(list(utils.unique_everseen([x.macaddr
                                                  for x in client_list.values()]))),
                  len(list(utils.unique_everseen([x.ipaddr
                                                  for x in client_list.values()])))))
    if args.long:
        print(client_list.format_by_row())
    else:
        client_list = utils.select_columns(client_list,
                                           ["macaddr",
                                            "hostname",
                                            "ipaddr",
                                            "tx_packets",
                                            "tx_fail",
                                            "rx_unicast_pkts",
                                            "last_rx_rate",
                                            "rssi"])
        best_rssi = max([row['rssi'] for row in client_list.values()])
        worst_rssi = min([row['rssi'] for row in client_list.values()])
        scale = 10

        def rss_scale(rssi):
            val = int(scale * (rssi - worst_rssi) / (best_rssi - worst_rssi) + 0.5)
            return "*" * val

        for row in client_list.values():
            row['signal'] = rss_scale(row['rssi'])

        client_list = utils.sort_table(client_list,
                                       key=lambda row: (row['macaddr'],
                                                        row['ipaddr']))
        print(utils.format_table(client_list))

@subcommand
@Args([
    (("--long", "-l"), dict(help="Show the 'long' format of the table",
                            action="store_true"))
    ])
def portforward_list(hub, args):
    """List current port forwardings."""

    pflist = hub.portforwards
    if args.long:
        print(utils.format_by_row(pflist))
    else:
        pflist = utils.sort_table(pflist,
                                  key=lambda x: (x['ext_port_start']))
        def portsummary(start, end):
            if start == end:
                return start
            return "{0}-{1}".format(start, end)

        pflist = utils.select_columns(pflist,
                                      ["rowstatus",
                                       "proto",
                                       "ext_port_start",
                                       "ext_port_end",
                                       "local_addr",
                                       "local_port_start",
                                       "local_port_end"])
        pflist = utils.filter_table(pflist,
                                    lambda k, v: v['rowstatus'] == snmp.RowStatus.ACTIVE)
        for row in pflist.values():
            row['ext_ports'] = portsummary(row['ext_port_start'], row['ext_port_end'])
            row['local_ports'] = portsummary(row['local_port_start'], row['local_port_end'])

        pflist = utils.select_columns(pflist,
                                      ["proto",
                                       "ext_ports",
                                       "local_addr",
                                       "local_ports"])
        print(utils.format_table(pflist))

    if not hub.firewall_enabled:
        print("Warning: Port forwarding will not work: firewalling is disabled...")

@subcommand
@Args([
    ("protocol", {"help":"Protocol to forward",
                  "choices": [p.name for p in snmp.IPProtocol]}),
    ("external_port", {"type": int,
                       "help": "The external port number to map"}),
    ("internal_ip", {"help": "Internal IP address to map it to"}),
    ("internal_port", {"type": int,
                       "help": "The internal port number to map it to"})
])
def portforward_add(hub, args):
    """Add a port forwarding entry.

    This directs the hub to forward incoming traffic (arriving at the
    external interface) on that port to some port internally.

    """
    if not hub.firewall_enabled:
        print("Warning: Port forwarding will not work: firewalling is disabled...")

    pflist = hub.portforwards
    pflist.append(
        proto=snmp.IPProtocol[args.protocol],
        ext_port_start=args.external_port,
        local_port_start=args.internal_port,
        local_addr=args.internal_ip)
    hub.apply_settings()

@subcommand
@Args([
    ("protocol", {"help":"Protocol to no longer forward",
                  "choices": [p.name for p in snmp.IPProtocol]}),
    ("external_port", {"type": int,
                       "help": "The external port number"})
    ])
def portforward_del(hub, args):
    """Remove a port forwarding entry.

    The entry is identified by giving the protocol and the first port
    in the range.

    """
    args.protocol = snmp.IPProtocol[args.protocol]
    pflist = hub.portforwards

    for rowid, entry in pflist.items():
        if entry.proto == args.protocol and entry.ext_port_start == args.external_port:
            print("Removing port forward entry", entry)
            del pflist[rowid]
            hub.apply_settings()
            return

    print("Port forward for %s port %d was not found - nothing removed."
          % (utils.human(args.protocol), args.external_port))

@subcommand
def ether_ports(hub, _args):
    """List ethernet ports on the hub."""
    print(utils.format_table(hub.etherports))

@subcommand
@Args([
    (("--long", "-l"), dict(help="Show the 'long' format of the table",
                            action="store_true"))
    ])
def wifi_networks(hub, args):
    """List WIFI networks."""
    networks = hub.bsstable
    if args.long:
        print(utils.format_by_row(networks))
    else:
        print(utils.format_table(networks))

@subcommand
@Args([
    (("--long", "-l"), dict(help="Show the long version of the table",
                            action="store_true")),
    (("--all", "-a"), dict(help="Also show offline clients",
                           action="store_true"))])
def clients(hub, args):
    """List known clients.

    Lists the clients on the internal LANs - both wired and wireless.

    Beware that extracting this information from the hub can take a
    long time - 90 seconds is not unheard of.

    """
    client_list = utils.sort_table(hub.clients, key=lambda x: x.get('mac_address', netaddr.EUI(addr=0)))
    client_list = utils.unselect_columns(client_list, ['rowstatus',
                                                       'device_name',
                                                       'last_change_secs',
                                                       'connected_secs'])
    client_count = len(list(utils.unique_everseen([e.get('mac_address')
                                                   for e in client_list.values()
                                                   if e.get('mac_address')])))
    online_count = len(list(utils.unique_everseen([e.get('mac_address')
                                                   for e in client_list.values()
                                                   if e.get('online') and e.get('mac_address')])))
    if not args.all:
        client_list = utils.unselect_columns(
            utils.filter_table(client_list,
                               lambda k, v: v.get('online')),
            ['online'])

    print("There are {0:d} known clients - {1:d} clients online)"
          .format(client_count, online_count))
    if args.long:
        print(utils.format_by_row(client_list))
    else:
        print(utils.format_table(client_list))
    print("Note: This list includes both wired and wireless clients")

@subcommand
@Args([
    (("--quiet", "-q"), dict(action="store_true"))])
def reboot(hub, args):
    """Instructs the hub to reboot.

    The hub will start to reboot - it will take a few minutes before
    it is fully back.

    """
    hub.reboot()
    if not args.quiet:
        print("{0} is now rebooting. It will be down for a few minutes".format(hub))

@subcommand
@nohub
def property_list(_args):
    """Get a list of the known property names.

    Not all properties will be settable.

    """
    for prop in sorted(virginmedia.HUB_PROPERTIES):
        print(prop)

@subcommand
@Args([
    ("property", {"nargs": "+",
                  "help": "The property to retrieve",
                  "choices": virginmedia.HUB_PROPERTIES})
])
def property_get(hub, args):
    """Get one or more properties."""
    for prop in args.property:
        propvalue = getattr(hub, prop)
        if isinstance(propvalue, snmp.Table):
            print(utils.format_table(propvalue))
        else:
            print(utils.human(getattr(hub, prop)))

@subcommand
@Args([
    ("--skip-get", {"help": "Do not try to retrieve the property first",
                    "action": "store_true"}),
    ("property", {"help": "The name of the property to set",
                  "choices": virginmedia.HUB_PROPERTIES}),
    ("value", {"help": "Value to set the property to"})])
def property_set(hub, args):
    """Set a specific property.

    Note that not all properties are settable.

    """
    if not args.skip_get:
        oldvalue = getattr(hub, args.property)

        if oldvalue == args.value:
            print("Property", args.property, "is already set to", args.value)
            return

    try:
        setattr(hub, args.property, args.value)
    except AttributeError:
        raise SystemExit("Property {0} is not settable".format(args.property))

    if args.skip_get:
        print("Set", args.property, "to", args.value)
    else:
        print("Changed", args.property, "from", oldvalue, "to", args.value)
    hub.apply_settings()

@subcommand
@Args([
    ("oid", {"help": "OID of the SNMP property to retrieve. " \
             "This should be given as a dot-separated number string, " \
             "e.g. '1.3.6.1.4.1.4115.1.20.1.1.5.10.0'"})])
def snmp_get(hub, args):
    """Retrieve an SNMP property from the hub.

    This is mostly useful for developers. The output of this command
    may vary between versions.

    """
    args.oid = args.oid.strip('.')
    print("{o} = {v}".format(o=args.oid, v=hub.snmp_get(args.oid)))

@subcommand
@Args([
    ("oid", {"help": "OID of the SNMP property to retrieve. " \
             "This should be given as a dot-separated number string, " \
             "e.g. '1.3.6.1.4.1.4115.1.20.1.1.5.10.0'"}),
    ("value", {"help": "The value to set it to. Is will be the raw string as sent to the hub"}),
    ("type", {"help": "SNMP Data type (I think)",
              "choices": snmp.DataType.__members__.keys(),
              "nargs": "?"})
])
def snmp_set(hub, args):
    """Set an SNMP attribute.

    This is mostly useful for developers, as it requires knowledge of
    SNMP and the relevant MIBs.

    """
    args.oid = args.oid.strip('.')
    if args.type is not None:
        args.type = snmp.DataType[args.type]

    res = hub.snmp_set(args.oid, value=args.value, datatype=args.type)
    print("Result:", res)

@subcommand
@Args([
    ("--byrow", {"help": "Arrange output on a row-by-row basis",
                 "action": "store_true"}),
    ("oid", {"help": "OID of the SNMP property walk down from. " \
             "This should be given as a dot-separated number string, " \
             "e.g. '1.3.6.1.4.1.4115.1.20.1.1.5.10.0'"})])
def snmp_walk(hub, args):
    """Do an SNMP walk on the hub.

    This is mostly useful for developers. The output of this command
    may vary between versions.

    This will produce a json-representation of the result
    """
    args.oid = args.oid.strip('.')
    walk_result = hub.snmp_walk(args.oid)
    if args.byrow:
        walk_result = snmp.parse_table(args.oid, walk_result)

    print(json.dumps(walk_result, sort_keys=True, indent=2))

@subcommand
def ddns_status(hub, _args):
    """Show Dynamic DNS settings"""
    dump_properties(hub,
                    [
                        "ddns_enabled",
                        "ddns_type",
                        "ddns_username",
                        "ddns_password",
                        "ddns_domain_name",
                        "ddns_addr_type",
                        "ddns_address",
                        "ddns_current_status"
                        ])

@subcommand
def mso_log(hub, _args):
    """Show MSO log.

    The MSO log is a log of configuration changes that are not done by
    the user. Assumed to be the MSO remotely or a technician.

    """
    log = hub.mso_log
    for entry in log.values():
        print(entry.stamp, entry.text)

@subcommand
def fw_log(hub, _args):
    """Show the firewall log
    """
    log = hub.fw_log
    for entry in log.values():
        print(entry.stamp, entry.text)

def main():
    """Main function. Obviously!"""

    parser = argparse.ArgumentParser()
    parser.add_argument("--host", "-H",
                        help="IP Address/dns name of the hub. "
                        "Uses the HUB environment variable as a default value"
                        " - and 192.168.0.1 if that is not set",
                        default=os.environ.get("HUB", "192.168.0.1"))
    parser.add_argument("--username", "-u",
                        help="User name to login as. "
                        "Uses the HUB_USER environment variable as a default value"
                        " - and 'admin' if that is not set",
                        default=os.environ.get("HUB_USER", "admin"))
    parser.add_argument("--password", "-p",
                        help="Password to authenticate on the hub. "
                        "Uses the HUB_PASSWORD environment variable if not specified. "
                        "If no password is given, no login will be attempted, and "
                        "some commands may fail as a result",
                        default=os.environ.get("HUB_PASSWORD"))
    parser.add_argument("--timeout",
                        help="Timeout for HTTP requests to the router in seconds. "
                        "Defaults to 30 seconds.",
                        default=30,
                        type=int)
    subparsers = parser.add_subparsers(description="The subcommands specify what should be done:")

    for cmd in sorted(SUBCOMMANDS, key=lambda x: x.__name__):
        cmd_parser = subparsers.add_parser(cmd.__name__.lower().replace('_', '-'),
                                           help=cmd.__doc__.split('\n')[0],
                                           description=cmd.__doc__)

        if hasattr(cmd, 'args'):
            for arg in cmd.args:
                if isinstance(arg[0], str):
                    cmd_parser.add_argument(arg[0], **arg[1])
                else:
                    cmd_parser.add_argument(*arg[0], **arg[1])
        cmd_parser.set_defaults(func=cmd)

    args = parser.parse_args()
    if not hasattr(args, "func"):
        parser.print_usage()
        raise SystemExit()

    try:
        needs_login = args.func.needs_login
    except AttributeError:
        needs_login = True

    try:
        needs_hub = args.func.needs_hub
    except AttributeError:
        needs_hub = True

    if needs_hub:
        with virginmedia.Hub(args.host) as hub:
            if args.timeout:
                hub.http_timeout = args.timeout
            if needs_login and args.password:
                hub.login(username=args.username,
                          password=args.password)
            args.func(hub, args)
    else:
        args.func(args)

if __name__ == '__main__':
    main()