infra
/
node-external-ip-controller
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: add ip address validation

main
Malar Invention 2025-01-12 17:47:03 +05:30
parent 924c24b16b
commit 7efde36ca3
1 changed files with 18 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
node_external_ip_controller_async.py
View File

@ -2,6 +2,7 @@ import asyncio
from kubernetes_asyncio import client, config, watch from kubernetes_asyncio import client, config, watch
import os import os
import logging import logging
import ipaddress
# Configuration # Configuration
ANNOTATION_KEY = os.getenv("ANNOTATION_KEY", "kube-vip.io/loadbalancerIPs") ANNOTATION_KEY = os.getenv("ANNOTATION_KEY", "kube-vip.io/loadbalancerIPs")
@ -17,6 +18,14 @@ logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def is_valid_ip(ip):
try:
ipaddress.ip_address(ip)
return True
except ValueError:
return False
async def update_service_annotation(v1, service, external_ips): async def update_service_annotation(v1, service, external_ips):
try: try:
service_name = service.metadata.name service_name = service.metadata.name
@ -27,7 +36,15 @@ async def update_service_annotation(v1, service, external_ips):
current_annotation = service_obj.metadata.annotations.get(ANNOTATION_KEY) current_annotation = service_obj.metadata.annotations.get(ANNOTATION_KEY)
zlan_gateway_ip = service_obj.metadata.labels.get(SERVICE_LABEL_KEY) zlan_gateway_ip = service_obj.metadata.labels.get(SERVICE_LABEL_KEY)
logger.debug(f"Zlan Gateway IP: {zlan_gateway_ip}") logger.debug(f"Zlan Gateway IP: {zlan_gateway_ip}")
if is_valid_ip(zlan_gateway_ip):
target_annotation = ",".join(external_ips) + "," + zlan_gateway_ip target_annotation = ",".join(external_ips) + "," + zlan_gateway_ip
else:
target_annotation = ",".join(external_ips)
logger.debug(
f"Invalid Zlan Gateway IP: {zlan_gateway_ip}, excluding from target annotation"
)
logger.debug(f"Current annotation: {current_annotation}") logger.debug(f"Current annotation: {current_annotation}")
logger.debug(f"Target annotation: {target_annotation}") logger.debug(f"Target annotation: {target_annotation}")
if current_annotation != target_annotation: if current_annotation != target_annotation: