Merge pull request #2349 from erikwilson/fix-data-extract

Fix race condition in data extraction

cmd/k3s/main.go

@@ -12,6 +12,8 @@ import (
 	"github.com/rancher/k3s/pkg/cli/cmds"
 	"github.com/rancher/k3s/pkg/data"
 	"github.com/rancher/k3s/pkg/datadir"
+	"github.com/rancher/k3s/pkg/dataverify"
+	"github.com/rancher/k3s/pkg/flock"
 	"github.com/rancher/k3s/pkg/untar"
 	"github.com/rancher/k3s/pkg/version"
 	"github.com/sirupsen/logrus"
@@ -88,6 +90,7 @@ func stageAndRun(dataDir string, cmd string, args []string) error {
 	if err != nil {
 		return errors.Wrap(err, "extracting data")
 	}
+	logrus.Debugf("Asset dir %s", dir)
 
 	if err := os.Setenv("PATH", filepath.Join(dir, "bin")+":"+os.Getenv("PATH")+":"+filepath.Join(dir, "bin/aux")); err != nil {
 		return err
@@ -102,6 +105,7 @@ func stageAndRun(dataDir string, cmd string, args []string) error {
 	}
 
 	logrus.Debugf("Running %s %v", cmd, args)
+
 	return syscall.Exec(cmd, args, os.Environ())
 }
 
@@ -115,13 +119,27 @@ func extract(dataDir string) (string, error) {
 	// first look for global asset folder so we don't create a HOME version if not needed
 	_, dir := getAssetAndDir(datadir.DefaultDataDir)
 	if _, err := os.Stat(dir); err == nil {
-		logrus.Debugf("Asset dir %s", dir)
 		return dir, nil
 	}
 
 	asset, dir := getAssetAndDir(dataDir)
+	// check if target directory already exists
+	if _, err := os.Stat(dir); err == nil {
+		return dir, nil
+	}
+
+	// acquire a data directory lock
+	os.MkdirAll(filepath.Join(dataDir, "data"), 0755)
+	lockFile := filepath.Join(dataDir, "data", ".lock")
+	logrus.Infof("Acquiring lock file %s", lockFile)
+	lock, err := flock.Acquire(lockFile)
+	if err != nil {
+		return "", err
+	}
+	defer flock.Release(lock)
+
+	// check again if target directory exists
 	if _, err := os.Stat(dir); err == nil {
-		logrus.Debugf("Asset dir %s", dir)
 		return dir, nil
 	}
 
@@ -135,12 +153,14 @@ func extract(dataDir string) (string, error) {
 
 	tempDest := dir + "-tmp"
 	defer os.RemoveAll(tempDest)
-
 	os.RemoveAll(tempDest)
 
 	if err := untar.Untar(buf, tempDest); err != nil {
 		return "", err
 	}
+	if err := dataverify.Verify(filepath.Join(tempDest, "bin")); err != nil {
+		return "", err
+	}
 
 	currentSymLink := filepath.Join(dataDir, "data", "current")
 	previousSymLink := filepath.Join(dataDir, "data", "previous")
@@ -152,6 +172,5 @@ func extract(dataDir string) (string, error) {
 	if err := os.Symlink(dir, currentSymLink); err != nil {
 		return "", err
 	}
-
 	return dir, os.Rename(tempDest, dir)
 }

pkg/dataverify/dataverify.go

@@ -0,0 +1,117 @@
+package dataverify
+
+import (
+	"bufio"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/sirupsen/logrus"
+)
+
+// Verify will check the sha256sums and links from the files in a given directory
+func Verify(dir string) error {
+	failed := false
+	if err := VerifySums(dir, ".sha256sums"); err != nil {
+		logrus.Errorf("Unable to verify sums: %s", err)
+		failed = true
+	}
+	if err := VerifyLinks(dir, ".links"); err != nil {
+		logrus.Errorf("Unable to verify links: %s", err)
+		failed = true
+	}
+	if failed {
+		return fmt.Errorf("failed to verify directory %s", dir)
+	}
+	return nil
+}
+
+// VerifySums will take a file which contains a list of hash sums for files and verify they match
+func VerifySums(root, sumListFile string) error {
+	sums, err := fileMapFields(filepath.Join(root, sumListFile), 1, 0)
+	if err != nil {
+		return err
+	}
+	if len(sums) == 0 {
+		return fmt.Errorf("no entries found in %s", sumListFile)
+	}
+	numFailed := 0
+	for sumFile, sumExpected := range sums {
+		file := filepath.Join(root, sumFile)
+		sumActual, _ := sha256Sum(file)
+		if sumExpected != sumActual {
+			logrus.Errorf("Hash for file %s expected to be %s (fail)", sumFile, sumExpected)
+			numFailed++
+		} else {
+			logrus.Debugf("Verified hash %s is correct", sumFile)
+		}
+	}
+	if numFailed != 0 {
+		return fmt.Errorf("failed %d hash verifications", numFailed)
+	}
+	return nil
+}
+
+// VerifyLinks will take a file which contains a list of target links for files and verify they match
+func VerifyLinks(root, linkListFile string) error {
+	links, err := fileMapFields(filepath.Join(root, linkListFile), 0, 1)
+	if err != nil {
+		return err
+	}
+	if len(links) == 0 {
+		return fmt.Errorf("no entries found in %s", linkListFile)
+	}
+	numFailed := 0
+	for linkFile, linkExpected := range links {
+		file := filepath.Join(root, linkFile)
+		linkActual, _ := os.Readlink(file)
+		if linkExpected != linkActual {
+			logrus.Errorf("Link for file %s expected to be %s (fail)", linkFile, linkExpected)
+			numFailed++
+		} else {
+			logrus.Debugf("Verified link %s is correct", linkFile)
+		}
+	}
+	if numFailed != 0 {
+		return fmt.Errorf("failed %d link verifications", numFailed)
+	}
+	return nil
+}
+
+func fileMapFields(fileName string, key, val int) (map[string]string, error) {
+	file, err := os.Open(fileName)
+	if err != nil {
+		return nil, err
+	}
+	defer file.Close()
+	result := map[string]string{}
+	scanner := bufio.NewScanner(file)
+	for scanner.Scan() {
+		fields := strings.Fields(scanner.Text())
+		if len(fields) == 0 {
+			continue
+		}
+		if len(fields) <= key || len(fields) <= val {
+			return nil, fmt.Errorf("fields for file %s (%d) smaller than required index (key: %d, val: %d)", fileName, len(fields), key, val)
+		}
+		result[fields[key]] = fields[val]
+	}
+	return result, scanner.Err()
+}
+
+func sha256Sum(filePath string) (string, error) {
+	file, err := os.Open(filePath)
+	if err != nil {
+		return "", err
+	}
+	defer file.Close()
+	hash := sha256.New()
+	if _, err := io.Copy(hash, file); err != nil {
+		return "", err
+	}
+	return hex.EncodeToString(hash.Sum(nil)), nil
+}

pkg/flock/flock_other.go

@@ -0,0 +1,29 @@
+// +build !linux,!darwin,!freebsd,!openbsd,!netbsd,!dragonfly
+
+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package flock
+
+// Acquire is not implemented on non-unix systems.
+func Acquire(path string) (int, error) {
+	return -1, nil
+}
+
+// Release is not implemented on non-unix systems.
+func Release(lock int) error {
+	return nil
+}

pkg/flock/flock_unix.go

@@ -0,0 +1,36 @@
+// +build linux darwin freebsd openbsd netbsd dragonfly
+
+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package flock
+
+import "golang.org/x/sys/unix"
+
+// Acquire creates an exclusive lock on a file for the duration of the process, or until Release(d).
+// This method is reentrant.
+func Acquire(path string) (int, error) {
+	lock, err := unix.Open(path, unix.O_CREAT|unix.O_RDWR|unix.O_CLOEXEC, 0600)
+	if err != nil {
+		return -1, err
+	}
+	return lock, unix.Flock(lock, unix.LOCK_EX)
+}
+
+// Release removes an existing lock held by this process.
+func Release(lock int) error {
+	return unix.Flock(lock, unix.LOCK_UN)
+}