focker/focker/steps.py

#
# Copyright (C) Stanislaw Adaszewski, 2020
# License: GNU General Public License v3.0
# URL: https://github.com/sadaszewski/focker
# URL: https://adared.ch/focker
#

import hashlib
import json
from .jail import jail_run
import shutil
import os
import shlex
from .misc import filehash


class RunStep(object):
    def __init__(self, spec):
        if not isinstance(spec, list) and \
            not isinstance(spec, str):
            raise ValueError('Run spec must be a list or a string')
        self.spec = spec

    def hash(self, base, **kwargs):
        res = hashlib.sha256(
            json.dumps(( base, self.spec ))
            .encode('utf-8')).hexdigest()
        return res

    def execute(self, path, **kwargs):
        spec = self.spec
        if isinstance(spec, list):
            spec = ' && ' .join(self.spec)
        jail_run(path, spec)


class CopyStep(object):
    def __init__(self, spec):
        if not isinstance(spec, list):
            raise ValueError('CopyStep spec should be a list')
        self.spec = spec

    def hash(self, base, args, **kwargs):
        if len(self.spec) == 0:
            fh = []
        elif isinstance(self.spec[0], list):
            fh = list(map(lambda a: filehash(os.path.join(args.focker_dir, a[0])), self.spec))
        else:
            fh = [ filehash(os.path.join(args.focker_dir, self.spec[0])) ]
        res = hashlib.sha256(
            json.dumps(( base, fh, self.spec ))
            .encode('utf-8')).hexdigest()
        return res

    def execute(self, path, **kwargs):
        lst = [ self.spec ] \
            if not isinstance(self.spec[0], list) \
            else self.spec
        for entry in lst:
            (source, target) = entry[:2]
            options = entry[2] if len(entry) > 2 else {}
            target = target.strip('/')
            os.makedirs(os.path.split(os.path.join(path, target))[0], exist_ok=True)
            shutil.copyfile(os.path.join(kwargs['args'].focker_dir, source),
                os.path.join(path, target))
            if 'chmod' in options:
                os.chmod(os.path.join(path, target), options['chmod'])
            if 'chown' in options:
                uid, gid = options['chown'].split(':').map(int)
                os.chown(os.path.join(path, target), uid, gid)


def create_step(spec):
    if not isinstance(spec, dict):
        raise ValueError('Step specification must be a dictionary')
    if 'copy' in spec:
        return CopyStep(spec['copy'])
    elif 'run' in spec:
        return RunStep(spec['run'])
    raise ValueError('Unrecognized step spec: ' + json.dumps(spec))