In my last blog post I described external-DNS, which is a way to have Kubernetes create and update DNS entries for its services. But as I mentioned, it got me thinking a bit on ways to extend this concept to handle other external aspects of my Kubernetes environment.
My ISP is in total control over my external IP addresses. I don’t pay for permanent IP addresses, and while they haven’t so far changed neither my IPv4 address or my IPv6 network, it can happen. Probably by mistake, since I have no kept my current ones for three months.
But accidents happen, and usually at the worst possible time, so can I be prepared for it?
Of course I can. However, it did require a bit of programming!
Firewall
My firewall at home is a Unifi Cloud Gateway Max, which I’m pretty happy with so far.
My firewall policy for IPv4 is pretty simple, it’s port forwarding on a single external IP to private addresses externally.
For IPv6, it’s quite a bit more complex. While the underlying operating system of the gateway supports specifying only the host part of a destination address, Unifi doesn’t expose that functionality. So my firewall rules for IPv6 has hardcoded addresses in it, that may change unexpectedly.
Luckily, there is an API. So I’ll be good there!
Kubernetes Resources
There’s tons of places in Kubernetes you can specify IP addresses. Since I only have one official IPv4 address, my Kubernetes services public IPv4 addresses are all private, on my networks that is defined in my Unify gateway. I have opted to run all my services exposed on a network kubernetesdmz, which I pull in on a VLAN interface on the server. All publically facing ingress IP addresses reside on that network. My IPv6 addresses on the same network are all public addresses, and will have to change should my ISP change my IPv6 network one day.
Luckily, there’s tons of APIs in Kubernetes. This should all be possible to automate.
Discovering changes
There’s tons of services on the network you can call to have it tell your publically facing IP address. However, since I am going to have to use the Unifi API to change the firewall rules, I might as well pull that information from there too.
For another project, I had implemented MQTT notifications that I used in home assistant to give me notifications whenever something happens, so I have opted to do so here too, should my IPs ever change.
Starting out
There were a few attempts at using the Unifi API out there. What I ended up using (and adapting to my own need) was a pretty simple class that handles logon and allows you to do API requests to an authenticated session. There were tons of more or less finished python APIs out there. I had already experimented with a tool called Unifi2Netbox, which has a simple Unifi class that I got working for that project, so I decided to just reuse that. I have made my own modifications on it to handle logins with local users and without 2 factor authentication for now, as I struggled quite a bit with the 2FA setup in that class. I’m not going to go into detail on this for now, but I’ll attach it to the end of this article if you’re interested.
The program
I run the code as a Kubernetes deployment. This gives me easy access to the Kubernetes API itself, for when I need to modify the Kubernetes resources. My deployment looks like this:
apiVersion: apps/v1
kind: Deployment
metadata:
name: ipchanger
namespace: unifi
spec:
replicas: 1
selector:
matchLabels:
app: ipchanger
strategy:
type: Recreate
template:
metadata:
labels:
app: ipchanger
spec:
containers:
- name: ipchanger
image: registry.engen.priv.no/ipchanger:latest
imagePullPolicy: Always
env:
- name: MQTT_BROKER
value: "mosquitto.mosquitto.svc.cluster.local"
- name: MQTT_PORT
value: "1883"
- name: MQTT_TOPIC
value: "k8s/unifi-ipchanger"
- name: NAMESPACE
value: "unifi"
- name: UNIFI_URL
value: "https://gw.engen.priv.no"
- name: UNIFI_GW_MAC
value: "0c:ea:14:3a:12:13"
- name: UNIFI_INGRESS_NETWORK
value: "kubernetesdmz"
- name: UNIFI_IPV6_PREFIX_LEN
value: "56"
- name: ANNOTATION_KEY
value: "ipchanger.alpha.kubernetes.io/patch"
- name: MQTT_USERNAME
valueFrom:
secretKeyRef:
name: mqtt-secret
key: MQTT_USERNAME
- name: MQTT_PASSWORD
valueFrom:
secretKeyRef:
name: mqtt-secret
key: MQTT_PASSWORD
- name: UNIFI_USERNAME
valueFrom:
secretKeyRef:
name: unifi-credentials
key: UNIFI_USERNAME
- name: UNIFI_PASSWORD
valueFrom:
secretKeyRef:
name: unifi-credentials
key: UNIFI_PASSWORD
resources:
limits:
cpu: "250m"
memory: "128Mi"
volumeMounts:
- mountPath: /storage
name: ipchange-state
volumes:
- name: ipchange-state
persistentVolumeClaim:
claimName: ipchange-state
I have a ton of environment variables, and it could probably be better if I just put most of that in a config map (which I’ll probably do at some point). The state of my network, i.e. my current IPv4 address and my IPV6 network, I persist in a volume called ipchange-state. One of the triggers for an IP address/IPv6 prefix change is likely a major outage somewhere, and it’s entirely possible that this will also take down my home infrastructure, so it’s good to persist this so that it knows what the world looked like when it was last running.
The script itself, I have packaged in a docker container that I host in my own docker registry, for Kubernetes to pull it whenever the POD starts.
But, on to the script. I’ll go through it in bits, to explain the functionality along with it.
Here’s the b
import json
import os
import re
import sys
import time
import requests
import warnings
import logging
import yaml
import pprint
import paho.mqtt.client as mqtt
import ipaddress
from http.client import HTTPConnection
from kubernetes import client, config, watch
import urllib3
from kubernetes.client.rest import ApiException
from unifi import Unifi
# Suppress only the InsecureRequestWarning
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
logger = logging.getLogger(__name__)
def setup_logging(level=logging.INFO):
"""Sets up basic logging configuration with a configurable log level."""
logging.basicConfig(
format="%(asctime)s - %(levelname)s - %(message)s",
level=level
)
This just imports my requirements and sets up logging. The Unifi API is attached to the end of this article, the rest is just standard modules that is either builtin or installed with pi. My requirements.txt looks like this:
requests~=2.32.3
PyYAML~=6.0.2
kubernetes
paho-mqtt
I’m a noob in python, and this was basically hacked together as an experiment, so you’ll have to forgive my lack of structure in my main script. First, we set up the environment – the logging in and pulling in the environment variables:
if __name__ == "__main__":
setup_logging(logging.INFO)
config.load_incluster_config()
api = client.CustomObjectsApi()
core_api = client.CoreV1Api()
apps_api = client.AppsV1Api()
try:
unifi_url = os.getenv("UNIFI_URL")
unifi_username = os.getenv('UNIFI_USERNAME')
unifi_password = os.getenv('UNIFI_PASSWORD')
mqtt_broker = os.getenv('MQTT_BROKER')
mqtt_username = os.getenv('MQTT_USERNAME')
mqtt_password = os.getenv('MQTT_PASSWORD')
mqtt_port = os.getenv('MQTT_PORT')
mqtt_topic = os.getenv('MQTT_TOPIC')
unifi_gw_mac = os.getenv('UNIFI_GW_MAC')
unifi_ingress_network = os.getenv('UNIFI_INGRESS_NETWORK')
unifi_ipv6_prefix_len = int(os.getenv('UNIFI_IPV6_PREFIX_LEN'))
annotation_key=os.getenv('ANNOTATION_KEY')
except KeyError:
logger.exception("Required environment parameter is missing")
raise SystemExit(1)
STATE_FILE="/storage/network_state.json"
Then we define some utilities.
Sending an mqtt notification
def send_mqtt_message(broker,username,password,topic,message):
# Set authentication if provided
mqtt_client = mqtt.Client()
if username and password:
mqtt_client.username_pw_set(username, password)
mqtt_client.connect(broker, 1883, 60)
"""Publish a message to the MQTT topic."""
mqtt_client.publish(topic, message)
logger.debug(f"Sent MQTT message: {message}")
Storing and retrieving the state
def load_last_state():
try:
with open(STATE_FILE, "r") as f:
return json.load(f)
except FileNotFoundError:
return None
# Helper function to store the last seen state to file
def store_last_state(state):
with open(STATE_FILE, "w") as f:
json.dump(state, f)
Rewriting an ip address or network
This function takes an address or a network and the old and new network state. It will return the address, possibly rewritten, and a boolean variable that tells whether or not it is modified.
It utilizes the functionality in the builtin python module ipaddress, which gives me a toolbox to calculate and convert ip addresses to networks etc.
It’s the IPv6 functionality that is complex. Basically, I treat one local network special, the Kubernetes ingress network, so I’ll check the complete /64 on this one. Unifi claims they’ll never change the 8 bits parts after the ISP /56 on a configured, but I’ll make sure to handle it they do. I might also do mistakes to cause it, so I’d better make sure. I call this network the ipv6_ingress_net, while the /56 I get from the ISP is the isp_ipv6_net. I have specified in an environment variable the netmask I get from the ISP, so it’s not hardcoded to be 56 bits – some ISPs might decide to hand out /48s, others might be cheap and go with a /60.
I am however working on the assumption that it’ll be divisible by 4, and not some substandard /49, /51, /62 or similar. It makes part of this a bit easier.
For IPv4, there is only one address, so I either replace it or I don’t.
def rewrite_address(address,old_network_state,network_state):
logger.debug("Checking %s", address)
old_ipv6_ingress_net = ipaddress.IPv6Network(old_network_state["ipv6"])
new_ipv6_ingress_net = ipaddress.IPv6Network(network_state["ipv6"])
old_isp_ipv6_net = old_ipv6_ingress_net.supernet(new_prefix=unifi_ipv6_prefix_len)
new_isp_ipv6_net = new_ipv6_ingress_net.supernet(new_prefix=unifi_ipv6_prefix_len)
logger.debug("%s %s %s %s", format(old_ipv6_ingress_net), format(new_ipv6_ingress_net), format(old_isp_ipv6_net), format(new_isp_ipv6_net))
# Handle when ISP sends out normal sized ranges.
isp_replace_str = "::"
if unifi_ipv6_prefix_len % 16 == 4:
isp_replace_str = "000::"
if unifi_ipv6_prefix_len % 16 == 8:
isp_replace_str = "00::"
if unifi_ipv6_prefix_len % 16 == 12:
isp_replace_str = "0::"
network = ipaddress.ip_network(address, strict=False)
# We are not rewriting ipv4 addresses
changed = False
if network.version == 4:
if address == old_network_state["ipv4"]:
address = network_state["ipv4"]
changed = True
return address, changed
logger.debug("%s %d",pprint.pformat(network),network.prefixlen)
if network.prefixlen > 64:
network = network.supernet(new_prefix=64)
logger.debug("%s %s", address, network)
if network == old_ipv6_ingress_net:
old_network_text = format(old_ipv6_ingress_net.network_address).replace("::","")
new_network_text = format(new_ipv6_ingress_net.network_address).replace("::","")
logger.debug("Changing address %s with: %s to %s", address, old_network_text, new_network_text)
address = address.replace(old_network_text,new_network_text)
changed = True
else:
if network.prefixlen > unifi_ipv6_prefix_len:
network = network.supernet(new_prefix=unifi_ipv6_prefix_len)
if network == old_isp_ipv6_net:
old_network_text = format(old_isp_ipv6_net.network_address).replace(isp_replace_str,"")
new_network_text = format(new_isp_ipv6_net.network_address).replace(isp_replace_str,"")
logger.info("Changing address %s with: %s to %s", address, old_network_text, new_network_text)
address = address.replace(old_network_text,new_network_text)
changed = True
return address, changed
Rewriting lists of networks and addresses.
This is just a logical extension, since I have this kind of list several places, especially in firewall policies. This one will change the addresses inline on the list, and return true or false to tell if it’s changed.
The true or false both in this and in the previous is of course used to tell if I need to call an API to update the information.
def rewrite_array(array,old_network_state,network_state):
changed = False
for i in range(len(array)):
array[i],is_changed = rewrite_address(array[i],old_network_state,network_state)
if is_changed:
changed = True
return changed
Updating Unifi Firewall Objects and Rules.
Unifi has firewall objects that can contain ip addresses and network, but it can also be contained in the rules directly. Fortunately they are pretty neatly structured, so I handle it with this function, that takes a unifi context and the old and new network state as input. The unifi context is presumed to be already authenticated, but it will handle reauthentication if it isn’t, by using the username/password set in environment variables from Kubernetes secrets.
I am not handling ipv4 firewall rules, because my one and only external ipv4 address is not specified in any ruleset – there’s no need to.
Unifi also has network zones, but there is no need to bother with that, because that’s just groupings. The actual ip addresses we need to rewrite is still either in the source or the destination of the firewall rule.
Now, to the functionality. Thanks to the utilities, it’s actually not that complex:
def change_unifi(unifi,old_network_state,network_state):
if old_network_state["ipv6"] == network_state["ipv6"]:
logger.debug("Only ipv4 change. No firewall rule changes needed!")
return
else:
firewall_groups = unifi.make_request("/proxy/network/api/s/default/rest/firewallgroup",method="GET")
firewall_policies = unifi.make_request("/proxy/network/v2/api/site/default/firewall-policies",method="GET")
logger.debug("Firewall groups: %s", pprint.pformat(firewall_groups))
logger.debug("Firewall Policies: %s", pprint.pformat(firewall_policies))
for group in firewall_groups["data"]:
if group["group_type"] == 'ipv6-address-group':
changed = rewrite_array(group["group_members"], old_network_state,network_state)
if changed:
logger.debug("Changed array into: %s", pprint.pformat(group["group_members"]))
logger.debug(pprint.pformat(group))
result = unifi.make_request(f"/proxy/network/api/s/default/rest/firewallgroup/{group['_id']}",method="PUT",data=group)
logger.info("Updated firewall group %s", group["name"])
for policy in firewall_policies:
if policy['predefined']:
continue
changed = False
if "ips" in policy["source"]:
changed = rewrite_array(policy["source"]["ips"],old_network_state,network_state)
if "ips" in policy["destination"]:
changed = rewrite_array(policy["destination"]["ips"],old_network_state,network_state) or changed
if changed:
logger.debug(pprint.pformat(policy))
result = unifi.make_request(f"/proxy/network/v2/api/site/default/firewall-policies/{policy['_id']}",method="PUT",data=policy)
logger.info("Updated firewall policy %s", policy["name"])
Kubernetes
Kubernetes properties, I will handle with a similar change_kubernetes function, that uses a function to change services and another to change metallb. This is the only place I have my ISP applied IP addresses specified. I might have to handle more later – i.e. security policies, but for now I don’t have that need.
patching a service
I’ll only handle services with a special annotation, which is a pretty common way to do similar things in Kubernetes. I use two kinds of Services, load balancers that I need to change loadBalancerIP property in, and externalName for using with external-dns so that I can update DNS records when my IPv4 addresses change. My IPv6 load balancers all have real IP addresses and have the external-dns annotations directly on that service, but for IPv4 I have to use an ExternalName pointing to it as a helper for external-dns.
def patch_service(service,old_network_state,network_state):
name, namespace = service.metadata.name, service.metadata.namespace
annotations = service.metadata.annotations or {}
if annotations.get(annotation_key) == "true":
loadBalancerIP = service.spec.load_balancer_ip if service.spec.load_balancer_ip else None
if loadBalancerIP is not None:
loadBalancerIP, changed = rewrite_address(loadBalancerIP,old_network_state,network_state)
if changed:
patch_data = {"spec": {"loadBalancerIP": loadBalancerIP}}
logger.debug(patch_data)
core_api.patch_namespaced_service(name, namespace, patch_data)
logger.info(f"Service patched: {name} in {namespace}")
else:
logger.debug(f"Service not patched: {name} in {namespace}")
externalName = service.spec.external_name if service.spec.external_name else None
if externalName is not None:
try:
logger.debug("Trying to convert %s to an ip address object", externalName)
externalname_ip = ipaddress.ip_address(externalName)
except:
logger.info("Externalname is not an ip address")
return
externalName, changed = rewrite_address(externalName,old_network_state,network_state)
if changed:
patch_data = {"spec": {"externalName": externalName}}
logger.info(patch_data)
core_api.patch_namespaced_service(name, namespace, patch_data)
logger.info(f"Service patched: {name} in {namespace}")
else:
logger.debug(f"Service not patched: {name} in {namespace}")
else:
logger.debug(f"Skipped service: {name} (annotation missing or false)")
MetalLB
MetalLB, which I use to define load balancers, have some internal configuration specifying pools load balancers can go in, and some configuration to announce load balancers on layer two on the network (ARP and neighbour discovery). I am not using BGP, so I am not supporting that method yet. I will also restart the services, since it needs to reread the patched configuration.
def patch_metallb(old_network_state,network_state):
changed_anything = False
try:
configMap = core_api.read_namespaced_config_map(name="config", namespace="metallb-system")
metallb_config = yaml.safe_load(configMap.data["config"])
changed = False
for pool in metallb_config.get("address-pools", []):
if ":" in pool["addresses"][0]:
changed = rewrite_array(pool["addresses"], old_network_state,network_state) or changed
if changed:
changed_anything = True
updated_yaml = yaml.dump(metallb_config, default_flow_style=False)
patch_data = {"data": {"config": updated_yaml}}
logger.debug(patch_data)
core_api.patch_namespaced_config_map("config", "metallb-system", patch_data)
logger.info(f"ConfigMap patched: config in metallb-system")
else:
logger.info(f"ConfigMap not patched: config in metallb-system")
except ApiException as e:
if e.status == 404:
logger.info('configMap config does not exist in namespace metallb-system')
else:
logger.error('Error checking configMap')
# Update metallb address pools
try:
address_pools = api.list_namespaced_custom_object(
group="metallb.io",
version="v1beta1", # Make sure to use the correct version for your setup
namespace="metallb-system",
plural="ipaddresspools"
)
for pool in address_pools["items"]:
annotations = pool["metadata"]["annotations"] or {}
if annotations.get(annotation_key) == "true":
if ":" in pool["spec"]["addresses"][0]:
changed = rewrite_array(pool["spec"]["addresses"],old_network_state,network_state)
if changed:
patch = {"spec": {"addresses": pool["spec"]["addresses"]}}
api.patch_namespaced_custom_object(
group="metallb.io",
version="v1beta1", # Use the correct version
namespace="metallb-system",
plural="ipaddresspools",
name=pool["metadata"]["name"],
body=patch
)
changed_anything = True
except:
logger.info("No metallb address pools were updated.",exc_info=1)
if changed_anything:
# Restart metallb
logger.info("Restarting metalLB")
namespace = "metallb-system"
# Restart the MetalLB Controller (Deployment)
deployment_name = "metallb-controller"
try:
apps_api.patch_namespaced_deployment(
name=deployment_name,
namespace=namespace,
body={"spec": {"template": {"metadata": {"annotations": {"kubectl.kubernetes.io/restartedAt": "now"}}}}}
)
logger.info(f"Restarted deployment: {deployment_name}")
except Exception as e:
logger.error(f"Failed to restart deployment {deployment_name}: {e}")
# Restart the MetalLB Speaker (DaemonSet pods)
label_selector = "app.kubernetes.io/component=speaker"
try:
pod_list = core_api.list_namespaced_pod(namespace=namespace, label_selector=label_selector)
for pod in pod_list.items:
core_api.delete_namespaced_pod(name=pod.metadata.name, namespace=namespace)
logger.info(f"Deleted speaker pod: {pod.metadata.name}")
except Exception as e:
logger.error(f"Failed to restart speaker pods: {e}")
modifying kubernetes resources
The master script for updating Kubernetes is pretty simple, all the logic is in the two preceding functions
def change_kubernetes(old_network_state,network_state):
for service in core_api.list_service_for_all_namespaces().items:
patch_service(service,old_network_state,network_state)
patch_metallb(old_network_state,network_state)
Getting the network information from Unifi Gateway
This is the function that calls the API to get the network information from unifi. It will populate a deviceinfo structure that I can use to find the needed information.
def update_unifi_info(unifi):
deviceinfo = unifi.make_request("/proxy/network/api/s/default/stat/device",method="GET")
logger.debug(pprint.pformat(deviceinfo))
return deviceinfo
Piecing it all together
Once all the pieces are written, we can write the main functionality. It first gets the needed information. Then it runs in a loop to check for changed network information from Unifi, and will use the previously defined functions to update my infrastructure.
It will also nominally handle losing IPv6, because I have heard anecdotes of that happening. That is not well tested, though. I will not act on that when losing it, but rather if it changes when it comes back, should the ipv6 addressing be different.
network_state = load_last_state()
ipv6_gone_notification_sent = False
while True:
logger.debug("Network state: %s",pprint.pformat(network_state))
deviceinfo = update_unifi_info(unifi)
devices = deviceinfo["data"]
gateway_info = next((device for device in devices if device["mac"] == unifi_gw_mac), None)
ipv4 = gateway_info["ip"]
try:
ingress_network_info = next((network for network in gateway_info["network_table"] if network["name"] == unifi_ingress_network), None)
ipv6_subnets = ingress_network_info["ipv6_subnets"]
ingress_ipv6_subnet = ipaddress.IPv6Network(ipv6_subnets[0], strict=False)
isp_ipv6 = ingress_ipv6_subnet.supernet(new_prefix=unifi_ipv6_prefix_len)
except:
ingress_ipv6_subnet = None
changed = False
old_network_state = network_state.copy()
if network_state is None:
network_state = {
"ipv4": ipv4,
"ipv6": format(ingress_ipv6_subnet)
}
logger.info("Saving ipv4 address: %s", ipv4)
logger.info("Saving ipv6 network: %s", ingress_ipv6_subnet)
changed = True
if network_state["ipv4"] != ipv4:
logger.info("IPV4 has changed from %s to %s", network_state["ipv4"], ipv4)
send_mqtt_message(mqtt_broker,mqtt_username,mqtt_password,mqtt_topic,f"IPV4 address has changed to {ipv4}")
network_state["ipv4"] = ipv4
changed = True
if ingress_ipv6_subnet is not None and ipv6_gone_notification_sent:
logger.info("IPV6 is back!")
send_mqtt_message(mqtt_broker,mqtt_username,mqtt_password,mqtt_topic,f"IPV6 seems to be back. Any changes will be handled.")
if ingress_ipv6_subnet is None and network_state["ipv6"] is not None:
if not ipv6_gone_notification_sent:
ipv6_gone_notification_sent = True
logger.info("IPV6 has gone! I won't do anything with that")
send_mqtt_message(mqtt_broker,mqtt_username,mqtt_password,mqtt_topic,f"IPV6 seems to be gone. Will not act on it.")
if ingress_ipv6_subnet is not None and ipaddress.IPv6Network(network_state["ipv6"]) != ingress_ipv6_subnet:
logger.info("IPV6 ingress network has changed from %s to %s", network_state["ipv6"], ingress_ipv6_subnet)
logger.info("The ISP provided IPV6 range is now %s", isp_ipv6)
network_state["ipv6"] = format(ingress_ipv6_subnet)
changed = True
if changed:
change_unifi(unifi,old_network_state,network_state)
change_kubernetes(old_network_state,network_state)
store_last_state(network_state)
else:
logger.info("No ip address changes")
time.sleep(30);
Summary
I’ll probably package this up in somehing a bit more usable pretty soon. Stay tuned!