kubernetes
/
kops
mirror of https://github.com/kubernetes/kops.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GCE: Use object-level permissions for files in GCS 

This lets us configure cross-project permissions while ourselves needing
minimal permissions, but also gives us a nice hook for future lockdown
of object-level permissions.
This commit is contained in:
Justin Santa Barbara 2017-10-29 17:59:38 -04:00
parent d1ee8026ac
commit b2bcba4a6d
49 changed files with 448 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/kops/create_cluster.go
View File

@ -996,7 +996,7 @@ func RunCreateCluster(f *util.Factory, out io.Writer, c *CreateClusterOptions) e
return err
}
err = registry.WriteConfigDeprecated(configBase.Join(registry.PathClusterCompleted), fullCluster)
err = registry.WriteConfigDeprecated(cluster, configBase.Join(registry.PathClusterCompleted), fullCluster)
if err != nil {
return fmt.Errorf("error writing completed cluster spec: %v", err)
}

2
cmd/kops/edit_cluster.go
View File

@ -250,7 +250,7 @@ func RunEditCluster(f *util.Factory, cmd *cobra.Command, args []string, out io.W
return preservedFile(err, file, out)
}
err = registry.WriteConfigDeprecated(configBase.Join(registry.PathClusterCompleted), fullCluster)
err = registry.WriteConfigDeprecated(newCluster, configBase.Join(registry.PathClusterCompleted), fullCluster)
if err != nil {
return preservedFile(fmt.Errorf("error writing completed cluster spec: %v", err), file, out)
}

1
cmd/kops/util/BUILD.bazel
View File

@ -5,6 +5,7 @@ go_library(
srcs = ["factory.go"],
visibility = ["//visibility:public"],
deps = [
"//pkg/acls/gce:go_default_library",
"//pkg/client/clientset_generated/clientset:go_default_library",
"//pkg/client/simple:go_default_library",
"//pkg/client/simple/api:go_default_library",

21
cmd/kops/util/factory.go
View File

@ -18,17 +18,18 @@ package util
import (
"fmt"
"k8s.io/apimachinery/pkg/util/validation/field"
"k8s.io/kops/pkg/client/simple"
"k8s.io/kops/pkg/client/simple/vfsclientset"
"k8s.io/kops/util/pkg/vfs"
"github.com/golang/glog"
"k8s.io/client-go/rest"
kopsclient "k8s.io/kops/pkg/client/clientset_generated/clientset"
"k8s.io/kops/pkg/client/simple/api"
"net/url"
"strings"
"github.com/golang/glog"
"k8s.io/apimachinery/pkg/util/validation/field"
"k8s.io/client-go/rest"
gceacls "k8s.io/kops/pkg/acls/gce"
kopsclient "k8s.io/kops/pkg/client/clientset_generated/clientset"
"k8s.io/kops/pkg/client/simple"
"k8s.io/kops/pkg/client/simple/api"
"k8s.io/kops/pkg/client/simple/vfsclientset"
"k8s.io/kops/util/pkg/vfs"
)
type FactoryOptions struct {
@ -41,6 +42,8 @@ type Factory struct {
}
func NewFactory(options *FactoryOptions) *Factory {
gceacls.Register()
return &Factory{
options: options,
}

2
federation/apply_federation.go
View File

@ -207,7 +207,7 @@ func (o *ApplyFederationOperation) federationContextForCluster(cluster *kopsapi.
federationKeystore := k8sapi.NewKubernetesKeystore(target.KubernetesClient, o.namespace)
checkExisting := true
context, err := fi.NewContext(target, nil, federationKeystore, nil, nil, checkExisting, nil)
context, err := fi.NewContext(target, nil, nil, federationKeystore, nil, nil, checkExisting, nil)
if err != nil {
return nil, err
}

2
hack/.packages
View File

@ -23,6 +23,8 @@ k8s.io/kops/nodeup/pkg/bootstrap
k8s.io/kops/nodeup/pkg/distros
k8s.io/kops/nodeup/pkg/model
k8s.io/kops/nodeup/pkg/model/resources
k8s.io/kops/pkg/acls
k8s.io/kops/pkg/acls/gce
k8s.io/kops/pkg/apis/kops
k8s.io/kops/pkg/apis/kops/install
k8s.io/kops/pkg/apis/kops/model

2
nodeup/pkg/bootstrap/install.go
View File

@ -80,7 +80,7 @@ func (i *Installation) Run() error {
}
checkExisting := true
context, err := fi.NewContext(target, cloud, keyStore, secretStore, configBase, checkExisting, tasks)
context, err := fi.NewContext(target, nil, cloud, keyStore, secretStore, configBase, checkExisting, tasks)
if err != nil {
return fmt.Errorf("error building context: %v", err)
}

14
pkg/acls/BUILD.bazel Normal file
View File

@ -0,0 +1,14 @@
load("@io_bazel_rules_go//go:def.bzl", "go_library")
go_library(
name = "go_default_library",
srcs = [
"interface.go",
"plugins.go",
],
visibility = ["//visibility:public"],
deps = [
"//pkg/apis/kops:go_default_library",
"//util/pkg/vfs:go_default_library",
],
)

15
pkg/acls/gce/BUILD.bazel Normal file
View File

@ -0,0 +1,15 @@
load("@io_bazel_rules_go//go:def.bzl", "go_library")
go_library(
name = "go_default_library",
srcs = ["storage.go"],
visibility = ["//visibility:public"],
deps = [
"//pkg/acls:go_default_library",
"//pkg/apis/kops:go_default_library",
"//upup/pkg/fi/cloudup:go_default_library",
"//upup/pkg/fi/cloudup/gce:go_default_library",
"//util/pkg/vfs:go_default_library",
"//vendor/google.golang.org/api/storage/v1:go_default_library",
],
)

84
pkg/acls/gce/storage.go Normal file
View File

@ -0,0 +1,84 @@
/*
Copyright 2017 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package gce
import (
"fmt"
storage "google.golang.org/api/storage/v1"
"k8s.io/kops/pkg/acls"
"k8s.io/kops/pkg/apis/kops"
"k8s.io/kops/upup/pkg/fi/cloudup"
"k8s.io/kops/upup/pkg/fi/cloudup/gce"
"k8s.io/kops/util/pkg/vfs"
)
// gcsAclStrategy is the AclStrategy for objects written to google cloud storage
type gcsAclStrategy struct {
}
var _ acls.ACLStrategy = &gcsAclStrategy{}
// GetACL returns the ACL to use if this is a google cloud storage path
func (s *gcsAclStrategy) GetACL(p vfs.Path, cluster *kops.Cluster) (vfs.ACL, error) {
if kops.CloudProviderID(cluster.Spec.CloudProvider) != kops.CloudProviderGCE {
return nil, nil
}
gcsPath, ok := p.(*vfs.GSPath)
if !ok {
return nil, nil
}
bucketName := gcsPath.Bucket()
client := gcsPath.Client()
// TODO: Cache?
bucket, err := client.Buckets.Get(bucketName).Do()
if err != nil {
return nil, fmt.Errorf("error querying bucket %q: %v", bucketName, err)
}
// TODO: Cache?
cloud, err := cloudup.BuildCloud(cluster)
if err != nil {
return nil, err
}
serviceAccount, err := cloud.(gce.GCECloud).ServiceAccount()
if err != nil {
return nil, err
}
var acls []*storage.ObjectAccessControl
for _, a := range bucket.DefaultObjectAcl {
acls = append(acls, a)
}
acls = append(acls, &storage.ObjectAccessControl{
Email: serviceAccount,
Entity: "user-" + serviceAccount,
Role: "READER",
})
return &vfs.GSAcl{
Acl: acls,
}, nil
}
func Register() {
acls.RegisterPlugin("k8s.io/kops/acl/gce", &gcsAclStrategy{})
}

28
pkg/acls/interface.go Normal file
View File

@ -0,0 +1,28 @@
/*
Copyright 2017 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package acls
import (
"k8s.io/kops/pkg/apis/kops"
"k8s.io/kops/util/pkg/vfs"
)
// ACLStrategy is the interface implemented by ACL strategy providers
type ACLStrategy interface {
// GetACL returns the ACL if this strategy handles the vfs.Path, when writing for the specified cluster
GetACL(p vfs.Path, cluster *kops.Cluster) (vfs.ACL, error)
}

57
pkg/acls/plugins.go Normal file
View File

@ -0,0 +1,57 @@
/*
Copyright 2017 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package acls
import (
"fmt"
"sync"
"k8s.io/kops/pkg/apis/kops"
"k8s.io/kops/util/pkg/vfs"
)
var strategies map[string]ACLStrategy
var strategiesMutex sync.Mutex
// GetACL returns the ACL for the vfs.Path, by consulting all registered strategies
func GetACL(p vfs.Path, cluster *kops.Cluster) (vfs.ACL, error) {
strategiesMutex.Lock()
defer strategiesMutex.Unlock()
for k, strategy := range strategies {
acl, err := strategy.GetACL(p, cluster)
if err != nil {
return nil, fmt.Errorf("error from acl provider %q: %v", k, err)
}
if acl != nil {
return acl, nil
}
}
return nil, nil
}
// RegisterPlugin adds the strategy to the registered strategies
func RegisterPlugin(key string, strategy ACLStrategy) {
strategiesMutex.Lock()
defer strategiesMutex.Unlock()
if strategies == nil {
strategies = make(map[string]ACLStrategy)
}
strategies[key] = strategy
}

1
pkg/apis/kops/registry/BUILD.bazel
View File

@ -9,6 +9,7 @@ go_library(
],
visibility = ["//visibility:public"],
deps = [
"//pkg/acls:go_default_library",
"//pkg/apis/kops:go_default_library",
"//pkg/client/simple:go_default_library",
"//upup/pkg/fi/utils:go_default_library",

18
pkg/apis/kops/registry/statestore.go
View File

@ -18,10 +18,13 @@ package registry
import (
"fmt"
"k8s.io/kops/upup/pkg/fi/utils"
"k8s.io/kops/util/pkg/vfs"
"os"
"strings"
"k8s.io/kops/pkg/acls"
"k8s.io/kops/pkg/apis/kops"
"k8s.io/kops/upup/pkg/fi/utils"
"k8s.io/kops/util/pkg/vfs"
)
func ReadConfigDeprecated(configPath vfs.Path, config interface{}) error {
@ -49,7 +52,7 @@ func ReadConfigDeprecated(configPath vfs.Path, config interface{}) error {
// WriteConfigDeprecated writes a config file as yaml.
// It is deprecated because it is unversioned, but it is still used, in particular for writing the completed config.
func WriteConfigDeprecated(configPath vfs.Path, config interface{}, writeOptions ...vfs.WriteOption) error {
func WriteConfigDeprecated(cluster *kops.Cluster, configPath vfs.Path, config interface{}, writeOptions ...vfs.WriteOption) error {
data, err := utils.YamlMarshal(config)
if err != nil {
return fmt.Errorf("error marshalling configuration: %v", err)
@ -73,10 +76,15 @@ func WriteConfigDeprecated(configPath vfs.Path, config interface{}, writeOptions
}
}
acl, err := acls.GetACL(configPath, cluster)
if err != nil {
return err
}
if create {
err = configPath.CreateFile(data)
err = configPath.CreateFile(data, acl)
} else {
err = configPath.WriteFile(data)
err = configPath.WriteFile(data, acl)
}
if err != nil {
return fmt.Errorf("error writing configuration file %s: %v", configPath, err)

4
pkg/client/simple/api/clientset.go
View File

@ -109,12 +109,12 @@ func (c *RESTClientset) GetFederation(name string) (*kops.Federation, error) {
func (c *RESTClientset) SecretStore(cluster *kops.Cluster) (fi.SecretStore, error) {
namespace := restNamespaceForClusterName(cluster.Name)
return secrets.NewClientsetSecretStore(c.KopsClient, namespace), nil
return secrets.NewClientsetSecretStore(cluster, c.KopsClient, namespace), nil
}
func (c *RESTClientset) KeyStore(cluster *kops.Cluster) (fi.CAStore, error) {
namespace := restNamespaceForClusterName(cluster.Name)
return fi.NewClientsetCAStore(c.KopsClient, namespace), nil
return fi.NewClientsetCAStore(cluster, c.KopsClient, namespace), nil
}
func (c *RESTClientset) DeleteCluster(cluster *kops.Cluster) error {

1
pkg/client/simple/vfsclientset/BUILD.bazel
View File

@ -13,6 +13,7 @@ go_library(
],
visibility = ["//visibility:public"],
deps = [
"//pkg/acls:go_default_library",
"//pkg/apis/kops:go_default_library",
"//pkg/apis/kops/registry:go_default_library",
"//pkg/apis/kops/v1alpha1:go_default_library",

7
pkg/client/simple/vfsclientset/clientset.go
View File

@ -70,8 +70,7 @@ func (c *VFSClientset) ConfigBaseFor(cluster *kops.Cluster) (vfs.Path, error) {
// InstanceGroupsFor implements the InstanceGroupsFor method of simple.Clientset for a VFS-backed state store
func (c *VFSClientset) InstanceGroupsFor(cluster *kops.Cluster) kopsinternalversion.InstanceGroupInterface {
clusterName := cluster.Name
return newInstanceGroupVFS(c, clusterName)
return newInstanceGroupVFS(c, cluster)
}
func (c *VFSClientset) federations() kopsinternalversion.FederationInterface {
@ -99,7 +98,7 @@ func (c *VFSClientset) SecretStore(cluster *kops.Cluster) (fi.SecretStore, error
return nil, err
}
basedir := configBase.Join("secrets")
return secrets.NewVFSSecretStore(basedir), nil
return secrets.NewVFSSecretStore(cluster, basedir), nil
}
func (c *VFSClientset) KeyStore(cluster *kops.Cluster) (fi.CAStore, error) {
@ -108,7 +107,7 @@ func (c *VFSClientset) KeyStore(cluster *kops.Cluster) (fi.CAStore, error) {
return nil, err
}
basedir := configBase.Join("pki")
return fi.NewVFSCAStore(basedir), nil
return fi.NewVFSCAStore(cluster, basedir), nil
}
func DeleteAllClusterState(basePath vfs.Path) error {

4
pkg/client/simple/vfsclientset/cluster.go
View File

@ -101,7 +101,7 @@ func (r *ClusterVFS) Create(c *api.Cluster) (*api.Cluster, error) {
return nil, fmt.Errorf("clusterName is required")
}
if err := r.writeConfig(r.basePath.Join(clusterName, registry.PathCluster), c, vfs.WriteOptionCreate); err != nil {
if err := r.writeConfig(c, r.basePath.Join(clusterName, registry.PathCluster), c, vfs.WriteOptionCreate); err != nil {
if os.IsExist(err) {
return nil, err
}
@ -126,7 +126,7 @@ func (r *ClusterVFS) Update(c *api.Cluster, status *api.ClusterStatus) (*api.Clu
return nil, err
}
if err := r.writeConfig(r.basePath.Join(clusterName, registry.PathCluster), c, vfs.WriteOptionOnlyIfExists); err != nil {
if err := r.writeConfig(c, r.basePath.Join(clusterName, registry.PathCluster), c, vfs.WriteOptionOnlyIfExists); err != nil {
if os.IsNotExist(err) {
return nil, err
}

20
pkg/client/simple/vfsclientset/commonvfs.go
View File

@ -25,6 +25,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/kops/pkg/acls"
kops "k8s.io/kops/pkg/apis/kops"
"k8s.io/kops/pkg/apis/kops/v1alpha2"
"k8s.io/kops/pkg/kopscodecs"
@ -76,7 +77,7 @@ func (c *commonVFS) list(items interface{}, options metav1.ListOptions) (interfa
return c.readAll(items)
}
func (c *commonVFS) create(i runtime.Object) error {
func (c *commonVFS) create(cluster *kops.Cluster, i runtime.Object) error {
objectMeta, err := meta.Accessor(i)
if err != nil {
return err
@ -94,7 +95,7 @@ func (c *commonVFS) create(i runtime.Object) error {
objectMeta.SetCreationTimestamp(v1.NewTime(time.Now().UTC()))
}
err = c.writeConfig(c.basePath.Join(objectMeta.GetName()), i, vfs.WriteOptionCreate)
err = c.writeConfig(cluster, c.basePath.Join(objectMeta.GetName()), i, vfs.WriteOptionCreate)
if err != nil {
if os.IsExist(err) {
return err
@ -131,7 +132,7 @@ func (c *commonVFS) readConfig(configPath vfs.Path) (runtime.Object, error) {
return object, nil
}
func (c *commonVFS) writeConfig(configPath vfs.Path, o runtime.Object, writeOptions ...vfs.WriteOption) error {
func (c *commonVFS) writeConfig(cluster *kops.Cluster, configPath vfs.Path, o runtime.Object, writeOptions ...vfs.WriteOption) error {
data, err := c.serialize(o)
if err != nil {
return fmt.Errorf("error marshalling object: %v", err)
@ -155,10 +156,15 @@ func (c *commonVFS) writeConfig(configPath vfs.Path, o runtime.Object, writeOpti
}
}
acl, err := acls.GetACL(configPath, cluster)
if err != nil {
return err
}
if create {
err = configPath.CreateFile(data)
err = configPath.CreateFile(data, acl)
} else {
err = configPath.WriteFile(data)
err = configPath.WriteFile(data, acl)
}
if err != nil {
if create && os.IsExist(err) {
@ -170,7 +176,7 @@ func (c *commonVFS) writeConfig(configPath vfs.Path, o runtime.Object, writeOpti
return nil
}
func (c *commonVFS) update(i runtime.Object) error {
func (c *commonVFS) update(cluster *kops.Cluster, i runtime.Object) error {
objectMeta, err := meta.Accessor(i)
if err != nil {
return err
@ -188,7 +194,7 @@ func (c *commonVFS) update(i runtime.Object) error {
objectMeta.SetCreationTimestamp(v1.NewTime(time.Now().UTC()))
}
err = c.writeConfig(c.basePath.Join(objectMeta.GetName()), i, vfs.WriteOptionOnlyIfExists)
err = c.writeConfig(cluster, c.basePath.Join(objectMeta.GetName()), i, vfs.WriteOptionOnlyIfExists)
if err != nil {
return fmt.Errorf("error writing %s: %v", c.kind, err)
}

4
pkg/client/simple/vfsclientset/federation.go
View File

@ -72,7 +72,7 @@ func (c *FederationVFS) List(options metav1.ListOptions) (*api.FederationList, e
}
func (c *FederationVFS) Create(g *api.Federation) (*api.Federation, error) {
err := c.create(g)
err := c.create(nil, g)
if err != nil {
return nil, err
}
@ -80,7 +80,7 @@ func (c *FederationVFS) Create(g *api.Federation) (*api.Federation, error) {
}
func (c *FederationVFS) Update(g *api.Federation) (*api.Federation, error) {
err := c.update(g)
err := c.update(nil, g)
if err != nil {
return nil, err
}

23
pkg/client/simple/vfsclientset/instancegroup.go
View File

@ -35,6 +35,7 @@ type InstanceGroupVFS struct {
commonVFS
clusterName string
cluster *kops.Cluster
}
type InstanceGroupMirror interface {
@ -43,10 +44,16 @@ type InstanceGroupMirror interface {
var _ InstanceGroupMirror = &InstanceGroupVFS{}
func NewInstanceGroupMirror(clusterName string, configBase vfs.Path) InstanceGroupMirror {
func NewInstanceGroupMirror(cluster *kops.Cluster, configBase vfs.Path) InstanceGroupMirror {
if cluster == nil || cluster.Name == "" {
glog.Fatalf("cluster / cluster.Name is required")
}
clusterName := cluster.Name
kind := "InstanceGroup"
r := &InstanceGroupVFS{
cluster: cluster,
clusterName: clusterName,
}
r.init(kind, configBase.Join("instancegroup"), StoreVersion)
@ -58,14 +65,16 @@ func NewInstanceGroupMirror(clusterName string, configBase vfs.Path) InstanceGro
return r
}
func newInstanceGroupVFS(c *VFSClientset, clusterName string) *InstanceGroupVFS {
if clusterName == "" {
glog.Fatalf("clusterName is required")
func newInstanceGroupVFS(c *VFSClientset, cluster *kops.Cluster) *InstanceGroupVFS {
if cluster == nil || cluster.Name == "" {
glog.Fatalf("cluster / cluster.Name is required")
}
clusterName := cluster.Name
kind := "InstanceGroup"
r := &InstanceGroupVFS{
cluster: cluster,
clusterName: clusterName,
}
r.init(kind, c.basePath.Join(clusterName, "instancegroup"), StoreVersion)
@ -119,7 +128,7 @@ func (c *InstanceGroupVFS) List(options metav1.ListOptions) (*api.InstanceGroupL
}
func (c *InstanceGroupVFS) Create(g *api.InstanceGroup) (*api.InstanceGroup, error) {
err := c.create(g)
err := c.create(c.cluster, g)
if err != nil {
return nil, err
}
@ -127,7 +136,7 @@ func (c *InstanceGroupVFS) Create(g *api.InstanceGroup) (*api.InstanceGroup, err
}
func (c *InstanceGroupVFS) Update(g *api.InstanceGroup) (*api.InstanceGroup, error) {
err := c.update(g)
err := c.update(c.cluster, g)
if err != nil {
return nil, err
}
@ -135,7 +144,7 @@ func (c *InstanceGroupVFS) Update(g *api.InstanceGroup) (*api.InstanceGroup, err
}
func (c *InstanceGroupVFS) WriteMirror(g *api.InstanceGroup) error {
err := c.writeConfig(c.basePath.Join(g.Name), g)
err := c.writeConfig(c.cluster, c.basePath.Join(g.Name), g)
if err != nil {
return fmt.Errorf("error writing %s: %v", c.kind, err)
}

4
upup/models/vfs.go
View File

@ -49,11 +49,11 @@ func (p *AssetPath) Join(relativePath ...string) vfs.Path {
return &AssetPath{location: joined}
}
func (p *AssetPath) WriteFile(data []byte) error {
func (p *AssetPath) WriteFile(data []byte, acl vfs.ACL) error {
return ReadOnlyError
}
func (p *AssetPath) CreateFile(data []byte) error {
func (p *AssetPath) CreateFile(data []byte, acl vfs.ACL) error {
return ReadOnlyError
}

1
upup/pkg/fi/BUILD.bazel
View File

@ -32,6 +32,7 @@ go_library(
],
visibility = ["//visibility:public"],
deps = [
"//pkg/acls:go_default_library",
"//pkg/apis/kops:go_default_library",
"//pkg/assets:go_default_library",
"//pkg/client/clientset_generated/clientset/typed/kops/internalversion:go_default_library",

1
upup/pkg/fi/assettasks/BUILD.bazel
View File

@ -12,6 +12,7 @@ go_library(
],
visibility = ["//visibility:public"],
deps = [
"//pkg/acls:go_default_library",
"//upup/pkg/fi:go_default_library",
"//util/pkg/hashing:go_default_library",
"//util/pkg/vfs:go_default_library",

32
upup/pkg/fi/assettasks/copyfile.go
View File

@ -17,17 +17,16 @@ limitations under the License.
package assettasks
import (
"fmt"
"net/url"
"bytes"
"fmt"
"net/url"
"strings"
"github.com/golang/glog"
"k8s.io/kops/pkg/acls"
"k8s.io/kops/upup/pkg/fi"
"k8s.io/kops/util/pkg/hashing"
"k8s.io/kops/util/pkg/vfs"
"strings"
)
// CopyFile copies an from a source file repository, to a target repository,
@ -83,14 +82,14 @@ func (_ *CopyFile) Render(c *fi.Context, a, e, changes *CopyFile) error {
glog.Infof("copying bits from %q to %q", source, target)
if err := transferFile(source, target, sourceSha, sourceSHALocation); err != nil {
if err := transferFile(c, source, target, sourceSha, sourceSHALocation); err != nil {
return fmt.Errorf("unable to transfer %q to %q: %v", source, target, err)
}
return nil
}
func transferFile(source string, target string, sourceSHA string, sourceSHALocation string) error {
func transferFile(c *fi.Context, source string, target string, sourceSHA string, sourceSHALocation string) error {
data, err := vfs.Context.ReadFile(source)
if err != nil {
return fmt.Errorf("Error unable to read path %q: %v", source, err)
@ -139,28 +138,33 @@ func transferFile(source string, target string, sourceSHA string, sourceSHALocat
}
b := bytes.NewBufferString(sha)
if err := writeFile(shaVFS, b.Bytes()); err != nil {
if err := writeFile(c, shaVFS, b.Bytes()); err != nil {
return fmt.Errorf("Error uploading file %q: %v", shaVFS, err)
}
}
}
if err := writeFile(uploadVFS, data); err != nil {
if err := writeFile(c, uploadVFS, data); err != nil {
return fmt.Errorf("Error uploading file %q: %v", uploadVFS, err)
}
return nil
}
func writeFile(vfsPath string, data []byte) error {
func writeFile(c *fi.Context, vfsPath string, data []byte) error {
glog.V(2).Infof("uploading to %q", vfsPath)
destinationRegistry, err := vfs.Context.BuildVfsPath(vfsPath)
p, err := vfs.Context.BuildVfsPath(vfsPath)
if err != nil {
return fmt.Errorf("Error parsing registry path %q: %v", vfsPath, err)
return fmt.Errorf("error building path %q: %v", vfsPath, err)
}
if err = destinationRegistry.WriteFile(data); err != nil {
return fmt.Errorf("Error destination path %q: %v", vfsPath, err)
acl, err := acls.GetACL(p, c.Cluster)
if err != nil {
return err
}
if err = p.WriteFile(data, acl); err != nil {
return fmt.Errorf("error writing path %q: %v", vfsPath, err)
}
glog.V(2).Infof("upload complete: %q", vfsPath)

26
upup/pkg/fi/clientset_castore.go
View File

@ -31,6 +31,7 @@ import (
"golang.org/x/crypto/ssh"
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/kops/pkg/acls"
"k8s.io/kops/pkg/apis/kops"
kopsinternalversion "k8s.io/kops/pkg/client/clientset_generated/clientset/typed/kops/internalversion"
"k8s.io/kops/pkg/pki"
@ -39,6 +40,7 @@ import (
// ClientsetCAStore is a CAStore implementation that stores keypairs in Keyset on a API server
type ClientsetCAStore struct {
cluster *kops.Cluster
namespace string
clientset kopsinternalversion.KopsInterface
@ -49,8 +51,9 @@ type ClientsetCAStore struct {
var _ CAStore = &ClientsetCAStore{}
// NewClientsetCAStore is the constructor for ClientsetCAStore
func NewClientsetCAStore(clientset kopsinternalversion.KopsInterface, namespace string) CAStore {
func NewClientsetCAStore(cluster *kops.Cluster, clientset kopsinternalversion.KopsInterface, namespace string) CAStore {
c := &ClientsetCAStore{
cluster: cluster,
clientset: clientset,
namespace: namespace,
cachedCaKeysets: make(map[string]*keyset),
@ -643,14 +646,24 @@ func (c *ClientsetCAStore) MirrorTo(basedir vfs.Path) error {
item := &keyset.Spec.Keys[i]
{
p := basedir.Join("issued", keyset.Name, item.Id+".crt")
err = p.WriteFile(item.PublicMaterial)
acl, err := acls.GetACL(p, c.cluster)
if err != nil {
return err
}
err = p.WriteFile(item.PublicMaterial, acl)
if err != nil {
return fmt.Errorf("error writing %q: %v", p, err)
}
}
{
p := basedir.Join("private", keyset.Name, item.Id+".key")
err = p.WriteFile(item.PrivateMaterial)
acl, err := acls.GetACL(p, c.cluster)
if err != nil {
return err
}
err = p.WriteFile(item.PrivateMaterial, acl)
if err != nil {
return fmt.Errorf("error writing %q: %v", p, err)
}
@ -684,7 +697,12 @@ func (c *ClientsetCAStore) MirrorTo(basedir vfs.Path) error {
id := formatFingerprint(h.Sum(nil))
p := basedir.Join("ssh", "public", sshCredential.Name, id)
err = p.WriteFile([]byte(sshCredential.Spec.PublicKey))
acl, err := acls.GetACL(p, c.cluster)
if err != nil {
return err
}
err = p.WriteFile([]byte(sshCredential.Spec.PublicKey), acl)
if err != nil {
return fmt.Errorf("error writing %q: %v", p, err)
}

15
upup/pkg/fi/cloudup/apply_cluster.go
View File

@ -22,6 +22,8 @@ import (
"strings"
"time"
"github.com/blang/semver"
"github.com/golang/glog"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
kopsbase "k8s.io/kops"
"k8s.io/kops/pkg/apis/kops"
@ -31,6 +33,7 @@ import (
"k8s.io/kops/pkg/apis/nodeup"
"k8s.io/kops/pkg/assets"
"k8s.io/kops/pkg/client/simple"
"k8s.io/kops/pkg/client/simple/vfsclientset"
"k8s.io/kops/pkg/dns"
"k8s.io/kops/pkg/featureflag"
"k8s.io/kops/pkg/model"
@ -45,6 +48,7 @@ import (
"k8s.io/kops/upup/pkg/fi"
"k8s.io/kops/upup/pkg/fi/cloudup/awstasks"
"k8s.io/kops/upup/pkg/fi/cloudup/awsup"
"k8s.io/kops/upup/pkg/fi/cloudup/baremetal"
"k8s.io/kops/upup/pkg/fi/cloudup/cloudformation"
"k8s.io/kops/upup/pkg/fi/cloudup/do"
"k8s.io/kops/upup/pkg/fi/cloudup/dotasks"
@ -56,11 +60,6 @@ import (
"k8s.io/kops/upup/pkg/fi/fitasks"
"k8s.io/kops/util/pkg/hashing"
"k8s.io/kops/util/pkg/vfs"
"github.com/blang/semver"
"github.com/golang/glog"
"k8s.io/kops/pkg/client/simple/vfsclientset"
"k8s.io/kops/upup/pkg/fi/cloudup/baremetal"
)
const (
@ -808,12 +807,12 @@ func (c *ApplyClusterCmd) Run() error {
c.Target = target
if !dryRun {
err = registry.WriteConfigDeprecated(configBase.Join(registry.PathClusterCompleted), c.Cluster)
err = registry.WriteConfigDeprecated(cluster, configBase.Join(registry.PathClusterCompleted), c.Cluster)
if err != nil {
return fmt.Errorf("error writing completed cluster spec: %v", err)
}
vfsMirror := vfsclientset.NewInstanceGroupMirror(cluster.Name, configBase)
vfsMirror := vfsclientset.NewInstanceGroupMirror(cluster, configBase)
for _, g := range c.InstanceGroups {
// TODO: We need to update the mirror (below), but do we need to update the primary?
@ -829,7 +828,7 @@ func (c *ApplyClusterCmd) Run() error {
}
}
context, err := fi.NewContext(target, cloud, keyStore, secretStore, configBase, checkExisting, taskMap)
context, err := fi.NewContext(target, cluster, cloud, keyStore, secretStore, configBase, checkExisting, taskMap)
if err != nil {
return fmt.Errorf("error building context: %v", err)
}

4
upup/pkg/fi/cloudup/awstasks/elastic_ip_test.go
View File

@ -69,7 +69,7 @@ func TestElasticIPCreate(t *testing.T) {
Cloud: cloud,
}
context, err := fi.NewContext(target, cloud, nil, nil, nil, true, allTasks)
context, err := fi.NewContext(target, nil, cloud, nil, nil, nil, true, allTasks)
if err != nil {
t.Fatalf("error building context: %v", err)
}
@ -106,7 +106,7 @@ func TestElasticIPCreate(t *testing.T) {
func checkNoChanges(t *testing.T, cloud fi.Cloud, allTasks map[string]fi.Task) {
assetBuilder := assets.NewAssetBuilder(nil)
target := fi.NewDryRunTarget(assetBuilder, os.Stderr)
context, err := fi.NewContext(target, cloud, nil, nil, nil, true, allTasks)
context, err := fi.NewContext(target, nil, cloud, nil, nil, nil, true, allTasks)
if err != nil {
t.Fatalf("error building context: %v", err)
}

2
upup/pkg/fi/cloudup/awstasks/securitygroup_test.go
View File

@ -125,7 +125,7 @@ func TestSecurityGroupCreate(t *testing.T) {
Cloud: cloud,
}
context, err := fi.NewContext(target, cloud, nil, nil, nil, true, allTasks)
context, err := fi.NewContext(target, nil, cloud, nil, nil, nil, true, allTasks)
if err != nil {
t.Fatalf("error building context: %v", err)
}

2
upup/pkg/fi/cloudup/awstasks/vpc_test.go
View File

@ -51,7 +51,7 @@ func TestVPCCreate(t *testing.T) {
Cloud: cloud,
}
context, err := fi.NewContext(target, cloud, nil, nil, nil, true, allTasks)
context, err := fi.NewContext(target, nil, cloud, nil, nil, nil, true, allTasks)
if err != nil {
t.Fatalf("error building context: %v", err)
}

5
upup/pkg/fi/context.go
View File

@ -21,6 +21,7 @@ import (
"fmt"
"github.com/golang/glog"
"io/ioutil"
"k8s.io/kops/pkg/apis/kops"
"k8s.io/kops/util/pkg/vfs"
"k8s.io/kubernetes/federation/pkg/dnsprovider"
"os"
@ -35,6 +36,7 @@ type Context struct {
Target Target
DNS dnsprovider.Interface
Cloud Cloud
Cluster *kops.Cluster
Keystore Keystore
SecretStore SecretStore
ClusterConfigBase vfs.Path
@ -44,9 +46,10 @@ type Context struct {
tasks map[string]Task
}
func NewContext(target Target, cloud Cloud, keystore Keystore, secretStore SecretStore, clusterConfigBase vfs.Path, checkExisting bool, tasks map[string]Task) (*Context, error) {
func NewContext(target Target, cluster *kops.Cluster, cloud Cloud, keystore Keystore, secretStore SecretStore, clusterConfigBase vfs.Path, checkExisting bool, tasks map[string]Task) (*Context, error) {
c := &Context{
Cloud: cloud,
Cluster: cluster,
Target: target,
Keystore: keystore,
SecretStore: secretStore,

1
upup/pkg/fi/fitasks/BUILD.bazel
View File

@ -17,6 +17,7 @@ go_library(
],
visibility = ["//visibility:public"],
deps = [
"//pkg/acls:go_default_library",
"//pkg/pki:go_default_library",
"//upup/pkg/fi:go_default_library",
"//upup/pkg/fi/secrets:go_default_library",

10
upup/pkg/fi/fitasks/managedfile.go
View File

@ -19,6 +19,7 @@ package fitasks
import (
"fmt"
"k8s.io/apimachinery/pkg/util/validation/field"
"k8s.io/kops/pkg/acls"
"k8s.io/kops/upup/pkg/fi"
"os"
)
@ -87,7 +88,14 @@ func (_ *ManagedFile) Render(c *fi.Context, a, e, changes *ManagedFile) error {
return fmt.Errorf("error reading contents of ManagedFile: %v", err)
}
err = c.ClusterConfigBase.Join(location).WriteFile(data)
p := c.ClusterConfigBase.Join(location)
acl, err := acls.GetACL(p, c.Cluster)
if err != nil {
return err
}
err = p.WriteFile(data, acl)
if err != nil {
return fmt.Errorf("error creating ManagedFile %q: %v", location, err)
}

1
upup/pkg/fi/fitasks/mirrorsecrets.go
View File

@ -75,6 +75,5 @@ func (s *MirrorSecrets) CheckChanges(a, e, changes *MirrorSecrets) error {
// Render implements fi.Task::Render
func (_ *MirrorSecrets) Render(c *fi.Context, a, e, changes *MirrorSecrets) error {
secrets := c.SecretStore
return secrets.MirrorTo(e.MirrorPath)
}

2
upup/pkg/fi/nodeup/command.go
View File

@ -258,7 +258,7 @@ func (c *NodeUpCommand) Run(out io.Writer) error {
return fmt.Errorf("unsupported target type %q", c.Target)
}
context, err := fi.NewContext(target, cloud, keyStore, secretStore, configBase, checkExisting, taskMap)
context, err := fi.NewContext(target, nil, cloud, keyStore, secretStore, configBase, checkExisting, taskMap)
if err != nil {
glog.Exitf("error building context: %v", err)
}

4
upup/pkg/fi/nodeup/template_functions.go
View File

@ -67,7 +67,7 @@ func newTemplateFunctions(nodeupConfig *nodeup.Config, cluster *api.Cluster, ins
return nil, fmt.Errorf("error building secret store path: %v", err)
}
t.secretStore = secrets.NewVFSSecretStore(p)
t.secretStore = secrets.NewVFSSecretStore(cluster, p)
} else {
return nil, fmt.Errorf("SecretStore not set")
}
@ -79,7 +79,7 @@ func newTemplateFunctions(nodeupConfig *nodeup.Config, cluster *api.Cluster, ins
return nil, fmt.Errorf("error building key store path: %v", err)
}
t.keyStore = fi.NewVFSCAStore(p)
t.keyStore = fi.NewVFSCAStore(cluster, p)
} else {
return nil, fmt.Errorf("KeyStore not set")
}

1
upup/pkg/fi/secrets/BUILD.bazel
View File

@ -8,6 +8,7 @@ go_library(
],
visibility = ["//visibility:public"],
deps = [
"//pkg/acls:go_default_library",
"//pkg/apis/kops:go_default_library",
"//pkg/client/clientset_generated/clientset/typed/kops/internalversion:go_default_library",
"//pkg/pki:go_default_library",

12
upup/pkg/fi/secrets/clientset_secretstore.go
View File

@ -25,6 +25,7 @@ import (
"github.com/golang/glog"
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/kops/pkg/acls"
"k8s.io/kops/pkg/apis/kops"
kopsinternalversion "k8s.io/kops/pkg/client/clientset_generated/clientset/typed/kops/internalversion"
"k8s.io/kops/pkg/pki"
@ -37,6 +38,7 @@ const NamePrefix = "token-"
// ClientsetSecretStore is a SecretStore backed by Keyset objects in an API server
type ClientsetSecretStore struct {
cluster *kops.Cluster
namespace string
clientset kopsinternalversion.KopsInterface
}
@ -44,8 +46,9 @@ type ClientsetSecretStore struct {
var _ fi.SecretStore = &ClientsetSecretStore{}
// NewClientsetSecretStore is the constructor for ClientsetSecretStore
func NewClientsetSecretStore(clientset kopsinternalversion.KopsInterface, namespace string) fi.SecretStore {
func NewClientsetSecretStore(cluster *kops.Cluster, clientset kopsinternalversion.KopsInterface, namespace string) fi.SecretStore {
c := &ClientsetSecretStore{
cluster: cluster,
clientset: clientset,
namespace: namespace,
}
@ -81,7 +84,12 @@ func (c *ClientsetSecretStore) MirrorTo(basedir vfs.Path) error {
return fmt.Errorf("error serializing secret: %v", err)
}
if err := p.WriteFile(data); err != nil {
acl, err := acls.GetACL(p, c.cluster)
if err != nil {
return err
}
if err := p.WriteFile(data, acl); err != nil {
return fmt.Errorf("error writing secret to %q: %v", p, err)
}
}

19
upup/pkg/fi/secrets/vfs_secretstore.go
View File

@ -20,19 +20,23 @@ import (
"encoding/json"
"fmt"
"github.com/golang/glog"
"k8s.io/kops/pkg/acls"
"k8s.io/kops/pkg/apis/kops"
"k8s.io/kops/upup/pkg/fi"
"k8s.io/kops/util/pkg/vfs"
"os"
)
type VFSSecretStore struct {
cluster *kops.Cluster
basedir vfs.Path
}
var _ fi.SecretStore = &VFSSecretStore{}
func NewVFSSecretStore(basedir vfs.Path) fi.SecretStore {
func NewVFSSecretStore(cluster *kops.Cluster, basedir vfs.Path) fi.SecretStore {
c := &VFSSecretStore{
cluster: cluster,
basedir: basedir,
}
return c
@ -48,7 +52,7 @@ func (c *VFSSecretStore) MirrorTo(basedir vfs.Path) error {
}
glog.V(2).Infof("Mirroring secret store from %q to %q", c.basedir, basedir)
return vfs.CopyTree(c.basedir, basedir)
return vfs.CopyTree(c.basedir, basedir, func(p vfs.Path) (vfs.ACL, error) { return acls.GetACL(p, c.cluster) })
}
func BuildVfsSecretPath(basedir vfs.Path, name string) vfs.Path {
@ -117,7 +121,12 @@ func (c *VFSSecretStore) GetOrCreateSecret(id string, secret *fi.Secret) (*fi.Se
return s, false, nil
}
err = c.createSecret(secret, p)
acl, err := acls.GetACL(p, c.cluster)
if err != nil {
return nil, false, err
}
err = c.createSecret(secret, p, acl)
if err != nil {
if os.IsExist(err) && i == 0 {
glog.Infof("Got already-exists error when writing secret; likely due to concurrent creation. Will retry")
@ -157,10 +166,10 @@ func (c *VFSSecretStore) loadSecret(p vfs.Path) (*fi.Secret, error) {
}
// createSecret writes the secret, but only if it does not exists
func (c *VFSSecretStore) createSecret(s *fi.Secret, p vfs.Path) error {
func (c *VFSSecretStore) createSecret(s *fi.Secret, p vfs.Path, acl vfs.ACL) error {
data, err := json.Marshal(s)
if err != nil {
return fmt.Errorf("error serializing secret: %v", err)
}
return p.CreateFile(data)
return p.CreateFile(data, acl)
}

35
upup/pkg/fi/vfs_castore.go
View File

@ -33,12 +33,15 @@ import (
"github.com/golang/glog"
"golang.org/x/crypto/ssh"
"k8s.io/kops/pkg/acls"
"k8s.io/kops/pkg/apis/kops"
"k8s.io/kops/pkg/pki"
"k8s.io/kops/util/pkg/vfs"
)
type VFSCAStore struct {
basedir vfs.Path
cluster *kops.Cluster
mutex sync.Mutex
cachedCAs map[string]*cachedEntry
@ -51,9 +54,10 @@ type cachedEntry struct {
var _ CAStore = &VFSCAStore{}
func NewVFSCAStore(basedir vfs.Path) CAStore {
func NewVFSCAStore(cluster *kops.Cluster, basedir vfs.Path) CAStore {
c := &VFSCAStore{
basedir: basedir,
cluster: cluster,
cachedCAs: make(map[string]*cachedEntry),
}
@ -417,7 +421,10 @@ func (c *VFSCAStore) MirrorTo(basedir vfs.Path) error {
}
glog.V(2).Infof("Mirroring key store from %q to %q", c.basedir, basedir)
return vfs.CopyTree(c.basedir, basedir)
aclOracle := func(p vfs.Path) (vfs.ACL, error) {
return acls.GetACL(p, c.cluster)
}
return vfs.CopyTree(c.basedir, basedir, aclOracle)
}
func (c *VFSCAStore) IssueCert(signer string, id string, serial *big.Int, privateKey *pki.PrivateKey, template *x509.Certificate) (*pki.Certificate, error) {
@ -625,7 +632,11 @@ func (c *VFSCAStore) storePrivateKey(privateKey *pki.PrivateKey, p vfs.Path) err
return err
}
return p.WriteFile(data.Bytes())
acl, err := acls.GetACL(p, c.cluster)
if err != nil {
return err
}
return p.WriteFile(data.Bytes(), acl)
}
func (c *VFSCAStore) storeCertificate(cert *pki.Certificate, p vfs.Path) error {
@ -636,7 +647,11 @@ func (c *VFSCAStore) storeCertificate(cert *pki.Certificate, p vfs.Path) error {
return err
}
return p.WriteFile(data.Bytes())
acl, err := acls.GetACL(p, c.cluster)
if err != nil {
return err
}
return p.WriteFile(data.Bytes(), acl)
}
func (c *VFSCAStore) buildSerial() *big.Int {
@ -699,7 +714,13 @@ func (c *VFSCAStore) AddSSHPublicKey(name string, pubkey []byte) error {
}
p := c.buildSSHPublicKeyPath(name, id)
return c.storeData(pubkey, p)
acl, err := acls.GetACL(p, c.cluster)
if err != nil {
return err
}
return p.WriteFile(pubkey, acl)
}
func (c *VFSCAStore) buildSSHPublicKeyPath(name string, id string) vfs.Path {
@ -708,10 +729,6 @@ func (c *VFSCAStore) buildSSHPublicKeyPath(name string, id string) vfs.Path {
return c.basedir.Join("ssh", "public", name, id)
}
func (c *VFSCAStore) storeData(data []byte, p vfs.Path) error {
return p.WriteFile(data)
}
func (c *VFSCAStore) FindSSHPublicKeys(name string) ([]*KeystoreItem, error) {
p := c.basedir.Join("ssh", "public", name)

2
upup/pkg/kutil/convert_kubeup_cluster.go
View File

@ -471,7 +471,7 @@ func (x *ConvertKubeupCluster) Upgrade() error {
}
// TODO: No longer needed?
err = registry.WriteConfigDeprecated(newConfigBase.Join(registry.PathClusterCompleted), fullCluster)
err = registry.WriteConfigDeprecated(cluster, newConfigBase.Join(registry.PathClusterCompleted), fullCluster)
if err != nil {
return fmt.Errorf("error writing completed cluster spec: %v", err)
}

6
util/pkg/vfs/fs.go
View File

@ -45,7 +45,7 @@ func (p *FSPath) Join(relativePath ...string) Path {
return &FSPath{location: joined}
}
func (p *FSPath) WriteFile(data []byte) error {
func (p *FSPath) WriteFile(data []byte, acl ACL) error {
dir := path.Dir(p.location)
err := os.MkdirAll(dir, 0755)
if err != nil {
@ -94,7 +94,7 @@ func (p *FSPath) WriteFile(data []byte) error {
// TODO: should we take a file lock or equivalent here? Can we use RENAME_NOREPLACE ?
var createFileLock sync.Mutex
func (p *FSPath) CreateFile(data []byte) error {
func (p *FSPath) CreateFile(data []byte, acl ACL) error {
createFileLock.Lock()
defer createFileLock.Unlock()
@ -108,7 +108,7 @@ func (p *FSPath) CreateFile(data []byte) error {
return err
}
return p.WriteFile(data)
return p.WriteFile(data, acl)
}
func (p *FSPath) ReadFile() ([]byte, error) {

27
util/pkg/vfs/gsfs.go
View File

@ -55,6 +55,13 @@ var gcsReadBackoff = wait.Backoff{
Steps: 4,
}
// GSAcl is an ACL implementation for objects on Google Cloud Storage
type GSAcl struct {
Acl []*storage.ObjectAccessControl
}
var _ ACL = &GSAcl{}
// gcsWriteBackoff is the backoff strategy for GCS write retries
var gcsWriteBackoff = wait.Backoff{
Duration: time.Second,
@ -86,6 +93,11 @@ func (p *GSPath) Object() string {
return p.key
}
// Client returns the storage.Service bound to this path
func (p *GSPath) Client() *storage.Service {
return p.client
}
func (p *GSPath) String() string {
return p.Path()
}
@ -122,7 +134,7 @@ func (p *GSPath) Join(relativePath ...string) Path {
}
}
func (p *GSPath) WriteFile(data []byte) error {
func (p *GSPath) WriteFile(data []byte, acl ACL) error {
done, err := RetryWithBackoff(gcsWriteBackoff, func() (bool, error) {
glog.V(4).Infof("Writing file %q", p)
@ -135,6 +147,15 @@ func (p *GSPath) WriteFile(data []byte) error {
Name: p.key,
Md5Hash: base64.StdEncoding.EncodeToString(md5Hash.HashValue),
}
if acl != nil {
gsAcl, ok := acl.(*GSAcl)
if !ok {
return true, fmt.Errorf("write to %s with ACL of unexpected type %T", p, acl)
}
obj.Acl = gsAcl.Acl
}
r := bytes.NewReader(data)
_, err = p.client.Objects.Insert(p.bucket, obj).Media(r).Do()
if err != nil {
@ -159,7 +180,7 @@ func (p *GSPath) WriteFile(data []byte) error {
// TODO: should we enable versioning?
var createFileLockGCS sync.Mutex
func (p *GSPath) CreateFile(data []byte) error {
func (p *GSPath) CreateFile(data []byte, acl ACL) error {
createFileLockGCS.Lock()
defer createFileLockGCS.Unlock()
@ -173,7 +194,7 @@ func (p *GSPath) CreateFile(data []byte) error {
return err
}
return p.WriteFile(data)
return p.WriteFile(data, acl)
}
// ReadFile implements Path::ReadFile

4
util/pkg/vfs/k8sfs.go
View File

@ -76,11 +76,11 @@ func (p *KubernetesPath) Join(relativePath ...string) Path {
}
}
func (p *KubernetesPath) WriteFile(data []byte) error {
func (p *KubernetesPath) WriteFile(data []byte, acl ACL) error {
return fmt.Errorf("KubernetesPath::WriteFile not supported")
}
func (p *KubernetesPath) CreateFile(data []byte) error {
func (p *KubernetesPath) CreateFile(data []byte, acl ACL) error {
return fmt.Errorf("KubernetesPath::CreateFile not supported")
}

6
util/pkg/vfs/memfs.go
View File

@ -87,18 +87,18 @@ func (p *MemFSPath) Join(relativePath ...string) Path {
return current
}
func (p *MemFSPath) WriteFile(data []byte) error {
func (p *MemFSPath) WriteFile(data []byte, acl ACL) error {
p.contents = data
return nil
}
func (p *MemFSPath) CreateFile(data []byte) error {
func (p *MemFSPath) CreateFile(data []byte, acl ACL) error {
// Check if exists
if p.contents != nil {
return os.ErrExist
}
return p.WriteFile(data)
return p.WriteFile(data, acl)
}
func (p *MemFSPath) ReadFile() ([]byte, error) {

6
util/pkg/vfs/s3fs.go
View File

@ -103,7 +103,7 @@ func (p *S3Path) Join(relativePath ...string) Path {
}
}
func (p *S3Path) WriteFile(data []byte) error {
func (p *S3Path) WriteFile(data []byte, aclObj ACL) error {
client, err := p.client()
if err != nil {
return err
@ -149,7 +149,7 @@ func (p *S3Path) WriteFile(data []byte) error {
// TODO: should we enable versioning?
var createFileLockS3 sync.Mutex
func (p *S3Path) CreateFile(data []byte) error {
func (p *S3Path) CreateFile(data []byte, acl ACL) error {
createFileLockS3.Lock()
defer createFileLockS3.Unlock()
@ -163,7 +163,7 @@ func (p *S3Path) CreateFile(data []byte) error {
return err
}
return p.WriteFile(data)
return p.WriteFile(data, acl)
}
func (p *S3Path) ReadFile() ([]byte, error) {

6
util/pkg/vfs/sshfs.go
View File

@ -142,7 +142,7 @@ func mkdirAll(sftpClient *sftp.Client, dir string) error {
return nil
}
func (p *SSHPath) WriteFile(data []byte) error {
func (p *SSHPath) WriteFile(data []byte, acl ACL) error {
sftpClient, err := p.newClient()
if err != nil {
return err
@ -197,7 +197,7 @@ func (p *SSHPath) WriteFile(data []byte) error {
// Not a great approach, but fine for a single process (with low concurrency)
var createFileLockSSH sync.Mutex
func (p *SSHPath) CreateFile(data []byte) error {
func (p *SSHPath) CreateFile(data []byte, acl ACL) error {
createFileLockSSH.Lock()
defer createFileLockSSH.Unlock()
@ -211,7 +211,7 @@ func (p *SSHPath) CreateFile(data []byte) error {
return err
}
return p.WriteFile(data)
return p.WriteFile(data, acl)
}
func (p *SSHPath) ReadFile() ([]byte, error) {

10
util/pkg/vfs/vfs.go
View File

@ -34,13 +34,19 @@ func IsDirectory(p Path) bool {
return err == nil
}
type ACL interface {
}
type ACLOracle func(Path) (ACL, error)
// Path is a path in the VFS space, which we can read, write, list etc
type Path interface {
Join(relativePath ...string) Path
ReadFile() ([]byte, error)
WriteFile(data []byte) error
WriteFile(data []byte, acl ACL) error
// CreateFile writes the file contents, but only if the file does not already exist
CreateFile(data []byte) error
CreateFile(data []byte, acl ACL) error
// Remove deletes the file
Remove() error

11
util/pkg/vfs/vfssync.go
View File

@ -161,7 +161,7 @@ func SyncDir(src *VFSScan, destBase Path) error {
if destData == nil || !bytes.Equal(srcData, destData) {
glog.V(2).Infof("Copying data from %s to %s", f, destFile)
err = destFile.WriteFile(srcData)
err = destFile.WriteFile(srcData, nil)
if err != nil {
return fmt.Errorf("error writing dest file %q: %v", f, err)
}
@ -215,7 +215,7 @@ func hashesMatch(src, dest Path) (bool, error) {
}
// CopyTree copies all files in src to dest. It copies the whole recursive subtree of files.
func CopyTree(src Path, dest Path) error {
func CopyTree(src Path, dest Path, aclOracle ACLOracle) error {
srcFiles, err := src.ReadTree()
if err != nil {
return fmt.Errorf("error reading source directory %q: %v", src, err)
@ -269,8 +269,13 @@ func CopyTree(src Path, dest Path) error {
}
if destData == nil || !bytes.Equal(srcData, destData) {
acl, err := aclOracle(destFile)
if err != nil {
return err
}
glog.V(2).Infof("Copying data from %s to %s", srcFile, destFile)
err = destFile.WriteFile(srcData)
err = destFile.WriteFile(srcData, acl)
if err != nil {
return fmt.Errorf("error writing dest file %q: %v", destFile, err)
}