mirror of
https://github.com/pgadmin-org/pgadmin4.git
synced 2026-08-17 16:34:44 -05:00
Fix OS command injection in the MASTER_PASSWORD_HOOK feature
The MASTER_PASSWORD_HOOK setting lets administrators specify an external command that returns a per-user encryption key, with %u in the configured string replaced by the current user's name. The previous implementation substituted the username into the command string and executed the result with subprocess.Popen(..., shell=True). Because the username can originate from an external authentication source (OAuth/OIDC claims, Kerberos, webserver auth), a username containing shell metacharacters allowed an authenticated user to execute arbitrary commands as the pgAdmin service account in deployments where the hook uses %u. Tokenise the trusted hook string into an argument vector first, substitute the untrusted username into the individual arguments, and execute with shell=False. The username is therefore always confined to a single argv element and any shell metacharacters it contains are inert. Note for administrators: hooks that previously relied on shell features (pipes, redirection, environment-variable expansion, globbing) in the MASTER_PASSWORD_HOOK string itself will no longer have those interpreted; such logic should be moved into the hook script. The documented form, '<PATH>/script.sh %u', is unaffected. Adds regression tests covering usernames containing ';', '$()', backticks, pipes, '&&' and newlines, plus an end-to-end marker-file proof that no shell execution occurs. Reported-by: B1gN0Se
This commit is contained in:
@@ -636,6 +636,12 @@ USE_OS_SECRET_STORAGE = True
|
||||
# You can pass the current username as an argument to the external script
|
||||
# by specifying %u in config value.
|
||||
# E.g. - MASTER_PASSWORD_HOOK = '<PATH>/passwdgen_script.sh %u'
|
||||
#
|
||||
# The command is split into arguments and executed directly, without a
|
||||
# shell, so shell features (pipes, redirection, environment-variable
|
||||
# expansion, globbing) in this value are not interpreted; put any such
|
||||
# logic inside the hook script itself. On Windows, a batch file must be
|
||||
# invoked via an executable wrapper rather than directly.
|
||||
##########################################################################
|
||||
MASTER_PASSWORD_HOOK = None
|
||||
|
||||
|
||||
@@ -179,16 +179,39 @@ def process_masterpass_disabled():
|
||||
|
||||
def get_master_password_from_master_hook():
|
||||
"""
|
||||
This method executes specified command & returns output.
|
||||
:param command: Shell command with absolute path
|
||||
:return: Output of command.
|
||||
This method executes the configured MASTER_PASSWORD_HOOK command and
|
||||
returns its output.
|
||||
|
||||
The hook is parsed into an argument vector and executed *without* a shell
|
||||
(shell=False). The username (which may originate from an external
|
||||
authentication source such as OAuth/OIDC, Kerberos or webserver auth, and
|
||||
is therefore untrusted) is substituted for %u only after the trusted hook
|
||||
string has been tokenised, so it is always confined to a single argument.
|
||||
Any shell metacharacters it contains are then inert rather than
|
||||
interpreted, which prevents OS command injection via the username.
|
||||
|
||||
:return: Output of the command, or None on failure.
|
||||
"""
|
||||
import os
|
||||
import shlex
|
||||
import subprocess
|
||||
|
||||
cmd = config.MASTER_PASSWORD_HOOK
|
||||
command = cmd.replace('%u', current_user.username) \
|
||||
if '%u' in cmd else cmd
|
||||
if not cmd:
|
||||
return None
|
||||
try:
|
||||
p = subprocess.Popen(command, stdout=subprocess.PIPE, shell=True)
|
||||
# Tokenise the admin-configured hook first, then substitute the
|
||||
# untrusted username into the individual arguments. Substituting
|
||||
# before tokenising would let metacharacters in the username alter
|
||||
# the command, so the order here is security-critical.
|
||||
args = [
|
||||
token.replace('%u', current_user.username)
|
||||
for token in shlex.split(cmd, posix=(os.name != 'nt'))
|
||||
]
|
||||
if not args:
|
||||
return None
|
||||
|
||||
p = subprocess.Popen(args, stdout=subprocess.PIPE, shell=False)
|
||||
out, err = p.communicate()
|
||||
if p.returncode == 0:
|
||||
output = out.decode() if hasattr(out, 'decode') else out
|
||||
@@ -196,7 +219,7 @@ def get_master_password_from_master_hook():
|
||||
return output
|
||||
else:
|
||||
error = "Command '{0}' failed, exit-code={1} error = {2}".format(
|
||||
command, p.returncode, str(err))
|
||||
cmd, p.returncode, str(err))
|
||||
current_app.logger.error(error)
|
||||
except Exception as e:
|
||||
current_app.logger.exception(
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
##########################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2026, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
##########################################################################
|
||||
|
||||
"""Regression tests for the MASTER_PASSWORD_HOOK OS command injection.
|
||||
|
||||
The hook substitutes the externally-supplied username for %u. The fix
|
||||
tokenises the (trusted) hook string first and only then substitutes the
|
||||
(untrusted) username into the individual arguments, executing the result
|
||||
with shell=False. As a result, shell metacharacters in the username are
|
||||
passed through as a single, literal argv element rather than being
|
||||
interpreted by a shell."""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
import pgadmin.utils.master_password as mp
|
||||
|
||||
|
||||
class _FakeProc:
|
||||
"""Minimal stand-in for a subprocess.Popen result."""
|
||||
returncode = 0
|
||||
|
||||
def communicate(self):
|
||||
return (b'secret-key', b'')
|
||||
|
||||
|
||||
class TestMasterPasswordHookConfinesUsername(BaseTestGenerator):
|
||||
"""The username, whatever shell syntax it contains, must reach the hook
|
||||
as exactly one argv element, and the hook must run with shell=False. This
|
||||
is verified by capturing the arguments passed to subprocess.Popen, so the
|
||||
check is deterministic and platform-independent (nothing is executed)."""
|
||||
|
||||
scenarios = [
|
||||
('semicolon command separator',
|
||||
dict(username='attacker; touch /tmp/x')),
|
||||
('command substitution $()',
|
||||
dict(username='x$(touch /tmp/x)')),
|
||||
('backtick command substitution',
|
||||
dict(username='x`touch /tmp/x`')),
|
||||
('pipe to another command',
|
||||
dict(username='x | touch /tmp/x')),
|
||||
('logical-and chain',
|
||||
dict(username='x && touch /tmp/x')),
|
||||
('embedded newline',
|
||||
dict(username='x\ntouch /tmp/x')),
|
||||
('embedded whitespace only',
|
||||
dict(username='first last')),
|
||||
]
|
||||
|
||||
def runTest(self):
|
||||
captured = {}
|
||||
|
||||
def fake_popen(args, *a, **kw):
|
||||
captured['args'] = args
|
||||
captured['shell'] = kw.get('shell')
|
||||
return _FakeProc()
|
||||
|
||||
fake_user = MagicMock()
|
||||
fake_user.username = self.username
|
||||
|
||||
with patch.object(mp, 'current_user', fake_user), \
|
||||
patch.object(mp.config, 'MASTER_PASSWORD_HOOK',
|
||||
'/opt/get-secret %u'), \
|
||||
patch('subprocess.Popen', side_effect=fake_popen):
|
||||
output = mp.get_master_password_from_master_hook()
|
||||
|
||||
# A shell must never be involved.
|
||||
self.assertIs(captured.get('shell'), False)
|
||||
# The command must be passed as an argument vector, not a string.
|
||||
self.assertIsInstance(captured.get('args'), list)
|
||||
# The hook program token is preserved untouched as argv[0].
|
||||
self.assertEqual(captured['args'][0], '/opt/get-secret')
|
||||
# The entire username lands in a single argv element ...
|
||||
self.assertEqual(captured['args'][1], self.username)
|
||||
# ... and nothing was split out into extra tokens.
|
||||
self.assertEqual(len(captured['args']), 2)
|
||||
# The hook output is still returned to the caller.
|
||||
self.assertEqual(output, 'secret-key')
|
||||
|
||||
|
||||
class TestMasterPasswordHookNoShellExecution(BaseTestGenerator):
|
||||
"""End-to-end PoC regression (POSIX only): a username containing shell
|
||||
syntax must not cause command execution. Mirrors the reported proof of
|
||||
concept using a harmless marker file: if a shell were involved, the ';'
|
||||
would start a second command that creates the marker."""
|
||||
|
||||
def runTest(self):
|
||||
if os.name == 'nt':
|
||||
self.skipTest('POSIX shell-injection PoC; skipped on Windows.')
|
||||
|
||||
echo = shutil.which('echo')
|
||||
if echo is None:
|
||||
self.skipTest('echo binary not available on PATH.')
|
||||
|
||||
tmpdir = tempfile.mkdtemp(prefix='pgadmin-hook-test-')
|
||||
marker = os.path.join(tmpdir, 'master_hook_rce')
|
||||
username = 'attacker; touch {0}; #'.format(marker)
|
||||
|
||||
fake_user = MagicMock()
|
||||
fake_user.username = username
|
||||
|
||||
try:
|
||||
with patch.object(mp, 'current_user', fake_user), \
|
||||
patch.object(mp.config, 'MASTER_PASSWORD_HOOK',
|
||||
'{0} hook-%u'.format(echo)):
|
||||
output = mp.get_master_password_from_master_hook()
|
||||
|
||||
self.assertFalse(
|
||||
os.path.exists(marker),
|
||||
'Injected command executed: the marker file was created, '
|
||||
'which means the username was interpreted by a shell.')
|
||||
# The username was handled as inert data and echoed verbatim.
|
||||
self.assertEqual(output, 'hook-' + username)
|
||||
finally:
|
||||
if os.path.exists(marker):
|
||||
os.remove(marker)
|
||||
os.rmdir(tmpdir)
|
||||
Reference in New Issue
Block a user