diff --git a/.changeset/cloud-quic.md b/.changeset/cloud-quic.md new file mode 100644 index 00000000..ebe6e17b --- /dev/null +++ b/.changeset/cloud-quic.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +The FTW Cloud uplink tries QUIC on UDP 443 first and falls back to the WebSocket when UDP does not get through within 250 ms, remembering per network which one opened. A QUIC session that ends before `hello_ok`, or that the path drops within a minute of it, puts the WebSocket first on that network for a day. Over QUIC, `hb` and a `live` without its source table go as datagrams, and the tunnel's bulk on a stream of its own, so history never holds up a command. Noise sessions accept frames up to 1024 out of order, since bulk can overtake lane 0. `GET /api/cloud` says which transport carries the link. The registry names them in `cloud_transports`. diff --git a/contract/registry.yaml b/contract/registry.yaml index 10941bad..ca5bdb6c 100644 --- a/contract/registry.yaml +++ b/contract/registry.yaml @@ -363,6 +363,11 @@ source_states: [live, lagging, stale, down, never] # tunnel is the box's own session carried by FTW Cloud, which cannot read it. # cloud is the box-signed copy FTW Cloud holds and can read. carrier_states: [webrtc, relay, tunnel, cloud, cache, none] +# How a session with FTW Cloud travels, under the tunnel and cloud carriers: +# QUIC on UDP 443, or the WebSocket on TCP 443 that a client falls back to +# (srcfl/ftw docs/protocol-cloud.md, "QUIC"). It says nothing about +# freshness, which stays carrier and source state. +cloud_transports: [quic, websocket] # --------------------------------------------------------------------------- # History resolutions backed by what the box actually stores. diff --git a/docs/protocol-cloud.md b/docs/protocol-cloud.md index ce450287..f36cdb0a 100644 --- a/docs/protocol-cloud.md +++ b/docs/protocol-cloud.md @@ -92,6 +92,151 @@ data. The network sees when messages go and how large they are, which shows which class changed and when. The privacy notice says so (ADR 0008, decision 5). +## QUIC + +The uplink and the app session also run over QUIC (RFC 9000) on UDP 443 at +`cloud.ftw.energy`, with the same certificate as TCP. The messages and every +rule in this document stay the same; only the framing below changes. The +WebSocket stays too, and a client falls back to it when UDP does not get +through. The one-shot read and the extension read are plain HTTPS, which the +service also serves over HTTP/3 (RFC 9114); they need nothing here. +`go/protocol/cloudwire/quic.go` holds the names and the codec, and +`cloud-quic-vectors.json` the bytes. + +| ALPN | Connection | WebSocket it stands for | +|---|---|---| +| `ftw-cloud-box/1` | Uplink | `/v1/box` | +| `ftw-cloud-app/1` | App session | `/v1/app?home=` | +| `h3` | HTTP/3 | — | + +The service offers only the ALPNs it serves. With the sessions turned off it +offers `h3` alone, so a client's QUIC handshake fails at once and it takes +the WebSocket. + +- **TLS 1.3 inside QUIC.** `ekm` is the exporter of that TLS session, with the + label and length of [Handshake](#handshake), and a box binds `hello` to it + as it does over TCP. Nothing may end QUIC in front of the service; a load + balancer that forwards UDP packets is fine. +- **Message.** One WebSocket binary message becomes one message: its length + as a QUIC variable-length integer (RFC 9000 §16), then its bytes. The size + limits of the WebSocket apply. A length over them closes the connection + with 4400. +- **Control stream.** The client opens bidirectional stream 0 and sends the + *target* first, as one message: the path and query the WebSocket would name, + such as `/v1/app?home=`, in ASCII. The service checks it as it checks + the WebSocket's URL. After it, every message of the connection goes on this + stream, both ways, except those below. +- **Bulk stream.** A sender may open one unidirectional stream and send on it, + in order, the `tun` messages whose `d` carries a lane 1 frame of the + phone's session: snapshots, history and answers too large for lane 0. It + opens it the first time it has bulk to send. The box already tells bulk + from lane 0 for its pacing (see [Backpressure](#backpressure-and-limits)). + The service sends a `tun` that came on the box's bulk stream on the phone's + bulk stream. A lost bulk packet then never holds up a command, its answer + or a tick. A phone sends everything on the control stream. +- **Datagrams (RFC 9221).** The box sends `hb`, and `live` without `sources`, + as one QUIC DATAGRAM each when it fits, and otherwise on the control stream. + The service sends those same envelopes to a phone the same way, each as one + Noise transport message. A datagram is never sent again, and the newest + wins. The [freshness](#freshness) rules already cover one that is lost or + late: a phone sees the gap through `hb.liveSeq` and ages what it holds + instead of showing it as live. A `live` that carries `sources` goes on the + control stream, because a lost table would leave sources stale until the + next full `live`. Nothing else goes as a datagram, and tunnel traffic never + does: lane 0 carries commands, their answers and deltas that must arrive, + and two paths for lane 0 would show the service which frames are which. + - The service closes the uplink with 4400 on a datagram that is not `hb` + or a `live` without `sources`. It keeps an `hb` or `live`, on either + path, only when its `seq` is above the newest of its type on this + connection, and drops a late one, with one exception: a late `live` + whose `srcRev` is above the table it holds still gives its table, and + nothing else. A phone does the same (Freshness case 17), so a table + overtaken by a datagram is not lost. + - Each end offers datagrams in its transport parameters. A sender whose + peer does not uses the control stream. + - Until a phone's first transport message arrives, the service sends it + everything on the control stream, behind message 2. quic-go and other + stacks send datagrams ahead of stream data, so an early `hb` could + otherwise reach a phone before the message 2 it needs to read it. +- **Order.** Messages on one stream arrive in order; across streams and + datagrams they may not. Noise transport carries its own counter, one for + all three paths, so a burst on the bulk stream can overtake a control + message by a few hundred. A receiver of a session that can travel over + QUIC keeps a window of at least 1024 (`appwire.ReplayWindow`), so the app + session and the tunnel take messages out of order. It drops a datagram + that falls outside the window or repeats; the same on a stream ends the + session, as over the WebSocket. The service's rule to forward a home's + messages in the order it received them holds per stream. +- **Streams.** A peer that opens a second bidirectional stream, or a second + unidirectional one, breaks the protocol. The service grants one of each. +- **Closing.** A close code becomes the QUIC application error code, with + the WebSocket close reason as the reason phrase. 1000 closes normally. +- **Liveness.** QUIC stands in for the WebSocket ping. The client sends a QUIC + PING after 15 s with nothing to send, which keeps a NAT's UDP mapping + open; many drop one after 30 s. Both ends close a connection silent for + 45 s, as the WebSocket's read deadline does. +- **Migration.** A client may move its connection to a new address (RFC 9000 + §9): a phone from Wi-Fi to cellular, or a box whose NAT rebinds. The + service validates the new path and follows it. The Noise sessions inside + never notice: no handshake, no new `sid`. + +### Racing the WebSocket + +A client starts QUIC first. If the QUIC handshake has not finished after +250 ms, it also dials the WebSocket, takes the first that opens and closes +the other. It remembers which won on each network for a day and starts the +next time with that one, giving the other the same 250 ms; after a day QUIC +goes first again, so a network that blocked UDP once is tried again. A QUIC +dial that loses to the WebSocket goes on for up to 3 s, only to learn: if it +opens, QUIC goes first next time, so one lost packet does not keep QUIC +second for a day. The session stays on the WebSocket either way. + +A QUIC handshake that finishes proves less than a session that works. A +QUIC connection that ends before the first session message (the `challenge` +for a box, message 2 for a phone), or that the path drops within 60 s of +it, counts as QUIC failing on that network: the client starts there with +the WebSocket for a day, and a learning dial in that day does not move QUIC +first. A close the service sends on purpose, with a code from 1000 up, and a +stateless reset after a restart, do not count. A box names its network by +its local address; a phone by its path's interfaces and gateways, kept as a +hash. A client +says which carries it, by the registry's `cloud_transports`: `quic` or +`websocket`. Core shows it in `GET /api/cloud`; the app shows it beside the +carrier. The carrier stays `tunnel` or `cloud` over both. + +### Early data + +A client that holds a session ticket from the service may send 0-RTT data. It +may put there only the target and, on the app session, Noise message 1, with +`now` and the box session's message 1 in `now.tun`. Nothing else can go +there: every later message is a Noise transport message, which the client +can encrypt only once message 2 has come back. So no `cmd` or `api.req` ever +rides in early data. A box puts only its target there, and the service's +`challenge` comes back in its first flight. + +The network can replay early data. These rules make a replay change nothing: + +- **A phone's session waits for the handshake.** The service reads a phone's + message 1 from early data but acts on it only once the QUIC handshake has + finished, which the sender of a replay cannot do. Until then nothing is + forwarded, opened or answered: no session is replaced, no pairing code + spent, no `watch` sent. This holds for the first copy too, which matters + when the network dropped the original and the phone dialled again. +- **A box's target is all it sends early.** The service may answer it with a + `challenge` before the handshake ends. A challenge changes nothing, and + only the box can answer it with a `hello`. +- **For a day.** A ticket allows early data for 24 hours after the service + issued it, and resumption without early data after that. A restart draws + new ticket keys, so every client makes a full handshake once. +- **The box holds a tunnel session as pending** until its first transport + message decrypts (see [Tunnel](#tunnel)), so a `now.tun` takes no slot + and spends no code, whoever sent it. +- **HTTP/3 requests wait for the handshake**, since no plain route has said + which of its requests survive a replay. + +A client whose early data the service rejects sends it again once the +handshake is done. + ## Envelope and signature ``` @@ -1165,6 +1310,11 @@ In all of them the phone uses rule N with 100 ms in transit unless stated. the generic line with no buttons. 16. **The box's state wins.** The box reports a source as `down`, and t = 1 s. The source shows `down`. +17. **A late table still counts.** The box changes the table: a `live` with + `seq` 2 and `srcRev` 2 goes on the stream, and a `live` without a table, + `seq` 3 and `srcRev` 2, overtakes it as a datagram. The source shows + `stale` (rule 1) until the late `live` arrives; the phone then takes its + table, keeps `seq` 3's values, and the source shows `live`. ## Grant on the box diff --git a/go/go.mod b/go/go.mod index 09f766ee..9df9d7ae 100644 --- a/go/go.mod +++ b/go/go.mod @@ -10,6 +10,7 @@ require ( github.com/gorilla/websocket v1.5.3 github.com/mochi-mqtt/server/v2 v2.7.9 github.com/parquet-go/parquet-go v0.32.0 + github.com/quic-go/quic-go v0.63.0 github.com/shirou/gopsutil/v4 v4.26.8 github.com/simonvetter/modbus v1.6.4 github.com/yuin/gopher-lua v1.1.2 @@ -21,12 +22,10 @@ require ( ) require ( - github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/go-playground/locales v0.12.1 // indirect github.com/go-playground/universal-translator v0.16.0 // indirect github.com/gorilla/mux v1.8.1 // indirect github.com/leodido/go-urn v1.1.0 // indirect - github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/relvacode/iso8601 v1.6.0 // indirect github.com/stretchr/objx v0.5.2 // indirect gopkg.in/go-playground/validator.v9 v9.30.0 // indirect diff --git a/go/go.sum b/go/go.sum index 07d24f44..62dd3371 100644 --- a/go/go.sum +++ b/go/go.sum @@ -12,8 +12,6 @@ github.com/caarlos0/env/v11 v11.3.1 h1:cArPWC15hWmEt+gWk7YBi7lEXTXCvpaSdCiZE2X5m github.com/caarlos0/env/v11 v11.3.1/go.mod h1:qupehSf/Y0TUTsxKywqRt/vJjN5nz6vauiYEUUr8P4U= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= -github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/ebitengine/purego v0.10.2 h1:W809HbnvzAxgdm+aOvlSekrM16wGCdT/e76+9tS7gzE= @@ -78,10 +76,12 @@ github.com/parquet-go/parquet-go v0.32.0/go.mod h1:navtkAYr2LGoJVp141oXPlO/sxLva github.com/pierrec/lz4/v4 v4.1.25 h1:kocOqRffaIbU5djlIBr7Wh+cx82C0vtFb0fOurZHqD0= github.com/pierrec/lz4/v4 v4.1.25/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= -github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU= github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/quic-go/go-ossfuzz-seeds v0.1.0 h1:APacT+iIaNF6fd8AGEiN3bT/Jtkd2jz4v4TzM7MFjy0= +github.com/quic-go/go-ossfuzz-seeds v0.1.0/go.mod h1:3IOHRbJIc+L6YKMwfDtJAM9Vj9k0YY4muhuyUYk5tbk= +github.com/quic-go/quic-go v0.63.0 h1:LIFGHI4PFUhhw2dDD1ARHdCff143ffMHwZtbnbuJ78A= +github.com/quic-go/quic-go v0.63.0/go.mod h1:RAro2j2yN9a9EiPACLHT9IB2NXCvGQmmo/alT0yYI0w= github.com/relvacode/iso8601 v1.6.0 h1:eFXUhMJN3Gz8Rcq82f9DTMW0svjtAVuIEULglM7QHTU= github.com/relvacode/iso8601 v1.6.0/go.mod h1:FlNp+jz+TXpyRqgmM7tnzHHzBnz776kmAH2h3sZCn0I= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= @@ -103,8 +103,8 @@ github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/ github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= -github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= -github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYICU0nA= github.com/tklauser/go-sysconf v0.3.16/go.mod h1:/qNL9xxDhc7tx3HSRsLWNnuzbVfh3e7gh/BmM179nYI= github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9RXw= @@ -119,6 +119,10 @@ github.com/yuin/gopher-lua v1.1.2 h1:yF/FjE3hD65tBbt0VXLE13HWS9h34fdzJmrWRXwobGA github.com/yuin/gopher-lua v1.1.2/go.mod h1:7aRmXIWl37SqRf0koeyylBEzJ+aPt8A+mmkQ4f1ntR8= github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko= +go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk= diff --git a/go/internal/appproto/gencontract/gencontract.go b/go/internal/appproto/gencontract/gencontract.go index fc409deb..ca88d1d8 100644 --- a/go/internal/appproto/gencontract/gencontract.go +++ b/go/internal/appproto/gencontract/gencontract.go @@ -59,6 +59,8 @@ type Registry struct { // with FTW Cloud, and what for. CloudScopes []string `yaml:"cloud_scopes"` CloudPurposes []string `yaml:"cloud_purposes"` + // CloudTransports are what a session with FTW Cloud travels over. + CloudTransports []string `yaml:"cloud_transports"` // AgentArgs is what an agent may put in each op it can name, and // Reads the reads it may send with the scope each needs. AgentArgs map[string][]struct { @@ -338,6 +340,20 @@ func writeCloud(b *bytes.Buffer, reg Registry) { fmt.Fprintf(b, "\tCloudPurpose%s,\n", goName(p)) } b.WriteString("}\n\n") + + b.WriteString("// What a session with FTW Cloud travels over, under its carrier.\n") + b.WriteString("const (\n") + for _, t := range reg.CloudTransports { + fmt.Fprintf(b, "\tCloudTransport%s = %q\n", goName(t), t) + } + b.WriteString(")\n\n") + + b.WriteString("// CloudTransports lists them in the registry's order.\n") + b.WriteString("var CloudTransports = []string{\n") + for _, t := range reg.CloudTransports { + fmt.Fprintf(b, "\tCloudTransport%s,\n", goName(t)) + } + b.WriteString("}\n\n") } // writeOps emits the operation table as data rather than constants. The op diff --git a/go/internal/cloudlink/conn.go b/go/internal/cloudlink/conn.go index 5f751e00..7af1636f 100644 --- a/go/internal/cloudlink/conn.go +++ b/go/internal/cloudlink/conn.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "crypto/rand" - "crypto/tls" "errors" "fmt" "math/big" @@ -54,11 +53,8 @@ func (l *Link) dialer() *websocket.Dialer { // once dials, serves until the connection ends and says how long to wait. func (l *Link) once(ctx context.Context, snap grantSnapshot) (time.Duration, error) { dialCtx, cancel := context.WithTimeout(ctx, dialTimeout) - conn, resp, err := l.dialer().DialContext(dialCtx, l.endpoint(), nil) + conn, err := l.dial(dialCtx) cancel() - if resp != nil && resp.Body != nil { - _ = resp.Body.Close() - } if err != nil { l.hostFailed() return l.backoff(), fmt.Errorf("dialling FTW Cloud: %w", err) @@ -69,11 +65,16 @@ func (l *Link) once(ctx context.Context, snap grantSnapshot) (time.Duration, err err = l.serve(ctx, conn, snap) l.mu.Lock() up := l.connected + upFor := time.Since(l.upAt) l.connected = false if time.Since(connectedAt) >= backoffCap { l.attempt = 0 } l.mu.Unlock() + if q, ok := conn.(*quicConn); ok && q.network != "" && quicFailed(err, up, upFor) { + l.demote(q.network) + l.log.Info("QUIC failed on this network; the WebSocket goes first here for a day", "err", err) + } if errors.Is(err, errLinkEnded) || errors.Is(err, errClosed) { return 0, nil } @@ -237,7 +238,7 @@ const ( // serve runs one connection: the handshake, then the reader here and the // sender on its own goroutine. -func (l *Link) serve(ctx context.Context, conn *websocket.Conn, snap grantSnapshot) error { +func (l *Link) serve(ctx context.Context, conn uplinkConn, snap grantSnapshot) error { ctx, cancel := context.WithCancel(ctx) defer cancel() go func() { @@ -302,12 +303,14 @@ func (l *Link) serve(ctx context.Context, conn *websocket.Conn, snap grantSnapsh tunOut, tunBulk := make(chan []byte, tunQueue), make(chan []byte, tunBulkQueue) l.mu.Lock() l.connected = true + l.upAt = time.Now() + l.transport = transportOf(conn) l.shardHops, l.hostFails = 0, 0 if l.opts.Tunnel != nil { l.tunOut, l.tunBulk = tunOut, tunBulk } l.mu.Unlock() - l.log.Info("FTW Cloud uplink connected", "home", snap.home) + l.log.Info("FTW Cloud uplink connected", "home", snap.home, "transport", transportOf(conn)) defer func() { l.mu.Lock() l.tunOut, l.tunBulk = nil, nil @@ -351,19 +354,6 @@ func (l *Link) serve(ctx context.Context, conn *websocket.Conn, snap grantSnapsh return err } -// exporterValue is the TLS 1.3 exporter value both ends bind the handshake to. -func exporterValue(conn *websocket.Conn) ([]byte, error) { - tc, ok := conn.NetConn().(*tls.Conn) - if !ok { - return nil, errors.New("cloudlink: the uplink is not TLS") - } - cs := tc.ConnectionState() - if cs.Version != tls.VersionTLS13 { - return nil, errors.New("cloudlink: the uplink is not TLS 1.3") - } - return cs.ExportKeyingMaterial(cloudwire.EKMLabel, nil, cloudwire.EKMBytes) -} - func (l *Link) verifiedByService(home string, env cloudwire.Envelope) bool { for _, k := range l.opts.ServiceKeys { if cloudwire.Verify(k, home, env) == nil { @@ -383,7 +373,7 @@ func (l *Link) clockName() string { // readEnvelope reads one binary frame. A text message, a lane 0 frame, a tun // frame over MaxTunFrame or any other over 4 KiB is the service breaking the // protocol. -func readEnvelope(conn *websocket.Conn) (cloudwire.Envelope, error) { +func readEnvelope(conn uplinkConn) (cloudwire.Envelope, error) { kind, msg, err := conn.ReadMessage() if err != nil { return cloudwire.Envelope{}, err @@ -392,8 +382,9 @@ func readEnvelope(conn *websocket.Conn) (cloudwire.Envelope, error) { } func parseDownlink(kind int, msg []byte) (cloudwire.Envelope, error) { - if kind != websocket.BinaryMessage { - return cloudwire.Envelope{}, fmt.Errorf("%w: a text message", errService) + if kind != websocket.BinaryMessage && kind != msgBulk { + // A text message, or a datagram: the service sends the box none. + return cloudwire.Envelope{}, fmt.Errorf("%w: a message of type %d", errService, kind) } if len(msg) > cloudwire.MaxTunFrame { return cloudwire.Envelope{}, fmt.Errorf("%w: %d bytes", errService, len(msg)) @@ -409,6 +400,9 @@ func parseDownlink(kind int, msg []byte) (cloudwire.Envelope, error) { if !slices.Contains(cloudwire.Downlink, env.T) { return cloudwire.Envelope{}, fmt.Errorf("%w: %q is not a message the service may send", errService, env.T) } + if kind == msgBulk && env.T != cloudwire.MsgTun { + return cloudwire.Envelope{}, fmt.Errorf("%w: a %s on the bulk stream", errService, env.T) + } if env.T != cloudwire.MsgTun && len(msg) > cloudwire.MaxServiceFrame { return cloudwire.Envelope{}, fmt.Errorf("%w: a %d-byte %s", errService, len(msg), env.T) } @@ -420,7 +414,7 @@ func parseDownlink(kind int, msg []byte) (cloudwire.Envelope, error) { // like any stranger. watch and resync go to the sender, which owns the // cadence. A purge receipt is taken here, before the service hangs up, so the // box never misses one it was sent. A tun goes to the Tunnel unread. -func (l *Link) read(conn *websocket.Conn, snap grantSnapshot, events chan<- cloudwire.Envelope) error { +func (l *Link) read(conn uplinkConn, snap grantSnapshot, events chan<- cloudwire.Envelope) error { _ = conn.SetReadDeadline(time.Now().Add(readTimeout)) conn.SetPingHandler(func(data string) error { _ = conn.SetReadDeadline(time.Now().Add(readTimeout)) @@ -455,7 +449,7 @@ func (l *Link) read(conn *websocket.Conn, snap grantSnapshot, events chan<- clou // writer is the one goroutine that writes data frames on a connection. type writer struct { - conn *websocket.Conn + conn uplinkConn link *Link mu sync.Mutex } @@ -480,7 +474,13 @@ func (w *writer) send(snap grantSnapshot, t string, body any) error { return err } - if err := w.write(frame); err != nil { + // Over QUIC an hb, or a live without its table, goes as a datagram: the + // newest wins, and a lost one shows as a gap on the phone. + kind := websocket.BinaryMessage + if cloudwire.Datagram(env) { + kind = msgDatagram + } + if err := w.writeAs(kind, frame); err != nil { return err } l := w.link @@ -596,17 +596,11 @@ func (p *pacer) take(n int, now time.Time) time.Duration { // first and at once, even while bulk waits for the pacer. It counts both // against the tunnel's budget, never the copy's. func (w *writer) tunnel(ctx context.Context, out, bulk <-chan []byte) error { - l := w.link pace := newPacer(time.Now()) - send := func(frame []byte) error { - if err := w.write(frame); err != nil { - return err - } - l.mu.Lock() - l.tunBudget.add(l.uptimeMs(), int64(len(frame))) - l.mu.Unlock() - return nil - } + // Lane 0 goes on the control stream, bulk on the bulk stream over QUIC, + // so a lost bulk packet never holds up a command or its answer. + send := func(frame []byte) error { return w.tunAs(websocket.BinaryMessage, frame) } + sendBulk := func(frame []byte) error { return w.tunAs(msgBulk, frame) } for { select { case frame := <-out: @@ -644,19 +638,37 @@ func (w *writer) tunnel(ctx context.Context, out, bulk <-chan []byte) error { } } } - if err := send(frame); err != nil { + if err := sendBulk(frame); err != nil { return err } } } } +// tunAs writes one tun frame the way kind says and counts it against the +// tunnel's budget, never the copy's. +func (w *writer) tunAs(kind int, frame []byte) error { + if err := w.writeAs(kind, frame); err != nil { + return err + } + l := w.link + l.mu.Lock() + l.tunBudget.add(l.uptimeMs(), int64(len(frame))) + l.mu.Unlock() + return nil +} + // write sends one frame under the socket's one write lock. -func (w *writer) write(frame []byte) error { +func (w *writer) write(frame []byte) error { return w.writeAs(websocket.BinaryMessage, frame) } + +// writeAs sends one frame the way kind says where the connection carries +// it: over QUIC, on the bulk stream or as a datagram. A WebSocket carries +// every kind as a binary message. +func (w *writer) writeAs(kind int, frame []byte) error { w.mu.Lock() defer w.mu.Unlock() if err := w.conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { return err } - return w.conn.WriteMessage(websocket.BinaryMessage, frame) + return w.conn.WriteMessage(kindFor(w.conn, kind), frame) } diff --git a/go/internal/cloudlink/link.go b/go/internal/cloudlink/link.go index e7068973..b15c7682 100644 --- a/go/internal/cloudlink/link.go +++ b/go/internal/cloudlink/link.go @@ -200,7 +200,12 @@ type Options struct { Arrivals <-chan struct{} Logger *slog.Logger - // Seams for tests. Production leaves them zero. + // Seams for tests. Production leaves them zero, and races QUIC against + // the WebSocket. A test that names its Endpoint gets the WebSocket alone + // unless it sets QUIC; QUICAddr names the service's UDP address when it + // is not the Endpoint's host and port. + QUIC bool + QUICAddr string Endpoint string ServiceKeys []*ecdsa.PublicKey RootCAs *x509.CertPool @@ -228,8 +233,22 @@ type Link struct { grant *grantRecord notice string connected bool - lastSend time.Time - budget budget + // transport is how the uplink travels, or last travelled: quic or + // websocket. winners is the race's memory, the transport that opened + // first on each network. + transport string + quicOn bool + winners map[string]winner + // tickets and quicTickets hold the service's TLS tickets, one cache for + // each transport, so a reconnect resumes and, over QUIC, sends its + // target as early data. A ticket from TLS over TCP cannot carry 0-RTT + // into QUIC, nor the other way. + tickets tls.ClientSessionCache + quicTickets tls.ClientSessionCache + // upAt is when the service's hello_ok arrived on this connection. + upAt time.Time + lastSend time.Time + budget budget // tunOut and tunBulk carry framed tun messages, lane 0 and bulk, to // the connection's tun writer. Nil while no uplink is up. tunBudget counts what left, apart from // the readable copy's budget, and tunDropped what never did. @@ -275,6 +294,7 @@ func New(opts Options) (*Link, error) { if opts.Logger == nil { opts.Logger = slog.Default() } + quicOn := opts.QUIC || opts.Endpoint == "" if opts.Endpoint == "" { opts.Endpoint = Endpoint } @@ -314,7 +334,8 @@ func New(opts Options) (*Link, error) { } l := &Link{ opts: opts, log: opts.Logger.With("component", "cloudlink"), boot: boot, - wake: make(chan struct{}, 1), sent: map[string]sentFrame{}, + wake: make(chan struct{}, 1), sent: map[string]sentFrame{}, quicOn: quicOn, + tickets: tls.NewLRUClientSessionCache(4), quicTickets: tls.NewLRUClientSessionCache(4), } if err := l.load(); err != nil { return nil, err @@ -785,6 +806,9 @@ type Status struct { GrantedBy string `json:"grantedBy,omitempty"` Policy string `json:"policy,omitempty"` Connected bool `json:"connected"` + // Transport is how the uplink travels, the registry's cloud_transports: + // quic or websocket. Empty before the first connection. + Transport string `json:"transport,omitempty"` // LastSendAgeMs is how long ago the box last sent anything, -1 never. LastSendAgeMs int64 `json:"lastSendAgeMs"` // BytesLast24h counts what the box sent in the last 24 hours of uptime. @@ -811,7 +835,7 @@ func (l *Link) Status() Status { defer l.mu.Unlock() s := Status{ State: "off", Notice: l.notice, Available: len(l.opts.ServiceKeys) > 0, - Scopes: []string{}, Purposes: []string{}, Connected: l.connected, + Scopes: []string{}, Purposes: []string{}, Connected: l.connected, Transport: l.transport, LastSendAgeMs: -1, PurgedAtMs: l.purgedAt, EventsDropped: l.eventsDropped, BytesLast24h: l.budget.total(l.uptimeMs()), BudgetBytes: budgetBytes, Link: "off", TunBytesLast24h: l.tunBudget.total(l.uptimeMs()), TunBudgetBytes: tunBudgetBytes, @@ -1021,7 +1045,14 @@ func (l *Link) caps(snap grantSnapshot) []string { // tlsConfig is TLS 1.3 only, with the system roots unless a test or a dev // build adds its own. func (l *Link) tlsConfig() *tls.Config { - return &tls.Config{MinVersion: tls.VersionTLS13, RootCAs: l.opts.RootCAs} + return &tls.Config{MinVersion: tls.VersionTLS13, RootCAs: l.opts.RootCAs, ClientSessionCache: l.tickets} +} + +// quicTLSConfig is tlsConfig with QUIC's own ticket cache. +func (l *Link) quicTLSConfig() *tls.Config { + c := l.tlsConfig() + c.ClientSessionCache = l.quicTickets + return c } // now is the wall clock, or a test's. diff --git a/go/internal/cloudlink/link_test.go b/go/internal/cloudlink/link_test.go index 8b4e3808..46805d4c 100644 --- a/go/internal/cloudlink/link_test.go +++ b/go/internal/cloudlink/link_test.go @@ -182,7 +182,7 @@ type fakeService struct { badEKM bool wrongKey bool purge bool - afterHello func(conn *websocket.Conn, home string) + afterHello func(conn uplinkConn, home string) rosterSeqOK uint64 // wrongShard, when set, closes after hello with CloseWrongShard and // this host, as an instance that does not hold the home; @@ -200,6 +200,16 @@ type fakeService struct { ch chan cloudwire.Envelope home string pub *ecdsa.PublicKey + // kinds is how each message from the box came: by the control + // stream, the bulk stream or as a datagram over QUIC. + kinds map[string][]int + // transports counts the connections each transport carried. + transports map[string]int + // kick drops this many connections right after hello_ok, so the box + // dials again. + kick int + // usedEarly says, per QUIC connection, whether it took early data. + usedEarly []bool } func newFakeService(t *testing.T) *fakeService { @@ -209,7 +219,8 @@ func newFakeService(t *testing.T) *fakeService { // newFakeServiceCert serves with cert, or httptest's own when nil. func newFakeServiceCert(t *testing.T, cert *tls.Certificate) *fakeService { key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) - s := &fakeService{t: t, key: key, ch: make(chan cloudwire.Envelope, 256)} + s := &fakeService{t: t, key: key, ch: make(chan cloudwire.Envelope, 256), kinds: map[string][]int{}, + transports: map[string]int{}} s.srv = httptest.NewUnstartedServer(http.HandlerFunc(s.handle)) s.srv.TLS = &tls.Config{MinVersion: tls.VersionTLS13} if cert != nil { @@ -230,7 +241,7 @@ func (s *fakeService) roots() *x509.CertPool { return pool } -func (s *fakeService) write(conn *websocket.Conn, key *ecdsa.PrivateKey, home, t string, body any) { +func (s *fakeService) write(conn uplinkConn, key *ecdsa.PrivateKey, home, t string, body any) { env, err := cloudwire.Seal(key, home, t, body) if err != nil { s.t.Error(err) @@ -248,11 +259,17 @@ func (s *fakeService) handle(w http.ResponseWriter, r *http.Request) { if err != nil { return } + ekm, _ := r.TLS.ExportKeyingMaterial(cloudwire.EKMLabel, nil, cloudwire.EKMBytes) + s.serve(conn, ekm) +} + +// serve is the service's side of one uplink, over either transport. +func (s *fakeService) serve(conn uplinkConn, ekm []byte) { defer conn.Close() s.mu.Lock() s.conns++ + s.transports[transportOf(conn)]++ s.mu.Unlock() - ekm, _ := r.TLS.ExportKeyingMaterial(cloudwire.EKMLabel, nil, cloudwire.EKMBytes) nonce := bytes.Repeat([]byte{7}, cloudwire.NonceBytes) s.write(conn, nil, "", cloudwire.MsgChallenge, cloudwire.Challenge{V: 1, Nonce: nonce}) @@ -305,13 +322,22 @@ func (s *fakeService) handle(w http.ResponseWriter, r *http.Request) { if s.afterHello != nil { go s.afterHello(conn, home) } + s.mu.Lock() + kick := s.kick > 0 + if kick { + s.kick-- + } + s.mu.Unlock() + if kick { + return + } for { kind, msg, err := conn.ReadMessage() if err != nil { return } - if kind != websocket.BinaryMessage { + if kind != websocket.BinaryMessage && kind != msgBulk && kind != msgDatagram { s.t.Errorf("the box sent a text message") return } @@ -337,6 +363,7 @@ func (s *fakeService) handle(w http.ResponseWriter, r *http.Request) { s.mu.Lock() s.got = append(s.got, env) s.raw = append(s.raw, msg) + s.kinds[env.T] = append(s.kinds[env.T], kind) s.mu.Unlock() s.ch <- env if !s.purge { @@ -639,7 +666,7 @@ func TestAReleaseBuildTrustsOnlyTheServicesKey(t *testing.T) { func TestUnknownDownlinkClosesTheUplink(t *testing.T) { svc := newFakeService(t) - svc.afterHello = func(conn *websocket.Conn, home string) { + svc.afterHello = func(conn uplinkConn, home string) { time.Sleep(200 * time.Millisecond) svc.write(conn, nil, home, "cmd", map[string]any{"op": "site.mode.set"}) } @@ -657,7 +684,7 @@ func TestUnknownDownlinkClosesTheUplink(t *testing.T) { func TestWatchRaisesTheCadence(t *testing.T) { svc := newFakeService(t) - svc.afterHello = func(conn *websocket.Conn, home string) { + svc.afterHello = func(conn uplinkConn, home string) { svc.write(conn, nil, home, cloudwire.MsgWatch, cloudwire.Watch{Hz: 1}) } h := newHarness(t, svc) @@ -676,7 +703,7 @@ func TestWatchRaisesTheCadence(t *testing.T) { // the next look at the box, and never sooner than a period after the last. func TestAWatchedReadingGoesAsItsPollLands(t *testing.T) { svc := newFakeService(t) - svc.afterHello = func(conn *websocket.Conn, home string) { + svc.afterHello = func(conn uplinkConn, home string) { svc.write(conn, nil, home, cloudwire.MsgWatch, cloudwire.Watch{Hz: 1}) } h := newHarness(t, svc) diff --git a/go/internal/cloudlink/quic.go b/go/internal/cloudlink/quic.go new file mode 100644 index 00000000..6a7f4411 --- /dev/null +++ b/go/internal/cloudlink/quic.go @@ -0,0 +1,409 @@ +package cloudlink + +import ( + "bufio" + "context" + "crypto/tls" + "encoding/binary" + "errors" + "io" + "net" + "sync" + "time" + + "github.com/gorilla/websocket" + "github.com/quic-go/quic-go" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// quicConn is the uplink over QUIC: the WebSocket's messages on a control +// stream that starts with the target, bulk tun on a unidirectional stream, +// and hb and live as QUIC DATAGRAMs (docs/protocol-cloud.md, "QUIC"). +// +// QUIC stands in for the service's WebSocket ping. The box sends a QUIC +// PING after cloudwire.QUICKeepAlive of quiet, which keeps the NAT's UDP +// mapping open, and the connection ends after cloudwire.QUICIdleTimeout with +// nothing from the service. While it lives, quicConn calls the ping handler +// on the keepalive's period, as the service's pings would. +type quicConn struct { + qc *quic.Conn + tr *quic.Transport + // network is where the race dialled it, for demote. + network string + ctrl *quic.Stream + r *bufio.Reader + + start sync.Once + in chan inbound + kick chan struct{} + + mu sync.Mutex + deadline time.Time + writeDL time.Time + limit int64 + ping func(string) error + readErr error + + writeMu sync.Mutex + bulkMu sync.Mutex + bulk *quic.SendStream + noBulk bool + closed sync.Once +} + +type inbound struct { + kind int + msg []byte +} + +// quicConfig is how the box dials. +func quicConfig() *quic.Config { + return &quic.Config{ + HandshakeIdleTimeout: dialTimeout, + MaxIdleTimeout: cloudwire.QUICIdleTimeout, + KeepAlivePeriod: cloudwire.QUICKeepAlive, + EnableDatagrams: true, + // The service sends the box everything on the control stream, and + // may open one bulk stream. + MaxIncomingStreams: -1, + MaxIncomingUniStreams: 1, + } +} + +// dialQUIC opens the uplink over QUIC to addr, a host and port, on a UDP +// socket of its own (pc, or a new one when nil), and sends the target. With +// a ticket in tlsConf's session cache the target goes as early data, so the +// service's challenge comes back in its first flight; the box sends nothing +// else early. A service that rejects the early data gets the target again. +func dialQUIC(ctx context.Context, addr string, tlsConf *tls.Config, pc net.PacketConn) (*quicConn, error) { + udp, err := net.ResolveUDPAddr("udp", addr) + if err != nil { + return nil, err + } + if pc == nil { + if pc, err = net.ListenUDP("udp", nil); err != nil { + return nil, err + } + } + tc := tlsConf.Clone() + tc.NextProtos = []string{cloudwire.ALPNBox} + tr := &quic.Transport{Conn: pc} + dial := tr.Dial + if tc.ClientSessionCache != nil { + dial = tr.DialEarly + } + qc, err := dial(ctx, udp, tc, quicConfig()) + if err != nil { + _ = tr.Close() + return nil, err + } + fail := func(err error) (*quicConn, error) { + _ = qc.CloseWithError(0, "") + _ = tr.Close() + return nil, err + } + // DialEarly comes back before the handshake only when it can send early + // data, with a ticket that allows it. + early := false + select { + case <-qc.HandshakeComplete(): + default: + early = true + } + target := cloudwire.AppendMessage(nil, []byte(cloudwire.TargetBox)) + ctrl, err := qc.OpenStreamSync(ctx) + if err == nil { + _, err = ctrl.Write(target) + } + if err != nil { + return fail(err) + } + select { + case <-qc.HandshakeComplete(): + case <-ctx.Done(): + return fail(ctx.Err()) + } + if early && !qc.ConnectionState().Used0RTT { + // The early data, if any went, was rejected: the connection quic-go + // hands over takes the target again. + if qc, err = qc.NextConnection(ctx); err != nil { + return fail(err) + } + if ctrl, err = qc.OpenStreamSync(ctx); err == nil { + _, err = ctrl.Write(target) + } + if err != nil { + return fail(err) + } + } + c := &quicConn{qc: qc, tr: tr, ctrl: ctrl, r: bufio.NewReader(ctrl), in: make(chan inbound), + kick: make(chan struct{}, 1), limit: cloudwire.MaxTunFrame} + go func() { + <-qc.Context().Done() + _ = tr.Close() + }() + return c, nil +} + +func (c *quicConn) transport() string { return cloudwire.TransportQUIC } + +func (c *quicConn) exporter() ([]byte, error) { + cs := c.qc.ConnectionState().TLS + return cs.ExportKeyingMaterial(cloudwire.EKMLabel, nil, cloudwire.EKMBytes) +} + +func (c *quicConn) carries(messageType int) bool { + switch messageType { + case msgDatagram: + d := c.qc.ConnectionState().SupportsDatagrams + return d.Local && d.Remote + case msgBulk: + c.bulkMu.Lock() + defer c.bulkMu.Unlock() + return !c.noBulk + } + return messageType == websocket.BinaryMessage +} + +func (c *quicConn) ReadMessage() (int, []byte, error) { + c.start.Do(c.readers) + for { + c.mu.Lock() + err, deadline := c.readErr, c.deadline + c.mu.Unlock() + if err != nil { + return 0, nil, err + } + var expired <-chan time.Time + var timer *time.Timer + if !deadline.IsZero() { + wait := time.Until(deadline) + if wait <= 0 { + return 0, nil, c.fail(errQUICTimeout) + } + timer = time.NewTimer(wait) + expired = timer.C + } + select { + case m := <-c.in: + if timer != nil { + timer.Stop() + } + return m.kind, m.msg, nil + case <-expired: + return 0, nil, c.fail(errQUICTimeout) + case <-c.kick: + if timer != nil { + timer.Stop() + } + } + } +} + +func (c *quicConn) fail(err error) error { + c.mu.Lock() + if c.readErr == nil { + c.readErr = err + } + err = c.readErr + c.mu.Unlock() + select { + case c.kick <- struct{}{}: + default: + } + return err +} + +func (c *quicConn) readers() { + ctx := c.qc.Context() + deliver := func(kind int, msg []byte) bool { + select { + case c.in <- inbound{kind: kind, msg: msg}: + return true + case <-ctx.Done(): + return false + } + } + readStream := func(r *bufio.Reader, kind int) { + for { + c.mu.Lock() + limit := c.limit + c.mu.Unlock() + msg, err := cloudwire.ReadMessage(r, int(limit)) + if err != nil { + if kind == websocket.BinaryMessage || !errors.Is(err, io.EOF) { + c.fail(c.readError(err)) + } + return + } + if !deliver(kind, msg) { + return + } + } + } + go readStream(c.r, websocket.BinaryMessage) + go func() { + for { + d, err := c.qc.ReceiveDatagram(ctx) + if err != nil || !deliver(msgDatagram, d) { + return + } + } + }() + go func() { + if str, err := c.qc.AcceptUniStream(ctx); err == nil { + readStream(bufio.NewReader(str), msgBulk) + } + }() + go func() { + t := time.NewTicker(cloudwire.QUICKeepAlive) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + c.mu.Lock() + ping := c.ping + c.mu.Unlock() + if ping != nil { + _ = ping("") + } + } + } + }() +} + +// readError is the service's close as a WebSocket close, so the link reads +// a close code the same over both transports. +func (c *quicConn) readError(err error) error { + if ae, ok := errors.AsType[*quic.ApplicationError](err); ok { + return &websocket.CloseError{Code: int(ae.ErrorCode), Text: ae.ErrorMessage} + } + if errors.Is(err, io.EOF) { + return &websocket.CloseError{Code: websocket.CloseNormalClosure} + } + if errors.Is(err, cloudwire.ErrQUICMessage) { + c.closeWith(cloudwire.CloseProtocol, "") + } + return err +} + +func (c *quicConn) WriteMessage(messageType int, data []byte) error { + switch messageType { + case msgDatagram: + if c.carries(msgDatagram) { + err := c.qc.SendDatagram(data) + if err == nil { + return nil + } + if _, tooLarge := errors.AsType[*quic.DatagramTooLargeError](err); !tooLarge { + return err + } + } + case msgBulk: + if sent, err := c.writeBulk(data); sent || err != nil { + return err + } + case websocket.BinaryMessage: + default: + return errors.New("cloudlink: no such message type") + } + c.writeMu.Lock() + defer c.writeMu.Unlock() + if err := c.ctrl.SetWriteDeadline(c.writeDeadline()); err != nil { + return err + } + _, err := c.ctrl.Write(cloudwire.AppendMessage(nil, data)) + return err +} + +func (c *quicConn) writeDeadline() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.writeDL +} + +// writeBulk sends on the bulk stream, opening it the first time; false +// when the service grants no stream for it. +func (c *quicConn) writeBulk(data []byte) (bool, error) { + c.bulkMu.Lock() + defer c.bulkMu.Unlock() + if c.noBulk { + return false, nil + } + if c.bulk == nil { + s, err := c.qc.OpenUniStream() + if err != nil { + c.noBulk = true + return false, nil + } + c.bulk = s + } + if err := c.bulk.SetWriteDeadline(c.writeDeadline()); err != nil { + return true, err + } + _, err := c.bulk.Write(cloudwire.AppendMessage(nil, data)) + return true, err +} + +func (c *quicConn) WriteControl(messageType int, data []byte, _ time.Time) error { + switch messageType { + case websocket.CloseMessage: + code, text := websocket.CloseNoStatusReceived, "" + if len(data) >= 2 { + code, text = int(binary.BigEndian.Uint16(data)), string(data[2:]) + } + c.closeWith(code, text) + case websocket.PingMessage, websocket.PongMessage: + // QUIC's keepalive does this. + } + return nil +} + +func (c *quicConn) closeWith(code int, text string) { + c.closed.Do(func() { _ = c.qc.CloseWithError(quic.ApplicationErrorCode(code), text) }) +} + +func (c *quicConn) SetReadDeadline(t time.Time) error { + c.mu.Lock() + c.deadline = t + c.mu.Unlock() + select { + case c.kick <- struct{}{}: + default: + } + return nil +} + +func (c *quicConn) SetWriteDeadline(t time.Time) error { + c.mu.Lock() + c.writeDL = t + c.mu.Unlock() + return nil +} + +func (c *quicConn) SetReadLimit(limit int64) { + c.mu.Lock() + c.limit = limit + c.mu.Unlock() +} + +func (c *quicConn) SetPingHandler(h func(appData string) error) { + c.mu.Lock() + c.ping = h + c.mu.Unlock() +} + +func (c *quicConn) Close() error { + c.closeWith(websocket.CloseGoingAway, "") + return nil +} + +var errQUICTimeout error = quicTimeout{} + +type quicTimeout struct{} + +func (quicTimeout) Error() string { return "cloudlink: read deadline passed" } +func (quicTimeout) Timeout() bool { return true } +func (quicTimeout) Temporary() bool { return true } diff --git a/go/internal/cloudlink/quic_test.go b/go/internal/cloudlink/quic_test.go new file mode 100644 index 00000000..f23d97a7 --- /dev/null +++ b/go/internal/cloudlink/quic_test.go @@ -0,0 +1,483 @@ +package cloudlink + +import ( + "bufio" + "context" + "crypto/ecdsa" + "errors" + "io" + "log/slog" + "net" + "path/filepath" + "slices" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/quic-go/quic-go" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// The uplink over QUIC against a service that speaks it (docs/protocol-cloud.md, +// "QUIC"): the race with the WebSocket, the bindings, and how each message +// travels. + +// serveQUIC makes the fake service answer QUIC on the UDP port that matches +// its TCP one, with the same handshake and loop as its WebSocket. A dropping +// service reads every packet and answers none, as a network that blocks UDP. +func (s *fakeService) serveQUIC(dropping bool) string { return s.serveQUICWith(dropping, false) } + +// serveQUICWith is serveQUIC with early data on when early, and each +// connection's use of it kept in usedEarly. +func (s *fakeService) serveQUICWith(dropping, early bool) string { + s.t.Helper() + addr := s.srv.Listener.Addr().String() + pc, err := net.ListenPacket("udp", addr) + if err != nil { + s.t.Fatal(err) + } + s.t.Cleanup(func() { _ = pc.Close() }) + if dropping { + go func() { + buf := make([]byte, 2048) + for { + if _, _, err := pc.ReadFrom(buf); err != nil { + return + } + } + }() + return addr + } + tr := &quic.Transport{Conn: pc} + s.t.Cleanup(func() { _ = tr.Close() }) + tc := s.srv.TLS.Clone() + tc.NextProtos = []string{cloudwire.ALPNBox} + conf := &quic.Config{MaxIdleTimeout: cloudwire.QUICIdleTimeout, EnableDatagrams: true, Allow0RTT: early} + ln, err := tr.ListenEarly(tc, conf) + if err != nil { + s.t.Fatal(err) + } + go func() { + for { + qc, err := ln.Accept(context.Background()) + if err != nil { + return + } + go func() { + ctrl, err := qc.AcceptStream(context.Background()) + if err != nil { + return + } + r := bufio.NewReader(ctrl) + target, err := cloudwire.ReadMessage(r, cloudwire.MaxTarget) + if err != nil || string(target) != cloudwire.TargetBox { + s.t.Errorf("the box's target: %q, %v", target, err) + return + } + c := &quicConn{qc: qc, ctrl: ctrl, r: r, in: make(chan inbound), kick: make(chan struct{}, 1), + limit: cloudwire.MaxTunFrame} + // The exporter exists once the server's flight is out, and + // hello can only come after the handshake. + <-qc.HandshakeComplete() + s.mu.Lock() + s.usedEarly = append(s.usedEarly, qc.ConnectionState().Used0RTT) + s.mu.Unlock() + ekm, _ := c.exporter() + s.serve(c, ekm) + }() + } + }() + return addr +} + +// A box that dials again sends its target as early data, with the ticket +// from the first connection; hello still binds to the TLS session it rides. +func TestAReconnectSendsItsTargetEarly(t *testing.T) { + svc := newFakeService(t) + svc.serveQUICWith(false, true) + svc.kick = 1 + h := newQUICHarness(t, svc) + h.grant(t, protocol.ScopeLiveRead) + waitWithin(t, 10*time.Second, "a second connection", func() bool { return svc.count(cloudwire.TransportQUIC) >= 2 }) + waitUntil(t, "the uplink up", func() bool { return h.link.Status().Connected }) + svc.mu.Lock() + early := slices.Clone(svc.usedEarly) + svc.mu.Unlock() + if len(early) < 2 || early[0] || !early[1] { + t.Fatalf("early data per connection: %v, want the second only", early) + } + if got := h.link.Status().Transport; got != cloudwire.TransportQUIC { + t.Fatalf("transport %q", got) + } +} + +func (s *fakeService) kindsOf(t string) []int { + s.mu.Lock() + defer s.mu.Unlock() + return slices.Clone(s.kinds[t]) +} + +func (s *fakeService) count(transport string) int { + s.mu.Lock() + defer s.mu.Unlock() + return s.transports[transport] +} + +// newQUICHarness is a box that races QUIC against the WebSocket. +func newQUICHarness(t *testing.T, svc *fakeService) *harness { + t.Helper() + h := &harness{svc: svc, site: &fakeSite{grid: 1200, state: protocol.SourceLive}, kv: &memKV{}, + keyPath: filepath.Join(t.TempDir(), "cloud.key"), pushed: make(chan string, 8), + roster: &fakeRosterState{}, tunnel: &fakeTunnel{ch: make(chan cloudwire.Tun, 64)}} + link, err := New(Options{ + Site: h.site, Roster: fakeRoster{state: h.roster}, KV: h.kv, KeyPath: h.keyPath, Build: "test", + Started: time.Now().Add(-time.Minute), ClockSynced: func() bool { return true }, + Notify: func(kind string) { h.pushed <- kind }, Tunnel: h.tunnel, + Endpoint: svc.endpoint(), QUIC: true, ServiceKeys: []*ecdsa.PublicKey{&svc.key.PublicKey}, + RootCAs: svc.roots(), Tick: 20 * time.Millisecond, + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + }) + if err != nil { + t.Fatal(err) + } + h.link = link + ctx, cancel := context.WithCancel(context.Background()) + h.cancel, h.done = cancel, make(chan error, 1) + go func() { h.done <- link.Run(ctx) }() + t.Cleanup(func() { + cancel() + <-h.done + }) + return h +} + +func waitUntil(t *testing.T, what string, ok func() bool) { + t.Helper() + waitWithin(t, 5*time.Second, what, ok) +} + +func waitWithin(t *testing.T, d time.Duration, what string, ok func() bool) { + t.Helper() + deadline := time.Now().Add(d) + for !ok() { + if time.Now().After(deadline) { + t.Fatalf("never: %s", what) + } + time.Sleep(10 * time.Millisecond) + } +} + +// Where UDP gets through, the uplink rides QUIC: hello binds to the TLS +// session inside it (the fake service checks), the status says quic, hb and +// a live without its table go as datagrams, and a live with it and the +// roster on the control stream. +func TestTheUplinkRidesQUIC(t *testing.T) { + svc := newFakeService(t) + svc.serveQUIC(false) + h := newQUICHarness(t, svc) + h.grant(t, protocol.ScopeLiveRead) + svc.waitFor(cloudwire.MsgLive, 5*time.Second) + if got := h.link.Status(); !got.Connected || got.Transport != cloudwire.TransportQUIC { + t.Fatalf("status %+v", got) + } + if svc.count(cloudwire.TransportWebSocket) != 0 { + t.Fatal("the WebSocket was dialled while QUIC opened at once") + } + // Without a watch a changed live goes within the 5 s period. + h.site.setGrid(2400) + waitWithin(t, 12*time.Second, "a live without its table", func() bool { return len(svc.kindsOf(cloudwire.MsgLive)) >= 2 }) + svc.waitFor(cloudwire.MsgHB, 20*time.Second) + lives := svc.kindsOf(cloudwire.MsgLive) + if lives[0] != websocket.BinaryMessage || lives[1] != msgDatagram { + t.Fatalf("live kinds %v: the first carries its table, the next does not", lives) + } + if hb := svc.kindsOf(cloudwire.MsgHB); hb[0] != msgDatagram { + t.Fatalf("hb kinds %v", hb) + } + if r := svc.kindsOf(cloudwire.MsgRoster); r[0] != websocket.BinaryMessage { + t.Fatalf("roster kinds %v", r) + } +} + +// Where UDP is blocked, the WebSocket starts after the race delay and +// carries the uplink; the next connection on the same network starts with +// it. +func TestTheWebSocketTakesOverWhenUDPIsBlocked(t *testing.T) { + svc := newFakeService(t) + svc.serveQUIC(true) + h := newQUICHarness(t, svc) + start := time.Now() + h.grant(t, protocol.ScopeLiveRead) + waitUntil(t, "the uplink up", func() bool { return h.link.Status().Connected }) + took := time.Since(start) + if got := h.link.Status().Transport; got != cloudwire.TransportWebSocket { + t.Fatalf("transport %q", got) + } + if took < cloudwire.RaceDelay { + t.Fatalf("the WebSocket opened %v in, before QUIC had its lead", took) + } + addr, _ := h.link.quicAddr() + if got := h.link.preferred(localNetwork(addr)); got != cloudwire.TransportWebSocket { + t.Fatalf("the race remembered %q", got) + } + + // The same network again: the WebSocket goes first, with no lead to wait. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + again := time.Now() + conn, err := h.link.dial(ctx) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if transportOf(conn) != cloudwire.TransportWebSocket || time.Since(again) >= cloudwire.RaceDelay { + t.Fatalf("redial: %s after %v", transportOf(conn), time.Since(again)) + } +} + +// A slow QUIC path loses the race to the WebSocket, then opens: the uplink +// stays on the WebSocket, and QUIC goes first on this network next time. +func TestALateQUICTeachesThatUDPGetsThrough(t *testing.T) { + svc := newFakeService(t) + // The QUIC listener sits behind a proxy on the TCP port's UDP twin that + // holds every packet 400 ms, so QUIC opens well after the WebSocket. + back, err := net.ListenPacket("udp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = back.Close() }) + tr := &quic.Transport{Conn: back} + t.Cleanup(func() { _ = tr.Close() }) + tc := svc.srv.TLS.Clone() + tc.NextProtos = []string{cloudwire.ALPNBox} + ln, err := tr.Listen(tc, &quic.Config{MaxIdleTimeout: cloudwire.QUICIdleTimeout, EnableDatagrams: true}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ln.Close() }) + front, err := net.ListenPacket("udp", svc.srv.Listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = front.Close() }) + go slowProxy(front, back.LocalAddr(), 400*time.Millisecond) + + h := newQUICHarness(t, svc) + h.grant(t, protocol.ScopeLiveRead) + waitUntil(t, "the uplink up", func() bool { return h.link.Status().Connected }) + if got := h.link.Status().Transport; got != cloudwire.TransportWebSocket { + t.Fatalf("the slow path carried the uplink: %q", got) + } + addr, _ := h.link.quicAddr() + waitWithin(t, 5*time.Second, "QUIC learned", func() bool { + return h.link.preferred(localNetwork(addr)) == cloudwire.TransportQUIC + }) +} + +// slowProxy forwards UDP between clients of front and back, holding each +// packet d. +func slowProxy(front net.PacketConn, back net.Addr, d time.Duration) { + var mu sync.Mutex + ups := map[string]net.PacketConn{} + buf := make([]byte, 2048) + for { + n, from, err := front.ReadFrom(buf) + if err != nil { + return + } + pkt := append([]byte(nil), buf[:n]...) + mu.Lock() + up := ups[from.String()] + if up == nil { + if up, err = net.ListenPacket("udp", "127.0.0.1:0"); err != nil { + mu.Unlock() + continue + } + ups[from.String()] = up + go func(up net.PacketConn, client net.Addr) { + b := make([]byte, 2048) + for { + n, _, err := up.ReadFrom(b) + if err != nil { + return + } + p := append([]byte(nil), b[:n]...) + time.AfterFunc(d, func() { _, _ = front.WriteTo(p, client) }) + } + }(up, from) + } + mu.Unlock() + time.AfterFunc(d, func() { _, _ = up.WriteTo(pkt, back) }) + } +} + +// A tun carrying bulk goes on the bulk stream; lane 0 stays on the control +// stream. +func TestBulkTunTakesTheBulkStream(t *testing.T) { + svc := newFakeService(t) + svc.serveQUIC(false) + h := newQUICHarness(t, svc) + h.grant(t, protocol.ScopeLiveRead) + waitUntil(t, "the uplink up", func() bool { return h.link.Status().Connected }) + waitUntil(t, "the tunnel's queues", func() bool { + return h.link.SendTun(cloudwire.Tun{Dev: "phone", Sid: 1, D: []byte{1}}, false) == nil + }) + if err := h.link.SendTun(cloudwire.Tun{Dev: "phone", Sid: 1, D: make([]byte, 4096)}, true); err != nil { + t.Fatal(err) + } + waitUntil(t, "two tuns", func() bool { return len(svc.kindsOf(cloudwire.MsgTun)) >= 2 }) + // The two travel on two streams, which QUIC keeps in no order between + // them, so each is told by its size rather than by which came first. + bySize := svc.tunKindsBySize() + if bySize[1] != websocket.BinaryMessage || bySize[4096] != msgBulk { + t.Fatalf("tun kinds by size %v: lane 0 should come by the control stream (%d), bulk by the bulk stream (%d)", + bySize, websocket.BinaryMessage, msgBulk) + } +} + +// tunKindsBySize is how each tun from the box came, by the size of its d. +func (s *fakeService) tunKindsBySize() map[int]int { + s.mu.Lock() + defer s.mu.Unlock() + out := map[int]int{} + n := 0 + for _, env := range s.got { + if env.T != cloudwire.MsgTun { + continue + } + var tun cloudwire.Tun + if cloudwire.Unmarshal(env.B, &tun) == nil && n < len(s.kinds[cloudwire.MsgTun]) { + out[len(tun.D)] = s.kinds[cloudwire.MsgTun][n] + } + n++ + } + return out +} + +// A datagram from the service, or anything but tun on its bulk stream, +// breaks the protocol, and the box closes. +func TestTheBoxRefusesDatagramsFromTheService(t *testing.T) { + for _, kind := range []int{msgDatagram, msgBulk} { + if _, err := parseDownlink(kind, mustFrame(t, cloudwire.MsgWatch, cloudwire.Watch{Hz: 1})); err == nil { + t.Fatalf("a watch as message type %d was taken", kind) + } + } + if _, err := parseDownlink(msgBulk, mustFrame(t, cloudwire.MsgTun, cloudwire.Tun{Dev: "p", Sid: 1, D: []byte{1}})); err != nil { + t.Fatalf("a tun on the bulk stream: %v", err) + } +} + +func mustFrame(t *testing.T, typ string, body any) []byte { + t.Helper() + b, err := cloudwire.Marshal(body) + if err != nil { + t.Fatal(err) + } + raw, _ := cloudwire.Envelope{T: typ, B: b}.Encode() + frame, _ := cloudwire.EncodeFrame(raw) + return frame +} + +// countingConn counts the bytes and packets a UDP socket moves. +type countingConn struct { + net.PacketConn + mu sync.Mutex + sent, recv int + sentPkts, recvPkts int +} + +func (c *countingConn) WriteTo(p []byte, addr net.Addr) (int, error) { + n, err := c.PacketConn.WriteTo(p, addr) + c.mu.Lock() + c.sent += n + c.sentPkts++ + c.mu.Unlock() + return n, err +} + +func (c *countingConn) ReadFrom(p []byte) (int, net.Addr, error) { + n, addr, err := c.PacketConn.ReadFrom(p) + c.mu.Lock() + c.recv += n + c.recvPkts++ + c.mu.Unlock() + return n, addr, err +} + +// What counts as QUIC failing on a network, and what does not. +func TestWhatCountsAsQUICFailing(t *testing.T) { + for _, c := range []struct { + name string + err error + up bool + upFor time.Duration + want bool + }{ + {"closed before hello_ok, below 1000", &websocket.CloseError{Code: 0x110}, false, 0, true}, + {"a path error before hello_ok", errors.New("timeout: no recent network activity"), false, 0, true}, + {"the path dropped it soon after", errors.New("timeout: no recent network activity"), true, 10 * time.Second, true}, + {"the path dropped it after a while", errors.New("timeout: no recent network activity"), true, 2 * time.Minute, false}, + {"the service restarts", &websocket.CloseError{Code: cloudwire.CloseRestart}, true, time.Second, false}, + {"the service refuses on purpose", &websocket.CloseError{Code: cloudwire.CloseDenied}, false, 0, false}, + {"a reset after a restart", &quic.StatelessResetError{}, true, time.Second, false}, + {"the link stopped", errLinkEnded, false, 0, false}, + } { + if got := quicFailed(c.err, c.up, c.upFor); got != c.want { + t.Errorf("%s: %v, want %v", c.name, got, c.want) + } + } +} + +// A service whose QUIC side finishes the handshake and then closes before +// the challenge, as a door with the sessions turned off once did, does not +// hold the box on QUIC: the box goes over to the WebSocket, and QUIC stays +// second on this network even when a QUIC dial wins again. +func TestAQUICSessionThatEndsBeforeHelloOKDemotesQUIC(t *testing.T) { + svc := newFakeService(t) + addr := svc.srv.Listener.Addr().String() + pc, err := net.ListenPacket("udp", addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = pc.Close() }) + tr := &quic.Transport{Conn: pc} + t.Cleanup(func() { _ = tr.Close() }) + tc := svc.srv.TLS.Clone() + tc.NextProtos = []string{cloudwire.ALPNBox} + ln, err := tr.Listen(tc, &quic.Config{MaxIdleTimeout: cloudwire.QUICIdleTimeout}) + if err != nil { + t.Fatal(err) + } + var refused atomic.Int32 + go func() { + for { + qc, err := ln.Accept(context.Background()) + if err != nil { + return + } + refused.Add(1) + _ = qc.CloseWithError(0x110, "") + } + }() + + h := newQUICHarness(t, svc) + h.grant(t, protocol.ScopeLiveRead) + waitWithin(t, 20*time.Second, "the uplink up", func() bool { return h.link.Status().Connected }) + if got := h.link.Status().Transport; got != cloudwire.TransportWebSocket { + t.Fatalf("transport %q", got) + } + if refused.Load() == 0 { + t.Fatal("QUIC was never tried") + } + network := localNetwork(addr) + if got := h.link.preferred(network); got != cloudwire.TransportWebSocket { + t.Fatalf("after QUIC failed, %q goes first", got) + } + h.link.rememberWinner(network, cloudwire.TransportQUIC) + if got := h.link.preferred(network); got != cloudwire.TransportWebSocket { + t.Fatalf("a QUIC win moved %q first within the day", got) + } +} diff --git a/go/internal/cloudlink/race.go b/go/internal/cloudlink/race.go new file mode 100644 index 00000000..0076305f --- /dev/null +++ b/go/internal/cloudlink/race.go @@ -0,0 +1,239 @@ +package cloudlink + +import ( + "context" + "errors" + "net" + "net/url" + "time" + + "github.com/gorilla/websocket" + "github.com/quic-go/quic-go" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// The race (docs/protocol-cloud.md, "Racing the WebSocket"): QUIC starts +// first; after cloudwire.RaceDelay without a QUIC handshake the WebSocket +// starts too, and the first to open carries the uplink. The link remembers +// the winner for each network, named by the box's local address towards the +// service, and starts the next time with it, for a day: a network that +// blocked UDP once gets QUIC first again after that. + +// winnerHolds is how long a network's winner goes first. +const winnerHolds = 24 * time.Hour + +type winner struct { + transport string + at time.Time + // demoted marks a network where QUIC failed after its handshake (see + // quicFailed): the WebSocket goes first, and no QUIC win moves QUIC + // back until the entry is a day old. + demoted bool +} + +type dialed struct { + transport string + conn uplinkConn + err error +} + +// dial opens the uplink by whichever transport opens first. +func (l *Link) dial(ctx context.Context) (uplinkConn, error) { + addr, quicOn := l.quicAddr() + if !quicOn { + return l.dialWebSocket(ctx) + } + network := localNetwork(addr) + first, second := cloudwire.TransportQUIC, cloudwire.TransportWebSocket + if l.preferred(network) == cloudwire.TransportWebSocket { + first, second = second, first + } + + // QUIC dials on a context of its own: when the WebSocket wins, a QUIC + // dial still under way goes on for learnFor, only to learn whether UDP + // gets through here. One lost packet should not keep QUIC second for a + // day. + wsCtx, cancelWS := context.WithCancel(ctx) + // Not ctx itself: the caller ends ctx as soon as dial returns, which + // would end the dial that is still learning. + quicCtx, cancelQUIC := context.WithTimeout(context.WithoutCancel(ctx), dialTimeout) + results := make(chan dialed, 2) + start := func(transport string) { + go func() { + var conn uplinkConn + var err error + if transport == cloudwire.TransportQUIC { + var q *quicConn + if q, err = dialQUIC(quicCtx, addr, l.quicTLSConfig(), nil); err == nil { + q.network = network + conn = q + } + } else { + conn, err = l.dialWebSocket(wsCtx) + } + results <- dialed{transport: transport, conn: conn, err: err} + }() + } + start(first) + started := 1 + lead := time.NewTimer(cloudwire.RaceDelay) + defer lead.Stop() + var errs []error + for { + select { + case <-lead.C: + if started == 1 { + start(second) + started = 2 + } + case r := <-results: + if r.err == nil { + l.rememberWinner(network, r.transport) + if started == 2 { + l.afterRace(network, r.transport, results, cancelQUIC) + } else { + cancelQUIC() + } + cancelWS() + return r.conn, nil + } + errs = append(errs, r.err) + if started == 1 { + start(second) + started = 2 + continue + } + if len(errs) == 2 { + cancelQUIC() + cancelWS() + return nil, errors.Join(errs...) + } + } + } +} + +// learnFor is how long a QUIC dial that lost goes on. +const learnFor = 3 * time.Second + +// afterRace waits for the loser. A QUIC dial that opens within learnFor +// teaches that UDP gets through here, so QUIC goes first next time; either +// way the loser is closed. +func (l *Link) afterRace(network, won string, results <-chan dialed, cancelQUIC context.CancelFunc) { + go func() { + t := time.NewTimer(learnFor) + defer t.Stop() + defer cancelQUIC() + select { + case late := <-results: + if late.err == nil { + if late.transport == cloudwire.TransportQUIC && won == cloudwire.TransportWebSocket { + l.rememberWinner(network, cloudwire.TransportQUIC) + } + _ = late.conn.Close() + } + case <-t.C: + cancelQUIC() + if late := <-results; late.err == nil { + _ = late.conn.Close() + } + } + }() +} + +func (l *Link) dialWebSocket(ctx context.Context) (uplinkConn, error) { + conn, resp, err := l.dialer().DialContext(ctx, l.endpoint(), nil) + if resp != nil && resp.Body != nil { + _ = resp.Body.Close() + } + if err != nil { + return nil, err + } + return conn, nil +} + +// quicAddr is the service's QUIC address: the WebSocket endpoint's host and +// port, over UDP, on the shard that holds the home. False when QUIC is off: +// a test that names its own endpoint and does not ask for QUIC. +func (l *Link) quicAddr() (string, bool) { + if !l.quicOn { + return "", false + } + if l.opts.QUICAddr != "" { + return l.opts.QUICAddr, true + } + u, err := url.Parse(l.endpoint()) + if err != nil || u.Hostname() == "" { + return "", false + } + port := u.Port() + if port == "" { + port = "443" + } + return net.JoinHostPort(u.Hostname(), port), true +} + +// localNetwork names the network the box reaches addr through: its local +// address on the route there. Connecting a UDP socket sends nothing. +func localNetwork(addr string) string { + c, err := net.Dial("udp", addr) + if err != nil { + return "" + } + defer c.Close() + if a, ok := c.LocalAddr().(*net.UDPAddr); ok { + return a.IP.String() + } + return "" +} + +// preferred is the transport that won last on network, QUIC when none has. +func (l *Link) preferred(network string) string { + l.mu.Lock() + defer l.mu.Unlock() + if w, ok := l.winners[network]; ok && time.Since(w.at) < winnerHolds { + return w.transport + } + return cloudwire.TransportQUIC +} + +func (l *Link) rememberWinner(network, transport string) { + l.mu.Lock() + defer l.mu.Unlock() + if l.winners == nil { + l.winners = map[string]winner{} + } + if w, ok := l.winners[network]; ok && w.demoted && time.Since(w.at) < winnerHolds { + return + } + l.winners[network] = winner{transport: transport, at: time.Now()} +} + +// demote puts the WebSocket first on network for a day, whatever a QUIC +// dial wins there meanwhile. +func (l *Link) demote(network string) { + l.mu.Lock() + defer l.mu.Unlock() + if l.winners == nil { + l.winners = map[string]winner{} + } + l.winners[network] = winner{transport: cloudwire.TransportWebSocket, at: time.Now(), demoted: true} +} + +// quicFailed reports whether a QUIC uplink that ended with err counts as QUIC +// failing on its network (docs/protocol-cloud.md, "Racing the WebSocket"): a +// handshake that finished proves less than a session that works. It counts +// when the uplink ended before hello_ok, or when the path dropped it within +// QUICProvenAfter of it. A close the service sent on purpose, with a code +// from 1000 up, and a stateless reset after a restart do not count. +func quicFailed(err error, up bool, upFor time.Duration) bool { + if err == nil || errors.Is(err, errLinkEnded) || errors.Is(err, errClosed) { + return false + } + if closed, ok := errors.AsType[*websocket.CloseError](err); ok && closed.Code >= 1000 { + return false + } + if _, ok := errors.AsType[*quic.StatelessResetError](err); ok { + return false + } + return !up || upFor < cloudwire.QUICProvenAfter +} diff --git a/go/internal/cloudlink/shard_test.go b/go/internal/cloudlink/shard_test.go index 3c1e4098..75b667dd 100644 --- a/go/internal/cloudlink/shard_test.go +++ b/go/internal/cloudlink/shard_test.go @@ -22,7 +22,7 @@ import ( // closeWith ends a connection as the service does, with a close code and a // reason. The caller's read loop takes the box's answer. -func closeWith(conn *websocket.Conn, code int, reason string) { +func closeWith(conn uplinkConn, code int, reason string) { _ = conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(code, reason), time.Now().Add(time.Second)) } @@ -245,7 +245,7 @@ func TestARestartCloseRedialsWithinTheSpreadWithoutBackoff(t *testing.T) { svc := newFakeService(t) var mu sync.Mutex restarts := 0 - svc.afterHello = func(conn *websocket.Conn, _ string) { + svc.afterHello = func(conn uplinkConn, _ string) { mu.Lock() restarts++ n := restarts diff --git a/go/internal/cloudlink/transport.go b/go/internal/cloudlink/transport.go new file mode 100644 index 00000000..bce7c9cb --- /dev/null +++ b/go/internal/cloudlink/transport.go @@ -0,0 +1,77 @@ +package cloudlink + +import ( + "crypto/tls" + "errors" + "time" + + "github.com/gorilla/websocket" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// uplinkConn is what the uplink needs of a connection: gorilla/websocket's +// method set, which a WebSocket has as it is, plus the TLS exporter and the +// transport's name. quicConn is the other one (docs/protocol-cloud.md, +// "QUIC"). +type uplinkConn interface { + ReadMessage() (messageType int, p []byte, err error) + WriteMessage(messageType int, data []byte) error + WriteControl(messageType int, data []byte, deadline time.Time) error + SetReadDeadline(t time.Time) error + SetWriteDeadline(t time.Time) error + SetReadLimit(limit int64) + SetPingHandler(h func(appData string) error) + Close() error +} + +// Message types beyond the WebSocket's, for what QUIC carries apart. +const ( + // msgBulk is a tun carrying a lane 1 frame, on the bulk stream. + msgBulk = 0x100 + // msgDatagram is a QUIC DATAGRAM: never sent again, the newest wins. + msgDatagram = 0x101 +) + +// carrier is what a connection that carries more than binary messages says. +type carrier interface { + carries(messageType int) bool + exporter() ([]byte, error) + transport() string +} + +// kindFor is the message type to write a message of the wanted type with: +// that one where conn carries it, a binary message anywhere else. +func kindFor(conn uplinkConn, want int) int { + if c, ok := conn.(carrier); ok && c.carries(want) { + return want + } + return websocket.BinaryMessage +} + +// transportOf is the registry's name for how conn travels. +func transportOf(conn uplinkConn) string { + if c, ok := conn.(carrier); ok { + return c.transport() + } + return cloudwire.TransportWebSocket +} + +// exporterValue is the TLS 1.3 exporter value both ends bind the handshake to. +func exporterValue(conn uplinkConn) ([]byte, error) { + if c, ok := conn.(carrier); ok { + return c.exporter() + } + ws, ok := conn.(*websocket.Conn) + if !ok { + return nil, errors.New("cloudlink: the uplink is not TLS") + } + tc, ok := ws.NetConn().(*tls.Conn) + if !ok { + return nil, errors.New("cloudlink: the uplink is not TLS") + } + cs := tc.ConnectionState() + if cs.Version != tls.VersionTLS13 { + return nil, errors.New("cloudlink: the uplink is not TLS 1.3") + } + return cs.ExportKeyingMaterial(cloudwire.EKMLabel, nil, cloudwire.EKMBytes) +} diff --git a/go/internal/cloudlink/tunnel_test.go b/go/internal/cloudlink/tunnel_test.go index 377b45a6..c968e8aa 100644 --- a/go/internal/cloudlink/tunnel_test.go +++ b/go/internal/cloudlink/tunnel_test.go @@ -10,7 +10,6 @@ import ( "testing" "time" - "github.com/gorilla/websocket" "github.com/srcfl/ftw/go/protocol" "github.com/srcfl/ftw/go/protocol/cloudwire" ) @@ -92,7 +91,7 @@ func TestAPendingCodeIsListedByItsHash(t *testing.T) { // tunnel past its own does not stop the copy. func TestTunGoesBothWaysApartFromTheBudget(t *testing.T) { svc := newFakeService(t) - svc.afterHello = func(conn *websocket.Conn, home string) { + svc.afterHello = func(conn uplinkConn, home string) { svc.write(conn, nil, home, cloudwire.MsgTun, cloudwire.Tun{Dev: "q5Lw0Zb1", Sid: 9, D: []byte{1, 2, 3}}) } h := newHarness(t, svc) @@ -202,7 +201,7 @@ func TestSendTunWithoutAnUplinkIsRefused(t *testing.T) { // Every session the uplink carried ends with it. func TestTheTunnelDropsWithTheUplink(t *testing.T) { svc := newFakeService(t) - svc.afterHello = func(conn *websocket.Conn, _ string) { + svc.afterHello = func(conn uplinkConn, _ string) { time.Sleep(300 * time.Millisecond) _ = conn.Close() } diff --git a/go/protocol/appwire/transport.go b/go/protocol/appwire/transport.go index b372da08..2c1899e9 100644 --- a/go/protocol/appwire/transport.go +++ b/go/protocol/appwire/transport.go @@ -29,13 +29,19 @@ const ( TransportOverhead = SeqBytes + TagBytes // ReplayWindow is how far out of order a frame may arrive and still be - // accepted: one machine word. Over the relay lane 0 goes once a second, - // so that is a minute. Through FTW Cloud it goes up to five times a - // second, about 13 s, while a bulk frame the link paces waits at most - // cloudlink.MaxBulkWait, about 9 s (appuplink tests the two). - ReplayWindow = 64 + // accepted. Through FTW Cloud lane 0 goes up to five times a second, + // while a bulk frame the link paces waits at most cloudlink.MaxBulkWait, + // about 9 s (appuplink tests the two). Over QUIC one session's frames + // also travel on two streams and in datagrams, so a burst of bulk + // frames, a few hundred at most, can overtake a frame on the control + // stream. 1024 covers both in 128 bytes, far beyond any reordering a + // DataChannel produces. + ReplayWindow = 1024 ) +// seenWords is the replay window in 64-bit words. +const seenWords = ReplayWindow / 64 + // ErrReplay is a frame already accepted, or one older than the window. var ErrReplay = errors.New("appwire: replayed or out-of-window sequence") @@ -54,8 +60,10 @@ type Transport struct { // sentinel because sequence zero is a real, and the first, frame. highestSeq uint64 haveSeq bool - seen uint64 - closed bool + // seen holds a bit for each sequence number in the window: bit i (word + // i/64, bit i%64) is highestSeq-i, set once that frame was accepted. + seen [seenWords]uint64 + closed bool handshakeHash []byte remoteStatic []byte @@ -164,7 +172,7 @@ func (t *Transport) checkReplay(seq uint64) error { if behind >= ReplayWindow { return fmt.Errorf("%w: sequence %d is outside the window", ErrReplay, seq) } - if t.seen>>behind&1 == 1 { + if t.seen[behind/64]>>(behind%64)&1 == 1 { return fmt.Errorf("%w: sequence %d was already accepted", ErrReplay, seq) } return nil @@ -175,20 +183,37 @@ func (t *Transport) markSeen(seq uint64) { if !t.haveSeq { t.haveSeq = true t.highestSeq = seq - t.seen = 1 + t.seen = [seenWords]uint64{1} return } if seq > t.highestSeq { - shift := seq - t.highestSeq - if shift >= ReplayWindow { - t.seen = 1 - } else { - t.seen = t.seen<= ReplayWindow { + t.seen = [seenWords]uint64{} + return + } + words, bits := int(n/64), n%64 + for i := seenWords - 1; i >= 0; i-- { + var w uint64 + if j := i - words; j >= 0 { + w = t.seen[j] << bits + if bits > 0 && j > 0 { + w |= t.seen[j-1] >> (64 - bits) + } + } + t.seen[i] = w + } } diff --git a/go/protocol/appwire/transport_test.go b/go/protocol/appwire/transport_test.go index 59c2bcad..6bf7229f 100644 --- a/go/protocol/appwire/transport_test.go +++ b/go/protocol/appwire/transport_test.go @@ -4,6 +4,7 @@ import ( "bytes" "crypto/rand" "errors" + mrand "math/rand/v2" "testing" ) @@ -373,3 +374,70 @@ func TestAPassiveObserverLearnsOnlyThatAFrameWentBy(t *testing.T) { t.Fatalf("a run of %d zero bytes survived encryption", longest) } } + +// Over QUIC a burst on the bulk stream can overtake a frame on the control +// stream. The late frame still opens, once. +func TestABurstOfAHundredCanOvertakeAFrame(t *testing.T) { + app, box := connect(t) + late, err := app.Encrypt(tickFrame(t)) + if err != nil { + t.Fatal(err) + } + for i := range 100 { + wire, err := app.Encrypt(deltaFrame(t, map[string]int{"2": i})) + if err != nil { + t.Fatal(err) + } + if _, err := box.Decrypt(wire); err != nil { + t.Fatalf("burst frame %d: %v", i, err) + } + } + if _, err := box.Decrypt(late); err != nil { + t.Fatalf("the overtaken frame: %v", err) + } + if _, err := box.Decrypt(late); !errors.Is(err, ErrReplay) { + t.Fatalf("the overtaken frame again: err = %v, want ErrReplay", err) + } +} + +// The window spans 16 words. Frames in a shuffled order, spread over many +// shifts of every size, open exactly as a plain record of what was seen +// says they should. +func TestTheWindowMatchesARecordOfWhatWasSeen(t *testing.T) { + app, box := connect(t) + const n = 3000 + wires := make([][]byte, n) + for i := range wires { + w, err := app.Encrypt(deltaFrame(t, map[string]int{"2": i})) + if err != nil { + t.Fatal(err) + } + wires[i] = w + } + rng := mrand.New(mrand.NewPCG(1, 2)) + order := make([]int, 0, 2*n) + for i := range n { + order = append(order, i) + } + // Shuffle locally, up to twice the window, and replay a tenth. + for i := range order { + j := min(len(order)-1, i+rng.IntN(2*ReplayWindow)) + order[i], order[j] = order[j], order[i] + } + for range n / 10 { + order = append(order, rng.IntN(n)) + } + seen := map[int]bool{} + highest := -1 + for _, i := range order { + want := !seen[i] && (highest < 0 || i > highest || highest-i < ReplayWindow) + _, err := box.Decrypt(wires[i]) + if got := err == nil; got != want { + t.Fatalf("frame %d with highest %d: opened %v, want %v (err %v)", i, highest, got, want, err) + } + if err == nil { + seen[i] = true + highest = max(highest, i) + } + } +} diff --git a/go/protocol/cloudwire/quic.go b/go/protocol/cloudwire/quic.go new file mode 100644 index 00000000..632c33f6 --- /dev/null +++ b/go/protocol/cloudwire/quic.go @@ -0,0 +1,156 @@ +package cloudwire + +import ( + "bufio" + "errors" + "fmt" + "io" + "time" + + "github.com/fxamacker/cbor/v2" + "github.com/srcfl/ftw/go/protocol" +) + +// QUIC: the uplink and the app session over QUIC on UDP 443, beside the +// WebSocket. docs/protocol-cloud.md, "QUIC", is the normative text. The +// messages are the WebSocket's; only the framing is new. + +// ALPN names, one per connection. +const ( + ALPNBox = "ftw-cloud-box/1" + ALPNApp = "ftw-cloud-app/1" +) + +// The target is the first message on the control stream: the path and query +// the WebSocket would name. +const ( + TargetBox = "/v1/box" + // AppPath is the app session's path; AppTarget adds the home. + AppPath = "/v1/app" + // MaxTarget bounds the target message. + MaxTarget = 256 +) + +// AppTarget is the target of an app session with home. +func AppTarget(home string) string { return AppPath + "?" + AppHomeParam + "=" + home } + +// Timing, shared by every client and the service. +const ( + // QUICKeepAlive is how long a client stays silent before it sends a + // QUIC PING, well inside the 30 s after which many NATs forget a UDP + // mapping. + QUICKeepAlive = 15 * time.Second + // QUICIdleTimeout closes a connection silent this long, as the + // WebSocket's read deadline does. + QUICIdleTimeout = 45 * time.Second + // RaceDelay is how long QUIC leads before a client also dials the + // WebSocket. + RaceDelay = 250 * time.Millisecond + // EarlyDataFor is how long a session ticket allows 0-RTT data. + EarlyDataFor = 24 * time.Hour + // QUICProvenAfter is how long a QUIC session must last past its first + // session message before the path dropping it stops counting as QUIC + // failing on that network. + QUICProvenAfter = 60 * time.Second +) + +// The transports under a carrier, the registry's cloud_transports, as +// generated from it. +const ( + TransportQUIC = protocol.CloudTransportQuic + TransportWebSocket = protocol.CloudTransportWebsocket +) + +// Transports lists them in the registry's order. +var Transports = protocol.CloudTransports + +// ErrQUICMessage is a message the QUIC framing refuses: a length over the +// limit, or one cut short. +var ErrQUICMessage = errors.New("cloudwire: not a QUIC message") + +// maxVarint is the largest QUIC variable-length integer. +const maxVarint = 1<<62 - 1 + +// AppendVarint appends v as a QUIC variable-length integer (RFC 9000 §16): +// the two high bits of the first byte give the length, 1, 2, 4 or 8 bytes. +func AppendVarint(dst []byte, v uint64) []byte { + switch { + case v < 1<<6: + return append(dst, byte(v)) + case v < 1<<14: + return append(dst, 0x40|byte(v>>8), byte(v)) + case v < 1<<30: + return append(dst, 0x80|byte(v>>24), byte(v>>16), byte(v>>8), byte(v)) + case v <= maxVarint: + return append(dst, 0xc0|byte(v>>56), byte(v>>48), byte(v>>40), byte(v>>32), + byte(v>>24), byte(v>>16), byte(v>>8), byte(v)) + } + panic("cloudwire: varint out of range") +} + +// ReadVarint reads one QUIC variable-length integer. +func ReadVarint(r io.ByteReader) (uint64, error) { + b, err := r.ReadByte() + if err != nil { + return 0, err + } + n := 1 << (b >> 6) + v := uint64(b & 0x3f) + for i := 1; i < n; i++ { + b, err := r.ReadByte() + if err != nil { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + return 0, err + } + v = v<<8 | uint64(b) + } + return v, nil +} + +// AppendMessage appends one message as it goes on a stream: its length as +// a varint, then its bytes. +func AppendMessage(dst, msg []byte) []byte { + return append(AppendVarint(dst, uint64(len(msg))), msg...) +} + +// ReadMessage reads one message from a stream. A length over max is +// ErrQUICMessage and nothing of the message is read. io.EOF means the +// stream ended cleanly between messages. +func ReadMessage(r *bufio.Reader, max int) ([]byte, error) { + n, err := ReadVarint(r) + if err != nil { + return nil, err + } + if n > uint64(max) { + return nil, fmt.Errorf("%w: %d bytes, over %d", ErrQUICMessage, n, max) + } + msg := make([]byte, n) + if _, err := io.ReadFull(r, msg); err != nil { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + return nil, err + } + return msg, nil +} + +// Datagram reports whether an envelope may go as a QUIC DATAGRAM: an hb, or +// a live without sources. Both are newest-wins: a lost one shows as a gap in +// the freshness rules, never as a value that looks live. A live that carries +// its source table must arrive, and so must everything else. +func Datagram(env Envelope) bool { + switch env.T { + case MsgHB: + return true + case MsgLive: + var keys map[string]cbor.RawMessage + if decMode.Unmarshal(env.B, &keys) != nil { + return false + } + _, table := keys["sources"] + return !table + } + return false +} diff --git a/go/protocol/cloudwire/quic_test.go b/go/protocol/cloudwire/quic_test.go new file mode 100644 index 00000000..986f8269 --- /dev/null +++ b/go/protocol/cloudwire/quic_test.go @@ -0,0 +1,181 @@ +package cloudwire + +import ( + "bufio" + "bytes" + "encoding/hex" + "encoding/json" + "errors" + "io" + "os" + "strings" + "testing" +) + +// testdata/cloud-quic-vectors.json freezes the QUIC framing: the ALPN names, +// varints at each width, message prefixes at the lengths where they change, +// the two targets, and which of the uplink's envelopes may go as a datagram. +// srcfl/ftw-cloud and the phone replay it. +// +// go test ./cloudwire -run TestQUICVectors -update + +const quicVectorsPath = "testdata/cloud-quic-vectors.json" + +type quicVectorFile struct { + ALPN map[string]string `json:"alpn"` + Transports []string `json:"transports"` + Varints []quicVectorVarint `json:"varints"` + Messages []quicVectorMessage `json:"messages"` + Targets []quicVectorTarget `json:"targets"` + Datagrams []quicVectorDatagram `json:"datagrams"` + Notes map[string]string `json:"notes"` +} + +type quicVectorVarint struct { + Value uint64 `json:"value"` + Hex string `json:"hex"` +} + +type quicVectorMessage struct { + Len int `json:"len"` + // Prefix is the varint the message starts with. + Prefix string `json:"prefix"` +} + +type quicVectorTarget struct { + Name string `json:"name"` + Target string `json:"target"` + // Message is the target as it goes first on the control stream. + Message string `json:"message"` +} + +type quicVectorDatagram struct { + // Name is the envelope's name in cloud-vectors.json. + Name string `json:"name"` + Envelope string `json:"envelope"` + Datagram bool `json:"datagram"` +} + +func TestQUICVectors(t *testing.T) { + raw, err := os.ReadFile(vectorsPath) + if err != nil { + t.Fatal(err) + } + var signed vectorFile + if err := json.Unmarshal(raw, &signed); err != nil { + t.Fatal(err) + } + + want := quicVectorFile{ + ALPN: map[string]string{"box": ALPNBox, "app": ALPNApp}, + Transports: Transports, + Notes: map[string]string{ + "message": "one WebSocket binary message: its length as a QUIC varint (RFC 9000 §16), then its bytes", + "target": "the first message on the control stream (client bidirectional stream 0), ASCII", + "datagram": "hb, and live without the sources key, may go as one QUIC DATAGRAM; nothing else", + }, + } + // RFC 9000 §A.1's own examples, then each width's edges. + for _, v := range []uint64{0, 37, 63, 64, 15293, 16383, 16384, 494878333, 1<<30 - 1, 1 << 30, 151288809941952652, maxVarint} { + want.Varints = append(want.Varints, quicVectorVarint{Value: v, Hex: hex.EncodeToString(AppendVarint(nil, v))}) + } + for _, n := range []int{0, 1, 63, 64, 536, 4096, 16383, 16384, MaxTunFrame} { + want.Messages = append(want.Messages, quicVectorMessage{Len: n, Prefix: hex.EncodeToString(AppendVarint(nil, uint64(n)))}) + } + for _, tg := range []struct{ name, target string }{ + {"box", TargetBox}, + {"app", AppTarget(signed.BoxKey.Home)}, + } { + want.Targets = append(want.Targets, quicVectorTarget{Name: tg.name, Target: tg.target, + Message: hex.EncodeToString(AppendMessage(nil, []byte(tg.target)))}) + } + for _, m := range signed.Messages { + b, _ := hex.DecodeString(m.Envelope) + env, err := DecodeEnvelope(b) + if err != nil { + t.Fatal(err) + } + want.Datagrams = append(want.Datagrams, quicVectorDatagram{Name: m.Name, Envelope: m.Envelope, Datagram: Datagram(env)}) + } + + out, err := json.MarshalIndent(want, "", " ") + if err != nil { + t.Fatal(err) + } + out = append(out, '\n') + if *update { + if err := os.WriteFile(quicVectorsPath, out, 0o644); err != nil { + t.Fatal(err) + } + } + got, err := os.ReadFile(quicVectorsPath) + if err != nil { + t.Fatalf("%v (run with -update)", err) + } + if !bytes.Equal(got, out) { + t.Fatalf("%s is out of date; run go test ./cloudwire -run TestQUICVectors -update", quicVectorsPath) + } + + // The file says what the rules say, read back through the decoder. + var file quicVectorFile + if err := json.Unmarshal(got, &file); err != nil { + t.Fatal(err) + } + for _, v := range file.Varints { + b, _ := hex.DecodeString(v.Hex) + n, err := ReadVarint(bytes.NewReader(b)) + if err != nil || n != v.Value { + t.Fatalf("varint %s reads %d, %v", v.Hex, n, err) + } + } + datagrams := map[string]bool{} + for _, d := range file.Datagrams { + datagrams[d.Name] = d.Datagram + } + for name, ok := range map[string]bool{"hb": true, "live delta": true, "live full": false, "prices": false, + "tun down": false, "tun up": false, "roster": false, "event connected": false} { + if datagrams[name] != ok { + t.Errorf("%s as a datagram: %v, want %v", name, datagrams[name], ok) + } + } +} + +func TestMessagesRoundTripOnAStream(t *testing.T) { + var stream []byte + msgs := [][]byte{{}, []byte("x"), bytes.Repeat([]byte{7}, 64), bytes.Repeat([]byte{9}, 16384), []byte(TargetBox)} + for _, m := range msgs { + stream = AppendMessage(stream, m) + } + r := bufio.NewReader(bytes.NewReader(stream)) + for i, m := range msgs { + got, err := ReadMessage(r, MaxTunFrame) + if err != nil || !bytes.Equal(got, m) { + t.Fatalf("message %d: %d bytes, %v", i, len(got), err) + } + } + if _, err := ReadMessage(r, MaxTunFrame); err != io.EOF { + t.Fatalf("after the last message: %v", err) + } +} + +// A length over the limit is refused before a byte of the message is read, +// so a peer cannot make the reader allocate what it declares. +func TestALengthOverTheLimitIsRefused(t *testing.T) { + r := bufio.NewReader(bytes.NewReader(AppendVarint(nil, maxVarint))) + if _, err := ReadMessage(r, MaxTunFrame); !errors.Is(err, ErrQUICMessage) { + t.Fatalf("a 2^62 length: %v", err) + } + cut := AppendMessage(nil, []byte("hello"))[:3] + if _, err := ReadMessage(bufio.NewReader(bytes.NewReader(cut)), 64); !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("a message cut short: %v", err) + } + if _, err := ReadVarint(bytes.NewReader([]byte{0x40})); !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("a varint cut short: %v", err) + } +} + +func TestTheAppTargetNamesTheHome(t *testing.T) { + if got := AppTarget("abc"); got != "/v1/app?home=abc" || !strings.HasPrefix(got, AppPath) { + t.Fatal(got) + } +} diff --git a/go/protocol/cloudwire/testdata/cloud-quic-vectors.json b/go/protocol/cloudwire/testdata/cloud-quic-vectors.json new file mode 100644 index 00000000..bf679d45 --- /dev/null +++ b/go/protocol/cloudwire/testdata/cloud-quic-vectors.json @@ -0,0 +1,252 @@ +{ + "alpn": { + "app": "ftw-cloud-app/1", + "box": "ftw-cloud-box/1" + }, + "transports": [ + "quic", + "websocket" + ], + "varints": [ + { + "value": 0, + "hex": "00" + }, + { + "value": 37, + "hex": "25" + }, + { + "value": 63, + "hex": "3f" + }, + { + "value": 64, + "hex": "4040" + }, + { + "value": 15293, + "hex": "7bbd" + }, + { + "value": 16383, + "hex": "7fff" + }, + { + "value": 16384, + "hex": "80004000" + }, + { + "value": 494878333, + "hex": "9d7f3e7d" + }, + { + "value": 1073741823, + "hex": "bfffffff" + }, + { + "value": 1073741824, + "hex": "c000000040000000" + }, + { + "value": 151288809941952652, + "hex": "c2197c5eff14e88c" + }, + { + "value": 4611686018427387903, + "hex": "ffffffffffffffff" + } + ], + "messages": [ + { + "len": 0, + "prefix": "00" + }, + { + "len": 1, + "prefix": "01" + }, + { + "len": 63, + "prefix": "3f" + }, + { + "len": 64, + "prefix": "4040" + }, + { + "len": 536, + "prefix": "4218" + }, + { + "len": 4096, + "prefix": "5000" + }, + { + "len": 16383, + "prefix": "7fff" + }, + { + "len": 16384, + "prefix": "80004000" + }, + { + "len": 16664, + "prefix": "80004118" + } + ], + "targets": [ + { + "name": "box", + "target": "/v1/box", + "message": "072f76312f626f78" + }, + { + "name": "app", + "target": "/v1/app?home=fjzyr67buzpyncqmjliwgezvde", + "message": "272f76312f6170703f686f6d653d666a7a7972363762757a70796e63716d6a6c697767657a766465" + } + ], + "datagrams": [ + { + "name": "challenge", + "envelope": "a36162582ca2617601656e6f6e6365582011111111111111111111111111111111111111111111111111111111111111116174696368616c6c656e67656373696740", + "datagram": false + }, + { + "name": "hello", + "envelope": "a36162590155ae61760162706b584104cd7a7c09b7c1fed854fcba642c1da25b48df5bf3208c67bcc2158374011e8c84bf6490130896243c527842441c170a64252874fdf719aa61444e329fe6bc02b863656b6d58202222222222222222222222222222222222222222222222222222222222222222637365710164626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0646361707385646c697665626c706670726963657362686266726f73746572656275696c6467302e3134322e3065636c6f636b636e7470656e6f6e6365582011111111111111111111111111111111111111111111111111111111111111116677616c6c4d731b000001a130d7a000676772616e744964504444444444444444444444444444444468626f784e6f6e63655820333333333333333333333333333333333333333333333333333333333333333368757074696d654d7319ea6069726f737465725365710461746568656c6c6f63736967584032b835d4f0c887e458a7af2f52f53368a7c188d85ec502bcec39fd43e6f0fd00a0205eb354a8ea6351acedecf5b3c8c5c663bd2c66d507e282f65794a87f540d", + "datagram": false + }, + { + "name": "hello_ok", + "envelope": "a361625863a463656b6d582022222222222222222222222222222222222222222222222222222222222222226468617665a068626f784e6f6e63655820333333333333333333333333333333333333333333333333333333333333333369726f737465725365710361746868656c6c6f5f6f6b637369675840230af93c21c9b3b977d9c89cfdd637ab53d0be86fee865f5b3bd0f931993476fd439a631ce64cd949df8047b48ce083610b71316cd20cd9ca12c49242c7d283d", + "datagram": false + }, + { + "name": "roster", + "envelope": "a361625901edaa637365710264626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b06474746c531a00015180656772616e74a462696450444444444444444444444444444444446673636f706573826d6674772e6c6976652e726561646f6674772e6576656e74732e7265616468707572706f736573816573657276656a636f6e74726f6c6c657269736f7572636566756c656e6f697365582077777777777777777777777777777777777777777777777777777777777777776677616c6c4d731b000001a130d7a00a676465766963657382a66269646871354c77305a6231636b6579582055555555555555555555555555555555555555555555555555555555555555556374756ef464726f6c65656f776e65726565706f6368016673636f706573856d6674772e6c6976652e72656164706674772e686973746f72792e726561646d6674772e706c616e2e726561646f6674772e6865616c74682e726561646f6674772e6576656e74732e72656164a6626964685a66303978513261636b6579582066666666666666666666666666666666666666666666666666666666666666666374756ef464726f6c65667669657765726565706f6368026673636f706573816d6674772e6c6976652e726561646770656e64696e678068757074696d654d7319ea6a69726f7374657253657104617466726f7374657263736967584096a9ec9cde7b6b2b62bf55ed8d74aeaba0a8d0ded9e5ec32d1f868f758051c146e323ee9aca2c7a009b78190cde63c4b9c882ec29843f07fd15608ed9cf4dd11", + "datagram": false + }, + { + "name": "live full", + "envelope": "a361625901fdaa63736571036461676573a36b41436648743444574c37671903986b4455595757516d6e2d586719e68c6b662d6f484d4471562d6d41190a3c64626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0666669656c6473a76131006132190730613339090561341903b66135190264613619044c62313019083466737263526576016677616c6c4d731b000001a130d7a01467736f7572636573a36b41436648743444574c3767a4646b696e6466647269766572657374617465646c697665686c6173744f6b4d7319e6dc6c7374616c6541667465724d731927106b4455595757516d6e2d5867a4646b696e6466647269766572657374617465676c616767696e67686c6173744f6b4d731903e86c7374616c6541667465724d731975306b662d6f484d4471562d6d41a4646b696e6466647269766572657374617465646c697665686c6173744f6b4d7319e0386c7374616c6541667465724d73193a9868757074696d654d7319ea746c6669656c64536f7572636573a66132816b41436648743444574c37676133816b662d6f484d4471562d6d416134816b662d6f484d4471562d6d416135816b662d6f484d4471562d6d416136836b4455595757516d6e2d58676b662d6f484d4471562d6d416b41436648743444574c3767623130816b4455595757516d6e2d5867716469737061746368426c6f636b65644279806174646c69766563736967584002a8fa0091fbaa932e9ce91bf214e7b1e09f1b17767f49e330bebfdec31c0d7ae738f0a34e3c3f9331a1a8ddf7195152821318c815a61891dc3cd91610516b95", + "datagram": false + }, + { + "name": "live delta", + "envelope": "a3616258a1a863736571046461676573a26b41436648743444574c37671901906b662d6f484d4471562d6d4119083464626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0666669656c6473a7613100613219070361333908f161341903b66135190264613619040b62313019083466737263526576016677616c6c4d731b000001a130d7b19468757074696d654d7319fbf4716469737061746368426c6f636b65644279806174646c6976656373696758400dc02a9f9207e8694654a8e64125db8b165f7302b8b952c1d5f3d90e8d8de7f9d7969ad8006334681b9b20732e24efddbdf46904686dd01092dd2cf7edfa49f6", + "datagram": true + }, + { + "name": "lp", + "envelope": "a361625901ffa7637365710564626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b06677616c6c4d731b000001a130d7b19e68666565646261636b81a9637372636b4455595757516d6e2d5867646b696e64626576656167654d73190384656672657368f566726561736f6e64706c616e6673746174757369666f6c6c6f77696e676761637475616c571908346865766964656e6365686d65617375726564687365766572697479626f6b68757074696d654d7319fbfe6a636f6e74726f6c52657619032c6a6c6f6164706f696e747381b66269646b6a33485a4f524e43574873637372636b4455595757516d6e2d586765626f6f7374f4666d616e75616cf6666e6578745768191ce866706f7765725719083467706c7567676564f56773657373696f6e62733167736f6341676553190708696e657874456e644d731b000001a133030b8069726561647942794d731b000001a133a1bc8069726563757272696e67f56973657373696f6e5768190d4869736f63536f7572636568696e6665727265646a636f6d6d616e646564571908166b6e65787453746172744d731b000001a132952e806b706c616e50656e64696e67f46b736f635065726d696c6c651901ae6b737572706c75734f6e6c79f46c677269644465666572726564f46e7461726765745065726d696c6c651903206f636f6d6d616e646564526561736f6e64706c616e6174626c70637369675840b3754b61b1e615d66e94d1e78bbb04f06d466710276ff5807be02499af3863bb27bd183f773095e145e3415b54ec49991c7f6e4e70e2f2db235cd7d80c379c79", + "datagram": false + }, + { + "name": "prices", + "envelope": "a36162588caa637365710664626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b06474304d731b000001a130d7a000647a6f6e656353453366736c6f744d731a000dbba06677616c6c4d731b000001a130d7b1a86863757272656e63796353454b68757074696d654d7319fc086973706f744d696e6f7284182a1828f6226a746f74616c4d696e6f728418761873f61847617466707269636573637369675840e53619e522b5665e9599d2e04c4a62a1040f797a5742f39fde5b56b250b60c5426e008ae623bcaf0def871c596b371763f589cf280cd7dee1381064fdca2c7e3", + "datagram": false + }, + { + "name": "hb", + "envelope": "a361625880a963736571076461676573a26b41436648743444574c37671905786b662d6f484d4471562d6d41190c1c64626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b065636c6f636b636e7470656c705365710566737263526576016677616c6c4d731b000001a130d7c33c676c6976655365710468757074696d654d731a00010d9c6174626862637369675840e82b54b463aaf1c315cb0a049e304609482ecf9b3f6a1f5ea9c727d618d9f69ff7c256f1ac88cef858d5168ccec7528951f987e360891aacb00dd03370403fd8", + "datagram": true + }, + { + "name": "watch", + "envelope": "a3616245a162687a0161746577617463686373696740", + "datagram": false + }, + { + "name": "resync", + "envelope": "a361624ca16477686174816466756c6c617466726573796e636373696740", + "datagram": false + }, + { + "name": "bye", + "envelope": "a36162584da6637365710863776879677265766f6b656464626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0657075726765f56677616c6c4d731b000001a130d7c71068757074696d654d731a0001117061746362796563736967584019b5ea48e5c3b113712f5815840c2339d654a3438a516dff0b13c444c604d94f5f25b72689a8dd9e02de38740f3b420836d0e40e12e8d236ac21e8685e781b04", + "datagram": false + }, + { + "name": "purged", + "envelope": "a361625832a36461744d731b000001a130daad40647768617464686f6d65676772616e7449645044444444444444444444444444444444617466707572676564637369675840910f5b966d04ed95f1e5d2e33fd542391976b4013c82b9beb0f5c42c365dd1f35c5f71d5efe89a44a4fb67cd35f93f85087549e8e4283f6c20e794d92d12cc32", + "datagram": false + }, + { + "name": "event connected", + "envelope": "a3616258a6ab626c706b6a33485a4f524e435748736274731b000001a130d7c9686373657109637372636b4455595757516d6e2d58676461726773a064626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0646b696e64726368617267696e672e636f6e6e65637465646677616c6c4d731b000001a130d7caf8676576656e7449646c51325a724d6a5230647a566868736576657269747964696e666f68757074696d654d731a000115586174656576656e74637369675840490916f3211c463ee9de60249e27a7a44e405a175679f57e238cb910fa9c05891209946aa986336d246eed44413583089e7c7777859cd13172eb8750a684dc23", + "datagram": false + }, + { + "name": "event interrupted", + "envelope": "a3616258abab626c706b6a33485a4f524e435748736274731b000001a130d7cdb4637365710a637372636b4455595757516d6e2d58676461726773a064626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0646b696e64746368617267696e672e696e7465727275707465646677616c6c4d731b000001a130d7cee0676576656e7449646c61573530636e42305a444178687365766572697479677761726e696e6768757074696d654d731a000119406174656576656e74637369675840ae81be2140544a903e180cf33bf97a43d075ead5d1547226ce9df7a3e0b40d18c3e3727c607138fcf68806b7a5068d7cf0ec12e3536f73229ea3ff40d1a31f25", + "datagram": false + }, + { + "name": "event session complete", + "envelope": "a3616258b7ab626c706b6a33485a4f524e435748736274731b000001a130d7d200637365710b637372636b4455595757516d6e2d58676461726773a1636b77686431382e3464626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0646b696e6478196368617267696e672e73657373696f6e5f636f6d706c6574656677616c6c4d731b000001a130d7d2c8676576656e7449646c5a4739755a5441774d44417868736576657269747964696e666f68757074696d654d731a00011d286174656576656e74637369675840e75a581948135cbd2b54eec888fa0396f57b9159de07881ed78cc51b271cad39f1e81431d1125b51040c7794b109ff106702004dccb0285a61a879d29afb04bb", + "datagram": false + }, + { + "name": "roster tunnel only", + "envelope": "a36162590190aa637365710c64626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b06474746c531a00015180656772616e74a4626964406673636f7065738068707572706f736573806a636f6e74726f6c6c657269736f7572636566756c656e6f697365582077777777777777777777777777777777777777777777777777777777777777776677616c6c4d731b000001a130d7d6b0676465766963657382a66269646871354c77305a6231636b6579582055555555555555555555555555555555555555555555555555555555555555556374756ef564726f6c65656f776e65726565706f6368016673636f70657380a6626964685a66303978513261636b6579582066666666666666666666666666666666666666666666666666666666666666666374756ef564726f6c65667669657765726565706f6368026673636f706573806770656e64696e6781a264686173685820b4e707c0ac2241df0e0325b42b951b3c71bbd5099bc3c6f2894a48ee23fa35636474746c5319025868757074696d654d731a0001211069726f7374657253657105617466726f73746572637369675840729ecdf8ab1a41194cc3da99bf86d5e3593f9bb9e417e70512905e0a6c664f2bfa390316d794493370c9beb8da904e8a01f11af360503b1705cf463834502ac7", + "datagram": false + }, + { + "name": "roster revoking", + "envelope": "a3616259011daa637365710d64626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b06474746c531a00015180656772616e74a462696450444444444444444444444444444444446673636f7065738068707572706f736573806a636f6e74726f6c6c657269736f7572636566756c656e6f697365582077777777777777777777777777777777777777777777777777777777777777776677616c6c4d731b000001a130d7da98676465766963657381a66269646871354c77305a6231636b6579582055555555555555555555555555555555555555555555555555555555555555556374756ef564726f6c65656f776e65726565706f6368016673636f706573806770656e64696e678068757074696d654d731a000124f869726f7374657253657106617466726f73746572637369675840c96e88f759f66bb580111e4d63cb45bb4b160d9cd17e211a394df3901b07c2e5317aab96aad398b1a1aca1827a68ba3fe7632ba3b12e886a9968d2c617b70d3c", + "datagram": false + }, + { + "name": "purged copy", + "envelope": "a361625832a36461744d731b000001a130dc33e0647768617464636f7079676772616e744964504444444444444444444444444444444461746670757267656463736967584025e59e26aa33f7aaa1f4e463b721ea04c125f4cf1cec9f547dc49d07143e786dc35c05cd4320753d703759116c312e77f386579e61f1649763435fcbac5079c1", + "datagram": false + }, + { + "name": "tun down", + "envelope": "a361625887a36164587099999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999636465766871354c77305a6231637369640761746374756e6373696740", + "datagram": false + }, + { + "name": "tun up", + "envelope": "a361625847a3616458309a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a636465766871354c77305a6231637369640761746374756e6373696740", + "datagram": false + }, + { + "name": "tun end", + "envelope": "a3616256a3616440636465766871354c77305a6231637369640761746374756e6373696740", + "datagram": false + }, + { + "name": "roster with an agent", + "envelope": "a36162590256aa637365710e64626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b06474746c531a00015180656772616e74a4626964406673636f7065738068707572706f736573806a636f6e74726f6c6c657269736f7572636566756c656e6f697365582077777777777777777777777777777777777777777777777777777777777777776677616c6c4d731b000001a130d7de80676465766963657382a66269646871354c77305a6231636b6579582055555555555555555555555555555555555555555555555555555555555555556374756ef564726f6c65656f776e65726565706f6368016673636f70657380a7626964686f4b47696f36536c636b65795820a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a06374756ef464726f6c6567737570706f7274656167656e74a5636f7073816e6c6f6164706f696e742e686f6c646463617073a3646d617857191ce865686f6c6453190384667065724d696e066673636f70657383716674772e64697370617463682e686f6c646d6674772e6c6976652e726561646d6674772e6c6f67732e72656164697072696e636970616ca366636c69656e7471536f7572636566756c20737570706f7274667469636b6574674654572d31323368636c69656e744964783268747470733a2f2f636c6f75642e6674772e656e657267792f636c69656e74732f736f7572636566756c2d737570706f72746b6578706972657341744d731b000001a17720ca006565706f6368016673636f706573806770656e64696e678068757074696d654d731a000128e069726f7374657253657107617466726f73746572637369675840f2f15bfcafd92e077fb41218fb5eca3735d45484cab4074541f0908b8e6a1107a31f8bf3caf605405bd7cbd02963ff4466a40de915aa9d9af96166e5fc5965a7", + "datagram": false + }, + { + "name": "lp car asking", + "envelope": "a361625901bda7637365710f64626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b06677616c6c4d731b000001a130d7de8068666565646261636b8068757074696d654d731a000128e06a636f6e74726f6c52657619032d6a6c6f6164706f696e747381b8186269646b6a33485a4f524e43574873637372636b4455595757516d6e2d586765626f6f7374f46663686f6f7365826b464e4b69626974666555306b446d34516733447a336263666d616e75616cf6666e6578745768f666706f776572570067706c7567676564f56773657373696f6e703764316630633261396233653466353067736f6341676553f6696e657874456e644d73f669726561647942794d73f669726563757272696e67f46973657373696f6e57680069736f63536f7572636567756e6b6e6f776e6a636f6d6d616e64656457f66b6e65787453746172744d73f66b706c616e50656e64696e67f46b736f635065726d696c6c65f66b737572706c75734f6e6c79f56c677269644465666572726564f46e7461726765745065726d696c6c65f66f6361724964656e74696669656442796661736b696e676f636f6d6d616e646564526561736f6e7070765f737572706c75735f70617573656174626c70637369675840fa1b746e25aa7f02deb3fbc543e233315f26f31e0325f4921fb0d5a9ef01e249c129184c88d23396bcc1dc3193bdf78fa2f5805760dd9bea3d05137343bbe420", + "datagram": false + }, + { + "name": "car", + "envelope": "a36162590177a5637365711064626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0646361727382a96269646b464e4b696269746665553064676f616ca46464617973187f6a726561647942794d696e1901a46d66696e69736841744c696d6974f46e7461726765745065726d696c6c6519032067736f634167655319021c69736f63536f757263656776656869636c656b736f63457374696d617465f46b736f635065726d696c6c65188c6b737572706c75734f6e6c79f46d656e6572677953696e6365576800736368617267654c696d69745065726d696c6c65190320a96269646b446d34516733447a33626364676f616cf667736f634167655319778869736f63536f757263656776656869636c656b736f63457374696d617465f56b736f635065726d696c6c6519026c6b737572706c75734f6e6c79f56d656e6572677953696e6365576800736368617267654c696d69745065726d696c6c65f66677616c6c4d731b000001a130d7de8a68757074696d654d731a000128ea6174636361726373696758407ee438570fc0f1072201594add376235e687cb87168f882c88e94e1e087b595be184505d295ecc901cb8c8a424a8fa3df0b65b28b1c2352a7ecfc4cb0b53ae01", + "datagram": false + }, + { + "name": "event car which", + "envelope": "a3616258d4ac626c706b6a33485a4f524e435748736274731b000001a130d7de306373657111637372636b4455595757516d6e2d58676461726773a16773657373696f6e703764316630633261396233653466353064626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b06463617273826b464e4b69626974666555306b446d34516733447a336263646b696e64696361722e77686963686677616c6c4d731b000001a130d7de94676576656e7449646c643268705932686a5958497868736576657269747964696e666f68757074696d654d731a000128f46174656576656e7463736967584070a21cc4e288f019d1ffeb59456f8555b07c9f450f6182e6d2cddb5dcca37c31a4a3465c5080c429d6f01063c571edf44e4012dd0198b6bdc1fdfcc6ca3e4b88", + "datagram": false + }, + { + "name": "lp car tapped", + "envelope": "a361625901ada7637365711264626f6f7450b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b06677616c6c4d731b000001a130d7f20868666565646261636b8068757074696d654d731a00013c686a636f6e74726f6c52657619032e6a6c6f6164706f696e747381b8186269646b6a33485a4f524e43574873637372636b4455595757516d6e2d586765626f6f7374f46563617249646b464e4b6962697466655530666d616e75616cf6666e6578745768f666706f776572570067706c7567676564f56773657373696f6e703764316630633261396233653466353067736f6341676553190221696e657874456e644d73f669726561647942794d731b000001a133a1bc8069726563757272696e67f56973657373696f6e57680069736f63536f757263656776656869636c656a636f6d6d616e64656457f66b6e65787453746172744d73f66b706c616e50656e64696e67f46b736f635065726d696c6c65188c6b737572706c75734f6e6c79f46c677269644465666572726564f46e7461726765745065726d696c6c651903206f6361724964656e7469666965644279637461706f636f6d6d616e646564526561736f6e64706c616e6174626c7063736967584089b4a6faa2af341cdc4da9ad0247aa2135f40e1aec94554cf1855392d22412cb2191b2a594084e4490faada200f68abbdcdd80b7b0bc3e7fb4ff65e5572f3fcf", + "datagram": false + } + ], + "notes": { + "datagram": "hb, and live without the sources key, may go as one QUIC DATAGRAM; nothing else", + "message": "one WebSocket binary message: its length as a QUIC varint (RFC 9000 §16), then its bytes", + "target": "the first message on the control stream (client bidirectional stream 0), ASCII" + } +} diff --git a/go/protocol/cloudwire/view.go b/go/protocol/cloudwire/view.go index 218a4c10..7c394262 100644 --- a/go/protocol/cloudwire/view.go +++ b/go/protocol/cloudwire/view.go @@ -142,6 +142,13 @@ func (v *View) Accept(e Envelope, arrivalMs int64) error { } v.reset(h.Boot) } else if h.Seq <= v.seq[e.T] { + // Over QUIC a live without its table can overtake the one that + // carries it. The late one still brings a newer table, and nothing + // else. + if live != nil && live.Sources != nil && live.SrcRev > v.tableRev { + v.table, v.tableRev, v.fieldSrc = live.Sources, live.SrcRev, live.FieldSources + return nil + } return ErrNotNewer } if lead { diff --git a/go/protocol/cloudwire/view_test.go b/go/protocol/cloudwire/view_test.go index 4495cb54..c7bafd9d 100644 --- a/go/protocol/cloudwire/view_test.go +++ b/go/protocol/cloudwire/view_test.go @@ -371,3 +371,28 @@ func TestAnUnknownStateIsNeverLive(t *testing.T) { t.Fatalf("an unknown state graded as %s, field %s", g.Sources["m"], g.Fields["2"]) } } + +// 17. A late table still counts. +func TestFreshnessALateTableStillCounts(t *testing.T) { + box := newFakeBox(t, 1) + v := box.view() + t0 := int64(2_000_000) + accept(t, v, box.fullLive(t0, 1_000, protocol.SourceLive, 1), t0+transit) + withTable := box.fullLive(t0+3_000, 1_000, protocol.SourceLive, 2) // on the stream, late + overtaking := box.seal(MsgLive, Live{ + Header: box.header(t0 + 4_000), Fields: map[string]int64{"2": 1200, "1": 0}, + SrcRev: 2, Ages: map[string]int64{"m": 1_000}, + }) + accept(t, v, overtaking, t0+4_000+transit) + wantSource(t, v, t0+4_200, protocol.SourceStale) + + accept(t, v, withTable, t0+4_300) + wantSource(t, v, t0+4_400, protocol.SourceLive) + if got := v.live.Fields["2"]; got != 1200 { + t.Fatalf("the late live moved the values back: field 2 is %d", got) + } + // The same table again, or an older one, brings nothing. + if err := v.Accept(withTable, t0+4_500); !errors.Is(err, ErrNotNewer) { + t.Fatalf("the late table twice: err = %v, want ErrNotNewer", err) + } +} diff --git a/go/protocol/contract_gen.go b/go/protocol/contract_gen.go index 10f80a1d..6f5d07a8 100644 --- a/go/protocol/contract_gen.go +++ b/go/protocol/contract_gen.go @@ -246,6 +246,18 @@ var CloudPurposes = []string{ CloudPurposeStats, } +// What a session with FTW Cloud travels over, under its carrier. +const ( + CloudTransportQuic = "quic" + CloudTransportWebsocket = "websocket" +) + +// CloudTransports lists them in the registry's order. +var CloudTransports = []string{ + CloudTransportQuic, + CloudTransportWebsocket, +} + // RegistryOps is the app's copy of the operations a cmd frame may name, // each mapped to the scope its grant must carry. defaultOps() in command.go // remains the authority on what this box accepts — this exists only so