diff options
Diffstat (limited to 'hosts/hosts.py')
| -rwxr-xr-x | hosts/hosts.py | 88 |
1 files changed, 83 insertions, 5 deletions
diff --git a/hosts/hosts.py b/hosts/hosts.py index 424a674..0d1b664 100755 --- a/hosts/hosts.py +++ b/hosts/hosts.py @@ -15,13 +15,23 @@ the ~/.ssh/config aliases that point at it, plus one entry per alias that points at no note, so every alias the rofi menu listed is still here. hosts.py print the list + hosts.py --probe print {name: reachable} for the same list hosts.py --check run the self-check + +Each entry carries a `group`: the Proxmox node it runs on, its own name if +other notes name it as their node, "Devices" for anything else on a private +address, "Remote" for the rest. """ +import ipaddress import json import os +import socket +import subprocess import sys +from concurrent.futures import ThreadPoolExecutor from pathlib import Path +from urllib.parse import urlsplit VAULT = os.environ.get( "HOSTS_DIR", "~/Documents/Obsidian/my KB/Knowledge base/Local Network/services") @@ -95,15 +105,69 @@ def merge(notes, aliases): "hostname": target, "node": "", "ct_id": "", "aliases": [a["alias"]], "user": a["user"], }) - return out - - -def main(): + return group(out) + + +def private(ip): + try: + return ipaddress.ip_address(ip).is_private + except ValueError: + return False + + +def group(hosts): + """Set each host's group and sort: nodes in name order, each with its own + card first, then Devices, then Remote.""" + nodes = sorted({h["node"] for h in hosts if h["node"]}) + for h in hosts: + h["group"] = (h["node"] or (h["name"] if h["name"] in nodes else "") + or ("Devices" if private(h["ip"]) else "Remote")) + order = nodes + ["Devices", "Remote"] + # sorted() is stable, so within a group the note order holds. + return sorted(hosts, key=lambda h: (order.index(h["group"]), + h["name"] != h["group"])) + + +def target(h): + """What to probe: (host, tcp port), port None meaning ICMP. A service is + up when its port answers, not merely when its CT does.""" + url = urlsplit(h["url"]) if h["url"] else None + port = h["port"] or (url and (url.port or {"http": 80, "https": 443}.get(url.scheme))) + return h["ip"] or h["hostname"], int(port) if port else None + + +def reachable(h): + host, port = target(h) + if not host: + return False + if port: + try: + socket.create_connection((host, port), timeout=1).close() + return True + except OSError: + return False + return subprocess.run(["ping", "-c1", "-W1", host], stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL).returncode == 0 + + +def probe(hosts): + # ponytail: one thread per host, ~1s for the whole list; a pool cap + # if the list ever reaches the hundreds. + with ThreadPoolExecutor(max_workers=max(1, len(hosts))) as pool: + return dict(zip((h["name"] for h in hosts), pool.map(reachable, hosts))) + + +def load(): notes = [(p.stem, frontmatter(p.read_text(errors="replace"))) for p in sorted(Path(VAULT).expanduser().glob("*.md"))] cfg = Path(SSH_CONFIG).expanduser() aliases = ssh_hosts(cfg.read_text(errors="replace")) if cfg.exists() else [] - json.dump(merge(notes, aliases), sys.stdout) + return merge(notes, aliases) + + +def main(): + hosts = load() + json.dump(probe(hosts) if "--probe" in sys.argv else hosts, sys.stdout) def check(): @@ -128,6 +192,20 @@ def check(): assert by["Media"]["aliases"] == ["media"], "short hostname match" assert by["lone"]["port"] == "2222" and by["lone"]["node"] == "" assert len(hosts) == 4 + assert by["lone"]["group"] == "Remote" and by["Box"]["group"] == "HP server" + + g = group([{"name": "dev", "node": "", "ip": "10.0.0.3"}, + {"name": "svc", "node": "N", "ip": "10.0.0.4"}, + {"name": "far", "node": "", "ip": "example.org"}, + {"name": "N", "node": "", "ip": "10.0.0.2"}]) + assert [(h["name"], h["group"]) for h in g] == [ + ("N", "N"), ("svc", "N"), ("dev", "Devices"), ("far", "Remote")], g + + t = lambda **kw: target({"ip": "", "hostname": "", "port": "", "url": "", **kw}) + assert t(ip="10.0.0.7", port="5000") == ("10.0.0.7", 5000) + assert t(ip="10.0.0.7", url="https://x.example.org") == ("10.0.0.7", 443) + assert t(ip="10.0.0.7", url="http://x.example.org:81") == ("10.0.0.7", 81) + assert t(hostname="example.org") == ("example.org", None) print("ok") |