Add servicenow-cmdb-export CLI

servicenow-cmdb-export/.env.example

@@ -0,0 +1,3 @@
+SNOW_INSTANCE=acme.service-now.com
+SNOW_USER=cmdb_reader
+SNOW_PASSWORD=changeme

servicenow-cmdb-export/.gitignore

@@ -0,0 +1,9 @@
+__pycache__/
+*.py[cod]
+*.egg-info/
+.venv/
+venv/
+.env
+*.xlsx
+!sample.xlsx
+.pytest_cache/

servicenow-cmdb-export/README.md

@@ -0,0 +1,55 @@
+# servicenow-cmdb-export
+
+CLI that pulls CMDB reference records from ServiceNow and writes a denormalized
+device-to-business-app-and-environment mapping to an Excel file. Servers and
+client devices in one sheet, joined to apps via `cmdb_rel_ci` (with an
+assigned-user fallback for clients).
+
+## Install
+
+```bash
+pip install -e .
+```
+
+## Configure
+
+Set environment variables (see `.env.example`):
+
+- `SNOW_INSTANCE` — e.g. `acme.service-now.com`
+- `SNOW_USER` / `SNOW_PASSWORD` — basic-auth credentials with read on
+  `cmdb_ci_server`, `cmdb_ci_pc_hardware`, `cmdb_ci_business_app`,
+  `cmdb_ci_environment`, `cmdb_rel_ci`, `sys_user`.
+
+## Run
+
+```bash
+snow-cmdb-export server-app-map --out cmdb.xlsx
+snow-cmdb-export server-app-map --out servers-only.xlsx --device-type server
+snow-cmdb-export server-app-map --out filtered.xlsx \
+    --query "operational_status=1^company.name=Acme"
+snow-cmdb-export server-app-map --out audit.xlsx --raw
+```
+
+## Output schema (`device_app_map` sheet)
+
+| column | source |
+| --- | --- |
+| device_type | `server` or `client` |
+| name | `cmdb_ci_*.name` |
+| fqdn | `cmdb_ci_*.fqdn` (falls back to `dns_domain`) |
+| environment | `cmdb_ci_environment.name` joined via `cmdb_rel_ci` |
+| assigned_user | `sys_user.name` resolved from `assigned_to` |
+| assigned_user_email | `sys_user.email` |
+| business_app | `cmdb_ci_business_app.name` |
+| app_owner | `business_owner` (fallback `managed_by`) |
+| criticality | `business_criticality` |
+| support_group | `support_group` on the CI |
+
+## Joining logic
+
+- **Servers → apps**: relationship rows on `cmdb_rel_ci` whose `type` display
+  value is one of `Runs on::Runs`, `Depends on::Used by`, `Hosted on::Hosts`.
+- **Clients → apps**: same relationship lookup first; if none, fall back to
+  business apps where the client's assigned user is `business_owner`,
+  `managed_by`, or `owned_by`. Capped by `--max-apps-per-user` (default 10).
+- **Environment**: any `cmdb_rel_ci` edge from the CI to a `cmdb_ci_environment`.

servicenow-cmdb-export/pyproject.toml

@@ -0,0 +1,20 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "servicenow-cmdb-export"
+version = "0.1.0"
+description = "Export CMDB server/client to business-app and environment mappings from ServiceNow into Excel."
+requires-python = ">=3.10"
+dependencies = [
+    "requests>=2.31",
+    "openpyxl>=3.1",
+    "click>=8.1",
+]
+
+[project.scripts]
+snow-cmdb-export = "servicenow_cmdb_export.cli:main"
+
+[tool.setuptools.packages.find]
+include = ["servicenow_cmdb_export*"]

servicenow-cmdb-export/requirements.txt

@@ -0,0 +1,3 @@
+requests>=2.31
+openpyxl>=3.1
+click>=8.1

servicenow-cmdb-export/servicenow_cmdb_export/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.1.0"

servicenow-cmdb-export/servicenow_cmdb_export/__main__.py

@@ -0,0 +1,4 @@
+from servicenow_cmdb_export.cli import main
+
+if __name__ == "__main__":
+    main()

servicenow-cmdb-export/servicenow_cmdb_export/cli.py

@@ -0,0 +1,74 @@
+"""CLI: snow-cmdb-export."""
+from __future__ import annotations
+
+import sys
+
+import click
+
+from servicenow_cmdb_export.client import ServiceNowClient, ServiceNowError
+from servicenow_cmdb_export.exporter import write_workbook
+from servicenow_cmdb_export.queries import build_rows, fetch_snapshot
+
+
+@click.group()
+@click.version_option()
+def main() -> None:
+    """Export CMDB device-to-app-and-environment mappings from ServiceNow."""
+
+
+@main.command("server-app-map")
+@click.option(
+    "--out",
+    "-o",
+    required=True,
+    type=click.Path(dir_okay=False, writable=True),
+    help="Output .xlsx path.",
+)
+@click.option(
+    "--device-type",
+    type=click.Choice(["all", "server", "client"]),
+    default="all",
+    show_default=True,
+)
+@click.option(
+    "--query",
+    default=None,
+    help="Optional ServiceNow encoded query applied to device tables (sysparm_query).",
+)
+@click.option(
+    "--max-apps-per-user",
+    type=int,
+    default=10,
+    show_default=True,
+    help="Cap apps per client (via assigned-user fallback). 0 = unlimited.",
+)
+@click.option(
+    "--raw/--no-raw",
+    default=False,
+    help="Also dump raw CMDB tables as additional sheets.",
+)
+def server_app_map(
+    out: str, device_type: str, query: str | None, max_apps_per_user: int, raw: bool
+) -> None:
+    """Pull CMDB and write a denormalized device/app/env mapping to Excel."""
+    try:
+        client = ServiceNowClient()
+    except ServiceNowError as e:
+        click.echo(f"error: {e}", err=True)
+        sys.exit(2)
+
+    click.echo("Fetching CMDB snapshot from ServiceNow...", err=True)
+    snap = fetch_snapshot(client, device_filter=device_type, extra_query=query)
+    click.echo(
+        f"  servers={len(snap.servers)} clients={len(snap.clients)} "
+        f"apps={len(snap.apps_by_id)} envs={len(snap.envs_by_id)}",
+        err=True,
+    )
+
+    rows = build_rows(snap, max_apps_per_user=max_apps_per_user)
+    n = write_workbook(out, rows, snapshot=snap, include_raw=raw)
+    click.echo(f"Wrote {n} rows to {out}", err=True)
+
+
+if __name__ == "__main__":
+    main()

servicenow-cmdb-export/servicenow_cmdb_export/client.py

@@ -0,0 +1,94 @@
+"""Thin ServiceNow Table API client.
+
+Auth: HTTP Basic via env vars (SNOW_USER / SNOW_PASSWORD). Instance from
+SNOW_INSTANCE (e.g. "acme.service-now.com" or full https URL).
+"""
+from __future__ import annotations
+
+import os
+from typing import Iterable, Iterator
+from urllib.parse import urljoin
+
+import requests
+
+
+class ServiceNowError(RuntimeError):
+    pass
+
+
+class ServiceNowClient:
+    DEFAULT_PAGE_SIZE = 1000
+
+    def __init__(
+        self,
+        instance: str | None = None,
+        user: str | None = None,
+        password: str | None = None,
+        timeout: int = 60,
+    ) -> None:
+        instance = instance or os.environ.get("SNOW_INSTANCE")
+        user = user or os.environ.get("SNOW_USER")
+        password = password or os.environ.get("SNOW_PASSWORD")
+        if not instance:
+            raise ServiceNowError("SNOW_INSTANCE not set")
+        if not user or not password:
+            raise ServiceNowError("SNOW_USER and SNOW_PASSWORD must be set")
+
+        if not instance.startswith("http"):
+            instance = f"https://{instance}"
+        self.base_url = instance.rstrip("/") + "/"
+        self.timeout = timeout
+        self._session = requests.Session()
+        self._session.auth = (user, password)
+        self._session.headers.update(
+            {"Accept": "application/json", "Content-Type": "application/json"}
+        )
+
+    def table(
+        self,
+        name: str,
+        query: str | None = None,
+        fields: Iterable[str] | None = None,
+        page_size: int = DEFAULT_PAGE_SIZE,
+    ) -> Iterator[dict]:
+        """Yield rows from a ServiceNow table, paging through results."""
+        url = urljoin(self.base_url, f"api/now/table/{name}")
+        params: dict[str, str] = {
+            "sysparm_limit": str(page_size),
+            "sysparm_display_value": "all",
+            "sysparm_exclude_reference_link": "true",
+        }
+        if query:
+            params["sysparm_query"] = query
+        if fields:
+            params["sysparm_fields"] = ",".join(fields)
+
+        offset = 0
+        while True:
+            params["sysparm_offset"] = str(offset)
+            resp = self._session.get(url, params=params, timeout=self.timeout)
+            if resp.status_code >= 400:
+                raise ServiceNowError(
+                    f"GET {name} failed: {resp.status_code} {resp.text[:300]}"
+                )
+            batch = resp.json().get("result", [])
+            if not batch:
+                return
+            for row in batch:
+                yield row
+            if len(batch) < page_size:
+                return
+            offset += page_size
+
+    @staticmethod
+    def display(field) -> str:
+        """Pull display_value from a ServiceNow field that may be a dict or scalar."""
+        if isinstance(field, dict):
+            return field.get("display_value") or field.get("value") or ""
+        return field or ""
+
+    @staticmethod
+    def value(field) -> str:
+        if isinstance(field, dict):
+            return field.get("value") or ""
+        return field or ""

servicenow-cmdb-export/servicenow_cmdb_export/exporter.py

@@ -0,0 +1,79 @@
+"""Excel writer for CMDB export rows."""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Iterable
+
+from openpyxl import Workbook
+from openpyxl.styles import Font
+from openpyxl.utils import get_column_letter
+
+from servicenow_cmdb_export.queries import CMDBSnapshot, Row
+
+
+def write_workbook(
+    path: str | Path,
+    rows: Iterable[Row],
+    snapshot: CMDBSnapshot | None = None,
+    include_raw: bool = False,
+) -> int:
+    wb = Workbook()
+    ws = wb.active
+    ws.title = "device_app_map"
+
+    headers = Row.headers()
+    ws.append(headers)
+    for cell in ws[1]:
+        cell.font = Font(bold=True)
+
+    count = 0
+    for row in rows:
+        ws.append(row.as_list())
+        count += 1
+
+    _autosize(ws, len(headers))
+    ws.freeze_panes = "A2"
+
+    if include_raw and snapshot is not None:
+        _add_raw_sheet(wb, "raw_servers", snapshot.servers)
+        _add_raw_sheet(wb, "raw_clients", snapshot.clients)
+        _add_raw_sheet(wb, "raw_apps", list(snapshot.apps_by_id.values()))
+        _add_raw_sheet(wb, "raw_envs", list(snapshot.envs_by_id.values()))
+
+    wb.save(path)
+    return count
+
+
+def _autosize(ws, col_count: int, max_width: int = 50) -> None:
+    for idx in range(1, col_count + 1):
+        letter = get_column_letter(idx)
+        longest = 0
+        for cell in ws[letter]:
+            v = cell.value
+            if v is None:
+                continue
+            longest = max(longest, len(str(v)))
+        ws.column_dimensions[letter].width = min(longest + 2, max_width)
+
+
+def _add_raw_sheet(wb: Workbook, title: str, records: list[dict]) -> None:
+    ws = wb.create_sheet(title=title)
+    if not records:
+        ws.append(["(no records)"])
+        return
+    keys = sorted({k for r in records for k in r.keys()})
+    ws.append(keys)
+    for cell in ws[1]:
+        cell.font = Font(bold=True)
+    for r in records:
+        ws.append([_flatten(r.get(k)) for k in keys])
+    _autosize(ws, len(keys))
+    ws.freeze_panes = "A2"
+
+
+def _flatten(field) -> str:
+    if isinstance(field, dict):
+        return field.get("display_value") or field.get("value") or ""
+    if field is None:
+        return ""
+    return str(field)