WIP: fix encrypt-on-recv

fixes https://github.com/zrepl/zrepl/issues/504

docs/configuration/prune.rst

@@ -124,51 +124,43 @@ The syntax to describe the bucket list is as follows:
 
 ::
 
-   Assume the following grid spec:
+          This grid spec produces the following list of adjacent buckets. For the sake of simplicity,
+          we subject all snapshots to the grid pruning policy by settings `regex: .*`.
 
+          `
            grid: 1x1h(keep=all) | 2x2h | 1x3h
-
-   This grid spec produces the following constellation of buckets:
+           regex: .*
+          `
 
           0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
           |         |         |         |         |         |         |         |         |         |
-   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
+          |-Bucket1-|-----Bucket 2------|------Bucket 3-----|-----------Bucket 4----------|
           | keep=all|      keep=1       |       keep=1      |            keep=1           |
 
 
 
-   Now assume that we have a set of snapshots @a, @b, ..., @D.
-   Snapshot @a is the most recent snapshot.
-   Snapshot @D is the oldest snapshot, it is almost 9 hours older than snapshot @a.
-   We place the snapshots on the same timeline as the buckets:
+          Let us consider the following set of snapshots @a-zA-C:
 
 
-   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
-   |         |         |         |         |         |         |         |         |         |
-   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
-   | keep=all|      keep=1       |       keep=1      |            keep=1           |
-   |         |                   |                   |                             |
-   |  a  b  c| d  e  f  g  h  i  j  k  l  m  n  o  p |q  r  s  t  u  v  w  x  y  z |A  B  C  D
+          |  a  b  c  d  e  f  g  h  i  j  k  l  m  n  o  p  q  r  s  t  u  v  w  x  y  z  A  B  C  D        |
 
-   The result is the following mapping of snapshots to buckets:
+          The `grid` algorithm maps them to their respective buckets:
 
           Bucket 1: a, b, c
           Bucket 2: d,e,f,g,h,i,j
           Bucket 3: k,l,m,n,o,p
-   Bucket4:   q,r,s,t,u,v,w,x,y,z
-   No bucket: A,B,C,D
+          Bucket 4: q,r, q,r,s,t,u,v,w,x,y,z
+          None:     A,B,C,D
 
-   For each bucket, we now prune snapshots until it only contains `keep` snapshots.
-   Newer snapshots are destroyed first.
-   Snapshots that do not fall into a bucket are always destroyed.
+          It then applies the per-bucket pruning logic described above which resulting in the
+          following list of remaining snapshots.
 
-   Result after pruning:
+          |  a  b  c                    j                 p                             z                    |
+
+          Note that it only makes sense to grow (not shorten) the interval duration for buckets
+          further in the past since each bucket acts like a low-pass filter for incoming snapshots
+          and adding a less-low-pass-filter after a low-pass one has no effect.
 
-   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
-   |         |         |         |         |         |         |         |         |         |
-   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
-   |         |                   |                   |                             |
-   |  a  b  c|                   j                 p |                           z  |
 
 .. _prune-keep-last-n:
 

platformtest/tests/generated_cases.go

@@ -34,10 +34,5 @@ var Cases = []Case{BatchDestroy,
 	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden__EncryptionSupported_true,
 	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
 	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
-	SendStreamCloseAfterBlockedOnPipeWrite,
-	SendStreamCloseAfterEOFRead,
-	SendStreamMultipleCloseAfterEOF,
-	SendStreamMultipleCloseBeforeEOF,
-	SendStreamNonEOFReadErrorHandling,
 	UndestroyableSnapshotParsing,
 }

platformtest/tests/recvRollback.go

@@ -40,7 +40,6 @@ func ReceiveForceRollbackWorksUnencrypted(ctx *platformtest.Context) {
 
 	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
 	require.NoError(ctx, err)
-	defer sendStream.Close()
 
 	recvOpts := zfs.RecvOptions{
 		RollbackAndForceRecv: true,

platformtest/tests/replication.go

@@ -7,7 +7,6 @@ import (
 	"os"
 	"path"
 	"sort"
-	"strings"
 	"time"
 
 	"github.com/kr/pretty"
@@ -91,7 +90,6 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		ReplicationConfig: &pdu.ReplicationConfig{
 			Protection: i.guarantee,
 		},
-		SizeEstimationConcurrency: 1,
 	}
 
 	report, wait := replication.Do(
@@ -1056,9 +1054,6 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
 		rep := fsByName[fs]
 		require.Len(ctx, rep.Steps, 1)
 		require.Nil(ctx, rep.PlanError)
-		if rep.StepError != nil && strings.Contains(rep.StepError.Error(), "invalid option 'x'") {
-			ctx.SkipNow() // XXX feature detection
-		}
 		require.Nil(ctx, rep.StepError)
 		require.Len(ctx, rep.Steps, 1)
 		require.Equal(ctx, 1, rep.CurrentStep)

platformtest/tests/sendStream.go

@@ -1,186 +0,0 @@
-package tests
-
-import (
-	"fmt"
-	"io"
-	"io/ioutil"
-	"os"
-	"os/exec"
-	"path"
-	"syscall"
-	"time"
-
-	"github.com/stretchr/testify/require"
-	"golang.org/x/sys/unix"
-
-	"github.com/zrepl/zrepl/platformtest"
-	"github.com/zrepl/zrepl/util/nodefault"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-func sendStreamTest(ctx *platformtest.Context) *zfs.SendStream {
-
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-		DESTROYROOT
-		CREATEROOT
-		+  "sender"
-	`)
-
-	fs := fmt.Sprintf("%s/sender", ctx.RootDataset)
-
-	fsmpo, err := zfs.ZFSGetMountpoint(ctx, fs)
-	require.NoError(ctx, err)
-
-	writeDummyData(path.Join(fsmpo.Mountpoint, "dummy.data"), 1<<26)
-	mustSnapshot(ctx, fs+"@1")
-	snap := fsversion(ctx, fs, "@1")
-	snapSendArg := snap.ToSendArgVersion()
-
-	sendArgs, err := zfs.ZFSSendArgsUnvalidated{
-		FS:   fs,
-		From: nil,
-		To:   &snapSendArg,
-		ZFSSendFlags: zfs.ZFSSendFlags{
-			Encrypted: &nodefault.Bool{B: false},
-		},
-	}.Validate(ctx)
-	require.NoError(ctx, err)
-
-	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
-	require.NoError(ctx, err)
-
-	return sendStream
-
-}
-
-func SendStreamCloseAfterBlockedOnPipeWrite(ctx *platformtest.Context) {
-
-	sendStream := sendStreamTest(ctx)
-
-	// let the pipe buffer fill and the zfs process block uninterruptibly
-
-	ctx.Logf("waiting for pipe write to block")
-	time.Sleep(5 * time.Second) // XXX need a platform-neutral way to detect that the pipe is full and the writer is blocked
-
-	ctx.Logf("closing send stream")
-	err := sendStream.Close() // this is what this test case is about
-	ctx.Logf("close error: %T %s", err, err)
-	require.NoError(ctx, err)
-
-	exitErrZfsError := sendStream.TestOnly_ExitErr()
-	require.Contains(ctx, exitErrZfsError.Error(), "signal")
-	require.Error(ctx, exitErrZfsError)
-	exitErr, ok := exitErrZfsError.WaitErr.(*exec.ExitError)
-	require.True(ctx, ok)
-	if exitErr.Exited() {
-		// some ZFS impls (FreeBSD 12) behaves that way
-		return
-	}
-
-	// ProcessState is only available after exit
-	// => use as proxy that the process was wait()ed upon and is gone
-	ctx.Logf("%#v", exitErr.ProcessState)
-	require.NotNil(ctx, exitErr.ProcessState)
-	// and let's verify that the process got killed, so that we know it was the call to .Close() above
-	waitStatus := exitErr.ProcessState.Sys().(syscall.WaitStatus)
-	ctx.Logf("wait status: %#v", waitStatus)
-	ctx.Logf("exit status: %v", waitStatus.ExitStatus())
-	require.True(ctx, waitStatus.Signaled())
-	switch waitStatus.Signal() {
-	case unix.SIGKILL:
-		fallthrough
-	case unix.SIGPIPE:
-		// ok
-
-	default:
-		ctx.Errorf("%T %s\n%v", waitStatus.Signal(), waitStatus.Signal(), waitStatus.Signal())
-		ctx.FailNow()
-	}
-}
-
-func SendStreamCloseAfterEOFRead(ctx *platformtest.Context) {
-
-	sendStream := sendStreamTest(ctx)
-
-	_, err := io.Copy(ioutil.Discard, sendStream)
-	require.NoError(ctx, err)
-
-	var buf [128]byte
-	n, err := sendStream.Read(buf[:])
-	require.Zero(ctx, n)
-	require.Equal(ctx, io.EOF, err)
-
-	err = sendStream.Close()
-	require.NoError(ctx, err)
-
-	n, err = sendStream.Read(buf[:])
-	require.Zero(ctx, n)
-	require.Equal(ctx, os.ErrClosed, err, "same read error should be returned")
-}
-
-func SendStreamMultipleCloseAfterEOF(ctx *platformtest.Context) {
-
-	sendStream := sendStreamTest(ctx)
-
-	_, err := io.Copy(ioutil.Discard, sendStream)
-	require.NoError(ctx, err)
-
-	var buf [128]byte
-	n, err := sendStream.Read(buf[:])
-	require.Zero(ctx, n)
-	require.Equal(ctx, io.EOF, err)
-
-	err = sendStream.Close()
-	require.NoError(ctx, err)
-
-	err = sendStream.Close()
-	require.Equal(ctx, os.ErrClosed, err)
-}
-
-func SendStreamMultipleCloseBeforeEOF(ctx *platformtest.Context) {
-
-	sendStream := sendStreamTest(ctx)
-
-	err := sendStream.Close()
-	require.NoError(ctx, err)
-
-	err = sendStream.Close()
-	require.Equal(ctx, os.ErrClosed, err)
-}
-
-type failingReadCloser struct {
-	err error
-}
-
-var _ io.ReadCloser = &failingReadCloser{}
-
-func (c *failingReadCloser) Read(p []byte) (int, error) { return 0, c.err }
-func (c *failingReadCloser) Close() error               { return c.err }
-
-func SendStreamNonEOFReadErrorHandling(ctx *platformtest.Context) {
-
-	sendStream := sendStreamTest(ctx)
-
-	var buf [128]byte
-	n, err := sendStream.Read(buf[:])
-	require.Equal(ctx, len(buf), n)
-	require.NoError(ctx, err)
-
-	var mockError = fmt.Errorf("taeghaefow4piesahwahjocu7ul5tiachaiLipheijae8ooZ8Pies8shohGee9feeTeirai5aiFeiyaecai4kiaLoh4azeih0tea")
-	mock := &failingReadCloser{err: mockError}
-	orig := sendStream.TestOnly_ReplaceStdoutReader(mock)
-
-	n, err = sendStream.Read(buf[:])
-	require.Equal(ctx, 0, n)
-	require.Equal(ctx, mockError, err)
-
-	if sendStream.TestOnly_ReplaceStdoutReader(orig) != mock {
-		panic("incorrect test impl")
-	}
-
-	err = sendStream.Close()
-	require.NoError(ctx, err) // if we can't kill the child then this will be a flaky test, but let's assume we can kill the child
-
-	err = sendStream.Close()
-	require.Equal(ctx, os.ErrClosed, err)
-}

zfs/placeholder.go

@@ -90,6 +90,40 @@ func ZFSCreatePlaceholderFilesystem(ctx context.Context, fs *DatasetPath, parent
 		"-o", fmt.Sprintf("%s=%s", PlaceholderPropertyName, placeholderPropertyOn),
 		"-o", "mountpoint=none",
 	}
+
+	// xxx handle encryption not supported
+	props, err := zfsGet(ctx, parent.ToString(), []string{"keystatus"}, SourceAny)
+	if err != nil {
+		return errors.Wrap(err, "cannot determine key status")
+	}
+	keystatus := props.Get("keystatus") // xxx ability to distringuish `-` from ``
+	if keystatus == "" {
+		// parent is unencrypted => placeholder inherits encryption
+	} else if keystatus == "available" {
+		// parent is encrypted but since the key is loaded we can create an encrypted placeholder dataset
+		// without `-o encryption=off`
+	} else if keystatus == "unavailable" {
+		// parent is encrypted but keys are not loaded, either because
+		//  1) it's a send-encrypted dataset, or because
+		//  2) the user forgot to zfs load-key the root_fs or above
+		// In both cases we can't create an encrypted placeholder dataset.
+		// In case 1), we want to create an unencrypted placeholder through `-o encryption=off`.
+		// In case 2), `-o encryption=off` is harmful security-wise because it breaks the encrypt-on-receiver use case (https://github.com/zrepl/zrepl/issues/504)
+		// 			   I.e., all children of the placeholder won't be encrypted because they inherit encryption=off
+		//
+		// => we could attempt to distinguish the cases by being more context sensitive
+		//    (i.e., check whether the encryption root is root_fs or a parent thereof)
+		//    However, that's always going to be imprecise, and a wrong decision there is harmful security-wise.
+		//
+		// => thus the safe choice is to never use `-o encryption=off` by default
+		//    only if we know for sure that the stream is encrypted should we create placeholders with `-o encryption=off`
+		//    => this knowledge can be achieved through one of the following means:
+		//		a) have the sender indicate it to us in the RPC request (it's ok to trust them in this particular case since lying only hurts _their_ data's confidentiality)
+		// 		b) have the user acknowledge it in the receiver config
+	} else {
+		return errors.Errorf("unknown keystatus value %q for dataset %q", keystatus, parent.ToString())
+	}
+
 	if parentEncrypted, err := ZFSGetEncryptionEnabled(ctx, parent.ToString()); err != nil {
 		return errors.Wrap(err, "cannot determine encryption support")
 	} else if parentEncrypted {

zfs/zfs.go

@@ -14,6 +14,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -333,122 +334,60 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	return stdoutReader, stdoutWriter, nil
 }
 
-type sendStreamState int
-
-const (
-	sendStreamOpen sendStreamState = iota
-	sendStreamClosed
-)
-
 type SendStream struct {
 	cmd  *zfscmd.Cmd
 	kill context.CancelFunc
-	stdoutReader io.ReadCloser // not *os.File for mocking during platformtest
+
+	closeMtx     sync.Mutex
+	stdoutReader *os.File
 	stderrBuf    *circlog.CircularLog
-
-	mtx     sync.Mutex
-	state   sendStreamState
-	exitErr *ZFSError
+	opErr        error
 }
 
-func (s *SendStream) Read(p []byte) (n int, _ error) {
-	s.mtx.Lock()
-	defer s.mtx.Unlock()
+func (s *SendStream) Read(p []byte) (n int, err error) {
+	s.closeMtx.Lock()
+	opErr := s.opErr
+	s.closeMtx.Unlock()
+	if opErr != nil {
+		return 0, opErr
+	}
 
-	switch s.state {
-	case sendStreamClosed:
-		return 0, os.ErrClosed
-
-	case sendStreamOpen:
-		n, readErr := s.stdoutReader.Read(p)
-		if readErr != nil {
-			debug("sendStream: read: readErr=%T %s", readErr, readErr)
-			if readErr == io.EOF {
-				// io.EOF must be bubbled up as is so that consumers can handle it properly.
-				return n, readErr
-			}
-			// Assume that the error is not retryable.
-			// Try to kill now so that we can return a nice *ZFSError with captured stderr.
-			// If the kill doesn't work, it doesn't matter because the caller must by contract call Close() anyways.
-			killErr := s.killAndWait()
-			debug("sendStream: read: killErr=%T %s", killErr, killErr)
-			if killErr == nil {
-				s.state = sendStreamClosed
-				return n, s.exitErr // return the nice error
-			} else {
-				// we remain open so that we retry
-				return n, readErr // return the normal error
-			}
-		}
-		return n, readErr
-
-	default:
-		panic("unreachable")
+	n, err = s.stdoutReader.Read(p)
+	if err != nil {
+		debug("sendStream: read err: %T %s", err, err)
+		// TODO we assume here that any read error is permanent
+		// which is most likely the case for a local zfs send
+		kwerr := s.killAndWait(err)
+		debug("sendStream: killAndWait n=%v err= %T %s", n, kwerr, kwerr)
+		// TODO we assume here that any read error is permanent
+		return n, kwerr
 	}
+	return n, err
 }
 
 func (s *SendStream) Close() error {
 	debug("sendStream: close called")
-	s.mtx.Lock()
-	defer s.mtx.Unlock()
-
-	switch s.state {
-	case sendStreamOpen:
-		err := s.killAndWait()
-		if err != nil {
-			return err
-		} else {
-			s.state = sendStreamClosed
-			return nil
-		}
-	case sendStreamClosed:
-		return os.ErrClosed
-	default:
-		panic("unreachable")
-	}
+	return s.killAndWait(nil)
 }
 
-// returns nil iff the child process is gone (has been successfully waited upon)
-// in that case, s.exitErr is set
-func (s *SendStream) killAndWait() error {
+func (s *SendStream) killAndWait(precedingReadErr error) error {
 
 	debug("sendStream: killAndWait enter")
 	defer debug("sendStream: killAndWait leave")
-
-	// send SIGKILL
-	s.kill()
-
-	// Close our read-end of the pipe.
-	//
-	// We must do this before .Wait() because in some (not all) versions/build configs of ZFS,
-	// `zfs send` uses a separate kernel thread (taskq) to write the send stream (function `dump_bytes`).
-	// The `zfs send` thread then waits uinterruptably for the taskq thread to finish the write.
-	// And signalling the `zfs send` thread doesn't propagate to the taskq thread.
-	// So we end up in a state where we .Wait() forever.
-	// (See https://github.com/openzfs/zfs/issues/12500 and
-	//  https://github.com/zrepl/zrepl/issues/495#issuecomment-902530043)
-	//
-	// By closing our read end of the pipe before .Wait(), we unblock the taskq thread if there is any.
-	// If there is no separate taskq thread, the SIGKILL to `zfs end` would suffice and be most precise,
-	// but due to the circumstances above, there is no other portable & robust way.
-	//
-	// However, the fallout from closing the pipe is that (in non-taskq builds) `zfs sends` will get a SIGPIPE.
-	// And on Linux, that SIGPIPE appears to win over the previously issued SIGKILL.
-	// And thus, on Linux, the `zfs send` will be killed by the default SIGPIPE handler.
-	// We can observe this in the WaitStatus below.
-	// This behavior is slightly annoying because the *exec.ExitError's message ("signal: broken pipe")
-	// isn't as clear as ("signal: killed").
-	// However, it seems like we just have to live with that. (covered by platformtest)
-	var closePipeErr error
-	if s.stdoutReader != nil {
-		closePipeErr = s.stdoutReader.Close()
-		if closePipeErr == nil {
-			// avoid double-closes in case waiting below doesn't work
-			// and someone attempts Close again
-			s.stdoutReader = nil
+	if precedingReadErr == io.EOF {
+		// give the zfs process a little bit of time to terminate itself
+		// if it holds this deadline, exitErr will be nil
+		time.AfterFunc(200*time.Millisecond, s.kill)
 	} else {
-			return closePipeErr
+		s.kill()
 	}
+
+	// allow async kills from Close(), that's why we only take the mutex here
+	s.closeMtx.Lock()
+	defer s.closeMtx.Unlock()
+
+	if s.opErr != nil {
+		return s.opErr
 	}
 
 	waitErr := s.cmd.Wait()
@@ -463,30 +402,39 @@ func (s *SendStream) killAndWait() error {
 		}
 	}
 
-	// invariant: at this point, the child is gone and we cleaned up everything related to the SendStream
+	// now, after we know the program exited do we close the pipe
+	var closePipeErr error
+	if s.stdoutReader != nil {
+		closePipeErr = s.stdoutReader.Close()
+		if closePipeErr == nil {
+			// avoid double-closes in case anything below doesn't work
+			// and someone calls Close again
+			s.stdoutReader = nil
+		} else {
+			return closePipeErr
+		}
+	}
 
+	// we managed to tear things down, no let's give the user some pretty *ZFSError
 	if exitErr != nil {
-		// zfs send exited with an error or was killed by a signal.
-		s.exitErr = &ZFSError{
+		s.opErr = &ZFSError{
 			Stderr:  []byte(s.stderrBuf.String()),
 			WaitErr: exitErr,
 		}
 	} else {
-		// zfs send exited successfully (we know that since waitErr was either nil or wasn't an *exec.ExitError)
-		s.exitErr = nil
+		s.opErr = precedingReadErr
 	}
 
-	return nil
+	// detect the edge where we're called from s.Read
+	// after the pipe EOFed and zfs send exited without errors
+	// this is actually the "hot" / nice path
+	if exitErr == nil && precedingReadErr == io.EOF {
+		return precedingReadErr
 	}
 
-func (s *SendStream) TestOnly_ReplaceStdoutReader(f io.ReadCloser) (prev io.ReadCloser) {
-	prev = s.stdoutReader
-	s.stdoutReader = f
-	return prev
+	return s.opErr
 }
 
-func (s *SendStream) TestOnly_ExitErr() *ZFSError { return s.exitErr }
-
 // NOTE: When updating this struct, make sure to update funcs Validate ValidateCorrespondsToResumeToken
 type ZFSSendArgVersion struct {
 	RelName string

zfs/zfscmd/zfscmd.go

@@ -209,7 +209,3 @@ func (c *Cmd) Runtime() time.Duration {
 	}
 	return c.waitReturnedAt.Sub(c.startedAt)
 }
-
-func (c *Cmd) TestOnly_ExecCmd() *exec.Cmd {
-	return c.cmd
-}