Path Encryption

go.mod

@@ -98,9 +98,9 @@ require (
 	go.uber.org/atomic v1.3.2 // indirect
 	go.uber.org/multierr v1.1.0 // indirect
 	go.uber.org/zap v1.9.1
-	golang.org/x/crypto v0.0.0-20181001203147-e3636079e1a4
+	golang.org/x/crypto v0.0.0-20181009213950-7c1a557ab941
 	golang.org/x/net v0.0.0-20181003013248-f5e5bdd77824
-	golang.org/x/sys v0.0.0-20181004145325-8469e314837c
+	golang.org/x/sys v0.0.0-20181005133103-4497e2df6f9e
 	golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2 // indirect
 	google.golang.org/grpc v1.15.0
 	gopkg.in/Shopify/sarama.v1 v1.18.0 // indirect

go.sum

@@ -268,11 +268,14 @@ go.uber.org/zap v1.9.1 h1:XCJQEf3W6eZaVwhRBof6ImoYGJSITeKWsyeh3HFu/5o=
 go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 golang.org/x/crypto v0.0.0-20181001203147-e3636079e1a4 h1:Vk3wNqEZwyGyei9yq5ekj7frek2u7HUfffJ1/opblzc=
 golang.org/x/crypto v0.0.0-20181001203147-e3636079e1a4/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/crypto v0.0.0-20181009213950-7c1a557ab941 h1:qBTHLajHecfu+xzRI9PqVDcqx7SdHj9d4B+EzSn3tAc=
+golang.org/x/crypto v0.0.0-20181009213950-7c1a557ab941/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181003013248-f5e5bdd77824 h1:MkjFNbaZJyH98M67Q3umtwZ+EdVdrNJLqSwZp5vcv60=
 golang.org/x/net v0.0.0-20181003013248-f5e5bdd77824/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20181005035420-146acd28ed58 h1:otZG8yDCO4LVps5+9bxOeNiCvgmOyt96J3roHTYs7oE=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be h1:vEDujvNQGv4jgYKudGeI/+DAX4Jffq6hpD55MmoEvKs=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f h1:wMNYb4v58l5UBM7MYRLPG6ZhfOqbKu7X5eyFl8ZhKvA=
@@ -284,6 +287,8 @@ golang.org/x/sys v0.0.0-20180906133057-8cf3aee42992/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181004145325-8469e314837c h1:SJ7JoQNVl3mC7EWkkONgBWgCno8LcABIJwFMkWBC+EY=
 golang.org/x/sys v0.0.0-20181004145325-8469e314837c/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20181005133103-4497e2df6f9e h1:EfdBzeKbFSvOjoIqSZcfS8wp0FBLokGBEs9lz1OtSg0=
+golang.org/x/sys v0.0.0-20181005133103-4497e2df6f9e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2 h1:+DCIGbF/swA92ohVg0//6X2IVY3KZs6p9mix0ziNYJM=

pkg/storage/streams/store.go

@@ -174,12 +174,17 @@ func (s *streamStore) Put(ctx context.Context, path paths.Path, data io.Reader,
 		}
 
 		putMeta, err = s.segments.Put(ctx, transformedReader, expiration, func() (paths.Path, []byte, error) {
+			encPath, err := encryptAfterBucket(path, s.rootKey)
+			if err != nil {
+				return nil, nil, err
+			}
+
 			if !eofReader.isEOF() {
-				segmentPath := getSegmentPath(path, currentSegment)
+				segmentPath := getSegmentPath(encPath, currentSegment)
 				return segmentPath, encryptedEncKey, nil
 			}
 
-			lastSegmentPath := path.Prepend("l")
+			lastSegmentPath := encPath.Prepend("l")
 			msi := pb.MetaStreamInfo{
 				NumberOfSegments:         currentSegment + 1,
 				SegmentsSize:             s.segmentSize,
@@ -227,7 +232,12 @@ func getSegmentPath(p paths.Path, segNum int64) paths.Path {
 func (s *streamStore) Get(ctx context.Context, path paths.Path) (rr ranger.Ranger, meta Meta, err error) {
 	defer mon.Task()(&ctx)(&err)
 
-	lastSegmentRanger, lastSegmentMeta, err := s.segments.Get(ctx, path.Prepend("l"))
+	encPath, err := encryptAfterBucket(path, s.rootKey)
+	if err != nil {
+		return nil, Meta{}, err
+	}
+
+	lastSegmentRanger, lastSegmentMeta, err := s.segments.Get(ctx, encPath.Prepend("l"))
 	if err != nil {
 		return nil, Meta{}, err
 	}
@@ -250,7 +260,7 @@ func (s *streamStore) Get(ctx context.Context, path paths.Path) (rr ranger.Range
 
 	var rangers []ranger.Ranger
 	for i := int64(0); i < msi.NumberOfSegments-1; i++ {
-		currentPath := getSegmentPath(path, i)
+		currentPath := getSegmentPath(encPath, i)
 		size := msi.SegmentsSize
 		var nonce eestream.Nonce
 		_, err := nonce.Increment(i)
@@ -295,8 +305,15 @@ func (s *streamStore) Get(ctx context.Context, path paths.Path) (rr ranger.Range
 }
 
 // Meta implements Store.Meta
-func (s *streamStore) Meta(ctx context.Context, path paths.Path) (Meta, error) {
-	lastSegmentMeta, err := s.segments.Meta(ctx, path.Prepend("l"))
+func (s *streamStore) Meta(ctx context.Context, path paths.Path) (meta Meta, err error) {
+	defer mon.Task()(&ctx)(&err)
+
+	encPath, err := encryptAfterBucket(path, s.rootKey)
+	if err != nil {
+		return Meta{}, err
+	}
+
+	lastSegmentMeta, err := s.segments.Meta(ctx, encPath.Prepend("l"))
 	if err != nil {
 		return Meta{}, err
 	}
@@ -313,7 +330,11 @@ func (s *streamStore) Meta(ctx context.Context, path paths.Path) (Meta, error) {
 func (s *streamStore) Delete(ctx context.Context, path paths.Path) (err error) {
 	defer mon.Task()(&ctx)(&err)
 
-	lastSegmentMeta, err := s.segments.Meta(ctx, path.Prepend("l"))
+	encPath, err := encryptAfterBucket(path, s.rootKey)
+	if err != nil {
+		return err
+	}
+	lastSegmentMeta, err := s.segments.Meta(ctx, encPath.Prepend("l"))
 	if err != nil {
 		return err
 	}
@@ -325,14 +346,18 @@ func (s *streamStore) Delete(ctx context.Context, path paths.Path) (err error) {
 	}
 
 	for i := 0; i < int(msi.NumberOfSegments-1); i++ {
-		currentPath := getSegmentPath(path, int64(i))
+		encPath, err = encryptAfterBucket(path, s.rootKey)
+		if err != nil {
+			return err
+		}
+		currentPath := getSegmentPath(encPath, int64(i))
 		err := s.segments.Delete(ctx, currentPath)
 		if err != nil {
 			return err
 		}
 	}
 
-	return s.segments.Delete(ctx, path.Prepend("l"))
+	return s.segments.Delete(ctx, encPath.Prepend("l"))
 }
 
 // ListItem is a single item in a listing
@@ -352,7 +377,26 @@ func (s *streamStore) List(ctx context.Context, prefix, startAfter, endBefore pa
 		metaFlags |= meta.UserDefined
 	}
 
-	segments, more, err := s.segments.List(ctx, prefix.Prepend("l"), startAfter, endBefore, recursive, limit, metaFlags)
+	encPrefix, err := encryptAfterBucket(prefix, s.rootKey)
+	if err != nil {
+		return nil, false, err
+	}
+
+	prefixKey, err := prefix.DeriveKey(s.rootKey, len(prefix))
+	if err != nil {
+		return nil, false, err
+	}
+
+	encStartAfter, err := s.encryptMarker(startAfter, prefixKey)
+	if err != nil {
+		return nil, false, err
+	}
+	encEndBefore, err := s.encryptMarker(endBefore, prefixKey)
+	if err != nil {
+		return nil, false, err
+	}
+
+	segments, more, err := s.segments.List(ctx, encPrefix.Prepend("l"), encStartAfter, encEndBefore, recursive, limit, metaFlags)
 	if err != nil {
 		return nil, false, err
 	}
@@ -363,12 +407,32 @@ func (s *streamStore) List(ctx context.Context, prefix, startAfter, endBefore pa
 		if err != nil {
 			return nil, false, err
 		}
-		items[i] = ListItem{Path: item.Path, Meta: newMeta, IsPrefix: item.IsPrefix}
+		decPath, err := s.decryptMarker(item.Path, prefixKey)
+		if err != nil {
+			return nil, false, err
+		}
+		items[i] = ListItem{Path: decPath, Meta: newMeta, IsPrefix: item.IsPrefix}
 	}
 
 	return items, more, nil
 }
 
+// encryptMarker is a helper method for encrypting startAfter and endBefore markers
+func (s *streamStore) encryptMarker(marker paths.Path, prefixKey []byte) (paths.Path, error) {
+	if bytes.Equal(s.rootKey, prefixKey) { // empty prefix
+		return encryptAfterBucket(marker, s.rootKey)
+	}
+	return marker.Encrypt(prefixKey)
+}
+
+// decryptMarker is a helper method for decrypting listed path markers
+func (s *streamStore) decryptMarker(marker paths.Path, prefixKey []byte) (paths.Path, error) {
+	if bytes.Equal(s.rootKey, prefixKey) { // empty prefix
+		return decryptAfterBucket(marker, s.rootKey)
+	}
+	return marker.Decrypt(prefixKey)
+}
+
 type lazySegmentRanger struct {
 	ranger        ranger.Ranger
 	segments      segments.Store
@@ -437,6 +501,45 @@ func decryptRanger(ctx context.Context, rr ranger.Ranger, decryptedSize int64, c
 	return eestream.Unpad(rd, int(rd.Size()-decryptedSize))
 }
 
+// encryptAfterBucket encrypts a path without encrypting its first element
+func encryptAfterBucket(p paths.Path, key []byte) (encrypted paths.Path, err error) {
+	if len(p) <= 1 {
+		return p, nil
+	}
+	bucket := p[0]
+	toEncrypt := p[1:]
+
+	// derive a key from the bucket so the same path in different buckets is encrypted differently
+	bucketKey, err := p.DeriveKey(key, 1)
+	if err != nil {
+		return nil, err
+	}
+	encPath, err := toEncrypt.Encrypt(bucketKey)
+	if err != nil {
+		return nil, err
+	}
+	return encPath.Prepend(bucket), nil
+}
+
+// decryptAfterBucket decrypts a path without modifying its first element
+func decryptAfterBucket(p paths.Path, key []byte) (decrypted paths.Path, err error) {
+	if len(p) <= 1 {
+		return p, nil
+	}
+	bucket := p[0]
+	toDecrypt := p[1:]
+
+	bucketKey, err := p.DeriveKey(key, 1)
+	if err != nil {
+		return nil, err
+	}
+	decPath, err := toDecrypt.Decrypt(bucketKey)
+	if err != nil {
+		return nil, err
+	}
+	return decPath.Prepend(bucket), nil
+}
+
 // CancelHandler handles clean up of segments on receiving CTRL+C
 func (s *streamStore) cancelHandler(ctx context.Context, totalSegments int64, path paths.Path) {
 	for i := int64(0); i < totalSegments; i++ {