Add CLI for local storage (#4243)

<!-- PR description-->

New commands to initialize & connect to a repo on local or network attached storage.
* `repo init filesystem --path /tmp/repo`
* `repo connect filesystem --path /tmp/repo`

Includes basic unit & e2e tests. More coverage to be added in a following PR to keep the size contained.

**Updates:**
* Added Repo path sanitization i.e. handle relative paths, make paths cross platform compatible, etc. 
* Removed retention artifacts, not supported for filesystem storage.
* cli docs - auto updated.
* Manually tested with all corso backup/restore/export commands.

**Doesn't include** 
1. Symlinks
2. User ids wiring into repo.
3. Repos documentation update - in an upcoming PR.
4. Prefix support -> kopia doesn't support prefixes for `filesystem` storage
5. More E2E tests. 

---

#### Does this PR need a docs update or release note?

- [ ] ✅ Yes, it's included
- [x] 🕐 Yes, but in a later PR
- [ ] ⛔ No

#### Type of change

<!--- Please check the type of change your PR introduces: --->
- [x] 🌻 Feature
- [ ] 🐛 Bugfix
- [ ] 🗺️ Documentation
- [ ] 🤖 Supportability/Tests
- [ ] 💻 CI/Deployment
- [ ] 🧹 Tech Debt/Cleanup

#### Issue(s)

<!-- Can reference multiple issues. Use one of the following "magic words" - "closes, fixes" to auto-close the Github issue. -->
* #1416 

#### Test Plan

<!-- How will this be tested prior to merging.-->
- [x] 💪 Manual
- [x] ⚡ Unit test
- [x] 💚 E2E

src/cli/cli.go

@@ -65,6 +65,8 @@ func preRun(cc *cobra.Command, args []string) error {
 		"Initialize a repository.",
 		"Initialize a S3 repository",
 		"Connect to a S3 repository",
+		"Initialize a repository on local or network storage.",
+		"Connect to a repository on local or network storage.",
 		"Help about any command",
 		"Free, Secure, Open-Source Backup for M365.",
 		"env var guide",

src/cli/config/config.go

@@ -297,7 +297,12 @@ func getStorageAndAccountWithViper(
 		return config, clues.Wrap(err, "retrieving account configuration details")
 	}
 
-	config.Storage, err = configureStorage(vpr, provider, readConfigFromViper, mustMatchFromConfig, overrides)
+	config.Storage, err = configureStorage(
+		vpr,
+		provider,
+		readConfigFromViper,
+		mustMatchFromConfig,
+		overrides)
 	if err != nil {
 		return config, clues.Wrap(err, "retrieving storage provider details")
 	}

src/cli/config/storage.go

@@ -95,9 +95,6 @@ func GetStorageProviderFromConfigFile(ctx context.Context) (storage.ProviderType
 	}
 
 	provider := vpr.GetString(storage.StorageProviderTypeKey)
-	if provider != storage.ProviderS3.String() {
-		return storage.ProviderUnknown, clues.New("unsupported storage provider: " + provider)
-	}
 
 	return storage.StringToProviderType[provider], nil
 }

src/cli/flags/filesystem.go

@@ -0,0 +1,55 @@
+package flags
+
+import (
+	"github.com/spf13/cobra"
+
+	"github.com/alcionai/corso/src/pkg/storage"
+)
+
+// filesystem flag names
+const (
+	FilesystemPathFN = "path"
+)
+
+// filesystem flag values
+var (
+	FilesystemPathFV string
+)
+
+func AddFilesystemFlags(cmd *cobra.Command) {
+	fs := cmd.Flags()
+
+	AddAzureCredsFlags(cmd)
+	AddCorsoPassphaseFlags(cmd)
+
+	fs.StringVar(
+		&FilesystemPathFV,
+		FilesystemPathFN,
+		"",
+		"path to local or network storage")
+	cobra.CheckErr(cmd.MarkFlagRequired(FilesystemPathFN))
+
+	fs.BoolVar(
+		&SucceedIfExistsFV,
+		SucceedIfExistsFN,
+		false,
+		"Exit with success if the repo has already been initialized.")
+	cobra.CheckErr(fs.MarkHidden("succeed-if-exists"))
+}
+
+func FilesystemFlagOverrides(cmd *cobra.Command) map[string]string {
+	fs := GetPopulatedFlags(cmd)
+	return PopulateFilesystemFlags(fs)
+}
+
+func PopulateFilesystemFlags(flagset PopulatedFlags) map[string]string {
+	fsOverrides := map[string]string{
+		storage.StorageProviderTypeKey: storage.ProviderFilesystem.String(),
+	}
+
+	if _, ok := flagset[FilesystemPathFN]; ok {
+		fsOverrides[FilesystemPathFN] = FilesystemPathFV
+	}
+
+	return fsOverrides
+}

src/cli/flags/repo.go

@@ -12,6 +12,7 @@ const (
 
 	// Corso Flags
 	CorsoPassphraseFN = "passphrase"
+	SucceedIfExistsFN = "succeed-if-exists"
 )
 
 var (
@@ -20,6 +21,7 @@ var (
 	AWSSecretAccessKeyFV string
 	AWSSessionTokenFV    string
 	CorsoPassphraseFV    string
+	SucceedIfExistsFV    bool
 )
 
 // AddBackupIDFlag adds the --backup flag.

src/cli/flags/s3.go

@@ -11,22 +11,20 @@ import (
 
 // S3 bucket flags
 const (
-	BucketFN          = "bucket"
-	EndpointFN        = "endpoint"
-	PrefixFN          = "prefix"
-	DoNotUseTLSFN     = "disable-tls"
-	DoNotVerifyTLSFN  = "disable-tls-verification"
-	SucceedIfExistsFN = "succeed-if-exists"
+	BucketFN         = "bucket"
+	EndpointFN       = "endpoint"
+	PrefixFN         = "prefix"
+	DoNotUseTLSFN    = "disable-tls"
+	DoNotVerifyTLSFN = "disable-tls-verification"
 )
 
 // S3 bucket flag values
 var (
-	BucketFV          string
-	EndpointFV        string
-	PrefixFV          string
-	DoNotUseTLSFV     bool
-	DoNotVerifyTLSFV  bool
-	SucceedIfExistsFV bool
+	BucketFV         string
+	EndpointFV       string
+	PrefixFV         string
+	DoNotUseTLSFV    bool
+	DoNotVerifyTLSFV bool
 )
 
 // S3 bucket flags

src/cli/repo/filesystem.go

@@ -0,0 +1,213 @@
+package repo
+
+import (
+	"github.com/alcionai/clues"
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+
+	"github.com/alcionai/corso/src/cli/config"
+	"github.com/alcionai/corso/src/cli/flags"
+	. "github.com/alcionai/corso/src/cli/print"
+	"github.com/alcionai/corso/src/cli/utils"
+	"github.com/alcionai/corso/src/internal/events"
+	ctrlRepo "github.com/alcionai/corso/src/pkg/control/repository"
+	"github.com/alcionai/corso/src/pkg/repository"
+	"github.com/alcionai/corso/src/pkg/storage"
+)
+
+const (
+	fsProviderCommand      = "filesystem"
+	fsProviderCmdUseSuffix = "--path <path>"
+)
+
+const (
+	fsProviderCmdInitExamples = `# Create a new Corso repository on local or network attached storage
+corso repo init filesystem --path /tmp/corso-repo`
+
+	fsProviderCmdConnectExamples = `# Connect to a Corso repository on local or network attached storage
+corso repo connect filesystem --path /tmp/corso-repo`
+)
+
+func addFilesystemCommands(cmd *cobra.Command) *cobra.Command {
+	var c *cobra.Command
+
+	switch cmd.Use {
+	case initCommand:
+		init := filesystemInitCmd()
+		c, _ = utils.AddCommand(cmd, init)
+
+	case connectCommand:
+		c, _ = utils.AddCommand(cmd, filesystemConnectCmd())
+	}
+
+	c.Use = c.Use + " " + fsProviderCmdUseSuffix
+	c.SetUsageTemplate(cmd.UsageTemplate())
+
+	flags.AddFilesystemFlags(c)
+
+	return c
+}
+
+// `corso repo init filesystem [<flag>...]`
+func filesystemInitCmd() *cobra.Command {
+	return &cobra.Command{
+		Use:     fsProviderCommand,
+		Short:   "Initialize a repository on local or network storage.",
+		Long:    `Bootstraps a new repository on local or network storage and connects it to your m365 account.`,
+		RunE:    initFilesystemCmd,
+		Args:    cobra.NoArgs,
+		Example: fsProviderCmdInitExamples,
+	}
+}
+
+// initializes a filesystem repo.
+func initFilesystemCmd(cmd *cobra.Command, args []string) error {
+	ctx := cmd.Context()
+	overrides := flags.FilesystemFlagOverrides(cmd)
+
+	// TODO(pandeyabs): Move filepath conversion to FilesystemConfig scope.
+	abs, err := utils.MakeAbsoluteFilePath(overrides[flags.FilesystemPathFN])
+	if err != nil {
+		return Only(ctx, clues.Wrap(err, "getting absolute repo path"))
+	}
+
+	overrides[flags.FilesystemPathFN] = abs
+
+	cfg, err := config.GetConfigRepoDetails(
+		ctx,
+		storage.ProviderFilesystem,
+		true,
+		false,
+		flags.FilesystemFlagOverrides(cmd))
+	if err != nil {
+		return Only(ctx, err)
+	}
+
+	opt := utils.ControlWithConfig(cfg)
+	// Retention is not supported for filesystem repos.
+	retention := ctrlRepo.Retention{}
+
+	// SendStartCorsoEvent uses distict ID as tenant ID because repoID is still not generated
+	utils.SendStartCorsoEvent(
+		ctx,
+		cfg.Storage,
+		cfg.Account.ID(),
+		map[string]any{"command": "init repo"},
+		cfg.Account.ID(),
+		opt)
+
+	sc, err := cfg.Storage.StorageConfig()
+	if err != nil {
+		return Only(ctx, clues.Wrap(err, "Retrieving filesystem configuration"))
+	}
+
+	storageCfg := sc.(*storage.FilesystemConfig)
+
+	m365, err := cfg.Account.M365Config()
+	if err != nil {
+		return Only(ctx, clues.Wrap(err, "Failed to parse m365 account config"))
+	}
+
+	r, err := repository.Initialize(
+		ctx,
+		cfg.Account,
+		cfg.Storage,
+		opt,
+		retention)
+	if err != nil {
+		if flags.SucceedIfExistsFV && errors.Is(err, repository.ErrorRepoAlreadyExists) {
+			return nil
+		}
+
+		return Only(ctx, clues.Wrap(err, "Failed to initialize a new filesystem repository"))
+	}
+
+	defer utils.CloseRepo(ctx, r)
+
+	Infof(ctx, "Initialized a repository at path %s", storageCfg.Path)
+
+	if err = config.WriteRepoConfig(ctx, sc, m365, opt.Repo, r.GetID()); err != nil {
+		return Only(ctx, clues.Wrap(err, "Failed to write repository configuration"))
+	}
+
+	return nil
+}
+
+// ---------------------------------------------------------------------------------------------------------
+// Connect
+// ---------------------------------------------------------------------------------------------------------
+
+// `corso repo connect filesystem [<flag>...]`
+func filesystemConnectCmd() *cobra.Command {
+	return &cobra.Command{
+		Use:     fsProviderCommand,
+		Short:   "Connect to a repository on local or network storage.",
+		Long:    `Ensures a connection to an existing repository on local or network storage.`,
+		RunE:    connectFilesystemCmd,
+		Args:    cobra.NoArgs,
+		Example: fsProviderCmdConnectExamples,
+	}
+}
+
+// connects to an existing filesystem repo.
+func connectFilesystemCmd(cmd *cobra.Command, args []string) error {
+	ctx := cmd.Context()
+	overrides := flags.FilesystemFlagOverrides(cmd)
+
+	// TODO(pandeyabs): Move filepath conversion to FilesystemConfig scope.
+	abs, err := utils.MakeAbsoluteFilePath(overrides[flags.FilesystemPathFN])
+	if err != nil {
+		return Only(ctx, clues.Wrap(err, "getting absolute repo path"))
+	}
+
+	overrides[flags.FilesystemPathFN] = abs
+
+	cfg, err := config.GetConfigRepoDetails(
+		ctx,
+		storage.ProviderFilesystem,
+		true,
+		true,
+		overrides)
+	if err != nil {
+		return Only(ctx, err)
+	}
+
+	repoID := cfg.RepoID
+	if len(repoID) == 0 {
+		repoID = events.RepoIDNotFound
+	}
+
+	sc, err := cfg.Storage.StorageConfig()
+	if err != nil {
+		return Only(ctx, clues.Wrap(err, "Retrieving filesystem configuration"))
+	}
+
+	storageCfg := sc.(*storage.FilesystemConfig)
+
+	m365, err := cfg.Account.M365Config()
+	if err != nil {
+		return Only(ctx, clues.Wrap(err, "Failed to parse m365 account config"))
+	}
+
+	opts := utils.ControlWithConfig(cfg)
+
+	r, err := repository.ConnectAndSendConnectEvent(
+		ctx,
+		cfg.Account,
+		cfg.Storage,
+		repoID,
+		opts)
+	if err != nil {
+		return Only(ctx, clues.Wrap(err, "Failed to connect to the filesystem repository"))
+	}
+
+	defer utils.CloseRepo(ctx, r)
+
+	Infof(ctx, "Connected to repository at path %s", storageCfg.Path)
+
+	if err = config.WriteRepoConfig(ctx, sc, m365, opts.Repo, r.GetID()); err != nil {
+		return Only(ctx, clues.Wrap(err, "Failed to write repository configuration"))
+	}
+
+	return nil
+}

src/cli/repo/filesystem_e2e_test.go

@@ -0,0 +1,155 @@
+package repo_test
+
+import (
+	"os"
+	"testing"
+
+	"github.com/alcionai/clues"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+
+	"github.com/alcionai/corso/src/cli"
+	"github.com/alcionai/corso/src/cli/config"
+	cliTD "github.com/alcionai/corso/src/cli/testdata"
+	"github.com/alcionai/corso/src/internal/tester"
+	"github.com/alcionai/corso/src/internal/tester/tconfig"
+	"github.com/alcionai/corso/src/pkg/account"
+	"github.com/alcionai/corso/src/pkg/control"
+	ctrlRepo "github.com/alcionai/corso/src/pkg/control/repository"
+	"github.com/alcionai/corso/src/pkg/repository"
+	"github.com/alcionai/corso/src/pkg/storage"
+	storeTD "github.com/alcionai/corso/src/pkg/storage/testdata"
+)
+
+type FilesystemE2ESuite struct {
+	tester.Suite
+}
+
+func TestFilesystemE2ESuite(t *testing.T) {
+	suite.Run(t, &FilesystemE2ESuite{Suite: tester.NewE2ESuite(
+		t,
+		[][]string{storeTD.AWSStorageCredEnvs, tconfig.M365AcctCredEnvs})})
+}
+
+func (suite *FilesystemE2ESuite) TestInitFilesystemCmd() {
+	table := []struct {
+		name          string
+		hasConfigFile bool
+	}{
+		{
+			name:          "NoConfigFile",
+			hasConfigFile: false,
+		},
+		{
+			name:          "hasConfigFile",
+			hasConfigFile: true,
+		},
+	}
+
+	for _, test := range table {
+		suite.Run(test.name, func() {
+			t := suite.T()
+
+			ctx, flush := tester.NewContext(t)
+			defer flush()
+
+			st := storeTD.NewFilesystemStorage(t)
+
+			sc, err := st.StorageConfig()
+			require.NoError(t, err, clues.ToCore(err))
+			cfg := sc.(*storage.FilesystemConfig)
+
+			force := map[string]string{
+				tconfig.TestCfgStorageProvider: storage.ProviderFilesystem.String(),
+			}
+
+			vpr, configFP := tconfig.MakeTempTestConfigClone(t, force)
+			if !test.hasConfigFile {
+				// Ideally we could use `/dev/null`, but you need a
+				// toml file plus this works cross platform
+				os.Remove(configFP)
+			}
+
+			ctx = config.SetViper(ctx, vpr)
+
+			cmd := cliTD.StubRootCmd(
+				"repo", "init", "filesystem",
+				"--config-file", configFP,
+				"--path", cfg.Path)
+			cli.BuildCommandTree(cmd)
+
+			// run the command
+			err = cmd.ExecuteContext(ctx)
+			require.NoError(t, err, clues.ToCore(err))
+
+			// a second initialization should result in an error
+			err = cmd.ExecuteContext(ctx)
+			assert.ErrorIs(t, err, repository.ErrorRepoAlreadyExists, clues.ToCore(err))
+		})
+	}
+}
+
+func (suite *FilesystemE2ESuite) TestConnectFilesystemCmd() {
+	table := []struct {
+		name          string
+		hasConfigFile bool
+	}{
+		{
+			name:          "NoConfigFile",
+			hasConfigFile: false,
+		},
+		{
+			name:          "HasConfigFile",
+			hasConfigFile: true,
+		},
+	}
+
+	for _, test := range table {
+		suite.Run(test.name, func() {
+			t := suite.T()
+
+			ctx, flush := tester.NewContext(t)
+			defer flush()
+
+			st := storeTD.NewFilesystemStorage(t)
+			sc, err := st.StorageConfig()
+			require.NoError(t, err, clues.ToCore(err))
+			cfg := sc.(*storage.FilesystemConfig)
+
+			force := map[string]string{
+				tconfig.TestCfgAccountProvider: account.ProviderM365.String(),
+				tconfig.TestCfgStorageProvider: storage.ProviderFilesystem.String(),
+				tconfig.TestCfgFilesystemPath:  cfg.Path,
+			}
+			vpr, configFP := tconfig.MakeTempTestConfigClone(t, force)
+			if !test.hasConfigFile {
+				// Ideally we could use `/dev/null`, but you need a
+				// toml file plus this works cross platform
+				os.Remove(configFP)
+			}
+
+			ctx = config.SetViper(ctx, vpr)
+
+			// init the repo first
+			_, err = repository.Initialize(
+				ctx,
+				account.Account{},
+				st,
+				control.DefaultOptions(),
+				ctrlRepo.Retention{})
+			require.NoError(t, err, clues.ToCore(err))
+
+			// then test it
+			cmd := cliTD.StubRootCmd(
+				"repo", "connect", "filesystem",
+				"--config-file", configFP,
+				"--path", cfg.Path)
+			cli.BuildCommandTree(cmd)
+
+			// run the command
+			err = cmd.ExecuteContext(ctx)
+			require.NoError(t, err, clues.ToCore(err))
+		})
+	}
+}

src/cli/repo/filesystem_test.go

@@ -0,0 +1,53 @@
+package repo
+
+import (
+	"testing"
+
+	"github.com/spf13/cobra"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+
+	"github.com/alcionai/corso/src/internal/tester"
+)
+
+type FilesystemSuite struct {
+	tester.Suite
+}
+
+func TestFilesystemSuite(t *testing.T) {
+	suite.Run(t, &FilesystemSuite{Suite: tester.NewUnitSuite(t)})
+}
+
+func (suite *FilesystemSuite) TestAddFilesystemCommands() {
+	expectUse := fsProviderCommand + " " + fsProviderCmdUseSuffix
+
+	table := []struct {
+		name        string
+		use         string
+		expectUse   string
+		expectShort string
+		expectRunE  func(*cobra.Command, []string) error
+	}{
+		{"init filesystem", initCommand, expectUse, filesystemInitCmd().Short, initFilesystemCmd},
+		{"connect filesystem", connectCommand, expectUse, filesystemConnectCmd().Short, connectFilesystemCmd},
+	}
+	for _, test := range table {
+		suite.Run(test.name, func() {
+			t := suite.T()
+
+			cmd := &cobra.Command{Use: test.use}
+
+			c := addFilesystemCommands(cmd)
+			require.NotNil(t, c)
+
+			cmds := cmd.Commands()
+			require.Len(t, cmds, 1)
+
+			child := cmds[0]
+			assert.Equal(t, test.expectUse, child.Use)
+			assert.Equal(t, test.expectShort, child.Short)
+			tester.AreSameFunc(t, test.expectRunE, child.RunE)
+		})
+	}
+}

src/cli/repo/repo.go

@@ -22,6 +22,7 @@ const (
 
 var repoCommands = []func(cmd *cobra.Command) *cobra.Command{
 	addS3Commands,
+	addFilesystemCommands,
 }
 
 // AddCommands attaches all `corso repo * *` commands to the parent.

src/cli/utils/flags_test.go

@@ -101,9 +101,9 @@ func (suite *FlagUnitSuite) TestAddS3BucketFlags() {
 			assert.Equal(t, "bucket1", flags.BucketFV, flags.BucketFN)
 			assert.Equal(t, "endpoint1", flags.EndpointFV, flags.EndpointFN)
 			assert.Equal(t, "prefix1", flags.PrefixFV, flags.PrefixFN)
-			assert.Equal(t, true, flags.DoNotUseTLSFV, flags.DoNotUseTLSFN)
-			assert.Equal(t, true, flags.DoNotVerifyTLSFV, flags.DoNotVerifyTLSFN)
-			assert.Equal(t, true, flags.SucceedIfExistsFV, flags.SucceedIfExistsFN)
+			assert.True(t, flags.DoNotUseTLSFV, flags.DoNotUseTLSFN)
+			assert.True(t, flags.DoNotVerifyTLSFV, flags.DoNotVerifyTLSFN)
+			assert.True(t, flags.SucceedIfExistsFV, flags.SucceedIfExistsFN)
 		},
 	}
 
@@ -122,3 +122,34 @@ func (suite *FlagUnitSuite) TestAddS3BucketFlags() {
 	err := cmd.Execute()
 	require.NoError(t, err, clues.ToCore(err))
 }
+
+func (suite *FlagUnitSuite) TestFilesystemFlags() {
+	t := suite.T()
+
+	cmd := &cobra.Command{
+		Use: "test",
+		Run: func(cmd *cobra.Command, args []string) {
+			assert.Equal(t, "/tmp/test", flags.FilesystemPathFV, flags.FilesystemPathFN)
+			assert.True(t, flags.SucceedIfExistsFV, flags.SucceedIfExistsFN)
+			assert.Equal(t, "tenantID", flags.AzureClientTenantFV, flags.AzureClientTenantFN)
+			assert.Equal(t, "clientID", flags.AzureClientIDFV, flags.AzureClientIDFN)
+			assert.Equal(t, "secret", flags.AzureClientSecretFV, flags.AzureClientSecretFN)
+			assert.Equal(t, "passphrase", flags.CorsoPassphraseFV, flags.CorsoPassphraseFN)
+		},
+	}
+
+	flags.AddFilesystemFlags(cmd)
+
+	cmd.SetArgs([]string{
+		"test",
+		"--" + flags.FilesystemPathFN, "/tmp/test",
+		"--" + flags.SucceedIfExistsFN,
+		"--" + flags.AzureClientIDFN, "clientID",
+		"--" + flags.AzureClientTenantFN, "tenantID",
+		"--" + flags.AzureClientSecretFN, "secret",
+		"--" + flags.CorsoPassphraseFN, "passphrase",
+	})
+
+	err := cmd.Execute()
+	require.NoError(t, err, clues.ToCore(err))
+}

src/cli/utils/utils.go

@@ -3,6 +3,8 @@ package utils
 import (
 	"context"
 	"fmt"
+	"os"
+	"path/filepath"
 
 	"github.com/alcionai/clues"
 	"github.com/spf13/cobra"
@@ -95,8 +97,6 @@ func AccountConnectAndWriteRepoConfig(
 		return nil, nil, err
 	}
 
-	s3Config := sc.(*storage.S3Config)
-
 	m365Config, err := acc.M365Config()
 	if err != nil {
 		logger.CtxErr(ctx, err).Info("getting m365 configuration")
@@ -105,7 +105,7 @@ func AccountConnectAndWriteRepoConfig(
 
 	// repo config gets set during repo connect and init.
 	// This call confirms we have the correct values.
-	err = config.WriteRepoConfig(ctx, s3Config, m365Config, opts.Repo, r.GetID())
+	err = config.WriteRepoConfig(ctx, sc, m365Config, opts.Repo, r.GetID())
 	if err != nil {
 		logger.CtxErr(ctx, err).Info("writing to repository configuration")
 		return nil, nil, err
@@ -258,12 +258,40 @@ func GetStorageProviderAndOverrides(
 		return provider, nil, clues.Stack(err)
 	}
 
-	overrides := map[string]string{}
-
 	switch provider {
 	case storage.ProviderS3:
-		overrides = flags.S3FlagOverrides(cmd)
+		return provider, flags.S3FlagOverrides(cmd), nil
+	case storage.ProviderFilesystem:
+		return provider, flags.FilesystemFlagOverrides(cmd), nil
 	}
 
-	return provider, overrides, nil
+	return provider, nil, clues.New("unknown storage provider: " + provider.String())
+}
+
+// MakeAbsoluteFilePath does directory path expansions & conversions, namely:
+// 1. Expands "~" prefix to the user's home directory, and converts to absolute path.
+// 2. Relative paths are converted to absolute paths.
+// 3. Absolute paths are returned as-is.
+// 4. Empty paths are not allowed, an error is returned.
+func MakeAbsoluteFilePath(p string) (string, error) {
+	if len(p) == 0 {
+		return "", clues.New("empty path")
+	}
+
+	// Special case handling for "~". filepath.Abs will not expand it.
+	if p[0] == '~' {
+		homeDir, err := os.UserHomeDir()
+		if err != nil {
+			return "", clues.Wrap(err, "getting user home directory")
+		}
+
+		p = filepath.Join(homeDir, p[1:])
+	}
+
+	abs, err := filepath.Abs(p)
+	if err != nil {
+		return "", clues.Stack(err)
+	}
+
+	return abs, nil
 }

src/cli/utils/utils_test.go

@@ -1,6 +1,8 @@
 package utils
 
 import (
+	"os"
+	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -65,3 +67,73 @@ func (suite *CliUtilsSuite) TestSplitFoldersIntoContainsAndPrefix() {
 		})
 	}
 }
+
+// Test MakeAbsoluteFilePath
+func (suite *CliUtilsSuite) TestMakeAbsoluteFilePath() {
+	currentDir, err := os.Getwd()
+	assert.NoError(suite.T(), err)
+
+	homeDir, err := os.UserHomeDir()
+	assert.NoError(suite.T(), err)
+
+	table := []struct {
+		name        string
+		input       string
+		expected    string
+		expectedErr assert.ErrorAssertionFunc
+	}{
+		{
+			name:        "empty path",
+			input:       "",
+			expected:    "",
+			expectedErr: assert.Error,
+		},
+		{
+			name:        "absolute path",
+			input:       "/tmp/dir",
+			expected:    "/tmp/dir",
+			expectedErr: assert.NoError,
+		},
+		{
+			name:        "relative path",
+			input:       "subdir/file.txt",
+			expected:    filepath.Join(currentDir, "subdir/file.txt"),
+			expectedErr: assert.NoError,
+		},
+		{
+			name:        "relative path 2",
+			input:       ".",
+			expected:    currentDir,
+			expectedErr: assert.NoError,
+		},
+		{
+			name:        "home dir",
+			input:       "~/file.txt",
+			expected:    filepath.Join(homeDir, "file.txt"),
+			expectedErr: assert.NoError,
+		},
+		{
+			name:        "home dir 2",
+			input:       "~",
+			expected:    homeDir,
+			expectedErr: assert.NoError,
+		},
+		{
+			name:        "relative path with home dir",
+			input:       "~/test/..",
+			expected:    homeDir,
+			expectedErr: assert.NoError,
+		},
+	}
+
+	for _, test := range table {
+		suite.Run(test.name, func() {
+			t := suite.T()
+
+			actual, err := MakeAbsoluteFilePath(test.input)
+			assert.Equal(t, test.expected, actual)
+
+			test.expectedErr(t, err)
+		})
+	}
+}

src/internal/kopia/conn.go

@@ -222,6 +222,8 @@ func blobStoreByProvider(
 	switch s.Provider {
 	case storage.ProviderS3:
 		return s3BlobStorage(ctx, opts, s)
+	case storage.ProviderFilesystem:
+		return filesystemStorage(ctx, opts, s)
 	default:
 		return nil, clues.New("storage provider details are required").WithClues(ctx)
 	}

src/internal/kopia/filesystem.go

@@ -0,0 +1,35 @@
+package kopia
+
+import (
+	"context"
+
+	"github.com/alcionai/clues"
+	"github.com/kopia/kopia/repo/blob"
+	"github.com/kopia/kopia/repo/blob/filesystem"
+
+	"github.com/alcionai/corso/src/pkg/control/repository"
+	"github.com/alcionai/corso/src/pkg/storage"
+)
+
+func filesystemStorage(
+	ctx context.Context,
+	repoOpts repository.Options,
+	s storage.Storage,
+) (blob.Storage, error) {
+	cfg, err := s.StorageConfig()
+	if err != nil {
+		return nil, clues.Stack(err).WithClues(ctx)
+	}
+
+	fsCfg := cfg.(*storage.FilesystemConfig)
+	opts := filesystem.Options{
+		Path: fsCfg.Path,
+	}
+
+	store, err := filesystem.New(ctx, &opts, true)
+	if err != nil {
+		return nil, clues.Stack(err).WithClues(ctx)
+	}
+
+	return store, nil
+}

src/internal/tester/tconfig/config.go

@@ -20,6 +20,7 @@ const (
 	TestCfgEndpoint        = "endpoint"
 	TestCfgPrefix          = "prefix"
 	TestCfgStorageProvider = "provider"
+	TestCfgFilesystemPath  = "path"
 
 	// M365 config
 	TestCfgAzureTenantID    = "azure_tenantid"

src/pkg/storage/filesystem.go

@@ -0,0 +1,104 @@
+package storage
+
+import (
+	"github.com/alcionai/clues"
+	"github.com/spf13/cast"
+
+	"github.com/alcionai/corso/src/internal/common/str"
+)
+
+const (
+	FilesystemPath = "path"
+)
+
+var fsConstToTomlKeyMap = map[string]string{
+	StorageProviderTypeKey: StorageProviderTypeKey,
+	FilesystemPath:         FilesystemPath,
+}
+
+type FilesystemConfig struct {
+	Path string
+}
+
+func buildFilesystemConfigFromMap(config map[string]string) (*FilesystemConfig, error) {
+	c := &FilesystemConfig{}
+
+	if len(config) > 0 {
+		c.Path = orEmptyString(config[FilesystemPath])
+	}
+
+	return c, c.validate()
+}
+
+func (c FilesystemConfig) validate() error {
+	check := map[string]string{
+		FilesystemPath: c.Path,
+	}
+
+	for k, v := range check {
+		if len(v) == 0 {
+			return clues.Stack(errMissingRequired, clues.New(k))
+		}
+	}
+
+	return nil
+}
+
+func (c *FilesystemConfig) fsConfigsFromStore(g Getter) {
+	c.Path = cast.ToString(g.Get(FilesystemPath))
+}
+
+// TODO(pandeyabs): Remove this. It's not adding any value.
+func fsOverrides(in map[string]string) map[string]string {
+	return map[string]string{
+		FilesystemPath: in[FilesystemPath],
+	}
+}
+
+var _ Configurer = &FilesystemConfig{}
+
+func (c *FilesystemConfig) ApplyConfigOverrides(
+	g Getter,
+	readConfigFromStore bool,
+	matchFromConfig bool,
+	overrides map[string]string,
+) error {
+	if readConfigFromStore {
+		c.fsConfigsFromStore(g)
+
+		if matchFromConfig {
+			providerType := cast.ToString(g.Get(StorageProviderTypeKey))
+			if providerType != ProviderFilesystem.String() {
+				return clues.New("unsupported storage provider in config file: " + providerType)
+			}
+
+			// This is matching override values from config file.
+			if err := mustMatchConfig(g, fsConstToTomlKeyMap, fsOverrides(overrides)); err != nil {
+				return clues.Wrap(err, "verifying storage configs in corso config file")
+			}
+		}
+	}
+
+	c.Path = str.First(overrides[FilesystemPath], c.Path)
+
+	return c.validate()
+}
+
+// TODO(pandeyabs): We need to sanitize paths e.g. handle relative paths,
+// make paths cross platform compatible, etc.
+func (c FilesystemConfig) StringConfig() (map[string]string, error) {
+	cfg := map[string]string{
+		FilesystemPath: c.Path,
+	}
+
+	return cfg, c.validate()
+}
+
+var _ WriteConfigToStorer = FilesystemConfig{}
+
+func (c FilesystemConfig) WriteConfigToStore(
+	s Setter,
+) {
+	s.Set(StorageProviderTypeKey, ProviderFilesystem.String())
+	s.Set(FilesystemPath, c.Path)
+}

src/pkg/storage/storage.go

@@ -98,6 +98,8 @@ func (s Storage) StorageConfig() (Configurer, error) {
 	switch s.Provider {
 	case ProviderS3:
 		return buildS3ConfigFromMap(s.Config)
+	case ProviderFilesystem:
+		return buildFilesystemConfigFromMap(s.Config)
 	}
 
 	return nil, clues.New("unsupported storage provider: " + s.Provider.String())
@@ -107,6 +109,8 @@ func NewStorageConfig(provider ProviderType) (Configurer, error) {
 	switch provider {
 	case ProviderS3:
 		return &S3Config{}, nil
+	case ProviderFilesystem:
+		return &FilesystemConfig{}, nil
 	}
 
 	return nil, clues.New("unsupported storage provider: " + provider.String())

src/pkg/storage/testdata/storage.go

@@ -2,6 +2,7 @@ package testdata
 
 import (
 	"os"
+	"path/filepath"
 
 	"github.com/alcionai/clues"
 	"github.com/stretchr/testify/require"
@@ -46,6 +47,28 @@ func NewPrefixedS3Storage(t tester.TestT) storage.Storage {
 			Corso:       GetAndInsertCorso(""),
 			KopiaCfgDir: t.TempDir(),
 		})
+	require.NoErrorf(t, err, "creating storage: %+v", clues.ToCore(err))
+
+	return st
+}
+
+func NewFilesystemStorage(t tester.TestT) storage.Storage {
+	now := tester.LogTimeOfTest(t)
+	repoPath := filepath.Join(t.TempDir(), now)
+
+	err := os.MkdirAll(repoPath, 0700)
+	require.NoErrorf(t, err, "creating filesystem repo: %+v", clues.ToCore(err))
+
+	t.Logf("testing at filesystem repo [%s]", repoPath)
+
+	st, err := storage.NewStorage(
+		storage.ProviderFilesystem,
+		&storage.FilesystemConfig{
+			Path: repoPath,
+		},
+		storage.CommonConfig{
+			Corso: GetAndInsertCorso(""),
+		})
 	require.NoError(t, err, "creating storage", clues.ToCore(err))
 
 	return st