146 lines
10 KiB
Python
146 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Probe private Podman confinement without images, shared storage or policy loads.
|
|
|
|
Requires root, Podman, a C compiler, dynamic BusyBox, an installed default seccomp
|
|
profile and one already loaded enforced AppArmor profile. Copies only the local
|
|
BusyBox/runtime libraries into a disposable read-only rootfs. No app data, host
|
|
bind mounts, network, image pulls, running services or host profiles are changed.
|
|
Generic AppArmor attachment does not qualify application policies or Docker.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
parser=argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--receipt',type=Path,required=True)
|
|
parser.add_argument('--apparmor-profile',required=True,help='already loaded profile; never loaded or replaced by this probe')
|
|
parser.add_argument('--policy-source',type=Path,help='load exact selected application policy only into a new disposable AppArmor namespace')
|
|
args=parser.parse_args()
|
|
REPORT=args.receipt
|
|
PROFILE=args.apparmor_profile
|
|
if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.+-]{0,127}',PROFILE):
|
|
parser.error('Expected one plain existing AppArmor profile name')
|
|
if not REPORT.parent.is_dir():
|
|
parser.error('Receipt parent directory must already exist')
|
|
if os.geteuid() != 0:
|
|
parser.error('Run with sudo -n; only the private temporary root/store is mutated')
|
|
if REPORT.exists():
|
|
parser.error('Refusing to overwrite previous evidence')
|
|
report={'started':time.time(),'mode':'private-rootfs; no shared image store, pulls, host mounts or policy loads','checks':{},'commands':[],'apparmor_positive_qualified':False,'docker_qualified':False}
|
|
def capture(argv,env=None,timeout=45):
|
|
started=time.time()
|
|
try:
|
|
r=subprocess.run(argv,env=env,capture_output=True,text=True,timeout=timeout)
|
|
result={'returncode':r.returncode,'stdout':r.stdout[-12000:],'stderr':r.stderr[-12000:]}
|
|
except subprocess.TimeoutExpired as e:
|
|
def txt(v): return v.decode(errors='replace') if isinstance(v,bytes) else (v or '')
|
|
result={'returncode':None,'timeout':timeout,'stdout':txt(e.stdout)[-12000:],'stderr':txt(e.stderr)[-12000:]}
|
|
result['elapsed']=time.time()-started; report['commands'].append({'argv':argv,**result}); return result
|
|
cleanup_failed=False
|
|
policy_namespace=None
|
|
original_profile=PROFILE
|
|
try:
|
|
work=Path(tempfile.mkdtemp(prefix='archy-private-confinement-'))
|
|
try:
|
|
root=work/'rootfs'; root.mkdir()
|
|
for d in ['bin','proc','sys','dev','tmp','etc','run']: (root/d).mkdir()
|
|
busybox=Path(shutil.which('busybox')); deps=subprocess.run(['ldd',str(busybox)],check=True,capture_output=True,text=True).stdout
|
|
paths=[busybox]+[Path(p) for p in re.findall(r'(/[^\s()]+)',deps)]
|
|
report['rootfs_inputs']={}
|
|
for source in paths:
|
|
target=root/'bin/busybox' if source==busybox else root/str(source).lstrip('/')
|
|
target.parent.mkdir(parents=True,exist_ok=True); shutil.copy2(source,target)
|
|
report['rootfs_inputs'][str(source)]={'sha256':hashlib.sha256(source.read_bytes()).hexdigest(),'target':str(target.relative_to(root))}
|
|
for command in ['sh','uname','cat','true']: (root/'bin'/command).symlink_to('busybox')
|
|
source=work/'uname_probe.c'
|
|
source.write_text('#include <errno.h>\n#include <stdio.h>\n#include <sys/syscall.h>\n#include <sys/utsname.h>\n#include <unistd.h>\nint main(void){struct utsname u;if(syscall(SYS_uname,&u)==0){puts(u.sysname);return 0;}int e=errno;fprintf(stderr,"errno=%d\\n",e);return 1;}\n')
|
|
built=capture(['cc','-O0','-o',str(root/'bin/uname-probe'),str(source)],timeout=30)
|
|
if built['returncode']!=0: raise RuntimeError('direct syscall fixture compile failed')
|
|
report['syscall_fixture_source']=source.read_text()
|
|
report['rootfs_inputs']['direct-syscall-fixture']={'sha256':hashlib.sha256((root/'bin/uname-probe').read_bytes()).hexdigest(),'target':'bin/uname-probe'}
|
|
if args.policy_source:
|
|
policy=args.policy_source.resolve()
|
|
report['policy_source_sha256']=hashlib.sha256(policy.read_bytes()).hexdigest()
|
|
namespace='archy-probe-'+uuid.uuid4().hex
|
|
policy_namespace=Path('/sys/kernel/security/apparmor/policy/namespaces')/namespace
|
|
policy_namespace.mkdir()
|
|
loaded=capture(['/usr/sbin/apparmor_parser','--add','--skip-cache','-n',namespace,str(policy)])
|
|
if loaded['returncode']!=0: raise RuntimeError('isolated application policy load failed')
|
|
PROFILE=original_profile
|
|
report['policy_namespace']=namespace
|
|
for directory in ('etc/archipelago','data'):
|
|
(root/directory).mkdir(parents=True,exist_ok=True)
|
|
(root/'etc/archipelago/probe-secret').write_text('synthetic secret')
|
|
(root/'data/probe').write_text('application data')
|
|
profiles=Path('/sys/kernel/security/apparmor/profiles').read_text().splitlines()
|
|
report['apparmor_selected_profile_enforced']=((':'+namespace+':'+PROFILE if args.policy_source else PROFILE)+' (enforce)') in profiles
|
|
config=work/'containers.conf'; config.write_text('[engine]\ncgroup_manager="cgroupfs"\nevents_logger="file"\nlock_type="file"\nhooks_dir=[]\n')
|
|
deny=work/'deny-uname.json'; deny.write_text(json.dumps({'defaultAction':'SCMP_ACT_ALLOW','syscalls':[{'names':['uname'],'action':'SCMP_ACT_ERRNO','errnoRet':1}]}))
|
|
env=os.environ.copy()
|
|
for key in ['CONTAINER_HOST','CONTAINER_CONNECTION','DOCKER_HOST','CONTAINERS_CONF','CONTAINERS_CONF_OVERRIDE','CONTAINERS_STORAGE_CONF','STORAGE_DRIVER']: env.pop(key,None)
|
|
env.update(CONTAINERS_CONF=str(config),LC_ALL='C')
|
|
base=['podman','--remote=false','--root',str(work/'store'),'--runroot',str(work/'runroot'),'--tmpdir',str(work/'tmp'),'--storage-driver=vfs','--cgroup-manager=cgroupfs','--events-backend=file']
|
|
if args.policy_source:
|
|
base=['aa-exec','-p',':'+namespace+'://unconfined','--',*base]
|
|
def run(options,command):
|
|
global cleanup_failed
|
|
name='archy-confinement-'+uuid.uuid4().hex
|
|
try:
|
|
return capture(base+['run','--rm','--name',name,'--network=none','--cap-drop=ALL','--read-only','--pids-limit=64','--memory=128m','--security-opt=no-new-privileges=true',*options,'--rootfs',str(root),*command],env)
|
|
finally:
|
|
cleaned=capture(base+['rm','--force','--ignore',name],env,30)
|
|
if cleaned['returncode']!=0:
|
|
cleanup_failed=True
|
|
raise RuntimeError('private container cleanup failed; preserve private store and command evidence')
|
|
baseline=run(['--security-opt=seccomp=/usr/share/containers/seccomp.json'],['/bin/uname-probe'])
|
|
report['checks']['baseline_command_succeeds']=baseline['returncode']==0 and baseline['stdout'].strip()=='Linux'
|
|
if not report['checks']['baseline_command_succeeds']: raise RuntimeError('private runtime baseline failed; no confinement claim')
|
|
denied=run(['--security-opt=seccomp='+str(deny)],['/bin/uname-probe'])
|
|
report['checks']['seccomp_denies_uname']=denied['returncode'] not in (None,0) and denied['stderr'].strip()=='errno=1'
|
|
state=run(['--security-opt=seccomp='+str(deny)],['/bin/cat','/proc/self/status'])
|
|
report['checks']['kernel_no_new_privileges_and_seccomp']=state['returncode']==0 and 'NoNewPrivs:\t1' in state['stdout'] and 'Seccomp:\t2' in state['stdout']
|
|
label=run(['--security-opt=apparmor='+PROFILE],['/bin/cat','/proc/self/attr/current'])
|
|
report['checks']['existing_apparmor_profile_attached']=label['returncode']==0 and PROFILE in label['stdout'].split(' (')[0].split('//&') and report['apparmor_selected_profile_enforced']
|
|
report['apparmor_kernel_label']=label['stdout'].strip()
|
|
report['apparmor_scope']='Existing generic enforced profile attachment only; no Archipelago app policy or negative-access qualification'
|
|
if args.policy_source:
|
|
positive=run(['--security-opt=apparmor='+PROFILE],['/bin/cat','/data/probe'])
|
|
control=run([],['/bin/cat','/etc/archipelago/probe-secret'])
|
|
negative=run(['--security-opt=apparmor='+PROFILE],['/bin/cat','/etc/archipelago/probe-secret'])
|
|
report['checks']['application_data_read']=positive['returncode']==0 and positive['stdout']=='application data'
|
|
report['checks']['secret_read_control']=control['returncode']==0 and control['stdout']=='synthetic secret'
|
|
report['checks']['application_profile_denies_control_secret']=negative['returncode'] not in (0,None) and 'Permission denied' in negative['stderr']
|
|
report['apparmor_scope']='Exact selected application baseline in disposable namespace: positive data and denied control-secret access; not full application-image compatibility'
|
|
report['apparmor_positive_qualified']=all(report['checks'].values())
|
|
missing=run(['--security-opt=apparmor=archipelago-probe-not-installed'],['/bin/true'])
|
|
report['checks']['missing_apparmor_refuses']=missing['returncode'] not in (None,0) and 'apparmor' in missing['stderr'].lower()
|
|
report['checks']['probe_completed']=True
|
|
finally:
|
|
if cleanup_failed:
|
|
report['preserved_private_workdir']=str(work)
|
|
else:
|
|
shutil.rmtree(work)
|
|
report['private_workdir_removed']=not work.exists()
|
|
except Exception as e:
|
|
report['checks']['probe_completed']=False; report['error']=repr(e)
|
|
if policy_namespace is not None:
|
|
try:
|
|
policy_namespace.rmdir()
|
|
report['policy_namespace_removed']=True
|
|
except OSError as error:
|
|
report['checks']['policy_namespace_removed']=False
|
|
report['cleanup_error']=str(error)
|
|
report['finished']=time.time(); report['passed']=all(report['checks'].values())
|
|
with REPORT.open('x') as output:
|
|
json.dump(report,output,indent=2); output.write('\n')
|
|
os.chown(REPORT,int(os.environ.get('SUDO_UID','0')),int(os.environ.get('SUDO_GID','0')))
|
|
print(json.dumps({k:v for k,v in report.items() if k!='commands'},indent=2))
|
|
raise SystemExit(0 if report['passed'] else 1)
|