mirror of
https://github.com/pgadmin-org/pgadmin4.git
synced 2026-08-17 16:34:44 -05:00
Add RBAC regression tests for tool routes, sockets and adhoc ownership
Add regression coverage for the authorisation fixes:
* test_tool_permissions_required: a consolidated, per-blueprint check
that logs in as a user with no roles (hence no tool permissions) and
asserts every gated backend HTTP route across the query tool, grant
wizard, schema diff, ERD, PSQL and debugger returns 403. This catches
any future route added to these blueprints without the decorator.
* test_tool_socket_permissions_required: asserts the schema diff
compare_database/compare_schema and psql start_process Socket.IO
handlers refuse a user lacking the tool permission.
* test_adhoc_connect_server_ownership: asserts that an adhoc connect
triggered by a non-owner against an administrator-owned shared server
persists a server row owned by the caller and not shared.
All three are skipped in DESKTOP mode, where every request is
auto-authenticated as the all-permissions DESKTOP_USER.
This commit is contained in:
Dave Page
Ashesh Vashi
@@ -0,0 +1,8 @@
|
||||
##########################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2026, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
##########################################################################
|
||||
@@ -0,0 +1,138 @@
|
||||
##########################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2026, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
##########################################################################
|
||||
|
||||
"""Regression: adhoc connect must not persist cross-tenant server rows.
|
||||
|
||||
/misc/workspace/adhoc_connect_server clones an existing server when a
|
||||
'sid' is supplied. Server.clone() copies every column of the source row,
|
||||
including user_id/shared/shared_username. When a non-owner triggers an
|
||||
adhoc connect against an administrator-owned *shared* server, the clone
|
||||
must be re-homed to the current user and made private; otherwise pgAdmin
|
||||
persists a new, administrator-owned (user_id of the admin), shared adhoc
|
||||
server row created at the behest of another user, a cross-tenant
|
||||
integrity problem.
|
||||
|
||||
This test creates a shared server as the admin, then, as a non-admin
|
||||
user, calls adhoc_connect_server with that server's id and asserts that
|
||||
every resulting adhoc server row belongs to the non-admin user and is not
|
||||
shared.
|
||||
|
||||
Skipped in DESKTOP mode (single user; no cross-tenant boundary).
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
import config
|
||||
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from regression.python_test_utils import test_utils as utils
|
||||
from regression.test_setup import config_data
|
||||
from regression.python_test_utils.test_utils import \
|
||||
create_user_wise_test_client
|
||||
|
||||
test_user_details = None
|
||||
if config.SERVER_MODE:
|
||||
test_user_details = config_data['pgAdmin4_test_non_admin_credentials']
|
||||
|
||||
|
||||
class AdhocConnectServerOwnershipTestCase(BaseTestGenerator):
|
||||
"""A non-owner adhoc connect against a shared server must not leave a
|
||||
persisted server row owned by another (admin) user."""
|
||||
|
||||
scenarios = [
|
||||
('adhoc clone of a shared server is re-homed to the caller',
|
||||
dict()),
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
self.shared_sid = None
|
||||
if not config.SERVER_MODE:
|
||||
self.skipTest(
|
||||
'Adhoc ownership isolation only applies to server mode.')
|
||||
|
||||
# Create a shared server as the admin user.
|
||||
self.server['shared'] = True
|
||||
url = "/browser/server/obj/{0}/".format(utils.SERVER_GROUP)
|
||||
response = self.tester.post(
|
||||
url, data=json.dumps(self.server), content_type='html/json')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.shared_sid = json.loads(
|
||||
response.data.decode('utf-8'))['node']['_id']
|
||||
|
||||
def _user_id(self, email):
|
||||
from pgadmin.model import User
|
||||
with self.app.app_context():
|
||||
user = User.query.filter_by(username=email).first()
|
||||
return user.id if user else None
|
||||
|
||||
def _adhoc_servers(self):
|
||||
from pgadmin.model import Server
|
||||
with self.app.app_context():
|
||||
return [
|
||||
dict(id=s.id, user_id=s.user_id, shared=bool(s.shared))
|
||||
for s in Server.query.filter_by(is_adhoc=1).all()
|
||||
]
|
||||
|
||||
@create_user_wise_test_client(test_user_details)
|
||||
def runTest(self):
|
||||
if not self.shared_sid:
|
||||
raise Exception('Shared server was not created.')
|
||||
|
||||
admin_email = \
|
||||
config_data['pgAdmin4_login_credentials']['login_username']
|
||||
non_admin_email = test_user_details['login_username']
|
||||
admin_id = self._user_id(admin_email)
|
||||
non_admin_id = self._user_id(non_admin_email)
|
||||
self.assertIsNotNone(non_admin_id)
|
||||
|
||||
# As the non-admin, trigger an adhoc connect that clones the
|
||||
# admin-owned shared server. The connection attempt itself may
|
||||
# fail; what matters is the persisted row.
|
||||
data = dict(
|
||||
server_name='adhoc_isolation_probe',
|
||||
did=self.server.get('did', 1),
|
||||
sid=self.shared_sid,
|
||||
host=self.server['host'],
|
||||
port=self.server['port'],
|
||||
user=self.server['username'],
|
||||
)
|
||||
self.tester.post(
|
||||
'/misc/workspace/adhoc_connect_server',
|
||||
data=json.dumps(data), content_type='application/json')
|
||||
|
||||
adhoc = self._adhoc_servers()
|
||||
self.assertGreaterEqual(
|
||||
len(adhoc), 1,
|
||||
'Expected an adhoc server row to have been persisted.')
|
||||
for row in adhoc:
|
||||
self.assertNotEqual(
|
||||
row['user_id'], admin_id,
|
||||
'Adhoc server row {0} is owned by the administrator '
|
||||
'(user_id={1}); a non-owner created a cross-tenant '
|
||||
'server record.'.format(row['id'], row['user_id']))
|
||||
self.assertEqual(
|
||||
row['user_id'], non_admin_id,
|
||||
'Adhoc server row {0} should be owned by the calling '
|
||||
'user (id={1}), got user_id={2}.'.format(
|
||||
row['id'], non_admin_id, row['user_id']))
|
||||
self.assertFalse(
|
||||
row['shared'],
|
||||
'Adhoc server row {0} must not be shared.'.format(
|
||||
row['id']))
|
||||
|
||||
def tearDown(self):
|
||||
# Remove any adhoc rows left behind, then the shared server.
|
||||
from pgadmin.model import db, Server
|
||||
with self.app.app_context():
|
||||
for s in Server.query.filter_by(is_adhoc=1).all():
|
||||
db.session.delete(s)
|
||||
db.session.commit()
|
||||
if self.shared_sid:
|
||||
utils.delete_server_with_api(
|
||||
self.__class__.tester, self.shared_sid)
|
||||
@@ -0,0 +1,8 @@
|
||||
##########################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2026, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
##########################################################################
|
||||
@@ -0,0 +1,164 @@
|
||||
##########################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2026, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
##########################################################################
|
||||
|
||||
"""RBAC regression: tool backend routes must enforce the tool permission.
|
||||
|
||||
pgAdmin gates each tool behind a permission (tools_query_tool,
|
||||
tools_grant_wizard, tools_schema_diff, tools_erd_tool, tools_psql_tool,
|
||||
tools_debugger, ...). Historically the @permissions_required decorator was
|
||||
applied only to a single 'front door' route per tool, while alternate
|
||||
initialisation routes, object-discovery/SQL/apply routes and Socket.IO
|
||||
handlers relied on @pga_login_required alone. That let a user who had been
|
||||
denied a tool still drive the rest of that tool's backend workflow:
|
||||
|
||||
* View/Edit Data via sqleditor.initialize_viewdata (Query Tool),
|
||||
* object discovery, SQL preview and the real GRANT via the grant wizard,
|
||||
* schema diff initialisation, enumeration and comparison,
|
||||
* ERD initialisation and DDL generation,
|
||||
* the PSQL panel,
|
||||
* the debugger's stored-argument routes.
|
||||
|
||||
This test logs in as a user with no roles (therefore no tool permissions)
|
||||
and asserts that every one of those backend routes returns HTTP 403, i.e.
|
||||
the permission gate fires before the route body runs. It is deliberately a
|
||||
single consolidated 'blanket' check so that a newly-added route in any of
|
||||
these blueprints which forgets the decorator is caught here.
|
||||
|
||||
The permission check is the outermost decorator, so it runs before any
|
||||
connection/transaction lookup; dummy path parameters (ids of 1, a random
|
||||
trans_id) are sufficient to reach and trip it.
|
||||
|
||||
Skipped in DESKTOP mode, where every request is auto-authenticated as the
|
||||
all-permissions DESKTOP_USER and no permission decorator is exercisable.
|
||||
"""
|
||||
|
||||
import secrets
|
||||
|
||||
import config
|
||||
import flask
|
||||
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from regression.test_setup import config_data
|
||||
from regression.python_test_utils.test_utils import \
|
||||
create_user_wise_test_client
|
||||
|
||||
test_user_details = None
|
||||
if config.SERVER_MODE:
|
||||
test_user_details = config_data['pgAdmin4_test_non_admin_credentials']
|
||||
|
||||
|
||||
# A throwaway transaction id; these routes never get far enough to use it.
|
||||
_TRANS = secrets.choice(range(1, 9999999))
|
||||
|
||||
|
||||
class ToolPermissionRequiredTestCase(BaseTestGenerator):
|
||||
"""A user lacking a tool's permission must get 403 from every backend
|
||||
route in that tool, not just its primary entry point."""
|
||||
|
||||
scenarios = [
|
||||
# --- Query Tool: View/Edit Data alternate init (AC-001) ---
|
||||
('sqleditor.initialize_viewdata requires tools_query_tool',
|
||||
dict(method='post', endpoint='sqleditor.initialize_viewdata',
|
||||
url_kwargs=dict(trans_id=_TRANS, cmd_type=1, obj_type='table',
|
||||
sgid=1, sid=1, did=1, obj_id=1))),
|
||||
|
||||
# --- Grant Wizard: discovery, SQL preview, apply (AC-002) ---
|
||||
('grant_wizard.objects requires tools_grant_wizard',
|
||||
dict(method='get', endpoint='grant_wizard.objects',
|
||||
url_kwargs=dict(sid=1, did=1, node_id=1, node_type='table'))),
|
||||
('grant_wizard.modified_sql requires tools_grant_wizard',
|
||||
dict(method='post', endpoint='grant_wizard.modified_sql',
|
||||
url_kwargs=dict(sid=1, did=1))),
|
||||
('grant_wizard.apply requires tools_grant_wizard',
|
||||
dict(method='post', endpoint='grant_wizard.apply',
|
||||
url_kwargs=dict(sid=1, did=1))),
|
||||
|
||||
# --- Schema Diff: init, enumeration, connect, ddl (AC-005) ---
|
||||
('schema_diff.initialize requires tools_schema_diff',
|
||||
dict(method='get', endpoint='schema_diff.initialize',
|
||||
url_kwargs=dict(trans_id=_TRANS))),
|
||||
('schema_diff.servers requires tools_schema_diff',
|
||||
dict(method='get', endpoint='schema_diff.servers',
|
||||
url_kwargs=dict())),
|
||||
('schema_diff.get_server requires tools_schema_diff',
|
||||
dict(method='get', endpoint='schema_diff.get_server',
|
||||
url_kwargs=dict(sid=1, did=1))),
|
||||
('schema_diff.connect_server requires tools_schema_diff',
|
||||
dict(method='post', endpoint='schema_diff.connect_server',
|
||||
url_kwargs=dict(sid=1))),
|
||||
('schema_diff.connect_database requires tools_schema_diff',
|
||||
dict(method='post', endpoint='schema_diff.connect_database',
|
||||
url_kwargs=dict(sid=1, did=1))),
|
||||
('schema_diff.databases requires tools_schema_diff',
|
||||
dict(method='get', endpoint='schema_diff.databases',
|
||||
url_kwargs=dict(sid=1))),
|
||||
('schema_diff.schemas requires tools_schema_diff',
|
||||
dict(method='get', endpoint='schema_diff.schemas',
|
||||
url_kwargs=dict(sid=1, did=1))),
|
||||
('schema_diff.ddl_compare requires tools_schema_diff',
|
||||
dict(method='get', endpoint='schema_diff.ddl_compare',
|
||||
url_kwargs=dict(trans_id=_TRANS, source_sid=1, source_did=1,
|
||||
source_scid=1, target_sid=1, target_did=1,
|
||||
target_scid=1, source_oid=1, target_oid=1,
|
||||
node_type='table', comp_status='Different'))),
|
||||
|
||||
# --- ERD: init and DDL generation (audit, new) ---
|
||||
('erd.initialize requires tools_erd_tool',
|
||||
dict(method='post', endpoint='erd.initialize',
|
||||
url_kwargs=dict(trans_id=_TRANS, sgid=1, sid=1, did=1))),
|
||||
('erd.prequisite requires tools_erd_tool',
|
||||
dict(method='get', endpoint='erd.prequisite',
|
||||
url_kwargs=dict(trans_id=_TRANS, sgid=1, sid=1, did=1))),
|
||||
('erd.sql requires tools_erd_tool',
|
||||
dict(method='post', endpoint='erd.sql',
|
||||
url_kwargs=dict(trans_id=_TRANS, sgid=1, sid=1, did=1))),
|
||||
|
||||
# --- PSQL: panel (audit, new) ---
|
||||
('psql.panel requires tools_psql_tool',
|
||||
dict(method='post', endpoint='psql.panel',
|
||||
url_kwargs=dict(trans_id=_TRANS))),
|
||||
|
||||
# --- Debugger: directly-addressable stored-argument routes ---
|
||||
('debugger.get_arguments requires tools_debugger',
|
||||
dict(method='get', endpoint='debugger.get_arguments',
|
||||
url_kwargs=dict(sid=1, did=1, scid=1, func_id=1))),
|
||||
('debugger.set_arguments requires tools_debugger',
|
||||
dict(method='post', endpoint='debugger.set_arguments',
|
||||
url_kwargs=dict(sid=1, did=1, scid=1, func_id=1))),
|
||||
('debugger.clear_arguments requires tools_debugger',
|
||||
dict(method='post', endpoint='debugger.clear_arguments',
|
||||
url_kwargs=dict(sid=1, did=1, scid=1, func_id=1))),
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
if not config.SERVER_MODE:
|
||||
self.skipTest(
|
||||
'Tool permission decorators are only exercisable in SERVER '
|
||||
'mode; DESKTOP mode auto-authenticates as the '
|
||||
'all-permissions DESKTOP_USER on every request.'
|
||||
)
|
||||
|
||||
@create_user_wise_test_client(test_user_details)
|
||||
def runTest(self):
|
||||
# Build the URL through the URL map so we exercise the registered
|
||||
# route exactly and don't hard-code blueprint prefixes.
|
||||
with self.app.test_request_context():
|
||||
url = flask.url_for(self.endpoint, **self.url_kwargs)
|
||||
|
||||
http = getattr(self.tester, self.method)
|
||||
response = http(url, follow_redirects=False)
|
||||
|
||||
self.assertEqual(
|
||||
response.status_code, 403,
|
||||
'Route {0} ({1} {2}) did not enforce its tool permission for a '
|
||||
'user lacking it: expected 403, got {3}. Body: {4!r}'.format(
|
||||
self.endpoint, self.method.upper(), url,
|
||||
response.status_code, response.data[:200]
|
||||
)
|
||||
)
|
||||
@@ -0,0 +1,110 @@
|
||||
##########################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2026, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
##########################################################################
|
||||
|
||||
"""RBAC regression: Socket.IO tool handlers must enforce the permission.
|
||||
|
||||
The HTTP routes of the schema diff, ERD and PSQL tools are guarded by
|
||||
@permissions_required, but those tools also expose Socket.IO event
|
||||
handlers (schema_diff 'compare_database'/'compare_schema', erd 'tables',
|
||||
the psql '/pty' namespace). Those handlers previously carried only
|
||||
@socket_login_required (or nothing at all), so an authenticated user who
|
||||
had been denied the tool could still drive them. They now use
|
||||
@socket_permissions_required, which refuses the connection when the user
|
||||
lacks the permission.
|
||||
|
||||
This test connects to each namespace as a user with no roles (hence no
|
||||
tool permissions) and asserts that emitting a guarded event causes the
|
||||
handler to refuse and disconnect the socket, rather than executing.
|
||||
|
||||
Skipped in DESKTOP mode, where every request is auto-authenticated as the
|
||||
all-permissions DESKTOP_USER.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
import config
|
||||
|
||||
from flask_socketio import ConnectionRefusedError
|
||||
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from pgadmin import socketio
|
||||
from regression.test_setup import config_data
|
||||
from regression.python_test_utils import test_utils as utils
|
||||
|
||||
test_user_details = None
|
||||
if config.SERVER_MODE:
|
||||
test_user_details = config_data['pgAdmin4_test_non_admin_credentials']
|
||||
|
||||
|
||||
class ToolSocketPermissionRequiredTestCase(BaseTestGenerator):
|
||||
"""A guarded Socket.IO handler must refuse a user lacking the tool
|
||||
permission instead of running."""
|
||||
|
||||
scenarios = [
|
||||
('schema_diff compare_database requires tools_schema_diff',
|
||||
dict(namespace='/schema_diff', event='compare_database',
|
||||
params=dict(trans_id=1, source_sid=1, source_did=1,
|
||||
target_sid=1, target_did=1,
|
||||
ignore_owner=0, ignore_whitespaces=0,
|
||||
ignore_tablespace=0, ignore_grants=0))),
|
||||
('schema_diff compare_schema requires tools_schema_diff',
|
||||
dict(namespace='/schema_diff', event='compare_schema',
|
||||
params=dict(trans_id=1, source_sid=1, source_did=1,
|
||||
source_scid=1, target_sid=1, target_did=1,
|
||||
target_scid=1, ignore_owner=0,
|
||||
ignore_whitespaces=0, ignore_tablespace=0,
|
||||
ignore_grants=0))),
|
||||
('psql start_process requires tools_psql_tool',
|
||||
dict(namespace='/pty', event='start_process',
|
||||
params=dict(sid=1, did=1))),
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
if not config.SERVER_MODE:
|
||||
self.skipTest(
|
||||
'Socket permission decorators are only exercisable in '
|
||||
'SERVER mode; DESKTOP mode auto-authenticates as the '
|
||||
'all-permissions DESKTOP_USER.'
|
||||
)
|
||||
if self.namespace == '/pty' and sys.platform == 'win32':
|
||||
self.skipTest('PSQL is disabled on Windows.')
|
||||
config.ENABLE_PSQL = True
|
||||
|
||||
def runTest(self):
|
||||
# Log in a brand-new user that has no roles, therefore no tool
|
||||
# permissions, and bind a Socket.IO test client to that session.
|
||||
non_admin_client = utils.get_test_user(self, test_user_details)
|
||||
self.assertIsNotNone(
|
||||
non_admin_client, 'Could not create the non-admin test user.')
|
||||
|
||||
sclient = socketio.test_client(
|
||||
self.app, namespace=self.namespace,
|
||||
flask_test_client=non_admin_client)
|
||||
|
||||
# The namespace 'connect' handler is not permission-gated, so the
|
||||
# connection itself is allowed; the guard is on the event handler.
|
||||
self.assertTrue(
|
||||
sclient.is_connected(self.namespace),
|
||||
'Expected the namespace connection to succeed for an '
|
||||
'authenticated user.')
|
||||
|
||||
# Emitting the guarded event must trip socket_permissions_required,
|
||||
# which raises ConnectionRefusedError instead of running the
|
||||
# handler. The flask-socketio test client surfaces that as an
|
||||
# exception out of emit(); if the handler had run unguarded, no
|
||||
# exception would be raised here.
|
||||
try:
|
||||
sclient.emit(self.event, self.params, namespace=self.namespace)
|
||||
except ConnectionRefusedError:
|
||||
pass
|
||||
else:
|
||||
self.fail(
|
||||
'Socket handler {0} on {1} did not refuse a user lacking '
|
||||
'the tool permission; the handler ran instead of raising '
|
||||
'ConnectionRefusedError.'.format(self.event, self.namespace))
|
||||
Reference in New Issue
Block a user