#!/usr/bin/env python3 """Run the public flasher against disposable NVMe and USB disks in an ARM VM.""" import io import json from pathlib import Path import shlex import subprocess import sys import tarfile import tempfile import time project = Path(__file__).resolve().parents[2] sys.path.insert(0, str(project/'tools')) from image_formats import digest, gpt, LINUX_FILESYSTEM from vm_test import VM work = Path(tempfile.mkdtemp(prefix='workstation-flash-vm.', dir=project/'out')) fixtures = Path((project/'out/workstation-flash-current.txt').read_text().strip()) binary = project/'target/aarch64-unknown-linux-musl/release/fds-flash' archive = project/'out/rootfs-cli.tar' for path in (binary, archive, fixtures/'internal.img', fixtures/'system.img'): assert path.is_file(), path inputs = [binary, archive, project/'out/kernel/boot/kernel_2712.img', project/'out/fds-initramfs.img'] (work/'inputs.sha256').write_text(''.join(f'{digest(p)} {p}\n' for p in inputs)) # This is an isolated fixture with a root serial console and the new workstation # binary. It does not alter production login, rootfs archives or attached disks. additions = { 'usr/libexec/fds/console-session': (b"#!/bin/bash\nexec env HOME=/root PS1='FLASH# ' bash --noprofile --norc\n", 0o755), 'usr/bin/fds-flash': (binary.read_bytes(), 0o755), 'usr/share/fds/flash-internal.img': ((fixtures/'internal.img').read_bytes(), 0o644), 'usr/share/fds/flash-system.img': ((fixtures/'system.img').read_bytes(), 0o644), } with tarfile.open(archive) as source, tarfile.open(work/'fixture.tar','w',format=tarfile.PAX_FORMAT) as target: for member in source: if member.name not in additions: target.addfile(member, source.extractfile(member) if member.isfile() else None) for name,(data,mode) in additions.items(): member=tarfile.TarInfo(name);member.size=len(data);member.mode=mode target.addfile(member,io.BytesIO(data)) (work/'system').mkdir() with (work/'build.log').open('w') as log: subprocess.run([str(project/'image/build-system-cartridge'),'--rootfs',str(work/'fixture.tar'), '--output-directory',str(work/'system')],check=True,stdout=log,stderr=subprocess.STDOUT) filesystem=work/'mountable.ext4' with filesystem.open('xb') as stream:stream.truncate(32*1024*1024) subprocess.run([str(project/'tools/in-image-tools'),'mke2fs','-q','-F','-t','ext4',str(filesystem)],check=True) disk=work/'nvme.img' gpt(disk,[('FDS_DATA',LINUX_FILESYSTEM,filesystem)]) with (work/'4k.img').open('xb') as stream:stream.truncate(32*1024*1024) extra=['-device','qemu-xhci,id=xhci,addr=05.0', '-drive',f'file={disk},if=none,id=nvme,format=raw', '-device','nvme,drive=nvme,serial=FDS-FLASH', '-drive',f'file={work/"4k.img"},if=none,id=fourk,format=raw', '-device','nvme,drive=fourk,serial=FDS-4K,logical_block_size=4096,physical_block_size=4096'] def wait(read, predicate, timeout=60): deadline=time.monotonic()+timeout while True: result=read() if predicate(result):return result assert time.monotonic()/tmp/flash-error',ok=ok) except AssertionError as error: raise AssertionError((arguments, vm.capture('cat /tmp/flash-error'))) from error def preview(vm,image,path,ok=True): text=cmd(vm,['--image',image,'--device',path,'--dry-run','--json'],ok) return json.loads(text)['plan'] if ok else text def write(vm,image,path,plan,ok=True): return cmd(vm,['--image',image,'--device',path,'--unattended', '--expect-target',plan['target_id'],'--sha256',plan['sha256'],'--json'],ok) record = {} with VM(work,'flash',work/'system/system.img',extra=extra) as vm: vm.expect(rb'FLASH# ') vm.capture('s6-rc -b -l /run/s6-rc -d change cartridged') rows=json.loads(cmd(vm,['list','--json'])) nvme=next(row for row in rows if row['target'].get('serial')=='FDS-FLASH')['target']['path'] fourk=next(row for row in rows if row['target'].get('serial')=='FDS-4K')['target']['path'] source='/usr/share/fds/flash-internal.img' before=digest(disk) preview(vm,source,'/dev/vda',False) assert 'read-only' in vm.capture('cat /tmp/flash-error') preview(vm,source,fourk,False) assert '512-byte' in vm.capture('cat /tmp/flash-error') preview(vm,source,nvme+'p1',False) assert 'whole disk' in vm.capture('cat /tmp/flash-error') vm.capture('mkdir /tmp/flash-mounted; mount -o ro,noload '+shlex.quote(nvme+'p1')+' /tmp/flash-mounted') preview(vm,source,nvme,False) assert 'mounted' in vm.capture('cat /tmp/flash-error') vm.capture('umount /tmp/flash-mounted') assert digest(disk)==before plan=preview(vm,source,nvme) assert json.loads(write(vm,source,nvme,plan))['status']=='verified' # Kernel partition names are independent of the asynchronous udev/blkid # cache (the deliberately inert payloads are not mountable filesystems). def labels(): return vm.capture('cat /sys/class/block/'+Path(nvme).name+"p*/uevent | sed -n 's/^PARTNAME=//p'").split() labels=wait(labels,lambda names:names==['FDS_BOOT','FDS_RECOVERY','FDS_INTERNAL']) assert labels==['FDS_BOOT','FDS_RECOVERY','FDS_INTERNAL'],labels record['nvme_internal_write_and_kernel_partition_reread']=True # Reinstalling the same offline FDS disk is an intended operation. source='/usr/share/fds/flash-system.img' plan=preview(vm,source,nvme) assert json.loads(write(vm,source,nvme,plan))['status']=='verified' assert wait(lambda:vm.capture('cat /sys/class/block/'+Path(nvme).name+"p*/uevent | sed -n 's/^PARTNAME=//p'").split(), lambda names:names==['FDS_SYSTEM'])==['FDS_SYSTEM'] record['reflash_internal_disk_with_system']=True record['mounted_partition_readonly_and_4k_refused']=True # Hotplug only generated files into this VM. No host block paths are used. def attach(name, rule=None): path=work/(name+'.img') with path.open('xb') as stream:stream.truncate(32*1024*1024) vm.qmp('blockdev-add',{'driver':'raw','node-name':name,'file':{'driver':'file','filename':str(path)}}) top=name if rule: top=name+'-fault' vm.qmp('blockdev-add',{'driver':'blkdebug','node-name':top,'image':name, 'inject-error':[{'event':'none','errno':5,'once':False,**rule}]}) vm.qmp('device_add',{'driver':'usb-storage','id':name,'drive':top,'bus':'xhci.0','port':'2','serial':name}) def rows():return json.loads(cmd(vm,['list','--json'])) entries=wait(rows,lambda entries:any(row['target'].get('serial')==name for row in entries)) return path,next(row['target']['path'] for row in entries if row['target'].get('serial')==name) def detach(name): vm.qmp('device_del',{'id':name}) wait(lambda:json.loads(cmd(vm,['list','--json'])),lambda entries:all(row['target'].get('serial')!=name for row in entries)) usb,path=attach('usb-first') plan=preview(vm,source,path) detach('usb-first') usb2,path2=attach('usb-replacement') write(vm,source,path2,plan,False) assert 'Target identity' in vm.capture('cat /tmp/flash-error') fresh=preview(vm,source,path2) assert json.loads(write(vm,source,path2,fresh))['status']=='verified' record['usb_replacement_refused_and_fresh_write_verified']=True detach('usb-replacement') for name,rule in [('write-error',{'iotype':'write','sector':2056}), ('flush-error',{'iotype':'flush'}), ('readback-error',{'iotype':'read','sector':2056})]: target,path=attach(name,rule) plan=preview(vm,source,path) output=write(vm,source,path,plan,False) error=vm.capture('cat /tmp/flash-error') assert 'os error 5' in error,(name,error) assert 'verified' not in output record[name]=True detach(name) # The complete wizard must select the intended disk without CLI paths. rows=json.loads(cmd(vm,['list','--json'])) selection=next(n+1 for n,row in enumerate(rows) if row['target']['path']==nvme) vm.send('fds-flash; printf "\\nFLASH_WIZARD_STATUS:%s\\n" "$?"') vm.expect(rb'Image path: ');vm.send(source) vm.expect(rb'Disk number \(no default\): ');vm.send(str(selection)) vm.expect(rb' to proceed: ');vm.send('ERASE '+nvme) vm.expect(rb'VERIFIED: image written') vm.expect(rb'FLASH_WIZARD_STATUS:0\r?\n') record['interactive_image_disk_selection_and_confirmation']=True vm.capture('s6-rc -b -l /run/s6-rc -u change cartridged') vm.send('fds poweroff');vm.expect(rb'reboot: Power down');assert vm.child.wait(timeout=20)==0 # The final successful NVMe write has a valid GPT on the actual backing file. subprocess.run(['sfdisk','--verify',str(disk)],check=True) record.update(status='passed',physical_hardware='not tested',source_binary_sha256=digest(binary)) (work/'acceptance.json').write_text(json.dumps(record,indent=2)+'\n') (project/'out/workstation-flash-vm-current.txt').write_text(str(work)+'\n') print('PASS: real virtual NVMe/USB flashing, protected disks, replacement and I/O failures:',work) print('SKIP: physical Pi, drive/controller flush behavior and power-loss recovery')