X-Git-Url: http://www.git.cypherpunks.ru/?a=blobdiff_plain;f=src%2Ftx.go;h=20482efb234bc5a99e4b2fc5b792d7a976ff8c3a;hb=0fad171c0d79ad583c0faf5427e22d1d62a0a52d;hp=4c31d92ad3e127a983dcd7edb4f9f7ac54673c0e;hpb=c3abbbda3f7dc6c03702acbd534cbf196283eba8;p=nncp.git diff --git a/src/tx.go b/src/tx.go index 4c31d92..20482ef 100644 --- a/src/tx.go +++ b/src/tx.go @@ -1,6 +1,6 @@ /* NNCP -- Node to Node copy, utilities for store-and-forward data exchange -Copyright (C) 2016-2020 Sergey Matveev +Copyright (C) 2016-2021 Sergey Matveev This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -23,6 +23,7 @@ import ( "bytes" "crypto/rand" "errors" + "fmt" "hash" "io" "io/ioutil" @@ -33,8 +34,8 @@ import ( "time" xdr "github.com/davecgh/go-xdr/xdr2" + "github.com/dustin/go-humanize" "github.com/klauspost/compress/zstd" - "golang.org/x/crypto/blake2b" "golang.org/x/crypto/chacha20poly1305" ) @@ -62,7 +63,9 @@ func (ctx *Ctx) Tx( } expectedSize := size for i := 0; i < len(hops); i++ { - expectedSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+expectedSize) + expectedSize = PktEncOverhead + + PktSizeOverhead + + sizeWithTags(PktOverhead+expectedSize) } padSize := minSize - expectedSize if padSize < 0 { @@ -79,16 +82,30 @@ func (ctx *Ctx) Tx( errs := make(chan error) curSize := size pipeR, pipeW := io.Pipe() + var pktEncRaw []byte go func(size int64, src io.Reader, dst io.WriteCloser) { - ctx.LogD("tx", SDS{ - "node": hops[0].Id, - "nice": int(nice), - "size": size, - }, "wrote") - errs <- PktEncWrite(ctx.Self, hops[0], pkt, nice, size, padSize, src, dst) + ctx.LogD("tx", LEs{ + {"Node", hops[0].Id}, + {"Nice", int(nice)}, + {"Size", size}, + }, func(les LEs) string { + return fmt.Sprintf( + "Tx packet to %s (%s) nice: %s", + ctx.NodeName(hops[0].Id), + humanize.IBytes(uint64(size)), + NicenessFmt(nice), + ) + }) + pktEncRaw, err = PktEncWrite( + ctx.Self, hops[0], pkt, nice, size, padSize, src, dst, + ) + errs <- err dst.Close() // #nosec G104 }(curSize, src, pipeW) - curSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+curSize) + padSize + curSize = PktEncOverhead + + PktSizeOverhead + + sizeWithTags(PktOverhead+curSize) + + padSize var pipeRPrev io.Reader for i := 1; i < len(hops); i++ { @@ -96,12 +113,20 @@ func (ctx *Ctx) Tx( pipeRPrev = pipeR pipeR, pipeW = io.Pipe() go func(node *Node, pkt *Pkt, size int64, src io.Reader, dst io.WriteCloser) { - ctx.LogD("tx", SDS{ - "node": node.Id, - "nice": int(nice), - "size": size, - }, "trns wrote") - errs <- PktEncWrite(ctx.Self, node, pkt, nice, size, 0, src, dst) + ctx.LogD("tx", LEs{ + {"Node", node.Id}, + {"Nice", int(nice)}, + {"Size", size}, + }, func(les LEs) string { + return fmt.Sprintf( + "Tx trns packet to %s (%s) nice: %s", + ctx.NodeName(node.Id), + humanize.IBytes(uint64(size)), + NicenessFmt(nice), + ) + }) + _, err := PktEncWrite(ctx.Self, node, pkt, nice, size, 0, src, dst) + errs <- err dst.Close() // #nosec G104 }(hops[i], pktTrns, curSize, pipeRPrev, pipeW) curSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+curSize) @@ -109,7 +134,7 @@ func (ctx *Ctx) Tx( go func() { _, err := CopyProgressed( tmp.W, pipeR, "Tx", - SDS{"pkt": pktName, "fullsize": curSize}, + LEs{{"Pkt", pktName}, {"FullSize", curSize}}, ctx.ShowPrgrs, ) errs <- err @@ -124,6 +149,12 @@ func (ctx *Ctx) Tx( nodePath := filepath.Join(ctx.Spool, lastNode.Id.String()) err = tmp.Commit(filepath.Join(nodePath, string(TTx))) os.Symlink(nodePath, filepath.Join(ctx.Spool, lastNode.Name)) // #nosec G104 + if err != nil { + return lastNode, err + } + if ctx.HdrUsage { + ctx.HdrWrite(pktEncRaw, filepath.Join(nodePath, string(TTx), tmp.Checksum())) + } return lastNode, err } @@ -131,49 +162,63 @@ type DummyCloser struct{} func (dc DummyCloser) Close() error { return nil } -func prepareTxFile(srcPath string) (reader io.Reader, closer io.Closer, fileSize int64, archived bool, rerr error) { - if srcPath == "-" { - // Read content from stdin, saving to temporary file, encrypting - // on the fly - src, err := ioutil.TempFile("", "nncp-file") - if err != nil { - rerr = err - return - } - os.Remove(src.Name()) // #nosec G104 - tmpW := bufio.NewWriter(src) - tmpKey := make([]byte, chacha20poly1305.KeySize) - if _, rerr = rand.Read(tmpKey[:]); rerr != nil { - return - } - aead, err := chacha20poly1305.New(tmpKey) - if err != nil { - rerr = err - return - } - nonce := make([]byte, aead.NonceSize()) - written, err := aeadProcess(aead, nonce, true, bufio.NewReader(os.Stdin), tmpW) - if err != nil { - rerr = err - return - } - fileSize = int64(written) - if err = tmpW.Flush(); err != nil { - rerr = err - return - } - if _, err = src.Seek(0, io.SeekStart); err != nil { - rerr = err - return +func throughTmpFile(r io.Reader) ( + reader io.Reader, + closer io.Closer, + fileSize int64, + rerr error, +) { + src, err := ioutil.TempFile("", "nncp-file") + if err != nil { + rerr = err + return + } + os.Remove(src.Name()) // #nosec G104 + tmpW := bufio.NewWriter(src) + tmpKey := make([]byte, chacha20poly1305.KeySize) + if _, rerr = rand.Read(tmpKey[:]); rerr != nil { + return + } + aead, err := chacha20poly1305.New(tmpKey) + if err != nil { + rerr = err + return + } + nonce := make([]byte, aead.NonceSize()) + written, err := aeadProcess(aead, nonce, true, r, tmpW) + if err != nil { + rerr = err + return + } + fileSize = int64(written) + if err = tmpW.Flush(); err != nil { + rerr = err + return + } + if _, err = src.Seek(0, io.SeekStart); err != nil { + rerr = err + return + } + r, w := io.Pipe() + go func() { + if _, err := aeadProcess(aead, nonce, false, bufio.NewReader(src), w); err != nil { + w.CloseWithError(err) // #nosec G104 } - r, w := io.Pipe() - go func() { - if _, err := aeadProcess(aead, nonce, false, bufio.NewReader(src), w); err != nil { - w.CloseWithError(err) // #nosec G104 - } - }() - reader = r - closer = src + }() + reader = r + closer = src + return +} + +func prepareTxFile(srcPath string) ( + reader io.Reader, + closer io.Closer, + fileSize int64, + archived bool, + rerr error, +) { + if srcPath == "-" { + reader, closer, fileSize, rerr = throughTmpFile(bufio.NewReader(os.Stdin)) return } @@ -333,31 +378,40 @@ func (ctx *Ctx) TxFile( return err } _, err = ctx.Tx(node, pkt, nice, fileSize, minSize, reader, dstPath) - sds := SDS{ - "type": "file", - "node": node.Id, - "nice": int(nice), - "src": srcPath, - "dst": dstPath, - "size": fileSize, + les := LEs{ + {"Type", "file"}, + {"Node", node.Id}, + {"Nice", int(nice)}, + {"Src", srcPath}, + {"Dst", dstPath}, + {"Size", fileSize}, + } + logMsg := func(les LEs) string { + return fmt.Sprintf( + "File %s (%s) sent to %s:%s", + srcPath, + humanize.IBytes(uint64(fileSize)), + ctx.NodeName(node.Id), + dstPath, + ) } if err == nil { - ctx.LogI("tx", sds, "sent") + ctx.LogI("tx", les, logMsg) } else { - ctx.LogE("tx", sds, err, "sent") + ctx.LogE("tx", les, err, logMsg) } return err } leftSize := fileSize metaPkt := ChunkedMeta{ - Magic: MagicNNCPMv1, + Magic: MagicNNCPMv2, FileSize: uint64(fileSize), ChunkSize: uint64(chunkSize), - Checksums: make([][32]byte, 0, (fileSize/chunkSize)+1), + Checksums: make([][MTHSize]byte, 0, (fileSize/chunkSize)+1), } for i := int64(0); i < (fileSize/chunkSize)+1; i++ { - hsh := new([32]byte) + hsh := new([MTHSize]byte) metaPkt.Checksums = append(metaPkt.Checksums, *hsh) } var sizeToSend int64 @@ -376,10 +430,7 @@ func (ctx *Ctx) TxFile( if err != nil { return err } - hsh, err = blake2b.New256(nil) - if err != nil { - return err - } + hsh = MTHNew(0, 0) _, err = ctx.Tx( node, pkt, @@ -389,18 +440,27 @@ func (ctx *Ctx) TxFile( io.TeeReader(reader, hsh), path, ) - sds := SDS{ - "type": "file", - "node": node.Id, - "nice": int(nice), - "src": srcPath, - "dst": path, - "size": sizeToSend, + les := LEs{ + {"Type", "file"}, + {"Node", node.Id}, + {"Nice", int(nice)}, + {"Src", srcPath}, + {"Dst", path}, + {"Size", sizeToSend}, + } + logMsg := func(les LEs) string { + return fmt.Sprintf( + "File %s (%s) sent to %s:%s", + srcPath, + humanize.IBytes(uint64(sizeToSend)), + ctx.NodeName(node.Id), + path, + ) } if err == nil { - ctx.LogI("tx", sds, "sent") + ctx.LogI("tx", les, logMsg) } else { - ctx.LogE("tx", sds, err, "sent") + ctx.LogE("tx", les, err, logMsg) return err } hsh.Sum(metaPkt.Checksums[chunkNum][:0]) @@ -422,18 +482,27 @@ func (ctx *Ctx) TxFile( } metaPktSize := int64(metaBuf.Len()) _, err = ctx.Tx(node, pkt, nice, metaPktSize, minSize, &metaBuf, path) - sds := SDS{ - "type": "file", - "node": node.Id, - "nice": int(nice), - "src": srcPath, - "dst": path, - "size": metaPktSize, + les := LEs{ + {"Type", "file"}, + {"Node", node.Id}, + {"Nice", int(nice)}, + {"Src", srcPath}, + {"Dst", path}, + {"Size", metaPktSize}, + } + logMsg := func(les LEs) string { + return fmt.Sprintf( + "File %s (%s) sent to %s:%s", + srcPath, + humanize.IBytes(uint64(metaPktSize)), + ctx.NodeName(node.Id), + path, + ) } if err == nil { - ctx.LogI("tx", sds, "sent") + ctx.LogI("tx", les, logMsg) } else { - ctx.LogE("tx", sds, err, "sent") + ctx.LogE("tx", les, err, logMsg) } return err } @@ -458,18 +527,25 @@ func (ctx *Ctx) TxFreq( src := strings.NewReader(dstPath) size := int64(src.Len()) _, err = ctx.Tx(node, pkt, nice, size, minSize, src, srcPath) - sds := SDS{ - "type": "freq", - "node": node.Id, - "nice": int(nice), - "replynice": int(replyNice), - "src": srcPath, - "dst": dstPath, + les := LEs{ + {"Type", "freq"}, + {"Node", node.Id}, + {"Nice", int(nice)}, + {"ReplyNice", int(replyNice)}, + {"Src", srcPath}, + {"Dst", dstPath}, + } + logMsg := func(les LEs) string { + return fmt.Sprintf( + "File request from %s:%s to %s sent", + ctx.NodeName(node.Id), srcPath, + dstPath, + ) } if err == nil { - ctx.LogI("tx", sds, "sent") + ctx.LogI("tx", les, logMsg) } else { - ctx.LogE("tx", sds, err, "sent") + ctx.LogE("tx", les, err, logMsg) } return err } @@ -481,60 +557,136 @@ func (ctx *Ctx) TxExec( args []string, in io.Reader, minSize int64, + useTmp bool, + noCompress bool, ) error { path := make([][]byte, 0, 1+len(args)) path = append(path, []byte(handle)) for _, arg := range args { path = append(path, []byte(arg)) } - pkt, err := NewPkt(PktTypeExec, replyNice, bytes.Join(path, []byte{0})) - if err != nil { - return err + pktType := PktTypeExec + if noCompress { + pktType = PktTypeExecFat } - var compressed bytes.Buffer - compressor, err := zstd.NewWriter( - &compressed, - zstd.WithEncoderLevel(zstd.SpeedDefault), - ) + pkt, err := NewPkt(pktType, replyNice, bytes.Join(path, []byte{0})) if err != nil { return err } - if _, err = io.Copy(compressor, in); err != nil { - compressor.Close() // #nosec G104 - return err + var size int64 + + if !noCompress && !useTmp { + var compressed bytes.Buffer + compressor, err := zstd.NewWriter( + &compressed, + zstd.WithEncoderLevel(zstd.SpeedDefault), + ) + if err != nil { + return err + } + if _, err = io.Copy(compressor, in); err != nil { + compressor.Close() // #nosec G104 + return err + } + if err = compressor.Close(); err != nil { + return err + } + size = int64(compressed.Len()) + _, err = ctx.Tx(node, pkt, nice, size, minSize, &compressed, handle) } - if err = compressor.Close(); err != nil { - return err + if noCompress && !useTmp { + var data bytes.Buffer + if _, err = io.Copy(&data, in); err != nil { + return err + } + size = int64(data.Len()) + _, err = ctx.Tx(node, pkt, nice, size, minSize, &data, handle) + } + if !noCompress && useTmp { + r, w := io.Pipe() + compressor, err := zstd.NewWriter(w, zstd.WithEncoderLevel(zstd.SpeedDefault)) + if err != nil { + return err + } + copyErr := make(chan error) + go func() { + _, err := io.Copy(compressor, in) + if err != nil { + compressor.Close() // #nosec G104 + copyErr <- err + } + err = compressor.Close() + w.Close() + copyErr <- err + }() + tmpReader, closer, fileSize, err := throughTmpFile(r) + if closer != nil { + defer closer.Close() + } + if err != nil { + return err + } + err = <-copyErr + if err != nil { + return err + } + size = fileSize + _, err = ctx.Tx(node, pkt, nice, size, minSize, tmpReader, handle) } - size := int64(compressed.Len()) - _, err = ctx.Tx(node, pkt, nice, size, minSize, &compressed, handle) - sds := SDS{ - "type": "exec", - "node": node.Id, - "nice": int(nice), - "replynice": int(replyNice), - "dst": strings.Join(append([]string{handle}, args...), " "), - "size": size, + if noCompress && useTmp { + tmpReader, closer, fileSize, err := throughTmpFile(in) + if closer != nil { + defer closer.Close() + } + if err != nil { + return err + } + size = fileSize + _, err = ctx.Tx(node, pkt, nice, size, minSize, tmpReader, handle) + } + + dst := strings.Join(append([]string{handle}, args...), " ") + les := LEs{ + {"Type", "exec"}, + {"Node", node.Id}, + {"Nice", int(nice)}, + {"ReplyNice", int(replyNice)}, + {"Dst", dst}, + {"Size", size}, + } + logMsg := func(les LEs) string { + return fmt.Sprintf( + "Exec sent to %s@%s (%s)", + ctx.NodeName(node.Id), dst, humanize.IBytes(uint64(size)), + ) } if err == nil { - ctx.LogI("tx", sds, "sent") + ctx.LogI("tx", les, logMsg) } else { - ctx.LogE("tx", sds, err, "sent") + ctx.LogE("tx", les, err, logMsg) } return err } func (ctx *Ctx) TxTrns(node *Node, nice uint8, size int64, src io.Reader) error { - sds := SDS{ - "type": "trns", - "node": node.Id, - "nice": int(nice), - "size": size, + les := LEs{ + {"Type", "trns"}, + {"Node", node.Id}, + {"Nice", int(nice)}, + {"Size", size}, + } + logMsg := func(les LEs) string { + return fmt.Sprintf( + "Transitional packet to %s (%s) (nice %s)", + ctx.NodeName(node.Id), + humanize.IBytes(uint64(size)), + NicenessFmt(nice), + ) } - ctx.LogD("tx", sds, "taken") + ctx.LogD("tx", les, logMsg) if !ctx.IsEnoughSpace(size) { err := errors.New("is not enough space") - ctx.LogE("tx", sds, err, err.Error()) + ctx.LogE("tx", les, err, logMsg) return err } tmp, err := ctx.NewTmpFileWHash() @@ -543,7 +695,7 @@ func (ctx *Ctx) TxTrns(node *Node, nice uint8, size int64, src io.Reader) error } if _, err = CopyProgressed( tmp.W, src, "Tx trns", - SDS{"pkt": node.Id.String(), "fullsize": size}, + LEs{{"Pkt", node.Id.String()}, {"FullSize", size}}, ctx.ShowPrgrs, ); err != nil { return err @@ -551,9 +703,9 @@ func (ctx *Ctx) TxTrns(node *Node, nice uint8, size int64, src io.Reader) error nodePath := filepath.Join(ctx.Spool, node.Id.String()) err = tmp.Commit(filepath.Join(nodePath, string(TTx))) if err == nil { - ctx.LogI("tx", sds, "sent") + ctx.LogI("tx", les, logMsg) } else { - ctx.LogI("tx", SdsAdd(sds, SDS{"err": err}), "sent") + ctx.LogI("tx", append(les, LE{"Err", err}), logMsg) } os.Symlink(nodePath, filepath.Join(ctx.Spool, node.Name)) // #nosec G104 return err