From 832f0c3f47abe3eded61f4be585fde3c4f00b91a Mon Sep 17 00:00:00 2001 From: fwazb Date: Tue, 21 Jul 2026 12:58:07 -0700 Subject: [PATCH] Support Cloud SQL scoped role provisioning --- ops/gcp_leoclean_runtime_role.sql | 11 +++-- ...test_gcp_leoclean_runtime_role_postgres.py | 42 +++++++++++++++++++ .../test_hermes_leoclean_kb_bridge_source.py | 20 +++++++-- 3 files changed, 66 insertions(+), 7 deletions(-) diff --git a/ops/gcp_leoclean_runtime_role.sql b/ops/gcp_leoclean_runtime_role.sql index 1dd8fd2..20b4b66 100644 --- a/ops/gcp_leoclean_runtime_role.sql +++ b/ops/gcp_leoclean_runtime_role.sql @@ -19,10 +19,15 @@ select 'create role leoclean_kb_stage_owner nologin nosuperuser nocreatedb nocre where not exists (select 1 from pg_catalog.pg_roles where rolname = 'leoclean_kb_stage_owner') \gexec +-- Cloud SQL's postgres administrator has CREATEROLE but is deliberately not a +-- true PostgreSQL superuser. It may not spell SUPERUSER, REPLICATION, or +-- BYPASSRLS attributes in ALTER ROLE, even when setting their safe false form. +-- CREATE ROLE above establishes those attributes as false; the verification +-- blocks below fail closed if an existing role ever carries any of them. alter role leoclean_kb_runtime - with nologin nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls connection limit 8; + with nologin nocreatedb nocreaterole noinherit connection limit 8; alter role leoclean_kb_stage_owner - with nologin nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls connection limit -1 + with nologin nocreatedb nocreaterole noinherit connection limit -1 password null; -- Snapshot every non-superuser principal connected to either scoped role before @@ -1432,7 +1437,7 @@ $verification$; -- canonical privilege, large-object, and topology assertion. Any failure rolls -- back the complete privilege migration and leaves the role NOLOGIN. alter role leoclean_kb_runtime - with login nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls connection limit 8; + with login nocreatedb nocreaterole noinherit connection limit 8; do $login_verification$ begin diff --git a/tests/test_gcp_leoclean_runtime_role_postgres.py b/tests/test_gcp_leoclean_runtime_role_postgres.py index 29b351b..74acee8 100644 --- a/tests/test_gcp_leoclean_runtime_role_postgres.py +++ b/tests/test_gcp_leoclean_runtime_role_postgres.py @@ -230,6 +230,47 @@ def _provision(container: str, remote_sql_path: str) -> subprocess.CompletedProc ) +def _assert_cloudsql_admin_attribute_compatibility(container: str) -> None: + probe_role = "cloudsql_admin_probe" + target_role = "cloudsql_attribute_target" + _require_success( + _psql( + container, + f""" +create role {probe_role} + login createrole createdb nosuperuser noreplication nobypassrls; +create role {target_role} + nologin nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls; +grant {target_role} to {probe_role} with admin option; +""", + ), + "Cloud SQL administrator attribute fixture", + ) + + _require_success( + _psql( + container, + f""" +alter role {target_role} + with nologin nocreatedb nocreaterole noinherit connection limit 8; +alter role {target_role} + with login nocreatedb nocreaterole noinherit connection limit 8; +""", + role=probe_role, + ), + "Cloud SQL-compatible scoped attribute changes", + ) + + for forbidden_attribute in ("nosuperuser", "noreplication", "nobypassrls"): + denied = _psql( + container, + f"alter role {target_role} with {forbidden_attribute};", + role=probe_role, + ) + assert denied.returncode != 0 + assert re.search(r"(?:ERROR|FATAL):\s+42501:", denied.stderr), denied.stderr + + def _read_table_rows_state(container: str) -> dict[str, dict[str, object]]: table_output = _require_success( _psql( @@ -720,6 +761,7 @@ def test_digest_pinned_network_disabled_postgres_16_14_permission_contract(tmp_p _require_success(_psql(container, "create database teleo_kb;"), "legacy database fixture") _require_success(_psql(container, _bootstrap_sql()), "permission fixture bootstrap") + _assert_cloudsql_admin_attribute_compatibility(container) _require_success( _run(["docker", "cp", str(ROLE_SQL), f"{container}:/tmp/gcp_leoclean_runtime_role.sql"]), "role SQL copy", diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index 89aff9c..2e3d6a9 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -1000,14 +1000,26 @@ def test_gcp_runtime_role_uses_one_narrow_staging_function() -> None: assert "grant update" not in sql.lower() assert "grant delete" not in sql.lower() assert "create role leoclean_kb_runtime nologin" in sql.lower() - disable_statement = "alter role leoclean_kb_runtime\n with nologin nosuperuser" + disable_statement = "alter role leoclean_kb_runtime\n with nologin nocreatedb" + enable_statement = "alter role leoclean_kb_runtime\n with login nocreatedb" password_statement = "\\password leoclean_kb_runtime" assert sql.lower().index(disable_statement) < sql.index(password_statement) assert "\\getenv runtime_password" not in sql assert "password :'runtime_password'" not in sql.lower() - assert sql.lower().index("with nologin nosuperuser") < sql.index("\\connect teleo_canonical") - assert sql.index("\\connect teleo_canonical") < sql.lower().rindex("with login nosuperuser") - assert sql.lower().rindex("with login nosuperuser") < sql.rindex("commit;") + assert sql.lower().index(disable_statement) < sql.index("\\connect teleo_canonical") + assert sql.index("\\connect teleo_canonical") < sql.lower().rindex(enable_statement) + assert sql.lower().rindex(enable_statement) < sql.rindex("commit;") + alter_statements = re.findall(r"alter\s+role\s+[^;]+;", sql, re.IGNORECASE | re.DOTALL) + scoped_attribute_statements = [ + statement + for statement in alter_statements + if "leoclean_kb_runtime" in statement or "leoclean_kb_stage_owner" in statement + ] + assert scoped_attribute_statements + for statement in scoped_attribute_statements: + assert "nosuperuser" not in statement.lower() + assert "noreplication" not in statement.lower() + assert "nobypassrls" not in statement.lower() assert "leoclean_kb_runtime final login attributes are unsafe" in sql