X-Git-Url: http://www.git.cypherpunks.ru/?a=blobdiff_plain;f=src%2Fcmd%2Fnncp-bundle%2Fmain.go;h=8c5e7a2a98bf0d0585678bebc4a859420339ceb5;hb=7c507e7ddba9fe5557df06c860fe9fa6197ba2e1;hp=0dbe3ba34a45b9ac1e3ab97562891cab77385937;hpb=ff2139ccf41d72a3c1c4b56c2106effd1ef2e841;p=nncp.git

diff --git a/src/cmd/nncp-bundle/main.go b/src/cmd/nncp-bundle/main.go
index 0dbe3ba..8c5e7a2 100644
--- a/src/cmd/nncp-bundle/main.go
+++ b/src/cmd/nncp-bundle/main.go
@@ -1,6 +1,6 @@
 /*
 NNCP -- Node to Node copy, utilities for store-and-forward data exchange
-Copyright (C) 2016-2021 Sergey Matveev <stargrave@stargrave.org>
+Copyright (C) 2016-2023 Sergey Matveev <stargrave@stargrave.org>
 
 This program is free software: you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
@@ -26,24 +26,19 @@ import (
 	"flag"
 	"fmt"
 	"io"
-	"io/ioutil"
+	"io/fs"
 	"log"
 	"os"
 	"path/filepath"
 	"strings"
 
 	xdr "github.com/davecgh/go-xdr/xdr2"
-	"go.cypherpunks.ru/nncp/v5"
-	"golang.org/x/crypto/blake2b"
-)
-
-const (
-	CopyBufSize = 1 << 17
+	"github.com/dustin/go-humanize"
+	"go.cypherpunks.ru/nncp/v8"
 )
 
 func usage() {
-	fmt.Fprintf(os.Stderr, nncp.UsageHeader())
-	fmt.Fprintf(os.Stderr, "nncp-bundle -- Create/digest stream of NNCP encrypted packets\n\n")
+	fmt.Fprint(os.Stderr, "nncp-bundle -- Create/digest stream of NNCP encrypted packets\n\n")
 	fmt.Fprintf(os.Stderr, "Usage: %s [options] -tx [-delete] NODE [NODE ...] > ...\n", os.Args[0])
 	fmt.Fprintf(os.Stderr, "       %s [options] -rx -delete [-dryrun] [NODE ...] < ...\n", os.Args[0])
 	fmt.Fprintf(os.Stderr, "       %s [options] -rx [-check] [-dryrun] [NODE ...] < ...\n", os.Args[0])
@@ -69,6 +64,7 @@ func main() {
 		version   = flag.Bool("version", false, "Print version information")
 		warranty  = flag.Bool("warranty", false, "Print warranty information")
 	)
+	log.SetFlags(log.Lshortfile)
 	flag.Usage = usage
 	flag.Parse()
 	if *warranty {
@@ -107,7 +103,7 @@ func main() {
 	for i := 0; i < flag.NArg(); i++ {
 		node, err := ctx.FindNode(flag.Arg(i))
 		if err != nil {
-			log.Fatalln("Invalid specified:", err)
+			log.Fatalln("Invalid node specified:", err)
 		}
 		nodeIds[*node.Id] = struct{}{}
 	}
@@ -119,19 +115,28 @@ func main() {
 		bufStdout := bufio.NewWriter(os.Stdout)
 		tarWr := tar.NewWriter(bufStdout)
 		for nodeId := range nodeIds {
-			les := nncp.LEs{
-				{K: "XX", V: string(nncp.TTx)},
-				{K: "Node", V: nodeId.String()},
-				{K: "Pkt", V: "dummy"},
-			}
 			for job := range ctx.Jobs(&nodeId, nncp.TTx) {
-				pktName = filepath.Base(job.Fd.Name())
-				les[len(les)-1].V = pktName
+				pktName = filepath.Base(job.Path)
+				les := nncp.LEs{
+					{K: "XX", V: string(nncp.TTx)},
+					{K: "Node", V: nodeId.String()},
+					{K: "Pkt", V: pktName},
+				}
 				if job.PktEnc.Nice > nice {
-					ctx.LogD("nncp-bundle", les, "too nice")
-					job.Fd.Close() // #nosec G104
+					ctx.LogD("bundle-tx-too-nice", les, func(les nncp.LEs) string {
+						return fmt.Sprintf(
+							"Bundle transfer %s/tx/%s: too nice %s",
+							ctx.NodeName(&nodeId),
+							pktName,
+							nncp.NicenessFmt(job.PktEnc.Nice),
+						)
+					})
 					continue
 				}
+				fd, err := os.Open(job.Path)
+				if err != nil {
+					log.Fatalln("Error during opening:", err)
+				}
 				if err = tarWr.WriteHeader(&tar.Header{
 					Format:   tar.FormatUSTAR,
 					Name:     nncp.NNCPBundlePrefix,
@@ -155,7 +160,7 @@ func main() {
 					log.Fatalln("Error writing tar header:", err)
 				}
 				if _, err = nncp.CopyProgressed(
-					tarWr, job.Fd, "Tx",
+					tarWr, bufio.NewReaderSize(fd, nncp.MTHBlockSize), "Tx",
 					append(les, nncp.LEs{
 						{K: "Pkt", V: nncp.Base32Codec.EncodeToString(job.HshValue[:])},
 						{K: "FullSize", V: job.Size},
@@ -164,7 +169,9 @@ func main() {
 				); err != nil {
 					log.Fatalln("Error during copying to tar:", err)
 				}
-				job.Fd.Close() // #nosec G104
+				if err = fd.Close(); err != nil {
+					log.Fatalln("Error during closing:", err)
+				}
 				if err = tarWr.Flush(); err != nil {
 					log.Fatalln("Error during tar flushing:", err)
 				}
@@ -172,22 +179,35 @@ func main() {
 					log.Fatalln("Error during stdout flushing:", err)
 				}
 				if *doDelete {
-					if err = os.Remove(job.Fd.Name()); err != nil {
+					if err = os.Remove(job.Path); err != nil {
 						log.Fatalln("Error during deletion:", err)
+					} else if ctx.HdrUsage {
+						os.Remove(nncp.JobPath2Hdr(job.Path))
 					}
 				}
-				ctx.LogI("nncp-bundle", append(les, nncp.LE{K: "Size", V: job.Size}), "")
+				ctx.LogI(
+					"bundle-tx",
+					append(les, nncp.LE{K: "Size", V: job.Size}),
+					func(les nncp.LEs) string {
+						return fmt.Sprintf(
+							"Bundle transfer, sent to node %s %s (%s)",
+							ctx.NodeName(&nodeId),
+							pktName,
+							humanize.IBytes(uint64(job.Size)),
+						)
+					},
+				)
 			}
 		}
 		if err = tarWr.Close(); err != nil {
 			log.Fatalln("Error during tar closing:", err)
 		}
 	} else {
-		bufStdin := bufio.NewReaderSize(os.Stdin, CopyBufSize*2)
+		bufStdin := bufio.NewReaderSize(os.Stdin, nncp.MTHBlockSize*2)
 		pktEncBuf := make([]byte, nncp.PktEncOverhead)
 		var pktEnc *nncp.PktEnc
 		for {
-			peeked, err := bufStdin.Peek(CopyBufSize)
+			peeked, err := bufStdin.Peek(nncp.MTHBlockSize)
 			if err != nil && err != io.EOF {
 				log.Fatalln("Error during reading:", err)
 			}
@@ -196,7 +216,7 @@ func main() {
 				if err == io.EOF {
 					break
 				}
-				bufStdin.Discard(bufStdin.Buffered() - (len(nncp.NNCPBundlePrefix) - 1)) // #nosec G104
+				bufStdin.Discard(bufStdin.Buffered() - (len(nncp.NNCPBundlePrefix) - 1))
 				continue
 			}
 			if _, err = bufStdin.Discard(prefixIdx); err != nil {
@@ -207,18 +227,25 @@ func main() {
 			if err != nil {
 				if err != io.EOF {
 					ctx.LogD(
-						"nncp-bundle",
+						"bundle-rx-read-tar",
 						nncp.LEs{{K: "XX", V: string(nncp.TRx)}, {K: "Err", V: err}},
-						"error reading tar",
+						func(les nncp.LEs) string {
+							return "Bundle transfer rx: reading tar"
+						},
 					)
 				}
 				continue
 			}
 			if entry.Typeflag != tar.TypeDir {
 				ctx.LogD(
-					"nncp-bundle",
-					nncp.LEs{{K: "XX", V: string(nncp.TRx)}},
-					"Expected NNCP/",
+					"bundle-rx-read-tar",
+					nncp.LEs{
+						{K: "XX", V: string(nncp.TRx)},
+						{K: "Err", V: errors.New("expected NNCP/")},
+					},
+					func(les nncp.LEs) string {
+						return "Bundle transfer rx: reading tar"
+					},
 				)
 				continue
 			}
@@ -226,47 +253,89 @@ func main() {
 			if err != nil {
 				if err != io.EOF {
 					ctx.LogD(
-						"nncp-bundle",
+						"bundle-rx-read-tar",
 						nncp.LEs{{K: "XX", V: string(nncp.TRx)}, {K: "Err", V: err}},
-						"error reading tar",
+						func(les nncp.LEs) string {
+							return "Bundle transfer rx: reading tar"
+						},
 					)
 				}
 				continue
 			}
 			les := nncp.LEs{{K: "XX", V: string(nncp.TRx)}, {K: "Pkt", V: entry.Name}}
+			logMsg := func(les nncp.LEs) string {
+				return "Bundle transfer rx/" + entry.Name
+			}
 			if entry.Size < nncp.PktEncOverhead {
-				ctx.LogD("nncp-bundle", les, "Too small packet")
+				ctx.LogD("bundle-rx-too-small", les, func(les nncp.LEs) string {
+					return logMsg(les) + ": too small packet"
+				})
 				continue
 			}
 			if !ctx.IsEnoughSpace(entry.Size) {
-				ctx.LogE("nncp-bundle", les, errors.New("not enough spool space"), "")
+				ctx.LogE("bundle-rx", les, errors.New("not enough spool space"), logMsg)
 				continue
 			}
 			pktName := filepath.Base(entry.Name)
 			if _, err = nncp.Base32Codec.DecodeString(pktName); err != nil {
-				ctx.LogD("nncp-bundle", append(les, nncp.LE{K: "Err", V: "bad packet name"}), "")
+				ctx.LogD(
+					"bundle-rx",
+					append(les, nncp.LE{K: "Err", V: "bad packet name"}),
+					logMsg,
+				)
 				continue
 			}
 			if _, err = io.ReadFull(tarR, pktEncBuf); err != nil {
-				ctx.LogD("nncp-bundle", append(les, nncp.LE{K: "Err", V: err}), "read")
+				ctx.LogD(
+					"bundle-rx",
+					append(les, nncp.LE{K: "Err", V: err}),
+					logMsg,
+				)
 				continue
 			}
 			if _, err = xdr.Unmarshal(bytes.NewReader(pktEncBuf), &pktEnc); err != nil {
-				ctx.LogD("nncp-bundle", les, "Bad packet structure")
+				ctx.LogD(
+					"bundle-rx",
+					append(les, nncp.LE{K: "Err", V: "Bad packet structure"}),
+					logMsg,
+				)
 				continue
 			}
-			if pktEnc.Magic != nncp.MagicNNCPEv4 {
-				ctx.LogD("nncp-bundle", les, "Bad packet magic number")
+			switch pktEnc.Magic {
+			case nncp.MagicNNCPEv1.B:
+				err = nncp.MagicNNCPEv1.TooOld()
+			case nncp.MagicNNCPEv2.B:
+				err = nncp.MagicNNCPEv2.TooOld()
+			case nncp.MagicNNCPEv3.B:
+				err = nncp.MagicNNCPEv3.TooOld()
+			case nncp.MagicNNCPEv4.B:
+				err = nncp.MagicNNCPEv4.TooOld()
+			case nncp.MagicNNCPEv5.B:
+				err = nncp.MagicNNCPEv5.TooOld()
+			case nncp.MagicNNCPEv6.B:
+			default:
+				err = errors.New("Bad packet magic number")
+			}
+			if err != nil {
+				ctx.LogD(
+					"bundle-rx",
+					append(les, nncp.LE{K: "Err", V: err.Error()}),
+					logMsg,
+				)
 				continue
 			}
 			if pktEnc.Nice > nice {
-				ctx.LogD("nncp-bundle", les, "too nice")
+				ctx.LogD("bundle-rx-too-nice", les, func(les nncp.LEs) string {
+					return logMsg(les) + ": too nice"
+				})
 				continue
 			}
 			if *pktEnc.Sender == *ctx.SelfId && *doDelete {
 				if len(nodeIds) > 0 {
 					if _, exists := nodeIds[*pktEnc.Recipient]; !exists {
-						ctx.LogD("nncp-bundle", les, "Recipient is not requested")
+						ctx.LogD("bundle-tx-skip", les, func(les nncp.LEs) string {
+							return logMsg(les) + ": recipient is not requested"
+						})
 						continue
 					}
 				}
@@ -276,15 +345,17 @@ func main() {
 					{K: "Node", V: nodeId32},
 					{K: "Pkt", V: pktName},
 				}
+				logMsg = func(les nncp.LEs) string {
+					return fmt.Sprintf("Bundle transfer %s/tx/%s", nodeId32, pktName)
+				}
 				dstPath := filepath.Join(ctx.Spool, nodeId32, string(nncp.TTx), pktName)
 				if _, err = os.Stat(dstPath); err != nil {
-					ctx.LogD("nncp-bundle", les, "Packet is already missing")
+					ctx.LogD("bundle-tx-missing", les, func(les nncp.LEs) string {
+						return logMsg(les) + ": packet is already missing"
+					})
 					continue
 				}
-				hsh, err := blake2b.New256(nil)
-				if err != nil {
-					log.Fatalln("Error during hasher creation:", err)
-				}
+				hsh := nncp.MTHNew(entry.Size, 0)
 				if _, err = hsh.Write(pktEncBuf); err != nil {
 					log.Fatalln("Error during writing:", err)
 				}
@@ -296,22 +367,31 @@ func main() {
 					log.Fatalln("Error during copying:", err)
 				}
 				if nncp.Base32Codec.EncodeToString(hsh.Sum(nil)) == pktName {
-					ctx.LogI("nncp-bundle", les, "removed")
+					ctx.LogI("bundle-tx-removed", les, func(les nncp.LEs) string {
+						return logMsg(les) + ": removed"
+					})
 					if !*dryRun {
-						os.Remove(dstPath) // #nosec G104
+						os.Remove(dstPath)
+						if ctx.HdrUsage {
+							os.Remove(nncp.JobPath2Hdr(dstPath))
+						}
 					}
 				} else {
-					ctx.LogE("nncp-bundle", les, errors.New("bad checksum"), "")
+					ctx.LogE("bundle-tx", les, errors.New("bad checksum"), logMsg)
 				}
 				continue
 			}
 			if *pktEnc.Recipient != *ctx.SelfId {
-				ctx.LogD("nncp-bundle", les, "Unknown recipient")
+				ctx.LogD("nncp-bundle", les, func(les nncp.LEs) string {
+					return logMsg(les) + ": unknown recipient"
+				})
 				continue
 			}
 			if len(nodeIds) > 0 {
 				if _, exists := nodeIds[*pktEnc.Sender]; !exists {
-					ctx.LogD("nncp-bundle", les, "Sender is not requested")
+					ctx.LogD("bundle-rx-skip", les, func(les nncp.LEs) string {
+						return logMsg(les) + ": sender is not requested"
+					})
 					continue
 				}
 			}
@@ -322,22 +402,28 @@ func main() {
 				{K: "Pkt", V: pktName},
 				{K: "FullSize", V: entry.Size},
 			}
+			logMsg = func(les nncp.LEs) string {
+				return fmt.Sprintf("Bundle transfer %s/rx/%s", sender, pktName)
+			}
 			dstDirPath := filepath.Join(ctx.Spool, sender, string(nncp.TRx))
 			dstPath := filepath.Join(dstDirPath, pktName)
-			if _, err = os.Stat(dstPath); err == nil || !os.IsNotExist(err) {
-				ctx.LogD("nncp-bundle", les, "Packet already exists")
+			if _, err = os.Stat(dstPath); err == nil || !errors.Is(err, fs.ErrNotExist) {
+				ctx.LogD("bundle-rx-exists", les, func(les nncp.LEs) string {
+					return logMsg(les) + ": packet already exists"
+				})
 				continue
 			}
-			if _, err = os.Stat(dstPath + nncp.SeenSuffix); err == nil || !os.IsNotExist(err) {
-				ctx.LogD("nncp-bundle", les, "Packet already exists")
+			if _, err = os.Stat(filepath.Join(
+				dstDirPath, nncp.SeenDir, pktName,
+			)); err == nil || !errors.Is(err, fs.ErrNotExist) {
+				ctx.LogD("bundle-rx-seen", les, func(les nncp.LEs) string {
+					return logMsg(les) + ": packet already seen"
+				})
 				continue
 			}
 			if *doCheck {
 				if *dryRun {
-					hsh, err := blake2b.New256(nil)
-					if err != nil {
-						log.Fatalln("Error during hasher creation:", err)
-					}
+					hsh := nncp.MTHNew(entry.Size, 0)
 					if _, err = hsh.Write(pktEncBuf); err != nil {
 						log.Fatalln("Error during writing:", err)
 					}
@@ -345,7 +431,7 @@ func main() {
 						log.Fatalln("Error during copying:", err)
 					}
 					if nncp.Base32Codec.EncodeToString(hsh.Sum(nil)) != pktName {
-						ctx.LogE("nncp-bundle", les, errors.New("bad checksum"), "")
+						ctx.LogE("bundle-rx", les, errors.New("bad checksum"), logMsg)
 						continue
 					}
 				} else {
@@ -367,14 +453,14 @@ func main() {
 							log.Fatalln("Error during commiting:", err)
 						}
 					} else {
-						ctx.LogE("nncp-bundle", les, errors.New("bad checksum"), "")
+						ctx.LogE("bundle-rx", les, errors.New("bad checksum"), logMsg)
 						tmp.Cancel()
 						continue
 					}
 				}
 			} else {
 				if *dryRun {
-					if _, err = nncp.CopyProgressed(ioutil.Discard, tarR, "Rx", les, ctx.ShowPrgrs); err != nil {
+					if _, err = nncp.CopyProgressed(io.Discard, tarR, "Rx", les, ctx.ShowPrgrs); err != nil {
 						log.Fatalln("Error during copying:", err)
 					}
 				} else {
@@ -382,7 +468,7 @@ func main() {
 					if err != nil {
 						log.Fatalln("Error during temporary file creation:", err)
 					}
-					bufTmp := bufio.NewWriterSize(tmp, CopyBufSize)
+					bufTmp := bufio.NewWriterSize(tmp, nncp.MTHBlockSize)
 					if _, err = bufTmp.Write(pktEncBuf); err != nil {
 						log.Fatalln("Error during writing:", err)
 					}
@@ -392,8 +478,10 @@ func main() {
 					if err = bufTmp.Flush(); err != nil {
 						log.Fatalln("Error during flushing:", err)
 					}
-					if err = tmp.Sync(); err != nil {
-						log.Fatalln("Error during syncing:", err)
+					if !nncp.NoSync {
+						if err = tmp.Sync(); err != nil {
+							log.Fatalln("Error during syncing:", err)
+						}
 					}
 					if err = tmp.Close(); err != nil {
 						log.Fatalln("Error during closing:", err)
@@ -407,6 +495,9 @@ func main() {
 					if err = nncp.DirSync(dstDirPath); err != nil {
 						log.Fatalln("Error during syncing:", err)
 					}
+					if ctx.HdrUsage {
+						ctx.HdrWrite(pktEncBuf, dstPath)
+					}
 				}
 			}
 			for _, le := range les {
@@ -415,7 +506,12 @@ func main() {
 					break
 				}
 			}
-			ctx.LogI("nncp-bundle", les, "")
+			ctx.LogI("bundle-rx", les, func(les nncp.LEs) string {
+				return fmt.Sprintf(
+					"Bundle transfer, received from %s %s (%s)",
+					sender, pktName, humanize.IBytes(uint64(entry.Size)),
+				)
+			})
 		}
 	}
 }