diff --git a/share b/share index e746ea0..985056d 100755 Binary files a/share and b/share differ diff --git a/src/cli/cli.go b/src/cli/cli.go index 8d226cf..9ccd3b4 100644 --- a/src/cli/cli.go +++ b/src/cli/cli.go @@ -4,13 +4,13 @@ import ( "fmt" "os" "runtime" - "strconv" - "strings" "time" "github.com/schollz/cli/v2" "github.com/tzgyn/share/src/models" + "github.com/tzgyn/share/src/swarm" "github.com/tzgyn/share/src/tcp" + "github.com/tzgyn/share/src/utils" log "github.com/schollz/logger" ) @@ -26,44 +26,26 @@ func Run() (err error) { } app.Version = Version app.Compiled = time.Now() - app.Usage = "easily and securely transfer stuff from one computer to another" - app.UsageText = `share [GLOBAL OPTIONS] [COMMAND] [COMMAND OPTIONS] [filename(s) or folder] + app.Usage = "a torrent-like swarm: everyone who joins a room can send and receive" + app.UsageText = `share [GLOBAL OPTIONS] [COMMAND] [COMMAND OPTIONS] [filename(s)] USAGE EXAMPLES: - Send a file: + Seed a file into a new room (prints a code): share send file.txt - Send multiple files: - share send file1.txt file2.txt file3.txt + Seed multiple files, or into a room with a chosen code: + share send --code secret-code file1.txt file2.txt - Send everything in a folder: - share send example-folder-name - - Send a file with a custom code: - share send --code secret-code file.txt - - Receive a file using code: + Join a room to receive whatever's shared there (and help re-share it): share secret-code` app.Commands = []*cli.Command{ { Name: "send", - Usage: "send file(s), or folder (see options with share send -h)", - Description: "send file(s), or folder, over the relay", - ArgsUsage: "[filename(s) or folder]", + Usage: "seed file(s) into a room (see options with share send -h)", + Description: "seed file(s) into a swarm room so others can join and receive them", + ArgsUsage: "[filename(s)]", Flags: []cli.Flag{ - &cli.BoolFlag{Name: "zip", Usage: "zip folder before sending"}, - &cli.StringFlag{Name: "code", Aliases: []string{"c"}, Usage: "codephrase used to connect to relay (at least 6 characters)"}, - &cli.StringFlag{Name: "hash", Value: "xxhash", Usage: "hash algorithm (xxhash, imohash, md5, highway)"}, - &cli.StringFlag{Name: "text", Aliases: []string{"t"}, Usage: "send some text"}, - &cli.BoolFlag{Name: "no-local", Usage: "disable local relay when sending"}, - &cli.BoolFlag{Name: "no-multi", Usage: "disable multiplexing"}, - &cli.BoolFlag{Name: "git", Usage: "enable .gitignore respect / don't send ignored files"}, - &cli.IntFlag{Name: "port", Value: 9009, Usage: "base port for the relay"}, - &cli.IntFlag{Name: "transfers", Value: 4, Usage: "number of ports to use for transfers"}, - &cli.BoolFlag{Name: "qrcode", Aliases: []string{"qr"}, Usage: "show receive code as a qrcode"}, - &cli.StringFlag{Name: "exclude", Value: "", Usage: "exclude files if they contain any of the comma separated strings"}, - &cli.StringFlag{Name: "socks5", Value: "", Usage: "add a socks5 proxy", EnvVars: []string{"SOCKS5_PROXY"}}, - &cli.StringFlag{Name: "connect", Value: "", Usage: "add a http proxy", EnvVars: []string{"HTTP_PROXY"}}, + &cli.StringFlag{Name: "code", Aliases: []string{"c"}, Usage: "room codephrase (generated if not given)"}, }, HelpName: "share send", Action: send, @@ -76,9 +58,7 @@ func Run() (err error) { Action: relay, Flags: []cli.Flag{ &cli.StringFlag{Name: "host", Usage: "host of the relay"}, - &cli.StringFlag{Name: "ports", Value: "9009,9010,9011,9012,9013", Usage: "ports of the relay", EnvVars: []string{"SHARE_PORTS"}}, - &cli.IntFlag{Name: "port", Value: 9009, Usage: "base port for the relay", EnvVars: []string{"SHARE_PORT"}}, - &cli.IntFlag{Name: "transfers", Value: 5, Usage: "number of ports to use for relay"}, + &cli.IntFlag{Name: "port", Value: 9009, Usage: "port of the relay", EnvVars: []string{"SHARE_PORT"}}, }, }, } @@ -104,6 +84,7 @@ func Run() (err error) { &cli.StringFlag{Name: "relay", Value: models.DEFAULT_RELAY, Usage: "address of the relay", EnvVars: []string{"SHARE_RELAY"}}, &cli.StringFlag{Name: "relay6", Value: models.DEFAULT_RELAY6, Usage: "ipv6 address of the relay", EnvVars: []string{"SHARE_RELAY6"}}, &cli.StringFlag{Name: "out", Value: ".", Usage: "specify an output folder to receive the file"}, + &cli.BoolFlag{Name: "one-time", Usage: "exit once everything currently in the room is downloaded, instead of staying to help reseed it"}, &cli.StringFlag{Name: "socks5", Value: "", Usage: "add a socks5 proxy", EnvVars: []string{"SOCKS5_PROXY"}}, &cli.StringFlag{Name: "connect", Value: "", Usage: "add a http proxy", EnvVars: []string{"HTTP_PROXY"}}, &cli.StringFlag{Name: "throttleUpload", Value: "", Usage: "throttle the upload speed e.g. 500k"}, @@ -129,28 +110,44 @@ func setDebugLevel(c *cli.Context) { } } -func parseRelayPorts(portsFlag string) []string { - var ports []string - for _, p := range strings.Split(portsFlag, ",") { - if p = strings.TrimSpace(p); p != "" { - ports = append(ports, p) - } - } - return ports -} - func send(c *cli.Context) (err error) { setDebugLevel(c) - fmt.Println("Send functionality not yet implemented in this fork.") - fmt.Println("This is a placeholder for the send command.") - return nil + files := c.Args().Slice() + if len(files) == 0 { + return fmt.Errorf("no files specified; usage: share send file1 [file2 ...]") + } + code := c.String("code") + if code == "" { + code = utils.GetRandomName() + } + fmt.Printf("Code is: %s\n", code) + fmt.Println("On the other computer, run:") + fmt.Printf("\tshare %s\n", code) + + return swarm.Run(swarm.Options{ + RelayAddress: c.String("relay"), + Code: code, + Files: files, + }) } func receive(c *cli.Context) (err error) { setDebugLevel(c) - fmt.Println("Receive functionality not yet implemented in this fork.") - fmt.Println("This is a placeholder for the receive command.") - return nil + code := c.Args().First() + if code == "" { + return cli.ShowAppHelp(c) + } + outDir := c.String("out") + if outDir == "" { + outDir = "." + } + + return swarm.Run(swarm.Options{ + RelayAddress: c.String("relay"), + Code: code, + OutDir: outDir, + OneTime: c.Bool("one-time"), + }) } func relay(c *cli.Context) (err error) { @@ -160,35 +157,9 @@ func relay(c *cli.Context) (err error) { debugString = "debug" } host := c.String("host") - var ports []string - - if c.IsSet("ports") { - ports = parseRelayPorts(c.String("ports")) - } else { - portString := c.Int("port") - if portString == 0 { - portString = 9009 - } - transfersString := c.Int("transfers") - if transfersString == 0 { - transfersString = 4 - } - ports = make([]string, transfersString) - for i := range ports { - ports[i] = strconv.Itoa(portString + i) - } + port := c.Int("port") + if port == 0 { + port = 9009 } - if len(ports) < 2 { - return fmt.Errorf("relay requires at least two ports; specify --ports with two or more ports or set --transfers to 2+") - } - - for _, port := range ports[1:] { - go func(portStr string) { - err := tcp.Run(debugString, host, portStr) - if err != nil { - panic(err) - } - }(port) - } - return tcp.Run(debugString, host, ports[0]) + return tcp.Run(debugString, host, fmt.Sprintf("%d", port)) } diff --git a/src/swarm/chunks.go b/src/swarm/chunks.go new file mode 100644 index 0000000..ec89b78 --- /dev/null +++ b/src/swarm/chunks.go @@ -0,0 +1,160 @@ +package swarm + +import ( + "bytes" + "io" + "os" + "sync" + + "github.com/tzgyn/share/src/utils" +) + +// ChunkSize is fixed at 1MB: large enough to keep protocol overhead low, +// small enough that resuming a transfer never re-fetches much. +const ChunkSize = 1 << 20 + +// NumChunks returns how many ChunkSize pieces a file of size bytes has. +func NumChunks(size int64) int64 { + if size <= 0 { + return 0 + } + return (size + ChunkSize - 1) / ChunkSize +} + +func chunkLen(size, offset int64) int64 { + if offset+ChunkSize > size { + return size - offset + } + return ChunkSize +} + +// ChunkStore tracks on-disk chunk presence for one file in the swarm and +// reads/writes it at chunk granularity. +type ChunkStore struct { + mu sync.Mutex + path string + size int64 + have map[int64]bool // chunk offset -> present +} + +// OpenChunkStore opens (or creates) path as a file of exactly size bytes +// for receiving into, and figures out which chunks are already there. This +// reuses the same "a chunk-sized block that's still all zero bytes hasn't +// arrived yet" scan croc itself uses, so resuming an interrupted transfer +// needs no separate manifest of what's already done. +func OpenChunkStore(path string, size int64) (*ChunkStore, error) { + info, statErr := os.Stat(path) + fresh := statErr != nil || info.Size() != size + + f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o644) + if err != nil { + return nil, err + } + if fresh { + err = f.Truncate(size) + } + f.Close() + if err != nil { + return nil, err + } + + cs := &ChunkStore{path: path, size: size, have: make(map[int64]bool)} + if !fresh { + if err := cs.scanExisting(); err != nil { + return nil, err + } + } + return cs, nil +} + +// NewSeedChunkStore wraps an already-complete local file being shared into +// the swarm: no scanning, every chunk is simply present. The file is never +// truncated or modified. +func NewSeedChunkStore(path string, size int64) *ChunkStore { + cs := &ChunkStore{path: path, size: size, have: make(map[int64]bool)} + for offset := int64(0); offset < size; offset += ChunkSize { + cs.have[offset] = true + } + return cs +} + +func (cs *ChunkStore) scanExisting() error { + f, err := os.Open(cs.path) + if err != nil { + return err + } + defer f.Close() + + zero := make([]byte, ChunkSize) + buf := make([]byte, ChunkSize) + for offset := int64(0); offset < cs.size; offset += ChunkSize { + l := chunkLen(cs.size, offset) + if _, err := f.ReadAt(buf[:l], offset); err != nil && err != io.EOF { + return err + } + if !bytes.Equal(buf[:l], zero[:l]) { + cs.have[offset] = true + } + } + return nil +} + +func (cs *ChunkStore) HaveOffsets() []int64 { + cs.mu.Lock() + defer cs.mu.Unlock() + list := make([]int64, 0, len(cs.have)) + for offset := range cs.have { + list = append(list, offset) + } + return list +} + +func (cs *ChunkStore) HasChunk(offset int64) bool { + cs.mu.Lock() + defer cs.mu.Unlock() + return cs.have[offset] +} + +func (cs *ChunkStore) Complete() bool { + cs.mu.Lock() + defer cs.mu.Unlock() + return int64(len(cs.have)) == NumChunks(cs.size) +} + +func (cs *ChunkStore) ReadChunk(offset int64) ([]byte, error) { + f, err := os.Open(cs.path) + if err != nil { + return nil, err + } + defer f.Close() + buf := make([]byte, chunkLen(cs.size, offset)) + if _, err := f.ReadAt(buf, offset); err != nil && err != io.EOF { + return nil, err + } + return buf, nil +} + +func (cs *ChunkStore) WriteChunk(offset int64, data []byte) error { + f, err := os.OpenFile(cs.path, os.O_WRONLY, 0o644) + if err != nil { + return err + } + defer f.Close() + if _, err := f.WriteAt(data, offset); err != nil { + return err + } + cs.mu.Lock() + cs.have[offset] = true + cs.mu.Unlock() + return nil +} + +// VerifyHash confirms a completed file matches the expected whole-file +// hash from the manifest. +func (cs *ChunkStore) VerifyHash(expected []byte) (bool, error) { + got, err := utils.HashFile(cs.path, "xxhash") + if err != nil { + return false, err + } + return bytes.Equal(got, expected), nil +} diff --git a/src/swarm/client.go b/src/swarm/client.go new file mode 100644 index 0000000..2ca3ef1 --- /dev/null +++ b/src/swarm/client.go @@ -0,0 +1,390 @@ +package swarm + +import ( + "crypto/rand" + "encoding/hex" + "fmt" + "net" + "os" + "path/filepath" + "strconv" + "sync" + "time" + + log "github.com/schollz/logger" + + "github.com/tzgyn/share/src/comm" + "github.com/tzgyn/share/src/tcp" + "github.com/tzgyn/share/src/utils" +) + +// Options configures one participant in a swarm room. Everyone who joins +// with the same Code can both send and receive: peers who pass Files seed +// them into the room, and every peer (seeding or not) helps re-share +// whatever chunks it has already received to others who join later. +type Options struct { + RelayAddress string + Code string // room name and PAKE codephrase, both at once + Files []string // local files to seed; empty means pure receiver + OutDir string // where received files are written + 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 +} + +type peerLink struct { + conn *comm.Comm + key []byte + remote string // remote address, for logging who's talking to whom +} + +func newPeerLink(c *comm.Comm, key []byte) *peerLink { + return &peerLink{conn: c, key: key, remote: c.Connection().RemoteAddr().String()} +} + +type client struct { + opts Options + selfID string + + mu sync.Mutex + files map[string]*sharedFile + pending map[string]map[int64]*peerLink // fileID -> offset -> link we asked + links map[*peerLink]bool + + dialedMu sync.Mutex + dialed map[string]bool // relay peer IDs already dialed, so we don't redial every peer-list update +} + +func newClient(opts Options) *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), + } +} + +func randomID() string { + b := make([]byte, 8) + if _, err := rand.Read(b); err != nil { + panic(err) + } + return hex.EncodeToString(b) +} + +func (cl *client) run() error { + for _, path := range cl.opts.Files { + if err := cl.addSeedFile(path); err != nil { + return err + } + } + + listener, listenAddr, err := startListener() + if err != nil { + return 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}) + if err != nil { + return fmt.Errorf("could not join room: %w", err) + } + defer relayConn.Close() + + for { + select { + case peers, ok := <-peerUpdates: + if !ok { + return fmt.Errorf("lost connection to relay") + } + for _, p := range peers { + if p.ID == cl.selfID || p.Addr == "" { + continue + } + cl.dialedMu.Lock() + already := cl.dialed[p.ID] + cl.dialed[p.ID] = true + cl.dialedMu.Unlock() + if !already { + go cl.dialPeer(p) + } + } + case <-time.After(500 * time.Millisecond): + if cl.opts.OneTime && cl.allKnownFilesComplete() { + return nil + } + } + } +} + +func (cl *client) addSeedFile(path string) error { + info, err := os.Stat(path) + if err != nil { + return err + } + if info.IsDir() { + return fmt.Errorf("%s is a directory; folder sharing isn't supported yet", path) + } + hash, err := utils.HashFile(path, "xxhash") + 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}, + store: NewSeedChunkStore(path, info.Size()), + } + log.Infof("seeding %s (%s)", name, utils.ByteCountDecimal(info.Size())) + return nil +} + +func (cl *client) allKnownFilesComplete() bool { + cl.mu.Lock() + defer cl.mu.Unlock() + if len(cl.files) == 0 { + return false + } + for _, f := range cl.files { + if !f.store.Complete() { + return false + } + } + return true +} + +func startListener() (net.Listener, string, error) { + l, err := net.Listen("tcp", ":0") + if err != nil { + return nil, "", err + } + port := l.Addr().(*net.TCPAddr).Port + return l, net.JoinHostPort(utils.LocalIP(), strconv.Itoa(port)), nil +} + +func (cl *client) acceptLoop(l net.Listener) { + for { + conn, err := l.Accept() + if err != nil { + return + } + go func(conn net.Conn) { + c := comm.New(conn) + key, err := handshake(c, cl.opts.Code, false) + if err != nil { + log.Debugf("peer handshake failed: %v", err) + conn.Close() + return + } + cl.handlePeerLink(newPeerLink(c, key)) + }(conn) + } +} + +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) + return + } + c := comm.New(conn) + key, err := handshake(c, cl.opts.Code, true) + if err != nil { + log.Debugf("handshake with peer %s failed: %v", p.ID, err) + conn.Close() + return + } + cl.handlePeerLink(newPeerLink(c, key)) +} + +func (cl *client) handlePeerLink(link *peerLink) { + cl.mu.Lock() + cl.links[link] = true + cl.mu.Unlock() + log.Infof("connected to peer %s", link.remote) + defer func() { + cl.mu.Lock() + delete(cl.links, link) + // free up anything only this link was asked for, so others can serve it + for _, byOffset := range cl.pending { + for offset, l := range byOffset { + if l == link { + delete(byOffset, offset) + } + } + } + cl.mu.Unlock() + link.conn.Close() + log.Infof("disconnected from peer %s", link.remote) + }() + + if err := cl.sendManifest(link); err != nil { + log.Debugf("send manifest failed: %v", err) + return + } + + for { + msg, err := receiveMessage(link.conn, link.key) + if err != nil { + log.Debugf("peer link closed: %v", err) + return + } + if err := cl.handleMessage(link, msg); err != nil { + log.Debugf("handling %s message: %v", msg.Type, err) + } + } +} + +func (cl *client) sendManifest(link *peerLink) error { + cl.mu.Lock() + files := make([]FileManifestEntry, 0, len(cl.files)) + for _, f := range cl.files { + e := f.entry + e.Have = f.store.HaveOffsets() + files = append(files, e) + } + cl.mu.Unlock() + return sendMessage(link.conn, link.key, Message{Type: TypeManifest, Files: files}) +} + +func (cl *client) handleMessage(link *peerLink, msg Message) error { + switch msg.Type { + case TypeManifest: + for _, entry := range msg.Files { + cl.learnFile(entry) + cl.considerRequests(entry.ID, entry.Have, link) + } + case TypeHave: + cl.considerRequests(msg.FileID, []int64{msg.Offset}, link) + case TypeRequest: + return cl.serveRequest(link, msg.FileID, msg.Offset) + case TypeChunk: + return cl.receiveChunk(link, msg.FileID, msg.Offset, msg.Data) + } + return nil +} + +func (cl *client) learnFile(entry FileManifestEntry) { + cl.mu.Lock() + _, known := cl.files[entry.ID] + cl.mu.Unlock() + if known { + return + } + path := filepath.Join(cl.opts.OutDir, entry.Name) + store, err := OpenChunkStore(path, entry.Size) + if err != nil { + log.Errorf("could not open %s for receiving: %v", path, err) + return + } + cl.mu.Lock() + cl.files[entry.ID] = &sharedFile{ + entry: FileManifestEntry{ID: entry.ID, Name: entry.Name, Size: entry.Size, Hash: entry.Hash}, + store: store, + } + cl.mu.Unlock() + log.Infof("new file in room: %s (%s)", entry.Name, utils.ByteCountDecimal(entry.Size)) +} + +// 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) { + cl.mu.Lock() + f, ok := cl.files[fileID] + if !ok { + cl.mu.Unlock() + return + } + if cl.pending[fileID] == nil { + cl.pending[fileID] = make(map[int64]*peerLink) + } + toRequest := make([]int64, 0, len(offsets)) + for _, off := range offsets { + if f.store.HasChunk(off) { + continue + } + if _, already := cl.pending[fileID][off]; already { + continue + } + cl.pending[fileID][off] = link + toRequest = append(toRequest, off) + } + cl.mu.Unlock() + + for _, off := range toRequest { + if err := sendMessage(link.conn, link.key, Message{Type: TypeRequest, FileID: fileID, Offset: off}); err != nil { + log.Debugf("request failed: %v", err) + return + } + } +} + +func (cl *client) serveRequest(link *peerLink, fileID string, offset int64) error { + cl.mu.Lock() + f, ok := cl.files[fileID] + cl.mu.Unlock() + if !ok || !f.store.HasChunk(offset) { + return nil + } + data, err := f.store.ReadChunk(offset) + if err != nil { + return err + } + log.Infof("sending %s chunk %d/%d to %s", f.entry.Name, offset/ChunkSize+1, NumChunks(f.entry.Size), link.remote) + 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 { + cl.mu.Lock() + f, ok := cl.files[fileID] + if ok { + delete(cl.pending[fileID], offset) + } + links := make([]*peerLink, 0, len(cl.links)) + for l := range cl.links { + links = append(links, l) + } + cl.mu.Unlock() + if !ok { + 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 + } + + // let everyone else in the swarm know we can now serve this chunk too + for _, l := range links { + go func(l *peerLink) { + _ = sendMessage(l.conn, l.key, Message{Type: TypeHave, FileID: fileID, Offset: offset}) + }(l) + } + + if f.store.Complete() { + verified, err := f.store.VerifyHash(f.entry.Hash) + if err != nil { + log.Errorf("could not verify %s: %v", f.entry.Name, err) + } else if !verified { + log.Errorf("%s failed hash verification after transfer", f.entry.Name) + } else { + log.Infof("%s complete and verified", f.entry.Name) + } + } + return nil +} diff --git a/src/swarm/handshake.go b/src/swarm/handshake.go new file mode 100644 index 0000000..e80be60 --- /dev/null +++ b/src/swarm/handshake.go @@ -0,0 +1,49 @@ +package swarm + +import ( + "github.com/schollz/pake/v3" + + "github.com/tzgyn/share/src/comm" +) + +// handshake runs a PAKE key exchange over c using the room's codephrase as +// the shared weak secret, deriving a session key for this one link that +// only two peers who know the codephrase can arrive at. Exactly one side +// of any given link must pass initiator=true — the side that dialed out. +// The relay is never involved in this exchange and never sees the result. +func handshake(c *comm.Comm, codephrase string, initiator bool) (key []byte, err error) { + role := 1 + if initiator { + role = 0 + } + p, err := pake.InitCurve([]byte(codephrase), role, "siec") + if err != nil { + return nil, err + } + + if initiator { + if err = c.Send(p.Bytes()); err != nil { + return nil, err + } + peerBytes, err := c.Receive() + if err != nil { + return nil, err + } + if err = p.Update(peerBytes); err != nil { + return nil, err + } + } else { + peerBytes, err := c.Receive() + if err != nil { + return nil, err + } + if err = p.Update(peerBytes); err != nil { + return nil, err + } + if err = c.Send(p.Bytes()); err != nil { + return nil, err + } + } + + return p.SessionKey() +} diff --git a/src/swarm/message.go b/src/swarm/message.go new file mode 100644 index 0000000..b263be0 --- /dev/null +++ b/src/swarm/message.go @@ -0,0 +1,69 @@ +package swarm + +import ( + "encoding/json" + + "github.com/tzgyn/share/src/comm" + "github.com/tzgyn/share/src/crypt" +) + +// Type is the entire vocabulary peers speak to each other once directly +// connected and past the handshake. +type Type string + +const ( + // TypeManifest introduces what a peer has: known files and, per file, + // which chunk offsets it already holds. + TypeManifest Type = "manifest" + // TypeHave announces a single newly-finished chunk for one file. + TypeHave Type = "have" + // TypeRequest asks a peer to send one chunk of one file. + TypeRequest Type = "request" + // TypeChunk carries the chunk payload requested. + TypeChunk Type = "chunk" +) + +// FileManifestEntry describes one file known to the swarm: its identity, +// size, whole-file hash (for final verification), and which chunk offsets +// the sender of this manifest currently holds. +type FileManifestEntry struct { + ID string `json:"id"` + Name string `json:"name"` + Size int64 `json:"size"` + Hash []byte `json:"hash"` + Have []int64 `json:"have"` +} + +// Message is what flows over an established, encrypted peer link. +type Message struct { + Type Type `json:"type"` + Files []FileManifestEntry `json:"files,omitempty"` // TypeManifest + FileID string `json:"fileId,omitempty"` + Offset int64 `json:"offset,omitempty"` + Data []byte `json:"data,omitempty"` // TypeChunk +} + +func sendMessage(c *comm.Comm, key []byte, m Message) error { + b, err := json.Marshal(m) + if err != nil { + return err + } + enc, err := crypt.Encrypt(b, key) + if err != nil { + return err + } + return c.Send(enc) +} + +func receiveMessage(c *comm.Comm, key []byte) (m Message, err error) { + enc, err := c.Receive() + if err != nil { + return + } + b, err := crypt.Decrypt(enc, key) + if err != nil { + return + } + err = json.Unmarshal(b, &m) + return +} diff --git a/src/tcp/defaults.go b/src/tcp/defaults.go index 27e23c8..3c5cd5d 100644 --- a/src/tcp/defaults.go +++ b/src/tcp/defaults.go @@ -1,9 +1,5 @@ package tcp -import "time" - const ( - DEFAULT_LOG_LEVEL = "debug" - DEFAULT_ROOM_CLEANUP_INTERVAL = 10 * time.Minute - DEFAULT_ROOM_TTL = 3 * time.Hour + DEFAULT_LOG_LEVEL = "debug" ) diff --git a/src/tcp/options.go b/src/tcp/options.go index bfa1be9..eae45c0 100644 --- a/src/tcp/options.go +++ b/src/tcp/options.go @@ -2,7 +2,6 @@ package tcp import ( "fmt" - "time" ) var availableLogLevels = []string{"info", "error", "warn", "debug", "trace"} @@ -19,20 +18,6 @@ func WithLogLevel(level string) serverOptsFunc { } } -func WithRoomCleanupInterval(interval time.Duration) serverOptsFunc { - return func(s *server) error { - s.roomCleanupInterval = interval - return nil - } -} - -func WithRoomTTL(ttl time.Duration) serverOptsFunc { - return func(s *server) error { - s.roomTTL = ttl - return nil - } -} - func containsSlice(s []string, e string) bool { for _, ss := range s { if e == ss { diff --git a/src/tcp/tcp.go b/src/tcp/tcp.go index 4456182..82da94f 100644 --- a/src/tcp/tcp.go +++ b/src/tcp/tcp.go @@ -1,8 +1,8 @@ package tcp import ( - "bytes" "context" + "encoding/json" "fmt" "net" "strings" @@ -12,43 +12,53 @@ import ( log "github.com/schollz/logger" "github.com/tzgyn/share/src/comm" - "github.com/tzgyn/share/src/models" ) -// server is a minimal relay: it has no notion of a password. Its only job -// is to pair two connections that show up with the same room name and then -// pipe raw bytes between them. Encrypting what flows through that pipe is -// the clients' responsibility, not the relay's. +// PeerInfo is what the relay tells other room members about one peer: +// an identifier and the address others can dial to reach it directly +// (empty if this peer can't accept incoming connections, e.g. behind NAT). +type PeerInfo struct { + ID string `json:"id"` + Addr string `json:"addr"` +} + +// ControlMessage is the entire vocabulary spoken between a peer and the +// relay. A peer sends exactly one "join" to announce itself; the relay +// pushes a "peers" message with the current room membership whenever it +// changes. The relay never sees file content or chunk data at all — peers +// exchange that directly with each other once they have each other's Addr. +type ControlMessage struct { + Type string `json:"type"` // "join" or "peers" + Room string `json:"room,omitempty"` + Peer PeerInfo `json:"peer,omitempty"` + Peers []PeerInfo `json:"peers,omitempty"` +} + type server struct { host string port string debugLevel string rooms roomMap - roomCleanupInterval time.Duration - roomTTL time.Duration - *stop } +type peerEntry struct { + info PeerInfo + conn *comm.Comm +} + type roomInfo struct { - first *comm.Comm - second *comm.Comm - opened time.Time - full bool + peers map[string]*peerEntry } type roomMap struct { - rooms map[string]roomInfo + rooms map[string]*roomInfo sync.Mutex } -const pingRoom = "pinglkasjdlfjsaldjf" - func newDefaultServer() *server { s := new(server) - s.roomCleanupInterval = DEFAULT_ROOM_CLEANUP_INTERVAL - s.roomTTL = DEFAULT_ROOM_TTL s.debugLevel = DEFAULT_LOG_LEVEL s.stop = newStop(context.Background()) return s @@ -76,14 +86,9 @@ func (s *server) start() (err error) { log.SetLevel(s.debugLevel) s.rooms.Lock() - s.rooms.rooms = make(map[string]roomInfo) + s.rooms.rooms = make(map[string]*roomInfo) s.rooms.Unlock() - s.stop.wg.Add(1) - go func() { - defer s.stop.wg.Done() - s.deleteOldRooms() - }() defer s.stop.Cancel() if s.stop.gui { defer s.stop.wg.Wait() @@ -139,7 +144,6 @@ func (s *server) run() (err error) { } }() - // spawn a new goroutine whenever a client connects for { connection, err := s.stop.server.Accept() if err != nil { @@ -149,274 +153,97 @@ func (s *server) run() (err error) { s.stop.wg.Add(1) go func(connection net.Conn) { defer s.stop.wg.Done() - c := comm.New(connection) - room, errCommunication := s.clientCommunication(c) - if errCommunication != nil { - log.Debugf("relay-%s: %s", connection.RemoteAddr().String(), errCommunication.Error()) - connection.Close() - return - } - if room == pingRoom { - log.Debugf("got ping") - connection.Close() - return - } - ticker := time.NewTicker(1 * time.Second) - defer ticker.Stop() - for { - // check connection - deleteIt := false - s.rooms.Lock() - roomData, ok := s.rooms.rooms[room] - if !ok { - log.Debug("room is gone") - s.rooms.Unlock() - return - } - if roomData.first != nil && roomData.second != nil { - log.Debug("rooms ready") - s.rooms.Unlock() - break - } - if roomData.first != nil { - if errSend := roomData.first.Send([]byte{1}); errSend != nil { - log.Debug(errSend) - deleteIt = true - } - } - s.rooms.Unlock() - if deleteIt { - s.deleteRoom(room) - break - } - select { - case <-s.stop.ctx.Done(): - s.deleteRoom(room) - return - case <-ticker.C: - } - } + s.handleConnection(connection) }(connection) } } -// deleteOldRooms checks for rooms at a regular interval and removes those -// that have exceeded their allocated TTL. -func (s *server) deleteOldRooms() { - ticker := time.NewTicker(s.roomCleanupInterval) - defer func() { - ticker.Stop() - log.Debug("room cleanup stopped") - }() - for next := true; next; { - roomsToDelete := []string{} - select { - case <-ticker.C: - s.rooms.Lock() - for room, roomData := range s.rooms.rooms { - if time.Since(roomData.opened) > s.roomTTL { - roomsToDelete = append(roomsToDelete, room) - } - } - s.rooms.Unlock() - case <-s.stop.ctx.Done(): - if s.server != nil { - log.Debugf("stop TCP server on %s", s.server.Addr()) - s.server.Close() - time.Sleep(time.Millisecond) - } - log.Debug("stop room cleanup fired") - s.rooms.Lock() - for room := range s.rooms.rooms { - roomsToDelete = append(roomsToDelete, room) - } - s.rooms.Unlock() - next = false - } - for _, room := range roomsToDelete { - s.deleteRoom(room) - log.Debugf("room cleaned up: %s", room) - } - } -} +// handleConnection is the entire relay protocol: read one join message, +// register the peer in its room, broadcast membership, then just wait for +// the connection to close so we know when to remove the peer again. No +// data ever flows through here beyond membership control messages. +func (s *server) handleConnection(connection net.Conn) { + c := comm.New(connection) + defer connection.Close() -// clientCommunication is the entire relay protocol: a client sends the room -// name it wants to join (or "ping" to check liveness), and the relay either -// parks it as the first occupant or, once a second client shows up, wires -// the two connections together and gets out of the way. -func (s *server) clientCommunication(c *comm.Comm) (room string, err error) { - roomBytes, err := c.Receive() - if err != nil { - return - } - if bytes.Equal(roomBytes, []byte("ping")) { - room = pingRoom - log.Debug("sending back pong") - c.Send([]byte("pong")) - return - } - room = string(roomBytes) - - s.rooms.Lock() - // create the room if it is new - if _, ok := s.rooms.rooms[room]; !ok { - s.rooms.rooms[room] = roomInfo{ - first: c, - opened: time.Now(), - } - s.rooms.Unlock() - err = c.Send([]byte("ok|||" + c.Connection().RemoteAddr().String())) - if err != nil { - log.Error(err) - s.deleteRoom(room) - return - } - log.Debugf("room %s has 1", room) - return - } - if s.rooms.rooms[room].full { - s.rooms.Unlock() - if err = c.Send([]byte("room full")); err != nil { - log.Error(err) - } - return - } - log.Debugf("room %s has 2", room) - s.rooms.rooms[room] = roomInfo{ - first: s.rooms.rooms[room].first, - second: c, - opened: s.rooms.rooms[room].opened, - full: true, - } - otherConnection := s.rooms.rooms[room].first - s.rooms.Unlock() - - // second connection is the sender, time to staple connections - var wg sync.WaitGroup - wg.Add(1) - - // start piping - go func(com1, com2 *comm.Comm, wg *sync.WaitGroup) { - log.Debug("starting pipes") - pipe(com1.Connection(), com2.Connection()) - wg.Done() - log.Debug("done piping") - }(otherConnection, c, &wg) - - // tell the joiner everything is ready - err = c.Send([]byte("ok|||" + c.Connection().RemoteAddr().String())) - if err != nil { - s.deleteRoom(room) - return - } - wg.Wait() - - // delete room - s.deleteRoom(room) - return -} - -func (s *server) deleteRoom(room string) { - s.rooms.Lock() - defer s.rooms.Unlock() - roomData, ok := s.rooms.rooms[room] - if !ok { - return - } - log.Debugf("deleting room: %s", room) - if roomData.first != nil { - roomData.first.Close() - } - if roomData.second != nil { - roomData.second.Close() - } - delete(s.rooms.rooms, room) -} - -// chanFromConn creates a channel from a Conn object, and sends everything it -// Read()s from the socket to the channel. -func chanFromConn(conn net.Conn) chan []byte { - c := make(chan []byte, 1) - if err := conn.SetReadDeadline(time.Now().Add(3 * time.Hour)); err != nil { - log.Warnf("can't set read deadline: %v", err) - } - - go func() { - b := make([]byte, models.TCP_BUFFER_SIZE) - for { - n, err := conn.Read(b) - if n > 0 { - res := make([]byte, n) - // Copy the buffer so it doesn't get changed while read by the recipient. - copy(res, b[:n]) - c <- res - } - if err != nil { - log.Debug(err) - c <- nil - break - } - } - log.Debug("exiting") - }() - - return c -} - -// pipe creates a full-duplex pipe between the two sockets and -// transfers data from one to the other. -func pipe(conn1 net.Conn, conn2 net.Conn) { - chan1 := chanFromConn(conn1) - chan2 := chanFromConn(conn2) - - for { - select { - case b1 := <-chan1: - if b1 == nil { - return - } - if _, err := conn2.Write(b1); err != nil { - log.Errorf("write error on channel 1: %v", err) - } - - case b2 := <-chan2: - if b2 == nil { - return - } - if _, err := conn1.Write(b2); err != nil { - log.Errorf("write error on channel 2: %v", err) - } - } - } -} - -func PingServer(address string) (err error) { - log.Debugf("pinging %s", address) - c, err := comm.NewConnection(address, 300*time.Millisecond) - if err != nil { - log.Debug(err) - return - } - err = c.Send([]byte("ping")) - if err != nil { - log.Debug(err) - return - } b, err := c.Receive() if err != nil { log.Debug(err) return } - if bytes.Equal(b, []byte("pong")) { - return nil + var msg ControlMessage + if err := json.Unmarshal(b, &msg); err != nil || msg.Type != "join" || msg.Room == "" || msg.Peer.ID == "" { + log.Debugf("bad join message: %v", err) + return + } + log.Debugf("peer %s joining room %s", msg.Peer.ID, msg.Room) + + s.rooms.Lock() + r, ok := s.rooms.rooms[msg.Room] + if !ok { + r = &roomInfo{peers: make(map[string]*peerEntry)} + s.rooms.rooms[msg.Room] = r + } + r.peers[msg.Peer.ID] = &peerEntry{info: msg.Peer, conn: c} + s.rooms.Unlock() + + s.broadcastPeers(msg.Room) + defer func() { + s.rooms.Lock() + if r, ok := s.rooms.rooms[msg.Room]; ok { + delete(r.peers, msg.Peer.ID) + if len(r.peers) == 0 { + delete(s.rooms.rooms, msg.Room) + } + } + s.rooms.Unlock() + s.broadcastPeers(msg.Room) + }() + + // nothing more to read from a peer; block here until it disconnects + for { + if _, err := c.Receive(); err != nil { + log.Debugf("peer %s left room %s: %v", msg.Peer.ID, msg.Room, err) + return + } } - return fmt.Errorf("no pong") } -// ConnectToTCPServer will initiate a new connection to the specified -// address and join the given room, no password required. Once the room -// pairs up, the returned *comm.Comm is a raw pipe to the other peer. -func ConnectToTCPServer(address, room string, timelimit ...time.Duration) (c *comm.Comm, ipaddr string, err error) { +func (s *server) broadcastPeers(room string) { + s.rooms.Lock() + r, ok := s.rooms.rooms[room] + var peers []PeerInfo + var conns []*comm.Comm + if ok { + peers = make([]PeerInfo, 0, len(r.peers)) + conns = make([]*comm.Comm, 0, len(r.peers)) + for _, p := range r.peers { + peers = append(peers, p.info) + conns = append(conns, p.conn) + } + } + s.rooms.Unlock() + if !ok { + return + } + + b, err := json.Marshal(ControlMessage{Type: "peers", Peers: peers}) + if err != nil { + log.Error(err) + return + } + for _, c := range conns { + if err := c.Send(b); err != nil { + log.Debug(err) + } + } +} + +// JoinRoom connects to the relay and announces self in room. The returned +// channel always holds the most recent peer-list broadcast (it never +// blocks the relay's reader goroutine and never queues up stale updates). +// Closing c leaves the room. +func JoinRoom(address, room string, self PeerInfo, timelimit ...time.Duration) (c *comm.Comm, peers <-chan []PeerInfo, err error) { if len(timelimit) > 0 { c, err = comm.NewConnection(address, timelimit[0]) } else { @@ -427,25 +254,36 @@ func ConnectToTCPServer(address, room string, timelimit ...time.Duration) (c *co return } - log.Debugf("joining room: %s", room) - if err = c.Send([]byte(room)); err != nil { + b, err := json.Marshal(ControlMessage{Type: "join", Room: room, Peer: self}) + if err != nil { + return + } + if err = c.Send(b); err != nil { log.Debug(err) return } - data, err := c.Receive() - if err != nil { - log.Debug(err) - return - } - if !bytes.HasPrefix(data, []byte("ok")) { - err = fmt.Errorf("got bad response: %s", data) - log.Debug(err) - return - } - if parts := strings.SplitN(string(data), "|||", 2); len(parts) == 2 { - ipaddr = parts[1] - } - log.Debug("all set") + ch := make(chan []PeerInfo, 1) + go func() { + defer close(ch) + for { + data, err := c.Receive() + if err != nil { + log.Debugf("relay connection closed: %v", err) + return + } + var msg ControlMessage + if err := json.Unmarshal(data, &msg); err != nil || msg.Type != "peers" { + continue + } + // keep only the latest peer list in the channel + select { + case <-ch: + default: + } + ch <- msg.Peers + } + }() + peers = ch return }