cdn-ops/global/overlay/etc/puppet/modules/cdn/files/l4lb/sunet-l4lb-namespace

#!/usr/bin/env python3
# pylint:disable=invalid-name
# pylint:enable=invalid-name
# pylint:disable=invalid-name
# pylint:enable=invalid-name
"""
netplan does not have network namespace support so configure by hand

Tools used before committing code:
black sunet-l4lb-namespace
isort sunet-l4lb-namespace
pylint sunet-l4lb-namespace
mypy --strict sunet-l4lb-namespace
"""

import json
import shlex
import subprocess
import sys


def run_command(cmd: str) -> subprocess.CompletedProcess[str]:
    """Execute subprocess command"""
    args = shlex.split(cmd)
    try:
        proc = subprocess.run(args, capture_output=True, check=True, encoding="utf-8")
    except subprocess.CalledProcessError as exc:
        stderr_str = exc.stderr.rstrip()
        print(
            f"command failed: cmd='{cmd}', rc={exc.returncode}, stderr='{stderr_str}'",
            file=sys.stderr,
        )
        sys.exit(1)

    return proc


def configure_interfaces(
    namespace: str, if_data: dict[str, dict[str, list[str]]]
) -> None:
    """Configure interfaces"""
    proc = run_command("ip netns exec l4lb ip -j addr show")
    namespace_ifs = json.loads(proc.stdout)
    for if_name, data in if_data.items():
        if_exists = next(
            (True for interface in namespace_ifs if interface["ifname"] == if_name),
            False,
        )
        if not if_exists:
            if if_name.startswith("dummy"):
                run_command(
                    f"ip netns exec {namespace} ip link add {if_name} type dummy"
                )
            else:
                run_command(f"ip link set {if_name} netns {namespace}")

        proc = run_command(f"ip netns exec {namespace} ip -j addr show dev {if_name}")
        if_conf = json.loads(proc.stdout)
        for ipv4_cidr in data["ipv4"]:
            ip4, prefix = ipv4_cidr.split("/")
            v4_addr_exists = next(
                (
                    True
                    for addr in if_conf[0]["addr_info"]
                    if addr["local"] == ip4 and addr["prefixlen"] == int(prefix)
                ),
                False,
            )
            if not v4_addr_exists:
                run_command(
                    f"ip netns exec {namespace} ip addr add {ipv4_cidr} dev {if_name}"
                )
        for ipv6_cidr in data["ipv6"]:
            ip6, prefix = ipv6_cidr.split("/")
            v6_addr_exists = next(
                (
                    True
                    for addr in if_conf[0]["addr_info"]
                    if addr["local"] == ip6 and addr["prefixlen"] == int(prefix)
                ),
                False,
            )
            if not v6_addr_exists:
                run_command(
                    f"ip netns exec {namespace} ip addr add {ipv6_cidr} dev {if_name}"
                )

        run_command(f"ip netns exec {namespace} ip link set {if_name} up")


def setup_namespaces(netns_data: dict[str, dict[str, dict[str, list[str]]]]) -> None:
    """Setup network namespaces"""
    proc = run_command("ip -j netns list")
    existing_netns = json.loads(proc.stdout)
    for namespace, if_data in netns_data.items():
        netns_exists = next(
            (True for netns in existing_netns if netns["name"] == namespace), False
        )
        if not netns_exists:
            run_command(f"ip netns add {namespace}")

        # Make localhost available
        run_command(f"ip netns exec {namespace} ip link set lo up")

        configure_interfaces(namespace, if_data)


def main() -> None:
    """Starting point of the program"""

    # JSON file format:
    # {
    #  "namespace1": {
    #    "interface1": {
    #      "ipv4": [
    #        "192.168.10.1/31"
    #      ],
    #      "ipv6": [
    #        "2001:db8:1337:74::1/127"
    #      ]
    #    },
    #    "interface2": {
    #      "ipv4": [
    #        "192.168.10.3/31"
    #      ],
    #      "ipv6": [
    #        "2001:db8:1338:75::1/127"
    #      ]
    #    }
    #  }
    # }
    with open("/opt/sunet-cdn/l4lb/conf/netns.json", encoding="utf-8") as f:
        netns_data = json.load(f)

    setup_namespaces(netns_data)


if __name__ == "__main__":
    main()