compute_ccf.py

#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""CCF v1.0-RC metric reference computation and canonical test vectors.

Computes, for each synthetic estate:
  - CCI over exclusive executing units (HHI, 0-10,000 scale)
  - largest unit share s1, band
  - flagged-only CCI over qualifying assets
  - failure-domain reach per upstream node: known share and upper bound
  - diversity ladder: observed brands -> resolved units -> max reach (effective view)
Writes ccf-test-vectors-v1.0rc-draft.md and test-vectors.json.
"""
import json, pathlib

def shares(counts):
    N = sum(counts.values())
    return {k: 100.0 * v / N for k, v in counts.items()}, N

def hhi(counts):
    s, _ = shares(counts)
    return round(sum(x * x for x in s.values()))

def band(cci):
    if cci >= 8100: return "Single-source-equivalent"
    if cci >= 4900: return "Highly concentrated"
    if cci >= 2500: return "Concentrated"
    return "Diversified"

def pct(n, N): return round(100.0 * n / N, 1)

V = []  # (id, title, rule_certified, description, units, brands, flagged, reach, notes)

# TV-01 Monoculture
V.append(dict(id="TV-01", title="Monoculture",
  rule="CCI ceiling; reach degenerate",
  desc="One executing family serves every asset.",
  units={"F1": 1000}, brands=3, flagged={}, reach=[
    dict(node="U (ancestor of F1)", mode="lineage", known=1000, upper=1000)],
  note="Three brands resell one stack: observed 3, resolved 1."))

# TV-02 Perfect diversification
V.append(dict(id="TV-02", title="Perfect diversification",
  rule="CCI floor for five units; reach bounded by unit share when no upstream is shared",
  desc="Five equal, fully independent families, lineage disclosed and distinct.",
  units={f"F{i}": 200 for i in range(1, 6)}, brands=5, flagged={},
  reach=[dict(node=f"U{i}", mode="lineage", known=200, upper=200) for i in range(1, 6)],
  note="Every reach node reaches only its own family."))

# TV-03 Hidden shared ancestry
V.append(dict(id="TV-03", title="Hidden shared ancestry",
  rule="CCI stable while reach exposes the false diversity",
  desc="Four equal families; F1 and F2 share evidenced ancestor U12.",
  units={f"F{i}": 250 for i in range(1, 5)}, brands=4, flagged={},
  reach=[dict(node="U12", mode="lineage", known=500, upper=500),
         dict(node="U3", mode="lineage", known=250, upper=250),
         dict(node="U4", mode="lineage", known=250, upper=250)],
  note="Observed 4, resolved 4, largest failure domain 50%."))

# TV-04 Undisclosed lineage bound (worked-example shape)
V.append(dict(id="TV-04", title="Undisclosed lineage bound",
  rule="Non-disclosure moves the reach bound, never the CCI",
  desc="Families 2,400 / 900 / 900. F1 descends from U (disclosed). F2 independent (disclosed). F3 lineage undisclosed.",
  units={"F1": 2400, "F2": 900, "F3": 900}, brands=3, flagged={},
  reach=[dict(node="U", mode="lineage", known=2400, upper=3300),
         dict(node="U(F2)", mode="lineage", known=900, upper=900)],
  note="Replaces v0.9 Reading A/B: CCI is single-valued; U carries a bounded reach [57.1, 78.6]."))

# TV-05 Merge (multi-parent ancestry)
V.append(dict(id="TV-05", title="Merge: multi-parent ancestry",
  rule="A merge contributes its assets to every parent's reach; CCI unaffected",
  desc="M (400) descends from P1 AND P2 (flat pedigree, merge). X (300) descends from P1. Y (300) independent.",
  units={"M": 400, "X": 300, "Y": 300}, brands=3, flagged={},
  reach=[dict(node="P1", mode="lineage", known=700, upper=700),
         dict(node="P2", mode="lineage", known=400, upper=400),
         dict(node="U(Y)", mode="lineage", known=300, upper=300)],
  note="P1 reaches M and X: 70%."))

# TV-06 Chain depth and evidenced divergence
V.append(dict(id="TV-06", title="Chain depth with evidenced divergence",
  rule="Reach requires relevant code present; evidenced divergence severs the edge",
  desc="C -> B -> F1 (nested chain), F1 500 assets; F2 (500) descends from C directly. F1's relevant primitive rewritten since the fork from C (divergence evidenced by patches); B's code present in F1.",
  units={"F1": 500, "F2": 500}, brands=2, flagged={},
  reach=[dict(node="B", mode="lineage", known=500, upper=500),
         dict(node="C", mode="lineage", known=500, upper=500)],
  note="C reaches F2 only: the F1 edge is severed by evidenced divergence, and B never reached F2. Without the divergence evidence, C's reach would be bounded [500, 1000]."))

# TV-07 Cross-signed anchors (layer 3)
V.append(dict(id="TV-07", title="Cross-signed trust anchors",
  rule="Anchor CCI over primary accepted paths; bridge reach = assets where its compromise invalidates every accepted path",
  desc="Anchors A1 (600) and A2 (400), both cross-signed by bridge P. For 100 of A2's assets, relying parties also hold an independently distributed A2 root path not via P.",
  units={"A1": 600, "A2": 400}, brands=2, flagged={},
  reach=[dict(node="P (bridge)", mode="path-invalidation", known=900, upper=900),
         dict(node="A1", mode="key-compromise", known=600, upper=600),
         dict(node="A2", mode="key-compromise", known=400, upper=400)],
  note="P reaches 90%: every asset except the 100 holding a surviving independent path."))

# TV-08 Policy root vs cryptographic root
V.append(dict(id="TV-08", title="Policy root vs cryptographic root",
  rule="Reach is failure-mode-specific: distinct under key compromise, one domain under compulsion",
  desc="Anchors A1 (500) and A2 (500), cryptographically independent, both governed by policy authority PA (trust-list operator).",
  units={"A1": 500, "A2": 500}, brands=2, flagged={},
  reach=[dict(node="A1", mode="key-compromise", known=500, upper=500),
         dict(node="A2", mode="key-compromise", known=500, upper=500),
         dict(node="PA", mode="compulsion / mis-issuance / trust-list withdrawal", known=1000, upper=1000)],
  note="The reach table carries both modes; the tolerance test selects the mode under assessment."))

# TV-09 Layer 6: certification profile is not design identity
V.append(dict(id="TV-09", title="Key generation: profile-only evidence",
  rule="Certification-profile conformance bounds design reach; it never establishes it",
  desc="Generation points G1 (500), G2 (300), G3 (200). All three conform to one RNG certification profile; no design disclosure.",
  units={"G1": 500, "G2": 300, "G3": 200}, brands=3, flagged={},
  reach=[dict(node="D? (possible shared design)", mode="generation-design", known=500, upper=1000)],
  note="Known reach = largest single point (a design at least spans its own point); upper = all profile-conformant points. Disclosure collapses the interval either way."))

# TV-10 Flagged material and hybrid counting
V.append(dict(id="TV-10", title="Flagged concentration under hybrid counting",
  rule="Flagged-only CCI over qualifying assets; hybrid composite is one asset, both lineages reach it",
  desc="F1 (700, of which 20 flagged), F2 (200, of which 80 flagged), F3 (100, 0 flagged). F2 assets are hybrid composites: classical component from ancestor UC, post-quantum from UQ.",
  units={"F1": 700, "F2": 200, "F3": 100}, brands=3,
  flagged={"F1": 20, "F2": 80},
  reach=[dict(node="U1", mode="lineage", known=700, upper=700),
         dict(node="UC (classical component)", mode="lineage", known=200, upper=200),
         dict(node="UQ (post-quantum component)", mode="lineage", known=200, upper=200)],
  note="Estate looks diversified; the flagged view is single-source-equivalent in F2. Both hybrid component lineages carry full reach to the composite assets."))

# ---- compute and emit ----
out_md = ["# CCF Canonical Test Vectors – v1.0-RC",
"",
"**Generated by `compute_ccf.py` and never hand-edited.** Canonical fixtures for the CCF reference implementation, on the Universal Framework v1.0-RC baseline (executing-unit CCI, co-equal reach).",
"",
"Scale: CCI 0–10,000. Bands: Diversified < 2,500 ≤ Concentrated < 4,900 ≤ Highly concentrated < 8,100 ≤ Single-source-equivalent, boundaries closed on the left per Universal C.4.",
""]
out_json = []

for v in V:
    s, N = shares(v["units"])
    cci = hhi(v["units"])
    s1name = max(v["units"], key=v["units"].get)
    s1 = max(s.values())
    fl = v["flagged"]
    flcci = hhi(fl) if fl else None
    flN = sum(fl.values()) if fl else 0
    ladder_obs = v["brands"]; ladder_res = len(v["units"])
    maxreach = max((pct(r["upper"], N) for r in v["reach"]), default=None)
    out_md += [f"## {v['id']} – {v['title']}", "",
      f"*Certifies:* {v['rule']}.", "", v["desc"], "",
      f"**Population:** {N} assets. **Units:** " + ", ".join(f"{k} {c} ({s[k]:.1f}%)" for k, c in v["units"].items()) + ".", "",
      f"**CCI {cci}** ({band(cci)}); largest unit share {s1:.1f}% ({s1name})."]
    if fl:
        out_md += [f"**Flagged-only CCI {flcci}** ({band(flcci)}) over {flN} qualifying assets: " +
                   ", ".join(f"{k} {c} ({pct(c, flN):.1f}%)" for k, c in fl.items()) + "."]
    out_md += ["", "| Reach node | Failure mode | Known | Upper bound | State |", "|---|---|---|---|---|"]
    for r in v["reach"]:
        state = "evidenced" if r["known"] == r["upper"] else "bounded"
        out_md += [f"| {r['node']} | {r['mode']} | {r['known']} ({pct(r['known'],N)}%) | {r['upper']} ({pct(r['upper'],N)}%) | {state} |"]
    out_md += ["", f"**Diversity ladder:** observed {ladder_obs} → resolved {ladder_res} → largest failure domain {maxreach}% (upper bound).", "",
               f"*{v['note']}*", "", "---", ""]
    out_json.append(dict(id=v["id"], title=v["title"], population=N,
        units=v["units"], cci=cci, band=band(cci), s1=round(s1,1),
        flagged=fl, flagged_cci=flcci,
        reach=[dict(node=r["node"], mode=r["mode"], known=r["known"],
                    known_pct=pct(r["known"],N), upper=r["upper"],
                    upper_pct=pct(r["upper"],N),
                    state="evidenced" if r["known"]==r["upper"] else "bounded")
               for r in v["reach"]],
        ladder=dict(observed=ladder_obs, resolved=ladder_res, max_reach_upper_pct=maxreach)))

pathlib.Path("ccf-test-vectors-v1.0rc.md").write_text("\n".join(out_md))
pathlib.Path("test-vectors.json").write_text(json.dumps(out_json, indent=2))
for j in out_json:
    fl = f" flagged={j['flagged_cci']}" if j['flagged_cci'] else ""
    mr = j['ladder']['max_reach_upper_pct']
    print(f"{j['id']}: CCI={j['cci']} ({j['band']}) s1={j['s1']}%{fl} maxreach≤{mr}%")


# ---------------- FS.6 worked example: correspondent banking ----------------
def fs_data():
    """Correspondent banking at a mid-size universal bank. Synthetic, illustrative."""
    est = dict(N=2600, qv=2450, sym=150, flagged=140, tolerance="four hours")
    L = {}
    L[2] = dict(units={"ICSF mainframe crypto services": 1180, "OpenSSL-derived distributed stack": 640,
                       "JCE provider family": 340, "HSM firmware module family": 240,
                       "cloud provider library": 200},
                fl=None, reach=[("OpenSSL 1.1 lineage", 640, 880), ("provider boundary", 0, 200)],
                tol="Fail, on the bound")
    L[3] = dict(units={"internal enterprise CA": 980, "SWIFT PKI": 760, "central bank RTGS PKI": 320,
                       "public web PKI root": 300, "QTSP root": 240},
                fl={"SWIFT PKI": 110, "QTSP root": 30}, reach=[], tol="Fail, flagged")
    L[4] = dict(units={"mainframe coprocessor family": 1800, "HSM firmware family A": 420,
                       "cloud KMS (module undisclosed)": 280, "enterprise KMS software module": 100},
                fl={"mainframe coprocessor family": 100, "HSM firmware family A": 40},
                reach=[("provider boundary", 0, 280)], tol="Fail")
    L[5] = dict(units={"TLS 1.2 RSA key exchange": 1120, "TLS 1.3 x25519": 900,
                       "IPsec proposal set A": 340, "SSH kex set": 240},
                fl=None, reach=[], tol="Pass")
    L[6] = dict(units={"mainframe generation design": 1860, "HSM family A TRNG design": 400,
                       "cloud provider entropy (undisclosed)": 240, "application DRBG": 100},
                fl=None, reach=[("certification-profile P design", 0, 640), ("provider boundary", 0, 240)],
                tol="Fail, on the design bound")
    out = dict(est=est, layers={})
    for k, v in L.items():
        assert sum(v["units"].values()) == est["N"], k
        sh, _ = shares(v["units"]); s1n = max(v["units"], key=v["units"].get)
        cci = hhi(v["units"])
        fl = hhi(v["fl"]) if v["fl"] else None
        rmax = max(v["reach"], key=lambda r: r[2]) if v["reach"] else None
        out["layers"][k] = dict(units=v["units"], cci=cci, band=band(cci),
            s1=round(sh[s1n], 1), s1name=s1n, flcci=fl, flband=band(fl) if fl else None,
            reach=rmax and dict(node=rmax[0], known_pct=pct(rmax[1], est["N"]), upper_pct=pct(rmax[2], est["N"])),
            allreach=[dict(node=r[0], known_pct=pct(r[1], est["N"]), upper_pct=pct(r[2], est["N"])) for r in v["reach"]],
            tol=v["tol"])
    out["reported"] = max(out["layers"][k]["cci"] for k in out["layers"])
    out["reported_layer"] = max(out["layers"], key=lambda k: out["layers"][k]["cci"])
    out["cp"] = dict(blocking=3, peer=14, contractual=180, population=0)
    return out

def fs_table_md(d):
    rows = ["| Layer | Units (assets) | CCI | Largest share | Band | Flagged-only | Largest reach (bound) | Tolerance |",
            "|---|---|---|---|---|---|---|---|",
            "| 1 Algorithm *(baseline, excluded from the reported figure)* | one class | 10,000 | 100% | Single-source-equivalent | 10,000 | \u2013 | Fail |"]
    names = {2: "2 Lineage", 3: "3 Trust root", 4: "4 Custody", 5: "5 Protocol", 6: "6 Key generation"}
    for k in (2, 3, 4, 5, 6):
        l = d["layers"][k]
        u = "/".join(str(c) for c in l["units"].values())
        flc = f"{l['flcci']:,}" if l["flcci"] else "\u2013"
        r = l["reach"]
        rc = f"{r['known_pct']}% evid \u00b7 {r['upper_pct']}% bound" if r else "\u2013"
        rows.append(f"| {names[k]} | {u} | {l['cci']:,} | {l['s1']}% | {l['band']} | {flc} | {rc} | {l['tol']} |")
    return "\n".join(rows)

import sys as _sys
if '--fs-example' in _sys.argv:
    _d = fs_data()
    for _k in (2, 3, 4, 5, 6):
        _l = _d['layers'][_k]
        print(f"FS L{_k}: CCI={_l['cci']} ({_l['band']}) s1={_l['s1']}% ({_l['s1name']}) fl={_l['flcci']} tol={_l['tol']}")
    print("FS reported", _d['reported'], "at layer", _d['reported_layer'])
    print("FS L2 reach", _d['layers'][2]['allreach'])
    print("FS L6 reach", _d['layers'][6]['allreach'])