sashatrask
242 lines
10 KiB
Python
242 lines
10 KiB
Python
import ast
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest import mock
|
|
import venv
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP_DIR = ROOT / 'app'
|
|
sys.path.insert(0, str(APP_DIR))
|
|
|
|
import owned_process
|
|
|
|
|
|
CANONICAL_DSN_KEYS = (
|
|
'TRUF_MANAGED_POSTGRES_DSN',
|
|
'SCANNER_DB_URL',
|
|
'DATABASE_URL',
|
|
'SCANNER_DASHBOARD_DB_URL',
|
|
'KEYCHECK_DB_URL',
|
|
)
|
|
LIBPQ_PRIVATE_KEYS = (
|
|
'PGPASSWORD',
|
|
'PGUSER',
|
|
'PGDATABASE',
|
|
'PGHOST',
|
|
'PGHOSTADDR',
|
|
'PGPORT',
|
|
'PGSERVICE',
|
|
'PGSERVICEFILE',
|
|
'PGPASSFILE',
|
|
'PGOPTIONS',
|
|
'PGSSLMODE',
|
|
'PGSSLKEY',
|
|
'PGSSLCERT',
|
|
'PGSSLROOTCERT',
|
|
)
|
|
|
|
|
|
def mixed_case(value):
|
|
return ''.join(character.lower() if index % 2 else character for index, character in enumerate(value))
|
|
|
|
|
|
def sensitive_environment():
|
|
keys = CANONICAL_DSN_KEYS + LIBPQ_PRIVATE_KEYS + (
|
|
'TRUF_POSTGRES_PASSWORD',
|
|
'TRUF_POSTGRES_FUTURE_CREDENTIAL',
|
|
)
|
|
return {mixed_case(key): f'secret-{index}' for index, key in enumerate(keys)}
|
|
|
|
|
|
def assert_credentials_absent(test_case, environment):
|
|
normalized = {str(key).upper() for key in environment}
|
|
test_case.assertTrue(normalized.isdisjoint(CANONICAL_DSN_KEYS))
|
|
test_case.assertTrue(normalized.isdisjoint(LIBPQ_PRIVATE_KEYS))
|
|
test_case.assertFalse(any(key.startswith('TRUF_POSTGRES_') for key in normalized))
|
|
|
|
|
|
class OwnedProcessHostBoundaryTests(unittest.TestCase):
|
|
def test_host_command_puts_isolation_flags_before_script_and_control_handles(self):
|
|
host = mock.Mock(pid=701)
|
|
prefix = 'windows-filetime:' if os.name == 'nt' else 'proc-start-ticks:'
|
|
identities = [
|
|
{'pid': pid, 'creation_time': prefix + '123', 'creation_time_unix': 1700000000.0,
|
|
'executable': os.path.abspath(sys.executable)}
|
|
for pid in (701, 702)
|
|
]
|
|
with mock.patch.object(owned_process.subprocess, 'Popen', return_value=host) as popen, \
|
|
mock.patch.object(owned_process, '_write_packet'), \
|
|
mock.patch.object(
|
|
owned_process.OwnedProcess,
|
|
'_wait_for_startup',
|
|
return_value={
|
|
'ok': True,
|
|
'pid': 702,
|
|
'job_membership_verified': True,
|
|
'payload_identity': identities[1],
|
|
'host_identity': identities[0],
|
|
'job_members': identities,
|
|
'job_accounting': {'child_subreaper_verified': True},
|
|
},
|
|
):
|
|
process = owned_process.OwnedProcess(['payload'])
|
|
|
|
command = popen.call_args.args[0]
|
|
host_python = sys._base_executable if os.name == 'nt' else sys.executable
|
|
self.assertEqual(command[:4], [host_python, '-I', '-S', '-B'])
|
|
self.assertEqual(command[4], os.path.abspath(owned_process.__file__))
|
|
self.assertEqual(command[5], owned_process._HOST_FLAG)
|
|
self.assertEqual(len(command[6:]), 2)
|
|
if os.name != 'nt':
|
|
self.assertIs(popen.call_args.kwargs['start_new_session'], True)
|
|
process.terminate()
|
|
|
|
def test_nonisolated_host_rejects_before_reading_control_packet(self):
|
|
flags = SimpleNamespace(isolated=0, no_site=0, dont_write_bytecode=0)
|
|
with mock.patch.object(owned_process.sys, 'flags', flags), \
|
|
mock.patch.object(owned_process, '_open_inherited_fd', side_effect=(40, 41)), \
|
|
mock.patch.object(owned_process, '_read_packet') as read_packet, \
|
|
mock.patch.object(owned_process, '_write_packet'), \
|
|
mock.patch.object(owned_process.signal, 'signal'), \
|
|
mock.patch.object(owned_process.os, 'close'):
|
|
self.assertEqual(owned_process._host_main('control', 'status'), 127)
|
|
read_packet.assert_not_called()
|
|
|
|
def test_host_module_imports_only_the_standard_library(self):
|
|
tree = ast.parse(Path(owned_process.__file__).read_text(encoding='utf-8'))
|
|
imports = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
imports.update(alias.name.partition('.')[0] for alias in node.names)
|
|
elif isinstance(node, ast.ImportFrom) and node.module:
|
|
imports.add(node.module.partition('.')[0])
|
|
self.assertEqual(imports - set(sys.stdlib_module_names), set())
|
|
|
|
def test_host_ignores_sitecustomize_and_pth_from_its_temporary_venv(self):
|
|
payload_python = sys.executable
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
temp_path = Path(temp_dir)
|
|
venv_dir = temp_path / 'venv'
|
|
venv.EnvBuilder(with_pip=False).create(venv_dir)
|
|
if os.name == 'nt':
|
|
venv_python = venv_dir / 'Scripts' / 'python.exe'
|
|
site_packages = venv_dir / 'Lib' / 'site-packages'
|
|
else:
|
|
venv_python = venv_dir / 'bin' / 'python'
|
|
version = f'python{sys.version_info.major}.{sys.version_info.minor}'
|
|
site_packages = venv_dir / 'lib' / version / 'site-packages'
|
|
|
|
site_marker = temp_path / 'site.marker'
|
|
pth_marker = temp_path / 'pth.marker'
|
|
(site_packages / 'sitecustomize.py').write_text(
|
|
'import os\nopen(os.environ["OWNED_PROCESS_SITE_MARKER"], "w").write("ran")\n',
|
|
encoding='ascii',
|
|
)
|
|
(site_packages / 'malicious.pth').write_text(
|
|
'import os; open(os.environ["OWNED_PROCESS_PTH_MARKER"], "w").write("ran")\n',
|
|
encoding='ascii',
|
|
)
|
|
# Prefer the fixture over a distribution's stdlib sitecustomize.
|
|
marker_environment = {
|
|
'PYTHONPATH': str(site_packages),
|
|
'OWNED_PROCESS_SITE_MARKER': str(site_marker),
|
|
'OWNED_PROCESS_PTH_MARKER': str(pth_marker),
|
|
}
|
|
|
|
control_environment = os.environ.copy()
|
|
control_environment.update(marker_environment)
|
|
subprocess.run(
|
|
[str(venv_python), '-c', 'pass'],
|
|
env=control_environment,
|
|
check=True,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
self.assertTrue(site_marker.exists())
|
|
self.assertTrue(pth_marker.exists())
|
|
site_marker.unlink()
|
|
pth_marker.unlink()
|
|
|
|
with mock.patch.dict(os.environ, marker_environment, clear=False), \
|
|
mock.patch.object(owned_process.sys, 'executable', str(venv_python)):
|
|
process = owned_process.OwnedProcess(
|
|
[payload_python, '-I', '-S', '-B', '-c', 'import sys; sys.stdin.buffer.read()'],
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
self.assertEqual(process.host_identity['pid'], process.host_pid)
|
|
if os.name == 'nt':
|
|
self.assertEqual(process._host_process.args[0], sys._base_executable)
|
|
process.communicate(timeout=10)
|
|
self.assertEqual(process.returncode, 0)
|
|
|
|
self.assertFalse(site_marker.exists())
|
|
self.assertFalse(pth_marker.exists())
|
|
|
|
|
|
class CredentialBoundaryTests(unittest.TestCase):
|
|
def test_host_environment_strips_mixed_case_database_credentials_only(self):
|
|
environment = sensitive_environment()
|
|
environment.update({
|
|
mixed_case('TRUF_SUPERVISOR_TOKEN'): 'supervisor-secret',
|
|
'UNRELATED_SETTING': 'preserved',
|
|
})
|
|
with mock.patch.object(owned_process.os, 'environ', environment):
|
|
host_environment = owned_process._host_environment()
|
|
|
|
assert_credentials_absent(self, host_environment)
|
|
self.assertNotIn(mixed_case('TRUF_SUPERVISOR_TOKEN'), host_environment)
|
|
self.assertEqual(host_environment['UNRELATED_SETTING'], 'preserved')
|
|
|
|
def test_scanner_strips_serialized_external_payload_without_mutating_trusted_env(self):
|
|
import scanner
|
|
|
|
environment = sensitive_environment()
|
|
environment['UNRELATED_SETTING'] = 'preserved'
|
|
|
|
class CompleteProcess:
|
|
pid = 703
|
|
returncode = 0
|
|
job_membership_verified = True
|
|
payload_identity = {'pid': 703}
|
|
|
|
@staticmethod
|
|
def poll():
|
|
return 0
|
|
|
|
trusted_dsn = 'postgresql://trusted/managed'
|
|
with tempfile.TemporaryDirectory() as temp_dir, \
|
|
mock.patch.dict(
|
|
os.environ,
|
|
{'TRUF_MANAGED_POSTGRES_DSN': trusted_dsn},
|
|
clear=False,
|
|
), \
|
|
mock.patch.object(scanner, 'scoped_scan_slot_lease', return_value=(True, None)), \
|
|
mock.patch.object(scanner, 'create_command_work_dir', return_value=temp_dir), \
|
|
mock.patch.object(scanner, '_shared_staging_owners', return_value=[]), \
|
|
mock.patch.object(scanner, 'cleanup_command_work_dir'), \
|
|
mock.patch.object(scanner, 'require_trufflehog_launch_authority'), \
|
|
mock.patch.object(scanner, 'write_temp_owner'), \
|
|
mock.patch.object(scanner, 'OwnedProcess', return_value=CompleteProcess()) as process_factory, \
|
|
mock.patch.object(scanner.scan_config, 'trufflehog_job_memory_limit_bytes', 1024):
|
|
self.assertEqual(scanner.run_command(['fixture'], 5, env=environment), ('', '', 0))
|
|
self.assertEqual(os.environ['TRUF_MANAGED_POSTGRES_DSN'], trusted_dsn)
|
|
|
|
process_factory.assert_called_once()
|
|
payload_environment = process_factory.call_args.kwargs['env']
|
|
serialized_environment = owned_process._serializable_environment(payload_environment)
|
|
assert_credentials_absent(self, serialized_environment)
|
|
self.assertEqual(serialized_environment['UNRELATED_SETTING'], 'preserved')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|