From f5e5ab5fe318595ba7fe497baae3da2de14c47bf Mon Sep 17 00:00:00 2001 From: Ning Zhou Date: Wed, 29 Jul 2026 21:45:32 +0800 Subject: [PATCH] feat(platform): gate production object-store readiness --- .github/workflows/ci.yml | 5 + ...tadata-fabric-object-store.production.yaml | 118 +++ .../metadata_fabric_object_store_gate.py | 895 ++++++++++++++++++ .../test_metadata_fabric_object_store_gate.py | 407 ++++++++ ...-production-object-store-readiness-gate.md | 94 ++ docs/roadmap-ar0-platform-truth-2026-07-24.md | 9 +- docs/system-of-record-matrix-2026-07-24.md | 11 +- scripts/metadata-fabric-object-store-gate.sh | 22 + 8 files changed, 1552 insertions(+), 9 deletions(-) create mode 100644 config/metadata-fabric-object-store.production.yaml create mode 100644 data_agent/metadata_fabric_object_store_gate.py create mode 100644 data_agent/test_metadata_fabric_object_store_gate.py create mode 100644 docs/architecture-decisions/adr-057-production-object-store-readiness-gate.md create mode 100755 scripts/metadata-fabric-object-store-gate.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 312f995b..413b13e9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,6 +26,7 @@ on: - feat/ar1-metadata-fabric-jdbc-catalog-restart - feat/ar1-metadata-fabric-spark-iceberg-rest-interoperability - feat/ar1-metadata-fabric-spark-object-store-interoperability + - feat/ar1-metadata-fabric-object-store-readiness-gate env: PYTHON_VERSION: "3.13" @@ -153,6 +154,9 @@ jobs: - name: Validate metadata fabric Spark/object-store interoperability evidence run: python -m data_agent.metadata_fabric_spark_object_store_interoperability validate + - name: Validate metadata fabric production object-store gate + run: python -m data_agent.metadata_fabric_object_store_gate validate + - name: Validate DolphinScheduler adapter boundary run: python -m data_agent.dolphinscheduler_adapter validate @@ -214,6 +218,7 @@ jobs: data_agent/test_metadata_fabric_gravitino_jdbc_restart.py \ data_agent/test_metadata_fabric_spark_iceberg_rest_interoperability.py \ data_agent/test_metadata_fabric_spark_object_store_interoperability.py \ + data_agent/test_metadata_fabric_object_store_gate.py \ data_agent/test_metadata_fabric_otel_failure_rehearsal.py \ data_agent/test_metadata_fabric_otel_metrics.py \ data_agent/test_metadata_fabric_provider_metrics.py \ diff --git a/config/metadata-fabric-object-store.production.yaml b/config/metadata-fabric-object-store.production.yaml new file mode 100644 index 00000000..48ecc4f9 --- /dev/null +++ b/config/metadata-fabric-object-store.production.yaml @@ -0,0 +1,118 @@ +schema: gda.metadata_fabric_object_store_production_profile.v1 +environment: production + +scope: + engines: + spark: 3.5.0 + iceberg: 1.6.1 + gravitino: 1.3.0 + local_evidence: + path: docs/evidence/metadata-fabric-spark-object-store-interoperability-2026-07-29.json + evidence_fingerprint: 05844457efb378581fb7fc2e7ed3c706819b2d8fa5a52b2f82577051d38c2cd1 + required_claim: local_spark_object_store_interoperability_verified + +provider: + decision_status: pending + provider_type: null + account_reference: null + region: null + endpoint: null + bucket: null + warehouse_prefix: warehouse/ + infrastructure_reference: null + failure_domain_reference: null + recovery_region: null + multi_az_required: true + source_cluster_independent: true + +identity: + integration_mode: null + workload_identity_reference: null + kubernetes_service_account: null + least_privilege_policy_reference: null + bucket_policy_reference: null + allowed_operations: + - s3:AbortMultipartUpload + - s3:DeleteObject + - s3:GetBucketLocation + - s3:GetObject + - s3:ListBucket + - s3:ListBucketMultipartUploads + - s3:ListMultipartUploadParts + - s3:PutObject + static_credentials_forbidden: true + maximum_session_ttl_seconds: 900 + +transport: + tls_required: true + minimum_version: TLSv1.2 + endpoint: null + private_connectivity_reference: null + dns_policy_reference: null + trust_bundle_reference: null + certificate_policy_reference: null + +encryption: + server_side_required: true + mode: kms + key_reference: null + key_policy_reference: null + rotation_days: null + bucket_key_enabled: true + +durability: + versioning_enabled: true + delete_protection_mode: versioned_recovery + retention_policy_reference: null + replication_mode: asynchronous_cross_region + replication_policy_reference: null + recovery_bucket_reference: null + maximum_rpo_minutes: 15 + maximum_rto_minutes: 60 + +consistency: + strong_read_after_write_required: true + strong_list_after_write_required: true + atomic_rename_required: false + multipart_upload_cleanup_reference: null + orphan_file_cleanup_reference: null + +tenancy: + isolation_mode: bucket_prefix_and_provider_policy + tenant_prefix_template: tenants/{tenant_id}/warehouse/ + policy_reference: null + cross_tenant_denial_required: true + public_access_blocked: true + +operations: + platform_owner: null + security_owner: null + storage_owner: null + incident_owner: null + audit_log_reference: null + metrics_alert_reference: null + availability_slo_percent: null + latency_slo_ms: null + runbook: + uri: null + version: null + recovery_runbook: + uri: null + version: null + rollback_runbook: + uri: null + version: null + attestation_environment: production-object-store + attestation_policy_reference: null + +claims: + object_store_decision_frozen: false + protected_workload_identity_verified: false + tls_verified: false + kms_encryption_verified: false + tenant_isolation_verified: false + object_store_durability_verified: false + object_store_failure_recovery_verified: false + production_object_store_verified: false + production_object_store_gate_passed: false + production_ready: false diff --git a/data_agent/metadata_fabric_object_store_gate.py b/data_agent/metadata_fabric_object_store_gate.py new file mode 100644 index 00000000..9e6e2804 --- /dev/null +++ b/data_agent/metadata_fabric_object_store_gate.py @@ -0,0 +1,895 @@ +"""Evaluate the fail-closed production object-store readiness gate. + +The checked profile binds the local Spark/MinIO interoperability evidence and +freezes the external decisions required for a production Iceberg warehouse. +It deploys nothing, accepts no credentials, and derives production claims only +from a fresh attestation bound to the exact profile and protected environment. +""" + +from __future__ import annotations + +import argparse +import ipaddress +import json +import re +import sys +from collections.abc import Mapping +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any +from urllib.parse import urlparse + +import yaml + +from . import metadata_fabric_recovery_rehearsal as recovery +from . import metadata_fabric_spark_object_store_interoperability as local_interop + + +PROFILE_SCHEMA = "gda.metadata_fabric_object_store_production_profile.v1" +ATTESTATION_SCHEMA = "gda.metadata_fabric_object_store_attestation.v1" +REPORT_SCHEMA = "gda.metadata_fabric_object_store_readiness_report.v1" +ENVIRONMENT = "production" +REPOSITORY = "zhouning/gisdataagent" +PROTECTED_ENVIRONMENT = "production-object-store" + +REPO_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_PROFILE_PATH = REPO_ROOT / "config/metadata-fabric-object-store.production.yaml" +DEFAULT_WRAPPER_PATH = REPO_ROOT / "scripts/metadata-fabric-object-store-gate.sh" + +LOCAL_EVIDENCE = { + "path": "docs/evidence/metadata-fabric-spark-object-store-interoperability-2026-07-29.json", + "evidence_fingerprint": ( + "05844457efb378581fb7fc2e7ed3c706819b2d8fa5a52b2f82577051d38c2cd1" + ), + "required_claim": "local_spark_object_store_interoperability_verified", +} +EXPECTED_ENGINES = {"spark": "3.5.0", "iceberg": "1.6.1", "gravitino": "1.3.0"} +ALLOWED_PROVIDERS = { + "aws_s3", + "huawei_obs_s3_compatible", + "managed_s3_compatible", +} +ALLOWED_OPERATIONS = [ + "s3:AbortMultipartUpload", + "s3:DeleteObject", + "s3:GetBucketLocation", + "s3:GetObject", + "s3:ListBucket", + "s3:ListBucketMultipartUploads", + "s3:ListMultipartUploadParts", + "s3:PutObject", +] +EXPECTED_CHECKS = { + "provider_account_isolation", + "bucket_outside_source_cluster", + "multi_az_durability", + "private_network_path", + "tls_transport", + "workload_identity_exchange", + "static_credentials_absent", + "least_privilege_allow", + "administrative_action_denied", + "cross_tenant_denial", + "public_access_blocked", + "kms_encrypt_decrypt", + "kms_key_rotation", + "versioning_enabled", + "cross_region_replication", + "read_after_write_consistency", + "list_after_write_consistency", + "multipart_abort_cleanup", + "orphan_file_cleanup", + "spark_gravitino_read_write", + "commit_failure_recovery", + "source_cluster_loss_recovery", + "audit_log_delivery", + "metrics_alert_delivery", + "backup_restore", + "rollback_rehearsal", +} +PROFILE_CLAIMS = { + "object_store_decision_frozen", + "protected_workload_identity_verified", + "tls_verified", + "kms_encryption_verified", + "tenant_isolation_verified", + "object_store_durability_verified", + "object_store_failure_recovery_verified", + "production_object_store_verified", + "production_object_store_gate_passed", + "production_ready", +} +REPORT_CLAIMS = PROFILE_CLAIMS - {"production_ready"} +BINDING_SECTIONS = ( + "provider", + "identity", + "transport", + "encryption", + "durability", + "consistency", + "tenancy", + "operations", +) +REPORT_INVENTORY = { + "schema", + "environment", + "profile_fingerprint", + "local_evidence_fingerprint", + "attestation_fingerprint", + *{f"{name}_fingerprint" for name in BINDING_SECTIONS}, + "profile_valid", + "profile_errors", + "profile_blockers", + "ready_for_protected_verification", + "attestation_valid", + "attestation_errors", + *REPORT_CLAIMS, + "production_ready", + "report_fingerprint", +} + +SHA40_PATTERN = re.compile(r"^[0-9a-f]{40}$") +SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") +DNS_LABEL_PATTERN = re.compile(r"^[a-z0-9](?:[-a-z0-9.]{1,61}[a-z0-9])?$") +REGION_PATTERN = re.compile(r"^[a-z][a-z0-9-]{1,31}$") +REFERENCE_PATTERN = re.compile(r"^[a-z][a-z0-9+.-]*://[^\s]+$", re.IGNORECASE) +PLACEHOLDER_PATTERN = re.compile( + r"(^|[-_.:/])(pending|placeholder|replace|tbd|todo|changeme)([-_.:/]|$)|" + r"[<>]|\.example(?=[:/]|$)", + re.IGNORECASE, +) +SENSITIVE_KEY_PATTERN = re.compile( + r"(^|[-_.])(password|passwd|secret|client[-_.]?secret|private[-_.]?key|" + r"access[-_.]?key|access[-_.]?token|refresh[-_.]?token|authorization)" + r"($|[-_.])", + re.IGNORECASE, +) + + +class MetadataFabricObjectStoreGateError(RuntimeError): + """The production object-store readiness contract failed closed.""" + + +def _mapping(value: Any) -> Mapping[str, Any]: + return value if isinstance(value, Mapping) else {} + + +def _load_yaml_object(path: Path) -> dict[str, Any]: + value = yaml.safe_load(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise TypeError("YAML document is not an object") + return value + + +def _load_json_object(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise TypeError("JSON document is not an object") + return value + + +def _inventory_errors( + value: Mapping[str, Any], expected: set[str], label: str +) -> list[str]: + return [] if set(value) == expected else [f"{label} inventory does not match"] + + +def _placeholder(value: Any) -> bool: + return not isinstance(value, str) or not value.strip() or bool( + PLACEHOLDER_PATTERN.search(value.strip()) + ) + + +def _reference(value: Any) -> bool: + return not _placeholder(value) and bool(REFERENCE_PATTERN.fullmatch(str(value))) + + +def _production_https_url(value: Any) -> bool: + if _placeholder(value): + return False + parsed = urlparse(str(value)) + if ( + parsed.scheme != "https" + or not parsed.hostname + or parsed.username + or parsed.password + or parsed.query + or parsed.fragment + ): + return False + hostname = parsed.hostname.lower() + if hostname in {"localhost", "0.0.0.0", "::1"} or hostname.endswith( + (".localhost", ".local", ".svc", ".cluster.local") + ): + return False + if hostname in {"example.com", "example.net", "example.org"} or hostname.endswith( + (".example.com", ".example.net", ".example.org") + ): + return False + try: + if ipaddress.ip_address(hostname).is_private: + return False + except ValueError: + pass + return True + + +def _sensitive_paths(value: Any, path: tuple[str, ...] = ()) -> list[str]: + found: list[str] = [] + if isinstance(value, Mapping): + for key, nested in value.items(): + child = (*path, str(key)) + if SENSITIVE_KEY_PATTERN.search(str(key)): + found.append(".".join(child)) + found.extend(_sensitive_paths(nested, child)) + elif isinstance(value, list): + for index, nested in enumerate(value): + found.extend(_sensitive_paths(nested, (*path, str(index)))) + return found + + +def _local_evidence_errors(value: Mapping[str, Any]) -> list[str]: + errors = _inventory_errors(value, set(LOCAL_EVIDENCE), "local evidence") + if dict(value) != LOCAL_EVIDENCE: + errors.append("local object-store evidence binding does not match") + return errors + try: + path = (REPO_ROOT / LOCAL_EVIDENCE["path"]).resolve() + path.relative_to(REPO_ROOT) + evidence = _load_json_object(path) + integrity_errors = local_interop.verify_evidence_integrity(evidence) + except (OSError, TypeError, ValueError, json.JSONDecodeError): + errors.append("local object-store evidence is unavailable") + return errors + if integrity_errors: + errors.append("local object-store evidence integrity does not match") + if ( + evidence.get("evidence_fingerprint") + != LOCAL_EVIDENCE["evidence_fingerprint"] + or evidence.get(LOCAL_EVIDENCE["required_claim"]) is not True + ): + errors.append("local object-store evidence claim does not match") + for claim in ( + "production_object_store_verified", + "spark_conformance_verified", + "production_ready", + ): + if evidence.get(claim) is not False: + errors.append(f"local object-store evidence overclaims {claim}") + return errors + + +def _runbook_errors(value: Mapping[str, Any], label: str) -> list[str]: + errors = _inventory_errors(value, {"uri", "version"}, label) + if value.get("uri") is not None and not _production_https_url(value.get("uri")): + errors.append(f"{label} URI is invalid") + if value.get("version") is not None and _placeholder(value.get("version")): + errors.append(f"{label} version is invalid") + return errors + + +def _profile_errors(profile: Mapping[str, Any]) -> list[str]: + errors: list[str] = [] + if _sensitive_paths(profile): + errors.append("object-store profile contains credential-bearing fields") + errors.extend( + _inventory_errors( + profile, + { + "schema", + "environment", + "scope", + *BINDING_SECTIONS, + "claims", + }, + "object-store profile", + ) + ) + if profile.get("schema") != PROFILE_SCHEMA or profile.get("environment") != ENVIRONMENT: + errors.append("object-store profile schema or environment does not match") + + scope = _mapping(profile.get("scope")) + errors.extend(_inventory_errors(scope, {"engines", "local_evidence"}, "scope")) + if dict(_mapping(scope.get("engines"))) != EXPECTED_ENGINES: + errors.append("object-store engine version binding does not match") + errors.extend(_local_evidence_errors(_mapping(scope.get("local_evidence")))) + + provider = _mapping(profile.get("provider")) + provider_keys = { + "decision_status", + "provider_type", + "account_reference", + "region", + "endpoint", + "bucket", + "warehouse_prefix", + "infrastructure_reference", + "failure_domain_reference", + "recovery_region", + "multi_az_required", + "source_cluster_independent", + } + errors.extend(_inventory_errors(provider, provider_keys, "object-store provider")) + if provider.get("decision_status") not in {"pending", "approved"}: + errors.append("object-store provider decision status is invalid") + if provider.get("provider_type") not in {None, *ALLOWED_PROVIDERS}: + errors.append("object-store provider type is invalid") + for key in ( + "account_reference", + "infrastructure_reference", + "failure_domain_reference", + ): + if provider.get(key) is not None and not _reference(provider.get(key)): + errors.append(f"object-store provider {key} is invalid") + for key in ("region", "recovery_region"): + if provider.get(key) is not None and not REGION_PATTERN.fullmatch( + str(provider.get(key)) + ): + errors.append(f"object-store provider {key} is invalid") + if provider.get("endpoint") is not None and not _production_https_url( + provider.get("endpoint") + ): + errors.append("object-store provider endpoint is invalid") + if provider.get("bucket") is not None and not DNS_LABEL_PATTERN.fullmatch( + str(provider.get("bucket")) + ): + errors.append("object-store bucket name is invalid") + prefix = provider.get("warehouse_prefix") + if ( + not isinstance(prefix, str) + or not prefix.endswith("/") + or prefix.startswith("/") + or ".." in prefix + ): + errors.append("object-store warehouse prefix is invalid") + if ( + provider.get("region") is not None + and provider.get("region") == provider.get("recovery_region") + ): + errors.append("object-store recovery region must be distinct") + if ( + provider.get("multi_az_required") is not True + or provider.get("source_cluster_independent") is not True + ): + errors.append("object-store provider failure-domain baseline does not match") + + identity = _mapping(profile.get("identity")) + identity_keys = { + "integration_mode", + "workload_identity_reference", + "kubernetes_service_account", + "least_privilege_policy_reference", + "bucket_policy_reference", + "allowed_operations", + "static_credentials_forbidden", + "maximum_session_ttl_seconds", + } + errors.extend(_inventory_errors(identity, identity_keys, "object-store identity")) + if identity.get("integration_mode") not in {None, "oidc_workload_federation"}: + errors.append("object-store identity integration mode is invalid") + for key in ( + "workload_identity_reference", + "least_privilege_policy_reference", + "bucket_policy_reference", + ): + if identity.get(key) is not None and not _reference(identity.get(key)): + errors.append(f"object-store identity {key} is invalid") + service_account = identity.get("kubernetes_service_account") + if service_account is not None and not DNS_LABEL_PATTERN.fullmatch(str(service_account)): + errors.append("object-store Kubernetes ServiceAccount is invalid") + if identity.get("allowed_operations") != ALLOWED_OPERATIONS: + errors.append("object-store least-privilege operation inventory does not match") + if ( + identity.get("static_credentials_forbidden") is not True + or identity.get("maximum_session_ttl_seconds") != 900 + ): + errors.append("object-store credential baseline does not match") + + transport = _mapping(profile.get("transport")) + transport_keys = { + "tls_required", + "minimum_version", + "endpoint", + "private_connectivity_reference", + "dns_policy_reference", + "trust_bundle_reference", + "certificate_policy_reference", + } + errors.extend(_inventory_errors(transport, transport_keys, "object-store transport")) + if ( + transport.get("tls_required") is not True + or transport.get("minimum_version") not in {"TLSv1.2", "TLSv1.3"} + ): + errors.append("object-store TLS baseline does not match") + if transport.get("endpoint") is not None and not _production_https_url( + transport.get("endpoint") + ): + errors.append("object-store transport endpoint is invalid") + if ( + provider.get("endpoint") is not None + and transport.get("endpoint") != provider.get("endpoint") + ): + errors.append("object-store provider and transport endpoints differ") + for key in transport_keys - {"tls_required", "minimum_version", "endpoint"}: + if transport.get(key) is not None and not _reference(transport.get(key)): + errors.append(f"object-store transport {key} is invalid") + + encryption = _mapping(profile.get("encryption")) + encryption_keys = { + "server_side_required", + "mode", + "key_reference", + "key_policy_reference", + "rotation_days", + "bucket_key_enabled", + } + errors.extend(_inventory_errors(encryption, encryption_keys, "object-store encryption")) + if ( + encryption.get("server_side_required") is not True + or encryption.get("mode") != "kms" + or encryption.get("bucket_key_enabled") is not True + ): + errors.append("object-store encryption baseline does not match") + for key in ("key_reference", "key_policy_reference"): + if encryption.get(key) is not None and not _reference(encryption.get(key)): + errors.append(f"object-store encryption {key} is invalid") + rotation = encryption.get("rotation_days") + if rotation is not None and ( + not isinstance(rotation, int) or isinstance(rotation, bool) or not 1 <= rotation <= 365 + ): + errors.append("object-store encryption rotation is invalid") + + durability = _mapping(profile.get("durability")) + durability_keys = { + "versioning_enabled", + "delete_protection_mode", + "retention_policy_reference", + "replication_mode", + "replication_policy_reference", + "recovery_bucket_reference", + "maximum_rpo_minutes", + "maximum_rto_minutes", + } + errors.extend(_inventory_errors(durability, durability_keys, "object-store durability")) + if ( + durability.get("versioning_enabled") is not True + or durability.get("delete_protection_mode") != "versioned_recovery" + or durability.get("replication_mode") != "asynchronous_cross_region" + or durability.get("maximum_rpo_minutes") != 15 + or durability.get("maximum_rto_minutes") != 60 + ): + errors.append("object-store durability baseline does not match") + for key in ( + "retention_policy_reference", + "replication_policy_reference", + "recovery_bucket_reference", + ): + if durability.get(key) is not None and not _reference(durability.get(key)): + errors.append(f"object-store durability {key} is invalid") + + consistency = _mapping(profile.get("consistency")) + consistency_keys = { + "strong_read_after_write_required", + "strong_list_after_write_required", + "atomic_rename_required", + "multipart_upload_cleanup_reference", + "orphan_file_cleanup_reference", + } + errors.extend(_inventory_errors(consistency, consistency_keys, "object-store consistency")) + if ( + consistency.get("strong_read_after_write_required") is not True + or consistency.get("strong_list_after_write_required") is not True + or consistency.get("atomic_rename_required") is not False + ): + errors.append("object-store consistency baseline does not match") + for key in ("multipart_upload_cleanup_reference", "orphan_file_cleanup_reference"): + if consistency.get(key) is not None and not _reference(consistency.get(key)): + errors.append(f"object-store consistency {key} is invalid") + + tenancy = _mapping(profile.get("tenancy")) + tenancy_keys = { + "isolation_mode", + "tenant_prefix_template", + "policy_reference", + "cross_tenant_denial_required", + "public_access_blocked", + } + errors.extend(_inventory_errors(tenancy, tenancy_keys, "object-store tenancy")) + if ( + tenancy.get("isolation_mode") != "bucket_prefix_and_provider_policy" + or "{tenant_id}" not in str(tenancy.get("tenant_prefix_template", "")) + or tenancy.get("cross_tenant_denial_required") is not True + or tenancy.get("public_access_blocked") is not True + ): + errors.append("object-store tenancy baseline does not match") + if tenancy.get("policy_reference") is not None and not _reference( + tenancy.get("policy_reference") + ): + errors.append("object-store tenancy policy reference is invalid") + + operations = _mapping(profile.get("operations")) + operations_keys = { + "platform_owner", + "security_owner", + "storage_owner", + "incident_owner", + "audit_log_reference", + "metrics_alert_reference", + "availability_slo_percent", + "latency_slo_ms", + "runbook", + "recovery_runbook", + "rollback_runbook", + "attestation_environment", + "attestation_policy_reference", + } + errors.extend(_inventory_errors(operations, operations_keys, "object-store operations")) + for key in ( + "platform_owner", + "security_owner", + "storage_owner", + "incident_owner", + "audit_log_reference", + "metrics_alert_reference", + "attestation_policy_reference", + ): + if operations.get(key) is not None and not _reference(operations.get(key)): + errors.append(f"object-store operations {key} is invalid") + availability = operations.get("availability_slo_percent") + if availability is not None and ( + not isinstance(availability, (int, float)) + or isinstance(availability, bool) + or not 99.9 <= availability <= 100 + ): + errors.append("object-store availability SLO is invalid") + latency = operations.get("latency_slo_ms") + if latency is not None and ( + not isinstance(latency, int) or isinstance(latency, bool) or not 1 <= latency <= 5000 + ): + errors.append("object-store latency SLO is invalid") + for name in ("runbook", "recovery_runbook", "rollback_runbook"): + errors.extend(_runbook_errors(_mapping(operations.get(name)), name)) + if operations.get("attestation_environment") != PROTECTED_ENVIRONMENT: + errors.append("object-store protected environment does not match") + + claims = _mapping(profile.get("claims")) + errors.extend(_inventory_errors(claims, PROFILE_CLAIMS, "object-store claims")) + for claim in PROFILE_CLAIMS: + if claims.get(claim) is not False: + errors.append(f"object-store profile may not self-assert {claim}") + return errors + + +def _profile_blockers(profile: Mapping[str, Any]) -> list[str]: + blockers: list[str] = [] + provider = _mapping(profile.get("provider")) + if provider.get("decision_status") != "approved": + blockers.append("provider.decision_status") + required = { + "provider": ( + "provider_type", + "account_reference", + "region", + "endpoint", + "bucket", + "infrastructure_reference", + "failure_domain_reference", + "recovery_region", + ), + "identity": ( + "integration_mode", + "workload_identity_reference", + "kubernetes_service_account", + "least_privilege_policy_reference", + "bucket_policy_reference", + ), + "transport": ( + "endpoint", + "private_connectivity_reference", + "dns_policy_reference", + "trust_bundle_reference", + "certificate_policy_reference", + ), + "encryption": ("key_reference", "key_policy_reference", "rotation_days"), + "durability": ( + "retention_policy_reference", + "replication_policy_reference", + "recovery_bucket_reference", + ), + "consistency": ( + "multipart_upload_cleanup_reference", + "orphan_file_cleanup_reference", + ), + "tenancy": ("policy_reference",), + "operations": ( + "platform_owner", + "security_owner", + "storage_owner", + "incident_owner", + "audit_log_reference", + "metrics_alert_reference", + "availability_slo_percent", + "latency_slo_ms", + "attestation_policy_reference", + ), + } + for section, names in required.items(): + item = _mapping(profile.get(section)) + for name in names: + value = item.get(name) + if value is None or (isinstance(value, str) and _placeholder(value)): + blockers.append(f"{section}.{name}") + operations = _mapping(profile.get("operations")) + for name in ("runbook", "recovery_runbook", "rollback_runbook"): + runbook = _mapping(operations.get(name)) + for key in ("uri", "version"): + value = runbook.get(key) + if value is None or (isinstance(value, str) and _placeholder(value)): + blockers.append(f"operations.{name}.{key}") + return blockers + + +def _binding_fingerprints(profile: Mapping[str, Any]) -> dict[str, str]: + return { + f"{name}_fingerprint": recovery._canonical_sha256( + dict(_mapping(profile.get(name))) + ) + for name in BINDING_SECTIONS + } + + +def _parse_timestamp(value: Any) -> datetime | None: + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return parsed.astimezone(UTC) if parsed.tzinfo else None + + +def _attestation_errors( + attestation: Mapping[str, Any], + *, + profile: Mapping[str, Any], + profile_fingerprint: str, + bindings: Mapping[str, str], + now: datetime, +) -> list[str]: + errors: list[str] = [] + expected_inventory = { + "schema", + "environment", + "repository", + "protected_environment", + "source_revision", + "profile_fingerprint", + "local_evidence_fingerprint", + "engine_versions", + *bindings, + "observed_at", + "expires_at", + "evidence_uri", + "checks", + "claims", + "runbook_versions", + } + errors.extend(_inventory_errors(attestation, expected_inventory, "object-store attestation")) + if _sensitive_paths(attestation): + errors.append("object-store attestation contains credential-bearing fields") + if ( + attestation.get("schema") != ATTESTATION_SCHEMA + or attestation.get("environment") != ENVIRONMENT + or attestation.get("repository") != REPOSITORY + or attestation.get("protected_environment") != PROTECTED_ENVIRONMENT + ): + errors.append("object-store attestation authority does not match") + if not SHA40_PATTERN.fullmatch(str(attestation.get("source_revision", ""))): + errors.append("object-store attestation source revision is invalid") + if attestation.get("profile_fingerprint") != profile_fingerprint: + errors.append("object-store attestation does not bind the current profile") + if ( + attestation.get("local_evidence_fingerprint") + != LOCAL_EVIDENCE["evidence_fingerprint"] + or dict(_mapping(attestation.get("engine_versions"))) != EXPECTED_ENGINES + ): + errors.append("object-store attestation dependency bindings do not match") + for key, expected in bindings.items(): + if attestation.get(key) != expected: + errors.append(f"object-store attestation binding does not match: {key}") + observed = _parse_timestamp(attestation.get("observed_at")) + expires = _parse_timestamp(attestation.get("expires_at")) + if observed is None or observed > now or now - observed > timedelta(hours=24): + errors.append("object-store attestation is outside the 24-hour freshness window") + if ( + expires is None + or expires <= now + or observed is None + or expires - observed > timedelta(days=7) + ): + errors.append("object-store attestation expiry is invalid") + if not _production_https_url(attestation.get("evidence_uri")): + errors.append("object-store attestation evidence URI is invalid") + checks = _mapping(attestation.get("checks")) + errors.extend(_inventory_errors(checks, EXPECTED_CHECKS, "object-store checks")) + for check in EXPECTED_CHECKS: + if checks.get(check) != "passed": + errors.append(f"object-store attestation check did not pass: {check}") + claims = _mapping(attestation.get("claims")) + errors.extend(_inventory_errors(claims, PROFILE_CLAIMS, "object-store attestation claims")) + for claim in PROFILE_CLAIMS - {"production_ready"}: + if claims.get(claim) is not True: + errors.append(f"object-store attestation claim did not pass: {claim}") + if claims.get("production_ready") is not False: + errors.append("object-store attestation may not claim overall production readiness") + operations = _mapping(profile.get("operations")) + expected_runbooks = { + name: _mapping(operations.get(name)).get("version") + for name in ("runbook", "recovery_runbook", "rollback_runbook") + } + if dict(_mapping(attestation.get("runbook_versions"))) != expected_runbooks: + errors.append("object-store attestation runbook versions do not match") + return errors + + +def _stable_report( + *, + profile_fingerprint: str | None, + bindings: Mapping[str, str | None], + profile_valid: bool, + profile_errors: list[str], + blockers: list[str], + attestation_valid: bool, + attestation_errors: list[str], + attestation_fingerprint: str | None, +) -> dict[str, Any]: + passed = profile_valid and not blockers and attestation_valid + stable: dict[str, Any] = { + "schema": REPORT_SCHEMA, + "environment": ENVIRONMENT, + "profile_fingerprint": profile_fingerprint, + "local_evidence_fingerprint": LOCAL_EVIDENCE["evidence_fingerprint"], + "attestation_fingerprint": attestation_fingerprint, + **bindings, + "profile_valid": profile_valid, + "profile_errors": profile_errors, + "profile_blockers": blockers, + "ready_for_protected_verification": profile_valid and not blockers, + "attestation_valid": attestation_valid, + "attestation_errors": attestation_errors, + **{claim: passed for claim in REPORT_CLAIMS}, + "production_ready": False, + } + return {**stable, "report_fingerprint": recovery._canonical_sha256(stable)} + + +def build_object_store_readiness_report( + *, + profile_path: Path = DEFAULT_PROFILE_PATH, + attestation: Mapping[str, Any] | None = None, + now: datetime | None = None, +) -> dict[str, Any]: + try: + profile = _load_yaml_object(profile_path.resolve()) + except (OSError, TypeError, ValueError, yaml.YAMLError) as exc: + return _stable_report( + profile_fingerprint=None, + bindings={f"{name}_fingerprint": None for name in BINDING_SECTIONS}, + profile_valid=False, + profile_errors=[f"object-store profile is invalid: {type(exc).__name__}"], + blockers=[], + attestation_valid=False, + attestation_errors=["object-store profile is not ready for attestation"], + attestation_fingerprint=None, + ) + profile_errors = _profile_errors(profile) + profile_valid = not profile_errors + blockers = _profile_blockers(profile) if profile_valid else [] + profile_fingerprint = recovery._canonical_sha256(profile) + bindings = _binding_fingerprints(profile) + attestation_errors: list[str] + attestation_valid = False + attestation_fingerprint: str | None = None + if attestation is None: + attestation_errors = ["production object-store attestation is required"] + elif not profile_valid or blockers: + attestation_errors = ["object-store profile is not ready for attestation"] + else: + attestation_value = dict(attestation) + attestation_fingerprint = recovery._canonical_sha256(attestation_value) + attestation_errors = _attestation_errors( + attestation_value, + profile=profile, + profile_fingerprint=profile_fingerprint, + bindings=bindings, + now=(now or datetime.now(UTC)).astimezone(UTC), + ) + attestation_valid = not attestation_errors + return _stable_report( + profile_fingerprint=profile_fingerprint, + bindings=bindings, + profile_valid=profile_valid, + profile_errors=profile_errors, + blockers=blockers, + attestation_valid=attestation_valid, + attestation_errors=attestation_errors, + attestation_fingerprint=attestation_fingerprint, + ) + + +def verify_report_integrity(report: Mapping[str, Any]) -> list[str]: + errors = _inventory_errors(report, REPORT_INVENTORY, "object-store readiness report") + stable = {key: value for key, value in report.items() if key != "report_fingerprint"} + if report.get("report_fingerprint") != recovery._canonical_sha256(stable): + errors.append("object-store readiness report fingerprint does not match") + if report.get("production_ready") is not False: + errors.append("object-store gate may not claim overall production readiness") + expected = ( + report.get("profile_valid") is True + and report.get("ready_for_protected_verification") is True + and report.get("attestation_valid") is True + ) + for claim in REPORT_CLAIMS: + if report.get(claim) is not expected: + errors.append(f"object-store gate result is inconsistent: {claim}") + if report.get("ready_for_protected_verification") is not ( + report.get("profile_valid") is True and not report.get("profile_blockers") + ): + errors.append("object-store readiness derivation is inconsistent") + for key in ( + "profile_fingerprint", + "local_evidence_fingerprint", + *{f"{name}_fingerprint" for name in BINDING_SECTIONS}, + ): + value = report.get(key) + if value is not None and not SHA256_PATTERN.fullmatch(str(value)): + errors.append(f"object-store report fingerprint field is invalid: {key}") + return errors + + +def _write_report(report: Mapping[str, Any], output: Path | None) -> None: + rendered = json.dumps(report, ensure_ascii=True, indent=2, sort_keys=True) + "\n" + if output is None: + print(rendered, end="") + return + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(rendered, encoding="utf-8") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + validate = subparsers.add_parser("validate") + validate.add_argument("--profile", type=Path, default=DEFAULT_PROFILE_PATH) + validate.add_argument("--output", type=Path) + evaluate = subparsers.add_parser("evaluate") + evaluate.add_argument("--profile", type=Path, default=DEFAULT_PROFILE_PATH) + evaluate.add_argument("--attestation", type=Path, required=True) + evaluate.add_argument("--output", type=Path) + verify = subparsers.add_parser("verify") + verify.add_argument("--report", type=Path, required=True) + args = parser.parse_args(argv) + try: + if args.command == "verify": + report = _load_json_object(args.report) + errors = verify_report_integrity(report) + print(json.dumps({"verified": not errors, "errors": errors}, indent=2)) + return 0 if not errors else 1 + attestation = ( + _load_json_object(args.attestation) if args.command == "evaluate" else None + ) + report = build_object_store_readiness_report( + profile_path=args.profile, + attestation=attestation, + ) + _write_report(report, args.output) + if args.command == "validate": + return 0 if report["profile_valid"] else 1 + return 0 if report["production_object_store_gate_passed"] else 1 + except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: + print(f"metadata fabric object-store gate: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/data_agent/test_metadata_fabric_object_store_gate.py b/data_agent/test_metadata_fabric_object_store_gate.py new file mode 100644 index 00000000..5cdc26fd --- /dev/null +++ b/data_agent/test_metadata_fabric_object_store_gate.py @@ -0,0 +1,407 @@ +from __future__ import annotations + +import json +from copy import deepcopy +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest +import yaml + +from data_agent import metadata_fabric_object_store_gate as gate + + +NOW = datetime(2026, 7, 29, 14, 0, tzinfo=UTC) + + +def _write_profile(tmp_path: Path, profile: dict) -> Path: + path = tmp_path / "object-store-profile.yaml" + path.write_text(yaml.safe_dump(profile, sort_keys=False), encoding="utf-8") + return path + + +def _checked_profile() -> dict: + return gate._load_yaml_object(gate.DEFAULT_PROFILE_PATH) + + +def _complete_profile(provider_type: str = "aws_s3") -> dict: + profile = deepcopy(_checked_profile()) + profile["provider"].update( + { + "decision_status": "approved", + "provider_type": provider_type, + "account_reference": "cloud://aws/accounts/gda-production", + "region": "ap-northeast-1", + "endpoint": "https://s3.ap-northeast-1.amazonaws.com", + "bucket": "gda-prod-lakehouse-01", + "infrastructure_reference": "iac://gda-production/object-store/v1", + "failure_domain_reference": "cloud://aws/failure-domains/ap-northeast-1", + "recovery_region": "ap-northeast-3", + } + ) + profile["identity"].update( + { + "integration_mode": "oidc_workload_federation", + "workload_identity_reference": "iam://gda-production/lakehouse-writer", + "kubernetes_service_account": "gda-lakehouse-writer", + "least_privilege_policy_reference": "policy://object-store/lakehouse-writer/v1", + "bucket_policy_reference": "policy://object-store/bucket-boundary/v1", + } + ) + profile["transport"].update( + { + "endpoint": profile["provider"]["endpoint"], + "private_connectivity_reference": "network://gda-production/object-store-private-path", + "dns_policy_reference": "dns://gda-production/object-store/v1", + "trust_bundle_reference": "pki://gda-production/object-store-ca/v1", + "certificate_policy_reference": "pki://gda-production/object-store-certificate/v1", + } + ) + profile["encryption"].update( + { + "key_reference": "kms://gda-production/lakehouse-key/v1", + "key_policy_reference": "policy://kms/lakehouse-key/v1", + "rotation_days": 365, + } + ) + profile["durability"].update( + { + "retention_policy_reference": "policy://object-store/retention/v1", + "replication_policy_reference": "policy://object-store/cross-region-replication/v1", + "recovery_bucket_reference": "s3://gda-prod-lakehouse-recovery", + } + ) + profile["consistency"].update( + { + "multipart_upload_cleanup_reference": "policy://object-store/multipart-cleanup/v1", + "orphan_file_cleanup_reference": "policy://iceberg/orphan-file-cleanup/v1", + } + ) + profile["tenancy"]["policy_reference"] = "policy://object-store/tenant-isolation/v1" + profile["operations"].update( + { + "platform_owner": "owner://teams/data-platform", + "security_owner": "owner://teams/security", + "storage_owner": "owner://teams/storage-sre", + "incident_owner": "owner://teams/platform-oncall", + "audit_log_reference": "observability://object-store/audit/v1", + "metrics_alert_reference": "observability://object-store/alerts/v1", + "availability_slo_percent": 99.95, + "latency_slo_ms": 250, + "runbook": { + "uri": "https://runbooks.gisdataagent.cn/object-store/operations", + "version": "v1", + }, + "recovery_runbook": { + "uri": "https://runbooks.gisdataagent.cn/object-store/recovery", + "version": "v1", + }, + "rollback_runbook": { + "uri": "https://runbooks.gisdataagent.cn/object-store/rollback", + "version": "v1", + }, + "attestation_policy_reference": "policy://protected-environments/object-store/v1", + } + ) + return profile + + +def _attestation(profile_path: Path, **overrides) -> dict: + profile = gate._load_yaml_object(profile_path) + report = gate.build_object_store_readiness_report( + profile_path=profile_path, now=NOW + ) + attestation = { + "schema": gate.ATTESTATION_SCHEMA, + "environment": gate.ENVIRONMENT, + "repository": gate.REPOSITORY, + "protected_environment": gate.PROTECTED_ENVIRONMENT, + "source_revision": "a" * 40, + "profile_fingerprint": report["profile_fingerprint"], + "local_evidence_fingerprint": gate.LOCAL_EVIDENCE["evidence_fingerprint"], + "engine_versions": dict(gate.EXPECTED_ENGINES), + **{ + f"{name}_fingerprint": report[f"{name}_fingerprint"] + for name in gate.BINDING_SECTIONS + }, + "observed_at": (NOW - timedelta(minutes=5)).isoformat(), + "expires_at": (NOW + timedelta(days=1)).isoformat(), + "evidence_uri": "https://evidence.gisdataagent.cn/object-store/run-20260729", + "checks": {name: "passed" for name in sorted(gate.EXPECTED_CHECKS)}, + "claims": { + name: name != "production_ready" for name in sorted(gate.PROFILE_CLAIMS) + }, + "runbook_versions": { + name: profile["operations"][name]["version"] + for name in ("runbook", "recovery_runbook", "rollback_runbook") + }, + } + attestation.update(overrides) + return attestation + + +def test_checked_profile_is_valid_pending_and_fail_closed(): + report = gate.build_object_store_readiness_report(now=NOW) + + assert report["profile_valid"] is True + assert report["profile_errors"] == [] + assert len(report["profile_blockers"]) == 43 + assert report["ready_for_protected_verification"] is False + assert report["attestation_valid"] is False + assert report["production_object_store_gate_passed"] is False + assert report["production_object_store_verified"] is False + assert report["protected_workload_identity_verified"] is False + assert report["tls_verified"] is False + assert report["production_ready"] is False + assert report["profile_fingerprint"] == ( + "668e194b3c688307014148391e7f389c9d6e9ca69c95d7b4cc92b4acae93181a" + ) + assert report["report_fingerprint"] == ( + "85362dd10b7dc565f9fa567673d90b774cdec714bd1e70fb2c3c83c1af48b5ea" + ) + assert gate.verify_report_integrity(report) == [] + + +def test_checked_profile_binds_verified_local_interoperability_evidence(): + evidence = gate._load_json_object(gate.REPO_ROOT / gate.LOCAL_EVIDENCE["path"]) + + assert gate.local_interop.verify_evidence_integrity(evidence) == [] + assert evidence[gate.LOCAL_EVIDENCE["required_claim"]] is True + assert evidence["production_object_store_verified"] is False + assert evidence["spark_conformance_verified"] is False + assert evidence["production_ready"] is False + + +def test_complete_profile_is_ready_but_cannot_self_attest(tmp_path): + profile_path = _write_profile(tmp_path, _complete_profile()) + + report = gate.build_object_store_readiness_report( + profile_path=profile_path, now=NOW + ) + + assert report["profile_valid"] is True + assert report["profile_blockers"] == [] + assert report["ready_for_protected_verification"] is True + assert report["attestation_valid"] is False + assert report["production_object_store_gate_passed"] is False + assert report["production_ready"] is False + + +def test_fresh_fully_bound_attestation_passes_only_object_store_gate(tmp_path): + profile_path = _write_profile(tmp_path, _complete_profile()) + + report = gate.build_object_store_readiness_report( + profile_path=profile_path, + attestation=_attestation(profile_path), + now=NOW, + ) + + assert report["attestation_valid"] is True + for claim in gate.REPORT_CLAIMS: + assert report[claim] is True + assert report["production_ready"] is False + assert gate.verify_report_integrity(report) == [] + + +@pytest.mark.parametrize("provider_type", sorted(gate.ALLOWED_PROVIDERS)) +def test_profile_accepts_only_s3_compatible_provider_families(tmp_path, provider_type): + profile_path = _write_profile(tmp_path, _complete_profile(provider_type)) + report = gate.build_object_store_readiness_report( + profile_path=profile_path, now=NOW + ) + assert report["profile_valid"] is True + assert report["ready_for_protected_verification"] is True + + +def test_profile_rejects_native_non_s3_provider_without_new_conformance(tmp_path): + profile = _complete_profile("gcs_native") + report = gate.build_object_store_readiness_report( + profile_path=_write_profile(tmp_path, profile), now=NOW + ) + assert report["profile_valid"] is False + assert "provider type is invalid" in "\n".join(report["profile_errors"]) + + +def test_profile_rejects_static_credentials_http_public_access_and_same_region(tmp_path): + profile = _complete_profile() + profile["identity"]["static_credentials_forbidden"] = False + profile["provider"]["endpoint"] = "http://localhost:9000" + profile["transport"]["endpoint"] = "http://localhost:9000" + profile["tenancy"]["public_access_blocked"] = False + profile["provider"]["recovery_region"] = profile["provider"]["region"] + + report = gate.build_object_store_readiness_report( + profile_path=_write_profile(tmp_path, profile), now=NOW + ) + rendered = "\n".join(report["profile_errors"]) + assert report["profile_valid"] is False + assert "credential baseline" in rendered + assert "provider endpoint is invalid" in rendered + assert "transport endpoint is invalid" in rendered + assert "tenancy baseline" in rendered + assert "recovery region must be distinct" in rendered + + +def test_profile_rejects_privilege_expansion_and_local_evidence_drift(tmp_path): + profile = _complete_profile() + profile["identity"]["allowed_operations"].append("s3:PutBucketPolicy") + profile["scope"]["local_evidence"]["evidence_fingerprint"] = "0" * 64 + + report = gate.build_object_store_readiness_report( + profile_path=_write_profile(tmp_path, profile), now=NOW + ) + rendered = "\n".join(report["profile_errors"]) + assert report["profile_valid"] is False + assert "least-privilege operation inventory" in rendered + assert "local object-store evidence binding" in rendered + + +def test_profile_rejects_sensitive_fields_and_self_asserted_claims(tmp_path): + profile = _complete_profile() + profile["identity"]["secret_access_key"] = "must-not-enter-profile" + profile["claims"]["production_object_store_verified"] = True + + report = gate.build_object_store_readiness_report( + profile_path=_write_profile(tmp_path, profile), now=NOW + ) + rendered = "\n".join(report["profile_errors"]) + assert report["profile_valid"] is False + assert "credential-bearing fields" in rendered + assert "identity inventory" in rendered + assert "may not self-assert production_object_store_verified" in rendered + + +@pytest.mark.parametrize( + "binding", + ["profile_fingerprint", *[f"{name}_fingerprint" for name in gate.BINDING_SECTIONS]], +) +def test_attestation_rejects_every_profile_binding_drift(tmp_path, binding): + profile_path = _write_profile(tmp_path, _complete_profile()) + attestation = _attestation(profile_path) + attestation[binding] = "0" * 64 + + report = gate.build_object_store_readiness_report( + profile_path=profile_path, attestation=attestation, now=NOW + ) + + assert report["attestation_valid"] is False + assert "does not bind the current profile" in "\n".join( + report["attestation_errors"] + ) or "binding does not match" in "\n".join(report["attestation_errors"]) + + +def test_attestation_rejects_sensitive_material_expiry_and_dependency_drift(tmp_path): + profile_path = _write_profile(tmp_path, _complete_profile()) + attestation = _attestation( + profile_path, + observed_at=(NOW - timedelta(days=2)).isoformat(), + expires_at=(NOW - timedelta(days=1)).isoformat(), + secret_access_key="must-not-enter-attestation", + ) + attestation["local_evidence_fingerprint"] = "0" * 64 + + report = gate.build_object_store_readiness_report( + profile_path=profile_path, attestation=attestation, now=NOW + ) + rendered = "\n".join(report["attestation_errors"]) + assert report["attestation_valid"] is False + assert "credential-bearing fields" in rendered + assert "freshness window" in rendered + assert "expiry is invalid" in rendered + assert "dependency bindings" in rendered + + +@pytest.mark.parametrize( + "check", + [ + "static_credentials_absent", + "administrative_action_denied", + "cross_tenant_denial", + "kms_key_rotation", + "commit_failure_recovery", + "source_cluster_loss_recovery", + "rollback_rehearsal", + ], +) +def test_attestation_requires_security_durability_and_recovery_checks(tmp_path, check): + profile_path = _write_profile(tmp_path, _complete_profile()) + attestation = _attestation(profile_path) + attestation["checks"][check] = "failed" + + report = gate.build_object_store_readiness_report( + profile_path=profile_path, attestation=attestation, now=NOW + ) + + assert report["attestation_valid"] is False + assert check in "\n".join(report["attestation_errors"]) + assert report["production_object_store_verified"] is False + + +def test_attestation_rejects_runbook_version_and_overall_production_overclaim(tmp_path): + profile_path = _write_profile(tmp_path, _complete_profile()) + attestation = _attestation(profile_path) + attestation["runbook_versions"]["recovery_runbook"] = "v2" + attestation["claims"]["production_ready"] = True + + report = gate.build_object_store_readiness_report( + profile_path=profile_path, attestation=attestation, now=NOW + ) + rendered = "\n".join(report["attestation_errors"]) + assert report["attestation_valid"] is False + assert "runbook versions do not match" in rendered + assert "may not claim overall production readiness" in rendered + + +def test_report_integrity_rejects_tampering_inventory_and_production_overclaim(): + report = gate.build_object_store_readiness_report(now=NOW) + report["production_ready"] = True + report["tls_verified"] = True + + errors = gate.verify_report_integrity(report) + + assert "object-store readiness report fingerprint does not match" in errors + assert "object-store gate may not claim overall production readiness" in errors + assert "object-store gate result is inconsistent: tls_verified" in errors + + forged = gate.build_object_store_readiness_report(now=NOW) + forged["unexpected_claim"] = False + stable = {key: value for key, value in forged.items() if key != "report_fingerprint"} + forged["report_fingerprint"] = gate.recovery._canonical_sha256(stable) + assert "object-store readiness report inventory does not match" in ( + gate.verify_report_integrity(forged) + ) + + +def test_wrapper_is_fail_closed_and_malformed_profile_is_blocked(tmp_path): + text = gate.DEFAULT_WRAPPER_PATH.read_text(encoding="utf-8") + assert "set -euo pipefail" in text + assert "metadata_fabric_object_store_gate" in text + + target = tmp_path / "profile.yaml" + target.write_text("provider: [\n", encoding="utf-8") + report = gate.build_object_store_readiness_report(profile_path=target, now=NOW) + assert report["profile_valid"] is False + assert report["production_object_store_gate_passed"] is False + assert gate.verify_report_integrity(report) == [] + + +def test_attestation_never_records_material_in_success_or_failure_report(tmp_path): + profile_path = _write_profile(tmp_path, _complete_profile()) + successful = gate.build_object_store_readiness_report( + profile_path=profile_path, + attestation=_attestation(profile_path), + now=NOW, + ) + failed_attestation = _attestation(profile_path, secret_access_key="not-recorded") + failed = gate.build_object_store_readiness_report( + profile_path=profile_path, + attestation=failed_attestation, + now=NOW, + ) + + for report in (successful, failed): + rendered = json.dumps(report, sort_keys=True) + assert "secret_access_key" not in rendered + assert "not-recorded" not in rendered + assert successful["production_object_store_gate_passed"] is True + assert failed["production_object_store_gate_passed"] is False diff --git a/docs/architecture-decisions/adr-057-production-object-store-readiness-gate.md b/docs/architecture-decisions/adr-057-production-object-store-readiness-gate.md new file mode 100644 index 00000000..c79ef5e5 --- /dev/null +++ b/docs/architecture-decisions/adr-057-production-object-store-readiness-gate.md @@ -0,0 +1,94 @@ +# ADR-057: Production Object-Store Readiness Gate + +**Status**: Accepted + +**Date**: 2026-07-29 + +**Decision owners**: Metadata Platform, Data Engineering, Security, SRE, Platform Architecture + +**Related decisions**: [ADR-006](adr-006-openmetadata-governance-and-active-metadata-platform.md) · [ADR-053](adr-053-production-metadata-fabric-identity-readiness-gate.md) · [ADR-056](adr-056-local-spark-object-store-interoperability.md) + +## Context + +M3-10 已证明 Spark `3.5.0`、Iceberg `1.6.1` 与 Gravitino `1.3.0` 可以在 Docker Desktop Kubernetes 中通过跨节点 MinIO S3 API 读写同一 Iceberg warehouse，并由直接对象检查验证 data、metadata、manifest、schema 和 snapshot 一致。但 MinIO、两个 Kubernetes node 和 PVC 仍位于同一台主机；运行时使用临时静态凭据、Basic IdP 和无认证 HTTP REST。因此该 evidence 不能证明生产对象存储、独立 failure domain、protected workload identity、TLS、KMS、tenant isolation 或灾难恢复。 + +在具体云账户、region、bucket、identity integration、KMS、replication、owner 和受保护环境尚未获批时，直接选择 AWS S3、华为云 OBS 或另一个 S3-compatible provider 会把技术默认值伪装成生产决策。M3-11 需要先把生产对象存储的最低边界和验收证据冻结为机器可验证、默认关闭的合同，同时明确不选择 provider、不部署基础设施、不持有凭据，也不制造 production attestation。 + +## Options Considered + +| 方案 | 优点 | 代价/风险 | 结论 | +|---|---|---|---| +| 将 M3-10 MinIO evidence 直接提升为生产对象存储 | 无新增实现 | 同主机 failure domain、静态凭据和 HTTP 不具备生产语义 | 拒绝 | +| 在未获批时直接选择并部署一个云 provider | 快速得到真实 endpoint | 账户、区域、安全、成本、合规和 owner 决策均越权 | 拒绝 | +| 继续只在 roadmap 维护对象存储缺口 | 无新增代码 | CI 无法拒绝 placeholder、权限扩大、过期证明或配置漂移 | 拒绝 | +| 版本化 pending profile + 独立、新鲜且完全绑定的 attestation | 决策与运行证据分离；未选型时可诚实 fail closed | 需要 owner 后续批准 profile 并维护证明生命周期 | 采用，限定为 M3-11 | + +## Decision + +### 1. Checked-in profile 固定生产对象存储边界，不代表选型或部署 + +`config/metadata-fabric-object-store.production.yaml` 固定： + +- Spark `3.5.0`、Iceberg `1.6.1`、Gravitino `1.3.0` 与 M3-10 local evidence fingerprint； +- provider account、region、HTTPS endpoint、bucket、warehouse prefix、基础设施和 failure-domain reference，以及独立 recovery region； +- 只允许 `aws_s3`、`huawei_obs_s3_compatible` 或 `managed_s3_compatible`；原生非 S3 provider 必须进入新的 conformance slice； +- OIDC workload federation、Kubernetes ServiceAccount、最长 900 秒 session、禁止 static credential，以及精确的八项 S3 data-plane permission； +- TLS 1.2+、private connectivity、DNS/trust/certificate policy； +- KMS server-side encryption、key policy/rotation、versioning、delete recovery、cross-region replication 与 `RPO <= 15 min`、`RTO <= 60 min`； +- strong read/list-after-write、multipart cleanup、Iceberg orphan cleanup、tenant prefix + provider policy、cross-tenant denial 和 public-access block； +- platform/security/storage/incident owner、audit、metrics/alert、availability/latency SLO、operations/recovery/rollback runbook 与 protected-environment policy； +- 所有 self-reported production claim 固定为 `false`。 + +`null` 与 `decision_status=pending` 是合法的显式 blockers，所以当前 profile 可以结构有效而不假装外部决策已完成。placeholder、HTTP/loopback/cluster-local endpoint、credential-bearing 字段、扩大后的 S3 permission、local evidence 或 engine version 漂移、自报生产结论、同 region recovery 或弱化 identity/TLS/KMS/durability/tenancy baseline 都使 profile 无效。 + +### 2. 生产结论只能从受保护 attestation 派生 + +`data_agent.metadata_fabric_object_store_gate` 将结果分为三层： + +1. `profile_valid`：profile 的结构、安全基线和 M3-10 evidence binding 可信； +2. `ready_for_protected_verification`：provider、identity、transport、encryption、durability、consistency、tenancy 和 operations 的 43 项外部输入均已明确； +3. `production_object_store_gate_passed`：另有新鲜的 `production-object-store` protected-environment attestation，且绑定当前 profile、source revision、M3-10 evidence、engine versions、八组 section fingerprint 和三个 runbook version。 + +attestation 的精确 26 项检查必须全部为 `passed`，覆盖 provider/account/failure domain、private network/TLS、workload identity/static credential absence、least-privilege allow 与 administrative/cross-tenant/public denial、KMS/rotation、versioning/cross-region replication、read/list consistency、multipart/orphan cleanup、Spark/Gravitino read-write、commit failure recovery、source-cluster loss recovery、audit、metrics/alert、backup/restore 和 rollback rehearsal。观测时间不得早于验证时刻 24 小时，expiry 必须在未来且有效期最长七天，evidence URI 必须是非本地 HTTPS；任何 binding 漂移、过期或失败都会关闭门禁。 + +### 3. 对象存储门禁通过也不等于平台生产就绪 + +同一有效 attestation 可派生 object-store decision、protected workload identity、TLS、KMS、tenant isolation、durability、failure recovery 和 production object-store claims。报告中的 `production_ready` 始终固定为 `false`；生产 identity 全链、observability、NetworkPolicy、provider ingestion、持久 binding、cancel/reconcile/lineage、完整 Spark/Flink conformance、upgrade、registry provenance 和其他退出门仍须独立通过。 + +`validate` 只验证 checked-in profile，因此有效的 pending contract 在 CI 中成功；`evaluate` 必须提供 attestation，且仅在对象存储门禁实际通过时成功；`verify` 拒绝 fingerprint 漂移、派生 claim 不一致和 overall production overclaim。本切片不提交真实 attestation、不连接 provider、不创建 bucket/key/policy，也不修改 Kubernetes workload。 + +## Verification + +当前 checked-in profile： + +- profile fingerprint：`668e194b3c688307014148391e7f389c9d6e9ca69c95d7b4cc92b4acae93181a`； +- report fingerprint：`85362dd10b7dc565f9fa567673d90b774cdec714bd1e70fb2c3c83c1af48b5ea`； +- `profile_valid=true` 且无 profile errors； +- 43 项 provider/identity/transport/encryption/durability/consistency/tenancy/operations 外部输入以 blockers 暴露； +- `ready_for_protected_verification=false`、`attestation_valid=false`； +- 全部对象存储生产 claims 与 `production_ready` 固定为 `false`。 + +32 个定向测试覆盖 pending/complete profile、三类允许的 S3-compatible provider、原生非 S3 provider 拒绝、新鲜且完全绑定的合成 attestation、least privilege 与 local evidence drift、static credential/HTTP/public access/same-region recovery、敏感字段、自报 claim、全部 section binding、关键 security/durability/recovery check、runbook version、过期证明、报告篡改、malformed YAML 和生产 overclaim。 + +## Claim Boundary + +允许声明： + +- M3-11 production object-store readiness contract 已建立； +- 当前 pending profile 结构有效，并机器可读地暴露 43 个 blockers； +- 合成完整 profile/attestation 验证了 fail-closed 门禁逻辑。 + +当前不得声明： + +- 已选择、采购、配置、部署或验证 AWS S3、华为云 OBS 或任何生产对象存储； +- 已验证生产 workload identity、TLS/private network、KMS、tenant isolation、versioning/replication、RPO/RTO、commit failure 或 source-cluster loss recovery； +- 已完成 cancel/reconcile/lineage、Flink 或完整 Spark conformance； +- `production_object_store_gate_passed=true`、`production_object_store_verified=true` 或 `production_ready=true`。 + +## Consequences + +**Positive**：本地 MinIO evidence 不再可能被误读为生产 storage；provider-neutral 的 S3-compatible 决策边界、最小权限、failure domain 和受保护证据生命周期可由 CI 与后续 protected runner 一致校验。 + +**Negative**：M3-11 本身不增加生产存储能力。门禁会保持 blocked，直到 Metadata Platform、Data Engineering、Security 和 SRE 完成 provider、账户、区域、identity、network、KMS、replication、tenancy、operations 和 owner 决策。 + +**Next gate**：批准并物化 production profile，在受保护环境部署选定的 S3-compatible 路径，生成绑定当前 source/profile 的真实 attestation并通过全部 26 项检查；随后以该路径执行 commit failure、source-cluster loss、cancel/reconcile/lineage 和完整 Spark/Flink conformance，同时继续独立完成其余 production gates。 diff --git a/docs/roadmap-ar0-platform-truth-2026-07-24.md b/docs/roadmap-ar0-platform-truth-2026-07-24.md index 9b329d33..3853d0f5 100644 --- a/docs/roadmap-ar0-platform-truth-2026-07-24.md +++ b/docs/roadmap-ar0-platform-truth-2026-07-24.md @@ -1,8 +1,8 @@ # GIS Data Agent 下一代 Data Platform Roadmap（AR-0 主线） 日期：2026-07-29 -分支：`feat/ar1-metadata-fabric-spark-object-store-interoperability` -基线：`feat/ar1-metadata-fabric-spark-iceberg-rest-interoperability@f2befab` +分支：`feat/ar1-metadata-fabric-object-store-readiness-gate` +基线：`feat/ar1-metadata-fabric-spark-object-store-interoperability@48d1279` ## 1. 决策摘要 @@ -199,7 +199,7 @@ Temporal 继续保持目标组件状态，不在这一包并行接入。OpenMeta 当前完成仅指本地合同、授权 evidence、outbox/callback 代码、数据库成功终局门、托管 worker 代码、默认关闭的部署模板及离线 activation/release preflight、candidate/registry/provenance/artifact-release/live observation evidence gate、合成 golden slice、定向测试、真实 PostgreSQL 16 事务边界和 canonical mainline 治理。`candidate_validated`、`registry_subject_bound`、本地合成 `provenance_verified`、`ready_for_activation`、`ready_for_staging_apply`、`verified_for_staging_apply` 和本地 live collection 都不等于真实镜像已 attested 或 staging 已部署；真实 IAM/OIDC 与 service token 生命周期、首次 GHCR publish/verify、真实 provenance artifact verify、registry-backed live staging revision、worker/callback 扩容运行、golden slice staging 运行链、受保护 release/live evidence provenance、独立 DolphinScheduler metadata PostgreSQL 和真实数据终局证据仍属于 4.7 后续切片。 -### 4.8 Metadata Fabric Bridge M1 + M2 + M3-10（本地跨节点 Spark/对象存储互操作已验证） +### 4.8 Metadata Fabric Bridge M1 + M2 + M3-11（生产对象存储就绪合同已建立，生产验证待执行） 第八块回到 AR-1 的 metadata control plane，以 [ADR-036](architecture-decisions/adr-036-read-only-metadata-fabric-bridge-contract.md) 固定 OpenMetadata + Gravitino + GDA Control Ledger 的首条 table slice： @@ -230,8 +230,9 @@ Temporal 继续保持目标组件状态，不在这一包并行接入。OpenMeta 25. [ADR-054](architecture-decisions/adr-054-local-gravitino-jdbc-catalog-restart-continuity.md) 已在隔离 namespace 中将 Gravitino Iceberg catalog metadata 落到 PostgreSQL JDBC、warehouse 落到独立 PVC，并复用 M3-6 精确 `USE_CATALOG`/`USE_SCHEMA`/`CREATE_TABLE` 角色。依次重启 PostgreSQL 与 Gravitino 后，两者 Pod UID 均变化而 StatefulSet/PVC UID 保持；同一 bounded user 重新认证并读取相同 table fingerprint，catalog create 前后均为 403，namespace/PV 完整清理。evidence fingerprint 为 `34792bb47ad71041a87adeb644439bf9b6aa3f4855cdc98782d6e3b4282bf1aa`。该结果仅证明 Docker Desktop 单集群、本地 Basic/HTTP/file warehouse 的 restart continuity，不等于 production persistent identity binding、OIDC、TLS、备份恢复、Spark/Flink conformance 或生产 ingestion。 26. [ADR-055](architecture-decisions/adr-055-local-spark-iceberg-rest-interoperability.md) 已将 Gravitino API 与 bundled Iceberg REST `1.11.0` 连接到同一 PostgreSQL JDBC catalog 和 file warehouse PVC，并在 `desktop-worker` 运行 Spark `3.5.0` + Iceberg `1.6.1`。bounded Basic user 先创建零行表且 catalog create 返回 403；Spark 经标准 `/iceberg` REST 读取该表、两次 append、增加 nullable `quality`、验证三行 current state、两个 snapshot 与 first-snapshot time travel；随后 Gravitino API 回读相同演进 schema，catalog create 仍为 403，namespace/PV 完整清理。该结果只证明本地同节点共享 RWO PVC 的 engine interoperability；Spark REST 路径仍是无认证 HTTP，cancel/reconcile/lineage、Flink、对象存储、生产身份/TLS 和完整 `spark_conformance_verified` 均未证明。 27. [ADR-056](architecture-decisions/adr-056-local-spark-object-store-interoperability.md) 已移除 Spark/Gravitino 的共享 warehouse PVC：MinIO 在 `desktop-control-plane`，PostgreSQL、Gravitino 和 Spark 在 `desktop-worker`，两端只经 S3-compatible ClusterIP 共享 `s3://gda-metadata-warehouse/warehouse`。Spark 保持两次 append、schema evolution、三行 current state、双 snapshot 和 first-snapshot time travel；Gravitino 回读演进 schema，直接 MinIO 检查确认 2 个 Parquet、4 个 metadata JSON 和 4 个 Avro manifest，并匹配 table location、schema 与 current snapshot。contract fingerprint 为 `9713cdb3040e1b6532489f329aef7ed7b5266e0757551252f537cb83476b4bee`，evidence fingerprint 为 `05844457efb378581fb7fc2e7ed3c706819b2d8fa5a52b2f82577051d38c2cd1`。该结果只证明同一 Docker Desktop 主机/集群内的跨节点 MinIO 互操作，不证明生产云对象存储、protected identity/TLS、故障注入、cancel/reconcile/lineage、Flink 或完整 Spark conformance。 +28. [ADR-057](architecture-decisions/adr-057-production-object-store-readiness-gate.md) 已将 M3-10 evidence、S3-compatible provider/account/region/bucket、独立 failure domain、OIDC workload federation、精确八项 S3 permission、TLS/private path、KMS、versioning、cross-region replication、strong read/list consistency、tenant isolation、owner/SLO/runbook 和 26 项 protected attestation check 冻结为 fail-closed profile。当前 profile fingerprint 为 `668e194b3c688307014148391e7f389c9d6e9ca69c95d7b4cc92b4acae93181a`，report fingerprint 为 `85362dd10b7dc565f9fa567673d90b774cdec714bd1e70fb2c3c83c1af48b5ea`，合同有效但 43 项生产输入仍 blocked，全部 production claims 为 `false`。这只是 provider-neutral 决策和验收合同：没有选择、部署或验证 AWS S3、华为云 OBS 或其他生产对象存储；原生非 S3 provider 必须进入新的 conformance slice。 -此处 M1 只证明静态合同和只读 HTTP 边界；M2a 只证明本地 live foundation 与 PVC 重挂载连续性；M2b-1/M2b-2 分别限定在同集群新 PVC 和同集群隔离 repository；M2b-3 的 `local_cross_cluster_recovery_verified=true` 只限定在 `local_same_host_distinct_kubernetes_clusters_external_s3_repository`；M2c-1/M2c-2/M2c-3 分别限定本地 provider metrics、临时双周期 OTel 和单 job scrape recovery；M2c-4/M2d-2 只证明 production observability/NetworkPolicy profile 与 attestation 合同可校验；M2d-1 只证明本地两节点 kindnet 的隔离合成流量；M3-1 的 terminal evidence 与 M3-2 的 PolicyDecision/Approval 仍是 deterministic local fixtures。M3-2 只把 projection 写入本地 provider 并证明 retained target 的单次零写入 replay；M3-3 只把该本地 evidence 对应的 binding 写入临时 GDA Control 账本；M3-4 只向无认证 loopback receiver 发送精确 candidate 并验证 503 后幂等恢复；M3-5 只证明 OpenMetadata 在 provider 强制默认 role 之上的项目新增 grant 限定为 `table/Create`，以及本地 JWT 轮换/吊销和越权拒绝；M3-6 只证明隔离 Gravitino Basic IdP 的 bounded table-create、catalog-create 拒绝、登录轮换/吊销和完整清理；M3-7 只证明 pending production identity profile、profile-bound attestation 和派生 claim 的 fail-closed 合同可校验，没有部署或证明真实身份路径；M3-8 只证明同一 Docker Desktop 集群内 Basic 用户、JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 file warehouse PVC 的 Spark interoperability；M3-10 移除了该共享 PVC，并证明同一 Docker Desktop 主机/集群内 Spark 与 MinIO 的跨节点 S3-compatible 互操作，但不证明生产云对象存储、独立 failure domain、持久 identity binding、Flink 或完整 engine conformance。M3-2 ingestion 仍使用 bootstrap admin，生产持久 binding、ResourceVersion 和 legacy authority 都未写入；双 provider/生产最小权限、protected workload identity、OIDC、TLS、生产持久 catalog、tenant isolation、真实 receiver/alert/SLO、受保护 provider policy、生产 OpenLineage、生产 ingest/conformance、三项 production gate 和 `production_ready` 仍为 `false`。 +此处 M1 只证明静态合同和只读 HTTP 边界；M2a 只证明本地 live foundation 与 PVC 重挂载连续性；M2b-1/M2b-2 分别限定在同集群新 PVC 和同集群隔离 repository；M2b-3 的 `local_cross_cluster_recovery_verified=true` 只限定在 `local_same_host_distinct_kubernetes_clusters_external_s3_repository`；M2c-1/M2c-2/M2c-3 分别限定本地 provider metrics、临时双周期 OTel 和单 job scrape recovery；M2c-4/M2d-2 只证明 production observability/NetworkPolicy profile 与 attestation 合同可校验；M2d-1 只证明本地两节点 kindnet 的隔离合成流量；M3-1 的 terminal evidence 与 M3-2 的 PolicyDecision/Approval 仍是 deterministic local fixtures。M3-2 只把 projection 写入本地 provider 并证明 retained target 的单次零写入 replay；M3-3 只把该本地 evidence 对应的 binding 写入临时 GDA Control 账本；M3-4 只向无认证 loopback receiver 发送精确 candidate 并验证 503 后幂等恢复；M3-5 只证明 OpenMetadata 在 provider 强制默认 role 之上的项目新增 grant 限定为 `table/Create`，以及本地 JWT 轮换/吊销和越权拒绝；M3-6 只证明隔离 Gravitino Basic IdP 的 bounded table-create、catalog-create 拒绝、登录轮换/吊销和完整清理；M3-7 只证明 pending production identity profile、profile-bound attestation 和派生 claim 的 fail-closed 合同可校验，没有部署或证明真实身份路径；M3-8 只证明同一 Docker Desktop 集群内 Basic 用户、JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 file warehouse PVC 的 Spark interoperability；M3-10 移除了该共享 PVC，并证明同一 Docker Desktop 主机/集群内 Spark 与 MinIO 的跨节点 S3-compatible 互操作，但不证明生产云对象存储、独立 failure domain、持久 identity binding、Flink 或完整 engine conformance；M3-11 只冻结 provider-neutral production object-store profile、精确 attestation binding 与 fail-closed claims，没有选择 provider、部署 bucket/KMS/policy 或提交真实 attestation。M3-2 ingestion 仍使用 bootstrap admin，生产持久 binding、ResourceVersion 和 legacy authority 都未写入；生产对象存储、双 provider/生产最小权限、protected workload identity、OIDC、TLS、生产持久 catalog、tenant isolation、真实 receiver/alert/SLO、受保护 provider policy、commit failure、source-loss recovery、cancel/reconcile/lineage、完整 Spark/Flink conformance、生产 ingest、四项 production gate 和 `production_ready` 仍为 `false`。 ## 5. 重新评估条件 diff --git a/docs/system-of-record-matrix-2026-07-24.md b/docs/system-of-record-matrix-2026-07-24.md index 44e104ec..58c00f6f 100644 --- a/docs/system-of-record-matrix-2026-07-24.md +++ b/docs/system-of-record-matrix-2026-07-24.md @@ -2,9 +2,9 @@ 日期：2026-07-29 -阶段：AR-0 `in_progress`；AR-1 gateway、成功终局 evidence gate、DolphinScheduler adapter sandbox POC、Metadata Fabric M1/M2、M2c-4/M2d-2 production readiness contracts、M3-1/M3-2、M3-3 local binding ledger、M3-4 local OpenLineage wire delivery、M3-5 local OpenMetadata bounded identity、M3-6 local Gravitino Basic bounded identity、M3-7 production identity readiness contract、M3-8 local Gravitino JDBC restart continuity、M3-9 local Spark/Iceberg REST interoperability 与 M3-10 local cross-node Spark/object-store interoperability 已验证，生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity attestation 和生产切换仍 `in_progress` +阶段：AR-0 `in_progress`；AR-1 gateway、成功终局 evidence gate、DolphinScheduler adapter sandbox POC、Metadata Fabric M1/M2、M2c-4/M2d-2 production readiness contracts、M3-1/M3-2、M3-3 local binding ledger、M3-4 local OpenLineage wire delivery、M3-5 local OpenMetadata bounded identity、M3-6 local Gravitino Basic bounded identity、M3-7 production identity readiness contract、M3-8 local Gravitino JDBC restart continuity、M3-9 local Spark/Iceberg REST interoperability、M3-10 local cross-node Spark/object-store interoperability 与 M3-11 production object-store readiness contract 已验证，生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity/object-store attestation 和生产切换仍 `in_progress` -适用分支：`feat/ar1-metadata-fabric-spark-object-store-interoperability` +适用分支：`feat/ar1-metadata-fabric-object-store-readiness-gate` ## 判定规则 @@ -25,7 +25,7 @@ | 湖仓表与 snapshot | Iceberg/STAC/S3A 有局部实现，尚无通用发布权威 | STAC item、GeoParquet export | Iceberg catalog snapshot 是分析表版本权威；对象是物理内容，STAC 是发现投影 | Data Platform | AR-2 | | 在线空间数据 | PostGIS 业务表是当前编辑/查询事实，部分临时表混入 | Martin MVT、API JSON、导出文件 | 已批准 DataProductVersion 物化到 PostGIS；不能由瓦片或临时表反向定义产品版本 | GIS/Data Platform | AR-2 -> AR-4 | | 数据资产身份与版本 | `gda_control.resource/resource_version` 已实现 identity、hash、predecessor、tenant FK 和幂等 gateway 写入；`agent_data_assets`、`agent_asset_versions` 仍是兼容写路径 | UI catalog、search index、STAC | GDA ledger 管身份与版本绑定；旧行只有在 tenant、authority identity、checksum 和 version evidence 完整时才可形成 eligible plan；OpenMetadata 管治理目录，Gravitino 管技术对象映射 | Metadata Platform | AR-1 gateway 已验证 -> 生产切换待验收 | -| 技术元数据 | M1 已冻结 Gravitino table ref/reconciliation；M2 已验证本地 foundation/recovery/metrics/policy 和 production readiness contracts；M3-1 固定 technical projection intent，M3-2 已在 Gravitino memory catalog 创建/read-back，M3-3 将验证后的 ref 追加到 tenant-scoped 本地 binding ledger；M3-6 又在隔离 Gravitino Basic IdP 中验证 bounded table-create、catalog-create 拒绝、密码轮换/用户吊销和完整清理；M3-7 已冻结 production identity profile/attestation gate；M3-8 已验证同一 Basic role 与 Iceberg JDBC table 在 PostgreSQL/Gravitino Pod restart 后保持；M3-9 已验证 Spark 经标准 Iceberg REST 对同一 JDBC catalog 做 read/write/schema evolution/snapshot/time travel；M3-10 已移除共享 warehouse PVC，并由跨节点 MinIO 对象检查与 Gravitino API 回读验证 Spark 结果 | harvester 结果、合成 response、本地 sandbox/recovery/metrics/policy/ingestion/identity/JDBC restart/Spark/object-store interoperability observation、projection plan、provider evidence、binding ledger 与 readiness report | 源系统技术对象是原始证据；Gravitino 映射并联邦，不能覆盖业务 ResourceVersion；GDA binding ledger 只记录已验证关系，本地 memory/JDBC/file-PVC/MinIO catalog evidence 不得冒充生产持久技术权威；Basic IdP、无认证 REST/HTTP、本地主机对象存储、pending profile 和合成 attestation 都不是生产身份或生产 storage | Metadata Platform | AR-1 M1/M2 + M3-6 identity + M3-7 gate + M3-8 local persistence + M3-9/M3-10 local Spark interoperability 已验证 -> 受保护身份/生产对象存储/完整 Spark-Flink conformance 待执行 | +| 技术元数据 | M1 已冻结 Gravitino table ref/reconciliation；M2 已验证本地 foundation/recovery/metrics/policy 和 production readiness contracts；M3-1 固定 technical projection intent，M3-2 已在 Gravitino memory catalog 创建/read-back，M3-3 将验证后的 ref 追加到 tenant-scoped 本地 binding ledger；M3-6 又在隔离 Gravitino Basic IdP 中验证 bounded table-create、catalog-create 拒绝、密码轮换/用户吊销和完整清理；M3-7 已冻结 production identity profile/attestation gate；M3-8 已验证同一 Basic role 与 Iceberg JDBC table 在 PostgreSQL/Gravitino Pod restart 后保持；M3-9 已验证 Spark 经标准 Iceberg REST 对同一 JDBC catalog 做 read/write/schema evolution/snapshot/time travel；M3-10 已移除共享 warehouse PVC，并由跨节点 MinIO 对象检查与 Gravitino API 回读验证 Spark 结果；M3-11 已冻结 provider-neutral production object-store profile/attestation gate | harvester 结果、合成 response、本地 sandbox/recovery/metrics/policy/ingestion/identity/JDBC restart/Spark/object-store interoperability observation、projection plan、provider evidence、binding ledger 与 readiness report | 源系统技术对象是原始证据；Gravitino 映射并联邦，不能覆盖业务 ResourceVersion；GDA binding ledger 只记录已验证关系，本地 memory/JDBC/file-PVC/MinIO catalog evidence 不得冒充生产持久技术权威；Basic IdP、无认证 REST/HTTP、本地主机对象存储、pending profile 和合成 attestation 都不是生产身份或生产 storage，M3-11 合同不构成 provider selection/deployment | Metadata Platform | AR-1 M1/M2 + M3-6 identity + M3-7 gate + M3-8 local persistence + M3-9/M3-10 local Spark interoperability + M3-11 object-store gate 已验证 -> 受保护身份/生产对象存储 attestation/完整 Spark-Flink conformance 待执行 | | 治理目录 | M1 已冻结 OpenMetadata table ref/reconciliation；M2 已验证本地 foundation/recovery/metrics/policy 和 production readiness contracts；M3-2 已用 bootstrap admin 创建目标并回读真实 UUID；M3-3 将该 UUID 经 evidence gate 追加到本地 GDA binding ledger；M3-4 将精确 OpenLineage candidate 经 outbox 投递到本地 HTTP receiver；M3-5 已验证临时非管理员 bot 的 scoped `table/Create` grant、policy-create 拒绝及 JWT 轮换/吊销；M3-7 将其 allow/deny 范围纳入双 provider production identity gate | 搜索/页面视图、合成 response、本地 sandbox/recovery/metrics/policy/ingestion/identity observation、projection/provider evidence、binding ledger、lineage outbox/receipt、OpenLineage event 与 readiness report | OpenMetadata 为 owner/glossary/classification/quality discoverability 权威；GDA ledger 保留审批/provider identity，outbox 只拥有投递状态，receiver 拥有接收状态；pending identity profile 和合成 attestation 均不反写 ResourceVersion 或建立生产权威 | Governance | AR-1 M1/M2 + M3-5 local bounded identity + M3-7 readiness contract 已验证 -> protected identity ingestion/生产持久 binding/受保护 production receiver 待执行 | | 血缘 | `gda_control.lineage_event` 已实现 immutable version edge 和幂等 gateway ingest；`agent_asset_lineage` 旧记录仍是可变 asset edge | OpenMetadata lineage graph、UI DAG | 只有 source/target ResourceVersion 与 event checksum 证据完整的旧记录可形成 eligible plan；目录图只作可重建投影 | Data Platform | AR-1 gateway 已验证 -> adapter 待接入 | | Definition | `gda_control.platform_definition_version` 已绑定 definition ResourceVersion、完整逻辑 hash 和原子 gateway registration；3.4.2 adapter 可编译、创建并上线 provider DAG；binding 已以 append-only `execution_plan` Artifact 持久化并可按 tenant + artifact UUID 读取，旧 workflow/template/YAML 仍在写入 | 编辑器状态、DolphinScheduler DAG/definition | 旧 workflow 必须规范化并完整 hash 后才可形成 PlatformDefinitionVersion；provider binding 作为 ExecutionPlanArtifact/evidence，不可反写 definition | DataOps | AR-1 binding persistence 代码已验证 -> staging 调用链待验收 | @@ -56,7 +56,7 @@ 11. `platform_command_outbox` 只拥有投递状态；callback 只触发 reconcile，不能把 provider payload 直接写成 PlatformRun 状态或平台终局。 12. QualityResult evaluator 必须是 workload，且成功终局中的 evaluator 不能等于 Run workload；该代码级职责分离不替代生产 IAM。 13. `candidate_validated`、`registry_subject_bound`、GitHub provenance action 成功、CI artifact、离线 preflight、未独立 attested 的 live observation JSON 或人工批准都不能单独授权 production；缺少同一 source revision 的 OCI subject 独立验证、registry/live revision/identity/health/golden-slice 绑定及受保护 provenance 时，promotion 必须失败。 -14. Metadata Fabric M1 只允许 OpenMetadata/Gravitino GET；M2 只执行本地 foundation/recovery/metrics/policy 演练或验证 production readiness profile；M3-1 只从 synthetic terminal evidence 生成 plan/candidate；M3-2 只允许 exact local PolicyDecision/Approval 后向本地 provider 写 projection；M3-3 只将同一 source evidence 经 PlatformGateway 写入临时 append-only binding ledger；M3-4 只经 tenant-scoped outbox 向无认证 loopback receiver 投递精确 candidate，并验证 at-least-once + receiver idempotency；M3-5 只证明临时 OpenMetadata bot 在 provider 强制 `DefaultBotRole` 之上的项目新增 grant 是 `table/Create`，并验证 policy-create 拒绝与本地 JWT 轮换/吊销；M3-6 只证明隔离 Gravitino Basic user 的 bounded table-create、catalog-create 拒绝、密码轮换/用户吊销和完整清理；M3-7 只冻结 production identity profile、精确 attestation binding 与 fail-closed 派生 claims，既不部署 identity path，也不提交真实 production attestation；M3-8 只证明 Docker Desktop 单集群中 Basic role、PostgreSQL JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 RWO PVC 的 Spark/Iceberg REST 互操作；M3-10 移除 Spark/Gravitino 共享 warehouse PVC，且只证明同一 Docker Desktop 主机/集群内跨节点 MinIO 的 read/write/schema evolution/snapshot/time travel 与对象级 metadata 一致，不覆盖生产对象存储、failure injection、cancel/reconcile/lineage 或 Flink。Gravitino `1.3.0` Basic IdP 不算 OIDC，生产必须明确选择并证明 custom OIDC authenticator 或 identity-aware proxy。本地 bootstrap provisioner、Basic IdP、loopback/cluster HTTP、memory/file-backed JDBC catalog、同节点 PVC/MinIO、临时 identity/ledger/outbox、loopback receiver、pending profile、合成 attestation 和 local evidence 都不等于双 provider/生产最小权限、protected workload identity/OIDC、生产持久 catalog/binding、TLS、受保护 OpenLineage receiver、tenant isolation、alert/SLO、生产 ingestion/conformance 或生产写权威。 +14. Metadata Fabric M1 只允许 OpenMetadata/Gravitino GET；M2 只执行本地 foundation/recovery/metrics/policy 演练或验证 production readiness profile；M3-1 只从 synthetic terminal evidence 生成 plan/candidate；M3-2 只允许 exact local PolicyDecision/Approval 后向本地 provider 写 projection；M3-3 只将同一 source evidence 经 PlatformGateway 写入临时 append-only binding ledger；M3-4 只经 tenant-scoped outbox 向无认证 loopback receiver 投递精确 candidate，并验证 at-least-once + receiver idempotency；M3-5 只证明临时 OpenMetadata bot 在 provider 强制 `DefaultBotRole` 之上的项目新增 grant 是 `table/Create`，并验证 policy-create 拒绝与本地 JWT 轮换/吊销；M3-6 只证明隔离 Gravitino Basic user 的 bounded table-create、catalog-create 拒绝、密码轮换/用户吊销和完整清理；M3-7 只冻结 production identity profile、精确 attestation binding 与 fail-closed 派生 claims，既不部署 identity path，也不提交真实 production attestation；M3-8 只证明 Docker Desktop 单集群中 Basic role、PostgreSQL JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 RWO PVC 的 Spark/Iceberg REST 互操作；M3-10 移除 Spark/Gravitino 共享 warehouse PVC，且只证明同一 Docker Desktop 主机/集群内跨节点 MinIO 的 read/write/schema evolution/snapshot/time travel 与对象级 metadata 一致；M3-11 只冻结 S3-compatible production profile、精确 attestation binding 与 fail-closed claims，既不选择/部署 provider，也不创建 bucket/KMS/policy 或提交真实 attestation。生产对象存储、identity/TLS、failure injection、source-loss recovery、cancel/reconcile/lineage 与 Flink 仍未证明。Gravitino `1.3.0` Basic IdP 不算 OIDC，生产必须明确选择并证明 custom OIDC authenticator 或 identity-aware proxy。本地 bootstrap provisioner、Basic IdP、loopback/cluster HTTP、memory/file-backed JDBC catalog、同节点 PVC/MinIO、临时 identity/ledger/outbox、loopback receiver、pending profile、合成 attestation 和 local evidence 都不等于双 provider/生产最小权限、protected workload identity/OIDC、生产持久 catalog/binding、TLS、受保护 OpenLineage receiver、tenant isolation、alert/SLO、生产 ingestion/conformance 或生产写权威。 ## 已建立的 AR-0/AR-1 entry 证据 @@ -90,6 +90,7 @@ - Metadata Fabric M3-8 已在隔离 namespace 将 Iceberg catalog metadata 落到 PostgreSQL JDBC、warehouse 落到独立 PVC；PostgreSQL/Gravitino Pod UID 在顺序 restart 后均变化，StatefulSet/PVC UID 保持，同一 bounded Basic user 重认证并读取相同 table fingerprint，catalog create 前后均为 403，临时 namespace/PV 清理完成。contract fingerprint 为 `f622d8a61bae49171bc76a16bfe64280c616c028bddf88479f1ad04acb1dadf0`，evidence fingerprint 为 `34792bb47ad71041a87adeb644439bf9b6aa3f4855cdc98782d6e3b4282bf1aa`。这不证明 protected workload identity、OIDC/TLS、生产 failure domain、backup/PITR、Spark/Flink conformance、生产 ingestion 或 `production_ready`。 - Metadata Fabric M3-9 已在隔离 namespace 让 bounded Gravitino Basic user 先创建 Iceberg JDBC table，再由 Spark `3.5.0` + Iceberg `1.6.1` 经 Gravitino Iceberg REST `1.11.0` 读取、两次 append、增加 nullable column、验证三行 current state、两个 snapshot 与 first-snapshot time travel；Gravitino API 随后回读演进 schema，catalog create 前后均为 403，Spark/Gravitino 共用 `desktop-worker` 上的 warehouse PVC，namespace/PV 完整清理。contract fingerprint 为 `a78b95d36a6a5d5f4b5e303be21263d00fdd7102c3a70ce282ba69f2d8cdcd2e`，evidence fingerprint 为 `50f9d0021db11e22364697d1ad8928ee068d28dc8046556bbca1a4e1c819f8e0`。这只证明本地同节点共享 RWO PVC 互操作；无认证 HTTP REST、Basic IdP、file warehouse 不证明 protected identity、OIDC/TLS、对象存储、cancel/reconcile/lineage、Flink、完整 `spark_conformance_verified`、生产 ingestion 或 `production_ready`。 - Metadata Fabric M3-10 已将 warehouse 切换为 MinIO S3 API：MinIO 位于 `desktop-control-plane`，PostgreSQL、Gravitino 与 Spark 位于 `desktop-worker`，Spark/Gravitino 均无 warehouse PVC。Spark 保持两次 append、nullable `quality` 演进、三行 current state、双 snapshot 与 first-snapshot time travel；Gravitino 回读演进 schema，直接 MinIO 检查确认 2 个 Parquet、4 个 metadata JSON、4 个 Avro manifest，最新 location/schema/current snapshot 与 Spark 一致，catalog create 前后仍为 403，namespace、两块 PV 和 port-forward 均清理。contract fingerprint 为 `9713cdb3040e1b6532489f329aef7ed7b5266e0757551252f537cb83476b4bee`，evidence fingerprint 为 `05844457efb378581fb7fc2e7ed3c706819b2d8fa5a52b2f82577051d38c2cd1`。这只证明本地同主机/同集群跨节点 S3-compatible interoperability；生产云对象存储、protected identity/OIDC/TLS、故障注入、cancel/reconcile/lineage、Flink、完整 `spark_conformance_verified`、生产 ingestion 与 `production_ready` 仍未证明。 +- Metadata Fabric M3-11 已建立 production object-store profile/attestation gate；checked-in profile fingerprint 为 `668e194b3c688307014148391e7f389c9d6e9ca69c95d7b4cc92b4acae93181a`，report fingerprint 为 `85362dd10b7dc565f9fa567673d90b774cdec714bd1e70fb2c3c83c1af48b5ea`，`profile_valid=true`，43 项 provider/identity/transport/encryption/durability/consistency/tenancy/operations 外部输入以 blockers 暴露，`ready_for_protected_verification=false`、`attestation_valid=false`、`production_object_store_gate_passed=false`、`production_ready=false`。该合同绑定 M3-10 evidence，但没有选择或部署 provider；合成完整 attestation 只验证门禁逻辑，不计入生产证据。下一项真实证据是经 owner 批准并物化的 provider profile，以及来自 `production-object-store` 受保护环境、绑定当前 source/profile 并通过全部 26 项检查的 attestation。 ## 下一验收证据 @@ -98,5 +99,5 @@ - staging 的 migration role、应用 login membership、连接池 role/tenant 复位、双租户 API 和 success finalization 运行产物； - DolphinScheduler adapter 的真实 IAM/OIDC、service token provisioning/轮换、provider 最小权限、binding artifact staging 接入、managed outbox worker/provider callback 实际扩容部署、唯一 worker ID、status/lease 故障恢复和无双写证据； - 首条真实图斑链对 golden slice 的 output hash、独立质量结果/evidence、血缘、发布 revision 和 rollback 演练； -- OpenMetadata/Gravitino 的 source host/cluster 外生产 backup account/bucket、生产对象存储、KMS/TLS/workload identity、PITR/source-loss recovery、RPO/RTO、OIDC、受保护环境 provider NetworkPolicy/tenant isolation、upgrade/rollback、registry provenance、持续 metrics backend/retention/query、真实 alert delivery/SLO owner/runbook，以及受保护 PolicyDecision/Approval、双 provider 最小权限 ingestion、生产持久 binding、受保护 production OpenLineage receiver、无双写 read-back、commit failure injection、cancel/reconcile/lineage 和完整 Spark/Flink conformance；M1 fixture、M2 本地 evidence/readiness contracts、M3-1 projection candidate、M3-2 local replay、M3-3 临时 binding ledger、M3-4 loopback delivery、M3-5/M3-6 本地临时 provider identity、M3-7 pending profile/合成 attestation、M3-8 本地 JDBC restart continuity、M3-9 本地同节点 Spark interoperability 与 M3-10 本地同主机跨节点 MinIO interoperability 均不计入生产退出门； +- OpenMetadata/Gravitino 的 source host/cluster 外生产 backup account/bucket、已批准 provider profile 与受保护对象存储 attestation、生产对象存储、KMS/TLS/workload identity、PITR/source-loss recovery、RPO/RTO、OIDC、受保护环境 provider NetworkPolicy/tenant isolation、upgrade/rollback、registry provenance、持续 metrics backend/retention/query、真实 alert delivery/SLO owner/runbook，以及受保护 PolicyDecision/Approval、双 provider 最小权限 ingestion、生产持久 binding、受保护 production OpenLineage receiver、无双写 read-back、commit failure injection、cancel/reconcile/lineage 和完整 Spark/Flink conformance；M1 fixture、M2 本地 evidence/readiness contracts、M3-1 projection candidate、M3-2 local replay、M3-3 临时 binding ledger、M3-4 loopback delivery、M3-5/M3-6 本地临时 provider identity、M3-7 pending profile/合成 attestation、M3-8 本地 JDBC restart continuity、M3-9 本地同节点 Spark interoperability、M3-10 本地同主机跨节点 MinIO interoperability 与 M3-11 pending object-store profile/合成 attestation 均不计入生产退出门； - DolphinScheduler/Temporal sandbox 的独立数据库、备份恢复、身份、版本和升级责任证明；DolphinScheduler standalone/H2 不计入此退出门。 diff --git a/scripts/metadata-fabric-object-store-gate.sh b/scripts/metadata-fabric-object-store-gate.sh new file mode 100755 index 00000000..bf0cf5b3 --- /dev/null +++ b/scripts/metadata-fabric-object-store-gate.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +COMMON_GIT_DIR="$(git -C "$ROOT" rev-parse --path-format=absolute --git-common-dir 2>/dev/null || true)" +SHARED_ROOT="" +if [ -n "$COMMON_GIT_DIR" ]; then + SHARED_ROOT="$(cd "$COMMON_GIT_DIR/.." && pwd)" +fi + +if [ -n "${PYTHON:-}" ]; then + : +elif [ -x "$ROOT/.venv/bin/python" ]; then + PYTHON="$ROOT/.venv/bin/python" +elif [ -n "$SHARED_ROOT" ] && [ -x "$SHARED_ROOT/.venv/bin/python" ]; then + PYTHON="$SHARED_ROOT/.venv/bin/python" +else + PYTHON="python" +fi + +cd "$ROOT" +exec "$PYTHON" -m data_agent.metadata_fabric_object_store_gate "$@"