sashatrask
561 lines
32 KiB
Python
561 lines
32 KiB
Python
import copy
|
|
import hashlib
|
|
import ntpath
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import posixpath
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
APP_DIR = Path(__file__).resolve().parents[1] / 'app'
|
|
sys.path.insert(0, str(APP_DIR))
|
|
|
|
import lifecycle_authority as authority
|
|
import runtime_security as security
|
|
|
|
|
|
def platform_os(name, path=None):
|
|
# Do not mutate the shared os.name used by pathlib and the test runner.
|
|
result = SimpleNamespace(**vars(os))
|
|
result.name = name
|
|
result.path = path or os.path
|
|
return result
|
|
|
|
|
|
class NativeExecutablePolicyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.executable = '/usr/bin/git'
|
|
self.entries = {
|
|
path: SimpleNamespace(st_uid=0, st_gid=0, st_mode=kind | 0o755)
|
|
for path, kind in (
|
|
('/', stat.S_IFDIR), ('/usr', stat.S_IFDIR),
|
|
('/usr/bin', stat.S_IFDIR), (self.executable, stat.S_IFREG),
|
|
)
|
|
}
|
|
self.writable = set()
|
|
self.noexec = set()
|
|
self.os = platform_os('posix', posixpath)
|
|
self.os.sep = '/'
|
|
self.os.geteuid = lambda: 10001
|
|
self.os.lstat = mock.Mock(side_effect=self.lstat)
|
|
self.os.stat = mock.Mock(side_effect=self.lstat)
|
|
self.os.access = mock.Mock(side_effect=self.access)
|
|
for name in ('chmod', 'chown', 'makedirs'):
|
|
setattr(self.os, name, mock.Mock(side_effect=AssertionError('unexpected filesystem mutation')))
|
|
self.enterContext(mock.patch.object(security, 'os', self.os))
|
|
self.enterContext(mock.patch.object(security, 'canonical_path', side_effect=posixpath.normpath))
|
|
|
|
def lstat(self, path, **kwargs):
|
|
try:
|
|
return self.entries[posixpath.normpath(path)]
|
|
except KeyError:
|
|
raise FileNotFoundError(path) from None
|
|
|
|
def access(self, path, mode, *, effective_ids=False):
|
|
self.assertTrue(effective_ids)
|
|
details = self.lstat(path)
|
|
shift = 6 if details.st_uid == 10001 else 3 if details.st_gid == 10001 else 0
|
|
permissions = (details.st_mode >> shift) & 0o7
|
|
if mode == os.W_OK:
|
|
return path in self.writable or bool(permissions & 0o2)
|
|
self.assertEqual(mode, os.X_OK)
|
|
return path not in self.noexec and bool(permissions & 0o1)
|
|
|
|
def test_root_owned_executable_and_traversable_parent_chain_are_accepted_without_repairs(self):
|
|
for mode, gid in ((0o755, 0), (0o555, 0), (0o750, 10001)):
|
|
with self.subTest(mode=oct(mode), gid=gid):
|
|
self.entries[self.executable].st_mode = stat.S_IFREG | mode
|
|
self.entries[self.executable].st_gid = gid
|
|
self.assertEqual(
|
|
security.require_trusted_native_executable(PurePosixPath(self.executable)),
|
|
self.executable,
|
|
)
|
|
self.assertEqual(security.require_trusted_native_executable('/usr/bin/./git'), self.executable)
|
|
self.assertIn(mock.call('/'), self.os.lstat.call_args_list)
|
|
self.os.chmod.assert_not_called()
|
|
self.os.chown.assert_not_called()
|
|
self.os.makedirs.assert_not_called()
|
|
|
|
def test_relative_empty_and_invalid_paths_fail_closed(self):
|
|
for path in ('git', './git', '', None, '/usr/bin/\x00git'):
|
|
with self.subTest(path=path), self.assertRaises(security.PrivateFileError):
|
|
security.require_trusted_native_executable(path)
|
|
self.os.lstat.assert_not_called()
|
|
|
|
def test_nonregular_or_untrusted_executables_are_rejected(self):
|
|
cases = [
|
|
(stat.S_IFREG, 0o755, 10001, 'root-owned'),
|
|
(stat.S_IFREG, 0o755, 10002, 'root-owned'),
|
|
(stat.S_IFLNK, 0o755, 0, 'link'),
|
|
]
|
|
cases.extend((kind, 0o755, 0, 'regular file') for kind in (
|
|
stat.S_IFDIR, stat.S_IFIFO, stat.S_IFSOCK, stat.S_IFCHR, stat.S_IFBLK,
|
|
))
|
|
cases.extend((stat.S_IFREG, mode, 0, 'unsafe permissions') for mode in (
|
|
0o775, 0o757, 0o777, 0o4755, 0o2755, 0o6755,
|
|
))
|
|
for kind, mode, uid, error in cases:
|
|
details = SimpleNamespace(st_mode=kind | mode, st_uid=uid, st_gid=0)
|
|
with self.subTest(kind=kind, mode=oct(mode), uid=uid), \
|
|
mock.patch.dict(self.entries, {self.executable: details}), \
|
|
self.assertRaisesRegex(security.PrivateFileError, error):
|
|
security.require_trusted_native_executable(self.executable)
|
|
|
|
def test_every_parent_including_filesystem_root_must_be_trusted(self):
|
|
for path in ('/', '/usr', '/usr/bin'):
|
|
for mode, uid in ((0o775, 0), (0o777, 0), (0o1777, 0), (0o755, 10001), (0o755, 10002)):
|
|
details = SimpleNamespace(st_mode=stat.S_IFDIR | mode, st_uid=uid, st_gid=0)
|
|
with self.subTest(path=path, mode=oct(mode), uid=uid), \
|
|
mock.patch.dict(self.entries, {path: details}), \
|
|
self.assertRaises(security.PrivateFileError):
|
|
security.require_trusted_native_executable(self.executable)
|
|
with mock.patch.dict(self.entries, {'/usr': SimpleNamespace(st_mode=stat.S_IFREG | 0o755)}), \
|
|
self.assertRaisesRegex(security.PrivateFileError, 'parent is not a directory'):
|
|
security.require_trusted_native_executable(self.executable)
|
|
|
|
def test_parent_links_cannot_be_hidden_by_canonicalization(self):
|
|
for path, executable in (
|
|
('/usr', self.executable), ('/usr/bin', self.executable),
|
|
('/usr/linked', '/usr/linked/../bin/git'),
|
|
):
|
|
details = SimpleNamespace(st_mode=stat.S_IFLNK | 0o755, st_uid=0, st_gid=0)
|
|
with self.subTest(path=path), mock.patch.dict(self.entries, {path: details}), \
|
|
self.assertRaisesRegex(security.PrivateFileError, 'link'):
|
|
security.require_trusted_native_executable(executable)
|
|
|
|
def test_execute_access_is_for_runtime_credentials_not_root_or_any_execute_bit(self):
|
|
for mode in (0o644, 0o700, 0o750):
|
|
with self.subTest(mode=oct(mode)):
|
|
self.entries[self.executable].st_mode = stat.S_IFREG | mode
|
|
with self.assertRaisesRegex(security.PrivateFileError, 'not executable/searchable'):
|
|
security.require_trusted_native_executable(self.executable)
|
|
|
|
def test_effective_access_checks_reject_writes_and_noexec_even_with_safe_modes(self):
|
|
for paths, error in ((self.writable, 'writable'), (self.noexec, 'not executable/searchable')):
|
|
for path in self.entries:
|
|
with self.subTest(path=path, error=error):
|
|
paths.add(path)
|
|
try:
|
|
with self.assertRaisesRegex(security.PrivateFileError, error):
|
|
security.require_trusted_native_executable(self.executable)
|
|
finally:
|
|
paths.remove(path)
|
|
|
|
def test_missing_inaccessible_and_unsupported_inspections_fail_closed(self):
|
|
for error in (FileNotFoundError('missing'), PermissionError('denied'), ValueError('invalid')):
|
|
with self.subTest(error=type(error).__name__), \
|
|
mock.patch.object(self.os, 'lstat', side_effect=error), \
|
|
self.assertRaises(security.PrivateFileError):
|
|
security.require_trusted_native_executable(self.executable)
|
|
with mock.patch.object(self.os, 'access', side_effect=NotImplementedError('effective IDs')), \
|
|
self.assertRaises(security.PrivateFileError):
|
|
security.require_trusted_native_executable(self.executable)
|
|
|
|
def test_private_file_and_directory_policy_remains_owner_only(self):
|
|
with mock.patch.object(security, 'reject_reparse_components', side_effect=lambda path: path):
|
|
for mode, uid, expected in ((0o400, 10001, True), (0o600, 10001, True),
|
|
(0o644, 10001, False), (0o755, 0, False), (0o600, 0, False)):
|
|
self.entries[self.executable] = SimpleNamespace(st_mode=stat.S_IFREG | mode, st_uid=uid)
|
|
with self.subTest(mode=oct(mode), uid=uid):
|
|
self.assertEqual(security.private_file_ready(self.executable), expected)
|
|
for mode, uid, expected in ((0o700, 10001, True), (0o755, 10001, False), (0o700, 0, False)):
|
|
self.entries['/usr/bin'] = SimpleNamespace(st_mode=stat.S_IFDIR | mode, st_uid=uid)
|
|
with self.subTest(directory_mode=oct(mode), uid=uid):
|
|
self.assertEqual(security.private_directory_ready('/usr/bin'), expected)
|
|
|
|
def test_posix_endpoint_lock_root_is_fixed_across_configs_and_environment(self):
|
|
identity = 'fixture-endpoint'
|
|
expected = '/run/truf/authority/endpoint-' + hashlib.sha256(identity.encode()).hexdigest() + '.lock'
|
|
with mock.patch.dict(os.environ, {
|
|
'TRUF_AUTHORITY_LOCK_ROOT': '/tmp/alternate', 'TRUF_AUTHORITY_DIR': '/tmp/other',
|
|
}):
|
|
self.assertEqual(security._cluster_endpoint_lock_root(), '/run/truf/authority')
|
|
for root in ('/opt/truf', '/tmp/alternate'):
|
|
config = {'global': {'root_dir': root, 'runtime_dir': root + '/runtime', 'authority_dir': root}}
|
|
self.assertEqual(security.cluster_endpoint_authority_lock_path(config, identity), expected)
|
|
self.assertNotEqual(security.cluster_endpoint_authority_lock_path(endpoint_identity='other'), expected)
|
|
|
|
|
|
class WindowsNativePolicyTests(unittest.TestCase):
|
|
def test_native_helper_delegates_to_existing_private_file_policy(self):
|
|
path = r'C:\private\git.exe'
|
|
with mock.patch.object(security, 'os', platform_os('nt', ntpath)), \
|
|
mock.patch.object(security, 'require_private_file', return_value=path) as private, \
|
|
mock.patch.object(security, 'canonical_path', return_value=path) as canonical:
|
|
self.assertEqual(security.require_trusted_native_executable(path), path)
|
|
private.assert_called_once_with(path)
|
|
canonical.assert_called_once_with(path)
|
|
private.side_effect = security.PrivateFileError('not private')
|
|
with self.assertRaisesRegex(security.PrivateFileError, 'not private'):
|
|
security.require_trusted_native_executable(path)
|
|
for error in (PermissionError('denied'), TypeError('invalid path'), ValueError('invalid path')):
|
|
private.side_effect = error
|
|
with self.subTest(error=type(error).__name__), self.assertRaises(security.PrivateFileError):
|
|
security.require_trusted_native_executable(path)
|
|
|
|
def test_windows_endpoint_lock_root_is_unchanged(self):
|
|
with mock.patch.object(security, 'os', platform_os('nt', ntpath)), \
|
|
mock.patch.object(security, '_windows_common_appdata', return_value=r'C:\ProgramData'):
|
|
self.assertEqual(security._cluster_endpoint_lock_root(), r'C:\ProgramData\Truf\authority')
|
|
|
|
|
|
class ContainerLifecyclePolicyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
temporary = self.enterContext(tempfile.TemporaryDirectory())
|
|
self.root = Path(temporary)
|
|
self.app = self.root / 'app'
|
|
self.app.mkdir()
|
|
self.enterContext(mock.patch.object(authority, 'CODE_AUTHORITY_FILES', ('supervisor.py', 'runtime_security.py')))
|
|
for name in (*authority.CODE_AUTHORITY_FILES, *authority.EXTERNAL_CODE_AUTHORITY_FILES):
|
|
path = self.app / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text('# inert fixture\n', encoding='ascii')
|
|
self.git = self.root / 'git.exe'
|
|
self.trufflehog = self.root / 'trufflehog.exe'
|
|
self.policy = self.app / 'policy.yaml'
|
|
self.config_path = self.app / 'config.yaml'
|
|
self.secrets = self.app / 'secrets.yaml'
|
|
for path in (self.git, self.trufflehog, self.policy, self.config_path, self.secrets):
|
|
path.write_bytes(b'inert fixture\n')
|
|
self.manifest = authority.build_code_manifest(self.app, self.trufflehog, [self.policy], git_path=self.git)
|
|
self.native = {name: entry['path'] for name, entry in self.manifest['executables'].items()}
|
|
self.private = {entry['path'] for entry in self.manifest['files'].values()} | {security.canonical_path(self.policy)}
|
|
self.config = {'global': {
|
|
'root_dir': str(self.root), 'project_dir': str(self.app),
|
|
'trufflehog_path': str(self.trufflehog), 'trufflehog_config': str(self.policy),
|
|
'secrets_file': str(self.secrets),
|
|
}}
|
|
|
|
def test_external_authority_files_are_windows_only(self):
|
|
expected = (
|
|
'../runtime/check-openrouter-keys.ps1', '../start_core_runtime.ps1',
|
|
'../start_runtime.ps1', '../stop_runtime.ps1',
|
|
) if os.name == 'nt' else ()
|
|
self.assertEqual(authority.EXTERNAL_CODE_AUTHORITY_FILES, expected)
|
|
self.assertEqual(authority.verify_code_manifest(self.manifest), self.manifest)
|
|
|
|
def test_project_private_git_preference_is_windows_only(self):
|
|
private_git = self.root / 'runtime' / 'git' / 'cmd' / 'git.exe'
|
|
private_git.parent.mkdir(parents=True, exist_ok=True)
|
|
private_git.write_bytes(b'private Git fixture')
|
|
for platform in ('nt', 'posix'):
|
|
with self.subTest(platform=platform), \
|
|
mock.patch.object(authority, 'os', platform_os(platform)), \
|
|
mock.patch.object(authority.shutil, 'which', return_value=str(self.git)) as which:
|
|
resolved = authority.resolve_manifest_executable(None, name='git', app_dir=self.app)
|
|
self.assertEqual(resolved, security.canonical_path(private_git if platform == 'nt' else self.git))
|
|
if platform == 'nt':
|
|
which.assert_not_called()
|
|
else:
|
|
which.assert_called_once_with('git')
|
|
self.assertEqual(authority.resolve_manifest_executable(self.git, name='git'), self.native['git'])
|
|
|
|
def test_posix_resolution_rejects_inexact_paths_and_links_before_canonicalization(self):
|
|
with mock.patch.object(authority, 'os', platform_os('posix', posixpath)):
|
|
for path in ('./git', '/usr/bin/../bin/git'):
|
|
with self.subTest(path=path), self.assertRaisesRegex(authority.LifecycleAuthorityError, 'exact and absolute'):
|
|
authority.resolve_manifest_executable(path, name='git')
|
|
with mock.patch.object(authority, 'reject_reparse_components', side_effect=security.PrivateFileError('link')), \
|
|
mock.patch.object(authority, 'canonical_path') as canonical, \
|
|
self.assertRaisesRegex(authority.LifecycleAuthorityError, 'unsafe'):
|
|
authority.resolve_manifest_executable('/usr/bin/git', name='git')
|
|
canonical.assert_not_called()
|
|
|
|
def test_posix_manifest_splits_native_and_private_policies(self):
|
|
with mock.patch.object(authority, 'os', platform_os('posix')), \
|
|
mock.patch.object(authority, 'private_file_ready', side_effect=lambda path: path in self.private) as private, \
|
|
mock.patch.object(authority, 'require_trusted_native_executable', side_effect=lambda path: path) as native:
|
|
self.assertEqual(authority.verify_code_manifest(self.manifest, require_private_acl=True), self.manifest)
|
|
self.assertEqual({call.args[0] for call in private.call_args_list}, self.private)
|
|
self.assertEqual({call.args[0] for call in native.call_args_list}, set(self.native.values()))
|
|
|
|
def test_windows_manifest_still_requires_private_native_files(self):
|
|
with mock.patch.object(authority, 'os', platform_os('nt')), \
|
|
mock.patch.object(authority, 'private_file_ready', return_value=True) as private, \
|
|
mock.patch.object(authority, 'require_trusted_native_executable') as native:
|
|
authority.verify_code_manifest(self.manifest, require_private_acl=True)
|
|
self.assertEqual({call.args[0] for call in private.call_args_list}, self.private | set(self.native.values()))
|
|
native.assert_not_called()
|
|
private.side_effect = lambda path: path != self.native['git']
|
|
with self.assertRaisesRegex(authority.LifecycleAuthorityError, 'exact-private: executable:git'):
|
|
authority.verify_code_manifest(self.manifest, require_private_acl=True)
|
|
|
|
def test_native_policy_failures_are_rejected_before_hashing(self):
|
|
for name, path in self.native.items():
|
|
def check(value):
|
|
if value == path:
|
|
raise security.PrivateFileError('unsafe fixture')
|
|
return value
|
|
|
|
with self.subTest(name=name), mock.patch.object(authority, 'os', platform_os('posix')), \
|
|
mock.patch.object(authority, 'private_file_ready', return_value=True), \
|
|
mock.patch.object(authority, 'require_trusted_native_executable', side_effect=check), \
|
|
mock.patch.object(authority, 'sha256_file', wraps=security.sha256_file) as hashed:
|
|
with self.assertRaisesRegex(authority.LifecycleAuthorityError, 'executable:' + name):
|
|
authority.verify_code_manifest(self.manifest, require_private_acl=True)
|
|
self.assertNotIn(mock.call(path), hashed.call_args_list)
|
|
|
|
def test_code_and_policy_do_not_inherit_native_exemptions(self):
|
|
for path in (security.canonical_path(self.app / 'supervisor.py'), security.canonical_path(self.policy)):
|
|
with self.subTest(path=path), mock.patch.object(authority, 'os', platform_os('posix')), \
|
|
mock.patch.object(authority, 'private_file_ready', side_effect=lambda value: value != path), \
|
|
mock.patch.object(authority, 'require_trusted_native_executable'), \
|
|
self.assertRaisesRegex(authority.LifecycleAuthorityError, 'exact-private'):
|
|
authority.verify_code_manifest(self.manifest, require_private_acl=True)
|
|
manifest = copy.deepcopy(self.manifest)
|
|
manifest['assets'][self.native['git']] = manifest['executables']['git'].copy()
|
|
with mock.patch.object(authority, 'os', platform_os('posix')), \
|
|
mock.patch.object(authority, 'private_file_ready', side_effect=lambda path: path in self.private), \
|
|
mock.patch.object(authority, 'require_trusted_native_executable'), \
|
|
self.assertRaisesRegex(authority.LifecycleAuthorityError, 'exact-private: asset:'):
|
|
authority.verify_code_manifest(manifest, require_private_acl=True)
|
|
|
|
def test_supported_native_identity_sets_are_exact_and_validated(self):
|
|
self.assertEqual(set(self.manifest['executables']), {'git', 'trufflehog'})
|
|
server_manifest = authority.build_code_manifest(
|
|
self.app, policy_paths=[self.policy], git_path=self.git,
|
|
include_trufflehog=False,
|
|
)
|
|
self.assertEqual(set(server_manifest['executables']), {'git'})
|
|
self.assertEqual(authority.verify_code_manifest(server_manifest), server_manifest)
|
|
|
|
without_trufflehog = copy.deepcopy(self.manifest)
|
|
del without_trufflehog['executables']['trufflehog']
|
|
self.assertEqual(authority.verify_code_manifest(without_trufflehog), without_trufflehog)
|
|
|
|
for label, original in (('full', self.manifest), ('server', server_manifest)):
|
|
for name in original['executables']:
|
|
for change in ('relative', 'digest', 'entry'):
|
|
manifest = copy.deepcopy(original)
|
|
if change == 'entry':
|
|
manifest['executables'][name] = None
|
|
else:
|
|
field, value = ('path', 'relative') if change == 'relative' else ('sha256', 'z' * 64)
|
|
manifest['executables'][name][field] = value
|
|
with self.subTest(label=label, name=name, change=change), \
|
|
self.assertRaises(authority.LifecycleAuthorityError):
|
|
authority.verify_code_manifest(manifest)
|
|
|
|
extra = copy.deepcopy(original)
|
|
extra['executables']['shell'] = original['executables']['git'].copy()
|
|
with self.subTest(label=label, change='extra'), \
|
|
self.assertRaises(authority.LifecycleAuthorityError):
|
|
authority.verify_code_manifest(extra)
|
|
|
|
missing_git = copy.deepcopy(original)
|
|
del missing_git['executables']['git']
|
|
with self.subTest(label=label, change='missing-git'), \
|
|
self.assertRaises(authority.LifecycleAuthorityError):
|
|
authority.verify_code_manifest(missing_git)
|
|
|
|
def test_server_preflight_requires_git_without_trufflehog(self):
|
|
config = copy.deepcopy(self.config)
|
|
config['global'].pop('trufflehog_path')
|
|
config['global'].pop('trufflehog_config')
|
|
|
|
with mock.patch.object(security, 'os', platform_os('posix')), \
|
|
mock.patch.object(security, 'require_private_directory'), \
|
|
mock.patch.object(security, 'require_private_file'), \
|
|
mock.patch.object(security, 'require_trusted_native_executable'), \
|
|
mock.patch.object(
|
|
authority, 'resolve_manifest_executable',
|
|
side_effect=lambda value, **kwargs: self.native[kwargs['name']],
|
|
) as resolve:
|
|
self.assertTrue(security.preflight_lifecycle_paths(
|
|
str(self.config_path), config, authority_profile='server',
|
|
))
|
|
|
|
self.assertEqual(
|
|
[call.kwargs['name'] for call in resolve.call_args_list],
|
|
['git'],
|
|
)
|
|
|
|
def test_both_native_hashes_detect_drift_with_unchanged_size_and_mtime(self):
|
|
for name, path in self.native.items():
|
|
original = Path(path).read_bytes()
|
|
before = os.stat(path)
|
|
Path(path).write_bytes(b'I' + original[1:])
|
|
os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns))
|
|
with self.subTest(name=name), mock.patch.object(authority, 'os', platform_os('posix')), \
|
|
mock.patch.object(authority, 'private_file_ready', return_value=True), \
|
|
mock.patch.object(authority, 'require_trusted_native_executable'):
|
|
self.assertEqual(os.stat(path).st_size, before.st_size)
|
|
self.assertEqual(os.stat(path).st_mtime_ns, before.st_mtime_ns)
|
|
for private_acl in (False, True):
|
|
with self.assertRaisesRegex(authority.LifecycleAuthorityError, 'drifted: executable:' + name):
|
|
authority.verify_code_manifest(self.manifest, require_private_acl=private_acl)
|
|
Path(path).write_bytes(original)
|
|
|
|
def test_manifest_digest_binds_exact_native_paths_even_with_identical_bytes(self):
|
|
expected = authority.code_manifest_sha256(self.manifest)
|
|
alternate = self.root / 'alternate.exe'
|
|
alternate.write_bytes(self.git.read_bytes())
|
|
manifest = copy.deepcopy(self.manifest)
|
|
manifest['executables']['git']['path'] = security.canonical_path(alternate)
|
|
with self.assertRaisesRegex(authority.LifecycleAuthorityError, 'manifest digest mismatch'):
|
|
authority.verify_code_manifest(manifest, expected_sha256=expected)
|
|
with mock.patch.object(authority, 'os', platform_os('posix')), \
|
|
mock.patch.object(authority, 'canonical_path', side_effect=lambda path: (
|
|
security.canonical_path(alternate) if path == self.native['git'] else security.canonical_path(path)
|
|
)), self.assertRaisesRegex(authority.LifecycleAuthorityError, 'path is not exact'):
|
|
authority.normalize_code_manifest(self.manifest)
|
|
|
|
def test_private_inventory_excludes_native_only_when_explicitly_requested(self):
|
|
with mock.patch.object(authority, 'resolve_manifest_executable', side_effect=AssertionError('native lookup')):
|
|
paths = authority.manifest_authority_paths(
|
|
self.app, self.trufflehog, [self.policy], existing_only=True, include_executables=False,
|
|
)
|
|
self.assertEqual(set(paths), self.private)
|
|
self.assertEqual(set(authority.manifest_authority_paths(
|
|
self.app, self.trufflehog, [self.policy], existing_only=True, git_path=self.git,
|
|
)), self.private | set(self.native.values()))
|
|
|
|
def test_preflight_splits_native_from_private_code_config_policy_and_secrets(self):
|
|
for platform in ('posix', 'nt'):
|
|
def private_check(path):
|
|
if platform == 'posix' and security.canonical_path(path) in self.native.values():
|
|
raise security.PrivateFileError('native file is not private')
|
|
return path
|
|
|
|
def native_check(path):
|
|
return security.require_private_file(path) if platform == 'nt' else path
|
|
|
|
with self.subTest(platform=platform), mock.patch.object(security, 'os', platform_os(platform)), \
|
|
mock.patch.object(security, 'require_private_directory') as directory, \
|
|
mock.patch.object(security, 'require_private_file', side_effect=private_check) as private, \
|
|
mock.patch.object(security, 'require_trusted_native_executable', side_effect=native_check) as native, \
|
|
mock.patch.object(authority, 'resolve_manifest_executable', side_effect=lambda value, **kw: self.native[kw['name']]) as resolve, \
|
|
mock.patch.object(security, 'harden_private_file') as harden_file, \
|
|
mock.patch.object(security, 'harden_private_directory') as harden_directory:
|
|
self.assertTrue(security.preflight_lifecycle_paths(str(self.config_path), self.config))
|
|
expected = self.private | {security.canonical_path(self.config_path), security.canonical_path(self.secrets)}
|
|
if platform == 'nt':
|
|
expected |= set(self.native.values())
|
|
else:
|
|
directory.assert_any_call(os.path.abspath('/run/truf/authority'), create=False)
|
|
self.assertEqual({security.canonical_path(call.args[0]) for call in private.call_args_list}, expected)
|
|
self.assertEqual({call.args[0] for call in native.call_args_list}, set(self.native.values()))
|
|
self.assertEqual(resolve.call_count, 2)
|
|
harden_file.assert_not_called()
|
|
harden_directory.assert_not_called()
|
|
|
|
def test_preflight_rejects_public_config_or_policy_even_when_native_is_trusted(self):
|
|
for target in (self.config_path, self.policy):
|
|
def private_check(path):
|
|
if security.canonical_path(path) == security.canonical_path(target):
|
|
raise security.PrivateFileError('not private')
|
|
return path
|
|
|
|
with self.subTest(path=str(target)), mock.patch.object(security, 'os', platform_os('posix')), \
|
|
mock.patch.object(security, 'require_private_directory'), \
|
|
mock.patch.object(security, 'require_private_file', side_effect=private_check), \
|
|
mock.patch.object(security, 'require_trusted_native_executable') as native, \
|
|
self.assertRaisesRegex(security.PrivateFileError, 'offline hardening'):
|
|
security.preflight_lifecycle_paths(str(self.config_path), self.config)
|
|
native.assert_not_called()
|
|
|
|
def test_preflight_requires_both_native_tools_even_without_explicit_configuration(self):
|
|
config = copy.deepcopy(self.config)
|
|
config['global'].pop('trufflehog_path')
|
|
for missing in self.native:
|
|
def resolve(value, **kwargs):
|
|
if kwargs['name'] == missing:
|
|
raise authority.LifecycleAuthorityError('missing native fixture')
|
|
return self.native[kwargs['name']]
|
|
|
|
with self.subTest(missing=missing), mock.patch.object(security, 'os', platform_os('posix')), \
|
|
mock.patch.object(security, 'require_private_directory'), \
|
|
mock.patch.object(security, 'require_private_file'), \
|
|
mock.patch.object(security, 'require_trusted_native_executable'), \
|
|
mock.patch.object(authority, 'resolve_manifest_executable', side_effect=resolve), \
|
|
self.assertRaisesRegex(security.PrivateFileError, 'missing native fixture'):
|
|
security.preflight_lifecycle_paths(str(self.config_path), config)
|
|
|
|
def test_preflight_requires_prepared_private_authority_root_without_repair(self):
|
|
def require_directory(path, create=False):
|
|
self.assertFalse(create)
|
|
if path == os.path.abspath('/run/truf/authority'):
|
|
raise security.PrivateFileError('authority directory is not private')
|
|
|
|
with mock.patch.object(security, 'os', platform_os('posix')), \
|
|
mock.patch.object(security, 'require_private_directory', side_effect=require_directory), \
|
|
mock.patch.object(security, 'require_private_file'), \
|
|
mock.patch.object(security, 'harden_private_directory') as harden, \
|
|
self.assertRaisesRegex(security.PrivateFileError, 'authority directory is not private'):
|
|
security.preflight_lifecycle_paths(str(self.config_path), self.config)
|
|
harden.assert_not_called()
|
|
|
|
|
|
@unittest.skipUnless(sys.platform.startswith('linux'), 'native Linux filesystem checks')
|
|
class LinuxFilesystemSecurityTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.root = Path(self.enterContext(tempfile.TemporaryDirectory(prefix='truf-security-')))
|
|
|
|
def test_private_paths_require_runtime_ownership_and_owner_only_modes(self):
|
|
directory = self.root / 'private'
|
|
security.ensure_private_directory(directory, reject_reparse=True)
|
|
path = directory / 'secret'
|
|
path.write_bytes(b'fake secret fixture')
|
|
for mode, expected in ((0o400, True), (0o600, True), (0o644, False), (0o755, False)):
|
|
path.chmod(mode)
|
|
with self.subTest(mode=oct(mode)):
|
|
self.assertEqual(path.stat().st_uid, os.geteuid())
|
|
self.assertEqual(stat.S_IMODE(path.stat().st_mode), mode)
|
|
self.assertEqual(security.private_file_ready(path), expected)
|
|
self.assertTrue(security.private_directory_ready(directory))
|
|
directory.chmod(0o755)
|
|
with self.assertRaises(security.PrivateFileError):
|
|
security.require_private_directory(directory)
|
|
self.assertEqual(stat.S_IMODE(directory.stat().st_mode), 0o755)
|
|
|
|
def test_runtime_owned_executable_is_not_immutable_native_code(self):
|
|
if os.geteuid() == 0:
|
|
self.skipTest('requires an unprivileged runtime user')
|
|
path = self.root / 'fake-native'
|
|
path.write_bytes(b'not executed')
|
|
path.chmod(0o700)
|
|
self.assertTrue(security.private_file_ready(path))
|
|
with self.assertRaisesRegex(security.PrivateFileError, 'root-owned'):
|
|
security.require_trusted_native_executable(path)
|
|
self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o700)
|
|
|
|
def test_installed_root_owned_native_file_is_trusted_but_not_private(self):
|
|
if os.geteuid() == 0:
|
|
self.skipTest('requires an unprivileged runtime user')
|
|
path = '/usr/bin/true'
|
|
if not os.path.isfile(path):
|
|
self.skipTest('coreutils fixture is unavailable')
|
|
before = os.stat(path)
|
|
self.assertEqual(security.require_trusted_native_executable(path), security.canonical_path(path))
|
|
self.assertFalse(security.private_file_ready(path))
|
|
after = os.stat(path)
|
|
self.assertEqual((before.st_mode, before.st_uid, before.st_mtime_ns), (after.st_mode, after.st_uid, after.st_mtime_ns))
|
|
|
|
def test_native_leaf_and_parent_links_are_rejected_without_touching_targets(self):
|
|
if os.geteuid() == 0:
|
|
self.skipTest('requires an unprivileged runtime user')
|
|
target = Path('/usr/bin/true')
|
|
if not target.is_file():
|
|
self.skipTest('coreutils fixture is unavailable')
|
|
before = target.stat()
|
|
leaf = self.root / 'native-link'
|
|
leaf.symlink_to(target)
|
|
parent = self.root / 'bin'
|
|
parent.symlink_to(target.parent, target_is_directory=True)
|
|
for path in (str(leaf), str(parent / 'true'), str(parent) + '/../bin/true'):
|
|
with self.subTest(path=path), self.assertRaisesRegex(security.PrivateFileError, 'link'):
|
|
security.require_trusted_native_executable(path)
|
|
with self.assertRaises(authority.LifecycleAuthorityError):
|
|
authority.resolve_manifest_executable(path, name='git')
|
|
after = target.stat()
|
|
self.assertEqual((before.st_mode, before.st_uid, before.st_mtime_ns), (after.st_mode, after.st_uid, after.st_mtime_ns))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|