You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
76 lines
4.5 KiB
76 lines
4.5 KiB
"""Run against the migrated development database; all DML is rolled back.""" |
|
import os |
|
import uuid |
|
|
|
import psycopg |
|
from psycopg.errors import InsufficientPrivilege |
|
|
|
|
|
def check_permissions(): |
|
with psycopg.connect(host=os.environ["DB_HOST"], dbname=os.environ["DB_NAME"], |
|
user=os.environ["DB_USER"], password=os.environ["DB_PASSWORD"], |
|
autocommit=True) as conn: |
|
row = conn.execute("SELECT rolsuper, rolcreatedb, rolcreaterole FROM pg_roles WHERE rolname = current_user").fetchone() |
|
assert row == (False, False, False), "Runtime role has elevated privileges" |
|
with conn.transaction(force_rollback=True): |
|
name = "synthetic_permission_" + uuid.uuid4().hex |
|
pk = conn.execute("INSERT INTO auth_group (name) VALUES (%s) RETURNING id", (name,)).fetchone()[0] |
|
assert conn.execute("SELECT name FROM auth_group WHERE id = %s", (pk,)).fetchone() == (name,) |
|
conn.execute("UPDATE auth_group SET name = %s WHERE id = %s", (name + "_updated", pk)) |
|
assert conn.execute("SELECT name FROM auth_group WHERE id = %s", (pk,)).fetchone() == (name + "_updated",) |
|
conn.execute("DELETE FROM auth_group WHERE id = %s", (pk,)) |
|
assert conn.execute("SELECT id FROM auth_group WHERE id = %s", (pk,)).fetchone() is None |
|
for statement in ["CREATE TABLE public.permission_probe (id integer)", |
|
"CREATE SCHEMA permission_probe", |
|
"SET ROLE gizmoduck_migrate"]: |
|
try: |
|
with conn.transaction(force_rollback=True): |
|
conn.execute(statement) |
|
except InsufficientPrivilege: |
|
pass |
|
else: |
|
raise AssertionError("Runtime unexpectedly permitted: " + statement) |
|
application_tables = conn.execute( |
|
"SELECT tablename FROM pg_tables " |
|
"WHERE schemaname = 'public' " |
|
"AND (tablename LIKE 'finance_%' OR tablename LIKE 'ingestion_%')" |
|
).fetchall() |
|
assert len(application_tables) >= 11, "Application migrations are missing" |
|
for (table,) in application_tables: |
|
for privilege in ("SELECT", "INSERT", "UPDATE", "DELETE"): |
|
assert conn.execute("SELECT has_table_privilege(current_user, %s, %s)", ("public." + table, privilege)).fetchone()[0], (table, privilege) |
|
with conn.transaction(force_rollback=True): |
|
account_id = uuid.uuid4() |
|
conn.execute("INSERT INTO finance_account (id, name, kind, currency, archived, created_at, updated_at) VALUES (%s, %s, 'BANK', 'BRL', false, now(), now())", (account_id, "synthetic_" + account_id.hex)) |
|
assert conn.execute("SELECT currency FROM finance_account WHERE id = %s", (account_id,)).fetchone() == ("BRL",) |
|
conn.execute("UPDATE finance_account SET archived = true WHERE id = %s", (account_id,)) |
|
assert conn.execute("SELECT archived FROM finance_account WHERE id = %s", (account_id,)).fetchone() == (True,) |
|
conn.execute("DELETE FROM finance_account WHERE id = %s", (account_id,)) |
|
with conn.transaction(force_rollback=True): |
|
batch_id = uuid.uuid4() |
|
conn.execute( |
|
"INSERT INTO ingestion_importbatch " |
|
"(id, created_at, updated_at, document_hash, source_format, " |
|
"profile, parser_version, encoding, context, section_mapping, " |
|
"summary, snapshot, actor_kind, actor_identifier, state, revision, result) " |
|
"VALUES (%s, now(), now(), %s, 'BANK_CSV', 'BANK_CSV_V1', " |
|
"'1', 'utf-8', '{}'::jsonb, '{}'::jsonb, '{}'::jsonb, " |
|
"'{}'::jsonb, 'AGENT', 'synthetic-permission-check', " |
|
"'PREVIEW', 1, '{}'::jsonb)", |
|
(batch_id, "0" * 64), |
|
) |
|
assert conn.execute( |
|
"SELECT state FROM ingestion_importbatch WHERE id = %s", (batch_id,) |
|
).fetchone() == ("PREVIEW",) |
|
conn.execute( |
|
"UPDATE ingestion_importbatch SET state = 'CANCELLED' WHERE id = %s", |
|
(batch_id,), |
|
) |
|
conn.execute("DELETE FROM ingestion_importbatch WHERE id = %s", (batch_id,)) |
|
# PostgreSQL CREATE DATABASE cannot run in a rollback transaction. The |
|
# server's role attributes above verify that capability without creating one. |
|
print("PASS: finance/ingestion DML, sequences, denied DDL and role escalation") |
|
|
|
|
|
if __name__ == "__main__": |
|
check_permissions()
|
|
|