diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6f5b730..04acef3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,6 +44,7 @@ jobs: ops/apply_gcp_iam_split.py \ ops/check_gcp_infra_readiness.py \ ops/apply_gcp_runtime_baseline.py \ + ops/check_gcp_service_communications.py \ ops/plan_gcp_iam_split.py \ ops/sqlite_to_postgres_dump.py \ ops/verify_gcp_cloudsql_restore_readback.py \ @@ -57,6 +58,7 @@ jobs: tests/test_evaluate_agent_routing.py \ tests/test_gcp_artifact_workflow.py \ tests/test_gcp_runtime_baseline_apply.py \ + tests/test_gcp_service_communications.py \ tests/test_gcp_cloudsql_restore_drill.py \ tests/test_gcp_cloudsql_restore_readback.py \ tests/test_gcp_iam_split_apply.py \ diff --git a/config/gcp-service-communications.json b/config/gcp-service-communications.json new file mode 100644 index 0000000..f5637d8 --- /dev/null +++ b/config/gcp-service-communications.json @@ -0,0 +1,185 @@ +{ + "artifact": "teleo_gcp_service_communication_contract", + "schema_version": 1, + "project": "teleo-501523", + "region": "europe-west6", + "network": "teleo-staging-net", + "global_invariants": { + "no_public_database_ip": true, + "no_broad_ssh_or_rdp": true, + "no_default_compute_service_accounts": true, + "secret_values_not_stored_in_contract": true, + "database_connections_encrypted_only": true + }, + "allowed_paths": [ + { + "name": "github_actions_to_artifact_registry", + "purpose": "Build, smoke-run, and push Teleo staging Docker image.", + "source": { + "type": "github_actions_wif", + "repository": "living-ip/teleo-infrastructure" + }, + "identity": "sa-artifact-builder@teleo-501523.iam.gserviceaccount.com", + "target": { + "type": "artifact_registry", + "repository": "projects/teleo-501523/locations/europe-west6/repositories/teleo" + }, + "protocol": "https", + "ports": [443], + "network_path": "google_apis", + "encryption": "tls", + "allowed_by": [".github/workflows/gcp-artifact.yml", "roles/artifactregistry.writer"], + "forbidden": ["cloudsql.editor", "secretmanager.secretAccessor", "compute.admin"] + }, + { + "name": "github_actions_to_readiness_probe", + "purpose": "Read-only GCP readiness probe with retained artifacts.", + "source": { + "type": "github_actions_wif", + "repository": "living-ip/teleo-infrastructure" + }, + "identity": "sa-teleo-readiness@teleo-501523.iam.gserviceaccount.com", + "target": { + "type": "gcp_control_plane", + "resources": ["artifact_registry", "compute_metadata", "cloudsql_metadata", "secret_metadata", "storage_metadata"] + }, + "protocol": "https", + "ports": [443], + "network_path": "google_apis", + "encryption": "tls", + "allowed_by": [".github/workflows/gcp-readiness.yml", "ops/plan_gcp_iam_split.py"], + "forbidden": ["artifactregistry.writer", "cloudsql.editor", "secretmanager.secretAccessor"] + }, + { + "name": "operator_ssh_to_teleo_vms", + "purpose": "Emergency/operator shell access only from one retained operator IPv4 /32.", + "source": { + "type": "operator_ip", + "cidrs": ["/32"] + }, + "identity": "oslogin_operator", + "target": { + "type": "compute_instance_tags", + "tags": ["teleo-prod-ssh", "teleo-staging-ssh"] + }, + "protocol": "tcp", + "ports": [22], + "network_path": "vpc_firewall", + "encryption": "ssh", + "allowed_by": ["ops/apply_gcp_runtime_baseline.py", "ops/check_gcp_infra_readiness.py"], + "forbidden": ["broad_ipv4_ingress", "broad_ipv6_ingress", "tcp:3389"] + }, + { + "name": "teleo_vms_to_artifact_registry", + "purpose": "Runtime pulls immutable images from Artifact Registry over Private Google Access.", + "source": { + "type": "compute_service_accounts", + "service_accounts": [ + "sa-teleo-prod-vm@teleo-501523.iam.gserviceaccount.com", + "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com" + ] + }, + "identity": "dedicated_vm_service_accounts", + "target": { + "type": "artifact_registry", + "repository": "projects/teleo-501523/locations/europe-west6/repositories/teleo" + }, + "protocol": "https", + "ports": [443], + "network_path": "private_google_access", + "encryption": "tls", + "allowed_by": ["roles/artifactregistry.reader", "ops/apply_gcp_runtime_baseline.py"], + "forbidden": ["default_compute_service_account", "public_image_registry_required"] + }, + { + "name": "teleo_vms_to_cloudsql_private", + "purpose": "Application and restore readback connections to the private Cloud SQL standby.", + "source": { + "type": "compute_service_accounts", + "service_accounts": [ + "sa-teleo-prod-vm@teleo-501523.iam.gserviceaccount.com", + "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com" + ] + }, + "identity": "dedicated_vm_service_accounts", + "target": { + "type": "cloudsql_postgres", + "instance": "teleo-pgvector-standby", + "database": "teleo_kb", + "private_network": "projects/teleo-501523/global/networks/teleo-staging-net", + "public_ip": false + }, + "protocol": "postgres", + "ports": [5432], + "network_path": "private_vpc", + "encryption": "encrypted_only", + "allowed_by": ["ops/apply_gcp_runtime_baseline.py", "ops/check_gcp_infra_readiness.py"], + "forbidden": ["public_ip", "ssl_disabled", "authorized_networks_public"] + }, + { + "name": "cloudsql_import_from_backup_bucket", + "purpose": "Cloud SQL import operation reads generated restore SQL from the versioned backup bucket.", + "source": { + "type": "cloudsql_instance_service_account", + "discovered_by": "gcloud sql instances describe teleo-pgvector-standby --format=value(serviceAccountEmailAddress)" + }, + "identity": "cloudsql_instance_service_account", + "target": { + "type": "gcs_bucket", + "bucket": "gs://teleo-501523-prod-backups", + "prefix": "kb-dumps/cloudsql-restore-drills/" + }, + "protocol": "https", + "ports": [443], + "network_path": "google_apis", + "encryption": "tls", + "allowed_by": ["roles/storage.objectAdmin", "ops/apply_gcp_iam_split.py"], + "forbidden": ["allUsers", "allAuthenticatedUsers", "publicAccessPreventionDisabled"] + }, + { + "name": "restore_drill_operator_to_cloudsql_admin", + "purpose": "Operator-triggered Cloud SQL import drill and post-import readback.", + "source": { + "type": "operator_or_ci_with_explicit_auth", + "requires_retained_execute_proof": true + }, + "identity": "sa-teleo-restore-drill@teleo-501523.iam.gserviceaccount.com", + "target": { + "type": "cloudsql_admin_api", + "instance": "teleo-pgvector-standby" + }, + "protocol": "https", + "ports": [443], + "network_path": "google_apis", + "encryption": "tls", + "allowed_by": ["roles/cloudsql.editor", "ops/run_gcp_cloudsql_restore_drill.sh"], + "forbidden": ["unretained_execute", "raw_password_in_logs"] + }, + { + "name": "vps_backup_to_gcs_bucket", + "purpose": "Source backup artifacts are uploaded to versioned GCS only through an approved backup identity.", + "source": { + "type": "hetzner_vps_or_operator_export", + "host": "teleo@77.42.65.182" + }, + "identity": "approved_backup_writer_or_operator_session", + "target": { + "type": "gcs_bucket", + "bucket": "gs://teleo-501523-prod-backups", + "prefix": "kb-dumps/" + }, + "protocol": "https", + "ports": [443], + "network_path": "google_apis", + "encryption": "tls", + "allowed_by": ["ops/backup_vps_sqlite_kb.sh", "ops/run_gcp_cloudsql_restore_drill.sh"], + "forbidden": ["raw_private_key_in_contract", "database_public_ingress", "unversioned_bucket"] + } + ], + "not_proven_by_this_contract": [ + "that GCP resources currently exist", + "that firewall rules are currently applied", + "that Cloud SQL import/readback has succeeded", + "that production Telegram/Leo traffic has cut over to GCP" + ] +} diff --git a/docs/gcp-infra-hardening-20260707.md b/docs/gcp-infra-hardening-20260707.md index 266d0f5..f18df2a 100644 --- a/docs/gcp-infra-hardening-20260707.md +++ b/docs/gcp-infra-hardening-20260707.md @@ -236,11 +236,30 @@ drill/readback verifier before claiming database redundancy. ## Communication Posture -Current GCP communication hardening is limited to the surfaces already read back: +The service-to-service communication contract is declared in: -- default SSH/RDP firewall rules are disabled; -- Teleo SSH access is scoped by instance tags and current source IP rules; -- the custom Teleo subnet has Private Google Access enabled. -- prod/staging VMs run as dedicated Teleo service accounts, not the default Compute Engine service account. +- `config/gcp-service-communications.json` -Before production cutover, add a dedicated runtime communication contract for service-to-service paths, including exact source tags/service accounts, allowed ports, health checks, and rollback. +Validate it locally or in CI with: + +```bash +python3 ops/check_gcp_service_communications.py \ + --contract config/gcp-service-communications.json \ + --output outputs/gcp-infra-hardening-20260707/proofs/gcp-service-communications-check.json +``` + +The contract currently requires: + +- GitHub Actions artifact publishing only through `sa-artifact-builder`. +- GitHub Actions readiness only through `sa-teleo-readiness`. +- emergency SSH only from one operator IPv4 `/32` to Teleo target tags. +- VM image pulls over Private Google Access. +- Cloud SQL only on private VPC paths with encrypted-only PostgreSQL. +- Cloud SQL imports only from the versioned backup bucket through approved + service accounts. +- no broad SSH/RDP, no public database IP, no default Compute Engine service + accounts, and no raw secret values in the contract. + +This is still a contract until GCP readiness passes. Live proof requires the +current `gcp-readiness` artifact to show the matching firewall, VM service +accounts, bucket, and Cloud SQL checks passing. diff --git a/ops/check_gcp_infra_readiness.py b/ops/check_gcp_infra_readiness.py index e9b1c84..08aa65b 100755 --- a/ops/check_gcp_infra_readiness.py +++ b/ops/check_gcp_infra_readiness.py @@ -695,6 +695,58 @@ def check_gcp_runtime_baseline_contract() -> Check: ) +def check_gcp_service_communications_contract() -> Check: + contract = Path("config/gcp-service-communications.json") + checker = Path("ops/check_gcp_service_communications.py") + runbook = Path("docs/gcp-infra-hardening-20260707.md") + missing = [str(path) for path in (contract, checker, runbook) if not path.exists()] + if missing: + return Check( + "gcp_service_communications_contract", + "blocked", + f"missing={missing}", + required_tier="T2_communication_contract", + current_tier="T1_foundation", + ) + completed = subprocess.run( + [ + "python3", + str(checker), + "--contract", + str(contract), + "--output", + "/tmp/teleo-gcp-service-communications-check.json", + ], + text=True, + capture_output=True, + check=False, + ) + if completed.returncode != 0: + detail = completed.stdout[-500:] if completed.stdout else completed.stderr[-500:] + return Check( + "gcp_service_communications_contract", + "fail", + detail, + required_tier="T2_communication_contract", + current_tier="T1_foundation", + ) + if "check_gcp_service_communications.py" not in runbook.read_text(): + return Check( + "gcp_service_communications_contract", + "fail", + "runbook_reference_missing", + required_tier="T2_communication_contract", + current_tier="T1_foundation", + ) + return Check( + "gcp_service_communications_contract", + "pass", + "service-to-service matrix forbids broad SSH/RDP, public Cloud SQL, unencrypted DB paths, default compute service accounts, and raw secret values", + required_tier="T2_communication_contract", + current_tier="T2_communication_contract", + ) + + def check_kb_restore_or_replication() -> Check: return Check( "kb_restore_or_replication", @@ -721,6 +773,7 @@ def main() -> int: safe_check("local_sqlite_postgres_restore_canary", check_local_sqlite_postgres_restore_canary), safe_check("gcp_cloudsql_restore_drill_contract", check_gcp_cloudsql_restore_drill_contract), safe_check("gcp_runtime_baseline_contract", check_gcp_runtime_baseline_contract), + safe_check("gcp_service_communications_contract", check_gcp_service_communications_contract), safe_check("kb_restore_or_replication", check_kb_restore_or_replication), ] payload = { diff --git a/ops/check_gcp_service_communications.py b/ops/check_gcp_service_communications.py new file mode 100644 index 0000000..94ba5f2 --- /dev/null +++ b/ops/check_gcp_service_communications.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +"""Validate the Teleo GCP service-to-service communication contract.""" + +from __future__ import annotations + +import argparse +import ipaddress +import json +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +BROAD_CIDRS = {"0.0.0.0/0", "::/0"} +SECRET_FIELD_TOKENS = ("password", "private_key", "secret_value", "token_value", "seed_phrase") +DEFAULT_COMPUTE_SUFFIX = "-compute@developer.gserviceaccount.com" +REQUIRED_PATHS = { + "github_actions_to_artifact_registry", + "github_actions_to_readiness_probe", + "operator_ssh_to_teleo_vms", + "teleo_vms_to_artifact_registry", + "teleo_vms_to_cloudsql_private", + "cloudsql_import_from_backup_bucket", + "restore_drill_operator_to_cloudsql_admin", + "vps_backup_to_gcs_bucket", +} + + +def load_contract(path: Path) -> dict[str, Any]: + return json.loads(path.read_text()) + + +def as_list(value: Any) -> list[Any]: + return value if isinstance(value, list) else [] + + +def nested_values(value: Any) -> list[Any]: + if isinstance(value, dict): + result: list[Any] = [] + for item in value.values(): + result.extend(nested_values(item)) + return result + if isinstance(value, list): + result = [] + for item in value: + result.extend(nested_values(item)) + return result + return [value] + + +def find_secret_fields(value: Any, prefix: str = "") -> list[str]: + problems: list[str] = [] + if isinstance(value, dict): + for key, item in value.items(): + path = f"{prefix}.{key}" if prefix else str(key) + if str(key).lower() in SECRET_FIELD_TOKENS: + problems.append(path) + problems.extend(find_secret_fields(item, path)) + elif isinstance(value, list): + for index, item in enumerate(value): + problems.extend(find_secret_fields(item, f"{prefix}[{index}]")) + return problems + + +def validate_cidrs(path: dict[str, Any]) -> list[str]: + problems: list[str] = [] + cidrs = as_list(path.get("source", {}).get("cidrs")) + for cidr in cidrs: + if cidr in BROAD_CIDRS: + problems.append(f"{path['name']}:broad_source_cidr={cidr}") + continue + if cidr == "/32": + continue + try: + network = ipaddress.ip_network(str(cidr), strict=False) + except ValueError: + problems.append(f"{path['name']}:invalid_source_cidr={cidr}") + continue + if path.get("ports") == [22] and (network.version != 4 or network.prefixlen != 32): + problems.append(f"{path['name']}:ssh_source_not_single_ipv4_32={cidr}") + return problems + + +def validate_path(path: dict[str, Any]) -> list[str]: + problems: list[str] = [] + name = path.get("name", "") + for field in ("name", "purpose", "source", "identity", "target", "protocol", "ports", "network_path", "encryption", "allowed_by"): + if field not in path: + problems.append(f"{name}:missing_field={field}") + + ports = as_list(path.get("ports")) + if any(not isinstance(port, int) for port in ports): + problems.append(f"{name}:ports_must_be_integers") + if 22 in ports and name != "operator_ssh_to_teleo_vms": + problems.append(f"{name}:tcp_22_only_allowed_for_operator_ssh") + if 3389 in ports: + problems.append(f"{name}:rdp_forbidden") + + problems.extend(validate_cidrs(path)) + + values = [str(value) for value in nested_values(path)] + if any(value in BROAD_CIDRS for value in values): + problems.append(f"{name}:contains_broad_cidr") + if any(value.endswith(DEFAULT_COMPUTE_SUFFIX) for value in values): + problems.append(f"{name}:default_compute_service_account_forbidden") + + target = path.get("target", {}) + if target.get("type") == "cloudsql_postgres": + if target.get("public_ip") is not False: + problems.append(f"{name}:cloudsql_public_ip_not_false") + if not target.get("private_network"): + problems.append(f"{name}:cloudsql_missing_private_network") + if path.get("encryption") != "encrypted_only": + problems.append(f"{name}:cloudsql_encryption_not_encrypted_only") + + if path.get("protocol") in {"https", "postgres"} and path.get("encryption") in {None, "none", "disabled"}: + problems.append(f"{name}:encrypted_protocol_without_encryption") + return problems + + +def build_payload(contract_path: Path) -> dict[str, Any]: + contract = load_contract(contract_path) + paths = as_list(contract.get("allowed_paths")) + names = {str(path.get("name")) for path in paths if isinstance(path, dict)} + problems: list[str] = [] + + missing_paths = sorted(REQUIRED_PATHS - names) + if missing_paths: + problems.append(f"missing_required_paths={missing_paths}") + extra_secret_fields = find_secret_fields(contract) + if extra_secret_fields: + problems.append(f"secret_value_fields_present={extra_secret_fields}") + + invariants = contract.get("global_invariants", {}) + for invariant in ( + "no_public_database_ip", + "no_broad_ssh_or_rdp", + "no_default_compute_service_accounts", + "secret_values_not_stored_in_contract", + "database_connections_encrypted_only", + ): + if invariants.get(invariant) is not True: + problems.append(f"invariant_not_true={invariant}") + + for path in paths: + if not isinstance(path, dict): + problems.append("allowed_paths_contains_non_object") + continue + problems.extend(validate_path(path)) + + return { + "artifact": "teleo_gcp_service_communications_check", + "generated_at_utc": datetime.now(UTC).isoformat(), + "contract_path": str(contract_path), + "project": contract.get("project"), + "network": contract.get("network"), + "allowed_path_count": len(paths), + "required_path_count": len(REQUIRED_PATHS), + "status": "pass" if not problems else "fail", + "problems": problems, + "not_proven_by_this_artifact": contract.get("not_proven_by_this_contract", []), + } + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--contract", default="config/gcp-service-communications.json", type=Path) + parser.add_argument( + "--output", + default="outputs/gcp-infra-hardening-20260707/proofs/gcp-service-communications-check.json", + type=Path, + ) + args = parser.parse_args() + payload = build_payload(args.contract) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + print(json.dumps(payload, indent=2, sort_keys=True)) + return 0 if payload["status"] == "pass" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_gcp_service_communications.py b/tests/test_gcp_service_communications.py new file mode 100644 index 0000000..ade7042 --- /dev/null +++ b/tests/test_gcp_service_communications.py @@ -0,0 +1,83 @@ +import json +import subprocess +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +CONTRACT = REPO_ROOT / "config/gcp-service-communications.json" + + +def run_checker(contract: Path, output: Path) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [ + "python3", + "ops/check_gcp_service_communications.py", + "--contract", + str(contract), + "--output", + str(output), + ], + cwd=REPO_ROOT, + text=True, + capture_output=True, + check=False, + ) + + +def test_gcp_service_communications_contract_passes(tmp_path: Path) -> None: + output = tmp_path / "communications.json" + completed = run_checker(CONTRACT, output) + + assert completed.returncode == 0, completed.stderr + payload = json.loads(output.read_text()) + assert payload["status"] == "pass" + assert payload["allowed_path_count"] == 8 + assert payload["problems"] == [] + + +def test_gcp_service_communications_rejects_broad_ssh(tmp_path: Path) -> None: + contract = json.loads(CONTRACT.read_text()) + for path in contract["allowed_paths"]: + if path["name"] == "operator_ssh_to_teleo_vms": + path["source"]["cidrs"] = ["0.0.0.0/0"] + contract_path = tmp_path / "bad-communications.json" + output = tmp_path / "communications.json" + contract_path.write_text(json.dumps(contract)) + + completed = run_checker(contract_path, output) + + assert completed.returncode == 1 + payload = json.loads(output.read_text()) + assert payload["status"] == "fail" + assert any("broad_source_cidr" in problem for problem in payload["problems"]) + + +def test_gcp_service_communications_rejects_public_or_unencrypted_cloudsql(tmp_path: Path) -> None: + contract = json.loads(CONTRACT.read_text()) + for path in contract["allowed_paths"]: + if path["name"] == "teleo_vms_to_cloudsql_private": + path["target"]["public_ip"] = True + path["encryption"] = "disabled" + contract_path = tmp_path / "bad-cloudsql.json" + output = tmp_path / "communications.json" + contract_path.write_text(json.dumps(contract)) + + completed = run_checker(contract_path, output) + + assert completed.returncode == 1 + payload = json.loads(output.read_text()) + assert "teleo_vms_to_cloudsql_private:cloudsql_public_ip_not_false" in payload["problems"] + assert "teleo_vms_to_cloudsql_private:cloudsql_encryption_not_encrypted_only" in payload["problems"] + + +def test_gcp_service_communications_rejects_secret_value_fields(tmp_path: Path) -> None: + contract = json.loads(CONTRACT.read_text()) + contract["secret_value"] = "do-not-store-this" + contract_path = tmp_path / "bad-secret.json" + output = tmp_path / "communications.json" + contract_path.write_text(json.dumps(contract)) + + completed = run_checker(contract_path, output) + + assert completed.returncode == 1 + payload = json.loads(output.read_text()) + assert any("secret_value_fields_present" in problem for problem in payload["problems"])