llssh

#!/usr/bin/python
# llssh (part of ossobv/vcutil) // wdoekes/2024-2025 // Public Domain
#
# Helper script to ssh to (otherwise unconfigured) link-local IPv6 peers
# from Cumulus switches.
#
# Steps:
# - the peer must have lldpd running so we can get the mac-address
# - we ask our lldpd for mac address on the interface
# - we calculate the link-local IPv6 address
# - we construct the right 'ip vrf exec ... ssh link-local-address' invocation
#
# Usage:
#   llssh iface [ssh_options] [command [arguments]]
#   (where iface is the network interface/port)
#
# Example:
#   llssh swp12 -l myuser echo 'hi from $(hostname)'
#
# Notes:
# - This script is in python because of the non-trivial (for (ba)sh) MAC
#   to IPv6 LL address conversion.
# - The script assumes that the user has "lldpcli" access; it may need
#   to be in the adm group for that. We might also amend this with options
#   to pass the IPv6 address or MAC on the command line directly.
#
from __future__ import print_function
import sys
from json import dumps, loads
from os import environ, execve
from subprocess import CalledProcessError, check_output
from unittest import TestCase, main as unittest_main

try:
    FileNotFoundError
except NameError:
    FileNotFoundError = IOError  # python2


def lldpcli_show_neighbors():
    """
    Run 'lldpcli show neighbors -f json0'

    Previously we would do -f json, but json0 seems to provide the same
    output on several different systems.
    """
    try:
        output = check_output(['lldpcli', 'show', 'neighbors', '-f', 'json0'])
    except (CalledProcessError, FileNotFoundError):
        output = check_output([
            'sudo', 'lldpcli', 'show', 'neighbors', '-f', 'json0'])
    output = output.decode('utf-8', 'replace')
    return output


def mac_from_lldpcli(lldpcli_show_neighbors_js, iface):
    """
    Extract the mac address from the lldpcli show neighbors output

    lldpcli sh ne -f json0 | jq -r '
        .lldp[] | .interface[] | select(.name == "%s") | .port[0].id[] |
        select(.type == "mac") | .value'
    """
    data = loads(lldpcli_show_neighbors_js)
    for lldp in data['lldp']:
        for interface in lldp['interface']:
            if interface['name'] == iface:
                # NOTE: There's also a mac in the interface['chassis'], but it
                # shows the management IP-mac, not the mac connected to this
                # port.
                for port in interface['port']:
                    for id_ in port['id']:
                        if id_['type'] == 'mac':
                            return id_['value']
                else:
                    raise ValueError(
                        'lldp iface {iface} found but no "mac" type: {ports}'
                        .format(iface=iface, ports=dumps(interface['port'])))
    raise ValueError('lldp iface not found')


def link_local_v6_from_mac(mac):
    """
    Translate mac address to standard link local IP

    This way we only need lldpd on the other end: we can get the mac
    address and calculate what the link-local IP will be.
    """
    mac_as_ints = [int(x, 16) for x in mac.split(':')]
    ip6addr = 'fe80::%02x%02x:%02xff:fe%02x:%02x%02x' % tuple(
        [mac_as_ints[0] ^ 2] + mac_as_ints[1:])
    return ip6addr


def link_local_v6_addresses_from_arp(iface):
    """
    Fetch ip neighbor list and try to find a v6 address there
    """
    try:
        # ip -brief [-family inet6] neighbor show dev swp49.2102 [vrf MY_VF]
        output = check_output([
            'ip', '-brief', '-family', 'inet6', 'neighbor',
            'show', 'dev', iface])
    except (CalledProcessError, FileNotFoundError):
        raise ValueError('failed to run "ip neighbor show"')
    else:
        lines = output.strip().split('\n')
        lines = [line for line in lines if line]

    # Maybe there wasn't a neighbor for the main interface, but there is one
    # for a subinterface.
    if not lines:
        try:
            # ip -br [-family inet6] neighbor show
            output = check_output([
                'ip', '-brief', '-family', 'inet6', 'neighbor', 'show'])
        except (CalledProcessError, FileNotFoundError):
            raise ValueError('failed to run "ip neighbor show"')
        matches = [' dev {} '.format(iface), ' dev {}.'.format(iface)]
        lines = [line for line in output.strip().split('\n')]
        lines = [
            line for line in lines if any(match in line for match in matches)]

    if not lines:
        raise ValueError('ip neighbor not found')

    # fe80::bae9:24ff:fe0b:xxxx lladdr b8:e9:24:0b:xx:xx REACHABLE
    # fe80::bae9:24ff:fe0b:xxxx dev swp1 lladdr b8:e9:24:0b:xx:xx REACHABLE
    return [line.split(' ', 1)[0] for line in lines]


class MiscTests(TestCase):
    def test_mac_from_lldpcli(self):
        iface = 'swp41'
        mac = 'c0:ff:ee:be:ee:ff'
        input_ = '''{"lldp":[{"interface":[{"name":"swp41","via":"LLDP",
            "rid":"52","age":"158 days, 21:57:30","chassis":[{"id":[{"type":
            "mac","value":"c0:ff:ee:be:ee:ff"}],"name":[{"value":
            "natgw.dr.osso.nl"}],"descr":[{"value":"Ubuntu 20.04.3"}],
            "mgmt-ip":[{"value":"10.20.30.1"},{"value":
            "fe80::c0ff:eeba:beb0:0b1e"}],"capability":[{"type":"Bridge",
            "enabled":false},{"type":"Router","enabled":true},{"type":"Wlan",
            "enabled":false},{"type":"Station","enabled":false}]}],"port":
            [{"id":[{"type":"mac","value":"c0:ff:ee:be:ee:ff"}],"descr":
            [{"value":"enp1s0"}],"ttl":[{"value":"120"}]}]}]}]}'''
        self.assertEqual(mac, mac_from_lldpcli(input_, iface))

    def test_link_local_v6_from_mac(self):
        mac = 'c0:ff:ee:be:ee:ff'
        ip6addr = 'fe80::c2ff:eeff:febe:eeff'
        self.assertEqual(ip6addr, link_local_v6_from_mac(mac))


def main():
    iface = sys.argv[1]  # "swp12"
    assert all(ch in 'abcdefghijklmnopqrstuvwxyz0123456789' for ch in iface), (
        iface)

    try:
        # Fetch v6 IPs from the arp/neighbor table.
        addresses = link_local_v6_addresses_from_arp(iface)
    except ValueError:
        addresses = []

    try:
        # Get the mac address from LLDP, using lldpcli and fetching the MAC
        # from the selected port/interface.
        hwaddr = mac_from_lldpcli(lldpcli_show_neighbors(), iface)
        # Turn hardware address into LL v6 address.
        ip6addr = link_local_v6_from_mac(hwaddr)
        # Add it to the list.
        addresses.append(ip6addr)
    except ValueError:
        if not addresses:
            raise

    # Candidates?
    addresses = list(sorted(set(addresses)))
    if len(addresses) > 1:
        print(
            'Multiple candidates: {}'.format(dumps(addresses)),
            file=sys.stderr)

    # Add interface (this only works for v6 addresses).
    dstaddr = '{}%{}'.format(addresses[0], iface)  # "fe80::1%swp12"

    # On Cumulus switches, we must select the correct VRF or we get the
    # 'mgmt' VRF. Using 'default' works. This also works on other modern
    # systems. You can add '-l myusername' and other command line
    # options as appropriate.
    execve(
        '/sbin/ip',
        ['ip', 'vrf', 'exec', 'default',
         'ssh', '-oStrictHostKeychecking=no', '-oUserKnownHostsFile=/dev/null',
         dstaddr] + sys.argv[2:],
        environ)


if __name__ == '__main__':
    if environ.get('RUNTESTS', '') not in ('', '0'):
        unittest_main()
    else:
        main()