src/cmd/nncp-xfer/main.go

   1 // NNCP -- Node to Node copy, utilities for store-and-forward data exchange
   2 // Copyright (C) 2016-2024 Sergey Matveev <stargrave@stargrave.org>
   3 //
   4 // This program is free software: you can redistribute it and/or modify
   5 // it under the terms of the GNU General Public License as published by
   6 // the Free Software Foundation, version 3 of the License.
   7 //
   8 // This program is distributed in the hope that it will be useful,
   9 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 // GNU General Public License for more details.
  12 //
  13 // You should have received a copy of the GNU General Public License
  14 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
  15
  16 // Exchange NNCP inbound and outbounds packets with external directory.
  17 package main
  18
  19 import (
  20         "bufio"
  21         "errors"
  22         "flag"
  23         "fmt"
  24         "io"
  25         "io/fs"
  26         "log"
  27         "os"
  28         "path/filepath"
  29
  30         "github.com/dustin/go-humanize"
  31         "go.cypherpunks.ru/nncp/v8"
  32 )
  33
  34 func usage() {
  35         fmt.Fprint(os.Stderr, "nncp-xfer -- copy inbound and outbounds packets\n\n")
  36         fmt.Fprintf(os.Stderr, "Usage: %s [options] DIR\nOptions:\n", os.Args[0])
  37         flag.PrintDefaults()
  38 }
  39
  40 func main() {
  41         var (
  42                 cfgPath   = flag.String("cfg", nncp.DefaultCfgPath, "Path to configuration file")
  43                 nodeRaw   = flag.String("node", "", "Process only that node")
  44                 niceRaw   = flag.String("nice", nncp.NicenessFmt(255), "Minimal required niceness")
  45                 rxOnly    = flag.Bool("rx", false, "Only receive packets")
  46                 txOnly    = flag.Bool("tx", false, "Only transfer packets")
  47                 mkdir     = flag.Bool("mkdir", false, "Create necessary outbound directories")
  48                 keep      = flag.Bool("keep", false, "Do not delete transferred packets")
  49                 spoolPath = flag.String("spool", "", "Override path to spool")
  50                 logPath   = flag.String("log", "", "Override path to logfile")
  51                 quiet     = flag.Bool("quiet", false, "Print only errors")
  52                 showPrgrs = flag.Bool("progress", false, "Force progress showing")
  53                 omitPrgrs = flag.Bool("noprogress", false, "Omit progress showing")
  54                 debug     = flag.Bool("debug", false, "Print debug messages")
  55                 version   = flag.Bool("version", false, "Print version information")
  56                 warranty  = flag.Bool("warranty", false, "Print warranty information")
  57         )
  58         log.SetFlags(log.Lshortfile)
  59         flag.Usage = usage
  60         flag.Parse()
  61         if *warranty {
  62                 fmt.Println(nncp.Warranty)
  63                 return
  64         }
  65         if *version {
  66                 fmt.Println(nncp.VersionGet())
  67                 return
  68         }
  69         if flag.NArg() != 1 {
  70                 usage()
  71                 os.Exit(1)
  72         }
  73         nice, err := nncp.NicenessParse(*niceRaw)
  74         if err != nil {
  75                 log.Fatalln(err)
  76         }
  77         if *rxOnly && *txOnly {
  78                 log.Fatalln("-rx and -tx can not be set simultaneously")
  79         }
  80
  81         ctx, err := nncp.CtxFromCmdline(
  82                 *cfgPath,
  83                 *spoolPath,
  84                 *logPath,
  85                 *quiet,
  86                 *showPrgrs,
  87                 *omitPrgrs,
  88                 *debug,
  89         )
  90         if err != nil {
  91                 log.Fatalln("Error during initialization:", err)
  92         }
  93
  94         var nodeOnly *nncp.Node
  95         if *nodeRaw != "" {
  96                 nodeOnly, err = ctx.FindNode(*nodeRaw)
  97                 if err != nil {
  98                         log.Fatalln("Invalid -node specified:", err)
  99                 }
 100         }
 101
 102         ctx.Umask()
 103         selfPath := filepath.Join(flag.Arg(0), ctx.SelfId.String())
 104         isBad := false
 105         var dir *os.File
 106         var entries []os.DirEntry
 107         var les nncp.LEs
 108         var logMsg func(les nncp.LEs) string
 109         if *txOnly {
 110                 goto Tx
 111         }
 112         les = nncp.LEs{
 113                 {K: "XX", V: string(nncp.TRx)},
 114                 {K: "Dir", V: selfPath},
 115         }
 116         logMsg = func(les nncp.LEs) string {
 117                 return "Packet transfer, received from self"
 118         }
 119         ctx.LogD("xfer-self", les, logMsg)
 120         if _, err = os.Stat(selfPath); err != nil {
 121                 if errors.Is(err, fs.ErrNotExist) {
 122                         ctx.LogD("xfer-self-no-dir", les, func(les nncp.LEs) string {
 123                                 return logMsg(les) + ": no directory"
 124                         })
 125                         goto Tx
 126                 }
 127                 ctx.LogE("xfer-self-stat", les, err, func(les nncp.LEs) string {
 128                         return logMsg(les) + ": stating"
 129                 })
 130                 isBad = true
 131                 goto Tx
 132         }
 133         dir, err = os.Open(selfPath)
 134         if err != nil {
 135                 ctx.LogE("xfer-self-open", les, err, func(les nncp.LEs) string {
 136                         return logMsg(les) + ": opening"
 137                 })
 138                 isBad = true
 139                 goto Tx
 140         }
 141         entries, err = dir.ReadDir(0)
 142         dir.Close()
 143         if err != nil {
 144                 ctx.LogE("xfer-self-read", les, err, func(les nncp.LEs) string {
 145                         return logMsg(les) + ": reading"
 146                 })
 147                 isBad = true
 148                 goto Tx
 149         }
 150         for _, entry := range entries {
 151                 if !entry.IsDir() {
 152                         continue
 153                 }
 154                 nodeId, err := nncp.NodeIdFromString(entry.Name())
 155                 les := append(les, nncp.LE{K: "Node", V: entry.Name()})
 156                 logMsg := func(les nncp.LEs) string {
 157                         return "Packet transfer, received from " + ctx.NodeName(nodeId)
 158                 }
 159                 if err != nil {
 160                         ctx.LogD("xfer-rx-not-node", les, func(les nncp.LEs) string {
 161                                 return logMsg(les) + ": is not NodeId"
 162                         })
 163                         continue
 164                 }
 165                 if nodeOnly != nil && *nodeId != *nodeOnly.Id {
 166                         ctx.LogD("xfer-rx-skip", les, func(les nncp.LEs) string {
 167                                 return logMsg(les) + ": skipping"
 168                         })
 169                         continue
 170                 }
 171                 if _, known := ctx.Neigh[*nodeId]; !known {
 172                         ctx.LogD("xfer-rx-unknown", les, func(les nncp.LEs) string {
 173                                 return logMsg(les) + ": unknown"
 174                         })
 175                         continue
 176                 }
 177                 dir, err = os.Open(filepath.Join(selfPath, entry.Name()))
 178                 if err != nil {
 179                         ctx.LogE("xfer-rx-open", les, err, func(les nncp.LEs) string {
 180                                 return logMsg(les) + ": opening"
 181                         })
 182                         isBad = true
 183                         continue
 184                 }
 185                 fisInt, err := dir.Readdir(0)
 186                 dir.Close()
 187                 if err != nil {
 188                         ctx.LogE("xfer-rx-read", les, err, func(les nncp.LEs) string {
 189                                 return logMsg(les) + ": reading"
 190                         })
 191                         isBad = true
 192                         continue
 193                 }
 194                 for _, fiInt := range fisInt {
 195                         if fiInt.IsDir() {
 196                                 continue
 197                         }
 198                         // Check that it is valid Base32 encoding
 199                         if _, err = nncp.NodeIdFromString(fiInt.Name()); err != nil {
 200                                 continue
 201                         }
 202                         filename := filepath.Join(dir.Name(), fiInt.Name())
 203                         les := append(les, nncp.LE{K: "File", V: filename})
 204                         logMsg := func(les nncp.LEs) string {
 205                                 return fmt.Sprintf(
 206                                         "Packet transfer, received from %s: %s",
 207                                         ctx.NodeName(nodeId), filename,
 208                                 )
 209                         }
 210                         if _, err = os.Stat(filepath.Join(
 211                                 ctx.Spool,
 212                                 nodeId.String(),
 213                                 string(nncp.TRx),
 214                                 nncp.SeenDir,
 215                                 fiInt.Name(),
 216                         )); err == nil || !errors.Is(err, fs.ErrNotExist) {
 217                                 ctx.LogI("xfer-rx-seen", les, func(les nncp.LEs) string {
 218                                         return logMsg(les) + ": packet already seen"
 219                                 })
 220                                 if !*keep {
 221                                         if err = os.Remove(filename); err != nil {
 222                                                 ctx.LogE("xfer-rx-remove", les, err, logMsg)
 223                                                 isBad = true
 224                                         }
 225                                 }
 226                                 continue
 227                         }
 228                         fd, err := os.Open(filename)
 229                         if err != nil {
 230                                 ctx.LogE("xfer-rx-open", les, err, func(les nncp.LEs) string {
 231                                         return logMsg(les) + ": opening"
 232                                 })
 233                                 isBad = true
 234                                 continue
 235                         }
 236                         pktEnc, pktEncRaw, err := ctx.HdrRead(fd)
 237                         if err == nil {
 238                                 switch pktEnc.Magic {
 239                                 case nncp.MagicNNCPEv1.B:
 240                                         err = nncp.MagicNNCPEv1.TooOld()
 241                                 case nncp.MagicNNCPEv2.B:
 242                                         err = nncp.MagicNNCPEv2.TooOld()
 243                                 case nncp.MagicNNCPEv3.B:
 244                                         err = nncp.MagicNNCPEv3.TooOld()
 245                                 case nncp.MagicNNCPEv4.B:
 246                                         err = nncp.MagicNNCPEv4.TooOld()
 247                                 case nncp.MagicNNCPEv5.B:
 248                                         err = nncp.MagicNNCPEv5.TooOld()
 249                                 case nncp.MagicNNCPEv6.B:
 250                                 default:
 251                                         err = errors.New("is not an encrypted packet")
 252                                 }
 253                         }
 254                         if err != nil {
 255                                 ctx.LogD(
 256                                         "xfer-rx-not-packet",
 257                                         append(les, nncp.LE{K: "Err", V: err}),
 258                                         func(les nncp.LEs) string {
 259                                                 return logMsg(les) + ": not valid packet: " + err.Error()
 260                                         },
 261                                 )
 262                                 fd.Close()
 263                                 continue
 264                         }
 265                         if pktEnc.Nice > nice {
 266                                 ctx.LogD("xfer-rx-too-nice", les, func(les nncp.LEs) string {
 267                                         return logMsg(les) + ": too nice"
 268                                 })
 269                                 fd.Close()
 270                                 continue
 271                         }
 272                         les = append(les, nncp.LE{K: "Size", V: fiInt.Size()})
 273                         logMsg = func(les nncp.LEs) string {
 274                                 return fmt.Sprintf(
 275                                         "Packet transfer, received from %s: %s (%s)",
 276                                         ctx.NodeName(nodeId), filename,
 277                                         humanize.IBytes(uint64(fiInt.Size())),
 278                                 )
 279                         }
 280                         if !ctx.IsEnoughSpace(fiInt.Size()) {
 281                                 ctx.LogE("xfer-rx", les, errors.New("is not enough space"), logMsg)
 282                                 fd.Close()
 283                                 continue
 284                         }
 285                         if _, err = fd.Seek(0, io.SeekStart); err != nil {
 286                                 log.Fatalln(err)
 287                         }
 288                         tmp, err := ctx.NewTmpFileWHash()
 289                         if err != nil {
 290                                 log.Fatalln(err)
 291                         }
 292                         r, w := io.Pipe()
 293                         go func() {
 294                                 _, err := io.CopyN(
 295                                         w, bufio.NewReaderSize(fd, nncp.MTHBlockSize), fiInt.Size(),
 296                                 )
 297                                 if err == nil {
 298                                         err = w.Close()
 299                                 }
 300                                 if err != nil {
 301                                         ctx.LogE("xfer-rx", les, err, logMsg)
 302                                         w.CloseWithError(err)
 303                                 }
 304                         }()
 305                         _, err = nncp.CopyProgressed(
 306                                 tmp.W, r, "Rx",
 307                                 append(
 308                                         les,
 309                                         nncp.LE{K: "Pkt", V: filename},
 310                                         nncp.LE{K: "FullSize", V: fiInt.Size()},
 311                                 ),
 312                                 ctx.ShowPrgrs,
 313                         )
 314                         fd.Close()
 315                         if err != nil {
 316                                 ctx.LogE("xfer-rx", les, err, logMsg)
 317                                 tmp.Cancel()
 318                                 isBad = true
 319                                 continue
 320                         }
 321                         if err = tmp.W.Flush(); err != nil {
 322                                 ctx.LogE("xfer-rx", les, err, logMsg)
 323                                 tmp.Cancel()
 324                                 isBad = true
 325                                 continue
 326                         }
 327                         if tmp.Checksum() != fiInt.Name() {
 328                                 ctx.LogE("xfer-rx", les, errors.New("checksum mismatch"), logMsg)
 329                                 tmp.Cancel()
 330                                 isBad = true
 331                                 continue
 332                         }
 333                         if err = tmp.Commit(filepath.Join(
 334                                 ctx.Spool,
 335                                 nodeId.String(),
 336                                 string(nncp.TRx),
 337                         )); err != nil {
 338                                 log.Fatalln(err)
 339                         }
 340                         ctx.LogI("xfer-rx", les, logMsg)
 341                         if !*keep {
 342                                 if err = os.Remove(filename); err != nil {
 343                                         ctx.LogE("xfer-rx-remove", les, err, logMsg)
 344                                         isBad = true
 345                                 }
 346                         }
 347                         if ctx.HdrUsage {
 348                                 ctx.HdrWrite(pktEncRaw, filepath.Join(
 349                                         ctx.Spool,
 350                                         nodeId.String(),
 351                                         string(nncp.TRx),
 352                                         tmp.Checksum(),
 353                                 ))
 354                         }
 355                 }
 356         }
 357
 358 Tx:
 359         if *rxOnly {
 360                 if isBad {
 361                         os.Exit(1)
 362                 }
 363                 return
 364         }
 365         for nodeId := range ctx.Neigh {
 366                 les := nncp.LEs{{K: "XX", V: string(nncp.TTx)}, {K: "Node", V: nodeId}}
 367                 logMsg := func(les nncp.LEs) string {
 368                         return "Packet transfer, sent to " + ctx.NodeName(&nodeId)
 369                 }
 370                 if nodeOnly != nil && nodeId != *nodeOnly.Id {
 371                         ctx.LogD("xfer-tx-skip", les, func(les nncp.LEs) string {
 372                                 return logMsg(les) + ": skipping"
 373                         })
 374                         continue
 375                 }
 376                 dirLock, err := ctx.LockDir(&nodeId, string(nncp.TTx))
 377                 if err != nil {
 378                         continue
 379                 }
 380                 nodePath := filepath.Join(flag.Arg(0), nodeId.String())
 381                 les = append(les, nncp.LE{K: "Dir", V: nodePath})
 382                 logMsg = func(les nncp.LEs) string {
 383                         return fmt.Sprintf(
 384                                 "Packet transfer, sent to %s: directory %s",
 385                                 ctx.NodeName(&nodeId), nodePath,
 386                         )
 387                 }
 388                 _, err = os.Stat(nodePath)
 389                 if err != nil {
 390                         if errors.Is(err, fs.ErrNotExist) {
 391                                 ctx.LogD("xfer-tx-not-exist", les, func(les nncp.LEs) string {
 392                                         return logMsg(les) + ": does not exist"
 393                                 })
 394                                 if !*mkdir {
 395                                         ctx.UnlockDir(dirLock)
 396                                         continue
 397                                 }
 398                                 if err = os.Mkdir(nodePath, os.FileMode(0777)); err != nil {
 399                                         ctx.UnlockDir(dirLock)
 400                                         ctx.LogE("xfer-tx-mkdir", les, err, logMsg)
 401                                         isBad = true
 402                                         continue
 403                                 }
 404                         } else {
 405                                 ctx.UnlockDir(dirLock)
 406                                 ctx.LogE("xfer-tx", les, err, logMsg)
 407                                 isBad = true
 408                                 continue
 409                         }
 410                 }
 411                 dstPath := filepath.Join(nodePath, ctx.SelfId.String())
 412                 les[len(les)-1].V = dstPath
 413                 logMsg = func(les nncp.LEs) string {
 414                         return fmt.Sprintf(
 415                                 "Packet transfer, sent to %s: directory %s",
 416                                 ctx.NodeName(&nodeId), dstPath,
 417                         )
 418                 }
 419                 _, err = os.Stat(dstPath)
 420                 if err != nil {
 421                         if errors.Is(err, fs.ErrNotExist) {
 422                                 if err = os.Mkdir(dstPath, os.FileMode(0777)); err != nil {
 423                                         ctx.UnlockDir(dirLock)
 424                                         ctx.LogE("xfer-tx-mkdir", les, err, logMsg)
 425                                         isBad = true
 426                                         continue
 427                                 }
 428                         } else {
 429                                 ctx.UnlockDir(dirLock)
 430                                 ctx.LogE("xfer-tx", les, err, logMsg)
 431                                 isBad = true
 432                                 continue
 433                         }
 434                 }
 435                 les = les[:len(les)-1]
 436                 for job := range ctx.Jobs(&nodeId, nncp.TTx) {
 437                         pktName := filepath.Base(job.Path)
 438                         les := append(les, nncp.LE{K: "Pkt", V: pktName})
 439                         logMsg = func(les nncp.LEs) string {
 440                                 return fmt.Sprintf(
 441                                         "Packet transfer, sent to %s: %s",
 442                                         ctx.NodeName(&nodeId), pktName,
 443                                 )
 444                         }
 445                         if job.PktEnc.Nice > nice {
 446                                 ctx.LogD("xfer-tx-too-nice", les, func(les nncp.LEs) string {
 447                                         return logMsg(les) + ": too nice"
 448                                 })
 449                                 continue
 450                         }
 451                         if _, err = os.Stat(filepath.Join(dstPath, pktName)); err == nil || !errors.Is(err, fs.ErrNotExist) {
 452                                 ctx.LogD("xfer-tx-exists", les, func(les nncp.LEs) string {
 453                                         return logMsg(les) + ": already exists"
 454                                 })
 455                                 continue
 456                         }
 457                         tmp, err := nncp.TempFile(dstPath, "xfer")
 458                         if err != nil {
 459                                 ctx.LogE("xfer-tx-mktemp", les, err, func(les nncp.LEs) string {
 460                                         return logMsg(les) + ": mktemp"
 461                                 })
 462                                 isBad = true
 463                                 break
 464                         }
 465                         les = append(les, nncp.LE{K: "Tmp", V: tmp.Name()})
 466                         ctx.LogD("xfer-tx-tmp-create", les, func(les nncp.LEs) string {
 467                                 return fmt.Sprintf("%s: temporary %s created", logMsg(les), tmp.Name())
 468                         })
 469                         fd, err := os.Open(job.Path)
 470                         if err != nil {
 471                                 ctx.LogE("xfer-tx-open", les, err, func(les nncp.LEs) string {
 472                                         return logMsg(les) + ": opening"
 473                                 })
 474                                 tmp.Close()
 475                                 isBad = true
 476                                 continue
 477                         }
 478                         bufW := bufio.NewWriter(tmp)
 479                         copied, err := nncp.CopyProgressed(
 480                                 bufW, bufio.NewReaderSize(fd, nncp.MTHBlockSize), "Tx",
 481                                 append(les, nncp.LE{K: "FullSize", V: job.Size}),
 482                                 ctx.ShowPrgrs,
 483                         )
 484                         fd.Close()
 485                         if err != nil {
 486                                 ctx.LogE("xfer-tx-copy", les, err, func(les nncp.LEs) string {
 487                                         return logMsg(les) + ": copying"
 488                                 })
 489                                 tmp.Close()
 490                                 isBad = true
 491                                 continue
 492                         }
 493                         if err = bufW.Flush(); err != nil {
 494                                 tmp.Close()
 495                                 ctx.LogE("xfer-tx-flush", les, err, func(les nncp.LEs) string {
 496                                         return logMsg(les) + ": flushing"
 497                                 })
 498                                 isBad = true
 499                                 continue
 500                         }
 501                         if !nncp.NoSync {
 502                                 if err = tmp.Sync(); err != nil {
 503                                         tmp.Close()
 504                                         ctx.LogE("xfer-tx-sync", les, err, func(les nncp.LEs) string {
 505                                                 return logMsg(les) + ": syncing"
 506                                         })
 507                                         isBad = true
 508                                         continue
 509                                 }
 510                         }
 511                         if err = tmp.Close(); err != nil {
 512                                 ctx.LogE("xfer-tx-close", les, err, func(les nncp.LEs) string {
 513                                         return logMsg(les) + ": closing"
 514                                 })
 515                         }
 516                         if err = os.Rename(tmp.Name(), filepath.Join(dstPath, pktName)); err != nil {
 517                                 ctx.LogE("xfer-tx-rename", les, err, func(les nncp.LEs) string {
 518                                         return logMsg(les) + ": renaming"
 519                                 })
 520                                 isBad = true
 521                                 continue
 522                         }
 523                         if err = nncp.DirSync(dstPath); err != nil {
 524                                 ctx.LogE("xfer-tx-dirsync", les, err, func(les nncp.LEs) string {
 525                                         return logMsg(les) + ": dirsyncing"
 526                                 })
 527                                 isBad = true
 528                                 continue
 529                         }
 530                         os.Remove(filepath.Join(dstPath, pktName+".part"))
 531                         les = les[:len(les)-1]
 532                         ctx.LogI(
 533                                 "xfer-tx",
 534                                 append(les, nncp.LE{K: "Size", V: copied}),
 535                                 func(les nncp.LEs) string {
 536                                         return fmt.Sprintf(
 537                                                 "%s (%s)", logMsg(les), humanize.IBytes(uint64(copied)),
 538                                         )
 539                                 },
 540                         )
 541                         if !*keep {
 542                                 if err = os.Remove(job.Path); err != nil {
 543                                         ctx.LogE("xfer-tx-remove", les, err, func(les nncp.LEs) string {
 544                                                 return logMsg(les) + ": removing"
 545                                         })
 546                                         isBad = true
 547                                 } else if ctx.HdrUsage {
 548                                         os.Remove(nncp.JobPath2Hdr(job.Path))
 549                                 }
 550                         }
 551                 }
 552                 ctx.UnlockDir(dirLock)
 553         }
 554         if isBad {
 555                 os.Exit(1)
 556         }
 557 }