#!/usr/bin/env python3 """ Levo: export your Kong tags. Read-only. Every call is a GET; nothing in Kong is changed. pip install requests python levo_kong_export.py --admin-url https://your-kong-admin:8444 --gateway-id kong-prod If your Kong needs a token, add: --auth-mode admin_token --token YOUR_TOKEN Use `--auth-mode bearer` for Kong Konnect. --gateway-id is a name you choose for this Kong; keep it the same every time you export from it. Writes kong-services.json, which you send back to Levo. Generated from levoai_e7s.gateway by standalone.py -- edit the module, not this file. """ from __future__ import annotations import logging from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Dict, List, Optional from urllib.parse import urlparse import requests logger = logging.getLogger(__name__) FORMAT_VERSION = 1 COLLECTOR_NAME = "levo-gateway-export" COLLECTOR_VERSION = "0.1.0" SOURCE_KIND_KONG = "KONG" AUTH_NONE = "none" AUTH_ADMIN_TOKEN = "admin_token" AUTH_BEARER = "bearer" AUTH_MODES = (AUTH_NONE, AUTH_ADMIN_TOKEN, AUTH_BEARER) # Kong caps page size at 1000. The page bound stops a non-advancing cursor spinning forever. PAGE_SIZE = 1000 MAX_PAGES = 1000 class KongExportError(Exception): """Raised when the gateway cannot be read at all, so no bundle should be written.""" @dataclass class KongClient: """Minimal read-only Kong Admin API client.""" admin_url: str auth_mode: str = AUTH_ADMIN_TOKEN token: Optional[str] = None timeout: float = 20.0 session: requests.Session = field(default_factory=requests.Session) def __post_init__(self) -> None: if self.auth_mode not in AUTH_MODES: raise KongExportError( f"unsupported auth mode {self.auth_mode!r}; expected one of {', '.join(AUTH_MODES)}" ) if self.auth_mode != AUTH_NONE and not (self.token or "").strip(): raise KongExportError(f"a token is required for auth mode {self.auth_mode!r}") @property def base_url(self) -> str: """A bare host resolves to https: the token travels on the first call, so plaintext is opt-in and must be spelled out.""" configured = (self.admin_url or "").strip() if not configured: raise KongExportError("admin URL is required") parsed = urlparse(configured) if parsed.scheme.lower() not in ("http", "https"): configured = f"https://{configured}" return configured.rstrip("/") def _headers(self) -> Dict[str, str]: headers = {"Accept": "application/json"} if self.auth_mode == AUTH_ADMIN_TOKEN: headers["Kong-Admin-Token"] = self.token or "" elif self.auth_mode == AUTH_BEARER: headers["Authorization"] = f"Bearer {self.token or ''}" return headers def get_json(self, path: str) -> Dict[str, Any]: url = path if path.startswith("http") else self.base_url + path try: response = self.session.get(url, headers=self._headers(), timeout=self.timeout) except requests.RequestException as exc: raise KongExportError(f"Kong is unreachable at {self.base_url}: {exc}") from exc if response.status_code in (401, 403): raise KongExportError( f"Kong rejected the credentials (HTTP {response.status_code}) for auth mode " f"{self.auth_mode!r}" ) if not response.ok: raise KongExportError(f"Kong returned HTTP {response.status_code} for {path}") if not (response.text or "").strip(): # An empty body is not an empty gateway; treating it as one would later read as a # deletion of every label. raise KongExportError(f"Kong returned an empty body for {path}") try: parsed = response.json() except ValueError as exc: raise KongExportError(f"Kong returned a non-JSON body for {path}") from exc if not isinstance(parsed, dict): raise KongExportError(f"Kong returned a non-object body for {path}") return parsed def fetch_all(self, collection: str) -> List[Dict[str, Any]]: """Walk Kong's ``next`` cursor. Raises rather than returning a partial collection.""" items: List[Dict[str, Any]] = [] path: Optional[str] = f"/{collection}?size={PAGE_SIZE}" seen: set = set() for _ in range(MAX_PAGES): if path is None: return items if path in seen: logger.warning("Kong pagination cursor did not advance at %s; stopping", path) return items seen.add(path) body = self.get_json(path) data = body.get("data") if not isinstance(data, list): raise KongExportError( f'Kong response for {path} has no "data" array; refusing to read it as empty' ) items.extend(data) nxt = body.get("next") path = nxt if isinstance(nxt, str) and nxt else None raise KongExportError( f"stopped after {MAX_PAGES} pages of /{collection}; the gateway is larger than expected" ) def _string_list(value: Any, drop_blanks: bool) -> List[str]: """Kong sends null rather than [] for unset lists.""" if not isinstance(value, list): return [] out = [] for item in value: text = item if isinstance(item, str) else str(item) if drop_blanks and not text.strip(): continue out.append(text) return out def build_bundle( routes: List[Dict[str, Any]], services: List[Dict[str, Any]], gateway_id: str, gateway_version: str = "", complete: bool = True, ) -> Dict[str, Any]: """Assemble the gateway-metadata bundle described in the file contract.""" services_by_id = { svc.get("id"): svc for svc in services if isinstance(svc.get("id"), str) and svc.get("id") } entries = [] for route in routes: service_ref = route.get("service") or {} service_id = service_ref.get("id") if isinstance(service_ref, dict) else None service = services_by_id.get(service_id) entries.append( { "route_id": route.get("id") or "", "route_name": route.get("name") or "", "service_id": service_id or "", "service_name": (service or {}).get("name") or "", # Verbatim, regex forms included: the server decides what is usable. "paths": _string_list(route.get("paths"), drop_blanks=True), # Empty means ALL methods, not none. "methods": _string_list(route.get("methods"), drop_blanks=True), # Blanks kept: a bad tag must not read as "untagged", which means deletion. "route_tags": _string_list(route.get("tags"), drop_blanks=False), "service_tags": _string_list((service or {}).get("tags"), drop_blanks=False), # Reserved: not used for matching today, carried so adding it later does not # invalidate files customers have already produced. "hosts": _string_list(route.get("hosts"), drop_blanks=True), } ) return { "format_version": FORMAT_VERSION, "source": { "kind": SOURCE_KIND_KONG, "collected_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "collector": f"{COLLECTOR_NAME}/{COLLECTOR_VERSION}", "gateway_version": gateway_version, "gateway_id": gateway_id, }, "snapshot": { "complete": complete, "route_count": len(entries), "service_count": len(services), }, "routes": entries, } def export(client: KongClient, gateway_id: str) -> Dict[str, Any]: """Read the gateway and return the bundle. Raises if it could not be read completely.""" if not (gateway_id or "").strip(): raise KongExportError("a gateway id is required; it scopes which labels a sync owns") version = "" try: version = str(client.get_json("/").get("version") or "") except KongExportError: # Not fatal: the root read is informational only. logger.warning("could not read the Kong version; continuing") # Services first: a route references its service by id only, never by name. services = client.fetch_all("services") routes = client.fetch_all("routes") return build_bundle(routes, services, gateway_id=gateway_id, gateway_version=version) import argparse import json import logging import sys from typing import List, Optional logger = logging.getLogger(__name__) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="kong-export", description="Export Kong route and service tags for upload to Levo. Read-only.", ) parser.add_argument( "--admin-url", required=True, help="Kong Admin API URL. A bare host is treated as https; use http:// explicitly for a " "plaintext admin API on a private network.", ) parser.add_argument( "--gateway-id", required=True, help="A stable name for this gateway, e.g. kong-prod-eu. Keep it the same across exports: " "it tells Levo which labels came from which gateway.", ) parser.add_argument("--auth-mode", choices=list(AUTH_MODES), default=AUTH_ADMIN_TOKEN) parser.add_argument( "--token", default=None, help=f"Admin token or bearer token. Not needed for --auth-mode {AUTH_NONE}.", ) parser.add_argument("--out", default="kong-services.json", help="Output file.") parser.add_argument("--timeout", type=float, default=20.0, help="Per-request timeout, seconds.") parser.add_argument("--verbose", action="store_true") return parser def main(argv: Optional[List[str]] = None) -> int: args = build_parser().parse_args(argv) logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(levelname)s %(message)s" ) try: client = KongClient( admin_url=args.admin_url, auth_mode=args.auth_mode, token=args.token, timeout=args.timeout, ) bundle = export(client, gateway_id=args.gateway_id) except KongExportError as exc: # No partial file: an incomplete export uploaded as complete would let Levo treat missing # routes as deletions. logger.error("export failed, nothing written: %s", exc) return 1 with open(args.out, "w", encoding="utf-8") as handle: json.dump(bundle, handle, indent=2) handle.write("\n") snapshot = bundle["snapshot"] logger.info( "wrote %s: %d routes, %d services from %s", args.out, snapshot["route_count"], snapshot["service_count"], bundle["source"]["gateway_id"], ) tagged = sum(1 for r in bundle["routes"] if r["route_tags"] or r["service_tags"]) logger.info("%d of %d routes carry at least one tag", tagged, snapshot["route_count"]) if tagged == 0: logger.warning("no tags found — check that tags are set on your Kong routes or services") return 0 if __name__ == "__main__": sys.exit(main())