"""Explicit read-only USB partition mounts; no disk formatting or system-disk writes.""" import json import os from pathlib import Path import re import stat import subprocess import field_profiles as profiles from field_ui import print, input MOUNT_ROOT = Path('/mnt/contextra-usb') FILESYSTEMS = frozenset(('vfat','exfat','ntfs','ntfs3','ext4','ext3','ext2','iso9660','udf')) PROTECTED = ('/','/boot','/boot/efi','/usr','/var','/home','/workspace','/backup','/training','/data','[SWAP]') def candidates(data): result=[] def all_rows(row): yield row for child in row.get('children',[]):yield from all_rows(child) for disk in data.get('blockdevices',[]): if disk.get('tran')!='usb' or disk.get('type')!='disk':continue members=list(all_rows(disk)) mounts=[m for row in members for m in (row.get('mountpoints') or []) if m] if any(m in PROTECTED or any(m.startswith(p+'/') for p in PROTECTED if p!='/') for m in mounts):continue if any(r.get('type') not in ('disk','part') or r.get('fstype')=='swap' for r in members):continue for row in members: if row.get('fstype') not in FILESYSTEMS:continue if row['type']=='disk' and row.get('children'):continue name=row.get('name','') if not re.fullmatch(r'/dev/[A-Za-z0-9_-]+',name):continue result.append({k:row.get(k) for k in ('name','maj:min','uuid','fstype','size','label','mountpoints')}) return result def inventory(): p=subprocess.run(['lsblk','-J','-p','-b','-o','NAME,TYPE,TRAN,FSTYPE,SIZE,LABEL,UUID,MAJ:MIN,MOUNTPOINTS'], check=True,capture_output=True,text=True,timeout=10) return candidates(json.loads(p.stdout)) def mount(selected): current=next((x for x in inventory() if x==selected),None) if current is None:raise ValueError('USB changed; select again') mounted=[m for m in selected.get('mountpoints') or [] if m] if mounted: return profiles.safe_path(mounted[0]) device=profiles.safe_path(selected['name']);info=device.stat() if not stat.S_ISBLK(info.st_mode) or f'{os.major(info.st_rdev)}:{os.minor(info.st_rdev)}'!=selected['maj:min']: raise ValueError('USB device identity mismatch') target=profiles.safe_path(MOUNT_ROOT/device.name) target.mkdir(parents=True,exist_ok=True,mode=0o755) if any(target.iterdir()) or os.path.ismount(target):raise ValueError('Mount destination is not empty') subprocess.run(['mount','-o','ro,nosuid,nodev,noexec',str(device),str(target)],check=True,timeout=20) doc=json.loads(subprocess.run(['findmnt','-J','-M',str(target),'-o','SOURCE,TARGET,OPTIONS'], check=True,capture_output=True,text=True,timeout=5).stdout) rows=doc.get('filesystems',[]) if len(rows)!=1 or not {'ro','nosuid','nodev','noexec'}<=set(rows[0]['options'].split(',')): raise RuntimeError('Mount protection verification failed; no CA read. Inspect this mount explicitly.') print('Read-only mount:',target,'No fstab change; USB original preserved.') return target def choose(): rows=inventory() if not rows: print('No eligible USB filesystem. OS/data/LVM/swap disks are excluded. No format operation.') return None for i,row in enumerate(rows,1): print(i,row['name'],row['fstype'],str(row['size'])+' bytes',row.get('label') or '',row.get('mountpoints') or []) raw=input('USB number (Enter=cancel): ').strip() if not raw:return None if not raw.isdecimal() or not 1<=int(raw)<=len(rows):raise ValueError('Invalid USB number') selected=rows[int(raw)-1] if not any(selected.get('mountpoints') or []): if input('Mount this USB read-only? Type MOUNT (Enter=cancel): ')!='MOUNT':return None return mount(selected) def unmountable(path): p=profiles.safe_path(path) allowed=(MOUNT_ROOT,Path('/media')/profiles.server.operator(),Path('/run/media')/profiles.server.operator()) return str(p)=='/mnt/ct-usb' or any(p!=root and p.is_relative_to(root) for root in allowed) def unmount(selected,target): target=profiles.safe_path(target) if not unmountable(target):raise ValueError('Protected mount path; only explicit USB locations are allowed') current=next((x for x in inventory() if x==selected),None) if current is None or str(target) not in (selected.get('mountpoints') or []): raise ValueError('USB/mount identity changed; no unmount') doc=json.loads(subprocess.run(['findmnt','-J','-R','-M',str(target),'-o','SOURCE,TARGET,MAJ:MIN'], check=True,capture_output=True,text=True,timeout=5).stdout) rows=doc.get('filesystems',[]) if (len(rows)!=1 or rows[0].get('target')!=str(target) or rows[0].get('children') or rows[0].get('maj:min')!=selected['maj:min']): raise ValueError('Mount has children or identity differs; no unmount') # No recursive, force, lazy unmount or disk-eject operation is permitted. subprocess.run(['umount','--',str(target)],check=True,timeout=20) check=subprocess.run(['findmnt','-n','-M',str(target)],capture_output=True,text=True,timeout=5) if check.returncode==0:raise RuntimeError('Mount remains; no force/lazy retry') if check.returncode!=1:raise RuntimeError('Unmount verification unavailable; inspect status') print('USB unmounted:',target,'Other disks and mountpoints unchanged.') def main(args): action=args[0] if args else 'status' if len(args)>1:raise ValueError('ct usb status | mount | unmount') if action=='mount':return choose() rows=inventory() if action=='status': for row in rows:print(row) print('ct usb mount | ct usb unmount. /backup, /training, system/data disks are protected.') return if action!='unmount':raise ValueError('ct usb status | mount | unmount') choices=[] for row in rows: for path in row.get('mountpoints') or []: if path and unmountable(path):choices.append((row,path)) for i,(row,path) in enumerate(choices,1):print(i,row['name'],path,row.get('label') or '') if not choices:print('No eligible mounted USB. Protected disks are not selectable.');return raw=input('USB unmount number [Enter=cancel]: ').strip() if not raw:return if not raw.isdecimal() or not 1<=int(raw)<=len(choices):raise ValueError('Invalid USB number') row,path=choices[int(raw)-1] print('Close files/terminals using this USB. Busy mounts remain mounted; no force/lazy option.') if input('Unmount '+path+' only? Type UNMOUNT [Enter=cancel]: ')!='UNMOUNT':return unmount(row,path)