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("area has no encryption keys")
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
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
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
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
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++ {
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))
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.Stat(seenDir); err != nil {
261 if !os.IsNotExist(err) {
262 ctx.LogE("tx-mkdir", les, err, logMsg)
265 if err = os.MkdirAll(seenDir, os.FileMode(0777)); err != nil {
266 ctx.LogE("tx-mkdir", les, err, logMsg)
270 if fd, err := os.Create(seenPath); err == nil {
272 if err = DirSync(seenDir); err != nil {
273 ctx.LogE("tx-dirsync", les, err, logMsg)
277 ctx.LogI("tx-area", les, logMsg)
282 type DummyCloser struct{}
284 func (dc DummyCloser) Close() error { return nil }
286 func throughTmpFile(r io.Reader) (
292 src, err := ioutil.TempFile("", "nncp-file")
297 os.Remove(src.Name())
298 tmpW := bufio.NewWriter(src)
299 tmpKey := make([]byte, chacha20poly1305.KeySize)
300 if _, rerr = rand.Read(tmpKey[:]); rerr != nil {
303 aead, err := chacha20poly1305.New(tmpKey)
308 nonce := make([]byte, aead.NonceSize())
309 written, err := aeadProcess(aead, nonce, nil, true, r, tmpW)
314 fileSize = int64(written)
315 if err = tmpW.Flush(); err != nil {
319 if _, err = src.Seek(0, io.SeekStart); err != nil {
325 for i := 0; i < aead.NonceSize(); i++ {
328 if _, err := aeadProcess(aead, nonce, nil, false, bufio.NewReader(src), w); err != nil {
329 w.CloseWithError(err)
337 func prepareTxFile(srcPath string) (
345 reader, closer, fileSize, rerr = throughTmpFile(bufio.NewReader(os.Stdin))
349 srcStat, err := os.Stat(srcPath)
354 mode := srcStat.Mode()
356 if mode.IsRegular() {
357 // It is regular file, just send it
358 src, err := os.Open(srcPath)
363 fileSize = srcStat.Size()
364 reader = bufio.NewReader(src)
370 rerr = errors.New("unsupported file type")
374 // It is directory, create PAX archive with its contents
376 basePath := filepath.Base(srcPath)
377 rootPath, err := filepath.Abs(srcPath)
387 dirs := make([]einfo, 0, 1<<10)
388 files := make([]einfo, 0, 1<<10)
389 rerr = filepath.Walk(rootPath, func(path string, info os.FileInfo, err error) error {
394 // directory header, PAX record header+contents
395 fileSize += TarBlockSize + 2*TarBlockSize
396 dirs = append(dirs, einfo{path: path, modTime: info.ModTime()})
398 // file header, PAX record header+contents, file content
399 fileSize += TarBlockSize + 2*TarBlockSize + info.Size()
400 if n := info.Size() % TarBlockSize; n != 0 {
401 fileSize += TarBlockSize - n // padding
403 files = append(files, einfo{
405 modTime: info.ModTime(),
417 closer = DummyCloser{}
418 fileSize += 2 * TarBlockSize // termination block
421 tarWr := tar.NewWriter(w)
423 Typeflag: tar.TypeDir,
425 PAXRecords: map[string]string{
426 "comment": "Autogenerated by " + VersionGet(),
428 Format: tar.FormatPAX,
430 for _, e := range dirs {
431 hdr.Name = basePath + e.path[len(rootPath):]
432 hdr.ModTime = e.modTime
433 if err = tarWr.WriteHeader(&hdr); err != nil {
434 return w.CloseWithError(err)
437 hdr.Typeflag = tar.TypeReg
439 for _, e := range files {
440 hdr.Name = basePath + e.path[len(rootPath):]
441 hdr.ModTime = e.modTime
443 if err = tarWr.WriteHeader(&hdr); err != nil {
444 return w.CloseWithError(err)
446 fd, err := os.Open(e.path)
449 return w.CloseWithError(err)
451 if _, err = io.Copy(tarWr, bufio.NewReader(fd)); err != nil {
453 return w.CloseWithError(err)
457 if err = tarWr.Close(); err != nil {
458 return w.CloseWithError(err)
465 func (ctx *Ctx) TxFile(
468 srcPath, dstPath string,
470 minSize, maxSize int64,
473 dstPathSpecified := false
476 return errors.New("Must provide destination filename")
478 dstPath = filepath.Base(srcPath)
480 dstPathSpecified = true
482 dstPath = filepath.Clean(dstPath)
483 if filepath.IsAbs(dstPath) {
484 return errors.New("Relative destination path required")
486 reader, closer, fileSize, archived, err := prepareTxFile(srcPath)
493 if fileSize > maxSize {
494 return errors.New("Too big than allowed")
496 if archived && !dstPathSpecified {
500 if fileSize <= chunkSize {
501 pkt, err := NewPkt(PktTypeFile, nice, []byte(dstPath))
505 _, err = ctx.Tx(node, pkt, nice, fileSize, minSize, reader, dstPath, areaId)
514 logMsg := func(les LEs) string {
516 "File %s (%s) sent to %s:%s",
518 humanize.IBytes(uint64(fileSize)),
519 ctx.NodeName(node.Id),
524 ctx.LogI("tx", les, logMsg)
526 ctx.LogE("tx", les, err, logMsg)
532 metaPkt := ChunkedMeta{
533 Magic: MagicNNCPMv2.B,
534 FileSize: uint64(fileSize),
535 ChunkSize: uint64(chunkSize),
536 Checksums: make([][MTHSize]byte, 0, (fileSize/chunkSize)+1),
538 for i := int64(0); i < (fileSize/chunkSize)+1; i++ {
539 hsh := new([MTHSize]byte)
540 metaPkt.Checksums = append(metaPkt.Checksums, *hsh)
548 if leftSize <= chunkSize {
549 sizeToSend = leftSize
551 sizeToSend = chunkSize
553 path = dstPath + ChunkedSuffixPart + strconv.Itoa(chunkNum)
554 pkt, err = NewPkt(PktTypeFile, nice, []byte(path))
565 io.TeeReader(reader, hsh),
575 {"Size", sizeToSend},
577 logMsg := func(les LEs) string {
579 "File %s (%s) sent to %s:%s",
581 humanize.IBytes(uint64(sizeToSend)),
582 ctx.NodeName(node.Id),
587 ctx.LogI("tx", les, logMsg)
589 ctx.LogE("tx", les, err, logMsg)
592 hsh.Sum(metaPkt.Checksums[chunkNum][:0])
593 leftSize -= sizeToSend
599 var metaBuf bytes.Buffer
600 _, err = xdr.Marshal(&metaBuf, metaPkt)
604 path = dstPath + ChunkedSuffixMeta
605 pkt, err = NewPkt(PktTypeFile, nice, []byte(path))
609 metaPktSize := int64(metaBuf.Len())
610 _, err = ctx.Tx(node, pkt, nice, metaPktSize, minSize, &metaBuf, path, areaId)
617 {"Size", metaPktSize},
619 logMsg := func(les LEs) string {
621 "File %s (%s) sent to %s:%s",
623 humanize.IBytes(uint64(metaPktSize)),
624 ctx.NodeName(node.Id),
629 ctx.LogI("tx", les, logMsg)
631 ctx.LogE("tx", les, err, logMsg)
636 func (ctx *Ctx) TxFreq(
638 nice, replyNice uint8,
639 srcPath, dstPath string,
640 minSize int64) error {
641 dstPath = filepath.Clean(dstPath)
642 if filepath.IsAbs(dstPath) {
643 return errors.New("Relative destination path required")
645 srcPath = filepath.Clean(srcPath)
646 if filepath.IsAbs(srcPath) {
647 return errors.New("Relative source path required")
649 pkt, err := NewPkt(PktTypeFreq, replyNice, []byte(srcPath))
653 src := strings.NewReader(dstPath)
654 size := int64(src.Len())
655 _, err = ctx.Tx(node, pkt, nice, size, minSize, src, srcPath, nil)
660 {"ReplyNice", int(replyNice)},
664 logMsg := func(les LEs) string {
666 "File request from %s:%s to %s sent",
667 ctx.NodeName(node.Id), srcPath,
672 ctx.LogI("tx", les, logMsg)
674 ctx.LogE("tx", les, err, logMsg)
679 func (ctx *Ctx) TxExec(
681 nice, replyNice uint8,
690 path := make([][]byte, 0, 1+len(args))
691 path = append(path, []byte(handle))
692 for _, arg := range args {
693 path = append(path, []byte(arg))
695 pktType := PktTypeExec
697 pktType = PktTypeExecFat
699 pkt, rerr := NewPkt(pktType, replyNice, bytes.Join(path, []byte{0}))
705 if !noCompress && !useTmp {
706 var compressed bytes.Buffer
707 compressor, err := zstd.NewWriter(
709 zstd.WithEncoderLevel(zstd.SpeedDefault),
714 if _, err = io.Copy(compressor, in); err != nil {
718 if err = compressor.Close(); err != nil {
721 size = int64(compressed.Len())
722 _, rerr = ctx.Tx(node, pkt, nice, size, minSize, &compressed, handle, areaId)
724 if noCompress && !useTmp {
725 var data bytes.Buffer
726 if _, err := io.Copy(&data, in); err != nil {
729 size = int64(data.Len())
730 _, rerr = ctx.Tx(node, pkt, nice, size, minSize, &data, handle, areaId)
732 if !noCompress && useTmp {
734 compressor, err := zstd.NewWriter(w, zstd.WithEncoderLevel(zstd.SpeedDefault))
738 copyErr := make(chan error)
740 _, err := io.Copy(compressor, in)
745 err = compressor.Close()
749 tmpReader, closer, fileSize, err := throughTmpFile(r)
761 _, rerr = ctx.Tx(node, pkt, nice, size, minSize, tmpReader, handle, areaId)
763 if noCompress && useTmp {
764 tmpReader, closer, fileSize, err := throughTmpFile(in)
772 _, rerr = ctx.Tx(node, pkt, nice, size, minSize, tmpReader, handle, areaId)
775 dst := strings.Join(append([]string{handle}, args...), " ")
780 {"ReplyNice", int(replyNice)},
784 logMsg := func(les LEs) string {
786 "Exec sent to %s@%s (%s)",
787 ctx.NodeName(node.Id), dst, humanize.IBytes(uint64(size)),
791 ctx.LogI("tx", les, logMsg)
793 ctx.LogE("tx", les, rerr, logMsg)
798 func (ctx *Ctx) TxTrns(node *Node, nice uint8, size int64, src io.Reader) error {
805 logMsg := func(les LEs) string {
807 "Transitional packet to %s (%s) (nice %s)",
808 ctx.NodeName(node.Id),
809 humanize.IBytes(uint64(size)),
813 ctx.LogD("tx", les, logMsg)
814 if !ctx.IsEnoughSpace(size) {
815 err := errors.New("is not enough space")
816 ctx.LogE("tx", les, err, logMsg)
819 tmp, err := ctx.NewTmpFileWHash()
823 if _, err = CopyProgressed(
824 tmp.W, src, "Tx trns",
825 LEs{{"Pkt", node.Id.String()}, {"FullSize", size}},
830 nodePath := filepath.Join(ctx.Spool, node.Id.String())
831 err = tmp.Commit(filepath.Join(nodePath, string(TTx)))
833 ctx.LogI("tx", les, logMsg)
835 ctx.LogI("tx", append(les, LE{"Err", err}), logMsg)
837 os.Symlink(nodePath, filepath.Join(ctx.Spool, node.Name))