diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..b77e73a --- /dev/null +++ b/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. \ No newline at end of file diff --git a/ipam_utils/utils.py b/ipam_utils/utils.py index eff9eb0..20e8a82 100644 --- a/ipam_utils/utils.py +++ b/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 diff --git a/kea_listener.py b/kea_listener.py index a22e1e8..c51b69c 100644 --- a/kea_listener.py +++ b/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) diff --git a/logging_setup.py b/logging_setup.py new file mode 100644 index 0000000..d28af32 --- /dev/null +++ b/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 \ No newline at end of file