src/cypherpunks.ru/nncp/tx.go

   1 /*
   2 NNCP -- Node to Node copy, utilities for store-and-forward data exchange
   3 Copyright (C) 2016-2019 Sergey Matveev <stargrave@stargrave.org>
   4
   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.
   9
  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.
  14
  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/>.
  17 */
  18
  19 package nncp
  20
  21 import (
  22         "bufio"
  23         "bytes"
  24         "compress/zlib"
  25         "crypto/rand"
  26         "errors"
  27         "hash"
  28         "io"
  29         "io/ioutil"
  30         "os"
  31         "path/filepath"
  32         "strconv"
  33         "strings"
  34
  35         "github.com/davecgh/go-xdr/xdr2"
  36         "golang.org/x/crypto/blake2b"
  37         "golang.org/x/crypto/chacha20poly1305"
  38 )
  39
  40 func (ctx *Ctx) Tx(
  41         node *Node,
  42         pkt *Pkt,
  43         nice uint8,
  44         size, minSize int64,
  45         src io.Reader) (*Node, error) {
  46         tmp, err := ctx.NewTmpFileWHash()
  47         if err != nil {
  48                 return nil, err
  49         }
  50         hops := make([]*Node, 0, 1+len(node.Via))
  51         hops = append(hops, node)
  52         lastNode := node
  53         for i := len(node.Via); i > 0; i-- {
  54                 lastNode = ctx.Neigh[*node.Via[i-1]]
  55                 hops = append(hops, lastNode)
  56         }
  57         expectedSize := size
  58         for i := 0; i < len(hops); i++ {
  59                 expectedSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+expectedSize)
  60         }
  61         padSize := minSize - expectedSize
  62         if padSize < 0 {
  63                 padSize = 0
  64         }
  65         errs := make(chan error)
  66         curSize := size
  67         pipeR, pipeW := io.Pipe()
  68         go func(size int64, src io.Reader, dst io.WriteCloser) {
  69                 ctx.LogD("tx", SDS{
  70                         "node": hops[0].Id,
  71                         "nice": strconv.Itoa(int(nice)),
  72                         "size": strconv.FormatInt(size, 10),
  73                 }, "wrote")
  74                 errs <- PktEncWrite(ctx.Self, hops[0], pkt, nice, size, padSize, src, dst)
  75                 dst.Close()
  76         }(curSize, src, pipeW)
  77         curSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+curSize) + padSize
  78
  79         var pipeRPrev io.Reader
  80         for i := 1; i < len(hops); i++ {
  81                 pktTrns, _ := NewPkt(PktTypeTrns, 0, hops[i-1].Id[:])
  82                 pipeRPrev = pipeR
  83                 pipeR, pipeW = io.Pipe()
  84                 go func(node *Node, pkt *Pkt, size int64, src io.Reader, dst io.WriteCloser) {
  85                         ctx.LogD("tx", SDS{
  86                                 "node": node.Id,
  87                                 "nice": strconv.Itoa(int(nice)),
  88                                 "size": strconv.FormatInt(size, 10),
  89                         }, "trns wrote")
  90                         errs <- PktEncWrite(ctx.Self, node, pkt, nice, size, 0, src, dst)
  91                         dst.Close()
  92                 }(hops[i], pktTrns, curSize, pipeRPrev, pipeW)
  93                 curSize = PktEncOverhead + PktSizeOverhead + sizeWithTags(PktOverhead+curSize)
  94         }
  95         go func() {
  96                 _, err := io.Copy(tmp.W, pipeR)
  97                 errs <- err
  98         }()
  99         for i := 0; i <= len(hops); i++ {
 100                 err = <-errs
 101                 if err != nil {
 102                         tmp.Fd.Close()
 103                         return nil, err
 104                 }
 105         }
 106         nodePath := filepath.Join(ctx.Spool, lastNode.Id.String())
 107         err = tmp.Commit(filepath.Join(nodePath, string(TTx)))
 108         os.Symlink(nodePath, filepath.Join(ctx.Spool, lastNode.Name))
 109         return lastNode, err
 110 }
 111
 112 func prepareTxFile(srcPath string) (io.Reader, *os.File, int64, error) {
 113         var reader io.Reader
 114         var src *os.File
 115         var fileSize int64
 116         var err error
 117         if srcPath == "-" {
 118                 src, err = ioutil.TempFile("", "nncp-file")
 119                 if err != nil {
 120                         return nil, nil, 0, err
 121                 }
 122                 os.Remove(src.Name())
 123                 tmpW := bufio.NewWriter(src)
 124                 tmpKey := make([]byte, chacha20poly1305.KeySize)
 125                 if _, err = rand.Read(tmpKey[:]); err != nil {
 126                         return nil, nil, 0, err
 127                 }
 128                 aead, err := chacha20poly1305.New(tmpKey)
 129                 if err != nil {
 130                         return nil, nil, 0, err
 131                 }
 132                 nonce := make([]byte, aead.NonceSize())
 133                 written, err := aeadProcess(aead, nonce, true, bufio.NewReader(os.Stdin), tmpW)
 134                 if err != nil {
 135                         return nil, nil, 0, err
 136                 }
 137                 fileSize = int64(written)
 138                 if err = tmpW.Flush(); err != nil {
 139                         return nil, nil, 0, err
 140                 }
 141                 src.Seek(0, io.SeekStart)
 142                 r, w := io.Pipe()
 143                 go func() {
 144                         if _, err := aeadProcess(aead, nonce, false, bufio.NewReader(src), w); err != nil {
 145                                 panic(err)
 146                         }
 147                 }()
 148                 reader = r
 149         } else {
 150                 src, err = os.Open(srcPath)
 151                 if err != nil {
 152                         return nil, nil, 0, err
 153                 }
 154                 srcStat, err := src.Stat()
 155                 if err != nil {
 156                         return nil, nil, 0, err
 157                 }
 158                 fileSize = srcStat.Size()
 159                 reader = bufio.NewReader(src)
 160         }
 161         return reader, src, fileSize, nil
 162 }
 163
 164 func (ctx *Ctx) TxFile(node *Node, nice uint8, srcPath, dstPath string, minSize int64) error {
 165         if dstPath == "" {
 166                 if srcPath == "-" {
 167                         return errors.New("Must provide destination filename")
 168                 }
 169                 dstPath = filepath.Base(srcPath)
 170         }
 171         dstPath = filepath.Clean(dstPath)
 172         if filepath.IsAbs(dstPath) {
 173                 return errors.New("Relative destination path required")
 174         }
 175         pkt, err := NewPkt(PktTypeFile, nice, []byte(dstPath))
 176         if err != nil {
 177                 return err
 178         }
 179         reader, src, fileSize, err := prepareTxFile(srcPath)
 180         if src != nil {
 181                 defer src.Close()
 182         }
 183         if err != nil {
 184                 return err
 185         }
 186         _, err = ctx.Tx(node, pkt, nice, fileSize, minSize, reader)
 187         sds := SDS{
 188                 "type": "file",
 189                 "node": node.Id,
 190                 "nice": strconv.Itoa(int(nice)),
 191                 "src":  srcPath,
 192                 "dst":  dstPath,
 193                 "size": strconv.FormatInt(fileSize, 10),
 194         }
 195         if err == nil {
 196                 ctx.LogI("tx", sds, "sent")
 197         } else {
 198                 sds["err"] = err
 199                 ctx.LogE("tx", sds, "sent")
 200         }
 201         return err
 202 }
 203
 204 func (ctx *Ctx) TxFileChunked(
 205         node *Node,
 206         nice uint8,
 207         srcPath, dstPath string,
 208         minSize int64,
 209         chunkSize int64) error {
 210         if dstPath == "" {
 211                 if srcPath == "-" {
 212                         return errors.New("Must provide destination filename")
 213                 }
 214                 dstPath = filepath.Base(srcPath)
 215         }
 216         dstPath = filepath.Clean(dstPath)
 217         if filepath.IsAbs(dstPath) {
 218                 return errors.New("Relative destination path required")
 219         }
 220         reader, src, fileSize, err := prepareTxFile(srcPath)
 221         if src != nil {
 222                 defer src.Close()
 223         }
 224         if err != nil {
 225                 return err
 226         }
 227
 228         if fileSize <= chunkSize {
 229                 pkt, err := NewPkt(PktTypeFile, nice, []byte(dstPath))
 230                 if err != nil {
 231                         return err
 232                 }
 233                 _, err = ctx.Tx(node, pkt, nice, fileSize, minSize, reader)
 234                 sds := SDS{
 235                         "type": "file",
 236                         "node": node.Id,
 237                         "nice": strconv.Itoa(int(nice)),
 238                         "src":  srcPath,
 239                         "dst":  dstPath,
 240                         "size": strconv.FormatInt(fileSize, 10),
 241                 }
 242                 if err == nil {
 243                         ctx.LogI("tx", sds, "sent")
 244                 } else {
 245                         sds["err"] = err
 246                         ctx.LogE("tx", sds, "sent")
 247                 }
 248                 return err
 249         }
 250
 251         leftSize := fileSize
 252         metaPkt := ChunkedMeta{
 253                 Magic:     MagicNNCPMv1,
 254                 FileSize:  uint64(fileSize),
 255                 ChunkSize: uint64(chunkSize),
 256                 Checksums: make([][32]byte, 0, (fileSize/chunkSize)+1),
 257         }
 258         for i := int64(0); i < (fileSize/chunkSize)+1; i++ {
 259                 hsh := new([32]byte)
 260                 metaPkt.Checksums = append(metaPkt.Checksums, *hsh)
 261         }
 262         var sizeToSend int64
 263         var hsh hash.Hash
 264         var pkt *Pkt
 265         var chunkNum int
 266         var path string
 267         for {
 268                 if leftSize <= chunkSize {
 269                         sizeToSend = leftSize
 270                 } else {
 271                         sizeToSend = chunkSize
 272                 }
 273                 path = dstPath + ChunkedSuffixPart + strconv.Itoa(chunkNum)
 274                 pkt, err = NewPkt(PktTypeFile, nice, []byte(path))
 275                 if err != nil {
 276                         return err
 277                 }
 278                 hsh, err = blake2b.New256(nil)
 279                 if err != nil {
 280                         return err
 281                 }
 282                 _, err = ctx.Tx(
 283                         node,
 284                         pkt,
 285                         nice,
 286                         sizeToSend,
 287                         minSize,
 288                         io.TeeReader(reader, hsh),
 289                 )
 290                 sds := SDS{
 291                         "type": "file",
 292                         "node": node.Id,
 293                         "nice": strconv.Itoa(int(nice)),
 294                         "src":  srcPath,
 295                         "dst":  path,
 296                         "size": strconv.FormatInt(sizeToSend, 10),
 297                 }
 298                 if err == nil {
 299                         ctx.LogI("tx", sds, "sent")
 300                 } else {
 301                         sds["err"] = err
 302                         ctx.LogE("tx", sds, "sent")
 303                         return err
 304                 }
 305                 hsh.Sum(metaPkt.Checksums[chunkNum][:0])
 306                 leftSize -= sizeToSend
 307                 chunkNum++
 308                 if leftSize == 0 {
 309                         break
 310                 }
 311         }
 312         var metaBuf bytes.Buffer
 313         _, err = xdr.Marshal(&metaBuf, metaPkt)
 314         if err != nil {
 315                 return err
 316         }
 317         path = dstPath + ChunkedSuffixMeta
 318         pkt, err = NewPkt(PktTypeFile, nice, []byte(path))
 319         if err != nil {
 320                 return err
 321         }
 322         metaPktSize := int64(metaBuf.Len())
 323         _, err = ctx.Tx(node, pkt, nice, metaPktSize, minSize, &metaBuf)
 324         sds := SDS{
 325                 "type": "file",
 326                 "node": node.Id,
 327                 "nice": strconv.Itoa(int(nice)),
 328                 "src":  srcPath,
 329                 "dst":  path,
 330                 "size": strconv.FormatInt(metaPktSize, 10),
 331         }
 332         if err == nil {
 333                 ctx.LogI("tx", sds, "sent")
 334         } else {
 335                 sds["err"] = err
 336                 ctx.LogE("tx", sds, "sent")
 337         }
 338         return err
 339 }
 340
 341 func (ctx *Ctx) TxFreq(
 342         node *Node,
 343         nice, replyNice uint8,
 344         srcPath, dstPath string,
 345         minSize int64) error {
 346         dstPath = filepath.Clean(dstPath)
 347         if filepath.IsAbs(dstPath) {
 348                 return errors.New("Relative destination path required")
 349         }
 350         srcPath = filepath.Clean(srcPath)
 351         if filepath.IsAbs(srcPath) {
 352                 return errors.New("Relative source path required")
 353         }
 354         pkt, err := NewPkt(PktTypeFreq, replyNice, []byte(srcPath))
 355         if err != nil {
 356                 return err
 357         }
 358         src := strings.NewReader(dstPath)
 359         size := int64(src.Len())
 360         _, err = ctx.Tx(node, pkt, nice, size, minSize, src)
 361         sds := SDS{
 362                 "type":      "freq",
 363                 "node":      node.Id,
 364                 "nice":      strconv.Itoa(int(nice)),
 365                 "replynice": strconv.Itoa(int(replyNice)),
 366                 "src":       srcPath,
 367                 "dst":       dstPath,
 368         }
 369         if err == nil {
 370                 ctx.LogI("tx", sds, "sent")
 371         } else {
 372                 sds["err"] = err
 373                 ctx.LogE("tx", sds, "sent")
 374         }
 375         return err
 376 }
 377
 378 func (ctx *Ctx) TxExec(
 379         node *Node,
 380         nice, replyNice uint8,
 381         handle string,
 382         args []string,
 383         body []byte,
 384         minSize int64) error {
 385         path := make([][]byte, 0, 1+len(args))
 386         path = append(path, []byte(handle))
 387         for _, arg := range args {
 388                 path = append(path, []byte(arg))
 389         }
 390         pkt, err := NewPkt(PktTypeExec, replyNice, bytes.Join(path, []byte{0}))
 391         if err != nil {
 392                 return err
 393         }
 394         var compressed bytes.Buffer
 395         compressor, err := zlib.NewWriterLevel(&compressed, zlib.BestCompression)
 396         if err != nil {
 397                 return err
 398         }
 399         if _, err = io.Copy(compressor, bytes.NewReader(body)); err != nil {
 400                 return err
 401         }
 402         compressor.Close()
 403         size := int64(compressed.Len())
 404         _, err = ctx.Tx(node, pkt, nice, size, minSize, &compressed)
 405         sds := SDS{
 406                 "type":      "exec",
 407                 "node":      node.Id,
 408                 "nice":      strconv.Itoa(int(nice)),
 409                 "replynice": strconv.Itoa(int(replyNice)),
 410                 "dst":       strings.Join(append([]string{handle}, args...), " "),
 411                 "size":      strconv.FormatInt(size, 10),
 412         }
 413         if err == nil {
 414                 ctx.LogI("tx", sds, "sent")
 415         } else {
 416                 sds["err"] = err
 417                 ctx.LogE("tx", sds, "sent")
 418         }
 419         return err
 420 }
 421
 422 func (ctx *Ctx) TxTrns(node *Node, nice uint8, size int64, src io.Reader) error {
 423         sds := SDS{
 424                 "type": "trns",
 425                 "node": node.Id,
 426                 "nice": strconv.Itoa(int(nice)),
 427                 "size": strconv.FormatInt(size, 10),
 428         }
 429         ctx.LogD("tx", sds, "taken")
 430         tmp, err := ctx.NewTmpFileWHash()
 431         if err != nil {
 432                 return err
 433         }
 434         if _, err = io.Copy(tmp.W, src); err != nil {
 435                 return err
 436         }
 437         nodePath := filepath.Join(ctx.Spool, node.Id.String())
 438         err = tmp.Commit(filepath.Join(nodePath, string(TTx)))
 439         if err == nil {
 440                 ctx.LogI("tx", sds, "sent")
 441         } else {
 442                 sds["err"] = err
 443                 ctx.LogI("tx", sds, "sent")
 444         }
 445         os.Symlink(nodePath, filepath.Join(ctx.Spool, node.Name))
 446         return err
 447 }