diff --git a/share b/share index 6c09780..0d60016 100755 Binary files a/share and b/share differ diff --git a/src/cli/cli.go b/src/cli/cli.go index 50cba4e..528804e 100644 --- a/src/cli/cli.go +++ b/src/cli/cli.go @@ -34,12 +34,16 @@ USAGE EXAMPLES: Join a room to receive whatever's shared there (and help re-share it): share secret-code + See what rooms are currently active: + share status + 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) + status show the rooms the background daemon is currently in stop stop the background daemon and leave every room daemon run the background daemon in the foreground (mostly for debugging) @@ -77,6 +81,8 @@ func Run() error { return runRelay(splice(args, i)) case "stop": return runStop(splice(args, i)) + case "status": + return runStatus(splice(args, i)) case "daemon": return runDaemon(splice(args, i)) } @@ -244,6 +250,42 @@ func runStop(args []string) error { return nil } +func runStatus(args []string) error { + fs := flag.NewFlagSet("status", flag.ExitOnError) + fs.Usage = func() { + fmt.Fprintln(os.Stderr, "share status") + } + if err := fs.Parse(args); err != nil { + return err + } + + rooms, err := daemon.Status() + if err != nil { + fmt.Println(err) + return nil + } + if len(rooms) == 0 { + fmt.Println("No active rooms.") + return nil + } + + for _, r := range rooms { + peerWord := "peers" + if r.Peers == 1 { + peerWord = "peer" + } + fmt.Printf("Room %s (%d %s)\n", r.Code, r.Peers, peerWord) + for _, f := range r.Files { + state := "in progress" + if f.Complete { + state = "complete" + } + fmt.Printf(" %-24s %10s %d/%d chunks %s\n", f.Name, utils.ByteCountDecimal(f.Size), f.HaveChunks, f.TotalChunks, state) + } + } + return nil +} + func runDaemon(args []string) error { fs := flag.NewFlagSet("daemon", flag.ExitOnError) fs.Usage = func() { diff --git a/src/daemon/client.go b/src/daemon/client.go index 599d0ed..9a884d3 100644 --- a/src/daemon/client.go +++ b/src/daemon/client.go @@ -8,6 +8,8 @@ import ( "os/exec" "syscall" "time" + + "github.com/tzgyn/share/src/swarm" ) // IsRunning reports whether a daemon is currently listening on the @@ -71,7 +73,7 @@ func EnsureRunning(debug bool) error { // 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{ + _, err := call(request{ Action: "join", Join: &joinRequest{ RelayAddress: relayAddress, @@ -81,6 +83,7 @@ func Join(relayAddress, code string, files []string, outDir string, oneTime bool OneTime: oneTime, }, }) + return err } // Stop tells the running daemon to leave every room it's in and exit. @@ -88,29 +91,43 @@ func Stop() error { if !IsRunning() { return fmt.Errorf("daemon is not running") } - return call(request{Action: "stop"}) + _, err := call(request{Action: "stop"}) + return err } -func call(req request) error { +// Status asks the running daemon which rooms it's currently in and how +// each one's transfers are progressing. +func Status() ([]swarm.RoomStatus, error) { + if !IsRunning() { + return nil, fmt.Errorf("daemon is not running") + } + resp, err := call(request{Action: "status"}) + if err != nil { + return nil, err + } + return resp.Rooms, nil +} + +func call(req request) (response, error) { path, err := socketPath() if err != nil { - return err + return response{}, err } conn, err := net.DialTimeout("unix", path, 2*time.Second) if err != nil { - return fmt.Errorf("could not reach daemon: %w", err) + return response{}, fmt.Errorf("could not reach daemon: %w", err) } defer conn.Close() if err := json.NewEncoder(conn).Encode(req); err != nil { - return err + return response{}, err } var resp response if err := json.NewDecoder(conn).Decode(&resp); err != nil { - return err + return response{}, err } if !resp.OK { - return fmt.Errorf("%s", resp.Error) + return resp, fmt.Errorf("%s", resp.Error) } - return nil + return resp, nil } diff --git a/src/daemon/protocol.go b/src/daemon/protocol.go index 5daabd9..894941f 100644 --- a/src/daemon/protocol.go +++ b/src/daemon/protocol.go @@ -1,9 +1,11 @@ package daemon +import "github.com/tzgyn/share/src/swarm" + // 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" + Action string `json:"action"` // "join", "stop", or "status" Join *joinRequest `json:"join,omitempty"` } @@ -20,6 +22,7 @@ type joinRequest struct { } type response struct { - OK bool `json:"ok"` - Error string `json:"error,omitempty"` + OK bool `json:"ok"` + Error string `json:"error,omitempty"` + Rooms []swarm.RoomStatus `json:"rooms,omitempty"` // "status" only } diff --git a/src/daemon/server.go b/src/daemon/server.go index c6ba780..943e888 100644 --- a/src/daemon/server.go +++ b/src/daemon/server.go @@ -81,6 +81,9 @@ func (s *server) handleConn(conn net.Conn) (shutdown bool) { case "stop": respond(conn, response{OK: true}) return true + case "status": + respond(conn, response{OK: true, Rooms: s.status()}) + return false default: respond(conn, response{OK: false, Error: fmt.Sprintf("unknown action %q", req.Action)}) return false @@ -140,3 +143,13 @@ func (s *server) stopAll() { cl.Stop() } } + +func (s *server) status() []swarm.RoomStatus { + s.mu.Lock() + defer s.mu.Unlock() + statuses := make([]swarm.RoomStatus, 0, len(s.rooms)) + for _, cl := range s.rooms { + statuses = append(statuses, cl.Status()) + } + return statuses +} diff --git a/src/swarm/client.go b/src/swarm/client.go index a4a6966..bac4904 100644 --- a/src/swarm/client.go +++ b/src/swarm/client.go @@ -134,6 +134,44 @@ func (cl *Client) Wait() error { return <-cl.done } +// FileStatus summarizes one file's transfer progress within a room. +type FileStatus struct { + Name string + Size int64 + HaveChunks int64 + TotalChunks int64 + Complete bool +} + +// RoomStatus summarizes one room this client is participating in. +type RoomStatus struct { + Code string + Peers int + Files []FileStatus +} + +// Status reports this room's current code, peer count, and per-file +// transfer progress. +func (cl *Client) Status() RoomStatus { + cl.mu.Lock() + defer cl.mu.Unlock() + files := make([]FileStatus, 0, len(cl.files)) + for _, f := range cl.files { + files = append(files, FileStatus{ + Name: f.entry.Name, + Size: f.entry.Size, + HaveChunks: int64(len(f.store.HaveOffsets())), + TotalChunks: NumChunks(f.entry.Size), + Complete: f.store.Complete(), + }) + } + return RoomStatus{ + Code: cl.opts.Code, + Peers: len(cl.links), + Files: files, + } +} + // Stop ends this room's participation. Wait will then return nil. func (cl *Client) Stop() { cl.stopOnce.Do(func() { close(cl.stopCh) }) @@ -154,7 +192,7 @@ func (cl *Client) loop() { return } for _, p := range peers { - if p.ID == cl.selfID || p.Addr == "" { + if p.ID == cl.selfID { continue } cl.dialedMu.Lock() @@ -162,7 +200,7 @@ func (cl *Client) loop() { cl.dialed[p.ID] = true cl.dialedMu.Unlock() if !already { - go cl.dialPeer(p) + go cl.connectToPeer(p) } } case <-time.After(500 * time.Millisecond): @@ -246,20 +284,77 @@ func (cl *Client) acceptLoop(l net.Listener) { } } -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 +// connectToPeer races a direct dial against a relay-piped fallback for +// reaching p; whichever succeeds first is used, and the other attempt is +// abandoned. This is what makes peers on different networks (no shared +// LAN, no forwarded port) work at all: when direct fails, the relay just +// pipes bytes between the two of them instead — the same as croc's own +// default behavior for anyone not on the same local network. +func (cl *Client) connectToPeer(p tcp.PeerInfo) { + type dialResult struct { + conn *comm.Comm + key []byte } - 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 + found := make(chan dialResult, 1) + + win := func(conn *comm.Comm, key []byte) { + select { + case found <- dialResult{conn, key}: + default: + conn.Close() + } } - cl.handlePeerLink(newPeerLink(c, key)) + + if p.Addr != "" { + go func() { + conn, err := net.DialTimeout("tcp", p.Addr, 5*time.Second) + if err != nil { + log.Debugf("could not dial peer %s directly 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 + } + win(c, key) + }() + } + + go func() { + c, err := tcp.RelayPipe(cl.opts.RelayAddress, pairKey(cl.opts.Code, cl.selfID, p.ID)) + if err != nil { + log.Debugf("could not open a relay pipe to peer %s: %v", p.ID, err) + return + } + // Both sides independently compute the same initiator role from + // the same two IDs, so exactly one of them ends up true. + key, err := handshake(c, cl.opts.Code, cl.selfID < p.ID) + if err != nil { + log.Debugf("relay-piped handshake with peer %s failed: %v", p.ID, err) + c.Close() + return + } + win(c, key) + }() + + select { + case r := <-found: + cl.handlePeerLink(newPeerLink(r.conn, r.key)) + case <-time.After(25 * time.Second): + log.Warnf("could not connect to peer %s: direct dial and relay-piped fallback both failed", p.ID) + } +} + +// pairKey deterministically derives the same relay pipe-pairing key for +// both sides of a peer pair, regardless of which one computes it first. +func pairKey(code, idA, idB string) string { + if idA > idB { + idA, idB = idB, idA + } + return code + ":" + idA + ":" + idB } func (cl *Client) handlePeerLink(link *peerLink) { diff --git a/src/tcp/tcp.go b/src/tcp/tcp.go index 4684746..8e8ed4f 100644 --- a/src/tcp/tcp.go +++ b/src/tcp/tcp.go @@ -12,8 +12,16 @@ import ( log "github.com/tzgyn/share/src/logger" "github.com/tzgyn/share/src/comm" + "github.com/tzgyn/share/src/models" ) +// pipeWaitTimeout bounds how long the relay holds a lone "pipe" request +// open waiting for its matching partner before giving up. Kept short: a +// peer only opens one of these right after a direct connection attempt to +// another peer has already failed, so both sides should arrive at nearly +// the same time. +const pipeWaitTimeout = 20 * time.Second + // 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). @@ -22,13 +30,21 @@ type PeerInfo struct { 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 +// ControlMessage is the vocabulary spoken between a peer and the relay. +// A peer sends exactly one "join" to announce itself in a room; 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. +// changes. The relay never sees file content or chunk data for that — +// peers exchange it directly with each other once they have each other's +// Addr. +// +// Separately, a peer can send a "pipe" instead of "join": this is for +// when two peers couldn't reach each other directly (different networks, +// no forwarded port). Two peers sending "pipe" with the same Room value +// (used here as an arbitrary pairing key, not an actual room) get piped +// together — the relay just forwards raw bytes between them from that +// point on, blind to whatever protocol the two peers speak over it. type ControlMessage struct { - Type string `json:"type"` // "join" or "peers" + Type string `json:"type"` // "join", "peers", or "pipe" Room string `json:"room,omitempty"` Peer PeerInfo `json:"peer,omitempty"` Peers []PeerInfo `json:"peers,omitempty"` @@ -39,6 +55,7 @@ type server struct { port string debugLevel string rooms roomMap + pipes pipeMap *stop } @@ -57,6 +74,17 @@ type roomMap struct { sync.Mutex } +// pipeSlot is a lone "pipe" request waiting for its matching partner. +type pipeSlot struct { + comm *comm.Comm + ready chan *comm.Comm // the partner's comm, posted here once it arrives +} + +type pipeMap struct { + slots map[string]*pipeSlot + sync.Mutex +} + func newDefaultServer() *server { s := new(server) s.debugLevel = DEFAULT_LOG_LEVEL @@ -89,6 +117,10 @@ func (s *server) start() (err error) { s.rooms.rooms = make(map[string]*roomInfo) s.rooms.Unlock() + s.pipes.Lock() + s.pipes.slots = make(map[string]*pipeSlot) + s.pipes.Unlock() + defer s.stop.Cancel() if s.stop.gui { defer s.stop.wg.Wait() @@ -149,7 +181,7 @@ func (s *server) run() (err error) { if err != nil { return fmt.Errorf("problem accepting connection: %w", err) } - log.Debugf("client %s connected", connection.RemoteAddr().String()) + log.Infof("client %s connected", connection.RemoteAddr().String()) s.stop.wg.Add(1) go func(connection net.Conn) { defer s.stop.wg.Done() @@ -158,25 +190,59 @@ func (s *server) run() (err error) { } } -// 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. +// handleConnection reads one control message and dispatches on its type: +// "join" registers the peer in a room and holds the connection open for +// membership control messages; "pipe" pairs this connection with another +// peer's matching "pipe" request and forwards raw bytes between them — +// the fallback for when two peers couldn't reach each other directly. func (s *server) handleConnection(connection net.Conn) { c := comm.New(connection) - defer connection.Close() b, err := c.Receive() if err != nil { - log.Debug(err) + // includes the relay's own startup self-check connection, which + // dials and closes without sending anything — routine, not a + // sign of a real client failing to join. + log.Infof("client %s disconnected before sending a control message: %v", connection.RemoteAddr().String(), err) + connection.Close() return } 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) + if err := json.Unmarshal(b, &msg); err != nil { + log.Warnf("client %s sent an unparseable control message: %v", connection.RemoteAddr().String(), err) + connection.Close() return } - log.Debugf("peer %s joining room %s", msg.Peer.ID, msg.Room) + + switch msg.Type { + case "join": + if msg.Room == "" || msg.Peer.ID == "" { + log.Warnf("client %s sent a bad join message (room=%q peerID=%q)", connection.RemoteAddr().String(), msg.Room, msg.Peer.ID) + connection.Close() + return + } + s.handleJoin(c, connection, msg) + case "pipe": + if msg.Room == "" { + log.Warnf("client %s sent a bad pipe request (empty key)", connection.RemoteAddr().String()) + connection.Close() + return + } + s.handlePipe(c, msg.Room) + default: + log.Warnf("client %s sent an unknown control message type %q", connection.RemoteAddr().String(), msg.Type) + connection.Close() + } +} + +// handleJoin registers the peer in its room, broadcasts membership, then +// just waits 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) handleJoin(c *comm.Comm, connection net.Conn, msg ControlMessage) { + defer connection.Close() + + log.Infof("peer %s (%s) joining room %s", msg.Peer.ID, connection.RemoteAddr().String(), msg.Room) s.rooms.Lock() r, ok := s.rooms.rooms[msg.Room] @@ -203,12 +269,105 @@ func (s *server) handleConnection(connection net.Conn) { // 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) + log.Infof("peer %s left room %s: %v", msg.Peer.ID, msg.Room, err) return } } } +// handlePipe pairs this connection with another peer's "pipe" request +// carrying the same key, then forwards raw bytes between them until +// either side disconnects. The relay never parses anything past this +// point — whatever protocol the two peers speak over the pipe is +// invisible to it, same as a direct connection between them would be. +func (s *server) handlePipe(c *comm.Comm, key string) { + s.pipes.Lock() + slot, waiting := s.pipes.slots[key] + if !waiting { + slot = &pipeSlot{comm: c, ready: make(chan *comm.Comm, 1)} + s.pipes.slots[key] = slot + s.pipes.Unlock() + log.Infof("pipe %s: waiting for a match", key) + + select { + case partner := <-slot.ready: + log.Infof("pipe %s: matched, forwarding bytes", key) + pipe(c.Connection(), partner.Connection()) + c.Close() + partner.Close() + log.Infof("pipe %s: finished", key) + case <-time.After(pipeWaitTimeout): + s.pipes.Lock() + if s.pipes.slots[key] == slot { + delete(s.pipes.slots, key) + } + s.pipes.Unlock() + log.Infof("pipe %s: timed out waiting for a match", key) + c.Close() + } + return + } + + delete(s.pipes.slots, key) + s.pipes.Unlock() + // hand off to the waiting goroutine above, which owns closing both + // connections once piping finishes — this one must not close its own. + slot.ready <- c +} + +// 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(res, b[:n]) + c <- res + } + if err != nil { + c <- nil + break + } + } + }() + + return c +} + +// pipe creates a full-duplex pipe between the two sockets, transferring +// data from one to the other until either side closes. +func pipe(conn1, 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.Warnf("pipe write error: %v", err) + } + case b2 := <-chan2: + if b2 == nil { + return + } + if _, err := conn1.Write(b2); err != nil { + log.Warnf("pipe write error: %v", err) + } + } + } +} + func (s *server) broadcastPeers(room string) { s.rooms.Lock() r, ok := s.rooms.rooms[room] @@ -224,9 +383,19 @@ func (s *server) broadcastPeers(room string) { } s.rooms.Unlock() if !ok { + log.Infof("room %s is now empty", room) return } + addrs := make([]string, len(peers)) + for i, p := range peers { + addrs[i] = p.Addr + if addrs[i] == "" { + addrs[i] = "(unreachable/no listener)" + } + } + log.Infof("room %s now has %d peer(s): %v", room, len(peers), addrs) + b, err := json.Marshal(ControlMessage{Type: "peers", Peers: peers}) if err != nil { log.Error(err) @@ -234,7 +403,7 @@ func (s *server) broadcastPeers(room string) { } for _, c := range conns { if err := c.Send(b); err != nil { - log.Debug(err) + log.Warnf("could not send peer list to %s: %v", c.Connection().RemoteAddr().String(), err) } } } @@ -287,3 +456,32 @@ func JoinRoom(address, room string, self PeerInfo, timelimit ...time.Duration) ( peers = ch return } + +// RelayPipe connects to the relay and asks it to pipe this connection +// together with whichever other peer sends the same pairKey — the +// fallback for when two peers can't reach each other directly (different +// networks, no forwarded port, etc). pairKey must be something both +// peers compute identically and independently (e.g. derived from both +// their IDs, sorted, so it doesn't matter who arrives first). The +// returned connection can be used exactly like a direct one — the relay +// forwards whatever either side writes to it verbatim once both sides +// have shown up. +func RelayPipe(address, pairKey string, timelimit ...time.Duration) (c *comm.Comm, err error) { + if len(timelimit) > 0 { + c, err = comm.NewConnection(address, timelimit[0]) + } else { + c, err = comm.NewConnection(address) + } + if err != nil { + return nil, err + } + + b, err := json.Marshal(ControlMessage{Type: "pipe", Room: pairKey}) + if err != nil { + return nil, err + } + if err := c.Send(b); err != nil { + return nil, err + } + return c, nil +}