97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
"""Publish on the LAN: services (JSON config) or a bare hostname->IP."""
|
|
import json
|
|
import socket
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from zeroconf import Zeroconf, ServiceInfo, NonUniqueNameException
|
|
|
|
|
|
def _local_ipv4() -> str:
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
try:
|
|
s.connect(("8.8.8.8", 80))
|
|
return s.getsockname()[0]
|
|
except OSError:
|
|
return "127.0.0.1"
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def _addr_bytes(ip: str) -> bytes:
|
|
if ":" in ip:
|
|
return socket.inet_pton(socket.AF_INET6, ip)
|
|
return socket.inet_aton(ip)
|
|
|
|
|
|
_DEFAULT_CONFIG = Path.home() / ".config" / "mdns" / "services.json"
|
|
|
|
|
|
def _load_config(path: str | None) -> dict:
|
|
p = Path(path).expanduser() if path else _DEFAULT_CONFIG
|
|
if not p.exists():
|
|
return {"hostname": "movil", "services": []}
|
|
try:
|
|
return json.loads(p.read_text())
|
|
except Exception as e:
|
|
print(f"warning: bad config {p}: {e}")
|
|
return {"hostname": "movil", "services": []}
|
|
|
|
|
|
def _serve(zc: Zeroconf, infos: list[ServiceInfo], verbose: bool, label: str) -> None:
|
|
for i in infos:
|
|
try: # tolerate name conflicts per RFC 6762 (probe + rename)
|
|
zc.register_service(i)
|
|
except NonUniqueNameException:
|
|
zc.register_service(i, allow_name_change=True)
|
|
if verbose:
|
|
print(f"[mdns] {label} ({len(infos)} record(s)). Ctrl+C to stop.")
|
|
try:
|
|
while True:
|
|
time.sleep(60)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
for i in infos:
|
|
try:
|
|
zc.unregister_service(i)
|
|
except Exception:
|
|
pass
|
|
zc.close()
|
|
if verbose:
|
|
print("[mdns] stopped, goodbye packets sent.")
|
|
|
|
|
|
def _workstation(host: str, ip: str) -> ServiceInfo:
|
|
return ServiceInfo(
|
|
type_="_workstation._tcp.local.",
|
|
name=f"{host} [{ip.replace('.', '-')}]._workstation._tcp.local.",
|
|
addresses=[_addr_bytes(ip)], port=9, server=f"{host}.local.",
|
|
)
|
|
|
|
|
|
def run(config_path: str | None = None, hostname: str | None = None,
|
|
verbose: bool = True) -> None:
|
|
cfg = _load_config(config_path)
|
|
host = hostname or cfg.get("hostname", "movil")
|
|
ip = _local_ipv4()
|
|
infos = [_workstation(host, ip)]
|
|
for svc in (cfg.get("services") or []):
|
|
try:
|
|
t = svc["type"].rstrip(".") + ".local."
|
|
infos.append(ServiceInfo(
|
|
type_=t, name=f'{svc.get("name", host)}.{t}',
|
|
addresses=[_addr_bytes(ip)], port=int(svc["port"]),
|
|
server=f"{host}.local.", properties=svc.get("txt", {}),
|
|
))
|
|
except Exception as e:
|
|
print(f"warning: could not register {svc}: {e}")
|
|
_serve(Zeroconf(), infos, verbose, f"published {host}.local ({ip})")
|
|
|
|
|
|
def publish_address(name: str, ip: str, verbose: bool = True) -> None:
|
|
"""Publish a bare hostname -> IP mapping (like `avahi-publish -a`)."""
|
|
host = name.rstrip(".").removesuffix(".local")
|
|
_serve(Zeroconf(), [_workstation(host, ip)], verbose,
|
|
f"published {host}.local -> {ip}")
|