#!/usr/bin/env python3 """Atomically configure and attest the Phase 1 publisher without printing secrets.""" from __future__ import annotations import argparse import json import os import re import shlex import tempfile from pathlib import Path CONFIG_KEYS = ( "RB_INTEGRATION_CAPTURE_ENABLED", "RB_INTEGRATION_PUBLISH_ENABLED", "RB_INTEGRATION_ACTIVATION_APPROVED", "RB_INTEGRATION_PHASE", "RB_INTEGRATION_REVERSE_DELIVERY_ENABLED", "RB_INTEGRATION_DEPLOYMENT_ID", "RB_INTEGRATION_SOURCE_SCOPE", "RB_INTEGRATION_SCHEMA_VERSION", "RB_INTEGRATION_TRANSPORT", "RB_INTEGRATION_KAFKA_TOPIC", "RB_INTEGRATION_KAFKA_BOOTSTRAP_SERVERS", "RB_INTEGRATION_KAFKA_SECURITY_PROTOCOL", "RB_INTEGRATION_KAFKA_SASL_MECHANISM", "RB_INTEGRATION_KAFKA_SASL_USERNAME", "RB_INTEGRATION_KAFKA_SASL_PASSWORD", "RB_INTEGRATION_KAFKA_SSL_CA_LOCATION", "RB_INTEGRATION_SNAPSHOT_SERVICE_TOKEN", "RB_INTEGRATION_PUBLISHER_CONCURRENCY", "RB_CONFIRMED_SCHEDULE_ENABLED", ) SECRET_KEYS = { "RB_INTEGRATION_KAFKA_SASL_USERNAME", "RB_INTEGRATION_KAFKA_SASL_PASSWORD", "RB_INTEGRATION_SNAPSHOT_SERVICE_TOKEN", } DEFAULTS = { "RB_INTEGRATION_CAPTURE_ENABLED": "True", "RB_INTEGRATION_PUBLISH_ENABLED": "False", "RB_INTEGRATION_ACTIVATION_APPROVED": "False", "RB_INTEGRATION_PHASE": "phase1", "RB_INTEGRATION_REVERSE_DELIVERY_ENABLED": "False", "RB_INTEGRATION_DEPLOYMENT_ID": "default", "RB_INTEGRATION_SOURCE_SCOPE": "default", "RB_INTEGRATION_SCHEMA_VERSION": "2", "RB_INTEGRATION_TRANSPORT": "disabled", "RB_INTEGRATION_KAFKA_TOPIC": "", "RB_INTEGRATION_KAFKA_BOOTSTRAP_SERVERS": "", "RB_INTEGRATION_KAFKA_SECURITY_PROTOCOL": "PLAINTEXT", "RB_INTEGRATION_KAFKA_SASL_MECHANISM": "", "RB_INTEGRATION_KAFKA_SASL_USERNAME": "", "RB_INTEGRATION_KAFKA_SASL_PASSWORD": "", "RB_INTEGRATION_KAFKA_SSL_CA_LOCATION": "", "RB_INTEGRATION_SNAPSHOT_SERVICE_TOKEN": "", "RB_INTEGRATION_PUBLISHER_CONCURRENCY": "1", "RB_CONFIRMED_SCHEDULE_ENABLED": "False", } def _boolean(value: str, *, key: str) -> bool: normalized = str(value).strip().lower() if normalized in {"1", "true", "yes", "on"}: return True if normalized in {"0", "false", "no", "off"}: return False raise ValueError(f"{key} must be a boolean") def requested_configuration(environment=os.environ) -> dict[str, str]: values = {} for key in CONFIG_KEYS: value = str(environment.get(key, DEFAULTS[key])) values[key] = value if key in SECRET_KEYS else value.strip() for key, value in values.items(): if "\n" in value or "\r" in value or "\x00" in value: raise ValueError(f"{key} contains an invalid control character") return values def validate(values: dict[str, str]) -> dict[str, bool | str]: capture = _boolean(values["RB_INTEGRATION_CAPTURE_ENABLED"], key="capture") publish = _boolean(values["RB_INTEGRATION_PUBLISH_ENABLED"], key="publish") approved = _boolean(values["RB_INTEGRATION_ACTIVATION_APPROVED"], key="approval") reverse = _boolean( values["RB_INTEGRATION_REVERSE_DELIVERY_ENABLED"], key="reverse delivery" ) confirmed = _boolean(values["RB_CONFIRMED_SCHEDULE_ENABLED"], key="confirmed schedule") if not capture: raise ValueError("Phase 1 deployment requires capture enabled") if values["RB_INTEGRATION_PHASE"] != "phase1": raise ValueError("Phase 1 deployment requires RB_INTEGRATION_PHASE=phase1") if reverse: raise ValueError("Phase 2 reverse delivery must remain disabled") if confirmed: raise ValueError("confirmed_schedule must remain disabled for this gate") if values["RB_INTEGRATION_SCHEMA_VERSION"] != "2": raise ValueError("Phase 1 deployment requires canonical schema version 2") if not values["RB_INTEGRATION_DEPLOYMENT_ID"]: raise ValueError("deployment identity is required") if not values["RB_INTEGRATION_SOURCE_SCOPE"]: raise ValueError("source scope is required") if values["RB_INTEGRATION_PUBLISHER_CONCURRENCY"] != "1": raise ValueError("the ordered publisher requires concurrency 1") if publish != approved: raise ValueError("publication and activation approval must be enabled/disabled together") protocol = values["RB_INTEGRATION_KAFKA_SECURITY_PROTOCOL"].upper() if protocol not in {"PLAINTEXT", "SSL", "SASL_PLAINTEXT", "SASL_SSL"}: raise ValueError("unsupported Kafka security protocol") if publish: required = ( "RB_INTEGRATION_KAFKA_TOPIC", "RB_INTEGRATION_KAFKA_BOOTSTRAP_SERVERS", "RB_INTEGRATION_SNAPSHOT_SERVICE_TOKEN", ) missing = [key for key in required if not values[key]] if values["RB_INTEGRATION_TRANSPORT"] != "kafka": missing.append("RB_INTEGRATION_TRANSPORT=kafka") if protocol.startswith("SASL"): for key in ( "RB_INTEGRATION_KAFKA_SASL_MECHANISM", "RB_INTEGRATION_KAFKA_SASL_USERNAME", "RB_INTEGRATION_KAFKA_SASL_PASSWORD", ): if not values[key]: missing.append(key) if missing: raise ValueError("enabled publisher is missing: " + ", ".join(missing)) return { "capture_enabled": capture, "publish_enabled": publish, "activation_approved": approved, "phase": values["RB_INTEGRATION_PHASE"], "reverse_delivery_enabled": reverse, "deployment_id": values["RB_INTEGRATION_DEPLOYMENT_ID"], "source_scope": values["RB_INTEGRATION_SOURCE_SCOPE"], "schema_version": values["RB_INTEGRATION_SCHEMA_VERSION"], "transport": values["RB_INTEGRATION_TRANSPORT"], "topic_configured": bool(values["RB_INTEGRATION_KAFKA_TOPIC"]), "broker_configured": bool(values["RB_INTEGRATION_KAFKA_BOOTSTRAP_SERVERS"]), "security_protocol": protocol, "sasl_mechanism": values["RB_INTEGRATION_KAFKA_SASL_MECHANISM"] or None, "sasl_identity_configured": bool(values["RB_INTEGRATION_KAFKA_SASL_USERNAME"]), "ssl_ca_configured": bool(values["RB_INTEGRATION_KAFKA_SSL_CA_LOCATION"]), "snapshot_token_configured": bool(values["RB_INTEGRATION_SNAPSHOT_SERVICE_TOKEN"]), "publisher_concurrency": values["RB_INTEGRATION_PUBLISHER_CONCURRENCY"], } def update_environment_file(path: Path, values: dict[str, str]) -> None: path = path.resolve() path.parent.mkdir(parents=True, exist_ok=True) lines = path.read_text().splitlines() if path.exists() else [] key_pattern = re.compile(r"^\s*(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)=") remaining = dict(values) output = [] for line in lines: match = key_pattern.match(line) key = match.group(1) if match else None if key in remaining: output.append(f"{key}={shlex.quote(remaining.pop(key))}") else: output.append(line) if output and output[-1] != "": output.append("") output.extend(f"{key}={shlex.quote(remaining[key])}" for key in values if key in remaining) content = "\n".join(output) + "\n" with tempfile.NamedTemporaryFile( "w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False ) as handle: handle.write(content) temporary = Path(handle.name) temporary.chmod(0o600) os.replace(temporary, path) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--env-file", type=Path) parser.add_argument("--check-only", action="store_true") parser.add_argument("--disable-only", action="store_true") options = parser.parse_args() if options.disable_only: if not options.env_file: parser.error("--disable-only requires --env-file") update_environment_file( options.env_file, { "RB_INTEGRATION_PUBLISH_ENABLED": "False", "RB_INTEGRATION_ACTIVATION_APPROVED": "False", "RB_INTEGRATION_REVERSE_DELIVERY_ENABLED": "False", }, ) print(json.dumps({"publish_enabled": False, "activation_approved": False})) return 0 values = requested_configuration() attestation = validate(values) if not options.check_only: if not options.env_file: parser.error("--env-file is required unless --check-only is used") update_environment_file(options.env_file, values) print(json.dumps(attestation, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())