2 NNCP -- Node to Node copy, utilities for store-and-forward data exchange
3 Copyright (C) 2016-2019 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, either version 3 of the License, or
8 (at your option) any later version.
10 This program is distributed in the hope that it will be useful,
11 but WITHOUT ANY WARRANTY; without even the implied warranty of
12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 GNU General Public License for more details.
15 You should have received a copy of the GNU General Public License
16 along with this program. If not, see <http://www.gnu.org/licenses/>.
35 "github.com/davecgh/go-xdr/xdr2"
36 "golang.org/x/crypto/blake2b"
37 "golang.org/x/crypto/chacha20poly1305"
47 tmp, err := ctx.NewTmpFileWHash()
51 hops := make([]*Node, 0, 1+len(node.Via))
52 hops = append(hops, node)
54 for i := len(node.Via); i > 0; i-- {
55 lastNode = ctx.Neigh[*node.Via[i-1]]
56 hops = append(hops, lastNode)
59 for i := 0; i < len(hops); i++ {
60 expectedSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+expectedSize)
62 padSize := minSize - expectedSize
66 errs := make(chan error)
68 pipeR, pipeW := io.Pipe()
69 go func(size int64, src io.Reader, dst io.WriteCloser) {
72 "nice": strconv.Itoa(int(nice)),
73 "size": strconv.FormatInt(size, 10),
75 errs <- PktEncWrite(ctx.Self, hops[0], pkt, nice, size, padSize, src, dst)
77 }(curSize, src, pipeW)
78 curSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+curSize) + padSize
80 var pipeRPrev io.Reader
81 for i := 1; i < len(hops); i++ {
82 pktTrns, _ := NewPkt(PktTypeTrns, 0, hops[i-1].Id[:])
84 pipeR, pipeW = io.Pipe()
85 go func(node *Node, pkt *Pkt, size int64, src io.Reader, dst io.WriteCloser) {
88 "nice": strconv.Itoa(int(nice)),
89 "size": strconv.FormatInt(size, 10),
91 errs <- PktEncWrite(ctx.Self, node, pkt, nice, size, 0, src, dst)
93 }(hops[i], pktTrns, curSize, pipeRPrev, pipeW)
94 curSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+curSize)
97 _, err := io.Copy(tmp.W, pipeR)
100 for i := 0; i <= len(hops); i++ {
107 nodePath := filepath.Join(ctx.Spool, lastNode.Id.String())
108 err = tmp.Commit(filepath.Join(nodePath, string(TTx)))
109 os.Symlink(nodePath, filepath.Join(ctx.Spool, lastNode.Name))
113 func prepareTxFile(srcPath string) (io.Reader, *os.File, int64, error) {
119 src, err = ioutil.TempFile("", "nncp-file")
121 return nil, nil, 0, err
123 os.Remove(src.Name())
124 tmpW := bufio.NewWriter(src)
125 tmpKey := make([]byte, chacha20poly1305.KeySize)
126 if _, err = rand.Read(tmpKey[:]); err != nil {
127 return nil, nil, 0, err
129 aead, err := chacha20poly1305.New(tmpKey)
131 return nil, nil, 0, err
133 nonce := make([]byte, aead.NonceSize())
134 written, err := aeadProcess(aead, nonce, true, bufio.NewReader(os.Stdin), tmpW)
136 return nil, nil, 0, err
138 fileSize = int64(written)
139 if err = tmpW.Flush(); err != nil {
140 return nil, nil, 0, err
142 src.Seek(0, io.SeekStart)
145 if _, err := aeadProcess(aead, nonce, false, bufio.NewReader(src), w); err != nil {
151 src, err = os.Open(srcPath)
153 return nil, nil, 0, err
155 srcStat, err := src.Stat()
157 return nil, nil, 0, err
159 fileSize = srcStat.Size()
160 reader = bufio.NewReader(src)
162 return reader, src, fileSize, nil
165 func (ctx *Ctx) TxFile(node *Node, nice uint8, srcPath, dstPath string, minSize int64) error {
168 return errors.New("Must provide destination filename")
170 dstPath = filepath.Base(srcPath)
172 dstPath = filepath.Clean(dstPath)
173 if filepath.IsAbs(dstPath) {
174 return errors.New("Relative destination path required")
176 pkt, err := NewPkt(PktTypeFile, nice, []byte(dstPath))
180 reader, src, fileSize, err := prepareTxFile(srcPath)
187 _, err = ctx.Tx(node, pkt, nice, fileSize, minSize, reader)
191 "nice": strconv.Itoa(int(nice)),
194 "size": strconv.FormatInt(fileSize, 10),
197 ctx.LogI("tx", sds, "sent")
200 ctx.LogE("tx", sds, "sent")
205 func (ctx *Ctx) TxFileChunked(
208 srcPath, dstPath string,
214 return errors.New("Must provide destination filename")
216 dstPath = filepath.Base(srcPath)
218 dstPath = filepath.Clean(dstPath)
219 if filepath.IsAbs(dstPath) {
220 return errors.New("Relative destination path required")
222 reader, src, fileSize, err := prepareTxFile(srcPath)
230 if fileSize <= chunkSize {
231 pkt, err := NewPkt(PktTypeFile, nice, []byte(dstPath))
235 _, err = ctx.Tx(node, pkt, nice, fileSize, minSize, reader)
239 "nice": strconv.Itoa(int(nice)),
242 "size": strconv.FormatInt(fileSize, 10),
245 ctx.LogI("tx", sds, "sent")
248 ctx.LogE("tx", sds, "sent")
254 metaPkt := ChunkedMeta{
256 FileSize: uint64(fileSize),
257 ChunkSize: uint64(chunkSize),
258 Checksums: make([][32]byte, 0, (fileSize/chunkSize)+1),
260 for i := int64(0); i < (fileSize/chunkSize)+1; i++ {
262 metaPkt.Checksums = append(metaPkt.Checksums, *hsh)
270 if leftSize <= chunkSize {
271 sizeToSend = leftSize
273 sizeToSend = chunkSize
275 path = dstPath + ChunkedSuffixPart + strconv.Itoa(chunkNum)
276 pkt, err = NewPkt(PktTypeFile, nice, []byte(path))
280 hsh, err = blake2b.New256(nil)
290 io.TeeReader(reader, hsh),
295 "nice": strconv.Itoa(int(nice)),
298 "size": strconv.FormatInt(sizeToSend, 10),
301 ctx.LogI("tx", sds, "sent")
304 ctx.LogE("tx", sds, "sent")
307 hsh.Sum(metaPkt.Checksums[chunkNum][:0])
308 leftSize -= sizeToSend
314 var metaBuf bytes.Buffer
315 _, err = xdr.Marshal(&metaBuf, metaPkt)
319 path = dstPath + ChunkedSuffixMeta
320 pkt, err = NewPkt(PktTypeFile, nice, []byte(path))
324 metaPktSize := int64(metaBuf.Len())
325 _, err = ctx.Tx(node, pkt, nice, metaPktSize, minSize, &metaBuf)
329 "nice": strconv.Itoa(int(nice)),
332 "size": strconv.FormatInt(metaPktSize, 10),
335 ctx.LogI("tx", sds, "sent")
338 ctx.LogE("tx", sds, "sent")
343 func (ctx *Ctx) TxFreq(
345 nice, replyNice uint8,
346 srcPath, dstPath string,
347 minSize int64) error {
348 dstPath = filepath.Clean(dstPath)
349 if filepath.IsAbs(dstPath) {
350 return errors.New("Relative destination path required")
352 srcPath = filepath.Clean(srcPath)
353 if filepath.IsAbs(srcPath) {
354 return errors.New("Relative source path required")
356 pkt, err := NewPkt(PktTypeFreq, replyNice, []byte(srcPath))
360 src := strings.NewReader(dstPath)
361 size := int64(src.Len())
362 _, err = ctx.Tx(node, pkt, nice, size, minSize, src)
366 "nice": strconv.Itoa(int(nice)),
367 "replynice": strconv.Itoa(int(replyNice)),
372 ctx.LogI("tx", sds, "sent")
375 ctx.LogE("tx", sds, "sent")
380 func (ctx *Ctx) TxExec(
382 nice, replyNice uint8,
388 path := make([][]byte, 0, 1+len(args))
389 path = append(path, []byte(handle))
390 for _, arg := range args {
391 path = append(path, []byte(arg))
393 pkt, err := NewPkt(PktTypeExec, replyNice, bytes.Join(path, []byte{0}))
397 var compressed bytes.Buffer
398 compressor, err := zlib.NewWriterLevel(&compressed, zlib.BestCompression)
402 if _, err = io.Copy(compressor, bytes.NewReader(body)); err != nil {
406 size := int64(compressed.Len())
407 _, err = ctx.Tx(node, pkt, nice, size, minSize, &compressed)
411 "nice": strconv.Itoa(int(nice)),
412 "replynice": strconv.Itoa(int(replyNice)),
413 "dst": strings.Join(append([]string{handle}, args...), " "),
414 "size": strconv.FormatInt(size, 10),
417 ctx.LogI("tx", sds, "sent")
420 ctx.LogE("tx", sds, "sent")
425 func (ctx *Ctx) TxTrns(node *Node, nice uint8, size int64, src io.Reader) error {
429 "nice": strconv.Itoa(int(nice)),
430 "size": strconv.FormatInt(size, 10),
432 ctx.LogD("tx", sds, "taken")
433 tmp, err := ctx.NewTmpFileWHash()
437 if _, err = io.Copy(tmp.W, src); err != nil {
440 nodePath := filepath.Join(ctx.Spool, node.Id.String())
441 err = tmp.Commit(filepath.Join(nodePath, string(TTx)))
443 ctx.LogI("tx", sds, "sent")
446 ctx.LogI("tx", sds, "sent")
448 os.Symlink(nodePath, filepath.Join(ctx.Spool, node.Name))