mirror of
https://github.com/pgadmin-org/pgadmin4.git
synced 2026-10-27 23:20:05 -05:00
Make the 9.19 security tests pass on Windows.
The tests added for the 9.19 storage and connection hardening were written on Linux and failed on Windows, where the behaviour they check is correct but expressed differently. The production code is unchanged. - Skip the Backup, Restore and Import/Export descriptor-based symlink-race tests on Windows: opening a file with O_NOFOLLOW and handing it to the utility as /dev/fd/N is POSIX-only, and the code already passes the real path there. - Compare credential-file paths after fs_short_path()+forward slashes, the short (8.3) form libpq is given on Windows; this is the identity on POSIX. Covers the --dbname conninfo call sites, get_complete_file_path, client-file isolation and the SSH tunnel identity file. - Normalise the File Manager closest-parent separator to '/'. - Drop the other inherited credential variable before checking which remove_inherited_libpq_credentials() removes, as the runner's own environment may hold PGPASSWORD. - Use forward slashes in the raw hostile --database value, whose backslashes libpq would otherwise treat as escapes, and skip the 'missing configured file passed through' case on Windows, where such a file is disabled (NO_CLIENT_FILE) instead.
This commit is contained in:
1 parent
36122bc29b
commit
eb66e6a347
11 files changed
+97
-23
No files matched your search
@@ -1571,7 +1571,13 @@ class TestGetClosestParentStaysInStorage(_StorageOpsMixin, BaseTestGenerator):
|
||||
scenarios = [('default', dict())]
|
||||
|
||||
def runTest(self):
|
||||
closest = Filemanager.get_closest_parent
|
||||
_closest = Filemanager.get_closest_parent
|
||||
|
||||
# The File Manager uses forward-slash paths; normalise the separator
|
||||
# so the expectations hold on Windows, where os.path.join uses "\".
|
||||
def closest(*args):
|
||||
return _closest(*args).replace('\\', '/')
|
||||
|
||||
os.makedirs(os.path.join(self.storage_dir, "a"))
|
||||
self.assertEqual(closest(self.storage_dir, "/a/"), "/a/")
|
||||
self.assertEqual(closest(self.storage_dir, "/a/gone/deeper/"), "/a/")
|
||||
|
||||
@@ -43,6 +43,13 @@ class BackupSymlinkRaceTest(BaseTestGenerator):
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
# Opening the file without following a leaf symbolic link and handing
|
||||
# it to the utility as /dev/fd/N needs O_NOFOLLOW and descriptor-based
|
||||
# opens, which exist only on POSIX; on Windows the real path is passed
|
||||
# and this protection does not apply.
|
||||
if os.name == 'nt':
|
||||
self.skipTest(
|
||||
'Descriptor-based symlink-race protection is POSIX-only.')
|
||||
if 'default_binary_paths' not in self.server or \
|
||||
self.server['default_binary_paths'] is None or \
|
||||
self.server['type'] not in self.server['default_binary_paths'] or \
|
||||
|
||||
@@ -45,6 +45,13 @@ class ImportExportSymlinkRaceTest(BaseTestGenerator):
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
# Opening the file without following a leaf symbolic link and handing
|
||||
# it to the utility as /dev/fd/N needs O_NOFOLLOW and descriptor-based
|
||||
# opens, which exist only on POSIX; on Windows the real path is passed
|
||||
# and this protection does not apply.
|
||||
if os.name == 'nt':
|
||||
self.skipTest(
|
||||
'Descriptor-based symlink-race protection is POSIX-only.')
|
||||
if 'default_binary_paths' not in self.server or \
|
||||
self.server['default_binary_paths'] is None or \
|
||||
self.server['type'] not in self.server['default_binary_paths'] or \
|
||||
|
||||
@@ -46,6 +46,13 @@ class RestoreSymlinkRaceTest(BaseTestGenerator):
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
# Opening the file without following a leaf symbolic link and handing
|
||||
# it to the utility as /dev/fd/N needs O_NOFOLLOW and descriptor-based
|
||||
# opens, which exist only on POSIX; on Windows the real path is passed
|
||||
# and this protection does not apply.
|
||||
if os.name == 'nt':
|
||||
self.skipTest(
|
||||
'Descriptor-based symlink-race protection is POSIX-only.')
|
||||
if 'default_binary_paths' not in self.server or \
|
||||
self.server['default_binary_paths'] is None or \
|
||||
self.server['type'] not in self.server['default_binary_paths'] or \
|
||||
|
||||
@@ -25,7 +25,7 @@ from unittest.mock import MagicMock, patch
|
||||
import psycopg
|
||||
from psycopg.conninfo import conninfo_to_dict, make_conninfo
|
||||
|
||||
from pgadmin.utils import NO_CLIENT_FILE
|
||||
from pgadmin.utils import NO_CLIENT_FILE, fs_short_path
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
|
||||
import pgadmin.utils.driver.psycopg3 as psycopg3_driver
|
||||
@@ -140,8 +140,11 @@ class ConnectionStringClientFilesTestCase(_StorageMixin, BaseTestGenerator):
|
||||
for key, value in self.expected.items():
|
||||
if value not in (NO_CLIENT_FILE, 'system'):
|
||||
# Expected files are given relative to the storage
|
||||
# directory.
|
||||
value = os.path.join(self.storage, value)
|
||||
# directory, and resolved as get_complete_file_path() returns
|
||||
# them: the short (8.3) form with forward slashes on Windows,
|
||||
# unchanged on POSIX.
|
||||
value = fs_short_path(
|
||||
os.path.join(self.storage, value)).replace('\\', '/')
|
||||
self.assertEqual(dsn.get(key), value, key)
|
||||
for key in self.absent:
|
||||
self.assertNotIn(key, dsn)
|
||||
|
||||
@@ -228,6 +228,12 @@ class InheritedCredentialsTestCase(_ServiceMixin, BaseTestGenerator):
|
||||
conn.close()
|
||||
|
||||
with patch.dict(os.environ, env):
|
||||
# The runner's own environment may already hold the other
|
||||
# inherited credential variable (CI sets PGPASSWORD); drop it so
|
||||
# the scenario sees only the one it set.
|
||||
for other in ('PGPASSWORD', 'PGSERVICE'):
|
||||
if other != self.var:
|
||||
os.environ.pop(other, None)
|
||||
removed = remove_inherited_libpq_credentials()
|
||||
remaining = dict(os.environ)
|
||||
self.assertEqual(removed, [self.var])
|
||||
|
||||
@@ -22,6 +22,7 @@ from unittest.mock import MagicMock, patch
|
||||
|
||||
import paramiko
|
||||
|
||||
from pgadmin.utils import fs_short_path
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from pgadmin.utils.driver.psycopg3 import server_manager as sm
|
||||
|
||||
@@ -204,8 +205,12 @@ class TestSSHTunnelIdentityFileConfinement(
|
||||
self.assertNotIn(self.alice, result[1])
|
||||
else:
|
||||
self.assertEqual(result, (True, None))
|
||||
self.assertEqual(forwarder.call_args.kwargs['ssh_pkey'],
|
||||
os.path.join(self.alice, 'mykey'))
|
||||
# get_complete_file_path() returns the short (8.3) form with
|
||||
# forward slashes on Windows, unchanged on POSIX.
|
||||
self.assertEqual(
|
||||
forwarder.call_args.kwargs['ssh_pkey'],
|
||||
fs_short_path(
|
||||
os.path.join(self.alice, 'mykey')).replace('\\', '/'))
|
||||
|
||||
|
||||
class TestSSHTunnelRealSSHTunnelForwarder(
|
||||
|
||||
@@ -26,12 +26,19 @@ import flask
|
||||
|
||||
import config
|
||||
from pgadmin.utils import (
|
||||
connection_params_to_env, get_complete_file_path)
|
||||
connection_params_to_env, fs_short_path, get_complete_file_path)
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
|
||||
import pgadmin.utils.driver.psycopg3.server_manager as server_manager
|
||||
|
||||
|
||||
def _canon(path):
|
||||
"""Compare paths as get_complete_file_path() returns them: on Windows the
|
||||
short (8.3) form libpq is given, so an expected long path matches;
|
||||
fs_short_path() is the identity on POSIX."""
|
||||
return os.path.normpath(fs_short_path(path))
|
||||
|
||||
|
||||
class _StorageMixin:
|
||||
"""Builds a storage root holding two users and a shared directory."""
|
||||
|
||||
@@ -138,8 +145,7 @@ class CompleteFilePathContainmentTestCase(_StorageMixin, BaseTestGenerator):
|
||||
else:
|
||||
# Expected paths are given relative to the temporary root.
|
||||
expected = os.path.join(self.outer, *self.expected.split('/'))
|
||||
self.assertEqual(os.path.normpath(result),
|
||||
os.path.normpath(expected))
|
||||
self.assertEqual(_canon(result), _canon(expected))
|
||||
|
||||
|
||||
class CompleteFilePathDesktopTestCase(_StorageMixin, BaseTestGenerator):
|
||||
@@ -263,8 +269,8 @@ class CompleteFilePathRealStorageDirTestCase(_StorageMixin,
|
||||
|
||||
def runTest(self):
|
||||
self.assertEqual(
|
||||
os.path.normpath(self._resolve_real('/.pgpass')),
|
||||
os.path.join(self.alice, '.pgpass'))
|
||||
_canon(self._resolve_real('/.pgpass')),
|
||||
_canon(os.path.join(self.alice, '.pgpass')))
|
||||
|
||||
with self.assertRaises(PermissionError) as cm:
|
||||
self._resolve_real('../bob/.pgpass')
|
||||
@@ -272,8 +278,8 @@ class CompleteFilePathRealStorageDirTestCase(_StorageMixin,
|
||||
str(cm.exception))
|
||||
|
||||
self.assertEqual(
|
||||
os.path.normpath(self._resolve_real('Team:/.pgpass')),
|
||||
os.path.join(self.shared, '.pgpass'))
|
||||
_canon(self._resolve_real('Team:/.pgpass')),
|
||||
_canon(os.path.join(self.shared, '.pgpass')))
|
||||
|
||||
# A name that is not configured is just part of a relative path.
|
||||
self.assertIsNone(self._resolve_real('Nope:/.pgpass'))
|
||||
@@ -405,8 +411,8 @@ class CompleteFilePathWriteTargetTestCase(_StorageMixin, BaseTestGenerator):
|
||||
else:
|
||||
result = get_complete_file_path('Team:/keylog', validate=False,
|
||||
for_write=True)
|
||||
self.assertEqual(os.path.normpath(result),
|
||||
os.path.join(self.shared, 'keylog'))
|
||||
self.assertEqual(_canon(result),
|
||||
_canon(os.path.join(self.shared, 'keylog')))
|
||||
|
||||
|
||||
def _user(is_admin):
|
||||
|
||||
@@ -23,7 +23,7 @@ from unittest.mock import patch
|
||||
from psycopg.conninfo import conninfo_to_dict
|
||||
|
||||
from pgadmin.utils import (
|
||||
utility_conninfo, connection_params_to_env, NO_CLIENT_FILE)
|
||||
utility_conninfo, connection_params_to_env, fs_short_path, NO_CLIENT_FILE)
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
|
||||
CREDENTIAL_FILES = ('passfile', 'sslcert', 'sslkey')
|
||||
@@ -55,7 +55,7 @@ class UtilityConninfoTestCase(BaseTestGenerator):
|
||||
'sslkey': NO_CLIENT_FILE})),
|
||||
('Server mode, a configured file that does not exist is passed on '
|
||||
'as the env translation gives it',
|
||||
dict(server_mode=True, database='postgres',
|
||||
dict(server_mode=True, database='postgres', posix_only=True,
|
||||
params={'sslcert': 'missing.crt'},
|
||||
expected={'dbname': 'postgres',
|
||||
'passfile': NO_CLIENT_FILE,
|
||||
@@ -129,8 +129,23 @@ class UtilityConninfoTestCase(BaseTestGenerator):
|
||||
for p in reversed(patches):
|
||||
p.stop()
|
||||
|
||||
expected = {k: v.replace('<storage>', self.storage)
|
||||
for k, v in self.expected.items()}
|
||||
# On Windows a configured file that cannot be found is disabled
|
||||
# (NO_CLIENT_FILE), not passed through as its resolved path; both are
|
||||
# unopenable, so the connection still cannot use another file.
|
||||
if getattr(self, 'posix_only', False) and os.name == 'nt':
|
||||
self.skipTest(
|
||||
'A missing configured file is disabled on Windows.')
|
||||
|
||||
def _exp(v):
|
||||
# Storage files resolve as get_complete_file_path() returns them:
|
||||
# the short (8.3) form with forward slashes on Windows, unchanged
|
||||
# on POSIX.
|
||||
if isinstance(v, str) and '<storage>' in v:
|
||||
return fs_short_path(
|
||||
v.replace('<storage>', self.storage)).replace('\\', '/')
|
||||
return v
|
||||
|
||||
expected = {k: _exp(v) for k, v in self.expected.items()}
|
||||
self.assertEqual(result, expected)
|
||||
|
||||
|
||||
|
||||
@@ -28,13 +28,21 @@ from psycopg.conninfo import conninfo_to_dict
|
||||
|
||||
from pgadmin.misc.bgprocess.processes import BatchProcess
|
||||
from pgadmin.utils import (
|
||||
NO_CLIENT_FILE, database_conninfo, utility_connection_params)
|
||||
NO_CLIENT_FILE, database_conninfo, fs_short_path,
|
||||
utility_connection_params)
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
|
||||
DATABASE = 'host=evil dbname=x'
|
||||
CONNECTION_PARAMS = {'passfile': '.pgpass'}
|
||||
|
||||
|
||||
def _resolved(path):
|
||||
"""A storage file path as get_complete_file_path() returns it: on
|
||||
Windows the short (8.3) form with forward slashes, which libpq is given;
|
||||
unchanged on POSIX, where fs_short_path() is the identity."""
|
||||
return fs_short_path(path).replace('\\', '/')
|
||||
|
||||
|
||||
SCENARIOS = [
|
||||
('Server mode', dict(server_mode=True)),
|
||||
('Desktop mode', dict(server_mode=False)),
|
||||
@@ -105,7 +113,7 @@ class _CallSiteBase:
|
||||
if not self.server_mode:
|
||||
return {'dbname': DATABASE}
|
||||
return {'dbname': DATABASE,
|
||||
'passfile': os.path.join(self.storage, '.pgpass'),
|
||||
'passfile': _resolved(os.path.join(self.storage, '.pgpass')),
|
||||
'sslcert': NO_CLIENT_FILE, 'sslkey': NO_CLIENT_FILE}
|
||||
|
||||
def _assert_dbname(self, args, expected=None):
|
||||
@@ -199,7 +207,7 @@ class ConnectionParamsSourceTestCase(_CallSiteBase, BaseTestGenerator):
|
||||
lambda: utility_conninfo(
|
||||
'postgres', utility_connection_params(manager, server)))
|
||||
self.assertEqual(conninfo_to_dict(conninfo)['passfile'],
|
||||
os.path.join(self.storage, '.pgpass'))
|
||||
_resolved(os.path.join(self.storage, '.pgpass')))
|
||||
|
||||
|
||||
class RestoreArgsTestCase(_CallSiteBase, BaseTestGenerator):
|
||||
|
||||
@@ -304,7 +304,11 @@ class PgDumpallMaintenanceDatabaseTestCase(BaseTestGenerator):
|
||||
f.write('[pgadmin_probe]\npassfile={0}\n'.format(
|
||||
service_passfile))
|
||||
|
||||
hostile = 'dbname=postgres passfile={0}'.format(service_passfile)
|
||||
# The raw (control) value is parsed by libpq as a connection string,
|
||||
# where a backslash escapes the next character; a Windows path must
|
||||
# use forward slashes, which libpq accepts, to survive it unchanged.
|
||||
hostile = 'dbname=postgres passfile={0}'.format(
|
||||
service_passfile.replace(os.sep, '/'))
|
||||
database = database_conninfo(hostile) if self.fixed else hostile
|
||||
dbname, server_env = self._utility_dbname()
|
||||
env = {k: v for k, v in os.environ.items() if not k.startswith('PG')}
|
||||
|
||||
Reference in new issue
Block a user