#!/usr/bin/env python3
"""
review-bundle.py -- comprehensive pre-deploy review of the Charmed OpenStack
Caracal 2024.1 IPv4-only bundle (VR0 / DC0 / Omega test cloud).
READ-ONLY. Encodes every lesson learned from the 2026-05-28/29/30 deploy
sessions as a fail-closed check. Superset of audit-bundle-fixes.py.
Severities:
FAIL deploy-blocker or known regression -> exit 1
WARN review item / possible issue -> exit 1 only under --strict
INFO informational summary -> never affects exit
Dependencies: PyYAML only (already used by the existing fix scripts); rest stdlib.
ASCII-only output by design (non-ASCII has caused silent daemon failures here).
Usage:
python3 review-bundle.py [BUNDLE] [--strict] [--quiet]
BUNDLE path to bundle.yaml (default: ./bundle.yaml)
--strict treat WARN as failing for exit code
--quiet suppress PASS/INFO lines (show only WARN/FAIL)
"""
import sys
import argparse
import ipaddress
try:
import yaml
except ImportError:
sys.stderr.write("ERROR: PyYAML not installed (pip install pyyaml --break-system-packages)\n")
sys.exit(2)
# --------------------------------------------------------------------------- #
# Config -- the known-good baseline. Adjust here if the design changes.
# --------------------------------------------------------------------------- #
EXPECTED_APPS = 51
EXPECTED_RELATIONS = 98
PROVIDER_NET = ipaddress.ip_network("10.12.4.0/22")
METAL_NET = ipaddress.ip_network("10.12.8.0/22")
VIP_OCTET_MIN = 224 # MAAS reserved metal VIP range 10.12.8.224-254 (D-020)
VIP_OCTET_MAX = 254
# BUNDLEFIX-001: the 7 per-endpoint binding keys that were phantom and removed.
# Final anchors are {"":metal} and {"":metal, public:provider} -> none of these
# should reappear in any app's effective bindings.
PHANTOM_BINDING_KEYS = {
"admin", "internal", "shared-db", "amqp", "certificates", "cluster", "ha",
}
# D-020 clustered-API charm -> provider VIP last octet (metal mirrors it).
EXPECTED_CLUSTERED = {
"barbican": 224, "cinder": 226, "glance": 228, "keystone": 229,
"magnum": 230, "neutron-api": 231, "nova-cloud-controller": 232,
"octavia": 233, "openstack-dashboard": 234, "placement": 235,
}
# Verified Caracal channel matrix (from prior charmhub verification).
# WARN-only: channels can be intentionally pinned; flag deviation, do not block.
OPENSTACK_CORE_CHANNEL = "2024.1/stable"
OPENSTACK_CORE_CHARMS = {
"keystone", "glance", "cinder", "cinder-ceph", "nova-cloud-controller",
"nova-compute", "neutron-api", "neutron-api-plugin-ovn", "placement",
"octavia", "barbican", "magnum", "magnum-dashboard", "openstack-dashboard",
"ceph-radosgw",
}
CHANNEL_MATRIX = {
"ovn-central": "24.03/stable", "ovn-chassis": "24.03/stable",
"ceph-mon": "squid/stable", "ceph-osd": "squid/stable", "ceph-fs": "squid/stable",
"mysql-innodb-cluster": "8.0/stable", "mysql-router": "8.0/stable",
"rabbitmq-server": "3.9/stable", "vault": "1.8/stable",
}
EXPECTED_BASE = "ubuntu@22.04" # jammy; Caracal-bundle paradigm (not noble)
MAC_RE = None # compiled below
import re
MAC_RE = re.compile(r"([0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}")
# --------------------------------------------------------------------------- #
# Duplicate-key-detecting YAML loader (PyYAML silently keeps the last dup).
# --------------------------------------------------------------------------- #
_DUP_KEYS = []
class DupKeyLoader(yaml.SafeLoader):
def construct_mapping(self, node, deep=False):
seen = set()
for key_node, _ in node.value:
try:
key = self.construct_object(key_node, deep=deep)
except Exception:
continue
if isinstance(key, (str, int, float, bool)) or key is None:
if key in seen:
_DUP_KEYS.append((str(key), key_node.start_mark.line + 1))
seen.add(key)
return super().construct_mapping(node, deep)
# --------------------------------------------------------------------------- #
# Reporter
# --------------------------------------------------------------------------- #
class Reporter:
def __init__(self, quiet=False):
self.quiet = quiet
self.rows = [] # (section, level, code, msg)
self.counts = {"PASS": 0, "WARN": 0, "FAIL": 0, "INFO": 0}
def add(self, section, level, code, msg):
self.rows.append((section, level, code, msg))
self.counts[level] = self.counts.get(level, 0) + 1
def emit(self):
section = None
for sec, level, code, msg in self.rows:
if self.quiet and level in ("PASS", "INFO"):
continue
if sec != section:
print("\n--- %s ---" % sec)
section = sec
print(" [%-4s] %-10s %s" % (level, code, msg))
print("\n==================== SUMMARY ====================")
print(" PASS=%d WARN=%d FAIL=%d INFO=%d"
% (self.counts["PASS"], self.counts["WARN"],
self.counts["FAIL"], self.counts["INFO"]))
# --------------------------------------------------------------------------- #
# Helpers
# --------------------------------------------------------------------------- #
def ep_app(endpoint):
"""'keystone:shared-db' -> 'keystone'. Non-str -> None."""
if not isinstance(endpoint, str):
return None
return endpoint.split(":", 1)[0]
def in_net(addr, net):
try:
return ipaddress.ip_address(addr) in net
except ValueError:
return False
# --------------------------------------------------------------------------- #
# Checks
# --------------------------------------------------------------------------- #
def check_ascii(R, text):
sec = "0. Structure / integrity"
bad = []
for i, line in enumerate(text.splitlines(), 1):
for ch in line:
if ord(ch) > 127:
bad.append((i, repr(ch)))
break
if bad:
for ln, ch in bad[:20]:
R.add(sec, "WARN", "NON-ASCII",
"non-ASCII char %s on line %d (non-ASCII has caused silent daemon failures here)" % (ch, ln))
if len(bad) > 20:
R.add(sec, "WARN", "NON-ASCII", "...and %d more non-ASCII line(s)" % (len(bad) - 20))
else:
R.add(sec, "PASS", "ASCII", "file is pure ASCII")
def check_structure(R, doc):
sec = "0. Structure / integrity"
if not isinstance(doc, dict):
R.add(sec, "FAIL", "STRUCT-00", "top-level YAML is not a mapping")
return None, None
if _DUP_KEYS:
for k, ln in _DUP_KEYS:
R.add(sec, "FAIL", "DUPKEY", "duplicate key '%s' near line %d" % (k, ln))
else:
R.add(sec, "PASS", "DUPKEY", "no duplicate keys")
apps = doc.get("applications")
rels = doc.get("relations")
if not isinstance(apps, dict):
R.add(sec, "FAIL", "STRUCT-APPS", "no 'applications' mapping")
apps = {}
if not isinstance(rels, list):
R.add(sec, "FAIL", "STRUCT-RELS", "no 'relations' list")
rels = []
na, nr = len(apps), len(rels)
R.add(sec, "INFO" if na == EXPECTED_APPS else "WARN", "APP-COUNT",
"applications=%d (baseline %d)" % (na, EXPECTED_APPS))
R.add(sec, "INFO" if nr == EXPECTED_RELATIONS else "WARN", "REL-COUNT",
"relations=%d (baseline %d)" % (nr, EXPECTED_RELATIONS))
return apps, rels
def check_relations(R, apps, rels):
sec = "1. Relation integrity"
bad_shape = miss_colon = dangling = 0
for r in rels:
if not (isinstance(r, list) and len(r) == 2):
R.add(sec, "FAIL", "REL-SHAPE", "relation not a 2-element list: %r" % (r,))
bad_shape += 1
continue
for e in r:
if not isinstance(e, str) or ":" not in e:
R.add(sec, "FAIL", "REL-COLON", "endpoint missing colon: %r in %r" % (e, r))
miss_colon += 1
else:
a = ep_app(e)
if a not in apps:
R.add(sec, "FAIL", "REL-DANGLE",
"endpoint references unknown app '%s' in %r" % (a, r))
dangling += 1
if not (bad_shape or miss_colon or dangling):
R.add(sec, "PASS", "REL-INT",
"all relations well-formed, colon-explicit, both ends resolve to apps")
def check_bindings_phantom(R, apps):
sec = "2. BUNDLEFIX-001 (phantom binding keys)"
hits = 0
for name, spec in apps.items():
b = (spec or {}).get("bindings")
if not isinstance(b, dict):
continue
bad = sorted(set(b.keys()) & PHANTOM_BINDING_KEYS)
if bad:
R.add(sec, "FAIL", "PHANTOM",
"%s has phantom per-endpoint binding key(s): %s" % (name, ", ".join(bad)))
hits += 1
if not hits:
R.add(sec, "PASS", "PHANTOM",
"no app reintroduces a removed phantom binding key (%s)"
% ", ".join(sorted(PHANTOM_BINDING_KEYS)))
def check_vault(R, apps, rels):
sec = "3. BUNDLEFIX-002 (vault de-HA)"
v = apps.get("vault")
if v is None:
R.add(sec, "WARN", "VAULT", "no 'vault' app found")
return
opts = (v or {}).get("options") or {}
if "vip" in opts:
R.add(sec, "FAIL", "VAULT-VIP", "vault has a 'vip' option (must be de-HA'd): %r" % opts["vip"])
else:
R.add(sec, "PASS", "VAULT-VIP", "vault has no vip")
if "os-public-hostname" in opts:
R.add(sec, "WARN", "VAULT-HOST", "vault has os-public-hostname (expected removed)")
if "vault-hacluster" in apps:
R.add(sec, "FAIL", "VAULT-HA", "vault-hacluster application is present (must be removed)")
else:
R.add(sec, "PASS", "VAULT-HA", "no vault-hacluster application")
for r in rels:
if isinstance(r, list) and any(isinstance(e, str) and e.startswith("vault:ha") for e in r):
R.add(sec, "FAIL", "VAULT-HAREL", "vault:ha relation present: %r" % (r,))
def map_hacluster(apps, rels):
"""principal -> hacluster_app_name, using charm==hacluster + the :ha relation."""
hac_apps = {n for n, s in apps.items() if (s or {}).get("charm") == "hacluster"}
principal_of = {}
for r in rels:
if not (isinstance(r, list) and len(r) == 2):
continue
a0, a1 = ep_app(r[0]), ep_app(r[1])
if a0 in hac_apps and a1 and a1 not in hac_apps:
principal_of[a1] = a0
elif a1 in hac_apps and a0 and a0 not in hac_apps:
principal_of[a0] = a1
return hac_apps, principal_of
def check_hacluster(R, apps, rels):
sec = "4. BUNDLEFIX-003 (hacluster cluster_count)"
hac_apps, principal_of = map_hacluster(apps, rels)
if not hac_apps:
R.add(sec, "WARN", "HAC", "no hacluster apps found")
return principal_of
principal_for_hac = {h: p for p, h in principal_of.items()}
ok = 0
for h in sorted(hac_apps):
opts = (apps[h].get("options") or {})
cc = opts.get("cluster_count")
prin = principal_for_hac.get(h)
nu = (apps.get(prin, {}) or {}).get("num_units") if prin else None
if cc is None:
R.add(sec, "FAIL", "HAC-CC", "%s missing cluster_count" % h)
continue
if not prin:
R.add(sec, "WARN", "HAC-PRIN", "%s has no principal via :ha relation" % h)
if isinstance(nu, int) and cc > nu:
R.add(sec, "FAIL", "HAC-OVER",
"%s cluster_count=%s > principal %s num_units=%s" % (h, cc, prin, nu))
continue
if cc != 1:
R.add(sec, "WARN", "HAC-NE1",
"%s cluster_count=%s (testcloud baseline is 1)" % (h, cc))
else:
ok += 1
if ok:
R.add(sec, "PASS", "HAC", "%d hacluster app(s) cluster_count=1 and <= principal num_units" % ok)
def check_memcached(R, apps, rels):
sec = "5. BUNDLEFIX-004 (memcached)"
if "memcached" not in apps:
R.add(sec, "FAIL", "MEMCACHE-APP", "no 'memcached' application")
else:
R.add(sec, "PASS", "MEMCACHE-APP", "memcached application present")
found = False
for r in rels:
if not (isinstance(r, list) and len(r) == 2):
continue
s = set()
for e in r:
if isinstance(e, str):
s.add(e)
if {"nova-cloud-controller:memcache", "memcached:cache"} <= s:
found = True
R.add(sec, "PASS" if found else "FAIL", "MEMCACHE-REL",
"nova-cloud-controller:memcache <-> memcached:cache relation %s"
% ("present" if found else "MISSING"))
def check_router_bindings(R, apps):
sec = "6. BUNDLEFIX-005 (mysql-router metal binding)"
routers = [n for n, s in apps.items() if (s or {}).get("charm") == "mysql-router"]
if not routers:
R.add(sec, "WARN", "ROUTER", "no mysql-router apps found")
return
bad = 0
for n in sorted(routers):
b = (apps[n].get("bindings") or {})
# effective default space is the "" key; anchors already resolved by yaml
default = b.get("", None)
non_metal = {k: v for k, v in b.items() if v not in ("metal",)}
if default == "metal" and not non_metal:
continue
if default != "metal":
R.add(sec, "FAIL", "ROUTER-BIND",
"%s default space binding is %r (expected metal)" % (n, default))
bad += 1
elif non_metal:
R.add(sec, "WARN", "ROUTER-BIND",
"%s has non-metal endpoint binding(s): %r" % (n, non_metal))
if not bad:
R.add(sec, "PASS", "ROUTER-BIND",
"%d mysql-router app(s) bound to metal" % len(routers))
def check_vips(R, apps, rels):
sec = "7. BUNDLEFIX-006 / D-020 (dual provider+metal VIPs)"
_, principal_of = map_hacluster(apps, rels)
clustered = sorted(principal_of.keys())
# set comparison vs expected D-020 clustered set
got = set(clustered)
exp = set(EXPECTED_CLUSTERED)
if got != exp:
if exp - got:
R.add(sec, "WARN", "VIP-SET", "expected-clustered apps NOT detected as clustered: %s"
% ", ".join(sorted(exp - got)))
if got - exp:
R.add(sec, "WARN", "VIP-SET", "clustered apps beyond the D-020 set: %s"
% ", ".join(sorted(got - exp)))
ok = 0
for name in clustered:
opts = (apps[name].get("options") or {})
vip = opts.get("vip")
if not vip:
R.add(sec, "FAIL", "VIP-MISS", "%s is clustered but has no vip" % name)
continue
parts = str(vip).split()
if len(parts) != 2:
R.add(sec, "FAIL", "VIP-DUAL", "%s vip is not dual (got %r)" % (name, vip))
continue
prov, metal = parts
if not in_net(prov, PROVIDER_NET):
R.add(sec, "FAIL", "VIP-PROV", "%s provider vip %s not in %s" % (name, prov, PROVIDER_NET))
continue
if not in_net(metal, METAL_NET):
R.add(sec, "FAIL", "VIP-METAL", "%s metal vip %s not in %s" % (name, metal, METAL_NET))
continue
po, mo = int(prov.split(".")[-1]), int(metal.split(".")[-1])
if po != mo:
R.add(sec, "FAIL", "VIP-MIRROR", "%s octets differ: provider .%d vs metal .%d" % (name, po, mo))
continue
if not (VIP_OCTET_MIN <= mo <= VIP_OCTET_MAX):
R.add(sec, "FAIL", "VIP-RANGE",
"%s metal vip octet .%d outside reserved %d-%d" % (name, mo, VIP_OCTET_MIN, VIP_OCTET_MAX))
continue
expected_octet = EXPECTED_CLUSTERED.get(name)
if expected_octet is not None and po != expected_octet:
R.add(sec, "WARN", "VIP-OCTET",
"%s vip octet .%d != D-020 map .%d" % (name, po, expected_octet))
ok += 1
if ok:
R.add(sec, "PASS", "VIP-DUAL",
"%d clustered API charm(s) have mirrored dual VIPs in the reserved range" % ok)
def check_osd(R, apps):
sec = "8. Anti-pattern: ceph-osd osd-devices"
osds = [n for n, s in apps.items() if (s or {}).get("charm") == "ceph-osd"]
if not osds:
R.add(sec, "WARN", "OSD", "no ceph-osd app found")
return
for n in osds:
dev = (apps[n].get("options") or {}).get("osd-devices")
if not dev or not isinstance(dev, str) or not dev.strip().startswith("/"):
R.add(sec, "FAIL", "OSD-DEV", "%s osd-devices not a real path: %r" % (n, dev))
else:
note = ""
if "/dev/disk/by-" not in dev:
note = " (kernel-name; by-path/by-id is harder for bare metal -- Roosevelt note)"
R.add(sec, "PASS", "OSD-DEV", "%s osd-devices=%s%s" % (n, dev.strip(), note))
def check_ovn(R, apps):
sec = "9. Anti-pattern: ovn-chassis mappings (MAC over NIC name)"
chassis = [n for n, s in apps.items() if (s or {}).get("charm") == "ovn-chassis"]
if not chassis:
R.add(sec, "WARN", "OVN", "no ovn-chassis app found")
return
for n in sorted(chassis):
opts = (apps[n].get("options") or {})
bim = opts.get("bridge-interface-mappings")
if not bim:
R.add(sec, "INFO", "OVN-BIM", "%s has no bridge-interface-mappings (expected for octavia-side chassis)" % n)
continue
if MAC_RE.search(str(bim)):
R.add(sec, "PASS", "OVN-BIM", "%s bridge-interface-mappings is MAC-based" % n)
else:
R.add(sec, "WARN", "OVN-BIM",
"%s bridge-interface-mappings has no MAC (NIC-name? fragile): %r" % (n, bim))
def check_os_networks(R, apps, rels):
sec = "10. D-020: spaces-native (no os-*-network pinning)"
_, principal_of = map_hacluster(apps, rels)
flagged = 0
for name in sorted(principal_of):
opts = (apps[name].get("options") or {})
for k in ("os-internal-network", "os-admin-network", "os-public-network"):
if k in opts:
R.add(sec, "WARN", "OS-NET",
"%s sets %s (D-020 found spaces-native resolve sufficient; verify intent)" % (name, k))
flagged += 1
if not flagged:
R.add(sec, "PASS", "OS-NET", "no clustered charm pins os-*-network (spaces-native, per D-020)")
def expected_channel(charm):
if charm in CHANNEL_MATRIX:
return CHANNEL_MATRIX[charm]
if charm in OPENSTACK_CORE_CHARMS:
return OPENSTACK_CORE_CHANNEL
return None
def check_channels_base(R, apps):
sec = "11. Channels / base (verified Caracal matrix; WARN-only)"
mismatch = 0
for name, spec in sorted(apps.items()):
spec = spec or {}
charm = spec.get("charm")
ch = spec.get("channel")
exp = expected_channel(charm)
if exp and ch and ch != exp:
R.add(sec, "WARN", "CHANNEL", "%s (%s) channel=%s expected=%s" % (name, charm, ch, exp))
mismatch += 1
base = spec.get("base")
series = spec.get("series")
if base and base != EXPECTED_BASE:
R.add(sec, "WARN", "BASE", "%s base=%s expected=%s" % (name, base, EXPECTED_BASE))
if series and series not in ("jammy",):
R.add(sec, "WARN", "SERIES", "%s series=%s expected=jammy" % (name, series))
if not mismatch:
R.add(sec, "PASS", "CHANNEL", "no charm deviates from the known Caracal channel matrix")
def summary_tables(R, apps, rels):
sec = "12. Inventory (informational)"
_, principal_of = map_hacluster(apps, rels)
for name in sorted(principal_of):
vip = ((apps[name].get("options") or {}).get("vip"))
R.add(sec, "INFO", "CLUSTERED", "%-26s vip=%s" % (name, vip))
routers = sorted(n for n, s in apps.items() if (s or {}).get("charm") == "mysql-router")
R.add(sec, "INFO", "ROUTERS", "%d mysql-router apps: %s" % (len(routers), ", ".join(routers)))
# --------------------------------------------------------------------------- #
# Main
# --------------------------------------------------------------------------- #
def main():
ap = argparse.ArgumentParser(description="Comprehensive Caracal bundle reviewer (read-only).")
ap.add_argument("bundle", nargs="?", default="bundle.yaml")
ap.add_argument("--strict", action="store_true", help="treat WARN as failing for exit code")
ap.add_argument("--quiet", action="store_true", help="show only WARN/FAIL")
args = ap.parse_args()
try:
with open(args.bundle, "r", encoding="utf-8", errors="replace") as fh:
text = fh.read()
except FileNotFoundError:
sys.stderr.write("ERROR: bundle not found: %s\n" % args.bundle)
return 2
try:
doc = yaml.load(text, Loader=DupKeyLoader)
except yaml.YAMLError as e:
sys.stderr.write("ERROR: YAML parse failed: %s\n" % e)
return 2
R = Reporter(quiet=args.quiet)
print("================ Caracal v1 bundle review: %s ================" % args.bundle)
check_ascii(R, text)
apps, rels = check_structure(R, doc)
if apps is None:
R.emit()
return 1
check_relations(R, apps, rels)
check_bindings_phantom(R, apps)
check_vault(R, apps, rels)
check_hacluster(R, apps, rels)
check_memcached(R, apps, rels)
check_router_bindings(R, apps)
check_vips(R, apps, rels)
check_osd(R, apps)
check_ovn(R, apps)
check_os_networks(R, apps, rels)
check_channels_base(R, apps)
summary_tables(R, apps, rels)
R.emit()
fail = R.counts["FAIL"] > 0
warn = R.counts["WARN"] > 0
if fail or (args.strict and warn):
print("\nVERDICT: NOT CLEAN" + (" (--strict: WARN counts)" if (warn and not fail) else ""))
return 1
print("\nVERDICT: CLEAN" + (" (with WARN review items)" if warn else ""))
return 0
if __name__ == "__main__":
sys.exit(main())
