spoc/usr/lib/python3.8/spoc/imagebuilder.py

# -*- coding: utf-8 -*-

import os
import requests
import shutil
import stat
import tarfile
import tempfile
import zipfile

from .container import Container
from .image import Image
from .config import LAYERS_DIR

class ImageBuilder:
    def build(self, image, filename):
        # Reset internal state, read and process lines from filename
        self.image = image
        self.builddir = os.path.dirname(filename)
        self.script_eof = None
        self.script_lines = []
        with open(filename, 'r') as f:
            for line in f:
                self.process_line(line.strip())

    def process_line(self, line):
        # Parse a line from image file
        if self.script_eof:
            if line == self.script_eof:
                self.script_eof = None
                self.run_script(self.script_lines)
            else:
                self.script_lines.append(line)
        elif line:
            self.process_directive(*line.split(None, 1))

    def process_directive(self, directive, args):
        # Process a directive from image file
        if 'RUN' == directive:
            self.script_lines = []
            self.script_eof = args
        elif 'FROM' == directive:
            # Set the values of image from which this one inherits
            self.image.set_definition(Image(args).get_definition())
            self.image.layers.append(self.image.name)
        elif 'COPY' == directive:
            srcdst = args.split()
            self.copy_files(srcdst[0], srcdst[1] if len(srcdst) > 1 else '')
        elif 'ENV' == directive:
            # Sets/unsets environment variable
            self.set_env(*args.split(None, 1))
        elif 'USER' == directive:
            # Sets init UID / GID
            self.set_uidgid(*args.split())
        elif 'CMD' == directive:
            # Sets init command
            self.image.cmd = args
        elif 'WORKDIR' == directive:
            # Sets init working directory
            self.image.cwd = args
        elif 'HALT' == directive:
            # Sets signal to be sent to init when stopping the container
            self.image.halt = args
        elif 'READY' == directive:
            # Sets a command to check readiness of the container after it has been started
            self.image.ready = args

    def run_script(self, script_lines):
        # Creates a temporary container, runs a script in its namespace, and stores the files modified by it as part of the layer
        # Note: If USER or WORKDIR directive has already been set, the command is run under that UID/GID or working directory
        script_fd, script_path = tempfile.mkstemp(suffix='.sh', dir=self.image.layer_path, text=True)
        script_name = os.path.basename(script_path)
        script_lines = '\n'.join(script_lines)
        with os.fdopen(script_fd, 'w') as script:
            script.write(f'#!/bin/sh\nset -ev\n\n{script_lines}\n')
        os.chmod(script_path, 0o755)
        os.chown(script_path, 100000, 100000)
        # Create a temporary container from the current image definition and execute the script within the container
        container = Container(self.image.name, False)
        container.set_definition(self.image.get_definition())
        container.build = True
        container.create()
        container.execute(['/bin/sh', '-lc', os.path.join('/', script_name)], check=True)
        container.destroy()
        os.unlink(script_path)

    def set_env(self, key, value=None):
        # Set or unset environement variable
        if value:
            self.image.env[key] = value
        else:
            try:
                del self.image.env[key]
            except KeyError:
                pass

    def set_uidgid(self, uid, gid=''):
        # Set UID/GID for init
        if not uid.isdigit() or not gid.isdigit():
            # Resolve the UID/GID from container if either of them is entered as string
            container = Container(self.image.name, False)
            container.set_definition(self.image.get_definition())
            container.create()
            uid,gid = container.get_uidgid(uid, gid)
            container.destroy()
        self.image.uid = uid
        self.image.gid = gid

    def copy_files(self, src, dst):
        # Copy files from the host or download them from a http(s) URL
        dst = os.path.join(self.image.layer_path, dst.lstrip('/'))
        if src.startswith('http://') or src.startswith('https://'):
            unpack_http_archive(src, dst)
        else:
            src = os.path.join(self.builddir, src)
            if not os.path.isdir(src):
                shutil.copy2(src, dst)
            else:
                shutil.copytree(src, dst, symlinks=True, ignore_dangling_symlinks=True, dirs_exist_ok=True)
        # Shift UID/GID of the files to the unprivileged range
        shift_uid(dst, os.stat(dst, follow_symlinks=False))

def unpack_http_archive(src, dst):
    # Decompress an archive downloaded via http(s)
    with tempfile.TemporaryFile() as tmp_archive:
        # Download the file via http(s) and store as temporary file
        with requests.Session() as session:
            resource = session.get(src, stream=True)
            resource.raise_for_status()
            for chunk in resource.iter_content(chunk_size=None):
                if chunk:
                    tmp_archive.write(chunk)
        # Check if the magic bytes and determine if the file is zip
        tmp_archive.seek(0)
        is_zip = zipfile.is_zipfile(tmp_archive)
        # Extract the file. If it is not zip, assume tar (bzip2, gizp or xz)
        tmp_archive.seek(0)
        if is_zip:
            with zipfile.ZipFile(tmp_archive) as zip:
                zip.extractall(dst)
        else:
            with tarfile.open(fileobj=tmp_archive) as tar:
                tar.extractall(dst, numeric_owner=True)

def shift_uid(path, path_stat):
    # Shifts UID/GID of a file or a directory and its contents to the unprivileged range
    # The function parameters could arguably be more friendly, but os.scandir() already calls stat() on the entires,
    # so it would be wasteful to not reuse them for considerable performance gain
    uid = path_stat.st_uid
    gid = path_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.chown(path, uid, gid, follow_symlinks=False)
    if stat.S_ISDIR(path_stat.st_mode):
        for entry in os.scandir(path):
            shift_uid(entry.path, entry.stat(follow_symlinks=False))