src/toss.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, version 3 of the License.
   8
   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.
  13
  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/>.
  16 */
  17
  18 package nncp
  19
  20 import (
  21         "bufio"
  22         "bytes"
  23         "encoding/base64"
  24         "fmt"
  25         "io"
  26         "io/ioutil"
  27         "log"
  28         "mime"
  29         "os"
  30         "os/exec"
  31         "path"
  32         "path/filepath"
  33         "strconv"
  34         "strings"
  35
  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/poly1305"
  41 )
  42
  43 const (
  44         SeenSuffix = ".seen"
  45 )
  46
  47 func newNotification(fromTo *FromToJSON, subject string, body []byte) io.Reader {
  48         lines := []string{
  49                 "From: " + fromTo.From,
  50                 "To: " + fromTo.To,
  51                 "Subject: " + mime.BEncoding.Encode("UTF-8", subject),
  52         }
  53         if len(body) > 0 {
  54                 lines = append(lines, []string{
  55                         "MIME-Version: 1.0",
  56                         "Content-Type: text/plain; charset=utf-8",
  57                         "Content-Transfer-Encoding: base64",
  58                         "",
  59                         base64.StdEncoding.EncodeToString(body),
  60                 }...)
  61         }
  62         return strings.NewReader(strings.Join(lines, "\n"))
  63 }
  64
  65 func (ctx *Ctx) Toss(
  66         nodeId *NodeId,
  67         nice uint8,
  68         dryRun, doSeen, noFile, noFreq, noExec, noTrns bool,
  69 ) bool {
  70         isBad := false
  71         sendmail := ctx.Neigh[*ctx.SelfId].Exec["sendmail"]
  72         decompressor, err := zstd.NewReader(nil)
  73         if err != nil {
  74                 panic(err)
  75         }
  76         defer decompressor.Close()
  77         for job := range ctx.Jobs(nodeId, TRx) {
  78                 pktName := filepath.Base(job.Fd.Name())
  79                 sds := SDS{"node": job.PktEnc.Sender, "pkt": pktName}
  80                 if job.PktEnc.Nice > nice {
  81                         ctx.LogD("rx", SdsAdd(sds, SDS{
  82                                 "nice": strconv.Itoa(int(job.PktEnc.Nice)),
  83                         }), "too nice")
  84                         continue
  85                 }
  86                 pipeR, pipeW := io.Pipe()
  87                 errs := make(chan error, 1)
  88                 go func(job Job) {
  89                         pipeWB := bufio.NewWriter(pipeW)
  90                         _, _, err := PktEncRead(
  91                                 ctx.Self,
  92                                 ctx.Neigh,
  93                                 bufio.NewReader(job.Fd),
  94                                 pipeWB,
  95                         )
  96                         errs <- err
  97                         pipeWB.Flush()
  98                         pipeW.Close()
  99                         job.Fd.Close()
 100                         if err != nil {
 101                                 ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "decryption")
 102                         }
 103                 }(job)
 104                 var pkt Pkt
 105                 var err error
 106                 var pktSize int64
 107                 var pktSizeBlocks int64
 108                 if _, err = xdr.Unmarshal(pipeR, &pkt); err != nil {
 109                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "unmarshal")
 110                         isBad = true
 111                         goto Closing
 112                 }
 113                 pktSize = job.Size - PktEncOverhead - PktOverhead - PktSizeOverhead
 114                 pktSizeBlocks = pktSize / (EncBlkSize + poly1305.TagSize)
 115                 if pktSize%(EncBlkSize+poly1305.TagSize) != 0 {
 116                         pktSize -= poly1305.TagSize
 117                 }
 118                 pktSize -= pktSizeBlocks * poly1305.TagSize
 119                 sds["size"] = strconv.FormatInt(pktSize, 10)
 120                 ctx.LogD("rx", sds, "taken")
 121                 switch pkt.Type {
 122                 case PktTypeExec:
 123                         if noExec {
 124                                 goto Closing
 125                         }
 126                         path := bytes.Split(pkt.Path[:int(pkt.PathLen)], []byte{0})
 127                         handle := string(path[0])
 128                         args := make([]string, 0, len(path)-1)
 129                         for _, p := range path[1:] {
 130                                 args = append(args, string(p))
 131                         }
 132                         argsStr := strings.Join(append([]string{handle}, args...), " ")
 133                         sds := SdsAdd(sds, SDS{
 134                                 "type": "exec",
 135                                 "dst":  argsStr,
 136                         })
 137                         sender := ctx.Neigh[*job.PktEnc.Sender]
 138                         cmdline, exists := sender.Exec[handle]
 139                         if !exists || len(cmdline) == 0 {
 140                                 ctx.LogE("rx", SdsAdd(sds, SDS{"err": "No handle found"}), "")
 141                                 isBad = true
 142                                 goto Closing
 143                         }
 144                         if err = decompressor.Reset(pipeR); err != nil {
 145                                 log.Fatalln(err)
 146                         }
 147                         if !dryRun {
 148                                 cmd := exec.Command(
 149                                         cmdline[0],
 150                                         append(cmdline[1:len(cmdline)], args...)...,
 151                                 )
 152                                 cmd.Env = append(
 153                                         cmd.Env,
 154                                         "NNCP_SELF="+ctx.Self.Id.String(),
 155                                         "NNCP_SENDER="+sender.Id.String(),
 156                                         "NNCP_NICE="+strconv.Itoa(int(pkt.Nice)),
 157                                 )
 158                                 cmd.Stdin = decompressor
 159                                 output, err := cmd.Output()
 160                                 if err != nil {
 161                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "handle")
 162                                         isBad = true
 163                                         goto Closing
 164                                 }
 165                                 if len(sendmail) > 0 && ctx.NotifyExec != nil {
 166                                         notify, exists := ctx.NotifyExec[sender.Name+"."+handle]
 167                                         if !exists {
 168                                                 notify, exists = ctx.NotifyExec["*."+handle]
 169                                         }
 170                                         if exists {
 171                                                 cmd := exec.Command(
 172                                                         sendmail[0],
 173                                                         append(sendmail[1:len(sendmail)], notify.To)...,
 174                                                 )
 175                                                 cmd.Stdin = newNotification(notify, fmt.Sprintf(
 176                                                         "Exec from %s: %s", sender.Name, argsStr,
 177                                                 ), output)
 178                                                 cmd.Run()
 179                                         }
 180                                 }
 181                         }
 182                         ctx.LogI("rx", sds, "")
 183                         if !dryRun {
 184                                 if doSeen {
 185                                         if fd, err := os.Create(job.Fd.Name() + SeenSuffix); err == nil {
 186                                                 fd.Close()
 187                                         }
 188                                 }
 189                                 if err = os.Remove(job.Fd.Name()); err != nil {
 190                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove")
 191                                         isBad = true
 192                                 }
 193                         }
 194                 case PktTypeFile:
 195                         if noFile {
 196                                 goto Closing
 197                         }
 198                         dst := string(pkt.Path[:int(pkt.PathLen)])
 199                         sds := SdsAdd(sds, SDS{"type": "file", "dst": dst})
 200                         if filepath.IsAbs(dst) {
 201                                 ctx.LogE("rx", sds, "non-relative destination path")
 202                                 isBad = true
 203                                 goto Closing
 204                         }
 205                         incoming := ctx.Neigh[*job.PktEnc.Sender].Incoming
 206                         if incoming == nil {
 207                                 ctx.LogE("rx", sds, "incoming is not allowed")
 208                                 isBad = true
 209                                 goto Closing
 210                         }
 211                         dir := filepath.Join(*incoming, path.Dir(dst))
 212                         if err = os.MkdirAll(dir, os.FileMode(0777)); err != nil {
 213                                 ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "mkdir")
 214                                 isBad = true
 215                                 goto Closing
 216                         }
 217                         if !dryRun {
 218                                 tmp, err := TempFile(dir, "file")
 219                                 if err != nil {
 220                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "mktemp")
 221                                         isBad = true
 222                                         goto Closing
 223                                 }
 224                                 sds["tmp"] = tmp.Name()
 225                                 ctx.LogD("rx", sds, "created")
 226                                 bufW := bufio.NewWriter(tmp)
 227                                 if _, err = io.Copy(bufW, pipeR); err != nil {
 228                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "copy")
 229                                         isBad = true
 230                                         goto Closing
 231                                 }
 232                                 if err = bufW.Flush(); err != nil {
 233                                         tmp.Close()
 234                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "copy")
 235                                         isBad = true
 236                                         goto Closing
 237                                 }
 238                                 if err = tmp.Sync(); err != nil {
 239                                         tmp.Close()
 240                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "copy")
 241                                         isBad = true
 242                                         goto Closing
 243                                 }
 244                                 tmp.Close()
 245                                 dstPathOrig := filepath.Join(*incoming, dst)
 246                                 dstPath := dstPathOrig
 247                                 dstPathCtr := 0
 248                                 for {
 249                                         if _, err = os.Stat(dstPath); err != nil {
 250                                                 if os.IsNotExist(err) {
 251                                                         break
 252                                                 }
 253                                                 ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "stat")
 254                                                 isBad = true
 255                                                 goto Closing
 256                                         }
 257                                         dstPath = dstPathOrig + "." + strconv.Itoa(dstPathCtr)
 258                                         dstPathCtr++
 259                                 }
 260                                 if err = os.Rename(tmp.Name(), dstPath); err != nil {
 261                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "rename")
 262                                         isBad = true
 263                                 }
 264                                 delete(sds, "tmp")
 265                         }
 266                         ctx.LogI("rx", sds, "")
 267                         if !dryRun {
 268                                 if doSeen {
 269                                         if fd, err := os.Create(job.Fd.Name() + SeenSuffix); err == nil {
 270                                                 fd.Close()
 271                                         }
 272                                 }
 273                                 if err = os.Remove(job.Fd.Name()); err != nil {
 274                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove")
 275                                         isBad = true
 276                                 }
 277                                 if len(sendmail) > 0 && ctx.NotifyFile != nil {
 278                                         cmd := exec.Command(
 279                                                 sendmail[0],
 280                                                 append(sendmail[1:len(sendmail)], ctx.NotifyFile.To)...,
 281                                         )
 282                                         cmd.Stdin = newNotification(ctx.NotifyFile, fmt.Sprintf(
 283                                                 "File from %s: %s (%s)",
 284                                                 ctx.Neigh[*job.PktEnc.Sender].Name,
 285                                                 dst,
 286                                                 humanize.IBytes(uint64(pktSize)),
 287                                         ), nil)
 288                                         cmd.Run()
 289                                 }
 290                         }
 291                 case PktTypeFreq:
 292                         if noFreq {
 293                                 goto Closing
 294                         }
 295                         src := string(pkt.Path[:int(pkt.PathLen)])
 296                         if filepath.IsAbs(src) {
 297                                 ctx.LogE("rx", sds, "non-relative source path")
 298                                 isBad = true
 299                                 goto Closing
 300                         }
 301                         sds := SdsAdd(sds, SDS{"type": "freq", "src": src})
 302                         dstRaw, err := ioutil.ReadAll(pipeR)
 303                         if err != nil {
 304                                 ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "read")
 305                                 isBad = true
 306                                 goto Closing
 307                         }
 308                         dst := string(dstRaw)
 309                         sds["dst"] = dst
 310                         sender := ctx.Neigh[*job.PktEnc.Sender]
 311                         freqPath := sender.FreqPath
 312                         if freqPath == nil {
 313                                 ctx.LogE("rx", sds, "freqing is not allowed")
 314                                 isBad = true
 315                                 goto Closing
 316                         }
 317                         if !dryRun {
 318                                 err = ctx.TxFile(
 319                                         sender,
 320                                         pkt.Nice,
 321                                         filepath.Join(*freqPath, src),
 322                                         dst,
 323                                         sender.FreqChunked,
 324                                         sender.FreqMinSize,
 325                                         sender.FreqMaxSize,
 326                                 )
 327                                 if err != nil {
 328                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "tx file")
 329                                         isBad = true
 330                                         goto Closing
 331                                 }
 332                         }
 333                         ctx.LogI("rx", sds, "")
 334                         if !dryRun {
 335                                 if doSeen {
 336                                         if fd, err := os.Create(job.Fd.Name() + SeenSuffix); err == nil {
 337                                                 fd.Close()
 338                                         }
 339                                 }
 340                                 if err = os.Remove(job.Fd.Name()); err != nil {
 341                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove")
 342                                         isBad = true
 343                                 }
 344                                 if len(sendmail) > 0 && ctx.NotifyFreq != nil {
 345                                         cmd := exec.Command(
 346                                                 sendmail[0],
 347                                                 append(sendmail[1:len(sendmail)], ctx.NotifyFreq.To)...,
 348                                         )
 349                                         cmd.Stdin = newNotification(ctx.NotifyFreq, fmt.Sprintf(
 350                                                 "Freq from %s: %s", sender.Name, src,
 351                                         ), nil)
 352                                         cmd.Run()
 353                                 }
 354                         }
 355                 case PktTypeTrns:
 356                         if noTrns {
 357                                 goto Closing
 358                         }
 359                         dst := new([blake2b.Size256]byte)
 360                         copy(dst[:], pkt.Path[:int(pkt.PathLen)])
 361                         nodeId := NodeId(*dst)
 362                         node, known := ctx.Neigh[nodeId]
 363                         sds := SdsAdd(sds, SDS{"type": "trns", "dst": nodeId})
 364                         if !known {
 365                                 ctx.LogE("rx", sds, "unknown node")
 366                                 isBad = true
 367                                 goto Closing
 368                         }
 369                         ctx.LogD("rx", sds, "taken")
 370                         if !dryRun {
 371                                 if err = ctx.TxTrns(node, job.PktEnc.Nice, pktSize, pipeR); err != nil {
 372                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "tx trns")
 373                                         isBad = true
 374                                         goto Closing
 375                                 }
 376                         }
 377                         ctx.LogI("rx", sds, "")
 378                         if !dryRun {
 379                                 if doSeen {
 380                                         if fd, err := os.Create(job.Fd.Name() + SeenSuffix); err == nil {
 381                                                 fd.Close()
 382                                         }
 383                                 }
 384                                 if err = os.Remove(job.Fd.Name()); err != nil {
 385                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove")
 386                                         isBad = true
 387                                 }
 388                         }
 389                 default:
 390                         ctx.LogE("rx", sds, "unknown type")
 391                         isBad = true
 392                 }
 393         Closing:
 394                 pipeR.Close()
 395         }
 396         return isBad
 397 }