From b95da7effe52ab60828a62b4b07a59a21972f90b Mon Sep 17 00:00:00 2001 From: Joseph Lynch Date: Sat, 4 Apr 2026 09:53:26 -0400 Subject: [PATCH 1/4] Race detection and race fixes --- server/Makefile | 2 +- server/internal/cmd/filexfercli/cli.go | 141 ++++++++++++++------- server/internal/filexfer/ftcp/send_test.go | 14 +- 3 files changed, 104 insertions(+), 53 deletions(-) diff --git a/server/Makefile b/server/Makefile index 2dc719b..2772cbd 100644 --- a/server/Makefile +++ b/server/Makefile @@ -8,7 +8,7 @@ build: CGO_ENABLED=0 go build -a -tags netgo -ldflags='-s -w -extldflags "-static"' -o pinch-server test: build - go test ./... + go test -race ./... fuzz: go test ./internal/filexfer/encoding -run=^$$ -fuzz=FuzzRoundTrip -fuzztime=$(FUZZTIME) diff --git a/server/internal/cmd/filexfercli/cli.go b/server/internal/cmd/filexfercli/cli.go index adb8611..dbaf0c6 100644 --- a/server/internal/cmd/filexfercli/cli.go +++ b/server/internal/cmd/filexfercli/cli.go @@ -301,14 +301,14 @@ func resolveEncryptionOptions(mode string) (pubKey string, identity string, encM switch mode { case "", "none": return "", "", "", nil - case "age", "aes": + case "auto", "aes", "chacha20": id, genErr := age.GenerateX25519Identity() if genErr != nil { return "", "", "", fmt.Errorf("generate age identity: %w", genErr) } return id.Recipient().String(), id.String(), mode, nil default: - return "", "", "", fmt.Errorf("unsupported --encrypt value %q (supported: none, age, aes)", mode) + return "", "", "", fmt.Errorf("unsupported --encrypt value %q (supported: none, auto, aes, chacha20)", mode) } } @@ -459,7 +459,7 @@ func runCopyCLI(serverURL string, args []string, stdout io.Writer, stderr io.Wri cf.BoolVar(&cfg.verifyMeta, "", "verify-meta", false, "Run read-only metadata verification after copy; with --skip-fetch this is allowed only if LOCAL_DST already exists") cf.IntVar(&cfg.verifyDataSamplePct, "", "verify-data-sample", 0, "Percent of frame slots to sample per file for data verification (0-100); implies --verify-meta; not allowed with --skip-fetch or --skip-write") cf.StringVar(&cfg.modeRaw, "", "mode", LoadStrategyFast, "Server read strategy: fast|gentle") - cf.StringVar(&cfg.encryptMode, "", "encrypt", "", "Encryption algorithm: none|age|aes (default: none)") + cf.StringVar(&cfg.encryptMode, "", "encrypt", "", "Encryption algorithm: none|auto|aes|chacha20 (default: none)") cf.StringVar(&cfg.compressRaw, "", "compress", "", "Compression algorithm: adapt|none|lz4|zstd (default: adapt)") cf.IntVar(&cfg.concurrency, "", "concurrency", 0, "Parallel download / verification workers (0=adapt from server)") cf.BoolVar(&cfg.progress, "", "progress", true, "Show transfer progress every 2s") @@ -996,7 +996,7 @@ func runTransferCLI(serverURL string, args []string, stdout io.Writer, stderr io var maxChunk int var deadlineRaw string cf.StringVar(&sourceDir, "s", "source-directory", "", "Absolute source directory to transfer") - cf.StringVar(&encryptMode, "", "encrypt", "", "Encryption algorithm: none|age|aes (default: none)") + cf.StringVar(&encryptMode, "", "encrypt", "", "Encryption algorithm: none|auto|aes|chacha20 (default: none)") cf.StringVar(&loadStrategyRaw, "", "load-strategy", LoadStrategyFast, "Server load strategy (fast|gentle)") probeSizeRaw = encoding.HumanBytes(defaultCLIProbeBytes) cf.StringVar(&probeSizeRaw, "", "probe-size", probeSizeRaw, "Probe payload size for transfer metadata; 1B, 4KiB, 8MiB") @@ -1092,10 +1092,15 @@ func runTransfer(serverURL string, cfg transferArgs, stdout io.Writer, stderr io fmt.Fprintf(stderr, "probe failed: %v\n", err) return 1 } + cipherDisplay := "none" + if probeResult.Cipher != "" { + cipherDisplay = probeResult.Cipher + } fmt.Fprintf( stdout, - "transfer-probe : strategy=%s avg_ms=%d est_link=%dMbps srv-conc=(%d cpu * %d io = %d)\n", + "transfer-probe : strategy=%s cipher=%s avg_ms=%d est_link=%dMbps srv-conc=(%d cpu * %d io = %d)\n", cfg.loadStrategy, + cipherDisplay, probeResult.AvgLatencyMS, probeResult.LinkMbps, probeResult.ServerCPU, probeResult.ServerIODepth, probeResult.SuggestedConcurrency, @@ -1388,7 +1393,7 @@ func runGetCLI(serverURL string, args []string, stdout io.Writer, stderr io.Writ var progressFilePath string var progressFileIntervalRaw string cf.StringVar(&outFile, "o", "", "", "Output file path, or '-' for stdout") - cf.StringVar(&encryptMode, "", "encrypt", "", "Encryption algorithm: none|age|aes (default: none)") + cf.StringVar(&encryptMode, "", "encrypt", "", "Encryption algorithm: none|auto|aes|chacha20 (default: none)") cf.StringVar(&compressRaw, "", "compress", "", "Compression algorithm: adapt|none|lz4|zstd (default: adapt)") cf.IntVar(&concurrency, "", "concurrency", 0, "Parallel download workers (0=auto)") cf.BoolVar(&skipWrite, "", "skip-write", false, "Do not write the file; fetch to discard instead") @@ -1505,7 +1510,6 @@ func runGetCLI(serverURL string, args []string, stdout io.Writer, stderr io.Writ onProgressUpdate = progressReporter.ReportUpdate } forwardProgress := func(update DownloadProgressUpdate) { - applyProgressUpdateToManifest(manifest, update) if onProgressUpdate != nil { onProgressUpdate(update) } @@ -1592,7 +1596,7 @@ func runSyncCLI(serverURL string, args []string, stdout io.Writer, stderr io.Wri var progressFilePath string var progressFileIntervalRaw string cf.StringVar(&sourceDir, "s", "source-directory", "", "Absolute source directory on server (default: manifest root)") - cf.StringVar(&encryptMode, "", "encrypt", "", "Encryption algorithm: none|age|aes (default: none)") + cf.StringVar(&encryptMode, "", "encrypt", "", "Encryption algorithm: none|auto|aes|chacha20 (default: none)") cf.StringVar(&compressRaw, "", "compress", "", "Compression algorithm: adapt|none|lz4|zstd (default: adapt)") cf.IntVar(&concurrency, "", "concurrency", 0, "Parallel download workers (0=manifest default)") cf.BoolVar(&yes, "y", "yes", false, "Skip confirmation prompt") @@ -1741,11 +1745,11 @@ func runSync(serverURL string, cfg syncArgs, stdout io.Writer, stderr io.Writer) // SYNC: send old manifest, receive new manifest + removed paths. syncResp, err := client.SyncManifest(context.Background(), SyncManifestRequest{ - Directory: syncSourceDir, - OldManifest: oldManifest, - Mode: loadStrategy, - LinkMbps: probeResult.LinkMbps, - Concurrency: probeResult.SuggestedConcurrency, + Directory: syncSourceDir, + OldManifest: oldManifest, + Mode: loadStrategy, + LinkMbps: probeResult.LinkMbps, + Concurrency: probeResult.SuggestedConcurrency, }) if err != nil { fmt.Fprintf(stderr, "sync failed: %v\n", err) @@ -1878,20 +1882,22 @@ func runSync(serverURL string, cfg syncArgs, stdout io.Writer, stderr io.Writer) } progressUpdates := make(chan DownloadProgressUpdate, 1024) + entryByID := manifestEntriesByID(mergedManifest) var onProgressUpdate func(DownloadProgressUpdate) if cfg.verbose { progressReporter := newVerboseProgressReporter(stderr) onProgressUpdate = progressReporter.ReportUpdate } forwardProgress := func(update DownloadProgressUpdate) { - applyProgressUpdateToManifest(mergedManifest, update) if onProgressUpdate != nil { onProgressUpdate(update) } } - stopProgress, markMetadataDonePersisted := startProgressWriter(ps.ProgressPath, mergedProgress, progressUpdates, forwardProgress, stderr) + stopProgress, persistProgressAck, markMetadataDonePersisted := startProgressWriter(ps.ProgressPath, mergedProgress, progressUpdates, forwardProgress, stderr) + persistFileDone := func(fileID uint64, ackBytes int64) { + persistProgressAck(fileID, ackBytes) + } markMetadataDone := func(fileID uint64) { - markManifestEntryMetadataDone(mergedManifest, fileID) markMetadataDonePersisted(fileID) } @@ -1948,7 +1954,7 @@ func runSync(serverURL string, cfg syncArgs, stdout io.Writer, stderr io.Writer) BatchMaxBytes: batchSize, ProgressUpdates: progressUpdates, OnFileDone: func(evt StartFileDoneEvent) { - entry, ok := mergedManifest.EntryByID(evt.File.Meta.FileID) + entry, ok := entryByID[evt.File.Meta.FileID] if !ok { recordFailure(fmt.Errorf("id=%d metadata apply failed: file id not in manifest", evt.File.Meta.FileID)) return @@ -1961,11 +1967,13 @@ func runSync(serverURL string, cfg syncArgs, stdout io.Writer, stderr io.Writer) recordFailure(fmt.Errorf("id=%d metadata apply failed: %w", evt.File.Meta.FileID, err)) return } + persistFileDone(evt.File.Meta.FileID, entry.Size) markMetadataDone(evt.File.Meta.FileID) printStartFileSummary(stdout, evt.File.Meta.FileID, destPath, evt.File.Meta, evt.File.LocalFileHash, evt.File.WindowChecksumPassed, evt.File.WindowChecksumTotal, evt.Elapsed) }, }) stopProgress() + applyProgressStateToManifest(mergedManifest, mergedProgress) if err != nil { fmt.Fprintf(stderr, "sync download failed: %v\n", err) return 1 @@ -2028,7 +2036,7 @@ func runStartCLI(serverURL string, args []string, stdout io.Writer, stderr io.Wr var deadlineRaw string var progressFilePath string var progressFileIntervalRaw string - cf.StringVar(&encryptMode, "", "encrypt", "", "Encryption algorithm: none|age|aes (default: none)") + cf.StringVar(&encryptMode, "", "encrypt", "", "Encryption algorithm: none|auto|aes|chacha20 (default: none)") cf.BoolVar(&progress, "", "progress", true, "Show transfer progress every 2s") cf.BoolVar(&verbose, "v", "verbose", false, "Per-file progress output") cf.StringVar(&progressFilePath, "", "progress-file", "", "Write integer % to this file/pipe") @@ -2178,20 +2186,22 @@ func runStart(serverURL string, cfg startArgs, stdout io.Writer, stderr io.Write manifest.DeadlineMS = cfg.deadlineMS } progressUpdates := make(chan DownloadProgressUpdate, 1024) + entryByID := manifestEntriesByID(manifest) var onStartProgressUpdate func(DownloadProgressUpdate) if cfg.verbosity >= 2 { progressReporter := newVerboseProgressReporter(stderr) onStartProgressUpdate = progressReporter.ReportUpdate } forwardProgress := func(update DownloadProgressUpdate) { - applyProgressUpdateToManifest(manifest, update) if onStartProgressUpdate != nil { onStartProgressUpdate(update) } } - stopProgress, markMetadataDonePersisted := startProgressWriter(ps.ProgressPath, progressState, progressUpdates, forwardProgress, stderr) + stopProgress, persistProgressAck, markMetadataDonePersisted := startProgressWriter(ps.ProgressPath, progressState, progressUpdates, forwardProgress, stderr) + persistFileDone := func(fileID uint64, ackBytes int64) { + persistProgressAck(fileID, ackBytes) + } markMetadataDone := func(fileID uint64) { - markManifestEntryMetadataDone(manifest, fileID) markMetadataDonePersisted(fileID) } progressStopped := false @@ -2253,6 +2263,7 @@ func runStart(serverURL string, cfg startArgs, stdout io.Writer, stderr io.Write recordFailure(fmt.Errorf("id=%d metadata refresh failed: %w", entry.ID, err)) continue } + persistFileDone(entry.ID, entry.Size) markMetadataDone(entry.ID) completed++ continue @@ -2293,7 +2304,7 @@ func runStart(serverURL string, cfg startArgs, stdout io.Writer, stderr io.Write BatchMaxBytes: batchSize, ProgressUpdates: progressUpdates, OnFileDone: func(evt StartFileDoneEvent) { - entry, ok := manifest.EntryByID(evt.File.Meta.FileID) + entry, ok := entryByID[evt.File.Meta.FileID] if !ok { recordFailure(fmt.Errorf("id=%d metadata apply failed: file id not in manifest", evt.File.Meta.FileID)) return @@ -2303,6 +2314,7 @@ func runStart(serverURL string, cfg startArgs, stdout io.Writer, stderr io.Write recordFailure(fmt.Errorf("id=%d metadata apply failed: %w", evt.File.Meta.FileID, err)) return } + persistFileDone(evt.File.Meta.FileID, entry.Size) markMetadataDone(evt.File.Meta.FileID) if cfg.verbosity >= 2 { printStartFileSummary(stdout, evt.File.Meta.FileID, destPath, evt.File.Meta, evt.File.LocalFileHash, evt.File.WindowChecksumPassed, evt.File.WindowChecksumTotal, evt.Elapsed) @@ -2310,6 +2322,8 @@ func runStart(serverURL string, cfg startArgs, stdout io.Writer, stderr io.Write }, }) if err != nil { + stopProgress() + progressStopped = true fmt.Fprintf(stderr, "start failed: %v\n", err) return 1 } @@ -2320,6 +2334,7 @@ func runStart(serverURL string, cfg startArgs, stdout io.Writer, stderr io.Write } stopProgress() progressStopped = true + applyProgressStateToManifest(manifest, progressState) failuresMu.Lock() finalFailures := append([]error(nil), failures...) failuresMu.Unlock() @@ -2599,6 +2614,17 @@ func applyProgressUpdateToManifest(manifest *Manifest, update DownloadProgressUp } } +func manifestEntriesByID(manifest *Manifest) map[uint64]ManifestEntry { + if manifest == nil || len(manifest.Entries) == 0 { + return nil + } + entries := make(map[uint64]ManifestEntry, len(manifest.Entries)) + for _, entry := range manifest.Entries { + entries[entry.ID] = entry + } + return entries +} + func markManifestEntryMetadataDone(manifest *Manifest, fileID uint64) { if manifest == nil { return @@ -2669,13 +2695,19 @@ type metadataProgressUpdate struct { FileID uint64 } -func startProgressWriter(progressPath string, initial map[uint64]ManifestProgress, updates <-chan DownloadProgressUpdate, onUpdate func(DownloadProgressUpdate), stderr io.Writer) (func(), func(uint64)) { +type persistedProgressUpdate struct { + FileID uint64 + AckBytes int64 +} + +func startProgressWriter(progressPath string, initial map[uint64]ManifestProgress, updates <-chan DownloadProgressUpdate, onUpdate func(DownloadProgressUpdate), stderr io.Writer) (func(), func(uint64, int64), func(uint64)) { state := initial if state == nil { state = make(map[uint64]ManifestProgress) } stopCh := make(chan struct{}) doneCh := make(chan struct{}) + persistedProgressCh := make(chan persistedProgressUpdate, 1024) metadataDoneCh := make(chan metadataProgressUpdate, 1024) writeSnapshot := func() error { @@ -2717,6 +2749,22 @@ func startProgressWriter(progressPath string, initial map[uint64]ManifestProgres ticker := time.NewTicker(1 * time.Second) defer ticker.Stop() dirty := false + hasPersistedState := func() bool { + return len(state) > 0 + } + flushSnapshot := func(force bool) { + if !force && !dirty { + return + } + if !hasPersistedState() { + return + } + if err := writeSnapshot(); err != nil { + fmt.Fprintf(stderr, "progress flush failed: %v\n", err) + return + } + dirty = false + } applyProgress := func(update DownloadProgressUpdate) { if onUpdate != nil { onUpdate(update) @@ -2728,6 +2776,14 @@ func startProgressWriter(progressPath string, initial map[uint64]ManifestProgres dirty = true } } + applyPersistedProgress := func(update persistedProgressUpdate) { + prev := state[update.FileID] + if update.AckBytes > prev.AckBytes { + prev.AckBytes = update.AckBytes + state[update.FileID] = prev + dirty = true + } + } applyMetadataDone := func(update metadataProgressUpdate) { prev := state[update.FileID] if !prev.MetadataDone { @@ -2745,6 +2801,8 @@ func startProgressWriter(progressPath string, initial map[uint64]ManifestProgres continue } applyProgress(update) + case update := <-persistedProgressCh: + applyPersistedProgress(update) case update := <-metadataDoneCh: applyMetadataDone(update) default: @@ -2756,32 +2814,20 @@ func startProgressWriter(progressPath string, initial map[uint64]ManifestProgres select { case <-stopCh: drainPending() - if dirty { - if err := writeSnapshot(); err != nil { - fmt.Fprintf(stderr, "progress flush failed: %v\n", err) - } - } + flushSnapshot(true) return case update, ok := <-updates: if !ok { - if dirty { - if err := writeSnapshot(); err != nil { - fmt.Fprintf(stderr, "progress flush failed: %v\n", err) - } - } + flushSnapshot(hasPersistedState()) return } applyProgress(update) + case update := <-persistedProgressCh: + applyPersistedProgress(update) case update := <-metadataDoneCh: applyMetadataDone(update) case <-ticker.C: - if dirty { - if err := writeSnapshot(); err != nil { - fmt.Fprintf(stderr, "progress flush failed: %v\n", err) - } else { - dirty = false - } - } + flushSnapshot(false) } } }() @@ -2790,16 +2836,23 @@ func startProgressWriter(progressPath string, initial map[uint64]ManifestProgres close(stopCh) <-doneCh } + persistProgressAck := func(fileID uint64, ackBytes int64) { + update := persistedProgressUpdate{FileID: fileID, AckBytes: ackBytes} + select { + case <-doneCh: + return + case persistedProgressCh <- update: + } + } markMetadataDone := func(fileID uint64) { + update := metadataProgressUpdate{FileID: fileID} select { case <-doneCh: return - case metadataDoneCh <- metadataProgressUpdate{FileID: fileID}: - default: - // Do not block download workers on progress persistence. + case metadataDoneCh <- update: } } - return stop, markMetadataDone + return stop, persistProgressAck, markMetadataDone } func refreshCompletedFileMetadata(ctx context.Context, client *Client, manifest *Manifest, fileID uint64, outRoot string, outFile string) error { diff --git a/server/internal/filexfer/ftcp/send_test.go b/server/internal/filexfer/ftcp/send_test.go index 7b1bbf1..e873d17 100644 --- a/server/internal/filexfer/ftcp/send_test.go +++ b/server/internal/filexfer/ftcp/send_test.go @@ -41,7 +41,7 @@ func (d *sendTestDeps) RegisterTransferFileState(string, <-chan TransferFileStat func (d *sendTestDeps) ClipTransfer(string) bool { return false } func (d *sendTestDeps) GetTransfer(string) (Transfer, bool) { return Transfer{}, false } -func (d *sendTestDeps) ListTransfers() []Transfer { return nil } +func (d *sendTestDeps) ListTransfers() []Transfer { return nil } func (d *sendTestDeps) SetTransferHints(string, string, int64, int) bool { return true } @@ -231,11 +231,9 @@ func TestStreamSendItemRoundTripCompressionModes(t *testing.T) { } func TestStreamSendItemAdaptiveUpgradesFromNone(t *testing.T) { - size := (2 * defaultFileFrameLogicalSize) + 1 - data := make([]byte, size) - for i := range data { - data[i] = byte(i) - } + size := (5 * defaultFileFrameLogicalSize) + 1 + data := bytes.Repeat([]byte("compress-me-"), int(size/int64(len("compress-me-")))+1) + data = data[:size] tmp := writeTempSendFile(t, data) deps := &sendTestDeps{filePath: tmp} @@ -250,8 +248,8 @@ func TestStreamSendItemAdaptiveUpgradesFromNone(t *testing.T) { if err != nil { t.Fatalf("frameComps failed: %v", err) } - if len(comps) < 3 { - t.Fatalf("expected >=3 frames for adaptive test, got %d", len(comps)) + if len(comps) < 5 { + t.Fatalf("expected >=5 frames for adaptive test, got %d", len(comps)) } if comps[0] != "none" { t.Fatalf("expected first frame to start at none, got %q", comps[0]) From dfb529e84963a1e13cef55dc40dea8fa546e11b4 Mon Sep 17 00:00:00 2001 From: Joseph Lynch Date: Sat, 4 Apr 2026 09:57:55 -0400 Subject: [PATCH 2/4] Migrate from age and our own aes to our own aead Since our chunking format is self describing we don't need half close hacks like age does where it reads beyond the stream. Now the client no longer needs to close the connection and if we wanted to pipeline commands from the client in the future we could. --- .github/workflows/ci.yml | 1 + server/README.md | 2 +- server/bench.sh | 15 +- server/filexfer/client.go | 59 ++----- server/filexfer/client_tcp.go | 149 +++++++----------- server/filexfer/client_test.go | 40 +---- server/filexfer/docs/CLI.md | 2 +- server/filexfer/docs/FRAMING.md | 23 +-- server/filexfer/docs/PROTOCOL.md | 27 ++-- .../{filexfer/encoding => aead}/aead.go | 102 ++++++++---- .../{filexfer/encoding => aead}/aead_test.go | 43 ++++- server/internal/cmd/filexfercli/cli.go | 35 +--- server/internal/cmd/filexfercli/cli_test.go | 65 +++----- server/internal/filexfer/encoding/frame.go | 14 +- .../internal/filexfer/encoding/frame_test.go | 1 - server/internal/filexfer/ftcp/auth.go | 59 +++---- server/internal/filexfer/ftcp/auth_test.go | 80 ++++++++++ server/internal/filexfer/ftcp/cxsum.go | 1 - server/internal/filexfer/ftcp/request.go | 4 +- server/internal/filexfer/ftcp/request_test.go | 25 +++ server/internal/filexfer/ftcp/send.go | 4 +- server/internal/filexfer/ftcp/server.go | 45 ++---- server/main.go | 83 +++++++--- 23 files changed, 462 insertions(+), 417 deletions(-) rename server/internal/{filexfer/encoding => aead}/aead.go (86%) rename server/internal/{filexfer/encoding => aead}/aead_test.go (91%) create mode 100644 server/internal/filexfer/ftcp/auth_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1addb58..7232234 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,5 +15,6 @@ jobs: uses: actions/setup-go@v5 with: go-version: '1.26' + cache-dependency-path: "**/go.sum" - name: Build and test run: make acceptance diff --git a/server/README.md b/server/README.md index 729f751..224c6d9 100644 --- a/server/README.md +++ b/server/README.md @@ -72,7 +72,7 @@ curl localhost:8080/status/${FD} | jq . Compression and Encryption ========================== -Pinch can also encrypt and decrypt data using `age`. +Pinch can also encrypt and decrypt data using AEAD (AES-GCM or ChaCha20-Poly1305, auto-detected). ```bash $ curl -s 'localhost:8080/pinch?timeout=100s&age-public-key=age1630vztsaydze8r9qc3e865spc989mvcls3wg7hh9vu2w3luulqlqp0v6wh' | jq . diff --git a/server/bench.sh b/server/bench.sh index 464ddce..e9ea5b1 100755 --- a/server/bench.sh +++ b/server/bench.sh @@ -21,8 +21,8 @@ Options: --target-dir PATH Target directory for copy (default: /var/lib/pinch/data). --skip-fsync Skip fdatasync after each file/window (passed to copy). --no-sync Alias for --skip-fsync. - --concurrency N Copy command concurrency (default: 128). - --encrypt MODE Client encryption mode (supported: age|aes). + --concurrency N Copy command concurrency (default: adaptive). + --encrypt MODE Client encryption mode (supported: auto|aes|chacha20). --compress MODE Compression mode passed to copy (adapt|none|lz4|zstd). --disable-zero-copy Force server to use buffered send path (no tee/splice). --freq HZ perf sample frequency (default: 199). @@ -63,7 +63,7 @@ SERVER_URL="127.0.0.1:3453" SERVER_STARTUP_TIMEOUT_SEC=30 SOURCE_DIRECTORY="" TARGET_DIR="/var/lib/pinch/data" -CONCURRENCY="48" +CONCURRENCY="" ENCRYPT_MODE="" COMPRESS_MODE="" DISABLE_ZERO_COPY=false @@ -209,8 +209,8 @@ require_cmd go require_cmd rm require_cmd time -if [[ -n "${ENCRYPT_MODE}" && "${ENCRYPT_MODE}" != "age" && "${ENCRYPT_MODE}" != "aes" ]]; then - echo "unsupported --encrypt value: ${ENCRYPT_MODE} (supported: age, aes)" >&2 +if [[ -n "${ENCRYPT_MODE}" && "${ENCRYPT_MODE}" != "auto" && "${ENCRYPT_MODE}" != "aes" && "${ENCRYPT_MODE}" != "chacha20" ]]; then + echo "unsupported --encrypt value: ${ENCRYPT_MODE} (supported: auto, aes, chacha20)" >&2 exit 2 fi if [[ "${RSYNC}" == "true" && "${SKIP_WRITE}" == "true" ]]; then @@ -345,7 +345,10 @@ echo "bench: source=${SOURCE_DIRECTORY} target=${TARGET_DIR}" echo "Cleaning prior benchmark output..." rm -rf "${TARGET_DIR}/" "${STATE_DIR}/" -COPY_CMD=(./pinch filecli "${SERVER_URL}" copy --concurrency "${CONCURRENCY}") +COPY_CMD=(./pinch filecli "${SERVER_URL}" copy) +if [[ -n "${CONCURRENCY}" ]]; then + COPY_CMD+=(--concurrency "${CONCURRENCY}") +fi if [[ -n "${ENCRYPT_MODE}" ]]; then COPY_CMD+=(--encrypt "${ENCRYPT_MODE}") fi diff --git a/server/filexfer/client.go b/server/filexfer/client.go index 6d1484a..1751dca 100644 --- a/server/filexfer/client.go +++ b/server/filexfer/client.go @@ -166,7 +166,7 @@ type Client struct { Comp string // adapt|none|lz4|zstd; empty means server default (adapt) ClientAgePublicKey string ClientAgeIdentity string - EncryptMode string // "age" (default) or "aes" — selects post-AUTH stream cipher + EncryptMode string // "auto", "aes", or "chacha20" — selects post-AUTH stream cipher // Context dialer allows clients to setup custom connections // For example injecting TLS @@ -223,7 +223,6 @@ type FileFrameMeta struct { FileID uint64 Comp string CompCounts map[string]uint64 - Enc string Offset int64 Size int64 WireSize int64 @@ -316,6 +315,7 @@ type ProbeResponse struct { LinkMbps int64 SuggestedConcurrency int ServerSendBufBytes int64 + SuggestedCipher string // resolved cipher suggested for this connection (e.g. "aes", "chacha20", or "" if none) } type GetManifestResponse struct { @@ -403,7 +403,7 @@ var ErrFileMissing = errors.New("file missing") const ( // Window and batch sizes. The window is max in-flight bytes per file; the batch is // the unit of parallel work. parallelism = window / batch. - defaultClientRequestWindowBytes int64 = 1024 * 1024 * 1024 + defaultClientRequestWindowBytes int64 = 512 * 1024 * 1024 defaultClientMaxFrameReadBufferBytes int = 64 * 1024 * 1024 defaultClientBatchMaxBytes int64 = 64 * 1024 * 1024 @@ -850,7 +850,6 @@ func (c *Client) downloadManifestGroupSequential( meta := FileFrameMeta{ FileID: plan.entry.ID, Comp: "none", - Enc: "none", Offset: plan.resumeFrom, } offset := plan.resumeFrom @@ -888,7 +887,7 @@ func (c *Client) downloadManifestGroupSequential( } payloadReader := io.LimitReader(br, frameMeta.WireSize) - logicalReader, decodeErr := decodePayloadReader(payloadReader, frameMeta.Comp, frameMeta.Enc, nil) + logicalReader, decodeErr := decodePayloadReader(payloadReader, frameMeta.Comp) if decodeErr != nil { _ = closeWriter() return nil, nil, nil, fmt.Errorf("decode payload reader: %w", decodeErr) @@ -916,7 +915,6 @@ func (c *Client) downloadManifestGroupSequential( meta.Size += frameMeta.Size meta.WireSize += frameMeta.WireSize meta.Comp = frameMeta.Comp - meta.Enc = frameMeta.Enc offset += frameMeta.Size trailerLine, trailerReadErr := br.ReadString('\n') @@ -1487,7 +1485,6 @@ func aggregateSplitWindowResults( Meta: FileFrameMeta{ FileID: plan.entry.ID, Comp: "none", - Enc: "none", Offset: plan.resumeFrom, }, } @@ -1506,7 +1503,6 @@ func aggregateSplitWindowResults( aggregate.Meta.WireSize += meta.WireSize if idx == 0 { aggregate.Meta.Comp = meta.Comp - aggregate.Meta.Enc = meta.Enc } if len(meta.CompCounts) > 0 { if aggregate.Meta.CompCounts == nil { @@ -1804,6 +1800,12 @@ func (c *Client) ProbeLink(ctx context.Context, req ProbeRequest) (ProbeResponse } response := summarizeProbeSamples(probeResults, probeBytes) response.SuggestedConcurrency = clampConcurrency(suggestedConcurrencyFromProbe(response.ServerCPU, response.ServerIODepth, loadStrategy)) + + // Resolve the cipher name for display. If encryption is enabled, + // resolveTCPAuthState resolves "auto" to the server's recommendation. + if authState, authErr := c.resolveTCPAuthState(ctx); authErr == nil && authState.hasAuth { + response.SuggestedCipher = authState.encMode + } return response, nil } @@ -2549,10 +2551,9 @@ func (s *fileStream) openNextFrame() error { if s.expectOffset && meta.Offset != s.expectedOffset { return fmt.Errorf("non-contiguous frame offset: expected=%d got=%d", s.expectedOffset, meta.Offset) } - if s.meta.FileID == 0 && s.meta.Comp == "" && s.meta.Enc == "" && s.meta.Size == 0 && s.meta.WireSize == 0 { + if s.meta.FileID == 0 && s.meta.Comp == "" && s.meta.Size == 0 && s.meta.WireSize == 0 { s.meta.FileID = meta.FileID s.meta.Comp = meta.Comp - s.meta.Enc = meta.Enc s.meta.Offset = meta.Offset s.meta.MaxWireSizeHint = meta.MaxWireSizeHint s.meta.HeaderTS = meta.HeaderTS @@ -2560,13 +2561,10 @@ func (s *fileStream) openNextFrame() error { if meta.FileID != s.meta.FileID { return fmt.Errorf("file id mismatch across frames: expected=%d got=%d", s.meta.FileID, meta.FileID) } - if meta.Enc != s.meta.Enc { - return fmt.Errorf("encryption mode mismatch across frames: expected=%s got=%s", s.meta.Enc, meta.Enc) - } } payloadReader := io.LimitReader(s.br, meta.WireSize) - logicalReader, err := decodePayloadReader(payloadReader, meta.Comp, meta.Enc, s.identity) + logicalReader, err := decodePayloadReader(payloadReader, meta.Comp) if err != nil { return fmt.Errorf("decode payload reader: %w", err) } @@ -2650,7 +2648,6 @@ func parseFXHeader(line string) (FileFrameMeta, error) { } comp := props["comp"] - enc := props["enc"] offset, err := parseHeaderInt(props["offset"], "offset") if err != nil { return FileFrameMeta{}, err @@ -2680,13 +2677,12 @@ func parseFXHeader(line string) (FileFrameMeta, error) { if ts < 0 { return FileFrameMeta{}, errors.New("invalid header ts") } - if comp == "" || enc == "" { + if comp == "" { return FileFrameMeta{}, errors.New("missing required frame properties") } return FileFrameMeta{ FileID: fileID, Comp: comp, - Enc: enc, Offset: offset, Size: size, WireSize: wsize, @@ -2859,34 +2855,7 @@ func validHashToken(raw string) bool { return len(parts) == 2 && parts[0] != "" && parts[1] != "" } -func decodePayloadReader(payload io.Reader, comp string, enc string, identity age.Identity) (io.ReadCloser, error) { - switch enc { - case "none": - return decodePayloadReaderByComp(payload, comp) - case "age": - if identity == nil { - return nil, errors.New("missing age identity for encrypted frame") - } - decrypted, err := age.Decrypt(payload, identity) - if err != nil { - return nil, err - } - return decodePayloadReaderByComp(decrypted, comp) - case "aes": - if identity == nil { - return nil, errors.New("missing identity for AES encrypted frame") - } - decrypted, err := intencoding.Decrypt(payload, identity) - if err != nil { - return nil, err - } - return decodePayloadReaderByComp(decrypted, comp) - default: - return nil, fmt.Errorf("unsupported encryption mode: %s", enc) - } -} - -func decodePayloadReaderByComp(payload io.Reader, comp string) (io.ReadCloser, error) { +func decodePayloadReader(payload io.Reader, comp string) (io.ReadCloser, error) { switch comp { case "none": return io.NopCloser(payload), nil diff --git a/server/filexfer/client_tcp.go b/server/filexfer/client_tcp.go index 05808e9..076981c 100644 --- a/server/filexfer/client_tcp.go +++ b/server/filexfer/client_tcp.go @@ -16,7 +16,7 @@ import ( "time" "filippo.io/age" - intencoding "github.com/jolynch/pinch/internal/filexfer/encoding" + "github.com/jolynch/pinch/internal/aead" ftcp "github.com/jolynch/pinch/internal/filexfer/ftcp" ) @@ -28,7 +28,7 @@ type tcpAuthState struct { parsedIdentity age.Identity serverKey string // server's age public key (discovered via AUTH key) hasAuth bool - encMode string // "age" or "aes" + encMode string // resolved cipher: "aes" or "chacha20" } type probeResponse struct { @@ -163,7 +163,7 @@ func (c *Client) resolveTCPAuthState(ctx context.Context) (tcpAuthState, error) if encMode == "" || encMode == "none" { return tcpAuthState{}, nil } - if encMode != "age" && encMode != "aes" { + if encMode != "auto" && encMode != "aes" && encMode != "chacha20" { return tcpAuthState{}, fmt.Errorf("unsupported encrypt mode: %s", encMode) } @@ -184,12 +184,17 @@ func (c *Client) resolveTCPAuthState(ctx context.Context) (tcpAuthState, error) return tcpAuthState{}, err } - // Discover the server's public key via AUTH key. - serverKey, err := c.discoverServerKey(ctx) + // Discover the server's public key and recommended cipher via AUTH key. + recommendedCipher, serverKey, err := c.discoverServerKey(ctx) if err != nil { return tcpAuthState{}, fmt.Errorf("discover server key: %w", err) } + // Resolve "auto" to the server's recommended cipher. + if encMode == "auto" { + encMode = recommendedCipher + } + return tcpAuthState{ publicKey: requestPub, identity: requestIdentity, @@ -201,36 +206,42 @@ func (c *Client) resolveTCPAuthState(ctx context.Context) (tcpAuthState, error) } // discoverServerKey sends AUTH key on a fresh connection and reads back the -// server's age public key from the OK response. -func (c *Client) discoverServerKey(ctx context.Context) (string, error) { - conn, err := c.dialTCP(ctx) - if err != nil { - return "", fmt.Errorf("dial for key exchange: %w", err) +// server's recommended cipher and public key from the OK response. +// Response format: "OK \r\n" +func (c *Client) discoverServerKey(ctx context.Context) (recommendedCipher string, serverKey string, err error) { + conn, dialErr := c.dialTCP(ctx) + if dialErr != nil { + return "", "", fmt.Errorf("dial for key exchange: %w", dialErr) } defer conn.Close() if err := writeTCPLine(conn, "AUTH key"); err != nil { - return "", err + return "", "", err } br := bufio.NewReader(conn) line, err := readTCPLine(br, maxTCPLineBytes) if err != nil { - return "", fmt.Errorf("read key exchange response: %w", err) + return "", "", fmt.Errorf("read key exchange response: %w", err) } msg, ok := parseOKStatusLine(line) if !ok { if lineErr := parseErrControlFrame(line); lineErr != nil { - return "", lineErr + return "", "", lineErr } - return "", fmt.Errorf("unexpected key exchange response: %s", line) + return "", "", fmt.Errorf("unexpected key exchange response: %s", line) + } + parts := strings.Fields(msg) + if len(parts) != 2 { + return "", "", fmt.Errorf("malformed AUTH key response: expected ' ', got %q", msg) } - key := strings.TrimSpace(msg) - if key == "" { - return "", errors.New("server returned empty public key") + cipher := parts[0] + key := parts[1] + if cipher != "aes" && cipher != "chacha20" { + return "", "", fmt.Errorf("unsupported server cipher: %s", cipher) } if _, err := age.ParseX25519Recipient(key); err != nil { - return "", fmt.Errorf("invalid server public key: %w", err) + return "", "", fmt.Errorf("invalid server public key: %w", err) } - return key, nil + return cipher, key, nil } func (c *Client) sendTCPAuth(conn net.Conn, state tcpAuthState) error { @@ -241,48 +252,31 @@ func (c *Client) sendTCPAuth(conn net.Conn, state tcpAuthState) error { if err != nil { return fmt.Errorf("parse server key: %w", err) } - // Encrypt the client's public key to the server. + // Encrypt the client's public key to the server using AEAD. encrypted := c.acquireScratchBuffer(0) defer c.releaseScratchBuffer(encrypted) - switch state.encMode { - case "aes": - ew, encErr := intencoding.Encrypt(encrypted, recipient, intencoding.Options{Algorithm: intencoding.AlgorithmAES}) - if encErr != nil { - return encErr - } - if _, err := ew.Write([]byte(state.publicKey)); err != nil { - return err - } - if err := ew.Close(); err != nil { - return err - } - default: - ew, encErr := age.Encrypt(encrypted, recipient) - if encErr != nil { - return encErr - } - if _, err := ew.Write([]byte(state.publicKey)); err != nil { - return err - } - if err := ew.Close(); err != nil { - return err - } + ew, encErr := aead.Encrypt(encrypted, recipient, aeadOptionsForMode(state.encMode)) + if encErr != nil { + return encErr + } + if _, err := ew.Write([]byte(state.publicKey)); err != nil { + return err + } + if err := ew.Close(); err != nil { + return err } encoded := base64.StdEncoding.EncodeToString(encrypted.Bytes()) return writeTCPLine(conn, "AUTH "+state.encMode+" "+encoded) } -// halfCloseWrite sends TCP FIN for the write direction only, signaling EOF to -// the remote reader while keeping the connection open for reading responses. -// This is required for age-encrypted requests: age's DecryptReader performs an -// extra Read after the final chunk to confirm EOF, which would deadlock on a -// duplex connection that stays open. -func halfCloseWrite(conn net.Conn) { - type writeHalfCloser interface { - CloseWrite() error - } - if wc, ok := conn.(writeHalfCloser); ok { - _ = wc.CloseWrite() +func aeadOptionsForMode(mode string) aead.Options { + switch mode { + case "aes": + return aead.Options{Algorithm: aead.AlgorithmAES} + case "chacha20": + return aead.Options{Algorithm: aead.AlgorithmChaCha20} + default: + return aead.Options{} // RecommendedCipher } } @@ -294,13 +288,7 @@ func (c *Client) sendTCPCommand(conn net.Conn, state tcpAuthState, payload strin if err != nil { return err } - var ew io.WriteCloser - switch state.encMode { - case "aes": - ew, err = intencoding.Encrypt(conn, recipient, intencoding.Options{Algorithm: intencoding.AlgorithmAES}) - default: - ew, err = age.Encrypt(conn, recipient) - } + ew, err := aead.Encrypt(conn, recipient, aeadOptionsForMode(state.encMode)) if err != nil { return err } @@ -308,11 +296,7 @@ func (c *Client) sendTCPCommand(conn net.Conn, state tcpAuthState, payload strin _ = ew.Close() return err } - if err := ew.Close(); err != nil { - return err - } - halfCloseWrite(conn) - return nil + return ew.Close() } func (c *Client) responseReaderForTCP(conn net.Conn, state tcpAuthState) (io.Reader, error) { @@ -321,22 +305,13 @@ func (c *Client) responseReaderForTCP(conn net.Conn, state tcpAuthState) (io.Rea } identity := state.parsedIdentity if identity == nil { - return nil, errors.New("missing age identity for encrypted response") + return nil, errors.New("missing identity for encrypted response") } - switch state.encMode { - case "aes": - decReader, err := intencoding.Decrypt(conn, identity) - if err != nil { - return nil, fmt.Errorf("aes decryption failed: %w", err) - } - return decReader, nil - default: - decReader, err := age.Decrypt(conn, identity) - if err != nil { - return nil, err - } - return decReader, nil + decReader, err := aead.Decrypt(conn, identity) + if err != nil { + return nil, fmt.Errorf("decryption failed: %w", err) } + return decReader, nil } func (c *Client) getManifestTCP(ctx context.Context, request GetManifestRequest) (GetManifestResponse, error) { @@ -752,13 +727,7 @@ func (c *Client) sendTCPProbe(conn net.Conn, state tcpAuthState, cmd string, pro if err != nil { return err } - var ew io.WriteCloser - switch state.encMode { - case "aes": - ew, err = intencoding.Encrypt(conn, recipient, intencoding.Options{Algorithm: intencoding.AlgorithmAES}) - default: - ew, err = age.Encrypt(conn, recipient) - } + ew, err := aead.Encrypt(conn, recipient, aeadOptionsForMode(state.encMode)) if err != nil { return err } @@ -770,11 +739,7 @@ func (c *Client) sendTCPProbe(conn net.Conn, state tcpAuthState, cmd string, pro _ = ew.Close() return err } - if err := ew.Close(); err != nil { - return err - } - halfCloseWrite(conn) - return nil + return ew.Close() } type firstReadTimestampReader struct { diff --git a/server/filexfer/client_test.go b/server/filexfer/client_test.go index 6645ae4..4830dda 100644 --- a/server/filexfer/client_test.go +++ b/server/filexfer/client_test.go @@ -19,7 +19,6 @@ import ( "testing" "time" - "filippo.io/age" intencoding "github.com/jolynch/pinch/internal/filexfer/encoding" intftcp "github.com/jolynch/pinch/internal/filexfer/ftcp" "github.com/jolynch/pinch/utils" @@ -92,19 +91,8 @@ func serveFTCPTestConn(conn net.Conn, handler func(intftcp.Request, io.Writer) e closeResponse := func() error { return nil } cmdReq := firstReq if firstReq.Verb == intftcp.VerbAUTH { - if len(firstReq.Params) > 0 { - blob := strings.TrimSpace(firstReq.Params[0]["blob"]) - if blob != "" { - if recipient, parseErr := age.ParseX25519Recipient(blob); parseErr == nil { - ew, encErr := age.Encrypt(conn, recipient) - if encErr != nil { - return - } - responseOut = ew - closeResponse = ew.Close - } - } - } + // The test server doesn't exercise encryption, so just skip the + // AUTH blob and read the next command line. cmdLine, cmdErr := readCompatLine(br) if cmdErr != nil { _, _ = io.WriteString(responseOut, "ERR BAD_REQUEST missing command\r\n") @@ -283,7 +271,7 @@ func buildFXFrameWithTrailerTokens(t *testing.T, fileID uint64, comp string, off trailerTS := headerTS + 1 xsum := xxh128HexTest(logical) header := fmt.Sprintf( - "FX/1 %d offset=%d size=%d wsize=%d comp=%s enc=none hash=xxh128:%s ts=%d\n", + "FX/1 %d offset=%d size=%d wsize=%d comp=%s hash=xxh128:%s ts=%d\n", fileID, offset, len(logical), @@ -431,24 +419,8 @@ func downloadSingle(ctx context.Context, client *Client, req singleDownloadReque return resp.Files[0], nil } -func encryptAgeBlob(t *testing.T, plaintext []byte, recipient age.Recipient) []byte { - t.Helper() - var buf bytes.Buffer - w, err := age.Encrypt(&buf, recipient) - if err != nil { - t.Fatalf("age encrypt setup failed: %v", err) - } - if _, err := w.Write(plaintext); err != nil { - t.Fatalf("age encrypt write failed: %v", err) - } - if err := w.Close(); err != nil { - t.Fatalf("age encrypt close failed: %v", err) - } - return buf.Bytes() -} - func TestParseFXHeaderMaxWSizeHint(t *testing.T) { - meta, err := parseFXHeader("FX/1 7 offset=0 size=5 wsize=5 comp=none enc=none hash=xxh128:abc max-wsize=16777216 ts=1000") + meta, err := parseFXHeader("FX/1 7 offset=0 size=5 wsize=5 comp=none hash=xxh128:abc max-wsize=16777216 ts=1000") if err != nil { t.Fatalf("parseFXHeader failed: %v", err) } @@ -458,10 +430,10 @@ func TestParseFXHeaderMaxWSizeHint(t *testing.T) { } func TestParseFXHeaderInvalidMaxWSizeHint(t *testing.T) { - if _, err := parseFXHeader("FX/1 7 offset=0 size=5 wsize=5 comp=none enc=none hash=xxh128:abc max-wsize=-1 ts=1000"); err == nil { + if _, err := parseFXHeader("FX/1 7 offset=0 size=5 wsize=5 comp=none hash=xxh128:abc max-wsize=-1 ts=1000"); err == nil { t.Fatalf("expected parse error for negative max-wsize") } - if _, err := parseFXHeader("FX/1 7 offset=0 size=5 wsize=5 comp=none enc=none hash=xxh128:abc max-wsize=nope ts=1000"); err == nil { + if _, err := parseFXHeader("FX/1 7 offset=0 size=5 wsize=5 comp=none hash=xxh128:abc max-wsize=nope ts=1000"); err == nil { t.Fatalf("expected parse error for malformed max-wsize") } } diff --git a/server/filexfer/docs/CLI.md b/server/filexfer/docs/CLI.md index 4faea64..9b814e5 100644 --- a/server/filexfer/docs/CLI.md +++ b/server/filexfer/docs/CLI.md @@ -148,7 +148,7 @@ pinch filecli get --skip-write /srv/data/file.bin # fetch without writing Important flags: - `-o `: output file path, or `-` for stdout -- `--encrypt none|age`: encryption algorithm (default: none) +- `--encrypt none|auto|aes|chacha20`: encryption algorithm (default: none) - `--compress adapt|none|lz4|zstd`: compression algorithm (default: adapt) - `--concurrency N`: parallel download workers (0=auto) - `--skip-write`: fetch to discard without writing diff --git a/server/filexfer/docs/FRAMING.md b/server/filexfer/docs/FRAMING.md index fdde1e0..ef6651e 100644 --- a/server/filexfer/docs/FRAMING.md +++ b/server/filexfer/docs/FRAMING.md @@ -34,7 +34,7 @@ FX/1 Example: ```text -FX/1 12 offset=0 size=1048576 wsize=262144 comp=zstd enc=age hash=xxh128:9f12ab... deadline:30s +FX/1 12 offset=0 size=1048576 wsize=262144 comp=zstd hash=xxh128:9f12ab... deadline:30s <262144 payload bytes> FXT/1 12 status=ok hash=xxh64:1af3bc9d00ee42aa ``` @@ -46,7 +46,6 @@ Properties are ASCII and case-sensitive. ### Required - `comp=`: compression mode. -- `enc=`: encryption mode. - `offset=`: byte offset within the logical file where payload data belongs. - `size=`: number of original (logical, uncompressed) bytes represented by this frame. - `wsize=`: number of payload bytes on the wire for this frame. @@ -76,23 +75,11 @@ Receiver behavior: - `none`: write bytes directly. - `zstd` or `lz4`: decompress before writing to destination offset. -## Encryption - -`enc` allowed values: - -- `none` -- `age` - -Receiver behavior: - -- `none`: interpret payload as plaintext/compressed bytes per `comp`. -- `age`: decrypt payload before decompression and write. - ## Parsing Rules - Maximum header bytes: 16 KiB (defensive limit). - Unknown properties are ignored. -- Missing required fields (`comp`, `enc`, `offset`, `size`, `wsize`) reject frame. +- Missing required fields (`comp`, `offset`, `size`, `wsize`) reject frame. - Invalid `file_id` reject frame. - Invalid numeric value formats reject frame. - Invalid `deadline` duration format rejects frame when `deadline:` is present. @@ -105,7 +92,6 @@ Receiver behavior: - `size` is the logical uncompressed bytes covered by this frame. - `wsize` is the exact payload byte count that follows the header newline. - For `comp=none`, `size` must equal `wsize`. -- For `enc=age`, decrypt before applying `comp`. - For `comp=zstd|lz4`, decompressed bytes must equal `size`. - Trailer `hash=:` is used for frame-integrity validation/logging. - `deadline:` limits how long receiver should allow this frame to complete; exceeded deadline is a protocol timeout. @@ -152,7 +138,7 @@ FXR/1 Example: ```text -FXR/1 12 status=ok comp=zstd enc=age offset=0 size=1048576 wsize=262144 elapsed=420ms +FXR/1 12 status=ok comp=zstd offset=0 size=1048576 wsize=262144 elapsed=420ms <262144 payload bytes> FXT/1 12 status=ok hash=xxh64:1af3bc9d00ee42aa ``` @@ -167,7 +153,6 @@ FXT/1 12 status=ok hash=xxh64:1af3bc9d00ee42aa ### Optional Response Properties - `comp=`: compression mode actually applied. -- `enc=`: encryption mode actually applied. - `deadline:`: effective deadline used for this segment. - `elapsed=`: observed processing duration. - `detail=`: machine-readable short detail code. @@ -229,7 +214,7 @@ The server repeats these triplets until each requested window is complete. Default logical frame size cap is `8 MiB`. For `SEND` responses, header properties are emitted in this order: -`offset`, `size`, `wsize`, `comp`, `enc`, `hash`, optional `max-wsize`, then `ts`. +`offset`, `size`, `wsize`, `comp`, `hash`, optional `max-wsize`, then `ts`. Current implementation supports adaptive compression and may vary `comp` per frame. Current trailer shape for `SEND`: diff --git a/server/filexfer/docs/PROTOCOL.md b/server/filexfer/docs/PROTOCOL.md index b302732..9b854ca 100644 --- a/server/filexfer/docs/PROTOCOL.md +++ b/server/filexfer/docs/PROTOCOL.md @@ -54,26 +54,35 @@ For this line protocol, token bytes cannot span command newlines. Three forms: -- `AUTH key` — key exchange: server returns its age public key. -- `AUTH age ` — age encryption: blob is the client's age public key encrypted with age to the server's public key. -- `AUTH aes ` — AES-GCM encryption: blob is the client's age public key encrypted with AES-GCM to the server's public key. +- `AUTH key` — key exchange: server returns its recommended cipher and age public key. +- `AUTH aes ` — encrypted session setup using AES-GCM. +- `AUTH chacha20 ` — encrypted session setup using ChaCha20-Poly1305. `` is length-prefixed (`:`). ### Key Exchange Flow 1. Client sends `AUTH key\r\n`. -2. Server responds `OK \r\n` and closes the connection. -3. Client opens a new connection and sends `AUTH age \r\n` or `AUTH aes \r\n`. +2. Server responds `OK \r\n` and closes the connection. +3. Client opens a new connection and sends either `AUTH aes \r\n` or `AUTH chacha20 \r\n`. + +If the client requested `auto`, it first resolves `auto` to the server's +recommended cipher from step 2 and then uses that resolved value in the second +connection. ### Encrypted Connection Flow -After `AUTH age ` or `AUTH aes `: +After `AUTH aes ` or `AUTH chacha20 `: -- Server decrypts blob using its identity to recover the client's age public key. +- Server treats the first AUTH token (`aes` or `chacha20`) as the session + cipher. +- Server decrypts `` using its identity and the selected AEAD algorithm + to recover the client's age public key. - If valid: - - subsequent response bytes are encrypted to the client's public key using the selected cipher (age or AES-GCM). - - subsequent command bytes from the client must be encrypted to the server's public key using the same cipher. + - subsequent command bytes from the client must be encrypted to the server's + public key using that same AEAD algorithm. + - subsequent response bytes are encrypted to the client's public key using + that same AEAD algorithm. - If invalid: `ERR NOT_AUTHORIZED authorization failed`. ### Unencrypted Connection Flow diff --git a/server/internal/filexfer/encoding/aead.go b/server/internal/aead/aead.go similarity index 86% rename from server/internal/filexfer/encoding/aead.go rename to server/internal/aead/aead.go index 6873f90..afe0a83 100644 --- a/server/internal/filexfer/encoding/aead.go +++ b/server/internal/aead/aead.go @@ -1,4 +1,4 @@ -package encoding +package aead import ( "crypto/aes" @@ -120,13 +120,12 @@ func Encrypt(dst io.Writer, recipient age.Recipient, opts Options) (io.WriteClos } buf, chunkSize, release := acquireAEADBuffer(chunkSize) - half := len(buf) / 2 return &aeadWriter{ aead: aeadCipher, dst: dst, - plainBuf: buf[:half], - sealBuf: buf[half:], + plainBuf: buf[:chunkSize], + sealBuf: buf[chunkSize:], chunkSize: chunkSize, backingBuf: buf, releaseBuf: release, @@ -140,6 +139,16 @@ func Encrypt(dst io.Writer, recipient age.Recipient, opts Options) (io.WriteClos // // Not goroutine-safe. func Decrypt(src io.Reader, identity age.Identity) (io.Reader, error) { + return decryptWithOptions(src, identity, Options{}) +} + +// DecryptWithOptions creates a streaming AEAD decrypting reader and, when +// opts.Algorithm is set, requires the ciphertext header to match it. +func DecryptWithOptions(src io.Reader, identity age.Identity, opts Options) (io.Reader, error) { + return decryptWithOptions(src, identity, opts) +} + +func decryptWithOptions(src io.Reader, identity age.Identity, opts Options) (io.Reader, error) { if src == nil { return nil, errors.New("nil source reader") } @@ -151,6 +160,15 @@ func Decrypt(src io.Reader, identity age.Identity) (io.Reader, error) { if err != nil { return nil, fmt.Errorf("read stanza header: %w", err) } + if opts.Algorithm != "" { + expected, err := validateAlgorithm(opts.Algorithm) + if err != nil { + return nil, err + } + if algorithm != expected { + return nil, fmt.Errorf("unexpected AEAD algorithm: got %q want %q", algorithm, expected) + } + } fileKey, err := identity.Unwrap([]*age.Stanza{stanza}) if err != nil { @@ -163,13 +181,12 @@ func Decrypt(src io.Reader, identity age.Identity) (io.Reader, error) { } buf, chunkSize, release := acquireAEADBuffer(chunkSize) - half := len(buf) / 2 return &aeadReader{ aead: aeadCipher, src: src, - cipherBuf: buf[:half], - plainBuf: buf[half:], + cipherBuf: buf[:chunkSize+aeadTagSize], + plainBuf: buf[chunkSize+aeadTagSize:], chunkSize: chunkSize, backingBuf: buf, releaseBuf: release, @@ -290,7 +307,10 @@ func parseAlgorithmID(id byte) (Algorithm, error) { func acquireAEADBuffer(chunkSize int) ([]byte, int, func()) { chunkSize = Options{ChunkSize: chunkSize}.ResolveChunkSize() - bufSize := 2 * (chunkSize + aeadTagSize) + // Keep the logical chunk size exact while reserving one tag-sized margin for + // ciphertext expansion. A 64 KiB chunk therefore uses a 64 KiB plaintext + // working region plus a 64 KiB+tag ciphertext region. + bufSize := 2*chunkSize + aeadTagSize pool := aeadBufferPool(bufSize) raw := pool.Get() buf := raw.([]byte) @@ -522,6 +542,13 @@ func (w *aeadWriter) flushChunk(last bool) error { return err } sealed := w.aead.Seal(w.sealBuf[:0], w.nonce[:], w.plainBuf[:w.plainN], aad[:]) + // Write 4-byte big-endian length prefix so the reader knows exactly how + // many bytes to read for each chunk (no EOF signaling needed). + var lenBuf [4]byte + binary.BigEndian.PutUint32(lenBuf[:], uint32(len(sealed))) + if err := writeFull(w.dst, lenBuf[:]); err != nil { + return err + } if err := writeFull(w.dst, sealed); err != nil { return err } @@ -601,11 +628,26 @@ func (r *aeadReader) Read(p []byte) (int, error) { } func (r *aeadReader) readChunk() error { - sealedSize := r.chunkSize + aeadTagSize - n, err := io.ReadFull(r.src, r.cipherBuf[:sealedSize]) + // Read the 4-byte big-endian length prefix that precedes each sealed chunk. + var lenBuf [4]byte + if _, err := io.ReadFull(r.src, lenBuf[:]); err != nil { + if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { + return io.EOF + } + return fmt.Errorf("read chunk length: %w", err) + } + sealedLen := int(binary.BigEndian.Uint32(lenBuf[:])) + if sealedLen < aeadTagSize || sealedLen > r.chunkSize+aeadTagSize { + return fmt.Errorf("invalid sealed chunk length: %d", sealedLen) + } - switch { - case err == nil: + n, err := io.ReadFull(r.src, r.cipherBuf[:sealedLen]) + if err != nil { + return fmt.Errorf("read sealed chunk: %w", err) + } + + fullSize := r.chunkSize + aeadTagSize + if n == fullSize { // Full-size read. Try non-final nonce first. r.buildNonce(false) aad, aadErr := buildChunkAAD(r.chunkSize, r.chunkCount, false) @@ -633,30 +675,22 @@ func (r *aeadReader) readChunk() error { r.done = true r.chunkCount++ return nil + } - case errors.Is(err, io.ErrUnexpectedEOF) && n > 0: - // Short read — must be the final (possibly short) chunk. - r.buildNonce(true) - aad, aadErr := buildChunkAAD(r.chunkSize, r.chunkCount, true) - if aadErr != nil { - return aadErr - } - plaintext, openErr := r.aead.Open(r.plainBuf[:0], r.nonce[:], r.cipherBuf[:n], aad[:]) - if openErr != nil { - return fmt.Errorf("AEAD authentication failed on final chunk: %w", openErr) - } - r.unread = plaintext - r.done = true - r.chunkCount++ - return nil - - case errors.Is(err, io.EOF) && n == 0: - // Stream ended without a final chunk — truncated. - return io.ErrUnexpectedEOF - - default: - return err + // Short chunk — must be the final chunk. + r.buildNonce(true) + aad, aadErr := buildChunkAAD(r.chunkSize, r.chunkCount, true) + if aadErr != nil { + return aadErr } + plaintext, openErr := r.aead.Open(r.plainBuf[:0], r.nonce[:], r.cipherBuf[:n], aad[:]) + if openErr != nil { + return fmt.Errorf("AEAD authentication failed on final chunk: %w", openErr) + } + r.unread = plaintext + r.done = true + r.chunkCount++ + return nil } func (r *aeadReader) buildNonce(last bool) { diff --git a/server/internal/filexfer/encoding/aead_test.go b/server/internal/aead/aead_test.go similarity index 91% rename from server/internal/filexfer/encoding/aead_test.go rename to server/internal/aead/aead_test.go index dcc14ed..bcbdf6b 100644 --- a/server/internal/filexfer/encoding/aead_test.go +++ b/server/internal/aead/aead_test.go @@ -1,4 +1,4 @@ -package encoding +package aead import ( "bytes" @@ -213,6 +213,27 @@ func TestChunkSizeFromHeader(t *testing.T) { } } +func TestAcquireAEADBufferLayout(t *testing.T) { + buf, chunkSize, release := acquireAEADBuffer(64 * 1024) + defer release() + + if chunkSize != 64*1024 { + t.Fatalf("unexpected chunk size: got %d want %d", chunkSize, 64*1024) + } + if len(buf) != 2*chunkSize+aeadTagSize { + t.Fatalf("unexpected buffer len: got %d want %d", len(buf), 2*chunkSize+aeadTagSize) + } + + cipherBuf := buf[:chunkSize+aeadTagSize] + plainBuf := buf[chunkSize+aeadTagSize:] + if len(cipherBuf) != chunkSize+aeadTagSize { + t.Fatalf("unexpected cipher buffer len: got %d want %d", len(cipherBuf), chunkSize+aeadTagSize) + } + if len(plainBuf) != chunkSize { + t.Fatalf("unexpected plain buffer len: got %d want %d", len(plainBuf), chunkSize) + } +} + func TestHeaderChunkSizeTamper(t *testing.T) { for _, algorithm := range testAlgorithms { t.Run(algorithmName(algorithm), func(t *testing.T) { @@ -294,6 +315,26 @@ func TestReadStanzaHeaderRejectsUnsetAlgorithm(t *testing.T) { } } +func TestDecryptWithOptionsRejectsUnexpectedAlgorithm(t *testing.T) { + id, recipient := generateTestIdentity(t) + + var ciphertext bytes.Buffer + w, err := Encrypt(&ciphertext, recipient, Options{Algorithm: AlgorithmAES}) + if err != nil { + t.Fatalf("encrypt: %v", err) + } + if _, err := w.Write([]byte("tamper me")); err != nil { + t.Fatalf("write: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + if _, err := DecryptWithOptions(bytes.NewReader(ciphertext.Bytes()), id, Options{Algorithm: AlgorithmChaCha20}); err == nil { + t.Fatal("expected algorithm mismatch error") + } +} + func TestEncryptDefaultAlgorithmUsesRecommendation(t *testing.T) { id, recipient := generateTestIdentity(t) diff --git a/server/internal/cmd/filexfercli/cli.go b/server/internal/cmd/filexfercli/cli.go index dbaf0c6..c43ccb3 100644 --- a/server/internal/cmd/filexfercli/cli.go +++ b/server/internal/cmd/filexfercli/cli.go @@ -1093,8 +1093,8 @@ func runTransfer(serverURL string, cfg transferArgs, stdout io.Writer, stderr io return 1 } cipherDisplay := "none" - if probeResult.Cipher != "" { - cipherDisplay = probeResult.Cipher + if probeResult.SuggestedCipher != "" { + cipherDisplay = probeResult.SuggestedCipher } fmt.Fprintf( stdout, @@ -2478,10 +2478,6 @@ func startVerboseStatusPolling(txferID string, client *Client, stderr io.Writer) } } -func parseFileID(raw string) (uint64, error) { - return strconv.ParseUint(raw, 10, 64) -} - type noOpWriteCloser struct { io.Writer } @@ -2599,21 +2595,6 @@ func applyProgressStateToManifest(manifest *Manifest, state map[uint64]ManifestP } } -func applyProgressUpdateToManifest(manifest *Manifest, update DownloadProgressUpdate) { - if manifest == nil { - return - } - for i := range manifest.Entries { - if manifest.Entries[i].ID != update.FileID { - continue - } - if update.AckBytes > manifest.Entries[i].Progress.AckBytes { - manifest.Entries[i].Progress.AckBytes = update.AckBytes - } - return - } -} - func manifestEntriesByID(manifest *Manifest) map[uint64]ManifestEntry { if manifest == nil || len(manifest.Entries) == 0 { return nil @@ -2625,18 +2606,6 @@ func manifestEntriesByID(manifest *Manifest) map[uint64]ManifestEntry { return entries } -func markManifestEntryMetadataDone(manifest *Manifest, fileID uint64) { - if manifest == nil { - return - } - for i := range manifest.Entries { - if manifest.Entries[i].ID == fileID { - manifest.Entries[i].Progress.MetadataDone = true - return - } - } -} - func loadProgressState(progressPath string) (map[uint64]ManifestProgress, error) { state := make(map[uint64]ManifestProgress) fd, err := os.Open(progressPath) diff --git a/server/internal/cmd/filexfercli/cli_test.go b/server/internal/cmd/filexfercli/cli_test.go index 9b59e26..c5ad639 100644 --- a/server/internal/cmd/filexfercli/cli_test.go +++ b/server/internal/cmd/filexfercli/cli_test.go @@ -18,6 +18,7 @@ import ( "filippo.io/age" . "github.com/jolynch/pinch/filexfer" + "github.com/jolynch/pinch/internal/aead" "github.com/jolynch/pinch/internal/filexfer/encoding" intftcp "github.com/jolynch/pinch/internal/filexfer/ftcp" "github.com/zeebo/xxh3" @@ -92,44 +93,43 @@ func serveFTCPConn(conn net.Conn, serverID *age.X25519Identity, handler func(int protocol := req.Params[0]["protocol"] if protocol == "key" { - // Key exchange: return server public key and close. + // Key exchange: return recommended cipher and server public key. if serverID == nil { _, _ = io.WriteString(conn, "ERR NOT_AUTHORIZED no server identity\r\n") return } - _, _ = io.WriteString(conn, "OK "+serverID.Recipient().String()+"\r\n") + _, _ = io.WriteString(conn, "OK "+string(aead.RecommendedCipher())+" "+serverID.Recipient().String()+"\r\n") return } - // age or aes: decode and decrypt the blob to get the client's public key. + // aes/chacha20: decode and decrypt the blob to get the client's public key. blobRaw := req.Params[0]["blob"] if blobRaw == "" || serverID == nil { _, _ = io.WriteString(conn, "ERR NOT_AUTHORIZED\r\n") return } + opts := aead.Options{} + switch protocol { + case "aes": + opts.Algorithm = aead.AlgorithmAES + case "chacha20": + opts.Algorithm = aead.AlgorithmChaCha20 + default: + _, _ = io.WriteString(conn, "ERR NOT_AUTHORIZED\r\n") + return + } blobBytes, b64Err := base64.StdEncoding.DecodeString(strings.TrimSpace(blobRaw)) if b64Err != nil { _, _ = io.WriteString(conn, "ERR NOT_AUTHORIZED bad base64\r\n") return } - var plain []byte - switch protocol { - case "aes": - dec, decErr := encoding.Decrypt(bytes.NewReader(blobBytes), serverID) - if decErr != nil { - _, _ = io.WriteString(conn, "ERR NOT_AUTHORIZED\r\n") - return - } - plain, err = io.ReadAll(dec) - default: - dec, decErr := age.Decrypt(bytes.NewReader(blobBytes), serverID) - if decErr != nil { - _, _ = io.WriteString(conn, "ERR NOT_AUTHORIZED\r\n") - return - } - plain, err = io.ReadAll(dec) + dec, decErr := aead.DecryptWithOptions(bytes.NewReader(blobBytes), serverID, opts) + if decErr != nil { + _, _ = io.WriteString(conn, "ERR NOT_AUTHORIZED\r\n") + return } - if err != nil { + plain, readErr := io.ReadAll(dec) + if readErr != nil { _, _ = io.WriteString(conn, "ERR NOT_AUTHORIZED\r\n") return } @@ -140,14 +140,7 @@ func serveFTCPConn(conn net.Conn, serverID *age.X25519Identity, handler func(int } // Encrypt responses to client. - var ew io.WriteCloser - var encErr error - switch protocol { - case "aes": - ew, encErr = encoding.Encrypt(conn, recipient, encoding.Options{Algorithm: encoding.AlgorithmAES}) - default: - ew, encErr = age.Encrypt(conn, recipient) - } + ew, encErr := aead.Encrypt(conn, recipient, opts) if encErr != nil { return } @@ -155,14 +148,8 @@ func serveFTCPConn(conn net.Conn, serverID *age.X25519Identity, handler func(int closeOut = ew.Close // Decrypt the command from client. - var cmdReader io.Reader - switch protocol { - case "aes": - cmdReader, err = encoding.Decrypt(br, serverID) - default: - cmdReader, err = age.Decrypt(br, serverID) - } - if err != nil { + cmdReader, cmdDecErr := aead.DecryptWithOptions(br, serverID, opts) + if cmdDecErr != nil { _, _ = io.WriteString(out, "ERR NOT_AUTHORIZED request decryption failed\r\n") _ = closeOut() return @@ -237,7 +224,7 @@ func buildCLIFrame(fileID uint64, body []byte, offset int64) string { func buildCLIFrameWithMetadata(fileID uint64, body []byte, offset int64, meta *FileTrailerMetadata) string { xsum := xxh128HexCLI(body) header := fmt.Sprintf( - "FX/1 %d offset=%d size=%d wsize=%d comp=none enc=none hash=xxh128:%s ts=1000\n", + "FX/1 %d offset=%d size=%d wsize=%d comp=none hash=xxh128:%s ts=1000\n", fileID, offset, len(body), @@ -493,7 +480,7 @@ func TestRunCLIGetSkipWriteDiscardsOutput(t *testing.T) { } } -func TestRunCLITransferWithEncryptAge(t *testing.T) { +func TestRunCLITransferWithEncryptAuto(t *testing.T) { tmp := t.TempDir() targetDir := filepath.Join(tmp, "dst") manifestRaw := strings.Join([]string{ @@ -538,7 +525,7 @@ func TestRunCLITransferWithEncryptAge(t *testing.T) { var stdout bytes.Buffer var stderr bytes.Buffer - code := runTransferCLI(srv.URL, []string{"-s", "/remote", "--encrypt", "age", targetDir}, &stdout, &stderr) + code := runTransferCLI(srv.URL, []string{"-s", "/remote", "--encrypt", "auto", targetDir}, &stdout, &stderr) if code != 0 { t.Fatalf("transfer: expected 0, got %d stderr=%s", code, stderr.String()) } diff --git a/server/internal/filexfer/encoding/frame.go b/server/internal/filexfer/encoding/frame.go index b2a1b69..3bef8ec 100644 --- a/server/internal/filexfer/encoding/frame.go +++ b/server/internal/filexfer/encoding/frame.go @@ -90,7 +90,6 @@ type WriteArgs struct { Size int64 WSize int64 Comp string - Enc string HeaderHash string MaxWSizeHint *int64 HeaderTS int64 @@ -112,13 +111,13 @@ func WriteFrame(w io.Writer, args WriteArgs) (WriteStats, error) { header := "" if args.MaxWSizeHint != nil { header = fmt.Sprintf( - "FX/1 %d offset=%d size=%d wsize=%d comp=%s enc=%s hash=%s max-wsize=%d ts=%d\n", - args.FileID, args.Offset, args.Size, args.WSize, args.Comp, args.Enc, args.HeaderHash, *args.MaxWSizeHint, args.HeaderTS, + "FX/1 %d offset=%d size=%d wsize=%d comp=%s hash=%s max-wsize=%d ts=%d\n", + args.FileID, args.Offset, args.Size, args.WSize, args.Comp, args.HeaderHash, *args.MaxWSizeHint, args.HeaderTS, ) } else { header = fmt.Sprintf( - "FX/1 %d offset=%d size=%d wsize=%d comp=%s enc=%s hash=%s ts=%d\n", - args.FileID, args.Offset, args.Size, args.WSize, args.Comp, args.Enc, args.HeaderHash, args.HeaderTS, + "FX/1 %d offset=%d size=%d wsize=%d comp=%s hash=%s ts=%d\n", + args.FileID, args.Offset, args.Size, args.WSize, args.Comp, args.HeaderHash, args.HeaderTS, ) } if _, err := w.Write([]byte(header)); err != nil { @@ -177,7 +176,6 @@ func WriteFrame(w io.Writer, args WriteArgs) (WriteStats, error) { type FileFrameMeta struct { FileID uint64 Comp string - Enc string Offset int64 Size int64 WireSize int64 @@ -213,7 +211,6 @@ func ParseFXHeader(line string) (FileFrameMeta, error) { } comp := props["comp"] - enc := props["enc"] offset, err := parseHeaderInt(props["offset"], "offset") if err != nil { return FileFrameMeta{}, err @@ -243,13 +240,12 @@ func ParseFXHeader(line string) (FileFrameMeta, error) { if ts < 0 { return FileFrameMeta{}, errors.New("invalid header ts") } - if comp == "" || enc == "" { + if comp == "" { return FileFrameMeta{}, errors.New("missing required frame properties") } return FileFrameMeta{ FileID: fileID, Comp: comp, - Enc: enc, Offset: offset, Size: size, WireSize: wsize, diff --git a/server/internal/filexfer/encoding/frame_test.go b/server/internal/filexfer/encoding/frame_test.go index 8308e16..62ce72a 100644 --- a/server/internal/filexfer/encoding/frame_test.go +++ b/server/internal/filexfer/encoding/frame_test.go @@ -16,7 +16,6 @@ func TestWriteFrameReturnsStats(t *testing.T) { Size: int64(len(payload)), WSize: int64(len(payload)), Comp: "none", - Enc: "none", HeaderHash: "xxh128:00", HeaderTS: time.Now().UnixMilli(), Payload: payload, diff --git a/server/internal/filexfer/ftcp/auth.go b/server/internal/filexfer/ftcp/auth.go index 4eede40..97a9228 100644 --- a/server/internal/filexfer/ftcp/auth.go +++ b/server/internal/filexfer/ftcp/auth.go @@ -9,7 +9,7 @@ import ( "strings" "filippo.io/age" - "github.com/jolynch/pinch/internal/filexfer/encoding" + "github.com/jolynch/pinch/internal/aead" ) var errNotAuthorized = errors.New("not authorized") @@ -17,8 +17,8 @@ var errNotAuthorized = errors.New("not authorized") type authResult struct { recipient age.Recipient encryptedRequests bool - encryptMode string // "age" or "aes" - keyExchange bool // true for AUTH key — server should return its public key + keyExchange bool // true for AUTH key — server should return its public key + responseCipher aead.Algorithm } func processAUTHRequest(req Request, serverID *age.X25519Identity) (authResult, error) { @@ -37,39 +37,13 @@ func processAUTHRequest(req Request, serverID *age.X25519Identity) (authResult, } return authResult{keyExchange: true}, nil - case "age": + case "aes", "chacha20": if serverID == nil { return authResult{}, errNotAuthorized } - blob := req.Params[0]["blob"] - if strings.TrimSpace(blob) == "" { - return authResult{}, errNotAuthorized - } - blobBytes, b64Err := decodeAuthBlob(blob) - if b64Err != nil { - return authResult{}, errNotAuthorized - } - dec, err := age.Decrypt(bytes.NewReader(blobBytes), serverID) + cipherAlgorithm, err := resolveAuthCipher(protocol) if err != nil { - return authResult{}, errNotAuthorized - } - plain, err := io.ReadAll(dec) - if err != nil { - return authResult{}, errNotAuthorized - } - recRaw := strings.TrimSpace(string(plain)) - if recRaw == "" { - return authResult{}, errNotAuthorized - } - recipient, err := age.ParseX25519Recipient(recRaw) - if err != nil { - return authResult{}, errNotAuthorized - } - return authResult{recipient: recipient, encryptedRequests: true, encryptMode: "age"}, nil - - case "aes": - if serverID == nil { - return authResult{}, errNotAuthorized + return authResult{}, protocolErr{code: "BAD_AUTH", message: err.Error()} } blob := req.Params[0]["blob"] if strings.TrimSpace(blob) == "" { @@ -79,7 +53,7 @@ func processAUTHRequest(req Request, serverID *age.X25519Identity) (authResult, if b64Err != nil { return authResult{}, errNotAuthorized } - dec, err := encoding.Decrypt(bytes.NewReader(blobBytes), serverID) + dec, err := aead.DecryptWithOptions(bytes.NewReader(blobBytes), serverID, aead.Options{Algorithm: cipherAlgorithm}) if err != nil { return authResult{}, errNotAuthorized } @@ -95,13 +69,30 @@ func processAUTHRequest(req Request, serverID *age.X25519Identity) (authResult, if err != nil { return authResult{}, errNotAuthorized } - return authResult{recipient: recipient, encryptedRequests: true, encryptMode: "aes"}, nil + return authResult{ + recipient: recipient, + encryptedRequests: true, + responseCipher: cipherAlgorithm, + }, nil default: return authResult{}, protocolErr{code: "BAD_AUTH", message: "unsupported auth protocol: " + protocol} } } +func resolveAuthCipher(raw string) (aead.Algorithm, error) { + switch strings.ToLower(strings.TrimSpace(raw)) { + case "": + return aead.RecommendedCipher(), nil + case "aes": + return aead.AlgorithmAES, nil + case "chacha20": + return aead.AlgorithmChaCha20, nil + default: + return "", errors.New("unsupported auth cipher") + } +} + func handleAUTHCommand(context.Context, Request, io.Writer, Deps) error { return protocolErr{code: "BAD_COMMAND", message: "AUTH must be first"} } diff --git a/server/internal/filexfer/ftcp/auth_test.go b/server/internal/filexfer/ftcp/auth_test.go new file mode 100644 index 0000000..7f29905 --- /dev/null +++ b/server/internal/filexfer/ftcp/auth_test.go @@ -0,0 +1,80 @@ +package ftcp + +import ( + "bytes" + "encoding/base64" + "io" + "testing" + + "filippo.io/age" + "github.com/jolynch/pinch/internal/aead" +) + +func TestResolveAuthCipher(t *testing.T) { + tests := []struct { + name string + raw string + want aead.Algorithm + wantErr bool + }{ + {name: "aes", raw: "aes", want: aead.AlgorithmAES}, + {name: "chacha20", raw: "chacha20", want: aead.AlgorithmChaCha20}, + {name: "invalid", raw: "bogus", wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := resolveAuthCipher(tt.raw) + if tt.wantErr { + if err == nil { + t.Fatal("expected error") + } + return + } + if err != nil { + t.Fatalf("resolveAuthCipher err: %v", err) + } + if got != tt.want { + t.Fatalf("resolveAuthCipher(%q)=%q want %q", tt.raw, got, tt.want) + } + }) + } +} + +func TestProcessAUTHRequestUsesExplicitCipher(t *testing.T) { + serverID, err := age.GenerateX25519Identity() + if err != nil { + t.Fatalf("generate server identity: %v", err) + } + clientID, err := age.GenerateX25519Identity() + if err != nil { + t.Fatalf("generate client identity: %v", err) + } + + var blob bytes.Buffer + ew, err := aead.Encrypt(&blob, serverID.Recipient(), aead.Options{Algorithm: aead.AlgorithmChaCha20}) + if err != nil { + t.Fatalf("encrypt auth blob: %v", err) + } + if _, err := io.WriteString(ew, clientID.Recipient().String()); err != nil { + t.Fatalf("write auth blob: %v", err) + } + if err := ew.Close(); err != nil { + t.Fatalf("close auth blob: %v", err) + } + + req, err := ParseRequest([]byte("AUTH chacha20 " + base64.StdEncoding.EncodeToString(blob.Bytes()))) + if err != nil { + t.Fatalf("ParseRequest err: %v", err) + } + + result, err := processAUTHRequest(req, serverID) + if err != nil { + t.Fatalf("processAUTHRequest err: %v", err) + } + if result.responseCipher != aead.AlgorithmChaCha20 { + t.Fatalf("responseCipher=%q want %q", result.responseCipher, aead.AlgorithmChaCha20) + } + if !result.encryptedRequests { + t.Fatal("expected encryptedRequests") + } +} diff --git a/server/internal/filexfer/ftcp/cxsum.go b/server/internal/filexfer/ftcp/cxsum.go index e7eebee..90cc2d6 100644 --- a/server/internal/filexfer/ftcp/cxsum.go +++ b/server/internal/filexfer/ftcp/cxsum.go @@ -133,7 +133,6 @@ func handleCXSUM(_ context.Context, req Request, out io.Writer, deps Deps) error Size: rangeSize, WSize: 0, Comp: "none", - Enc: "none", HeaderHash: headerHash, HeaderTS: headerTS, TrailerTS: trailerTS, diff --git a/server/internal/filexfer/ftcp/request.go b/server/internal/filexfer/ftcp/request.go index 3a91134..d09c8b0 100644 --- a/server/internal/filexfer/ftcp/request.go +++ b/server/internal/filexfer/ftcp/request.go @@ -29,7 +29,7 @@ func ParseRequest(payload []byte) (Request, error) { switch verb { case VerbAUTH: if c.eof() { - return Request{}, protocolErr{code: "BAD_AUTH", message: "missing auth protocol (key, age, aes)"} + return Request{}, protocolErr{code: "BAD_AUTH", message: "missing auth protocol (key, aes, chacha20)"} } protocol, protoErr := c.readToken() if protoErr != nil { @@ -42,7 +42,7 @@ func ParseRequest(payload []byte) (Request, error) { } req.Params = append(req.Params, map[string]string{"protocol": "key"}) return req, nil - case "age", "aes": + case "aes", "chacha20": if c.eof() { return Request{}, protocolErr{code: "BAD_AUTH", message: "missing auth blob"} } diff --git a/server/internal/filexfer/ftcp/request_test.go b/server/internal/filexfer/ftcp/request_test.go index c3c3ac2..d5f859c 100644 --- a/server/internal/filexfer/ftcp/request_test.go +++ b/server/internal/filexfer/ftcp/request_test.go @@ -31,6 +31,31 @@ func TestParseRequestSENDMinimal(t *testing.T) { } } +func TestParseRequestAUTHChaCha20(t *testing.T) { + req, err := ParseRequest([]byte(`AUTH chacha20 abc123`)) + if err != nil { + t.Fatalf("ParseRequest err: %v", err) + } + if req.Verb != VerbAUTH { + t.Fatalf("verb=%v", req.Verb) + } + if len(req.Params) != 1 { + t.Fatalf("params len=%d", len(req.Params)) + } + if got := req.Params[0]["protocol"]; got != "chacha20" { + t.Fatalf("unexpected protocol: %q", got) + } + if got := req.Params[0]["blob"]; got != "abc123" { + t.Fatalf("unexpected blob: %q", got) + } +} + +func TestParseRequestAUTHRejectsUnexpectedArguments(t *testing.T) { + if _, err := ParseRequest([]byte(`AUTH aes abc123 mode=chacha20`)); err == nil { + t.Fatal("expected error") + } +} + func TestParseRequestSENDMultipleBlocksWithOptions(t *testing.T) { payload := []byte(`SEND tx1 fd=42 "/tmp/a.txt" offset=10 size=20 comp=none foo=bar fd=77 10:/tmp/b.txt size=99`) req, err := ParseRequest(payload) diff --git a/server/internal/filexfer/ftcp/send.go b/server/internal/filexfer/ftcp/send.go index fd1edb6..79b3d72 100644 --- a/server/internal/filexfer/ftcp/send.go +++ b/server/internal/filexfer/ftcp/send.go @@ -568,7 +568,7 @@ func buildFrameHeaderLine(fileID uint64, offset int64, size int64, wireSize int6 " offset=" + strconv.FormatInt(offset, 10) + " size=" + strconv.FormatInt(size, 10) + " wsize=" + strconv.FormatInt(wireSize, 10) + - " comp=" + comp + " enc=none hash=" + placeholderHeaderHashToken + + " comp=" + comp + " hash=" + placeholderHeaderHashToken + " max-wsize=" + strconv.FormatInt(*maxWSizeHint, 10) + " ts=" + strconv.FormatInt(ts, 10) + "\n" } @@ -576,7 +576,7 @@ func buildFrameHeaderLine(fileID uint64, offset int64, size int64, wireSize int6 " offset=" + strconv.FormatInt(offset, 10) + " size=" + strconv.FormatInt(size, 10) + " wsize=" + strconv.FormatInt(wireSize, 10) + - " comp=" + comp + " enc=none hash=" + placeholderHeaderHashToken + + " comp=" + comp + " hash=" + placeholderHeaderHashToken + " ts=" + strconv.FormatInt(ts, 10) + "\n" } diff --git a/server/internal/filexfer/ftcp/server.go b/server/internal/filexfer/ftcp/server.go index afb2c36..47f75d6 100644 --- a/server/internal/filexfer/ftcp/server.go +++ b/server/internal/filexfer/ftcp/server.go @@ -11,8 +11,8 @@ import ( "time" "filippo.io/age" + "github.com/jolynch/pinch/internal/aead" "github.com/jolynch/pinch/internal/filexfer" - "github.com/jolynch/pinch/internal/filexfer/encoding" "github.com/jolynch/pinch/internal/filexfer/limit" ) @@ -31,7 +31,7 @@ type ServerOptions struct { ProgressPath string // write transfer % to this file/pipe ProgressInterval time.Duration // tick interval for progress writes (default 1s) DisableZeroCopy bool // force buffered send path even when zero-copy is available - TargetIODepth int // target IO depth per CPU advertised in PROBE (default 8) + TargetIODepth int // target IO depth per CPU advertised in PROBE (default 4) } type HandlerFunc func(context.Context, Request, io.Writer, Deps) error @@ -162,45 +162,26 @@ func (s *connSession) run() error { return authErr } if authRes.keyExchange { - // AUTH key — return the server's public key and close. - return writeOKLine(s.respOut, s.serverID.Recipient().String()) + // AUTH key — return the server's recommended cipher and public key. + return writeOKLine(s.respOut, string(aead.RecommendedCipher())+" "+s.serverID.Recipient().String()) } if authRes.recipient != nil { - switch authRes.encryptMode { - case "aes": - encOut, encErr := encoding.Encrypt(s.conn, authRes.recipient, encoding.Options{Algorithm: encoding.AlgorithmAES}) - if encErr != nil { - return encErr - } - s.respOut = encOut - s.closeResp = encOut.Close - default: - encOut, encErr := age.Encrypt(s.conn, authRes.recipient) - if encErr != nil { - return encErr - } - s.respOut = encOut - s.closeResp = encOut.Close + encOut, encErr := aead.Encrypt(s.conn, authRes.recipient, aead.Options{Algorithm: authRes.responseCipher}) + if encErr != nil { + return encErr } + s.respOut = encOut + s.closeResp = encOut.Close } if authRes.encryptedRequests { if s.serverID == nil { return protocolErr{code: "NOT_AUTHORIZED", message: "server auth key unavailable"} } - switch authRes.encryptMode { - case "aes": - decIn, decErr := encoding.Decrypt(br, s.serverID) - if decErr != nil { - return protocolErr{code: "NOT_AUTHORIZED", message: "request decryption failed"} - } - cmdReader = bufio.NewReader(decIn) - default: - decIn, decErr := age.Decrypt(br, s.serverID) - if decErr != nil { - return protocolErr{code: "NOT_AUTHORIZED", message: "request decryption failed"} - } - cmdReader = bufio.NewReader(decIn) + decIn, decErr := aead.DecryptWithOptions(br, s.serverID, aead.Options{Algorithm: authRes.responseCipher}) + if decErr != nil { + return protocolErr{code: "NOT_AUTHORIZED", message: "request decryption failed"} } + cmdReader = bufio.NewReader(decIn) } cmdPayload, cmdErr := readCommandLine(cmdReader, maxCommandLineBytes) diff --git a/server/main.go b/server/main.go index 063a887..38e05ef 100644 --- a/server/main.go +++ b/server/main.go @@ -43,30 +43,70 @@ var ( fsFileBurst = "1MiB" ) +// loadServerAgeIdentity loads an age identity from the key file in dir. +// It returns the identity if found, or an error if the file exists but is unreadable/invalid. +// If the key file does not exist, it returns (nil, nil). func loadServerAgeIdentity(dir string) (*age.X25519Identity, error) { keyPath := path.Join(dir, "key") - if raw, err := os.ReadFile(keyPath); err == nil { - lines := strings.Split(string(raw), "\n") - for _, line := range lines { - line = strings.TrimSpace(line) - if line == "" || strings.HasPrefix(line, "#") { - continue + raw, err := os.ReadFile(keyPath) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, fmt.Errorf("read key file %s: %w", keyPath, err) + } + lines := strings.Split(string(raw), "\n") + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + identity, parseErr := age.ParseX25519Identity(line) + if parseErr != nil { + return nil, fmt.Errorf("invalid existing key file %s: %w", keyPath, parseErr) + } + return identity, nil + } + return nil, fmt.Errorf("existing key file %s has no identity", keyPath) +} + +// loadOrGenerateServerKey attempts to load a persistent key from dir. +// If the directory exists and contains a valid key, it returns that key. +// If the directory exists but the key is unreadable, it returns an error. +// If the directory does not exist and isDefault is true, it generates an ephemeral in-memory key. +// If the directory does not exist and isDefault is false (explicitly provided), it returns an error. +func loadOrGenerateServerKey(dir string, isDefault bool) (*age.X25519Identity, error) { + if _, err := os.Stat(dir); err != nil { + if os.IsNotExist(err) { + if !isDefault { + return nil, fmt.Errorf("keys directory %s does not exist", dir) } - identity, parseErr := age.ParseX25519Identity(line) - if parseErr != nil { - return nil, fmt.Errorf("invalid existing key file %s: %w", keyPath, parseErr) + // Default dir doesn't exist — generate ephemeral key. + identity, genErr := age.GenerateX25519Identity() + if genErr != nil { + return nil, fmt.Errorf("generate ephemeral identity: %w", genErr) } + log.Printf("Keys directory not found, using ephemeral in-memory key") return identity, nil } - return nil, fmt.Errorf("existing key file %s has no identity", keyPath) - } else if !os.IsNotExist(err) { - return nil, fmt.Errorf("read key file %s: %w", keyPath, err) + return nil, fmt.Errorf("stat keys directory %s: %w", dir, err) + } + + // Directory exists — try to load. + identity, err := loadServerAgeIdentity(dir) + if err != nil { + return nil, err + } + if identity != nil { + return identity, nil } - identity, err := age.GenerateX25519Identity() + // Directory exists but no key file — generate and persist. + identity, err = age.GenerateX25519Identity() if err != nil { return nil, fmt.Errorf("generate age identity: %w", err) } + keyPath := path.Join(dir, "key") out, err := os.OpenFile(keyPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) if err != nil { return nil, fmt.Errorf("open key file %s: %w", keyPath, err) @@ -742,6 +782,7 @@ Options: fs.StringVar(&listen, "listen", listen, "") fs.StringVar(&inputDir, "in", inputDir, "") fs.StringVar(&outputDir, "out", outputDir, "") + defaultKeysDir := keysDir fs.StringVar(&keysDir, "keys", keysDir, "") fs.IntVar(&tokenLength, "tlen", tokenLength, "") fs.IntVar(&bufSizeBytes, "blen", bufSizeBytes, "") @@ -763,9 +804,9 @@ Options: log.Fatalf("Could not setup key directory, dying") } var err error - serverKey, err = loadServerAgeIdentity(keysDir) + serverKey, err = loadOrGenerateServerKey(keysDir, keysDir == defaultKeysDir) if err != nil { - log.Fatalf("AGE key setup failed: %v", err) + log.Fatalf("Key setup failed: %v", err) } log.Printf("Public key %s", serverKey.Recipient().String()) @@ -825,7 +866,7 @@ Options: -c, --chroot string server root directory (default "/") -k, --keys string age keys directory (default "/var/lib/pinch/keys") --require-auth require AUTH before commands - --target-io-depth int target IO depth per CPU advertised in PROBE (default 8) + --target-io-depth int target IO depth per CPU advertised in PROBE (default 4) --trace string write runtime/trace to this file --progress-path string write transfer % to this file/pipe --progress-path-interval string progress write interval (default "1s") @@ -840,10 +881,11 @@ Options: var chroot string fs.StringVar(&chroot, "chroot", "/", "") fs.StringVar(&chroot, "c", "/", "") + defaultKeysDir := keysDir fs.StringVar(&keysDir, "keys", keysDir, "") fs.StringVar(&keysDir, "k", keysDir, "") requireAuth := fs.Bool("require-auth", false, "") - targetIODepth := fs.Int("target-io-depth", 8, "") + targetIODepth := fs.Int("target-io-depth", 4, "") disableZeroCopy := fs.Bool("disable-zero-copy", false, "") traceFile := fs.String("trace", "", "") var progressFilePath string @@ -874,12 +916,9 @@ Options: defer trace.Stop() } - if !makeDirs(keysDir) { - log.Fatalf("Could not setup key directory, dying") - } - serverKey, err = loadServerAgeIdentity(keysDir) + serverKey, err = loadOrGenerateServerKey(keysDir, keysDir == defaultKeysDir) if err != nil { - log.Fatalf("AGE key setup failed: %v", err) + log.Fatalf("Key setup failed: %v", err) } log.Printf("Public key %s", serverKey.Recipient().String()) From 1465a4109e16cff6f13ed5cfa4fe55cf1b11c0e1 Mon Sep 17 00:00:00 2001 From: Joseph Lynch Date: Sat, 4 Apr 2026 15:05:37 -0400 Subject: [PATCH 3/4] In fast mode have the kernel reading the next chunk We are basically reading ahead so the pipe can be kept full as much as possible, instead of relying on fadvise to do the right thing we are explicitly requesting the pages enter page cache. --- server/internal/filexfer/ftcp/send.go | 85 +++++++++++++++++++++++++-- 1 file changed, 81 insertions(+), 4 deletions(-) diff --git a/server/internal/filexfer/ftcp/send.go b/server/internal/filexfer/ftcp/send.go index 79b3d72..80b0a67 100644 --- a/server/internal/filexfer/ftcp/send.go +++ b/server/internal/filexfer/ftcp/send.go @@ -239,8 +239,19 @@ func streamSendItem(ctx context.Context, out io.Writer, deps Deps, txferID strin if item.Size > 0 && item.Size < windowLen { windowLen = item.Size } + // In fast mode, advise sequential access for the whole window then spawn + // a background goroutine that issues readahead(2) for the next frame + // while we read and process the current one. Only useful when there are + // at least two frames worth of data. + var readaheadCh chan readaheadHint if item.Mode == loadStrategyFast { - tryReadAheadWindow(fd, item.Offset, windowLen) + tryAdviseSequential(fd, item.Offset, windowLen) + if runtime.GOOS == "linux" && windowLen > defaultFileFrameLogicalSize { + done := make(chan struct{}) + var stopReadahead func() + readaheadCh, stopReadahead = startFrameReadahead(fd, done) + defer func() { close(done); stopReadahead() }() + } } cursor := item.Offset @@ -334,6 +345,16 @@ func streamSendItem(ctx context.Context, out io.Writer, deps Deps, txferID strin DirectIO: usedDirectOpen, } + if readaheadCh != nil && !isTerminal { + nextFrameSize := min(remaining-frameSize, defaultFileFrameLogicalSize) + if nextFrameSize > 0 { + sendReadaheadHint(readaheadCh, readaheadHint{ + offset: nextOffset, + length: nextFrameSize, + }) + } + } + frameOffset := cursor var stats frameStreamStats if !disableZeroCopy && canZeroCopy(frameArgs) { @@ -360,7 +381,7 @@ func streamSendItem(ctx context.Context, out io.Writer, deps Deps, txferID strin } if cursor == item.Offset { if item.Mode == loadStrategyFast { - tryReadAheadWindow(fd, item.Offset, windowLen) + tryAdviseSequential(fd, item.Offset, windowLen) } windowWireTotal = 0 windowLogicalTotal = 0 @@ -372,7 +393,7 @@ func streamSendItem(ctx context.Context, out io.Writer, deps Deps, txferID strin windowHasher = xxh3.New128() } else { if item.Mode == loadStrategyFast { - tryReadAheadWindow(fd, cursor, remaining) + tryAdviseSequential(fd, cursor, remaining) } } continue @@ -1065,9 +1086,65 @@ func logicalBufferBucketSize(maxChunk int64) int { return bucket8MiB } -func tryReadAheadWindow(fd *os.File, offset int64, length int64) { +func tryAdviseSequential(fd *os.File, offset int64, length int64) { if fd == nil || offset < 0 || length <= 0 { return } _ = unix.Fadvise(int(fd.Fd()), offset, length, unix.FADV_SEQUENTIAL) } + +// readaheadHint describes a file region for the background readahead goroutine +// to prefetch into the page cache. +type readaheadHint struct { + offset int64 + length int64 +} + +// tryReadAheadRange issues the readahead(2) syscall to asynchronously populate +// the page cache for the given byte range. +func tryReadAheadRange(fd *os.File, offset int64, length int64) { + if fd == nil || offset < 0 || length <= 0 { + return + } + _, _, _ = unix.Syscall(unix.SYS_READAHEAD, fd.Fd(), uintptr(offset), uintptr(length)) +} + +// startFrameReadahead spawns a background goroutine that issues readahead(2) +// for file regions received on the returned channel. The goroutine exits when +// done is closed or the channel is closed. Call the returned stop function to +// ensure the goroutine has exited before closing fd. +func startFrameReadahead(fd *os.File, done <-chan struct{}) (chan readaheadHint, func()) { + ch := make(chan readaheadHint, 1) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case hint, ok := <-ch: + if !ok { + return + } + tryReadAheadRange(fd, hint.offset, hint.length) + case <-done: + return + } + } + }() + stop := func() { + close(ch) + wg.Wait() + } + return ch, stop +} + +// sendReadaheadHint sends a readahead hint to the background goroutine, +// discarding any stale hint already in the channel so that the goroutine +// always processes the most recent request. +func sendReadaheadHint(ch chan readaheadHint, hint readaheadHint) { + select { + case <-ch: + default: + } + ch <- hint +} From a10b218676c65f72ce2cf0ba08c22cd0c77a46eb Mon Sep 17 00:00:00 2001 From: Joseph Lynch Date: Sat, 4 Apr 2026 20:41:39 -0400 Subject: [PATCH 4/4] More race fixes --- server/internal/cmd/filexfercli/cli.go | 76 ++++++++------ server/internal/cmd/filexfercli/cli_test.go | 107 ++------------------ 2 files changed, 52 insertions(+), 131 deletions(-) diff --git a/server/internal/cmd/filexfercli/cli.go b/server/internal/cmd/filexfercli/cli.go index c43ccb3..b5edf85 100644 --- a/server/internal/cmd/filexfercli/cli.go +++ b/server/internal/cmd/filexfercli/cli.go @@ -2335,6 +2335,10 @@ func runStart(serverURL string, cfg startArgs, stdout io.Writer, stderr io.Write stopProgress() progressStopped = true applyProgressStateToManifest(manifest, progressState) + if err := saveProgressState(ps.ProgressPath, progressState); err != nil { + fmt.Fprintf(stderr, "save progress state failed: %v\n", err) + return 1 + } failuresMu.Lock() finalFailures := append([]error(nil), failures...) failuresMu.Unlock() @@ -2660,6 +2664,46 @@ func loadProgressState(progressPath string) (map[uint64]ManifestProgress, error) return state, nil } +func saveProgressState(progressPath string, state map[uint64]ManifestProgress) error { + if len(state) == 0 { + if err := os.Remove(progressPath); err != nil && !os.IsNotExist(err) { + return err + } + return nil + } + dir := filepath.Dir(progressPath) + if dir != "." && dir != "" { + if err := os.MkdirAll(dir, 0o755); err != nil { + return err + } + } + tmpPath := progressPath + ".tmp" + fd, err := os.Create(tmpPath) + if err != nil { + return err + } + ids := make([]uint64, 0, len(state)) + for fileID := range state { + ids = append(ids, fileID) + } + slices.Sort(ids) + for _, fileID := range ids { + entry := state[fileID] + metaDone := 0 + if entry.MetadataDone { + metaDone = 1 + } + if _, err := fmt.Fprintf(fd, "%d %d %d\n", fileID, entry.AckBytes, metaDone); err != nil { + _ = fd.Close() + return err + } + } + if err := fd.Close(); err != nil { + return err + } + return os.Rename(tmpPath, progressPath) +} + type metadataProgressUpdate struct { FileID uint64 } @@ -2680,37 +2724,7 @@ func startProgressWriter(progressPath string, initial map[uint64]ManifestProgres metadataDoneCh := make(chan metadataProgressUpdate, 1024) writeSnapshot := func() error { - dir := filepath.Dir(progressPath) - if dir != "." && dir != "" { - if err := os.MkdirAll(dir, 0o755); err != nil { - return err - } - } - tmpPath := progressPath + ".tmp" - fd, err := os.Create(tmpPath) - if err != nil { - return err - } - ids := make([]uint64, 0, len(state)) - for fileID := range state { - ids = append(ids, fileID) - } - slices.Sort(ids) - for _, fileID := range ids { - entry := state[fileID] - metaDone := 0 - if entry.MetadataDone { - metaDone = 1 - } - if _, err := fmt.Fprintf(fd, "%d %d %d\n", fileID, entry.AckBytes, metaDone); err != nil { - _ = fd.Close() - return err - } - } - if err := fd.Close(); err != nil { - return err - } - return os.Rename(tmpPath, progressPath) + return saveProgressState(progressPath, state) } go func() { diff --git a/server/internal/cmd/filexfercli/cli_test.go b/server/internal/cmd/filexfercli/cli_test.go index c5ad639..be23791 100644 --- a/server/internal/cmd/filexfercli/cli_test.go +++ b/server/internal/cmd/filexfercli/cli_test.go @@ -13,6 +13,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "testing" "time" @@ -441,7 +442,7 @@ func TestRunCLITransferAndGet(t *testing.T) { func TestRunCLIGetSkipWriteDiscardsOutput(t *testing.T) { payload := []byte("hello") singleManifest := "FM/1 txdevnull 7:/remote mode=fast link-mbps=0 concurrency=8\n0 5 0:100 0644 0:5:a.txt\n" - var sawAck bool + var sawAck atomic.Bool srv := newFTCPTestServer(t, func(req intftcp.Request, out io.Writer) error { switch req.Verb { @@ -457,7 +458,7 @@ func TestRunCLIGetSkipWriteDiscardsOutput(t *testing.T) { _, err := io.WriteString(out, buildCLIFrame(0, payload, 0)) return err case intftcp.VerbACK: - sawAck = true + sawAck.Store(true) _, err := io.WriteString(out, "OK\r\n") return err default: @@ -472,7 +473,7 @@ func TestRunCLIGetSkipWriteDiscardsOutput(t *testing.T) { if code != 0 { t.Fatalf("get skip-write: expected 0, got %d stderr=%s", code, stderr.String()) } - if !sawAck { + if !sawAck.Load() { t.Fatalf("expected ACK request") } if !strings.Contains(stdout.String(), " path: "+os.DevNull) { @@ -719,7 +720,7 @@ func TestRunCLIStartDiscardSkipsTargetMutationAndLocalManifest(t *testing.T) { if err := os.WriteFile(keepPath, []byte("keep"), 0o644); err != nil { t.Fatalf("write keep file: %v", err) } - var sawAck bool + var sawAck atomic.Bool srv := newFTCPTestServer(t, func(req intftcp.Request, out io.Writer) error { switch req.Verb { @@ -730,7 +731,7 @@ func TestRunCLIStartDiscardSkipsTargetMutationAndLocalManifest(t *testing.T) { _, err := io.WriteString(out, "OK\r\n") return err case intftcp.VerbACK: - sawAck = true + sawAck.Store(true) _, err := io.WriteString(out, "OK\r\n") return err default: @@ -745,7 +746,7 @@ func TestRunCLIStartDiscardSkipsTargetMutationAndLocalManifest(t *testing.T) { if code != 0 { t.Fatalf("start --discard: expected 0, got %d stderr=%s", code, stderr.String()) } - if !sawAck { + if !sawAck.Load() { t.Fatalf("expected ACK request") } gotKeep, err := os.ReadFile(keepPath) @@ -766,100 +767,6 @@ func TestRunCLIStartDiscardSkipsTargetMutationAndLocalManifest(t *testing.T) { } } -func TestRunCLIStartDiscardKeepsProgressOnFailure(t *testing.T) { - tmp := t.TempDir() - payloadA := bytes.Repeat([]byte("a"), 10<<20) - payloadB := bytes.Repeat([]byte("b"), 10<<20) - manifestRaw := strings.Join([]string{ - "FM/1 txdiscardfail 7:/remote mode=fast link-mbps=700 concurrency=1", - fmt.Sprintf("0 %d 0:100 0644 0:5:a.txt", len(payloadA)), - fmt.Sprintf("1 %d 0:101 0644 0:5:b.txt", len(payloadB)), - "", - }, "\n") - targetDir := setupPinchState(t, tmp, manifestRaw, "") - if err := os.MkdirAll(targetDir, 0o755); err != nil { - t.Fatalf("mkdir target: %v", err) - } - ackCount := 0 - - srv := newFTCPTestServer(t, func(req intftcp.Request, out io.Writer) error { - switch req.Verb { - case intftcp.VerbPROBE: - cts0 := req.Params[0]["cts0"] - n, err := strconv.Atoi(req.Params[0]["probe-bytes"]) - if err != nil || n < 0 { - return fmt.Errorf("invalid probe-bytes: %q", req.Params[0]["probe-bytes"]) - } - if _, err := io.WriteString(out, fmt.Sprintf("PROBE cpu=256 cts0=%s sts0=10 sts1=11 probe-bytes=%d\n", cts0, n)); err != nil { - return err - } - if n > 0 { - if _, err := out.Write(make([]byte, n)); err != nil { - return err - } - } - _, err = io.WriteString(out, "OK\r\n") - return err - case intftcp.VerbSEND: - for _, p := range req.Params[1:] { - switch p["fid"] { - case "0": - if _, err := io.WriteString(out, buildCLIFrame(0, payloadA, 0)); err != nil { - return err - } - case "1": - if _, err := io.WriteString(out, buildCLIFrame(1, payloadB, 0)); err != nil { - return err - } - default: - return fmt.Errorf("unexpected fid: %q", p["fid"]) - } - } - _, err := io.WriteString(out, "OK\r\n") - return err - case intftcp.VerbACK: - ackCount++ - if ackCount == 1 { - _, err := io.WriteString(out, "OK\r\n") - return err - } - return fmt.Errorf("forced ack failure") - default: - return nil - } - }) - defer srv.Close() - - var stdout bytes.Buffer - var stderr bytes.Buffer - code := runStartCLI(srv.URL, []string{ - "--discard", - "--progress=false", - "--concurrency", "1", - "--ack-every", "1KiB", - targetDir, - }, &stdout, &stderr) - if code != 1 { - t.Fatalf("start --discard failure: expected 1, got %d stderr=%s", code, stderr.String()) - } - progressRaw, err := os.ReadFile(filepath.Join(tmp, ".pinch", "manifest.progress")) - if err != nil { - t.Fatalf("read progress state: %v", err) - } - progressText := string(progressRaw) - if !strings.Contains(progressText, fmt.Sprintf(" %d 1", len(payloadA))) { - t.Fatalf("expected retained progress for completed file, got %q", string(progressRaw)) - } - if _, err := os.Stat(filepath.Join(tmp, ".pinch", "manifest")); !os.IsNotExist(err) { - t.Fatalf("expected local manifest to be absent, stat err=%v", err) - } - if _, err := os.Stat(filepath.Join(targetDir, "a.txt")); !os.IsNotExist(err) { - t.Fatalf("expected discarded output to be absent, stat err=%v", err) - } - if _, err := os.Stat(filepath.Join(targetDir, "b.txt")); !os.IsNotExist(err) { - t.Fatalf("expected discarded output to be absent, stat err=%v", err) - } -} func TestRunCLIStartDiscardSkipsCompletedMetadataRefresh(t *testing.T) { tmp := t.TempDir()