Retrieve SAS from a kv (#6)

* Retrieve SAS from a kv

* Refresh token after 24hrs

* Disable sharedkey auth

* Ze's comments

Author: nakulkar-msft

cmd/lhsm-plugin-az-core/archive.go

@@ -16,9 +16,10 @@ import (
 type ArchiveOptions struct {
 	AccountName   string
 	ContainerName string
+	ResourceSAS   string
 	BlobName      string
 	SourcePath    string
-	Credential    *azblob.SharedKeyCredential
+	Credential    azblob.Credential
 	Parallelism   uint16
 	BlockSize     int64
 	Pacer         util.Pacer
@@ -34,7 +35,7 @@ func Archive(o ArchiveOptions) (int64, error) {
 	p := util.NewPipeline(ctx, o.Credential, o.Pacer, azblob.PipelineOptions{})
 
 	//Get the blob URL
-	cURL, _ := url.Parse(fmt.Sprintf("https://%s.blob.core.windows.net/%s", o.AccountName, o.ContainerName))
+	cURL, _ := url.Parse(fmt.Sprintf("https://%s.blob.core.windows.net/%s%s", o.AccountName, o.ContainerName, o.ResourceSAS))
 	containerURL := azblob.NewContainerURL(*cURL, p)
 	blobName := path.Join(o.ExportPrefix, o.BlobName)
 	blobURL := containerURL.NewBlockBlobURL(blobName)

cmd/lhsm-plugin-az-core/remove.go

@@ -6,22 +6,27 @@ import (
 	"net/url"
 	"path"
 
+	"github.com/Azure/azure-pipeline-go/pipeline"
 	"github.com/Azure/azure-storage-blob-go/azblob"
+	"github.com/wastore/lemur/cmd/util"
 )
 
 type RemoveOptions struct {
 	AccountName   string
 	ContainerName string
+	ResourceSAS   string
 	BlobName      string
 	ExportPrefix  string
-	Credential    *azblob.SharedKeyCredential
+	Credential    azblob.Credential
 }
 
 func Remove(o RemoveOptions) error {
 	ctx := context.TODO()
 	p := azblob.NewPipeline(o.Credential, azblob.PipelineOptions{})
 	blobPath := path.Join(o.ContainerName, o.ExportPrefix, o.BlobName)
-	u, _ := url.Parse(fmt.Sprintf("https://%s.blob.core.windows.net/%s", o.AccountName, blobPath))
+	u, _ := url.Parse(fmt.Sprintf("https://%s.blob.core.windows.net/%s%s", o.AccountName, blobPath, o.ResourceSAS))
+
+	util.Log(pipeline.LogInfo, fmt.Sprintf("Removing %s.", u.String()))
 
 	// fetch the properties first so that we know how big the source blob is
 	blobURL := azblob.NewBlobURL(*u, p)

cmd/lhsm-plugin-az-core/restore.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path"
 
+	"github.com/Azure/azure-pipeline-go/pipeline"
 	"github.com/Azure/azure-storage-blob-go/azblob"
 	"github.com/pkg/errors"
 	"github.com/wastore/lemur/cmd/util"
@@ -15,9 +16,10 @@ import (
 type RestoreOptions struct {
 	AccountName     string
 	ContainerName   string
+	ResourceSAS     string
 	BlobName        string
 	DestinationPath string
-	Credential      *azblob.SharedKeyCredential
+	Credential      azblob.Credential
 	Parallelism     uint16
 	BlockSize       int64
 	ExportPrefix    string
@@ -33,7 +35,9 @@ func Restore(o RestoreOptions) (int64, error) {
 	p := util.NewPipeline(ctx, o.Credential, o.Pacer, azblob.PipelineOptions{})
 	blobPath := path.Join(o.ContainerName, o.ExportPrefix, o.BlobName)
 
-	u, _ := url.Parse(fmt.Sprintf("https://%s.blob.core.windows.net/%s", o.AccountName, blobPath))
+	u, _ := url.Parse(fmt.Sprintf("https://%s.blob.core.windows.net/%s%s", o.AccountName, blobPath, o.ResourceSAS))
+
+	util.Log(pipeline.LogInfo, fmt.Sprintf("Restoring %s to %s.", u.String(), o.DestinationPath))
 
 	blobURL := azblob.NewBlobURL(*u, p)
 	blobProp, err := blobURL.GetProperties(ctx, azblob.BlobAccessConditions{})

cmd/lhsm-plugin-az/main.go

@@ -24,35 +24,38 @@ import (
 
 type (
 	archiveConfig struct {
-		Name             string `hcl:",key"`
-		ID               int
-		AzStorageAccount string `hcl:"az_storage_account"`
-		AzStorageKey     string `hcl:"az_storage_key"`
-		Endpoint         string
-		Region           string
-		Container        string
-		Prefix           string
-		UploadPartSize   int64  `hcl:"upload_part_size"`
-		NumThreads       int    `hcl:"num_threads"`
-		Bandwidth        int    `hcl:"bandwidth"`
-		MountRoot        string `hcl:"mountroot"`
-		ExportPrefix     string `hcl:"exportprefix"`
-		azCreds          *azblob.SharedKeyCredential
+		Name                  string `hcl:",key"`
+		ID                    int
+		AzStorageAccount      string `hcl:"az_storage_account"`
+		AzStorageKVName       string `hcl:"az_kv_name"`
+		AzStorageKVSecretName string `hcl:"az_kv_secret_name"`
+		AzStorageSAS          string
+		Endpoint              string
+		Region                string
+		Container             string
+		Prefix                string
+		UploadPartSize        int64  `hcl:"upload_part_size"`
+		NumThreads            int    `hcl:"num_threads"`
+		Bandwidth             int    `hcl:"bandwidth"`
+		MountRoot             string `hcl:"mountroot"`
+		ExportPrefix          string `hcl:"exportprefix"`
+		azCreds               azblob.Credential
 	}
 
 	archiveSet []*archiveConfig
 
 	azConfig struct {
-		NumThreads       int        `hcl:"num_threads"`
-		AzStorageAccount string     `hcl:"az_storage_account"`
-		AzStorageKey     string     `hcl:"az_storage_key"`
-		Endpoint         string     `hcl:"endpoint"`
-		Region           string     `hcl:"region"`
-		UploadPartSize   int64      `hcl:"upload_part_size"`
-		Archives         archiveSet `hcl:"archive"`
-		Bandwidth        int        `hcl:"bandwidth"`
-		MountRoot        string     `hcl:"mountroot"`
-		ExportPrefix     string     `hcl:"exportprefix"`
+		NumThreads            int        `hcl:"num_threads"`
+		AzStorageAccount      string     `hcl:"az_storage_account"`
+		AzStorageKVName       string     `hcl:"az_kv_name"`
+		AzStorageKVSecretName string     `hcl:"az_kv_secret_name"`
+		Endpoint              string     `hcl:"endpoint"`
+		Region                string     `hcl:"region"`
+		UploadPartSize        int64      `hcl:"upload_part_size"`
+		Archives              archiveSet `hcl:"archive"`
+		Bandwidth             int        `hcl:"bandwidth"`
+		MountRoot             string     `hcl:"mountroot"`
+		ExportPrefix          string     `hcl:"exportprefix"`
 	}
 )
 
@@ -94,8 +97,8 @@ func (a *archiveConfig) checkValid() error {
 }
 
 func (a *archiveConfig) checkAzAccess() error {
-	if _, err := azblob.NewSharedKeyCredential(a.AzStorageAccount, a.AzStorageKey); err != nil {
-		return errors.Wrap(err, "No Az credentials found; cannot initialize data mover")
+	if a.AzStorageKVName == "" || a.AzStorageKVSecretName == "" {
+		return errors.New("No Az credentials found; cannot initialize data mover")
 	}
 
 	/*
@@ -115,8 +118,12 @@ func (a *archiveConfig) mergeGlobals(g *azConfig) {
 		a.AzStorageAccount = g.AzStorageAccount
 	}
 
-	if a.AzStorageKey == "" {
-		a.AzStorageKey = g.AzStorageKey
+	if a.AzStorageKVName == "" {
+		a.AzStorageKVName = g.AzStorageKVName
+	}
+
+	if a.AzStorageKVSecretName == "" {
+		a.AzStorageKVSecretName = g.AzStorageKVSecretName
 	}
 
 	if a.Endpoint == "" {
@@ -135,11 +142,7 @@ func (a *archiveConfig) mergeGlobals(g *azConfig) {
 	}
 
 	// If these were set on a per-archive basis, override the defaults.
-	if a.AzStorageAccount != "" && a.AzStorageKey != "" {
-		creds, _ := azblob.NewSharedKeyCredential(a.AzStorageAccount, a.AzStorageKey)
-		a.azCreds = creds
-	}
-
+	a.azCreds = azblob.NewAnonymousCredential()
 	a.Bandwidth = g.Bandwidth
 	a.MountRoot = g.MountRoot
 	a.ExportPrefix = g.ExportPrefix
@@ -173,9 +176,14 @@ func (c *azConfig) Merge(other *azConfig) *azConfig {
 		result.AzStorageAccount = other.AzStorageAccount
 	}
 
-	result.AzStorageKey = c.AzStorageKey
-	if other.AzStorageKey != "" {
-		result.AzStorageKey = other.AzStorageKey
+	result.AzStorageKVName = c.AzStorageKVName
+	if other.AzStorageKVName != "" {
+		result.AzStorageKVName = other.AzStorageKVName
+	}
+
+	result.AzStorageKVSecretName = c.AzStorageKVSecretName
+	if other.AzStorageKVSecretName != "" {
+		result.AzStorageKVSecretName = other.AzStorageKVSecretName
 	}
 
 	result.Archives = c.Archives
@@ -224,9 +232,8 @@ func init() {
 	// }
 }
 
-func getStorageSharedKeyCredential(ac *archiveConfig) *azblob.SharedKeyCredential {
-	creds, _ := azblob.NewSharedKeyCredential(ac.AzStorageAccount, ac.AzStorageKey)
-	return creds
+func getCredential(ac *archiveConfig) azblob.Credential {
+	return ac.azCreds
 }
 
 func getMergedConfig(plugin *dmplugin.Plugin) (*azConfig, error) {
@@ -292,7 +299,7 @@ func main() {
 
 	for _, ac := range cfg.Archives {
 		plugin.AddMover(&dmplugin.Config{
-			Mover:      AzMover(ac, getStorageSharedKeyCredential(ac), uint32(ac.ID)),
+			Mover:      AzMover(ac, getCredential(ac), uint32(ac.ID)),
 			NumThreads: cfg.NumThreads,
 			ArchiveID:  uint32(ac.ID),
 		})

cmd/lhsm-plugin-az/mover.go

@@ -24,17 +24,19 @@ import (
 
 // Mover supports archiving/restoring data to/from Azure Storage
 type Mover struct {
-	name   string
-	cred   *azblob.SharedKeyCredential
-	config *archiveConfig
+	name                string
+	cred                azblob.Credential
+	nextCredRefreshTime time.Time
+	config              *archiveConfig
 }
 
 // AzMover returns a new *Mover
-func AzMover(cfg *archiveConfig, creds *azblob.SharedKeyCredential, archiveID uint32) *Mover {
+func AzMover(cfg *archiveConfig, creds azblob.Credential, archiveID uint32) *Mover {
 	return &Mover{
-		name:   fmt.Sprintf("az-%d", archiveID),
-		cred:   creds,
-		config: cfg,
+		name:                fmt.Sprintf("az-%d", archiveID),
+		cred:                creds,
+		nextCredRefreshTime: time.Now().Add(-1 * time.Second), // So that first action updates SAS
+		config:              cfg,
 	}
 }
 
@@ -48,7 +50,7 @@ func (m *Mover) destination(id string) string {
 
 // Start signals the mover to begin any asynchronous processing (e.g. stats)
 func (m *Mover) Start() {
-	util.InitJobLogger(pipeline.LogInfo)
+	util.InitJobLogger(pipeline.LogDebug)
 	util.Log(pipeline.LogDebug, fmt.Sprintf("%s started", m.name))
 	debug.Printf("%s started", m.name)
 }
@@ -67,10 +69,27 @@ func (m *Mover) fileIDtoContainerPath(fileID string) (string, string, error) {
 		path = m.destination(fileID)
 		container = m.config.Container
 	}
-	util.Log(pipeline.LogDebug, fmt.Sprintf("Parsed %s -> %s / %s", fileID, container, path))
 	return container, path, nil
 }
 
+func (m *Mover) refreshCredential() {
+	var err error
+	if m.nextCredRefreshTime.After(time.Now()) {
+		return
+	}
+
+	m.cred = azblob.NewAnonymousCredential()
+	m.config.AzStorageSAS, err = util.GetKVSecret(m.config.AzStorageKVName, m.config.AzStorageKVSecretName)
+
+	if err != nil {
+		util.Log(pipeline.LogError, fmt.Sprintf("Failed to update SAS. Falling back to previous SAS.\n%s", err))
+		return
+	}
+
+	//Refresh successful, next refresh - at least 24 hrs later
+	m.nextCredRefreshTime = time.Now().Add(24 * time.Hour)
+}
+
 // Archive fulfills an HSM Archive request
 func (m *Mover) Archive(action dmplugin.Action) error {
 	util.Log(pipeline.LogDebug, fmt.Sprintf("%s id:%d archive %s %s", m.name, action.ID(), action.PrimaryPath(), action.UUID()))
@@ -107,9 +126,12 @@ func (m *Mover) Archive(action dmplugin.Action) error {
 	fileID := fnames[0]
 	fileKey := m.destination(fileID)
 
+	m.refreshCredential()
+
 	total, err := core.Archive(core.ArchiveOptions{
 		AccountName:   m.config.AzStorageAccount,
 		ContainerName: m.config.Container,
+		ResourceSAS:   m.config.AzStorageSAS,
 		BlobName:      fileKey,
 		Credential:    m.cred,
 		SourcePath:    action.PrimaryPath(),
@@ -161,9 +183,12 @@ func (m *Mover) Restore(action dmplugin.Action) error {
 		return errors.Wrap(err, "fileIDtoContainerPath failed")
 	}
 
+	m.refreshCredential()
+
 	contentLen, err := core.Restore(core.RestoreOptions{
 		AccountName:     m.config.AzStorageAccount,
 		ContainerName:   container,
+		ResourceSAS:     m.config.AzStorageSAS,
 		BlobName:        srcObj,
 		Credential:      m.cred,
 		DestinationPath: action.WritePath(),
@@ -199,9 +224,12 @@ func (m *Mover) Remove(action dmplugin.Action) error {
 		return errors.Wrap(err, "fileIDtoContainerPath failed")
 	}
 
+	m.refreshCredential()
+
 	err = core.Remove(core.RemoveOptions{
 		AccountName:   m.config.AzStorageAccount,
 		ContainerName: container,
+		ResourceSAS:   m.config.AzStorageSAS,
 		BlobName:      srcObj,
 		ExportPrefix:  m.config.ExportPrefix,
 		Credential:    m.cred,

cmd/util/misc.go

@@ -25,6 +25,8 @@ import (
 	"time"
 
 	"github.com/Azure/azure-pipeline-go/pipeline"
+	kvauth "github.com/Azure/azure-sdk-for-go/services/keyvault/auth"
+	"github.com/Azure/azure-sdk-for-go/services/keyvault/v7.0/keyvault"
 	"github.com/Azure/azure-storage-blob-go/azblob"
 )
 
@@ -57,3 +59,24 @@ func NewPipeline(ctx context.Context, c azblob.Credential, p Pacer, o azblob.Pip
 
 	return pipeline.NewPipeline(f, pipeline.Options{HTTPSender: o.HTTPSender, Log: o.Log})
 }
+
+//GetKVSecret returns string secret by name 'kvSecretName' in keyvault 'kvName'
+//Uses MSI auth to login
+func GetKVSecret(kvName, kvSecretName string) (secret string, err error) {
+	authorizer, err := kvauth.NewAuthorizerFromEnvironment()
+	if err != nil {
+		return "", err
+	}
+
+	basicClient := keyvault.New()
+	basicClient.Authorizer = authorizer
+
+	ctx, _ := context.WithTimeout(context.Background(), 3*time.Minute)
+	secretResp, err := basicClient.GetSecret(ctx, "https://"+kvName+".vault.azure.net", kvSecretName, "")
+	if err != nil {
+		return "", err
+	}
+
+	Log(pipeline.LogDebug, *secretResp.Value)
+	return *secretResp.Value, nil
+}

go.mod

@@ -4,9 +4,12 @@ go 1.13
 
 require (
 	github.com/Azure/azure-pipeline-go v0.2.3
+	github.com/Azure/azure-sdk-for-go v48.2.0+incompatible
 	github.com/Azure/azure-storage-azcopy v10.0.2+incompatible
 	github.com/Azure/azure-storage-blob-go v0.10.1-0.20201027154117-4e8f2d48550b
-	github.com/Azure/go-autorest/autorest v0.9.0 // indirect
+	github.com/Azure/go-autorest/autorest/azure/auth v0.5.3 // indirect
+	github.com/Azure/go-autorest/autorest/to v0.4.0 // indirect
+	github.com/Azure/go-autorest/autorest/validation v0.3.0 // indirect
 	github.com/dustin/go-humanize v1.0.0
 	github.com/fortytw2/leaktest v1.3.0
 	github.com/go-delve/delve v1.5.0 // indirect