532 lines
20 KiB
Python
532 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Probe current GCP DB parity readiness without mutating GCP.
|
|
|
|
The probe is intentionally read-only. It copies the local gcloud config into a
|
|
temporary writable directory so sandboxed agents can run gcloud without writing
|
|
under ~/.config/gcloud, redirects access-token output out of artifacts, and
|
|
keeps GCP parity separate from VPS proof.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import configparser
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
REPORT_DIR = Path("docs/reports/leo-working-state-20260709")
|
|
DEFAULT_OUT = REPORT_DIR / "gcp-db-parity-current-probe-current.json"
|
|
DEFAULT_MARKDOWN_OUT = REPORT_DIR / "gcp-db-parity-current-probe-current.md"
|
|
DEFAULT_GCLOUD = Path("/opt/homebrew/bin/gcloud")
|
|
DEFAULT_SOURCE_CONFIG = Path.home() / ".config" / "gcloud"
|
|
DEFAULT_PROJECT = "teleo-501523"
|
|
DEFAULT_INSTANCE = "teleo-prod-1"
|
|
DEFAULT_ZONE = "europe-west6-a"
|
|
EXPECTED_ACCESS_ACCOUNT = "billy@livingip.xyz"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CommandResult:
|
|
returncode: int
|
|
stdout: str
|
|
stderr: str
|
|
|
|
|
|
Runner = Callable[[list[str], dict[str, str], int], CommandResult]
|
|
|
|
|
|
def subprocess_runner(command: list[str], env: dict[str, str], timeout: int) -> CommandResult:
|
|
proc = subprocess.run(command, text=True, capture_output=True, env=env, timeout=timeout, check=False)
|
|
return CommandResult(proc.returncode, proc.stdout, proc.stderr)
|
|
|
|
|
|
def _json_or_none(text: str) -> Any:
|
|
try:
|
|
return json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
|
|
def _excerpt(text: str, limit: int = 1200) -> str:
|
|
text = text.strip()
|
|
if len(text) <= limit:
|
|
return text
|
|
return text[:limit] + "...[truncated]"
|
|
|
|
|
|
def _run(
|
|
command: list[str],
|
|
*,
|
|
env: dict[str, str],
|
|
runner: Runner,
|
|
timeout: int,
|
|
redact_stdout: bool = False,
|
|
) -> dict[str, Any]:
|
|
try:
|
|
result = runner(command, env, timeout)
|
|
stdout = "" if redact_stdout else result.stdout
|
|
return {
|
|
"command": redact_command(command),
|
|
"returncode": result.returncode,
|
|
"ok": result.returncode == 0,
|
|
"stdout_redacted": redact_stdout,
|
|
"stdout_excerpt": _excerpt(stdout) if stdout else "",
|
|
"stderr_excerpt": _excerpt(result.stderr),
|
|
"stdout_json": None if redact_stdout else _json_or_none(result.stdout),
|
|
}
|
|
except FileNotFoundError as exc:
|
|
return {
|
|
"command": redact_command(command),
|
|
"returncode": 127,
|
|
"ok": False,
|
|
"stdout_redacted": redact_stdout,
|
|
"stdout_excerpt": "",
|
|
"stderr_excerpt": str(exc),
|
|
"stdout_json": None,
|
|
}
|
|
except subprocess.TimeoutExpired as exc:
|
|
return {
|
|
"command": redact_command(command),
|
|
"returncode": 124,
|
|
"ok": False,
|
|
"stdout_redacted": redact_stdout,
|
|
"stdout_excerpt": "",
|
|
"stderr_excerpt": f"timed out after {exc.timeout}s",
|
|
"stdout_json": None,
|
|
}
|
|
|
|
|
|
def redact_command(command: list[str]) -> list[str]:
|
|
redacted: list[str] = []
|
|
for part in command:
|
|
if part.startswith("--account="):
|
|
redacted.append(part)
|
|
else:
|
|
redacted.append(part)
|
|
return redacted
|
|
|
|
|
|
def copy_gcloud_config(source: Path) -> tuple[tempfile.TemporaryDirectory[str] | None, Path, bool]:
|
|
if not source.exists():
|
|
return None, source, False
|
|
tmp = tempfile.TemporaryDirectory(prefix="teleo-gcp-parity-", dir="/private/tmp")
|
|
target = Path(tmp.name)
|
|
|
|
def ignore(_: str, names: list[str]) -> set[str]:
|
|
return {"logs"} & set(names)
|
|
|
|
shutil.copytree(source, target, dirs_exist_ok=True, ignore=ignore)
|
|
return tmp, target, True
|
|
|
|
|
|
def read_static_config(config_dir: Path) -> dict[str, Any]:
|
|
active_name = "default"
|
|
active_file = config_dir / "active_config"
|
|
if active_file.exists():
|
|
active_name = active_file.read_text(encoding="utf-8").strip() or "default"
|
|
config_file = config_dir / "configurations" / f"config_{active_name}"
|
|
parser = configparser.ConfigParser()
|
|
parser.read(config_file)
|
|
core = parser["core"] if parser.has_section("core") else {}
|
|
return {
|
|
"active_config": active_name,
|
|
"config_file": str(config_file),
|
|
"account": core.get("account"),
|
|
"project": core.get("project"),
|
|
}
|
|
|
|
|
|
def accounts_from_auth_list(auth_list: dict[str, Any], fallback_account: str | None) -> list[str]:
|
|
parsed = auth_list.get("stdout_json")
|
|
accounts: list[str] = []
|
|
if isinstance(parsed, list):
|
|
for item in parsed:
|
|
if isinstance(item, dict) and item.get("account"):
|
|
accounts.append(str(item["account"]))
|
|
if fallback_account and fallback_account not in accounts:
|
|
accounts.insert(0, fallback_account)
|
|
return accounts
|
|
|
|
|
|
def has_network_dns_error(results: list[dict[str, Any]]) -> bool:
|
|
needle_set = ("Failed to resolve", "NameResolutionError", "nodename nor servname", "oauth2.googleapis.com")
|
|
return any(any(needle in item.get("stderr_excerpt", "") for needle in needle_set) for item in results)
|
|
|
|
|
|
def has_reauth_error(results: list[dict[str, Any]]) -> bool:
|
|
needle_set = ("Reauthentication failed", "cannot prompt during non-interactive execution", "gcloud auth login")
|
|
return any(any(needle in item.get("stderr_excerpt", "") for needle in needle_set) for item in results)
|
|
|
|
|
|
def summarize_project(project_result: dict[str, Any]) -> dict[str, Any] | None:
|
|
parsed = project_result.get("stdout_json")
|
|
if not isinstance(parsed, dict):
|
|
return None
|
|
return {
|
|
"projectId": parsed.get("projectId"),
|
|
"projectNumber": parsed.get("projectNumber"),
|
|
"lifecycleState": parsed.get("lifecycleState"),
|
|
}
|
|
|
|
|
|
def summarize_instance(instance_result: dict[str, Any]) -> dict[str, Any] | None:
|
|
parsed = instance_result.get("stdout_json")
|
|
if not isinstance(parsed, dict):
|
|
return None
|
|
return {
|
|
"name": parsed.get("name"),
|
|
"status": parsed.get("status"),
|
|
"zone": parsed.get("zone"),
|
|
"machineType": parsed.get("machineType"),
|
|
"serviceAccounts": [
|
|
{"email": item.get("email"), "scopes": item.get("scopes", [])}
|
|
for item in parsed.get("serviceAccounts", [])
|
|
if isinstance(item, dict)
|
|
],
|
|
}
|
|
|
|
|
|
def summarize_sql_instances(sql_result: dict[str, Any]) -> list[dict[str, Any]]:
|
|
parsed = sql_result.get("stdout_json")
|
|
if not isinstance(parsed, list):
|
|
return []
|
|
summary: list[dict[str, Any]] = []
|
|
for item in parsed:
|
|
if isinstance(item, dict):
|
|
summary.append(
|
|
{
|
|
"name": item.get("name"),
|
|
"state": item.get("state"),
|
|
"databaseVersion": item.get("databaseVersion"),
|
|
"region": item.get("region"),
|
|
}
|
|
)
|
|
return summary
|
|
|
|
|
|
def status_from_results(
|
|
*,
|
|
selected_account: str | None,
|
|
expected_account: str,
|
|
project: str | None,
|
|
expected_project: str,
|
|
token_checks: list[dict[str, Any]],
|
|
project_readback: dict[str, Any],
|
|
instance_readback: dict[str, Any] | None,
|
|
sql_readback: dict[str, Any] | None,
|
|
) -> str:
|
|
command_results = [*token_checks, project_readback]
|
|
if instance_readback is not None:
|
|
command_results.append(instance_readback)
|
|
if sql_readback is not None:
|
|
command_results.append(sql_readback)
|
|
if has_network_dns_error(command_results):
|
|
return "blocked_on_network_dns_before_auth_refresh"
|
|
if selected_account != expected_account:
|
|
return "blocked_on_wrong_active_account"
|
|
if project != expected_project:
|
|
return "blocked_on_wrong_active_project"
|
|
selected_token = next((item for item in token_checks if item["account"] == selected_account), None)
|
|
if selected_token and not selected_token["token_refresh"]["ok"]:
|
|
return "blocked_on_gcloud_reauth" if has_reauth_error([selected_token["token_refresh"]]) else "blocked_on_token_refresh"
|
|
if not project_readback["ok"]:
|
|
return "blocked_on_project_access"
|
|
if instance_readback is not None and not instance_readback["ok"]:
|
|
return "blocked_on_gce_instance_readback"
|
|
if sql_readback is not None and not sql_readback["ok"]:
|
|
return "blocked_on_cloudsql_readback"
|
|
return "ready_for_gcp_readonly_parity"
|
|
|
|
|
|
def probe_gcp_parity(
|
|
*,
|
|
gcloud_path: Path = DEFAULT_GCLOUD,
|
|
source_config: Path = DEFAULT_SOURCE_CONFIG,
|
|
expected_project: str = DEFAULT_PROJECT,
|
|
expected_account: str = EXPECTED_ACCESS_ACCOUNT,
|
|
instance: str = DEFAULT_INSTANCE,
|
|
zone: str = DEFAULT_ZONE,
|
|
runner: Runner = subprocess_runner,
|
|
use_temp_config: bool = True,
|
|
timeout: int = 30,
|
|
) -> dict[str, Any]:
|
|
temp_dir: tempfile.TemporaryDirectory[str] | None = None
|
|
temp_config_created = False
|
|
config_dir = source_config
|
|
try:
|
|
if use_temp_config:
|
|
temp_dir, config_dir, temp_config_created = copy_gcloud_config(source_config)
|
|
static_config = read_static_config(config_dir)
|
|
env = os.environ.copy()
|
|
env["CLOUDSDK_CONFIG"] = str(config_dir)
|
|
env["CLOUDSDK_CORE_DISABLE_PROMPTS"] = "1"
|
|
|
|
config_list = _run(
|
|
[str(gcloud_path), "config", "list", "--format=json"],
|
|
env=env,
|
|
runner=runner,
|
|
timeout=timeout,
|
|
)
|
|
auth_list = _run(
|
|
[str(gcloud_path), "auth", "list", "--format=json"],
|
|
env=env,
|
|
runner=runner,
|
|
timeout=timeout,
|
|
)
|
|
accounts = accounts_from_auth_list(auth_list, static_config.get("account"))
|
|
token_checks = []
|
|
for account in accounts:
|
|
token_checks.append(
|
|
{
|
|
"account": account,
|
|
"token_refresh": _run(
|
|
[str(gcloud_path), "auth", "print-access-token", f"--account={account}"],
|
|
env=env,
|
|
runner=runner,
|
|
timeout=timeout,
|
|
redact_stdout=True,
|
|
),
|
|
}
|
|
)
|
|
|
|
project_readback = _run(
|
|
[str(gcloud_path), "projects", "describe", expected_project, "--format=json"],
|
|
env=env,
|
|
runner=runner,
|
|
timeout=timeout,
|
|
)
|
|
|
|
instance_readback: dict[str, Any] | None = None
|
|
sql_readback: dict[str, Any] | None = None
|
|
if project_readback["ok"]:
|
|
instance_readback = _run(
|
|
[
|
|
str(gcloud_path),
|
|
"compute",
|
|
"instances",
|
|
"describe",
|
|
instance,
|
|
"--zone",
|
|
zone,
|
|
"--project",
|
|
expected_project,
|
|
"--format=json",
|
|
],
|
|
env=env,
|
|
runner=runner,
|
|
timeout=timeout,
|
|
)
|
|
sql_readback = _run(
|
|
[str(gcloud_path), "sql", "instances", "list", "--project", expected_project, "--format=json"],
|
|
env=env,
|
|
runner=runner,
|
|
timeout=timeout,
|
|
)
|
|
|
|
status = status_from_results(
|
|
selected_account=static_config.get("account"),
|
|
expected_account=expected_account,
|
|
project=static_config.get("project"),
|
|
expected_project=expected_project,
|
|
token_checks=token_checks,
|
|
project_readback=project_readback,
|
|
instance_readback=instance_readback,
|
|
sql_readback=sql_readback,
|
|
)
|
|
ready = status == "ready_for_gcp_readonly_parity"
|
|
clear_cta = (
|
|
"Run this probe from a network-enabled shell where oauth2.googleapis.com resolves, with "
|
|
f"`{expected_account}` authenticated for project `{expected_project}`. Then rerun "
|
|
"`python3 scripts/probe_gcp_db_parity.py` and proceed only if it reports "
|
|
"`ready_for_gcp_readonly_parity`."
|
|
)
|
|
if status == "blocked_on_gcloud_reauth":
|
|
clear_cta = (
|
|
f"In a local terminal, run `gcloud auth login {expected_account}` and complete Google "
|
|
f"reauth for project `{expected_project}`. Then rerun "
|
|
"`python3 scripts/probe_gcp_db_parity.py`."
|
|
)
|
|
elif status == "blocked_on_wrong_active_account":
|
|
clear_cta = (
|
|
f"Run `gcloud config set account {expected_account}` in the active gcloud config, then rerun "
|
|
"`python3 scripts/probe_gcp_db_parity.py`."
|
|
)
|
|
elif status == "blocked_on_project_access":
|
|
clear_cta = (
|
|
f"Grant `{static_config.get('account')}` read access to project `{expected_project}` or switch "
|
|
"to an account that already has it, then rerun `python3 scripts/probe_gcp_db_parity.py`."
|
|
)
|
|
|
|
return {
|
|
"generated_at_utc": datetime.now(timezone.utc).isoformat(),
|
|
"artifact": "gcp_db_parity_current_probe",
|
|
"mode": "read_only_gcloud_parity_probe",
|
|
"status": status,
|
|
"ready_for_gcp_readonly_parity": ready,
|
|
"mutates_gcp": False,
|
|
"mutates_db": False,
|
|
"current_canary": "gcloud account can read teleo-501523 project metadata plus target VM/Cloud SQL metadata",
|
|
"expected": {
|
|
"project": expected_project,
|
|
"project_access_account": expected_account,
|
|
"gce_instance": instance,
|
|
"gce_zone": zone,
|
|
},
|
|
"local_gcloud": {
|
|
"gcloud_path": str(gcloud_path),
|
|
"source_config": str(source_config),
|
|
"used_temporary_writable_config_copy": temp_config_created,
|
|
"temporary_config_removed_after_probe": True,
|
|
"static_active_config": static_config,
|
|
"config_list": config_list,
|
|
"auth_list": auth_list,
|
|
},
|
|
"accounts_tested": token_checks,
|
|
"project_readback": {
|
|
**project_readback,
|
|
"summary": summarize_project(project_readback),
|
|
},
|
|
"runtime_readbacks": {
|
|
"gce_instance": None
|
|
if instance_readback is None
|
|
else {**instance_readback, "summary": summarize_instance(instance_readback)},
|
|
"cloudsql_instances": None
|
|
if sql_readback is None
|
|
else {**sql_readback, "summary": summarize_sql_instances(sql_readback)},
|
|
},
|
|
"attempted_routes": [
|
|
"static active_config/config_default readback",
|
|
"temporary writable copy of local gcloud config to avoid sandbox writes under ~/.config/gcloud",
|
|
"gcloud auth list without printing secrets",
|
|
"gcloud auth print-access-token for configured accounts with stdout redacted",
|
|
"gcloud projects describe teleo-501523",
|
|
"if project readback works: gcloud compute instances describe teleo-prod-1",
|
|
"if project readback works: gcloud sql instances list",
|
|
],
|
|
"exact_gate": status,
|
|
"clear_CTA": clear_cta,
|
|
"next_non_user_action": [
|
|
"Run this same probe again and require ready_for_gcp_readonly_parity=true.",
|
|
"SSH into teleo-prod-1 or use Cloud SQL read-only access for DB/runtime parity proof.",
|
|
"Run the Working Leo benchmark against GCP only after read-only GCP metadata is proven current.",
|
|
],
|
|
"claim_ceiling": (
|
|
"This probe proves only current local gcloud/GCP read-only parity readiness. It does not prove "
|
|
"VPS behavior, Telegram-visible behavior, production DB mutation, or GCP-hosted Leo correctness."
|
|
),
|
|
}
|
|
finally:
|
|
if temp_dir is not None:
|
|
temp_dir.cleanup()
|
|
|
|
|
|
def write_json(path: Path, data: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
|
|
|
|
def write_markdown(path: Path, data: dict[str, Any]) -> None:
|
|
lines = [
|
|
"# GCP DB Parity Current Probe",
|
|
"",
|
|
f"Generated UTC: `{data['generated_at_utc']}`",
|
|
f"Status: `{data['status']}`",
|
|
f"Ready for GCP read-only parity: `{data['ready_for_gcp_readonly_parity']}`",
|
|
f"Mutates GCP: `{data['mutates_gcp']}`",
|
|
f"Mutates DB: `{data['mutates_db']}`",
|
|
f"Current canary: {data['current_canary']}",
|
|
"",
|
|
"## Active Config",
|
|
"",
|
|
f"- Account: `{data['local_gcloud']['static_active_config'].get('account')}`",
|
|
f"- Project: `{data['local_gcloud']['static_active_config'].get('project')}`",
|
|
f"- Temporary writable config copy: `{data['local_gcloud']['used_temporary_writable_config_copy']}`",
|
|
f"- Temporary config removed after probe: `{data['local_gcloud']['temporary_config_removed_after_probe']}`",
|
|
"",
|
|
"## Account Checks",
|
|
"",
|
|
]
|
|
for item in data["accounts_tested"]:
|
|
token = item["token_refresh"]
|
|
lines.append(
|
|
f"- `{item['account']}`: `ok={token['ok']}`, `returncode={token['returncode']}`, "
|
|
f"`stdout_redacted={token['stdout_redacted']}`"
|
|
)
|
|
if token["stderr_excerpt"]:
|
|
lines.append(f" - stderr: `{token['stderr_excerpt']}`")
|
|
project = data["project_readback"]
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Project Readback",
|
|
"",
|
|
f"- `ok={project['ok']}`, `returncode={project['returncode']}`",
|
|
]
|
|
)
|
|
if project.get("summary"):
|
|
lines.append(f"- Summary: `{project['summary']}`")
|
|
if project["stderr_excerpt"]:
|
|
lines.append(f"- stderr: `{project['stderr_excerpt']}`")
|
|
lines.extend(["", "## Gate", "", f"- Exact gate: `{data['exact_gate']}`", f"- Clear CTA: {data['clear_CTA']}"])
|
|
lines.extend(["", "## Next Non-User Action", ""])
|
|
lines.extend(f"- {item}" for item in data["next_non_user_action"])
|
|
lines.extend(["", "## Claim Ceiling", "", data["claim_ceiling"], ""])
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("\n".join(lines), encoding="utf-8")
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--gcloud", type=Path, default=DEFAULT_GCLOUD)
|
|
parser.add_argument("--source-config", type=Path, default=DEFAULT_SOURCE_CONFIG)
|
|
parser.add_argument("--project", default=DEFAULT_PROJECT)
|
|
parser.add_argument("--expected-account", default=EXPECTED_ACCESS_ACCOUNT)
|
|
parser.add_argument("--instance", default=DEFAULT_INSTANCE)
|
|
parser.add_argument("--zone", default=DEFAULT_ZONE)
|
|
parser.add_argument("--timeout", type=int, default=30)
|
|
parser.add_argument("--no-temp-config-copy", action="store_true")
|
|
parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
|
|
parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MARKDOWN_OUT)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
data = probe_gcp_parity(
|
|
gcloud_path=args.gcloud,
|
|
source_config=args.source_config,
|
|
expected_project=args.project,
|
|
expected_account=args.expected_account,
|
|
instance=args.instance,
|
|
zone=args.zone,
|
|
use_temp_config=not args.no_temp_config_copy,
|
|
timeout=args.timeout,
|
|
)
|
|
write_json(args.out, data)
|
|
write_markdown(args.markdown_out, data)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"out": str(args.out),
|
|
"markdown_out": str(args.markdown_out),
|
|
"status": data["status"],
|
|
"ready_for_gcp_readonly_parity": data["ready_for_gcp_readonly_parity"],
|
|
"mutates_gcp": data["mutates_gcp"],
|
|
"mutates_db": data["mutates_db"],
|
|
},
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
)
|
|
return 0 if data["ready_for_gcp_readonly_parity"] else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|