Docker-Net-DHCP/net-dhcp/network.py

import itertools
import ipaddress
import logging
import atexit
import socket
import time
import threading
import subprocess

from os import path

import pyroute2
from pyroute2.netlink.rtnl import rtypes
import docker
from flask import request, jsonify

from . import interface
from . import NetDhcpError, udhcpc, app

LIBRARY = 'IPR'

OPTS_KEY = 'com.docker.network.generic'
OPT_BRIDGE = 'bridge'
OPT_IPV6 = 'ipv6'

logger = logging.getLogger('net-dhcp')

interface.InitializeLibrary(LIBRARY)

dockerClient = None

def get_docker_client():
    global dockerClient
    if dockerClient is None:
        dockerClient = docker.from_env()

@atexit.register
def close_docker_client():
    global dockerClient
    if dockerClient is not None:
        dockerClient.close()
        dockerClient = None

gateway_hints = {}
container_dhcp_clients = {}
@atexit.register
def cleanup_dhcp():
    for endpoint, dhcp in container_dhcp_clients.items():
        logger.warning('cleaning up orphaned container DHCP client (endpoint "%s")', endpoint)
        dhcp.stop()

def veth_pair(e):
    return f'dh-{e[:12]}', f'{e[:12]}-dh'

def iface_addrs(iface):
    return list(map(lambda a: ipaddress.ip_interface((a['address'], a['prefixlen'])), iface.ipaddr))

def iface_nets(iface):
    return list(map(lambda n: n.network, iface_addrs(iface)))

def get_bridges():
    get_docker_client()
    reserved_nets = set(map(ipaddress.ip_network, map(lambda c: c['Subnet'], \
        itertools.chain.from_iterable(map(lambda x: [] if x is None else x, \
            map(lambda i: i['Config'], filter(lambda i: i['Driver'] != 'net-dhcp', \
                map(lambda n: n.attrs['IPAM'], dockerClient.networks.list()))))))))

    return dict(map(lambda i: (i['ifname'], i), filter(lambda i: i['kind'] == 'bridge' and not \
        set(iface_nets(i)).intersection(reserved_nets), map(lambda i: interface.GetInterface(LIBRARY, i['ifname']), \
            interface.GetInterfaces(LIBRARY)))))

def net_bridge(n):
    get_docker_client()
    return interface.GetInterface(LIBRARY, dockerClient.networks.get(n).attrs['Options'][OPT_BRIDGE])

def ipv6_enabled(n):
    get_docker_client()
    options = dockerClient.networks.get(n).attrs['Options']
    return OPT_IPV6 in options and options[OPT_IPV6] == 'true'

def endpoint_container_iface(n, e):
    get_docker_client()
    for cid, info in dockerClient.networks.get(n).attrs['Containers'].items():
        if info['EndpointID'] == e:
            container = dockerClient.containers.get(cid)
            netns = f'/proc/{container.attrs["State"]["Pid"]}/ns/net'

            with pyroute2.NetNS(netns) as rtnl:
                for link in rtnl.get_links():
                    attrs = dict(link['attrs'])
                    if attrs['IFLA_ADDRESS'] == info['MacAddress']:
                        return {
                            'netns': netns,
                            'ifname': attrs['IFLA_IFNAME'],
                            'address': attrs['IFLA_ADDRESS']
                        }
            break
    return None

def await_endpoint_container_iface(n, e, timeout=5):
    start = time.time()
    iface = None
    while time.time() - start < timeout:
        try:
            iface = endpoint_container_iface(n, e)
        except docker.errors.NotFound:
            time.sleep(0.5)
        except KeyError:
            time.sleep(0.5)
    if not iface:
        raise NetDhcpError('Timed out waiting for container to become availabile')
    return iface

def endpoint_container_hostname(n, e):
    get_docker_client()
    for cid, info in dockerClient.networks.get(n).attrs['Containers'].items():
        if info['EndpointID'] == e:
            return dockerClient.containers.get(cid).attrs['Config']['Hostname']
    return None

def endpoint_container_network_gwpriority(n, e):
    get_docker_client()
    for cid, info in dockerClient.networks.get(n).attrs['Containers'].items():
        if info['EndpointID'] == e:
            networkName = dockerClient.networks.get(n).attrs['Name']
            if 'GwPriority' in dockerClient.containers.get(cid).attrs['NetworkSettings']['Networks'][networkName]:
                return dockerClient.containers.get(cid).attrs['NetworkSettings']['Networks'][networkName]['GwPriority']
            else:
                return 0
    return None

def remove_veth(host_ifname):
    logger.info('Removing veth %s', host_ifname)
    if LIBRARY == 'NDB':
        if_host = interface.GetInterface(LIBRARY, if_host)
        bridge = net_bridge(req['NetworkID'])
        interface.DelPort(LIBRARY, bridge.ifname, if_host.ifname)
        interface.RemoveInterface(LIBRARY, if_host.ifname)
        logger.info('Removed veth for endpoint %s on ()', endpoint)
    else:
        logger.info('Deleting veth via shell: %s', host_ifname)
        try:
            subprocess.run(
                ['ip', 'link', 'del', host_ifname],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.PIPE,
                check=True
            )
            logger.info('Deleted veth %s', host_ifname)

        except subprocess.CalledProcessError as e:
            if b'Cannot find device' in e.stderr:
                logger.info('Veth %s already gone', host_ifname)
            else:
                logger.warning(
                    'Failed to delete veth %s: %s',
                    host_ifname,
                    e.stderr.decode().strip()
                )

@app.route('/NetworkDriver.GetCapabilities', methods=['POST'])
def net_get_capabilities():
    return jsonify({
        'Scope': 'local',
        'ConnectivityScope': 'global'
    })

@app.route('/NetworkDriver.CreateNetwork', methods=['POST'])
def create_net():
    req = request.get_json(force=True)
    for data in req['IPv4Data']:
        if data['AddressSpace'] != 'null' or data['Pool'] != '0.0.0.0/0':
            return jsonify({'Err': 'Only the null IPAM driver is supported'}), 400

    options = req['Options'][OPTS_KEY]
    if OPT_BRIDGE not in options:
        return jsonify({'Err': 'No bridge provided'}), 400
    # We have to use a custom "enable IPv6" option because Docker's null IPAM driver doesn't support IPv6 and a plugin
    # IPAM driver isn't allowed to return an empty address
    if OPT_IPV6 in options and options[OPT_IPV6] not in ('', 'true', 'false'):
        return jsonify({'Err': 'Invalid boolean value for ipv6'}), 400

    desired = options[OPT_BRIDGE]
    bridges = get_bridges()
    if desired not in bridges:
        return jsonify({'Err': f'Bridge "{desired}" not found (or the specified bridge is already used by Docker)'}), 400

    logger.info('Creating network "%s" (using bridge "%s")', req['NetworkID'], desired)
    return jsonify({})

@app.route('/NetworkDriver.DeleteNetwork', methods=['POST'])
def delete_net():
    return jsonify({})

@app.route('/NetworkDriver.CreateEndpoint', methods=['POST'])
def create_endpoint():
    req = request.get_json(force=True)
    network_id = req['NetworkID']
    endpoint_id = req['EndpointID']
    req_iface = req['Interface']

    bridge = net_bridge(network_id)
    bridge_addrs = iface_addrs(bridge)

    if_host, if_container = veth_pair(endpoint_id)
    logger.info('creating veth pair %s <=> %s', if_host, if_container)
    if_host = interface.CreateInterface(LIBRARY, if_host, 'veth', if_container)
    if_host.Up()

    try:
        start = time.time()
        while isinstance(if_container, str) and time.time() - start < 10:
            try:
                if_container = interface.GetInterface(LIBRARY, if_container)
                if_container.Up()
            except KeyError:
                time.sleep(0.5)
        if isinstance(if_container, str):
            raise NetDhcpError(f'timed out waiting for {if_container} to appear in host')

        interface.AddPort(LIBRARY, bridge.ifname, if_host.ifname)

        res_iface = {
            'MacAddress': '',
            'Address': '',
            'AddressIPv6': ''
        }

        if 'MacAddress' in req_iface and req_iface['MacAddress']:
            if_container.SetAddress(req_iface['MacAddress'])
        else:
            res_iface['MacAddress'] = if_container['address']

        def try_addr(type_):
            addr = None
            k = 'AddressIPv6' if type_ == 'v6' else 'Address'
            if k in req_iface and req_iface[k]:
                # TODO: Should we allow static IP's somehow?
                # Just validate the address, Docker will add it to the interface for us
                #addr = ipaddress.ip_interface(req_iface[k])
                #for bridge_addr in bridge_addrs:
                #    if addr.ip == bridge_addr.ip:
                #        raise NetDhcpError(400, f'Address {addr} is already in use on bridge {bridge["ifname"]}')
                raise NetDhcpError('Only the null IPAM driver is supported')
            else:
                dhcp = udhcpc.DHCPClient(if_container, v6=type_ == 'v6', once=True)
                addr = dhcp.finish()
                if not addr:
                    return
                res_iface[k] = str(addr)

                if dhcp.gateway:
                    gateway_hints[endpoint_id] = dhcp.gateway
            logger.info('Adding IP%s address %s to %s', type_, addr, if_container['ifname'])

        try_addr('v4')
        if ipv6_enabled(network_id):
            try_addr('v6')

        res = jsonify({
            'Interface': res_iface
        })
    except Exception as e:
        logger.exception(e)

        if not isinstance(if_container, str):
            interface.DelPort(LIBRARY, bridge.ifname, if_host.ifname)
        interface.RemoveInterface(LIBRARY, if_host.ifname)

        if isinstance(e, NetDhcpError):
            res = jsonify({'Err': str(e)}), e.status
        else:
            res = jsonify({'Err': str(e)}), 500
    finally:
        return res

@app.route('/NetworkDriver.EndpointOperInfo', methods=['POST'])
def endpoint_info():
    req = request.get_json(force=True)

    bridge = net_bridge(req['NetworkID'])
    if_host, _if_container = veth_pair(req['EndpointID'])
    if_host = interface.GetInterface(LIBRARY, if_host)

    return jsonify({
        'bridge': bridge['ifname'],
        'if_host': {
            'name': if_host['ifname'],
            'mac': if_host['address']
        }
    })

@app.route('/NetworkDriver.DeleteEndpoint', methods=['POST'])
def delete_endpoint():
    req = request.get_json(force=True)

    if_host, _ = veth_pair(req['EndpointID'])
    remove_veth(if_host)

    return jsonify({})

@app.route('/NetworkDriver.Join', methods=['POST'])
def join():
    req = request.get_json(force=True)
    network = req['NetworkID']
    endpoint = req['EndpointID']

    bridge = net_bridge(req['NetworkID'])
    _if_host, if_container = veth_pair(req['EndpointID'])

    res = {
        'InterfaceName': {
            'SrcName': if_container,
            'DstPrefix': 'eth'
        },
        'StaticRoutes': []
    }

    if endpoint in gateway_hints:
        gateway = gateway_hints[endpoint]
        logger.info('Setting IPv4 gateway from DHCP (%s)', gateway)
        res['Gateway'] = str(gateway)
        del gateway_hints[endpoint]

    ipv6 = ipv6_enabled(network)
    for route in bridge.routes:
        if route['type'] != rtypes['RTN_UNICAST'] or \
            (route['family'] == socket.AF_INET6 and not ipv6):
            continue

        if route['dst'] in ('', '/0'):
            if route['family'] == socket.AF_INET and 'Gateway' not in res:
                logger.info('Adding IPv4 gateway %s', route['gateway'])
                res['Gateway'] = route['gateway']
            elif route['family'] == socket.AF_INET6 and 'GatewayIPv6' not in res:
                logger.info('Adding IPv6 gateway %s', route['gateway'])
                res['GatewayIPv6'] = route['gateway']
        elif route['gateway']:
            dst = f'{route["dst"]}/{route["dst_len"]}'
            logger.info('Adding route to %s via %s', dst, route['gateway'])
            res['StaticRoutes'].append({
                'Destination': dst,
                'RouteType': 0,
                'NextHop': route['gateway']
            })

    container_dhcp_clients[endpoint] = ContainerDHCPManager(network, endpoint)
    return jsonify(res)

@app.route('/NetworkDriver.Leave', methods=['POST'])
def leave():
    req = request.get_json(force=True)
    endpoint = req['EndpointID']

    if endpoint in container_dhcp_clients:
        container_dhcp_clients[endpoint].stop()
        del container_dhcp_clients[endpoint]

    return jsonify({})

# Trying to grab the container's attributes (to get the network namespace)
# will deadlock (since Docker is waiting on us), so we must defer starting
# the DHCP client
class ContainerDHCPManager:
    def __init__(self, network, endpoint):
        self.network = network
        self.endpoint = endpoint
        self.ipv6 = ipv6_enabled(network)

        self.dhcp = None
        self.dhcp6 = None
        self.gwPriority = None
        self._thread = threading.Thread(target=self.run)
        self._thread.start()

    def _on_event(self, dhcp, event_type, _event):
        if event_type == udhcpc.EventType.DECONFIG:
            logger.info('[dhcp container] DECONFIG Event %s', _event)
            if path.lexists('{dhcp.netns}'):
                logger.info('[dhcp container] Flushing IP addresses')
                subprocess.check_call(['nsenter', f'-n{dhcp.netns}', '--', '/sbin/ip', 'address', 'flush', 'dev',
                    str(dhcp.iface['ifname']), 'label', str(dhcp.iface['ifname'])],
                    timeout=1, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            else:
                logger.info('[dhcp container] Container gone, can\'t flush IP addresses')
        elif event_type == udhcpc.EventType.RENEW or event_type == udhcpc.EventType.BOUND:
            if event_type == udhcpc.EventType.RENEW:
                logger.info('[dhcp container] RENEW Event %s', _event)
            elif event_type == udhcpc.EventType.BOUND:
                logger.info('[dhcp container] BOUND Event %s', _event)

            result = subprocess.run(['nsenter', f'-n{dhcp.netns}', '--', '/sbin/ip', 'route', 'show', 'type', 'unicast', 'proto', 'static', 'table', 'all'], timeout=1, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
            if result.returncode == 0:
                routes = result.stdout
                routes = routes.splitlines()
                if routes:
                    logger.info('[dhcp container] Store existing static routes')
                    for route in routes:
                        logger.info('[dhcp container] Stored static route: %s', route)

            logger.info('[dhcp container] Flushing IP addresses')
            subprocess.check_call(['nsenter', f'-n{dhcp.netns}', '--', '/sbin/ip', 'address', 'flush', 'dev',
                str(dhcp.iface['ifname']), 'label', str(dhcp.iface['ifname'])],
                timeout=1, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

            logger.info('[dhcp container] Adding IP addresses %s', dhcp.ip)
            subprocess.check_call(['nsenter', f'-n{dhcp.netns}', '--', '/sbin/ip', 'address', 'add', str(dhcp.ip), 'dev',
                str(dhcp.iface['ifname'])],
                timeout=1, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

            if dhcp.gateway and self.gwPriority >= 0:
                logger.info('[dhcp container] Replacing gateway with %s', dhcp.gateway)
                subprocess.check_call(['nsenter', f'-n{dhcp.netns}', '--', '/sbin/ip', 'route', 'replace', 'default', 'via',
                    str(dhcp.gateway)],
                    timeout=1, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

            if routes:
                logger.info('[dhcp container] Restore static routes')
                for route in routes:
                    parts = route.split()
                    command = ['nsenter', f'-n{dhcp.netns}', '--', '/sbin/ip', 'route', 'add'] + parts + ['proto', 'static']
                    output = subprocess.run(command, timeout=1, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
                    logger.info('[dhcp container] Restored static route: %s', route)
        else:
            logger.info('[dhcp container] Unhandled Event %s: %s', event_type,_event)

    def run(self):
        try:
            iface = await_endpoint_container_iface(self.network, self.endpoint)
            hostname = endpoint_container_hostname(self.network, self.endpoint)
            self.gwPriority = endpoint_container_network_gwpriority(self.network, self.endpoint)

            if self.gwPriority >= 0:
                self.dhcp = udhcpc.DHCPClient(iface, event_listener=self._on_event, hostname=hostname)
            else:
                self.dhcp = udhcpc.DHCPClient(iface, event_listener=self._on_event)
            logger.info('Starting DHCPv4 client on %s in container namespace %s', iface['ifname'], \
                self.dhcp.netns)

            if self.ipv6:
                self.dhcp6 = udhcpc.DHCPClient(iface, v6=True, event_listener=self._on_event, hostname=hostname)
                logger.info('Starting DHCPv6 client on %s in container namespace %s', iface['ifname'], \
                    self.dhcp6.netns)
        except Exception as e:
            logger.exception(e)
            if self.dhcp:
                self.dhcp.finish(timeout=1)

    def stop(self):
        if not self.dhcp:
            return

        try:
            logger.info('Shutting down DHCPv4 client on %s in container namespace %s', \
                self.dhcp.iface['ifname'], self.dhcp.netns)
            self.dhcp.finish(timeout=1)
        finally:
            try:
                if self.ipv6:
                    logger.info('Shutting down DHCPv6 client on %s in container namespace %s', \
                        self.dhcp6.iface['ifname'], self.dhcp.netns)
                    self.dhcp6.finish(timeout=1)
            finally:
                self._thread.join()

                # we have to do this since the docker client leaks sockets...
                close_docker_client()
                get_docker_client()