#!/usr/bin/env python3 """Fail-closed validation of the exact three-service disposable Immich recipe.""" import json import os from pathlib import Path import re import stat import sys MIB = 1024 ** 2 ROLES = { 'db': (1024*MIB, 128, 1, {'CHOWN','DAC_OVERRIDE','FOWNER','SETUID','SETGID'}, {'/tmp':64*MIB, '/var/lib/postgresql/data':768*MIB, '/run/postgresql':16*MIB, '/etc/postgresql':8*MIB}), 'redis': (128*MIB, 128, 1, {'CHOWN','SETUID','SETGID'}, {'/tmp':64*MIB, '/data':64*MIB}), 'server': (2048*MIB, 256, 2, set(), {'/tmp':64*MIB, '/usr/src/app/upload':256*MIB}), } def require(value, message): if not value: raise ValueError(message) def validate(items, name, db_image, redis_image, server_image): require(re.fullmatch(r'archy-cloud-immich-[A-Za-z0-9]+', name), 'Invalid fixture name') images = dict(zip(ROLES, (db_image, redis_image, server_image))) require(all(re.fullmatch(r'(sha256:)?[a-f0-9]{64}', image) for image in images.values()), 'Immutable images required') require(isinstance(items, list) and len(items) == 3, 'Exactly three services required') by_name = {item.get('Name','').lstrip('/'):item for item in items} require(set(by_name) == {name+'-'+role for role in ROLES}, 'Unexpected services') db_id = by_name[name+'-db'].get('Id') require(isinstance(db_id,str) and re.fullmatch(r'[a-f0-9]{64}',db_id), 'Invalid namespace owner') pids = [] for role, (memory,pids_limit,cpus,caps,scratch) in ROLES.items(): item = by_name[name+'-'+role] host = item.get('HostConfig', {}) require(item.get('Image','').removeprefix('sha256:') == images[role].removeprefix('sha256:'), 'Image mismatch') require(item.get('Config',{}).get('Labels',{}).get('org.archipelago.cloud-fixture') == 'immich274-v1', 'Missing fixture label') require(host.get('NetworkMode') == ('none' if role == 'db' else 'container:'+db_id), 'Unexpected network namespace') require(not host.get('PortBindings') and not host.get('PublishAllPorts'), 'Host ports forbidden') require(host.get('Privileged') is False and host.get('ReadonlyRootfs') is True, 'Readonly unprivileged root required') require(host.get('PidMode','') in ('','private') and host.get('IpcMode') == 'private', 'Private PID/IPC required') require(type(host.get('Memory')) is int and 0 < host['Memory'] <= memory, 'Memory bound exceeded') require(host.get('MemorySwap') == host['Memory'], 'Swap forbidden') require(type(host.get('PidsLimit')) is int and 0 < host['PidsLimit'] <= pids_limit, 'PID bound exceeded') nano,quota,period = (host.get(key,0) for key in ('NanoCpus','CpuQuota','CpuPeriod')) require(all(type(value) is int and value >= 0 for value in (nano,quota,period)), 'Invalid CPU cap') require(nano <= cpus*10**9 and (not quota or 1000 <= period <= 1000000 and 1000 <= quota <= cpus*period) and (nano > 0 or quota > 0), 'CPU bound exceeded') require(host.get('SecurityOpt') in (['no-new-privileges'], ['no-new-privileges=true']), 'Security policy mismatch') for key in ('EffectiveCaps','BoundingCaps'): # Podman serializes a present empty capability set as null. # A missing inspect field remains an error, not an empty grant. actual = [] if key in item and item[key] is None else item.get(key) require(isinstance(actual,list) and len(actual) <= len(caps) and all(isinstance(cap,str) and cap.removeprefix('CAP_') in caps for cap in actual), 'Capability escalation') require(not host.get('Devices') and not host.get('VolumesFrom'), 'Devices/borrowed volumes forbidden') tmpfs = host.get('Tmpfs',{}) require(set(tmpfs) == set(scratch), 'Scratch destinations mismatch') for target, options in tmpfs.items(): require(isinstance(options,str), 'Invalid scratch options') sizes = [option[5:] for option in options.split(',') if option.startswith('size=')] require(len(sizes) == 1, 'One scratch bound required') match = re.fullmatch(r'([0-9]+)([kKmMgG]?)',sizes[0]) require(match is not None, 'Invalid scratch size') size = int(match[1])*{'':1,'k':1024,'m':MIB,'g':1024*MIB}[match[2].lower()] require(0 < size <= scratch[target], 'Scratch bound exceeded') allowed = {'/etc/hosts':'hosts','/etc/resolv.conf':'resolv.conf'} if role == 'server': allowed.update({'/contract.mjs':'immich-upstream-contract.mjs','/fixture.png':'immich-fixture.png'}) binds = [] for mount in item.get('Mounts',[]): target = mount.get('Destination') if mount.get('Type') == 'tmpfs': require(target in scratch, 'Unexpected scratch mount') continue require(mount.get('Type') == 'bind' and mount.get('RW') is False and target in allowed, 'Unreviewed mount') source = Path(mount.get('Source','')) require(source.name == allowed[target], 'Unexpected fixture input') metadata = source.lstat() require(stat.S_ISREG(metadata.st_mode) and metadata.st_nlink == 1, 'Input must be a regular unaliased file') if target == '/etc/hosts': require(source.read_bytes() == b'127.0.0.1 localhost\n::1 localhost\n', 'Host inventory forbidden') if target == '/etc/resolv.conf': require(metadata.st_size == 0, 'External resolver forbidden') binds.append(target) require(len(binds) == len(allowed) and set(binds) == set(allowed), 'Missing/duplicate fixture inputs') state = item.get('State',{}) require(state.get('Running') is True and type(state.get('Pid')) is int and state['Pid'] > 1, 'Running services required') pids.append(state['Pid']) return pids if __name__ == '__main__': try: require(len(sys.argv) == 5, 'Expected fixture prefix and exact db/redis/server images') pids = validate(json.load(sys.stdin), *sys.argv[1:]) # The trusted launcher must check these kernel namespace identities and # recheck its execution container before running any test code. print(' '.join(map(str, pids))) except (ValueError, OSError, TypeError, KeyError, AttributeError) as error: print(f'Immich fixture refused: {error}', file=sys.stderr) sys.exit(2)