From 3469905bbb0f165911123bc8f32863c3db340fec Mon Sep 17 00:00:00 2001 From: Stephen J Day Date: Thu, 3 Nov 2016 17:02:34 -0700 Subject: [PATCH] content: break up into multiple files Break up the content store prototype into a few logical files. We have a file for the store, the writer and helpers. Also, the writer has been modified to remove write and exec permissions on blobs in the store. Signed-off-by: Stephen J Day --- content/content.go | 100 ---------------------------------------- content/content_test.go | 30 +++++++++--- content/helpers.go | 53 +++++++++++++++++++++ content/writer.go | 92 ++++++++++++++++++++++++++++++++++++ 4 files changed, 168 insertions(+), 107 deletions(-) create mode 100644 content/helpers.go create mode 100644 content/writer.go diff --git a/content/content.go b/content/content.go index 202707c01..ecb8b9a72 100644 --- a/content/content.go +++ b/content/content.go @@ -40,50 +40,6 @@ func OpenContentStore(root string) (*ContentStore, error) { }, nil } -// OpenBlob opens the blob for reading identified by dgst. -// -// The opened blob may also implement seek. Callers can detect with io.Seeker. -func OpenBlob(cs *ContentStore, dgst digest.Digest) (io.ReadCloser, error) { - path, err := cs.GetPath(dgst) - if err != nil { - return nil, err - } - - fp, err := os.Open(path) - return fp, err -} - -// WriteBlob writes data with the expected digest into the content store. If -// expected already exists, the method returns immediately and the reader will -// not be consumed. -// -// This is useful when the digest and size are known beforehand. -// -// Copy is buffered, so no need to wrap reader in buffered io. -func WriteBlob(cs *ContentStore, r io.Reader, size int64, expected digest.Digest) error { - cw, err := cs.Begin(expected.Hex()) - if err != nil { - return err - } - buf := bufPool.Get().([]byte) - defer bufPool.Put(buf) - - nn, err := io.CopyBuffer(cw, r, buf) - if err != nil { - return err - } - - if nn != size { - return errors.Errorf("failed size verification: %v != %v", nn, size) - } - - if err := cw.Commit(size, expected); err != nil { - return err - } - - return nil -} - func (cs *ContentStore) GetPath(dgst digest.Digest) (string, error) { p := filepath.Join(cs.root, "blobs", dgst.Algorithm().String(), dgst.Hex()) if _, err := os.Stat(p); err != nil { @@ -190,59 +146,3 @@ func (cs *ContentStore) ingestPaths(ref string) (string, string, error) { return fp, filepath.Join(fp, "data"), nil } - -// ContentWriter represents a write transaction against the blob store. -// -// -type ContentWriter struct { - cs *ContentStore - fp *os.File // opened data file - path string // path to writer dir - offset int64 - digester digest.Digester -} - -func (cw *ContentWriter) Write(p []byte) (n int, err error) { - n, err = cw.fp.Write(p) - cw.digester.Hash().Write(p[:n]) - return n, err -} - -func (cw *ContentWriter) Commit(size int64, expected digest.Digest) error { - if err := cw.fp.Sync(); err != nil { - return errors.Wrap(err, "sync failed") - } - - fi, err := cw.fp.Stat() - if err != nil { - return errors.Wrap(err, "stat on data file failed") - } - - if size != fi.Size() { - return errors.Errorf("failed size validation: %v != %v", fi.Size(), size) - } - - dgst := cw.digester.Digest() - if expected != dgst { - return errors.Errorf("unexpected digest: %v != %v", dgst, expected) - } - - apath := filepath.Join(cw.cs.root, "blobs", dgst.Algorithm().String()) - if err := os.MkdirAll(apath, 0755); err != nil { - return err - } - - dpath := filepath.Join(apath, dgst.Hex()) - - // clean up!! - defer os.RemoveAll(cw.path) - return os.Rename(filepath.Join(cw.path, "data"), dpath) -} - -// Close the writer, leaving the progress in tact. -// -// If one needs to resume the transaction, a new writer can be obtained from -// `ContentStore.Resume` using the same key. -func (cw *ContentWriter) Close() error { - return cw.fp.Close() -} diff --git a/content/content_test.go b/content/content_test.go index 98a6a57db..4d7eeee90 100644 --- a/content/content_test.go +++ b/content/content_test.go @@ -4,6 +4,7 @@ import ( "bufio" "bytes" "crypto/rand" + _ "crypto/sha256" // required for digest package "fmt" "io" "io/ioutil" @@ -73,13 +74,7 @@ func TestContentWriter(t *testing.T) { t.Fatal(err) } - path, err := cs.GetPath(expected) - if err != nil { - t.Fatal(err) - } - if path != filepath.Join(tmpdir, "blobs", expected.Algorithm().String(), expected.Hex()) { - t.Fatalf("unxpected path: %q", path) - } + path := checkBlobPath(t, cs, expected) // read the data back, make sure its the same pp, err := ioutil.ReadFile(path) @@ -105,6 +100,27 @@ func checkCopy(t *testing.T, size int64, dst io.Writer, src io.Reader) { } } +func checkBlobPath(t *testing.T, cs *ContentStore, dgst digest.Digest) string { + path, err := cs.GetPath(dgst) + if err != nil { + t.Fatal(err) + } + if path != filepath.Join(cs.root, "blobs", dgst.Algorithm().String(), dgst.Hex()) { + t.Fatalf("unxpected path: %q", path) + } + fi, err := os.Stat(path) + if err != nil { + t.Fatalf("error stating blob path: %v", err) + } + + // ensure that only read bits are set. + if ((fi.Mode() & os.ModePerm) & 0333) != 0 { + t.Fatalf("incorrect permissions: %v", fi.Mode()) + } + + return path +} + func dumpDir(root string) error { return filepath.Walk(root, func(path string, fi os.FileInfo, err error) error { if err != nil { diff --git a/content/helpers.go b/content/helpers.go new file mode 100644 index 000000000..65f13e759 --- /dev/null +++ b/content/helpers.go @@ -0,0 +1,53 @@ +package content + +import ( + "io" + "os" + + "github.com/docker/distribution/digest" + "github.com/pkg/errors" +) + +// OpenBlob opens the blob for reading identified by dgst. +// +// The opened blob may also implement seek. Callers can detect with io.Seeker. +func OpenBlob(cs *ContentStore, dgst digest.Digest) (io.ReadCloser, error) { + path, err := cs.GetPath(dgst) + if err != nil { + return nil, err + } + + fp, err := os.Open(path) + return fp, err +} + +// WriteBlob writes data with the expected digest into the content store. If +// expected already exists, the method returns immediately and the reader will +// not be consumed. +// +// This is useful when the digest and size are known beforehand. +// +// Copy is buffered, so no need to wrap reader in buffered io. +func WriteBlob(cs *ContentStore, r io.Reader, size int64, expected digest.Digest) error { + cw, err := cs.Begin(expected.Hex()) + if err != nil { + return err + } + buf := bufPool.Get().([]byte) + defer bufPool.Put(buf) + + nn, err := io.CopyBuffer(cw, r, buf) + if err != nil { + return err + } + + if nn != size { + return errors.Errorf("failed size verification: %v != %v", nn, size) + } + + if err := cw.Commit(size, expected); err != nil { + return err + } + + return nil +} diff --git a/content/writer.go b/content/writer.go new file mode 100644 index 000000000..83b2ed658 --- /dev/null +++ b/content/writer.go @@ -0,0 +1,92 @@ +package content + +import ( + "os" + "path/filepath" + + "github.com/docker/distribution/digest" + "github.com/pkg/errors" +) + +// ContentWriter represents a write transaction against the blob store. +// +// +type ContentWriter struct { + cs *ContentStore + fp *os.File // opened data file + path string // path to writer dir + offset int64 + digester digest.Digester +} + +// Write p to the transaction. +// +// Note that writes are unbuffered to the backing file. When writing, it is +// recommended to wrap in a bufio.Writer or, preferrably, use io.CopyBuffer. +func (cw *ContentWriter) Write(p []byte) (n int, err error) { + n, err = cw.fp.Write(p) + cw.digester.Hash().Write(p[:n]) + return n, err +} + +func (cw *ContentWriter) Commit(size int64, expected digest.Digest) error { + if err := cw.fp.Sync(); err != nil { + return errors.Wrap(err, "sync failed") + } + + fi, err := cw.fp.Stat() + if err != nil { + return errors.Wrap(err, "stat on ingest file failed") + } + + // change to readonly, more important for read, but provides _some_ + // protection from this point on. We use the existing perms with a mask + // only allowing reads honoring the umask on creation. + // + // This removes write and exec, only allowing read per the creation umask. + if err := cw.fp.Chmod((fi.Mode() & os.ModePerm) &^ 0333); err != nil { + return errors.Wrap(err, "failed to change ingest file permissions") + } + + if size != fi.Size() { + return errors.Errorf("failed size validation: %v != %v", fi.Size(), size) + } + + dgst := cw.digester.Digest() + if expected != dgst { + return errors.Errorf("unexpected digest: %v != %v", dgst, expected) + } + + apath := filepath.Join(cw.cs.root, "blobs", dgst.Algorithm().String()) + if err := os.MkdirAll(apath, 0755); err != nil { + return err + } + + var ( + ingest = filepath.Join(cw.path, "data") + target = filepath.Join(apath, dgst.Hex()) + ) + + // clean up!! + defer os.RemoveAll(cw.path) + if err := os.Rename(ingest, target); err != nil { + if os.IsExist(err) { + // collision with the target file! + return nil + } + return err + } + + return nil +} + +// Close the writer, flushing any unwritten data and leaving the progress in +// tact. +// +// If one needs to resume the transaction, a new writer can be obtained from +// `ContentStore.Resume` using the same key. The write can then be continued +// from it was left off. +func (cw *ContentWriter) Close() (err error) { + cw.fp.Sync() + return cw.fp.Close() +}