Compare commits
8 Commits
3a105e2c9a
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2b77c3e102 | ||
|
|
932a4bd7bb | ||
|
|
0ffe9fc29d | ||
|
|
06e7ce83e9 | ||
|
|
c8205c703b | ||
|
|
bd7879ce7b | ||
|
|
6a3ad787d5 | ||
|
|
153426367d |
21
LICENSE
Normal file
21
LICENSE
Normal file
@@ -0,0 +1,21 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2026 explewd
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
219
PROPOSAL-watch.md
Normal file
219
PROPOSAL-watch.md
Normal file
@@ -0,0 +1,219 @@
|
||||
# Proposal: `flit watch` — one-way folder sync over flit
|
||||
|
||||
**Status:** proposed, not built. Self-contained — written so a fresh agent
|
||||
with no prior conversation context can pick this up and implement it
|
||||
without anything explained first.
|
||||
|
||||
## Context (read this first)
|
||||
|
||||
Full background on flit is in [README.md](./README.md) and
|
||||
[PROTOCOL.md](./PROTOCOL.md) — read both before touching code. The
|
||||
relevant existing pieces for this proposal, all in `cli/`:
|
||||
|
||||
- `cmd/flit/main.go` — `send(ctx, path)`: creates a `transport.Session`,
|
||||
generates a one-shot room + QR invite, and on `OnConnected` calls
|
||||
`sess.SendFile(path)`. **`SendFile` is synchronous and blocking** — it
|
||||
returns only once the transfer is fully complete (or errors). Its
|
||||
return value alone is a sufficient "did this actually get there"
|
||||
signal; no separate completion callback exists or is needed.
|
||||
- `cmd/flit/daemon.go` — the pattern this proposal reuses almost
|
||||
wholesale, just aimed the other direction:
|
||||
- `daemonConfig`/`peerConfig` — TOML config (`~/.flit/daemon.toml`),
|
||||
a list of trusted peers by `id`/`label`.
|
||||
- `runPeerLoop(ctx, tc, myID, peer, downloadDir)` — maintains a
|
||||
**persistent connection to one named peer**, reconnecting with
|
||||
exponential backoff (2s → 30s cap) on disconnect. Sets
|
||||
`OnFileOffer` to auto-accept anything the peer sends and save it to
|
||||
`downloadDir`.
|
||||
- This is receive-only today. **`flit watch` needs the send-side
|
||||
sibling of this exact loop** — same persistent-connection/backoff
|
||||
shape, but triggered by local filesystem events instead of incoming
|
||||
offers.
|
||||
- `internal/transport/transport.go` — `Session.Join(ctx, roomName,
|
||||
trustedPeerID)`, `transport.PairRoomName(a, b)` (deterministic
|
||||
pairwise room from two peer IDs — this is what makes "known device,
|
||||
no fresh QR" work at all), `Session.SendFile(path)`,
|
||||
`Session.PeerID()`.
|
||||
|
||||
## Motivation
|
||||
|
||||
At one point the idea on the table was two-way directory sync — watch
|
||||
a folder, keep it mirrored to a known host, in both directions. That
|
||||
was deliberately walked back to **one-way sync**: source → destination
|
||||
only, push only, never reconciling anything back. That sidesteps the
|
||||
entire class of problem full bidirectional sync brings with it
|
||||
(conflict resolution, delete propagation, catch-up ordering). The
|
||||
moment this proposal starts growing bidirectional-anything, that's a
|
||||
sign it's drifting toward re-implementing a general sync tool — flit's
|
||||
own case study already draws that line explicitly ("Syncthing syncs
|
||||
continuously; too heavy for ad hoc one-offs" — flit is deliberately
|
||||
*not* trying to be that). This proposal should stay on the "one
|
||||
watched folder pushes new files to one named peer, and that's the
|
||||
whole feature" side of that line.
|
||||
|
||||
The actual want: drop a file into a watched folder, it gets encrypted
|
||||
and pushed to a specific already-known device automatically — no
|
||||
manual `flit send`, no QR, no re-running anything by hand per file.
|
||||
|
||||
## Design
|
||||
|
||||
### 1. The receiving side needs no changes
|
||||
|
||||
Worth stating plainly since it simplifies scope: an already-running
|
||||
`flit daemon`, configured with the sender as a trusted peer, already
|
||||
does exactly the right thing on receipt — auto-accept, save to
|
||||
`download_dir`. This proposal is entirely about the sending side.
|
||||
|
||||
### 2. `flit watch` — a new subcommand, new config file
|
||||
|
||||
`~/.flit/watch.toml`, same shape as `daemon.toml` deliberately, for
|
||||
familiarity:
|
||||
|
||||
```toml
|
||||
signal_url = "wss://your-anchor.example.com/ws"
|
||||
turn_url = ""
|
||||
turn_secret = ""
|
||||
|
||||
[[watch]]
|
||||
dir = "/home/user/backups/documents"
|
||||
peer_id = "<hex Ed25519 pubkey of the destination device>"
|
||||
peer_label = "home-server"
|
||||
pattern = "*.pdf,*.docx" # optional glob include-list; default "*" if omitted
|
||||
stable_after = "10s" # optional quiet period before considering a file ready
|
||||
```
|
||||
|
||||
Multiple `[[watch]]` blocks are allowed — different folders can go to
|
||||
different peers, or the same peer, independently.
|
||||
|
||||
### 3. Local state = the send queue, no separate queue structure needed
|
||||
|
||||
One state file per watch entry:
|
||||
`~/.flit/watch-state/<sha256(dir + peer_id)>.json` —
|
||||
`{ "<relative-path>": { "size": N, "mtime": N, "sha256": "...", "sent_at": N|null } }`.
|
||||
|
||||
**The invariant that makes this simple**: `sent_at` is only ever set
|
||||
*after* `SendFile` returns without error. A file with `sent_at: null`,
|
||||
or whose current size/mtime/hash no longer matches what's recorded, is
|
||||
"needs sending" — that's the entire queue. No separate retry logic:
|
||||
if `SendFile` fails partway (peer disconnects mid-transfer), the state
|
||||
was never advanced, so the file is exactly as eligible for retry on
|
||||
the next connection as it was before the attempt.
|
||||
|
||||
### 4. Startup: reconcile state against reality, don't trust only live events
|
||||
|
||||
On start, before subscribing to filesystem events, walk `dir`
|
||||
(respecting `pattern`) and diff against the state file. This catches
|
||||
anything that changed while the watcher wasn't running — a live-only
|
||||
`fsnotify` subscription would silently miss files dropped in during
|
||||
downtime.
|
||||
|
||||
### 5. The watch-and-send loop — `runPeerLoop`'s sibling
|
||||
|
||||
```go
|
||||
// runWatchLoop mirrors daemon.go's runPeerLoop: a persistent connection
|
||||
// to one named peer, reconnecting with the same backoff. Differs only in
|
||||
// what drives sends — a local file-ready queue instead of OnFileOffer.
|
||||
func runWatchLoop(ctx context.Context, tc transport.Config, myID string, w watchConfig, state *watchState) {
|
||||
room := transport.PairRoomName(myID, w.PeerID)
|
||||
backoff := 2 * time.Second
|
||||
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
sess, err := transport.NewSession(dataDir(), tc)
|
||||
// ... same connect/reconnect shape as runPeerLoop ...
|
||||
sess.OnConnected = func(verified bool) {
|
||||
backoff = 2 * time.Second
|
||||
drainQueue(sess, w, state) // send everything currently queued, in order
|
||||
}
|
||||
// fsnotify events (and the startup reconciliation) push into the
|
||||
// same queue; if already connected when a new file becomes ready,
|
||||
// send it immediately rather than waiting for the next OnConnected.
|
||||
...
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
`drainQueue` sends queued files one at a time via `sess.SendFile(path)`
|
||||
— sequential, not concurrent, since flit's session model is one peer
|
||||
connection at a time and concurrent sends would need to be queued
|
||||
DataChannel-side anyway (see "Multi-file send" in the flit case study
|
||||
for the existing precedent of handling several files across one
|
||||
session, just not built for a filesystem-driven trigger).
|
||||
|
||||
### 6. Debounce — don't send a file mid-write
|
||||
|
||||
`stable_after` is a quiet-period debounce: after the *last* fsnotify
|
||||
write event for a given file, wait that long before considering it
|
||||
ready. Without this, a large file being actively written would get
|
||||
offered (and likely fail hash verification on the receiving end, or
|
||||
worse, silently transfer a truncated version) the moment its first
|
||||
bytes land. This is a real requirement, not a nice-to-have — flag it
|
||||
as such if a fresh implementation is tempted to skip it for a first
|
||||
cut.
|
||||
|
||||
### 7. Never touches deletes
|
||||
|
||||
The watcher only reacts to create/write events. A file deleted from
|
||||
the source directory is never deleted from the destination, and never
|
||||
should be — that's the entire point of one-way sync rather than a
|
||||
mirror. Don't wire up `fsnotify`'s remove/rename events to anything
|
||||
that deletes on the far end.
|
||||
|
||||
## Security considerations
|
||||
|
||||
- **This sends files with no per-file human confirmation** — a
|
||||
deliberate, meaningful departure from interactive `flit send`
|
||||
(always a conscious action) and even from `daemon`'s auto-accept
|
||||
(still gated behind that peer being explicitly marked trusted with
|
||||
`autoAccept` in the keyring). The safety knob here is entirely
|
||||
`watch.toml` itself: which directories are named, and each entry's
|
||||
`pattern`. Recommend a **`flit watch --dry-run`** mode that walks the
|
||||
configured directories and prints what *would* be sent, without
|
||||
connecting to anything — a way to sanity-check a new watch config
|
||||
before it goes live and silently starts pushing files.
|
||||
- **`pattern` should default to something narrower than `*` in
|
||||
practice**, even though `*` is the technical default if omitted —
|
||||
worth calling out in whatever docs ship with this, since an
|
||||
unconstrained watch on a broad directory (e.g. a whole home folder)
|
||||
turns "sync my documents" into "sync everything that ever lands in
|
||||
this tree," which is a much bigger promise than most `[[watch]]`
|
||||
entries are meant to make.
|
||||
- **The destination is exactly as trusted as any other flit peer** —
|
||||
this doesn't introduce a new trust primitive, it reuses the existing
|
||||
Ed25519 identity + pairwise-room model as-is. A `peer_id` here is
|
||||
the same kind of already-known, already-paired device as anything in
|
||||
the PWA's "Known devices" list or `daemon.toml`.
|
||||
- **No encryption-at-rest concern beyond what already exists** — files
|
||||
are E2E encrypted in transit exactly like every other flit transfer;
|
||||
this proposal doesn't change flit's crypto at all, only adds a new
|
||||
trigger for when `SendFile` gets called.
|
||||
|
||||
## Open questions
|
||||
|
||||
- **Large initial backlogs** — if `dir` already has thousands of files
|
||||
on first run, the startup reconciliation queues all of them at once.
|
||||
Sequential sending (per §5) means this could take a long time before
|
||||
catching up to "live." Acceptable for a first cut; a
|
||||
`--max-startup-queue` or similar throttle could be added later if
|
||||
this turns out to matter in practice — not designing it ahead of an
|
||||
actual backlog being a problem.
|
||||
- **Multiple watchers targeting the same peer simultaneously** — each
|
||||
`[[watch]]` entry in §5's design opens its own `runWatchLoop`, meaning
|
||||
its own separate connection/room to the same peer if two entries
|
||||
share a `peer_id`. `transport.PairRoomName` is deterministic per
|
||||
peer pair, so two simultaneous sessions to the same peer would
|
||||
collide on the same room name — this needs either a single shared
|
||||
connection per unique `peer_id` (multiple watch entries feeding one
|
||||
`runWatchLoop`/queue) or confirmation that flit's transport already
|
||||
tolerates multiple simultaneous sessions to the same room gracefully.
|
||||
**Needs resolving before implementation, not deferred** — this is a
|
||||
correctness question about the connection model, not a nice-to-have.
|
||||
- **State file corruption/loss** — if `~/.flit/watch-state/*.json` is
|
||||
lost (disk issue, manual deletion), the next startup reconciliation
|
||||
will treat every file as unsent and re-push the entire directory.
|
||||
Harmless for correctness (the destination just re-receives what it
|
||||
already has) but potentially slow/wasteful for a large directory.
|
||||
Worth a passing mention in user-facing docs, not worth engineering
|
||||
around for a first version.
|
||||
40
README.md
40
README.md
@@ -20,6 +20,7 @@ It is deliberately not a replacement for [Zipline](https://github.com/diced/zipl
|
||||
- Auto-downloads on the receiver side — no "click to save" step.
|
||||
- In-app QR scanner (Chrome/Android) — scan the sender's QR without leaving flit.
|
||||
- Android share sheet integration — "Share → flit" works from Photos, Files, any app.
|
||||
- One-way folder sync (`flit watch`) — drop a file into a watched folder, it gets pushed to a trusted peer automatically, no manual send.
|
||||
|
||||
---
|
||||
|
||||
@@ -52,9 +53,9 @@ Each device generates an **Ed25519 keypair** on first run. The hex-encoded publi
|
||||
|
||||
Pairing uses [yaw/2.1](PROTOCOL.md) forward-secret signaling, trimmed for flit's 1:1 ephemeral use case:
|
||||
|
||||
1. Connecting peers exchange **ephemeral X25519 keys** (`ekey`) signed with their Ed25519 identity keys — providing forward secrecy for the signaling channel.
|
||||
2. Once the WebRTC data channel opens, each side sends a **`hello`** message containing their identity and a signature over a known prefix. This binds the WebRTC DTLS fingerprint to the Ed25519 identity, confirming "the device on the other end is who the QR said it was."
|
||||
3. File data flows over the WebRTC data channel — encrypted by DTLS, with the identity-confirmed binding from step 2.
|
||||
1. Connecting peers exchange **ephemeral X25519 keys** (`ekey`) signed with their Ed25519 identity keys over the sealed signaling channel — this is what actually authenticates "the device on the other end is who the QR said it was."
|
||||
2. Once the WebRTC data channel opens, each side sends a **`hello`** message confirming its identity, plus a best-effort signature binding it to the session's DTLS fingerprint when both sides report that fingerprint under a hash algorithm flit can parse. `verified` status rests on step 1 (the authenticated sealed exchange) and `hello.id` matching it — not on the fingerprint signature, which different WebRTC stacks may report under different hash algorithms (WebKit: sha-512, Chromium/pion: sha-256) and so can't be relied on to match even between two legitimate peers.
|
||||
3. File data flows over the WebRTC data channel — encrypted by DTLS, with the identity already confirmed by step 1.
|
||||
|
||||
The anchor never sees plaintext message content — it only routes sealed (encrypted) blobs by peer id and hashed room name.
|
||||
|
||||
@@ -101,6 +102,7 @@ For headless machines (homelab boxes, servers) that can't run a browser.
|
||||
flit send path/to/file.tar.gz # print QR + invite, wait for peer, send
|
||||
flit recv "flit:eyJhbmNob3..." # join from invite string, accept file
|
||||
flit daemon # persistent receiver for trusted peers
|
||||
flit watch [--dry-run] # watch folders, auto-push new files to trusted peers
|
||||
```
|
||||
|
||||
**Built with:** Go, pion/webrtc, libsodium (via nhooyr.io/websocket + internal crypto).
|
||||
@@ -132,6 +134,38 @@ Copy `cli/daemon.toml.example` to `~/.flit/daemon.toml` to get started. The daem
|
||||
|
||||
**Bootstrapping:** to get a peer's `id`, pair once via QR (`flit send` / PWA "Invite new") and copy the hex ID from the "peer connected" log line, or read it from the PWA's Known devices list after tapping "Remember this device."
|
||||
|
||||
### Watch mode
|
||||
|
||||
`flit watch` is the send-side sibling of `flit daemon`: it watches one or more local folders and automatically pushes new or changed files to a specific trusted peer — no manual `flit send`, no QR, no per-file interaction. It's **one-way sync only** (source → destination, push only) — deliberately not a general bidirectional sync tool, and it never propagates deletes.
|
||||
|
||||
The receiving side needs no changes: an already-running `flit daemon`, configured with the watcher's device as a trusted peer, already does the right thing on receipt.
|
||||
|
||||
Config lives at `~/.flit/watch.toml`:
|
||||
|
||||
```toml
|
||||
signal_url = "wss://your-anchor.example.com/ws"
|
||||
turn_url = "" # optional
|
||||
turn_secret = "" # optional
|
||||
|
||||
[[watch]]
|
||||
dir = "/home/user/backups/documents"
|
||||
peer_id = "<hex Ed25519 pubkey of the destination device>"
|
||||
peer_label = "home-server"
|
||||
pattern = "*.pdf,*.docx" # optional glob include-list; default "*" if omitted
|
||||
stable_after = "10s" # optional quiet period before considering a file ready
|
||||
```
|
||||
|
||||
Multiple `[[watch]]` blocks are allowed — different folders can go to different peers, or the same peer (entries sharing a `peer_id` share a single connection).
|
||||
|
||||
Copy `cli/watch.toml.example` to `~/.flit/watch.toml` to get started. Notes on behavior:
|
||||
|
||||
- **Startup reconciliation** — on start, before subscribing to filesystem events, `flit watch` walks each `dir` and diffs it against local state, so anything that changed while the watcher wasn't running gets picked up (not just live events going forward).
|
||||
- **Debounce (`stable_after`)** — a file is only considered ready to send after this quiet period has passed since its last write, so a large file being actively written isn't offered mid-write.
|
||||
- **State tracking** — one JSON state file per `[[watch]]` entry lives in `~/.flit/watch-state/`, recording what's been sent. If a send fails partway (peer disconnects mid-transfer), the file is retried on the next successful connection — no separate retry logic needed. Losing this state is harmless (worst case: the whole directory gets re-sent) but not something to engineer around.
|
||||
- **`flit watch --dry-run`** walks the configured directories and prints what *would* be sent, without connecting to anything — use it to sanity-check a new `watch.toml` before it goes live.
|
||||
- **`pattern` should be as narrow as your use case allows.** `*` is the default if omitted, but an unconstrained watch on a broad directory turns "sync my documents" into "sync everything that ever lands in this tree." There's no per-file confirmation in watch mode — the directory and pattern in `watch.toml` are the only safety knob.
|
||||
- Reconnects with the same exponential backoff (2s → 30s cap) as daemon mode, and reuses the same trusted-peer identity model — a `peer_id` here is the same kind of already-known device as anything in the PWA's Known devices list or `daemon.toml`.
|
||||
|
||||
---
|
||||
|
||||
## Self-hosting
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
// flit send <path> generate a one-shot room, print QR + invite, wait for a peer, send
|
||||
// flit recv <invite|code> 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)
|
||||
// flit watch [--dry-run] watch folders and auto-push new files to trusted peers (reads ~/.flit/watch.toml)
|
||||
package main
|
||||
|
||||
import (
|
||||
@@ -93,7 +94,7 @@ func cfg() transport.Config {
|
||||
|
||||
func main() {
|
||||
if len(os.Args) < 2 {
|
||||
fmt.Println("usage: flit send <path> | flit recv <invite> | flit daemon")
|
||||
fmt.Println("usage: flit id | flit send <path> | flit recv <invite> | flit daemon | flit watch [--dry-run]")
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
@@ -117,8 +118,11 @@ func main() {
|
||||
recv(ctx, os.Args[2])
|
||||
case "daemon":
|
||||
runDaemon(ctx)
|
||||
case "watch":
|
||||
dryRun := len(os.Args) > 2 && os.Args[2] == "--dry-run"
|
||||
runWatch(ctx, dryRun)
|
||||
default:
|
||||
fmt.Println("usage: flit id | flit send <path> | flit recv <invite> | flit daemon")
|
||||
fmt.Println("usage: flit id | flit send <path> | flit recv <invite> | flit daemon | flit watch [--dry-run]")
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
532
cli/cmd/flit/watch.go
Normal file
532
cli/cmd/flit/watch.go
Normal file
@@ -0,0 +1,532 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/BurntSushi/toml"
|
||||
"github.com/fsnotify/fsnotify"
|
||||
|
||||
"flit/internal/transport"
|
||||
)
|
||||
|
||||
// watchFileConfig is read from ~/.flit/watch.toml.
|
||||
type watchFileConfig struct {
|
||||
SignalURL string `toml:"signal_url"`
|
||||
TurnURL string `toml:"turn_url"`
|
||||
TurnSecret string `toml:"turn_secret"`
|
||||
Watches []watchEntryConfig `toml:"watch"`
|
||||
}
|
||||
|
||||
type watchEntryConfig struct {
|
||||
Dir string `toml:"dir"`
|
||||
PeerID string `toml:"peer_id"`
|
||||
PeerLabel string `toml:"peer_label"`
|
||||
Pattern string `toml:"pattern"`
|
||||
StableAfter string `toml:"stable_after"`
|
||||
}
|
||||
|
||||
func watchConfigPath() string {
|
||||
home, _ := os.UserHomeDir()
|
||||
return filepath.Join(home, ".flit", "watch.toml")
|
||||
}
|
||||
|
||||
func watchStateDir() string {
|
||||
home, _ := os.UserHomeDir()
|
||||
return filepath.Join(home, ".flit", "watch-state")
|
||||
}
|
||||
|
||||
func loadWatchConfig() (*watchFileConfig, error) {
|
||||
path := watchConfigPath()
|
||||
var cfg watchFileConfig
|
||||
if _, err := toml.DecodeFile(path, &cfg); err != nil {
|
||||
return nil, fmt.Errorf("reading %s: %w", path, err)
|
||||
}
|
||||
if cfg.SignalURL == "" {
|
||||
return nil, fmt.Errorf("%s: signal_url is required", path)
|
||||
}
|
||||
if len(cfg.Watches) == 0 {
|
||||
return nil, fmt.Errorf("%s: no [[watch]] entries configured", path)
|
||||
}
|
||||
for i := range cfg.Watches {
|
||||
w := &cfg.Watches[i]
|
||||
if w.Dir == "" {
|
||||
return nil, fmt.Errorf("%s: watch entry %d missing dir", path, i)
|
||||
}
|
||||
w.Dir = expandHome(w.Dir)
|
||||
if w.PeerID == "" {
|
||||
return nil, fmt.Errorf("%s: watch entry for %s missing peer_id", path, w.Dir)
|
||||
}
|
||||
if w.PeerLabel == "" {
|
||||
w.PeerLabel = w.PeerID[:min16(w.PeerID)]
|
||||
}
|
||||
if w.Pattern == "" {
|
||||
w.Pattern = "*"
|
||||
}
|
||||
if w.StableAfter == "" {
|
||||
w.StableAfter = "10s"
|
||||
}
|
||||
}
|
||||
return &cfg, nil
|
||||
}
|
||||
|
||||
// fileState is the per-file record kept in a watch entry's state file.
|
||||
// SentAt is nil until SendFile has returned successfully for exactly this
|
||||
// size/mtime — that's the entire "needs sending" queue (see PROPOSAL-watch.md §3).
|
||||
type fileState struct {
|
||||
Size int64 `json:"size"`
|
||||
Mtime int64 `json:"mtime"`
|
||||
SHA256 string `json:"sha256,omitempty"`
|
||||
SentAt *int64 `json:"sent_at"`
|
||||
}
|
||||
|
||||
// watchEntry is one [[watch]] block: a directory watched for one peer.
|
||||
type watchEntry struct {
|
||||
cfg watchEntryConfig
|
||||
stableAfter time.Duration
|
||||
statePath string
|
||||
|
||||
mu sync.Mutex
|
||||
state map[string]fileState
|
||||
}
|
||||
|
||||
func newWatchEntry(cfg watchEntryConfig) (*watchEntry, error) {
|
||||
d, err := time.ParseDuration(cfg.StableAfter)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("dir %s: invalid stable_after %q: %w", cfg.Dir, cfg.StableAfter, err)
|
||||
}
|
||||
sum := sha256.Sum256([]byte(cfg.Dir + cfg.PeerID))
|
||||
e := &watchEntry{
|
||||
cfg: cfg,
|
||||
stableAfter: d,
|
||||
statePath: filepath.Join(watchStateDir(), hex.EncodeToString(sum[:])+".json"),
|
||||
state: map[string]fileState{},
|
||||
}
|
||||
return e, nil
|
||||
}
|
||||
|
||||
func (e *watchEntry) load() error {
|
||||
data, err := os.ReadFile(e.statePath)
|
||||
if os.IsNotExist(err) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
return json.Unmarshal(data, &e.state)
|
||||
}
|
||||
|
||||
func (e *watchEntry) save() error {
|
||||
e.mu.Lock()
|
||||
data, err := json.MarshalIndent(e.state, "", " ")
|
||||
e.mu.Unlock()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(e.statePath), 0700); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(e.statePath, data, 0600)
|
||||
}
|
||||
|
||||
// matches checks a file's base name against the entry's comma-separated
|
||||
// glob pattern list.
|
||||
func (e *watchEntry) matches(name string) bool {
|
||||
for _, pat := range strings.Split(e.cfg.Pattern, ",") {
|
||||
pat = strings.TrimSpace(pat)
|
||||
if pat == "" {
|
||||
continue
|
||||
}
|
||||
if ok, _ := filepath.Match(pat, name); ok {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// noteObserved records the current size/mtime for relPath and, if they
|
||||
// differ from what was last successfully sent (or nothing was ever sent),
|
||||
// marks the file ready to send by clearing sent_at.
|
||||
func (e *watchEntry) noteObserved(relPath string, size, mtime int64) (readyNow bool) {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
cur, ok := e.state[relPath]
|
||||
if ok && cur.SentAt != nil && cur.Size == size && cur.Mtime == mtime {
|
||||
return false // already sent this exact version
|
||||
}
|
||||
e.state[relPath] = fileState{Size: size, Mtime: mtime, SHA256: cur.SHA256, SentAt: nil}
|
||||
return true
|
||||
}
|
||||
|
||||
func (e *watchEntry) markSent(relPath string, size, mtime int64, sha string) {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
now := time.Now().Unix()
|
||||
e.state[relPath] = fileState{Size: size, Mtime: mtime, SHA256: sha, SentAt: &now}
|
||||
}
|
||||
|
||||
// pendingSorted returns relative paths currently marked as needing a send,
|
||||
// in a deterministic order.
|
||||
func (e *watchEntry) pendingSorted() []string {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
var out []string
|
||||
for rel, st := range e.state {
|
||||
if st.SentAt == nil {
|
||||
out = append(out, rel)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// reconcile walks the watched directory and updates in-memory state for
|
||||
// anything that's new or changed since it was last recorded (PROPOSAL-watch.md §4).
|
||||
// It does not touch the filesystem beyond reading; call save() afterward to persist.
|
||||
func (e *watchEntry) reconcile() error {
|
||||
entries, err := os.ReadDir(e.cfg.Dir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, de := range entries {
|
||||
if de.IsDir() || !e.matches(de.Name()) {
|
||||
continue
|
||||
}
|
||||
info, err := de.Info()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
e.noteObserved(de.Name(), info.Size(), info.ModTime().Unix())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// watchPeerGroup shares one persistent connection across every [[watch]]
|
||||
// entry targeting the same peer_id (PROPOSAL-watch.md open question #2).
|
||||
type watchPeerGroup struct {
|
||||
peerID string
|
||||
label string
|
||||
entries []*watchEntry
|
||||
|
||||
mu sync.Mutex
|
||||
sess *transport.Session
|
||||
|
||||
sendMu sync.Mutex
|
||||
}
|
||||
|
||||
func (g *watchPeerGroup) currentSession() *transport.Session {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
return g.sess
|
||||
}
|
||||
|
||||
// sendOne sends one file and, on success, advances that file's state past
|
||||
// "needs sending". On failure the state is left untouched, so the file
|
||||
// remains eligible for retry on the next drain (PROPOSAL-watch.md §3).
|
||||
func (g *watchPeerGroup) sendOne(sess *transport.Session, e *watchEntry, rel string) error {
|
||||
g.sendMu.Lock()
|
||||
defer g.sendMu.Unlock()
|
||||
|
||||
full := filepath.Join(e.cfg.Dir, rel)
|
||||
info, err := os.Stat(full)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("[%s] → %s", g.label, rel)
|
||||
if err := sess.SendFile(full); err != nil {
|
||||
return err
|
||||
}
|
||||
e.markSent(rel, info.Size(), info.ModTime().Unix(), "")
|
||||
if err := e.save(); err != nil {
|
||||
log.Printf("[%s] warning: could not persist state for %s: %v", g.label, e.cfg.Dir, err)
|
||||
}
|
||||
log.Printf("[%s] ✓ sent %s", g.label, rel)
|
||||
return nil
|
||||
}
|
||||
|
||||
// drainAll sends every currently-pending file across every entry in the
|
||||
// group, sequentially (PROPOSAL-watch.md §5). Stops early on the first
|
||||
// send error, since that almost always means the connection just dropped;
|
||||
// remaining files stay queued for the next successful connect.
|
||||
func (g *watchPeerGroup) drainAll() {
|
||||
sess := g.currentSession()
|
||||
if sess == nil {
|
||||
return
|
||||
}
|
||||
for _, e := range g.entries {
|
||||
for _, rel := range e.pendingSorted() {
|
||||
if g.currentSession() != sess {
|
||||
return // disconnected mid-drain
|
||||
}
|
||||
if err := g.sendOne(sess, e, rel); err != nil {
|
||||
log.Printf("[%s] send failed for %s: %v", g.label, rel, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// attemptSendNow sends rel immediately if the group is currently connected.
|
||||
// If not connected, the file simply stays marked pending — the next
|
||||
// successful connect's drainAll will pick it up.
|
||||
func (g *watchPeerGroup) attemptSendNow(e *watchEntry, rel string) {
|
||||
sess := g.currentSession()
|
||||
if sess == nil {
|
||||
return
|
||||
}
|
||||
if err := g.sendOne(sess, e, rel); err != nil {
|
||||
log.Printf("[%s] send failed for %s: %v", g.label, rel, err)
|
||||
}
|
||||
}
|
||||
|
||||
// runWatchGroup mirrors daemon.go's runPeerLoop: a persistent connection to
|
||||
// one trusted peer, reconnecting with the same backoff. It differs only in
|
||||
// what drives sends — pending state across the group's entries, rather than
|
||||
// incoming OnFileOffer.
|
||||
func runWatchGroup(ctx context.Context, tc transport.Config, myID string, g *watchPeerGroup) {
|
||||
room := transport.PairRoomName(myID, g.peerID)
|
||||
log.Printf("[%s] room: %s", g.label, room)
|
||||
backoff := 2 * time.Second
|
||||
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf("[%s] connecting…", g.label)
|
||||
disconnected := make(chan struct{})
|
||||
|
||||
sess, err := transport.NewSession(dataDir(), tc)
|
||||
if err != nil {
|
||||
log.Printf("[%s] session error: %v", g.label, err)
|
||||
} else {
|
||||
sess.OnConnected = func(verified bool) {
|
||||
status := "unverified"
|
||||
if verified {
|
||||
status = "verified"
|
||||
}
|
||||
log.Printf("[%s] connected (%s)", g.label, status)
|
||||
backoff = 2 * time.Second
|
||||
g.mu.Lock()
|
||||
g.sess = sess
|
||||
g.mu.Unlock()
|
||||
go g.drainAll()
|
||||
}
|
||||
sess.OnDisconnected = func() {
|
||||
g.mu.Lock()
|
||||
if g.sess == sess {
|
||||
g.sess = nil
|
||||
}
|
||||
g.mu.Unlock()
|
||||
select {
|
||||
case <-disconnected:
|
||||
default:
|
||||
close(disconnected)
|
||||
}
|
||||
}
|
||||
|
||||
if err := sess.Join(ctx, room, g.peerID); err != nil {
|
||||
log.Printf("[%s] join error: %v", g.label, err)
|
||||
} else {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-disconnected:
|
||||
log.Printf("[%s] disconnected", g.label)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(backoff):
|
||||
}
|
||||
if backoff < 30*time.Second {
|
||||
backoff *= 2
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// watchDirLive subscribes to filesystem events for one entry and, after a
|
||||
// stable_after quiet period with no further writes to a given file, marks
|
||||
// it ready and (if connected) sends it immediately. Deletes/renames are
|
||||
// deliberately ignored (PROPOSAL-watch.md §7) — this is push-only, one-way.
|
||||
func watchDirLive(ctx context.Context, e *watchEntry, g *watchPeerGroup) error {
|
||||
w, err := fsnotify.NewWatcher()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer w.Close()
|
||||
if err := w.Add(e.cfg.Dir); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var mu sync.Mutex
|
||||
timers := map[string]*time.Timer{}
|
||||
|
||||
fire := func(name string) {
|
||||
full := filepath.Join(e.cfg.Dir, name)
|
||||
info, err := os.Stat(full)
|
||||
if err != nil {
|
||||
return // removed again before it went stable
|
||||
}
|
||||
if e.noteObserved(name, info.Size(), info.ModTime().Unix()) {
|
||||
_ = e.save()
|
||||
g.attemptSendNow(e, name)
|
||||
}
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
mu.Lock()
|
||||
for _, t := range timers {
|
||||
t.Stop()
|
||||
}
|
||||
mu.Unlock()
|
||||
return nil
|
||||
case ev, ok := <-w.Events:
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
if ev.Op&(fsnotify.Write|fsnotify.Create) == 0 {
|
||||
continue
|
||||
}
|
||||
name := filepath.Base(ev.Name)
|
||||
if !e.matches(name) {
|
||||
continue
|
||||
}
|
||||
mu.Lock()
|
||||
if t, ok := timers[name]; ok {
|
||||
t.Stop()
|
||||
}
|
||||
timers[name] = time.AfterFunc(e.stableAfter, func() { fire(name) })
|
||||
mu.Unlock()
|
||||
case err, ok := <-w.Errors:
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
log.Printf("watch %s: fsnotify error: %v", e.cfg.Dir, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func runWatch(ctx context.Context, dryRun bool) {
|
||||
cfg, err := loadWatchConfig()
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "flit watch:", err)
|
||||
fmt.Fprintf(os.Stderr, "create %s — see watch.toml.example\n", watchConfigPath())
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
entries := make([]*watchEntry, 0, len(cfg.Watches))
|
||||
for _, wc := range cfg.Watches {
|
||||
if _, err := os.Stat(wc.Dir); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "flit watch: %s: %v\n", wc.Dir, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
e, err := newWatchEntry(wc)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "flit watch:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
entries = append(entries, e)
|
||||
}
|
||||
|
||||
if dryRun {
|
||||
for _, e := range entries {
|
||||
if err := e.load(); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "flit watch: %s: reading state: %v\n", e.cfg.Dir, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
if err := e.reconcile(); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "flit watch: %s: %v\n", e.cfg.Dir, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
pending := e.pendingSorted()
|
||||
fmt.Printf("%s → %s (%d file(s) would be sent):\n", e.cfg.Dir, e.cfg.PeerLabel, len(pending))
|
||||
for _, rel := range pending {
|
||||
fmt.Printf(" %s\n", rel)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
for _, e := range entries {
|
||||
if err := e.load(); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "flit watch: %s: reading state: %v\n", e.cfg.Dir, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
tc := transport.Config{SignalURL: cfg.SignalURL, TurnURL: cfg.TurnURL, TurnSecret: cfg.TurnSecret}
|
||||
|
||||
probe, err := transport.NewSession(dataDir(), tc)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "flit watch: identity error:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
myID := probe.PeerID()
|
||||
|
||||
groups := map[string]*watchPeerGroup{}
|
||||
var order []string
|
||||
for _, e := range entries {
|
||||
g, ok := groups[e.cfg.PeerID]
|
||||
if !ok {
|
||||
g = &watchPeerGroup{peerID: e.cfg.PeerID, label: e.cfg.PeerLabel}
|
||||
groups[e.cfg.PeerID] = g
|
||||
order = append(order, e.cfg.PeerID)
|
||||
}
|
||||
g.entries = append(g.entries, e)
|
||||
}
|
||||
|
||||
log.Printf("flit watch starting — id %s…", myID[:16])
|
||||
log.Printf(" anchor : %s", cfg.SignalURL)
|
||||
log.Printf(" dirs : %d, peers : %d", len(entries), len(groups))
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for _, e := range entries {
|
||||
if err := e.reconcile(); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "flit watch: %s: %v\n", e.cfg.Dir, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
if err := e.save(); err != nil {
|
||||
log.Printf("warning: could not persist state for %s: %v", e.cfg.Dir, err)
|
||||
}
|
||||
}
|
||||
|
||||
for _, peerID := range order {
|
||||
g := groups[peerID]
|
||||
wg.Add(1)
|
||||
go func(g *watchPeerGroup) {
|
||||
defer wg.Done()
|
||||
runWatchGroup(ctx, tc, myID, g)
|
||||
}(g)
|
||||
|
||||
for _, e := range g.entries {
|
||||
wg.Add(1)
|
||||
go func(e *watchEntry, g *watchPeerGroup) {
|
||||
defer wg.Done()
|
||||
if err := watchDirLive(ctx, e, g); err != nil {
|
||||
log.Printf("watch %s: %v", e.cfg.Dir, err)
|
||||
}
|
||||
}(e, g)
|
||||
}
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
log.Println("flit watch stopped")
|
||||
}
|
||||
@@ -13,6 +13,7 @@ require (
|
||||
require (
|
||||
github.com/BurntSushi/toml v1.6.0 // indirect
|
||||
github.com/davecgh/go-spew v1.1.1 // indirect
|
||||
github.com/fsnotify/fsnotify v1.10.1 // indirect
|
||||
github.com/google/uuid v1.3.1 // indirect
|
||||
github.com/pion/datachannel v1.5.8 // indirect
|
||||
github.com/pion/dtls/v2 v2.2.12 // indirect
|
||||
|
||||
@@ -5,6 +5,8 @@ github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho=
|
||||
github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo=
|
||||
github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4=
|
||||
github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
|
||||
|
||||
@@ -172,6 +172,7 @@ type Session struct {
|
||||
dc *webrtc.DataChannel
|
||||
peerID string
|
||||
verified bool
|
||||
peerAuthed bool // set once a sealed box from peerID has been opened successfully
|
||||
esk, epk *[32]byte
|
||||
peerEPK *[32]byte
|
||||
ekeySent bool
|
||||
@@ -423,6 +424,9 @@ func (s *Session) onBox(from, box string) {
|
||||
if plain == nil {
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
s.peerAuthed = true
|
||||
s.mu.Unlock()
|
||||
var obj map[string]any
|
||||
if err := json.Unmarshal(plain, &obj); err != nil {
|
||||
return
|
||||
@@ -573,20 +577,25 @@ func (s *Session) sendHello() {
|
||||
if pc == nil {
|
||||
return
|
||||
}
|
||||
lfp, err1 := dtlsFP(pc.LocalDescription().SDP)
|
||||
rfp, err2 := dtlsFP(pc.RemoteDescription().SDP)
|
||||
if err1 != nil || err2 != nil {
|
||||
log.Printf("flit: sendHello: fingerprint error: %v / %v", err1, err2)
|
||||
return
|
||||
}
|
||||
// §6: sender signs prefix || local_fp || remote_fp
|
||||
bindMsg := append([]byte(bindPrefix), lfp...)
|
||||
bindMsg = append(bindMsg, rfp...)
|
||||
s.controlSend(map[string]string{
|
||||
msg := map[string]string{
|
||||
"type": "hello",
|
||||
"id": s.identity.PeerID(),
|
||||
"sig": s.identity.Sign(bindMsg),
|
||||
})
|
||||
}
|
||||
// §6: sign prefix || local_fp || remote_fp when both are extractable.
|
||||
// This is best-effort, not required — different WebRTC stacks report the
|
||||
// DTLS certificate hash under different algorithms (WebKit: sha-512,
|
||||
// pion/Chromium: sha-256), so a legitimate cross-stack peer may have no
|
||||
// fingerprint we know how to parse. Trust rests on the sealed signaling
|
||||
// exchange that already authenticated s.peerID (see onBox/peerAuthed)
|
||||
// plus hello.id matching it, not on this signature — see onControl.
|
||||
if lfp, err1 := dtlsFP(pc.LocalDescription().SDP); err1 == nil {
|
||||
if rfp, err2 := dtlsFP(pc.RemoteDescription().SDP); err2 == nil {
|
||||
bindMsg := append([]byte(bindPrefix), lfp...)
|
||||
bindMsg = append(bindMsg, rfp...)
|
||||
msg["sig"] = s.identity.Sign(bindMsg)
|
||||
}
|
||||
}
|
||||
s.controlSend(msg)
|
||||
}
|
||||
|
||||
func (s *Session) onControl(data []byte) {
|
||||
@@ -598,11 +607,18 @@ func (s *Session) onControl(data []byte) {
|
||||
case "hello":
|
||||
helloID, _ := m["id"].(string)
|
||||
sigHex, _ := m["sig"].(string)
|
||||
verified := false
|
||||
s.mu.Lock()
|
||||
pc := s.pc
|
||||
peerAuthed := s.peerAuthed
|
||||
s.mu.Unlock()
|
||||
if pc != nil && pc.RemoteDescription() != nil && pc.LocalDescription() != nil {
|
||||
// Trust rests on the sealed signaling exchange that already
|
||||
// authenticated s.peerID (peerAuthed, set in onBox) plus hello.id
|
||||
// matching it — not on the fingerprint-bound signature below,
|
||||
// which is only checkable when both sides report their DTLS cert
|
||||
// hash under an algorithm we can parse. See sendHello for why
|
||||
// that can't be a hard requirement.
|
||||
verified := helloID == s.peerID && peerAuthed
|
||||
if sigHex != "" && pc != nil && pc.RemoteDescription() != nil && pc.LocalDescription() != nil {
|
||||
// §6: verifier reconstructs prefix || remote_fp || local_fp
|
||||
// (sender's local = our remote; sender's remote = our local)
|
||||
rfp, err1 := dtlsFP(pc.RemoteDescription().SDP)
|
||||
@@ -610,9 +626,9 @@ func (s *Session) onControl(data []byte) {
|
||||
if err1 == nil && err2 == nil {
|
||||
bindMsg := append([]byte(bindPrefix), rfp...)
|
||||
bindMsg = append(bindMsg, lfp...)
|
||||
verified = helloID == s.peerID && crypto.Verify(helloID, bindMsg, sigHex) == nil
|
||||
} else {
|
||||
log.Printf("flit: hello verify: fingerprint error: %v / %v", err1, err2)
|
||||
if err := crypto.Verify(helloID, bindMsg, sigHex); err != nil {
|
||||
log.Printf("flit: hello: fingerprint-bind signature did not verify for %s (informational only): %v", helloID, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
s.mu.Lock()
|
||||
@@ -758,11 +774,12 @@ func (s *Session) SendFile(path string) error {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// §9: send file-done before closing the file DC so the control channel
|
||||
// is still reachable when the message is delivered.
|
||||
s.controlSend(map[string]string{"type": "file-done", "xid": xid, "sha256": sha256hex})
|
||||
if err := dc.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
// §9: send file-done with the pre-computed hash so the receiver can verify.
|
||||
s.controlSend(map[string]string{"type": "file-done", "xid": xid, "sha256": sha256hex})
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -776,13 +793,20 @@ func (s *Session) AcceptOffer(offer FileOffer, downloadDir string) {
|
||||
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)})
|
||||
// Register doneCh before sending file-accept so file-done can never arrive
|
||||
// before the channel exists (small files finish before wireFileRecv runs).
|
||||
doneCh := make(chan string, 1)
|
||||
s.mu.Lock()
|
||||
if s.pendingDones == nil {
|
||||
s.pendingDones = make(map[string]chan string)
|
||||
}
|
||||
s.pendingDones[offer.XID] = doneCh
|
||||
if s.pendingRecvs == nil {
|
||||
s.pendingRecvs = make(map[string]*recvState)
|
||||
}
|
||||
s.pendingRecvs[offer.XID] = &recvState{offer: offer, dir: downloadDir, have: offset}
|
||||
s.pendingRecvs[offer.XID] = &recvState{offer: offer, dir: downloadDir, have: offset, doneCh: doneCh}
|
||||
s.mu.Unlock()
|
||||
s.controlSend(map[string]string{"type": "file-accept", "xid": offer.XID, "offset": fmt.Sprint(offset)})
|
||||
}
|
||||
|
||||
func (s *Session) RejectOffer(xid string) {
|
||||
@@ -790,10 +814,11 @@ func (s *Session) RejectOffer(xid string) {
|
||||
}
|
||||
|
||||
type recvState struct {
|
||||
offer FileOffer
|
||||
dir string
|
||||
have int64
|
||||
f *os.File
|
||||
offer FileOffer
|
||||
dir string
|
||||
have int64
|
||||
f *os.File
|
||||
doneCh chan string
|
||||
}
|
||||
|
||||
func (s *Session) wireFileRecv(dc *webrtc.DataChannel) {
|
||||
@@ -884,13 +909,9 @@ func (s *Session) wireFileRecv(dc *webrtc.DataChannel) {
|
||||
}
|
||||
|
||||
// §9: wait for file-done and verify sha256 before reporting success.
|
||||
doneCh := make(chan string, 1)
|
||||
s.mu.Lock()
|
||||
if s.pendingDones == nil {
|
||||
s.pendingDones = make(map[string]chan string)
|
||||
}
|
||||
s.pendingDones[xid] = doneCh
|
||||
s.mu.Unlock()
|
||||
// doneCh was registered in AcceptOffer before file-accept was sent,
|
||||
// so file-done can never arrive before the channel exists.
|
||||
doneCh := rs.doneCh
|
||||
var expectedSHA256 string
|
||||
select {
|
||||
case expectedSHA256 = <-doneCh:
|
||||
|
||||
27
cli/watch.toml.example
Normal file
27
cli/watch.toml.example
Normal file
@@ -0,0 +1,27 @@
|
||||
# flit watch config — copy to ~/.flit/watch.toml and fill in.
|
||||
#
|
||||
# Run with: flit watch
|
||||
# Sanity-check what would be sent, without connecting to anything:
|
||||
# flit watch --dry-run
|
||||
#
|
||||
# One-way sync only: new/changed files in `dir` are pushed to `peer_id`.
|
||||
# Deletes are never propagated. The receiving device just needs to be
|
||||
# running `flit daemon` with this device listed as a trusted peer.
|
||||
|
||||
# Required — WebSocket URL of your waste-go anchor
|
||||
signal_url = "wss://your-anchor.example.com/ws"
|
||||
|
||||
# Optional — TURN relay for cross-network transfers
|
||||
turn_url = "turn:your-anchor.example.com:3478"
|
||||
turn_secret = "" # coturn use-auth-secret; leave empty to use STUN only
|
||||
|
||||
# One [[watch]] block per watched directory. Multiple entries may target
|
||||
# different peers, or the same peer — entries sharing a peer_id share one
|
||||
# connection.
|
||||
|
||||
[[watch]]
|
||||
dir = "~/backups/documents"
|
||||
peer_id = "aabbccddeeff..." # hex Ed25519 pubkey of the destination device
|
||||
peer_label = "home-server"
|
||||
pattern = "*.pdf,*.docx" # comma-separated glob include-list; default "*" if omitted
|
||||
stable_after = "10s" # quiet period after the last write before a file is considered ready
|
||||
24
pwa/package-lock.json
generated
24
pwa/package-lock.json
generated
@@ -611,6 +611,29 @@
|
||||
"node": "^20.19.0 || >=22.12.0"
|
||||
}
|
||||
},
|
||||
"node_modules/@rolldown/binding-wasm32-wasi/node_modules/@emnapi/core": {
|
||||
"version": "1.11.1",
|
||||
"resolved": "https://registry.npmjs.org/@emnapi/core/-/core-1.11.1.tgz",
|
||||
"integrity": "sha512-RSvbQmHzdKzNsLYa/wHrbc3KN4sYLKAdPZxqiM2HATqv/SBk2/ENSHpvXGaLOMcsAyz0poEGqkmmKYG3OWiJEQ==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"dependencies": {
|
||||
"@emnapi/wasi-threads": "1.2.2",
|
||||
"tslib": "^2.4.0"
|
||||
}
|
||||
},
|
||||
"node_modules/@rolldown/binding-wasm32-wasi/node_modules/@emnapi/runtime": {
|
||||
"version": "1.11.1",
|
||||
"resolved": "https://registry.npmjs.org/@emnapi/runtime/-/runtime-1.11.1.tgz",
|
||||
"integrity": "sha512-vgj7R3y3Wgx24IQaGPA/R6YFXLHVMOZ0uVEyIQPaWs+rd1AzfEMXlAC22FYwO1XkKR6NPsq7mUandH8oIRdZFw==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"dependencies": {
|
||||
"tslib": "^2.4.0"
|
||||
}
|
||||
},
|
||||
"node_modules/@rolldown/binding-win32-arm64-msvc": {
|
||||
"version": "1.1.3",
|
||||
"resolved": "https://registry.npmjs.org/@rolldown/binding-win32-arm64-msvc/-/binding-win32-arm64-msvc-1.1.3.tgz",
|
||||
@@ -676,7 +699,6 @@
|
||||
"integrity": "sha512-fRa09kZTgu8o71KFcDjUFuc7F+dEbZYZmkI0mg5YBTRs0yMKjYHsq/c0urDKeDb+D5qVgXOdFcuu+DZPKOITwA==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"undici-types": "~7.18.0"
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
"name": "pwa",
|
||||
"private": true,
|
||||
"version": "0.0.0",
|
||||
"license": "MIT",
|
||||
"type": "module",
|
||||
"scripts": {
|
||||
"dev": "vite",
|
||||
|
||||
@@ -278,6 +278,19 @@
|
||||
font-weight: 400;
|
||||
}
|
||||
|
||||
.presence-dot {
|
||||
display: inline-block;
|
||||
width: 8px;
|
||||
height: 8px;
|
||||
border-radius: 50%;
|
||||
margin-right: 6px;
|
||||
vertical-align: middle;
|
||||
}
|
||||
|
||||
.presence-online { background: var(--accent); box-shadow: 0 0 4px var(--accent); }
|
||||
.presence-offline { background: var(--muted); opacity: 0.5; }
|
||||
.presence-unknown { background: var(--muted); opacity: 0.25; }
|
||||
|
||||
.peer-id {
|
||||
font-family: var(--mono);
|
||||
font-size: 11px;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { useEffect, useRef, useState } from 'react'
|
||||
import QRCode from 'qrcode'
|
||||
import { FlitSession, type FileOfferEvent, type FileProgressEvent, type FileRecvEvent, type FileSendProgressEvent, type FileSentEvent, type FlitConfig } from './transport/flit'
|
||||
import { FlitSession, queryPresence, netHash, type FileOfferEvent, type FileProgressEvent, type FileRecvEvent, type FileSendProgressEvent, type FileSentEvent, type FlitConfig } from './transport/flit'
|
||||
import { createInvite, decodeInvite, pairRoomName } from './pairing/ephemeral'
|
||||
import { QrScanner } from './pairing/QrScanner'
|
||||
import { addTrusted, listTrusted, removeTrusted, setAutoAccept, shouldAutoAccept, type TrustedPeer } from './pairing/keyring'
|
||||
@@ -40,6 +40,7 @@ function App() {
|
||||
const [sentFiles, setSentFiles] = useState<FileSentEvent[]>([])
|
||||
const [scanning, setScanning] = useState(false)
|
||||
const [myId, setMyId] = useState<string | null>(null)
|
||||
const [presence, setPresence] = useState<Record<string, boolean | null>>({})
|
||||
const [addIdInput, setAddIdInput] = useState('')
|
||||
const [copied, setCopied] = useState<'snip' | 'id' | null>(null)
|
||||
const sessionRef = useRef<FlitSession | null>(null)
|
||||
@@ -65,6 +66,32 @@ function App() {
|
||||
return c
|
||||
}
|
||||
|
||||
// Presence check for known devices (waste-go EXT-009). Each pairing has its
|
||||
// own deterministic anchor room, so this is one short-lived query per
|
||||
// device — no persistent connection held while idle.
|
||||
async function refreshPresence() {
|
||||
if (!myId || trusted.length === 0) return
|
||||
const signalURL = cfg().signalURL
|
||||
const results = await Promise.all(trusted.map(async (p) => {
|
||||
try {
|
||||
const hash = await netHash(pairRoomName(myId, p.id))
|
||||
const online = await queryPresence(signalURL, hash, p.id)
|
||||
return [p.id, online] as const
|
||||
} catch {
|
||||
return [p.id, null] as const
|
||||
}
|
||||
}))
|
||||
setPresence(Object.fromEntries(results))
|
||||
}
|
||||
|
||||
useEffect(() => {
|
||||
if (mode !== 'known' || phase !== 'idle' || !myId || trusted.length === 0) return
|
||||
void refreshPresence()
|
||||
const iv = setInterval(() => void refreshPresence(), 15000)
|
||||
return () => clearInterval(iv)
|
||||
// eslint-disable-next-line react-hooks/exhaustive-deps
|
||||
}, [mode, phase, myId, trusted])
|
||||
|
||||
function peerEvents() {
|
||||
return {
|
||||
connected: (v: boolean, pid: string) => { setVerified(v); verifiedRef.current = v; setPeerId(pid); setPhase('connected') },
|
||||
@@ -259,7 +286,13 @@ function App() {
|
||||
<li key={p.id} className="known-row">
|
||||
<div className="known-top">
|
||||
<div className="known-info">
|
||||
<span className="known-label">{p.label}</span>
|
||||
<span className="known-label">
|
||||
<span
|
||||
className={`presence-dot ${presence[p.id] === true ? 'presence-online' : presence[p.id] === false ? 'presence-offline' : 'presence-unknown'}`}
|
||||
title={presence[p.id] === true ? 'Online' : presence[p.id] === false ? 'Offline' : 'Unknown'}
|
||||
/>
|
||||
{p.label}
|
||||
</span>
|
||||
<span className="peer-id">{p.id.slice(0, 16)}…</span>
|
||||
</div>
|
||||
<div className="known-secondary">
|
||||
|
||||
@@ -79,6 +79,38 @@ export async function netHash(name: string): Promise<string> {
|
||||
throw new Error('SHA-256 not available in this browser runtime')
|
||||
}
|
||||
|
||||
// queryPresence asks the anchor whether `id` is currently connected in
|
||||
// network `net` (waste-go EXT-009). Opens a short-lived WS connection —
|
||||
// no join, no identity signature required, since the query is anchor-local
|
||||
// and scoped to a (net, id) pair the caller must already know. Resolves
|
||||
// `null` on timeout/error (treat the same as "unknown", not "offline").
|
||||
export function queryPresence(url: string, net: string, id: string, timeoutMs = 4000): Promise<boolean | null> {
|
||||
return new Promise((resolve) => {
|
||||
let settled = false
|
||||
const ws = new WebSocket(url)
|
||||
const finish = (result: boolean | null) => {
|
||||
if (settled) return
|
||||
settled = true
|
||||
clearTimeout(timer)
|
||||
try { ws.close() } catch { /* ignore */ }
|
||||
resolve(result)
|
||||
}
|
||||
const timer = setTimeout(() => finish(null), timeoutMs)
|
||||
ws.onopen = () => {
|
||||
try { ws.send(JSON.stringify({ type: 'presence_query', net, id })) } catch { finish(null) }
|
||||
}
|
||||
ws.onmessage = (ev) => {
|
||||
let m: Record<string, unknown>
|
||||
try { m = JSON.parse(ev.data) } catch { return }
|
||||
if (m['type'] === 'presence' && m['id'] === id && m['net'] === net) {
|
||||
finish(Boolean(m['online']))
|
||||
}
|
||||
}
|
||||
ws.onerror = () => finish(null)
|
||||
ws.onclose = () => finish(null)
|
||||
})
|
||||
}
|
||||
|
||||
// ── Identity ─────────────────────────────────────────────────────────────────
|
||||
|
||||
export class Identity {
|
||||
@@ -403,15 +435,23 @@ export class PeerConn {
|
||||
}
|
||||
|
||||
private _sendHello() {
|
||||
// §6: sign prefix || local_fp || remote_fp when both are extractable.
|
||||
// Best-effort, not required — different WebRTC stacks report the DTLS
|
||||
// certificate hash under different algorithms (WebKit: sha-512,
|
||||
// pion/Chromium: sha-256), so a legitimate cross-stack peer may have no
|
||||
// fingerprint we know how to parse. Trust rests on the sealed signaling
|
||||
// exchange that already authenticated peerId (peerAuthed, set in onBox)
|
||||
// plus hello.id matching it, not on this signature — see _onControl.
|
||||
const msg: Record<string, string> = { type: 'hello', id: this.identity.id }
|
||||
try {
|
||||
const lfp = dtlsFP(this.pc.localDescription!.sdp)
|
||||
const rfp = dtlsFP(this.pc.remoteDescription!.sdp)
|
||||
// §6: sender signs prefix || local_fp || remote_fp
|
||||
const bindMsg = concat(enc(BIND_PREFIX), lfp, rfp)
|
||||
this._dc({ type: 'hello', id: this.identity.id, sig: sodium.to_hex(this.identity.sign(bindMsg)) })
|
||||
} catch (e) {
|
||||
console.error('flit: sendHello fingerprint error', e)
|
||||
msg['sig'] = sodium.to_hex(this.identity.sign(bindMsg))
|
||||
} catch {
|
||||
// no parseable fingerprint on one or both sides — hello still goes out
|
||||
}
|
||||
this._dc(msg)
|
||||
}
|
||||
|
||||
private _onControl(data: string) {
|
||||
@@ -420,17 +460,28 @@ export class PeerConn {
|
||||
|
||||
if (m['type'] === 'hello') {
|
||||
const helloId = m['id'] as string
|
||||
const helloSig = sodium.from_hex(m['sig'] as string)
|
||||
try {
|
||||
// §6: verifier reconstructs prefix || remote_fp || local_fp
|
||||
// (sender's local = our remote; sender's remote = our local)
|
||||
const rfp = dtlsFP(this.pc.remoteDescription!.sdp)
|
||||
const lfp = dtlsFP(this.pc.localDescription!.sdp)
|
||||
const bindMsg = concat(enc(BIND_PREFIX), rfp, lfp)
|
||||
const sigOk = Identity.verify(helloId, bindMsg, helloSig)
|
||||
this.verified = helloId === this.peerId && sigOk && this.peerAuthed
|
||||
} catch {
|
||||
this.verified = false
|
||||
// Trust rests on the sealed signaling exchange that already
|
||||
// authenticated peerId (peerAuthed, set in onBox) plus hello.id
|
||||
// matching it — not on the fingerprint-bound signature below, which
|
||||
// is only checkable when both sides report their DTLS cert hash
|
||||
// under an algorithm we can parse. See _sendHello for why that
|
||||
// can't be a hard requirement.
|
||||
this.verified = helloId === this.peerId && this.peerAuthed
|
||||
const sigHex = m['sig'] as string | undefined
|
||||
if (sigHex) {
|
||||
try {
|
||||
// §6: verifier reconstructs prefix || remote_fp || local_fp
|
||||
// (sender's local = our remote; sender's remote = our local)
|
||||
const rfp = dtlsFP(this.pc.remoteDescription!.sdp)
|
||||
const lfp = dtlsFP(this.pc.localDescription!.sdp)
|
||||
const bindMsg = concat(enc(BIND_PREFIX), rfp, lfp)
|
||||
const helloSig = sodium.from_hex(sigHex)
|
||||
if (!Identity.verify(helloId, bindMsg, helloSig)) {
|
||||
console.warn(`flit: hello: fingerprint-bind signature did not verify for ${helloId} (informational only)`)
|
||||
}
|
||||
} catch {
|
||||
// no parseable fingerprint on one or both sides — nothing to check
|
||||
}
|
||||
}
|
||||
this.events.connected?.(this.verified, this.peerId)
|
||||
} else if (m['type'] === 'file-offer') {
|
||||
@@ -510,9 +561,10 @@ export class PeerConn {
|
||||
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()
|
||||
// §9: send file-done with sha256 so receiver can verify integrity.
|
||||
// §9: send file-done before closing the file DC so the control channel
|
||||
// is still healthy when the message is dispatched.
|
||||
this._dc({ type: 'file-done', xid, sha256: hash })
|
||||
dc.close()
|
||||
this.events.fileSent?.({ peer: this.peerId, xid, name: file.name })
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user