diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..06d7405 Binary files /dev/null and b/app/__init__.py differ diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..0137d99 --- /dev/null +++ b/app/config.py @@ -0,0 +1,36 @@ +"""Konfiguration, wird aus Umgebungsvariablen / .env geladen.""" +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore") + + # --- phpipam --- + phpipam_base_url: str # z.B. https://ipam.firma.local + phpipam_app_id: str # App-ID der phpIPAM API-App (Administration > API) + + # Variante A: statischer App-Token (App-Security-Type "Token" in phpipam) + phpipam_app_token: str | None = None + + # Variante B: User-Login, Token wird zur Laufzeit geholt + phpipam_username: str | None = None + phpipam_password: str | None = None + + phpipam_verify_ssl: bool = True + + # --- Outline --- + outline_base_url: str # z.B. https://outline.firma.local + outline_api_token: str + outline_collection_id: str # Ziel-Collection für die Subnetz-Dokumente + outline_parent_document_id: str | None = None # z.B. die "Netze"-Seite; Docs werden darunter angelegt + outline_verify_ssl: bool = True + + # --- Sync --- + sync_dry_run: bool = False + + # Automatischer Zeitplan (Cron-Syntax, Standard: jede volle Stunde) + sync_schedule_enabled: bool = True + sync_schedule_cron: str = "0 * * * *" + + +settings = Settings() # type: ignore[call-arg] diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..b416a67 --- /dev/null +++ b/app/main.py @@ -0,0 +1,68 @@ +"""FastAPI App: phpipam -> Outline Sync.""" +from __future__ import annotations + +import logging +from contextlib import asynccontextmanager + +from fastapi import FastAPI, HTTPException + +from .config import settings +from .phpipam_client import PhpIpamClient +from .scheduler import scheduler, start_scheduler, stop_scheduler +from .sync import run_sync + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", +) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + start_scheduler() + yield + stop_scheduler() + + +app = FastAPI(title="phpipam-outline-sync", version="0.1.0", lifespan=lifespan) + + +@app.get("/health") +def health() -> dict: + return {"status": "ok"} + + +@app.get("/subnets") +def subnets() -> list[dict]: + """Nur zum Testen der phpipam-Verbindung, ohne Outline anzufassen.""" + try: + with PhpIpamClient() as ipam: + return ipam.list_subnets() + except Exception as exc: # noqa: BLE001 + raise HTTPException(status_code=502, detail=f"phpipam-Fehler: {exc}") from exc + + +@app.post("/sync") +def sync(dry_run: bool | None = None) -> dict: + """Synchronisiert alle Subnetze + Adressen aus phpipam nach Outline. + + ?dry_run=true -> nichts wird in Outline geschrieben, nur eine Vorschau. + """ + try: + results = run_sync(dry_run=dry_run) + except Exception as exc: # noqa: BLE001 + raise HTTPException(status_code=502, detail=str(exc)) from exc + return {"count": len(results), "results": results} + + +@app.get("/schedule") +def schedule_info() -> dict: + """Zeigt, ob der automatische Zeitplan aktiv ist und wann er als nächstes läuft.""" + if not settings.sync_schedule_enabled or not scheduler.running: + return {"enabled": False, "cron": settings.sync_schedule_cron} + job = scheduler.get_job("phpipam-outline-sync") + return { + "enabled": True, + "cron": settings.sync_schedule_cron, + "next_run": job.next_run_time.isoformat() if job and job.next_run_time else None, + } diff --git a/app/outline_client.py b/app/outline_client.py new file mode 100644 index 0000000..875f634 --- /dev/null +++ b/app/outline_client.py @@ -0,0 +1,118 @@ +"""Kleiner Client für die Outline API. + +Doku: https://www.getoutline.com/developers +""" +from __future__ import annotations + +import httpx + +from .config import settings + + +class OutlineClient: + def __init__(self) -> None: + self._base = f"{settings.outline_base_url.rstrip('/')}/api" + self._client = httpx.Client( + verify=settings.outline_verify_ssl, + timeout=30.0, + headers={ + "Authorization": f"Bearer {settings.outline_api_token}", + "Content-Type": "application/json", + }, + ) + + def close(self) -> None: + self._client.close() + + def __enter__(self) -> "OutlineClient": + return self + + def __exit__(self, *exc) -> None: + self.close() + + def _post(self, path: str, json: dict) -> dict: + resp = self._client.post(f"{self._base}{path}", json=json) + resp.raise_for_status() + data = resp.json() + if "data" not in data: + raise RuntimeError(f"Unerwartete Outline-Antwort von {path}: {data}") + return data + + def resolve_collection_id(self, collection_id_or_slug: str) -> str: + """Löst eine Collection-ID oder Friendly-URL-Slug in die echte UUID auf. + + `documents.search`/`documents.create` akzeptieren für `collectionId` + nur eine vollständige UUID. Die ID in der Browser-Adresszeile ist bei + neueren Outline-Versionen aber die gekürzte Friendly-URL-ID, keine + UUID. `collections.info` akzeptiert beides, daher lösen wir hier + einmal auf. + """ + data = self._post("/collections.info", {"id": collection_id_or_slug}) + return data["data"]["id"] + + def resolve_document_id(self, document_id_or_slug: str) -> str: + """Löst eine Dokument-ID oder Friendly-URL-Slug in die echte UUID auf.""" + data = self._post("/documents.info", {"id": document_id_or_slug}) + return data["data"]["id"] + + def find_document_by_title( + self, title: str, collection_id: str, parent_document_id: str | None = None + ) -> dict | None: + """Sucht ein bestehendes Dokument mit exakt diesem Titel in der Collection. + + Ist parent_document_id gesetzt, wird zusätzlich geprüft, dass der Treffer + auch wirklich unter dieser Elternseite hängt (z.B. "Netze"), damit gleich + benannte Dokumente an anderer Stelle in der Collection nicht versehentlich + überschrieben werden. + """ + data = self._post( + "/documents.search", + {"query": title, "collectionId": collection_id, "limit": 25}, + ) + for hit in data["data"]: + doc = hit.get("document", hit) + if doc.get("title") != title: + continue + if parent_document_id and doc.get("parentDocumentId") != parent_document_id: + continue + return doc + return None + + def create_document( + self, title: str, text: str, collection_id: str, parent_document_id: str | None = None + ) -> dict: + payload = { + "title": title, + "text": text, + "collectionId": collection_id, + "publish": True, + } + if parent_document_id: + payload["parentDocumentId"] = parent_document_id + data = self._post("/documents.create", payload) + return data["data"] + + def update_document(self, document_id: str, title: str, text: str) -> dict: + data = self._post( + "/documents.update", + {"id": document_id, "title": title, "text": text}, + ) + return data["data"] + + def upsert_document( + self, + title: str, + text: str, + collection_id: str, + parent_document_id: str | None = None, + ) -> tuple[dict, bool]: + """Legt ein Dokument an oder aktualisiert es, falls der Titel bereits existiert. + + Returns: (document, created) + """ + existing = self.find_document_by_title(title, collection_id, parent_document_id) + if existing: + doc = self.update_document(existing["id"], title, text) + return doc, False + doc = self.create_document(title, text, collection_id, parent_document_id) + return doc, True diff --git a/app/phpipam_client.py b/app/phpipam_client.py new file mode 100644 index 0000000..77f7799 --- /dev/null +++ b/app/phpipam_client.py @@ -0,0 +1,75 @@ +"""Kleiner Client für die phpIPAM REST API. + +Doku: https://phpipam.net/api/api_documentation/ +""" +from __future__ import annotations + +import httpx + +from .config import settings + + +class PhpIpamClient: + def __init__(self) -> None: + self._base = f"{settings.phpipam_base_url.rstrip('/')}/api/{settings.phpipam_app_id}" + self._token: str | None = settings.phpipam_app_token + self._client = httpx.Client(verify=settings.phpipam_verify_ssl, timeout=30.0) + + def close(self) -> None: + self._client.close() + + def __enter__(self) -> "PhpIpamClient": + return self + + def __exit__(self, *exc) -> None: + self.close() + + # ------------------------------------------------------------------ + # Auth + # ------------------------------------------------------------------ + def _ensure_token(self) -> str: + if self._token: + return self._token + if not (settings.phpipam_username and settings.phpipam_password): + raise RuntimeError( + "Weder PHPIPAM_APP_TOKEN noch PHPIPAM_USERNAME/PHPIPAM_PASSWORD gesetzt." + ) + resp = self._client.post( + f"{self._base}/user/", + auth=(settings.phpipam_username, settings.phpipam_password), + ) + resp.raise_for_status() + data = resp.json() + if not data.get("success"): + raise RuntimeError(f"phpipam Login fehlgeschlagen: {data}") + self._token = data["data"]["token"] + return self._token + + def _headers(self) -> dict[str, str]: + return {"token": self._ensure_token()} + + # ------------------------------------------------------------------ + # API calls + # ------------------------------------------------------------------ + def _get(self, path: str, params: dict | None = None) -> dict: + resp = self._client.get(f"{self._base}{path}", headers=self._headers(), params=params) + if resp.status_code == 404: + # phpipam liefert 404 z.B. wenn ein Subnetz keine Adressen hat + return {"success": False, "data": []} + resp.raise_for_status() + return resp.json() + + def list_subnets(self) -> list[dict]: + """Alle Subnetze (flach, inkl. verschachtelter unter allen Sections).""" + data = self._get("/subnets/") + return data.get("data") or [] + + def list_addresses(self, subnet_id: int | str) -> list[dict]: + """Alle belegten Adressen eines Subnetzes.""" + data = self._get(f"/subnets/{subnet_id}/addresses/") + return data.get("data") or [] + + def get_section_name(self, section_id: int | str) -> str | None: + data = self._get(f"/sections/{section_id}/") + section = data.get("data") + return section.get("name") if section else None diff --git a/app/scheduler.py b/app/scheduler.py new file mode 100644 index 0000000..838161c --- /dev/null +++ b/app/scheduler.py @@ -0,0 +1,54 @@ +"""Hintergrund-Scheduler: führt den Sync automatisch nach Zeitplan aus.""" +from __future__ import annotations + +import logging + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.cron import CronTrigger + +from .config import settings +from .sync import run_sync + +logger = logging.getLogger("phpipam_outline_sync.scheduler") + +scheduler = AsyncIOScheduler() +JOB_ID = "phpipam-outline-sync" + + +def _run_scheduled_sync() -> None: + """Wird vom Scheduler aufgerufen (läuft in einem Thread, blockiert den + Event-Loop also nicht).""" + try: + results = run_sync() + created = sum(1 for r in results if r.get("action") == "created") + updated = sum(1 for r in results if r.get("action") == "updated") + logger.info( + "Geplanter Sync abgeschlossen: %d Subnetze (%d neu, %d aktualisiert)", + len(results), + created, + updated, + ) + except Exception: # noqa: BLE001 + logger.exception("Geplanter Sync fehlgeschlagen") + + +def start_scheduler() -> None: + if not settings.sync_schedule_enabled: + logger.info("Scheduler deaktiviert (SYNC_SCHEDULE_ENABLED=false)") + return + trigger = CronTrigger.from_crontab(settings.sync_schedule_cron) + scheduler.add_job( + _run_scheduled_sync, + trigger=trigger, + id=JOB_ID, + replace_existing=True, + max_instances=1, + coalesce=True, + ) + scheduler.start() + logger.info("Scheduler gestartet, Zeitplan: '%s'", settings.sync_schedule_cron) + + +def stop_scheduler() -> None: + if scheduler.running: + scheduler.shutdown(wait=False) diff --git a/app/sync.py b/app/sync.py new file mode 100644 index 0000000..dd14731 --- /dev/null +++ b/app/sync.py @@ -0,0 +1,108 @@ +"""Sync-Logik: liest Subnetze + Adressen aus phpipam und schreibt sie als +ein Outline-Dokument pro Subnetz.""" +from __future__ import annotations + +from .config import settings +from .outline_client import OutlineClient +from .phpipam_client import PhpIpamClient + + +def _subnet_title(subnet: dict) -> str: + cidr = f"{subnet.get('subnet')}/{subnet.get('mask')}" + desc = subnet.get("description") or "" + return f"{cidr} – {desc}".rstrip(" –") if desc else cidr + + +def _subnet_markdown(subnet: dict, addresses: list[dict], section_name: str | None) -> str: + cidr = f"{subnet.get('subnet')}/{subnet.get('mask')}" + lines: list[str] = [] + lines.append(f"# {cidr}") + lines.append("") + lines.append(f"- **Beschreibung:** {subnet.get('description') or '–'}") + if section_name: + lines.append(f"- **Section:** {section_name}") + if subnet.get("vlanId") and subnet.get("vlanId") != "0": + lines.append(f"- **VLAN-ID:** {subnet.get('vlanId')}") + if subnet.get("location"): + lines.append(f"- **Standort:** {subnet.get('location')}") + lines.append("") + + if not addresses: + lines.append("_Keine belegten Adressen in phpipam hinterlegt._") + return "\n".join(lines) + + lines.append("| IP | Hostname | Beschreibung | MAC | Owner |") + lines.append("|---|---|---|---|---|") + for addr in sorted(addresses, key=lambda a: tuple(int(o) for o in a["ip"].split("."))): + lines.append( + "| {ip} | {hostname} | {desc} | {mac} | {owner} |".format( + ip=addr.get("ip", ""), + hostname=addr.get("hostname") or "", + desc=(addr.get("description") or "").replace("\n", " "), + mac=addr.get("mac") or "", + owner=addr.get("owner") or "", + ) + ) + return "\n".join(lines) + + +def run_sync(dry_run: bool | None = None) -> list[dict]: + """Führt den Sync durch. Gibt eine Liste von Ergebnis-Infos je Subnetz zurück.""" + dry_run = settings.sync_dry_run if dry_run is None else dry_run + results: list[dict] = [] + section_cache: dict[str, str | None] = {} + + with PhpIpamClient() as ipam: + subnets = ipam.list_subnets() + + outline = None if dry_run else OutlineClient() + collection_id = settings.outline_collection_id + parent_document_id = settings.outline_parent_document_id + try: + if outline is not None: + collection_id = outline.resolve_collection_id(settings.outline_collection_id) + if parent_document_id: + parent_document_id = outline.resolve_document_id(parent_document_id) + + for subnet in subnets: + section_id = subnet.get("sectionId") + if section_id not in section_cache: + section_cache[section_id] = ipam.get_section_name(section_id) if section_id else None + section_name = section_cache.get(section_id) + + addresses = ipam.list_addresses(subnet["id"]) + title = _subnet_title(subnet) + text = _subnet_markdown(subnet, addresses, section_name) + + if dry_run: + results.append( + { + "subnet_id": subnet["id"], + "title": title, + "address_count": len(addresses), + "action": "dry-run", + } + ) + continue + + doc, created = outline.upsert_document( # type: ignore[union-attr] + title, + text, + collection_id, + parent_document_id, + ) + results.append( + { + "subnet_id": subnet["id"], + "title": title, + "address_count": len(addresses), + "action": "created" if created else "updated", + "outline_document_id": doc.get("id"), + "outline_url": doc.get("url"), + } + ) + finally: + if outline is not None: + outline.close() + + return results