"""Real-Git regressions; no pytest or product execution dependency.""" import copy import hashlib import json import os import subprocess import sys import tempfile import unittest from pathlib import Path from unittest.mock import patch from harness.lib import test_inventory_binding as binding ROOT = Path(__file__).resolve().parents[3] class InventoryBindingTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.work = self.root / 'work' self.work.mkdir() self.git('init', '-q') self.git('config', 'user.email', 'fixture@example.invalid') self.git('config', 'user.name', 'Fixture') (self.work / 'tests').mkdir() (self.work / 'tests/sample.py').write_text('def test_sample():\n assert True\n') (self.work / 'config.ini').write_text('[test]\nvalue=1\n') self.git('add', '.') self.git('commit', '-qm', 'baseline') self.rev = self.git('rev-parse', 'HEAD').strip() self.gitdir = self.work / '.git' self.data = self.make_data(self.rev) def git(self, *args): return subprocess.check_output(['git', '-C', str(self.work), *args], text=True, stderr=subprocess.DEVNULL) def make_data(self, revision): hashes = {p: hashlib.sha256((self.work / p).read_bytes()).hexdigest() for p in ['config.ini', 'tests/sample.py']} return {'schema': 'gitea-agent.test-inventory-binding/v1', 'source_revision': revision, 'stage': 'baseline', 'ids': ['tests/sample.py::test_sample'], 'file_sha256': {'tests/sample.py': hashes['tests/sample.py']}, 'collection': {'source_revision': revision, 'stage': 'baseline', 'argv': ['python3', '-m', 'pytest', '--collect-only'], 'scope_paths': list(hashes), 'scope_sha256': self.scope_digest(hashes), 'evidence': 'retained-collection-record'}} @staticmethod def scope_digest(hashes): return hashlib.sha256(json.dumps(hashes, sort_keys=True, separators=(',', ':')).encode()).hexdigest() def verify(self, data=None, revision=None, stage='baseline', gitdir=None): return binding.verify_inventory(str(gitdir or self.gitdir), self.data if data is None else data, self.rev if revision is None else revision, stage) def cli(self, raw=None, extra=(), input_path=None): inp = self.root / 'input.json' if input_path is None: inp.write_text(json.dumps(self.data) if raw is None else raw) else: inp = input_path return subprocess.run([sys.executable, '-B', str(ROOT / 'harness/lib/test_inventory_binding.py'), '--git-dir', str(self.gitdir), '--input', str(inp), '--expected-revision', self.rev, '--expected-stage', 'baseline', *extra], capture_output=True, text=True, timeout=10) def test_valid_normal_and_bare_without_state_or_ref_changes(self): before = (self.git('status', '--porcelain'), self.git('show-ref')) (self.work / 'tests/sample.py').write_text('uncommitted edit') dirty = self.git('status', '--porcelain') self.assertEqual(self.verify()['status'], 'pass') bare = self.root / 'mirror.git' subprocess.run(['git', 'clone', '-q', '--bare', str(self.work), str(bare)], check=True) self.assertEqual(self.verify(gitdir=bare)['status'], 'pass') self.assertEqual(self.git('status', '--porcelain'), dirty) self.assertEqual(self.git('show-ref'), before[1]) def test_expected_revision_and_stage_cannot_be_self_attested(self): self.assertEqual(self.verify(revision='0' * 40)['reason_codes'], ['source_revision_mismatch']) self.assertEqual(self.verify(stage='current')['reason_codes'], ['stage_mismatch']) newer = copy.deepcopy(self.data) newer['stage'] = newer['collection']['stage'] = 'current' self.assertEqual(self.verify(newer)['status'], 'block') def test_tag_object_is_not_an_exact_commit_revision(self): self.git('tag', '-a', '-m', 'fixture tag', 'fixture-tag') tag = self.git('rev-parse', 'fixture-tag').strip() changed = copy.deepcopy(self.data) changed['source_revision'] = changed['collection']['source_revision'] = tag self.assertEqual(self.verify(changed, revision=tag)['status'], 'block') def test_missing_commit_and_malformed_sha(self): data = copy.deepcopy(self.data) data['source_revision'] = data['collection']['source_revision'] = '0' * 40 self.assertEqual(self.verify(data, revision='0' * 40)['reason_codes'], ['revision_not_found']) for rev in [self.rev.upper(), 'HEAD', self.rev + ' ', True]: with self.subTest(rev=rev), self.assertRaises(binding.BindingError): self.verify(revision=rev) def test_absent_file_and_tampered_recorded_hash_are_blocked(self): changed = copy.deepcopy(self.data) changed['file_sha256']['tests/sample.py'] = '0' * 64 self.assertIn('recorded_file_hash_mismatch', self.verify(changed)['reason_codes']) missing = copy.deepcopy(self.data) missing['ids'] = ['tests/absent.py::test_absent'] missing['file_sha256'] = {'tests/absent.py': '0' * 64} missing['collection']['scope_paths'] = ['tests/absent.py'] self.assertIn('scope_file_missing', self.verify(missing)['reason_codes']) def test_changed_collector_configuration_breaks_scope_digest(self): (self.work / 'config.ini').write_text('[test]\nvalue=2\n') self.git('add', 'config.ini') self.git('commit', '-qm', 'change collection config') revision = self.git('rev-parse', 'HEAD').strip() changed = copy.deepcopy(self.data) changed['source_revision'] = changed['collection']['source_revision'] = revision self.assertIn('scope_digest_mismatch', self.verify(changed, revision=revision)['reason_codes']) def test_nested_source_or_stage_and_omitted_scope_are_rejected(self): for key, value in [('source_revision', '0' * 40), ('stage', 'current'), ('scope_paths', ['config.ini'])]: changed = copy.deepcopy(self.data) changed['collection'][key] = value with self.subTest(key=key), self.assertRaises(binding.BindingError): self.verify(changed) def test_duplicates_unsafe_paths_extra_hash_and_wrong_types(self): variants = [] for value in [self.data['ids'] * 2, [{}], True, ['../outside::case'], ['/outside::case'], ['tests//sample.py::case']]: changed = copy.deepcopy(self.data); changed['ids'] = value; variants.append(changed) changed = copy.deepcopy(self.data); changed['extra'] = True; variants.append(changed) changed = copy.deepcopy(self.data); changed['file_sha256']['extra.py'] = '0' * 64; variants.append(changed) changed = copy.deepcopy(self.data); changed['collection']['scope_paths'] = [{}]; variants.append(changed) for changed in variants: with self.subTest(data_type=type(changed['ids']).__name__), self.assertRaises(binding.BindingError): self.verify(changed) def test_empty_collection_is_valid_but_not_a_collection_execution_claim(self): changed = copy.deepcopy(self.data) changed['ids'] = []; changed['file_sha256'] = {} self.assertEqual(self.verify(changed)['status'], 'pass') changed['collection']['scope_paths'] = [] changed['collection']['scope_sha256'] = self.scope_digest({}) self.assertEqual(self.verify(changed)['status'], 'pass') def test_symlink_tree_and_gitlink_are_not_regular_blobs(self): (self.work / 'tests/link.py').symlink_to('sample.py') self.git('add', 'tests/link.py'); self.git('commit', '-qm', 'add symlink') self.git('update-index', '--add', '--cacheinfo', f'160000,{self.rev},tests/submodule') self.git('commit', '-qm', 'add gitlink') revision = self.git('rev-parse', 'HEAD').strip() for path in ['tests/link.py', 'tests/submodule', 'tests']: changed = copy.deepcopy(self.data) changed['source_revision'] = changed['collection']['source_revision'] = revision changed['ids'] = [path + '::case']; changed['file_sha256'] = {path: '0' * 64} changed['collection']['scope_paths'] = [path] with self.subTest(path=path): self.assertEqual(self.verify(changed, revision=revision)['status'], 'block') def test_symlink_git_directory_ancestor_and_lexical_alias_rejected(self): alias = self.root / 'alias'; alias.symlink_to(self.work, target_is_directory=True) for path in [str(alias / '.git'), str(self.gitdir) + '/', str(self.work / 'x') + '/../.git']: with self.subTest(path=path), self.assertRaises(binding.BindingError): binding.verify_inventory(path, self.data, self.rev, 'baseline') def test_ambient_git_redirects_and_replacements_do_not_change_binding(self): self.git('replace', self.rev, self.rev) with patch.dict(os.environ, {'GIT_DIR': '/nonexistent', 'GIT_WORK_TREE': '/nonexistent', 'GIT_CONFIG_COUNT': '1', 'GIT_CONFIG_KEY_0': 'core.bare', 'GIT_CONFIG_VALUE_0': 'true'}): self.assertEqual(self.verify()['status'], 'pass') def test_json_symlink_ancestor_and_lexical_alias_rejected(self): real = self.root / 'records' real.mkdir() record = real / 'record.json' record.write_text(json.dumps(self.data)) alias = self.root / 'record-alias' alias.symlink_to(real, target_is_directory=True) for path in [str(alias / 'record.json'), str(real) + '/./record.json']: with self.subTest(path=path), self.assertRaises(binding.BindingError): binding._json_file(path) def test_lazy_fetch_disabled_in_read_only_git_environment(self): self.assertEqual(binding._env().get('GIT_NO_LAZY_FETCH'), '1') def test_json_duplicate_keys_symlink_fifo_and_oversize_rejected(self): duplicate = json.dumps(self.data)[:-1] + ',"schema":"secret-marker"}' for raw in [duplicate, '{"secret":"secret-marker"}', '[' * 2000, ' ' * (1024 * 1024 + 1)]: result = self.cli(raw) self.assertEqual(result.returncode, 2) self.assertNotIn('secret-marker', result.stdout + result.stderr) real = self.root / 'real.json'; real.write_text(json.dumps(self.data)) alias = self.root / 'alias.json'; alias.symlink_to(real) self.assertEqual(self.cli(input_path=alias).returncode, 2) fifo = self.root / 'fifo'; os.mkfifo(fifo) self.assertEqual(self.cli(input_path=fifo).returncode, 2) def test_cli_pass_block_and_rejected_input_are_json(self): self.assertEqual(self.cli().returncode, 0) changed = copy.deepcopy(self.data); changed['stage'] = changed['collection']['stage'] = 'current' result = self.cli(json.dumps(changed)) self.assertEqual(result.returncode, 1) self.assertEqual(json.loads(result.stdout)['status'], 'block') result = self.cli('{"secret":"secret-marker"}') self.assertEqual(result.returncode, 2) self.assertNotIn('secret-marker', result.stdout + result.stderr) def test_recorded_argv_is_not_a_shell_command_or_character_policy(self): changed = copy.deepcopy(self.data) changed['collection']['argv'] += ['-k', 'case[param$with|literal;chars]'] try: result = self.verify(changed) except binding.BindingError: self.fail('literal argv must not be rejected as a shell command') self.assertEqual(result['status'], 'pass') def test_cli_unknown_option_never_echoes_input(self): result = self.cli(extra=('--unknown=secret-marker',)) self.assertEqual(result.returncode, 2) self.assertNotIn('secret-marker', result.stdout + result.stderr) self.assertTrue(result.stdout.startswith('{'), 'CLI rejection must be bounded JSON') self.assertEqual(json.loads(result.stdout)['reason_codes'], ['rejected_input']) def test_git_output_limit_terminates_before_wall_clock_timeout(self): import time fake = self.root / 'bin' fake.mkdir() program = fake / 'git' program.write_text('#!' + sys.executable + '\nimport os,time\nos.write(1,b"x"*1024)\ntime.sleep(10)\n') program.chmod(0o700) started = time.monotonic() with patch.object(binding, '_GIT_EXECUTABLE', str(program)), patch.object(binding, '_MAX_OUTPUT', 32): self.assertIsNone(binding._git(str(self.gitdir), ['fixture-read'])) self.assertLess(time.monotonic() - started, 2.0) def test_hostile_path_git_is_never_executed(self): marker = self.root / 'hostile-marker' fake = self.root / 'bin' fake.mkdir() program = fake / 'git' program.write_text('#!' + sys.executable + '\nimport pathlib\npathlib.Path(' + repr(str(marker)) + ').write_text("executed")\n') program.chmod(0o700) with patch.dict(os.environ, {'PATH': str(fake)}): self.assertEqual(self.verify()['status'], 'pass') self.assertFalse(marker.exists()) def test_admitted_git_directory_remains_pinned_after_path_replacement(self): original = binding._git replacement_work = self.root / 'replacement-work' replacement_work.mkdir() subprocess.run(['git', '-C', str(replacement_work), 'init', '-q'], check=True) subprocess.run(['git', '-C', str(replacement_work), 'config', 'user.email', 'replacement@example.invalid'], check=True) subprocess.run(['git', '-C', str(replacement_work), 'config', 'user.name', 'Replacement'], check=True) (replacement_work / 'replacement.txt').write_text('replacement') subprocess.run(['git', '-C', str(replacement_work), 'add', '.'], check=True) subprocess.run(['git', '-C', str(replacement_work), 'commit', '-qm', 'replacement'], check=True) swapped = False def replace_after_admission(git_dir, args, **kwargs): nonlocal swapped result = original(git_dir, args, **kwargs) if not swapped and args[:2] == ['rev-parse', '--is-bare-repository']: swapped = True moved = self.root / 'admitted-original.git' os.rename(self.gitdir, moved) os.rename(replacement_work / '.git', self.gitdir) return result with patch.object(binding, '_git', side_effect=replace_after_admission): self.assertEqual(self.verify()['status'], 'pass') self.assertTrue(swapped) def test_each_scope_blob_is_read_once(self): original = binding._blob with patch.object(binding, '_blob', wraps=original) as read: self.assertEqual(self.verify()['status'], 'pass') self.assertEqual(read.call_count, 2) if __name__ == '__main__': unittest.main()