Turn the client into a background daemon; add per-chunk hash verification

Daemon (src/daemon): a Unix-socket control server holds every room this
machine is participating in. "share send"/"share <code>" resolve paths to
absolute, ask the daemon to join (auto-starting it if needed), and exit
immediately instead of blocking. "share stop" leaves every room and shuts
the daemon down; a stale socket from a crashed daemon self-heals on next
start.

swarm.Client is now a Start/Wait/Stop handle instead of a single blocking
Run, so the daemon can hold many rooms concurrently.

Also added per-chunk sha256 verification: the original seeder publishes a
hash for every chunk alongside the whole-file hash, and each chunk is
checked against it the moment it arrives — before it's written to disk or
reshared to other peers — rather than only catching corruption after the
whole file is assembled.

src/cli/cli.go

@@ -4,12 +4,13 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"path/filepath"
 	"runtime"
 	"strconv"
 
+	"github.com/tzgyn/share/src/daemon"
 	log "github.com/tzgyn/share/src/logger"
 	"github.com/tzgyn/share/src/models"
-	"github.com/tzgyn/share/src/swarm"
 	"github.com/tzgyn/share/src/tcp"
 	"github.com/tzgyn/share/src/utils"
 )
@@ -20,6 +21,9 @@ const usage = `share [GLOBAL OPTIONS] [COMMAND] [COMMAND OPTIONS] [filename(s)]
 
 a torrent-like swarm: everyone who joins a room can send and receive
 
+A background daemon does the actual work: joining a room hands it off to
+the daemon (starting it if it isn't already running) and exits right away.
+
 USAGE EXAMPLES:
    Seed a file into a new room (prints a code):
       share send file.txt
@@ -30,9 +34,14 @@ USAGE EXAMPLES:
    Join a room to receive whatever's shared there (and help re-share it):
       share secret-code
 
+   Stop the background daemon (leaves every room it's in):
+      share stop
+
 COMMANDS:
    send    seed file(s) into a room (see options with: share send -h)
    relay   start your own relay (optional; see options with: share relay -h)
+   stop    stop the background daemon and leave every room
+   daemon  run the background daemon in the foreground (mostly for debugging)
 
 Run "share [command] -h" for the options of a given command.
 `
@@ -66,6 +75,10 @@ func Run() error {
 			return runSend(splice(args, i))
 		case "relay":
 			return runRelay(splice(args, i))
+		case "stop":
+			return runStop(splice(args, i))
+		case "daemon":
+			return runDaemon(splice(args, i))
 		}
 	}
 	return runReceive(args)
@@ -119,19 +132,32 @@ func runSend(args []string) error {
 	if len(files) == 0 {
 		return fmt.Errorf("no files specified; usage: share send file1 [file2 ...]")
 	}
+	absFiles := make([]string, len(files))
+	for i, f := range files {
+		abs, err := filepath.Abs(f)
+		if err != nil {
+			return fmt.Errorf("could not resolve %s: %w", f, err)
+		}
+		absFiles[i] = abs
+	}
+
 	roomCode := *code
 	if roomCode == "" {
 		roomCode = utils.GetRandomName()
 	}
+
+	if err := daemon.EnsureRunning(*debug); err != nil {
+		return err
+	}
+	if err := daemon.Join(*relay, roomCode, absFiles, "", false); err != nil {
+		return err
+	}
+
 	fmt.Printf("Code is: %s\n", roomCode)
 	fmt.Println("On the other computer, run:")
 	fmt.Printf("\tshare %s\n", roomCode)
-
+	fmt.Println("Seeding in the background; run \"share stop\" to stop.")
-	return swarm.Run(swarm.Options{
+	return nil
-		RelayAddress: *relay,
-		Code:         roomCode,
-		Files:        files,
-	})
 }
 
 func runReceive(args []string) error {
@@ -155,13 +181,23 @@ func runReceive(args []string) error {
 		fs.Usage()
 		return nil
 	}
+	absOut, err := filepath.Abs(*out)
+	if err != nil {
+		return fmt.Errorf("could not resolve %s: %w", *out, err)
+	}
 
-	return swarm.Run(swarm.Options{
+	if err := daemon.EnsureRunning(*debug); err != nil {
-		RelayAddress: *relay,
+		return err
-		Code:         code,
+	}
-		OutDir:       *out,
+	if err := daemon.Join(*relay, code, nil, absOut, *oneTime); err != nil {
-		OneTime:      *oneTime,
+		return err
-	})
+	}
+
+	fmt.Printf("Joined room %s; downloading in the background.\n", code)
+	if !*oneTime {
+		fmt.Println("It'll keep reseeding to others; run \"share stop\" to stop.")
+	}
+	return nil
 }
 
 func runRelay(args []string) error {
@@ -190,3 +226,36 @@ func runRelay(args []string) error {
 	log.Infof("starting share relay version %v", Version)
 	return tcp.Run(debugString, *host, strconv.Itoa(*port))
 }
+
+func runStop(args []string) error {
+	fs := flag.NewFlagSet("stop", flag.ExitOnError)
+	fs.Usage = func() {
+		fmt.Fprintln(os.Stderr, "share stop")
+	}
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+
+	if err := daemon.Stop(); err != nil {
+		fmt.Println(err)
+		return nil
+	}
+	fmt.Println("Daemon stopped.")
+	return nil
+}
+
+func runDaemon(args []string) error {
+	fs := flag.NewFlagSet("daemon", flag.ExitOnError)
+	fs.Usage = func() {
+		fmt.Fprintln(os.Stderr, "share daemon [options]")
+		fs.PrintDefaults()
+	}
+	debug := fs.Bool("debug", false, "toggle debug mode")
+	quiet := fs.Bool("quiet", false, "disable all output")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	setDebugLevel(*quiet, *debug)
+
+	return daemon.Serve()
+}

src/daemon/client.go

@@ -0,0 +1,116 @@
+package daemon
+
+import (
+	"encoding/json"
+	"fmt"
+	"net"
+	"os"
+	"os/exec"
+	"syscall"
+	"time"
+)
+
+// IsRunning reports whether a daemon is currently listening on the
+// control socket.
+func IsRunning() bool {
+	path, err := socketPath()
+	if err != nil {
+		return false
+	}
+	conn, err := net.DialTimeout("unix", path, 200*time.Millisecond)
+	if err != nil {
+		return false
+	}
+	conn.Close()
+	return true
+}
+
+// EnsureRunning starts the daemon in the background if one isn't already
+// running, and waits until its control socket is reachable.
+func EnsureRunning(debug bool) error {
+	if IsRunning() {
+		return nil
+	}
+
+	exe, err := os.Executable()
+	if err != nil {
+		return fmt.Errorf("could not locate the share binary to start the daemon: %w", err)
+	}
+
+	args := []string{"daemon"}
+	if debug {
+		args = append(args, "--debug")
+	}
+	cmd := exec.Command(exe, args...)
+	cmd.Stdin = nil
+	if lf, err := logPath(); err == nil {
+		if out, err := os.OpenFile(lf, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644); err == nil {
+			cmd.Stdout = out
+			cmd.Stderr = out
+		}
+	}
+	// detach so the daemon survives this (short-lived) CLI process exiting
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
+
+	if err := cmd.Start(); err != nil {
+		return fmt.Errorf("could not start daemon: %w", err)
+	}
+	_ = cmd.Process.Release()
+
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if IsRunning() {
+			return nil
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	return fmt.Errorf("daemon did not start in time")
+}
+
+// Join asks the running daemon to start participating in a room. files
+// and outDir must already be absolute paths — the daemon may have been
+// started long ago from a different working directory.
+func Join(relayAddress, code string, files []string, outDir string, oneTime bool) error {
+	return call(request{
+		Action: "join",
+		Join: &joinRequest{
+			RelayAddress: relayAddress,
+			Code:         code,
+			Files:        files,
+			OutDir:       outDir,
+			OneTime:      oneTime,
+		},
+	})
+}
+
+// Stop tells the running daemon to leave every room it's in and exit.
+func Stop() error {
+	if !IsRunning() {
+		return fmt.Errorf("daemon is not running")
+	}
+	return call(request{Action: "stop"})
+}
+
+func call(req request) error {
+	path, err := socketPath()
+	if err != nil {
+		return err
+	}
+	conn, err := net.DialTimeout("unix", path, 2*time.Second)
+	if err != nil {
+		return fmt.Errorf("could not reach daemon: %w", err)
+	}
+	defer conn.Close()
+
+	if err := json.NewEncoder(conn).Encode(req); err != nil {
+		return err
+	}
+	var resp response
+	if err := json.NewDecoder(conn).Decode(&resp); err != nil {
+		return err
+	}
+	if !resp.OK {
+		return fmt.Errorf("%s", resp.Error)
+	}
+	return nil
+}

src/daemon/paths.go

@@ -0,0 +1,23 @@
+package daemon
+
+import (
+	"path/filepath"
+
+	"github.com/tzgyn/share/src/utils"
+)
+
+func socketPath() (string, error) {
+	dir, err := utils.GetConfigDir(true)
+	if err != nil {
+		return "", err
+	}
+	return filepath.Join(dir, "daemon.sock"), nil
+}
+
+func logPath() (string, error) {
+	dir, err := utils.GetConfigDir(true)
+	if err != nil {
+		return "", err
+	}
+	return filepath.Join(dir, "daemon.log"), nil
+}

src/daemon/protocol.go

@@ -0,0 +1,25 @@
+package daemon
+
+// request/response are the only vocabulary spoken over the control
+// socket between a CLI invocation and the background daemon.
+type request struct {
+	Action string       `json:"action"` // "join" or "stop"
+	Join   *joinRequest `json:"join,omitempty"`
+}
+
+// joinRequest asks the daemon to start participating in a room. Paths
+// must already be absolute — the daemon may have been started long ago
+// from a different working directory than whichever CLI invocation is
+// submitting this request.
+type joinRequest struct {
+	RelayAddress string   `json:"relay"`
+	Code         string   `json:"code"`
+	Files        []string `json:"files,omitempty"`
+	OutDir       string   `json:"out,omitempty"`
+	OneTime      bool     `json:"oneTime,omitempty"`
+}
+
+type response struct {
+	OK    bool   `json:"ok"`
+	Error string `json:"error,omitempty"`
+}

src/daemon/server.go

@@ -0,0 +1,142 @@
+package daemon
+
+import (
+	"encoding/json"
+	"fmt"
+	"net"
+	"os"
+	"sync"
+
+	log "github.com/tzgyn/share/src/logger"
+	"github.com/tzgyn/share/src/swarm"
+)
+
+type server struct {
+	mu    sync.Mutex
+	rooms map[string]*swarm.Client
+}
+
+// Serve runs the daemon: it listens on the control socket and services
+// join/stop requests until a stop is requested, then tears down every
+// room and returns. It's the whole daemon process's job — normally
+// started in the background by EnsureRunning, or run in the foreground
+// directly for debugging.
+func Serve() error {
+	path, err := socketPath()
+	if err != nil {
+		return err
+	}
+	if _, statErr := os.Stat(path); statErr == nil {
+		if IsRunning() {
+			return fmt.Errorf("a daemon is already running")
+		}
+		// stale socket left behind by a daemon that didn't shut down cleanly
+		os.Remove(path)
+	}
+
+	l, err := net.Listen("unix", path)
+	if err != nil {
+		return fmt.Errorf("could not listen on %s: %w", path, err)
+	}
+	defer os.Remove(path)
+
+	s := &server{rooms: make(map[string]*swarm.Client)}
+	log.Infof("daemon listening on %s", path)
+
+	for {
+		conn, err := l.Accept()
+		if err != nil {
+			return err
+		}
+		if s.handleConn(conn) {
+			s.stopAll()
+			return nil
+		}
+	}
+}
+
+// handleConn services one control request and reports whether the
+// daemon should shut down after responding (true only for "stop").
+func (s *server) handleConn(conn net.Conn) (shutdown bool) {
+	defer conn.Close()
+
+	var req request
+	if err := json.NewDecoder(conn).Decode(&req); err != nil {
+		respond(conn, response{OK: false, Error: err.Error()})
+		return false
+	}
+
+	switch req.Action {
+	case "join":
+		if req.Join == nil {
+			respond(conn, response{OK: false, Error: "missing join request"})
+			return false
+		}
+		if err := s.join(*req.Join); err != nil {
+			respond(conn, response{OK: false, Error: err.Error()})
+			return false
+		}
+		respond(conn, response{OK: true})
+		return false
+	case "stop":
+		respond(conn, response{OK: true})
+		return true
+	default:
+		respond(conn, response{OK: false, Error: fmt.Sprintf("unknown action %q", req.Action)})
+		return false
+	}
+}
+
+func respond(conn net.Conn, resp response) {
+	if err := json.NewEncoder(conn).Encode(resp); err != nil {
+		log.Debugf("could not send response: %v", err)
+	}
+}
+
+func (s *server) join(req joinRequest) error {
+	s.mu.Lock()
+	if _, already := s.rooms[req.Code]; already {
+		s.mu.Unlock()
+		return fmt.Errorf("already in room %s", req.Code)
+	}
+	s.mu.Unlock()
+
+	cl, err := swarm.Start(swarm.Options{
+		RelayAddress: req.RelayAddress,
+		Code:         req.Code,
+		Files:        req.Files,
+		OutDir:       req.OutDir,
+		OneTime:      req.OneTime,
+	})
+	if err != nil {
+		return err
+	}
+
+	s.mu.Lock()
+	s.rooms[req.Code] = cl
+	s.mu.Unlock()
+	log.Infof("joined room %s", req.Code)
+
+	go func() {
+		err := cl.Wait()
+		s.mu.Lock()
+		delete(s.rooms, req.Code)
+		s.mu.Unlock()
+		if err != nil {
+			log.Errorf("room %s ended: %v", req.Code, err)
+		} else {
+			log.Infof("room %s finished", req.Code)
+		}
+	}()
+
+	return nil
+}
+
+func (s *server) stopAll() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for code, cl := range s.rooms {
+		log.Infof("leaving room %s", code)
+		cl.Stop()
+	}
+}

src/swarm/chunks.go

@@ -2,6 +2,7 @@ package swarm
 
 import (
 	"bytes"
+	"crypto/sha256"
 	"io"
 	"os"
 	"sync"
@@ -158,3 +159,30 @@ func (cs *ChunkStore) VerifyHash(expected []byte) (bool, error) {
 	}
 	return bytes.Equal(got, expected), nil
 }
+
+// ChunkHash returns the sha256 hash of one chunk's content. A
+// cryptographic hash is used here (rather than the faster xxhash used
+// for the whole-file sanity check) because this is the check that
+// actually has to resist a peer deliberately handing out wrong data for
+// a chunk, not just catch accidental corruption.
+func ChunkHash(data []byte) []byte {
+	h := sha256.Sum256(data)
+	return h[:]
+}
+
+// ChunkHashes returns the hash of every chunk this store currently has,
+// in offset order. Only meaningful for a store that has every chunk
+// (i.e. a seeder's own file) — it's how the original hashes get
+// published so every other peer can verify each chunk as it arrives.
+func (cs *ChunkStore) ChunkHashes() ([][]byte, error) {
+	n := NumChunks(cs.size)
+	hashes := make([][]byte, n)
+	for i := int64(0); i < n; i++ {
+		data, err := cs.ReadChunk(i * ChunkSize)
+		if err != nil {
+			return nil, err
+		}
+		hashes[i] = ChunkHash(data)
+	}
+	return hashes, nil
+}

src/swarm/client.go

@@ -1,6 +1,7 @@
 package swarm
 
 import (
+	"bytes"
 	"crypto/rand"
 	"encoding/hex"
 	"fmt"
@@ -30,15 +31,6 @@ type Options struct {
 	OneTime      bool     // exit once everything currently known is complete, instead of staying to reseed
 }
 
-// Run joins the room described by opts and blocks until either an error
-// occurs, or (with OneTime set) every file known in the room has been
-// fully downloaded and verified. Without OneTime, Run only returns on
-// error — like a real torrent peer, it stays to keep reseeding whatever
-// it has to others until stopped.
-func Run(opts Options) error {
-	return newClient(opts).run()
-}
-
 type sharedFile struct {
 	entry FileManifestEntry // immutable identity: ID, Name, Size, Hash
 	store *ChunkStore
@@ -54,7 +46,10 @@ func newPeerLink(c *comm.Comm, key []byte) *peerLink {
 	return &peerLink{conn: c, key: key, remote: c.Connection().RemoteAddr().String()}
 }
 
-type client struct {
+// Client is a handle to one room this process is participating in. Start
+// returns one once the room is joined; the actual chunk exchange runs in
+// a background goroutine until Wait returns.
+type Client struct {
 	opts   Options
 	selfID string
 
@@ -65,16 +60,26 @@ type client struct {
 
 	dialedMu sync.Mutex
 	dialed   map[string]bool // relay peer IDs already dialed, so we don't redial every peer-list update
+
+	listener    net.Listener
+	relayConn   *comm.Comm
+	peerUpdates <-chan []tcp.PeerInfo
+
+	stopOnce sync.Once
+	stopCh   chan struct{}
+	done     chan error // buffered 1; sent to exactly once when the room's loop exits
 }
 
-func newClient(opts Options) *client {
+func newClient(opts Options) *Client {
-	return &client{
+	return &Client{
 		opts:    opts,
 		selfID:  randomID(),
 		files:   make(map[string]*sharedFile),
 		pending: make(map[string]map[int64]*peerLink),
 		links:   make(map[*peerLink]bool),
 		dialed:  make(map[string]bool),
+		stopCh:  make(chan struct{}),
+		done:    make(chan error, 1),
 	}
 }
 
@@ -86,31 +91,67 @@ func randomID() string {
 	return hex.EncodeToString(b)
 }
 
-func (cl *client) run() error {
+// Start seeds any local Files, opens a listener for other peers, and joins
-	for _, path := range cl.opts.Files {
+// the relay room described by opts. It returns as soon as that initial
+// join succeeds or fails — it does not wait for any transfer to complete.
+// The room then keeps running in the background (dialing peers, trading
+// chunks) until Stop is called or, with OneTime set, until everything
+// currently known has been downloaded; use Wait to block for that.
+func Start(opts Options) (*Client, error) {
+	cl := newClient(opts)
+
+	for _, path := range opts.Files {
 		if err := cl.addSeedFile(path); err != nil {
-			return err
+			return nil, err
 		}
 	}
 
 	listener, listenAddr, err := startListener()
 	if err != nil {
-		return fmt.Errorf("could not open a listener for peers: %w", err)
+		return nil, fmt.Errorf("could not open a listener for peers: %w", err)
 	}
-	defer listener.Close()
-	go cl.acceptLoop(listener)
 
-	relayConn, peerUpdates, err := tcp.JoinRoom(cl.opts.RelayAddress, cl.opts.Code, tcp.PeerInfo{ID: cl.selfID, Addr: listenAddr})
+	relayConn, peerUpdates, err := tcp.JoinRoom(opts.RelayAddress, opts.Code, tcp.PeerInfo{ID: cl.selfID, Addr: listenAddr})
 	if err != nil {
-		return fmt.Errorf("could not join room: %w", err)
+		listener.Close()
+		return nil, fmt.Errorf("could not join room: %w", err)
 	}
-	defer relayConn.Close()
 
+	cl.listener = listener
+	cl.relayConn = relayConn
+	cl.peerUpdates = peerUpdates
+
+	go cl.acceptLoop(listener)
+	go cl.loop()
+
+	return cl, nil
+}
+
+// Wait blocks until this room's participation ends, returning why: nil
+// for a clean stop or (with OneTime) a completed transfer, or an error if
+// the relay connection was lost.
+func (cl *Client) Wait() error {
+	return <-cl.done
+}
+
+// Stop ends this room's participation. Wait will then return nil.
+func (cl *Client) Stop() {
+	cl.stopOnce.Do(func() { close(cl.stopCh) })
+}
+
+func (cl *Client) finish(err error) {
+	cl.listener.Close()
+	cl.relayConn.Close()
+	cl.done <- err
+}
+
+func (cl *Client) loop() {
 	for {
 		select {
-		case peers, ok := <-peerUpdates:
+		case peers, ok := <-cl.peerUpdates:
 			if !ok {
-				return fmt.Errorf("lost connection to relay")
+				cl.finish(fmt.Errorf("lost connection to relay"))
+				return
 			}
 			for _, p := range peers {
 				if p.ID == cl.selfID || p.Addr == "" {
@@ -126,13 +167,17 @@ func (cl *client) run() error {
 			}
 		case <-time.After(500 * time.Millisecond):
 			if cl.opts.OneTime && cl.allKnownFilesComplete() {
-				return nil
+				cl.finish(nil)
+				return
 			}
+		case <-cl.stopCh:
+			cl.finish(nil)
+			return
 		}
 	}
 }
 
-func (cl *client) addSeedFile(path string) error {
+func (cl *Client) addSeedFile(path string) error {
 	info, err := os.Stat(path)
 	if err != nil {
 		return err
@@ -144,17 +189,22 @@ func (cl *client) addSeedFile(path string) error {
 	if err != nil {
 		return err
 	}
+	store := NewSeedChunkStore(path, info.Size())
+	chunkHashes, err := store.ChunkHashes()
+	if err != nil {
+		return err
+	}
 	name := filepath.Base(path)
 	id := utils.SHA256(fmt.Sprintf("%s:%d", name, info.Size()))
 	cl.files[id] = &sharedFile{
-		entry: FileManifestEntry{ID: id, Name: name, Size: info.Size(), Hash: hash},
+		entry: FileManifestEntry{ID: id, Name: name, Size: info.Size(), Hash: hash, ChunkHashes: chunkHashes},
-		store: NewSeedChunkStore(path, info.Size()),
+		store: store,
 	}
 	log.Infof("seeding %s (%s)", name, utils.ByteCountDecimal(info.Size()))
 	return nil
 }
 
-func (cl *client) allKnownFilesComplete() bool {
+func (cl *Client) allKnownFilesComplete() bool {
 	cl.mu.Lock()
 	defer cl.mu.Unlock()
 	if len(cl.files) == 0 {
@@ -177,7 +227,7 @@ func startListener() (net.Listener, string, error) {
 	return l, net.JoinHostPort(utils.LocalIP(), strconv.Itoa(port)), nil
 }
 
-func (cl *client) acceptLoop(l net.Listener) {
+func (cl *Client) acceptLoop(l net.Listener) {
 	for {
 		conn, err := l.Accept()
 		if err != nil {
@@ -196,7 +246,7 @@ func (cl *client) acceptLoop(l net.Listener) {
 	}
 }
 
-func (cl *client) dialPeer(p tcp.PeerInfo) {
+func (cl *Client) dialPeer(p tcp.PeerInfo) {
 	conn, err := net.DialTimeout("tcp", p.Addr, 5*time.Second)
 	if err != nil {
 		log.Debugf("could not dial peer %s at %s: %v", p.ID, p.Addr, err)
@@ -212,7 +262,7 @@ func (cl *client) dialPeer(p tcp.PeerInfo) {
 	cl.handlePeerLink(newPeerLink(c, key))
 }
 
-func (cl *client) handlePeerLink(link *peerLink) {
+func (cl *Client) handlePeerLink(link *peerLink) {
 	cl.mu.Lock()
 	cl.links[link] = true
 	cl.mu.Unlock()
@@ -250,7 +300,7 @@ func (cl *client) handlePeerLink(link *peerLink) {
 	}
 }
 
-func (cl *client) sendManifest(link *peerLink) error {
+func (cl *Client) sendManifest(link *peerLink) error {
 	cl.mu.Lock()
 	files := make([]FileManifestEntry, 0, len(cl.files))
 	for _, f := range cl.files {
@@ -262,7 +312,7 @@ func (cl *client) sendManifest(link *peerLink) error {
 	return sendMessage(link.conn, link.key, Message{Type: TypeManifest, Files: files})
 }
 
-func (cl *client) handleMessage(link *peerLink, msg Message) error {
+func (cl *Client) handleMessage(link *peerLink, msg Message) error {
 	switch msg.Type {
 	case TypeManifest:
 		for _, entry := range msg.Files {
@@ -279,7 +329,7 @@ func (cl *client) handleMessage(link *peerLink, msg Message) error {
 	return nil
 }
 
-func (cl *client) learnFile(entry FileManifestEntry) {
+func (cl *Client) learnFile(entry FileManifestEntry) {
 	cl.mu.Lock()
 	_, known := cl.files[entry.ID]
 	cl.mu.Unlock()
@@ -294,7 +344,7 @@ func (cl *client) learnFile(entry FileManifestEntry) {
 	}
 	cl.mu.Lock()
 	cl.files[entry.ID] = &sharedFile{
-		entry: FileManifestEntry{ID: entry.ID, Name: entry.Name, Size: entry.Size, Hash: entry.Hash},
+		entry: FileManifestEntry{ID: entry.ID, Name: entry.Name, Size: entry.Size, Hash: entry.Hash, ChunkHashes: entry.ChunkHashes},
 		store: store,
 	}
 	cl.mu.Unlock()
@@ -303,7 +353,7 @@ func (cl *client) learnFile(entry FileManifestEntry) {
 
 // considerRequests asks link for any of the given offsets of fileID that
 // we're still missing and haven't already asked someone else for.
-func (cl *client) considerRequests(fileID string, offsets []int64, link *peerLink) {
+func (cl *Client) considerRequests(fileID string, offsets []int64, link *peerLink) {
 	cl.mu.Lock()
 	f, ok := cl.files[fileID]
 	if !ok {
@@ -334,7 +384,7 @@ func (cl *client) considerRequests(fileID string, offsets []int64, link *peerLin
 	}
 }
 
-func (cl *client) serveRequest(link *peerLink, fileID string, offset int64) error {
+func (cl *Client) serveRequest(link *peerLink, fileID string, offset int64) error {
 	cl.mu.Lock()
 	f, ok := cl.files[fileID]
 	cl.mu.Unlock()
@@ -349,7 +399,7 @@ func (cl *client) serveRequest(link *peerLink, fileID string, offset int64) erro
 	return sendMessage(link.conn, link.key, Message{Type: TypeChunk, FileID: fileID, Offset: offset, Data: data})
 }
 
-func (cl *client) receiveChunk(link *peerLink, fileID string, offset int64, data []byte) error {
+func (cl *Client) receiveChunk(link *peerLink, fileID string, offset int64, data []byte) error {
 	cl.mu.Lock()
 	f, ok := cl.files[fileID]
 	if ok {
@@ -364,6 +414,12 @@ func (cl *client) receiveChunk(link *peerLink, fileID string, offset int64, data
 		return nil
 	}
 
+	idx := int(offset / ChunkSize)
+	if idx < len(f.entry.ChunkHashes) && !bytes.Equal(ChunkHash(data), f.entry.ChunkHashes[idx]) {
+		log.Warnf("chunk %d/%d of %s from %s failed hash verification; discarding", idx+1, NumChunks(f.entry.Size), f.entry.Name, link.remote)
+		return nil
+	}
+
 	log.Infof("received %s chunk %d/%d from %s", f.entry.Name, offset/ChunkSize+1, NumChunks(f.entry.Size), link.remote)
 	if err := f.store.WriteChunk(offset, data); err != nil {
 		return err

src/swarm/message.go

@@ -24,13 +24,18 @@ const (
 )
 
 // FileManifestEntry describes one file known to the swarm: its identity,
-// size, whole-file hash (for final verification), and which chunk offsets
+// size, whole-file hash (for a final sanity check once complete),
-// the sender of this manifest currently holds.
+// per-chunk hashes (so each chunk can be verified the moment it arrives,
+// before it's trusted or reshared), and which chunk offsets the sender of
+// this manifest currently holds. ID/Name/Size/Hash/ChunkHashes are set
+// once by the original seeder and simply carried along unchanged as the
+// file propagates through the swarm; only Have changes per sender.
 type FileManifestEntry struct {
 	ID          string   `json:"id"`
 	Name        string   `json:"name"`
 	Size        int64    `json:"size"`
 	Hash        []byte   `json:"hash"`
+	ChunkHashes [][]byte `json:"chunkHashes"`
 	Have        []int64  `json:"have"`
 }