#!/usr/bin/env python3
"""
provider-bundle-check.py -- focused, fail-closed QA for the Pattern A provider revert (D-060).
Asserts ONLY the post-revert (D-052/D-053 + Pattern A) provider invariants on a
Charmed-OpenStack bundle:
1. exactly 11 API charms bind public -> provider-public; none remain on provider-vip
2. every clustered VIP is a triple: provider-public(10.12.4/22) admin(10.12.8/22)
internal(10.12.12/22), all sharing one last octet in 50-60
3. ovn-chassis bridge-interface-mappings carries ALL FOUR chassis MACs, INCLUDING
openstack0's -- the Pattern A revert re-adds the openstack0 MAC that D-057 trimmed
for the now-dead provider-vip plane.
Structural invariants (absorbed from the retired scripts/review-bundle.py -- DOCFIX-070;
that linter's expectations were pre-D-052 and its NOT CLEAN verdict was pure noise):
4. every relation side names an EXISTING app and carries an explicit :endpoint
5. mysql-innodb-cluster deploys at num_units 3 (D-062 -- single-unit seed never bootstraps)
6. no VIP last-octet is shared between applications
7. keystone ships the D-051/D-064 policy IN the bundle (resources: policyd-override,
use-policyd-override true), and the committed zip content matches
policies/domain-manager-policy.yaml (DOCFIX-071 drift guard; content compare, not
byte compare, so zip mtimes cannot false-fail it)
It is the single deploy-gate: it REPLACED the retired scripts/d057-bundle-check.py (D-060)
and now also scripts/review-bundle.py (DOCFIX-070). FAIL -> exit 1. ASCII-only output.
(repo-lint: allow-stale-tokens -- guard checks name retired tokens by necessity)
"""
import sys, re, ipaddress
try:
import yaml
except ImportError:
sys.stderr.write("ERROR: PyYAML not installed (pip install pyyaml --break-system-packages)\n"); sys.exit(2)
PROVIDER = ipaddress.ip_network("10.12.4.0/22") # public API VIPs + FIPs share this plane (Pattern A)
ADMIN = ipaddress.ip_network("10.12.8.0/22")
INTERNAL = ipaddress.ip_network("10.12.12.0/22")
OCTET_LO, OCTET_HI = 50, 60
EXPECT_PUBLIC_VIP = 11
EXPECT_CHASSIS_MACS = {
"52:54:00:3d:fd:54", # openstack0 -- re-added by the Pattern A revert (D-057 had trimmed it)
"52:54:00:9d:63:77", # openstack1
"52:54:00:89:7f:ce", # openstack2
"52:54:00:99:fc:c2", # openstack3
}
MAC_RE = re.compile(r"[0-9a-f]{2}(?::[0-9a-f]{2}){5}")
def main():
path = sys.argv[1] if len(sys.argv) > 1 else "bundle.yaml"
try:
doc = yaml.safe_load(open(path, encoding="utf-8"))
except Exception as e:
sys.stderr.write("ERROR: cannot parse %s: %s\n" % (path, e)); return 2
apps = (doc or {}).get("applications", {}) or {}
fails, oks = [], []
def pub(s): return ((s or {}).get("bindings", {}) or {}).get("public")
on_vip = sorted(n for n, s in apps.items() if pub(s) == "provider-vip")
on_public = sorted(n for n, s in apps.items() if pub(s) == "provider-public")
if on_vip:
fails.append("public still on provider-vip (must be reverted to provider-public): %s" % ", ".join(on_vip))
if len(on_public) != EXPECT_PUBLIC_VIP:
fails.append("public->provider-public count=%d (expect %d): %s" % (len(on_public), EXPECT_PUBLIC_VIP, ", ".join(on_public)))
else:
oks.append("%d charms bind public->provider-public; none on provider-vip" % len(on_public))
vip_ok = 0
for n, s in apps.items():
vip = ((s or {}).get("options", {}) or {}).get("vip")
if not vip:
continue
parts = str(vip).split()
if len(parts) != 3:
fails.append("%s vip not a triple: %r" % (n, vip)); continue
prov, adm, intr = parts
try:
okp = ipaddress.ip_address(prov) in PROVIDER
oka = ipaddress.ip_address(adm) in ADMIN
oki = ipaddress.ip_address(intr) in INTERNAL
except ValueError as e:
fails.append("%s bad vip ip: %s" % (n, e)); continue
if not okp: fails.append("%s provider leg %s not in %s" % (n, prov, PROVIDER)); continue
if not oka: fails.append("%s admin leg %s not in %s" % (n, adm, ADMIN)); continue
if not oki: fails.append("%s internal leg %s not in %s" % (n, intr, INTERNAL)); continue
octs = {p.split(".")[-1] for p in parts}
if len(octs) != 1:
fails.append("%s vip octets differ: %r" % (n, vip)); continue
o = int(octs.pop())
if not (OCTET_LO <= o <= OCTET_HI):
fails.append("%s vip octet .%d outside %d-%d" % (n, o, OCTET_LO, OCTET_HI)); continue
vip_ok += 1
if vip_ok:
oks.append("%d clustered VIP(s) are provider-public/admin/internal triples, octet 50-60" % vip_ok)
for n, s in apps.items():
if (s or {}).get("charm") != "ovn-chassis":
continue
bim = str(((s or {}).get("options", {}) or {}).get("bridge-interface-mappings", ""))
if not bim:
continue
macs = set(MAC_RE.findall(bim.lower()))
missing = EXPECT_CHASSIS_MACS - macs
if missing:
fails.append("%s missing chassis MAC(s): %s" % (n, ", ".join(sorted(missing))))
else:
oks.append("%s bridge-interface-mappings: all 4 chassis MACs present (incl openstack0)" % n)
# -- 4. relations: existing apps, explicit endpoints (the magnum-shared-db class) --
rels = (doc or {}).get("relations", []) or []
rel_bad = 0
for r in rels:
if not isinstance(r, list) or len(r) != 2:
fails.append("relation not a 2-list: %r" % (r,)); rel_bad += 1; continue
for side in r:
side = str(side)
if ":" not in side:
fails.append("relation side lacks explicit :endpoint: %r" % side); rel_bad += 1; continue
if side.split(":")[0] not in apps:
fails.append("relation references unknown app: %r" % side); rel_bad += 1
if rels and not rel_bad:
oks.append("%d relations well-formed (explicit endpoints, all apps exist)" % len(rels))
# -- 5. D-062: mysql-innodb-cluster at target count 3 --
mi = apps.get("mysql-innodb-cluster") or {}
if mi.get("num_units") != 3:
fails.append("mysql-innodb-cluster num_units=%r (D-062 requires 3: single-unit seed never bootstraps)" % mi.get("num_units"))
else:
oks.append("mysql-innodb-cluster num_units=3 (D-062)")
# -- 6. VIP octet uniqueness --
seen_oct = {}
for n, s in apps.items():
vip = ((s or {}).get("options", {}) or {}).get("vip")
if not vip: continue
for part in str(vip).split():
o = part.rsplit(".", 1)[-1]
if o in seen_oct and seen_oct[o] != n:
fails.append("VIP last octet .%s shared by %s and %s" % (o, seen_oct[o], n))
seen_oct.setdefault(o, n)
# -- 7. DOCFIX-071: keystone policy ships in-bundle, zip content matches source --
import os, zipfile
ks = apps.get("keystone") or {}
res = ((ks.get("resources") or {}).get("policyd-override"))
upo = ((ks.get("options") or {}).get("use-policyd-override"))
if not upo:
fails.append("keystone use-policyd-override is not true (D-051)")
if not res:
fails.append("keystone has no resources: policyd-override (DOCFIX-071: policy unreachable on redeploy)")
else:
base = os.path.dirname(os.path.abspath(path))
zp = os.path.normpath(os.path.join(base, str(res)))
src = os.path.join(base, "policies", "domain-manager-policy.yaml")
if not os.path.isfile(zp):
fails.append("policyd-override zip missing at %s" % zp)
elif not os.path.isfile(src):
fails.append("policy source missing at %s" % src)
else:
try:
with zipfile.ZipFile(zp) as z:
inzip = z.read("domain-manager-policy.yaml")
ondisk = open(src, "rb").read()
if inzip != ondisk:
fails.append("policyd-override zip content DIFFERS from policies/domain-manager-policy.yaml (rebuild + recommit the zip)")
else:
oks.append("keystone policyd-override wired in-bundle; zip content matches source (DOCFIX-071)")
except KeyError:
fails.append("zip lacks top-level domain-manager-policy.yaml (keystone reads the top-level name)")
except Exception as e:
fails.append("cannot read policyd zip: %s" % e)
for o in oks: print(" [ok] %s" % o)
for f in fails: print(" [FAIL] %s" % f)
print("\n%s: Pattern A / D-052-D-053 bundle invariants (%s)" % ("PASS" if not fails else "FAIL", path))
return 1 if fails else 0
if __name__ == "__main__":
sys.exit(main())
