package plugin // import "github.com/docker/docker/plugin" import ( "archive/tar" "bytes" "compress/gzip" "context" "encoding/json" "io" "net/http" "os" "path" "path/filepath" "strings" "time" "github.com/containerd/containerd/content" "github.com/containerd/containerd/images" "github.com/containerd/containerd/platforms" "github.com/containerd/containerd/remotes" "github.com/containerd/containerd/remotes/docker" "github.com/docker/distribution/manifest/schema2" "github.com/docker/distribution/reference" "github.com/docker/docker/api/types" "github.com/docker/docker/api/types/filters" "github.com/docker/docker/api/types/registry" "github.com/docker/docker/dockerversion" "github.com/docker/docker/errdefs" "github.com/docker/docker/pkg/authorization" "github.com/docker/docker/pkg/chrootarchive" "github.com/docker/docker/pkg/containerfs" "github.com/docker/docker/pkg/pools" "github.com/docker/docker/pkg/progress" "github.com/docker/docker/pkg/stringid" v2 "github.com/docker/docker/plugin/v2" "github.com/moby/sys/mount" "github.com/opencontainers/go-digest" specs "github.com/opencontainers/image-spec/specs-go/v1" "github.com/pkg/errors" "github.com/sirupsen/logrus" ) var acceptedPluginFilterTags = map[string]bool{ "enabled": true, "capability": true, } // Disable deactivates a plugin. This means resources (volumes, networks) cant use them. func (pm *Manager) Disable(refOrID string, config *types.PluginDisableConfig) error { p, err := pm.config.Store.GetV2Plugin(refOrID) if err != nil { return err } pm.mu.RLock() c := pm.cMap[p] pm.mu.RUnlock() if !config.ForceDisable && p.GetRefCount() > 0 { return errors.WithStack(inUseError(p.Name())) } for _, typ := range p.GetTypes() { if typ.Capability == authorization.AuthZApiImplements { pm.config.AuthzMiddleware.RemovePlugin(p.Name()) } } if err := pm.disable(p, c); err != nil { return err } pm.publisher.Publish(EventDisable{Plugin: p.PluginObj}) pm.config.LogPluginEvent(p.GetID(), refOrID, "disable") return nil } // Enable activates a plugin, which implies that they are ready to be used by containers. func (pm *Manager) Enable(refOrID string, config *types.PluginEnableConfig) error { p, err := pm.config.Store.GetV2Plugin(refOrID) if err != nil { return err } c := &controller{timeoutInSecs: config.Timeout} if err := pm.enable(p, c, false); err != nil { return err } pm.publisher.Publish(EventEnable{Plugin: p.PluginObj}) pm.config.LogPluginEvent(p.GetID(), refOrID, "enable") return nil } // Inspect examines a plugin config func (pm *Manager) Inspect(refOrID string) (tp *types.Plugin, err error) { p, err := pm.config.Store.GetV2Plugin(refOrID) if err != nil { return nil, err } return &p.PluginObj, nil } func computePrivileges(c types.PluginConfig) types.PluginPrivileges { var privileges types.PluginPrivileges if c.Network.Type != "null" && c.Network.Type != "bridge" && c.Network.Type != "" { privileges = append(privileges, types.PluginPrivilege{ Name: "network", Description: "permissions to access a network", Value: []string{c.Network.Type}, }) } if c.IpcHost { privileges = append(privileges, types.PluginPrivilege{ Name: "host ipc namespace", Description: "allow access to host ipc namespace", Value: []string{"true"}, }) } if c.PidHost { privileges = append(privileges, types.PluginPrivilege{ Name: "host pid namespace", Description: "allow access to host pid namespace", Value: []string{"true"}, }) } for _, mnt := range c.Mounts { if mnt.Source != nil { privileges = append(privileges, types.PluginPrivilege{ Name: "mount", Description: "host path to mount", Value: []string{*mnt.Source}, }) } } for _, device := range c.Linux.Devices { if device.Path != nil { privileges = append(privileges, types.PluginPrivilege{ Name: "device", Description: "host device to access", Value: []string{*device.Path}, }) } } if c.Linux.AllowAllDevices { privileges = append(privileges, types.PluginPrivilege{ Name: "allow-all-devices", Description: "allow 'rwm' access to all devices", Value: []string{"true"}, }) } if len(c.Linux.Capabilities) > 0 { privileges = append(privileges, types.PluginPrivilege{ Name: "capabilities", Description: "list of additional capabilities required", Value: c.Linux.Capabilities, }) } return privileges } // Privileges pulls a plugin config and computes the privileges required to install it. func (pm *Manager) Privileges(ctx context.Context, ref reference.Named, metaHeader http.Header, authConfig *registry.AuthConfig) (types.PluginPrivileges, error) { var ( config types.PluginConfig configSeen bool ) h := func(ctx context.Context, desc specs.Descriptor) ([]specs.Descriptor, error) { switch desc.MediaType { case schema2.MediaTypeManifest, specs.MediaTypeImageManifest: data, err := content.ReadBlob(ctx, pm.blobStore, desc) if err != nil { return nil, errors.Wrapf(err, "error reading image manifest from blob store for %s", ref) } var m specs.Manifest if err := json.Unmarshal(data, &m); err != nil { return nil, errors.Wrapf(err, "error unmarshaling image manifest for %s", ref) } return []specs.Descriptor{m.Config}, nil case schema2.MediaTypePluginConfig: configSeen = true data, err := content.ReadBlob(ctx, pm.blobStore, desc) if err != nil { return nil, errors.Wrapf(err, "error reading plugin config from blob store for %s", ref) } if err := json.Unmarshal(data, &config); err != nil { return nil, errors.Wrapf(err, "error unmarshaling plugin config for %s", ref) } } return nil, nil } if err := pm.fetch(ctx, ref, authConfig, progress.DiscardOutput(), metaHeader, images.HandlerFunc(h)); err != nil { return types.PluginPrivileges{}, nil } if !configSeen { return types.PluginPrivileges{}, errors.Errorf("did not find plugin config for specified reference %s", ref) } return computePrivileges(config), nil } // Upgrade upgrades a plugin // // TODO: replace reference package usage with simpler url.Parse semantics func (pm *Manager) Upgrade(ctx context.Context, ref reference.Named, name string, metaHeader http.Header, authConfig *registry.AuthConfig, privileges types.PluginPrivileges, outStream io.Writer) (err error) { p, err := pm.config.Store.GetV2Plugin(name) if err != nil { return err } if p.IsEnabled() { return errors.Wrap(enabledError(p.Name()), "plugin must be disabled before upgrading") } // revalidate because Pull is public if _, err := reference.ParseNormalizedNamed(name); err != nil { return errors.Wrapf(errdefs.InvalidParameter(err), "failed to parse %q", name) } pm.muGC.RLock() defer pm.muGC.RUnlock() tmpRootFSDir, err := os.MkdirTemp(pm.tmpDir(), ".rootfs") if err != nil { return errors.Wrap(err, "error creating tmp dir for plugin rootfs") } var md fetchMeta ctx, cancel := context.WithCancel(ctx) out, waitProgress := setupProgressOutput(outStream, cancel) defer waitProgress() if err := pm.fetch(ctx, ref, authConfig, out, metaHeader, storeFetchMetadata(&md), childrenHandler(pm.blobStore), applyLayer(pm.blobStore, tmpRootFSDir, out)); err != nil { return err } pm.config.LogPluginEvent(reference.FamiliarString(ref), name, "pull") if err := validateFetchedMetadata(md); err != nil { return err } if err := pm.upgradePlugin(p, md.config, md.manifest, md.blobs, tmpRootFSDir, &privileges); err != nil { return err } p.PluginObj.PluginReference = ref.String() return nil } // Pull pulls a plugin, check if the correct privileges are provided and install the plugin. // // TODO: replace reference package usage with simpler url.Parse semantics func (pm *Manager) Pull(ctx context.Context, ref reference.Named, name string, metaHeader http.Header, authConfig *registry.AuthConfig, privileges types.PluginPrivileges, outStream io.Writer, opts ...CreateOpt) (err error) { pm.muGC.RLock() defer pm.muGC.RUnlock() // revalidate because Pull is public nameref, err := reference.ParseNormalizedNamed(name) if err != nil { return errors.Wrapf(errdefs.InvalidParameter(err), "failed to parse %q", name) } name = reference.FamiliarString(reference.TagNameOnly(nameref)) if err := pm.config.Store.validateName(name); err != nil { return errdefs.InvalidParameter(err) } tmpRootFSDir, err := os.MkdirTemp(pm.tmpDir(), ".rootfs") if err != nil { return errors.Wrap(errdefs.System(err), "error preparing upgrade") } defer os.RemoveAll(tmpRootFSDir) var md fetchMeta ctx, cancel := context.WithCancel(ctx) out, waitProgress := setupProgressOutput(outStream, cancel) defer waitProgress() if err := pm.fetch(ctx, ref, authConfig, out, metaHeader, storeFetchMetadata(&md), childrenHandler(pm.blobStore), applyLayer(pm.blobStore, tmpRootFSDir, out)); err != nil { return err } pm.config.LogPluginEvent(reference.FamiliarString(ref), name, "pull") if err := validateFetchedMetadata(md); err != nil { return err } refOpt := func(p *v2.Plugin) { p.PluginObj.PluginReference = ref.String() } optsList := make([]CreateOpt, 0, len(opts)+1) optsList = append(optsList, opts...) optsList = append(optsList, refOpt) // TODO: tmpRootFSDir is empty but should have layers in it p, err := pm.createPlugin(name, md.config, md.manifest, md.blobs, tmpRootFSDir, &privileges, optsList...) if err != nil { return err } pm.publisher.Publish(EventCreate{Plugin: p.PluginObj}) return nil } // List displays the list of plugins and associated metadata. func (pm *Manager) List(pluginFilters filters.Args) ([]types.Plugin, error) { if err := pluginFilters.Validate(acceptedPluginFilterTags); err != nil { return nil, err } enabledOnly := false disabledOnly := false if pluginFilters.Contains("enabled") { if pluginFilters.ExactMatch("enabled", "true") { enabledOnly = true } else if pluginFilters.ExactMatch("enabled", "false") { disabledOnly = true } else { return nil, invalidFilter{"enabled", pluginFilters.Get("enabled")} } } plugins := pm.config.Store.GetAll() out := make([]types.Plugin, 0, len(plugins)) next: for _, p := range plugins { if enabledOnly && !p.PluginObj.Enabled { continue } if disabledOnly && p.PluginObj.Enabled { continue } if pluginFilters.Contains("capability") { for _, f := range p.GetTypes() { if !pluginFilters.Match("capability", f.Capability) { continue next } } } out = append(out, p.PluginObj) } return out, nil } // Push pushes a plugin to the registry. func (pm *Manager) Push(ctx context.Context, name string, metaHeader http.Header, authConfig *registry.AuthConfig, outStream io.Writer) error { p, err := pm.config.Store.GetV2Plugin(name) if err != nil { return err } ref, err := reference.ParseNormalizedNamed(p.Name()) if err != nil { return errors.Wrapf(err, "plugin has invalid name %v for push", p.Name()) } statusTracker := docker.NewInMemoryTracker() resolver, err := pm.newResolver(ctx, statusTracker, authConfig, metaHeader, false) if err != nil { return err } pusher, err := resolver.Pusher(ctx, ref.String()) if err != nil { return errors.Wrap(err, "error creating plugin pusher") } pj := newPushJobs(statusTracker) ctx, cancel := context.WithCancel(ctx) out, waitProgress := setupProgressOutput(outStream, cancel) defer waitProgress() progressHandler := images.HandlerFunc(func(ctx context.Context, desc specs.Descriptor) ([]specs.Descriptor, error) { logrus.WithField("mediaType", desc.MediaType).WithField("digest", desc.Digest.String()).Debug("Preparing to push plugin layer") id := stringid.TruncateID(desc.Digest.String()) pj.add(remotes.MakeRefKey(ctx, desc), id) progress.Update(out, id, "Preparing") return nil, nil }) desc, err := pm.getManifestDescriptor(ctx, p) if err != nil { return errors.Wrap(err, "error reading plugin manifest") } progress.Messagef(out, "", "The push refers to repository [%s]", reference.FamiliarName(ref)) // TODO: If a layer already exists on the registry, the progress output just says "Preparing" go func() { timer := time.NewTimer(100 * time.Millisecond) defer timer.Stop() if !timer.Stop() { <-timer.C } var statuses []contentStatus for { timer.Reset(100 * time.Millisecond) select { case <-ctx.Done(): return case <-timer.C: statuses = pj.status() } for _, s := range statuses { out.WriteProgress(progress.Progress{ID: s.Ref, Current: s.Offset, Total: s.Total, Action: s.Status, LastUpdate: s.Offset == s.Total}) } } }() // Make sure we can authenticate the request since the auth scope for plugin repos is different than a normal repo. ctx = docker.WithScope(ctx, scope(ref, true)) if err := remotes.PushContent(ctx, pusher, desc, pm.blobStore, nil, nil, func(h images.Handler) images.Handler { return images.Handlers(progressHandler, h) }); err != nil { // Try fallback to http. // This is needed because the containerd pusher will only attempt the first registry config we pass, which would // typically be https. // If there are no http-only host configs found we'll error out anyway. resolver, _ := pm.newResolver(ctx, statusTracker, authConfig, metaHeader, true) if resolver != nil { pusher, _ := resolver.Pusher(ctx, ref.String()) if pusher != nil { logrus.WithField("ref", ref).Debug("Re-attmpting push with http-fallback") err2 := remotes.PushContent(ctx, pusher, desc, pm.blobStore, nil, nil, func(h images.Handler) images.Handler { return images.Handlers(progressHandler, h) }) if err2 == nil { err = nil } else { logrus.WithError(err2).WithField("ref", ref).Debug("Error while attempting push with http-fallback") } } } if err != nil { return errors.Wrap(err, "error pushing plugin") } } // For blobs that already exist in the registry we need to make sure to update the progress otherwise it will just say "pending" // TODO: How to check if the layer already exists? Is it worth it? for _, j := range pj.jobs { progress.Update(out, pj.names[j], "Upload complete") } // Signal the client for content trust verification progress.Aux(out, types.PushResult{Tag: ref.(reference.Tagged).Tag(), Digest: desc.Digest.String(), Size: int(desc.Size)}) return nil } // manifest wraps an OCI manifest, because... // Historically the registry does not support plugins unless the media type on the manifest is specifically schema2.MediaTypeManifest // So the OCI manifest media type is not supported. // Additionally, there is extra validation for the docker schema2 manifest than there is a mediatype set on the manifest itself // even though this is set on the descriptor // The OCI types do not have this field. type manifest struct { specs.Manifest MediaType string `json:"mediaType,omitempty"` } func buildManifest(ctx context.Context, s content.Manager, config digest.Digest, layers []digest.Digest) (manifest, error) { var m manifest m.MediaType = images.MediaTypeDockerSchema2Manifest m.SchemaVersion = 2 configInfo, err := s.Info(ctx, config) if err != nil { return m, errors.Wrapf(err, "error reading plugin config content for digest %s", config) } m.Config = specs.Descriptor{ MediaType: mediaTypePluginConfig, Size: configInfo.Size, Digest: configInfo.Digest, } for _, l := range layers { info, err := s.Info(ctx, l) if err != nil { return m, errors.Wrapf(err, "error fetching info for content digest %s", l) } m.Layers = append(m.Layers, specs.Descriptor{ MediaType: images.MediaTypeDockerSchema2LayerGzip, // TODO: This is assuming everything is a gzip compressed layer, but that may not be true. Digest: l, Size: info.Size, }) } return m, nil } // getManifestDescriptor gets the OCI descriptor for a manifest // It will generate a manifest if one does not exist func (pm *Manager) getManifestDescriptor(ctx context.Context, p *v2.Plugin) (specs.Descriptor, error) { logger := logrus.WithField("plugin", p.Name()).WithField("digest", p.Manifest) if p.Manifest != "" { info, err := pm.blobStore.Info(ctx, p.Manifest) if err == nil { desc := specs.Descriptor{ Size: info.Size, Digest: info.Digest, MediaType: images.MediaTypeDockerSchema2Manifest, } return desc, nil } logger.WithError(err).Debug("Could not find plugin manifest in content store") } else { logger.Info("Plugin does not have manifest digest") } logger.Info("Building a new plugin manifest") manifest, err := buildManifest(ctx, pm.blobStore, p.Config, p.Blobsums) if err != nil { return specs.Descriptor{}, err } desc, err := writeManifest(ctx, pm.blobStore, &manifest) if err != nil { return desc, err } if err := pm.save(p); err != nil { logger.WithError(err).Error("Could not save plugin with manifest digest") } return desc, nil } func writeManifest(ctx context.Context, cs content.Store, m *manifest) (specs.Descriptor, error) { platform := platforms.DefaultSpec() desc := specs.Descriptor{ MediaType: images.MediaTypeDockerSchema2Manifest, Platform: &platform, } data, err := json.Marshal(m) if err != nil { return desc, errors.Wrap(err, "error encoding manifest") } desc.Digest = digest.FromBytes(data) desc.Size = int64(len(data)) if err := content.WriteBlob(ctx, cs, remotes.MakeRefKey(ctx, desc), bytes.NewReader(data), desc); err != nil { return desc, errors.Wrap(err, "error writing plugin manifest") } return desc, nil } // Remove deletes plugin's root directory. func (pm *Manager) Remove(name string, config *types.PluginRmConfig) error { p, err := pm.config.Store.GetV2Plugin(name) pm.mu.RLock() c := pm.cMap[p] pm.mu.RUnlock() if err != nil { return err } if !config.ForceRemove { if p.GetRefCount() > 0 { return inUseError(p.Name()) } if p.IsEnabled() { return enabledError(p.Name()) } } if p.IsEnabled() { if err := pm.disable(p, c); err != nil { logrus.Errorf("failed to disable plugin '%s': %s", p.Name(), err) } } defer func() { go pm.GC() }() id := p.GetID() pluginDir := filepath.Join(pm.config.Root, id) if err := mount.RecursiveUnmount(pluginDir); err != nil { return errors.Wrap(err, "error unmounting plugin data") } if err := atomicRemoveAll(pluginDir); err != nil { return err } pm.config.Store.Remove(p) pm.config.LogPluginEvent(id, name, "remove") pm.publisher.Publish(EventRemove{Plugin: p.PluginObj}) return nil } // Set sets plugin args func (pm *Manager) Set(name string, args []string) error { p, err := pm.config.Store.GetV2Plugin(name) if err != nil { return err } if err := p.Set(args); err != nil { return err } return pm.save(p) } // CreateFromContext creates a plugin from the given pluginDir which contains // both the rootfs and the config.json and a repoName with optional tag. func (pm *Manager) CreateFromContext(ctx context.Context, tarCtx io.ReadCloser, options *types.PluginCreateOptions) (err error) { pm.muGC.RLock() defer pm.muGC.RUnlock() ref, err := reference.ParseNormalizedNamed(options.RepoName) if err != nil { return errors.Wrapf(err, "failed to parse reference %v", options.RepoName) } if _, ok := ref.(reference.Canonical); ok { return errors.Errorf("canonical references are not permitted") } name := reference.FamiliarString(reference.TagNameOnly(ref)) if err := pm.config.Store.validateName(name); err != nil { // fast check, real check is in createPlugin() return err } tmpRootFSDir, err := os.MkdirTemp(pm.tmpDir(), ".rootfs") if err != nil { return errors.Wrap(err, "failed to create temp directory") } defer os.RemoveAll(tmpRootFSDir) var configJSON []byte rootFS := splitConfigRootFSFromTar(tarCtx, &configJSON) rootFSBlob, err := pm.blobStore.Writer(ctx, content.WithRef(name)) if err != nil { return err } defer rootFSBlob.Close() gzw := gzip.NewWriter(rootFSBlob) rootFSReader := io.TeeReader(rootFS, gzw) if err := chrootarchive.Untar(rootFSReader, tmpRootFSDir, nil); err != nil { return err } if err := rootFS.Close(); err != nil { return err } if configJSON == nil { return errors.New("config not found") } if err := gzw.Close(); err != nil { return errors.Wrap(err, "error closing gzip writer") } var config types.PluginConfig if err := json.Unmarshal(configJSON, &config); err != nil { return errors.Wrap(err, "failed to parse config") } if err := pm.validateConfig(config); err != nil { return err } pm.mu.Lock() defer pm.mu.Unlock() if err := rootFSBlob.Commit(ctx, 0, ""); err != nil { return err } defer func() { if err != nil { go pm.GC() } }() config.Rootfs = &types.PluginConfigRootfs{ Type: "layers", DiffIds: []string{rootFSBlob.Digest().String()}, } config.DockerVersion = dockerversion.Version configBlob, err := pm.blobStore.Writer(ctx, content.WithRef(name+"-config.json")) if err != nil { return err } defer configBlob.Close() if err := json.NewEncoder(configBlob).Encode(config); err != nil { return errors.Wrap(err, "error encoding json config") } if err := configBlob.Commit(ctx, 0, ""); err != nil { return err } configDigest := configBlob.Digest() layers := []digest.Digest{rootFSBlob.Digest()} manifest, err := buildManifest(ctx, pm.blobStore, configDigest, layers) if err != nil { return err } desc, err := writeManifest(ctx, pm.blobStore, &manifest) if err != nil { return } p, err := pm.createPlugin(name, configDigest, desc.Digest, layers, tmpRootFSDir, nil) if err != nil { return err } p.PluginObj.PluginReference = name pm.publisher.Publish(EventCreate{Plugin: p.PluginObj}) pm.config.LogPluginEvent(p.PluginObj.ID, name, "create") return nil } func (pm *Manager) validateConfig(config types.PluginConfig) error { return nil // TODO: } func splitConfigRootFSFromTar(in io.ReadCloser, config *[]byte) io.ReadCloser { pr, pw := io.Pipe() go func() { tarReader := tar.NewReader(in) tarWriter := tar.NewWriter(pw) defer in.Close() hasRootFS := false for { hdr, err := tarReader.Next() if err == io.EOF { if !hasRootFS { pw.CloseWithError(errors.Wrap(err, "no rootfs found")) return } // Signals end of archive. tarWriter.Close() pw.Close() return } if err != nil { pw.CloseWithError(errors.Wrap(err, "failed to read from tar")) return } content := io.Reader(tarReader) name := path.Clean(hdr.Name) if path.IsAbs(name) { name = name[1:] } if name == configFileName { dt, err := io.ReadAll(content) if err != nil { pw.CloseWithError(errors.Wrapf(err, "failed to read %s", configFileName)) return } *config = dt } if parts := strings.Split(name, "/"); len(parts) != 0 && parts[0] == rootFSFileName { hdr.Name = path.Clean(path.Join(parts[1:]...)) if hdr.Typeflag == tar.TypeLink && strings.HasPrefix(strings.ToLower(hdr.Linkname), rootFSFileName+"/") { hdr.Linkname = hdr.Linkname[len(rootFSFileName)+1:] } if err := tarWriter.WriteHeader(hdr); err != nil { pw.CloseWithError(errors.Wrap(err, "error writing tar header")) return } if _, err := pools.Copy(tarWriter, content); err != nil { pw.CloseWithError(errors.Wrap(err, "error copying tar data")) return } hasRootFS = true } else { io.Copy(io.Discard, content) } } }() return pr } func atomicRemoveAll(dir string) error { renamed := dir + "-removing" err := os.Rename(dir, renamed) switch { case os.IsNotExist(err), err == nil: // even if `dir` doesn't exist, we can still try and remove `renamed` case os.IsExist(err): // Some previous remove failed, check if the origin dir exists if e := containerfs.EnsureRemoveAll(renamed); e != nil { return errors.Wrap(err, "rename target already exists and could not be removed") } if _, err := os.Stat(dir); os.IsNotExist(err) { // origin doesn't exist, nothing left to do return nil } // attempt to rename again if err := os.Rename(dir, renamed); err != nil { return errors.Wrap(err, "failed to rename dir for atomic removal") } default: return errors.Wrap(err, "failed to rename dir for atomic removal") } if err := containerfs.EnsureRemoveAll(renamed); err != nil { os.Rename(renamed, dir) return err } return nil }