Update .gitea/workflows/jobs.yaml

Reviewed-on: #1

.gitea/workflows/jobs.yaml

@@ -26,6 +26,7 @@ jobs:
         with:
           images: |
             gitea.ollis.id.au/${{ github.repository }}
+          github-server-url: ${{ github.server_url }}
           tags: |
             type=ref,event=branch
             type=ref,event=pr

AGENTS.md

@@ -0,0 +1,23 @@
+# Agent Guidelines for kea-phpipam
+
+This file summarizes non-obvious conventions, setup requirements, and execution quirks needed for efficient development with this repository.
+
+## ⚙️ Setup & Dependencies
+- **Dependency Management:** Always assume the virtual environment needs to be synced using `uv`. The authoritative source for dependencies is `pyproject.toml` combined with `uv.lock`.
+- **Installation Command:** Use `uv sync --locked` to ensure the environment exactly matches the locked dependencies.
+
+## 🚀 Running the Application
+- **Primary Entrypoint:** The main service is launched via `kea_listener.py`.
+- **Execution Command:** The service must be run using the virtual environment's runner to guarantee correct path resolution:
+    ```bash
+    uv run kea_listener.py
+    ```
+- **Docker Context:** When building/running via Docker, the command structure is locked into `uv run /app/kea_listener.py`.
+
+## 📁 Code Structure Notes
+- **Core Logic:** Shared utilities and domain logic are located in `ipam_utils/utils.py`.
+- **Project Root:** `kea_listener.py` serves as the primary application wrapper/listener.
+
+## ⚠️ Gotchas
+- **Environment Consistency:** If running locally, failing to use `uv sync --locked` (or the appropriate virtual environment activation) is the most common source of dependency errors.
+- **Version Reliance:** The application relies heavily on the specific MQTT and environment reading capabilities provided by the pinned `paho-mqtt` and `environs` versions.

ipam_utils/utils.py

@@ -1,97 +1,158 @@
 from requests import Session
+from typing import Optional, Dict, Any
+import sys
+import logging
 
 
 class IPAMUtils:
-    def __init__(self, base_url: str, token: str) -> None:
+    """
+    A client for interacting with the phpIPAM REST API.
+    Handles abstracting away the underlying HTTP request details.
+    """
+
+    # API Tags for record states (Magic Number Replacement)
+    COMMIT_TAG = 2  # Tag used when updating existing, already known IPs
+    NEW_TAG = 4  # Tag used when creating brand new records
+    CLEANUP_TAG = 5  # Tag used when marking records as abandoned/cleaned
+
+    def __init__(self, base_url: str, token: str):
+        """
+        Initializes the IPAM client session.
+
+        Args:
+            base_url: The base URL for the phpIPAM API.
+            token: The authentication token for the API.
+        """
         self.base_url = base_url
         self.token = token
+        # Using a specific session instance helps manage cookies and connection details
         self.session = Session()
-        self.session.verify = False
-
+        self.session.verify = False  # Caution: Use HTTPS and verify=True in production
         self.session.headers.update({"Token": self.token})
 
-    def get_ip_info(self, ip_address: str) -> dict | None:
+    def get_ip_info(self, ip_address: str) -> Optional[Dict[str, Any]]:
         """
-        Get information about the given IP address from phpIPAM.
+        Retrieves detailed information about a specific IP address.
+
+        Args:
+            ip_address: The IP address (e.g., "192.168.1.10").
+
+        Returns:
+            The IP address data dictionary if found, otherwise None.
         """
         if not ip_address:
             return None
-        response = self.session.get(f"{self.base_url}/addresses/search/{ip_address}/")
-        if response.status_code == 200:
-            return response.json().get("data")[0]
-        else:
+        try:
+            response = self.session.get(
+                f"{self.base_url}/addresses/search/{ip_address}/"
+            )
+            response.raise_for_status()
+            data = response.json()
+            return data.get("data")[0] if data.get("data") else None
+        except Exception as e:
+            logging.warning(f"Could not retrieve IP info for {ip_address}: {e}")
             return None
 
-    def get_ip_id(self, ip_address: str) -> int | None:
+    def get_ip_id(self, ip_address: str) -> Optional[int]:
         """
-        Get the ID of the IP address in phpIPAM.
+        Finds the internal database ID for a given IP address.
         """
         ip_info = self.get_ip_info(ip_address)
-        if ip_info:
-            return ip_info.get("id")
-        else:
-            return None
+        return ip_info.get("id") if ip_info else None
 
-    def get_containing_subnet_id(self, ip_address: str) -> int | None:
+    def get_containing_subnet_id(self, ip_address: str) -> Optional[int]:
         """
-        Get the ID of the subnet containing the given IP address in phpIPAM.
+        Finds the ID of the subnet pool containing the given IP address.
         """
         if not ip_address:
             return None
-        response = self.session.get(
-            f"{self.base_url}/subnets/overlapping/{ip_address}/32/",
-            params={"filter_by": "isPool", "filter_value": 1},
-        )
-        if response.status_code == 200:
-            return response.json().get("data")[0].get("id")
-        else:
+        try:
+            # Endpoint structure depends heavily on the API: using /32/ to target host IP
+            response = self.session.get(
+                f"{self.base_url}/subnets/overlapping/{ip_address}/32/",
+                params={"filter_by": "isPool", "filter_value": 1},
+            )
+            response.raise_for_status()
+            data = response.json()
+            return data.get("data")[0].get("id") if data.get("data") else None
+        except Exception as e:
+            logging.warning(f"Could not find subnet ID for {ip_address}: {e}")
             return None
 
     def update_ip_information(
-        self, ip_address: str, hostname: str = None, mac_address: str = None
-    ) -> dict | None:
+        self,
+        ip_address: str,
+        hostname: Optional[str] = None,
+        mac_address: Optional[str] = None,
+    ) -> Optional[Dict[str, Any]]:
         """
-        Mark the given IP address as used in phpIPAM.
+        Marks an existing IP record as active (COMMIT) or creates it (NEW).
+        Returns the refreshed IP object on success.
         """
         if not ip_address:
             return None
+
         ip_id = self.get_ip_id(ip_address)
+
         if ip_id:
-            response = self.session.patch(
-                f"{self.base_url}/addresses/{ip_id}/",
-                json={"hostname": hostname, "mac": mac_address, "tag": 2},
-            )
-            if response.status_code == 200:
+            # Case 1: IP record already exists -> Update (PATCH)
+            payload = {"hostname": hostname, "mac": mac_address, "tag": self.COMMIT_TAG}
+            try:
+                response = self.session.patch(
+                    f"{self.base_url}/addresses/{ip_id}/", json=payload
+                )
+                response.raise_for_status()
                 return self.get_ip_info(ip_address)
-            return None
+            except Exception as e:
+                print(f"Error updating existing IP {ip_address}: {e}", file=sys.stderr)
+                return None
         else:
+            # Case 2: IP record does not exist -> Check Subnet and Create (POST)
             subnet_id = self.get_containing_subnet_id(ip_address)
             if subnet_id:
-                response = self.session.post(
-                    f"{self.base_url}/addresses/",
-                    json={
-                        "ip": ip_address,
-                        "hostname": hostname,
-                        "mac": mac_address,
-                        "subnetId": subnet_id,
-                        "tag": 4,
-                    },
-                )
-                if response.status_code == 201:
+                payload = {
+                    "ip": ip_address,
+                    "hostname": hostname,
+                    "mac": mac_address,
+                    "subnetId": subnet_id,
+                    "tag": self.NEW_TAG,
+                }
+                try:
+                    response = self.session.post(
+                        f"{self.base_url}/addresses/", json=payload
+                    )
+                    response.raise_for_status()
+                    # On successful creation, fetch the new record details
                     return self.get_ip_info(ip_address)
-            return None
+                except Exception as e:
+                    logging.info(f"Error creating new IP record {ip_address}: {e}")
+
+                    return None
+            else:
+                logging.warning(
+                    f"Could not determine subnet ID for {ip_address}. Skipping update."
+                )
+                return None
 
     def clear_ip_information(self, ip_address: str) -> bool:
         """
-        Clear the information of the given IP address in phpIPAM.
+        Soft deletes the association of an IP record by clearing details (CLEANUP).
+        Returns True if the update was successful (HTTP 200), False otherwise.
         """
         if not ip_address:
             return False
+
         ip_id = self.get_ip_id(ip_address)
+
         if ip_id:
-            response = self.session.patch(
-                f"{self.base_url}/addresses/{ip_id}/",
-                json={"hostname": "", "mac": "", "tag": 5},
-            )
-            return response.status_code == 200
+            payload = {"hostname": "", "mac": "", "tag": self.CLEANUP_TAG}
+            try:
+                response = self.session.patch(
+                    f"{self.base_url}/addresses/{ip_id}/", json=payload
+                )
+                response.raise_for_status()
+                return response.status_code == 200
+            except Exception as e:
+                logging.warning(f"Error clearing IP {ip_address}: {e}")
+                return False
         return False

kea_listener.py

@@ -1,20 +1,21 @@
 #!/usr/bin/env python3
-import sys
 import argparse
 import json
 import logging
 import random
 import socket
 import time
+from typing import TypedDict
 import uuid
 import urllib3
 import threading
 from environs import env
-
 from dataclasses import dataclass
 import paho.mqtt.client as mqtt_client
 import paho.mqtt.publish as publish
 from ipam_utils import IPAMUtils
+from logging_setup import setup_split_logging
+from collections.abc import Callable
 
 
 @dataclass
@@ -38,12 +39,19 @@ def on_connect(client, userdata, flags, rc, properties=None):
     client.subscribe(userdata.mqtt_topic)
 
 
-def extract_lease_ip_info(msg_dict) -> dict[str, str | None]:
+class LeaseInfo(TypedDict):
+    HOOK_TYPE: str
+    ip_address: str | None
+    hostname: str | None
+    mac_address: str | None
+
+
+def extract_lease_ip_info(msg_dict) -> LeaseInfo | None:
     ip_address = None
     hostname = None
     mac_address = None
 
-    #    "QUERY4_TYPE": "DHCPREQUEST",
+    # "QUERY4_TYPE": "DHCPREQUEST",
 
     if msg_dict["HOOK_TYPE"] in ["RENEW"] and msg_dict["QUERY4_TYPE"] == "DHCPREQUEST":
         ip_address = msg_dict.get("LEASE4_ADDRESS")
@@ -63,97 +71,74 @@ def extract_lease_ip_info(msg_dict) -> dict[str, str | None]:
     }
 
 
-def on_message(client, userdata, msg):
-    msg_dict = json.loads(msg.payload.decode())
-    msg_dict["id"] = str(uuid.uuid1(node=uuid.getnode()))
+def _process_commit(userdata: KeaListenerConfig, lease_ip_info: LeaseInfo) -> None:
+    """Handles IP lease renewal events (COMMIT)."""
     if logging.getLogger().isEnabledFor(logging.DEBUG):
         logging.debug(
-            "HOOK: [" + msg_dict["id"] + "] " + msg.topic + " " + msg.payload.decode()
+            f"Processing COMMIT for IP: {lease_ip_info['ip_address']} Hostname: {lease_ip_info['hostname']} MAC: {lease_ip_info['mac_address']}"
         )
+    try:
+        ip_info = userdata.ipam_util.update_ip_information(
+            ip_address=lease_ip_info["ip_address"],
+            hostname=lease_ip_info["hostname"],
+            mac_address=lease_ip_info["mac_address"],
+        )
+        logging.info("Updated IP Info: " + str(ip_info))
+        userdata.increment_count()
+    except Exception as e:
+        logging.error("Error updating IP information: " + str(e))
+
+
+def _process_lease_change(
+    userdata: KeaListenerConfig, lease_ip_info: LeaseInfo
+) -> None:
+    """Handles lease expiry/release/recovery events (CLEANUP)."""
+    try:
+        userdata.ipam_util.clear_ip_information(ip_address=lease_ip_info["ip_address"])
+        userdata.increment_count()
+        logging.info(
+            "Processed RELEASE/EXPIRE/RECOVER for IP: " + lease_ip_info["ip_address"]
+        )
+    except Exception as e:
+        logging.error("Error clearing IP information: " + str(e))
+
+
+def on_message(client, userdata, msg):
+    """
+    Processes an incoming MQTT message, extracts IP lease information,
+    and dispatches the processing to the appropriate handler based on HOOK_TYPE.
+    """
+    try:
+        msg_dict = json.loads(msg.payload.decode())
+    except json.JSONDecodeError as e:
+        logging.error(f"Failed to decode MQTT payload from {msg.topic}: {e}")
+        return
+
+    # Assign a new unique ID for traceability
+    msg_dict["id"] = str(uuid.uuid1(node=uuid.getnode()))
+    if logging.getLogger().isEnabledFor(logging.DEBUG):
+        logging.debug(f"HOOK: [{msg_dict['id']}] {msg.topic} {msg.payload.decode()}")
 
     lease_ip_info = extract_lease_ip_info(msg_dict)
-    if lease_ip_info and lease_ip_info["HOOK_TYPE"] in ["RENEW"]:
-        if logging.getLogger().isEnabledFor(logging.DEBUG):
-            logging.debug(
-                "Processing COMMIT for IP: "
-                + lease_ip_info["ip_address"]
-                + " Hostname: "
-                + lease_ip_info["hostname"]
-                + " MAC: "
-                + lease_ip_info["mac_address"]
-            )
-        try:
-            ip_info = userdata.ipam_util.update_ip_information(
-                ip_address=lease_ip_info["ip_address"],
-                hostname=lease_ip_info["hostname"],
-                mac_address=lease_ip_info["mac_address"],
-            )
-            logging.info("Updated IP Info: " + str(ip_info))
-            userdata.increment_count()
-        except Exception as e:
-            logging.error("Error updating IP information: " + str(e))
-    if lease_ip_info and lease_ip_info["HOOK_TYPE"] in ["RELEASE", "EXPIRE", "RECOVER"]:
-        try:
-            userdata.ipam_util.clear_ip_information(
-                ip_address=lease_ip_info["ip_address"]
-            )
-            userdata.increment_count()
+    if not lease_ip_info:
+        return
+
+    # Dispatch function mapping HOOK_TYPE to processor methods
+    dispatch = {
+        "RENEW": _process_commit,
+        "RELEASE": _process_lease_change,
+        "EXPIRE": _process_lease_change,
+        "RECOVER": _process_lease_change,
+    }
+
+    processor = dispatch.get(lease_ip_info["HOOK_TYPE"])
+    if processor:
+        processor(userdata, lease_ip_info)
+    else:
+        if logging.getLogger().isEnabledFor(logging.INFO):
             logging.info(
-                "Processed RELEASE/EXPIRE/RECOVER for IP: "
-                + lease_ip_info["ip_address"]
+                f"Unhandled HOOK_TYPE '{lease_ip_info['HOOK_TYPE']}' received, skipping processing."
             )
-        except Exception as e:
-            logging.error("Error clearing IP information: " + str(e))
-
-
-class SpecificLevelFilter(logging.Filter):
-    """Filter to allow only specific level logs to pass through."""
-
-    def __init__(self, level: int):
-        super().__init__()
-        self.level = level
-
-    def filter(self, record: logging.LogRecord) -> bool:
-        # Only allow logs that are equal to the specified level
-        return record.levelno == self.level
-
-
-class ExcludeLevelFilter(logging.Filter):
-    """Filter to exclude specific level logs from passing through."""
-
-    def __init__(self, level: int):
-        super().__init__()
-        self.level = level
-
-    def filter(self, record: logging.LogRecord) -> bool:
-        # Only allow logs that are not equal to the specified level
-        return record.levelno != self.level
-
-
-def setup_split_logging(stderr_level=logging.WARNING, stdout_level=logging.INFO):
-    logger = logging.getLogger()
-    logger.setLevel(
-        min(stderr_level, stdout_level)
-    )  # Set the logger to the lowest level you want to capture
-
-    if logger.hasHandlers():
-        logger.handlers.clear()
-
-    stdout_formatter = logging.Formatter("%(asctime)s - %(message)s")
-    stderr_formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
-
-    stdout_handler = logging.StreamHandler(sys.stdout)
-    stdout_handler.setLevel(stdout_level)
-    stdout_handler.setFormatter(stdout_formatter)
-
-    stderr_handler = logging.StreamHandler(sys.stderr)
-    stderr_handler.setLevel(stderr_level)
-    stderr_handler.setFormatter(stderr_formatter)
-    stdout_handler.addFilter(SpecificLevelFilter(stdout_level))
-    stderr_handler.addFilter(ExcludeLevelFilter(stdout_level))
-
-    logger.addHandler(stdout_handler)
-    logger.addHandler(stderr_handler)
 
 
 def main():
@@ -227,15 +212,6 @@ def main():
             auth=mqtt_auth,
             payload=mqtt_payload,
         )
-        """
-        if logging.getLogger().isEnabledFor(logging.DEBUG):   
-            logging.debug(
-                "MARK: last update: "
-                + userdata.last_update_time
-                + " updates processed: "
-                + str(userdata.count)
-            )
-        """
         time.sleep(60)
     client.loop_stop(force=False)
 

logging_setup.py

@@ -0,0 +1,50 @@
+import logging
+import sys
+
+class SpecificLevelFilter(logging.Filter):
+    """Filter to allow only specific level logs to pass through."""
+    def __init__(self, level: int):
+        super().__init__()
+        self.level = level
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        # Only allow logs that are equal to the specified level
+        return record.levelno == self.level
+
+class ExcludeLevelFilter(logging.Filter):
+    """Filter to exclude specific level logs from passing through."""
+    def __init__(self, level: int):
+        super().__init__()
+        self.level = level
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        # Only allow logs that are not equal to the specified level
+        return record.levelno != self.level
+
+def setup_split_logging(stderr_level: int = logging.WARNING, stdout_level: int = logging.INFO):
+    """
+    Configures the root logger to split output between stdout and stderr
+    with different logging levels.
+    """
+    logger = logging.getLogger()
+    logger.setLevel(min(stderr_level, stdout_level))  # Set the logger to the lowest level you want to capture
+
+    if logger.hasHandlers():
+        logger.handlers.clear()
+
+    stdout_formatter = logging.Formatter("%(asctime)s - %(message)s")
+    stderr_formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
+
+    stdout_handler = logging.StreamHandler(sys.stdout)
+    stdout_handler.setLevel(stdout_level)
+    stdout_handler.setFormatter(stdout_formatter)
+    stdout_handler.addFilter(SpecificLevelFilter(stdout_level))
+
+    stderr_handler = logging.StreamHandler(sys.stderr)
+    stderr_handler.setLevel(stderr_level)
+    stderr_handler.setFormatter(stderr_formatter)
+    stderr_handler.addFilter(ExcludeLevelFilter(stdout_level))
+
+    logger.addHandler(stdout_handler)
+    logger.addHandler(stderr_handler)
+    return logger