Support Cloud SQL scoped role provisioning
Some checks are pending
CI / lint-and-test (push) Waiting to run

This commit is contained in:
fwazb 2026-07-21 12:58:07 -07:00
parent 658083e330
commit 832f0c3f47
3 changed files with 66 additions and 7 deletions

View file

@ -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

View file

@ -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",

View file

@ -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