"""Grant si workspace access while preserving boot-sensitive private permissions.""" import argparse import json import os from pathlib import Path import pwd import stat import subprocess import time ROOT = Path('/workspace') JOURNAL_ROOT = Path('/var/lib/contextra-maintenance') def bits(value): return sum(flag for char, flag in zip(value, (4, 2, 1)) if char != '-') def perms(value): return ''.join(char if value & flag else '-' for char, flag in zip('rwx', (4, 2, 1))) def transform(block, uid, mode, owner): lines = block.splitlines() header = [line for line in lines if line.startswith('# file:') or line.startswith('# flags:')] access, default = {}, {} for line in lines: if not line or line.startswith('#'): continue fields = line.split(':') target = access if fields[0] == 'default': target, fields = default, fields[1:] target[':'.join(fields[:2])] = bits(fields[2]) old_mask = access.get('mask:', 7) executable = bool((access['user:'] | access['other:']) & 1) or any( value & old_mask & 1 for key, value in access.items() if key.startswith('group:') or (key.startswith('user:') and key != 'user:')) desired = 6 | (1 if stat.S_ISDIR(mode) or executable else 0) def grant(entries, wanted): old_mask = entries.get('mask:', 7) for key in list(entries): if key.startswith('group:') or (key.startswith('user:') and key not in ('user:', 'user:' + str(uid))): entries[key] &= old_mask entries['user:' + str(uid)] = wanted entries['mask:'] = 0 for key, value in entries.items(): if key.startswith('group:') or (key.startswith('user:') and key != 'user:'): entries['mask:'] |= value return entries if owner == uid: access['user:'] |= desired original = access.copy() grant(access, desired) if stat.S_ISDIR(mode): if not default: default = {'user:': original['user:'], 'group:': original['group:'] & original.get('mask:', 7), 'other:': original['other:']} grant(default, 7) def render(entries, prefix=''): order = ['user:'] + sorted(k for k in entries if k.startswith('user:') and k != 'user:') order += ['group:'] + sorted(k for k in entries if k.startswith('group:') and k != 'group:') order += ['mask:', 'other:'] return [prefix + key + ':' + perms(entries[key]) for key in order] return '\n'.join(header + render(access) + (render(default, 'default:') if default else [])) + '\n\n' def main(): p = argparse.ArgumentParser() p.add_argument('--apply', action='store_true') args = p.parse_args() assert os.geteuid() == 0 and args.apply os.umask(0o077) uid = pwd.getpwnam('si').pw_uid site = json.loads(Path('/workspace/contextra/runtime/state/site.json').read_text()) protected = {Path(p) for p in site['env_files']} protected |= {Path('/workspace/contextra/data/site') / name for name in ('bridge', 'bridge-session', 'tls')} for mount in ('/backup', '/training'): if not os.path.ismount(mount): raise RuntimeError('Required data filesystem is not mounted') for entry in site['storage'].values(): if not os.path.samefile(entry['physical'], entry['logical']): raise RuntimeError('Data bind view mismatch') journal = JOURNAL_ROOT / ('workspace-acl-' + time.strftime('%Y%m%dT%H%M%S') + '-' + str(os.getpid())) journal.mkdir(mode=0o700, exist_ok=False) protected_modes = {str(p): stat.S_IMODE(p.stat().st_mode) for p in protected} counts = dict(files=0, directories=0, symlinks=0, special=0, protected=0, batches=0) pending = [] visited = set() def excluded(path): return (any(path == p or p in path.parents for p in protected) or path.name in ('runtime.env', 'secure-bridge.env') or path.name == '.env' or path.name.endswith('.env') or path.suffix == '.key') def flush(): if not pending: return index = counts['batches'] p = subprocess.run(['getfacl', '-p', '-n', '-E', '--', *[str(x[0]) for x in pending]], check=True, capture_output=True, text=True, timeout=30) (journal / f'{index:05d}.before.acl').write_text(p.stdout) blocks = [x for x in p.stdout.strip().split('\n\n') if x.strip()] assert len(blocks) == len(pending) changed = ''.join(transform(block, uid, info.st_mode, info.st_uid) for block, (path, info) in zip(blocks, pending)) target = journal / f'{index:05d}.apply.acl' target.write_text(changed) subprocess.run(['setfacl', '--restore=' + str(target)], check=True, timeout=30) for path, info in pending: assert (path.stat().st_uid, path.stat().st_gid) == (info.st_uid, info.st_gid) counts['batches'] += 1 pending.clear() if counts['batches'] % 20 == 0: print(json.dumps(dict(phase='ACL_PROGRESS', **counts)), flush=True) for parent, dirs, names in os.walk(ROOT, topdown=True, followlinks=False): parent = Path(parent) info = parent.lstat() identity = (info.st_dev, info.st_ino) if identity in visited: dirs[:] = [] continue visited.add(identity) if excluded(parent): counts['protected'] += 1 dirs[:] = [] continue for name in list(dirs): path = parent / name if path.is_symlink() or excluded(path): dirs.remove(name) counts['symlinks' if path.is_symlink() else 'protected'] += 1 for path in [parent] + [parent / name for name in names]: info = path.lstat() if stat.S_ISLNK(info.st_mode): counts['symlinks'] += 1 continue if excluded(path): counts['protected'] += 1 continue if not stat.S_ISREG(info.st_mode) and not stat.S_ISDIR(info.st_mode): counts['special'] += 1 continue counts['directories' if stat.S_ISDIR(info.st_mode) else 'files'] += 1 pending.append((path, info)) if len(pending) >= 256: flush() flush() assert {str(p): stat.S_IMODE(p.stat().st_mode) for p in protected} == protected_modes result = dict(status='APPLIED', root=str(ROOT), account='si', counts=counts, protected_roots=[str(p) for p in sorted(protected)], protected_modes_unchanged=True, backup=str(journal), owner_changed=False, symlinks_followed=False) (journal / 'result.json').write_text(json.dumps(result, indent=2)) print(json.dumps(result), flush=True) if __name__ == '__main__': main()