fedora-coreos-config/build-rootfs

#!/usr/bin/python3

# This script is called by the Containerfile to build FCOS. Here's what it does at a high-level:
# 1. It gathers the list of FCOS-specific packages using the manifests.
# 2. It gathers the list of FCOS-specific overlays using the manifests.
# 3. It runs `bootc-base-imagectl rebuild`, passing in the packages and overlays.
# 4. It injects various metadata (e.g. image.json, live/ bits, and platforms.json).
# 5. It runs the postprocess scripts defined in the manifest.

import glob
import json
import os
import shutil
import subprocess
import sys
import tempfile
import yaml


ARCH = os.uname().machine
CONTEXTDIR = '/run/src'


def main():
    manifest_name = sys.argv[1]
    version = sys.argv[2]
    target_rootfs = sys.argv[3]

    manifest_path = os.path.join(CONTEXTDIR, manifest_name)
    manifest = get_treefile(manifest_path)

    packages = list(manifest['packages'])

    repos = manifest.get('repos', []) + manifest.get('lockfile-repos', [])
    if repos:
        inject_yumrepos()

    local_overrides = prepare_local_rpm_overrides(target_rootfs)
    if local_overrides:
        repos += ['overrides']

    locked_nevras = get_locked_nevras(local_overrides)

    overlays = gather_overlays(manifest)
    nodocs = (manifest.get('documentation') is False)
    build_rootfs(target_rootfs, manifest_path, packages, locked_nevras, overlays, repos, nodocs)

    inject_live(target_rootfs)
    inject_image_json(target_rootfs, manifest_path)
    inject_platforms_json(target_rootfs)
    inject_content_manifest(target_rootfs, manifest)

    if version != "":
        inject_version_info(target_rootfs, manifest['mutate-os-release'], version)
    run_postprocess_scripts(target_rootfs, manifest)


def get_treefile(manifest_path):
    with tempfile.NamedTemporaryFile(suffix='.json', mode='w') as tmp_manifest:
        # This ensures that the treefile represents only the CoreOS bits and
        # doesn't recurse into fedora-bootc. We can drop this once we've fully
        # cut over to derivation.
        json.dump({
            "variables": {
                "deriving": True
            },
            "include": manifest_path
        }, tmp_manifest)
        tmp_manifest.flush()
        data = subprocess.check_output(['rpm-ostree', 'compose', 'tree',
                                        '--print-only', tmp_manifest.name])
    return json.loads(data)


def inject_yumrepos():
    # first delete all the default repos
    for repo in glob.glob('/etc/yum.repos.d/*.repo'):
        if os.path.basename(repo) == 'secret.repo':
            # this is a supported podman secret to inject repo files; see Containerfile
            continue
        os.unlink(repo)

    # and now inject our repos
    for repo in glob.glob(f'{CONTEXTDIR}/*.repo'):
        shutil.copy(repo, "/etc/yum.repos.d")


def build_rootfs(target_rootfs, manifest_path, packages, locked_nevras, overlays, repos, nodocs):
    passwd_group_dir = os.getenv('PASSWD_GROUP_DIR')
    if passwd_group_dir is not None:
        inject_passwd_group(os.path.join(CONTEXTDIR, passwd_group_dir))
    with tempfile.NamedTemporaryFile(mode='w') as argsfile:
        for pkg in packages:
            argsfile.write(f"--install={pkg}\n")
        for overlay in overlays:
            argsfile.write(f"--add-dir={overlay}\n")
        if nodocs:
            argsfile.write("--no-docs\n")
            # temporarily work around https://issues.redhat.com/browse/RHEL-97826
            tmpd = workaround_rhel_97826(argsfile)
        if repos and repo_arg_supported():
            for repo in repos:
                argsfile.write(f"--repo={repo}\n")
        if locked_nevras and lock_arg_supported():
            for locked_nevra in locked_nevras:
                argsfile.write(f"--lock={locked_nevra}\n")
        argsfile.flush()
        cache_arg = []
        if os.path.isdir('/cache') and rpm_ostree_has_cachedir_fix():
            cache_arg = ['--cachedir=/cache']
        subprocess.check_call(["/usr/libexec/bootc-base-imagectl",
                               "--args-file", argsfile.name, "build-rootfs",
                               "--manifest", 'minimal-plus',
                               target_rootfs] + cache_arg)
    if nodocs and tmpd is not None:
        del tmpd


def repo_arg_supported():
    # Detect if we have https://gitlab.com/fedora/bootc/base-images/-/merge_requests/248.
    # If not, then we can't use `--repo`. That's OK because that should only
    # happen on RHEL, where we don't have any default repos anyway and only rely on
    # the mounted secret repo file.
    help = subprocess.check_output(['/usr/libexec/bootc-base-imagectl', 'build-rootfs', '-h'], encoding='utf-8')
    return '--repo REPO' in help


def lock_arg_supported():
    # Detect if we have https://gitlab.com/fedora/bootc/base-images/-/merge_requests/279.
    # If not, then we can't use `--lock`. That should only happen in RHCOS,
    # where we only use this for autolocking and not base lockfile management.
    help = subprocess.check_output(['/usr/libexec/bootc-base-imagectl', 'build-rootfs', '-h'], encoding='utf-8')
    return '--lock NEVRA' in help


def workaround_rhel_97826(argsfile):
    basedir = 'usr/share/doc/bootc/baseimage/base'
    # Detect if we have https://github.com/bootc-dev/bootc/pull/1352.
    # This is not strictly correct because we're checking the _builder
    # image_ which could be different from the target. In practice, we don't
    # "cross-compose" and we're always going to be fetching equal or newer
    # NEVRAs from the repos so this is fine.
    if os.path.exists(f'/{basedir}/sysroot/.keepdir'):
        return
    print("Working around RHEL-97826", flush=True)
    tmpd = tempfile.TemporaryDirectory()
    shutil.copytree(f'/{basedir}', os.path.join(tmpd.name, basedir), symlinks=True)
    argsfile.write(f'--add-dir={tmpd.name}\n')
    return tmpd


def rpm_ostree_has_cachedir_fix():
    # we can only use --cachedir if we have rpm-ostree >= 2025.9 which has
    # https://github.com/coreos/rpm-ostree/pull/5391
    out = subprocess.check_output(['rpm-ostree', '--version'], encoding='utf-8')
    data = yaml.safe_load(out)
    version_str = data['rpm-ostree']['Version']
    # ideally, we could use `packaging.version`, but that's not in centos-bootc
    # but conveniently, Python list comparisons do the right thing here
    version = [int(c) for c in version_str.split('.')]
    return version >= [2025, 9]


# We want to keep our passwd/group as canonical for now. We should be
# able to clean this up when we migrate them to sysusers instead. See:
# https://github.com/coreos/rpm-ostree/pull/5427
def inject_passwd_group(parent_dir):
    minimal = '/usr/share/doc/bootc-base-imagectl/manifests/minimal'
    dst_passwd = os.path.join(minimal, 'passwd')
    dst_group = os.path.join(minimal, 'group')
    # unlink first instead of overwriting as a way to confirm they're still there
    os.unlink(dst_passwd)
    os.unlink(dst_group)
    print("Overriding passwd/group files", flush=True)
    shutil.copy(os.path.join(parent_dir, 'passwd'), dst_passwd)
    shutil.copy(os.path.join(parent_dir, 'group'), dst_group)


def run_postprocess_scripts(rootfs, manifest):
    # Since we have the derive-only manifest handy, just run the scripts now. An
    # alternative is to run it as a second stage, which would avoid the bwrap,
    # but operating on the raw rootfs means we don't pay for deleted files (nor
    # without requiring another rechunk).
    for i, script in enumerate(manifest.get('postprocess', [])):
        name = f'usr/libexec/coreos-postprocess-{i}'
        with open(os.path.join(rootfs, name), mode='w') as f:
            os.fchmod(f.fileno(), 0o755)
            f.write(script)
        print(f"Running CoreOS postprocess script {i}", flush=True)
        bwrap(rootfs, [f'/{name}'])
        os.unlink(os.path.join(rootfs, name))


def prepare_local_rpm_overrides(rootfs):
    overrides_repo = os.path.join(CONTEXTDIR, 'overrides/rpm')
    if not os.path.isdir(f'{overrides_repo}/repodata'):
        return None

    pkglist = subprocess.check_output(['dnf', 'repoquery', f'--repofrompath=overrides,file://{overrides_repo}',
                                       '--repo=overrides', '--latest-limit=1', f'--arch={ARCH},noarch',
                                       '--qf', 'pkg: %{name} %{evr} %{arch}\n'], encoding='utf-8')
    lockfile = {}
    for line in pkglist.splitlines():
        if not line.startswith("pkg: "):
            continue
        _, name, evr, arch = line.strip().split()
        lockfile[name] = {"evra": f"{evr}.{arch}"}

    if len(lockfile) == 0:
        return None

    with open('/etc/yum.repos.d/overrides.repo', 'w') as f:
        f.write(f'''
[overrides]
name=overrides
baseurl=file://{overrides_repo}
gpgcheck=0
cost=500
priority=1
''')

    print("Injected", len(lockfile), 'package overrides')
    return lockfile


# Could upstream this as e.g. `bootc-base-imagectl runroot /rootfs <cmd>` maybe?
# But we'd need to carry it anyway at least for RHCOS 9.6.
def bwrap(rootfs, args):
    subprocess.check_call(['bwrap', '--bind', f'{rootfs}', '/',
                           '--dev', '/dev', '--proc', '/proc',
                           '--tmpfs', '/tmp', '--tmpfs', '/var', '--tmpfs', '/run',
                           '--bind', '/run/.containerenv', '/run/.containerenv',
                           '--'] + args)


def get_locked_nevras(local_overrides):
    lockfile_path = os.path.join(CONTEXTDIR, f"manifest-lock.{ARCH}.json")
    overrides_path = os.path.join(CONTEXTDIR, "manifest-lock.overrides.yaml")
    overrides_arch_path = os.path.join(CONTEXTDIR, f"manifest-lock.overrides.{ARCH}.yaml")

    # we go from lowest priority to highest here: base lockfiles, overrides, local overrides
    locks = {}
    for path in [lockfile_path, overrides_path, overrides_arch_path]:
        if os.path.exists(path):
            with open(path) as f:
                if path.endswith('.yaml'):
                    data = yaml.safe_load(f)
                else:
                    data = json.load(f)
                # this essentially re-implements the merge semantics of rpm-ostree
                locks.update({pkgname: v['evra'] if 'evra' in v else v['evr']
                              for (pkgname, v) in data['packages'].items()})
    if local_overrides:
        locks.update({pkgname: v['evra'] if 'evra' in v else v['evr']
                      for (pkgname, v) in local_overrides.items()})
    return [f'{k}-{v}' for (k, v) in locks.items()]


# This re-implements rpm-ostree's mutate-os-release to preserve the historical
# /usr/lib/os-release API, but we may in the future completely sever off of that
# and only rely on org.opencontainers.image.version as argued in:
# https://gitlab.com/fedora/bootc/base-images/-/issues/40
# https://gitlab.com/fedora/bootc/base-images/-/issues/46
def inject_version_info(rootfs, base_version, version):
    os_release_path = os.path.join(rootfs, 'usr/lib/os-release')
    with open(os_release_path) as f:
        from collections import OrderedDict
        os_release = OrderedDict()
        for line in f:
            line = line.strip()
            if len(line) == 0 or line.startswith('#'):
                continue
            (k, v) = line.split('=', 1)
            os_release[k] = v

    for key in ['VERSION', 'PRETTY_NAME']:
        os_release[key] = os_release[key].replace(base_version, version)
    os_release['OSTREE_VERSION'] = f"'{version}'"
    os_release['IMAGE_VERSION'] = f"'{version}'"

    with open(os_release_path, mode='w', encoding='utf-8') as f:
        for (k, v) in os_release.items():
            f.write(f'{k}={v}\n')


# This re-implements cosa's overlay logic.
def gather_overlays(manifest):
    overlays = []
    for layer in manifest.get('ostree-layers', []):
        assert layer.startswith('overlay/')
        overlays.append(os.path.join(CONTEXTDIR, 'overlay.d', layer[len('overlay/'):]))

    rootfs_override = os.path.join(CONTEXTDIR, 'overrides/rootfs')
    if os.path.isdir(rootfs_override) and len(os.listdir(rootfs_override)) > 0:
        print("Injecting rootfs override")
        overlays.append(rootfs_override)

    return overlays


# Inject live/ bits.
def inject_live(rootfs):
    target_path = os.path.join(rootfs, 'usr/share/coreos-assembler/live')
    shutil.copytree(os.path.join(CONTEXTDIR, "live"), target_path)


def inject_image_json(rootfs, manifest_path):
    manifest_vars = yaml.safe_load(open(manifest_path))['variables']
    image = flatten_image_yaml(os.path.join(CONTEXTDIR, 'image.yaml'),
                               format_args=manifest_vars)
    fn = os.path.join(rootfs, 'usr/share/coreos-assembler/image.json')
    with open(fn, 'w') as f:
        json.dump(image, f, sort_keys=True)


def inject_content_manifest(target_rootfs, manifest):
    content_manifest_path = '/run/secrets/contentsets'
    if not os.path.exists(content_manifest_path):
        return

    with open(content_manifest_path) as f:
        data = yaml.safe_load(f)

    repos = []
    for base_repo in manifest['repos']:
        if base_repo in data['repo_mapping']:
            if data['repo_mapping'][base_repo]['name'] != '':
                repo_name = data['repo_mapping'][base_repo]['name'].replace('$ARCH', ARCH)
                repos.append(repo_name)
            else:
                print('Warning: No corresponding repo in repository-to-cpe.json for ' + base_repo)
        else:
            # Warning message for repositories with no entry in content_sets.yaml
            print('Warning: No corresponding entry in content_sets.yaml for ' + base_repo)

    dest = os.path.join(target_rootfs, "usr/share/buildinfo/content_manifest.json")
    os.makedirs(os.path.dirname(dest), exist_ok=True)
    with open(dest, 'w') as f:
        json.dump(fp=f, obj={
            'metadata': {
                'icm_version': 1,
                'icm_spec': 'https://raw.githubusercontent.com/containerbuildsystem/atomic-reactor/master/atomic_reactor/schemas/content_manifest.json',
                'image_layer_index': 1
            },
            'content_sets': repos,
            'image_contents': []
        })


# Imported from cosa
# Merge two lists, avoiding duplicates. Exact duplicate kargs could be valid
# but we have no use case for them right now in our official images.
def merge_lists(x, y, k):
    x[k] = x.get(k, [])
    assert isinstance(x[k], list)
    y[k] = y.get(k, [])
    assert isinstance(y[k], list)
    x[k].extend([i for i in y[k] if i not in x[k]])


# Imported from cosa
def flatten_image_yaml(srcfile, base=None, format_args={}):
    if base is None:
        base = {}

    with open(srcfile) as f:
        contents = f.read()
        srcyaml = yaml.safe_load(contents.format(**format_args))

    # first, special-case list values
    merge_lists(base, srcyaml, 'extra-kargs')

    # then handle all the non-list values
    base = merge_dicts(base, srcyaml)

    if 'include' not in srcyaml:
        return base

    fn = os.path.join(os.path.dirname(srcfile), srcyaml['include'])
    del base['include']
    return flatten_image_yaml(fn, base=base, format_args=format_args)


# Imported from cosa
# Credit to @arithx
def merge_dicts(x, y):
    """
    Merge two dicts recursively, but based on the difference.
    """
    sd = set(x.keys()).symmetric_difference(y.keys())
    ret = {}
    for d in [x, y]:
        for k, v in d.items():
            if k in sd:
                # the key is only present in one dict, add it directly
                ret.update({k: v})
            elif isinstance(x[k], dict) and isinstance(y[k], dict):
                # recursively merge
                ret.update({k: merge_dicts(x[k], y[k])})
            elif isinstance(x[k], list) and isinstance(y[k], list):
                ret.update({k: x[k]})
                merge_lists(ret, y, k)
            else:
                # first dictionary always takes precedence
                ret.update({k: x[k]})
    return ret


def inject_platforms_json(rootfs):
    with open(os.path.join(CONTEXTDIR, 'platforms.yaml')) as f:
        platforms = yaml.safe_load(f)
    fn = os.path.join(rootfs, 'usr/share/coreos-assembler/platforms.json')
    if ARCH in platforms:
        with open(fn, 'w') as f:
            json.dump(platforms[ARCH], f, sort_keys=True, indent=2)
            f.write('\n')


if __name__ == "__main__":
    sys.exit(main())