From cfbef13aff53a06095094501138993330e1dd34b Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Tue, 18 Jul 2017 15:40:50 +0200 Subject: [PATCH] Major refactoring to how repository connections are established and managed to simplify programmatic API usage and CLI. Removed unnecessary and confusing options, improved the API. --- README.md | 2 +- cli/command_backup.go | 12 +- cli/command_backups.go | 6 +- cli/command_cat.go | 2 +- cli/command_cleanup.go | 16 +- cli/command_connect.go | 39 +- cli/command_create.go | 142 +++---- cli/command_disconnect.go | 11 +- cli/command_expire.go | 4 +- cli/command_ls.go | 2 +- cli/command_metadata.go | 5 + cli/command_metadata_ls.go | 31 ++ cli/command_metadata_rm.go | 18 + cli/command_metadata_show.go | 43 ++ cli/command_mount.go | 2 +- cli/command_policy_ls.go | 2 +- cli/command_policy_remove.go | 2 +- cli/command_policy_set.go | 2 +- cli/command_policy_show.go | 2 +- cli/command_show.go | 2 +- cli/command_vault.go | 5 - cli/command_vault_ls.go | 31 -- cli/command_vault_rm.go | 24 -- cli/command_vault_show.go | 43 -- cli/config.go | 96 +---- cli/objref.go | 2 +- cli/progress.go | 2 +- internal/config/local_config.go | 73 ++++ .../config/metadata_format.go | 8 +- internal/units/units.go | 28 +- internal/units/units_test.go | 82 ++-- repo/blocks.go | 4 +- repo/connection.go | 156 ++++--- repo/connection_options.go | 16 - repo/initialize.go | 115 ++++++ repo/local_config.go | 50 --- repo/metadata_manager.go | 330 +++++++++++++++ repo/metadata_manager_test.go | 177 ++++++++ repo/{format.go => object_formatter.go} | 87 ++-- ...ormat_test.go => object_formatter_test.go} | 7 +- repo/{cas_manager.go => object_manager.go} | 121 +++--- ...manager_test.go => object_manager_test.go} | 51 +-- repo/object_reader.go | 2 +- repo/object_splitter.go | 55 ++- repo/object_splitter_test.go | 2 +- repo/object_writer.go | 21 +- repo/repository.go | 29 +- repo/vault.go | 381 ------------------ repo/vault_test.go | 207 ---------- snapshot/manager.go | 22 +- snapshot/stats.go | 2 +- snapshot/upload.go | 28 +- snapshot/upload_test.go | 29 +- 53 files changed, 1338 insertions(+), 1293 deletions(-) create mode 100644 cli/command_metadata.go create mode 100644 cli/command_metadata_ls.go create mode 100644 cli/command_metadata_rm.go create mode 100644 cli/command_metadata_show.go delete mode 100644 cli/command_vault.go delete mode 100644 cli/command_vault_ls.go delete mode 100644 cli/command_vault_rm.go delete mode 100644 cli/command_vault_show.go create mode 100644 internal/config/local_config.go rename repo/vault_format.go => internal/config/metadata_format.go (63%) delete mode 100644 repo/connection_options.go create mode 100644 repo/initialize.go delete mode 100644 repo/local_config.go create mode 100644 repo/metadata_manager.go create mode 100644 repo/metadata_manager_test.go rename repo/{format.go => object_formatter.go} (57%) rename repo/{format_test.go => object_formatter_test.go} (85%) rename repo/{cas_manager.go => object_manager.go} (65%) rename repo/{cas_manager_test.go => object_manager_test.go} (90%) delete mode 100644 repo/vault.go delete mode 100644 repo/vault_test.go diff --git a/README.md b/README.md index b49f9e064..83b84cd59 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ Unlike other cloud backup solutions, the user in full control of backup storage > **NOTICE**: > > Kopia is still in early stages of development and is **not ready for general use**. -> The repository and vault data format are subject to change, including backwards-incompatible changes. Use at your own risk. +> The repository data format is subject to change, including backwards-incompatible changes. Use at your own risk. Getting Started --- diff --git a/cli/command_backup.go b/cli/command_backup.go index 040f2c89b..b43996ad2 100644 --- a/cli/command_backup.go +++ b/cli/command_backup.go @@ -40,13 +40,9 @@ var ( ) func runBackupCommand(c *kingpin.ParseContext) error { - var repoOptions []repo.RepositoryOption - - if *backupWriteBack > 0 { - repoOptions = append(repoOptions, repo.WriteBack(*backupWriteBack)) - } - - rep := mustConnectToRepository(repoOptions) + rep := mustOpenRepository(&repo.Options{ + WriteBack: *backupWriteBack, + }) defer rep.Close() ctx := context.Background() @@ -66,7 +62,7 @@ func runBackupCommand(c *kingpin.ParseContext) error { } for _, backupDirectory := range sources { - rep.Stats.Reset() + rep.ResetStats() log.Printf("Backing up %v", backupDirectory) dir, err := filepath.Abs(backupDirectory) if err != nil { diff --git a/cli/command_backups.go b/cli/command_backups.go index 6e1ea8eb4..57d99519f 100644 --- a/cli/command_backups.go +++ b/cli/command_backups.go @@ -49,7 +49,7 @@ func findBackups(mgr *snapshot.Manager, sourceInfo snapshot.SourceInfo) (manifes } func runBackupsCommand(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) + rep := mustOpenRepository(nil) defer rep.Close() mgr := snapshot.NewManager(rep) @@ -98,7 +98,7 @@ func runBackupsCommand(context *kingpin.ParseContext) error { m.RootObjectID, relPath, m.StartTime.Format("2006-01-02 15:04:05 MST"), - units.BytesString(m.Stats.TotalFileSize), + units.BytesStringBase10(m.Stats.TotalFileSize), deltaBytes(m.Stats.Repository.WrittenBytes), ) count++ @@ -123,7 +123,7 @@ func (b manifestSorter) Swap(i, j int) { b[i], b[j] = b[j], b[i] } func deltaBytes(b int64) string { if b > 0 { - return "(+" + units.BytesString(b) + ")" + return "(+" + units.BytesStringBase10(b) + ")" } return "(no change)" diff --git a/cli/command_cat.go b/cli/command_cat.go index 1d15ca6ca..cc295f33b 100644 --- a/cli/command_cat.go +++ b/cli/command_cat.go @@ -13,7 +13,7 @@ var ( ) func runCatCommand(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) + rep := mustOpenRepository(nil) defer rep.Close() oid, err := parseObjectID(*catCommandPath, rep) diff --git a/cli/command_cleanup.go b/cli/command_cleanup.go index 04b3a36b2..856692e41 100644 --- a/cli/command_cleanup.go +++ b/cli/command_cleanup.go @@ -149,7 +149,7 @@ func findAliveBlocks(ctx *cleanupContext, wi *cleanupWorkItem) error { } func runCleanupCommand(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) + rep := mustOpenRepository(nil) defer rep.Close() mgr := snapshot.NewManager(rep) @@ -254,7 +254,7 @@ func runCleanupCommand(context *kingpin.ParseContext) error { totalBlocks++ totalBytes += b.Length - if strings.HasPrefix(b.BlockID, repo.VaultBlockPrefix) { + if strings.HasPrefix(b.BlockID, repo.MetadataBlockPrefix) { ignoredBlocks++ ignoredBytes += b.Length continue @@ -262,11 +262,11 @@ func runCleanupCommand(context *kingpin.ParseContext) error { if !ctx.inuse[b.BlockID] { if time.Since(b.TimeStamp) < *cleanupIgnoreAge { - log.Printf("Ignored unreferenced block: %v (%v) at %v", b.BlockID, units.BytesString(b.Length), b.TimeStamp.Local()) + log.Printf("Ignored unreferenced block: %v (%v) at %v", b.BlockID, units.BytesStringBase10(b.Length), b.TimeStamp.Local()) ignoredBlocks++ ignoredBytes += b.Length } else { - log.Printf("Unreferenced block: %v (%v) at %v", b.BlockID, units.BytesString(b.Length), b.TimeStamp.Local()) + log.Printf("Unreferenced block: %v (%v) at %v", b.BlockID, units.BytesStringBase10(b.Length), b.TimeStamp.Local()) unreferencedBlocks++ unreferencedBytes += b.Length } @@ -276,10 +276,10 @@ func runCleanupCommand(context *kingpin.ParseContext) error { } } - log.Printf("Found %v (%v) total blocks.", totalBlocks, units.BytesString(totalBytes)) - log.Printf("Ignored %v blocks (%v).", ignoredBlocks, units.BytesString(ignoredBytes)) - log.Printf("In-use objects: %v, %v blocks (%v)", len(ctx.queue.visited), inuseBlocks, units.BytesString(inuseBytes)) - log.Printf("Unreferenced: %v blocks (%v)", unreferencedBlocks, units.BytesString(unreferencedBytes)) + log.Printf("Found %v (%v) total blocks.", totalBlocks, units.BytesStringBase10(totalBytes)) + log.Printf("Ignored %v blocks (%v).", ignoredBlocks, units.BytesStringBase10(ignoredBytes)) + log.Printf("In-use objects: %v, %v blocks (%v)", len(ctx.queue.visited), inuseBlocks, units.BytesStringBase10(inuseBytes)) + log.Printf("Unreferenced: %v blocks (%v)", unreferencedBlocks, units.BytesStringBase10(unreferencedBytes)) return nil } diff --git a/cli/command_connect.go b/cli/command_connect.go index aa7180c3e..0e0c94b07 100644 --- a/cli/command_connect.go +++ b/cli/command_connect.go @@ -1,29 +1,54 @@ package cli import ( + "context" "fmt" + "github.com/kopia/kopia/repo" + "gopkg.in/alecthomas/kingpin.v2" ) var ( - connectCommand = app.Command("connect", "Connect to a vault.") + connectCommand = app.Command("connect", "Connect to a repository.") + connectRepositoryLocation = connectCommand.Arg("location", "Repository address").Required().String() + connectDontPersistCredentials *bool + connectCacheDirectory *string ) +func setupConnectOptions(cmd *kingpin.CmdClause) { + connectDontPersistCredentials = cmd.Flag("no-credentials", "Don't save credentials in the configuration file").Short('n').Bool() + connectCacheDirectory = cmd.Flag("cache-directory", "Cache directory").PlaceHolder("PATH").String() +} + +func connectOptions() repo.ConnectOptions { + return repo.ConnectOptions{ + PersistCredentials: !*connectDontPersistCredentials, + CacheDirectory: *connectCacheDirectory, + } +} + func init() { + setupConnectOptions(connectCommand) connectCommand.Action(runConnectCommand) } -func runConnectCommand(context *kingpin.ParseContext) error { - vlt, err := openVaultSpecifiedByFlag() +func runConnectCommand(_ *kingpin.ParseContext) error { + storage, err := newStorageFromURL(getContext(), *connectRepositoryLocation) if err != nil { - return fmt.Errorf("unable to open vault: %v", err) - } - if err := persistVaultConfig(vlt); err != nil { return err } - fmt.Println("Connected to vault:", *vaultPath) + creds, err := getRepositoryCredentials(false) + if err != nil { + return err + } + + if err := repo.Connect(context.Background(), repositoryConfigFileName(), storage, creds, connectOptions()); err != nil { + return err + } + + fmt.Println("Connected to repository:", *connectRepositoryLocation) return err } diff --git a/cli/command_create.go b/cli/command_create.go index 33bc1c984..c410a54cf 100644 --- a/cli/command_create.go +++ b/cli/command_create.go @@ -1,70 +1,48 @@ package cli import ( - "crypto/rand" "fmt" - "io" kingpin "gopkg.in/alecthomas/kingpin.v2" "github.com/kopia/kopia/blob" + "github.com/kopia/kopia/internal/units" "github.com/kopia/kopia/repo" ) var ( - createCommand = app.Command("create", "Create new vault and repository.") - createObjectFormat = createCommand.Flag("repo-format", "Format of repository objects.").PlaceHolder("FORMAT").Default(repo.DefaultObjectFormat).Enum(supportedObjectFormats()...) + createCommand = app.Command("create", "Create new repository in a specified location.") + createRepositoryLocation = createCommand.Arg("location", "Location where to create the repository").Required().String() - createMinBlockSize = createCommand.Flag("min-block-size", "Minimum size of a data block.").PlaceHolder("KB").Default("1024").Int() - createAvgBlockSize = createCommand.Flag("avg-block-size", "Average size of a data block.").PlaceHolder("KB").Default("10240").Int() - createMaxBlockSize = createCommand.Flag("max-block-size", "Maximum size of a data block.").PlaceHolder("KB").Default("20480").Int() - createObjectSplitter = createCommand.Flag("object-splitter", "The splitter to use for new objects in the repository").Default("DYNAMIC").Enum(supportedObjectSplitters()...) - createInlineBlobSize = createCommand.Flag("inline-blob-size", "Maximum size of an inline data object.").PlaceHolder("KB").Default("32").Int() - createVaultEncryptionFormat = createCommand.Flag("vault-encryption", "Vault encryption.").PlaceHolder("FORMAT").Default(repo.SupportedEncryptionAlgorithms[0]).Enum(repo.SupportedEncryptionAlgorithms...) - createOverwrite = createCommand.Flag("overwrite", "Overwrite existing data (DANGEROUS).").Bool() - createOnly = createCommand.Flag("create-only", "Create the vault, but don't connect to it.").Short('c').Bool() + createMetadataEncryptionFormat = createCommand.Flag("metadata-encryption", "Metadata item encryption.").PlaceHolder("FORMAT").Default(repo.SupportedMetadataEncryptionAlgorithms[0]).Enum(repo.SupportedMetadataEncryptionAlgorithms...) + createObjectFormat = createCommand.Flag("object-format", "Format of repository objects.").PlaceHolder("FORMAT").Default(repo.DefaultObjectFormat).Enum(repo.SupportedObjectFormats...) + createObjectSplitter = createCommand.Flag("object-splitter", "The splitter to use for new objects in the repository").Default("DYNAMIC").Enum(repo.SupportedObjectSplitters...) + + createMinBlockSize = createCommand.Flag("min-block-size", "Minimum size of a data block.").PlaceHolder("KB").Default("1024").Int() + createAvgBlockSize = createCommand.Flag("avg-block-size", "Average size of a data block.").PlaceHolder("KB").Default("10240").Int() + createMaxBlockSize = createCommand.Flag("max-block-size", "Maximum size of a data block.").PlaceHolder("KB").Default("20480").Int() + + createInlineBlobSize = createCommand.Flag("inline-blob-size", "Maximum size of an inline data object.").PlaceHolder("KB").Default("32").Int() + createOverwrite = createCommand.Flag("overwrite", "Overwrite existing data (DANGEROUS).").Bool() + createOnly = createCommand.Flag("create-only", "Create repository, but don't connect to it.").Short('c').Bool() ) func init() { + setupConnectOptions(createCommand) createCommand.Action(runCreateCommand) } -func vaultFormat() (*repo.VaultFormat, error) { - f := &repo.VaultFormat{ - Version: "1", - } - f.UniqueID = make([]byte, 32) - _, err := io.ReadFull(rand.Reader, f.UniqueID) - if err != nil { - return nil, err - } - f.EncryptionAlgorithm = *createVaultEncryptionFormat - return f, nil -} - -func repositoryFormat() (*repo.Format, error) { - f := &repo.Format{ - Version: 1, - Secret: make([]byte, 32), - MasterKey: make([]byte, 32), - MaxInlineContentLength: int32(*createInlineBlobSize * 1024), - ObjectFormat: *createObjectFormat, +func newRepositoryOptionsFromFlags() *repo.NewRepositoryOptions { + return &repo.NewRepositoryOptions{ + MetadataEncryptionAlgorithm: *createMetadataEncryptionFormat, + MaxInlineContentLength: *createInlineBlobSize * 1024, + ObjectFormat: *createObjectFormat, Splitter: *createObjectSplitter, - MinBlockSize: int32(*createMinBlockSize * 1024), - AvgBlockSize: int32(*createAvgBlockSize * 1024), - MaxBlockSize: int32(*createMaxBlockSize * 1024), + MinBlockSize: *createMinBlockSize * 1024, + AvgBlockSize: *createAvgBlockSize * 1024, + MaxBlockSize: *createMaxBlockSize * 1024, } - - if _, err := io.ReadFull(rand.Reader, f.Secret); err != nil { - return nil, err - } - - if _, err := io.ReadFull(rand.Reader, f.MasterKey); err != nil { - return nil, err - } - - return f, nil } func openStorageAndEnsureEmpty(url string) (blob.Storage, error) { @@ -83,72 +61,48 @@ func openStorageAndEnsureEmpty(url string) (blob.Storage, error) { return s, nil } -func runCreateCommand(context *kingpin.ParseContext) error { - if *vaultPath == "" { - return fmt.Errorf("--vault is required") - } - vaultStorage, err := openStorageAndEnsureEmpty(*vaultPath) +func runCreateCommand(_ *kingpin.ParseContext) error { + st, err := openStorageAndEnsureEmpty(*createRepositoryLocation) if err != nil { - return fmt.Errorf("unable to get vault storage: %v", err) + return fmt.Errorf("unable to get repository storage: %v", err) } - repoFormat, err := repositoryFormat() - if err != nil { - return fmt.Errorf("unable to initialize repository format: %v", err) - } + options := newRepositoryOptionsFromFlags() - fmt.Printf( - "Initializing repository in with format %q, splitter %q and maximum object size %v.\n", - repoFormat.ObjectFormat, - repoFormat.Splitter, - repoFormat.MaxBlockSize) - - vf, err := vaultFormat() - if err != nil { - return fmt.Errorf("unable to initialize vault format: %v", err) - } - - creds, err := getVaultCredentials(true) + creds, err := getRepositoryCredentials(true) if err != nil { return fmt.Errorf("unable to get credentials: %v", err) } - if err := repoFormat.Validate(); err != nil { - return fmt.Errorf("invalid format") + fmt.Printf("Initializing repository with:\n") + fmt.Printf(" metadata encryption: %v\n", options.MetadataEncryptionAlgorithm) + fmt.Printf(" object format: %v\n", options.ObjectFormat) + switch options.Splitter { + case "DYNAMIC": + fmt.Printf(" object splitter: DYNAMIC with block sizes (min:%v avg:%v max:%v)\n", + units.BytesStringBase2(int64(options.MinBlockSize)), + units.BytesStringBase2(int64(options.AvgBlockSize)), + units.BytesStringBase2(int64(options.MaxBlockSize))) + + case "FIXED": + fmt.Printf(" object splitter: FIXED with with block size: %v\n", units.BytesStringBase2(int64(options.MaxBlockSize))) + + case "NEVER": + fmt.Printf(" object splitter: NEVER\n") } - fmt.Printf( - "Initializing vault with encryption '%v'.\n", - vf.EncryptionAlgorithm) - - vlt, err := repo.Create(vaultStorage, vf, creds, repoFormat) - if err != nil { - return fmt.Errorf("cannot create vault: %v", err) + if err := repo.Initialize(st, options, creds); err != nil { + return fmt.Errorf("cannot initialize repository: %v", err) } if !*createOnly { - if err := persistVaultConfig(vlt); err != nil { + err := repo.Connect(getContext(), repositoryConfigFileName(), st, creds, connectOptions()) + if err != nil { return err } - fmt.Println("Connected to vault:", *vaultPath) + fmt.Println("Connected to repository:", *createRepositoryLocation) } return nil } - -func supportedObjectFormats() []string { - var r []string - for k := range repo.SupportedFormats { - r = append(r, k) - } - return r -} - -func supportedObjectSplitters() []string { - var r []string - for k := range repo.SupportedSplitters { - r = append(r, k) - } - return r -} diff --git a/cli/command_disconnect.go b/cli/command_disconnect.go index bf4028a02..507e90509 100644 --- a/cli/command_disconnect.go +++ b/cli/command_disconnect.go @@ -1,13 +1,12 @@ package cli import ( - "os" - + "github.com/kopia/kopia/repo" "gopkg.in/alecthomas/kingpin.v2" ) var ( - disconnectCommand = app.Command("disconnect", "Connect to a vault.") + disconnectCommand = app.Command("disconnect", "Disconnect from a repository.") ) func init() { @@ -15,9 +14,5 @@ func init() { } func runDisconnectCommand(context *kingpin.ParseContext) error { - fn := vaultConfigFileName() - if _, err := os.Stat(fn); err == nil { - return os.Remove(fn) - } - return nil + return repo.Disconnect(repositoryConfigFileName()) } diff --git a/cli/command_expire.go b/cli/command_expire.go index b0ec9ff09..f0039ca94 100644 --- a/cli/command_expire.go +++ b/cli/command_expire.go @@ -178,7 +178,7 @@ func expireSnapshots(mgr *snapshot.Manager, snapshots []*snapshot.Manifest, name } func runExpireCommand(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) + rep := mustOpenRepository(nil) defer rep.Close() mgr := snapshot.NewManager(rep) @@ -206,7 +206,7 @@ func runExpireCommand(context *kingpin.ParseContext) error { if *expireDelete == "yes" { fmt.Fprintf(os.Stderr, "Deleting %v snapshots...\n", len(toDelete)) - if err := rep.Vault.RemoveMany(toDelete); err != nil { + if err := rep.RemoveMany(toDelete); err != nil { return err } } else { diff --git a/cli/command_ls.go b/cli/command_ls.go index 00dcdc39d..13f2d777d 100644 --- a/cli/command_ls.go +++ b/cli/command_ls.go @@ -22,7 +22,7 @@ var ( ) func runLSCommand(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) + rep := mustOpenRepository(nil) defer rep.Close() oid, err := parseObjectID(*lsCommandPath, rep) diff --git a/cli/command_metadata.go b/cli/command_metadata.go new file mode 100644 index 000000000..ca352a8d4 --- /dev/null +++ b/cli/command_metadata.go @@ -0,0 +1,5 @@ +package cli + +var ( + metadataCommands = app.Command("metadata", "Low-level commands to manipulate metadata items.") +) diff --git a/cli/command_metadata_ls.go b/cli/command_metadata_ls.go new file mode 100644 index 000000000..5ab72ab42 --- /dev/null +++ b/cli/command_metadata_ls.go @@ -0,0 +1,31 @@ +package cli + +import ( + "fmt" + + "gopkg.in/alecthomas/kingpin.v2" +) + +var ( + metadataListCommand = metadataCommands.Command("list", "List metadata items").Alias("ls").Hidden() + metadataListPrefix = metadataListCommand.Flag("prefix", "Prefix").String() +) + +func init() { + metadataListCommand.Action(listMetadataItems) +} + +func listMetadataItems(context *kingpin.ParseContext) error { + rep := mustOpenRepository(nil) + + entries, err := rep.ListMetadata(*metadataListPrefix, -1) + if err != nil { + return err + } + + for _, e := range entries { + fmt.Println(e) + } + + return nil +} diff --git a/cli/command_metadata_rm.go b/cli/command_metadata_rm.go new file mode 100644 index 000000000..d5e92b108 --- /dev/null +++ b/cli/command_metadata_rm.go @@ -0,0 +1,18 @@ +package cli + +import "gopkg.in/alecthomas/kingpin.v2" + +var ( + metadataRemoveCommand = metadataCommands.Command("rm", "Remove metadata items").Hidden() + metadataRemoveItems = metadataRemoveCommand.Arg("item", "Items to remove").Strings() +) + +func init() { + metadataRemoveCommand.Action(removeMetadataItem) +} + +func removeMetadataItem(context *kingpin.ParseContext) error { + rep := mustOpenRepository(nil) + + return rep.RemoveMany(*metadataRemoveItems) +} diff --git a/cli/command_metadata_show.go b/cli/command_metadata_show.go new file mode 100644 index 000000000..5ee898953 --- /dev/null +++ b/cli/command_metadata_show.go @@ -0,0 +1,43 @@ +package cli + +import ( + "bytes" + "encoding/json" + "os" + + "gopkg.in/alecthomas/kingpin.v2" +) + +var ( + metadataShowCommand = metadataCommands.Command("show", "Show contents of a metadata item").Alias("cat").Hidden() + metadataShowID = metadataShowCommand.Arg("id", "ID of the metadata item to show").String() + metadataShowJSON = metadataShowCommand.Flag("json", "Pretty-print JSON").Short('j').Bool() + metadataShowNoNewLine = metadataShowCommand.Flag("nonewline", "Do not emit newline").Short('n').Bool() +) + +func init() { + metadataShowCommand.Action(showMetadataObject) +} + +func showMetadataObject(context *kingpin.ParseContext) error { + rep := mustOpenRepository(nil) + + b, err := rep.MetadataManager.GetMetadata(*metadataShowID) + if err != nil { + return err + } + + if *metadataShowJSON { + var buf bytes.Buffer + json.Indent(&buf, b, "", " ") + buf.WriteTo(os.Stdout) + } else { + os.Stdout.Write(b) + } + + if !*metadataShowNoNewLine { + os.Stdout.WriteString("\n") + } + + return nil +} diff --git a/cli/command_mount.go b/cli/command_mount.go index d6374c88e..14f6d64a9 100644 --- a/cli/command_mount.go +++ b/cli/command_mount.go @@ -33,7 +33,7 @@ func (r *root) Root() (fusefs.Node, error) { } func runMountCommand(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) + rep := mustOpenRepository(nil) fuseConnection, err := fuse.Mount( *mountPoint, diff --git a/cli/command_policy_ls.go b/cli/command_policy_ls.go index ba61aabc7..2af956d79 100644 --- a/cli/command_policy_ls.go +++ b/cli/command_policy_ls.go @@ -16,7 +16,7 @@ func init() { } func listPolicies(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) + rep := mustOpenRepository(nil) mgr := snapshot.NewManager(rep) policies, err := mgr.ListPolicies() diff --git a/cli/command_policy_remove.go b/cli/command_policy_remove.go index bce523d35..aebc8b9f7 100644 --- a/cli/command_policy_remove.go +++ b/cli/command_policy_remove.go @@ -18,7 +18,7 @@ func init() { } func removePolicy(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) + rep := mustOpenRepository(nil) mgr := snapshot.NewManager(rep) targets, err := policyTargets(policyRemoveGlobal, policyRemoveTargets) diff --git a/cli/command_policy_set.go b/cli/command_policy_set.go index 16f168073..4db295821 100644 --- a/cli/command_policy_set.go +++ b/cli/command_policy_set.go @@ -36,7 +36,7 @@ func init() { } func setPolicy(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) + rep := mustOpenRepository(nil) mgr := snapshot.NewManager(rep) _ = mgr diff --git a/cli/command_policy_show.go b/cli/command_policy_show.go index 0c82bd4cf..3fa07bf4c 100644 --- a/cli/command_policy_show.go +++ b/cli/command_policy_show.go @@ -19,7 +19,7 @@ func init() { } func showPolicy(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) + rep := mustOpenRepository(nil) mgr := snapshot.NewManager(rep) targets, err := policyTargets(policyShowGlobal, policyShowTargets) diff --git a/cli/command_show.go b/cli/command_show.go index d38d5a4ee..929eb48c3 100644 --- a/cli/command_show.go +++ b/cli/command_show.go @@ -20,7 +20,7 @@ var ( ) func runShowCommand(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) + rep := mustOpenRepository(nil) defer rep.Close() for _, oidString := range *showObjectIDs { diff --git a/cli/command_vault.go b/cli/command_vault.go deleted file mode 100644 index 627caa432..000000000 --- a/cli/command_vault.go +++ /dev/null @@ -1,5 +0,0 @@ -package cli - -var ( - vaultCommands = app.Command("vault", "Low-level commands to manipulate vault.") -) diff --git a/cli/command_vault_ls.go b/cli/command_vault_ls.go deleted file mode 100644 index 661490aef..000000000 --- a/cli/command_vault_ls.go +++ /dev/null @@ -1,31 +0,0 @@ -package cli - -import ( - "fmt" - - "gopkg.in/alecthomas/kingpin.v2" -) - -var ( - vaultListCommand = vaultCommands.Command("list", "List contents of a vault").Alias("ls").Hidden() - vaultListPrefix = vaultListCommand.Flag("prefix", "Prefix").String() -) - -func init() { - vaultListCommand.Action(listVaultContents) -} - -func listVaultContents(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) - - entries, err := rep.Vault.List(*vaultListPrefix, -1) - if err != nil { - return err - } - - for _, e := range entries { - fmt.Println(e) - } - - return nil -} diff --git a/cli/command_vault_rm.go b/cli/command_vault_rm.go deleted file mode 100644 index c79802f2a..000000000 --- a/cli/command_vault_rm.go +++ /dev/null @@ -1,24 +0,0 @@ -package cli - -import "gopkg.in/alecthomas/kingpin.v2" - -var ( - vaultRemoveCommand = vaultCommands.Command("rm", "Remove vault items").Hidden() - vaultRemoveItems = vaultRemoveCommand.Arg("item", "Items to remove").Strings() -) - -func init() { - vaultRemoveCommand.Action(removeVaultItem) -} - -func removeVaultItem(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) - - for _, v := range *vaultRemoveItems { - if err := rep.Vault.Remove(v); err != nil { - return err - } - } - - return nil -} diff --git a/cli/command_vault_show.go b/cli/command_vault_show.go deleted file mode 100644 index c7eff2eb7..000000000 --- a/cli/command_vault_show.go +++ /dev/null @@ -1,43 +0,0 @@ -package cli - -import ( - "bytes" - "encoding/json" - "os" - - "gopkg.in/alecthomas/kingpin.v2" -) - -var ( - vaultShowCommand = vaultCommands.Command("show", "Show contents of a vault item").Alias("cat").Hidden() - vaultShowID = vaultShowCommand.Arg("id", "ID of the vault item to show").String() - vaultShowJSON = vaultShowCommand.Flag("json", "Pretty-print JSON").Short('j').Bool() - vaultShowNoNewLine = vaultShowCommand.Flag("nonewline", "Do not emit newline").Short('n').Bool() -) - -func init() { - vaultShowCommand.Action(showVaultObject) -} - -func showVaultObject(context *kingpin.ParseContext) error { - rep := mustConnectToRepository(nil) - - b, err := rep.Vault.Get(*vaultShowID) - if err != nil { - return err - } - - if *vaultShowJSON { - var buf bytes.Buffer - json.Indent(&buf, b, "", " ") - buf.WriteTo(os.Stdout) - } else { - os.Stdout.Write(b) - } - - if !*vaultShowNoNewLine { - os.Stdout.WriteString("\n") - } - - return nil -} diff --git a/cli/config.go b/cli/config.go index 50dc27e54..1931ee8b0 100644 --- a/cli/config.go +++ b/cli/config.go @@ -3,7 +3,6 @@ package cli import ( "context" "encoding/hex" - "encoding/json" "fmt" "io/ioutil" "log" @@ -14,7 +13,6 @@ import ( "github.com/bgentry/speakeasy" "github.com/kopia/kopia/auth" - "github.com/kopia/kopia/blob/logging" "github.com/kopia/kopia/fs" "github.com/kopia/kopia/fs/localfs" "github.com/kopia/kopia/fs/loggingfs" @@ -25,12 +23,11 @@ var ( traceStorage = app.Flag("trace-storage", "Enables tracing of storage operations.").Hidden().Envar("KOPIA_TRACE_STORAGE").Bool() traceLocalFS = app.Flag("trace-localfs", "Enables tracing of local filesystem operations").Hidden().Envar("KOPIA_TRACE_STORAGE").Bool() - vaultConfigPath = app.Flag("vaultconfig", "Specify the vault config file to use.").PlaceHolder("PATH").Envar("KOPIA_VAULTCONFIG").String() - vaultPath = app.Flag("vault", "Specify the vault to use.").PlaceHolder("PATH").Envar("KOPIA_VAULT").Short('v').String() - password = app.Flag("password", "Vault password.").Envar("KOPIA_PASSWORD").Short('p').String() - passwordFile = app.Flag("passwordfile", "Read vault password from a file.").PlaceHolder("FILENAME").Envar("KOPIA_PASSWORD_FILE").ExistingFile() - key = app.Flag("key", "Specify vault master key (hexadecimal).").Envar("KOPIA_KEY").Short('k').String() - keyFile = app.Flag("keyfile", "Read vault master key from file.").PlaceHolder("FILENAME").Envar("KOPIA_KEY_FILE").ExistingFile() + configPath = app.Flag("config-file", "Specify the config file to use.").PlaceHolder("PATH").Envar("KOPIA_CONFIG_PATH").String() + password = app.Flag("password", "Repository password.").Envar("KOPIA_PASSWORD").Short('p').String() + passwordFile = app.Flag("passwordfile", "Read repository password from a file.").PlaceHolder("FILENAME").Envar("KOPIA_PASSWORD_FILE").ExistingFile() + key = app.Flag("key", "Specify master key (hexadecimal).").Envar("KOPIA_KEY").Short('k').String() + keyFile = app.Flag("keyfile", "Read master key from file.").PlaceHolder("FILENAME").Envar("KOPIA_KEY_FILE").ExistingFile() maxDownloadSpeed = app.Flag("max-download-speed", "Limit the download speed.").PlaceHolder("BYTES_PER_SEC").Int() maxUploadSpeed = app.Flag("max-upload-speed", "Limit the upload speed.").PlaceHolder("BYTES_PER_SEC").Int() @@ -48,15 +45,15 @@ func getContext() context.Context { return ctx } -func connectToRepository(options []repo.RepositoryOption) (*repo.Repository, error) { - return repo.Connect(getContext(), vaultConfigFileName(), connectOptionsFromFlags(options)) +func openRepository(opts *repo.Options) (*repo.Repository, error) { + return repo.Open(getContext(), repositoryConfigFileName(), applyOptionsFromFlags(opts)) } -func connectOptionsFromFlags(options []repo.RepositoryOption) *repo.ConnectOptions { - opts := &repo.ConnectOptions{ - CredentialsCallback: func() (auth.Credentials, error) { return getVaultCredentials(false) }, - RepositoryOptions: options, +func applyOptionsFromFlags(opts *repo.Options) *repo.Options { + if opts == nil { + opts = &repo.Options{} } + opts.CredentialsCallback = func() (auth.Credentials, error) { return getRepositoryCredentials(false) } if *traceStorage { opts.TraceStorage = log.Printf @@ -73,25 +70,12 @@ func connectOptionsFromFlags(options []repo.RepositoryOption) *repo.ConnectOptio return opts } -func mustConnectToRepository(repoOptions []repo.RepositoryOption) *repo.Repository { - s, err := connectToRepository(repoOptions) +func mustOpenRepository(opts *repo.Options) *repo.Repository { + s, err := openRepository(opts) failOnError(err) return s } -func repositoryOptionsFromFlags(extraOptions []repo.RepositoryOption) []repo.RepositoryOption { - var opts []repo.RepositoryOption - - for _, o := range extraOptions { - opts = append(opts, o) - } - - if *traceStorage { - opts = append(opts, repo.EnableLogging(logging.Prefix("[REPOSITORY] "))) - } - return opts -} - func getHomeDir() string { if runtime.GOOS == "windows" { home := os.Getenv("HOMEDRIVE") + os.Getenv("HOMEPATH") @@ -104,54 +88,14 @@ func getHomeDir() string { return os.Getenv("HOME") } -func vaultConfigFileName() string { - if len(*vaultConfigPath) > 0 { - return *vaultConfigPath +func repositoryConfigFileName() string { + if len(*configPath) > 0 { + return *configPath } - return filepath.Join(getHomeDir(), ".kopia/vault.config") + return filepath.Join(getHomeDir(), ".kopia/repository.config") } -func persistVaultConfig(r *repo.Vault) error { - cfg, err := r.Config() - if err != nil { - return err - } - - var lc repo.LocalConfig - lc.VaultConnection = cfg - - fname := vaultConfigFileName() - log.Printf("Saving vault configuration to '%v'.", fname) - if err := os.MkdirAll(filepath.Dir(fname), 0700); err != nil { - return err - } - - d, err := json.MarshalIndent(&lc, "", " ") - if err != nil { - return err - } - - return ioutil.WriteFile(fname, d, 0600) -} - -func openVaultSpecifiedByFlag() (*repo.Vault, error) { - if *vaultPath == "" { - return nil, fmt.Errorf("--vault must be specified") - } - storage, err := newStorageFromURL(getContext(), *vaultPath) - if err != nil { - return nil, err - } - - creds, err := getVaultCredentials(false) - if err != nil { - return nil, err - } - - return repo.Open(storage, creds) -} - -func getVaultCredentials(isNew bool) (auth.Credentials, error) { +func getRepositoryCredentials(isNew bool) (auth.Credentials, error) { if *key != "" { k, err := hex.DecodeString(*key) if err != nil { @@ -184,7 +128,7 @@ func getVaultCredentials(isNew bool) (auth.Credentials, error) { } if isNew { for { - p1, err := askPass("Enter password to create new vault: ") + p1, err := askPass("Enter password to create new repository: ") if err != nil { return nil, err } @@ -199,7 +143,7 @@ func getVaultCredentials(isNew bool) (auth.Credentials, error) { } } } else { - p1, err := askPass("Enter password to open vault: ") + p1, err := askPass("Enter password to open repository: ") if err != nil { return nil, err } diff --git a/cli/objref.go b/cli/objref.go index 2c9cdb316..ca4c29b65 100644 --- a/cli/objref.go +++ b/cli/objref.go @@ -18,7 +18,7 @@ func parseObjectID(id string, r *repo.Repository) (repo.ObjectID, error) { oid, err := repo.ParseObjectID(head) if err != nil { - return repo.NullObjectID, fmt.Errorf("can't retrieve vault object ID %v: %v", head, err) + return repo.NullObjectID, fmt.Errorf("can't parse object ID %v: %v", head, err) } if tail == "" { diff --git a/cli/progress.go b/cli/progress.go index 27ac990e7..c745c97ec 100644 --- a/cli/progress.go +++ b/cli/progress.go @@ -17,7 +17,7 @@ type uploadProgress struct { } func (p *uploadProgress) Cached(path string, length int64) { - log.Printf(" Cached: %v %v", path, units.BytesString(length)) + log.Printf(" Cached: %v %v", path, units.BytesStringBase10(length)) } func (p *uploadProgress) StartedDir(path string) { diff --git a/internal/config/local_config.go b/internal/config/local_config.go new file mode 100644 index 000000000..06892785a --- /dev/null +++ b/internal/config/local_config.go @@ -0,0 +1,73 @@ +package config + +import ( + "encoding/json" + "io" + "os" + + "github.com/kopia/kopia/blob" +) + +// LocalConfig is a configuration of Kopia. +type LocalConfig struct { + Connection *RepositoryConnectionInfo `json:"connection,omitempty"` + CacheDirectory string `json:"cacheDirectory,omitempty"` +} + +// RepositoryObjectFormat describes the format of objects in a repository. +type RepositoryObjectFormat struct { + Version int `json:"version,omitempty"` // version number, must be "1" + ObjectFormat string `json:"objectFormat,omitempty"` // identifier of object format + Secret []byte `json:"secret,omitempty"` // HMAC secret used to generate encryption keys + MaxInlineContentLength int `json:"maxInlineContentLength,omitempty"` // maximum size of object to be considered for inline storage within ObjectID + MasterKey []byte `json:"masterKey,omitempty"` // master encryption key (SIV-mode encryption only) + Splitter string `json:"splitter,omitempty"` // splitter used to break objects into storage blocks + + MinBlockSize int `json:"minBlockSize,omitempty"` // minimum block size used with dynamic splitter + AvgBlockSize int `json:"avgBlockSize,omitempty"` // approximate size of storage block (used with dynamic splitter) + MaxBlockSize int `json:"maxBlockSize,omitempty"` // maximum size of storage block +} + +// RepositoryConnectionInfo represents JSON-serializable configuration of the repository connection, including master key. +type RepositoryConnectionInfo struct { + ConnectionInfo blob.ConnectionInfo `json:"storage"` + Key []byte `json:"key,omitempty"` +} + +// EncryptedRepositoryConfig contains the configuration of repository that's persisted in encrypted format. +type EncryptedRepositoryConfig struct { + Format RepositoryObjectFormat `json:"format"` +} + +// Load reads local configuration from the specified reader. +func (lc *LocalConfig) Load(r io.Reader) error { + *lc = LocalConfig{} + return json.NewDecoder(r).Decode(lc) +} + +// Save writes the configuration to the specified writer. +func (lc *LocalConfig) Save(w io.Writer) error { + b, err := json.MarshalIndent(lc, "", " ") + if err != nil { + return nil + } + _, err = w.Write(b) + return err +} + +// LoadFromFile reads the local configuration from the specified file. +func LoadFromFile(fileName string) (*LocalConfig, error) { + f, err := os.Open(fileName) + if err != nil { + return nil, err + } + defer f.Close() + + var lc LocalConfig + + if err := lc.Load(f); err != nil { + return nil, err + } + + return &lc, nil +} diff --git a/repo/vault_format.go b/internal/config/metadata_format.go similarity index 63% rename from repo/vault_format.go rename to internal/config/metadata_format.go index b1a0f2c98..c47b14028 100644 --- a/repo/vault_format.go +++ b/internal/config/metadata_format.go @@ -1,12 +1,12 @@ -package repo +package config import ( "github.com/kopia/kopia/auth" ) -// VaultFormat describes the format of a Vault. -// Contents of this structure are serialized in plain text in the Vault storage. -type VaultFormat struct { +// MetadataFormat describes the format of metadata items in repository. +// Contents of this structure are serialized in plain text in the storage. +type MetadataFormat struct { auth.Options Version string `json:"version"` EncryptionAlgorithm string `json:"encryption"` diff --git a/internal/units/units.go b/internal/units/units.go index dffaa7ddc..6540e47b1 100644 --- a/internal/units/units.go +++ b/internal/units/units.go @@ -5,29 +5,35 @@ import ( "strings" ) -var unitPrefixes = []string{"", "K", "M", "G", "T"} +var base10UnitPrefixes = []string{"", "K", "M", "G", "T"} +var base2UnitPrefixes = []string{"", "Ki", "Mi", "Gi", "Ti"} func niceNumber(f float64) string { return strings.TrimRight(strings.TrimRight(fmt.Sprintf("%.1f", f), "0"), ".") } -func toDecimalUnitString(f float64, suffix string) string { - for i := range unitPrefixes { - if f < 900 { - return fmt.Sprintf("%v %v%v", niceNumber(f), unitPrefixes[i], suffix) +func toDecimalUnitString(f float64, thousand float64, prefixes []string, suffix string) string { + for i := range prefixes { + if f < 0.9*thousand { + return fmt.Sprintf("%v %v%v", niceNumber(f), prefixes[i], suffix) } - f = f / float64(1000) + f = f / thousand } - return fmt.Sprintf("%v %v%v", niceNumber(f), unitPrefixes[len(unitPrefixes)-1], suffix) + return fmt.Sprintf("%v %v%v", niceNumber(f), prefixes[len(prefixes)-1], suffix) } -// BytesString formats the given value as bytes with the appropriate suffix (KB, MB, GB, ...) -func BytesString(b int64) string { - return toDecimalUnitString(float64(b), "B") +// BytesStringBase10 formats the given value as bytes with the appropriate base-10 suffix (KB, MB, GB, ...) +func BytesStringBase10(b int64) string { + return toDecimalUnitString(float64(b), 1000, base10UnitPrefixes, "B") +} + +// BytesStringBase2 formats the given value as bytes with the appropriate base-2 suffix (KiB, MiB, GiB, ...) +func BytesStringBase2(b int64) string { + return toDecimalUnitString(float64(b), 1024.0, base2UnitPrefixes, "B") } // BitsPerSecondsString formats the given value bits per second with the appropriate suffix (Kbit/s, Mbit/s, Gbit/s, ...) func BitsPerSecondsString(bps float64) string { - return toDecimalUnitString(bps, "bit/s") + return toDecimalUnitString(bps, 1000, base10UnitPrefixes, "bit/s") } diff --git a/internal/units/units_test.go b/internal/units/units_test.go index 3e42dd1ba..4182cbb24 100644 --- a/internal/units/units_test.go +++ b/internal/units/units_test.go @@ -2,38 +2,70 @@ package units import "testing" -func TestToDecimalUnitString(t *testing.T) { +func TestBytesStringBase10(t *testing.T) { cases := []struct { - value float64 - suffix string + value int64 expected string }{ - {0, "B", "0 B"}, - {1, "B", "1 B"}, - {2, "B", "2 B"}, - {899, "B", "899 B"}, - {900, "B", "0.9 KB"}, - {999, "B", "1 KB"}, - {1000, "B", "1 KB"}, - {1200, "B", "1.2 KB"}, - {899999, "B", "900 KB"}, - {900000, "B", "0.9 MB"}, - {999000, "B", "1 MB"}, - {999999, "B", "1 MB"}, - {1000000, "B", "1 MB"}, - {99000000, "B", "99 MB"}, - {990000000, "B", "1 GB"}, - {9990000000, "B", "10 GB"}, - {99900000000, "B", "99.9 GB"}, - {1000000000000, "B", "1 TB"}, - {99000000000000, "B", "99 TB"}, - {98765432109876543912, "B", "98765.4 TB"}, + {0, "0 B"}, + {1, "1 B"}, + {2, "2 B"}, + {899, "899 B"}, + {900, "0.9 KB"}, + {999, "1 KB"}, + {1000, "1 KB"}, + {1200, "1.2 KB"}, + {899999, "900 KB"}, + {900000, "0.9 MB"}, + {999000, "1 MB"}, + {999999, "1 MB"}, + {1000000, "1 MB"}, + {99000000, "99 MB"}, + {990000000, "1 GB"}, + {9990000000, "10 GB"}, + {99900000000, "99.9 GB"}, + {1000000000000, "1 TB"}, + {99000000000000, "99 TB"}, } for i, c := range cases { - actual := toDecimalUnitString(c.value, c.suffix) + actual := BytesStringBase10(c.value) if actual != c.expected { - t.Errorf("case #%v failed, expected: '%v', got '%v'", i, c.expected, actual) + t.Errorf("case #%v failed for %v, expected: '%v', got '%v'", i, c.value, c.expected, actual) + } + } +} + +func TestBytesStringBase2(t *testing.T) { + cases := []struct { + value int64 + expected string + }{ + {0, "0 B"}, + {1, "1 B"}, + {2, "2 B"}, + {899, "899 B"}, + {900, "900 B"}, + {999, "1 KiB"}, + {1024, "1 KiB"}, + {1400, "1.4 KiB"}, + {900<<10 - 1, "900 KiB"}, + {900 << 10, "900 KiB"}, + {999000, "1 MiB"}, + {999999, "1 MiB"}, + {1000000, "1 MiB"}, + {99 << 20, "99 MiB"}, + {1 << 30, "1 GiB"}, + {10 << 30, "10 GiB"}, + {99900000000, "93 GiB"}, + {1000000000000, "0.9 TiB"}, + {99000000000000, "90 TiB"}, + } + + for i, c := range cases { + actual := BytesStringBase2(c.value) + if actual != c.expected { + t.Errorf("case #%v failed for %v, expected: '%v', got '%v'", i, c.value, c.expected, actual) } } } diff --git a/repo/blocks.go b/repo/blocks.go index 2ae5e66d3..9361b0923 100644 --- a/repo/blocks.go +++ b/repo/blocks.go @@ -1,7 +1,7 @@ package repo // GetStorageBlocks returns the list of storage blocks used by the specified object ID. -func (r *casManager) GetStorageBlocks(oid ObjectID) ([]string, error) { +func (r *ObjectManager) GetStorageBlocks(oid ObjectID) ([]string, error) { result := map[string]bool{} if err := r.addStorageBlocks(result, oid); err != nil { return nil, err @@ -14,7 +14,7 @@ func (r *casManager) GetStorageBlocks(oid ObjectID) ([]string, error) { return b, nil } -func (r *casManager) addStorageBlocks(result map[string]bool, oid ObjectID) error { +func (r *ObjectManager) addStorageBlocks(result map[string]bool, oid ObjectID) error { if oid.Section != nil { return r.addStorageBlocks(result, oid.Section.Base) } diff --git a/repo/connection.go b/repo/connection.go index 22b4d89c1..3bf2e534e 100644 --- a/repo/connection.go +++ b/repo/connection.go @@ -2,87 +2,153 @@ package repo import ( "context" + "encoding/json" "errors" "fmt" + "io/ioutil" "log" + "os" + "path/filepath" "github.com/kopia/kopia/auth" "github.com/kopia/kopia/blob" - "github.com/kopia/kopia/blob/caching" "github.com/kopia/kopia/blob/logging" + "github.com/kopia/kopia/internal/config" // Register well-known blob storage providers _ "github.com/kopia/kopia/blob/filesystem" _ "github.com/kopia/kopia/blob/gcs" ) -// Connect connects to the Repository specified in the specified configuration file. -func Connect(ctx context.Context, configFile string, options *ConnectOptions) (*Repository, error) { - lc, err := LoadFromFile(configFile) +// Options provides configuration parameters for connection to a repository. +type Options struct { + CredentialsCallback func() (auth.Credentials, error) // Provides credentials required to open the repository if not persisted. + TraceStorage func(f string, args ...interface{}) // Logs all storage access using provided Printf-style function + MaxDownloadSpeed int // Limits download speed. + MaxUploadSpeed int // Limits upload speed. + WriteBack int // Causes all object writes to be asynchronous with the specified number of workers. +} + +// Open opens a Repository specified in the configuration file. +func Open(ctx context.Context, configFile string, options *Options) (*Repository, error) { + if options == nil { + options = &Options{} + } + + configFile, err := filepath.Abs(configFile) + if err != nil { + return nil, err + } + + lc, err := config.LoadFromFile(configFile) if err != nil { return nil, err } var creds auth.Credentials - if len(lc.VaultConnection.Key) > 0 { - creds, err = auth.MasterKey(lc.VaultConnection.Key) + if len(lc.Connection.Key) > 0 { + creds, err = auth.MasterKey(lc.Connection.Key) } else { if options.CredentialsCallback == nil { - return nil, errors.New("vault key not persisted and no credentials specified") + return nil, errors.New("key not persisted and no credentials specified") } creds, err = options.CredentialsCallback() } if err != nil { - return nil, fmt.Errorf("invalid vault credentials: %v", err) + return nil, fmt.Errorf("invalid credentials: %v", err) } - rawVaultStorage, err := newStorageWithOptions(ctx, lc.VaultConnection.ConnectionInfo, options) + st, err := newStorageWithOptions(ctx, lc.Connection.ConnectionInfo, options) if err != nil { - return nil, fmt.Errorf("cannot open vault storage: %v", err) + return nil, fmt.Errorf("cannot open storage: %v", err) } - vaultStorage := rawVaultStorage - - if options.TraceStorage != nil { - vaultStorage = logging.NewWrapper(vaultStorage, logging.Prefix("[VAULT] "), logging.Output(options.TraceStorage)) - } - - vlt, err := Open(vaultStorage, creds) + r, err := connect(ctx, st, creds, options) if err != nil { - rawVaultStorage.Close() - return nil, fmt.Errorf("unable to open vault: %v", err) - } - - repositoryStorage := rawVaultStorage - if options.TraceStorage != nil { - repositoryStorage = logging.NewWrapper(repositoryStorage, logging.Prefix("[STORAGE] "), logging.Output(options.TraceStorage)) - } - - if lc.Caching != nil { - rs, err := caching.NewWrapper(ctx, repositoryStorage, lc.Caching) - if err != nil { - vaultStorage.Close() - repositoryStorage.Close() - return nil, err - } - repositoryStorage = rs - if options.TraceStorage != nil { - repositoryStorage = logging.NewWrapper(repositoryStorage, logging.Prefix("[CACHE] "), logging.Output(options.TraceStorage)) - } - } - - r, err := NewRepository(repositoryStorage, vlt.RepoConfig.Format) - if err != nil { - vaultStorage.Close() - repositoryStorage.Close() + st.Close() return nil, err } - r.Vault = vlt + + r.ConfigFile = configFile + r.CacheDirectory = applyDefaultString(lc.CacheDirectory, filepath.Join(filepath.Dir(configFile), "cache")) + return r, nil } -func newStorageWithOptions(ctx context.Context, cfg blob.ConnectionInfo, options *ConnectOptions) (blob.Storage, error) { +// ConnectOptions specifies options when persisting configuration to connect to a repository. +type ConnectOptions struct { + PersistCredentials bool + CacheDirectory string +} + +// Connect connects to the repository in the specified storage and persists the configuration and credentials in the file provided. +func Connect(ctx context.Context, configFile string, st blob.Storage, creds auth.Credentials, opt ConnectOptions) error { + r, err := connect(ctx, st, creds, nil) + if err != nil { + return err + } + + cfg, err := r.connectionConfiguration() + if err != nil { + return err + } + + var lc config.LocalConfig + lc.Connection = cfg + + if !opt.PersistCredentials { + lc.Connection.Key = nil + } + + d, err := json.MarshalIndent(&lc, "", " ") + if err != nil { + return err + } + + if err := os.MkdirAll(filepath.Dir(configFile), 0700); err != nil { + return err + } + + return ioutil.WriteFile(configFile, d, 0600) +} + +func connect(ctx context.Context, st blob.Storage, creds auth.Credentials, options *Options) (*Repository, error) { + if options == nil { + options = &Options{} + } + if options.TraceStorage != nil { + st = logging.NewWrapper(st, logging.Prefix("[STORAGE] "), logging.Output(options.TraceStorage)) + } + + mm, err := newMetadataManager(st, creds) + if err != nil { + return nil, fmt.Errorf("unable to open metadata manager: %v", err) + } + + om, err := newObjectManager(st, mm.repoConfig.Format, options) + if err != nil { + return nil, fmt.Errorf("unable to open object manager: %v", err) + } + + return &Repository{ + ObjectManager: om, + MetadataManager: mm, + Storage: st, + }, nil +} + +// Disconnect removes the specified configuration file and any local cache directories. +func Disconnect(configFile string) error { + _, err := config.LoadFromFile(configFile) + if err != nil { + return err + } + + return os.Remove(configFile) +} + +func newStorageWithOptions(ctx context.Context, cfg blob.ConnectionInfo, options *Options) (blob.Storage, error) { s, err := blob.NewStorage(ctx, cfg) if err != nil { return nil, err diff --git a/repo/connection_options.go b/repo/connection_options.go deleted file mode 100644 index eab40b366..000000000 --- a/repo/connection_options.go +++ /dev/null @@ -1,16 +0,0 @@ -package repo - -import ( - "github.com/kopia/kopia/auth" -) - -// ConnectOptions specifies the behavior of Connection. -type ConnectOptions struct { - CredentialsCallback func() (auth.Credentials, error) // credentials required to open the vault, unless persisted - - TraceStorage func(f string, args ...interface{}) - RepositoryOptions []RepositoryOption - - MaxDownloadSpeed int - MaxUploadSpeed int -} diff --git a/repo/initialize.go b/repo/initialize.go new file mode 100644 index 000000000..10108951c --- /dev/null +++ b/repo/initialize.go @@ -0,0 +1,115 @@ +package repo + +import ( + "crypto/rand" + "encoding/json" + "fmt" + "io" + + "github.com/kopia/kopia/auth" + "github.com/kopia/kopia/blob" + "github.com/kopia/kopia/internal/config" +) + +// NewRepositoryOptions specifies options that apply to newly created repositories. +// All fields are optional, when not provided, reasonable defaults will be used. +type NewRepositoryOptions struct { + MetadataEncryptionAlgorithm string // identifier of encryption algorithm + ObjectFormat string // identifier of object format + KeyDerivationAlgorithm string // identifier of key derivation algorithm + + MaxInlineContentLength int // maximum size of object to be considered for inline storage within ObjectID + Splitter string // splitter used to break objects into storage blocks + MinBlockSize int // minimum block size used with dynamic splitter + AvgBlockSize int // approximate size of storage block (used with dynamic splitter) + MaxBlockSize int // maximum size of storage block +} + +// Initialize creates initial repository data structures in the specified storage with given credentials. +func Initialize(st blob.Storage, opt *NewRepositoryOptions, creds auth.Credentials) error { + if opt == nil { + opt = &NewRepositoryOptions{} + } + + mm := MetadataManager{ + storage: st, + format: metadataFormatFromOptions(opt), + } + + var err error + mm.masterKey, err = creds.GetMasterKey(mm.format.Options) + if err != nil { + return err + } + + formatBytes, err := json.Marshal(&mm.format) + if err != nil { + return err + } + + if err := st.PutBlock(MetadataBlockPrefix+formatBlockID, formatBytes, blob.PutOptionsOverwrite); err != nil { + return err + } + + if err := mm.initCrypto(); err != nil { + return fmt.Errorf("unable to initialize crypto: %v", err) + } + + // Write encrypted repository configuration block. + rc := config.EncryptedRepositoryConfig{ + Format: repositoryObjectFormatFromOptions(opt), + } + + if err := mm.putJSON(repositoryConfigBlockID, &rc); err != nil { + return err + } + + return nil +} + +func metadataFormatFromOptions(opt *NewRepositoryOptions) config.MetadataFormat { + return config.MetadataFormat{ + Options: auth.Options{ + KeyDerivationAlgorithm: applyDefaultString(opt.KeyDerivationAlgorithm, auth.DefaultKeyDerivationAlgorithm), + UniqueID: randomBytes(32), + }, + Version: "1", + EncryptionAlgorithm: applyDefaultString(opt.MetadataEncryptionAlgorithm, DefaultMetadataEncryptionAlgorithm), + } +} + +func repositoryObjectFormatFromOptions(opt *NewRepositoryOptions) config.RepositoryObjectFormat { + return config.RepositoryObjectFormat{ + Version: 1, + Splitter: applyDefaultString(opt.Splitter, DefaultObjectSplitter), + ObjectFormat: applyDefaultString(opt.ObjectFormat, DefaultObjectFormat), + Secret: randomBytes(32), + MasterKey: randomBytes(32), + MaxInlineContentLength: applyDefaultInt(opt.MaxInlineContentLength, 32<<10), // 32KiB + MaxBlockSize: applyDefaultInt(opt.MaxBlockSize, 20<<20), // 20MiB + MinBlockSize: applyDefaultInt(opt.MinBlockSize, 10<<20), // 10MiB + AvgBlockSize: applyDefaultInt(opt.AvgBlockSize, 16<<20), // 16MiB + } +} + +func randomBytes(n int) []byte { + b := make([]byte, n) + io.ReadFull(rand.Reader, b) + return b +} + +func applyDefaultInt(v, def int) int { + if v == 0 { + return def + } + + return v +} + +func applyDefaultString(v, def string) string { + if v == "" { + return def + } + + return v +} diff --git a/repo/local_config.go b/repo/local_config.go deleted file mode 100644 index ec20119f2..000000000 --- a/repo/local_config.go +++ /dev/null @@ -1,50 +0,0 @@ -package repo - -import ( - "encoding/json" - "io" - "os" - - "github.com/kopia/kopia/blob" - "github.com/kopia/kopia/blob/caching" -) - -// LocalConfig is a configuration of Kopia. -type LocalConfig struct { - VaultConnection *VaultConfig `json:"vault,omitempty"` - RepoConnection *blob.ConnectionInfo `json:"repository,omitempty"` // can be nil indicating the same connection as for the vault - Caching *caching.Options `json:"caching,omitempty"` -} - -// Load reads local configuration from the specified reader. -func (lc *LocalConfig) Load(r io.Reader) error { - *lc = LocalConfig{} - return json.NewDecoder(r).Decode(lc) -} - -// Save writes the configuration to the specified writer. -func (lc *LocalConfig) Save(w io.Writer) error { - b, err := json.MarshalIndent(lc, "", " ") - if err != nil { - return nil - } - _, err = w.Write(b) - return err -} - -// LoadFromFile reads the local configuration from the specified file. -func LoadFromFile(fileName string) (*LocalConfig, error) { - f, err := os.Open(fileName) - if err != nil { - return nil, err - } - defer f.Close() - - var lc LocalConfig - - if err := lc.Load(f); err != nil { - return nil, err - } - - return &lc, nil -} diff --git a/repo/metadata_manager.go b/repo/metadata_manager.go new file mode 100644 index 000000000..c2484e977 --- /dev/null +++ b/repo/metadata_manager.go @@ -0,0 +1,330 @@ +package repo + +import ( + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "io" + "strings" + "sync" + + "github.com/kopia/kopia/auth" + "github.com/kopia/kopia/blob" + "github.com/kopia/kopia/internal/config" + + "golang.org/x/crypto/hkdf" +) + +const ( + formatBlockID = "format" + repositoryConfigBlockID = "repo" +) + +const ( + // MetadataBlockPrefix is a prefix used for metadata blocks in repository storage. + MetadataBlockPrefix = "VLT" +) + +var ( + purposeAESKey = []byte("AES") + purposeAuthData = []byte("CHECKSUM") +) + +// ErrMetadataNotFound is an error returned when a metadata item cannot be found. +var ErrMetadataNotFound = errors.New("metadata not found") + +// SupportedMetadataEncryptionAlgorithms is a list of supported metadata encryption algorithms including: +// +// AES256_GCM - AES-256 in GCM mode +// NONE - no encryption +var SupportedMetadataEncryptionAlgorithms []string + +// DefaultMetadataEncryptionAlgorithm is a metadata encryption algorithm used for new repositories. +const DefaultMetadataEncryptionAlgorithm = "AES256_GCM" + +func init() { + SupportedMetadataEncryptionAlgorithms = []string{ + "AES256_GCM", + "NONE", + } +} + +// MetadataManager manages JSON metadata, such as snapshot manifests, policies, object format etc. +// in a repository. +type MetadataManager struct { + storage blob.Storage + format config.MetadataFormat + repoConfig config.EncryptedRepositoryConfig + + masterKey []byte + + aead cipher.AEAD // authenticated encryption to use + authData []byte // additional data to authenticate +} + +// PutMetadata saves the specified metadata content under a provided name. +func (mm *MetadataManager) PutMetadata(itemID string, content []byte) error { + if err := checkReservedName(itemID); err != nil { + return err + } + + return mm.writeEncryptedBlock(itemID, content) +} + +func (mm *MetadataManager) writeEncryptedBlock(itemID string, content []byte) error { + if mm.aead != nil { + nonceLength := mm.aead.NonceSize() + noncePlusContentLength := nonceLength + len(content) + cipherText := make([]byte, noncePlusContentLength+mm.aead.Overhead()) + + // Store nonce at the beginning of ciphertext. + nonce := cipherText[0:nonceLength] + if _, err := io.ReadFull(rand.Reader, nonce); err != nil { + return err + } + + b := mm.aead.Seal(cipherText[nonceLength:nonceLength], nonce, content, mm.authData) + + content = nonce[0 : nonceLength+len(b)] + } + + return mm.storage.PutBlock(MetadataBlockPrefix+itemID, content, blob.PutOptionsOverwrite) +} + +func (mm *MetadataManager) readEncryptedBlock(itemID string) ([]byte, error) { + content, err := mm.storage.GetBlock(MetadataBlockPrefix + itemID) + if err != nil { + if err == blob.ErrBlockNotFound { + return nil, ErrMetadataNotFound + } + return nil, fmt.Errorf("unexpected error reading %v: %v", itemID, err) + } + + return mm.decryptBlock(content) +} + +func (mm *MetadataManager) decryptBlock(content []byte) ([]byte, error) { + if mm.aead != nil { + nonce := content[0:mm.aead.NonceSize()] + payload := content[mm.aead.NonceSize():] + return mm.aead.Open(payload[:0], nonce, payload, mm.authData) + } + + return content, nil +} + +// DeriveKey computes a key for a specific purpose and length using HKDF based on the master key. +func (mm *MetadataManager) DeriveKey(purpose []byte, length int) []byte { + key := make([]byte, length) + k := hkdf.New(sha256.New, mm.masterKey, mm.format.UniqueID, purpose) + io.ReadFull(k, key) + return key +} + +// GetMetadata returns the contents of a specified metadata item. +func (mm *MetadataManager) GetMetadata(itemID string) ([]byte, error) { + if err := checkReservedName(itemID); err != nil { + return nil, err + } + + return mm.readEncryptedBlock(itemID) +} + +func (mm *MetadataManager) getJSON(itemID string, content interface{}) error { + j, err := mm.readEncryptedBlock(itemID) + if err != nil { + return err + } + + return json.Unmarshal(j, content) +} + +// putJSON stores the contents of an item stored with a given ID. +func (mm *MetadataManager) putJSON(id string, content interface{}) error { + j, err := json.Marshal(content) + if err != nil { + return err + } + + return mm.writeEncryptedBlock(id, j) +} + +// ListMetadata returns the list of metadata items matching the specified prefix. +// The 'limit' parameter specifies the maximum number of items to retrieve (-1 == unlimited). +func (mm *MetadataManager) ListMetadata(prefix string, limit int) ([]string, error) { + var result []string + + ch, cancel := mm.storage.ListBlocks(MetadataBlockPrefix + prefix) + defer cancel() + for b := range ch { + if limit == 0 { + break + } + if b.Error != nil { + return result, b.Error + } + + itemID := strings.TrimPrefix(b.BlockID, MetadataBlockPrefix) + if !isReservedName(itemID) { + result = append(result, itemID) + } + if limit > 0 { + limit-- + } + } + return result, nil +} + +// Config returns a configuration of storage its credentials that's suitable +// for storing in configuration file. +func (mm *MetadataManager) connectionConfiguration() (*config.RepositoryConnectionInfo, error) { + cip, ok := mm.storage.(blob.ConnectionInfoProvider) + if !ok { + return nil, errors.New("repository does not support persisting configuration") + } + + ci := cip.ConnectionInfo() + + return &config.RepositoryConnectionInfo{ + ConnectionInfo: ci, + Key: mm.masterKey, + }, nil +} + +// RemoveMetadata removes the specified metadata item. +func (mm *MetadataManager) RemoveMetadata(itemID string) error { + if err := checkReservedName(itemID); err != nil { + return err + } + + return mm.storage.DeleteBlock(MetadataBlockPrefix + itemID) +} + +// RemoveMany efficiently removes multiple metadata items in parallel. +func (mm *MetadataManager) RemoveMany(itemIDs []string) error { + parallelism := 30 + ch := make(chan string) + var wg sync.WaitGroup + errch := make(chan error, len(itemIDs)) + + for i := 0; i < parallelism; i++ { + wg.Add(1) + go func() { + defer wg.Done() + + for id := range ch { + if err := mm.RemoveMetadata(id); err != nil { + errch <- err + } + } + }() + } + + for _, id := range itemIDs { + ch <- id + } + close(ch) + wg.Wait() + close(errch) + + return <-errch +} + +// newMetadataManager opens a MetadataManager for given storage and credentials. +func newMetadataManager(st blob.Storage, creds auth.Credentials) (*MetadataManager, error) { + mm := MetadataManager{ + storage: st, + } + + var wg sync.WaitGroup + + var blocks [4][]byte + + f := func(index int, name string) { + blocks[index], _ = st.GetBlock(name) + wg.Done() + } + + wg.Add(2) + go f(0, MetadataBlockPrefix+formatBlockID) + go f(1, MetadataBlockPrefix+repositoryConfigBlockID) + wg.Wait() + + if blocks[0] == nil { + return nil, fmt.Errorf("format block not found") + } + + var offset = 0 + err := json.Unmarshal(blocks[offset], &mm.format) + if err != nil { + return nil, err + } + + mm.masterKey, err = creds.GetMasterKey(mm.format.Options) + if err != nil { + return nil, err + } + + if err := mm.initCrypto(); err != nil { + return nil, fmt.Errorf("unable to initialize crypto: %v", err) + } + + cfgData, err := mm.decryptBlock(blocks[offset+1]) + if err != nil { + return nil, err + } + + var rc config.EncryptedRepositoryConfig + + if err := json.Unmarshal(cfgData, &rc); err != nil { + return nil, err + } + + mm.repoConfig = rc + + return &mm, nil +} + +func (mm *MetadataManager) initCrypto() error { + switch mm.format.EncryptionAlgorithm { + case "NONE": // do nothing + return nil + case "AES256_GCM": + aesKey := mm.DeriveKey(purposeAESKey, 32) + mm.authData = mm.DeriveKey(purposeAuthData, 32) + + blk, err := aes.NewCipher(aesKey) + if err != nil { + return fmt.Errorf("cannot create cipher: %v", err) + } + mm.aead, err = cipher.NewGCM(blk) + if err != nil { + return fmt.Errorf("cannot create cipher: %v", err) + } + return nil + default: + return fmt.Errorf("unknown encryption algorithm: '%v'", mm.format.EncryptionAlgorithm) + } +} + +func isReservedName(itemID string) bool { + switch itemID { + case formatBlockID, repositoryConfigBlockID: + return true + + default: + return false + } +} + +func checkReservedName(itemID string) error { + if isReservedName(itemID) { + return fmt.Errorf("invalid metadata item name: '%v'", itemID) + } + + return nil +} diff --git a/repo/metadata_manager_test.go b/repo/metadata_manager_test.go new file mode 100644 index 000000000..66708f92b --- /dev/null +++ b/repo/metadata_manager_test.go @@ -0,0 +1,177 @@ +package repo + +import ( + "context" + "io/ioutil" + "os" + "reflect" + "strings" + + "github.com/kopia/kopia/auth" + "github.com/kopia/kopia/blob" + "github.com/kopia/kopia/blob/filesystem" + + "testing" +) + +func TestMetadataManager(t *testing.T) { + tmpDir, err := ioutil.TempDir("", "metadata") + if err != nil { + t.Errorf("can't create temp dir: %v", err) + return + } + defer os.RemoveAll(tmpDir) + + st := mustCreateFileStorage(t, tmpDir) + + creds, err := auth.Password("foo.bar.baz.foo.bar.baz") + if err != nil { + t.Errorf("can't create password credentials: %v", err) + return + } + + otherCreds, err := auth.Password("foo.bar.baz.foo.bar.baz0") + if err != nil { + t.Errorf("can't create password credentials: %v", err) + return + } + + if err := Initialize(st, nil, creds); err != nil { + t.Errorf("can't initialize repository: %v", err) + return + } + + v1, err := newMetadataManager(st, creds) + if err != nil { + t.Errorf("can't open first metadata manager: %v", err) + return + } + + v2, err := newMetadataManager(st, creds) + if err != nil { + t.Errorf("can't open second metadata manager: %v", err) + return + } + + cfg, err := v1.connectionConfiguration() + if err != nil { + t.Errorf("error getting token1 %v", err) + } + + cfg2, err := v2.connectionConfiguration() + if err != nil { + t.Errorf("error getting token2 %v", err) + } + + if !reflect.DeepEqual(cfg, cfg2) { + t.Errorf("configurations are different: %+v vs %+v", cfg, cfg2) + } + + _, err = newMetadataManager(st, otherCreds) + if err == nil { + t.Errorf("unexpectedly opened repository with invalid credentials") + return + } + + if err := v1.PutMetadata("foo", []byte("test1")); err != nil { + t.Errorf("error putting: %v", err) + } + if err := v2.PutMetadata("bar", []byte("test2")); err != nil { + t.Errorf("error putting: %v", err) + } + if err := v1.PutMetadata("baz", []byte("test3")); err != nil { + t.Errorf("error putting: %v", err) + } + + // Verify contents of metadata items for both managers. + for _, v := range []*MetadataManager{v1, v2} { + assertMetadataItem(t, v, "foo", "test1") + assertMetadataItem(t, v, "bar", "test2") + assertMetadataItem(t, v, "baz", "test3") + assertMetadataItemNotFound(t, v, "x") + + assertMetadataItems(t, v, "x", nil) + assertMetadataItems(t, v, "f", []string{"foo"}) + assertMetadataItems(t, v, "ba", []string{"bar", "baz"}) + assertMetadataItems(t, v, "be", nil) + assertMetadataItems(t, v, "baz", []string{"baz"}) + assertMetadataItems(t, v, "bazx", nil) + + assertReservedName(t, v, formatBlockID) + assertReservedName(t, v, repositoryConfigBlockID) + } + + v1.RemoveMetadata("bar") + + for _, v := range []*MetadataManager{v1, v2} { + assertMetadataItem(t, v, "foo", "test1") + assertMetadataItemNotFound(t, v, "bar") + assertMetadataItem(t, v, "baz", "test3") + + assertMetadataItems(t, v, "x", nil) + assertMetadataItems(t, v, "f", []string{"foo"}) + assertMetadataItems(t, v, "ba", []string{"baz"}) + assertMetadataItems(t, v, "be", nil) + assertMetadataItems(t, v, "baz", []string{"baz"}) + assertMetadataItems(t, v, "bazx", nil) + } +} + +func assertMetadataItem(t *testing.T, v *MetadataManager, itemID string, expectedData string) { + b, err := v.GetMetadata(itemID) + if err != nil { + t.Errorf("error getting item %v: %v", itemID, err) + } + + bs := string(b) + if bs != expectedData { + t.Errorf("invalid data for '%v': expected: %v but got %v", itemID, expectedData, bs) + } +} + +func assertMetadataItemNotFound(t *testing.T, v *MetadataManager, itemID string) { + result, err := v.GetMetadata(itemID) + if err != ErrMetadataNotFound { + t.Errorf("invalid error getting item %v: %v (result=%v), but expected %v", itemID, err, result, ErrMetadataNotFound) + } +} + +func assertReservedName(t *testing.T, v *MetadataManager, itemID string) { + _, err := v.GetMetadata(itemID) + assertReservedNameError(t, "Get", itemID, err) + assertReservedNameError(t, "Put", itemID, v.PutMetadata(itemID, nil)) + assertReservedNameError(t, "Remove", itemID, v.RemoveMetadata(itemID)) +} + +func assertReservedNameError(t *testing.T, method string, itemID string, err error) { + if err == nil { + t.Errorf("call did not fail: %v(%v)", method, itemID) + return + } + + if !strings.Contains(err.Error(), "invalid metadata item name") { + t.Errorf("call did not fail the right way: %v(%v), was: %v", method, itemID, err) + } +} + +func assertMetadataItems(t *testing.T, v *MetadataManager, prefix string, expected []string) { + res, err := v.ListMetadata(prefix, -1) + if err != nil { + t.Errorf("error listing items beginning with %v: %v", prefix, err) + } + + if !reflect.DeepEqual(expected, res) { + t.Errorf("unexpected items returned for prefix '%v': %v, but expected %v", prefix, res, expected) + } +} + +func mustCreateFileStorage(t *testing.T, path string) blob.Storage { + os.MkdirAll(path, 0700) + s, err := filesystem.New(context.Background(), &filesystem.Options{ + Path: path, + }) + if err != nil { + t.Errorf("can't create file storage: %v", err) + } + return s +} diff --git a/repo/format.go b/repo/object_formatter.go similarity index 57% rename from repo/format.go rename to repo/object_formatter.go index 00c1e3c4a..39c95bec2 100644 --- a/repo/format.go +++ b/repo/object_formatter.go @@ -9,24 +9,16 @@ import ( "encoding/hex" "fmt" "hash" + + "strings" + + "sort" + + "github.com/kopia/kopia/internal/config" ) -// Format describes the format of objects in a repository. -type Format struct { - Version int32 `json:"version,omitempty"` // version number, must be "1" - ObjectFormat string `json:"objectFormat,omitempty"` // identifier of object format - Secret []byte `json:"secret,omitempty"` // HMAC secret used to generate encryption keys - MaxInlineContentLength int32 `json:"maxInlineContentLength,omitempty"` // maximum size of object to be considered for inline storage within ObjectID - MasterKey []byte `json:"masterKey,omitempty"` // master encryption key (SIV-mode encryption only) - Splitter string `json:"splitter,omitempty"` // splitter used to break objects into storage blocks - - MinBlockSize int32 `json:"minBlockSize,omitempty"` // minimum block size used with dynamic splitter - AvgBlockSize int32 `json:"avgBlockSize,omitempty"` // approximate size of storage block (used with dynamic splitter) - MaxBlockSize int32 `json:"maxBlockSize,omitempty"` // maximum size of storage block -} - -// Validate checks the validity of a Format and returns an error if invalid. -func (f *Format) Validate() error { +// validateFormat checks the validity of RepositoryObjectFormat and returns an error if invalid. +func validateFormat(f *config.RepositoryObjectFormat) error { if f.Version != 1 { return fmt.Errorf("unsupported version: %v", f.Version) } @@ -35,26 +27,22 @@ func (f *Format) Validate() error { return fmt.Errorf("MaxBlockSize is not set") } - sf := SupportedFormats[f.ObjectFormat] - if sf == nil { + if sf := objectFormatterFactories[f.ObjectFormat]; sf == nil { return fmt.Errorf("unknown object format: %v", f.ObjectFormat) } return nil } -// ObjectFormatter performs data block ID computation and encryption of a block of data when storing object in a repository, -type ObjectFormatter interface { - // ComputeObjectID computes ID of the storage block and encryption key for the specified block of data - // and returns them in ObjectID. +// ObjectFormatter performs data block ID computation and encryption of a block of data when storing object in a repository. +type objectFormatter interface { + // ComputeObjectID computes ID of the storage block for the specified block of data and returns it in ObjectID. ComputeObjectID(data []byte) ObjectID - // Encrypt returns encrypted bytes corresponding to the given plaintext. - // Encryption in-place is allowed within original slice's capacity. Encryption parameters are passed in ObjectID. + // Encrypt returns encrypted bytes corresponding to the given plaintext. May reuse the input slice. Encrypt(plainText []byte, oid ObjectID) ([]byte, error) - // Decrypt returns unencrypted bytes corresponding to the given ciphertext. - // Decryption in-place is allowed within original slice's capacity. Encryption parameters are passed in ObjectID. + // Decrypt returns unencrypted bytes corresponding to the given ciphertext. May reuse the input slice. Decrypt(cipherText []byte, oid ObjectID) ([]byte, error) } @@ -126,37 +114,42 @@ func decodeHexSuffix(s string, length int) ([]byte, error) { return hex.DecodeString(s[len(s)-length:]) } -// SupportedFormats is a map with an ObjectFormatter for each supported object format: +// SupportedObjectFormats is a list of supported object formats including: // // UNENCRYPTED_HMAC_SHA256_128 - unencrypted, block IDs are 128-bit (32 characters long) // UNENCRYPTED_HMAC_SHA256 - unencrypted, block IDs are 256-bit (64 characters long) // ENCRYPTED_HMAC_SHA256_AES256_SIV - encrypted with AES-256 (shared key), IV==FOLD(HMAC-SHA256(content), 128) -// -// Additional formats can be supported by adding them to the map. -var SupportedFormats map[string]func(f *Format) (ObjectFormatter, error) +var SupportedObjectFormats []string + +var objectFormatterFactories = map[string]func(f *config.RepositoryObjectFormat) (objectFormatter, error){ + "TESTONLY_MD5": func(f *config.RepositoryObjectFormat) (objectFormatter, error) { + return &unencryptedFormat{computeHash(md5.New, md5.Size)}, nil + }, + "UNENCRYPTED_HMAC_SHA256": func(f *config.RepositoryObjectFormat) (objectFormatter, error) { + return &unencryptedFormat{computeHMAC(sha256.New, f.Secret, sha256.Size)}, nil + }, + "UNENCRYPTED_HMAC_SHA256_128": func(f *config.RepositoryObjectFormat) (objectFormatter, error) { + return &unencryptedFormat{computeHMAC(sha256.New, f.Secret, 16)}, nil + }, + "ENCRYPTED_HMAC_SHA256_AES256_SIV": func(f *config.RepositoryObjectFormat) (objectFormatter, error) { + if len(f.MasterKey) < 32 { + return nil, fmt.Errorf("master key is not set") + } + return &syntheticIVEncryptionFormat{computeHMAC(sha256.New, f.Secret, aes.BlockSize), aes.NewCipher, f.MasterKey}, nil + }, +} func init() { - SupportedFormats = map[string]func(f *Format) (ObjectFormatter, error){ - "TESTONLY_MD5": func(f *Format) (ObjectFormatter, error) { - return &unencryptedFormat{computeHash(md5.New, md5.Size)}, nil - }, - "UNENCRYPTED_HMAC_SHA256": func(f *Format) (ObjectFormatter, error) { - return &unencryptedFormat{computeHMAC(sha256.New, f.Secret, sha256.Size)}, nil - }, - "UNENCRYPTED_HMAC_SHA256_128": func(f *Format) (ObjectFormatter, error) { - return &unencryptedFormat{computeHMAC(sha256.New, f.Secret, 16)}, nil - }, - "ENCRYPTED_HMAC_SHA256_AES256_SIV": func(f *Format) (ObjectFormatter, error) { - if len(f.MasterKey) < 32 { - return nil, fmt.Errorf("master key is not set") - } - return &syntheticIVEncryptionFormat{computeHMAC(sha256.New, f.Secret, aes.BlockSize), aes.NewCipher, f.MasterKey}, nil - }, + for k := range objectFormatterFactories { + if !strings.HasPrefix(k, "TESTONLY_") { + SupportedObjectFormats = append(SupportedObjectFormats, k) + } } + sort.Strings(SupportedObjectFormats) } // DefaultObjectFormat is the format that should be used by default when creating new repositories. -var DefaultObjectFormat = "ENCRYPTED_HMAC_SHA256_AES256_SIV" +const DefaultObjectFormat = "ENCRYPTED_HMAC_SHA256_AES256_SIV" // computeHash returns a digestFunction that computes a hash of a given block of bytes and truncates results to the given size. func computeHash(hf func() hash.Hash, truncate int) digestFunction { diff --git a/repo/format_test.go b/repo/object_formatter_test.go similarity index 85% rename from repo/format_test.go rename to repo/object_formatter_test.go index b030520a9..8dbe707b1 100644 --- a/repo/format_test.go +++ b/repo/object_formatter_test.go @@ -4,15 +4,16 @@ import ( "bytes" "crypto/rand" "crypto/sha1" - "testing" + + "github.com/kopia/kopia/internal/config" ) func TestObjectFormatters(t *testing.T) { secret := []byte("secret") - f := &Format{Secret: secret, MasterKey: make([]byte, 32)} + f := &config.RepositoryObjectFormat{Secret: secret, MasterKey: make([]byte, 32)} - for k, v := range SupportedFormats { + for k, v := range objectFormatterFactories { data := make([]byte, 100) rand.Read(data) diff --git a/repo/cas_manager.go b/repo/object_manager.go similarity index 65% rename from repo/cas_manager.go rename to repo/object_manager.go index 836e5a6ef..2ff28128a 100644 --- a/repo/cas_manager.go +++ b/repo/object_manager.go @@ -10,7 +10,7 @@ import ( "sync/atomic" "github.com/kopia/kopia/blob" - "github.com/kopia/kopia/blob/logging" + "github.com/kopia/kopia/internal/config" "github.com/kopia/kopia/internal/jsonstream" ) @@ -22,15 +22,15 @@ type ObjectReader interface { Length() int64 } -// casManager implements a content-addressable storage on top of blob storage. -type casManager struct { - Stats Stats // vital statistics - Storage blob.Storage // underlying blob storage +// ObjectManager implements a content-addressable storage on top of blob storage. +type ObjectManager struct { + stats Stats + storage blob.Storage verbose bool bufferManager *bufferManager - format Format - formatter ObjectFormatter + format config.RepositoryObjectFormat + formatter objectFormatter writeBack writebackManager @@ -38,7 +38,7 @@ type casManager struct { } // Close closes the connection to the underlying blob storage and releases any resources. -func (r *casManager) Close() error { +func (r *ObjectManager) Close() error { r.writeBack.flush() r.bufferManager.close() @@ -46,22 +46,18 @@ func (r *casManager) Close() error { } // NewWriter creates an ObjectWriter for writing to the repository. -func (r *casManager) NewWriter(options ...WriterOption) ObjectWriter { - result := &objectWriter{ +func (r *ObjectManager) NewWriter(opt WriterOptions) ObjectWriter { + return &objectWriter{ repo: r, blockTracker: &blockTracker{}, splitter: r.newSplitter(), + description: opt.Description, + prefix: opt.BlockNamePrefix, } - - for _, option := range options { - option(result) - } - - return result } // Open creates new ObjectReader for reading given object from a repository. -func (r *casManager) Open(objectID ObjectID) (ObjectReader, error) { +func (r *ObjectManager) Open(objectID ObjectID) (ObjectReader, error) { // log.Printf("Repository::Open %v", objectID.String()) // defer log.Printf("finished Repository::Open() %v", objectID.String()) @@ -101,60 +97,35 @@ func (r *casManager) Open(objectID ObjectID) (ObjectReader, error) { return r.newRawReader(objectID) } -// RepositoryOption controls the behavior of Repository. -type RepositoryOption func(o *casManager) - -// WriteBack is an RepositoryOption that enables asynchronous writes to the storage using the pool -// of goroutines. -func WriteBack(writeBackWorkers int) RepositoryOption { - return func(o *casManager) { - o.writeBack.workers = writeBackWorkers - } -} - -// EnableLogging is an RepositoryOption that causes all storage access to be logged. -func EnableLogging(options ...logging.Option) RepositoryOption { - return func(o *casManager) { - o.Storage = logging.NewWrapper(o.Storage, options...) - } -} - -// newCASManager creates a casManager with the specified storage, format and options. -func newCASManager(s blob.Storage, f *Format, options ...RepositoryOption) (*casManager, error) { - if err := f.Validate(); err != nil { +// newObjectManager creates an ObjectManager with the specified storage, format and options. +func newObjectManager(s blob.Storage, f config.RepositoryObjectFormat, opts *Options) (*ObjectManager, error) { + if err := validateFormat(&f); err != nil { return nil, err } - sf := SupportedFormats[f.ObjectFormat] - - r := &casManager{ - Storage: s, - format: *f, + sf := objectFormatterFactories[f.ObjectFormat] + r := &ObjectManager{ + storage: s, + format: f, } - sp := f.Splitter - if sp == "" { - sp = "FIXED" - } - - os := SupportedSplitters[sp] + os := objectSplitterFactories[applyDefaultString(f.Splitter, "FIXED")] if os == nil { - return nil, fmt.Errorf("unsupported splitter %q", sp) + return nil, fmt.Errorf("unsupported splitter %q", f.Splitter) } r.newSplitter = func() objectSplitter { - return os(f) + return os(&r.format) } var err error - - r.formatter, err = sf(f) + r.formatter, err = sf(&r.format) if err != nil { return nil, err } - for _, o := range options { - o(r) + if opts != nil { + r.writeBack.workers = opts.WriteBack } r.bufferManager = newBufferManager(int(r.format.MaxBlockSize)) @@ -168,7 +139,7 @@ func newCASManager(s blob.Storage, f *Format, options ...RepositoryOption) (*cas // hashEncryptAndWriteMaybeAsync computes hash of a given buffer, optionally encrypts and writes it to storage. // The write is not guaranteed to complete synchronously in case write-back is used, but by the time // Repository.Close() returns all writes are guaranteed be over. -func (r *casManager) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix string) (ObjectID, error) { +func (r *ObjectManager) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix string) (ObjectID, error) { var data []byte if buffer != nil { data = buffer.Bytes() @@ -181,8 +152,8 @@ func (r *casManager) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix // Hash the block and compute encryption key. objectID := r.formatter.ComputeObjectID(data) objectID.StorageBlock = prefix + objectID.StorageBlock - atomic.AddInt32(&r.Stats.HashedBlocks, 1) - atomic.AddInt64(&r.Stats.HashedBytes, int64(len(data))) + atomic.AddInt32(&r.stats.HashedBlocks, 1) + atomic.AddInt64(&r.stats.HashedBytes, int64(len(data))) if r.writeBack.enabled() { r.writeBack.waitGroup.Add(1) @@ -202,7 +173,7 @@ func (r *casManager) hashEncryptAndWriteMaybeAsync(buffer *bytes.Buffer, prefix return r.encryptAndMaybeWrite(objectID, buffer, prefix) } -func (r *casManager) encryptAndMaybeWrite(objectID ObjectID, buffer *bytes.Buffer, prefix string) (ObjectID, error) { +func (r *ObjectManager) encryptAndMaybeWrite(objectID ObjectID, buffer *bytes.Buffer, prefix string) (ObjectID, error) { defer r.bufferManager.returnBuffer(buffer) var data []byte @@ -211,10 +182,10 @@ func (r *casManager) encryptAndMaybeWrite(objectID ObjectID, buffer *bytes.Buffe } // Before performing encryption, check if the block is already there. - blockSize, err := r.Storage.BlockSize(objectID.StorageBlock) - atomic.AddInt32(&r.Stats.CheckedBlocks, int32(1)) + blockSize, err := r.storage.BlockSize(objectID.StorageBlock) + atomic.AddInt32(&r.stats.CheckedBlocks, int32(1)) if err == nil && blockSize == int64(len(data)) { - atomic.AddInt32(&r.Stats.PresentBlocks, int32(1)) + atomic.AddInt32(&r.stats.PresentBlocks, int32(1)) // Block already exists in storage, correct size, return without uploading. return objectID, nil } @@ -225,23 +196,23 @@ func (r *casManager) encryptAndMaybeWrite(objectID ObjectID, buffer *bytes.Buffe } // Encrypt the block in-place. - atomic.AddInt64(&r.Stats.EncryptedBytes, int64(len(data))) + atomic.AddInt64(&r.stats.EncryptedBytes, int64(len(data))) data, err = r.formatter.Encrypt(data, objectID) if err != nil { return NullObjectID, err } - atomic.AddInt32(&r.Stats.WrittenBlocks, int32(1)) - atomic.AddInt64(&r.Stats.WrittenBytes, int64(len(data))) + atomic.AddInt32(&r.stats.WrittenBlocks, int32(1)) + atomic.AddInt64(&r.stats.WrittenBytes, int64(len(data))) - if err := r.Storage.PutBlock(objectID.StorageBlock, data, blob.PutOptionsDefault); err != nil { + if err := r.storage.PutBlock(objectID.StorageBlock, data, blob.PutOptionsDefault); err != nil { r.writeBack.errors.add(err) } return objectID, nil } -func (r *casManager) flattenListChunk(rawReader io.Reader) ([]indirectObjectEntry, error) { +func (r *ObjectManager) flattenListChunk(rawReader io.Reader) ([]indirectObjectEntry, error) { pr, err := jsonstream.NewReader(bufio.NewReader(rawReader), indirectStreamType) if err != nil { return nil, err @@ -275,7 +246,7 @@ func removeIndirection(o ObjectID) ObjectID { return o } -func (r *casManager) newRawReader(objectID ObjectID) (ObjectReader, error) { +func (r *ObjectManager) newRawReader(objectID ObjectID) (ObjectReader, error) { if objectID.BinaryContent != nil { return newObjectReaderWithData(objectID.BinaryContent), nil } @@ -285,16 +256,16 @@ func (r *casManager) newRawReader(objectID ObjectID) (ObjectReader, error) { } blockID := objectID.StorageBlock - payload, err := r.Storage.GetBlock(blockID) + payload, err := r.storage.GetBlock(blockID) if err != nil { return nil, err } - atomic.AddInt32(&r.Stats.ReadBlocks, 1) - atomic.AddInt64(&r.Stats.ReadBytes, int64(len(payload))) + atomic.AddInt32(&r.stats.ReadBlocks, 1) + atomic.AddInt64(&r.stats.ReadBytes, int64(len(payload))) payload, err = r.formatter.Decrypt(payload, objectID) - atomic.AddInt64(&r.Stats.DecryptedBytes, int64(len(payload))) + atomic.AddInt64(&r.stats.DecryptedBytes, int64(len(payload))) if err != nil { return nil, err } @@ -308,14 +279,14 @@ func (r *casManager) newRawReader(objectID ObjectID) (ObjectReader, error) { return newObjectReaderWithData(payload), nil } -func (r *casManager) verifyChecksum(data []byte, blockID string) error { +func (r *ObjectManager) verifyChecksum(data []byte, blockID string) error { expected := r.formatter.ComputeObjectID(data) if !strings.HasSuffix(blockID, expected.StorageBlock) { - atomic.AddInt32(&r.Stats.InvalidBlocks, 1) + atomic.AddInt32(&r.stats.InvalidBlocks, 1) return fmt.Errorf("invalid checksum for blob: '%v', expected %v", blockID, expected.StorageBlock) } - atomic.AddInt32(&r.Stats.ValidBlocks, 1) + atomic.AddInt32(&r.stats.ValidBlocks, 1) return nil } diff --git a/repo/cas_manager_test.go b/repo/object_manager_test.go similarity index 90% rename from repo/cas_manager_test.go rename to repo/object_manager_test.go index ed1a8737b..0218535e6 100644 --- a/repo/cas_manager_test.go +++ b/repo/object_manager_test.go @@ -13,6 +13,7 @@ import ( "testing" "github.com/kopia/kopia/blob" + "github.com/kopia/kopia/internal/config" "github.com/kopia/kopia/internal/jsonstream" "github.com/kopia/kopia/internal/storagetesting" ) @@ -21,8 +22,8 @@ func init() { panicOnBufferLeaks = true } -func testFormat() *Format { - return &Format{ +func testFormat() config.RepositoryObjectFormat { + return config.RepositoryObjectFormat{ Version: 1, MaxBlockSize: 200, MaxInlineContentLength: 20, @@ -44,11 +45,13 @@ func getMd5LObjectID(data []byte) string { return fmt.Sprintf("L%s", getMd5Digest(data)) } -func setupTest(t *testing.T) (data map[string][]byte, repo *casManager) { +func setupTest(t *testing.T) (data map[string][]byte, om *ObjectManager) { data = map[string][]byte{} st := storagetesting.NewMapStorage(data) - repo, err := newCASManager(st, testFormat(), WriteBack(5)) + om, err := newObjectManager(st, testFormat(), &Options{ + WriteBack: 5, + }) if err != nil { t.Errorf("cannot create manager: %v", err) } @@ -76,9 +79,9 @@ func TestWriters(t *testing.T) { for _, c := range cases { data, repo := setupTest(t) - writer := repo.NewWriter( - WithBlockNamePrefix("X"), - ) + writer := repo.NewWriter(WriterOptions{ + BlockNamePrefix: "X", + }) writer.Write(c.data) @@ -114,9 +117,9 @@ func TestWriterCompleteChunkInTwoWrites(t *testing.T) { _, repo := setupTest(t) bytes := make([]byte, 100) - writer := repo.NewWriter( - WithBlockNamePrefix("X"), - ) + writer := repo.NewWriter(WriterOptions{ + BlockNamePrefix: "X", + }) writer.Write(bytes[0:50]) writer.Write(bytes[0:50]) result, err := writer.Result(false) @@ -125,7 +128,7 @@ func TestWriterCompleteChunkInTwoWrites(t *testing.T) { } } -func verifyIndirectBlock(t *testing.T, r *casManager, oid ObjectID) { +func verifyIndirectBlock(t *testing.T, r *ObjectManager, oid ObjectID) { for level := int32(0); level < oid.Indirect; level++ { direct := oid direct.Indirect = level @@ -175,7 +178,7 @@ func TestIndirection(t *testing.T) { contentBytes := make([]byte, c.dataLength) - writer := repo.NewWriter() + writer := repo.NewWriter(WriterOptions{}) writer.Write(contentBytes) result, err := writer.Result(false) repo.writeBack.flush() @@ -211,11 +214,11 @@ func TestHMAC(t *testing.T) { s := testFormat() s.ObjectFormat = "TESTONLY_MD5" - repo, err := newCASManager(storagetesting.NewMapStorage(data), s) + repo, err := newObjectManager(storagetesting.NewMapStorage(data), s, nil) if err != nil { t.Errorf("cannot create manager: %v", err) } - w := repo.NewWriter() + w := repo.NewWriter(WriterOptions{}) w.Write(content) result, err := w.Result(false) if result.String() != "D999732b72ceff665b3f7608411db66a4" { @@ -316,9 +319,9 @@ func TestEndToEndReadAndSeek(t *testing.T) { randomData := make([]byte, size) cryptorand.Read(randomData) - writer := repo.NewWriter( - WithBlockNamePrefix("X"), - ) + writer := repo.NewWriter(WriterOptions{ + BlockNamePrefix: "X", + }) writer.Write(randomData) objectID, err := writer.Result(forceStored) writer.Close() @@ -342,7 +345,7 @@ func TestEndToEndReadAndSeek(t *testing.T) { } } -func verify(t *testing.T, repo *casManager, objectID ObjectID, expectedData []byte, testCaseID string) { +func verify(t *testing.T, repo *ObjectManager, objectID ObjectID, expectedData []byte, testCaseID string) { reader, err := repo.Open(objectID) if err != nil { t.Errorf("cannot get reader for %v: %v", testCaseID, err) @@ -374,8 +377,8 @@ func verify(t *testing.T, repo *casManager, objectID ObjectID, expectedData []by } func TestFormats(t *testing.T) { - makeFormat := func(objectFormat string) *Format { - return &Format{ + makeFormat := func(objectFormat string) config.RepositoryObjectFormat { + return config.RepositoryObjectFormat{ Version: 1, ObjectFormat: objectFormat, Secret: []byte("key"), @@ -384,11 +387,11 @@ func TestFormats(t *testing.T) { } cases := []struct { - format *Format + format config.RepositoryObjectFormat oids map[string]ObjectID }{ { - format: &Format{ + format: config.RepositoryObjectFormat{ Version: 1, ObjectFormat: "TESTONLY_MD5", MaxBlockSize: 10000, @@ -422,7 +425,7 @@ func TestFormats(t *testing.T) { data := map[string][]byte{} st := storagetesting.NewMapStorage(data) - repo, err := newCASManager(st, c.format) + repo, err := newObjectManager(st, c.format, nil) if err != nil { t.Errorf("cannot create manager: %v", err) continue @@ -430,7 +433,7 @@ func TestFormats(t *testing.T) { for k, v := range c.oids { bytesToWrite := []byte(k) - w := repo.NewWriter() + w := repo.NewWriter(WriterOptions{}) w.Write(bytesToWrite) oid, err := w.Result(true) if err != nil { diff --git a/repo/object_reader.go b/repo/object_reader.go index 6f87f6160..f20f0b1fa 100644 --- a/repo/object_reader.go +++ b/repo/object_reader.go @@ -10,7 +10,7 @@ func (i *indirectObjectEntry) endOffset() int64 { } type objectReader struct { - repo *casManager + repo *ObjectManager seekTable []indirectObjectEntry diff --git a/repo/object_splitter.go b/repo/object_splitter.go index c221b404a..79f0a776b 100644 --- a/repo/object_splitter.go +++ b/repo/object_splitter.go @@ -3,14 +3,46 @@ package repo import ( "math" + "sort" + "github.com/chmduquesne/rollinghash" "github.com/chmduquesne/rollinghash/buzhash32" + "github.com/kopia/kopia/internal/config" ) type objectSplitter interface { add(b byte) bool } +// SupportedObjectSplitters is a list of supported object splitters including: +// +// NEVER - prevents objects from ever splitting +// FIXED - always splits large objects exactly at the maximum block size boundary +// DYNAMIC - dynamicaly splits large objects based on rolling hash of contents. +var SupportedObjectSplitters []string + +var objectSplitterFactories = map[string]func(*config.RepositoryObjectFormat) objectSplitter{ + "NEVER": func(f *config.RepositoryObjectFormat) objectSplitter { + return newNeverSplitter() + }, + "FIXED": func(f *config.RepositoryObjectFormat) objectSplitter { + return newFixedSplitter(int(f.MaxBlockSize)) + }, + "DYNAMIC": func(f *config.RepositoryObjectFormat) objectSplitter { + return newRollingHashSplitter(buzhash32.New(), f.MinBlockSize, f.AvgBlockSize, f.MaxBlockSize) + }, +} + +func init() { + for k := range objectSplitterFactories { + SupportedObjectSplitters = append(SupportedObjectSplitters, k) + } + sort.Strings(SupportedObjectSplitters) +} + +// DefaultObjectSplitter is the name of the splitter used by default for new repositories. +const DefaultObjectSplitter = "DYNAMIC" + type neverSplitter struct{} func (s *neverSplitter) add(b byte) bool { @@ -45,9 +77,9 @@ type rollingHashSplitter struct { mask uint32 allOnes uint32 - currentBlockSize int32 - minBlockSize int32 - maxBlockSize int32 + currentBlockSize int + minBlockSize int + maxBlockSize int } func (rs *rollingHashSplitter) add(b byte) bool { @@ -67,27 +99,14 @@ func (rs *rollingHashSplitter) add(b byte) bool { return false } -func newRollingHashSplitter(rh rollinghash.Hash32, minBlockSize int32, approxBlockSize int32, maxBlockSize int32) objectSplitter { +func newRollingHashSplitter(rh rollinghash.Hash32, minBlockSize int, approxBlockSize int, maxBlockSize int) objectSplitter { bits := rollingHashBits(approxBlockSize) mask := ^(^uint32(0) << bits) return &rollingHashSplitter{rh, mask, (uint32(0)) ^ mask, 0, minBlockSize, maxBlockSize} } -func rollingHashBits(n int32) uint { +func rollingHashBits(n int) uint { e := math.Log2(float64(n)) exp := math.Floor(e + 0.5) return uint(exp) } - -//SupportedSplitters is a map of supported splitters their factory functions. -var SupportedSplitters = map[string]func(*Format) objectSplitter{ - "NEVER": func(f *Format) objectSplitter { - return newNeverSplitter() - }, - "FIXED": func(f *Format) objectSplitter { - return newFixedSplitter(int(f.MaxBlockSize)) - }, - "DYNAMIC": func(f *Format) objectSplitter { - return newRollingHashSplitter(buzhash32.New(), f.MinBlockSize, f.AvgBlockSize, f.MaxBlockSize) - }, -} diff --git a/repo/object_splitter_test.go b/repo/object_splitter_test.go index 441311b88..f3f134d5b 100644 --- a/repo/object_splitter_test.go +++ b/repo/object_splitter_test.go @@ -117,7 +117,7 @@ func TestSplitterStability(t *testing.T) { func TestRollingHashBits(t *testing.T) { cases := []struct { - blockSize int32 + blockSize int bits uint }{ {256, 8}, diff --git a/repo/object_writer.go b/repo/object_writer.go index c40159f9b..ef8547560 100644 --- a/repo/object_writer.go +++ b/repo/object_writer.go @@ -37,7 +37,7 @@ func (t *blockTracker) blockIDs() []string { } type objectWriter struct { - repo *casManager + repo *ObjectManager buffer *bytes.Buffer totalLength int64 @@ -172,19 +172,8 @@ func (w *objectWriter) StorageBlocks() []string { return w.blockTracker.blockIDs() } -// WriterOption is an option that can be passed to Repository.NewWriter() -type WriterOption func(*objectWriter) - -// WithBlockNamePrefix causes the ObjectWriter to prefix any blocks emitted to the storage with a given string. -func WithBlockNamePrefix(prefix string) WriterOption { - return func(w *objectWriter) { - w.prefix = prefix - } -} - -// WithDescription is used for debugging only and causes the following string to be emitted in logs. -func WithDescription(description string) WriterOption { - return func(w *objectWriter) { - w.description = description - } +// WriterOptions can be passed to Repository.NewWriter() +type WriterOptions struct { + BlockNamePrefix string + Description string } diff --git a/repo/repository.go b/repo/repository.go index bd090276a..2ed5fd453 100644 --- a/repo/repository.go +++ b/repo/repository.go @@ -6,14 +6,17 @@ import ( // Repository represents storage where both content-addressable and user-addressable data is kept. type Repository struct { - *casManager - *Vault + *ObjectManager + *MetadataManager Storage blob.Storage + + ConfigFile string + CacheDirectory string } // Close closes the repository and releases all resources. func (r *Repository) Close() error { - if err := r.casManager.Close(); err != nil { + if err := r.ObjectManager.Close(); err != nil { return err } if err := r.Storage.Close(); err != nil { @@ -24,18 +27,16 @@ func (r *Repository) Close() error { // Flush waits for all in-flight writes to complete. func (r *Repository) Flush() error { - r.casManager.writeBack.flush() + r.ObjectManager.writeBack.flush() return nil } -// NewRepository initializes a new repository with a given format. -func NewRepository(s blob.Storage, f *Format, options ...RepositoryOption) (*Repository, error) { - cm, err := newCASManager(s, f, options...) - if err != nil { - return nil, err - } - return &Repository{ - Storage: s, - casManager: cm, - }, nil +// Stats returns repository-wide statistics. +func (r *Repository) Stats() Stats { + return r.ObjectManager.stats +} + +// ResetStats resets all repository-wide statistics to zero values. +func (r *Repository) ResetStats() { + r.ObjectManager.stats = Stats{} } diff --git a/repo/vault.go b/repo/vault.go deleted file mode 100644 index 29794fc97..000000000 --- a/repo/vault.go +++ /dev/null @@ -1,381 +0,0 @@ -package repo - -import ( - "crypto/aes" - "crypto/cipher" - "crypto/rand" - "crypto/sha256" - "encoding/json" - "errors" - "fmt" - "io" - "strings" - "sync" - - "github.com/kopia/kopia/auth" - "github.com/kopia/kopia/blob" - - "golang.org/x/crypto/hkdf" -) - -const ( - formatBlockID = "format" - repositoryConfigBlockID = "repo" -) - -const ( - // VaultBlockPrefix is a prefix used for vault blocks in a repository storage. - VaultBlockPrefix = "VLT" -) - -var ( - purposeAESKey = []byte("AES") - purposeAuthData = []byte("CHECKSUM") -) - -// ErrItemNotFound is an error returned when a vault item cannot be found. -var ErrItemNotFound = errors.New("item not found") - -// SupportedEncryptionAlgorithms lists supported key derivation algorithms. -var SupportedEncryptionAlgorithms = []string{ - "AES256_GCM", - "NONE", -} - -// Vault is a secure storage for secrets such as repository object identifiers. -type Vault struct { - storage blob.Storage - format VaultFormat - RepoConfig RepositoryConfig - - masterKey []byte - - aead cipher.AEAD // authenticated encryption to use - authData []byte // additional data to authenticate -} - -// Put saves the specified content in a vault under a specified name. -func (v *Vault) Put(itemID string, content []byte) error { - if err := checkReservedName(itemID); err != nil { - return err - } - - return v.writeEncryptedBlock(itemID, content) -} - -func (v *Vault) writeEncryptedBlock(itemID string, content []byte) error { - if v.aead != nil { - nonceLength := v.aead.NonceSize() - noncePlusContentLength := nonceLength + len(content) - cipherText := make([]byte, noncePlusContentLength+v.aead.Overhead()) - - // Store nonce at the beginning of ciphertext. - nonce := cipherText[0:nonceLength] - if _, err := io.ReadFull(rand.Reader, nonce); err != nil { - return err - } - - b := v.aead.Seal(cipherText[nonceLength:nonceLength], nonce, content, v.authData) - - content = nonce[0 : nonceLength+len(b)] - } - - return v.storage.PutBlock(VaultBlockPrefix+itemID, content, blob.PutOptionsOverwrite) -} - -func (v *Vault) readEncryptedBlock(itemID string) ([]byte, error) { - content, err := v.storage.GetBlock(VaultBlockPrefix + itemID) - if err != nil { - if err == blob.ErrBlockNotFound { - return nil, ErrItemNotFound - } - return nil, fmt.Errorf("unexpected error reading %v: %v", itemID, err) - } - - return v.decryptBlock(content) -} - -func (v *Vault) decryptBlock(content []byte) ([]byte, error) { - if v.aead != nil { - nonce := content[0:v.aead.NonceSize()] - payload := content[v.aead.NonceSize():] - return v.aead.Open(payload[:0], nonce, payload, v.authData) - } - - return content, nil -} - -// DeriveKey computes a key for a specific purpose and length using HKDF based on the master key. -func (v *Vault) DeriveKey(purpose []byte, length int) []byte { - key := make([]byte, length) - k := hkdf.New(sha256.New, v.masterKey, v.format.UniqueID, purpose) - io.ReadFull(k, key) - return key -} - -// Get returns the contents of a specified vault item. -func (v *Vault) Get(itemID string) ([]byte, error) { - if err := checkReservedName(itemID); err != nil { - return nil, err - } - - return v.readEncryptedBlock(itemID) -} - -func (v *Vault) getJSON(itemID string, content interface{}) error { - j, err := v.readEncryptedBlock(itemID) - if err != nil { - return err - } - - return json.Unmarshal(j, content) -} - -// Put stores the contents of an item stored in a vault with a given ID. -func (v *Vault) putJSON(id string, content interface{}) error { - j, err := json.Marshal(content) - if err != nil { - return err - } - - return v.writeEncryptedBlock(id, j) -} - -// List returns the list of vault items matching the specified prefix. -// The 'limit' parameter specifies the maximum number of items to retrieve (-1 == unlimited). -func (v *Vault) List(prefix string, limit int) ([]string, error) { - var result []string - - ch, cancel := v.storage.ListBlocks(VaultBlockPrefix + prefix) - defer cancel() - for b := range ch { - if limit == 0 { - break - } - if b.Error != nil { - return result, b.Error - } - - itemID := strings.TrimPrefix(b.BlockID, VaultBlockPrefix) - if !isReservedName(itemID) { - result = append(result, itemID) - } - if limit > 0 { - limit-- - } - } - return result, nil -} - -// VaultConfig represents JSON-compatible configuration of the vault connection, including vault key. -type VaultConfig struct { - ConnectionInfo blob.ConnectionInfo `json:"connection"` - Key []byte `json:"key,omitempty"` -} - -// Config returns a configuration of vault storage its credentials that's suitable -// for storing in configuration file. -func (v *Vault) Config() (*VaultConfig, error) { - cip, ok := v.storage.(blob.ConnectionInfoProvider) - if !ok { - return nil, errors.New("repository does not support persisting configuration") - } - - ci := cip.ConnectionInfo() - - return &VaultConfig{ - ConnectionInfo: ci, - Key: v.masterKey, - }, nil -} - -// Remove deletes the specified vault item. -func (v *Vault) Remove(itemID string) error { - if err := checkReservedName(itemID); err != nil { - return err - } - - return v.storage.DeleteBlock(VaultBlockPrefix + itemID) -} - -// RemoveMany efficiently removes multiple vault items in parallel. -func (v *Vault) RemoveMany(itemIDs []string) error { - parallelism := 30 - ch := make(chan string) - var wg sync.WaitGroup - errch := make(chan error, len(itemIDs)) - - for i := 0; i < parallelism; i++ { - wg.Add(1) - go func() { - defer wg.Done() - - for id := range ch { - if err := v.Remove(id); err != nil { - errch <- err - } - } - }() - } - - for _, id := range itemIDs { - ch <- id - } - close(ch) - wg.Wait() - close(errch) - - return <-errch -} - -// Create initializes a Vault attached to the specified repository. -func Create( - vaultStorage blob.Storage, - vaultFormat *VaultFormat, vaultCreds auth.Credentials, - repoFormat *Format, -) (*Vault, error) { - v := Vault{ - storage: vaultStorage, - format: *vaultFormat, - } - - v.format.Version = "1" - v.format.UniqueID = make([]byte, 32) - if _, err := io.ReadFull(rand.Reader, v.format.UniqueID); err != nil { - return nil, err - } - if v.format.KeyDerivationAlgorithm == "" { - v.format.KeyDerivationAlgorithm = auth.DefaultKeyDerivationAlgorithm - - } - var err error - v.masterKey, err = vaultCreds.GetMasterKey(v.format.Options) - if err != nil { - return nil, err - } - - formatBytes, err := json.Marshal(&v.format) - if err != nil { - return nil, err - } - - if err := vaultStorage.PutBlock(VaultBlockPrefix+formatBlockID, formatBytes, blob.PutOptionsOverwrite); err != nil { - return nil, err - } - - if err := v.initCrypto(); err != nil { - return nil, fmt.Errorf("unable to initialize crypto: %v", err) - } - - // Write encrypted repository configuration block. - rc := RepositoryConfig{ - Format: repoFormat, - } - - if err := v.putJSON(repositoryConfigBlockID, &rc); err != nil { - return nil, err - } - - v.RepoConfig = rc - return &v, nil -} - -// RepositoryConfig stores the configuration of the repository associated with the vault. -type RepositoryConfig struct { - Format *Format `json:"format"` -} - -// Open opens a vault. -func Open(vaultStorage blob.Storage, vaultCreds auth.Credentials) (*Vault, error) { - v := Vault{ - storage: vaultStorage, - } - - var wg sync.WaitGroup - - var blocks [4][]byte - - f := func(index int, name string) { - blocks[index], _ = vaultStorage.GetBlock(name) - wg.Done() - } - - wg.Add(2) - go f(0, VaultBlockPrefix+formatBlockID) - go f(1, VaultBlockPrefix+repositoryConfigBlockID) - wg.Wait() - - if blocks[0] == nil { - return nil, fmt.Errorf("vault format block not found") - } - - var offset = 0 - err := json.Unmarshal(blocks[offset], &v.format) - if err != nil { - return nil, err - } - - v.masterKey, err = vaultCreds.GetMasterKey(v.format.Options) - if err != nil { - return nil, err - } - - if err := v.initCrypto(); err != nil { - return nil, fmt.Errorf("unable to initialize crypto: %v", err) - } - - cfgData, err := v.decryptBlock(blocks[offset+1]) - if err != nil { - return nil, err - } - - var rc RepositoryConfig - - if err := json.Unmarshal(cfgData, &rc); err != nil { - return nil, err - } - - v.RepoConfig = rc - - return &v, nil -} - -func (v *Vault) initCrypto() error { - switch v.format.EncryptionAlgorithm { - case "NONE": // do nothing - return nil - case "AES256_GCM": - aesKey := v.DeriveKey(purposeAESKey, 32) - v.authData = v.DeriveKey(purposeAuthData, 32) - - blk, err := aes.NewCipher(aesKey) - if err != nil { - return fmt.Errorf("cannot create cipher: %v", err) - } - v.aead, err = cipher.NewGCM(blk) - if err != nil { - return fmt.Errorf("cannot create cipher: %v", err) - } - return nil - default: - return fmt.Errorf("unknown encryption algorithm: '%v'", v.format.EncryptionAlgorithm) - } -} - -func isReservedName(itemID string) bool { - switch itemID { - case formatBlockID, repositoryConfigBlockID: - return true - - default: - return false - } -} - -func checkReservedName(itemID string) error { - if isReservedName(itemID) { - return fmt.Errorf("invalid vault item name: '%v'", itemID) - } - - return nil -} diff --git a/repo/vault_test.go b/repo/vault_test.go deleted file mode 100644 index 9dee4712c..000000000 --- a/repo/vault_test.go +++ /dev/null @@ -1,207 +0,0 @@ -package repo - -import ( - "context" - "io/ioutil" - "os" - "path/filepath" - "reflect" - "strings" - - "github.com/kopia/kopia/auth" - "github.com/kopia/kopia/blob" - "github.com/kopia/kopia/blob/filesystem" - - "testing" -) - -func TestNonColocatedVault(t *testing.T) { - tmpDir, err := ioutil.TempDir("", "vault") - if err != nil { - t.Errorf("can't create temp dir: %v", err) - return - } - defer os.RemoveAll(tmpDir) - - verifyVault( - t, - filepath.Join(tmpDir, "vlt"), - filepath.Join(tmpDir, "repo")) -} - -func TestColocatedVault(t *testing.T) { - tmpDir, err := ioutil.TempDir("", "vault") - if err != nil { - t.Errorf("can't create temp dir: %v", err) - return - } - //defer os.RemoveAll(tmpDir) - - // vault and repository in one storage - verifyVault(t, tmpDir, tmpDir) -} - -func verifyVault(t *testing.T, vaultPath string, repoPath string) { - vaultStorage := mustCreateFileStorage(t, vaultPath) - - vaultCreds, err := auth.Password("foo.bar.baz.foo.bar.baz") - if err != nil { - t.Errorf("can't create password credentials: %v", err) - return - } - - otherVaultCreds, err := auth.Password("foo.bar.baz.foo.bar.baz0") - if err != nil { - t.Errorf("can't create password credentials: %v", err) - return - } - - vaultFormat := &VaultFormat{ - EncryptionAlgorithm: "AES256_GCM", - } - - repoFormat := &Format{ - Version: 1, - MaxBlockSize: 1000000, - ObjectFormat: "UNENCRYPTED_HMAC_SHA256", - Secret: []byte{1, 2, 3}, - } - - v1, err := Create(vaultStorage, vaultFormat, vaultCreds, repoFormat) - if err != nil { - t.Errorf("can't create vault: %v", err) - return - } - - v2, err := Open(vaultStorage, vaultCreds) - if err != nil { - t.Errorf("can't open vault: %v", err) - return - } - - cfg, err := v1.Config() - if err != nil { - t.Errorf("error getting token1 %v", err) - } - - cfg2, err := v2.Config() - if err != nil { - t.Errorf("error getting token2 %v", err) - } - - if !reflect.DeepEqual(cfg, cfg2) { - t.Errorf("configurations are different: %+v vs %+v", cfg, cfg2) - } - - _, err = Open(vaultStorage, otherVaultCreds) - if err == nil { - t.Errorf("unexpectedly opened vault with invalid credentials") - return - } - - if err := v1.Put("foo", []byte("test1")); err != nil { - t.Errorf("error putting: %v", err) - } - if err := v2.Put("bar", []byte("test2")); err != nil { - t.Errorf("error putting: %v", err) - } - if err := v1.Put("baz", []byte("test3")); err != nil { - t.Errorf("error putting: %v", err) - } - - // Verify contents of vault items for both created and opened vault. - for _, v := range []*Vault{v1, v2} { - rf := v.RepoConfig.Format - if !reflect.DeepEqual(rf, repoFormat) { - t.Errorf("invalid repository format: %v, but got %v", repoFormat, rf) - } - assertVaultItem(t, v, "foo", "test1") - assertVaultItem(t, v, "bar", "test2") - assertVaultItem(t, v, "baz", "test3") - assertVaultItemNotFound(t, v, "x") - - assertVaultItems(t, v, "x", nil) - assertVaultItems(t, v, "f", []string{"foo"}) - assertVaultItems(t, v, "ba", []string{"bar", "baz"}) - assertVaultItems(t, v, "be", nil) - assertVaultItems(t, v, "baz", []string{"baz"}) - assertVaultItems(t, v, "bazx", nil) - - assertReservedName(t, v, formatBlockID) - assertReservedName(t, v, repositoryConfigBlockID) - } - - v1.Remove("bar") - - for _, v := range []*Vault{v1, v2} { - assertVaultItem(t, v, "foo", "test1") - assertVaultItemNotFound(t, v, "bar") - assertVaultItem(t, v, "baz", "test3") - - assertVaultItems(t, v, "x", nil) - assertVaultItems(t, v, "f", []string{"foo"}) - assertVaultItems(t, v, "ba", []string{"baz"}) - assertVaultItems(t, v, "be", nil) - assertVaultItems(t, v, "baz", []string{"baz"}) - assertVaultItems(t, v, "bazx", nil) - } -} - -func assertVaultItem(t *testing.T, v *Vault, itemID string, expectedData string) { - b, err := v.Get(itemID) - if err != nil { - t.Errorf("error getting item %v: %v", itemID, err) - } - - bs := string(b) - if bs != expectedData { - t.Errorf("invalid data for '%v': expected: %v but got %v", itemID, expectedData, bs) - } -} - -func assertVaultItemNotFound(t *testing.T, v *Vault, itemID string) { - result, err := v.Get(itemID) - if err != ErrItemNotFound { - t.Errorf("invalid error getting item %v: %v (result=%v), but expected %v", itemID, err, result, ErrItemNotFound) - } -} - -func assertReservedName(t *testing.T, v *Vault, itemID string) { - _, err := v.Get(itemID) - assertReservedNameError(t, "Get", itemID, err) - assertReservedNameError(t, "Put", itemID, v.Put(itemID, nil)) - assertReservedNameError(t, "Remove", itemID, v.Remove(itemID)) -} - -func assertReservedNameError(t *testing.T, method string, itemID string, err error) { - if err == nil { - t.Errorf("call did not fail: %v(%v)", method, itemID) - return - } - - if !strings.Contains(err.Error(), "invalid vault item name") { - t.Errorf("call did not fail the right way: %v(%v), was: %v", method, itemID, err) - } -} - -func assertVaultItems(t *testing.T, v *Vault, prefix string, expected []string) { - res, err := v.List(prefix, -1) - if err != nil { - t.Errorf("error listing items beginning with %v: %v", prefix, err) - } - - if !reflect.DeepEqual(expected, res) { - t.Errorf("unexpected items returned for prefix '%v': %v, but expected %v", prefix, res, expected) - } -} - -func mustCreateFileStorage(t *testing.T, path string) blob.Storage { - os.MkdirAll(path, 0700) - s, err := filesystem.New(context.Background(), &filesystem.Options{ - Path: path, - }) - if err != nil { - t.Errorf("can't create file storage: %v", err) - } - return s -} diff --git a/snapshot/manager.go b/snapshot/manager.go index 5b3b3fad9..816bc9fdb 100644 --- a/snapshot/manager.go +++ b/snapshot/manager.go @@ -33,7 +33,7 @@ type Manager struct { // ListSources lists all snapshot sources. func (m *Manager) ListSources() ([]*SourceInfo, error) { - names, err := m.repository.List(snapshotPrefix, -1) + names, err := m.repository.ListMetadata(snapshotPrefix, -1) if err != nil { return nil, err } @@ -64,7 +64,7 @@ func (m *Manager) ListSources() ([]*SourceInfo, error) { // ListSnapshots lists all snapshots for a given source. func (m *Manager) ListSnapshots(si *SourceInfo, limit int) ([]*Manifest, error) { - names, err := m.repository.List(m.snapshotIDPrefix(si), limit) + names, err := m.repository.ListMetadata(m.snapshotIDPrefix(si), limit) if err != nil { return nil, err } @@ -74,7 +74,7 @@ func (m *Manager) ListSnapshots(si *SourceInfo, limit int) ([]*Manifest, error) // LoadSnapshot loads and parses a snapshot with a given ID. func (m *Manager) LoadSnapshot(manifestID string) (*Manifest, error) { - b, err := m.repository.Get(manifestID) + b, err := m.repository.GetMetadata(manifestID) if err != nil { return nil, fmt.Errorf("error loading previous backup: %v", err) } @@ -99,7 +99,7 @@ func (m *Manager) SaveSnapshot(manifest *Manifest) (string, error) { return "", fmt.Errorf("cannot marshal backup manifest to JSON: %v", err) } - if err := m.repository.Put(manifestID, b); err != nil { + if err := m.repository.PutMetadata(manifestID, b); err != nil { return "", err } @@ -148,7 +148,7 @@ func (m *Manager) ListSnapshotManifests(src *SourceInfo, limit int) ([]string, e prefix = m.snapshotIDPrefix(src) } - return m.repository.List(prefix, limit) + return m.repository.ListMetadata(prefix, limit) } // GetEffectivePolicy calculates effective snapshot policy for a given source by combining the source-specifc policy (if any) @@ -229,12 +229,12 @@ func (m *Manager) SavePolicy(p *Policy) error { return fmt.Errorf("cannot marshal policy to JSON: %v", err) } - return m.repository.Put(m.policyID(&p.Source), b) + return m.repository.PutMetadata(m.policyID(&p.Source), b) } // RemovePolicy removes the policy for a given source func (m *Manager) RemovePolicy(src *SourceInfo) error { - return m.repository.Remove(m.policyID(src)) + return m.repository.RemoveMetadata(m.policyID(src)) } // GetPolicy retrieves the Policy for a given source, if defined. @@ -252,8 +252,8 @@ func (m *Manager) policyID(src *SourceInfo) string { } func (m *Manager) getPolicyItem(itemID string) (*Policy, error) { - b, err := m.repository.Get(itemID) - if err == repo.ErrItemNotFound { + b, err := m.repository.GetMetadata(itemID) + if err == repo.ErrMetadataNotFound { return nil, ErrPolicyNotFound } @@ -269,9 +269,9 @@ func (m *Manager) getPolicyItem(itemID string) (*Policy, error) { return &s, nil } -// ListPolicies returns a list of all policies stored in a vault. +// ListPolicies returns a list of all snapshot policies. func (m *Manager) ListPolicies() ([]*Policy, error) { - names, err := m.repository.List(policyPrefix, -1) + names, err := m.repository.ListMetadata(policyPrefix, -1) if err != nil { return nil, err } diff --git a/snapshot/stats.go b/snapshot/stats.go index b497cde10..fb9db2935 100644 --- a/snapshot/stats.go +++ b/snapshot/stats.go @@ -4,7 +4,7 @@ import "github.com/kopia/kopia/repo" // Stats keeps track of snapshot generation statistics. type Stats struct { - Repository *repo.Stats `json:"repo,omitempty"` + Repository repo.Stats `json:"repo,omitempty"` TotalDirectoryCount int `json:"dirCount"` TotalFileCount int `json:"fileCount"` diff --git a/snapshot/upload.go b/snapshot/upload.go index d068b699c..1102d0343 100644 --- a/snapshot/upload.go +++ b/snapshot/upload.go @@ -71,9 +71,9 @@ func uploadFileInternal(u *uploadContext, f fs.File, relativePath string, forceS } defer file.Close() - writer := u.repo.NewWriter( - repo.WithDescription("FILE:" + f.Metadata().Name), - ) + writer := u.repo.NewWriter(repo.WriterOptions{ + Description: "FILE:" + f.Metadata().Name, + }) defer writer.Close() written, err := copyWithProgress(u, relativePath, writer, file, 0, f.Metadata().FileSize) @@ -152,9 +152,9 @@ func uploadBundleInternal(u *uploadContext, b *dir.Bundle, relativePath string) bundleMetadata := b.Metadata() u.progress.Started(relativePath, b.Metadata().FileSize) - writer := u.repo.NewWriter( - repo.WithDescription("BUNDLE:" + bundleMetadata.Name), - ) + writer := u.repo.NewWriter(repo.WriterOptions{ + Description: "BUNDLE:" + bundleMetadata.Name, + }) defer writer.Close() @@ -218,10 +218,10 @@ func uploadFile(u *uploadContext, file fs.File) (repo.ObjectID, error) { func uploadDir(u *uploadContext, dir fs.Directory) (repo.ObjectID, repo.ObjectID, error) { var err error - mw := u.repo.NewWriter( - repo.WithDescription("HASHCACHE:"+dir.Metadata().Name), - repo.WithBlockNamePrefix("H"), - ) + mw := u.repo.NewWriter(repo.WriterOptions{ + Description: "HASHCACHE:" + dir.Metadata().Name, + BlockNamePrefix: "H", + }) defer mw.Close() u.cacheWriter = hashcache.NewWriter(mw) oid, _, _, err := uploadDirInternal(u, dir, ".", true) @@ -254,9 +254,9 @@ func uploadDirInternal( entries = bundleEntries(u, entries) - writer := u.repo.NewWriter( - repo.WithDescription("DIR:" + relativePath), - ) + writer := u.repo.NewWriter(repo.WriterOptions{ + Description: "DIR:" + relativePath, + }) dw := dir.NewWriter(writer) defer writer.Close() @@ -570,7 +570,7 @@ func Upload( s.EndTime = time.Now() s.Stats = *u.stats - s.Stats.Repository = &u.repo.Stats + s.Stats.Repository = u.repo.Stats() return s, nil } diff --git a/snapshot/upload_test.go b/snapshot/upload_test.go index c99b40dfa..27265fc7c 100644 --- a/snapshot/upload_test.go +++ b/snapshot/upload_test.go @@ -5,6 +5,7 @@ import ( "fmt" "io/ioutil" "os" + "path/filepath" "reflect" "github.com/kopia/kopia/blob" @@ -13,6 +14,8 @@ import ( "github.com/kopia/kopia/repo" "testing" + + "github.com/kopia/kopia/auth" ) type uploadTestHarness struct { @@ -31,6 +34,7 @@ func (th *uploadTestHarness) cleanup() { } func newUploadTestHarness() *uploadTestHarness { + ctx := context.Background() repoDir, err := ioutil.TempDir("", "kopia-repo") if err != nil { panic("cannot create temp directory: " + err.Error()) @@ -44,18 +48,29 @@ func newUploadTestHarness() *uploadTestHarness { panic("cannot create storage directory: " + err.Error()) } - format := repo.Format{ - Version: 1, - ObjectFormat: "TESTONLY_MD5", - MaxBlockSize: 1000, - MaxInlineContentLength: 0, + creds, err := auth.Password("foofoofoofoofoofoofoofoo") + if err != nil { + panic("unable to create credentials: " + err.Error()) } - repo, err := repo.NewRepository(storage, &format) - if err != nil { + if err := repo.Initialize(storage, &repo.NewRepositoryOptions{ + MaxInlineContentLength: -1, + }, creds); err != nil { panic("unable to create repository: " + err.Error()) } + configFile := filepath.Join(repoDir, ".kopia.config") + if err := repo.Connect(ctx, configFile, storage, creds, repo.ConnectOptions{ + PersistCredentials: true, + }); err != nil { + panic("unable to connect to repository: " + err.Error()) + } + + repo, err := repo.Open(ctx, configFile, nil) + if err != nil { + panic("unable to open repository: " + err.Error()) + } + sourceDir := mockfs.NewDirectory() sourceDir.AddFile("f1", []byte{1, 2, 3}, 0777) sourceDir.AddFile("f2", []byte{1, 2, 3, 4}, 0777)