71 lines
4.1 KiB
Python
71 lines
4.1 KiB
Python
#!/usr/bin/python3
|
|
"""Actual immutable OCI through production user-manager jobs in disposable C01.
|
|
|
|
This is private execution qualification, not a fabricated reviewed source commit.
|
|
The production deployment entry still rejects unreviewed/fixture bundles.
|
|
"""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import tomllib
|
|
|
|
|
|
def qualify(c):
|
|
assert os.geteuid()==0 and Path('/run/.containerenv').exists()
|
|
assert os.uname().nodename=='c01-prerequisite-qualification'
|
|
package=c['PACKAGE'];sys.path.insert(0,str(package/'scripts'))
|
|
import uat_oci,c01_oci_job,c01_runtime
|
|
from c01_runtime_view import RuntimeLease
|
|
spec=importlib.util.spec_from_file_location('oci_importer',package/'scripts/import-uat-oci.py')
|
|
importer=importlib.util.module_from_spec(spec);spec.loader.exec_module(importer)
|
|
proof=uat_oci.check(package,package/'lnd-tor.oci.tar',package/'lnd-tor-identity.json',package/'lnd-tor-sources.json')
|
|
config,digest=c01_runtime.load_config()
|
|
lease=RuntimeLease(config['manager_uid'],config['manager_gid'],config['runtime_uid'],config['runtime_gid'],digest)
|
|
manager=None
|
|
try:
|
|
manager=c01_oci_job.UserManager(lease,config)
|
|
pid=lease.identity['pid'];conf=Path(config['home'])/'.config/containers/storage.conf'
|
|
viewed=Path(f'/proc/{pid}/root')/conf.relative_to('/')
|
|
storage=tomllib.loads(viewed.read_text());selected=importer.storage_declaration(config,storage)
|
|
script=package/'scripts/c01_oci_job.py'
|
|
context={'uid':config['runtime_uid'],'gid':config['runtime_gid'],'home':config['home'],'username':config['runtime_user'],
|
|
'runtime_dir':config['runtime_dir'],'mount_namespace':lease.identity['mount_namespace'],
|
|
'network_namespace':os.stat(f'/proc/{pid}/ns/net').st_ino,'helper_sha256':uat_oci.sha(script),
|
|
'podman_sha256':uat_oci.sha(Path(f'/proc/{pid}/root/usr/bin/podman')),
|
|
'config_path':str(conf),'storage_sha256':uat_oci.sha(viewed),'driver':selected['driver'],
|
|
'graphroot':selected['graphroot'],'runroot':selected['runroot'],
|
|
'image_id':proof['config_id'].removeprefix('sha256:'),
|
|
'tag':uat_oci.REPOSITORY+':uat-'+proof['manifest_digest'].removeprefix('sha256:'),'reference':proof['reference']}
|
|
context['archive_sha256']=proof['archive_sha256']
|
|
original=c['policy'].identity(Path(config['graphroot']).stat())
|
|
with (package/'lnd-tor.oci.tar').open('rb') as archive:
|
|
def call(args):
|
|
archive.seek(0)
|
|
with tempfile.TemporaryFile() as output,tempfile.TemporaryFile() as errors,open('/dev/null','rb') as empty:
|
|
try:manager.execute(script,context,args,archive if args[0]=='load' else empty,output,errors)
|
|
except Exception:
|
|
errors.seek(0)
|
|
c['write'](c['EVIDENCE']/'oci-job-error.log',errors.read(16384))
|
|
raise
|
|
output.seek(0);raw=output.read(4*1024*1024+1);assert len(raw)<=4*1024*1024
|
|
return raw.decode()
|
|
before=json.loads(call(['info','--format','json']))
|
|
importer.storage_check(before,config,storage,conf)
|
|
imported=importer.import_image(call,Path('/dev/stdin'),proof)
|
|
after=json.loads(call(['info','--format','json']))
|
|
importer.storage_check(after,config,storage,conf)
|
|
assert c['policy'].identity(Path(config['graphroot']).stat())==original
|
|
result={'status':'passed-private-appointed-store-import','source_qualified':False,'installed_import':False,
|
|
'image':imported,'archive_sha256':proof['archive_sha256'],'runtime_sha256':digest,
|
|
'storage_sha256':context['storage_sha256'],'store':after['store'],'jobs':manager.records,
|
|
'graphroot_identity':original,'container_activation_requested':False}
|
|
c['dump'](c['EVIDENCE']/'appointed-oci.json',result)
|
|
return result
|
|
finally:
|
|
try:
|
|
if manager is not None:manager.close()
|
|
finally:lease.close()
|