package oci
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"net/url"
"os"
"path"
"path/filepath"
"strings"
"text/template"
eventtypes "github.com/containerd/containerd/api/events"
"github.com/containerd/containerd/v2/client"
"github.com/containerd/containerd/v2/core/content"
"github.com/containerd/errdefs"
"github.com/containerd/typeurl/v2"
"github.com/go-logr/logr"
"github.com/opencontainers/go-digest"
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pelletier/go-toml/v2"
tomlu "github.com/pelletier/go-toml/v2/unstable"
"google.golang.org/grpc"
utilversion "k8s.io/apimachinery/pkg/util/version"
runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
)
const (
backupDir = "_backup"
)
var _ Client = &Containerd{}
type Containerd struct {
contentPath string
client *client.Client
clientGetter func() (*client.Client, error)
listFilter string
eventFilter string
registryConfigPath string
}
type Option func(*Containerd)
func WithContentPath(path string) Option {
return func(c *Containerd) {
c.contentPath = path
}
}
func NewContainerd(sock, namespace, registryConfigPath string, mirroredRegistries []url.URL, opts ...Option) (*Containerd, error) {
listFilter, eventFilter := createFilters(mirroredRegistries)
c := &Containerd{
clientGetter: func() (*client.Client, error) {
return client.New(sock, client.WithDefaultNamespace(namespace))
},
listFilter: listFilter,
eventFilter: eventFilter,
registryConfigPath: registryConfigPath,
}
for _, opt := range opts {
opt(c)
}
return c, nil
}
func (c *Containerd) Client() (*client.Client, error) {
var err error
if c.client == nil {
c.client, err = c.clientGetter()
}
return c.client, err
}
func (c *Containerd) Name() string {
return "containerd"
}
func (c *Containerd) Verify(ctx context.Context) error {
log := logr.FromContextOrDiscard(ctx)
client, err := c.Client()
if err != nil {
return err
}
ok, err := client.IsServing(ctx)
if err != nil {
return err
}
if !ok {
return errors.New("could not reach Containerd service")
}
grpcConn, ok := client.Conn().(*grpc.ClientConn)
if !ok {
return errors.New("client connection is not grpc")
}
srv := runtimeapi.NewRuntimeServiceClient(grpcConn)
versionResp, err := srv.Version(ctx, &runtimeapi.VersionRequest{})
if err != nil {
return err
}
ok, err = canVerifyContainerdConfiguration(versionResp.RuntimeVersion)
if err != nil {
return fmt.Errorf("could not check Containerd version %s: %w", versionResp.RuntimeVersion, err)
}
if !ok {
log.Info("skipping verification of Containerd configuration", "version", versionResp.RuntimeVersion)
return nil
}
statusResp, err := srv.Status(ctx, &runtimeapi.StatusRequest{Verbose: true})
if err != nil {
return err
}
err = verifyStatusResponse(statusResp, c.registryConfigPath)
if err != nil {
return err
}
return nil
}
func canVerifyContainerdConfiguration(version string) (bool, error) {
v, err := utilversion.Parse(version)
if err != nil {
return false, err
}
return v.LessThan(utilversion.MustParse("2.0")), nil
}
func verifyStatusResponse(resp *runtimeapi.StatusResponse, configPath string) error {
str, ok := resp.Info["config"]
if !ok {
return errors.New("could not get config data from info response")
}
cfg := &struct {
Registry struct {
ConfigPath *string `json:"configPath"`
} `json:"registry"`
Containerd struct {
DiscardUnpackedLayers *bool `json:"discardUnpackedLayers"`
} `json:"containerd"`
}{}
err := json.Unmarshal([]byte(str), cfg)
if err != nil {
return err
}
if cfg.Containerd.DiscardUnpackedLayers == nil {
return errors.New("field containerd.discardUnpackedLayers missing from config")
}
if *cfg.Containerd.DiscardUnpackedLayers {
return errors.New("Containerd discard unpacked layers cannot be enabled")
}
if cfg.Registry.ConfigPath == nil {
return errors.New("field registry.configPath missing from config")
}
if *cfg.Registry.ConfigPath == "" {
return errors.New("Containerd registry config path needs to be set for mirror configuration to take effect")
}
paths := filepath.SplitList(*cfg.Registry.ConfigPath)
for _, path := range paths {
if path != configPath {
continue
}
return nil
}
return fmt.Errorf("Containerd registry config path is %s but needs to contain path %s for mirror configuration to take effect", *cfg.Registry.ConfigPath, configPath)
}
func (c *Containerd) Subscribe(ctx context.Context) (<-chan ImageEvent, <-chan error, error) {
imgCh := make(chan ImageEvent)
errCh := make(chan error)
client, err := c.Client()
if err != nil {
return nil, nil, err
}
envelopeCh, cErrCh := client.EventService().Subscribe(ctx, c.eventFilter)
go func() {
for envelope := range envelopeCh {
var img Image
imageName, eventType, err := getEventImage(envelope.Event)
if err != nil {
errCh <- err
continue
}
switch eventType {
case CreateEvent, UpdateEvent:
cImg, err := client.GetImage(ctx, imageName)
if err != nil {
errCh <- err
continue
}
img, err = ParseImageRequireDigest(cImg.Name(), cImg.Target().Digest)
if err != nil {
errCh <- err
continue
}
case DeleteEvent:
img, err = ParseImageRequireDigest(imageName, "")
if err != nil {
errCh <- err
continue
}
}
imgCh <- ImageEvent{Image: img, Type: eventType}
}
close(imgCh)
}()
go func() {
for err := range cErrCh {
errCh <- err
}
close(errCh)
}()
return imgCh, errCh, nil
}
func (c *Containerd) ListImages(ctx context.Context) ([]Image, error) {
client, err := c.Client()
if err != nil {
return nil, err
}
cImgs, err := client.ListImages(ctx, c.listFilter)
if err != nil {
return nil, err
}
imgs := []Image{}
for _, cImg := range cImgs {
img, err := ParseImageRequireDigest(cImg.Name(), cImg.Target().Digest)
if err != nil {
return nil, err
}
imgs = append(imgs, img)
}
return imgs, nil
}
func (c *Containerd) Resolve(ctx context.Context, ref string) (digest.Digest, error) {
client, err := c.Client()
if err != nil {
return "", err
}
cImg, err := client.GetImage(ctx, ref)
if err != nil {
return "", err
}
return cImg.Target().Digest, nil
}
func (c *Containerd) Size(ctx context.Context, dgst digest.Digest) (int64, error) {
client, err := c.Client()
if err != nil {
return 0, err
}
info, err := client.ContentStore().Info(ctx, dgst)
if errors.Is(err, errdefs.ErrNotFound) {
return 0, errors.Join(ErrNotFound, err)
}
if err != nil {
return 0, err
}
return info.Size, nil
}
func (c *Containerd) GetManifest(ctx context.Context, dgst digest.Digest) ([]byte, string, error) {
client, err := c.Client()
if err != nil {
return nil, "", err
}
b, err := content.ReadBlob(ctx, client.ContentStore(), ocispec.Descriptor{Digest: dgst})
if errors.Is(err, errdefs.ErrNotFound) {
return nil, "", errors.Join(ErrNotFound, err)
}
if err != nil {
return nil, "", err
}
mt, err := DetermineMediaType(b)
if err != nil {
return nil, "", err
}
return b, mt, nil
}
func (c *Containerd) GetBlob(ctx context.Context, dgst digest.Digest) (io.ReadSeekCloser, error) {
if c.contentPath != "" {
path := filepath.Join(c.contentPath, "blobs", dgst.Algorithm().String(), dgst.Encoded())
file, err := os.Open(path)
if err != nil {
if errors.Is(err, os.ErrNotExist) {
return nil, ErrNotFound
}
return nil, err
}
return file, nil
}
client, err := c.Client()
if err != nil {
return nil, err
}
ra, err := client.ContentStore().ReaderAt(ctx, ocispec.Descriptor{Digest: dgst})
if err != nil {
if errors.Is(err, errdefs.ErrNotFound) {
return nil, ErrNotFound
}
return nil, err
}
return struct {
io.ReadSeeker
io.Closer
}{ReadSeeker: io.NewSectionReader(ra, 0, ra.Size()), Closer: ra}, nil
}
func getEventImage(e typeurl.Any) (string, EventType, error) {
if e == nil {
return "", "", errors.New("any cannot be nil")
}
evt, err := typeurl.UnmarshalAny(e)
if err != nil {
return "", "", fmt.Errorf("failed to unmarshal any: %w", err)
}
switch e := evt.(type) {
case *eventtypes.ImageCreate:
return e.Name, CreateEvent, nil
case *eventtypes.ImageUpdate:
return e.Name, UpdateEvent, nil
case *eventtypes.ImageDelete:
return e.Name, DeleteEvent, nil
default:
return "", "", errors.New("unsupported event type")
}
}
func createFilters(mirroredRegistries []url.URL) (string, string) {
registryHosts := []string{}
for _, registry := range mirroredRegistries {
registryHosts = append(registryHosts, strings.ReplaceAll(registry.Host, `.`, `\\.`))
}
listFilter := fmt.Sprintf(`name~="^(%s)/"`, strings.Join(registryHosts, "|"))
if len(registryHosts) == 0 {
listFilter = `name~="^.+/"`
}
eventFilter := fmt.Sprintf(`topic~="/images/create|/images/update|/images/delete",event.%s`, listFilter)
return listFilter, eventFilter
}
func AddMirrorConfiguration(ctx context.Context, configPath string, mirroredRegistries, mirrorTargets []url.URL, resolveTags, prependExisting bool, username, password string) error {
log := logr.FromContextOrDiscard(ctx)
err := validateRegistries(mirroredRegistries)
if err != nil {
return err
}
err = os.MkdirAll(configPath, 0o755)
if err != nil {
return err
}
err = backupConfig(log, configPath)
if err != nil {
return err
}
err = clearConfig(configPath)
if err != nil {
return err
}
capabilities := []string{"pull"}
if resolveTags {
capabilities = append(capabilities, "resolve")
}
if len(mirroredRegistries) == 0 {
mirroredRegistries = append(mirroredRegistries, url.URL{})
}
for _, mirroredRegistry := range mirroredRegistries {
templatedHosts, err := templateHosts(mirroredRegistry, mirrorTargets, capabilities, username, password)
if err != nil {
return err
}
if prependExisting {
existingHosts, err := existingHosts(configPath, mirroredRegistry)
if err != nil {
return err
}
if existingHosts != "" {
templatedHosts = templatedHosts + "\n\n" + existingHosts
}
log.Info("prepending to existing Containerd mirror configuration", "registry", mirroredRegistry.String())
}
hostComp := mirroredRegistry.Host
if hostComp == "" {
hostComp = "_default"
}
fp := path.Join(configPath, hostComp, "hosts.toml")
err = os.MkdirAll(filepath.Dir(fp), 0o755)
if err != nil {
return err
}
err = os.WriteFile(fp, []byte(templatedHosts), 0o644)
if err != nil {
return err
}
log.Info("added Containerd mirror configuration", "registry", mirroredRegistry.String(), "path", fp)
}
return nil
}
func CleanupMirrorConfiguration(ctx context.Context, configPath string) error {
log := logr.FromContextOrDiscard(ctx)
backupDirPath := path.Join(configPath, backupDir)
ok, err := dirExists(backupDirPath)
if err != nil {
return err
}
if !ok {
log.Info("skipping cleanup because backup directory does not exist")
return nil
}
err = clearConfig(configPath)
if err != nil {
return err
}
files, err := os.ReadDir(backupDirPath)
if err != nil {
return err
}
for _, fi := range files {
oldPath := path.Join(backupDirPath, fi.Name())
newPath := path.Join(configPath, fi.Name())
err := os.Rename(oldPath, newPath)
if err != nil {
return err
}
log.Info("recovering Containerd host configuration", "path", oldPath)
}
err = os.RemoveAll(backupDirPath)
if err != nil {
return err
}
return nil
}
func validateRegistries(urls []url.URL) error {
errs := []error{}
for _, u := range urls {
if u.Scheme != "http" && u.Scheme != "https" {
errs = append(errs, fmt.Errorf("invalid registry url scheme must be http or https: %s", u.String()))
}
if u.Path != "" {
errs = append(errs, fmt.Errorf("invalid registry url path has to be empty: %s", u.String()))
}
if len(u.Query()) != 0 {
errs = append(errs, fmt.Errorf("invalid registry url query has to be empty: %s", u.String()))
}
if u.User != nil {
errs = append(errs, fmt.Errorf("invalid registry url user has to be empty: %s", u.String()))
}
}
return errors.Join(errs...)
}
func backupConfig(log logr.Logger, configPath string) error {
backupDirPath := path.Join(configPath, backupDir)
ok, err := dirExists(backupDirPath)
if err != nil {
return err
}
if ok {
return nil
}
files, err := os.ReadDir(configPath)
if err != nil {
return err
}
err = os.MkdirAll(backupDirPath, 0o755)
if err != nil {
return err
}
for _, fi := range files {
oldPath := path.Join(configPath, fi.Name())
newPath := path.Join(backupDirPath, fi.Name())
err := os.Rename(oldPath, newPath)
if err != nil {
return err
}
log.Info("backing up Containerd host configuration", "path", oldPath)
}
return nil
}
func clearConfig(configPath string) error {
files, err := os.ReadDir(configPath)
if err != nil {
return err
}
for _, fi := range files {
if fi.Name() == backupDir {
continue
}
filePath := path.Join(configPath, fi.Name())
err := os.RemoveAll(filePath)
if err != nil {
return err
}
}
return nil
}
func templateHosts(mirroredRegistry url.URL, mirrorTargets []url.URL, capabilities []string, username, password string) (string, error) {
server := mirroredRegistry.String()
if mirroredRegistry.String() == "https://docker.io" {
server = "https://registry-1.docker.io"
}
authorization := ""
if username != "" || password != "" {
authorization = username + ":" + password
authorization = base64.StdEncoding.EncodeToString([]byte(authorization))
authorization = "Basic " + authorization
}
hc := struct {
Authorization string
Server string
Capabilities string
MirrorTargets []url.URL
}{
Server: server,
Capabilities: fmt.Sprintf("['%s']", strings.Join(capabilities, "', '")),
MirrorTargets: mirrorTargets,
Authorization: authorization,
}
tmpl, err := template.New("").Parse(`{{- with .Server }}server = '{{ . }}'{{ end }}
{{- $authorization := .Authorization }}
{{ range .MirrorTargets }}
[host.'{{ .String }}']
capabilities = {{ $.Capabilities }}
{{- if $authorization }}
[host.'{{ .String }}'.header]
Authorization = '{{ $authorization }}'
{{- end }}
{{ end }}`)
if err != nil {
return "", err
}
buf := bytes.NewBuffer(nil)
err = tmpl.Execute(buf, hc)
if err != nil {
return "", err
}
return strings.TrimSpace(buf.String()), nil
}
type hostFile struct {
Hosts map[string]any `toml:"host"`
}
func existingHosts(configPath string, mirroredRegistry url.URL) (string, error) {
fp := path.Join(configPath, backupDir, mirroredRegistry.Host, "hosts.toml")
b, err := os.ReadFile(fp)
if errors.Is(err, os.ErrNotExist) {
return "", nil
}
if err != nil {
return "", err
}
var hf hostFile
err = toml.Unmarshal(b, &hf)
if err != nil {
return "", err
}
if len(hf.Hosts) == 0 {
return "", nil
}
hosts := []string{}
parser := tomlu.Parser{}
parser.Reset(b)
for parser.NextExpression() {
err := parser.Error()
if err != nil {
return "", err
}
e := parser.Expression()
if e.Kind != tomlu.Table {
continue
}
ki := e.Key()
if ki.Next() && string(ki.Node().Data) == "host" && ki.Next() && ki.IsLast() {
hosts = append(hosts, string(ki.Node().Data))
}
}
ehs := []string{}
for _, h := range hosts {
data := hostFile{
Hosts: map[string]any{
h: hf.Hosts[h],
},
}
b, err := toml.Marshal(data)
if err != nil {
return "", err
}
eh := strings.TrimPrefix(string(b), "[host]\n")
ehs = append(ehs, eh)
}
return strings.TrimSpace(strings.Join(ehs, "\n")), nil
}
func dirExists(path string) (bool, error) {
info, err := os.Stat(path)
if err != nil {
if os.IsNotExist(err) {
return false, nil
}
return false, err
}
return info.IsDir(), nil
}