yunohost/bin/yunomdns
tituspijean 358885dc62
[mdns] Avoid miserably failing if another service exists on the network
also, service names do not clash anymore accross same device but different interfaces
2021-09-26 11:41:16 +02:00

113 lines
3.3 KiB
Python
Executable file

#!/usr/bin/env python3
"""
Pythonic declaration of mDNS .local domains for YunoHost
"""
import sys
import yaml
from time import sleep
from typing import List, Dict
import ifaddr
from zeroconf import Zeroconf, ServiceInfo
def get_network_local_interfaces() -> Dict[str, Dict[str, List[str]]]:
"""
Returns interfaces with their associated local IPs
"""
def islocal(ip: str) -> bool:
local_prefixes = ["192.168.", "10.", "172.16.", "fc00:"]
return any(ip.startswith(prefix) for prefix in local_prefixes)
interfaces = {
adapter.name: {
"ipv4": [ip.ip for ip in adapter.ips if ip.is_IPv4 and islocal(ip.ip)],
"ipv6": [ip.ip[0] for ip in adapter.ips if ip.is_IPv6 and islocal(ip.ip[0])],
}
for adapter in ifaddr.get_adapters()
if adapter.name != "lo"
}
return interfaces
def main() -> bool:
###
# CONFIG
###
with open("/etc/yunohost/mdns.yml", "r") as f:
config = yaml.safe_load(f) or {}
required_fields = ["interfaces", "domains"]
missing_fields = [field for field in required_fields if field not in config]
if missing_fields:
print("The fields %s are required" % ", ".join(missing_fields))
if config["interfaces"] is None:
print("No interface listed for broadcast.")
return True
if "yunohost.local" not in config["domains"]:
config["domains"].append("yunohost.local")
zcs: Dict[Zeroconf, List[ServiceInfo]] = {}
interfaces = get_network_local_interfaces()
for interface in config["interfaces"]:
if interface not in interfaces:
print(f"Interface {interface} of config file is not present on system.")
continue
ips: List[str] = interfaces[interface]["ipv4"] + interfaces[interface]["ipv6"]
# If at least one IP is listed
if not ips:
continue
# Create a Zeroconf object, and store the ServiceInfos
zc = Zeroconf(interfaces=ips) # type: ignore
zcs[zc] = []
for d in config["domains"]:
d_domain = d.replace(".local", "")
if "." in d_domain:
print(f"{d_domain}.local: subdomains are not supported.")
continue
# Create a ServiceInfo object for each .local domain
zcs[zc].append(
ServiceInfo(
type_="_device-info._tcp.local.",
name=f"{interface}: {d_domain}._device-info._tcp.local.",
parsed_addresses=ips,
port=80,
server=f"{d}.",
)
)
print(f"Adding {d} with addresses {ips} on interface {interface}")
# Run registration
print("Registering...")
for zc, infos in zcs.items():
for info in infos:
zc.register_service(info, allow_name_change=True, cooperating_responders=True)
try:
print("Registered. Press Ctrl+C or stop service to stop.")
while True:
sleep(1)
except KeyboardInterrupt:
pass
finally:
print("Unregistering...")
for zc, infos in zcs.items():
zc.unregister_all_services()
zc.close()
return True
if __name__ == "__main__":
sys.exit(0 if main() else 1)