108 lines
4.3 KiB
Python
108 lines
4.3 KiB
Python
"""Read-only worktree topology gate; caller owns admission and repository lock.
|
|
|
|
Reuse the harness inventory binder's bounded, fixed-Git reader and pinned,
|
|
no-symlink directory opening. This is not a standalone cache authenticator.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
|
|
if __package__:
|
|
from . import test_inventory_binding as binding
|
|
else:
|
|
import test_inventory_binding as binding
|
|
|
|
|
|
class PrefetchError(ValueError):
|
|
pass
|
|
|
|
|
|
def parse_worktrees(raw):
|
|
if type(raw) is not bytes or not raw or not raw.endswith(b'\0\0'):
|
|
raise PrefetchError('invalid_worktree_listing')
|
|
records = []
|
|
seen = set()
|
|
for chunk in raw[:-2].split(b'\0\0'):
|
|
fields = chunk.split(b'\0')
|
|
if not fields or not fields[0].startswith(b'worktree '):
|
|
raise PrefetchError('invalid_worktree_listing')
|
|
try:
|
|
path = fields[0][9:].decode('utf-8')
|
|
except UnicodeError:
|
|
raise PrefetchError('invalid_worktree_listing') from None
|
|
if not path or not os.path.isabs(path) or path != os.path.normpath(path) or path in seen:
|
|
raise PrefetchError('invalid_worktree_listing')
|
|
seen.add(path)
|
|
tokens = fields[1:]
|
|
if tokens == [b'bare']:
|
|
records.append({'path': path, 'kind': 'bare'})
|
|
continue
|
|
if len(tokens) < 2 or not re.fullmatch(rb'HEAD [0-9a-f]{40}|HEAD [0-9a-f]{64}', tokens[0]):
|
|
raise PrefetchError('invalid_worktree_listing')
|
|
branch = None
|
|
if tokens[1].startswith(b'branch refs/heads/'):
|
|
try:
|
|
branch = tokens[1][7:].decode('utf-8')
|
|
except UnicodeError:
|
|
raise PrefetchError('invalid_worktree_listing') from None
|
|
parts = branch.split('/')
|
|
if (not branch[len('refs/heads/'):] or '..' in branch or '@{' in branch
|
|
or re.search(r'[\x00-\x20\x7f~^:?*\[\\\\]', branch)
|
|
or any(not p or p.startswith('.') or p.endswith(('.', '.lock')) for p in parts)):
|
|
raise PrefetchError('invalid_worktree_listing')
|
|
elif tokens[1] != b'detached':
|
|
raise PrefetchError('invalid_worktree_listing')
|
|
optional = tokens[2:]
|
|
keys = [x.split(b' ', 1)[0] for x in optional]
|
|
if any(x not in (b'locked', b'prunable') for x in keys) or len(set(keys)) != len(keys):
|
|
raise PrefetchError('invalid_worktree_listing')
|
|
record = {'path': path, 'kind': 'attached' if branch is not None else 'detached'}
|
|
if branch is not None:
|
|
record['branch'] = branch
|
|
records.append(record)
|
|
if sum(x['kind'] == 'bare' for x in records) != 1:
|
|
raise PrefetchError('invalid_worktree_listing')
|
|
return records
|
|
|
|
|
|
def inspect_repository(git_dir):
|
|
fd = None
|
|
try:
|
|
fd = binding._canonical_git_dir(git_dir)
|
|
pinned = '/proc/self/fd/' + str(fd)
|
|
bare = binding._git(pinned, ['rev-parse', '--is-bare-repository'], check=True, git_fd=fd)
|
|
if bare is None or bare.stdout != b'true\n':
|
|
raise PrefetchError('not_bare_repository')
|
|
listing = binding._git(pinned, ['worktree', 'list', '--porcelain', '-z'], check=True, git_fd=fd)
|
|
if listing is None:
|
|
raise PrefetchError('git_read_unavailable')
|
|
records = parse_worktrees(listing.stdout)
|
|
attached = [{'branch': x['branch'], 'path': x['path']} for x in records if x['kind'] == 'attached']
|
|
return {'status': 'block' if attached else 'pass', 'attached_worktrees': attached,
|
|
'reason': 'attached_worktree' if attached else 'no_attached_worktrees'}
|
|
except PrefetchError:
|
|
raise
|
|
except (binding.BindingError, OSError, ValueError, TypeError):
|
|
raise PrefetchError('git_read_unavailable') from None
|
|
finally:
|
|
if fd is not None:
|
|
os.close(fd)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--git-dir', required=True, help='Already-admitted canonical absolute bare mirror')
|
|
args = parser.parse_args()
|
|
try:
|
|
result = inspect_repository(args.git_dir)
|
|
except PrefetchError as error:
|
|
print(json.dumps({'status': 'block', 'attached_worktrees': [], 'reason': str(error)}))
|
|
return 2
|
|
print(json.dumps(result, ensure_ascii=True))
|
|
return 0 if result['status'] == 'pass' else 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|