136 lines
8 KiB
Python
Executable file
136 lines
8 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Strict synthetic command boundary; only the private /tmp fixture is writable."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import sys
|
|
|
|
name = Path(sys.argv[0]).name
|
|
args = sys.argv[1:]
|
|
root = Path('/tmp/fixture')
|
|
state_path = root / 'state.json'
|
|
state = json.loads(state_path.read_text())
|
|
fail = os.environ.get('CASE_FAIL', '')
|
|
nvme = os.environ.get('CASE_STYLE', '') == 'nvme'
|
|
disk_path = '/dev/nvme9n9' if nvme else '/dev/target'
|
|
efi_path = disk_path + ('p1' if nvme else '1')
|
|
root_path = disk_path + ('p2' if nvme else '2')
|
|
with (root / 'calls').open('a') as stream:
|
|
stream.write(name + ' ' + ' '.join(args) + '\n')
|
|
|
|
def save():
|
|
state_path.write_text(json.dumps(state))
|
|
|
|
def die(message):
|
|
print(message, file=sys.stderr)
|
|
sys.exit(47)
|
|
|
|
if fail == name or fail == name + ':inspection' and name in ('lsblk', 'findmnt', 'swapon'):
|
|
die('injected failure')
|
|
if name == 'readlink':
|
|
if args[:2] != ['-e','--'] or len(args) != 3 or args[2] not in (disk_path,'/dev/alias',efi_path):
|
|
die('unexpected readlink')
|
|
print(efi_path if args[2] == efi_path else disk_path)
|
|
elif name == 'lsblk':
|
|
if args[:5] != ['--json','--tree','--bytes','--output','PATH,TYPE,MAJ:MIN,SIZE,MODEL,SERIAL,WWN,MOUNTPOINTS,PKNAME,PARTN'] or len(args) != 6:
|
|
die('unexpected lsblk args')
|
|
target = args[5]
|
|
state['inspections'] = state.get('inspections', 0) + (target == disk_path)
|
|
save()
|
|
disk = {'path':disk_path,'type':'disk','maj:min':'8:16','size':100*1024**3,'model':'SYNTHETIC','serial':'TEST-NOT-REAL','wwn':None,'mountpoints':[None]}
|
|
if fail == 'identity_change' and state['inspections'] >= 2:
|
|
disk['maj:min'] = '8:20'
|
|
if state['partitioned']:
|
|
disk['children'] = [
|
|
{'path':efi_path,'type':'part','maj:min':'8:17','size':1024**3,'partn':1,'pkname':Path(disk_path).name,'mountpoints':[None]},
|
|
{'path':root_path,'type':'part','maj:min':'8:18','size':99*1024**3,'partn':2,'pkname':Path(disk_path).name,'mountpoints':[None]},
|
|
]
|
|
if fail == 'bad_partitions': disk['children'][1]['partn'] = 3
|
|
if fail == 'mounted_child': disk['children'] = [{'path':'/dev/target1','type':'part','maj:min':'8:17','size':1024**3,'mountpoints':['/work']}]
|
|
if fail == 'wrong_type': disk['type'] = 'rom'
|
|
if target == efi_path: disk = {'path':efi_path,'type':'part','maj:min':'8:17','size':1024**3,'mountpoints':[None]}
|
|
if target == '/dev/live': disk = {'path':'/dev/live','type':'part','maj:min':'8:99','size':8*1024**3,'pkname':'live','mountpoints':['/run/archiso/bootmnt']}
|
|
if target not in (disk_path,efi_path,'/dev/live'): die('unexpected lsblk target')
|
|
print(json.dumps({'blockdevices':[disk]}))
|
|
elif name == 'findmnt':
|
|
if args != ['--json','--output','TARGET,SOURCE,MAJ:MIN,FSTYPE,OPTIONS,ID']:
|
|
die('unexpected findmnt')
|
|
entries = [{'target':'/','source':'/dev/live','maj:min':'8:99','fstype':'ext4','options':'rw'}, {'target':'/run/archiso','source':'/dev/live','maj:min':'8:99','fstype':'ext4','options':'rw'}]
|
|
if fail in ('overlay_live','overlay_unmapped'):
|
|
entries[0].update({'source':'overlay','maj:min':'0:99','fstype':'overlay',
|
|
'options':'rw,lowerdir=/run/archiso/bootmnt/airootfs,upperdir=/run/archiso/cowspace/upper,workdir=/run/archiso/cowspace/work'})
|
|
if fail == 'overlay_live':
|
|
entries.append({'target':'/run/archiso/bootmnt','source':'/dev/live','maj:min':'8:99','fstype':'iso9660','options':'ro'})
|
|
entries.append({'target':'/run/archiso/cowspace','source':'tmpfs','maj:min':'0:98','fstype':'tmpfs','options':'rw'})
|
|
if fail == 'source_target': entries[0]['maj:min'] = '8:16'
|
|
if fail == 'mounted_disk': entries.append({'target':'/work','source':'/dev/target1','maj:min':'8:17','fstype':'ext4'})
|
|
if fail == 'mnt_occupied': entries.append({'target':'/mnt/other','source':'/dev/live','maj:min':'8:99','fstype':'ext4'})
|
|
if fail == 'swapfile': entries.append({'target':'/work','source':'/dev/target2','maj:min':'8:18','fstype':'ext4'})
|
|
entries += state['mounts']
|
|
print(json.dumps({'filesystems':entries}))
|
|
elif name == 'swapon':
|
|
if args != ['--show=NAME','--noheadings','--raw']: die('unexpected swapon')
|
|
if fail == 'swap_partition': print(efi_path)
|
|
elif fail == 'swapfile': print('/work/swapfile')
|
|
elif name in ('wipefs','parted','partprobe','udevadm','mkfs.fat','mkfs.btrfs','btrfs','mount','umount'):
|
|
if fail == name or fail == name + ':mutate': die('injected mutation failure')
|
|
if name == 'wipefs' and args != ['-af','--',disk_path]: die('unexpected wipefs')
|
|
if name == 'parted':
|
|
allowed = (
|
|
['-s',disk_path,'mklabel','gpt'],
|
|
['-s',disk_path,'mkpart','ESP','fat32','1MiB','1025MiB'],
|
|
['-s',disk_path,'set','1','esp','on'],
|
|
['-s',disk_path,'mkpart','ARCH','btrfs','1025MiB','100%'],
|
|
)
|
|
if args not in allowed: die('unexpected parted')
|
|
if args == ['-s',disk_path,'mkpart','ARCH','btrfs','1025MiB','100%']:
|
|
state['partitioned'] = True; save()
|
|
if name == 'partprobe' and args != [disk_path]: die('unexpected partprobe')
|
|
if name == 'partprobe' and fail in ('signal_INT','signal_TERM'):
|
|
os.kill(os.getppid(), signal.SIGINT if fail == 'signal_INT' else signal.SIGTERM)
|
|
sys.exit(130 if fail == 'signal_INT' else 143)
|
|
if name == 'udevadm' and args != ['settle']: die('unexpected udevadm')
|
|
if name == 'mkfs.fat' and args != ['-F','32','-n','EFI',efi_path]: die('unexpected mkfs.fat')
|
|
if name == 'mkfs.btrfs' and args != ['-f','-L','ARCH',root_path]: die('unexpected mkfs.btrfs')
|
|
if name == 'btrfs':
|
|
if fail == 'snapshot_conflict' and args[-1] == '/mnt/@/.snapshots': Path(args[-1]).mkdir(exist_ok=True)
|
|
allowed = {'/mnt/@','/mnt/@home','/mnt/@log','/mnt/@pkg','/mnt/@/.snapshots'}
|
|
if args[:2] != ['subvolume','create'] or len(args) != 3 or args[2] not in allowed or Path(args[2]).exists():
|
|
die('subvolume destination already exists or invalid')
|
|
Path(args[2]).mkdir()
|
|
if name == 'mount':
|
|
if fail in ('mount:third','mount_cleanup_fail','mount_replaced') and len(state['mounts']) == 1 and state.get('mount_count', 0) >= 2:
|
|
if fail == 'mount_replaced':
|
|
state['mounts'][0]['id'] += 500
|
|
save()
|
|
die('injected mount failure')
|
|
if args[0] == '-o':
|
|
expected_options = {'rw,relatime,ssd,discard=async,space_cache=v2,subvol='+v for v in ('@','@home','@log','@pkg','@/.snapshots')}
|
|
if len(args) != 4 or args[1] not in expected_options:
|
|
die('unexpected mount options')
|
|
source, target = args[2:]
|
|
options = args[1]
|
|
else:
|
|
if len(args) != 2: die('unexpected mount')
|
|
source, target = args
|
|
options = 'rw'
|
|
allowed_mounts = {('/mnt',root_path),('/mnt/home',root_path),('/mnt/var/log',root_path),
|
|
('/mnt/var/cache/pacman/pkg',root_path),('/mnt/.snapshots',root_path),
|
|
('/mnt/boot',efi_path)}
|
|
if (target,source) not in allowed_mounts:
|
|
die('unexpected mount source/target')
|
|
if fail == 'canonical_subvol':
|
|
options = options.replace('subvol=', 'subvol=/')
|
|
state['mount_count'] = state.get('mount_count', 0) + 1
|
|
state['mounts'].append({'target':target,'source':source,'maj:min':'8:17' if source.endswith('1') else '8:18','fstype':'btrfs' if source.endswith('2') else 'vfat','options':options,'id':100+state['mount_count']})
|
|
save()
|
|
if name == 'umount':
|
|
allowed_targets = {'/mnt','/mnt/home','/mnt/var/log','/mnt/var/cache/pacman/pkg','/mnt/.snapshots','/mnt/boot'}
|
|
if len(args) != 2 or args[0] != '--' or args[1] not in allowed_targets: die('unexpected umount')
|
|
if fail == 'mount_cleanup_fail' and args[1] == '/mnt' and state.get('mount_count', 0) >= 2:
|
|
die('injected cleanup failure')
|
|
state['mounts'] = [m for m in state['mounts'] if m['target'] != args[1]]
|
|
save()
|
|
else:
|
|
die('unknown command')
|