"""Fail before a public build when its declared Cloud Run runtime is not deployable."""
from __future__ import annotations
import argparse
import json
import re
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence
from check_public_build_contract import ROOT, Target, load_contract
@dataclass(frozen=True)
class RuntimeBinding:
kind: str
reference: str | None = None
class RuntimePreflightError(Exception):
"""The live Cloud Run or Secret Manager runtime contract cannot be verified."""
def __init__(self, message: str, *, category: str = "unknown") -> None:
super().__init__(message)
self.category = category
def _sanitized_gcloud_diagnostic(error: subprocess.CalledProcessError) -> str:
"""Keep an actionable command failure without exposing credential-shaped text."""
raw = "\n".join(part for part in (error.stderr, error.stdout) if part)
compact = " ".join(raw.split())
compact = re.sub(r'(?i)(bearer\s+)[^\s,;"\']+', r"\1[REDACTED]", compact)
for marker in ("access_token", "authorization", "password", "secret", "token"):
compact = re.sub(
rf'(?i)({marker}["\']?\s*[=:]\s*["\']?)' rf'(?:bearer\s+)?' rf'[^\s,;"\']+' rf'(["\']?)',
r"\1[REDACTED]\2",
compact,
)
return compact[:500] or "no diagnostic returned"
def split_secret_reference(reference: str) -> tuple[str, str]:
secret, version = reference.rsplit(":", maxsplit=1)
return secret, version
def _containers(service: Mapping[str, Any]) -> list[Mapping[str, Any]]:
template = service.get("template")
if isinstance(template, Mapping) and isinstance(template.get("containers"), list):
return [container for container in template["containers"] if isinstance(container, Mapping)]
spec = service.get("spec")
if isinstance(spec, Mapping):
template = spec.get("template")
if isinstance(template, Mapping):
template_spec = template.get("spec")
if isinstance(template_spec, Mapping) and isinstance(template_spec.get("containers"), list):
return [container for container in template_spec["containers"] if isinstance(container, Mapping)]
return []
def _secret_reference(wrapper: str, value_source: Any) -> str | None:
"""Return a Secret Manager reference from one documented Cloud Run shape.
Cloud Run v1 serializes secret refs as ``name``/``key`` under
``valueFrom``. Cloud Run v2 uses ``secret``/``version`` under
``valueSource``. The wrapper and nested shape must match exactly so the
preflight never accepts an ambiguous source as a Secret Manager binding.
"""
expected_fields = {
"valueFrom": ("name", "key"),
"valueSource": ("secret", "version"),
}.get(wrapper)
if expected_fields is None or not isinstance(value_source, Mapping) or set(value_source) != {"secretKeyRef"}:
return None
secret_ref = value_source["secretKeyRef"]
if not isinstance(secret_ref, Mapping) or set(secret_ref) != set(expected_fields):
return None
secret, version = (secret_ref[field] for field in expected_fields)
if not isinstance(secret, str) or not secret or not isinstance(version, str) or not version:
return None
return f"{secret}:{version}"
def _runtime_binding(raw_item: Mapping[str, Any]) -> RuntimeBinding:
"""Classify one Cloud Run environment entry without reading literal values."""
source_wrappers = [wrapper for wrapper in ("valueFrom", "valueSource") if wrapper in raw_item]
if not source_wrappers:
return RuntimeBinding("literal")
if len(source_wrappers) != 1 or "value" in raw_item:
return RuntimeBinding("invalid")
wrapper = source_wrappers[0]
secret_reference = _secret_reference(wrapper, raw_item[wrapper])
if secret_reference is None:
return RuntimeBinding("invalid")
return RuntimeBinding("secret", secret_reference)
def current_bindings(service: Mapping[str, Any]) -> dict[str, RuntimeBinding]:
"""Extract Cloud Run env bindings without ever reading their values."""
bindings: dict[str, RuntimeBinding] = {}
for container in _containers(service):
environment = container.get("env")
if not isinstance(environment, list):
continue
for raw_item in environment:
if not isinstance(raw_item, Mapping) or not isinstance(raw_item.get("name"), str):
continue
name = raw_item["name"]
bindings[name] = _runtime_binding(raw_item)
return bindings
def validate_current_bindings(target: Target, service: Mapping[str, Any]) -> list[str]:
errors: list[str] = []
bindings = current_bindings(service)
for name, actual in bindings.items():
if actual.kind == "invalid":
errors.append(f"{target.service}: runtime binding {name} has an ambiguous or malformed value source")
for name, expected_reference in target.deployment.runtime_secrets.items():
actual = bindings.get(name)
if actual is None:
continue
if actual.kind == "invalid":
continue
if actual.kind != "secret":
errors.append(
f"{target.name}: runtime binding {name} is a literal; expected Secret Manager {expected_reference}"
)
elif actual.reference != expected_reference:
errors.append(
f"{target.name}: runtime binding {name} references {actual.reference}; expected {expected_reference}"
)
for name in target.deployment.preserve_runtime_secrets:
actual = bindings.get(name)
if actual is None:
errors.append(
f"{target.name}: preserved runtime secret {name} is absent; expected an enabled Secret Manager binding"
)
elif actual.kind == "invalid":
continue
elif actual.kind != "secret":
errors.append(
f"{target.name}: preserved runtime secret {name} is a literal; expected an enabled Secret Manager binding"
)
for name in target.deployment.runtime_env_vars:
actual = bindings.get(name)
if actual is not None and actual.kind == "invalid":
continue
if actual is not None and actual.kind != "literal":
errors.append(f"{target.name}: runtime config {name} is a Secret Manager binding; expected a literal value")
declared_or_removed = (
set(target.deployment.runtime_secrets)
| set(target.deployment.preserve_runtime_secrets)
| set(target.deployment.runtime_env_vars)
| set(target.deployment.remove_runtime_secrets)
)
for name, actual in bindings.items():
if actual.kind == "secret" and name not in declared_or_removed:
errors.append(f"{target.service}: secret binding {name} is missing from the deployment contract")
return errors
def _gcloud_json_document(arguments: Sequence[str]) -> Any:
try:
completed = subprocess.run(
["gcloud", *arguments, "--format=json"],
check=True,
capture_output=True,
text=True,
)
except subprocess.CalledProcessError as exc:
raw_message = f"{exc.stderr}\n{exc.stdout}".lower()
if "permission denied" in raw_message or "permissiondenied" in raw_message:
message, category = "permission denied", "permission_denied"
elif "unauthenticated" in raw_message or "authentication" in raw_message:
message, category = "authentication failed", "unauthenticated"
else:
message = f"gcloud command failed: {_sanitized_gcloud_diagnostic(exc)}"
category = "unknown"
raise RuntimePreflightError(message, category=category) from exc
try:
result = json.loads(completed.stdout)
except json.JSONDecodeError as exc:
raise RuntimePreflightError("gcloud returned invalid JSON") from exc
return result
def _gcloud_json(arguments: Sequence[str]) -> Mapping[str, Any]:
result = _gcloud_json_document(arguments)
if not isinstance(result, Mapping):
raise RuntimePreflightError("gcloud returned an unexpected JSON document")
return result
def _gcloud_access_token() -> str:
"""Return the current gcloud access token. Never log or print it."""
try:
completed = subprocess.run(
["gcloud", "auth", "print-access-token"],
check=True,
capture_output=True,
text=True,
)
except subprocess.CalledProcessError as exc:
raise RuntimePreflightError(
"gcloud auth print-access-token failed",
category="unauthenticated",
) from exc
token = ""
for line in completed.stdout.splitlines():
if line.strip():
token = line.strip()
if not token:
raise RuntimePreflightError(
"gcloud auth print-access-token returned no token",
category="unauthenticated",
)
return token
def _test_service_account_iam_permissions(
*,
service_account: str,
project_id: str,
permissions: Sequence[str],
) -> Mapping[str, Any]:
"""POST IAM testIamPermissions. Fake this in tests; never log the token."""
token = _gcloud_access_token()
quoted_project = urllib.parse.quote(project_id, safe=".-")
quoted_account = urllib.parse.quote(service_account, safe=".-")
url = (
f"https://iam.googleapis.com/v1/projects/{quoted_project}"
f"/serviceAccounts/{quoted_account}:testIamPermissions"
)
request = urllib.request.Request(
url,
data=json.dumps({"permissions": list(permissions)}).encode("utf-8"),
method="POST",
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"Accept": "application/json",
},
)
try:
with urllib.request.urlopen(request, timeout=30) as response:
raw = response.read()
except urllib.error.HTTPError as exc:
try:
exc.close()
except Exception:
pass
raise RuntimePreflightError(
f"IAM testIamPermissions HTTP {exc.code}",
category="unknown",
) from exc
except urllib.error.URLError as exc:
raise RuntimePreflightError("IAM testIamPermissions request failed", category="unknown") from exc
try:
result = json.loads(raw.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise RuntimePreflightError("IAM testIamPermissions returned invalid JSON") from exc
if not isinstance(result, Mapping):
raise RuntimePreflightError("IAM testIamPermissions returned an unexpected JSON document")
return result
def _cloud_run_service_exists(*, target: Target, project_id: str) -> bool:
"""Use an authenticated, name-filtered list to classify a first create."""
result = _gcloud_json_document(
[
"run",
"services",
"list",
f"--project={project_id}",
f"--region={target.deployment.region}",
f"--filter=metadata.name={target.service}",
]
)
if not isinstance(result, list) or not all(isinstance(item, Mapping) for item in result):
raise RuntimePreflightError("gcloud returned an unexpected Cloud Run service list")
return bool(result)
def load_current_service(*, target: Target, project_id: str) -> Mapping[str, Any] | None:
try:
if not _cloud_run_service_exists(target=target, project_id=project_id):
return None
return _gcloud_json(
[
"run",
"services",
"describe",
target.service,
f"--project={project_id}",
f"--region={target.deployment.region}",
]
)
except RuntimePreflightError as exc:
raise RuntimePreflightError(
f"{target.name}: cannot inspect current Cloud Run service {target.service}: {exc}", category=exc.category
) from exc
def validate_secret_versions(*, target: Target, project_id: str) -> list[str]:
return validate_secret_references(
service_name=target.service,
references=target.deployment.runtime_secrets,
project_id=project_id,
)
def validate_preserved_secret_versions(*, target: Target, service: Mapping[str, Any], project_id: str) -> list[str]:
bindings = current_bindings(service)
references = {
name: binding.reference
for name in target.deployment.preserve_runtime_secrets
if (binding := bindings.get(name)) is not None and binding.kind == "secret" and binding.reference is not None
}
return validate_secret_references(service_name=target.service, references=references, project_id=project_id)
def validate_fallback_secret_versions(*, target: Target, project_id: str) -> list[str]:
return validate_secret_references(
service_name=target.service,
references=target.deployment.fallback_runtime_secrets,
project_id=project_id,
)
def validate_secret_references(*, service_name: str, references: Mapping[str, str], project_id: str) -> list[str]:
errors: list[str] = []
results: dict[str, RuntimePreflightError | Mapping[str, Any]] = {}
for binding_name, reference in sorted(references.items()):
secret, version = split_secret_reference(reference)
result = results.get(reference)
if result is None:
try:
result = _gcloud_json(
[
"secrets",
"versions",
"describe",
version,
f"--secret={secret}",
f"--project={project_id}",
]
)
except RuntimePreflightError as exc:
result = exc
results[reference] = result
if isinstance(result, RuntimePreflightError):
errors.append(
f"{service_name}: runtime binding {binding_name} requires Secret Manager version {reference}, "
f"but it is unavailable ({result})"
)
continue
if result.get("state") != "ENABLED":
errors.append(
f"{service_name}: runtime binding {binding_name} requires enabled Secret Manager version {reference}"
)
return errors
def validate_service_account(*, service_account: str, project_id: str) -> list[str]:
"""Prove a Cloud Run runtime identity exists and the deployer can act as it."""
identity = service_account.strip()
if not identity:
return []
try:
_gcloud_json(["iam", "service-accounts", "describe", identity, f"--project={project_id}"])
except RuntimePreflightError as exc:
return [f"{identity}: Cloud Run runtime identity does not exist or cannot be described ({exc})"]
try:
result = _test_service_account_iam_permissions(
service_account=identity,
project_id=project_id,
permissions=("iam.serviceAccounts.actAs",),
)
except RuntimePreflightError as exc:
return [f"{identity}: cannot test iam.serviceAccounts.actAs ({exc})"]
granted = result.get("permissions")
if not isinstance(granted, list) or "iam.serviceAccounts.actAs" not in granted:
return [f"{identity}: deployer is missing permission iam.serviceAccounts.actAs"]
return []
def preflight_deployment_result(
*, target: Target, project_id: str, service_account: str = ""
) -> tuple[list[str], dict[str, str], bool]:
"""Validate the runtime and report whether the Cloud Run service exists."""
errors = validate_secret_versions(target=target, project_id=project_id)
errors.extend(validate_service_account(service_account=service_account, project_id=project_id))
fallback_runtime_secrets: dict[str, str] = {}
service_exists = False
try:
service = load_current_service(target=target, project_id=project_id)
except RuntimePreflightError as exc:
errors.append(str(exc))
else:
if service is None:
fallback_runtime_secrets = target.deployment.fallback_runtime_secrets
errors.extend(validate_fallback_secret_versions(target=target, project_id=project_id))
else:
service_exists = True
errors.extend(validate_current_bindings(target, service))
errors.extend(validate_preserved_secret_versions(target=target, service=service, project_id=project_id))
return errors, fallback_runtime_secrets, service_exists
def preflight_result(*, target: Target, project_id: str, service_account: str = "") -> tuple[list[str], dict[str, str]]:
"""Validate the runtime for callers that do not need deployment metadata."""
errors, fallback_runtime_secrets, _service_exists = preflight_deployment_result(
target=target, project_id=project_id, service_account=service_account
)
return errors, fallback_runtime_secrets
def preflight(*, target: Target, project_id: str, service_account: str = "") -> list[str]:
return preflight_result(target=target, project_id=project_id, service_account=service_account)[0]
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--target", required=True)
parser.add_argument("--project-id", required=True)
parser.add_argument("--service-account", default="")
parser.add_argument("--contract", type=Path, default=ROOT / "config" / "public-build-contract.json")
parser.add_argument("--github-output", type=Path)
args = parser.parse_args(argv)
try:
contract = load_contract(args.contract)
target = contract.targets[args.target]
except (KeyError, OSError, ValueError) as exc:
print(f"public-build runtime preflight failed: {exc}", file=sys.stderr)
return 1
errors, fallback_runtime_secrets, service_exists = preflight_deployment_result(
target=target, project_id=args.project_id, service_account=args.service_account
)
if errors:
print("public-build runtime preflight failed:", file=sys.stderr)
print("\n".join(f"- {error}" for error in errors), file=sys.stderr)
return 1
if args.github_output is not None:
fallback_lines = "\n".join(
f"{name}={reference}" for name, reference in sorted(fallback_runtime_secrets.items())
)
with args.github_output.open("a", encoding="utf-8") as output:
output.write(f"fallback_runtime_secrets<<EOF\n{fallback_lines}\nEOF\n")
output.write(f"service_exists={'true' if service_exists else 'false'}\n")
print(f"public-build runtime preflight passed: target={target.name}")
return 0
if __name__ == "__main__":
raise SystemExit(main())