只读根在其范围内限制文件系统修改

from pathlib import Path
from tempfile import mkdtemp
import errno,os,socket,subprocess,sys,time,uuid

def cleanup_steps(steps, primary):
    failures=[]
    for label, action in steps:
        try:action()
        except BaseException as error:
            failures.append(label+': '+type(error).__name__+': '+str(error))
    if failures:
        message='cleanup incomplete or uncertain: '+'; '.join(failures)
        try:print(message,file=sys.stderr)
        except (OSError,ValueError):pass
        if primary is None:raise RuntimeError(message)
    return failures

def remove_file(path):
    try:path.unlink()
    except FileNotFoundError:pass

def require_empty(value):
    if value:raise RuntimeError('residual resources: '+str(value))

assert os.geteuid()==0,'run inside a disposable read-only-root container as root'
assert os.statvfs('/').f_flag&os.ST_RDONLY
name='xbcatm-readonly-'+uuid.uuid4().hex
print('fixture root-tree probe:',name,flush=True)
for base in [Path('/'),Path('/usr'),Path('/etc')]:
    path=base/name
    fd=None
    try:
        try:fd=os.open(path,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600)
        except OSError as error:assert error.errno==errno.EROFS,error
        else:raise AssertionError('root filesystem accepted a new file')
    finally:
        primary=sys.exc_info()[1]
        if fd is not None:
            cleanup_steps([('probe descriptor',lambda:os.close(fd)),('unexpected root-tree file '+str(path),lambda:remove_file(path))],primary)
print('even this root process receives EROFS on paths in the read-only root, usr and etc tree')
for base in ['/tmp','/work']:
    assert not os.statvfs(base).f_flag&os.ST_RDONLY
    directory=None
    try:
        directory=Path(mkdtemp(prefix='xbcatm-write-',dir=base))
        path=directory/'data';path.write_bytes(b'allowed');assert path.read_bytes()==b'allowed'
        path.write_bytes(b'changed');assert path.read_bytes()==b'changed'
    finally:
        primary=sys.exc_info()[1]
        if directory is not None:
            cleanup_steps([('temporary data',lambda:remove_file(directory/'data')),('temporary directory',directory.rmdir)],primary)
print('explicit tmpfs and data-volume mounts remain writable, including overwriting their contents')
assert Path('/etc/os-release').read_text()
result=subprocess.run([sys.executable,'-c','print("executed")'],capture_output=True,text=True,check=True,timeout=5)
assert result.stdout.strip()=='executed'
def remaining(deadline):
    seconds=deadline-time.monotonic()
    if seconds<=0:raise TimeoutError('loopback exchange deadline expired')
    return seconds

def receive_exact(connection,count,deadline):
    chunks=[];received=0
    while received<count:
        connection.settimeout(remaining(deadline))
        chunk=connection.recv(count-received)
        if not chunk:raise EOFError('loopback peer closed before the complete payload')
        chunks.append(chunk);received+=len(chunk)
    return b''.join(chunks)

listener=client=connection=None
try:
    deadline=time.monotonic()+5
    listener=socket.socket();listener.settimeout(remaining(deadline))
    listener.bind(('127.0.0.1',0));listener.listen(1)
    client=socket.socket();client.settimeout(remaining(deadline));client.connect(listener.getsockname())
    listener.settimeout(remaining(deadline));connection,_=listener.accept()
    client.settimeout(remaining(deadline));client.sendall(b'local')
    assert receive_exact(connection,5,deadline)==b'local'
finally:
    primary=sys.exc_info()[1]
    cleanup_steps([(label,sock.close) for label,sock in [('connection',connection),('client',client),('listener',listener)] if sock is not None],primary)
print('reading packaged files, executing an existing program, and loopback network I/O still work')