plugin/backend_linux.go

   1 // +build linux
   2
   3 package plugin
   4
   5 import (
   6         "archive/tar"
   7         "compress/gzip"
   8         "encoding/json"
   9         "fmt"
  10         "io"
  11         "io/ioutil"
  12         "net/http"
  13         "os"
  14         "path"
  15         "path/filepath"
  16         "sort"
  17         "strings"
  18
  19         "github.com/Sirupsen/logrus"
  20         "github.com/docker/distribution/manifest/schema2"
  21         "github.com/docker/distribution/reference"
  22         "github.com/docker/docker/api/types"
  23         "github.com/docker/docker/api/types/filters"
  24         "github.com/docker/docker/distribution"
  25         progressutils "github.com/docker/docker/distribution/utils"
  26         "github.com/docker/docker/distribution/xfer"
  27         "github.com/docker/docker/dockerversion"
  28         "github.com/docker/docker/image"
  29         "github.com/docker/docker/layer"
  30         "github.com/docker/docker/pkg/authorization"
  31         "github.com/docker/docker/pkg/chrootarchive"
  32         "github.com/docker/docker/pkg/ioutils"
  33         "github.com/docker/docker/pkg/mount"
  34         "github.com/docker/docker/pkg/pools"
  35         "github.com/docker/docker/pkg/progress"
  36         "github.com/docker/docker/plugin/v2"
  37         refstore "github.com/docker/docker/reference"
  38         "github.com/opencontainers/go-digest"
  39         "github.com/pkg/errors"
  40         "golang.org/x/net/context"
  41 )
  42
  43 var acceptedPluginFilterTags = map[string]bool{
  44         "enabled":    true,
  45         "capability": true,
  46 }
  47
  48 // Disable deactivates a plugin. This means resources (volumes, networks) cant use them.
  49 func (pm *Manager) Disable(refOrID string, config *types.PluginDisableConfig) error {
  50         p, err := pm.config.Store.GetV2Plugin(refOrID)
  51         if err != nil {
  52                 return err
  53         }
  54         pm.mu.RLock()
  55         c := pm.cMap[p]
  56         pm.mu.RUnlock()
  57
  58         if !config.ForceDisable && p.GetRefCount() > 0 {
  59                 return fmt.Errorf("plugin %s is in use", p.Name())
  60         }
  61
  62         for _, typ := range p.GetTypes() {
  63                 if typ.Capability == authorization.AuthZApiImplements {
  64                         pm.config.AuthzMiddleware.RemovePlugin(p.Name())
  65                 }
  66         }
  67
  68         if err := pm.disable(p, c); err != nil {
  69                 return err
  70         }
  71         pm.config.LogPluginEvent(p.GetID(), refOrID, "disable")
  72         return nil
  73 }
  74
  75 // Enable activates a plugin, which implies that they are ready to be used by containers.
  76 func (pm *Manager) Enable(refOrID string, config *types.PluginEnableConfig) error {
  77         p, err := pm.config.Store.GetV2Plugin(refOrID)
  78         if err != nil {
  79                 return err
  80         }
  81
  82         c := &controller{timeoutInSecs: config.Timeout}
  83         if err := pm.enable(p, c, false); err != nil {
  84                 return err
  85         }
  86         pm.config.LogPluginEvent(p.GetID(), refOrID, "enable")
  87         return nil
  88 }
  89
  90 // Inspect examines a plugin config
  91 func (pm *Manager) Inspect(refOrID string) (tp *types.Plugin, err error) {
  92         p, err := pm.config.Store.GetV2Plugin(refOrID)
  93         if err != nil {
  94                 return nil, err
  95         }
  96
  97         return &p.PluginObj, nil
  98 }
  99
 100 func (pm *Manager) pull(ctx context.Context, ref reference.Named, config *distribution.ImagePullConfig, outStream io.Writer) error {
 101         if outStream != nil {
 102                 // Include a buffer so that slow client connections don't affect
 103                 // transfer performance.
 104                 progressChan := make(chan progress.Progress, 100)
 105
 106                 writesDone := make(chan struct{})
 107
 108                 defer func() {
 109                         close(progressChan)
 110                         <-writesDone
 111                 }()
 112
 113                 var cancelFunc context.CancelFunc
 114                 ctx, cancelFunc = context.WithCancel(ctx)
 115
 116                 go func() {
 117                         progressutils.WriteDistributionProgress(cancelFunc, outStream, progressChan)
 118                         close(writesDone)
 119                 }()
 120
 121                 config.ProgressOutput = progress.ChanOutput(progressChan)
 122         } else {
 123                 config.ProgressOutput = progress.DiscardOutput()
 124         }
 125         return distribution.Pull(ctx, ref, config)
 126 }
 127
 128 type tempConfigStore struct {
 129         config       []byte
 130         configDigest digest.Digest
 131 }
 132
 133 func (s *tempConfigStore) Put(c []byte) (digest.Digest, error) {
 134         dgst := digest.FromBytes(c)
 135
 136         s.config = c
 137         s.configDigest = dgst
 138
 139         return dgst, nil
 140 }
 141
 142 func (s *tempConfigStore) Get(d digest.Digest) ([]byte, error) {
 143         if d != s.configDigest {
 144                 return nil, fmt.Errorf("digest not found")
 145         }
 146         return s.config, nil
 147 }
 148
 149 func (s *tempConfigStore) GetTarSeekStream(d digest.Digest) (ioutils.ReadSeekCloser, error) {
 150         return nil, fmt.Errorf("unimplemented")
 151 }
 152
 153 func (s *tempConfigStore) RootFSAndPlatformFromConfig(c []byte) (*image.RootFS, layer.Platform, error) {
 154         return configToRootFS(c)
 155 }
 156
 157 func computePrivileges(c types.PluginConfig) (types.PluginPrivileges, error) {
 158         var privileges types.PluginPrivileges
 159         if c.Network.Type != "null" && c.Network.Type != "bridge" && c.Network.Type != "" {
 160                 privileges = append(privileges, types.PluginPrivilege{
 161                         Name:        "network",
 162                         Description: "permissions to access a network",
 163                         Value:       []string{c.Network.Type},
 164                 })
 165         }
 166         if c.IpcHost {
 167                 privileges = append(privileges, types.PluginPrivilege{
 168                         Name:        "host ipc namespace",
 169                         Description: "allow access to host ipc namespace",
 170                         Value:       []string{"true"},
 171                 })
 172         }
 173         if c.PidHost {
 174                 privileges = append(privileges, types.PluginPrivilege{
 175                         Name:        "host pid namespace",
 176                         Description: "allow access to host pid namespace",
 177                         Value:       []string{"true"},
 178                 })
 179         }
 180         for _, mount := range c.Mounts {
 181                 if mount.Source != nil {
 182                         privileges = append(privileges, types.PluginPrivilege{
 183                                 Name:        "mount",
 184                                 Description: "host path to mount",
 185                                 Value:       []string{*mount.Source},
 186                         })
 187                 }
 188         }
 189         for _, device := range c.Linux.Devices {
 190                 if device.Path != nil {
 191                         privileges = append(privileges, types.PluginPrivilege{
 192                                 Name:        "device",
 193                                 Description: "host device to access",
 194                                 Value:       []string{*device.Path},
 195                         })
 196                 }
 197         }
 198         if c.Linux.AllowAllDevices {
 199                 privileges = append(privileges, types.PluginPrivilege{
 200                         Name:        "allow-all-devices",
 201                         Description: "allow 'rwm' access to all devices",
 202                         Value:       []string{"true"},
 203                 })
 204         }
 205         if len(c.Linux.Capabilities) > 0 {
 206                 privileges = append(privileges, types.PluginPrivilege{
 207                         Name:        "capabilities",
 208                         Description: "list of additional capabilities required",
 209                         Value:       c.Linux.Capabilities,
 210                 })
 211         }
 212
 213         return privileges, nil
 214 }
 215
 216 // Privileges pulls a plugin config and computes the privileges required to install it.
 217 func (pm *Manager) Privileges(ctx context.Context, ref reference.Named, metaHeader http.Header, authConfig *types.AuthConfig) (types.PluginPrivileges, error) {
 218         // create image store instance
 219         cs := &tempConfigStore{}
 220
 221         // DownloadManager not defined because only pulling configuration.
 222         pluginPullConfig := &distribution.ImagePullConfig{
 223                 Config: distribution.Config{
 224                         MetaHeaders:      metaHeader,
 225                         AuthConfig:       authConfig,
 226                         RegistryService:  pm.config.RegistryService,
 227                         ImageEventLogger: func(string, string, string) {},
 228                         ImageStore:       cs,
 229                 },
 230                 Schema2Types: distribution.PluginTypes,
 231         }
 232
 233         if err := pm.pull(ctx, ref, pluginPullConfig, nil); err != nil {
 234                 return nil, err
 235         }
 236
 237         if cs.config == nil {
 238                 return nil, errors.New("no configuration pulled")
 239         }
 240         var config types.PluginConfig
 241         if err := json.Unmarshal(cs.config, &config); err != nil {
 242                 return nil, err
 243         }
 244
 245         return computePrivileges(config)
 246 }
 247
 248 // Upgrade upgrades a plugin
 249 func (pm *Manager) Upgrade(ctx context.Context, ref reference.Named, name string, metaHeader http.Header, authConfig *types.AuthConfig, privileges types.PluginPrivileges, outStream io.Writer) (err error) {
 250         p, err := pm.config.Store.GetV2Plugin(name)
 251         if err != nil {
 252                 return errors.Wrap(err, "plugin must be installed before upgrading")
 253         }
 254
 255         if p.IsEnabled() {
 256                 return fmt.Errorf("plugin must be disabled before upgrading")
 257         }
 258
 259         pm.muGC.RLock()
 260         defer pm.muGC.RUnlock()
 261
 262         // revalidate because Pull is public
 263         if _, err := reference.ParseNormalizedNamed(name); err != nil {
 264                 return errors.Wrapf(err, "failed to parse %q", name)
 265         }
 266
 267         tmpRootFSDir, err := ioutil.TempDir(pm.tmpDir(), ".rootfs")
 268         if err != nil {
 269                 return err
 270         }
 271         defer os.RemoveAll(tmpRootFSDir)
 272
 273         dm := &downloadManager{
 274                 tmpDir:    tmpRootFSDir,
 275                 blobStore: pm.blobStore,
 276         }
 277
 278         pluginPullConfig := &distribution.ImagePullConfig{
 279                 Config: distribution.Config{
 280                         MetaHeaders:      metaHeader,
 281                         AuthConfig:       authConfig,
 282                         RegistryService:  pm.config.RegistryService,
 283                         ImageEventLogger: pm.config.LogPluginEvent,
 284                         ImageStore:       dm,
 285                 },
 286                 DownloadManager: dm, // todo: reevaluate if possible to substitute distribution/xfer dependencies instead
 287                 Schema2Types:    distribution.PluginTypes,
 288         }
 289
 290         err = pm.pull(ctx, ref, pluginPullConfig, outStream)
 291         if err != nil {
 292                 go pm.GC()
 293                 return err
 294         }
 295
 296         if err := pm.upgradePlugin(p, dm.configDigest, dm.blobs, tmpRootFSDir, &privileges); err != nil {
 297                 return err
 298         }
 299         p.PluginObj.PluginReference = ref.String()
 300         return nil
 301 }
 302
 303 // Pull pulls a plugin, check if the correct privileges are provided and install the plugin.
 304 func (pm *Manager) Pull(ctx context.Context, ref reference.Named, name string, metaHeader http.Header, authConfig *types.AuthConfig, privileges types.PluginPrivileges, outStream io.Writer) (err error) {
 305         pm.muGC.RLock()
 306         defer pm.muGC.RUnlock()
 307
 308         // revalidate because Pull is public
 309         nameref, err := reference.ParseNormalizedNamed(name)
 310         if err != nil {
 311                 return errors.Wrapf(err, "failed to parse %q", name)
 312         }
 313         name = reference.FamiliarString(reference.TagNameOnly(nameref))
 314
 315         if err := pm.config.Store.validateName(name); err != nil {
 316                 return err
 317         }
 318
 319         tmpRootFSDir, err := ioutil.TempDir(pm.tmpDir(), ".rootfs")
 320         if err != nil {
 321                 return err
 322         }
 323         defer os.RemoveAll(tmpRootFSDir)
 324
 325         dm := &downloadManager{
 326                 tmpDir:    tmpRootFSDir,
 327                 blobStore: pm.blobStore,
 328         }
 329
 330         pluginPullConfig := &distribution.ImagePullConfig{
 331                 Config: distribution.Config{
 332                         MetaHeaders:      metaHeader,
 333                         AuthConfig:       authConfig,
 334                         RegistryService:  pm.config.RegistryService,
 335                         ImageEventLogger: pm.config.LogPluginEvent,
 336                         ImageStore:       dm,
 337                 },
 338                 DownloadManager: dm, // todo: reevaluate if possible to substitute distribution/xfer dependencies instead
 339                 Schema2Types:    distribution.PluginTypes,
 340         }
 341
 342         err = pm.pull(ctx, ref, pluginPullConfig, outStream)
 343         if err != nil {
 344                 go pm.GC()
 345                 return err
 346         }
 347
 348         p, err := pm.createPlugin(name, dm.configDigest, dm.blobs, tmpRootFSDir, &privileges)
 349         if err != nil {
 350                 return err
 351         }
 352         p.PluginObj.PluginReference = ref.String()
 353
 354         return nil
 355 }
 356
 357 // List displays the list of plugins and associated metadata.
 358 func (pm *Manager) List(pluginFilters filters.Args) ([]types.Plugin, error) {
 359         if err := pluginFilters.Validate(acceptedPluginFilterTags); err != nil {
 360                 return nil, err
 361         }
 362
 363         enabledOnly := false
 364         disabledOnly := false
 365         if pluginFilters.Include("enabled") {
 366                 if pluginFilters.ExactMatch("enabled", "true") {
 367                         enabledOnly = true
 368                 } else if pluginFilters.ExactMatch("enabled", "false") {
 369                         disabledOnly = true
 370                 } else {
 371                         return nil, fmt.Errorf("Invalid filter 'enabled=%s'", pluginFilters.Get("enabled"))
 372                 }
 373         }
 374
 375         plugins := pm.config.Store.GetAll()
 376         out := make([]types.Plugin, 0, len(plugins))
 377
 378 next:
 379         for _, p := range plugins {
 380                 if enabledOnly && !p.PluginObj.Enabled {
 381                         continue
 382                 }
 383                 if disabledOnly && p.PluginObj.Enabled {
 384                         continue
 385                 }
 386                 if pluginFilters.Include("capability") {
 387                         for _, f := range p.GetTypes() {
 388                                 if !pluginFilters.Match("capability", f.Capability) {
 389                                         continue next
 390                                 }
 391                         }
 392                 }
 393                 out = append(out, p.PluginObj)
 394         }
 395         return out, nil
 396 }
 397
 398 // Push pushes a plugin to the store.
 399 func (pm *Manager) Push(ctx context.Context, name string, metaHeader http.Header, authConfig *types.AuthConfig, outStream io.Writer) error {
 400         p, err := pm.config.Store.GetV2Plugin(name)
 401         if err != nil {
 402                 return err
 403         }
 404
 405         ref, err := reference.ParseNormalizedNamed(p.Name())
 406         if err != nil {
 407                 return errors.Wrapf(err, "plugin has invalid name %v for push", p.Name())
 408         }
 409
 410         var po progress.Output
 411         if outStream != nil {
 412                 // Include a buffer so that slow client connections don't affect
 413                 // transfer performance.
 414                 progressChan := make(chan progress.Progress, 100)
 415
 416                 writesDone := make(chan struct{})
 417
 418                 defer func() {
 419                         close(progressChan)
 420                         <-writesDone
 421                 }()
 422
 423                 var cancelFunc context.CancelFunc
 424                 ctx, cancelFunc = context.WithCancel(ctx)
 425
 426                 go func() {
 427                         progressutils.WriteDistributionProgress(cancelFunc, outStream, progressChan)
 428                         close(writesDone)
 429                 }()
 430
 431                 po = progress.ChanOutput(progressChan)
 432         } else {
 433                 po = progress.DiscardOutput()
 434         }
 435
 436         // TODO: replace these with manager
 437         is := &pluginConfigStore{
 438                 pm:     pm,
 439                 plugin: p,
 440         }
 441         ls := &pluginLayerProvider{
 442                 pm:     pm,
 443                 plugin: p,
 444         }
 445         rs := &pluginReference{
 446                 name:     ref,
 447                 pluginID: p.Config,
 448         }
 449
 450         uploadManager := xfer.NewLayerUploadManager(3)
 451
 452         imagePushConfig := &distribution.ImagePushConfig{
 453                 Config: distribution.Config{
 454                         MetaHeaders:      metaHeader,
 455                         AuthConfig:       authConfig,
 456                         ProgressOutput:   po,
 457                         RegistryService:  pm.config.RegistryService,
 458                         ReferenceStore:   rs,
 459                         ImageEventLogger: pm.config.LogPluginEvent,
 460                         ImageStore:       is,
 461                         RequireSchema2:   true,
 462                 },
 463                 ConfigMediaType: schema2.MediaTypePluginConfig,
 464                 LayerStore:      ls,
 465                 UploadManager:   uploadManager,
 466         }
 467
 468         return distribution.Push(ctx, ref, imagePushConfig)
 469 }
 470
 471 type pluginReference struct {
 472         name     reference.Named
 473         pluginID digest.Digest
 474 }
 475
 476 func (r *pluginReference) References(id digest.Digest) []reference.Named {
 477         if r.pluginID != id {
 478                 return nil
 479         }
 480         return []reference.Named{r.name}
 481 }
 482
 483 func (r *pluginReference) ReferencesByName(ref reference.Named) []refstore.Association {
 484         return []refstore.Association{
 485                 {
 486                         Ref: r.name,
 487                         ID:  r.pluginID,
 488                 },
 489         }
 490 }
 491
 492 func (r *pluginReference) Get(ref reference.Named) (digest.Digest, error) {
 493         if r.name.String() != ref.String() {
 494                 return digest.Digest(""), refstore.ErrDoesNotExist
 495         }
 496         return r.pluginID, nil
 497 }
 498
 499 func (r *pluginReference) AddTag(ref reference.Named, id digest.Digest, force bool) error {
 500         // Read only, ignore
 501         return nil
 502 }
 503 func (r *pluginReference) AddDigest(ref reference.Canonical, id digest.Digest, force bool) error {
 504         // Read only, ignore
 505         return nil
 506 }
 507 func (r *pluginReference) Delete(ref reference.Named) (bool, error) {
 508         // Read only, ignore
 509         return false, nil
 510 }
 511
 512 type pluginConfigStore struct {
 513         pm     *Manager
 514         plugin *v2.Plugin
 515 }
 516
 517 func (s *pluginConfigStore) Put([]byte) (digest.Digest, error) {
 518         return digest.Digest(""), errors.New("cannot store config on push")
 519 }
 520
 521 func (s *pluginConfigStore) Get(d digest.Digest) ([]byte, error) {
 522         if s.plugin.Config != d {
 523                 return nil, errors.New("plugin not found")
 524         }
 525         rwc, err := s.pm.blobStore.Get(d)
 526         if err != nil {
 527                 return nil, err
 528         }
 529         defer rwc.Close()
 530         return ioutil.ReadAll(rwc)
 531 }
 532
 533 func (s *pluginConfigStore) GetTarSeekStream(d digest.Digest) (ioutils.ReadSeekCloser, error) {
 534         return nil, fmt.Errorf("unimplemented")
 535 }
 536
 537 func (s *pluginConfigStore) RootFSAndPlatformFromConfig(c []byte) (*image.RootFS, layer.Platform, error) {
 538         return configToRootFS(c)
 539 }
 540
 541 type pluginLayerProvider struct {
 542         pm     *Manager
 543         plugin *v2.Plugin
 544 }
 545
 546 func (p *pluginLayerProvider) Get(id layer.ChainID) (distribution.PushLayer, error) {
 547         rootFS := rootFSFromPlugin(p.plugin.PluginObj.Config.Rootfs)
 548         var i int
 549         for i = 1; i <= len(rootFS.DiffIDs); i++ {
 550                 if layer.CreateChainID(rootFS.DiffIDs[:i]) == id {
 551                         break
 552                 }
 553         }
 554         if i > len(rootFS.DiffIDs) {
 555                 return nil, errors.New("layer not found")
 556         }
 557         return &pluginLayer{
 558                 pm:      p.pm,
 559                 diffIDs: rootFS.DiffIDs[:i],
 560                 blobs:   p.plugin.Blobsums[:i],
 561         }, nil
 562 }
 563
 564 type pluginLayer struct {
 565         pm      *Manager
 566         diffIDs []layer.DiffID
 567         blobs   []digest.Digest
 568 }
 569
 570 func (l *pluginLayer) ChainID() layer.ChainID {
 571         return layer.CreateChainID(l.diffIDs)
 572 }
 573
 574 func (l *pluginLayer) DiffID() layer.DiffID {
 575         return l.diffIDs[len(l.diffIDs)-1]
 576 }
 577
 578 func (l *pluginLayer) Parent() distribution.PushLayer {
 579         if len(l.diffIDs) == 1 {
 580                 return nil
 581         }
 582         return &pluginLayer{
 583                 pm:      l.pm,
 584                 diffIDs: l.diffIDs[:len(l.diffIDs)-1],
 585                 blobs:   l.blobs[:len(l.diffIDs)-1],
 586         }
 587 }
 588
 589 func (l *pluginLayer) Open() (io.ReadCloser, error) {
 590         return l.pm.blobStore.Get(l.blobs[len(l.diffIDs)-1])
 591 }
 592
 593 func (l *pluginLayer) Size() (int64, error) {
 594         return l.pm.blobStore.Size(l.blobs[len(l.diffIDs)-1])
 595 }
 596
 597 func (l *pluginLayer) MediaType() string {
 598         return schema2.MediaTypeLayer
 599 }
 600
 601 func (l *pluginLayer) Release() {
 602         // Nothing needs to be release, no references held
 603 }
 604
 605 // Remove deletes plugin's root directory.
 606 func (pm *Manager) Remove(name string, config *types.PluginRmConfig) error {
 607         p, err := pm.config.Store.GetV2Plugin(name)
 608         pm.mu.RLock()
 609         c := pm.cMap[p]
 610         pm.mu.RUnlock()
 611
 612         if err != nil {
 613                 return err
 614         }
 615
 616         if !config.ForceRemove {
 617                 if p.GetRefCount() > 0 {
 618                         return fmt.Errorf("plugin %s is in use", p.Name())
 619                 }
 620                 if p.IsEnabled() {
 621                         return fmt.Errorf("plugin %s is enabled", p.Name())
 622                 }
 623         }
 624
 625         if p.IsEnabled() {
 626                 if err := pm.disable(p, c); err != nil {
 627                         logrus.Errorf("failed to disable plugin '%s': %s", p.Name(), err)
 628                 }
 629         }
 630
 631         defer func() {
 632                 go pm.GC()
 633         }()
 634
 635         id := p.GetID()
 636         pm.config.Store.Remove(p)
 637         pluginDir := filepath.Join(pm.config.Root, id)
 638         if err := recursiveUnmount(pluginDir); err != nil {
 639                 logrus.WithField("dir", pluginDir).WithField("id", id).Warn(err)
 640         }
 641         if err := os.RemoveAll(pluginDir); err != nil {
 642                 logrus.Warnf("unable to remove %q from plugin remove: %v", pluginDir, err)
 643         }
 644         pm.config.LogPluginEvent(id, name, "remove")
 645         return nil
 646 }
 647
 648 func getMounts(root string) ([]string, error) {
 649         infos, err := mount.GetMounts()
 650         if err != nil {
 651                 return nil, errors.Wrap(err, "failed to read mount table")
 652         }
 653
 654         var mounts []string
 655         for _, m := range infos {
 656                 if strings.HasPrefix(m.Mountpoint, root) {
 657                         mounts = append(mounts, m.Mountpoint)
 658                 }
 659         }
 660
 661         return mounts, nil
 662 }
 663
 664 func recursiveUnmount(root string) error {
 665         mounts, err := getMounts(root)
 666         if err != nil {
 667                 return err
 668         }
 669
 670         // sort in reverse-lexicographic order so the root mount will always be last
 671         sort.Sort(sort.Reverse(sort.StringSlice(mounts)))
 672
 673         for i, m := range mounts {
 674                 if err := mount.Unmount(m); err != nil {
 675                         if i == len(mounts)-1 {
 676                                 return errors.Wrapf(err, "error performing recursive unmount on %s", root)
 677                         }
 678                         logrus.WithError(err).WithField("mountpoint", m).Warn("could not unmount")
 679                 }
 680         }
 681
 682         return nil
 683 }
 684
 685 // Set sets plugin args
 686 func (pm *Manager) Set(name string, args []string) error {
 687         p, err := pm.config.Store.GetV2Plugin(name)
 688         if err != nil {
 689                 return err
 690         }
 691         if err := p.Set(args); err != nil {
 692                 return err
 693         }
 694         return pm.save(p)
 695 }
 696
 697 // CreateFromContext creates a plugin from the given pluginDir which contains
 698 // both the rootfs and the config.json and a repoName with optional tag.
 699 func (pm *Manager) CreateFromContext(ctx context.Context, tarCtx io.ReadCloser, options *types.PluginCreateOptions) (err error) {
 700         pm.muGC.RLock()
 701         defer pm.muGC.RUnlock()
 702
 703         ref, err := reference.ParseNormalizedNamed(options.RepoName)
 704         if err != nil {
 705                 return errors.Wrapf(err, "failed to parse reference %v", options.RepoName)
 706         }
 707         if _, ok := ref.(reference.Canonical); ok {
 708                 return errors.Errorf("canonical references are not permitted")
 709         }
 710         name := reference.FamiliarString(reference.TagNameOnly(ref))
 711
 712         if err := pm.config.Store.validateName(name); err != nil { // fast check, real check is in createPlugin()
 713                 return err
 714         }
 715
 716         tmpRootFSDir, err := ioutil.TempDir(pm.tmpDir(), ".rootfs")
 717         if err != nil {
 718                 return errors.Wrap(err, "failed to create temp directory")
 719         }
 720         defer os.RemoveAll(tmpRootFSDir)
 721
 722         var configJSON []byte
 723         rootFS := splitConfigRootFSFromTar(tarCtx, &configJSON)
 724
 725         rootFSBlob, err := pm.blobStore.New()
 726         if err != nil {
 727                 return err
 728         }
 729         defer rootFSBlob.Close()
 730         gzw := gzip.NewWriter(rootFSBlob)
 731         layerDigester := digest.Canonical.Digester()
 732         rootFSReader := io.TeeReader(rootFS, io.MultiWriter(gzw, layerDigester.Hash()))
 733
 734         if err := chrootarchive.Untar(rootFSReader, tmpRootFSDir, nil); err != nil {
 735                 return err
 736         }
 737         if err := rootFS.Close(); err != nil {
 738                 return err
 739         }
 740
 741         if configJSON == nil {
 742                 return errors.New("config not found")
 743         }
 744
 745         if err := gzw.Close(); err != nil {
 746                 return errors.Wrap(err, "error closing gzip writer")
 747         }
 748
 749         var config types.PluginConfig
 750         if err := json.Unmarshal(configJSON, &config); err != nil {
 751                 return errors.Wrap(err, "failed to parse config")
 752         }
 753
 754         if err := pm.validateConfig(config); err != nil {
 755                 return err
 756         }
 757
 758         pm.mu.Lock()
 759         defer pm.mu.Unlock()
 760
 761         rootFSBlobsum, err := rootFSBlob.Commit()
 762         if err != nil {
 763                 return err
 764         }
 765         defer func() {
 766                 if err != nil {
 767                         go pm.GC()
 768                 }
 769         }()
 770
 771         config.Rootfs = &types.PluginConfigRootfs{
 772                 Type:    "layers",
 773                 DiffIds: []string{layerDigester.Digest().String()},
 774         }
 775
 776         config.DockerVersion = dockerversion.Version
 777
 778         configBlob, err := pm.blobStore.New()
 779         if err != nil {
 780                 return err
 781         }
 782         defer configBlob.Close()
 783         if err := json.NewEncoder(configBlob).Encode(config); err != nil {
 784                 return errors.Wrap(err, "error encoding json config")
 785         }
 786         configBlobsum, err := configBlob.Commit()
 787         if err != nil {
 788                 return err
 789         }
 790
 791         p, err := pm.createPlugin(name, configBlobsum, []digest.Digest{rootFSBlobsum}, tmpRootFSDir, nil)
 792         if err != nil {
 793                 return err
 794         }
 795         p.PluginObj.PluginReference = name
 796
 797         pm.config.LogPluginEvent(p.PluginObj.ID, name, "create")
 798
 799         return nil
 800 }
 801
 802 func (pm *Manager) validateConfig(config types.PluginConfig) error {
 803         return nil // TODO:
 804 }
 805
 806 func splitConfigRootFSFromTar(in io.ReadCloser, config *[]byte) io.ReadCloser {
 807         pr, pw := io.Pipe()
 808         go func() {
 809                 tarReader := tar.NewReader(in)
 810                 tarWriter := tar.NewWriter(pw)
 811                 defer in.Close()
 812
 813                 hasRootFS := false
 814
 815                 for {
 816                         hdr, err := tarReader.Next()
 817                         if err == io.EOF {
 818                                 if !hasRootFS {
 819                                         pw.CloseWithError(errors.Wrap(err, "no rootfs found"))
 820                                         return
 821                                 }
 822                                 // Signals end of archive.
 823                                 tarWriter.Close()
 824                                 pw.Close()
 825                                 return
 826                         }
 827                         if err != nil {
 828                                 pw.CloseWithError(errors.Wrap(err, "failed to read from tar"))
 829                                 return
 830                         }
 831
 832                         content := io.Reader(tarReader)
 833                         name := path.Clean(hdr.Name)
 834                         if path.IsAbs(name) {
 835                                 name = name[1:]
 836                         }
 837                         if name == configFileName {
 838                                 dt, err := ioutil.ReadAll(content)
 839                                 if err != nil {
 840                                         pw.CloseWithError(errors.Wrapf(err, "failed to read %s", configFileName))
 841                                         return
 842                                 }
 843                                 *config = dt
 844                         }
 845                         if parts := strings.Split(name, "/"); len(parts) != 0 && parts[0] == rootFSFileName {
 846                                 hdr.Name = path.Clean(path.Join(parts[1:]...))
 847                                 if hdr.Typeflag == tar.TypeLink && strings.HasPrefix(strings.ToLower(hdr.Linkname), rootFSFileName+"/") {
 848                                         hdr.Linkname = hdr.Linkname[len(rootFSFileName)+1:]
 849                                 }
 850                                 if err := tarWriter.WriteHeader(hdr); err != nil {
 851                                         pw.CloseWithError(errors.Wrap(err, "error writing tar header"))
 852                                         return
 853                                 }
 854                                 if _, err := pools.Copy(tarWriter, content); err != nil {
 855                                         pw.CloseWithError(errors.Wrap(err, "error copying tar data"))
 856                                         return
 857                                 }
 858                                 hasRootFS = true
 859                         } else {
 860                                 io.Copy(ioutil.Discard, content)
 861                         }
 862                 }
 863         }()
 864         return pr
 865 }