MobilityOps/n8n/workflows/check_drift.py

#!/usr/bin/env python3
"""Report drift between the repo's cleaned workflow definitions and the live n8n instance.

Read-only: this script only issues GET requests against n8n's Public API. It never writes,
imports, activates, or otherwise modifies anything in n8n -- fixing drift is a deliberate,
reviewed action a human takes in the n8n UI (or via a separate, explicit import step), not
something this script does automatically.

Usage:
    N8N_BASE_URL=https://n8n.itworx.tech N8N_API_KEY=... python n8n/workflows/check_drift.py

N8N_API_KEY must be an n8n Public API key (n8n UI -> Settings -> API), not a session cookie
and not a workflow credential. It is read from the environment only and is never printed.

Exit code is 0 when every checked workflow matches the live instance, 1 when any drift (or
a fetch error) is found, so this is safe to wire into CI as a non-blocking check.
"""
from __future__ import annotations

import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any

WORKFLOWS_DIR = Path(__file__).parent

# (repo file name, live workflow ID) -- kept in sync with MANIFEST.md by hand, since the
# manifest is the human-readable source of truth and this is just its machine-checkable echo.
KNOWN_WORKFLOWS = [
    ("fleet-ops-vehicle-return.json", "mobilityops-return-processing"),
    ("fleet-ops-data-quality-scan.json", "mobilityops-scheduled-quality-scan"),
    ("fleet-ops-error-handler.json", "Xppn2rAEqUuyiCJF"),
]

# Fields that legitimately differ between a committed definition and the live instance
# (instance-assigned identifiers, timestamps, UI-only cosmetics) and must not be reported
# as drift.
_VOLATILE_TOP_LEVEL_KEYS = {
    "versionId",
    "createdAt",
    "updatedAt",
    "pinData",
    "staticData",
    "shared",
    "triggerCount",
    "isArchived",
}
_VOLATILE_NODE_KEYS = {"position", "webhookId"}


def _strip_credential_ids(value: Any) -> Any:
    """Live workflows carry instance-specific credential IDs alongside the credential name
    (e.g. {"id": "3", "name": "Fleet Ops Service Token"}). The repo definitions intentionally
    omit the id, since it's meaningless outside the instance that issued it. Drop it from both
    sides so credential *references* are compared by name only."""
    if isinstance(value, dict):
        if set(value.keys()) <= {"id", "name"} and "name" in value:
            return {"name": value["name"]}
        return {k: _strip_credential_ids(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_strip_credential_ids(v) for v in value]
    return value


def _normalize_node(node: dict[str, Any]) -> dict[str, Any]:
    cleaned = {k: v for k, v in node.items() if k not in _VOLATILE_NODE_KEYS}
    return _strip_credential_ids(cleaned)


def _normalize_workflow(doc: dict[str, Any]) -> dict[str, Any]:
    nodes_by_name = {n["name"]: _normalize_node(n) for n in doc.get("nodes", [])}
    return {
        "name": doc.get("name"),
        "active": doc.get("active"),
        "nodes": nodes_by_name,
        "connections": doc.get("connections", {}),
        "settings": doc.get("settings", {}),
    }


def _diff(path: str, local: Any, live: Any, out: list[str]) -> None:
    if isinstance(local, dict) and isinstance(live, dict):
        for key in sorted(set(local) | set(live)):
            if key in _VOLATILE_TOP_LEVEL_KEYS:
                continue
            if key not in live:
                out.append(f"{path}.{key}: present in repo, missing live")
            elif key not in local:
                out.append(f"{path}.{key}: present live, missing in repo")
            else:
                _diff(f"{path}.{key}", local[key], live[key], out)
    elif local != live:
        out.append(f"{path}: repo={local!r} live={live!r}")


def fetch_live_workflow(base_url: str, api_key: str, workflow_id: str) -> dict[str, Any]:
    url = f"{base_url.rstrip('/')}/api/v1/workflows/{workflow_id}"
    request = urllib.request.Request(url, headers={"X-N8N-API-KEY": api_key, "Accept": "application/json"})
    with urllib.request.urlopen(request, timeout=15) as response:  # noqa: S310 (fixed https base url from env)
        return json.load(response)


def main() -> int:
    base_url = os.environ.get("N8N_BASE_URL")
    api_key = os.environ.get("N8N_API_KEY")
    if not base_url or not api_key:
        print(
            "Set N8N_BASE_URL and N8N_API_KEY (an n8n Public API key) in the environment.",
            file=sys.stderr,
        )
        return 1

    any_drift = False
    for filename, workflow_id in KNOWN_WORKFLOWS:
        repo_path = WORKFLOWS_DIR / filename
        local_doc = json.loads(repo_path.read_text(encoding="utf-8"))

        try:
            live_doc = fetch_live_workflow(base_url, api_key, workflow_id)
        except urllib.error.HTTPError as exc:
            print(f"[{filename}] FAILED to fetch live workflow {workflow_id}: HTTP {exc.code}")
            any_drift = True
            continue
        except urllib.error.URLError as exc:
            print(f"[{filename}] FAILED to fetch live workflow {workflow_id}: {exc.reason}")
            any_drift = True
            continue

        differences: list[str] = []
        _diff(filename, _normalize_workflow(local_doc), _normalize_workflow(live_doc), differences)

        if differences:
            any_drift = True
            print(f"[{filename}] DRIFT from live workflow {workflow_id}:")
            for line in differences:
                print(f"  - {line}")
        else:
            print(f"[{filename}] matches live workflow {workflow_id}")

    return 1 if any_drift else 0


if __name__ == "__main__":
    raise SystemExit(main())