X-Git-Url: http://www.git.cypherpunks.ru/?a=blobdiff_plain;f=src%2Ftoss.go;h=6abf278af8ab18cc61e785a80fb0c5565890bcca;hb=1d2ce674b042d07fd9b37a46578c8b62bb0345b7;hp=0ce446ac807b39df6e6a067a2745b0631101ede2;hpb=9c25694667c14c934fc122741b9736988172f7de;p=nncp.git diff --git a/src/toss.go b/src/toss.go index 0ce446a..6abf278 100644 --- a/src/toss.go +++ b/src/toss.go @@ -20,6 +20,8 @@ package nncp import ( "bufio" "bytes" + "encoding/base64" + "errors" "fmt" "io" "io/ioutil" @@ -32,7 +34,7 @@ import ( "strconv" "strings" - "github.com/davecgh/go-xdr/xdr2" + xdr "github.com/davecgh/go-xdr/xdr2" "github.com/dustin/go-humanize" "github.com/klauspost/compress/zstd" "golang.org/x/crypto/blake2b" @@ -43,13 +45,22 @@ const ( SeenSuffix = ".seen" ) -func newNotification(fromTo *FromToJSON, subject string) io.Reader { - return strings.NewReader(fmt.Sprintf( - "From: %s\nTo: %s\nSubject: %s\n", - fromTo.From, - fromTo.To, - mime.BEncoding.Encode("UTF-8", subject), - )) +func newNotification(fromTo *FromToJSON, subject string, body []byte) io.Reader { + lines := []string{ + "From: " + fromTo.From, + "To: " + fromTo.To, + "Subject: " + mime.BEncoding.Encode("UTF-8", subject), + } + if len(body) > 0 { + lines = append(lines, []string{ + "MIME-Version: 1.0", + "Content-Type: text/plain; charset=utf-8", + "Content-Transfer-Encoding: base64", + "", + base64.StdEncoding.EncodeToString(body), + }...) + } + return strings.NewReader(strings.Join(lines, "\n")) } func (ctx *Ctx) Toss( @@ -58,6 +69,7 @@ func (ctx *Ctx) Toss( dryRun, doSeen, noFile, noFreq, noExec, noTrns bool, ) bool { isBad := false + sendmail := ctx.Neigh[*ctx.SelfId].Exec["sendmail"] decompressor, err := zstd.NewReader(nil) if err != nil { panic(err) @@ -67,9 +79,7 @@ func (ctx *Ctx) Toss( pktName := filepath.Base(job.Fd.Name()) sds := SDS{"node": job.PktEnc.Sender, "pkt": pktName} if job.PktEnc.Nice > nice { - ctx.LogD("rx", SdsAdd(sds, SDS{ - "nice": strconv.Itoa(int(job.PktEnc.Nice)), - }), "too nice") + ctx.LogD("rx", SdsAdd(sds, SDS{"nice": int(job.PktEnc.Nice)}), "too nice") continue } pipeR, pipeW := io.Pipe() @@ -87,7 +97,7 @@ func (ctx *Ctx) Toss( pipeW.Close() job.Fd.Close() if err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "decryption") + ctx.LogE("rx", sds, err, "decryption") } }(job) var pkt Pkt @@ -95,7 +105,7 @@ func (ctx *Ctx) Toss( var pktSize int64 var pktSizeBlocks int64 if _, err = xdr.Unmarshal(pipeR, &pkt); err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "unmarshal") + ctx.LogE("rx", sds, err, "unmarshal") isBad = true goto Closing } @@ -105,7 +115,7 @@ func (ctx *Ctx) Toss( pktSize -= poly1305.TagSize } pktSize -= pktSizeBlocks * poly1305.TagSize - sds["size"] = strconv.FormatInt(pktSize, 10) + sds["size"] = pktSize ctx.LogD("rx", sds, "taken") switch pkt.Type { case PktTypeExec: @@ -118,14 +128,15 @@ func (ctx *Ctx) Toss( for _, p := range path[1:] { args = append(args, string(p)) } + argsStr := strings.Join(append([]string{handle}, args...), " ") sds := SdsAdd(sds, SDS{ "type": "exec", - "dst": strings.Join(append([]string{handle}, args...), " "), + "dst": argsStr, }) sender := ctx.Neigh[*job.PktEnc.Sender] cmdline, exists := sender.Exec[handle] if !exists || len(cmdline) == 0 { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": "No handle found"}), "") + ctx.LogE("rx", sds, errors.New("No handle found"), "") isBad = true goto Closing } @@ -144,11 +155,28 @@ func (ctx *Ctx) Toss( "NNCP_NICE="+strconv.Itoa(int(pkt.Nice)), ) cmd.Stdin = decompressor - if err = cmd.Run(); err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "handle") + output, err := cmd.Output() + if err != nil { + ctx.LogE("rx", sds, err, "handle") isBad = true goto Closing } + if len(sendmail) > 0 && ctx.NotifyExec != nil { + notify, exists := ctx.NotifyExec[sender.Name+"."+handle] + if !exists { + notify, exists = ctx.NotifyExec["*."+handle] + } + if exists { + cmd := exec.Command( + sendmail[0], + append(sendmail[1:len(sendmail)], notify.To)..., + ) + cmd.Stdin = newNotification(notify, fmt.Sprintf( + "Exec from %s: %s", sender.Name, argsStr, + ), output) + cmd.Run() + } + } } ctx.LogI("rx", sds, "") if !dryRun { @@ -158,7 +186,7 @@ func (ctx *Ctx) Toss( } } if err = os.Remove(job.Fd.Name()); err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove") + ctx.LogE("rx", sds, err, "remove") isBad = true } } @@ -169,46 +197,51 @@ func (ctx *Ctx) Toss( dst := string(pkt.Path[:int(pkt.PathLen)]) sds := SdsAdd(sds, SDS{"type": "file", "dst": dst}) if filepath.IsAbs(dst) { - ctx.LogE("rx", sds, "non-relative destination path") + ctx.LogE("rx", sds, errors.New("non-relative destination path"), "") isBad = true goto Closing } incoming := ctx.Neigh[*job.PktEnc.Sender].Incoming if incoming == nil { - ctx.LogE("rx", sds, "incoming is not allowed") + ctx.LogE("rx", sds, errors.New("incoming is not allowed"), "") isBad = true goto Closing } dir := filepath.Join(*incoming, path.Dir(dst)) if err = os.MkdirAll(dir, os.FileMode(0777)); err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "mkdir") + ctx.LogE("rx", sds, err, "mkdir") isBad = true goto Closing } if !dryRun { tmp, err := TempFile(dir, "file") if err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "mktemp") + ctx.LogE("rx", sds, err, "mktemp") isBad = true goto Closing } sds["tmp"] = tmp.Name() ctx.LogD("rx", sds, "created") bufW := bufio.NewWriter(tmp) - if _, err = io.Copy(bufW, pipeR); err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "copy") + if _, err = CopyProgressed( + bufW, + pipeR, + SdsAdd(sds, SDS{"fullsize": sds["size"]}), + ctx.ShowPrgrs, + ); err != nil { + ctx.LogE("rx", sds, err, "copy") isBad = true goto Closing } if err = bufW.Flush(); err != nil { tmp.Close() - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "copy") + ctx.LogE("rx", sds, err, "copy") isBad = true goto Closing } if err = tmp.Sync(); err != nil { tmp.Close() - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "copy") + ctx.LogE("rx", sds, err, "copy") isBad = true goto Closing } @@ -221,7 +254,7 @@ func (ctx *Ctx) Toss( if os.IsNotExist(err) { break } - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "stat") + ctx.LogE("rx", sds, err, "stat") isBad = true goto Closing } @@ -229,7 +262,11 @@ func (ctx *Ctx) Toss( dstPathCtr++ } if err = os.Rename(tmp.Name(), dstPath); err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "rename") + ctx.LogE("rx", sds, err, "rename") + isBad = true + } + if err = DirSync(*incoming); err != nil { + ctx.LogE("rx", sds, err, "sync") isBad = true } delete(sds, "tmp") @@ -242,11 +279,10 @@ func (ctx *Ctx) Toss( } } if err = os.Remove(job.Fd.Name()); err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove") + ctx.LogE("rx", sds, err, "remove") isBad = true } - sendmail, exists := ctx.Neigh[*ctx.SelfId].Exec["sendmail"] - if exists && len(sendmail) > 0 && ctx.NotifyFile != nil { + if len(sendmail) > 0 && ctx.NotifyFile != nil { cmd := exec.Command( sendmail[0], append(sendmail[1:len(sendmail)], ctx.NotifyFile.To)..., @@ -256,7 +292,7 @@ func (ctx *Ctx) Toss( ctx.Neigh[*job.PktEnc.Sender].Name, dst, humanize.IBytes(uint64(pktSize)), - )) + ), nil) cmd.Run() } } @@ -266,47 +302,38 @@ func (ctx *Ctx) Toss( } src := string(pkt.Path[:int(pkt.PathLen)]) if filepath.IsAbs(src) { - ctx.LogE("rx", sds, "non-relative source path") + ctx.LogE("rx", sds, errors.New("non-relative source path"), "") isBad = true goto Closing } sds := SdsAdd(sds, SDS{"type": "freq", "src": src}) dstRaw, err := ioutil.ReadAll(pipeR) if err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "read") + ctx.LogE("rx", sds, err, "read") isBad = true goto Closing } dst := string(dstRaw) sds["dst"] = dst sender := ctx.Neigh[*job.PktEnc.Sender] - freq := sender.Freq - if freq == nil { - ctx.LogE("rx", sds, "freqing is not allowed") + freqPath := sender.FreqPath + if freqPath == nil { + ctx.LogE("rx", sds, errors.New("freqing is not allowed"), "") isBad = true goto Closing } if !dryRun { - if sender.FreqChunked == 0 { - err = ctx.TxFile( - sender, - pkt.Nice, - filepath.Join(*freq, src), - dst, - sender.FreqMinSize, - ) - } else { - err = ctx.TxFileChunked( - sender, - pkt.Nice, - filepath.Join(*freq, src), - dst, - sender.FreqMinSize, - sender.FreqChunked, - ) - } + err = ctx.TxFile( + sender, + pkt.Nice, + filepath.Join(*freqPath, src), + dst, + sender.FreqChunked, + sender.FreqMinSize, + sender.FreqMaxSize, + ) if err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "tx file") + ctx.LogE("rx", sds, err, "tx file") isBad = true goto Closing } @@ -319,20 +346,17 @@ func (ctx *Ctx) Toss( } } if err = os.Remove(job.Fd.Name()); err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove") + ctx.LogE("rx", sds, err, "remove") isBad = true } - sendmail, exists := ctx.Neigh[*ctx.SelfId].Exec["sendmail"] - if exists && len(sendmail) > 0 && ctx.NotifyFreq != nil { + if len(sendmail) > 0 && ctx.NotifyFreq != nil { cmd := exec.Command( sendmail[0], append(sendmail[1:len(sendmail)], ctx.NotifyFreq.To)..., ) cmd.Stdin = newNotification(ctx.NotifyFreq, fmt.Sprintf( - "Freq from %s: %s", - ctx.Neigh[*job.PktEnc.Sender].Name, - src, - )) + "Freq from %s: %s", sender.Name, src, + ), nil) cmd.Run() } } @@ -346,14 +370,14 @@ func (ctx *Ctx) Toss( node, known := ctx.Neigh[nodeId] sds := SdsAdd(sds, SDS{"type": "trns", "dst": nodeId}) if !known { - ctx.LogE("rx", sds, "unknown node") + ctx.LogE("rx", sds, errors.New("unknown node"), "") isBad = true goto Closing } ctx.LogD("rx", sds, "taken") if !dryRun { if err = ctx.TxTrns(node, job.PktEnc.Nice, pktSize, pipeR); err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "tx trns") + ctx.LogE("rx", sds, err, "tx trns") isBad = true goto Closing } @@ -366,12 +390,12 @@ func (ctx *Ctx) Toss( } } if err = os.Remove(job.Fd.Name()); err != nil { - ctx.LogE("rx", SdsAdd(sds, SDS{"err": err}), "remove") + ctx.LogE("rx", sds, err, "remove") isBad = true } } default: - ctx.LogE("rx", sds, "unknown type") + ctx.LogE("rx", sds, errors.New("unknown type"), "") isBad = true } Closing: