#!/usr/bin/env python3 """Pip worker 2: weekly FortiGate config snapshot. Pulls the same 11 endpoints as Dex's weekly review, saves in the exact snapshot format Dex diffs against, and diffs vs the previous snapshot. Runs on the LAN (no Tailscale dependency at pull time). Read-only: all GETs. The API token is read from ~/.fortigate_token and never printed, logged, or written anywhere. """ import json, os, glob, ssl, time, urllib.request, urllib.error from datetime import datetime, timezone DEX = "/home/div/dex" RUNS = DEX + "/runs" LOG = DEX + "/logs/fortigate-snapshot.log" FGT = "https://172.31.0.14/api/v2" TODAY = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d") def norm(o): if isinstance(o, dict): return {k.replace("-", "_"): norm(v) for k, v in o.items()} if isinstance(o, list): return [norm(v) for v in o] return o def api(path, token, timeout=15): req = urllib.request.Request( FGT + path, headers={"Authorization": "Bearer " + token}) ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE try: with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r: d = json.loads(r.read().decode()) res = d.get("results", d) return norm(res), d.get("version") except Exception as e: return {"_error": str(e)[:200]}, None def pick(d, *keys): if isinstance(d, dict): return {k: d.get(k) for k in keys} return d def main(): out = {"ts": datetime.now(timezone.utc).isoformat(), "job": "fortigate-snapshot", "date": TODAY, "ok": True, "errors": []} try: with open(os.path.expanduser("~/.fortigate_token")) as f: token = f.read().strip() except Exception as e: out.update(ok=False, errors=["token_unreadable: " + str(e)[:100]]) finish(out, None) return if not token: out.update(ok=False, errors=["token_empty"]) finish(out, None) return S = {} # sections V = [None] def get(path): res, ver = api(path, token) if ver and not V[0]: V[0] = ver if isinstance(res, dict) and "_error" in res: out["errors"].append(path + ": " + res["_error"]) return res g = get("/cmdb/system/global") g = g[0] if isinstance(g, list) and g else (g if isinstance(g, dict) else {}) S["global"] = pick(g, "hostname", "timezone", "admintimeout") ifs = get("/cmdb/system/interface") ifs = ifs if isinstance(ifs, list) else [] S["interfaces"] = [ {"name": i.get("name"), "ip": (i.get("ip", "") + " " + i.get("netmask", "")).strip(), "allowaccess": " ".join(i.get("allowaccess", [])) if isinstance(i.get("allowaccess"), list) else i.get("allowaccess", "")} for i in ifs] pol = get("/cmdb/firewall/policy") pol = pol if isinstance(pol, list) else [] def ifnames(x): if isinstance(x, list): return [e.get("name", e) if isinstance(e, dict) else e for e in x] return x S["policies"] = {"count": len(pol), "policies": [ {"policyid": p.get("policyid"), "name": p.get("name"), "srcintf": ifnames(p.get("srcintf")), "dstintf": ifnames(p.get("dstintf")), "action": p.get("action"), "status": p.get("status"), "utm": {k: p.get(k) for k in ("utm_status", "ssl_ssh_profile", "av_profile", "webfilter_profile", "dnsfilter_profile", "ips_sensor", "application_list", "ips_voip_filter") if p.get(k) not in (None, "")}} for p in pol]} adm = get("/cmdb/system/admin") S["admins"] = adm if isinstance(adm, list) else [] ntp = get("/cmdb/system/ntp") ntp = ntp[0] if isinstance(ntp, list) and ntp else (ntp if isinstance(ntp, dict) else {}) S["ntp"] = pick(ntp, "ntpsync", "type", "syncinterval") t = get("/monitor/system/time") t = t if isinstance(t, dict) else {} epoch = t.get("time") or t.get("epoch") drift = round(time.time() - epoch, 1) if isinstance(epoch, (int, float)) else None S["time"] = {"epoch": epoch, "drift_seconds": drift} wp = get("/cmdb/wireless-controller/wtp-profile/Home") wp = wp[0] if isinstance(wp, list) and wp else (wp if isinstance(wp, dict) else {}) def chanlist(c): if isinstance(c, list): return [e.get("chan", e) if isinstance(e, dict) else e for e in c] return c def radio(r): r = r or {} return {"band": r.get("band"), "channel": chanlist(r.get("channel")), "power": r.get("power") or r.get("power_level"), "vap1": r.get("vap1"), "vap2": r.get("vap2"), "vap3": r.get("vap3"), "vap4": r.get("vap4")} S["wtp-profile Home"] = {"name": wp.get("name"), "handoff_roaming": wp.get("handoff_roaming"), "r1": radio(wp.get("radio_1")), "r2": radio(wp.get("radio_2"))} wo = get("/cmdb/wireless-controller/wtp/FP231FTF2309FD3E") wo = wo[0] if isinstance(wo, list) and wo else (wo if isinstance(wo, dict) else {}) def oradio(r): r = r or {} return {"ovr_chan": r.get("ovr_chan") or r.get("override_channel"), "chan": chanlist(r.get("chan") or r.get("channel")), "power": r.get("power") or r.get("power_level"), "vap1": r.get("vap1"), "vap2": r.get("vap2"), "vap3": r.get("vap3"), "vap4": r.get("vap4")} S["wtp overrides"] = {"name": wo.get("name"), "admin": wo.get("admin"), "r1": oradio(wo.get("radio_1")), "r2": oradio(wo.get("radio_2"))} ap = get("/monitor/wifi/managed_ap") ap = ap if isinstance(ap, list) else [] ap0 = ap[0] if ap else {} S["ap status"] = pick(ap0, "name", "serial", "status", "admin") lic = get("/monitor/license/status") lic = lic[0] if isinstance(lic, list) and lic else (lic if isinstance(lic, dict) else {}) def svcstr(v): v = v or {} parts = [v.get("status")] if v.get("support_level"): parts.append(v["support_level"]) if v.get("expires"): parts.append("expires:%s" % v["expires"]) return "/".join(str(x) for x in parts if x) fc, fsup = lic.get("forticare") or {}, (lic.get("forticare") or {}).get("support") or {} S["licenses"] = { "forticare": fc.get("status"), "forticloud": (lic.get("forticloud") or {}).get("status"), "fw_updates": (lic.get("firmware_updates") or {}).get("status"), "fortiguard_connected": (lic.get("fortiguard") or {}).get("connected"), "forticare_hardware": svcstr(fsup.get("hardware")), "forticare_enhanced": svcstr(fsup.get("enhanced")), "forticare_comprehensive": svcstr(fsup.get("comprehensive")), "av_db": svcstr(lic.get("antivirus")), } S["version"] = {"version": V[0]} if out["errors"]: out["ok"] = False # write snapshot in Dex's diffable format lines = ["===== pull %s (pip) =====" % TODAY] order = ["global", "interfaces", "policies", "admins", "ntp", "time", "wtp-profile Home", "wtp overrides", "ap status", "licenses", "version"] for sec in order: lines.append("--- %s ---" % sec) lines.append(json.dumps(S.get(sec), separators=(",", ":"))) snap_text = "\n".join(lines) + "\n" snap_path = "%s/fortigate-snapshot-%s.txt" % (RUNS, TODAY) with open(snap_path, "w") as f: f.write(snap_text) with open(RUNS + "/fortigate-snapshot-latest.txt", "w") as f: f.write(snap_text) # keep last 4 snapshots snaps = sorted(glob.glob(RUNS + "/fortigate-snapshot-2*.txt")) for old in snaps[:-4]: try: os.remove(old) except OSError: pass # diff vs previous snapshot: which sections changed changed = [] if len(snaps) >= 2: prev = snaps[-2] def sections(p): d, cur = {}, None with open(p) as f: for line in f: line = line.rstrip("\n") if line.startswith("--- ") and line.endswith(" ---"): cur = line[4:-4]; d[cur] = [] elif cur: d[cur].append(line) return {k: "\n".join(v) for k, v in d.items()} a, b = sections(prev), sections(snap_path) for sec in order: if a.get(sec) != b.get(sec): changed.append(sec) out["changed_sections"] = changed with open("%s/fortigate-diff-%s.json" % (RUNS, TODAY), "w") as f: json.dump({"date": TODAY, "changed_sections": changed, "snapshot": os.path.basename(snap_path)}, f) finish(out, snap_path) def finish(out, snap_path): with open("%s/fortigate-result-%s.json" % (RUNS, out["date"]), "w") as f: json.dump(out, f) with open(LOG, "a") as f: f.write("%s ok=%s errors=%d changed=%s snapshot=%s\n" % ( out["ts"], out["ok"], len(out["errors"]), ",".join(out.get("changed_sections", [])) or "none", os.path.basename(snap_path) if snap_path else "none")) if __name__ == "__main__": main()