debian-forge/inputs/org.osbuild.containers

#!/usr/bin/python3
"""Inputs for container images

This reads images from the `org.osbuild.containers` directory in the
sources store, or from oci-archive files in pipelines (typically
created by `org.osbuild.oci-archive`).

The store is indexed by the "container image id", which is the digest
of the container configuration file (rather than the outer manifest)
and is what will be shown in the "podman images" output when the image
is installed. This digest is stable as opposed to the manifest digest
which can change during transfer and storage due to
e.g. recompression.

When using pipeline sources, the first file (alphabetically) in the
root of the tree is used as the oci archive file to install.

"""

import os
import sys

from osbuild import inputs
from osbuild.util.mnt import MountGuard

SCHEMA = r"""
"definitions": {
  "source-options": {
    "type": "object",
    "required": ["name"],
    "additionalProperties": false,
    "properties": {
      "name": {
        "type": "string",
        "description": "The name to use for the image"
      }
    }
  },
  "source-object-ref": {
    "type": "object",
    "additionalProperties": false,
    "minProperties": 1,
    "patternProperties": {
      ".*": {
        "$ref": "#/definitions/source-options"
      }
    }
  },
  "source-array-ref": {
    "type": "array",
    "minItems": 1,
    "items": {
      "type": "object",
      "additionalProperties": false,
      "required": ["id"],
      "properties": {
        "id": {
          "type": "string"
        },
        "options": {
          "$ref": "#/definitions/source-options"
        }
      }
    }
  },
  "source-origin": {
    "type": "string",
    "description": "When the origin of the input is a source",
    "enum": ["org.osbuild.source"]
  },
  "pipeline-options": {
    "type": "object",
    "additionalProperties": false,
    "required": ["name"],
    "properties": {
      "name": {
        "type": "string",
        "description": "The name to use for the image"
      }
    }
  },
  "pipeline-object-ref": {
    "type": "object",
    "additionalProperties": false,
    "minProperties": 1,
    "patternProperties": {
      ".*": {
        "$ref": "#/definitions/pipeline-options"
      }
    }
  },
  "pipeline-array-ref": {
    "type": "array",
    "minItems": 1,
    "items": {
      "type": "object",
      "additionalProperties": false,
      "required": ["id"],
      "properties": {
        "id": {
          "type": "string"
        },
        "options": {
          "$ref": "#/definitions/pipeline-options"
        }
      }
    }
  },
  "pipeline-origin": {
    "type": "string",
    "description": "When the origin of the input is a pipeline",
    "enum": ["org.osbuild.pipeline"]
  }
},
"additionalProperties": true,
"oneOf": [
  {
    "additionalProperties": false,
    "required": ["type", "origin", "references"],
    "properties": {
      "type": {
        "enum": ["org.osbuild.containers"]
      },
      "origin": {
        "description": "The org.osbuild.source origin case",
        "$ref": "#/definitions/source-origin"
      },
      "references": {
        "description": "Container image id",
         "oneOf": [
          {"$ref": "#/definitions/source-array-ref"},
          {"$ref": "#/definitions/source-object-ref"}
        ]
      }
    }
  },
  {
    "additionalProperties": false,
    "required": ["type", "origin", "references"],
    "properties": {
      "type": {
        "enum": ["org.osbuild.containers"]
      },
      "origin": {
        "description": "The org.osbuild.source origin case",
        "$ref": "#/definitions/pipeline-origin"
      },
      "references": {
        "description": "References to pipelines",
        "oneOf": [
          {"$ref": "#/definitions/pipeline-array-ref"},
          {"$ref": "#/definitions/pipeline-object-ref"}
        ]
      }
    }
  }
]
"""


class ContainersInput(inputs.InputService):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.mg = MountGuard()

    def map_source_ref(self, source, ref, target):
        source_archive = os.path.join(source, ref, "image")
        dest = os.path.join(target, ref)

        # bind mount the input directory to the destination
        os.makedirs(dest)
        self.mg.mount(source_archive, dest)

        return ref, "dir"

    @staticmethod
    def map_pipeline_ref(store, ref, target):
        # prepare the mount point
        os.makedirs(target, exist_ok=True)
        print("target", target)

        store.read_tree_at(ref, target)

        # Find the archive file in target, we use the first alphabetical regular file
        files = sorted(filter(lambda f: os.path.isfile(os.path.join(target, f)),
                              os.listdir(target)))
        if len(files) == 0:
            raise RuntimeError("No archive files in source")

        return files[0], "oci-archive"

    def map(self, store, origin, refs, target, _options):
        source = store.source("org.osbuild.containers")
        images = {}

        for ref, data in refs.items():
            if origin == "org.osbuild.source":
                ref, container_format = self.map_source_ref(source, ref, target)
            else:
                ref, container_format = self.map_pipeline_ref(store, ref, target)

            images[ref] = {
                "format": container_format,
                "name": data["name"]
            }
            images[ref]["name"] = data["name"]

        reply = {
            "path": target,
            "data": {
                "archives": images
            }
        }
        return reply

    def unmap(self):
        self.mg.umount()


def main():
    service = ContainersInput.from_args(sys.argv[1:])
    service.main()


if __name__ == '__main__':
    main()