Implement staging and verificationv0.0.3

Ensure data integrity by supporting data staging and verification.

8 files changed, 214 insertions, 17 deletions

diff --git a/README.md b/README.md
index 307829b..bedc4fb 100644
--- a/README.md
+++ b/README.md
@@ -94,3 +94,37 @@ to = "/mirror/youtube-dl"
 Configuration may also be split across files in a directory.  By default
 loads configuration from /etc/mirror/mirror.toml and the /etc/mirror/conf.d
 directory.
+
+## Staging and Verification (Experimental)
+
+You can ensure mirror integrity by verifying the project in a staging directory
+before saving it.
+
+Use the 'verify' parameter to define Bash executed shell commands.  A non-zero
+exit value is considered a verification failure.  The standard error output is
+then written to the log.
+
+Use the 'staging-method' and 'staging-path' parameters to customize how the
+incoming mirror version is stored.  Both these values can be defined globally
+or specific to a mirror.
+
+The possible 'staging-method' values are:
+
+ - none: Save the data directly to the mirror destination.
+ - temporary: Save the data in a directory to be deleted after verification.
+ - persistent: Use a second location to store incoming data without clearing it after.
+
+The 'staging-path' defines the parent directory for where staging occurs, by default it is '/tmp/mirror'.
+
+```
+[[mirrors]]
+method = "rsync"
+from = "rsync://mirrors.kernel.org/slackware/slackware64-15.0"
+to = "/mirror/slackware/slackware64-15.0"
+staging-path = "/tmp/mirror"
+staging = "persistent"
+verify = """
+  (gpg --verify-files *.asc && tail +13 CHECKSUMS.md5 | md5sum -c --quiet -) && \
+  (find slackware64 -print0 -name '*.asc' | xargs -0L1 gpg2 --verify-files)
+"""
+```
diff --git a/internal/service/config.go b/internal/service/config.go
index 0f34997..2a8da6e 100644
--- a/internal/service/config.go
+++ b/internal/service/config.go
@@ -10,10 +10,21 @@ import (
 	"github.com/BurntSushi/toml"
 )
 
+type StagingMethod string
+
+const (
+	StagingMethodNone       = "none"
+	StagingMethodTemporary  = "temporary"
+	StagingMethodPersistent = "persistent"
+)
+
 // Global parameters
 type GlobalConfig struct {
-	MaxInterval *Duration `toml:"max-interval"`
-	MinInterval *Duration `toml:"min-interval"`
+	MaxInterval   *Duration `toml:"max-interval"`
+	MinInterval   *Duration `toml:"min-interval"`
+	StagingMethod string    `toml:"staging-method,omitempty"`
+	StagingPath   string    `toml:"staging-path,omitempty"`
+	Verify        string    `toml:"verify,omitempty"`
 }
 
 type Config struct {
@@ -27,8 +38,11 @@ func (c *Config) String() string {
 
 var DefaultConfig = Config{
 	GlobalConfig: GlobalConfig{
-		MaxInterval: DurationRef(24 * time.Hour),
-		MinInterval: DurationRef(time.Hour),
+		MaxInterval:   DurationRef(24 * time.Hour),
+		MinInterval:   DurationRef(time.Hour),
+		StagingMethod: StagingMethodNone,
+		StagingPath:   "/tmp/mirror",
+		Verify:        "",
 	},
 }
 
@@ -113,5 +127,11 @@ func (c *Config) Append(src *Config) {
 	if src.MinInterval != nil {
 		c.MinInterval = src.MinInterval
 	}
+	if src.StagingMethod != "" {
+		c.StagingMethod = src.StagingMethod
+	}
+	if src.StagingPath != "" {
+		c.StagingPath = src.StagingPath
+	}
 	c.Mirrors = append(c.Mirrors, src.Mirrors...)
 }
diff --git a/internal/service/git.go b/internal/service/git.go
index d56c96b..830f602 100644
--- a/internal/service/git.go
+++ b/internal/service/git.go
@@ -93,7 +93,10 @@ func MirrorGit(dst *internal.URL, src *internal.URL, description string) error {
 		return fmt.Errorf("'%s' scheme not supported", dst.Scheme)
 	}
 
-	if _, err := os.Stat(dst.Path); os.IsNotExist(err) {
+	entries, err := os.ReadDir(dst.Path)
+	if err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("creating directory '%s': %w", dst.Path, err)
+	} else if os.IsNotExist(err) || len(entries) == 0 {
 		err = os.MkdirAll(path.Join(dst.Path, ".."), 0750)
 		if err != nil {
 			return fmt.Errorf("creating new mirror: %w", err)
@@ -117,11 +120,6 @@ func MirrorGit(dst *internal.URL, src *internal.URL, description string) error {
 		}
 	}
 
-	err := setDescription(dst.Path, description)
-	if err != nil {
-		return err
-	}
-
 	err = setRemoteOrigin(dst.Path, src)
 	if err != nil {
 		return err
@@ -140,5 +138,10 @@ func MirrorGit(dst *internal.URL, src *internal.URL, description string) error {
 		return fmt.Errorf("fetching project: %s", err)
 	}
 
+	err = setDescription(dst.Path, description)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
diff --git a/internal/service/mirror.go b/internal/service/mirror.go
index 12aaa79..43dd059 100644
--- a/internal/service/mirror.go
+++ b/internal/service/mirror.go
@@ -8,10 +8,13 @@ import (
 )
 
 type Mirror struct {
-	Method      string        `toml:"method,omitempty"`
-	From        *internal.URL `toml:"from,omitempty"`
-	To          *internal.URL `toml:"to,omitempty"`
-	Description string        `toml:"description,omitempty"`
+	Method        string        `toml:"method,omitempty"`
+	From          *internal.URL `toml:"from,omitempty"`
+	To            *internal.URL `toml:"to,omitempty"`
+	Description   string        `toml:"description,omitempty"`
+	StagingMethod string        `toml:"staging-method,omitempty"`
+	StagingPath   string        `toml:"staging-path,omitempty"`
+	Verify        string        `toml:"verify,omitempty"`
 }
 
 func (m *Mirror) Equal(arg *Mirror) bool {
diff --git a/internal/service/service.go b/internal/service/service.go
index c5235f2..577a4b4 100644
--- a/internal/service/service.go
+++ b/internal/service/service.go
@@ -1,9 +1,12 @@
 package service
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"os"
+	"os/exec"
+	"path"
 	"strconv"
 	"strings"
 	"sync"
@@ -103,21 +106,82 @@ func (s *Service) scheduled(yield func(*mirrorRecord) bool) {
 	}
 }
 
+func runBash(runDir string, commands string) error {
+	cmd := exec.Command("bash", "-c", commands)
+	cmd.Dir = runDir
+
+	var stdErr bytes.Buffer
+	cmd.Stderr = &stdErr
+
+	err := cmd.Run()
+	if err != nil {
+		return errors.New(stdErr.String())
+	}
+
+	return nil
+}
+
 func (s *Service) Mirror(arg *Mirror) error {
 	if arg.From.Path == "" || arg.To.Path == "" || arg.Method == "" {
 		return fmt.Errorf("badly formatted mirror '%s'", arg.String())
 	}
 
+	if arg.To.Scheme != "" {
+		return fmt.Errorf("unsupported destination URL scheme '%s'", arg.To.Scheme)
+	}
+
+	cfgStagingMethod := s.cfg.StagingMethod
+	if arg.StagingMethod != "" {
+		cfgStagingMethod = arg.StagingMethod
+	}
+
+	cfgStagingPath := s.cfg.StagingPath
+	if arg.StagingPath != "" {
+		cfgStagingPath = arg.StagingPath
+	}
+
+	var downloadPath string
 	var err error
 
+	switch cfgStagingMethod {
+	case StagingMethodNone:
+		downloadPath = arg.To.Path
+	case StagingMethodTemporary:
+		tempPath, err := os.MkdirTemp("", "mirror.")
+		if err != nil {
+			return err
+		}
+		defer os.RemoveAll(tempPath)
+
+		downloadPath = tempPath
+	case StagingMethodPersistent:
+		downloadPath = path.Join(cfgStagingPath, arg.To.URL.Path)
+	default:
+		return fmt.Errorf("unknown staging method '%s'", cfgStagingMethod)
+	}
+
+	err = os.MkdirAll(downloadPath, 0755)
+	if err != nil {
+		return fmt.Errorf("creating directory '%s' for downloaded contents: %w", downloadPath, err)
+	}
+	err = os.MkdirAll(arg.To.URL.Path, 0755)
+	if err != nil {
+		return fmt.Errorf("creating directory '%s' for downloaded contents: %w", downloadPath, err)
+	}
+
+	err = Rsync(internal.MustURL(downloadPath), arg.To)
+	if err != nil {
+		return fmt.Errorf("copying current mirror data for staging to '%s': %w", downloadPath, err)
+	}
+
 	switch arg.Method {
 	case "git":
-		err = MirrorGit(arg.To, arg.From, arg.Description)
+		err = MirrorGit(internal.MustURL(downloadPath), arg.From, arg.Description)
 	case "github-assets":
 		client := github.NewClient()
-		err = client.MirrorAssets(arg.To, arg.From)
+		err = client.MirrorAssets(internal.MustURL(downloadPath), arg.From)
 	case "rsync":
-		err = Rsync(arg.To, arg.From)
+		err = Rsync(internal.MustURL(downloadPath), arg.From)
 	default:
 		err = fmt.Errorf("unknown method '%s'", arg.Method)
 	}
@@ -126,6 +190,20 @@ func (s *Service) Mirror(arg *Mirror) error {
 		return fmt.Errorf("could not clone from '%s': %w", arg.From, err)
 	}
 
+	if s.cfg.Verify != "" {
+		err = runBash(downloadPath, s.cfg.Verify)
+		if err != nil {
+			return fmt.Errorf("verification failed: %w", err)
+		}
+	}
+
+	if downloadPath != arg.To.String() {
+		err = Rsync(arg.To, internal.MustURL(downloadPath))
+		if err != nil {
+			return fmt.Errorf("committing staged mirror data: %w", err)
+		}
+	}
+
 	return nil
 }
 
diff --git a/internal/service/service_test.go b/internal/service/service_test.go
new file mode 100644
index 0000000..b73a753
--- /dev/null
+++ b/internal/service/service_test.go
@@ -0,0 +1,26 @@
+package service
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestService_Mirror(t *testing.T) {
+	cfgPath := os.Getenv("INTEGRATION_TEST_CONFIG")
+	if cfgPath == "" {
+		t.Skip("set INTEGRATION_TEST_CONFIG to run this test")
+	}
+
+	cfg, err := ReadConfig(cfgPath)
+	require.NoError(t, err, cfgPath)
+
+	srv, err := NewService(cfg)
+	require.NoError(t, err, "creating service")
+
+	for _, mirror := range srv.cfg.Mirrors {
+		err = srv.Mirror(mirror)
+		require.NoError(t, err)
+	}
+}
diff --git a/internal/service/testdata/integration_test_config.toml b/internal/service/testdata/integration_test_config.toml
new file mode 100644
index 0000000..87a93e5
--- /dev/null
+++ b/internal/service/testdata/integration_test_config.toml
@@ -0,0 +1,29 @@
+[[mirrors]]
+method = "git"
+from = "https://git.server.ky/slackcoder/mirror"
+to = "/tmp/local-mirror/slackcoder/mirror"
+description = "Mirror project"
+staging-method = "temporary"
+verify = "git verify-commit HEAD"
+
+[[mirrors]]
+method = "rsync"
+from = "rsync://ftp.gnu.org/gnu/taler"
+to = "/tmp/local-mirror/gnu/taler"
+staging-method = "persistent"
+verify = "find . -print0 -mindepth 1 -not -name '*.sig' | xargs -0L1 -d '{}' gpg2 --verify-files '{}'.sig"
+
+[[mirrors]]
+method = "github-assets"
+from = "https://github.com/yt-dlp/yt-dlp"
+to = "/tmp/local-mirror/yt-dlp"
+staging-method = "persistent"
+verify = """
+mapfile -d '' releases < <( find . -maxdepth 1 -mindepth 1 -type d -print0 )
+for release in "${releases[@]}"; do
+  (cd "$release";
+    gpg2 --quiet --verify-files SHA2-256SUMS.sig &&
+      sha256sum --quiet --check SHA2-256SUMS
+  ) || exit 1
+done
+"""
diff --git a/internal/url.go b/internal/url.go
index d7a712d..f5c0e17 100644
--- a/internal/url.go
+++ b/internal/url.go
@@ -19,6 +19,10 @@ func (u *URL) UnmarshalText(buf []byte) error {
 	return nil
 }
 
+func (u *URL) Copy() *URL {
+	return MustURL(u.String())
+}
+
 func MustURL(arg string) *URL {
 	u, err := url.Parse(arg)
 	if err != nil {