src/cypherpunks.ru/nncp/toss.go

   1 /*
   2 NNCP -- Node to Node copy, utilities for store-and-forward data exchange
   3 Copyright (C) 2016-2017 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         "compress/zlib"
  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         "github.com/davecgh/go-xdr/xdr2"
  37         "github.com/dustin/go-humanize"
  38         "golang.org/x/crypto/blake2b"
  39 )
  40
  41 func newNotification(fromTo *FromToYAML, subject string) io.Reader {
  42         return strings.NewReader(fmt.Sprintf(
  43                 "From: %s\nTo: %s\nSubject: %s\n",
  44                 fromTo.From,
  45                 fromTo.To,
  46                 mime.BEncoding.Encode("UTF-8", subject),
  47         ))
  48 }
  49
  50 func (ctx *Ctx) Toss(nodeId *NodeId, nice uint8, dryRun bool) bool {
  51         isBad := false
  52         for job := range ctx.Jobs(nodeId, TRx) {
  53                 pktName := filepath.Base(job.Fd.Name())
  54                 sds := SDS{"node": job.PktEnc.Sender, "pkt": pktName}
  55                 if job.PktEnc.Nice > nice {
  56                         ctx.LogD("rx", SdsAdd(sds, SDS{
  57                                 "nice": strconv.Itoa(int(job.PktEnc.Nice)),
  58                         }), "too nice")
  59                         continue
  60                 }
  61                 pipeR, pipeW := io.Pipe()
  62                 errs := make(chan error, 1)
  63                 go func(job Job) {
  64                         pipeWB := bufio.NewWriter(pipeW)
  65                         _, _, err := PktEncRead(
  66                                 ctx.Self,
  67                                 ctx.Neigh,
  68                                 bufio.NewReader(job.Fd),
  69                                 pipeWB,
  70                         )
  71                         errs <- err
  72                         pipeWB.Flush()
  73                         pipeW.Close()
  74                         job.Fd.Close()
  75                         if err != nil {
  76                                 ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "decryption")
  77                         }
  78                 }(job)
  79                 var pkt Pkt
  80                 var err error
  81                 var pktSize int64
  82                 if _, err = xdr.Unmarshal(pipeR, &pkt); err != nil {
  83                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "unmarshal")
  84                         isBad = true
  85                         goto Closing
  86                 }
  87                 pktSize = job.Size - PktEncOverhead - PktOverhead
  88                 sds["size"] = strconv.FormatInt(pktSize, 10)
  89                 ctx.LogD("rx", sds, "taken")
  90                 switch pkt.Type {
  91                 case PktTypeMail:
  92                         recipients := string(pkt.Path[:int(pkt.PathLen)])
  93                         sds := SdsAdd(sds, SDS{
  94                                 "type": "mail",
  95                                 "dst":  recipients,
  96                         })
  97                         decompressor, err := zlib.NewReader(pipeR)
  98                         if err != nil {
  99                                 log.Fatalln(err)
 100                         }
 101                         sender := ctx.Neigh[*job.PktEnc.Sender]
 102                         sendmail := sender.Sendmail
 103                         if len(sendmail) == 0 {
 104                                 ctx.LogE("rx", SdsAdd(sds, SDS{"err": "No sendmail configured"}), "")
 105                                 isBad = true
 106                                 goto Closing
 107                         }
 108                         if !dryRun {
 109                                 cmd := exec.Command(
 110                                         sendmail[0],
 111                                         append(
 112                                                 sendmail[1:len(sendmail)],
 113                                                 strings.Split(recipients, " ")...,
 114                                         )...,
 115                                 )
 116                                 cmd.Env = append(cmd.Env, "NNCP_SENDER=" + sender.Id.String())
 117                                 cmd.Stdin = decompressor
 118                                 if err = cmd.Run(); err != nil {
 119                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "sendmail")
 120                                         isBad = true
 121                                         goto Closing
 122                                 }
 123                         }
 124                         ctx.LogI("rx", sds, "")
 125                         if !dryRun {
 126                                 if err = os.Remove(job.Fd.Name()); err != nil {
 127                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove")
 128                                         isBad = true
 129                                 }
 130                         }
 131                 case PktTypeFile:
 132                         dst := string(pkt.Path[:int(pkt.PathLen)])
 133                         sds := SdsAdd(sds, SDS{"type": "file", "dst": dst})
 134                         if filepath.IsAbs(dst) {
 135                                 ctx.LogE("rx", sds, "non-relative destination path")
 136                                 isBad = true
 137                                 goto Closing
 138                         }
 139                         incoming := ctx.Neigh[*job.PktEnc.Sender].Incoming
 140                         if incoming == nil {
 141                                 ctx.LogE("rx", sds, "incoming is not allowed")
 142                                 isBad = true
 143                                 goto Closing
 144                         }
 145                         dir := filepath.Join(*incoming, path.Dir(dst))
 146                         if err = os.MkdirAll(dir, os.FileMode(0700)); err != nil {
 147                                 ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "mkdir")
 148                                 isBad = true
 149                                 goto Closing
 150                         }
 151                         if !dryRun {
 152                                 tmp, err := ioutil.TempFile(dir, "nncp-file")
 153                                 sds["tmp"] = tmp.Name()
 154                                 ctx.LogD("rx", sds, "created")
 155                                 if err != nil {
 156                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "mktemp")
 157                                         isBad = true
 158                                         goto Closing
 159                                 }
 160                                 bufW := bufio.NewWriter(tmp)
 161                                 if _, err = io.Copy(bufW, pipeR); err != nil {
 162                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "copy")
 163                                         isBad = true
 164                                         goto Closing
 165                                 }
 166                                 bufW.Flush()
 167                                 tmp.Sync()
 168                                 tmp.Close()
 169                                 dstPathOrig := filepath.Join(*incoming, dst)
 170                                 dstPath := dstPathOrig
 171                                 dstPathCtr := 0
 172                                 for {
 173                                         if _, err = os.Stat(dstPath); err != nil {
 174                                                 if os.IsNotExist(err) {
 175                                                         break
 176                                                 }
 177                                                 ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "stat")
 178                                                 isBad = true
 179                                                 goto Closing
 180                                         }
 181                                         dstPath = dstPathOrig + strconv.Itoa(dstPathCtr)
 182                                         dstPathCtr++
 183                                 }
 184                                 if err = os.Rename(tmp.Name(), dstPath); err != nil {
 185                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "rename")
 186                                         isBad = true
 187                                 }
 188                                 delete(sds, "tmp")
 189                         }
 190                         ctx.LogI("rx", sds, "")
 191                         if !dryRun {
 192                                 if err = os.Remove(job.Fd.Name()); err != nil {
 193                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove")
 194                                         isBad = true
 195                                 }
 196                                 sendmail := ctx.Neigh[*ctx.SelfId].Sendmail
 197                                 if ctx.NotifyFile != nil {
 198                                         cmd := exec.Command(
 199                                                 sendmail[0],
 200                                                 append(sendmail[1:len(sendmail)], ctx.NotifyFile.To)...,
 201                                         )
 202                                         cmd.Stdin = newNotification(ctx.NotifyFile, fmt.Sprintf(
 203                                                 "File from %s: %s (%s)",
 204                                                 ctx.Neigh[*job.PktEnc.Sender].Name,
 205                                                 dst,
 206                                                 humanize.IBytes(uint64(pktSize)),
 207                                         ))
 208                                         cmd.Run()
 209                                 }
 210                         }
 211                 case PktTypeFreq:
 212                         src := string(pkt.Path[:int(pkt.PathLen)])
 213                         if filepath.IsAbs(src) {
 214                                 ctx.LogE("rx", sds, "non-relative source path")
 215                                 isBad = true
 216                                 goto Closing
 217                         }
 218                         sds := SdsAdd(sds, SDS{"type": "freq", "src": src})
 219                         dstRaw, err := ioutil.ReadAll(pipeR)
 220                         if err != nil {
 221                                 ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "read")
 222                                 isBad = true
 223                                 goto Closing
 224                         }
 225                         dst := string(dstRaw)
 226                         sds["dst"] = dst
 227                         sender := ctx.Neigh[*job.PktEnc.Sender]
 228                         freq := sender.Freq
 229                         if freq == nil {
 230                                 ctx.LogE("rx", sds, "freqing is not allowed")
 231                                 isBad = true
 232                                 goto Closing
 233                         }
 234                         if !dryRun {
 235                                 if sender.FreqChunked == 0 {
 236                                         err = ctx.TxFile(
 237                                                 sender,
 238                                                 job.PktEnc.Nice,
 239                                                 filepath.Join(*freq, src),
 240                                                 dst,
 241                                                 sender.FreqMinSize,
 242                                         )
 243                                 } else {
 244                                         err = ctx.TxFileChunked(
 245                                                 sender,
 246                                                 job.PktEnc.Nice,
 247                                                 filepath.Join(*freq, src),
 248                                                 dst,
 249                                                 sender.FreqMinSize,
 250                                                 sender.FreqChunked,
 251                                         )
 252                                 }
 253                                 if err != nil {
 254                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "tx file")
 255                                         isBad = true
 256                                         goto Closing
 257                                 }
 258                         }
 259                         ctx.LogI("rx", sds, "")
 260                         if !dryRun {
 261                                 if err = os.Remove(job.Fd.Name()); err != nil {
 262                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove")
 263                                         isBad = true
 264                                 }
 265                                 if ctx.NotifyFreq != nil {
 266                                         sendmail := ctx.Neigh[*ctx.SelfId].Sendmail
 267                                         cmd := exec.Command(
 268                                                 sendmail[0],
 269                                                 append(sendmail[1:len(sendmail)], ctx.NotifyFreq.To)...,
 270                                         )
 271                                         cmd.Stdin = newNotification(ctx.NotifyFreq, fmt.Sprintf(
 272                                                 "Freq from %s: %s",
 273                                                 ctx.Neigh[*job.PktEnc.Sender].Name,
 274                                                 src,
 275                                         ))
 276                                         cmd.Run()
 277                                 }
 278                         }
 279                 case PktTypeTrns:
 280                         dst := new([blake2b.Size256]byte)
 281                         copy(dst[:], pkt.Path[:int(pkt.PathLen)])
 282                         nodeId := NodeId(*dst)
 283                         node, known := ctx.Neigh[nodeId]
 284                         sds := SdsAdd(sds, SDS{"type": "trns", "dst": nodeId})
 285                         if !known {
 286                                 ctx.LogE("rx", sds, "unknown node")
 287                                 isBad = true
 288                                 goto Closing
 289                         }
 290                         ctx.LogD("rx", sds, "taken")
 291                         if !dryRun {
 292                                 if err = ctx.TxTrns(node, job.PktEnc.Nice, pktSize, pipeR); err != nil {
 293                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "tx trns")
 294                                         isBad = true
 295                                         goto Closing
 296                                 }
 297                         }
 298                         ctx.LogI("rx", sds, "")
 299                         if !dryRun {
 300                                 if err = os.Remove(job.Fd.Name()); err != nil {
 301                                         ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove")
 302                                         isBad = true
 303                                 }
 304                         }
 305                 default:
 306                         ctx.LogE("rx", sds, "unknown type")
 307                         isBad = true
 308                 }
 309         Closing:
 310                 pipeR.Close()
 311         }
 312         return isBad
 313 }