113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
"""
|
|
B1.4 — create the unprivileged database role the application connects as.
|
|
|
|
**A superuser connection bypasses row-level security unconditionally.** Not
|
|
"mostly", not "unless forced" — the policies are simply not evaluated. So an
|
|
application connecting as `postgres` has every table marked protected, every
|
|
query returning plausible rows, and no isolation guarantee whatsoever. That is
|
|
risk R5, and it is invisible: nothing fails, nothing logs, and a probe run
|
|
against a superuser connection passes for the wrong reason.
|
|
|
|
This creates a NOSUPERUSER / NOBYPASSRLS role with exactly the rights the
|
|
application needs, and nothing else.
|
|
|
|
APP_ENV=rework python scripts/create_app_role.py --password '...'
|
|
|
|
Run it as an administrative user. The application's own connection string should
|
|
then point at the new role — that change is what actually makes RLS take effect.
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
os.environ.setdefault("APP_ENV", "rework")
|
|
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
DEFAULT_ROLE = "docqube_app"
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--role", default=DEFAULT_ROLE)
|
|
parser.add_argument("--password", required=True)
|
|
parser.add_argument(
|
|
"--database",
|
|
default=None,
|
|
help="Defaults to the database in the current APP_ENV configuration.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
from sqlalchemy import create_engine, text
|
|
|
|
from app.core.settings import settings
|
|
|
|
database = args.database or settings.DB_NAME
|
|
engine = create_engine(settings.DATABASE_URL, isolation_level="AUTOCOMMIT")
|
|
|
|
statements = [
|
|
# NOSUPERUSER and NOBYPASSRLS are the point of the exercise.
|
|
f"""
|
|
DO $$
|
|
BEGIN
|
|
IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{args.role}') THEN
|
|
CREATE ROLE {args.role} LOGIN PASSWORD '{args.password}'
|
|
NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS;
|
|
ELSE
|
|
ALTER ROLE {args.role} LOGIN PASSWORD '{args.password}'
|
|
NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS;
|
|
END IF;
|
|
END
|
|
$$;
|
|
""",
|
|
f"GRANT CONNECT ON DATABASE {database} TO {args.role}",
|
|
f"GRANT USAGE ON SCHEMA public TO {args.role}",
|
|
f"GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO {args.role}",
|
|
f"GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO {args.role}",
|
|
# Tables created by later migrations must be reachable too.
|
|
f"ALTER DEFAULT PRIVILEGES IN SCHEMA public "
|
|
f"GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO {args.role}",
|
|
f"ALTER DEFAULT PRIVILEGES IN SCHEMA public "
|
|
f"GRANT USAGE, SELECT ON SEQUENCES TO {args.role}",
|
|
]
|
|
|
|
with engine.connect() as conn:
|
|
for statement in statements:
|
|
conn.execute(text(statement))
|
|
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = :r"
|
|
),
|
|
{"r": args.role},
|
|
).first()
|
|
|
|
engine.dispose()
|
|
|
|
if row is None:
|
|
print(f"role {args.role} was not created", file=sys.stderr)
|
|
return 1
|
|
|
|
is_super, bypasses_rls = row
|
|
print(f"role {args.role}")
|
|
print(f"database {database}")
|
|
print(f"superuser {is_super}")
|
|
print(f"bypassrls {bypasses_rls}")
|
|
|
|
if is_super or bypasses_rls:
|
|
print(
|
|
"\nREFUSING TO REPORT SUCCESS: the role can bypass row-level "
|
|
"security, so the policies would not apply to it.",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
print("\nRLS applies to this role. Point the application's DB_USER at it.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|