0f16868f2d
docker-compose.yml passed POSTGRES_PASSWORD as APP_DB_PASSWORD, so the DML-only dtf_app role and the owning administrator shared one credential and the privilege separation bootstrap.py sets up was decorative. APP_DB_PASSWORD is now its own required variable, and bootstrap refuses to run when it matches the administrator password, in both the URL and discrete-field configuration forms. Deploying this requires APP_DB_PASSWORD to be set in the stack environment first; db-init rotates the role to it on the same deploy. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
49 lines
2.3 KiB
Python
49 lines
2.3 KiB
Python
"""One-shot schema/role setup. Only this job receives database admin credentials."""
|
|
import os
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
import psycopg
|
|
from psycopg import sql
|
|
|
|
|
|
def admin_connect():
|
|
if os.environ.get('DATABASE_ADMIN_HOST'):
|
|
return psycopg.connect(
|
|
host=os.environ['DATABASE_ADMIN_HOST'],
|
|
dbname=os.environ['DATABASE_ADMIN_NAME'],
|
|
user=os.environ['DATABASE_ADMIN_USER'],
|
|
password=os.environ['DATABASE_ADMIN_PASSWORD'],
|
|
)
|
|
return psycopg.connect(os.environ['DATABASE_ADMIN_URL'])
|
|
|
|
|
|
def admin_password():
|
|
if os.environ.get('DATABASE_ADMIN_HOST'):
|
|
return os.environ.get('DATABASE_ADMIN_PASSWORD')
|
|
url = os.environ.get('DATABASE_ADMIN_URL', '')
|
|
return urlparse(url).password
|
|
|
|
|
|
def main():
|
|
role = os.environ['APP_DB_USER']
|
|
password = os.environ['APP_DB_PASSWORD']
|
|
if password == admin_password():
|
|
raise RuntimeError('Application and database administrator passwords must differ')
|
|
with admin_connect() as c:
|
|
admin, database = c.execute('SELECT current_user,current_database()').fetchone()
|
|
if role == admin:
|
|
raise RuntimeError('Application and database administrator must differ')
|
|
if not c.execute('SELECT 1 FROM pg_roles WHERE rolname=%s', (role,)).fetchone():
|
|
c.execute(sql.SQL('CREATE ROLE {} LOGIN').format(sql.Identifier(role)))
|
|
c.execute(sql.SQL('ALTER ROLE {} NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION NOBYPASSRLS PASSWORD {}').format(
|
|
sql.Identifier(role), sql.Literal(password)))
|
|
c.execute(Path(__file__).with_name('schema.sql').read_text())
|
|
c.execute(sql.SQL('REVOKE CREATE ON SCHEMA public FROM PUBLIC'))
|
|
c.execute(sql.SQL('GRANT CONNECT ON DATABASE {} TO {}').format(sql.Identifier(database),sql.Identifier(role)))
|
|
c.execute(sql.SQL('GRANT USAGE ON SCHEMA dtf_local TO {}').format(sql.Identifier(role)))
|
|
c.execute(sql.SQL('GRANT SELECT,INSERT,UPDATE,DELETE ON ALL TABLES IN SCHEMA dtf_local TO {}').format(sql.Identifier(role)))
|
|
c.execute(sql.SQL('GRANT USAGE,SELECT ON ALL SEQUENCES IN SCHEMA dtf_local TO {}').format(sql.Identifier(role)))
|
|
print('Local schema migrated; runtime role has DML only.')
|
|
|
|
if __name__ == '__main__': main()
|