teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Merge pull request #173 from living-ip/codex/leo-gcp-data-parity-acl-preflight-20260715
Some checks are pending
CI / lint-and-test (push) Waiting to run
Details

Browse source 
Preserve role ACLs in canonical parity preflight
This commit is contained in:
twentyOne2x 2026-07-16 12:45:52 +02:00 committed by GitHub
commit 15ceffccd3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 974 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
.agents/skills/teleo-gcp-parity-ops/SKILL.md
View file

@ -136,8 +136,9 @@ Track these independently:
1. control-plane project, VM, Cloud SQL, private-IP, and TLS identity;
2. canonical database schema, counts, row hashes, constraints, indexes,
functions, extensions, exact roles and memberships, owners, normalized
schema/relation/column/function/type ACLs, and performance;
functions, extensions, exact role attributes, connection limits,
`default_transaction_read_only` settings and memberships, owners,
normalized schema/relation/column/function/type ACLs, and performance;
3. GCP service PID/restarts plus live profile and tool hashes;
4. `DC-01` through `DC-06` DB-read readiness;
5. real no-send model replies, strict score, and exact count consistency;
@ -218,12 +219,15 @@ input, supporting receipt, and verifier output; do not treat that evidence as
transient cleanup. The canonical restore streams through `pg_restore` and
creates no GCS import object.
The current snapshot and restore commands intentionally use `--no-owner` and
`--no-acl`/`--no-privileges`. The expanded manifest therefore fails GCP parity
until a separately reviewed Cloud-SQL-compatible authorization replay restores
the captured owner/grant semantics. Do not bypass this by dropping the new
manifest rows or by running the Docker/superuser gate bootstrap blindly on
shared Cloud SQL.
The source snapshot retains ACL commands while omitting owner commands. The
networkless local preflight can therefore replay captured ACLs after creating
only the exact allowlisted application roles, plus the bounded database-level
grants that `pg_restore` does not apply to an already-created database. The
shared-Cloud-SQL restore still uses `--no-owner` and `--no-privileges`, so GCP
parity must fail until a separately reviewed Cloud-SQL-compatible authorization
replay restores the captured owner/grant semantics. Do not bypass this by
dropping manifest rows or by running the Docker/superuser role bootstrap
blindly on shared Cloud SQL.
Static inspection or a dry run is insufficient for these helpers: verify a new
capture against the live VPS and run the focused tests. A current T3 claim also

20
docs/gcp-kb-cloudsql-restore-runbook.md
View file

@ -44,8 +44,8 @@ Do not call this redundancy complete until source data has been restored or repl
## Canonical Postgres Snapshot And Parity
Capture a custom-format dump and a full JSONL manifest from the same exported,
read-only PostgreSQL snapshot:
Capture an ACL-bearing custom-format dump and a full JSONL manifest from the
same exported, read-only PostgreSQL snapshot:
```bash
python3 ops/capture_vps_canonical_postgres_snapshot.py \
@ -83,13 +83,19 @@ it for GCP:
The local runner starts a uniquely named `postgres:16-alpine` container with
network mode `none` and tmpfs-only database storage. It waits for an actual
`psql` connection to the named database, restores with `pg_restore
--no-owner --no-privileges --exit-on-error`, compares the full parity manifest,
--no-owner --exit-on-error`, compares the full parity manifest,
then removes the container and proves it is absent. A passing local receipt is
the exact-recovery preflight; it is not semantic recompilation from raw source
documents. Full parity now includes `kb_gate_owner`, `kb_apply`, and
`kb_review`, their memberships, object ownership, and normalized database,
schema, relation, column, function, and type ACLs. A restore that recovers rows
but not those authorization semantics fails closed.
documents. The source dump retains ACL commands, and the isolated preflight
bootstraps only the exact allowlisted application roles needed to replay them.
Because `pg_restore` targets an already-created database, the preflight also
replays only the captured database-level grants for those allowlisted roles;
unsupported privileges or an unbound database ACL fail before verification.
Full parity includes `kb_gate_owner`, `kb_apply`, `kb_review`, and
`kb_observatory_read`, their connection limits, explicit
`default_transaction_read_only` settings, memberships, object ownership, and
normalized database, schema, relation, column, function, and type ACLs. A
restore that recovers rows but not those authorization semantics fails closed.
Choose one bounded suffix and use it consistently on the GCP replay VM. The
restore helper retains its mode-0700 evidence directory at the fixed private

2
ops/capture_vps_canonical_postgres_snapshot.py
View file

@ -217,7 +217,7 @@ printf '%s\n' "$snapshot_metadata" > "$remote_root/source-snapshot.json"
docker exec "$container" pg_dump \
-U postgres -d "$database" \
--format=custom --compress=9 --no-owner --no-acl \
--format=custom --compress=9 --no-owner \
--snapshot="$snapshot_id" \
--file="$container_dump"
docker exec "$container" psql \

64
ops/postgres_parity_manifest.sql
View file

@ -45,17 +45,55 @@ select jsonb_build_object(
'items', coalesce(
(select jsonb_agg(
jsonb_build_object(
'name', rolname,
'can_login', rolcanlogin,
'superuser', rolsuper,
'create_db', rolcreatedb,
'create_role', rolcreaterole,
'inherit', rolinherit,
'replication', rolreplication,
'bypass_rls', rolbypassrls
) order by rolname)
from pg_roles
where rolname in ('kb_gate_owner', 'kb_apply', 'kb_review')),
'name', role_row.rolname,
'can_login', role_row.rolcanlogin,
'superuser', role_row.rolsuper,
'create_db', role_row.rolcreatedb,
'create_role', role_row.rolcreaterole,
'inherit', role_row.rolinherit,
'replication', role_row.rolreplication,
'bypass_rls', role_row.rolbypassrls,
'connection_limit', role_row.rolconnlimit,
'default_transaction_read_only', (
select split_part(setting, '=', 2)
from unnest(coalesce(role_row.rolconfig, array[]::text[])) setting
where split_part(setting, '=', 1) = 'default_transaction_read_only'
limit 1
),
'settings', (
select coalesce(
jsonb_agg(
jsonb_build_object(
'name', split_part(setting, '=', 1),
'value', substring(setting from position('=' in setting) + 1)
) order by split_part(setting, '=', 1), setting
),
'[]'::jsonb
)
from unnest(coalesce(role_row.rolconfig, array[]::text[])) setting
),
'database_settings', (
select coalesce(
jsonb_agg(
jsonb_build_object(
'name', split_part(setting, '=', 1),
'value', substring(setting from position('=' in setting) + 1)
) order by split_part(setting, '=', 1), setting
),
'[]'::jsonb
)
from pg_db_role_setting database_setting
cross join lateral unnest(coalesce(database_setting.setconfig, array[]::text[])) setting
where database_setting.setrole = role_row.oid
and database_setting.setdatabase = (
select database_row.oid
from pg_database database_row
where database_row.datname = current_database()
)
)
) order by role_row.rolname)
from pg_roles role_row
where role_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review', 'kb_observatory_read')),
'[]'::jsonb
)
)::text;
@ -76,8 +114,8 @@ select jsonb_build_object(
join pg_roles role_row on role_row.oid = membership.roleid
join pg_roles member_row on member_row.oid = membership.member
join pg_roles grantor_row on grantor_row.oid = membership.grantor
where role_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review')
or member_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review')),
where role_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review', 'kb_observatory_read')
or member_row.rolname in ('kb_gate_owner', 'kb_apply', 'kb_review', 'kb_observatory_read')),
'[]'::jsonb
)
)::text;

391
ops/run_local_canonical_postgres_rebuild.py
View file

@ -31,7 +31,9 @@ MANIFEST_DESTINATION = "/tmp/postgres-parity-manifest.sql"
CANARY_LABEL = "com.livingip.teleo.canonical-rebuild-canary"
DATABASE_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,62}\Z")
CONTAINER_PREFIX_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,39}\Z")
ALLOWED_APPLICATION_ROLES = frozenset({"kb_gate_owner", "kb_apply", "kb_review"})
SETTING_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_.]{0,127}\Z")
ALLOWED_APPLICATION_ROLES = frozenset({"kb_gate_owner", "kb_apply", "kb_review", "kb_observatory_read"})
ALLOWED_DATABASE_PRIVILEGES = frozenset({"CONNECT", "CREATE", "TEMPORARY"})
ROLE_BOOLEAN_FIELDS = (
"can_login",
"superuser",
@ -226,11 +228,59 @@ def _role_keyword(value: bool, enabled: str, disabled: str) -> str:
return enabled if value else disabled
def application_role_sql(source_manifest: dict[str, Any]) -> str:
items = source_manifest["singleton"]["application_roles"].get("items", [])
def _quote_identifier(value: str) -> str:
return '"' + value.replace('"', '""') + '"'
def _quote_literal(value: str) -> str:
return "'" + value.replace("'", "''") + "'"
def _manifest_items(source_manifest: dict[str, Any], kind: str) -> list[Any]:
try:
items = source_manifest["singleton"][kind].get("items", [])
except (AttributeError, KeyError, TypeError) as exc:
raise ValueError(f"source manifest is missing {kind}.items") from exc
if not isinstance(items, list):
raise ValueError("source manifest application_roles.items must be a list")
statements = []
raise ValueError(f"source manifest {kind}.items must be a list")
return items
def _validated_role_name(value: Any, context: str) -> str:
if not isinstance(value, str) or not DATABASE_NAME_RE.fullmatch(value):
raise ValueError(f"source manifest {context} is not a safe PostgreSQL role name")
if value == "PUBLIC":
raise ValueError(f"source manifest {context} cannot use PUBLIC as a role name")
return value
def _validated_role_settings(item: dict[str, Any], field: str, role: str) -> list[tuple[str, str]]:
settings = item.get(field)
if not isinstance(settings, list):
raise ValueError(f"source manifest application role {role} {field} must be a list")
parsed: list[tuple[str, str]] = []
seen: set[str] = set()
for setting in settings:
if not isinstance(setting, dict):
raise ValueError(f"source manifest application role {role} {field} item must be an object")
name = setting.get("name")
value = setting.get("value")
if not isinstance(name, str) or not SETTING_NAME_RE.fullmatch(name):
raise ValueError(f"source manifest application role {role} has invalid setting name")
if not isinstance(value, str) or "\x00" in value:
raise ValueError(f"source manifest application role {role} setting {name} has invalid value")
if name in seen:
raise ValueError(f"source manifest application role {role} has duplicate {field} setting {name}")
seen.add(name)
parsed.append((name, value))
return sorted(parsed)
def _application_role_plan(source_manifest: dict[str, Any], database: str) -> dict[str, Any]:
if not DATABASE_NAME_RE.fullmatch(database):
raise ValueError("target database name is invalid")
items = _manifest_items(source_manifest, "application_roles")
parsed_roles: list[dict[str, Any]] = []
seen: set[str] = set()
for item in items:
if not isinstance(item, dict):
@ -244,6 +294,33 @@ def application_role_sql(source_manifest: dict[str, Any]) -> str:
for field in ROLE_BOOLEAN_FIELDS:
if type(item.get(field)) is not bool:
raise ValueError(f"source manifest application role {name} has invalid {field}")
connection_limit = item.get("connection_limit")
if type(connection_limit) is not int or connection_limit < -1:
raise ValueError(f"source manifest application role {name} has invalid connection_limit")
default_transaction_read_only = item.get("default_transaction_read_only")
if default_transaction_read_only not in {None, "on", "off"}:
raise ValueError(f"source manifest application role {name} has invalid default_transaction_read_only")
settings = _validated_role_settings(item, "settings", str(name))
database_settings = _validated_role_settings(item, "database_settings", str(name))
captured_read_only = [value for setting, value in settings if setting == "default_transaction_read_only"]
expected_read_only = [] if default_transaction_read_only is None else [default_transaction_read_only]
if captured_read_only != expected_read_only:
raise ValueError(f"source manifest application role {name} has inconsistent default_transaction_read_only")
parsed_roles.append(
{
"item": item,
"name": str(name),
"settings": settings,
"database_settings": database_settings,
}
)
statements: list[str] = []
role_setting_statement_count = 0
database_role_setting_statement_count = 0
for parsed in sorted(parsed_roles, key=lambda role: role["name"]):
item = parsed["item"]
name = parsed["name"]
attributes = [
_role_keyword(item["can_login"], "LOGIN", "NOLOGIN"),
_role_keyword(item["superuser"], "SUPERUSER", "NOSUPERUSER"),
@ -252,9 +329,30 @@ def application_role_sql(source_manifest: dict[str, Any]) -> str:
_role_keyword(item["inherit"], "INHERIT", "NOINHERIT"),
_role_keyword(item["replication"], "REPLICATION", "NOREPLICATION"),
_role_keyword(item["bypass_rls"], "BYPASSRLS", "NOBYPASSRLS"),
f"CONNECTION LIMIT {item['connection_limit']}",
]
statements.append(f'CREATE ROLE "{name}" {" ".join(attributes)};')
return "\n".join(statements)
for setting, value in parsed["settings"]:
statements.append(
f"ALTER ROLE {_quote_identifier(name)} SET {_quote_identifier(setting)} TO {_quote_literal(value)};"
)
role_setting_statement_count += 1
for setting, value in parsed["database_settings"]:
statements.append(
f"ALTER ROLE {_quote_identifier(name)} IN DATABASE {_quote_identifier(database)} "
f"SET {_quote_identifier(setting)} TO {_quote_literal(value)};"
)
database_role_setting_statement_count += 1
return {
"sql": "\n".join(statements),
"roles": sorted(parsed["name"] for parsed in parsed_roles),
"role_setting_statement_count": role_setting_statement_count,
"database_role_setting_statement_count": database_role_setting_statement_count,
}
def application_role_sql(source_manifest: dict[str, Any], database: str) -> str:
return str(_application_role_plan(source_manifest, database)["sql"])
def execute_application_role_bootstrap(
@ -265,7 +363,8 @@ def execute_application_role_bootstrap(
*,
timeout: float,
) -> list[str]:
sql = application_role_sql(source_manifest)
plan = _application_role_plan(source_manifest, database)
sql = str(plan["sql"])
if not sql:
return []
completed = _run(
@ -273,7 +372,256 @@ def execute_application_role_bootstrap(
timeout=timeout,
)
_require_success(completed, "application role bootstrap")
return sorted(item["name"] for item in source_manifest["singleton"]["application_roles"].get("items", []))
return list(plan["roles"])
def _membership_plan(source_manifest: dict[str, Any]) -> dict[str, Any]:
application_roles = _manifest_items(source_manifest, "application_roles")
application_role_names: set[str] = set()
initial_grantors = {"postgres"}
for item in application_roles:
if not isinstance(item, dict):
raise ValueError("source manifest application role must be an object")
name = item.get("name")
if name not in ALLOWED_APPLICATION_ROLES:
raise ValueError(f"source manifest contains non-allowlisted application role: {name!r}")
application_role_names.add(str(name))
if item.get("superuser") is True or item.get("create_role") is True:
initial_grantors.add(str(name))
parsed: list[dict[str, Any]] = []
seen: set[tuple[str, str, str]] = set()
for item in _manifest_items(source_manifest, "role_memberships"):
if not isinstance(item, dict):
raise ValueError("source manifest role membership must be an object")
role = _validated_role_name(item.get("role"), "role membership role")
member = _validated_role_name(item.get("member"), "role membership member")
grantor = _validated_role_name(item.get("grantor"), "role membership grantor")
if role not in application_role_names and member not in application_role_names:
raise ValueError("source manifest role membership is not bound to an application role")
for field in ("admin_option", "inherit_option", "set_option"):
if type(item.get(field)) is not bool:
raise ValueError(f"source manifest role membership has invalid {field}")
key = (role, member, grantor)
if key in seen:
raise ValueError("source manifest contains a duplicate role membership grant")
seen.add(key)
parsed.append(
{
"role": role,
"member": member,
"grantor": grantor,
"admin_option": item["admin_option"],
"inherit_option": item["inherit_option"],
"set_option": item["set_option"],
}
)
remaining = sorted(parsed, key=lambda row: (row["role"], row["member"], row["grantor"]))
ordered: list[dict[str, Any]] = []
authorized: dict[str, set[str]] = {row["role"]: set(initial_grantors) for row in remaining}
while remaining:
progress = False
for row in list(remaining):
if row["grantor"] not in authorized[row["role"]]:
continue
ordered.append(row)
remaining.remove(row)
if row["admin_option"]:
authorized[row["role"]].add(row["member"])
progress = True
if not progress:
blocked = ", ".join(f"{row['grantor']}->{row['role']}->{row['member']}" for row in remaining)
raise ValueError(f"source manifest role membership grantor dependencies cannot be replayed: {blocked}")
statements: list[str] = []
for row in ordered:
statements.extend(
[
f"SET ROLE {_quote_identifier(row['grantor'])};",
(
f"GRANT {_quote_identifier(row['role'])} TO {_quote_identifier(row['member'])} WITH "
f"ADMIN {str(row['admin_option']).upper()}, "
f"INHERIT {str(row['inherit_option']).upper()}, "
f"SET {str(row['set_option']).upper()} "
f"GRANTED BY {_quote_identifier(row['grantor'])};"
),
"RESET ROLE;",
]
)
return {"sql": "\n".join(statements), "statement_count": len(ordered)}
def role_membership_sql(source_manifest: dict[str, Any]) -> str:
return str(_membership_plan(source_manifest)["sql"])
def execute_role_membership_replay(
docker_bin: str,
container: str,
database: str,
source_manifest: dict[str, Any],
*,
timeout: float,
) -> int:
plan = _membership_plan(source_manifest)
sql = str(plan["sql"])
if not sql:
return 0
completed = _run(
_psql_command(container, database, "-c", sql, docker_bin=docker_bin),
timeout=timeout,
)
_require_success(completed, "role membership replay")
return int(plan["statement_count"])
def _database_owner(source_manifest: dict[str, Any]) -> str:
rows = [
item
for item in _manifest_items(source_manifest, "object_ownership")
if isinstance(item, dict) and item.get("object_type") == "database"
]
if len(rows) != 1:
raise ValueError("source manifest must contain exactly one database ownership row")
row = rows[0]
if row.get("name") != "canonical_database" or row.get("schema") is not None:
raise ValueError("source manifest database owner row is not bound to canonical_database")
owner = _validated_role_name(row.get("owner"), "database owner")
if owner != "postgres":
raise ValueError("local parity canary requires postgres as the captured database owner")
return owner
def _database_acl_plan(source_manifest: dict[str, Any], database: str) -> dict[str, Any]:
if not DATABASE_NAME_RE.fullmatch(database):
raise ValueError("target database name is invalid")
owner = _database_owner(source_manifest)
application_roles = _manifest_items(source_manifest, "application_roles")
initial_grantors = {owner, "postgres"}
for role in application_roles:
if not isinstance(role, dict) or role.get("name") not in ALLOWED_APPLICATION_ROLES:
raise ValueError("source manifest database ACL references require exact allowlisted roles")
if role.get("superuser") is True:
initial_grantors.add(str(role["name"]))
parsed: list[dict[str, Any]] = []
seen: set[tuple[str, str, str, bool]] = set()
for item in _manifest_items(source_manifest, "object_acl"):
if not isinstance(item, dict) or item.get("object_type") != "database":
continue
if item.get("name") != "canonical_database" or item.get("schema") is not None:
raise ValueError("source manifest database ACL row is not bound to canonical_database")
grantee_value = item.get("grantee")
grantee = "PUBLIC" if grantee_value == "PUBLIC" else _validated_role_name(grantee_value, "database ACL grantee")
grantor = _validated_role_name(item.get("grantor"), "database ACL grantor")
privilege = item.get("privilege_type")
if privilege not in ALLOWED_DATABASE_PRIVILEGES:
raise ValueError(f"source manifest contains unsupported database privilege: {privilege!r}")
is_grantable = item.get("is_grantable")
if type(is_grantable) is not bool:
raise ValueError("source manifest database ACL row has invalid is_grantable")
key = (grantor, grantee, str(privilege), is_grantable)
if key in seen:
raise ValueError("source manifest contains a duplicate database ACL grant")
seen.add(key)
parsed.append(
{
"grantor": grantor,
"grantee": grantee,
"privilege": str(privilege),
"is_grantable": is_grantable,
}
)
remaining = sorted(
parsed,
key=lambda row: (row["privilege"], row["grantor"], row["grantee"], row["is_grantable"]),
)
ordered: list[dict[str, Any]] = []
authorized: dict[str, set[str]] = {privilege: set(initial_grantors) for privilege in ALLOWED_DATABASE_PRIVILEGES}
while remaining:
progress = False
for row in list(remaining):
if row["grantor"] not in authorized[row["privilege"]]:
continue
ordered.append(row)
remaining.remove(row)
if row["is_grantable"] and row["grantee"] != "PUBLIC":
authorized[row["privilege"]].add(row["grantee"])
progress = True
if not progress:
blocked = ", ".join(f"{row['grantor']}->{row['grantee']}:{row['privilege']}" for row in remaining)
raise ValueError(f"source manifest database ACL grantor dependencies cannot be replayed: {blocked}")
reset_sql = """DO $database_acl_reset$
DECLARE
acl_grantee text;
BEGIN
FOR acl_grantee IN
SELECT DISTINCT CASE
WHEN acl.grantee = 0 THEN 'PUBLIC'
ELSE pg_get_userbyid(acl.grantee)
END
FROM pg_database database_row
CROSS JOIN LATERAL aclexplode(
coalesce(database_row.datacl, acldefault('d', database_row.datdba))
) acl
WHERE database_row.datname = current_database()
LOOP
IF acl_grantee = 'PUBLIC' THEN
EXECUTE format(
'REVOKE ALL PRIVILEGES ON DATABASE %I FROM PUBLIC CASCADE',
current_database()
);
ELSE
EXECUTE format(
'REVOKE ALL PRIVILEGES ON DATABASE %I FROM %I CASCADE',
current_database(),
acl_grantee
);
END IF;
END LOOP;
END
$database_acl_reset$;"""
statements = [reset_sql]
target = _quote_identifier(database)
for row in ordered:
grantee = "PUBLIC" if row["grantee"] == "PUBLIC" else _quote_identifier(row["grantee"])
suffix = " WITH GRANT OPTION" if row["is_grantable"] else ""
statements.extend(
[
f"SET ROLE {_quote_identifier(row['grantor'])};",
(
f"GRANT {row['privilege']} ON DATABASE {target} TO {grantee}{suffix} "
f"GRANTED BY {_quote_identifier(row['grantor'])};"
),
"RESET ROLE;",
]
)
return {"sql": "\n".join(statements), "grant_count": len(ordered)}
def database_acl_sql(source_manifest: dict[str, Any], database: str) -> str:
return str(_database_acl_plan(source_manifest, database)["sql"])
def execute_database_acl_replay(
docker_bin: str,
container: str,
database: str,
source_manifest: dict[str, Any],
*,
timeout: float,
) -> int:
plan = _database_acl_plan(source_manifest, database)
sql = str(plan["sql"])
completed = _run(
_psql_command(container, database, "-c", sql, docker_bin=docker_bin),
timeout=timeout,
)
_require_success(completed, "database ACL replay")
return int(plan["grant_count"])
def query_scalar(
@ -475,6 +823,10 @@ def _initial_receipt(args: argparse.Namespace, container: str) -> dict[str, Any]
"isolation": None,
},
"application_roles_bootstrapped": [],
"role_setting_statement_count": 0,
"database_role_setting_statement_count": 0,
"role_membership_statement_count": 0,
"database_acl_statement_count": 0,
"key_counts": {},
"canonical_schema_table_count": None,
"parity": {
@ -589,6 +941,7 @@ def run_canary(args: argparse.Namespace) -> dict[str, Any]:
if source_manifest is not None:
phase = "application_role_bootstrap"
role_plan = _application_role_plan(source_manifest, args.database)
receipt["application_roles_bootstrapped"] = execute_application_role_bootstrap(
args.docker_bin,
container,
@ -596,6 +949,17 @@ def run_canary(args: argparse.Namespace) -> dict[str, Any]:
source_manifest,
timeout=args.command_timeout,
)
receipt["role_setting_statement_count"] = role_plan["role_setting_statement_count"]
receipt["database_role_setting_statement_count"] = role_plan["database_role_setting_statement_count"]
phase = "role_membership_replay"
receipt["role_membership_statement_count"] = execute_role_membership_replay(
args.docker_bin,
container,
args.database,
source_manifest,
timeout=args.command_timeout,
)
phase = "dump_copy"
copy_result = _run(
@ -617,7 +981,6 @@ def run_canary(args: argparse.Namespace) -> dict[str, Any]:
"-d",
args.database,
"--no-owner",
"--no-privileges",
"--exit-on-error",
DUMP_DESTINATION,
],
@ -626,6 +989,16 @@ def run_canary(args: argparse.Namespace) -> dict[str, Any]:
_require_success(restore_result, "canonical PostgreSQL restore")
receipt["timings_seconds"]["pg_restore"] = round(time.monotonic() - restore_started, 6)
if source_manifest is not None:
phase = "database_acl_replay"
receipt["database_acl_statement_count"] = execute_database_acl_replay(
args.docker_bin,
container,
args.database,
source_manifest,
timeout=args.command_timeout,
)
phase = "key_count_readback"
count_started = time.monotonic()
key_counts, table_count = collect_key_counts(

6
ops/verify_postgres_parity_manifest.py
View file

@ -17,7 +17,7 @@ try:
except ImportError: # pragma: no cover - direct script execution
from private_receipt_io import print_private_receipt_summary, write_private_json
REVIEWED_MANIFEST_SQL_SHA256 = "2bd3e353627d44d4a292949952c06b1897d1a9a9e98243905c7a834d007db188"
REVIEWED_MANIFEST_SQL_SHA256 = "87bb9b7777d6d2e767c864de1d41d87ed157831b73926cd630354db942ddede4"
ROWSET_MD5_RE = re.compile(r"[0-9a-f]{32}\Z")
SINGLETON_KINDS = {
@ -249,6 +249,10 @@ def compare_manifests(
"structural_hashes": structural_hashes,
"required_extension_mismatches": extension_mismatches,
"target_extra_extensions": sorted(set(target_extensions) - set(source_extensions)),
"application_role_hashes": {
"source": canonical_hash([source_roles[name] for name in sorted(source_roles)]),
"target": canonical_hash([target_roles[name] for name in sorted(target_roles)]),
},
"application_role_mismatches": role_mismatches,
"target_extra_application_roles": sorted(set(target_roles) - set(source_roles)),
"performance": performance_rows,

7
tests/test_capture_vps_canonical_postgres_snapshot.py
View file

@ -35,7 +35,8 @@ def test_remote_capture_uses_one_exported_read_only_snapshot() -> None:
assert "pg_current_wal_lsn()::text" in script
assert "system_identifier::text from pg_control_system()" in script
assert "source-snapshot.json" in script
assert "--no-owner --no-acl" in script
assert "--no-owner" in script
assert "--no-acl" not in script
assert "cmp -s" in script
assert 'assert_service_healthy "$remote_root/service-before.txt"' in script
assert 'assert_service_healthy "$remote_root/service-after.txt"' in script
@ -49,7 +50,11 @@ def test_manifest_hash_order_is_collation_independent() -> None:
assert 'collate "C"' in manifest
assert "'server_address', host(inet_server_addr())" in manifest
assert "'kb_gate_owner', 'kb_apply', 'kb_review'" in manifest
assert "from pg_db_role_setting database_setting" in manifest
assert "'database_settings'" in manifest
assert "'kind', 'role_memberships'" in manifest
assert "'inherit_option', membership.inherit_option" in manifest
assert "'set_option', membership.set_option" in manifest
assert "'kind', 'object_ownership'" in manifest
assert "'kind', 'object_acl'" in manifest
assert "aclexplode(attribute.attacl)" in manifest

210
tests/test_run_local_canonical_postgres_rebuild.py
View file

@ -86,6 +86,10 @@ class FakeDocker:
return completed(command, stdout="teleo\n")
if sql.startswith("CREATE ROLE"):
return completed(command)
if 'GRANT "kb_observatory_read" TO "kb_apply"' in sql:
return completed(command)
if "ON DATABASE" in sql and "$database_acl_reset$" in sql:
return completed(command)
if sql.startswith("select to_regclass("):
return completed(command, stdout="t\n")
if "from pg_tables" in sql:
@ -111,16 +115,36 @@ def write_dump(path: Path, payload: bytes = b"fixture") -> Path:
def manifest_rows() -> list[dict]:
role = {
"name": "kb_apply",
"can_login": True,
"superuser": False,
"create_db": False,
"create_role": False,
"inherit": True,
"replication": False,
"bypass_rls": False,
}
roles = [
{
"name": "kb_apply",
"can_login": True,
"superuser": False,
"create_db": False,
"create_role": False,
"inherit": True,
"replication": False,
"bypass_rls": False,
"connection_limit": -1,
"default_transaction_read_only": None,
"settings": [],
"database_settings": [{"name": "statement_timeout", "value": "17s"}],
},
{
"name": "kb_observatory_read",
"can_login": False,
"superuser": False,
"create_db": False,
"create_role": False,
"inherit": False,
"replication": False,
"bypass_rls": False,
"connection_limit": -1,
"default_transaction_read_only": "on",
"settings": [{"name": "default_transaction_read_only", "value": "on"}],
"database_settings": [],
},
]
rows = [
{
"kind": "identity",
@ -136,10 +160,76 @@ def manifest_rows() -> list[dict]:
},
{"kind": "schemas", "items": ["kb_stage", "public"]},
{"kind": "extensions", "items": []},
{"kind": "application_roles", "items": [role]},
{"kind": "role_memberships", "items": []},
{"kind": "object_ownership", "items": []},
{"kind": "object_acl", "items": []},
{"kind": "application_roles", "items": roles},
{
"kind": "role_memberships",
"items": [
{
"role": "kb_observatory_read",
"member": "kb_apply",
"grantor": "postgres",
"admin_option": True,
"inherit_option": False,
"set_option": False,
}
],
},
{
"kind": "object_ownership",
"items": [
{
"object_type": "database",
"schema": None,
"name": "canonical_database",
"identity_arguments": None,
"owner": "postgres",
}
],
},
{
"kind": "object_acl",
"items": [
{
"object_type": "database",
"schema": None,
"name": "canonical_database",
"grantor": "postgres",
"grantee": "PUBLIC",
"privilege_type": "CONNECT",
"is_grantable": False,
},
*[
{
"object_type": "database",
"schema": None,
"name": "canonical_database",
"grantor": "postgres",
"grantee": "postgres",
"privilege_type": privilege,
"is_grantable": False,
}
for privilege in ("CONNECT", "CREATE", "TEMPORARY")
],
{
"object_type": "database",
"schema": None,
"name": "canonical_database",
"grantor": "postgres",
"grantee": "kb_observatory_read",
"privilege_type": "CONNECT",
"is_grantable": True,
},
{
"object_type": "database",
"schema": None,
"name": "canonical_database",
"grantor": "kb_observatory_read",
"grantee": "kb_apply",
"privilege_type": "CONNECT",
"is_grantable": False,
},
],
},
]
rows.extend(
{"kind": kind, "items": []}
@ -239,7 +329,7 @@ def test_success_uses_networkless_tmpfs_named_database_and_required_restore_flag
restore_command = fake.commands[restore_index]
assert "--no-owner" in restore_command
assert "--no-privileges" in restore_command
assert "--no-privileges" not in restore_command
assert "--exit-on-error" in restore_command
@ -254,7 +344,11 @@ def test_optional_manifest_bootstraps_allowlisted_roles_and_uses_existing_verifi
result = rebuild.run_canary(args_for(tmp_path, source_manifest=source_manifest))
assert result["status"] == "pass"
assert result["application_roles_bootstrapped"] == ["kb_apply"]
assert result["application_roles_bootstrapped"] == ["kb_apply", "kb_observatory_read"]
assert result["role_setting_statement_count"] == 1
assert result["database_role_setting_statement_count"] == 1
assert result["role_membership_statement_count"] == 1
assert result["database_acl_statement_count"] == 6
assert result["parity"]["status"] == "pass"
assert result["parity"]["verifier"] == "ops.verify_postgres_parity_manifest.compare_manifests"
assert result["parity"]["problems"] == []
@ -268,7 +362,37 @@ def test_optional_manifest_bootstraps_allowlisted_roles_and_uses_existing_verifi
)
role_sql = role_command[role_command.index("-c") + 1]
assert 'CREATE ROLE "kb_apply" LOGIN NOSUPERUSER' in role_sql
assert 'CREATE ROLE "kb_observatory_read" NOLOGIN NOSUPERUSER' in role_sql
assert 'ALTER ROLE "kb_observatory_read" SET "default_transaction_read_only" TO \'on\';' in role_sql
assert 'ALTER ROLE "kb_apply" IN DATABASE "teleo" SET "statement_timeout" TO \'17s\';' in role_sql
assert "PASSWORD" not in role_sql
membership_command = next(
command
for command in fake.commands
if command[1] == "exec"
and command[3] == "psql"
and any('GRANT "kb_observatory_read" TO "kb_apply"' in part for part in command)
)
membership_sql = membership_command[membership_command.index("-c") + 1]
assert 'SET ROLE "postgres";' in membership_sql
assert (
'GRANT "kb_observatory_read" TO "kb_apply" WITH ADMIN TRUE, INHERIT FALSE, SET FALSE GRANTED BY "postgres";'
) in membership_sql
acl_command = next(
command
for command in fake.commands
if command[1] == "exec" and command[3] == "psql" and any("$database_acl_reset$" in part for part in command)
)
acl_sql = acl_command[acl_command.index("-c") + 1]
assert "REVOKE ALL PRIVILEGES ON DATABASE %I FROM PUBLIC CASCADE" in acl_sql
assert 'GRANT CONNECT ON DATABASE "teleo" TO PUBLIC GRANTED BY "postgres";' in acl_sql
assert (
'GRANT CONNECT ON DATABASE "teleo" TO "kb_observatory_read" WITH GRANT OPTION GRANTED BY "postgres";'
) in acl_sql
assert (
'SET ROLE "kb_observatory_read";\n'
'GRANT CONNECT ON DATABASE "teleo" TO "kb_apply" GRANTED BY "kb_observatory_read";'
) in acl_sql
assert any(command[1] == "cp" and rebuild.MANIFEST_DESTINATION in command[-1] for command in fake.commands)
@ -360,7 +484,59 @@ def test_application_role_sql_rejects_roles_outside_exact_allowlist() -> None:
source = {"singleton": {"application_roles": {"items": [{"name": "postgres"}]}}}
with pytest.raises(ValueError, match="non-allowlisted"):
rebuild.application_role_sql(source)
rebuild.application_role_sql(source, "teleo")
def test_application_role_sql_fails_closed_without_database_scoped_settings() -> None:
source = {"singleton": {row["kind"]: row for row in manifest_rows() if "kind" in row}}
del source["singleton"]["application_roles"]["items"][0]["database_settings"]
with pytest.raises(ValueError, match="database_settings must be a list"):
rebuild.application_role_sql(source, "teleo")
def test_role_membership_sql_preserves_grantor_and_all_option_bits() -> None:
source = {"singleton": {row["kind"]: row for row in manifest_rows() if "kind" in row}}
sql = rebuild.role_membership_sql(source)
assert sql == (
'SET ROLE "postgres";\n'
'GRANT "kb_observatory_read" TO "kb_apply" WITH ADMIN TRUE, INHERIT FALSE, SET FALSE '
'GRANTED BY "postgres";\n'
"RESET ROLE;"
)
def test_database_acl_sql_resets_defaults_and_replays_public_and_grantor_chain() -> None:
source = {"singleton": {row["kind"]: row for row in manifest_rows() if "kind" in row}}
sql = rebuild.database_acl_sql(source, "teleo")
assert "$database_acl_reset$" in sql
assert 'TO PUBLIC GRANTED BY "postgres";' in sql
parent = 'GRANT CONNECT ON DATABASE "teleo" TO "kb_observatory_read" WITH GRANT OPTION GRANTED BY "postgres";'
child = 'GRANT CONNECT ON DATABASE "teleo" TO "kb_apply" GRANTED BY "kb_observatory_read";'
assert parent in sql
assert child in sql
assert sql.index(parent) < sql.index(child)
def test_database_acl_sql_fails_closed_on_unrepresented_grantor_dependency() -> None:
source = {"singleton": {row["kind"]: row for row in manifest_rows() if "kind" in row}}
child = source["singleton"]["object_acl"]["items"][-1]
child["grantor"] = "unrepresented_grantor"
with pytest.raises(ValueError, match="grantor dependencies cannot be replayed"):
rebuild.database_acl_sql(source, "teleo")
def test_database_acl_sql_rejects_unreviewed_privilege() -> None:
source = {"singleton": {row["kind"]: row for row in manifest_rows() if "kind" in row}}
source["singleton"]["object_acl"]["items"][0]["privilege_type"] = "DROP"
with pytest.raises(ValueError, match="unsupported database privilege"):
rebuild.database_acl_sql(source, "teleo")
def test_container_names_are_unique_and_bounded() -> None:

307
tests/test_run_local_canonical_postgres_rebuild_postgres.py Normal file
View file

@ -0,0 +1,307 @@
from __future__ import annotations
import json
import shutil
import subprocess
import sys
import time
import uuid
from pathlib import Path
import pytest
from ops import run_local_canonical_postgres_rebuild as rebuild
def _docker_available() -> bool:
if shutil.which("docker") is None:
return False
completed = subprocess.run(
["docker", "info", "--format", "{{.ServerVersion}}"],
text=True,
capture_output=True,
check=False,
)
return completed.returncode == 0
def _docker_psql(container: str, database: str, sql: str) -> subprocess.CompletedProcess[str]:
return subprocess.run(
[
"docker",
"exec",
"-i",
container,
"psql",
"-X",
"-U",
"postgres",
"-d",
database,
"-Atq",
"-v",
"ON_ERROR_STOP=1",
],
input=sql,
text=True,
capture_output=True,
check=False,
)
def _wait_for_postgres(container: str, database: str) -> None:
for _ in range(120):
completed = _docker_psql(container, database, "select 1;")
if completed.returncode == 0:
return
time.sleep(0.25)
pytest.fail(f"fixture PostgreSQL did not become ready: {container}")
def _require_success(completed: subprocess.CompletedProcess[str], action: str) -> None:
if completed.returncode != 0:
pytest.fail(
f"{action} failed ({completed.returncode})\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}"
)
def _capture_manifest(container: str, database: str, destination: Path) -> None:
copy = subprocess.run(
["docker", "cp", str(rebuild.DEFAULT_MANIFEST_SQL), f"{container}:/tmp/parity.sql"],
text=True,
capture_output=True,
check=False,
)
_require_success(copy, "manifest SQL copy")
capture = subprocess.run(
[
"docker",
"exec",
container,
"psql",
"-X",
"-U",
"postgres",
"-d",
database,
"-Atq",
"-v",
"ON_ERROR_STOP=1",
"-f",
"/tmp/parity.sql",
],
text=True,
capture_output=True,
check=False,
)
_require_success(capture, "source manifest capture")
destination.write_text(capture.stdout, encoding="utf-8")
def _capture_dump(container: str, database: str, destination: Path) -> None:
dump = subprocess.run(
[
"docker",
"exec",
container,
"pg_dump",
"-U",
"postgres",
"-d",
database,
"--format=custom",
"--file=/tmp/canonical.dump",
],
text=True,
capture_output=True,
check=False,
)
_require_success(dump, "fixture pg_dump")
copy = subprocess.run(
["docker", "cp", f"{container}:/tmp/canonical.dump", str(destination)],
text=True,
capture_output=True,
check=False,
)
_require_success(copy, "fixture dump copy")
def _fixture_sql(database: str) -> str:
return f"""
create schema kb_stage;
create table public.claims (id bigint primary key, body text not null);
create table public.sources (id bigint primary key, url text not null);
create table public.claim_evidence (
id bigint primary key,
claim_id bigint not null references public.claims(id),
source_id bigint not null references public.sources(id)
);
create table public.claim_edges (
id bigint primary key,
from_claim bigint not null references public.claims(id),
to_claim bigint not null references public.claims(id)
);
create table public.reasoning_tools (id bigint primary key, name text not null);
create table kb_stage.kb_proposals (
id bigint primary key,
status text not null,
created_at timestamptz not null
);
insert into public.claims values (1, 'fixture claim');
insert into public.sources values (1, 'https://example.invalid/source');
insert into public.claim_evidence values (1, 1, 1);
insert into public.claim_edges values (1, 1, 1);
insert into public.reasoning_tools values (1, 'fixture tool');
insert into kb_stage.kb_proposals values (1, 'pending_review', '2026-07-16T00:00:00Z');
create role kb_review nologin noinherit;
create role kb_apply login inherit;
create role kb_observatory_read nologin noinherit;
alter role kb_apply set application_name to 'teleo=parity-fixture';
alter role kb_apply in database "{database}" set statement_timeout to '17s';
alter role kb_observatory_read in database "{database}" set default_transaction_read_only to 'on';
grant kb_review to kb_apply
with admin true, inherit false, set false
granted by postgres;
set role kb_apply;
grant kb_review to kb_observatory_read
with admin false, inherit true, set false
granted by kb_apply;
reset role;
revoke all privileges on database "{database}" from public cascade;
grant connect on database "{database}" to public granted by postgres;
grant connect on database "{database}" to kb_review with grant option granted by postgres;
set role kb_review;
grant connect on database "{database}" to kb_apply granted by kb_review;
reset role;
"""
@pytest.mark.skipif(not _docker_available(), reason="Docker daemon is required for PostgreSQL parity")
def test_real_postgres_rebuild_replays_scoped_settings_memberships_public_acl_and_grantors(
tmp_path: Path,
) -> None:
source_container = f"teleo-pr173-source-{uuid.uuid4().hex[:12]}"
source_database = "teleo_source"
started = subprocess.run(
[
"docker",
"run",
"--rm",
"--detach",
"--network",
"none",
"--name",
source_container,
"--tmpfs",
f"{rebuild.DATA_DIR}:rw,nosuid,nodev,noexec,size=256m",
"--env",
f"POSTGRES_DB={source_database}",
"--env",
"POSTGRES_HOST_AUTH_METHOD=trust",
rebuild.DEFAULT_IMAGE,
],
text=True,
capture_output=True,
check=False,
)
_require_success(started, "source PostgreSQL start")
cleanup_verified = False
try:
_wait_for_postgres(source_container, source_database)
bootstrap = _docker_psql(source_container, source_database, _fixture_sql(source_database))
_require_success(bootstrap, "source parity fixture bootstrap")
source_manifest_path = tmp_path / "source-manifest.jsonl"
source_dump_path = tmp_path / "source.dump"
_capture_manifest(source_container, source_database, source_manifest_path)
_capture_dump(source_container, source_database, source_dump_path)
source_manifest = rebuild.load_manifest(source_manifest_path)
roles = {row["name"]: row for row in source_manifest["singleton"]["application_roles"]["items"]}
assert roles["kb_apply"]["settings"] == [{"name": "application_name", "value": "teleo=parity-fixture"}]
assert roles["kb_apply"]["database_settings"] == [{"name": "statement_timeout", "value": "17s"}]
assert roles["kb_observatory_read"]["database_settings"] == [
{"name": "default_transaction_read_only", "value": "on"}
]
memberships = source_manifest["singleton"]["role_memberships"]["items"]
assert {row["grantor"] for row in memberships} == {"postgres", "kb_apply"}
assert any(
row["admin_option"] is True and row["inherit_option"] is False and row["set_option"] is False
for row in memberships
)
database_acls = [
row for row in source_manifest["singleton"]["object_acl"]["items"] if row["object_type"] == "database"
]
assert any(row["grantee"] == "PUBLIC" and row["privilege_type"] == "CONNECT" for row in database_acls)
assert not any(row["grantee"] == "PUBLIC" and row["privilege_type"] == "TEMPORARY" for row in database_acls)
assert any(row["grantor"] == "kb_review" and row["grantee"] == "kb_apply" for row in database_acls)
receipt_path = tmp_path / "parity-receipt.json"
completed = subprocess.run(
[
sys.executable,
"ops/run_local_canonical_postgres_rebuild.py",
"--dump",
str(source_dump_path),
"--database",
"teleo_target",
"--source-manifest",
str(source_manifest_path),
"--manifest-sql",
str(rebuild.DEFAULT_MANIFEST_SQL),
"--tmpfs-mb",
"256",
"--max-target-query-ms",
"10000",
"--max-target-source-ratio",
"1000",
"--output",
str(receipt_path),
],
text=True,
capture_output=True,
check=False,
)
receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
assert completed.returncode == 0, completed.stdout + completed.stderr
assert receipt["status"] == "pass", receipt.get("error")
assert receipt["application_roles_bootstrapped"] == [
"kb_apply",
"kb_observatory_read",
"kb_review",
]
assert receipt["role_setting_statement_count"] == 1
assert receipt["database_role_setting_statement_count"] == 2
assert receipt["role_membership_statement_count"] == 2
assert receipt["database_acl_statement_count"] == len(database_acls)
assert receipt["parity"]["status"] == "pass"
assert receipt["parity"]["problems"] == []
structural = receipt["parity"]["details"]["structural_hashes"]
assert structural["role_memberships"]["source"] == structural["role_memberships"]["target"]
assert structural["object_acl"]["source"] == structural["object_acl"]["target"]
application_roles = receipt["parity"]["details"]["application_role_hashes"]
assert application_roles["source"] == application_roles["target"]
assert receipt["parity"]["details"]["application_role_mismatches"] == {}
assert receipt["parity"]["source_manifest"]["sha256"] != receipt["parity"]["target_manifest"]["sha256"]
assert receipt["cleanup"]["status"] == "pass"
assert receipt["cleanup"]["container_absent"] is True
finally:
subprocess.run(
["docker", "rm", "--force", source_container],
text=True,
capture_output=True,
check=False,
)
inspected = subprocess.run(
["docker", "container", "inspect", source_container],
text=True,
capture_output=True,
check=False,
)
cleanup_verified = inspected.returncode != 0
assert cleanup_verified is True

4
tests/test_verify_postgres_parity_manifest.py
View file

@ -106,6 +106,10 @@ def test_matching_local_manifests_pass(tmp_path: Path) -> None:
assert completed.returncode == 0, completed.stderr
assert payload["status"] == "pass"
assert payload["details"]["source_total_rows"] == 2
assert (
payload["details"]["application_role_hashes"]["source"]
== (payload["details"]["application_role_hashes"]["target"])
)
assert len(payload["source_manifest_sha256"]) == 64
assert len(payload["target_manifest_sha256"]) == 64
assert payload["private_connectivity"]["status"] == "not_applicable_local_restore"