2 NNCP -- Node to Node copy, utilities for store-and-forward data exchange
3 Copyright (C) 2016-2017 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"
39 func (ctx *Ctx) Tx(node *Node, pkt *Pkt, nice uint8, size, minSize int64, src io.Reader) (*Node, error) {
40 tmp, err := ctx.NewTmpFileWHash()
44 hops := make([]*Node, 0, 1+len(node.Via))
45 hops = append(hops, node)
47 for i := len(node.Via); i > 0; i-- {
48 lastNode = ctx.Neigh[*node.Via[i-1]]
49 hops = append(hops, lastNode)
51 padSize := minSize - size - int64(len(hops))*(PktOverhead+PktEncOverhead)
55 errs := make(chan error)
57 pipeR, pipeW := io.Pipe()
58 go func(size int64, src io.Reader, dst io.WriteCloser) {
61 "nice": strconv.Itoa(int(nice)),
62 "size": strconv.FormatInt(size, 10),
64 errs <- PktEncWrite(ctx.Self, hops[0], pkt, nice, size, padSize, src, dst)
66 }(curSize, src, pipeW)
69 var pipeRPrev io.Reader
70 for i := 1; i < len(hops); i++ {
74 PathLen: blake2b.Size256,
75 Path: new([MaxPathSize]byte),
77 copy(pktTrans.Path[:], hops[i-1].Id[:])
78 curSize += PktOverhead + PktEncOverhead
80 pipeR, pipeW = io.Pipe()
81 go func(node *Node, pkt *Pkt, size int64, src io.Reader, dst io.WriteCloser) {
84 "nice": strconv.Itoa(int(nice)),
85 "size": strconv.FormatInt(size, 10),
87 errs <- PktEncWrite(ctx.Self, node, pkt, nice, size, 0, src, dst)
89 }(hops[i], &pktTrans, curSize, pipeRPrev, pipeW)
92 _, err := io.Copy(tmp.W, pipeR)
95 for i := 0; i <= len(hops); i++ {
102 nodePath := filepath.Join(ctx.Spool, lastNode.Id.String())
103 err = tmp.Commit(filepath.Join(nodePath, string(TTx)))
104 os.Symlink(nodePath, filepath.Join(ctx.Spool, lastNode.Name))
108 func prepareTxFile(srcPath string) (io.Reader, *os.File, int64, error) {
114 src, err = ioutil.TempFile("", "nncp-file")
116 return nil, nil, 0, err
118 os.Remove(src.Name())
119 tmpW := bufio.NewWriter(src)
120 tmpKey := new([32]byte)
121 if _, err = rand.Read(tmpKey[:]); err != nil {
122 return nil, nil, 0, err
124 written, err := ae(tmpKey, bufio.NewReader(os.Stdin), tmpW)
126 return nil, nil, 0, err
128 fileSize = int64(written)
132 go ae(tmpKey, bufio.NewReader(src), w)
135 src, err = os.Open(srcPath)
137 return nil, nil, 0, err
139 srcStat, err := src.Stat()
141 return nil, nil, 0, err
143 fileSize = srcStat.Size()
144 reader = bufio.NewReader(src)
146 return reader, src, fileSize, nil
149 func (ctx *Ctx) TxFile(node *Node, nice uint8, srcPath, dstPath string, minSize int64) error {
152 return errors.New("Must provide destination filename")
154 dstPath = filepath.Base(srcPath)
156 dstPath = filepath.Clean(dstPath)
157 if filepath.IsAbs(dstPath) {
158 return errors.New("Relative destination path required")
160 pkt, err := NewPkt(PktTypeFile, dstPath)
164 reader, src, fileSize, err := prepareTxFile(srcPath)
171 _, err = ctx.Tx(node, pkt, nice, fileSize, minSize, reader)
176 "nice": strconv.Itoa(int(nice)),
179 "size": strconv.FormatInt(fileSize, 10),
185 "nice": strconv.Itoa(int(nice)),
188 "size": strconv.FormatInt(fileSize, 10),
195 func (ctx *Ctx) TxFileChunked(node *Node, nice uint8, srcPath, dstPath string, minSize int64, chunkSize int64) error {
198 return errors.New("Must provide destination filename")
200 dstPath = filepath.Base(srcPath)
202 dstPath = filepath.Clean(dstPath)
203 if filepath.IsAbs(dstPath) {
204 return errors.New("Relative destination path required")
206 reader, src, fileSize, err := prepareTxFile(srcPath)
215 metaPkt := ChunkedMeta{
217 FileSize: uint64(fileSize),
218 ChunkSize: uint64(chunkSize),
219 Checksums: make([][32]byte, 0, (fileSize/chunkSize)+1),
221 for i := int64(0); i < (fileSize/chunkSize)+1; i++ {
223 metaPkt.Checksums = append(metaPkt.Checksums, *hsh)
231 if leftSize <= chunkSize {
232 sizeToSend = leftSize
234 sizeToSend = chunkSize
236 path = dstPath + ChunkedSuffixPart + strconv.Itoa(chunkNum)
237 pkt, err = NewPkt(PktTypeFile, path)
241 hsh, err = blake2b.New256(nil)
251 io.TeeReader(reader, hsh),
257 "nice": strconv.Itoa(int(nice)),
260 "size": strconv.FormatInt(sizeToSend, 10),
266 "nice": strconv.Itoa(int(nice)),
269 "size": strconv.FormatInt(sizeToSend, 10),
274 hsh.Sum(metaPkt.Checksums[chunkNum][:0])
275 leftSize -= sizeToSend
281 var metaBuf bytes.Buffer
282 _, err = xdr.Marshal(&metaBuf, metaPkt)
286 path = dstPath + ChunkedSuffixMeta
287 pkt, err = NewPkt(PktTypeFile, path)
291 metaPktSize := int64(metaBuf.Len())
292 _, err = ctx.Tx(node, pkt, nice, metaPktSize, minSize, &metaBuf)
297 "nice": strconv.Itoa(int(nice)),
300 "size": strconv.FormatInt(metaPktSize, 10),
306 "nice": strconv.Itoa(int(nice)),
309 "size": strconv.FormatInt(metaPktSize, 10),
316 func (ctx *Ctx) TxFreq(node *Node, nice uint8, srcPath, dstPath string, minSize int64) error {
317 dstPath = filepath.Clean(dstPath)
318 if filepath.IsAbs(dstPath) {
319 return errors.New("Relative destination path required")
321 srcPath = filepath.Clean(srcPath)
322 if filepath.IsAbs(srcPath) {
323 return errors.New("Relative source path required")
325 pkt, err := NewPkt(PktTypeFreq, srcPath)
329 src := strings.NewReader(dstPath)
330 size := int64(src.Len())
331 _, err = ctx.Tx(node, pkt, nice, size, minSize, src)
336 "nice": strconv.Itoa(int(nice)),
344 "nice": strconv.Itoa(int(nice)),
353 func (ctx *Ctx) TxMail(node *Node, nice uint8, recipient string, body []byte, minSize int64) error {
354 pkt, err := NewPkt(PktTypeMail, recipient)
358 var compressed bytes.Buffer
359 compressor, err := zlib.NewWriterLevel(&compressed, zlib.BestCompression)
363 if _, err = io.Copy(compressor, bytes.NewReader(body)); err != nil {
367 size := int64(compressed.Len())
368 _, err = ctx.Tx(node, pkt, nice, size, minSize, &compressed)
373 "nice": strconv.Itoa(int(nice)),
375 "size": strconv.FormatInt(size, 10),
381 "nice": strconv.Itoa(int(nice)),
383 "size": strconv.FormatInt(size, 10),
390 func (ctx *Ctx) TxTrns(node *Node, nice uint8, size int64, src io.Reader) error {
394 "nice": strconv.Itoa(int(nice)),
395 "size": strconv.FormatInt(size, 10),
397 tmp, err := ctx.NewTmpFileWHash()
401 if _, err = io.Copy(tmp.W, src); err != nil {
404 nodePath := filepath.Join(ctx.Spool, node.Id.String())
405 err = tmp.Commit(filepath.Join(nodePath, string(TTx)))
410 "nice": strconv.Itoa(int(nice)),
411 "size": strconv.FormatInt(size, 10),
417 "nice": strconv.Itoa(int(nice)),
418 "size": strconv.FormatInt(size, 10),
422 os.Symlink(nodePath, filepath.Join(ctx.Spool, node.Name))