avahi-exporter.py

#!/usr/bin/env python3

import io
import json
import socket

def q(s):
  # For double quotes
  return json.dumps(s, ensure_ascii=False)

from gi.repository import GLib
import avahi, dbus
from dbus.mainloop.glib import DBusGMainLoop

from twisted.internet import gireactor
gireactor.install()
from twisted.internet import reactor
from twisted.web import server, resource

DBusGMainLoop(set_as_default=True)

bus = dbus.SystemBus()
avahi_server = dbus.Interface(
    bus.get_object(avahi.DBUS_NAME, avahi.DBUS_PATH_SERVER), avahi.DBUS_INTERFACE_SERVER
)

DOMAIN = avahi_server.GetDomainName()

def protocol_str(val):
    return ["IPv4", "IPv6"][val]

def interface_str(val):
    return socket.if_indextoname(val)

service_type_browser = dbus.Interface(
    bus.get_object(
        avahi.DBUS_NAME,
        avahi_server.ServiceTypeBrowserNew(
            avahi.IF_UNSPEC,
            avahi.PROTO_UNSPEC,
            DOMAIN,
            dbus.UInt32(0),
        ),
    ),
    avahi.DBUS_INTERFACE_SERVICE_TYPE_BROWSER,
)

service_types = {}
services = set()

def handle_new_service_type(interface, protocol, stype, domain, flags):
    if stype in service_types:
        return

    service_browser = dbus.Interface(
        bus.get_object(
            avahi.DBUS_NAME,
            avahi_server.ServiceBrowserNew(
                avahi.IF_UNSPEC,
                avahi.PROTO_UNSPEC,
                stype,
                DOMAIN,
                dbus.UInt32(0),
            ),
        ),
        avahi.DBUS_INTERFACE_SERVICE_BROWSER,
    )
    service_browser.connect_to_signal("ItemNew", handle_new_service)
    service_browser.connect_to_signal("ItemRemove", handle_remove_service)
    service_types[stype] = service_browser

def handle_new_service(interface, protocol, name, stype, domain, flags):
    key = (interface, protocol, name, stype, domain)
    services.add(key)

def handle_remove_service(interface, protocol, name, stype, domain, flags):
    key = (interface, protocol, name, stype, domain)
    services.remove(key)

service_type_browser.connect_to_signal("ItemNew", handle_new_service_type)

class Index(resource.Resource):
    def getChild(self, name, request):
        if name == b'':
            return self
        return super().getChild(name, request)
    def render_GET(self, request):
        return b'avahi exporter\n<a href="/metrics">/metrics</a>\n'

class AvahiExportServer(resource.Resource):
    isLeaf = True
    def render_GET(self, request):
        request.responseHeaders.addRawHeader("Content-Type", 'text/plain; charset=utf-8')
        out = io.StringIO()
        for service in services:
            (interface, protocol, name, stype, domain) = service
            interface = interface_str(interface)
            protocol = protocol_str(protocol)
            out.write(f'avahi_up{{interface={q(interface)},protocol={q(protocol)},name={q(name)},type={q(stype)},domain={q(domain)}}} 1\n')
        return out.getvalue().encode()

root = Index()
root.putChild(b"metrics", AvahiExportServer())

site = server.Site(root)
reactor.listenTCP(9713, site, interface='127.0.0.1')
reactor.run()