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