#!/usr/bin/env python3 """Regenerate intranet/index.html by polling each server over SSH.""" from __future__ import annotations import base64 import json import pathlib import shlex import subprocess import textwrap from dataclasses import dataclass from typing import Any, Dict, List, Optional REPO_ROOT = pathlib.Path(__file__).resolve().parents[1] OUTPUT_PATH = REPO_ROOT / "intranet" / "index.html" @dataclass class HostConfig: slug: str ssh: str title: str subtitle: str tags: List[str] vhost_notes: Dict[str, str] docker_notes: Dict[str, str] HOSTS: List[HostConfig] = [ HostConfig( slug="la", ssh="root@la.chelseawoodruff.net", title="Los Angeles", subtitle="la.chelseawoodruff.net", tags=["nginx", "FRP relay"], vhost_notes={ "chat.scorpi.us": "Public chat frontend with HTTPS redirect", }, docker_notes={ "frps": "Snowdreamtech FRP reverse proxy", }, ), HostConfig( slug="virginia", ssh="root@virginia.chelseawoodruff.net", title="Virginia", subtitle="virginia.chelseawoodruff.net", tags=["nginx", "Git", "Vikunja", "Dokuwiki", "Custom apps"], vhost_notes={ "hightimesfrom.space": "Personal landing page", "pm.scorpi.us": "Leantime / PM suite", "wiki.scorpi.us": "Dokuwiki knowledge base", "news.scorpi.us": "Pseudo news clone", "git.scorpi.us": "Gitea instance", "reddit.scorpi.us": "Reddit proxy (forced HTTPS)", "blocked.scorpi.us": "Block page mirror", "requests.scorpi.us": "Requesty helper", "youtube.scorpi.us": "PseudoTube experiment", }, docker_notes={ "claude-proxy": "Claude proxy forwarder", "solar_420": "Solartime demo app", "frps": "Snowdreamtech FRP relay", "balanceboard_app": "Balanceboard UI", "balanceboard_postgres": "Balanceboard DB", "vikunja_vikunja_1": "Vikunja task manager", "vikunja_db_1": "Vikunja Postgres", "gitea": "Gitea service", "gitea_db_1": "Gitea Postgres", "dokuwiki": "Dokuwiki container", "requesty": "Requesty API", "psuedo-tube": "PseudoTube UI", "block-page": "Block-page helper", }, ), HostConfig( slug="chicago", ssh="root@chicago.scorpi.us", title="Chicago", subtitle="chicago.scorpi.us", tags=["nginx", "ADHDbot", "ntfy", "IRC"], vhost_notes={ "adhd.scorpi.us": "ADHDbot UI/API", "matrix.scorpi.us": "Matrix homeserver proxy", "ntfy.scorpi.us": "ntfy notification hub", }, docker_notes={ "adhdbot-app_adhdbot_1": "ADHDbot FastAPI stack", "ntfy": "ntfy topic server", "inspircd": "InspIRCd daemon", }, ), HostConfig( slug="dallas", ssh="root@dallas.scorpi.us", title="Dallas", subtitle="dallas.scorpi.us", tags=["nginx", "Automation"], vhost_notes={ "n8n.dallas.scorpi.us": "n8n low-code automation suite", }, docker_notes={ "n8n_n8n_1": "n8n worker (localhost only)", }, ), HostConfig( slug="phoenix", ssh="root@phoenix.scorpi.us", title="Phoenix", subtitle="phoenix.scorpi.us", tags=["Discourse", "Discord bridge"], vhost_notes={}, docker_notes={ "app": "Discourse stack", "discord-discourse-bridge": "Discord ↔ Discourse bridge", }, ), ] REMOTE_PY = textwrap.dedent( """ import glob import json import os import subprocess def collect_nginx(): base = "/etc/nginx/sites-enabled" sites = [] if not os.path.isdir(base): return sites for path in sorted(glob.glob(os.path.join(base, "*"))): if not os.path.isfile(path): continue try: with open(path, "r", encoding="utf-8") as handle: lines = handle.readlines() except OSError: continue server_names = [] for line in lines: stripped = line.strip() if not stripped or stripped.startswith("#"): continue if stripped.startswith("server_name"): content = stripped.split(None, 1)[1] if " " in stripped else "" content = content.split("#", 1)[0].strip() if content.endswith(";"): content = content[:-1].strip() if content: server_names.extend([token for token in content.split() if token]) sites.append({"filename": os.path.basename(path), "server_names": server_names}) return sites def collect_docker(): try: output = subprocess.check_output( ["docker", "ps", "--format", "{{json .}}"], text=True, timeout=10 ) except Exception: return [] containers = [] for line in output.splitlines(): line = line.strip() if not line: continue try: data = json.loads(line) except json.JSONDecodeError: continue containers.append( { "name": data.get("Names"), "image": data.get("Image"), "ports": data.get("Ports"), } ) return containers payload = { "hostname": os.uname().nodename, "nginx": collect_nginx(), "docker": collect_docker(), } print("{{JSON}}" + json.dumps(payload) + "{{/JSON}}") """ ).strip() def run_remote_script(target: str) -> Dict[str, Any]: encoded = base64.b64encode(REMOTE_PY.encode("utf-8")).decode("ascii") remote_cmd = ( "python3 -c " + shlex.quote( "import base64, json; exec(base64.b64decode({}))".format(repr(encoded)) ) ) ssh_cmd = [ "ssh", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no", target, remote_cmd, ] completed = subprocess.run( ssh_cmd, capture_output=True, text=True, check=False, ) if completed.returncode != 0: raise RuntimeError(completed.stderr.strip() or completed.stdout.strip()) marker_start = "{{JSON}}" marker_end = "{{/JSON}}" stdout = completed.stdout start = stdout.find(marker_start) end = stdout.find(marker_end) if start == -1 or end == -1: raise RuntimeError("Could not locate JSON payload in ssh output") json_text = stdout[start + len(marker_start) : end] return json.loads(json_text) def format_ports(ports: Optional[str]) -> str: ports = (ports or "").strip() return ports if ports else "—" def render_host_card(config: HostConfig, payload: Optional[Dict[str, Any]], error: Optional[str]) -> str: tag_html = "".join(f'{tag}' for tag in config.tags) if error: body = f'

Unable to load data: {error}

' nginx_html = "" docker_html = "" else: nginx_entries = payload.get("nginx", []) if payload else [] docker_entries = payload.get("docker", []) if payload else [] if nginx_entries: items = [] for entry in nginx_entries: names = entry.get("server_names") or [entry.get("filename", "unknown")] label = ", ".join(names) note = config.vhost_notes.get(names[0]) or config.vhost_notes.get(entry.get("filename", "")) if note: items.append(f"
  • {label} – {note}
    • ") else: items.append(f"
  • {label}
    • ") nginx_html = "

    nginx sites

    " else: nginx_html = "

    nginx sites

    No vhosts detected.

    " if docker_entries: rows = [] for entry in docker_entries: name = entry.get("name") or "unknown" image = entry.get("image") or "?" ports = format_ports(entry.get("ports")) note = config.docker_notes.get(name) note_span = f'
    {note}' if note else "" rows.append( "" f"{name}{note_span}" f"{image}" f"{ports}" "" ) docker_html = ( "

    Docker

    " "" "" + "".join(rows) + "
    NameImagePorts
    " ) else: docker_html = "

    Docker

    No containers running.

    " body = nginx_html + docker_html return textwrap.dedent( f"""

    {config.title}

    {config.subtitle}

    {tag_html}
    {body}
    """ ).strip() def render_html(reports: List[str]) -> str: cards_html = "\n".join(reports) return textwrap.dedent( f""" Service Inventory

    Internal inventory

    Intranet Services

    Live snapshot of nginx vhosts and Docker workloads across Chelsea's network.

    {cards_html}
    """ ).strip() def main() -> None: reports: List[str] = [] for host in HOSTS: try: payload = run_remote_script(host.ssh) reports.append(render_host_card(host, payload, error=None)) except Exception as exc: # pragma: no cover - network dependent reports.append(render_host_card(host, payload=None, error=str(exc))) html = render_html(reports) OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True) OUTPUT_PATH.write_text(html, encoding="utf-8") print(f"Wrote {OUTPUT_PATH}") if __name__ == "__main__": main()