2 NNCP -- Node to Node copy, utilities for store-and-forward data exchange
3 Copyright (C) 2016-2021 Sergey Matveev <stargrave@stargrave.org>
5 This program is free software: you can redistribute it and/or modify
6 it under the terms of the GNU General Public License as published by
7 the Free Software Foundation, version 3 of the License.
9 This program is distributed in the hope that it will be useful,
10 but WITHOUT ANY WARRANTY; without even the implied warranty of
11 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 GNU General Public License for more details.
14 You should have received a copy of the GNU General Public License
15 along with this program. If not, see <http://www.gnu.org/licenses/>.
36 xdr "github.com/davecgh/go-xdr/xdr2"
37 "github.com/dustin/go-humanize"
38 "github.com/klauspost/compress/zstd"
39 "golang.org/x/crypto/blake2b"
40 "golang.org/x/crypto/chacha20poly1305"
61 area = ctx.AreaId2Area[*areaId]
63 return nil, errors.New("unknown area id")
66 hops := make([]*Node, 0, 1+len(node.Via))
67 hops = append(hops, node)
69 for i := len(node.Via); i > 0; i-- {
70 lastNode = ctx.Neigh[*node.Via[i-1]]
71 hops = append(hops, lastNode)
78 for i := 0; i < wrappers; i++ {
79 expectedSize = PktEncOverhead +
81 sizeWithTags(PktOverhead+expectedSize)
83 padSize := minSize - expectedSize
87 if !ctx.IsEnoughSpace(size + padSize) {
88 return nil, errors.New("is not enough space")
90 tmp, err := ctx.NewTmpFileWHash()
95 errs := make(chan error)
96 pktEncRaws := make(chan []byte)
98 pipeR, pipeW := io.Pipe()
99 var pipeRPrev io.Reader
101 go func(size int64, src io.Reader, dst io.WriteCloser) {
103 {"Node", hops[0].Id},
106 }, func(les LEs) string {
108 "Tx packet to %s (%s) nice: %s",
109 ctx.NodeName(hops[0].Id),
110 humanize.IBytes(uint64(size)),
114 pktEncRaw, err := PktEncWrite(
115 ctx.Self, hops[0], pkt, nice, size, padSize, src, dst,
117 pktEncRaws <- pktEncRaw
119 dst.Close() // #nosec G104
120 }(curSize, src, pipeW)
121 curSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+curSize)
124 go func(size, padSize int64, src io.Reader, dst io.WriteCloser) {
129 }, func(les LEs) string {
131 "Tx area packet to %s (%s) nice: %s",
132 ctx.AreaName(areaId),
133 humanize.IBytes(uint64(size)),
137 areaNode := Node{Id: new(NodeId), ExchPub: new([32]byte)}
138 copy(areaNode.Id[:], area.Id[:])
139 copy(areaNode.ExchPub[:], area.Pub[:])
140 pktEncRaw, err := PktEncWrite(
141 ctx.Self, &areaNode, pkt, nice, size, padSize, src, dst,
143 pktEncRaws <- pktEncRaw
145 dst.Close() // #nosec G104
146 }(curSize, padSize, src, pipeW)
147 curSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+curSize)
150 pipeR, pipeW = io.Pipe()
151 go func(size int64, src io.Reader, dst io.WriteCloser) {
152 pktArea, err := NewPkt(PktTypeArea, 0, area.Id[:])
157 {"Node", hops[0].Id},
160 }, func(les LEs) string {
162 "Tx packet to %s (%s) nice: %s",
163 ctx.NodeName(hops[0].Id),
164 humanize.IBytes(uint64(size)),
168 pktEncRaw, err := PktEncWrite(
169 ctx.Self, hops[0], pktArea, nice, size, 0, src, dst,
171 pktEncRaws <- pktEncRaw
173 dst.Close() // #nosec G104
174 }(curSize, pipeRPrev, pipeW)
175 curSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+curSize)
177 for i := 1; i < len(hops); i++ {
178 pktTrns, err := NewPkt(PktTypeTrns, 0, hops[i-1].Id[:])
183 pipeR, pipeW = io.Pipe()
184 go func(node *Node, pkt *Pkt, size int64, src io.Reader, dst io.WriteCloser) {
189 }, func(les LEs) string {
191 "Tx trns packet to %s (%s) nice: %s",
192 ctx.NodeName(node.Id),
193 humanize.IBytes(uint64(size)),
197 pktEncRaw, err := PktEncWrite(ctx.Self, node, pkt, nice, size, 0, src, dst)
198 pktEncRaws <- pktEncRaw
200 dst.Close() // #nosec G104
201 }(hops[i], pktTrns, curSize, pipeRPrev, pipeW)
202 curSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+curSize)
205 _, err := CopyProgressed(
207 LEs{{"Pkt", pktName}, {"FullSize", curSize}},
215 pktEncMsg = <-pktEncRaws
217 for i := 0; i < len(hops); i++ {
218 pktEncRaw = <-pktEncRaws
220 for i := 0; i <= wrappers; i++ {
223 tmp.Fd.Close() // #nosec G104
227 nodePath := filepath.Join(ctx.Spool, lastNode.Id.String())
228 err = tmp.Commit(filepath.Join(nodePath, string(TTx)))
229 os.Symlink(nodePath, filepath.Join(ctx.Spool, lastNode.Name)) // #nosec G104
234 ctx.HdrWrite(pktEncRaw, filepath.Join(nodePath, string(TTx), tmp.Checksum()))
237 msgHashRaw := blake2b.Sum256(pktEncMsg)
238 msgHash := Base32Codec.EncodeToString(msgHashRaw[:])
239 seenDir := filepath.Join(
240 ctx.Spool, ctx.SelfId.String(), AreaDir, areaId.String(),
242 seenPath := filepath.Join(seenDir, msgHash+SeenSuffix)
248 {"AreaMsg", msgHash},
250 logMsg := func(les LEs) string {
252 "Tx area packet to %s (%s) nice: %s, area %s: %s",
253 ctx.NodeName(node.Id),
254 humanize.IBytes(uint64(size)),
260 if err = os.MkdirAll(seenDir, os.FileMode(0777)); err != nil {
261 ctx.LogE("tx-mkdir", les, err, logMsg)
264 if fd, err := os.Create(seenPath); err == nil {
266 if err = DirSync(seenDir); err != nil {
267 ctx.LogE("tx-dirsync", les, err, logMsg)
271 ctx.LogI("tx-area", les, logMsg)
276 type DummyCloser struct{}
278 func (dc DummyCloser) Close() error { return nil }
280 func throughTmpFile(r io.Reader) (
286 src, err := ioutil.TempFile("", "nncp-file")
291 os.Remove(src.Name()) // #nosec G104
292 tmpW := bufio.NewWriter(src)
293 tmpKey := make([]byte, chacha20poly1305.KeySize)
294 if _, rerr = rand.Read(tmpKey[:]); rerr != nil {
297 aead, err := chacha20poly1305.New(tmpKey)
302 nonce := make([]byte, aead.NonceSize())
303 written, err := aeadProcess(aead, nonce, nil, true, r, tmpW)
308 fileSize = int64(written)
309 if err = tmpW.Flush(); err != nil {
313 if _, err = src.Seek(0, io.SeekStart); err != nil {
319 if _, err := aeadProcess(aead, nonce, nil, false, bufio.NewReader(src), w); err != nil {
320 w.CloseWithError(err) // #nosec G104
328 func prepareTxFile(srcPath string) (
336 reader, closer, fileSize, rerr = throughTmpFile(bufio.NewReader(os.Stdin))
340 srcStat, err := os.Stat(srcPath)
345 mode := srcStat.Mode()
347 if mode.IsRegular() {
348 // It is regular file, just send it
349 src, err := os.Open(srcPath)
354 fileSize = srcStat.Size()
355 reader = bufio.NewReader(src)
361 rerr = errors.New("unsupported file type")
365 // It is directory, create PAX archive with its contents
367 basePath := filepath.Base(srcPath)
368 rootPath, err := filepath.Abs(srcPath)
378 dirs := make([]einfo, 0, 1<<10)
379 files := make([]einfo, 0, 1<<10)
380 rerr = filepath.Walk(rootPath, func(path string, info os.FileInfo, err error) error {
385 // directory header, PAX record header+contents
386 fileSize += TarBlockSize + 2*TarBlockSize
387 dirs = append(dirs, einfo{path: path, modTime: info.ModTime()})
389 // file header, PAX record header+contents, file content
390 fileSize += TarBlockSize + 2*TarBlockSize + info.Size()
391 if n := info.Size() % TarBlockSize; n != 0 {
392 fileSize += TarBlockSize - n // padding
394 files = append(files, einfo{
396 modTime: info.ModTime(),
408 closer = DummyCloser{}
409 fileSize += 2 * TarBlockSize // termination block
412 tarWr := tar.NewWriter(w)
414 Typeflag: tar.TypeDir,
416 PAXRecords: map[string]string{
417 "comment": "Autogenerated by " + VersionGet(),
419 Format: tar.FormatPAX,
421 for _, e := range dirs {
422 hdr.Name = basePath + e.path[len(rootPath):]
423 hdr.ModTime = e.modTime
424 if err = tarWr.WriteHeader(&hdr); err != nil {
425 return w.CloseWithError(err)
428 hdr.Typeflag = tar.TypeReg
430 for _, e := range files {
431 hdr.Name = basePath + e.path[len(rootPath):]
432 hdr.ModTime = e.modTime
434 if err = tarWr.WriteHeader(&hdr); err != nil {
435 return w.CloseWithError(err)
437 fd, err := os.Open(e.path)
439 fd.Close() // #nosec G104
440 return w.CloseWithError(err)
442 if _, err = io.Copy(tarWr, bufio.NewReader(fd)); err != nil {
443 fd.Close() // #nosec G104
444 return w.CloseWithError(err)
446 fd.Close() // #nosec G104
448 if err = tarWr.Close(); err != nil {
449 return w.CloseWithError(err)
456 func (ctx *Ctx) TxFile(
459 srcPath, dstPath string,
461 minSize, maxSize int64,
464 dstPathSpecified := false
467 return errors.New("Must provide destination filename")
469 dstPath = filepath.Base(srcPath)
471 dstPathSpecified = true
473 dstPath = filepath.Clean(dstPath)
474 if filepath.IsAbs(dstPath) {
475 return errors.New("Relative destination path required")
477 reader, closer, fileSize, archived, err := prepareTxFile(srcPath)
484 if fileSize > maxSize {
485 return errors.New("Too big than allowed")
487 if archived && !dstPathSpecified {
491 if fileSize <= chunkSize {
492 pkt, err := NewPkt(PktTypeFile, nice, []byte(dstPath))
496 _, err = ctx.Tx(node, pkt, nice, fileSize, minSize, reader, dstPath, areaId)
505 logMsg := func(les LEs) string {
507 "File %s (%s) sent to %s:%s",
509 humanize.IBytes(uint64(fileSize)),
510 ctx.NodeName(node.Id),
515 ctx.LogI("tx", les, logMsg)
517 ctx.LogE("tx", les, err, logMsg)
523 metaPkt := ChunkedMeta{
524 Magic: MagicNNCPMv2.B,
525 FileSize: uint64(fileSize),
526 ChunkSize: uint64(chunkSize),
527 Checksums: make([][MTHSize]byte, 0, (fileSize/chunkSize)+1),
529 for i := int64(0); i < (fileSize/chunkSize)+1; i++ {
530 hsh := new([MTHSize]byte)
531 metaPkt.Checksums = append(metaPkt.Checksums, *hsh)
539 if leftSize <= chunkSize {
540 sizeToSend = leftSize
542 sizeToSend = chunkSize
544 path = dstPath + ChunkedSuffixPart + strconv.Itoa(chunkNum)
545 pkt, err = NewPkt(PktTypeFile, nice, []byte(path))
556 io.TeeReader(reader, hsh),
566 {"Size", sizeToSend},
568 logMsg := func(les LEs) string {
570 "File %s (%s) sent to %s:%s",
572 humanize.IBytes(uint64(sizeToSend)),
573 ctx.NodeName(node.Id),
578 ctx.LogI("tx", les, logMsg)
580 ctx.LogE("tx", les, err, logMsg)
583 hsh.Sum(metaPkt.Checksums[chunkNum][:0])
584 leftSize -= sizeToSend
590 var metaBuf bytes.Buffer
591 _, err = xdr.Marshal(&metaBuf, metaPkt)
595 path = dstPath + ChunkedSuffixMeta
596 pkt, err = NewPkt(PktTypeFile, nice, []byte(path))
600 metaPktSize := int64(metaBuf.Len())
601 _, err = ctx.Tx(node, pkt, nice, metaPktSize, minSize, &metaBuf, path, areaId)
608 {"Size", metaPktSize},
610 logMsg := func(les LEs) string {
612 "File %s (%s) sent to %s:%s",
614 humanize.IBytes(uint64(metaPktSize)),
615 ctx.NodeName(node.Id),
620 ctx.LogI("tx", les, logMsg)
622 ctx.LogE("tx", les, err, logMsg)
627 func (ctx *Ctx) TxFreq(
629 nice, replyNice uint8,
630 srcPath, dstPath string,
631 minSize int64) error {
632 dstPath = filepath.Clean(dstPath)
633 if filepath.IsAbs(dstPath) {
634 return errors.New("Relative destination path required")
636 srcPath = filepath.Clean(srcPath)
637 if filepath.IsAbs(srcPath) {
638 return errors.New("Relative source path required")
640 pkt, err := NewPkt(PktTypeFreq, replyNice, []byte(srcPath))
644 src := strings.NewReader(dstPath)
645 size := int64(src.Len())
646 _, err = ctx.Tx(node, pkt, nice, size, minSize, src, srcPath, nil)
651 {"ReplyNice", int(replyNice)},
655 logMsg := func(les LEs) string {
657 "File request from %s:%s to %s sent",
658 ctx.NodeName(node.Id), srcPath,
663 ctx.LogI("tx", les, logMsg)
665 ctx.LogE("tx", les, err, logMsg)
670 func (ctx *Ctx) TxExec(
672 nice, replyNice uint8,
681 path := make([][]byte, 0, 1+len(args))
682 path = append(path, []byte(handle))
683 for _, arg := range args {
684 path = append(path, []byte(arg))
686 pktType := PktTypeExec
688 pktType = PktTypeExecFat
690 pkt, err := NewPkt(pktType, replyNice, bytes.Join(path, []byte{0}))
696 if !noCompress && !useTmp {
697 var compressed bytes.Buffer
698 compressor, err := zstd.NewWriter(
700 zstd.WithEncoderLevel(zstd.SpeedDefault),
705 if _, err = io.Copy(compressor, in); err != nil {
706 compressor.Close() // #nosec G104
709 if err = compressor.Close(); err != nil {
712 size = int64(compressed.Len())
713 _, err = ctx.Tx(node, pkt, nice, size, minSize, &compressed, handle, areaId)
715 if noCompress && !useTmp {
716 var data bytes.Buffer
717 if _, err = io.Copy(&data, in); err != nil {
720 size = int64(data.Len())
721 _, err = ctx.Tx(node, pkt, nice, size, minSize, &data, handle, areaId)
723 if !noCompress && useTmp {
725 compressor, err := zstd.NewWriter(w, zstd.WithEncoderLevel(zstd.SpeedDefault))
729 copyErr := make(chan error)
731 _, err := io.Copy(compressor, in)
733 compressor.Close() // #nosec G104
736 err = compressor.Close()
740 tmpReader, closer, fileSize, err := throughTmpFile(r)
752 _, err = ctx.Tx(node, pkt, nice, size, minSize, tmpReader, handle, areaId)
754 if noCompress && useTmp {
755 tmpReader, closer, fileSize, err := throughTmpFile(in)
763 _, err = ctx.Tx(node, pkt, nice, size, minSize, tmpReader, handle, areaId)
766 dst := strings.Join(append([]string{handle}, args...), " ")
771 {"ReplyNice", int(replyNice)},
775 logMsg := func(les LEs) string {
777 "Exec sent to %s@%s (%s)",
778 ctx.NodeName(node.Id), dst, humanize.IBytes(uint64(size)),
782 ctx.LogI("tx", les, logMsg)
784 ctx.LogE("tx", les, err, logMsg)
789 func (ctx *Ctx) TxTrns(node *Node, nice uint8, size int64, src io.Reader) error {
796 logMsg := func(les LEs) string {
798 "Transitional packet to %s (%s) (nice %s)",
799 ctx.NodeName(node.Id),
800 humanize.IBytes(uint64(size)),
804 ctx.LogD("tx", les, logMsg)
805 if !ctx.IsEnoughSpace(size) {
806 err := errors.New("is not enough space")
807 ctx.LogE("tx", les, err, logMsg)
810 tmp, err := ctx.NewTmpFileWHash()
814 if _, err = CopyProgressed(
815 tmp.W, src, "Tx trns",
816 LEs{{"Pkt", node.Id.String()}, {"FullSize", size}},
821 nodePath := filepath.Join(ctx.Spool, node.Id.String())
822 err = tmp.Commit(filepath.Join(nodePath, string(TTx)))
824 ctx.LogI("tx", les, logMsg)
826 ctx.LogI("tx", append(les, LE{"Err", err}), logMsg)
828 os.Symlink(nodePath, filepath.Join(ctx.Spool, node.Name)) // #nosec G104