src/cypherpunks.ru/nncp/sp.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         "bytes"
  23         "crypto/subtle"
  24         "errors"
  25         "io"
  26         "net"
  27         "os"
  28         "path/filepath"
  29         "sort"
  30         "strconv"
  31         "sync"
  32         "time"
  33
  34         "github.com/davecgh/go-xdr/xdr2"
  35         "github.com/flynn/noise"
  36 )
  37
  38 const (
  39         MaxSPSize       = 2<<16 - 256
  40         PartSuffix      = ".part"
  41         DefaultDeadline = 10
  42 )
  43
  44 var (
  45         MagicNNCPLv1 [8]byte = [8]byte{'N', 'N', 'C', 'P', 'S', 0, 0, 1}
  46
  47         SPHeadOverhead    int
  48         SPInfoOverhead    int
  49         SPFreqOverhead    int
  50         SPFileOverhead    int
  51         SPHaltMarshalized []byte
  52
  53         NoiseCipherSuite noise.CipherSuite = noise.NewCipherSuite(
  54                 noise.DH25519,
  55                 noise.CipherChaChaPoly,
  56                 noise.HashBLAKE2b,
  57         )
  58 )
  59
  60 type SPType uint8
  61
  62 const (
  63         SPTypeInfo SPType = iota
  64         SPTypeFreq SPType = iota
  65         SPTypeFile SPType = iota
  66         SPTypeDone SPType = iota
  67         SPTypeHalt SPType = iota
  68 )
  69
  70 type SPHead struct {
  71         Type SPType
  72 }
  73
  74 type SPInfo struct {
  75         Nice uint8
  76         Size uint64
  77         Hash *[32]byte
  78 }
  79
  80 type SPFreq struct {
  81         Hash   *[32]byte
  82         Offset uint64
  83 }
  84
  85 type SPFile struct {
  86         Hash    *[32]byte
  87         Offset  uint64
  88         Payload []byte
  89 }
  90
  91 type SPDone struct {
  92         Hash *[32]byte
  93 }
  94
  95 type SPRaw struct {
  96         Magic   [8]byte
  97         Payload []byte
  98 }
  99
 100 func init() {
 101         var buf bytes.Buffer
 102         spHead := SPHead{Type: SPTypeHalt}
 103         if _, err := xdr.Marshal(&buf, spHead); err != nil {
 104                 panic(err)
 105         }
 106         copy(SPHaltMarshalized, buf.Bytes())
 107         SPHeadOverhead = buf.Len()
 108         buf.Reset()
 109
 110         spInfo := SPInfo{Nice: 123, Size: 123, Hash: new([32]byte)}
 111         if _, err := xdr.Marshal(&buf, spInfo); err != nil {
 112                 panic(err)
 113         }
 114         SPInfoOverhead = buf.Len()
 115         buf.Reset()
 116
 117         spFreq := SPFreq{Hash: new([32]byte), Offset: 123}
 118         if _, err := xdr.Marshal(&buf, spFreq); err != nil {
 119                 panic(err)
 120         }
 121         SPFreqOverhead = buf.Len()
 122         buf.Reset()
 123
 124         spFile := SPFile{Hash: new([32]byte), Offset: 123}
 125         if _, err := xdr.Marshal(&buf, spFile); err != nil {
 126                 panic(err)
 127         }
 128         SPFileOverhead = buf.Len()
 129 }
 130
 131 func MarshalSP(typ SPType, sp interface{}) []byte {
 132         var buf bytes.Buffer
 133         var err error
 134         if _, err = xdr.Marshal(&buf, SPHead{typ}); err != nil {
 135                 panic(err)
 136         }
 137         if _, err = xdr.Marshal(&buf, sp); err != nil {
 138                 panic(err)
 139         }
 140         return buf.Bytes()
 141 }
 142
 143 func payloadsSplit(payloads [][]byte) [][]byte {
 144         var outbounds [][]byte
 145         outbound := make([]byte, 0, MaxSPSize)
 146         for i, payload := range payloads {
 147                 outbound = append(outbound, payload...)
 148                 if i+1 < len(payloads) && len(outbound)+len(payloads[i+1]) > MaxSPSize {
 149                         outbounds = append(outbounds, outbound)
 150                         outbound = make([]byte, 0, MaxSPSize)
 151                 }
 152         }
 153         if len(outbound) > 0 {
 154                 outbounds = append(outbounds, outbound)
 155         }
 156         return outbounds
 157 }
 158
 159 type SPState struct {
 160         ctx            *Ctx
 161         Node           *Node
 162         onlineDeadline int
 163         nice           uint8
 164         hs             *noise.HandshakeState
 165         csOur          *noise.CipherState
 166         csTheir        *noise.CipherState
 167         payloads       chan []byte
 168         infosTheir     map[[32]byte]*SPInfo
 169         infosOurSeen   map[[32]byte]struct{}
 170         queueTheir     []*SPFreq
 171         wg             sync.WaitGroup
 172         RxBytes        int64
 173         RxLastSeen     time.Time
 174         TxBytes        int64
 175         TxLastSeen     time.Time
 176         started        time.Time
 177         Duration       time.Duration
 178         RxSpeed        int64
 179         TxSpeed        int64
 180         rxLock         *os.File
 181         txLock         *os.File
 182         xxOnly         *TRxTx
 183         isDead         bool
 184         sync.RWMutex
 185 }
 186
 187 func (state *SPState) NotAlive() bool {
 188         now := time.Now()
 189         return state.isDead || (int(now.Sub(state.RxLastSeen).Seconds()) >= state.onlineDeadline && int(now.Sub(state.TxLastSeen).Seconds()) >= state.onlineDeadline)
 190 }
 191
 192 func (state *SPState) dirUnlock() {
 193         state.ctx.UnlockDir(state.rxLock)
 194         state.ctx.UnlockDir(state.txLock)
 195 }
 196
 197 func (state *SPState) WriteSP(dst io.Writer, payload []byte) error {
 198         n, err := xdr.Marshal(dst, SPRaw{Magic: MagicNNCPLv1, Payload: payload})
 199         if err == nil {
 200                 state.TxLastSeen = time.Now()
 201                 state.TxBytes += int64(n)
 202         }
 203         return err
 204 }
 205
 206 func (state *SPState) ReadSP(src io.Reader) ([]byte, error) {
 207         var sp SPRaw
 208         n, err := xdr.UnmarshalLimited(src, &sp, 2<<17)
 209         if err != nil {
 210                 return nil, err
 211         }
 212         state.RxLastSeen = time.Now()
 213         state.RxBytes += int64(n)
 214         if sp.Magic != MagicNNCPLv1 {
 215                 return nil, BadMagic
 216         }
 217         return sp.Payload, nil
 218 }
 219
 220 func (ctx *Ctx) infosOur(nodeId *NodeId, nice uint8, seen *map[[32]byte]struct{}) [][]byte {
 221         var infos []*SPInfo
 222         var totalSize int64
 223         for job := range ctx.Jobs(nodeId, TTx) {
 224                 job.Fd.Close()
 225                 if job.PktEnc.Nice > nice {
 226                         continue
 227                 }
 228                 if _, known := (*seen)[*job.HshValue]; known {
 229                         continue
 230                 }
 231                 totalSize += job.Size
 232                 infos = append(infos, &SPInfo{
 233                         Nice: job.PktEnc.Nice,
 234                         Size: uint64(job.Size),
 235                         Hash: job.HshValue,
 236                 })
 237                 (*seen)[*job.HshValue] = struct{}{}
 238         }
 239         sort.Sort(ByNice(infos))
 240         var payloads [][]byte
 241         for _, info := range infos {
 242                 payloads = append(payloads, MarshalSP(SPTypeInfo, info))
 243                 ctx.LogD("sp-info-our", SDS{
 244                         "node": nodeId,
 245                         "name": ToBase32(info.Hash[:]),
 246                         "size": strconv.FormatInt(int64(info.Size), 10),
 247                 }, "")
 248         }
 249         if totalSize > 0 {
 250                 ctx.LogI("sp-infos", SDS{
 251                         "xx":   string(TTx),
 252                         "node": nodeId,
 253                         "pkts": strconv.Itoa(len(payloads)),
 254                         "size": strconv.FormatInt(totalSize, 10),
 255                 }, "")
 256         }
 257         return payloadsSplit(payloads)
 258 }
 259
 260 func (ctx *Ctx) StartI(conn net.Conn, nodeId *NodeId, nice uint8, xxOnly *TRxTx, onlineDeadline int) (*SPState, error) {
 261         err := ctx.ensureRxDir(nodeId)
 262         if err != nil {
 263                 return nil, err
 264         }
 265         var rxLock *os.File
 266         if xxOnly != nil && *xxOnly == TRx {
 267                 rxLock, err = ctx.LockDir(nodeId, TRx)
 268                 if err != nil {
 269                         return nil, err
 270                 }
 271         }
 272         var txLock *os.File
 273         if xxOnly != nil && *xxOnly == TTx {
 274                 txLock, err = ctx.LockDir(nodeId, TTx)
 275                 if err != nil {
 276                         return nil, err
 277                 }
 278         }
 279         started := time.Now()
 280         node := ctx.Neigh[*nodeId]
 281         conf := noise.Config{
 282                 CipherSuite: NoiseCipherSuite,
 283                 Pattern:     noise.HandshakeIK,
 284                 Initiator:   true,
 285                 StaticKeypair: noise.DHKey{
 286                         Private: ctx.Self.NoisePrv[:],
 287                         Public:  ctx.Self.NoisePub[:],
 288                 },
 289                 PeerStatic: node.NoisePub[:],
 290         }
 291         state := SPState{
 292                 ctx:            ctx,
 293                 hs:             noise.NewHandshakeState(conf),
 294                 Node:           node,
 295                 onlineDeadline: onlineDeadline,
 296                 nice:           nice,
 297                 payloads:       make(chan []byte),
 298                 infosTheir:     make(map[[32]byte]*SPInfo),
 299                 infosOurSeen:   make(map[[32]byte]struct{}),
 300                 started:        started,
 301                 rxLock:         rxLock,
 302                 txLock:         txLock,
 303                 xxOnly:         xxOnly,
 304         }
 305
 306         var infosPayloads [][]byte
 307         if xxOnly == nil || *xxOnly != TTx {
 308                 infosPayloads = ctx.infosOur(nodeId, nice, &state.infosOurSeen)
 309         }
 310         var firstPayload []byte
 311         if len(infosPayloads) > 0 {
 312                 firstPayload = infosPayloads[0]
 313         }
 314         // Pad first payload, to hide actual number of existing files
 315         for i := 0; i < (MaxSPSize-len(firstPayload))/SPHeadOverhead; i++ {
 316                 firstPayload = append(firstPayload, SPHaltMarshalized...)
 317         }
 318
 319         var buf []byte
 320         var payload []byte
 321         buf, _, _ = state.hs.WriteMessage(nil, firstPayload)
 322         sds := SDS{"node": nodeId, "nice": strconv.Itoa(int(nice))}
 323         ctx.LogD("sp-start", sds, "sending first message")
 324         conn.SetWriteDeadline(time.Now().Add(DefaultDeadline * time.Second))
 325         if err = state.WriteSP(conn, buf); err != nil {
 326                 ctx.LogE("sp-start", SdsAdd(sds, SDS{"err": err}), "")
 327                 state.dirUnlock()
 328                 return nil, err
 329         }
 330         ctx.LogD("sp-start", sds, "waiting for first message")
 331         conn.SetReadDeadline(time.Now().Add(DefaultDeadline * time.Second))
 332         if buf, err = state.ReadSP(conn); err != nil {
 333                 ctx.LogE("sp-start", SdsAdd(sds, SDS{"err": err}), "")
 334                 state.dirUnlock()
 335                 return nil, err
 336         }
 337         payload, state.csOur, state.csTheir, err = state.hs.ReadMessage(nil, buf)
 338         if err != nil {
 339                 ctx.LogE("sp-start", SdsAdd(sds, SDS{"err": err}), "")
 340                 state.dirUnlock()
 341                 return nil, err
 342         }
 343         ctx.LogD("sp-start", sds, "starting workers")
 344         err = state.StartWorkers(conn, infosPayloads, payload)
 345         if err != nil {
 346                 ctx.LogE("sp-start", SdsAdd(sds, SDS{"err": err}), "")
 347                 state.dirUnlock()
 348                 return nil, err
 349         }
 350         return &state, err
 351 }
 352
 353 func (ctx *Ctx) StartR(conn net.Conn, nice uint8, xxOnly *TRxTx) (*SPState, error) {
 354         started := time.Now()
 355         conf := noise.Config{
 356                 CipherSuite: NoiseCipherSuite,
 357                 Pattern:     noise.HandshakeIK,
 358                 Initiator:   false,
 359                 StaticKeypair: noise.DHKey{
 360                         Private: ctx.Self.NoisePrv[:],
 361                         Public:  ctx.Self.NoisePub[:],
 362                 },
 363         }
 364         state := SPState{
 365                 ctx:          ctx,
 366                 hs:           noise.NewHandshakeState(conf),
 367                 nice:         nice,
 368                 payloads:     make(chan []byte),
 369                 infosOurSeen: make(map[[32]byte]struct{}),
 370                 infosTheir:   make(map[[32]byte]*SPInfo),
 371                 started:      started,
 372                 xxOnly:       xxOnly,
 373         }
 374         var buf []byte
 375         var payload []byte
 376         var err error
 377         ctx.LogD(
 378                 "sp-start",
 379                 SDS{"nice": strconv.Itoa(int(nice))},
 380                 "waiting for first message",
 381         )
 382         conn.SetReadDeadline(time.Now().Add(DefaultDeadline * time.Second))
 383         if buf, err = state.ReadSP(conn); err != nil {
 384                 ctx.LogE("sp-start", SDS{"err": err}, "")
 385                 return nil, err
 386         }
 387         if payload, _, _, err = state.hs.ReadMessage(nil, buf); err != nil {
 388                 ctx.LogE("sp-start", SDS{"err": err}, "")
 389                 return nil, err
 390         }
 391
 392         var node *Node
 393         for _, node = range ctx.Neigh {
 394                 if subtle.ConstantTimeCompare(state.hs.PeerStatic(), node.NoisePub[:]) == 1 {
 395                         break
 396                 }
 397         }
 398         if node == nil {
 399                 peerId := ToBase32(state.hs.PeerStatic())
 400                 ctx.LogE("sp-start", SDS{"peer": peerId}, "unknown")
 401                 return nil, errors.New("Unknown peer: " + peerId)
 402         }
 403         state.Node = node
 404         state.onlineDeadline = node.OnlineDeadline
 405         sds := SDS{"node": node.Id, "nice": strconv.Itoa(int(nice))}
 406
 407         if ctx.ensureRxDir(node.Id); err != nil {
 408                 return nil, err
 409         }
 410         var rxLock *os.File
 411         if xxOnly != nil && *xxOnly == TRx {
 412                 rxLock, err = ctx.LockDir(node.Id, TRx)
 413                 if err != nil {
 414                         return nil, err
 415                 }
 416         }
 417         state.rxLock = rxLock
 418         var txLock *os.File
 419         if xxOnly != nil && *xxOnly == TTx {
 420                 txLock, err = ctx.LockDir(node.Id, TTx)
 421                 if err != nil {
 422                         return nil, err
 423                 }
 424         }
 425         state.txLock = txLock
 426
 427         var infosPayloads [][]byte
 428         if xxOnly == nil || *xxOnly != TTx {
 429                 infosPayloads = ctx.infosOur(node.Id, nice, &state.infosOurSeen)
 430         }
 431         var firstPayload []byte
 432         if len(infosPayloads) > 0 {
 433                 firstPayload = infosPayloads[0]
 434         }
 435         // Pad first payload, to hide actual number of existing files
 436         for i := 0; i < (MaxSPSize-len(firstPayload))/SPHeadOverhead; i++ {
 437                 firstPayload = append(firstPayload, SPHaltMarshalized...)
 438         }
 439
 440         ctx.LogD("sp-start", sds, "sending first message")
 441         buf, state.csTheir, state.csOur = state.hs.WriteMessage(nil, firstPayload)
 442         conn.SetWriteDeadline(time.Now().Add(DefaultDeadline * time.Second))
 443         if err = state.WriteSP(conn, buf); err != nil {
 444                 ctx.LogE("sp-start", SdsAdd(sds, SDS{"err": err}), "")
 445                 state.dirUnlock()
 446                 return nil, err
 447         }
 448         ctx.LogD("sp-start", sds, "starting workers")
 449         err = state.StartWorkers(conn, infosPayloads, payload)
 450         if err != nil {
 451                 state.dirUnlock()
 452                 return nil, err
 453         }
 454         return &state, err
 455 }
 456
 457 func (state *SPState) StartWorkers(conn net.Conn, infosPayloads [][]byte, payload []byte) error {
 458         sds := SDS{"node": state.Node.Id, "nice": strconv.Itoa(int(state.nice))}
 459         if len(infosPayloads) > 1 {
 460                 go func() {
 461                         for _, payload := range infosPayloads[1:] {
 462                                 state.ctx.LogD(
 463                                         "sp-work",
 464                                         SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 465                                         "queuing remaining payload",
 466                                 )
 467                                 state.payloads <- payload
 468                         }
 469                 }()
 470         }
 471         state.ctx.LogD(
 472                 "sp-work",
 473                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 474                 "processing first payload",
 475         )
 476         replies, err := state.ProcessSP(payload)
 477         if err != nil {
 478                 state.ctx.LogE("sp-work", SdsAdd(sds, SDS{"err": err}), "")
 479                 return err
 480         }
 481
 482         go func() {
 483                 for _, reply := range replies {
 484                         state.ctx.LogD(
 485                                 "sp-work",
 486                                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(reply))}),
 487                                 "queuing reply",
 488                         )
 489                         state.payloads <- reply
 490                 }
 491         }()
 492
 493         go func() {
 494                 for range time.Tick(time.Second) {
 495                         for _, payload := range state.ctx.infosOur(
 496                                 state.Node.Id,
 497                                 state.nice,
 498                                 &state.infosOurSeen,
 499                         ) {
 500                                 state.ctx.LogD(
 501                                         "sp-work",
 502                                         SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 503                                         "queuing new info",
 504                                 )
 505                                 state.payloads <- payload
 506                         }
 507                 }
 508         }()
 509
 510         state.wg.Add(1)
 511         go func() {
 512                 defer func() {
 513                         state.isDead = true
 514                         state.wg.Done()
 515                 }()
 516                 for {
 517                         if state.NotAlive() {
 518                                 return
 519                         }
 520                         var payload []byte
 521                         select {
 522                         case payload = <-state.payloads:
 523                                 state.ctx.LogD(
 524                                         "sp-xmit",
 525                                         SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 526                                         "got payload",
 527                                 )
 528                         default:
 529                         }
 530                         if payload == nil {
 531                                 state.RLock()
 532                                 if len(state.queueTheir) == 0 {
 533                                         state.ctx.LogD("sp-xmit", sds, "file queue is empty")
 534                                         state.RUnlock()
 535                                         time.Sleep(100 * time.Millisecond)
 536                                         continue
 537                                 }
 538                                 freq := state.queueTheir[0]
 539                                 state.RUnlock()
 540                                 sdsp := SdsAdd(sds, SDS{
 541                                         "xx":   string(TTx),
 542                                         "hash": ToBase32(freq.Hash[:]),
 543                                         "size": strconv.FormatInt(int64(freq.Offset), 10),
 544                                 })
 545                                 state.ctx.LogD("sp-file", sdsp, "queueing")
 546                                 fd, err := os.Open(filepath.Join(
 547                                         state.ctx.Spool,
 548                                         state.Node.Id.String(),
 549                                         string(TTx),
 550                                         ToBase32(freq.Hash[:]),
 551                                 ))
 552                                 if err != nil {
 553                                         state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 554                                         break
 555                                 }
 556                                 fi, err := fd.Stat()
 557                                 if err != nil {
 558                                         state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 559                                         break
 560                                 }
 561                                 fullSize := uint64(fi.Size())
 562                                 var buf []byte
 563                                 if freq.Offset < fullSize {
 564                                         state.ctx.LogD("sp-file", sdsp, "seeking")
 565                                         if _, err = fd.Seek(int64(freq.Offset), io.SeekStart); err != nil {
 566                                                 state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 567                                                 break
 568                                         }
 569                                         buf = make([]byte, MaxSPSize-SPHeadOverhead-SPFileOverhead)
 570                                         n, err := fd.Read(buf)
 571                                         if err != nil {
 572                                                 state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 573                                                 break
 574                                         }
 575                                         buf = buf[:n]
 576                                         state.ctx.LogD(
 577                                                 "sp-file",
 578                                                 SdsAdd(sdsp, SDS{"size": strconv.Itoa(n)}),
 579                                                 "read",
 580                                         )
 581                                 }
 582                                 fd.Close()
 583                                 payload = MarshalSP(SPTypeFile, SPFile{
 584                                         Hash:    freq.Hash,
 585                                         Offset:  freq.Offset,
 586                                         Payload: buf,
 587                                 })
 588                                 ourSize := freq.Offset + uint64(len(buf))
 589                                 sdsp["size"] = strconv.FormatInt(int64(ourSize), 10)
 590                                 sdsp["fullsize"] = strconv.FormatInt(int64(fullSize), 10)
 591                                 state.ctx.LogP("sp-file", sdsp, "")
 592                                 state.Lock()
 593                                 if len(state.queueTheir) > 0 && *state.queueTheir[0].Hash == *freq.Hash {
 594                                         if ourSize == fullSize {
 595                                                 state.ctx.LogD("sp-file", sdsp, "finished")
 596                                                 if len(state.queueTheir) > 1 {
 597                                                         state.queueTheir = state.queueTheir[1:]
 598                                                 } else {
 599                                                         state.queueTheir = state.queueTheir[:0]
 600                                                 }
 601                                         } else {
 602                                                 state.queueTheir[0].Offset += uint64(len(buf))
 603                                         }
 604                                 } else {
 605                                         state.ctx.LogD("sp-file", sdsp, "queue disappeared")
 606                                 }
 607                                 state.Unlock()
 608                         }
 609                         state.ctx.LogD(
 610                                 "sp-xmit",
 611                                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 612                                 "sending",
 613                         )
 614                         conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
 615                         if err := state.WriteSP(conn, state.csOur.Encrypt(nil, nil, payload)); err != nil {
 616                                 state.ctx.LogE("sp-xmit", SdsAdd(sds, SDS{"err": err}), "")
 617                                 break
 618                         }
 619                 }
 620         }()
 621
 622         state.wg.Add(1)
 623         go func() {
 624                 defer func() {
 625                         state.isDead = true
 626                         state.wg.Done()
 627                 }()
 628                 for {
 629                         if state.NotAlive() {
 630                                 return
 631                         }
 632                         state.ctx.LogD("sp-recv", sds, "waiting for payload")
 633                         conn.SetReadDeadline(time.Now().Add(DefaultDeadline * time.Second))
 634                         payload, err := state.ReadSP(conn)
 635                         if err != nil {
 636                                 unmarshalErr := err.(*xdr.UnmarshalError)
 637                                 netErr, ok := unmarshalErr.Err.(net.Error)
 638                                 if ok && netErr.Timeout() {
 639                                         continue
 640                                 }
 641                                 if unmarshalErr.ErrorCode == xdr.ErrIO {
 642                                         break
 643                                 }
 644                                 state.ctx.LogE("sp-recv", SdsAdd(sds, SDS{"err": err}), "")
 645                                 break
 646                         }
 647                         state.ctx.LogD(
 648                                 "sp-recv",
 649                                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 650                                 "got payload",
 651                         )
 652                         payload, err = state.csTheir.Decrypt(nil, nil, payload)
 653                         if err != nil {
 654                                 state.ctx.LogE("sp-recv", SdsAdd(sds, SDS{"err": err}), "")
 655                                 break
 656                         }
 657                         state.ctx.LogD(
 658                                 "sp-recv",
 659                                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(payload))}),
 660                                 "processing",
 661                         )
 662                         replies, err := state.ProcessSP(payload)
 663                         if err != nil {
 664                                 state.ctx.LogE("sp-recv", SdsAdd(sds, SDS{"err": err}), "")
 665                                 break
 666                         }
 667                         go func() {
 668                                 for _, reply := range replies {
 669                                         state.ctx.LogD(
 670                                                 "sp-recv",
 671                                                 SdsAdd(sds, SDS{"size": strconv.Itoa(len(reply))}),
 672                                                 "queuing reply",
 673                                         )
 674                                         state.payloads <- reply
 675                                 }
 676                         }()
 677                 }
 678         }()
 679
 680         return nil
 681 }
 682
 683 func (state *SPState) Wait() {
 684         state.wg.Wait()
 685         state.dirUnlock()
 686         state.Duration = time.Now().Sub(state.started)
 687         state.RxSpeed = state.RxBytes
 688         state.TxSpeed = state.TxBytes
 689         rxDuration := int64(state.RxLastSeen.Sub(state.started).Seconds())
 690         txDuration := int64(state.TxLastSeen.Sub(state.started).Seconds())
 691         if rxDuration > 0 {
 692                 state.RxSpeed = state.RxBytes / rxDuration
 693         }
 694         if txDuration > 0 {
 695                 state.TxSpeed = state.TxBytes / txDuration
 696         }
 697 }
 698
 699 func (state *SPState) ProcessSP(payload []byte) ([][]byte, error) {
 700         sds := SDS{"node": state.Node.Id, "nice": strconv.Itoa(int(state.nice))}
 701         r := bytes.NewReader(payload)
 702         var err error
 703         var replies [][]byte
 704         var infosGot bool
 705         for r.Len() > 0 {
 706                 state.ctx.LogD("sp-process", sds, "unmarshaling header")
 707                 var head SPHead
 708                 if _, err = xdr.Unmarshal(r, &head); err != nil {
 709                         state.ctx.LogE("sp-process", SdsAdd(sds, SDS{"err": err}), "")
 710                         return nil, err
 711                 }
 712                 switch head.Type {
 713                 case SPTypeInfo:
 714                         infosGot = true
 715                         sdsp := SdsAdd(sds, SDS{"type": "info"})
 716                         state.ctx.LogD("sp-process", sdsp, "unmarshaling packet")
 717                         var info SPInfo
 718                         if _, err = xdr.Unmarshal(r, &info); err != nil {
 719                                 state.ctx.LogE("sp-process", SdsAdd(sdsp, SDS{"err": err}), "")
 720                                 return nil, err
 721                         }
 722                         sdsp = SdsAdd(sds, SDS{
 723                                 "hash": ToBase32(info.Hash[:]),
 724                                 "size": strconv.FormatInt(int64(info.Size), 10),
 725                         })
 726                         if info.Nice > state.nice {
 727                                 state.ctx.LogD("sp-process", sdsp, "too nice")
 728                                 continue
 729                         }
 730                         state.ctx.LogD("sp-process", sdsp, "received")
 731                         if state.xxOnly != nil && *state.xxOnly == TTx {
 732                                 continue
 733                         }
 734                         state.Lock()
 735                         state.infosTheir[*info.Hash] = &info
 736                         state.Unlock()
 737                         state.ctx.LogD("sp-process", sdsp, "stating part")
 738                         if _, err = os.Stat(filepath.Join(
 739                                 state.ctx.Spool,
 740                                 state.Node.Id.String(),
 741                                 string(TRx),
 742                                 ToBase32(info.Hash[:]),
 743                         )); err == nil {
 744                                 state.ctx.LogD("sp-process", sdsp, "already done")
 745                                 replies = append(replies, MarshalSP(SPTypeDone, SPDone{info.Hash}))
 746                                 continue
 747                         }
 748                         fi, err := os.Stat(filepath.Join(
 749                                 state.ctx.Spool,
 750                                 state.Node.Id.String(),
 751                                 string(TRx),
 752                                 ToBase32(info.Hash[:])+PartSuffix,
 753                         ))
 754                         var offset int64
 755                         if err == nil {
 756                                 offset = fi.Size()
 757                                 state.ctx.LogD(
 758                                         "sp-process",
 759                                         SdsAdd(sdsp, SDS{"offset": strconv.FormatInt(offset, 10)}),
 760                                         "part exists",
 761                                 )
 762                         }
 763                         replies = append(replies, MarshalSP(
 764                                 SPTypeFreq,
 765                                 SPFreq{info.Hash, uint64(offset)},
 766                         ))
 767                 case SPTypeFile:
 768                         state.ctx.LogD(
 769                                 "sp-process",
 770                                 SdsAdd(sds, SDS{"type": "file"}),
 771                                 "unmarshaling packet",
 772                         )
 773                         var file SPFile
 774                         if _, err = xdr.Unmarshal(r, &file); err != nil {
 775                                 state.ctx.LogE("sp-process", SdsAdd(sds, SDS{
 776                                         "err":  err,
 777                                         "type": "file",
 778                                 }), "")
 779                                 return nil, err
 780                         }
 781                         sdsp := SdsAdd(sds, SDS{
 782                                 "xx":   string(TRx),
 783                                 "hash": ToBase32(file.Hash[:]),
 784                                 "size": strconv.Itoa(len(file.Payload)),
 785                         })
 786                         filePath := filepath.Join(
 787                                 state.ctx.Spool,
 788                                 state.Node.Id.String(),
 789                                 string(TRx),
 790                                 ToBase32(file.Hash[:]),
 791                         )
 792                         state.ctx.LogD("sp-file", sdsp, "opening part")
 793                         fd, err := os.OpenFile(
 794                                 filePath+PartSuffix,
 795                                 os.O_RDWR|os.O_CREATE,
 796                                 os.FileMode(0600),
 797                         )
 798                         if err != nil {
 799                                 state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 800                                 return nil, err
 801                         }
 802                         state.ctx.LogD(
 803                                 "sp-file",
 804                                 SdsAdd(sdsp, SDS{"offset": strconv.FormatInt(int64(file.Offset), 10)}),
 805                                 "seeking",
 806                         )
 807                         if _, err = fd.Seek(int64(file.Offset), io.SeekStart); err != nil {
 808                                 state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 809                                 fd.Close()
 810                                 return nil, err
 811                         }
 812                         state.ctx.LogD("sp-file", sdsp, "writing")
 813                         _, err = fd.Write(file.Payload)
 814                         if err != nil {
 815                                 state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "")
 816                                 fd.Close()
 817                                 return nil, err
 818                         }
 819                         ourSize := uint64(file.Offset) + uint64(len(file.Payload))
 820                         sdsp["fullsize"] = strconv.FormatInt(int64(state.infosTheir[*file.Hash].Size), 10)
 821                         sdsp["size"] = strconv.FormatInt(int64(ourSize), 10)
 822                         state.ctx.LogP("sp-file", sdsp, "")
 823                         if state.infosTheir[*file.Hash].Size != ourSize {
 824                                 fd.Close()
 825                                 continue
 826                         }
 827                         go func() {
 828                                 if err := fd.Sync(); err != nil {
 829                                         state.ctx.LogE("sp-file", SdsAdd(sdsp, SDS{"err": err}), "sync")
 830                                         fd.Close()
 831                                         return
 832                                 }
 833                                 state.wg.Add(1)
 834                                 defer state.wg.Done()
 835                                 fd.Seek(0, io.SeekStart)
 836                                 state.ctx.LogD("sp-file", sdsp, "checking")
 837                                 gut, err := Check(fd, file.Hash[:])
 838                                 fd.Close()
 839                                 if err != nil || !gut {
 840                                         state.ctx.LogE("sp-file", sdsp, "checksum mismatch")
 841                                         return
 842                                 }
 843                                 state.ctx.LogI("sp-done", SdsAdd(sdsp, SDS{"xx": string(TRx)}), "")
 844                                 os.Rename(filePath+PartSuffix, filePath)
 845                                 go func() {
 846                                         state.payloads <- MarshalSP(SPTypeDone, SPDone{file.Hash})
 847                                 }()
 848                         }()
 849                 case SPTypeDone:
 850                         state.ctx.LogD(
 851                                 "sp-process",
 852                                 SdsAdd(sds, SDS{"type": "done"}),
 853                                 "unmarshaling packet",
 854                         )
 855                         var done SPDone
 856                         if _, err = xdr.Unmarshal(r, &done); err != nil {
 857                                 state.ctx.LogE("sp-process", SdsAdd(sds, SDS{
 858                                         "type": "done",
 859                                         "err":  err,
 860                                 }), "")
 861                                 return nil, err
 862                         }
 863                         sdsp := SdsAdd(sds, SDS{"hash": ToBase32(done.Hash[:])})
 864                         state.ctx.LogD("sp-done", sdsp, "removing")
 865                         err := os.Remove(filepath.Join(
 866                                 state.ctx.Spool,
 867                                 state.Node.Id.String(),
 868                                 string(TTx),
 869                                 ToBase32(done.Hash[:]),
 870                         ))
 871                         if err == nil {
 872                                 state.ctx.LogI("sp-done", SdsAdd(sdsp, SDS{"xx": string(TTx)}), "")
 873                         } else {
 874                                 state.ctx.LogE("sp-done", SdsAdd(sdsp, SDS{"xx": string(TTx)}), "")
 875                         }
 876                 case SPTypeFreq:
 877                         sdsp := SdsAdd(sds, SDS{"type": "freq"})
 878                         state.ctx.LogD("sp-process", sdsp, "unmarshaling packet")
 879                         var freq SPFreq
 880                         if _, err = xdr.Unmarshal(r, &freq); err != nil {
 881                                 state.ctx.LogE("sp-process", SdsAdd(sdsp, SDS{"err": err}), "")
 882                                 return nil, err
 883                         }
 884                         state.ctx.LogD("sp-process", SdsAdd(sdsp, SDS{
 885                                 "hash":   ToBase32(freq.Hash[:]),
 886                                 "offset": strconv.FormatInt(int64(freq.Offset), 10),
 887                         }), "queueing")
 888                         state.Lock()
 889                         state.queueTheir = append(state.queueTheir, &freq)
 890                         state.Unlock()
 891                 case SPTypeHalt:
 892                         sdsp := SdsAdd(sds, SDS{"type": "halt"})
 893                         state.ctx.LogD("sp-process", sdsp, "")
 894                         state.Lock()
 895                         state.queueTheir = nil
 896                         state.Unlock()
 897                 default:
 898                         state.ctx.LogE(
 899                                 "sp-process",
 900                                 SdsAdd(sds, SDS{"type": head.Type}),
 901                                 "unknown",
 902                         )
 903                         return nil, BadPktType
 904                 }
 905         }
 906         if infosGot {
 907                 var pkts int
 908                 var size uint64
 909                 for _, info := range state.infosTheir {
 910                         pkts++
 911                         size += info.Size
 912                 }
 913                 state.ctx.LogI("sp-infos", SDS{
 914                         "xx":   string(TRx),
 915                         "node": state.Node.Id,
 916                         "pkts": strconv.Itoa(pkts),
 917                         "size": strconv.FormatInt(int64(size), 10),
 918                 }, "")
 919         }
 920         return payloadsSplit(replies), nil
 921 }