diff --git a/.gitignore b/.gitignore index 7e884d5..dfc9e36 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,7 @@ cli/flit # Host-specific deploy scripts (contain server addresses / SSH targets) deploy-pwa.sh +deploy-daemon.sh serve-pwa.sh # Runtime config — contains anchor URL; copy from example and fill in diff --git a/cli/cmd/flit/daemon.go b/cli/cmd/flit/daemon.go index 6fd7c54..644c4cd 100644 --- a/cli/cmd/flit/daemon.go +++ b/cli/cmd/flit/daemon.go @@ -115,6 +115,7 @@ func runDaemon(ctx context.Context) { // reconnecting with exponential backoff whenever the connection drops. func runPeerLoop(ctx context.Context, tc transport.Config, myID string, peer peerConfig, downloadDir string) { room := transport.PairRoomName(myID, peer.ID) + log.Printf("[%s] room: %s", peer.Label, room) backoff := 2 * time.Second for { diff --git a/cli/cmd/flit/main.go b/cli/cmd/flit/main.go index fb60129..58900f1 100644 --- a/cli/cmd/flit/main.go +++ b/cli/cmd/flit/main.go @@ -1,5 +1,6 @@ // flit — headless CLI for ephemeral, E2E-encrypted file transfer. // +// flit id print this device's peer ID (hex Ed25519 pubkey) // flit send generate a one-shot room, print QR + invite, wait for a peer, send // flit recv join a room from an invite string, accept the offered file // flit daemon run as a persistent receiver for trusted peers (reads ~/.flit/daemon.toml) @@ -100,6 +101,8 @@ func main() { defer stop() switch os.Args[1] { + case "id": + printID() case "send": if len(os.Args) < 3 { fmt.Println("usage: flit send ") @@ -115,11 +118,20 @@ func main() { case "daemon": runDaemon(ctx) default: - fmt.Println("usage: flit send | flit recv | flit daemon") + fmt.Println("usage: flit id | flit send | flit recv | flit daemon") os.Exit(1) } } +func printID() { + sess, err := transport.NewSession(dataDir(), transport.Config{}) + if err != nil { + fmt.Fprintln(os.Stderr, "flit:", err) + os.Exit(1) + } + fmt.Println(sess.PeerID()) +} + func send(ctx context.Context, path string) { if _, err := os.Stat(path); err != nil { fmt.Fprintln(os.Stderr, "flit:", err) diff --git a/cli/internal/transport/transport.go b/cli/internal/transport/transport.go index 76808e8..b65ef4c 100644 --- a/cli/internal/transport/transport.go +++ b/cli/internal/transport/transport.go @@ -15,6 +15,7 @@ import ( "io" "log" "os" + "path/filepath" "strconv" "strings" "sync" @@ -105,6 +106,26 @@ func dialSignaling(ctx context.Context, url string, id *crypto.Identity, netHash } } }() + // Keepalive: ping every 20s so proxies/NAT don't silently drop the connection. + go func() { + t := time.NewTicker(20 * time.Second) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + pingCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + err := conn.Ping(pingCtx) + cancel() + if err != nil { + // Close so the read loop errors and OnDisconnected fires. + _ = conn.Close(websocket.StatusGoingAway, "ping failed") + return + } + } + } + }() for { var msg anchorMsg @@ -145,18 +166,20 @@ type Session struct { identity *crypto.Identity cfg Config - mu sync.Mutex - pc *webrtc.PeerConnection - dc *webrtc.DataChannel - peerID string - verified bool - esk, epk *[32]byte - peerEPK *[32]byte - ekeySent bool - offered bool + mu sync.Mutex + pc *webrtc.PeerConnection + dc *webrtc.DataChannel + peerID string + verified bool + esk, epk *[32]byte + peerEPK *[32]byte + ekeySent bool + offered bool + offerByOrder bool // true if this side should create the yaw DC and SDP offer - sig *signaling - pendingRecv *recvState + sig *signaling + pendingRecvs map[string]*recvState + pendingAccepts map[string]chan int64 OnConnected func(verified bool) OnDisconnected func() @@ -197,8 +220,18 @@ func (s *Session) Join(ctx context.Context, roomName string, trustedPeerID strin go func() { for { + // 3-minute read deadline: if the anchor silently evicts us from + // the room (idle timeout) without closing the WS, we'll never + // receive a peer-join. Force a reconnect so we re-enter the room. + readCtx, cancel := context.WithTimeout(ctx, 3*time.Minute) var msg anchorMsg - if err := wsjson.Read(ctx, sig.conn, &msg); err != nil { + err := wsjson.Read(readCtx, sig.conn, &msg) + cancel() + if err != nil { + _ = sig.conn.Close(websocket.StatusGoingAway, "idle timeout") + if s.OnDisconnected != nil { + s.OnDisconnected() + } return } switch msg.Type { @@ -206,7 +239,16 @@ func (s *Session) Join(ctx context.Context, roomName string, trustedPeerID strin if trustedPeerID != "" && msg.ID != trustedPeerID { continue } + log.Printf("flit: peer-join: %s", msg.ID[:8]) _ = s.connectTo(ctx, msg.ID) + case "peer-left": + if trustedPeerID != "" && msg.ID != trustedPeerID { + continue + } + // Log but don't reset: active file transfers use the WebRTC + // path which is independent of signaling. The stale-PC case + // is handled in connectTo when peer-join arrives next. + log.Printf("flit: peer-left: %s", msg.ID[:8]) case "from": if trustedPeerID != "" && msg.From != trustedPeerID { continue @@ -220,45 +262,77 @@ func (s *Session) Join(ctx context.Context, roomName string, trustedPeerID strin func (s *Session) connectTo(ctx context.Context, peerID string) error { s.mu.Lock() + var oldPC *webrtc.PeerConnection if s.pc != nil { - s.mu.Unlock() - return nil + state := s.pc.ConnectionState() + if state == webrtc.PeerConnectionStateNew || + state == webrtc.PeerConnectionStateConnected { + s.mu.Unlock() + return nil // healthy or brand-new connection already exists + } + // Stale PC (peer left while we were connecting, or ICE failed) — replace it. + log.Printf("flit: connectTo %s: replacing stale PC (state=%s)", peerID[:8], state) + oldPC = s.pc + s.pc = nil } s.peerID = peerID - pc, err := webrtc.NewPeerConnection(webrtc.Configuration{ICEServers: iceServers(s.cfg)}) + var se webrtc.SettingEngine + se.DetachDataChannels() + api := webrtc.NewAPI(webrtc.WithSettingEngine(se)) + pc, err := api.NewPeerConnection(webrtc.Configuration{ICEServers: iceServers(s.cfg)}) if err != nil { s.mu.Unlock() + if oldPC != nil { + _ = oldPC.Close() + } return err } s.pc = pc kp, err := crypto.GenerateEphemeral() if err != nil { + s.pc = nil s.mu.Unlock() + if oldPC != nil { + _ = oldPC.Close() + } return err } s.esk, s.epk = kp.PrivateRaw(), kp.PublicRaw() + s.ekeySent = false + s.offered = false + s.peerEPK = nil + s.offerByOrder = s.identity.PeerID() < peerID s.mu.Unlock() + // Close the stale PC after s.pc has been updated. The Closed state callback + // will see isCurrent=false and skip OnDisconnected, so runPeerLoop keeps running. + if oldPC != nil { + _ = oldPC.Close() + } + pc.OnDataChannel(func(dc *webrtc.DataChannel) { s.wireDC(dc) }) pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { if state == webrtc.PeerConnectionStateFailed || state == webrtc.PeerConnectionStateDisconnected || state == webrtc.PeerConnectionStateClosed { s.mu.Lock() - s.pc = nil - s.ekeySent = false - s.offered = false - s.peerEPK = nil + isCurrent := s.pc == pc + if isCurrent { + s.pc = nil + s.ekeySent = false + s.offered = false + s.peerEPK = nil + } s.mu.Unlock() - if s.OnDisconnected != nil { + log.Printf("flit: PC state=%s isCurrent=%v", state, isCurrent) + if isCurrent && s.OnDisconnected != nil { s.OnDisconnected() } } }) s.sendEkey() - offerByOrder := s.identity.PeerID() < peerID - if offerByOrder { + if s.offerByOrder { go func() { time.Sleep(ekeyTimeout) s.maybeOffer() @@ -267,6 +341,23 @@ func (s *Session) connectTo(ctx context.Context, peerID string) error { return nil } +// resetPeer tears down the current WebRTC connection without triggering +// OnDisconnected, leaving the signaling connection open so that the next +// peer-join for this peer can reconnect immediately. +func (s *Session) resetPeer() { + s.mu.Lock() + pc := s.pc + s.pc = nil + s.ekeySent = false + s.offered = false + s.peerEPK = nil + s.mu.Unlock() + if pc != nil { + log.Printf("flit: resetPeer: closing stale PC (peer left signaling)") + _ = pc.Close() + } +} + func (s *Session) sendEkey() { s.mu.Lock() if s.ekeySent { @@ -290,7 +381,7 @@ func (s *Session) sendEkey() { func (s *Session) maybeOffer() { s.mu.Lock() - if s.offered || s.pc == nil { + if !s.offerByOrder || s.offered || s.pc == nil { s.mu.Unlock() return } @@ -434,8 +525,32 @@ func (s *Session) wireDC(dc *webrtc.DataChannel) { s.mu.Lock() s.dc = dc s.mu.Unlock() - dc.OnOpen(func() { s.sendHello() }) - dc.OnMessage(func(msg webrtc.DataChannelMessage) { s.onControl(msg.Data) }) + + startYaw := func() { + raw, err := dc.Detach() + if err != nil { + log.Printf("flit: yaw detach: %v", err) + return + } + s.sendHello() + go func() { + buf := make([]byte, 32768) + for { + n, err := raw.Read(buf) + if n > 0 { + s.onControl(append([]byte(nil), buf[:n]...)) + } + if err != nil { + return + } + } + }() + } + if dc.ReadyState() == webrtc.DataChannelStateOpen { + startYaw() + } else { + dc.OnOpen(func() { startYaw() }) + } return } if strings.HasPrefix(dc.Label(), "f:") { @@ -468,6 +583,19 @@ func (s *Session) onControl(data []byte) { size, _ := m["size"].(float64) s.OnFileOffer(FileOffer{XID: m["xid"].(string), Name: m["name"].(string), Size: int64(size)}) } + case "file-accept": + xid, _ := m["xid"].(string) + var offset int64 + if v, ok := m["offset"].(string); ok { + offset, _ = strconv.ParseInt(v, 10, 64) + } + s.mu.Lock() + ch := s.pendingAccepts[xid] + delete(s.pendingAccepts, xid) + s.mu.Unlock() + if ch != nil { + ch <- offset + } } } @@ -482,7 +610,9 @@ func (s *Session) controlSend(obj map[string]string) { _ = dc.SendText(string(data)) } -// SendFile offers and streams a file to the connected peer. +// SendFile offers and streams a file to the connected peer. If the receiver +// has a partial .part file it will reply with a non-zero offset and the send +// resumes from that byte position. func (s *Session) SendFile(path string) error { f, err := os.Open(path) if err != nil { @@ -496,6 +626,14 @@ func (s *Session) SendFile(path string) error { xid := fmt.Sprintf("push-%d", time.Now().UnixNano()) name := st.Name() + acceptCh := make(chan int64, 1) + s.mu.Lock() + if s.pendingAccepts == nil { + s.pendingAccepts = make(map[string]chan int64) + } + s.pendingAccepts[xid] = acceptCh + s.mu.Unlock() + s.controlSend(map[string]string{"type": "file-offer", "name": name, "size": fmt.Sprint(st.Size()), "xid": xid}) s.mu.Lock() @@ -507,10 +645,24 @@ func (s *Session) SendFile(path string) error { } opened := make(chan struct{}) dc.OnOpen(func() { close(opened) }) + + var offset int64 + select { + case offset = <-acceptCh: + case <-time.After(30 * time.Second): + return fmt.Errorf("timed out waiting for file-accept") + } select { case <-opened: case <-time.After(15 * time.Second): - return fmt.Errorf("timed out waiting for file-accept") + return fmt.Errorf("timed out waiting for data channel open") + } + + if offset > 0 { + if _, err := f.Seek(offset, io.SeekStart); err != nil { + return fmt.Errorf("seek to resume offset: %w", err) + } + log.Printf("flit: resuming %s from byte %d", name, offset) } buf := make([]byte, chunkSize) @@ -534,11 +686,22 @@ func (s *Session) SendFile(path string) error { return dc.Close() } -// AcceptOffer accepts an incoming file-offer and writes the transfer to downloadDir. +// AcceptOffer accepts an incoming file-offer and writes the transfer to +// downloadDir. If a .part file from a previous partial transfer exists for this +// filename and is smaller than the declared size, its byte count is sent back +// as the resume offset so the sender can skip ahead. func (s *Session) AcceptOffer(offer FileOffer, downloadDir string) { - s.controlSend(map[string]string{"type": "file-accept", "xid": offer.XID}) + partPath := filepath.Join(downloadDir, offer.Name+".part") + var offset int64 + if info, err := os.Stat(partPath); err == nil && info.Size() < offer.Size { + offset = info.Size() + } + s.controlSend(map[string]string{"type": "file-accept", "xid": offer.XID, "offset": fmt.Sprint(offset)}) s.mu.Lock() - s.pendingRecv = &recvState{offer: offer, dir: downloadDir} + if s.pendingRecvs == nil { + s.pendingRecvs = make(map[string]*recvState) + } + s.pendingRecvs[offer.XID] = &recvState{offer: offer, dir: downloadDir, have: offset} s.mu.Unlock() } @@ -556,35 +719,87 @@ type recvState struct { func (s *Session) wireFileRecv(dc *webrtc.DataChannel) { xid := strings.TrimPrefix(dc.Label(), "f:") s.mu.Lock() - rs := s.pendingRecv - s.pendingRecv = nil + rs := s.pendingRecvs[xid] + delete(s.pendingRecvs, xid) s.mu.Unlock() - if rs == nil || rs.offer.XID != xid { + if rs == nil { return } - path := rs.dir + "/" + rs.offer.Name - f, err := os.Create(path) - if err != nil { - log.Printf("flit: create %s: %v", path, err) - return - } - rs.f = f - dc.OnMessage(func(msg webrtc.DataChannelMessage) { - if _, err := f.Write(msg.Data); err != nil { + log.Printf("flit: wireFileRecv: %s size=%d state=%s", rs.offer.Name, rs.offer.Size, dc.ReadyState()) + + // Use pion's Detach API so data is buffered in pion's SCTP stream and + // read by our goroutine — no OnMessage/OnClose callback race. + startRecv := func() { + raw, err := dc.Detach() + if err != nil { + log.Printf("flit: file DC detach %s: %v", rs.offer.Name, err) return } - rs.have += int64(len(msg.Data)) - if s.OnFileProgress != nil { - s.OnFileProgress(xid, rs.have, rs.offer.Size) - } - }) - dc.OnClose(func() { - f.Close() - if s.OnFileDone != nil { - s.OnFileDone(rs.offer.Name, path) - } - }) + log.Printf("flit: file DC open: %s", rs.offer.Name) + + go func() { + finalPath := filepath.Join(rs.dir, rs.offer.Name) + resuming := rs.have > 0 + + var f *os.File + if resuming { + partPath := finalPath + ".part" + f, err = os.OpenFile(partPath, os.O_WRONLY|os.O_APPEND, 0644) + if err != nil { + log.Printf("flit: reopen %s: %v — restarting", partPath, err) + resuming = false + rs.have = 0 + f, err = os.Create(finalPath) + } else { + log.Printf("flit: resuming %s from byte %d", rs.offer.Name, rs.have) + } + } else { + f, err = os.Create(finalPath) + } + if err != nil { + log.Printf("flit: open output: %v", err) + return + } + + buf := make([]byte, chunkSize) + for { + n, readErr := raw.Read(buf) + if n > 0 { + if _, werr := f.Write(buf[:n]); werr != nil { + log.Printf("flit: write %s: %v", rs.offer.Name, werr) + break + } + rs.have += int64(n) + if s.OnFileProgress != nil { + s.OnFileProgress(xid, rs.have, rs.offer.Size) + } + } + if readErr != nil { + break + } + } + f.Close() + log.Printf("flit: recv done: %s wrote=%d of %d", rs.offer.Name, rs.have, rs.offer.Size) + + if resuming { + partPath := finalPath + ".part" + if rerr := os.Rename(partPath, finalPath); rerr != nil { + log.Printf("flit: rename %s: %v", partPath, rerr) + return + } + } + if s.OnFileDone != nil { + s.OnFileDone(rs.offer.Name, finalPath) + } + }() + } + + if dc.ReadyState() == webrtc.DataChannelStateOpen { + startRecv() + } else { + dc.OnOpen(func() { startRecv() }) + } } func mustHex(s string) []byte { diff --git a/pwa/src/App.tsx b/pwa/src/App.tsx index 63b5a7f..c77c89b 100644 --- a/pwa/src/App.tsx +++ b/pwa/src/App.tsx @@ -41,7 +41,7 @@ function App() { const [scanning, setScanning] = useState(false) const [myId, setMyId] = useState(null) const [addIdInput, setAddIdInput] = useState('') - const [copied, setCopied] = useState(false) + const [copied, setCopied] = useState<'snip' | 'id' | null>(null) const sessionRef = useRef(null) const verifiedRef = useRef(false) @@ -218,13 +218,14 @@ function App() { setMode('known') } - function copyMyId() { + function copyMyId(mode: 'snip' | 'id') { if (!myId) return - // Build a snip the other side can paste into "Add peer" - const snip = 'flit-peer:' + btoa(JSON.stringify({ id: myId })).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '') - navigator.clipboard.writeText(snip).then(() => { - setCopied(true) - setTimeout(() => setCopied(false), 2000) + const text = mode === 'snip' + ? 'flit-peer:' + btoa(JSON.stringify({ id: myId })).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '') + : myId + navigator.clipboard.writeText(text).then(() => { + setCopied(mode) + setTimeout(() => setCopied(null), 2000) }) } @@ -308,8 +309,11 @@ function App() {
your ID {myId.slice(0, 16)}… - +
)} diff --git a/pwa/src/transport/flit.ts b/pwa/src/transport/flit.ts index 8c7646d..4f4de12 100644 --- a/pwa/src/transport/flit.ts +++ b/pwa/src/transport/flit.ts @@ -400,7 +400,8 @@ export class PeerConn { this.events.fileOffer?.({ peer: this.peerId, xid: m['xid'] as string, name: m['name'] as string, size: m['size'] as number }) } else if (m['type'] === 'file-accept') { const xid = m['xid'] as string - if (this._pushQueue.has(xid)) void this._stream(xid) + const offset = typeof m['offset'] === 'string' ? parseInt(m['offset'] as string, 10) || 0 : 0 + if (this._pushQueue.has(xid)) void this._stream(xid, offset) } else if (m['type'] === 'file-cancel') { const xid = m['xid'] as string if (this._recv[xid]) { @@ -427,23 +428,22 @@ export class PeerConn { this._dc({ type: 'file-offer', name: file.name, size: file.size, xid }) } - private async _stream(xid: string) { + private async _stream(xid: string, resumeOffset = 0) { const file = this._pushQueue.get(xid) if (!file) return this._pushQueue.delete(xid) const dc = this.pc.createDataChannel(`f:${xid}`) dc.binaryType = 'arraybuffer' await new Promise(res => { dc.onopen = () => res() }) - const buf = await file.arrayBuffer() - let offset = 0 - while (offset < buf.byteLength) { + // Slice from resume offset so we only transfer the remaining bytes. + const slice = resumeOffset > 0 ? file.slice(resumeOffset) : file + const buf = await slice.arrayBuffer() + let localOffset = 0 + while (localOffset < buf.byteLength) { while (dc.bufferedAmount > 1024 * 1024) await new Promise(r => setTimeout(r, 10)) - dc.send(buf.slice(offset, offset + CHUNK)) - offset += CHUNK - this.events.fileSendProgress?.({ peer: this.peerId, xid, name: file.name, sent: offset, total: buf.byteLength }) - // Yield to the event loop every chunk so the browser can paint - // progress — without this, small files finish in one synchronous - // tick and the UI never shows intermediate state. + dc.send(buf.slice(localOffset, localOffset + CHUNK)) + localOffset += CHUNK + this.events.fileSendProgress?.({ peer: this.peerId, xid, name: file.name, sent: resumeOffset + localOffset, total: file.size }) await new Promise(r => setTimeout(r, 0)) } dc.close()