Spotter-VM/build/usr/lib/python3.6/lxcbuild/builder.py

175 lines
5.4 KiB
Python

# -*- coding: utf-8 -*-
import os
import shutil
import subprocess
import sys
from lxcmgr import lxcmgr
from lxcmgr.paths import PKG_STORAGE_DIR
class ImageExistsError(Exception):
pass
class ImageNotFoundError(Exception):
pass
class Builder:
def __init__(self):
self.image = None
self.script = []
self.script_eof = None
self.force = False
def build(self, image, force=False):
self.image = image
self.force = force
with open(self.image.lxcfile, 'r') as f:
for line in f:
line = line.strip()
if self.script_eof:
if line == self.script_eof:
self.script_eof = None
self.run_script(self.script)
else:
self.script.append(line)
elif line:
self.process_line(*line.split(None, 1))
def process_line(self, directive, args):
if 'RUN' == directive:
self.script = []
self.script_eof = args
elif 'IMAGE' == directive:
self.set_name(args)
elif 'LAYER' == directive:
self.add_layer(args)
elif 'FIXLAYER' == directive:
self.fix_layer(args.split())
elif 'COPY' == directive:
srcdst = args.split()
self.copy_files(srcdst[0], srcdst[1] if len(srcdst) == 2 else '')
elif 'ENV' == directive:
self.add_env(*args.split(None, 1))
elif 'USER' == directive:
self.set_user(*args.split())
elif 'CMD' == directive:
self.set_cmd(args)
elif 'WORKDIR' == directive:
self.set_cwd(args)
elif 'HALT' == directive:
self.set_halt(args)
elif 'READY' == directive:
self.set_ready(args)
def get_layer_path(self, layer):
return os.path.join(PKG_STORAGE_DIR, layer)
def run_script(self, script):
lxcmgr.register_container(self.image.name, self.image.conf)
sh = os.path.join(self.image.path, 'run.sh')
with open(sh, 'w') as f:
f.write('#!/bin/sh\nset -ev\n\n{}\n'.format('\n'.join(script)))
os.chmod(sh, 0o700)
os.chown(sh, 100000, 100000)
subprocess.run(['lxc-execute', self.image.name, '--', '/bin/sh', '-lc', '/run.sh'], check=True)
os.unlink(sh)
lxcmgr.unregister_container(self.image.name)
def set_name(self, name):
self.image.name = name
self.image.path = self.get_layer_path(name)
self.image.conf['layers'] = [name]
if os.path.exists(self.image.path):
if self.force:
self.clean()
else:
raise ImageExistsError(self.image.path)
os.makedirs(self.image.path, 0o755, True)
os.chown(self.image.path, 100000, 100000)
def add_layer(self, name):
layer_path = self.get_layer_path(name)
if not os.path.exists(layer_path):
raise ImageNotFoundError(layer_path)
self.image.conf['layers'].insert(0, name)
def fix_layer(self, cmd):
layers = [self.get_layer_path(layer) for layer in self.image.conf['layers']]
subprocess.run([cmd] + layers, check=True)
def copy_files(self, src, dst):
dst = os.path.join(self.image.path, dst)
if src.startswith('http://') or src.startswith('https://'):
unpack_http_archive(src, dst)
else:
copy_tree(os.path.join(self.image.build_dir, src), dst)
shift_uid(dst)
def add_env(self, key, value):
if 'env' not in self.image.conf:
self.image.conf['env'] = []
self.image.conf['env'].append('{}={}'.format(key, value))
def set_user(self, uid, gid):
self.image.conf['uid'] = uid
self.image.conf['gid'] = gid
def set_cmd(self, cmd):
self.image.conf['cmd'] = cmd
def set_cwd(self, cwd):
self.image.conf['cwd'] = cwd
def set_halt(self, halt):
self.image.conf['halt'] = halt
def set_ready(self, cmd):
self.image.conf['ready'] = cmd
def clean(self):
shutil.rmtree(self.image.path)
def unpack_http_archive(src, dst):
xf = 'xzf'
if src.endswith('.bz2'):
xf = 'xjf'
elif src.endswith('.xz'):
xf = 'xJf'
with subprocess.Popen(['wget', src, '-O', '-'], stdout=subprocess.PIPE) as wget:
with subprocess.Popen(['tar', xf, '-', '-C', dst], stdin=wget.stdout) as tar:
wget.stdout.close()
tar.wait()
def copy_tree(src, dst):
if not os.path.isdir(src):
shutil.copy2(src, dst)
else:
os.makedirs(dst, exist_ok=True)
for name in os.listdir(src):
copy_tree(os.path.join(src, name), os.path.join(dst, name))
shutil.copystat(src, dst)
def shift_uid(dir):
shift_uid_entry(dir, os.stat(dir, follow_symlinks=True))
shift_uid_recursively(dir)
def shift_uid_recursively(dir):
for entry in os.scandir(dir):
shift_uid_entry(entry.path, entry.stat(follow_symlinks=False))
if entry.is_dir():
shift_uid_recursively(entry.path)
def shift_uid_entry(path, stat):
uid = stat.st_uid
gid = stat.st_gid
do_chown = False
if uid < 100000:
uid = uid + 100000
do_chown = True
if gid < 100000:
gid = gid + 100000
do_chown = True
if do_chown:
os.lchown(path, uid, gid)