#!/usr/bin/env python3 """Compare active Test D3 Catalog sharing rules with mutable TiDB rows. This diagnostic is deliberately read-only. It resolves the exact active D3 Catalog artifacts, verifies every Runtime view by digest, and joins the identity-free serving rules back to live SharingRule rows by their complete semantic tuple. The result makes an incomplete metadata-authority cutover visible without teaching Runtime to read the retired repository authority. """ from __future__ import annotations import argparse import hashlib import json import sys from pathlib import Path from typing import Any REPOSITORY_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPOSITORY_ROOT / "deployments" / "offline" / "runtime")) from common import load_env_file # noqa: E402 sys.path.insert(0, str(REPOSITORY_ROOT / "scripts")) import run_serviceification_d3_test as test_d3 # noqa: E402 d3 = test_d3.impl RULE_FIELDS = ( "objectApiName", "criteria", "accessLevel", "sharedFromRoleId", "sharedFromGroupId", "shareWithRoleId", "shareWithGroupId", ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--runtime-env", type=Path, required=True) parser.add_argument("--object-api-name", default="Lead") parser.add_argument("--output", type=Path, required=True) return parser.parse_args() def canonical_rule(value: dict[str, Any]) -> dict[str, str]: result: dict[str, str] = {} for field in RULE_FIELDS: raw = value.get(field) result[field] = "" if raw is None else str(raw) return result def rule_key(value: dict[str, Any]) -> str: return json.dumps(canonical_rule(value), sort_keys=True, separators=(",", ":")) def verified_json(path: Path, digest: str, label: str) -> dict[str, Any]: if path.is_symlink() or not path.is_file(): raise SystemExit(f"{label} is missing or is not a regular file: {path}") raw = path.read_bytes() actual = "sha256:" + hashlib.sha256(raw).hexdigest() if actual != digest: raise SystemExit(f"{label} digest mismatch: {actual} != {digest}") try: value = json.loads(raw) except json.JSONDecodeError as exc: raise SystemExit(f"{label} is not valid JSON: {exc}") from exc if not isinstance(value, dict): raise SystemExit(f"{label} must be a JSON object") return value def decode_hex(value: str) -> str: if not value: return "" try: return bytes.fromhex(value).decode("utf-8") except (ValueError, UnicodeDecodeError) as exc: raise SystemExit("TiDB returned malformed hex-encoded sharing-rule data") from exc def live_rules(envv: dict[str, str], object_api_name: str) -> list[dict[str, str]]: escaped_object = object_api_name.replace("'", "''") raw = d3.db_query_scalar( envv, f""" SELECT CONCAT_WS('\t', HEX(COALESCE(`Id`, '')), HEX(COALESCE(`Name`, '')), HEX(COALESCE(`ObjectApiName`, '')), HEX(COALESCE(`Criteria`, '')), HEX(COALESCE(`AccessLevel`, '')), HEX(COALESCE(`SharedFromRoleId`, '')), HEX(COALESCE(`SharedFromGroupId`, '')), HEX(COALESCE(`ShareWithRoleId`, '')), HEX(COALESCE(`ShareWithGroupId`, ''))) FROM `SharingRule` WHERE `IsDeleted` = 0 AND LOWER(`ObjectApiName`) = LOWER('{escaped_object}') ORDER BY `Id` """, database=envv["TIDB_DATABASE"], ) result: list[dict[str, str]] = [] if not raw: return result names = ("id", "name", *RULE_FIELDS) for line in raw.splitlines(): columns = line.split("\t") if len(columns) != len(names): raise SystemExit("TiDB returned an unknown SharingRule row shape") result.append(dict(zip(names, (decode_hex(value) for value in columns), strict=True))) return result def main() -> None: args = parse_args() envv = load_env_file(args.runtime_env) target_env = test_d3.test_slot(envv) if target_env not in {"test2", "test3"}: raise SystemExit("sharing-rule diagnostic supports only Test2 and Test3") test_d3.bind_slot(target_env) test_d3._apply_impl_identity(target_env) active_path = Path(envv["ENV_ROOT"]) / ".runtime" / "serviceification-d3" / "active.json" state = d3.read_state(active_path) artifacts = state.get("catalogArtifacts") if not isinstance(artifacts, dict): raise SystemExit("active D3 state has no Catalog artifacts") output_path = Path(str(artifacts.get("output", ""))) store_root = Path(str(artifacts.get("store", ""))) if output_path.is_symlink() or not output_path.is_file(): raise SystemExit("active Catalog output is missing or not a regular file") if store_root.is_symlink() or not store_root.is_dir(): raise SystemExit("active Catalog store is missing or not a regular directory") output = json.loads(output_path.read_bytes()) capabilities = output.get("capabilities") if not isinstance(capabilities, list) or not capabilities: capabilities = [{ "appCapabilityId": output.get("appCapabilityId"), "runtimeCapabilityId": output.get("runtimeCapabilityId"), "runtimeMetadataViewDigest": output.get("runtimeMetadataViewDigest"), }] object_folded = args.object_api_name.casefold() catalog_rules: dict[str, dict[str, str]] = {} view_evidence: list[dict[str, Any]] = [] seen_digests: set[str] = set() for capability in capabilities: if not isinstance(capability, dict): raise SystemExit("Catalog output contains a malformed capability") digest = str(capability.get("runtimeMetadataViewDigest", "")) if not digest.startswith("sha256:") or len(digest) != 71: raise SystemExit("Catalog output contains a malformed Runtime view digest") if digest in seen_digests: continue seen_digests.add(digest) view_path = store_root / "runtime-metadata-view-v2" / f"{digest[7:]}.json" view = verified_json(view_path, digest, "Runtime metadata view") rules = view.get("sharingRules", []) if not isinstance(rules, list): raise SystemExit("Runtime metadata view sharingRules must be an array") selected = [] for rule in rules: if not isinstance(rule, dict): raise SystemExit("Runtime metadata view contains a malformed sharing rule") if str(rule.get("objectApiName", "")).casefold() != object_folded: continue normalized = canonical_rule(rule) catalog_rules[rule_key(normalized)] = normalized selected.append(normalized) view_evidence.append({ "runtimeCapabilityId": capability.get("runtimeCapabilityId"), "profile": view.get("profile"), "digest": digest, "objectPresent": any( isinstance(value, dict) and str(value.get("apiName", "")).casefold() == object_folded for value in view.get("objects", []) ), "sharingRuleCount": len(selected), }) rows = live_rules(envv, args.object_api_name) rows_by_key: dict[str, list[dict[str, str]]] = {} for row in rows: rows_by_key.setdefault(rule_key(row), []).append(row) joined = [] for key in sorted(catalog_rules): joined.append({ **catalog_rules[key], "liveRows": [ {"id": row["id"], "name": row["name"]} for row in rows_by_key.get(key, []) ], }) live_only = [ row for row in rows if rule_key(row) not in catalog_rules ] catalog_only = [ rule for rule in joined if not rule["liveRows"] ] report = { "schema": f"nexuscrm.{target_env}-sharing-rule-authority-diagnostic/v1", "targetEnv": target_env, "runId": state.get("runId"), "phase": state.get("phase"), "catalogRevision": output.get("catalogRevision"), "securityEpoch": output.get("securityEpoch"), "objectApiName": args.object_api_name, "runtimeViews": view_evidence, "catalogRules": joined, "liveOnlyRows": live_only, "catalogOnlyRules": catalog_only, "counts": { "catalogRules": len(catalog_rules), "liveRows": len(rows), "liveOnlyRows": len(live_only), "catalogOnlyRules": len(catalog_only), }, } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text( json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) if __name__ == "__main__": main()