[verified] Define fail-closed worktree creator-role compatibility

This commit is contained in:
2026-10-12 02:10:37 +00:00
parent 2ad836f332
commit aeab04289d
5 changed files with 62 additions and 4 deletions
@@ -252,7 +252,57 @@ class NativeReviewTests(unittest.TestCase):
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
lease = m.HeldIssueLease(fd, policy, time.monotonic() + 30)
canonical_bytes = state.read_bytes(); canonical_stat = state.stat()
lease.check(p)
self.assertEqual(state.read_bytes(), canonical_bytes)
self.assertEqual((state.stat().st_mode, state.stat().st_mtime_ns, state.stat().st_ino),
(canonical_stat.st_mode, canonical_stat.st_mtime_ns, canonical_stat.st_ino))
# The bounded compatibility admission is read-only: mutate only
# this scratch fixture from canonical producer state to legacy impl.
state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo',
'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'}))
state.chmod(0o600)
legacy_bytes = state.read_bytes(); legacy_stat = state.stat()
try:
lease.check(p)
except m.ValidationError as exc:
self.fail(f'valid legacy impl registration must be admitted: {exc}')
after_stat = state.stat()
self.assertEqual(state.read_bytes(), legacy_bytes)
self.assertEqual((after_stat.st_mode, after_stat.st_mtime_ns, after_stat.st_ino),
(legacy_stat.st_mode, legacy_stat.st_mtime_ns, legacy_stat.st_ino))
for bad_role in [None, True, False, 1, [], {}, '', ' impl', 'impl ', 'Impl', 'IMPL', 'gitea-impl-agent ', 'review', 'gitea-review-agent', 'gitea-spec-agent', 'gitea-release-agent', 'other']:
state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo',
'issue': 3, 'created_by_role': bad_role, 'mode': 'detached'}))
state.chmod(0o600)
with self.subTest(bad_role=repr(bad_role)), self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'):
lease.check(p)
missing_role = json.loads(legacy_bytes); missing_role.pop('created_by_role')
state.write_text(json.dumps(missing_role)); state.chmod(0o600)
with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p)
bool_issue = json.loads(legacy_bytes); bool_issue['issue'] = True
state.write_text(json.dumps(bool_issue)); state.chmod(0o600)
with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p)
state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'other', 'repository': 'repo',
'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600)
with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p)
state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'other',
'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600)
with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p)
state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo',
'issue': 4, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600)
with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p)
state.write_text(json.dumps({'schema': 'wrong/v1', 'owner': 'acme', 'repository': 'repo',
'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600)
with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p)
state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo',
'issue': 3, 'created_by_role': 'impl', 'mode': 'attached'})); state.chmod(0o600)
with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p)
state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo',
'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600)
wrong_head = copy.deepcopy(p); wrong_head['target']['head'] = 'b' * 40
wrong_head['body'] = wrong_head['body'].replace(head, 'b' * 40)
with self.assertRaisesRegex(m.ValidationError, 'worktree HEAD mismatch'): lease.check(wrong_head)
fcntl.flock(fd, fcntl.LOCK_UN)
with self.assertRaises(m.ValidationError): lease.check(p)
self.assertNotIn('lock:', Path(f'/proc/self/fdinfo/{fd}').read_text())