135 lines
3.9 KiB
Python
Executable file
135 lines
3.9 KiB
Python
Executable file
#!/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:
|
|
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("/etc/sunet-cdn-l4lb/netns.json", encoding="utf-8") as f:
|
|
netns_data = json.load(f)
|
|
|
|
setup_namespaces(netns_data)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|