In postgres, I created "test" database with superuser_name and then I created new user "app_user". Then I revoke app_user from create table for any public schema. And then I gave all permission on test database to app_user. So this way app_user can create tables in only test database. The steps:
- psql -h localhost -p 5432 -U superuser_name
- CREATE DATABASE test WITH OWNER superuser_name ENCODING 'UTF8'
- CREATE USER app_user WITH PASSWORD 'app_password'
- REVOKE CREATE ON schema public FROM public
- GRANT ALL PRIVILEGES ON DATABASE test TO app_user
- now quit and login back as app_user
- psql -h localhost -p 5432 -U app_user -d test
test=> \dn+
List of schemas
Name | Owner | Access privileges | Description
--------+----------+----------------------+------------------------
public | postgres | postgres=UC/postgres+| standard public schema
| | =UC/postgres |
(1 row)
- by following above steps, app_user is able to create table only in test database and not allow to create table in any other databases. That is what I was expecting. But When I run above steps via code then I am not getting same behaviors. Meaning, newly created app_user do not have permission to create table on newly created test database. And output for dn+ command as below.
test=> \dn+
List of schemas
Name | Owner | Access privileges | Description
--------+----------+----------------------+------------------------
public | postgres | postgres=UC/postgres+| standard public schema
| | =U/postgres |
(1 row)
Python code, to create database, user, give and revoke permission, that I am running.
from collections import namedtuple
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from contextlib import contextmanager
from urllib.parse import quote_plus
from dotenv import load_dotenv
import time, os
import logging
log = logging.getLogger(__name__)
load_dotenv()
SLEEP_TIME_BEFORE_CREATING_TABLES = 60
PG_ENGINE_URL = 'postgresql://{user}:{password}@{host}:{port}/{dbname}'
DbConfig = namedtuple('DbConfig', 'host port user password dbname')
def postgres_config():
"""
Return postgres connection config.
"""
pg_cfg = {
'host': os.environ.get('HOST','localhost'),
'port': os.environ.get('PORT', 5432),
'superuser_name': os.environ.get('SUPERUSER_NAME', 'postgres'),
'superuser_password': os.environ.get('SUPERUSER_PASSWORD','postgres'),
'app_user': os.environ.get('APP_USERNAME', 'visualrf'),
'app_password': os.environ.get('APP_PASSWORD','visualrf123'),
}
return pg_cfg
def engine_url(cfg):
url = PG_ENGINE_URL.format(host=cfg.host, port=cfg.port,
dbname=cfg.dbname, user=cfg.user, password=quote_plus(cfg.password))
return url
def engine(cfg, serialize=False):
"""
Return postgres engine url to connect with postgress databse.
"""
if serialize:
return create_engine(engine_url(cfg), isolation_level="SERIALIZABLE",
client_encoding='utf8')
else:
return create_engine(engine_url(cfg), client_encoding='utf8')
def session(cfg):
"""
Return sqlalchemy session to database using QueuePool.
"""
return Session(bind=engine(cfg), expire_on_commit=False)
def anon_session(host, port, user, password):
"""
Make a session to template1 db which is always present in postgres.
Useful to execute SQL before a particular db is created.
"""
return session(DbConfig(host, port, user, password, 'template1'))
@contextmanager
def terminating_sn(sn_or_cfg):
""" A contextlib which closes session and db connections after use. """
sn = session(sn_or_cfg) if isinstance(sn_or_cfg, DbConfig) else sn_or_cfg
try:
yield sn
finally:
sn.close()
sn.bind.dispose()
def db_exists(host, port, user, password, dbname):
with terminating_sn(anon_session(host, port, user, password)) as sn:
r = sn.execute("SELECT 1 FROM pg_catalog.pg_database d WHERE d.datname='%s'" % dbname).fetchone()
return True if r else False
def _db_exists(sn, dbname):
""" True if DB exists """
r = sn.execute("SELECT 1 FROM pg_catalog.pg_database d WHERE d.datname='%s'" % dbname).fetchone()
return True if r else False
def _user_exists(sn,app_user):
""" True if user/rolname exists """
r = sn.execute("SELECT 1 FROM pg_user WHERE pg_user.usename='%s'" % (app_user))
return True if r.rowcount!=0 else False
def create_db(dbname, dbconf, owner=None):
""" Pick a DB host and create a database in it """
owner = owner or dbconf['superuser_name']
with terminating_sn(anon_session(dbconf['host'], dbconf['port'], dbconf['superuser_name'], dbconf['superuser_password'])) as sn: # connecting to postgres session
sn.connection().connection.set_isolation_level(0)
if _db_exists(sn, dbname):
return False
sn.execute("CREATE DATABASE %s WITH OWNER %s ENCODING 'UTF8'" % (dbname, owner)) # create database
log.info('Provisioned db:%s for host:%s', dbname, dbconf['host'])
return True
def create_user(dbname, dbconf):
""" Create app user for <dbname> and grant att read write permition """
if db_exists(dbconf['host'], dbconf['port'], dbconf['superuser_name'], dbconf['superuser_password'], dbname): # create anon session to check db exist.
with terminating_sn(DbConfig(dbconf['host'], dbconf['port'], dbconf['superuser_name'], dbconf['superuser_password'], dbname)) as sn: # if db exist then make connection with mentioned dbname.
sn.connection().connection.set_isolation_level(0)
if not _user_exists(sn, dbconf['app_user']): # check if user already exist, if not then create one
sn.execute("CREATE USER %s WITH PASSWORD '%s'" % (dbconf['app_user'], dbconf['app_password']))
log.info("Created user:%s for db:%s on host:%s", dbconf['app_user'], dbname, dbconf['host'])
else:
log.info("User '%s' already exist for db:%s on host:%s", dbconf['app_user'], dbname, dbconf['host'])
#Grant all privileges to app_user on the database:
sn.execute("REVOKE CREATE ON schema public FROM public")
sn.execute("GRANT ALL PRIVILEGES ON DATABASE %s TO %s" % (dbname, dbconf['app_user']))
return True
else:
log.error("Can not create user:%s because db:%s for host:%s do not exist", dbconf['app_user'], dbname, dbconf['host'])
return False
def create_table(dbname, dbconf, sqla_metadata):
""" create tables for <dbname> database using created app user """
if db_exists(dbconf['host'], dbconf['port'], dbconf['app_user'], dbconf['app_password'], dbname) and sqla_metadata: # if database exist and we have metabadat to create tables
with terminating_sn(DbConfig(dbconf['host'], dbconf['port'], dbconf['app_user'], dbconf['app_password'], dbname)) as sn:
sqla_metadata.create_all(bind=sn.bind) # create tables
else:
log.error('Can not create tables because db:%s for host:%s do not exist', dbname, dbconf['host'])
return False
log.info('Created tables for db:%s on host:%s', dbname, dbconf['host'])
return True
And above python script will execute in this order:
created_db = pg.create_db(dbname, pg_cfg)
created_user = pg.create_user(dbname, pg_cfg) # creating app user and
grant all privileges for <dbname> database
created_table = pg.create_table(dbname, pg_cfg, AppBase.metadata)
So I am not able to figured out Why I am having this different behavior. Why app_user via python script not able to create table in test database?
Also, My requirements is, I just want to create new user and new db. then grant all privileges to new user for new db. and prevent new user to create tables in any other database except new db. what is the straightforward way to achieve this?
Best Answer
The statement
affects the database to which you are currently connected.
Since the script you are showing in the beginning does not contain a connect to the newly created database, the statement will affect the database to which you were connected when you ran
CREATE DATABASE
rather than the newly created database.After creating the database, add the following to your
psql
script: