Kea support in DHCPLSS plugin (or a new plugin)?

[void-spark]

Hi!

Kea is supposedly/possibly the future of DHCP (https://www.isc.org/kea/)
It would be nice if we can pull in DHCP leases from it :)

From a basic install on debian it creates this file:
cat /var/lib/kea/kea-leases4.csv
address,hwaddr,client_id,valid_lifetime,expire,subnet_id,fqdn_fwd,fqdn_rev,hostname,state,user_context,pool_id

But as I understand it Kea also supports databases, in which case scanning the above file won't help.
So another/additional approach could be API?
Seems someone had a look at that a long time ago:
https://www.reddit.com/r/opnsense/comments/1etkrpv/comment/lij7nzu/
Some actual API doc:
https://kea.readthedocs.io/en/latest/api.html#lease4-get-all
https://kea.readthedocs.io/en/latest/api.html#ref-reservation-get-all
Not sure yet what is required to use the API though :)

[void-spark]

Some hints..

To go the API way, you want to have this in your kea config, this is for debian 13:

    "control-socket": {
        "socket-type": "unix",
        "socket-name": "/run/kea/kea4-ctrl-socket"
    },
    
    "hooks-libraries": [
        {
            "library": "/usr/lib/x86_64-linux-gnu/kea/hooks/libdhcp_lease_cmds.so"
        }
    ],

And you need kea-ctrl-agent, with a config something like:

{
"Control-agent": {
    "http-host": "127.0.0.1",
    "http-port": 8000,

    "authentication": {
        "type": "basic",
        "realm": "Kea Control Agent",
        "directory": "/etc/kea",
        "clients": [
            {
                "user": "kea-api",
                "password-file": "kea-api-password"
            }
        ]
    },
    "control-sockets": {
        "dhcp4": {
            "socket-type": "unix",
            "socket-name": "/run/kea/kea4-ctrl-socket"
        }
    },
    "loggers": [
    {
        "name": "kea-ctrl-agent",
        "output-options": [
            {
                "output": "stdout",
                "pattern": "%-5p %m\n"
            }
        ],
        "severity": "INFO",
        "debuglevel": 0
    }
  ]
}
}

A basic python script to pull the info we want looks like this then:

import requests

KEA_API_URL = "http://127.0.0.1:8000"
KEA_USER = "kea-api"
KEA_PASS = "yourpasswordhere"

def get_online_devices():
    payload = {
        "command": "lease4-get-all",
        "service": ["dhcp4"]
    }

    try:
        response = requests.post(
            KEA_API_URL,
            json=payload,
            auth=(KEA_USER, KEA_PASS),
            timeout=5
        )
        response.raise_for_status()

        # Kea returns a list of results (one per service requested)
        results = response.json()
        for entry in results:
            if entry.get("result") == 0:
                leases = entry["arguments"].get("leases", [])
                print(f"{'IP':<15} | {'MAC':<17} | {'Hostname'}")
                print("-" * 50)
                for l in leases:
                    print(f"{l['ip-address']:<15} | {l['hw-address']:<17} | {l.get('hostname', 'N/A')}")
            else:
                print(f"Kea Error: {entry.get('text')}")

    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 401:
            print("Auth Failed: Check your username and the content of your password-file.")
        else:
            print(f"HTTP Error: {e}")
    except Exception as e:
        print(f"Connection Error: {e}")

if __name__ == "__main__":
    get_online_devices()

[void-spark]

And a start of an integration:

{
  "code_name": "kea_api",
  "unique_prefix": "KEALSS",
  "plugin_type": "device_scanner",
  "execution_order" : "Layer_3",
  "enabled": true,
  "data_source": "script",
  "show_ui": true,
  "localized": ["display_name", "description", "icon"],
  "display_name": [{"language_code": "en_us", "string": "Kea DHCP API"}],
  "description": [{"language_code": "en_us", "string": "Imports leases via Kea Control Agent REST API"}],
  "icon": [{"language_code": "en_us", "string": "<i class=\"fa-solid fa-hourglass-half\"></i>"}],
  "settings": [
    {
      "function": "RUN",
      "events": ["run"],
      "type": {
        "dataType": "string",
        "elements": [{"elementType": "select", "elementOptions": [], "transformers": []}]
      },
      "default_value": "disabled",
      "options": [
        "disabled",
        "once",
        "schedule",
        "always_after_scan",
        "on_new_device"
      ],
      "localized": ["name", "description"],
      "name": [{"language_code": "en_us", "string": "When to run"}],
      "description": [
        {
          "language_code": "en_us",
          "string": "Enable import of devices from <code>Kea API</code>. If you select <code>schedule</code> the scheduling settings from below are applied. If you select <code>once</code> the scan is run only once on start of the application (container) or after you update your settings. ⚠ Use the same schedule if you have multiple <i class=\"fa-solid fa-magnifying-glass-plus\"></i> Device scanners enabled."
        }      
      ]
    },
    {
      "function": "CMD",
      "type": {
        "dataType": "string",
        "elements": [
          { "elementType": "input", "elementOptions": [], "transformers": [] }
        ]
      },
      "default_value": "python3 /app/front/plugins/kea_api/script.py",
      "options": [],
      "localized": ["name", "description"],
      "name": [{ "language_code": "en_us", "string": "Command" }],
      "description": [{"language_code": "en_us", "string": "Command to run"}]
    },
    {
        "function": "URL",
        "localized": ["name", "description"],
        "name": [{"language_code": "en_us", "string": "API URL"}],
        "description": [{"language_code": "en_us", "string": "Kea Control Agent URL"}],
        "type": {
          "dataType": "string",
          "elements": [
            {
              "elementType": "input",
              "elementOptions": [],
              "transformers": []
            }
          ]
        },
        "default_value": "http://127.0.0.1:8000"
    },
    {
        "function": "USER",
        "localized": ["name", "description"],
        "name": [{"language_code": "en_us", "string": "API User"}],
        "description": [{"language_code": "en_us", "string": "Basic Auth Username"}],
        "type": {
          "dataType": "string",
          "elements": [
            {
              "elementType": "input",
              "elementOptions": [],
              "transformers": []
            }
          ]
        },
        "default_value": "kea-api"
    },
    {
        "function": "PASS",
        "localized": ["name", "description"],
        "name": [{"language_code": "en_us", "string": "API Password"}],
        "description": [{"language_code": "en_us", "string": "Basic Auth Password"}],
        "type": {
          "dataType": "string",
          "elements": [
            {
              "elementType": "input",
              "elementOptions": [{"type": "password"}],
              "transformers": []
            }
          ]
        },
        "default_value": ""
    }
  ]
}

#!/usr/bin/env python3
import os
import sys
import requests

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../server'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../plugins'))

from plugin_helper import Plugin_Objects, mylog, handleEmpty, is_mac
from helper import get_setting_value
from const import logPath

pluginName = "KEALSS"
LOG_PATH = logPath + "/plugins"
LOG_FILE = os.path.join(LOG_PATH, f'script.{pluginName}.log')
RESULT_FILE = os.path.join(LOG_PATH, f'last_result.{pluginName}.log')

plugin_objects = Plugin_Objects(RESULT_FILE)


def main():
    try:
        url = get_setting_value(f'{pluginName}_URL')
        user = get_setting_value(f'{pluginName}_USER')
        password = get_setting_value(f'{pluginName}_PASS')

        mylog('verbose', [f'[{pluginName}] Querying Kea API at {url}'])

        payload = {'command': 'lease4-get-all', 'service': ['dhcp4']}

        response = requests.post(url, json=payload, auth=(user, password), timeout=10)
        response.raise_for_status()
        data = response.json()

        count = 0
        for entry in data:
            if entry.get("result") == 0:
                leases = entry["arguments"].get("leases", [])
                for l in leases:
                    mac = l['hw-address']
                    if is_mac(mac):
                        plugin_objects.add_object(
                            primaryId   = handleEmpty(mac),
                            secondaryId = handleEmpty(l['ip-address']),
                            # Default (or assigned) (0), declined (1), expired-reclaimed (2), released (3), and registered (4)).
                            watched1     = handleEmpty(l['state']),
                            watched2     = handleEmpty(l['hostname']),
                            watched3     = handleEmpty(l['valid-lft']),
                            watched4     = None,
                            extra        = None,
                            foreignKey   = handleEmpty(mac)
                        )

                    count += 1
        
        plugin_objects.write_result_file()


        mylog('verbose', [f'[{pluginName}] Successfully imported {count} devices'])

    except Exception as e:
        mylog('none', [f'[{pluginName}] ⚠ ERROR: ', e.output])



if __name__ == '__main__':
    main()

[jokob-sk]

Hi - thanks for the idea - I think the API route is cleaner. Happy to review a PR for it. See the plugins guide for details: https://docs.netalertx.com/PLUGINS_DEV/?h=plugin+deve#plugin-development-guide

[void-spark]

I noticed the plugin development documentation is a bit of state of flux, not everything is correct anymore, and there's mildy different approaches in different places :)

Not sure I have the time to go to a full review worthy PR :) Maybe someone else wants this too and can take it a bit further :)

[jokob-sk]

The documentation should be up to date, so if you spot inconsistent information please let me know. Regrettably I don't have a kea server so can't do it myself. Let's see, maybe someone from the community can help.

[void-spark]

For documentation, bits I think I noticed, and I could be wrong, I spent till four at midnight on this yesterday, you get a bit blurry vision :) It was not 5 minutes :D

https://github.com/netalertx/NetAlertX/blob/main/docs/PLUGINS_DEV_QUICK_START.md

This was the biggest one, and started to make me doubt everything else, I think this bit has changed, on main anyways?

    plugin_objects.add_object(
        objectPrimaryId="example_id",
        objectSecondaryId=None,
        DateTime="2023-01-02 15:56:30",
        watchedValue1="value1",
        watchedValue2=None,
        watchedValue3=None,
        watchedValue4=None,
        Extra="additional_data",
        ForeignKey=None
    )

Seems to be this now, or do both work, of so which is preferred?

      plugin_objects.add_object(
          primaryId   = ..
          secondaryId = ..
          watched1     = ..
          watched2     = ..
          watched3     = ..
          watched4     = ..
          extra        = ..
          foreignKey   = ..
      )

I see a lot of files like this:

def main():
    mylog('verbose', [f'[{pluginName}] In script'])
..
    
if __name__ == '__main__':
    main()

But the quick start has everything at the root level, not main.
I realize now that it's almost the same thing, but since I don't do a lot of python it had me confused at first :)

https://github.com/netalertx/NetAlertX/blob/main/docs/PLUGINS_DEV_SETTINGS.md

It has this bit:

And I see now there's many examples that show it correctly, but somehow I hit this one and thought the other values were optional :) A small "... " or so would help that :)

https://github.com/netalertx/NetAlertX/blob/main/docs/PLUGINS_DEV_DATA_CONTRACT.md
This page also confused me on if I should use objectPrimaryId or primaryId, etc.
I guess in the tables still the long names are used, and in the python scripts the short names.
It makes more sense now, but trying to get your feet on the ground it was all quite confusing :)

https://github.com/netalertx/NetAlertX/blob/main/front/plugins/__template/rename_me.py
Has (I removed some logging related bits):

import os
import sys

INSTALL_PATH = os.getenv('NETALERTX_APP', '/app')
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])

from const import logPath
from plugin_helper import Plugin_Objects
from logger import mylog, Logger
from helper import get_setting_value

https://github.com/netalertx/NetAlertX/blob/main/docs/PLUGINS_DEV_QUICK_START.md
Has:

import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(file), '../../server'))
sys.path.insert(0, os.path.join(os.path.dirname(file), '../plugins'))

from plugin_helper import Plugin_Objects, mylog
from helper import get_setting_value
from const import logPath

I wasn't sure if there was a difference, if one is an old style, especially since I'm again not that good in python :)

[jokob-sk]

Thanks for this - will clean this up!

[void-spark]

@jokob-sk Ok, fine, created a PR :)
See if it's enough for you, I'm out of weekend now :)
#1636

[jokob-sk]

Thanks a LOT 🙏 left a couple of comments, once fixed, I think good to go

[void-spark]

@jokob-sk Ok, I think I covered everything, even most of what the naggy rabbit had to say :)

Still works:
22:34:58 [ 2026-05-11 20:34:58 ] START Run: KEALSS
22:34:58 [Plugins] Executing: python3 /app/front/plugins/kea_api/script.py
22:34:59 [Plugins] Output: [plugin_helper] reading config file
20:34:59 [KEALSS] Querying Kea API at http://127.0.0.1:8000
20:34:59 [KEALSS] Kea API response: 40 IPv4 lease(s) found.
20:34:59 [KEALSS] Successfully imported 40 devices reported by Kea API

22:34:59 [Plugins] Processed and deleted file: /tmp/log/plugins/last_result.KEALSS.log
22:34:59 [Plugins] SUCCESS for KEALSS received 40 entries
22:34:59 [Plugins] Processing : KEALSS
22:34:59 [Plugins] Should I update API (userUpdatedDevices): False
22:34:59 [ 2026-05-11 20:34:59 ] END Run: KEALSS