diff --git a/.gitignore b/.gitignore index a521cc9..1e193a8 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ /cachefs +/cachectl /metadatagen /run.sh diff --git a/Makefile b/Makefile index 56fc960..9331c3a 100644 --- a/Makefile +++ b/Makefile @@ -7,9 +7,11 @@ PROGRAM=cachefs CMD=-o $(PROGRAM) cmd/$(PROGRAM)/main.go MD_PROGRAM=metadatagen MD_CMD=-o $(MD_PROGRAM) cmd/$(MD_PROGRAM)/main.go +CTL_PROGRAM=cachectl +CTL_CMD=-o $(CTL_PROGRAM) cmd/$(CTL_PROGRAM)/main.go ENV=CGO_ENABLED=0 GO111MODULE=on -all: $(PROGRAM) $(MD_PROGRAM) +all: $(PROGRAM) $(MD_PROGRAM) $(CTL_PROGRAM) setup: $(CC) install github.com/client9/misspell/cmd/misspell@latest @@ -30,6 +32,9 @@ $(PROGRAM): clean gofmt $(MD_PROGRAM): clean gofmt $(ENV) $(CC) $(BUILD) $(GCFLAGS) $(LDFLAGS) $(MD_CMD) +$(CTL_PROGRAM): clean gofmt + $(ENV) $(CC) $(BUILD) $(GCFLAGS) $(LDFLAGS) $(CTL_CMD) + clean: $(CC) clean -x diff --git a/TODO.txt b/TODO.txt index 49f271b..90a2f37 100644 --- a/TODO.txt +++ b/TODO.txt @@ -3,6 +3,10 @@ * crypto * flush last chunk in streaming mode + * curl download does not finish - missing EOF?? + +* sftp + * implement read ahead buffer * add breadcrumb * add copyright footer diff --git a/cmd/cachectl/main.go b/cmd/cachectl/main.go new file mode 100644 index 0000000..9bf2b32 --- /dev/null +++ b/cmd/cachectl/main.go @@ -0,0 +1,196 @@ +// Copyright (C) 2022 Marius Schellenberger + +package main + +import ( + "cachefs/pkg/provider/parse" + "encoding/hex" + "errors" + "flag" + "fmt" + "io" + "os" + "path/filepath" + "strings" +) + +var ( + version string +) + +const gib = 1024 * 1024 * 1024 + +const ( + srcKey = "SRC_KEY" + dstKey = "DST_KEY" +) + +func fatal(err error) { + fmt.Println(err) + os.Exit(1) +} + +func getKey(key string) (b []byte, err error) { + k := os.Getenv(key) + if k == "" { + return + } + b, err = hex.DecodeString(k) + if err != nil { + err = fmt.Errorf("error decoding %s: %w", key, err) + } + return +} + +func main() { + flag.Usage = func() { + fmt.Fprintf(flag.CommandLine.Output(), "Usage of %s:\n", os.Args[0]) + fmt.Fprintf(flag.CommandLine.Output(), ` +optional environment variables: + + `+srcKey+` (source encryption key) + `+dstKey+` (destination encryption key) + +list files and directories (uses `+srcKey+`): + ls [ ] + + ls file:///mnt/nfs /foo + +copy file from/to directory: + cp + + cp file:///mnt/nfs foo.txt file:///mnt/cache bar.txt + + cp file:///mnt/cache fizz.txt sftp:///home/user fizz.txt + +remove file (uses `+srcKey+`): + rm + + rm file:///mnt/nfs /buzz.txt + +`) + os.Exit(1) + } + flag.Parse() + + args := flag.Args() + if len(args) < 2 { + flag.Usage() + } + + var err error + switch args[0] { + case "ls": + err = ls(args[1:]) + case "cp": + if len(args) < 5 { + flag.Usage() + } + err = cp(args[1:]) + case "rm": + if len(args) < 3 { + flag.Usage() + } + err = rm(args[1:]) + } + if err != nil { + fatal(err) + } +} + +func ls(args []string) (err error) { + k, err := getKey(srcKey) + if err != nil { + return + } + fs, err := parse.FS(args[0], k, true) + if err != nil { + return + } + defer fs.Close() + path := "" + if len(args) == 2 { + path = args[1] + } + f, err := fs.Open(path) + if err != nil { + return + } + defer f.Close() + fis, err := f.Readdir(0) + if err != nil { + return + } + for _, fi := range fis { + p := filepath.Join(fs.Root(), path, fi.Name()) + if fi.IsDir() { + fmt.Println(fi.Mode().String(), fi.Size(), p+"/") + continue + } + fmt.Println(fi.Mode().String(), fi.Size(), p) + } + return +} + +func cp(args []string) (err error) { + src := args[0] + sp := args[1] + dst := args[2] + dp := args[3] + sk, err := getKey(srcKey) + if err != nil { + return + } + dk, err := getKey(dstKey) + if err != nil { + return + } + sfs, err := parse.FS(src, sk, true) + if err != nil { + return + } + defer sfs.Close() + dfs, err := parse.FS(dst, dk, true) + if err != nil { + return + } + defer dfs.Close() + + sf, err := sfs.Open(sp) + if err != nil { + return + } + defer sf.Close() + if i := strings.LastIndex(dp, "/"); i > 0 { + dir := dp[:i] + _, err = dfs.Stat(dir) + if errors.Is(err, os.ErrNotExist) { + err = dfs.MkdirAll(dir, 0o755) + if err != nil { + return + } + } + } + df, err := dfs.OpenFile(dp, os.O_RDWR|os.O_CREATE, 0o644) + if err != nil { + return + } + defer df.Close() + _, err = io.Copy(df, sf) + return +} + +func rm(args []string) (err error) { + src := args[0] + path := args[1] + k, err := getKey(srcKey) + if err != nil { + return + } + fs, err := parse.FS(src, k, true) + if err != nil { + return + } + defer fs.Close() + return fs.Remove(path) +} diff --git a/cmd/cachefs/main.go b/cmd/cachefs/main.go index 8a23ec9..1f28261 100644 --- a/cmd/cachefs/main.go +++ b/cmd/cachefs/main.go @@ -24,8 +24,6 @@ var ( src string dst string - srckey string - dstkey string metadata string listenHttp string listenWebdav string @@ -33,6 +31,7 @@ var ( quota int64 max int bs int + block bool log logr.Logger ) @@ -42,16 +41,15 @@ const gib = 1024 * 1024 * 1024 func main() { klog.InitFlags(nil) flag.StringVar(&src, "src", "", "url path to source files (example: file:///mnt/nfs)") - flag.StringVar(&srckey, "srckey", "", "hex encoded encryption key for src") flag.StringVar(&dst, "dst", "", "url path to cache files (example: file:///mnt/cache)") - flag.StringVar(&dstkey, "dstkey", "", "hex encoded encryption key for dst") flag.StringVar(&metadata, "data", "", "path to metadata file") flag.StringVar(&listenHttp, "listen", "127.0.0.1:8080", "listen addr:port") flag.StringVar(&listenWebdav, "webdav", "", "listen addr:port for webdav") flag.StringVar(&listenCache, "cache", "", "listen addr:port for cache only") flag.IntVar(&max, "max", -1, "max parallel preloads") - flag.IntVar(&bs, "bs", -1, "tune reload buffer size in bytes (default: 8192)") - flag.Int64Var("a, "quota", 1, "max disk usage quota in GiB") + flag.IntVar(&bs, "bs", -1, "tune preload buffer size in bytes (default: 8192)") + flag.Int64Var("a, "quota", 1, "max disk usage quota for the dst cache in GiB") + flag.BoolVar(&block, "block", false, "block until sftp is connected") flag.Parse() log = klogr.New().WithName("main") @@ -65,10 +63,11 @@ func main() { quota*gib, max, src, - srckey, + os.Getenv("SRC_KEY"), dst, - dstkey, + os.Getenv("DST_KEY"), metadata, + block, klogr.New().WithName("fs"), ) if err != nil { diff --git a/cmd/metadatagen/main.go b/cmd/metadatagen/main.go index 785bd52..308db1a 100644 --- a/cmd/metadatagen/main.go +++ b/cmd/metadatagen/main.go @@ -16,13 +16,11 @@ var ( version string dst string - dstkey string metadata string ) func main() { flag.StringVar(&dst, "dst", "", "url path to cache files (example: file:///mnt/cache)") - flag.StringVar(&dstkey, "dstkey", "", "hex encoded encryption key for dst") flag.StringVar(&metadata, "data", "", "path to metadata file") flag.Parse() @@ -30,12 +28,13 @@ func main() { dstk []byte err error ) + dstkey := os.Getenv("DST_KEY") if dstkey != "" { dstk, err = hex.DecodeString(dstkey) fmt.Println(err) os.Exit(1) } - dstfs, err := parse.FS(dst, dstk) + dstfs, err := parse.FS(dst, dstk, true) if err != nil { fmt.Println(err) os.Exit(1) diff --git a/pkg/fs/file.go b/pkg/fs/file.go index 9f006c5..4cb7380 100644 --- a/pkg/fs/file.go +++ b/pkg/fs/file.go @@ -59,6 +59,12 @@ func (f *File) Preload(ctx context.Context, unlock func()) { log.V(2).Info("skipped preload for fully cached file") return } + + // TODO + //fi, _ := f.Stat() + //s := fi.Size() + //log.V(2).Info("preload started", "size", s, "crypto.Size", crypto.Size(s), "crypto.RealSize", crypto.RealSize(crypto.Size(s))) + log.V(2).Info("preload started") p := &preload{f: f, ctx: ctx} _, err := io.Copy(Discard, p) @@ -69,7 +75,10 @@ func (f *File) Preload(ctx context.Context, unlock func()) { if err != nil && err != io.EOF { log.Error(err, "error preloading file") } - f.md.Close() + err = f.md.Close() + if err != nil { + log.Error(err, "error closing cache file") + } log.V(2).Info("preload finished", "skipped", p.skipped, "written", p.written) } @@ -77,6 +86,9 @@ func (f *File) Read(p []byte) (n int, err error) { log := f.log if f.hasChunk(len(p)) { n, err = f.md.ReadAt(p, f.offset) + if err == io.EOF { + return + } if err != nil { if !IsIOErr(err) { log.Error(err, "error reading cache file") diff --git a/pkg/fs/fs.go b/pkg/fs/fs.go index d0b51c5..8a8c76e 100644 --- a/pkg/fs/fs.go +++ b/pkg/fs/fs.go @@ -32,7 +32,7 @@ type FS struct { ph *PreloadHandler } -func NewFS(quota int64, max int, src, srckey, dst, dstkey, metadata string, log logr.Logger) (fs *FS, err error) { +func NewFS(quota int64, max int, src, srckey, dst, dstkey, metadata string, block bool, log logr.Logger) (fs *FS, err error) { if src == dst { return nil, errors.New("src and dst path can not be equal") } @@ -42,18 +42,20 @@ func NewFS(quota int64, max int, src, srckey, dst, dstkey, metadata string, log if err != nil { return nil, fmt.Errorf("srckey: %w", err) } + log.V(2).Info("using source encryption") } if dstkey != "" { dstk, err = hex.DecodeString(dstkey) if err != nil { return nil, fmt.Errorf("dstkey: %w", err) } + log.V(2).Info("using destination encryption") } - srcfs, err := parse.FS(src, srck) + srcfs, err := parse.FS(src, srck, block) if err != nil { return nil, fmt.Errorf("srcfs: %w", err) } - dstfs, err := parse.FS(dst, dstk) + dstfs, err := parse.FS(dst, dstk, block) if err != nil { return nil, fmt.Errorf("dstfs: %w", err) } diff --git a/pkg/provider/crypto/file.go b/pkg/provider/crypto/file.go index 4bb3254..8e58b42 100644 --- a/pkg/provider/crypto/file.go +++ b/pkg/provider/crypto/file.go @@ -4,6 +4,7 @@ import ( "cachefs/pkg/provider" "io" "io/fs" + "os" ) var _ provider.File = (*file)(nil) @@ -16,6 +17,7 @@ type file struct { nonce [NonceSize]byte offset int64 name string + stat *stat dir bool } @@ -41,41 +43,81 @@ func newFile(key []byte, name string, fs *FS, f provider.File, dir bool) (ef pro if err != nil { return } - return &file{ + ef = &file{ File: f, fs: fs, w: w, r: r, name: name, - }, nil + } + return ef, nil } func (f *file) Stat() (fs.FileInfo, error) { + if f.stat != nil { + return f.stat, nil + } fi, err := f.File.Stat() - return stat{FileInfo: fi, name: f.name}, err + if err != nil { + return nil, err + } + if f.dir { + return dir{FileInfo: fi, name: f.name}, nil + } + + rs := RealSize(fi.Size()) + f.stat = &stat{ + FileInfo: fi, + size: rs, + name: f.name, + } + return f.stat, nil + + //fmt.Println("rs", rs) + //_, err = f.ReadAt(nil, rs) + //if err != nil && err != io.EOF { + // return nil, err + //} + //f.stat = &stat{ + // FileInfo: fi, + // size: rs - (ChunkSize - int64(len(f.r.unread))), + // name: f.name, + //} + //fmt.Println("size", f.stat.Size()) + //f.Seek(0, io.SeekStart) + //return f.stat, nil } func (f *file) Name() string { return f.name } -func (file) Fd() uintptr { +func (f *file) Fd() uintptr { + osf, ok := f.File.(*os.File) + if ok { + return osf.Fd() + } return 0 } type stat struct { fs.FileInfo + size int64 name string } func (s stat) Size() int64 { - return RealSize(s.FileInfo.Size()) + return s.size } func (s stat) Name() string { return s.name } +func (f *file) Truncate(n int64) error { + return f.File.Truncate(Size(n)) +} + func (f *file) Close() error { if f.dir { return f.File.Close() @@ -104,16 +146,23 @@ func (f *file) Seek(offset int64, whence int) (n int64, err error) { _, err = f.File.Seek(f.offset, io.SeekStart) n = offset } + f.reset() + if err != nil { + return + } f.nonce = [NonceSize]byte{} inc(&f.nonce, cn) f.r.nonce = f.nonce - f.r.off = roff - f.r.err = nil f.w.nonce = f.nonce - f.r.unread = f.r.unread[:0] + f.r.off = roff return } +func (f *file) reset() { + f.r.err = nil + f.r.unread = f.r.unread[:0] +} + func (f *file) Read(p []byte) (n int, err error) { n, err = f.r.Read(p) return @@ -125,21 +174,28 @@ func (f *file) Write(p []byte) (n int, err error) { } func (f *file) ReadAt(p []byte, pos int64) (n int, err error) { - _, _, roff := align(pos) - //cn, off, roff := align(pos) - f.Seek(pos, io.SeekStart) - //fmt.Println("readAt:", cn, off+KDFNonceSize, roff, pos) - _, err = f.r.readChunk() - if err != nil { - return + cn, _, roff := align(pos) + var last bool + if f.r.cn != cn { + _, err = f.Seek(pos, io.SeekStart) + if err != nil { + return + } + last, err = f.r.readChunk() + if err != nil { + return + } + f.r.cn = cn } n = copy(p, f.r.unread[roff:]) + if last && len(f.r.unread) == n+int(roff) { + err = io.EOF + } return } func (f *file) WriteAt(data []byte, pos int64) (n int, err error) { cn, off, woff := align(pos) - //fmt.Println("writeAt:", cn, off+KDFNonceSize, woff, pos) c, ok := f.w.cm[cn] if !ok { c = &chunkWriter{offset: off + KDFNonceSize, w: f.w, cn: cn} @@ -153,8 +209,11 @@ func (f *file) WriteAt(data []byte, pos int64) (n int, err error) { } n, err = c.writeAt(data, woff) if c.full() { - _, err = c.flush(notLastChunk) - //fmt.Println("flush:", cn) + // async flush + go func() { + //_, err = c.flush(notLastChunk) + c.flush(notLastChunk) + }() delete(f.w.cm, cn) } return @@ -171,7 +230,7 @@ func (f *file) Readdir(n int) ([]fs.FileInfo, error) { if err != nil { return nil, err } - fio[i] = dir{FileInfo: fi, name: name} + fio[i] = stat{FileInfo: fi, size: RealSize(fi.Size()), name: name} } return fio, nil } diff --git a/pkg/provider/crypto/fs.go b/pkg/provider/crypto/fs.go index 6f25890..16cc592 100644 --- a/pkg/provider/crypto/fs.go +++ b/pkg/provider/crypto/fs.go @@ -2,6 +2,7 @@ package crypto import ( "cachefs/pkg/provider" + osp "cachefs/pkg/provider/os" "encoding/base64" "errors" "io/fs" @@ -102,7 +103,11 @@ func (fs *FS) Root() string { return fs.fs.Root() } -func (fs *FS) Fstat(_ uintptr) (int64, error) { +func (fs *FS) Fstat(fd uintptr) (int64, error) { + osp, ok := fs.fs.(*osp.FS) + if ok { + return osp.Fstat(fd) + } return 0, provider.ErrFstat } diff --git a/pkg/provider/crypto/reader.go b/pkg/provider/crypto/reader.go index 4c70b65..ba89ea8 100644 --- a/pkg/provider/crypto/reader.go +++ b/pkg/provider/crypto/reader.go @@ -3,6 +3,7 @@ package crypto import ( "crypto/cipher" "errors" + "fmt" "io" "golang.org/x/crypto/chacha20poly1305" @@ -23,6 +24,7 @@ type reader struct { err error nonce [NonceSize]byte + cn int64 } func newReader(key []byte, src cryptoReader) (*reader, error) { @@ -33,6 +35,7 @@ func newReader(key []byte, src cryptoReader) (*reader, error) { return &reader{ a: aead, src: src, + cn: -1, }, nil } @@ -84,7 +87,9 @@ func (r *reader) readChunk() (last bool, err error) { // The last chunk can be short. in = in[:n] last = true + fmt.Println("last nonce", r.nonce) setLastChunkFlag(&r.nonce) + fmt.Println("last nonce flag", r.nonce) case err != nil: return false, err } diff --git a/pkg/provider/crypto/writer.go b/pkg/provider/crypto/writer.go index e2d943e..1efa359 100644 --- a/pkg/provider/crypto/writer.go +++ b/pkg/provider/crypto/writer.go @@ -4,6 +4,7 @@ import ( "cachefs/pkg/chunk" "crypto/cipher" "errors" + "fmt" "io" "golang.org/x/crypto/chacha20poly1305" @@ -71,7 +72,9 @@ func (c *chunkWriter) full() bool { func (c *chunkWriter) flush(last bool) (n int, err error) { wb := c.buf[:0] if last { + fmt.Println("last nonce", c.nonce) setLastChunkFlag(&c.nonce) + fmt.Println("last nonce flag", c.nonce) s := c.chunks.Size() wb = c.buf[:s] n = int(s) @@ -112,15 +115,19 @@ func (w *writer) Write(p []byte) (n int, err error) { } func (w *writer) Close() error { + fmt.Println("writer closed", w.err) if w.err != nil { return w.err } if w.last != nil { + fmt.Println("flushing last chunk") _, w.err = w.last.flush(lastChunk) if w.err != nil { + fmt.Println("error flushing last chunk", w.err) return w.err } + w.flush = false } if w.flush { diff --git a/pkg/provider/parse/parse.go b/pkg/provider/parse/parse.go index 147ec58..7d8b86b 100644 --- a/pkg/provider/parse/parse.go +++ b/pkg/provider/parse/parse.go @@ -20,7 +20,23 @@ var ( ErrUnsupportedScheme = errors.New("unsupported scheme") ) -func FS(url string, key []byte) (provider.FS, error) { +func TryParse(url string) error { + u, err := neturl.Parse(url) + if err != nil { + return err + } + if !filepath.IsAbs(u.Path) { + return ErrPathNotAbsolute + } + switch u.Scheme { + case "file", "sftp": + default: + return fmt.Errorf("%w: %s", ErrUnsupportedScheme, u.Scheme) + } + return nil +} + +func FS(url string, key []byte, block bool) (provider.FS, error) { u, err := neturl.Parse(url) if err != nil { return nil, err @@ -33,7 +49,7 @@ func FS(url string, key []byte) (provider.FS, error) { case "file": fs, err = os.NewFS(u.Path) case "sftp": - fs, err = sftp.NewFS(u, klogr.New().WithName("sftp")) + fs, err = sftp.NewFS(u, block, klogr.New().WithName("sftp")) default: return nil, fmt.Errorf("%w: %s", ErrUnsupportedScheme, u.Scheme) } diff --git a/pkg/provider/sftp/sftp.go b/pkg/provider/sftp/sftp.go index b0756d6..07397b5 100644 --- a/pkg/provider/sftp/sftp.go +++ b/pkg/provider/sftp/sftp.go @@ -51,7 +51,7 @@ type FS struct { keepalive bool } -func NewFS(u *url.URL, log logr.Logger) (fs *FS, err error) { +func NewFS(u *url.URL, block bool, log logr.Logger) (fs *FS, err error) { cfg, err := sshConfig(u) if err != nil { return @@ -66,6 +66,14 @@ func NewFS(u *url.URL, log logr.Logger) (fs *FS, err error) { ctx, cancel := context.WithCancel(context.Background()) fs.cancel = cancel go fs.connect(ctx) + if block { + for { + if _, e := fs.Stat("."); e != ErrNotConnected { + return + } + time.Sleep(time.Millisecond * 100) + } + } return } @@ -95,7 +103,12 @@ func (fs *FS) connect(ctx context.Context) { go fs.keepAlive(ctx) fs.keepalive = true } - fs.client, err = sftp.NewClient(fs.c, sftp.UseFstat(true)) + fs.client, err = sftp.NewClient(fs.c, + sftp.UseFstat(true), + sftp.UseConcurrentReads(true), + //sftp.UseConcurrentWrites(true), + //sftp.MaxPacket(23552), + ) select { case <-ctx.Done(): return