I know it's not very likely that you'd check database_exists with a user-controlled database name, but I think it's enough of an issue and an easy enough fix to bring attention to. The relevant code boils down to:
url = make_url(url)
database = url.database
text = "SELECT 1 FROM pg_database WHERE datname='%s'" % database
with engine.connect() as conn:
return conn.scalar(sa.text(text))
A user-controlled URL can run arbitrary SQL:
url_str = """postgresql://username:password@localhost:5432/test'; CREATE TABLE hello (id int); COMMIT; SELECT 1 FROM pg_database;--"""
database_exists(url_str) # has now created table postgres.public.hello
I know it's not very likely that you'd check database_exists with a user-controlled database name, but I think it's enough of an issue and an easy enough fix to bring attention to. The relevant code boils down to:
A user-controlled URL can run arbitrary SQL: