From c1352f26cb78df3efa0e714fbd4865c6c05c1095 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 12 Oct 2026 21:00:19 +0200 Subject: [PATCH] feat: ftwcloudd holds a signed copy of each home in memory The first version of FTW Cloud (srcfl/ftw ADR 0008, step C2): - /v1/box takes one uplink per home. The service checks that the box signed hello over this TLS session's exporter value, answers with a hello_ok signed by the service key, and accepts only messages signed by that box's key. The home id is a hash of the key. - store.For(principal) is the only way to a home. Each home's newest message of each class is kept in memory as the exact bytes the box sent; readings go an hour after the box does and never reach disk. - The registry (SQLite, secure_delete) holds the home id, the box key, the signed roster and tombstones. bye with purge deletes the home and returns a signed receipt, again if the first one was lost. - The downlink is a closed set of five types; what a box can sign is fuzzed. /source serves the running source. A test checks that no route reads a home. - /v1/app (C3) and /v1/now (C5) are marked and answer 501. The wire format comes from github.com/srcfl/ftw/go/protocol, pinned to a commit; CI checks that nothing else from srcfl/ftw is imported. Co-Authored-By: Claude Opus 5.5 --- .dockerignore | 4 + .github/workflows/ci.yml | 29 ++ .gitignore | 4 + CLAUDE.md | 47 ++++ CONTRIBUTING.md | 39 +++ Dockerfile | 24 ++ README.md | 107 ++++++++ cmd/ftwcloudd/harden_linux.go | 16 ++ cmd/ftwcloudd/harden_other.go | 6 + cmd/ftwcloudd/main.go | 184 +++++++++++++ deploy/README.md | 39 +++ deploy/compose.yml | 50 ++++ go.mod | 26 ++ go.sum | 64 +++++ internal/registry/registry.go | 188 +++++++++++++ internal/registry/registry_test.go | 72 +++++ internal/server/inspect_dev.go | 29 ++ internal/server/server.go | 141 ++++++++++ internal/server/server_test.go | 83 ++++++ internal/store/inspect_dev.go | 26 ++ internal/store/store.go | 285 ++++++++++++++++++++ internal/store/store_test.go | 167 ++++++++++++ internal/uplink/downlink.go | 68 +++++ internal/uplink/uplink.go | 414 +++++++++++++++++++++++++++++ internal/uplink/uplink_test.go | 382 ++++++++++++++++++++++++++ scripts/check-imports.sh | 19 ++ 26 files changed, 2513 insertions(+) create mode 100644 .dockerignore create mode 100644 .github/workflows/ci.yml create mode 100644 .gitignore create mode 100644 CLAUDE.md create mode 100644 CONTRIBUTING.md create mode 100644 Dockerfile create mode 100644 README.md create mode 100644 cmd/ftwcloudd/harden_linux.go create mode 100644 cmd/ftwcloudd/harden_other.go create mode 100644 cmd/ftwcloudd/main.go create mode 100644 deploy/README.md create mode 100644 deploy/compose.yml create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/registry/registry.go create mode 100644 internal/registry/registry_test.go create mode 100644 internal/server/inspect_dev.go create mode 100644 internal/server/server.go create mode 100644 internal/server/server_test.go create mode 100644 internal/store/inspect_dev.go create mode 100644 internal/store/store.go create mode 100644 internal/store/store_test.go create mode 100644 internal/uplink/downlink.go create mode 100644 internal/uplink/uplink.go create mode 100644 internal/uplink/uplink_test.go create mode 100755 scripts/check-imports.sh diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..09339a1 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,4 @@ +.git +state +*.key +*.pem diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..932108b --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,29 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v5 + - uses: actions/setup-go@v6 + with: + go-version: '1.26' + - run: go vet ./... + - run: go test -race -count=1 ./... + # The local inspection route exists only under this tag; the suite + # must pass with it too, and TestNoRouteReadsAHome must pass without. + - run: go test -count=1 -tags ftwclouddev ./... + - name: FTW Cloud imports only the wire format from srcfl/ftw + run: bash scripts/check-imports.sh + - name: What a box can send never panics the service + run: go test -run XXX -fuzz FuzzBoxMessage -fuzztime 30s ./internal/uplink + - name: Builds for the host + run: GOOS=linux GOARCH=arm64 CGO_ENABLED=0 go build -o /dev/null ./cmd/ftwcloudd diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..facee62 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +/ftwcloudd +/state/ +*.key +*.pem diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..5f88b21 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,47 @@ +# FTW Cloud guide + +`ftwcloudd` keeps a signed, readable copy of each home that turned FTW Cloud +on, and serves it to that home's phones. Read [README.md](README.md) first, +then the normative +[FTW Cloud protocol](https://github.com/srcfl/ftw/blob/master/docs/protocol-cloud.md) +and [ADR 0008](https://github.com/srcfl/ftw/blob/master/docs/adr/0008-ftw-cloud.md) +in `srcfl/ftw`. Fredrik owns FTW's direction; Sourceful develops and +maintains it. + +## Invariants + +- **FTW Cloud reads and never controls.** It sends a box only `challenge`, + `hello_ok`, `watch`, `resync` and `purged` (`internal/uplink/downlink.go`). + Commands go from the app to the box over Noise, never through here. +- **One way to a home:** `store.For(principal)`. No listing, no second lookup, + no principal for staff. A new route needs a row in `server.Routes` and must + pass `TestNoRouteReadsAHome`. +- **The home id comes from the key that signed hello** on this connection, + never from a body field. Every later box message must verify with that key. +- **No readings on disk.** Memory holds readings; SQLite holds only the home + id, the box key, the signed roster and tombstones. Nothing logs addresses. +- **TLS ends in `ftwcloudd`.** The handshake is bound to the TLS session; a + terminating proxy breaks it on purpose. A TCP load balancer is fine. +- **Forward the box's bytes unchanged.** The service cannot sign for a box, + so it cannot make data look fresher; keep it that way. +- **Import only `go/protocol` from `srcfl/ftw`.** `scripts/check-imports.sh` + fails on anything else. + +## Work + +- Check open PRs on the files you touch; an open PR has right of way. +- A wire change is a pair: `srcfl/ftw` (`go/protocol` and its golden vectors) + first, then this repository, moving the pin in `go.mod`. Say so in both PRs. +- Keep the service small. The budget is about 3,000 lines of Go; tiles go + first if it is overrun. +- Nothing is deployed from here without the owner. `deploy/` describes the + intended host; it does not run itself. +- Keep task breakdowns and agent notes out of the repository. + +## Check + +```bash +go vet ./... && go test -race ./... +go test -tags ftwclouddev ./... +bash scripts/check-imports.sh +``` diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..f94c391 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,39 @@ +# Contributing to FTW Cloud + +Sourceful maintains this repository. Fredrik owns FTW's direction in +[the shared vision](https://github.com/srcfl/ftw/blob/master/VISION.md). +External PRs are welcome, preferably based on an +[issue](https://github.com/srcfl/ftw-cloud/issues). Small fixes need no +separate issue. + +This service holds readable household data, so a change to what it stores, +logs, forwards or exposes needs a reason in the PR and a test that shows the +limit holds. Broad ideas may start as short Markdown PRs; mark proposed +behaviour as proposed. A wire change starts in `srcfl/ftw` (`go/protocol`) +and is paired with a PR here. Keep task breakdowns and agent scratch notes +out of the repository. + +Work is agentic first: give people and agents a clear problem, scope, +reproducible test steps and actual results. Check agent output before +submitting it. Sourceful reviews and maintains changes. A contribution grants +no merge, release or deployment authority. See +[FTW's contribution guide](https://github.com/srcfl/ftw/blob/master/CONTRIBUTING.md). + +## License of contributions + +This repository is licensed under the GNU Affero General Public License, +version 3 only (AGPL-3.0-only); see [LICENSE](LICENSE). Contributions use the +same terms. Preserve third-party licenses and notices. + +## Developer Certificate of Origin (DCO) + +We use the [Developer Certificate of Origin 1.1](https://developercertificate.org/) +instead of a CLA. Sign off every commit with your real name and email: + +```bash +git commit -s -m "fix(uplink): close on a frame over the limit" +``` + +The sign-off confirms that you wrote the contribution or have the right to +submit it under these terms. A sign-off does not assign copyright or grant +commercial relicensing rights. diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..144f538 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,24 @@ +# ftwcloudd, built from this repository. +# +# The image carries the source it was built from, served at /source, because +# AGPL-3.0 asks that everyone who uses the service over a network can get the +# code running it. + +FROM golang:1.26-alpine AS build +WORKDIR /src +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +# No cgo: the SQLite driver is pure Go, so the binary runs on a static base. +RUN CGO_ENABLED=0 go build -trimpath -ldflags "-s -w" -o /out/ftwcloudd ./cmd/ftwcloudd +RUN mkdir -p /out/share && tar -czf /out/share/source.tar.gz \ + --exclude=./.git --exclude=./state . \ + && mkdir -p /out/state + +FROM gcr.io/distroless/static-debian12:nonroot +COPY --from=build /out/ftwcloudd /usr/local/bin/ftwcloudd +COPY --from=build /out/share/source.tar.gz /usr/share/ftw-cloud/source.tar.gz +COPY --from=build --chown=nonroot:nonroot /out/state /state +USER nonroot +EXPOSE 443 +ENTRYPOINT ["/usr/local/bin/ftwcloudd"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..cbd5fca --- /dev/null +++ b/README.md @@ -0,0 +1,107 @@ +# FTW Cloud + +FTW Cloud keeps a signed, readable copy of each home that turns it on, and +serves it to that home's phones. It reads and never controls. The box opens +the connection, picks and signs every message, and can revoke at any time. + +This is `ftwcloudd`, the service. The decision is +[ADR 0008](https://github.com/srcfl/ftw/blob/master/docs/adr/0008-ftw-cloud.md) +and the wire format is the normative +[FTW Cloud protocol](https://github.com/srcfl/ftw/blob/master/docs/protocol-cloud.md), +both in `srcfl/ftw`. Both are proposed; nothing here is deployed. + +## What works today (C2) + +- A box with FTW Cloud on keeps one WebSocket at `/v1/box`. The service + checks that the box signed `hello` over this TLS session's exporter value, + answers with a `hello_ok` signed by the service key, and then accepts only + messages signed by that box's key. +- Each home's newest message of each class lives in memory, as the exact bytes + the box sent. Readings are dropped an hour after the box goes quiet and are + never written to disk. +- The registry (SQLite) holds the home id, the box key and the signed roster. + A box that turns FTW Cloud off gets its home deleted and a signed receipt. +- `/source` serves the source of the running build, as AGPL-3.0 asks. + +Not built yet, with a marked place in the code: + +| Step | What | Where | +|---|---|---| +| C3 | The app's Noise session, `/v1/app` | `internal/server`, `store.For` with a `Device` principal | +| C4 | Alerts over APNs, `box.unreachable` | not started | +| C5 | Widget token reads, `/v1/now`, tiles | `internal/server` | +| C7–C8 | Statistics | not started | + +## How it holds a home + +``` +box ──wss, TLS 1.3, signed frames──▶ /v1/box ──▶ store (memory) ◀── store.For(principal) + └────▶ registry (SQLite: key, roster, tombstones) +``` + +- **The home id is a hash of the box's key.** It comes from the key that + signed `hello`, never from a field anyone sends. A second box cannot claim + a home that is not its own. +- **One way in.** `store.For(principal)` is the only access to a home: a + `Box` principal may write its own home, and a `Device` principal (C3) may + read the classes its role and the owner's grant both allow. There is no + listing and no principal for staff; `TestNoRouteReadsAHome` checks every + route. +- **The service sends five things.** `challenge`, `hello_ok`, `watch`, + `resync` and `purged` are the whole downlink, and none asks the box to act. + The box closes the uplink on anything else. +- **No TLS terminator in front.** Box and service bind the handshake to the + TLS session, so `ftwcloudd` ends TLS itself. A TCP load balancer is fine. +- **Root on the host can read memory.** The privacy notice says so. + +## Run it locally + +```bash +go build -tags ftwclouddev -o ftwcloudd ./cmd/ftwcloudd +./ftwcloudd keygen -out service.key # prints the public key +./ftwcloudd -listen 127.0.0.1:18443 -state ./state -service-key service.key \ + -tls-cert leaf.pem -tls-key leaf.key +``` + +Build Core from `srcfl/ftw` against it. A release build names no service, so +a local build sets the endpoint, the service's public key and a root for the +local certificate at link time: + +```bash +go build -ldflags "\ + -X github.com/srcfl/ftw/go/internal/cloudlink.Endpoint=wss://127.0.0.1:18443/v1/box \ + -X github.com/srcfl/ftw/go/internal/cloudlink.ServiceKeys= \ + -X github.com/srcfl/ftw/go/internal/cloudlink.DevRootCA=$PWD/ca.pem" ./cmd/ftw +``` + +Then turn FTW Cloud on from the box itself: + +```bash +curl -X POST http://127.0.0.1:8080/api/cloud/grant \ + -d '{"scopes":["ftw.live.read"],"policy":"local-test"}' +curl http://127.0.0.1:8080/api/cloud/preview +``` + +`-tags ftwclouddev` adds `/dev/held?home=` on loopback, which shows what +the service holds so a test can compare it with the preview. Release builds +do not have it. + +## Develop + +```bash +go vet ./... && go test -race ./... +go test -run XXX -fuzz FuzzBoxMessage -fuzztime 30s ./internal/uplink +bash scripts/check-imports.sh +``` + +The wire types come from `github.com/srcfl/ftw/go/protocol`, Apache-2.0, +pinned to a commit in `go.mod`. A wire change is a pair of pull requests: +first in `srcfl/ftw` (`go/protocol` and its golden vectors), then here, moving +the pin. The service may import nothing else from `srcfl/ftw`; CI checks that +with `go list -deps`, so no control code can reach this build. + +## License + +`ftwcloudd` is licensed under the GNU Affero General Public License, version 3 +only (AGPL-3.0-only). See [LICENSE](LICENSE) and +[CONTRIBUTING.md](CONTRIBUTING.md). diff --git a/cmd/ftwcloudd/harden_linux.go b/cmd/ftwcloudd/harden_linux.go new file mode 100644 index 0000000..8cb337d --- /dev/null +++ b/cmd/ftwcloudd/harden_linux.go @@ -0,0 +1,16 @@ +package main + +import ( + "log/slog" + + "golang.org/x/sys/unix" +) + +// harden makes the process not dumpable: no core file, and no ptrace attach +// from another process of the same user. Household readings live in this +// process's memory and nowhere else. +func harden() { + if err := unix.Prctl(unix.PR_SET_DUMPABLE, 0, 0, 0, 0); err != nil { + slog.Warn("could not turn off core dumps", "err", err) + } +} diff --git a/cmd/ftwcloudd/harden_other.go b/cmd/ftwcloudd/harden_other.go new file mode 100644 index 0000000..bb16421 --- /dev/null +++ b/cmd/ftwcloudd/harden_other.go @@ -0,0 +1,6 @@ +//go:build !linux + +package main + +// harden has nothing to set outside Linux, where the service runs. +func harden() {} diff --git a/cmd/ftwcloudd/main.go b/cmd/ftwcloudd/main.go new file mode 100644 index 0000000..e758bd8 --- /dev/null +++ b/cmd/ftwcloudd/main.go @@ -0,0 +1,184 @@ +// Command ftwcloudd is FTW Cloud: it keeps a signed, readable copy of each +// home that turned it on, in memory, and serves it to that home's phones. It +// reads and never controls. See README.md. +// +// ftwcloudd keygen -out service.key make the service key; prints its public half +// ftwcloudd [flags] serve +package main + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "encoding/base64" + "encoding/pem" + "errors" + "flag" + "fmt" + "io" + "log" + "log/slog" + "net/http" + "os" + "os/signal" + "path/filepath" + "syscall" + "time" + + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/server" + "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw-cloud/internal/uplink" + "github.com/srcfl/ftw/go/protocol/cloudwire" + "golang.org/x/crypto/acme/autocert" +) + +func main() { + if len(os.Args) > 1 && os.Args[1] == "keygen" { + if err := keygen(os.Args[2:]); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + return + } + if err := serve(os.Args[1:]); err != nil { + slog.Error("ftwcloudd stopped", "err", err) + os.Exit(1) + } +} + +func serve(args []string) error { + fs := flag.NewFlagSet("ftwcloudd", flag.ExitOnError) + listen := fs.String("listen", ":443", "address to listen on") + domain := fs.String("domain", "cloud.ftw.energy", "name to get a certificate for") + stateDir := fs.String("state", "/state", "directory for the registry and certificates") + keyPath := fs.String("service-key", "/run/secrets/ftw-cloud-service.key", "service key, PEM") + certFile := fs.String("tls-cert", "", "certificate file, instead of ACME (local tests)") + keyFile := fs.String("tls-key", "", "certificate key file, with -tls-cert") + email := fs.String("acme-email", "", "contact for the certificate authority") + _ = fs.Parse(args) + + harden() + log := slog.New(slog.NewTextHandler(os.Stderr, nil)) + key, err := loadServiceKey(*keyPath) + if err != nil { + return err + } + reg, err := registry.Open(filepath.Join(*stateDir, "registry.db")) + if err != nil { + return err + } + defer reg.Close() + + st := store.New(nil) + srv := &server.Server{Store: st, Registry: reg, Uplink: uplink.New(st, reg, key, log), Log: log} + + // TLS ends here and nowhere else: each box binds its handshake to this + // TLS session, so a terminating proxy in front would break it. A plain + // TCP load balancer is fine. + tlsConfig := &tls.Config{MinVersion: tls.VersionTLS13, NextProtos: []string{"http/1.1"}} + if *certFile != "" { + cert, err := tls.LoadX509KeyPair(*certFile, *keyFile) + if err != nil { + return err + } + tlsConfig.Certificates = []tls.Certificate{cert} + } else { + m := &autocert.Manager{ + Prompt: autocert.AcceptTOS, + HostPolicy: autocert.HostWhitelist(*domain), + Cache: autocert.DirCache(filepath.Join(*stateDir, "acme")), + Email: *email, + } + tlsConfig.GetCertificate = m.GetCertificate + tlsConfig.NextProtos = append(tlsConfig.NextProtos, "acme-tls/1") + } + + httpSrv := &http.Server{ + Addr: *listen, + Handler: srv.Handler(), + TLSConfig: tlsConfig, + ReadHeaderTimeout: 10 * time.Second, + // Go's server logs "TLS handshake error from
". An address + // beside a time is what links a household to its traffic, so the + // server's own log goes nowhere. + ErrorLog: log0(), + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + go srv.Upkeep(ctx, time.Minute) + go func() { + <-ctx.Done() + shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = httpSrv.Shutdown(shutdown) + }() + + pub, _ := cloudwire.PublicKeyBytesOf(&key.PublicKey) + log.Info("ftwcloudd listening", "addr", *listen, "service_key", base64.RawURLEncoding.EncodeToString(pub)) + if err := httpSrv.ListenAndServeTLS("", ""); err != nil && !errors.Is(err, http.ErrServerClosed) { + return err + } + return nil +} + +func log0() *log.Logger { return log.New(io.Discard, "", 0) } + +func loadServiceKey(path string) (*ecdsa.PrivateKey, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("service key: %w", err) + } + block, _ := pem.Decode(raw) + if block == nil || block.Type != "EC PRIVATE KEY" { + return nil, fmt.Errorf("service key: %s is not an EC private key", path) + } + key, err := x509.ParseECPrivateKey(block.Bytes) + if err != nil { + return nil, fmt.Errorf("service key: %w", err) + } + if key.Curve != elliptic.P256() { + return nil, fmt.Errorf("service key: %s is not P-256", path) + } + return key, nil +} + +// keygen writes a new service key and prints its public half, which Core +// carries in cloudlink.ServiceKeys. +func keygen(args []string) error { + fs := flag.NewFlagSet("keygen", flag.ExitOnError) + out := fs.String("out", "", "file to write the key to") + _ = fs.Parse(args) + if *out == "" { + return errors.New("keygen: -out is required") + } + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return err + } + der, err := x509.MarshalECPrivateKey(key) + if err != nil { + return err + } + f, err := os.OpenFile(*out, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) + if err != nil { + return err + } + if err := pem.Encode(f, &pem.Block{Type: "EC PRIVATE KEY", Bytes: der}); err != nil { + _ = f.Close() + return err + } + if err := f.Close(); err != nil { + return err + } + pub, err := cloudwire.PublicKeyBytesOf(&key.PublicKey) + if err != nil { + return err + } + fmt.Println(base64.RawURLEncoding.EncodeToString(pub)) + return nil +} diff --git a/deploy/README.md b/deploy/README.md new file mode 100644 index 0000000..97fe1ba --- /dev/null +++ b/deploy/README.md @@ -0,0 +1,39 @@ +# Running FTW Cloud + +Proposed, not deployed. ADR 0008 in `srcfl/ftw` chose one `t4g.small` in AWS +eu-north-1 (Stockholm), in an AWS account of its own, apart from the relay and +escrow. The account is not decided yet; nothing here creates anything. + +## Shape + +- One instance, one container (`compose.yml`), no standby. A rebuilt host + starts empty and boxes fill it again when they reconnect. +- `ftwcloudd` ends TLS 1.3 itself and gets its certificate by ACME + (TLS-ALPN-01 on 443). **No TLS-terminating proxy in front**: box and service + bind the handshake to the TLS session, so Caddy, an ALB or an orange + Cloudflare cloud would break every box's connection, on purpose. An NLB in + TCP mode is fine. DNS for `cloud.ftw.energy` is a grey-cloud A record. +- The service key comes from Secrets Manager into `/run/ftw-cloud/service.key` + (tmpfs) at boot. Its public half is compiled into Core + (`cloudlink.ServiceKeys`); Core carries up to two, so it can rotate across a + release. +- The volume holds the SQLite registry (home ids, box keys, signed rosters, + tombstones) and the ACME cache. No readings. Use an encrypted EBS volume. + +## Host rules + +- Swap off (`swapoff -a`, no swap in fstab) and core dumps off (the compose + file sets `ulimits: core: 0`; `ftwcloudd` also marks itself not dumpable). + Household readings exist only in this process's memory. +- Security group: 443 from anywhere. Nothing else; administration is through + SSM, with every session logged to a separate account. +- Outbound only to AWS endpoints, Let's Encrypt and, from C4, APNs. +- Deploy by image digest, built in CI from a reviewed commit. The running + build's source is at `https://cloud.ftw.energy/source`. + +## Not here yet + +APNs (C4), the app session (C3), widget tokens (C5) and statistics (C7–C8) +add secrets, egress and storage when they land. The erase test, canary homes +and the external heartbeat that pages Sourceful belong with the first +deployment. diff --git a/deploy/compose.yml b/deploy/compose.yml new file mode 100644 index 0000000..d632f35 --- /dev/null +++ b/deploy/compose.yml @@ -0,0 +1,50 @@ +# FTW Cloud on one small instance. Nothing deploys this by itself. +# +# One container, and no TLS terminator in front of it: each box binds its +# handshake to the TLS session it opened, so ftwcloudd ends TLS itself. The +# relay can sit behind Caddy because what it carries is Noise inside; this +# service reads what it carries, so the box checks that nobody in between +# can. A TCP load balancer is fine; an HTTP one is not. + +name: ftw-cloud + +services: + ftwcloudd: + image: ${FTW_CLOUD_IMAGE:?set FTW_CLOUD_IMAGE to an image digest} + restart: unless-stopped + ports: + - '443:443' + command: + - -listen=:443 + - -domain=cloud.ftw.energy + - -state=/state + - -service-key=/run/secrets/ftw-cloud-service.key + - -acme-email=${FTW_CLOUD_ACME_EMAIL:-} + volumes: + # The registry and certificates. No readings are ever written here. + - ftw-cloud-state:/state + secrets: + - ftw-cloud-service.key + read_only: true + tmpfs: + - /tmp + # Household readings live in memory. No core file may carry them out. + ulimits: + core: 0 + cap_drop: [ALL] + security_opt: + - no-new-privileges:true + logging: + driver: json-file + options: + max-size: '10m' + max-file: '3' + +secrets: + # Fetched from Secrets Manager at boot into a tmpfs path; never baked + # into the image or the volume. + ftw-cloud-service.key: + file: /run/ftw-cloud/service.key + +volumes: + ftw-cloud-state: diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..2c89de0 --- /dev/null +++ b/go.mod @@ -0,0 +1,26 @@ +module github.com/srcfl/ftw-cloud + +go 1.26.0 + +require ( + github.com/fxamacker/cbor/v2 v2.9.4 + github.com/gorilla/websocket v1.5.3 + github.com/srcfl/ftw/go/protocol v0.0.0-20261012183740-8a030e2e4805 + golang.org/x/crypto v0.57.0 + golang.org/x/sys v0.48.0 + modernc.org/sqlite v1.59.0 +) + +require ( + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/mattn/go-isatty v0.0.24 // indirect + github.com/ncruces/go-strftime v1.0.0 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + github.com/x448/float16 v0.8.4 // indirect + golang.org/x/net v0.58.0 // indirect + golang.org/x/text v0.42.0 // indirect + modernc.org/libc v1.75.7 // indirect + modernc.org/mathutil v1.7.1 // indirect + modernc.org/memory v1.12.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..70f63f7 --- /dev/null +++ b/go.sum @@ -0,0 +1,64 @@ +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/fxamacker/cbor/v2 v2.9.4 h1:xwjVlxEMR3S605oUlgBjKLTTeGFciYPGYCtF/35LKGo= +github.com/fxamacker/cbor/v2 v2.9.4/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= +github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3 h1:LMLX+LgTNWpfvCBdFebv6EsYotImrt/Ppc5cXIriCSo= +github.com/google/pprof v0.0.0-20260802141513-ef3492d7dac3/go.mod h1:jl5iWTm0/hd5PjEYEOuwAJ57L/CibdZfrqZ5XA5GrCk= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= +github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI= +github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A= +github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= +github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/srcfl/ftw/go/protocol v0.0.0-20261012183740-8a030e2e4805 h1:DmeVlx+LhctSDD4sQS1ltcN8QD2EdaKiHtqU1Qn3K1U= +github.com/srcfl/ftw/go/protocol v0.0.0-20261012183740-8a030e2e4805/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU= +github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= +github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= +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.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c= +golang.org/x/mod v0.41.0/go.mod h1:Ek9pY8RKWXwsWvd3rQiHYtMqkjSUV+s1Rj7j4H5Ur6o= +golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= +golang.org/x/tools v0.49.0 h1:3NI7VXzL9+1WZD52Dx2ttoPwD5DWrFGpl9mFZDlmisI= +golang.org/x/tools v0.49.0/go.mod h1:SJNXV9DBKT0UbdttsQjbfJlAE/q+y36++zo3uL3N0Oo= +modernc.org/cc/v4 v4.29.2 h1:h6+9ciCnPKutf4I03CvheAvDLX7+IHlqR6Iy6J+cgd8= +modernc.org/cc/v4 v4.29.2/go.mod h1:OnovgIhbbMXMu1aISnJ0wvVD1KnW+cAUJkIrAWh+kVI= +modernc.org/ccgo/v4 v4.35.0 h1:F+TUsmw09QxLzmi3aeYYGxjAXarmZaKgj3mKQHNaA8w= +modernc.org/ccgo/v4 v4.35.0/go.mod h1:qrVGs9S3Sr2Ztcg9ve+kTAYMp5a3YvWjo+SoN06kJ5I= +modernc.org/fileutil v1.4.0 h1:j6ZzNTftVS054gi281TyLjHPp6CPHr2KCxEXjEbD6SM= +modernc.org/fileutil v1.4.0/go.mod h1:EqdKFDxiByqxLk8ozOxObDSfcVOv/54xDs/DUHdvCUU= +modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI= +modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito= +modernc.org/gc/v3 v3.1.5 h1:21ldfPfRYE31Tb7B3mwAK8gy1AxP4+dKjrOQPfqakoc= +modernc.org/gc/v3 v3.1.5/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY= +modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks= +modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI= +modernc.org/libc v1.75.7 h1:o3DTP9/0p9pKmY2WCKQaySW6wIiZhNM7wc2lUoyhfew= +modernc.org/libc v1.75.7/go.mod h1:bO5o2ztHxBb2rjz0PgdHN0sSMw57CgxGFLZ3Qd/QpVQ= +modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= +modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= +modernc.org/memory v1.12.1 h1:nFMiWrpStgZczNl6XI9GnIk/rWhYIyHGUaR04pGbp9g= +modernc.org/memory v1.12.1/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/opt v0.2.0 h1:tGyef5ApycA7FSEOMraay9SaTk5zmbx7Tu+cJs4QKZg= +modernc.org/opt v0.2.0/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns= +modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w= +modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE= +modernc.org/sqlite v1.59.0 h1:X1es1GpqBlS/5T+vbM4HLUdaa8OtQx468DF2vrx+38A= +modernc.org/sqlite v1.59.0/go.mod h1:+paeT2A3iPRHkQDwG7oA6Tk0zQd5woMEI8q7orfry8k= +modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0= +modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A= +modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= +modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= diff --git a/internal/registry/registry.go b/internal/registry/registry.go new file mode 100644 index 0000000..6808993 --- /dev/null +++ b/internal/registry/registry.go @@ -0,0 +1,188 @@ +// Package registry is the service's only disk state: which homes exist, the +// box key each was claimed with, and the roster the box last signed. It holds +// no readings. Deleted rows are overwritten (secure_delete), so a purge does +// not leave a home's roster in free pages. +package registry + +import ( + "database/sql" + "errors" + "fmt" + "time" + + _ "modernc.org/sqlite" +) + +// Retention, from the protocol's "Numbers in one place". +const ( + // NewHomeGrace is how long a new home may go without a roster. + NewHomeGrace = 24 * time.Hour + // RosterGrace is how long a home lives without a valid roster. + RosterGrace = 90 * 24 * time.Hour + // TombstoneKept is how long a purged home's receipt can be asked for + // again. + TombstoneKept = 90 * 24 * time.Hour +) + +// Registry is the SQLite file. +type Registry struct { + db *sql.DB +} + +// Home is one registered home. +type Home struct { + ID string + PK []byte + CreatedAt time.Time + // Roster is the roster envelope exactly as the box signed it, nil when + // none has arrived; RosterSeq its number and RosterAt when it arrived. + Roster []byte + RosterSeq uint64 + RosterAt time.Time + GrantID []byte +} + +// Open opens or creates the registry at path. +func Open(path string) (*Registry, error) { + db, err := sql.Open("sqlite", "file:"+path+"?_pragma=journal_mode(WAL)&_pragma=secure_delete(ON)&_pragma=busy_timeout(5000)") + if err != nil { + return nil, err + } + db.SetMaxOpenConns(1) + _, err = db.Exec(` +CREATE TABLE IF NOT EXISTS homes ( + home TEXT PRIMARY KEY, + pk BLOB NOT NULL, + created_ms INTEGER NOT NULL, + roster BLOB, + roster_seq INTEGER NOT NULL DEFAULT 0, + roster_ms INTEGER NOT NULL DEFAULT 0, + grant_id BLOB +); +CREATE TABLE IF NOT EXISTS tombstones ( + home TEXT PRIMARY KEY, + grant_id BLOB, + deleted_ms INTEGER NOT NULL +);`) + if err != nil { + _ = db.Close() + return nil, fmt.Errorf("registry: schema: %w", err) + } + return &Registry{db: db}, nil +} + +// Close closes the file. +func (r *Registry) Close() error { return r.db.Close() } + +// Claim registers a home at its box's first hello. A home id is a hash of the +// key, so a second claim with another key is impossible by construction, and +// one with the same key is a no-op. A purged home is not claimed again. +func (r *Registry) Claim(home string, pk []byte, now time.Time) error { + if _, ok, err := r.Tombstone(home); err != nil || ok { + if ok { + return ErrPurged + } + return err + } + _, err := r.db.Exec(`INSERT INTO homes (home, pk, created_ms) VALUES (?, ?, ?) ON CONFLICT (home) DO NOTHING`, + home, pk, now.UnixMilli()) + return err +} + +// ErrPurged is a home whose box asked for it to be deleted. +var ErrPurged = errors.New("registry: home was purged") + +// Get returns one home. +func (r *Registry) Get(home string) (Home, bool, error) { + var h Home + var created, rosterMs int64 + var seq int64 + err := r.db.QueryRow(`SELECT home, pk, created_ms, roster, roster_seq, roster_ms, grant_id FROM homes WHERE home = ?`, home). + Scan(&h.ID, &h.PK, &created, &h.Roster, &seq, &rosterMs, &h.GrantID) + if errors.Is(err, sql.ErrNoRows) { + return Home{}, false, nil + } + if err != nil { + return Home{}, false, err + } + h.CreatedAt, h.RosterSeq = time.UnixMilli(created), uint64(seq) + if rosterMs > 0 { + h.RosterAt = time.UnixMilli(rosterMs) + } + return h, true, nil +} + +// SetRoster keeps the signed roster envelope. +func (r *Registry) SetRoster(home string, seq uint64, grantID, envelope []byte, now time.Time) error { + _, err := r.db.Exec(`UPDATE homes SET roster = ?, roster_seq = ?, roster_ms = ?, grant_id = ? WHERE home = ? AND roster_seq <= ?`, + envelope, int64(seq), now.UnixMilli(), grantID, home, int64(seq)) + return err +} + +// Purge deletes a home and leaves a tombstone, so the receipt can be sent +// again if the first one is lost. +func (r *Registry) Purge(home string, grantID []byte, now time.Time) error { + tx, err := r.db.Begin() + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + if _, err := tx.Exec(`DELETE FROM homes WHERE home = ?`, home); err != nil { + return err + } + if _, err := tx.Exec(`INSERT INTO tombstones (home, grant_id, deleted_ms) VALUES (?, ?, ?) + ON CONFLICT (home) DO NOTHING`, home, grantID, now.UnixMilli()); err != nil { + return err + } + return tx.Commit() +} + +// Tombstone returns a purged home's grant id and when it was deleted. +func (r *Registry) Tombstone(home string) (time.Time, bool, error) { + var ms int64 + err := r.db.QueryRow(`SELECT deleted_ms FROM tombstones WHERE home = ?`, home).Scan(&ms) + if errors.Is(err, sql.ErrNoRows) { + return time.Time{}, false, nil + } + if err != nil { + return time.Time{}, false, err + } + return time.UnixMilli(ms), true, nil +} + +// TombstoneGrant returns the grant id a purged home was deleted under. +func (r *Registry) TombstoneGrant(home string) []byte { + var id []byte + _ = r.db.QueryRow(`SELECT grant_id FROM tombstones WHERE home = ?`, home).Scan(&id) + return id +} + +// Expire deletes homes that sent no roster within NewHomeGrace, or none for +// RosterGrace, and tombstones past TombstoneKept. It returns the homes it +// deleted so memory can forget them too. +func (r *Registry) Expire(now time.Time) ([]string, error) { + newCutoff := now.Add(-NewHomeGrace).UnixMilli() + rosterCutoff := now.Add(-RosterGrace).UnixMilli() + rows, err := r.db.Query(`SELECT home FROM homes + WHERE (roster_ms = 0 AND created_ms < ?) OR (roster_ms > 0 AND roster_ms < ?)`, newCutoff, rosterCutoff) + if err != nil { + return nil, err + } + var gone []string + for rows.Next() { + var h string + if err := rows.Scan(&h); err != nil { + _ = rows.Close() + return nil, err + } + gone = append(gone, h) + } + _ = rows.Close() + for _, h := range gone { + if _, err := r.db.Exec(`DELETE FROM homes WHERE home = ?`, h); err != nil { + return nil, err + } + } + _, err = r.db.Exec(`DELETE FROM tombstones WHERE deleted_ms < ?`, now.Add(-TombstoneKept).UnixMilli()) + return gone, err +} diff --git a/internal/registry/registry_test.go b/internal/registry/registry_test.go new file mode 100644 index 0000000..b878673 --- /dev/null +++ b/internal/registry/registry_test.go @@ -0,0 +1,72 @@ +package registry + +import ( + "errors" + "path/filepath" + "testing" + "time" +) + +func open(t *testing.T) *Registry { + r, err := Open(filepath.Join(t.TempDir(), "registry.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = r.Close() }) + return r +} + +func TestAHomeIsClaimedOnceAndKeepsItsRoster(t *testing.T) { + r := open(t) + now := time.Unix(1_791_820_800, 0) + if err := r.Claim("h1", []byte("pk"), now); err != nil { + t.Fatal(err) + } + if err := r.Claim("h1", []byte("pk"), now.Add(time.Hour)); err != nil { + t.Fatal(err) + } + if err := r.SetRoster("h1", 3, []byte("grant"), []byte("signed roster"), now); err != nil { + t.Fatal(err) + } + _ = r.SetRoster("h1", 2, []byte("grant"), []byte("older"), now) // ignored + h, ok, err := r.Get("h1") + if err != nil || !ok || h.RosterSeq != 3 || string(h.Roster) != "signed roster" || !h.CreatedAt.Equal(now) { + t.Fatalf("home %+v, %v, %v", h, ok, err) + } +} + +// A purged home leaves only a tombstone, and its box cannot claim it again. +func TestAPurgeLeavesATombstone(t *testing.T) { + r := open(t) + now := time.Unix(1_791_820_800, 0) + _ = r.Claim("h1", []byte("pk"), now) + _ = r.SetRoster("h1", 1, []byte("grant-1"), []byte("roster"), now) + if err := r.Purge("h1", []byte("grant-1"), now); err != nil { + t.Fatal(err) + } + if _, ok, _ := r.Get("h1"); ok { + t.Fatal("the home survived its purge") + } + if string(r.TombstoneGrant("h1")) != "grant-1" { + t.Fatal("no tombstone") + } + if err := r.Claim("h1", []byte("pk"), now); !errors.Is(err, ErrPurged) { + t.Fatalf("claim after purge: %v", err) + } +} + +func TestHomesWithoutARosterExpire(t *testing.T) { + r := open(t) + now := time.Unix(1_791_820_800, 0) + _ = r.Claim("silent", []byte("a"), now) + _ = r.Claim("talks", []byte("b"), now) + _ = r.SetRoster("talks", 1, nil, []byte("r"), now) + gone, err := r.Expire(now.Add(NewHomeGrace + time.Minute)) + if err != nil || len(gone) != 1 || gone[0] != "silent" { + t.Fatalf("after a day: %v %v", gone, err) + } + gone, _ = r.Expire(now.Add(RosterGrace + time.Minute)) + if len(gone) != 1 || gone[0] != "talks" { + t.Fatalf("after 90 days: %v", gone) + } +} diff --git a/internal/server/inspect_dev.go b/internal/server/inspect_dev.go new file mode 100644 index 0000000..f0e8893 --- /dev/null +++ b/internal/server/inspect_dev.go @@ -0,0 +1,29 @@ +//go:build ftwclouddev + +package server + +import ( + "encoding/json" + "net" + "net/http" +) + +// A local test binary answers /dev/held?home= from loopback with the +// envelopes the store holds. Release builds do not compile this file. +func init() { + devRoutes = func(mux *http.ServeMux, s *Server) { + mux.HandleFunc("/dev/held", func(w http.ResponseWriter, r *http.Request) { + host, _, _ := net.SplitHostPort(r.RemoteAddr) + if ip := net.ParseIP(host); ip == nil || !ip.IsLoopback() { + http.NotFound(w, r) + return + } + latest, online, ok := s.Store.Inspect(r.URL.Query().Get("home")) + if !ok { + http.NotFound(w, r) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{"online": online, "envelopes": latest}) + }) + } +} diff --git a/internal/server/server.go b/internal/server/server.go new file mode 100644 index 0000000..eb4a0c0 --- /dev/null +++ b/internal/server/server.go @@ -0,0 +1,141 @@ +// Package server is ftwcloudd's HTTP surface. It is small on purpose: every +// route is listed in Routes, and TestNoRouteReadsAHome holds the list. +// +// /v1/box a box's uplink (package uplink) +// /v1/app the app's Noise session to the service: C3, not built yet +// /v1/now an extension's token read: C5, not built yet +// /source the source of the running service, as AGPL-3.0 asks +// /healthz liveness, and nothing about any home +// +// There is no route for staff. Reading a home needs its box's key or a +// phone in its roster. +package server + +import ( + "context" + "log/slog" + "net/http" + "os" + "runtime/debug" + "time" + + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw-cloud/internal/uplink" +) + +// SourceArchive is where the image keeps the source it was built from. +const SourceArchive = "/usr/share/ftw-cloud/source.tar.gz" + +// Routes are every path the service answers. +var Routes = []string{"/v1/box", "/v1/app", "/v1/now", "/source", "/healthz"} + +// devRoutes adds a local inspection route in binaries built with +// -tags ftwclouddev. It is nil in every other build. +var devRoutes func(mux *http.ServeMux, s *Server) + +// Server wires the store, the registry and the routes. +type Server struct { + Store *store.Store + Registry *registry.Registry + Uplink *uplink.Uplink + Log *slog.Logger +} + +// Handler is the whole HTTP surface. +func (s *Server) Handler() http.Handler { + mux := http.NewServeMux() + mux.Handle("/v1/box", s.Uplink) + // C3: the app opens a WebSocket here and runs Noise IK to the service's + // static key, with prologue "ftw.cloud.v1:" and the home id. The + // service admits a phone only through store.For with a Device + // principal, and forwards the box's envelopes unchanged. + mux.HandleFunc("/v1/app", notYet) + // C5: widgets read the newest live, prices and hb here with a 7-day + // token bound to one roster entry. + mux.HandleFunc("/v1/now", notYet) + mux.HandleFunc("/source", source) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-store") + _, _ = w.Write([]byte("ok\n")) + }) + if devRoutes != nil { + devRoutes(mux, s) + } + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Link", `; rel="copyright"; title="AGPL-3.0 source"`) + mux.ServeHTTP(w, r) + }) +} + +func notYet(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "not available yet", http.StatusNotImplemented) +} + +// source serves the archive the image was built from. A development build +// has none and says where the source lives. +func source(w http.ResponseWriter, r *http.Request) { + if rev := revision(); rev != "" { + w.Header().Set("X-Source-Revision", rev) + } + f, err := os.Open(SourceArchive) + if err != nil { + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = w.Write([]byte("Development build: matching source must be supplied by the operator. https://github.com/srcfl/ftw-cloud\n")) + return + } + defer f.Close() + w.Header().Set("Content-Type", "application/gzip") + w.Header().Set("Content-Disposition", `attachment; filename="ftw-cloud-source.tar.gz"`) + http.ServeContent(w, r, "ftw-cloud-source.tar.gz", time.Time{}, f) +} + +// revision is the commit this binary was built from, when the build recorded +// one. +func revision() string { + info, ok := debug.ReadBuildInfo() + if !ok { + return "" + } + rev, dirty := "", false + for _, s := range info.Settings { + switch s.Key { + case "vcs.revision": + rev = s.Value + case "vcs.modified": + dirty = s.Value == "true" + } + } + if rev != "" && dirty { + rev += "+modified" + } + return rev +} + +// Upkeep applies the retention rules until ctx ends: readings go an hour +// after the box does, homes without a roster go after 24 hours or 90 days, +// and old tombstones go. +func (s *Server) Upkeep(ctx context.Context, every time.Duration) { + t := time.NewTicker(every) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + s.Store.Sweep() + gone, err := s.Registry.Expire(time.Now()) + if err != nil { + s.Log.Warn("registry upkeep failed", "err", err) + continue + } + for _, h := range gone { + s.Store.Forget(h) + } + if len(gone) > 0 { + s.Log.Info("homes expired", "count", len(gone)) + } + } + } +} diff --git a/internal/server/server_test.go b/internal/server/server_test.go new file mode 100644 index 0000000..1322d9c --- /dev/null +++ b/internal/server/server_test.go @@ -0,0 +1,83 @@ +package server + +import ( + "bytes" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "path/filepath" + "slices" + "testing" + + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw-cloud/internal/uplink" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +func newServer(t *testing.T) (*Server, string) { + t.Helper() + key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = reg.Close() }) + st := store.New(nil) + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + s := &Server{Store: st, Registry: reg, Uplink: uplink.New(st, reg, key, log), Log: log} + + // A home with readings in it, held the way a box's uplink holds them. + box, _ := st.For(store.Principal{Kind: store.Box, Home: "canaryhome"}) + box.Connected() + _ = box.Put(cloudwire.MsgLive, []byte("CANARY-READING"), 1, true) + return s, "canaryhome" +} + +// No route reads a home. Every path answers without the canary, whatever the +// method or the query asks for. +func TestNoRouteReadsAHome(t *testing.T) { + if devRoutes != nil { + t.Skip("built with -tags ftwclouddev, which adds a local inspection route on purpose") + } + s, home := newServer(t) + h := s.Handler() + paths := append(slices.Clone(Routes), "/", "/v1", "/v1/home", "/admin", "/debug/pprof/", "/metrics", + "/dev/held", "/dev/held?home="+home, "/v1/now?home="+home, "/v1/app?home="+home) + for _, p := range paths { + for _, m := range []string{http.MethodGet, http.MethodPost} { + w := httptest.NewRecorder() + r := httptest.NewRequest(m, p, nil) + r.RemoteAddr = "127.0.0.1:1" + h.ServeHTTP(w, r) + if bytes.Contains(w.Body.Bytes(), []byte("CANARY")) { + t.Fatalf("%s %s returned a home's readings", m, p) + } + if w.Header().Get("Link") == "" { + t.Fatalf("%s %s has no source link", m, p) + } + } + } +} + +func TestSourceSaysWhereTheSourceIs(t *testing.T) { + s, _ := newServer(t) + w := httptest.NewRecorder() + s.Handler().ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/source", nil)) + if w.Code != http.StatusServiceUnavailable || !bytes.Contains(w.Body.Bytes(), []byte("github.com/srcfl/ftw-cloud")) { + t.Fatalf("development /source: %d %s", w.Code, w.Body.String()) + } +} + +func TestTheUplinkRefusesPlainHTTP(t *testing.T) { + s, _ := newServer(t) + w := httptest.NewRecorder() + s.Handler().ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/v1/box", nil)) + if w.Code != http.StatusUpgradeRequired { + t.Fatalf("plain HTTP to /v1/box: %d", w.Code) + } +} diff --git a/internal/store/inspect_dev.go b/internal/store/inspect_dev.go new file mode 100644 index 0000000..9b34672 --- /dev/null +++ b/internal/store/inspect_dev.go @@ -0,0 +1,26 @@ +//go:build ftwclouddev + +package store + +// Inspect returns every envelope a home holds. It exists only in local test +// binaries built with -tags ftwclouddev, so an end-to-end run can compare +// what the service holds with what the box says it sent. A release build has +// no way to call it. +func (s *Store) Inspect(id string) (latest map[string][]byte, online bool, ok bool) { + s.mu.Lock() + h := s.homes[id] + s.mu.Unlock() + if h == nil { + return nil, false, false + } + h.mu.Lock() + defer h.mu.Unlock() + latest = map[string][]byte{} + for t, e := range h.latest { + latest[t] = e.Envelope + } + if h.full != nil { + latest["live+sources"] = h.full.Envelope + } + return latest, h.online, true +} diff --git a/internal/store/store.go b/internal/store/store.go new file mode 100644 index 0000000..0713484 --- /dev/null +++ b/internal/store/store.go @@ -0,0 +1,285 @@ +// Package store holds each home's state in memory. Nothing here is written to +// disk: a restart empties it, and boxes fill it again when they reconnect. +// +// The only way to a home is For, with a principal that proves whose home it +// is: the box, by the signed hello on its own connection, or a phone in the +// home's roster, by its Noise handshake. There is no listing, no lookup by +// anything else, and no principal for staff. One bug in a second access path +// would expose every home at once, so there is no second access path. +package store + +import ( + "bytes" + "errors" + "slices" + "sort" + "sync" + "time" + + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// ReadingsKeptFor is how long readings outlive the box's last socket. After +// that a phone gets only when the box was last seen. +const ReadingsKeptFor = time.Hour + +// Kind says how a principal proved itself. +type Kind int + +const ( + // Box is the box, proven by the hello it signed on this connection. + Box Kind = iota + 1 + // Device is a phone in the home's roster, proven by its Noise static key. + Device +) + +// Principal is who is asking for a home. +type Principal struct { + Kind Kind + Home string + // DeviceKey is the phone's X25519 static key, for Device. + DeviceKey []byte +} + +// ErrDenied is every refusal. It does not say whether the home exists. +var ErrDenied = errors.New("store: denied") + +// Store is every home the service holds. +type Store struct { + mu sync.Mutex + homes map[string]*home + now func() time.Time +} + +// New starts an empty store. now is for tests; nil uses the clock. +func New(now func() time.Time) *Store { + if now == nil { + now = time.Now + } + return &Store{homes: map[string]*home{}, now: now} +} + +type home struct { + mu sync.Mutex + // latest is the newest envelope of each class, exactly as the box sent + // it; full is the newest live that carried its source table. + latest map[string]Held + full *Held + roster *cloudwire.Roster + rosterAt time.Time + online bool + conn uint64 + lastSeen time.Time +} + +// Held is one envelope as received. +type Held struct { + T string + Envelope []byte + Seq uint64 +} + +// Handle is one principal's access to one home. +type Handle struct { + s *Store + id string + h *home + kind Kind + // scopes is a device's effective scope: its roster scopes intersected + // with the grant's. + scopes []string +} + +// For returns the principal's access to its home, or ErrDenied. A box may +// create its home; a device reaches only a home whose current roster lists +// its key. +func (s *Store) For(p Principal) (*Handle, error) { + s.mu.Lock() + h := s.homes[p.Home] + if h == nil && p.Kind == Box && p.Home != "" { + h = &home{latest: map[string]Held{}} + s.homes[p.Home] = h + } + s.mu.Unlock() + if h == nil { + return nil, ErrDenied + } + switch p.Kind { + case Box: + return &Handle{s: s, id: p.Home, h: h, kind: Box}, nil + case Device: + h.mu.Lock() + defer h.mu.Unlock() + r := h.roster + if r == nil || s.now().After(h.rosterAt.Add(time.Duration(r.TTLS)*time.Second)) { + return nil, ErrDenied + } + for _, d := range r.Devices { + if bytes.Equal(d.Key, p.DeviceKey) && len(p.DeviceKey) == 32 { + var scopes []string + for _, sc := range d.Scopes { + if slices.Contains(r.Grant.Scopes, sc) { + scopes = append(scopes, sc) + } + } + return &Handle{s: s, id: p.Home, h: h, kind: Device, scopes: scopes}, nil + } + } + } + return nil, ErrDenied +} + +// -------------------------------------------------------------------------- +// The box's handle +// -------------------------------------------------------------------------- + +// Connected marks the box's socket up and returns its number. A later +// connection replaces this one; Disconnected with an older number is ignored. +func (x *Handle) Connected() uint64 { + x.h.mu.Lock() + defer x.h.mu.Unlock() + x.h.conn++ + x.h.online = true + x.h.lastSeen = x.s.now() + return x.h.conn +} + +// Disconnected marks the socket down, if it is still the current one. +func (x *Handle) Disconnected(conn uint64) { + x.h.mu.Lock() + defer x.h.mu.Unlock() + if x.h.conn == conn { + x.h.online = false + x.h.lastSeen = x.s.now() + } +} + +// Put keeps env as the newest of its class. +func (x *Handle) Put(t string, env []byte, seq uint64, carriesTable bool) error { + if x.kind != Box { + return ErrDenied + } + x.h.mu.Lock() + defer x.h.mu.Unlock() + held := Held{T: t, Envelope: env, Seq: seq} + x.h.latest[t] = held + if t == cloudwire.MsgLive && carriesTable { + x.h.full = &held + } + x.h.lastSeen = x.s.now() + return nil +} + +// Roster returns the roster held, and its number, 0 when none. +func (x *Handle) Roster() (*cloudwire.Roster, uint64) { + x.h.mu.Lock() + defer x.h.mu.Unlock() + if x.h.roster == nil { + return nil, 0 + } + return x.h.roster, x.h.roster.RosterSeq +} + +// SetRoster keeps r when its number is at least the one held. An equal +// number renews its time to live. +func (x *Handle) SetRoster(r *cloudwire.Roster) bool { + if x.kind != Box { + return false + } + x.h.mu.Lock() + defer x.h.mu.Unlock() + if x.h.roster != nil && r.RosterSeq < x.h.roster.RosterSeq { + return false + } + x.h.roster, x.h.rosterAt = r, x.s.now() + return true +} + +// Delete forgets the home. Only its box may. +func (x *Handle) Delete() { + if x.kind != Box { + return + } + x.s.mu.Lock() + delete(x.s.homes, x.id) + x.s.mu.Unlock() + x.h.mu.Lock() + x.h.latest, x.h.full, x.h.roster = map[string]Held{}, nil, nil + x.h.mu.Unlock() +} + +// -------------------------------------------------------------------------- +// A device's handle +// -------------------------------------------------------------------------- + +// Snapshot is what a phone gets first: for each class it may read, the +// newest envelope; the newest live that carried sources; and the newest hb, +// in rising seq. After an hour without the box, no readings at all. +type Snapshot struct { + Envelopes [][]byte + LastSeenMs int64 + Online bool +} + +// Snapshot returns the device's view. +func (x *Handle) Snapshot() (Snapshot, error) { + if x.kind != Device { + return Snapshot{}, ErrDenied + } + x.h.mu.Lock() + defer x.h.mu.Unlock() + out := Snapshot{LastSeenMs: x.h.lastSeen.UnixMilli(), Online: x.h.online} + if !x.h.online && x.s.now().Sub(x.h.lastSeen) > ReadingsKeptFor { + return out, nil + } + var held []Held + for t, e := range x.h.latest { + if x.mayRead(t) { + held = append(held, e) + } + } + if x.h.full != nil && x.mayRead(cloudwire.MsgLive) && x.h.latest[cloudwire.MsgLive].Seq != x.h.full.Seq { + held = append(held, *x.h.full) + } + sort.Slice(held, func(a, b int) bool { return held[a].Seq < held[b].Seq }) + for _, e := range held { + out.Envelopes = append(out.Envelopes, e.Envelope) + } + return out, nil +} + +func (x *Handle) mayRead(class string) bool { + scope, forwarded := cloudwire.ClassScope[class] + return forwarded && (scope == "" || slices.Contains(x.scopes, scope)) +} + +// -------------------------------------------------------------------------- +// Upkeep +// -------------------------------------------------------------------------- + +// Sweep drops the readings of every home whose box has been gone longer than +// ReadingsKeptFor. What the box was last seen at stays. +func (s *Store) Sweep() { + now := s.now() + s.mu.Lock() + homes := make([]*home, 0, len(s.homes)) + for _, h := range s.homes { + homes = append(homes, h) + } + s.mu.Unlock() + for _, h := range homes { + h.mu.Lock() + if !h.online && now.Sub(h.lastSeen) > ReadingsKeptFor { + h.latest, h.full = map[string]Held{}, nil + } + h.mu.Unlock() + } +} + +// Forget drops a home the registry has deleted, as after 90 days without a +// roster. It is called by upkeep, never by a request. +func (s *Store) Forget(id string) { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.homes, id) +} diff --git a/internal/store/store_test.go b/internal/store/store_test.go new file mode 100644 index 0000000..d0c1078 --- /dev/null +++ b/internal/store/store_test.go @@ -0,0 +1,167 @@ +package store + +import ( + "bytes" + "errors" + "testing" + "time" + + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +type clock struct{ t time.Time } + +func (c *clock) now() time.Time { return c.t } + +var ( + ownerKey = bytes.Repeat([]byte{1}, 32) + viewerKey = bytes.Repeat([]byte{2}, 32) +) + +func roster(seq uint64, grant ...string) *cloudwire.Roster { + return &cloudwire.Roster{RosterSeq: seq, TTLS: 86_400, + Devices: []cloudwire.Device{ + {ID: "owner", Key: ownerKey, Role: "owner", Scopes: protocol.CloudScopes}, + {ID: "viewer", Key: viewerKey, Role: "viewer", Scopes: []string{protocol.ScopeLiveRead}}, + }, + Grant: cloudwire.Grant{ID: make([]byte, 16), Scopes: grant, Controller: "sourceful"}} +} + +func filled(t *testing.T) (*Store, *Handle, *clock) { + c := &clock{t: time.Unix(1_791_820_800, 0)} + s := New(c.now) + box, err := s.For(Principal{Kind: Box, Home: "h1"}) + if err != nil { + t.Fatal(err) + } + box.Connected() + box.SetRoster(roster(1, protocol.ScopeLiveRead, protocol.ScopePlanRead)) + _ = box.Put(cloudwire.MsgLive, []byte("live-full"), 3, true) + _ = box.Put(cloudwire.MsgPlan, []byte("plan"), 4, false) + _ = box.Put(cloudwire.MsgLive, []byte("live-delta"), 5, false) + _ = box.Put(cloudwire.MsgHB, []byte("hb"), 6, false) + _ = box.Put(cloudwire.MsgEvent, []byte("event"), 7, false) + return s, box, c +} + +func TestOnlyThePrincipalsOwnHome(t *testing.T) { + s, _, _ := filled(t) + for name, p := range map[string]Principal{ + "no kind": {Home: "h1"}, + "unknown home": {Kind: Device, Home: "h2", DeviceKey: ownerKey}, + "key not in roster": {Kind: Device, Home: "h1", DeviceKey: bytes.Repeat([]byte{9}, 32)}, + "empty key": {Kind: Device, Home: "h1"}, + "empty home": {Kind: Box}, + } { + if _, err := s.For(p); !errors.Is(err, ErrDenied) { + t.Errorf("%s: %v", name, err) + } + } +} + +// A device reads its effective scope: its role's scopes within the grant. +// The owner reads plan here because the grant has it; nobody reads event, +// which the grant lacks. +func TestADeviceReadsOnlyItsEffectiveScope(t *testing.T) { + s, _, _ := filled(t) + read := func(key []byte) []string { + x, err := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: key}) + if err != nil { + t.Fatal(err) + } + snap, err := x.Snapshot() + if err != nil { + t.Fatal(err) + } + var got []string + for _, e := range snap.Envelopes { + got = append(got, string(e)) + } + return got + } + if got := read(ownerKey); len(got) != 4 || got[0] != "live-full" || got[1] != "plan" || got[2] != "live-delta" || got[3] != "hb" { + t.Fatalf("owner read %v", got) + } + if got := read(viewerKey); len(got) != 3 || got[0] != "live-full" || got[2] != "hb" { + t.Fatalf("viewer read %v", got) + } +} + +func TestABoxCannotReadAndADeviceCannotWrite(t *testing.T) { + s, box, _ := filled(t) + if _, err := box.Snapshot(); !errors.Is(err, ErrDenied) { + t.Fatal("the box's handle read the home") + } + dev, _ := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}) + if err := dev.Put(cloudwire.MsgLive, []byte("forged"), 99, true); !errors.Is(err, ErrDenied) { + t.Fatal("a device wrote the home") + } + if dev.SetRoster(roster(9)) { + t.Fatal("a device set the roster") + } +} + +func TestAnExpiredRosterLocksDevicesOut(t *testing.T) { + s, _, c := filled(t) + c.t = c.t.Add(24*time.Hour + time.Second) + if _, err := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}); !errors.Is(err, ErrDenied) { + t.Fatalf("expired roster: %v", err) + } +} + +func TestAnOlderRosterIsIgnoredAndAnEqualOneRenews(t *testing.T) { + s, box, c := filled(t) + if box.SetRoster(roster(0)) { + t.Fatal("an older roster replaced a newer one") + } + c.t = c.t.Add(20 * time.Hour) + if !box.SetRoster(roster(1, protocol.ScopeLiveRead)) { + t.Fatal("an equal roster was refused") + } + c.t = c.t.Add(20 * time.Hour) + if _, err := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}); err != nil { + t.Fatal("an equal roster did not renew its time to live") + } +} + +// An hour after the box went quiet a phone gets when it was last seen, and +// no readings. +func TestReadingsGoAnHourAfterTheBox(t *testing.T) { + s, box, c := filled(t) + gen := box.Connected() + box.Disconnected(gen) + c.t = c.t.Add(ReadingsKeptFor + time.Second) + dev, _ := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}) + snap, _ := dev.Snapshot() + if len(snap.Envelopes) != 0 || snap.Online || snap.LastSeenMs == 0 { + t.Fatalf("after an hour: %+v", snap) + } + s.Sweep() + c.t = c.t.Add(-time.Hour) // even read back early, the sweep dropped them + snap, _ = dev.Snapshot() + if len(snap.Envelopes) != 0 { + t.Fatal("the sweep left readings in memory") + } +} + +func TestANewerConnectionOutlivesAnOlderOne(t *testing.T) { + s, box, c := filled(t) + first := box.Connected() + second := box.Connected() + box.Disconnected(first) + c.t = c.t.Add(2 * ReadingsKeptFor) + dev, _ := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}) + if snap, _ := dev.Snapshot(); !snap.Online || len(snap.Envelopes) == 0 { + t.Fatalf("the old connection's close took the home offline: %+v", snap) + } + box.Disconnected(second) +} + +func TestDeleteForgetsTheHome(t *testing.T) { + s, box, _ := filled(t) + box.Delete() + if _, err := s.For(Principal{Kind: Device, Home: "h1", DeviceKey: ownerKey}); !errors.Is(err, ErrDenied) { + t.Fatal("a deleted home still answers") + } +} diff --git a/internal/uplink/downlink.go b/internal/uplink/downlink.go new file mode 100644 index 0000000..40dae17 --- /dev/null +++ b/internal/uplink/downlink.go @@ -0,0 +1,68 @@ +package uplink + +import ( + "fmt" + + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// down is a message the service may send a box. The method is unexported, so +// only the five types below implement it and a sixth cannot be sent by +// accident. TestDownlinkIsTheProtocolsClosedSet holds the list to +// cloudwire.Downlink, and the box closes the uplink on anything else. +// +// None of the five asks the box to act. watch asks it to send sooner, resync +// to send again, and purged confirms a deletion the box asked for. +type down interface { + down() (t string, body any, signed bool) +} + +type ( + challenge cloudwire.Challenge + helloOK cloudwire.HelloOK + watch cloudwire.Watch + resync cloudwire.Resync + purged cloudwire.Purged +) + +func (m challenge) down() (string, any, bool) { + return cloudwire.MsgChallenge, cloudwire.Challenge(m), false +} +func (m helloOK) down() (string, any, bool) { return cloudwire.MsgHelloOK, cloudwire.HelloOK(m), true } +func (m watch) down() (string, any, bool) { return cloudwire.MsgWatch, cloudwire.Watch(m), false } +func (m resync) down() (string, any, bool) { return cloudwire.MsgResync, cloudwire.Resync(m), false } +func (m purged) down() (string, any, bool) { return cloudwire.MsgPurged, cloudwire.Purged(m), true } + +// encodeDown frames one downlink message. hello_ok and purged are signed with +// the service key over the box's home id; the rest are not. +func (u *Uplink) encodeDown(home string, m down) ([]byte, error) { + t, body, signed := m.down() + key := u.key + if !signed { + key = nil + } + env, err := cloudwire.Seal(key, home, t, body) + if err != nil { + return nil, err + } + raw, err := env.Encode() + if err != nil { + return nil, err + } + frame, err := cloudwire.EncodeFrame(raw) + if err != nil { + return nil, err + } + if len(frame) > cloudwire.MaxServiceFrame { + return nil, fmt.Errorf("uplink: %s is %d bytes, over the box's limit", t, len(frame)) + } + return frame, nil +} + +func (c *boxConn) send(m down) error { + frame, err := c.u.encodeDown(c.home, m) + if err != nil { + return err + } + return c.write(frame) +} diff --git a/internal/uplink/uplink.go b/internal/uplink/uplink.go new file mode 100644 index 0000000..da1e4f6 --- /dev/null +++ b/internal/uplink/uplink.go @@ -0,0 +1,414 @@ +// Package uplink serves /v1/box, where each box that has turned FTW Cloud on +// keeps one connection. docs/protocol-cloud.md in srcfl/ftw is the wire +// format; go/protocol/cloudwire holds its bytes. +// +// The service learns a home only from its box: the home id is a hash of the +// key that signed hello on this TLS session, never a field anyone sends, and +// every message after it must carry that key's signature. What the service +// may say back is the closed set in downlink.go. +package uplink + +import ( + "bytes" + "crypto/ecdsa" + "crypto/rand" + "crypto/tls" + "errors" + "fmt" + "log/slog" + "net/http" + "slices" + "sync" + "sync/atomic" + "time" + + "github.com/fxamacker/cbor/v2" + "github.com/gorilla/websocket" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +const ( + // maxBoxFrame bounds one message from a box. A roster for many phones + // is the largest thing a box sends. + maxBoxFrame = 64 << 10 + // handshakeTimeout bounds the hello. + handshakeTimeout = 15 * time.Second + // pingEvery and readTimeout: the box's read deadline is three pings, + // and the service gives the box as long. + pingEvery = 15 * time.Second + readTimeout = 45 * time.Second + // writeTimeout: a write blocked this long is a dead socket. + writeTimeout = 10 * time.Second + // A box sends a few messages a second at most. Past the burst at this + // rate the connection closes; one home cannot occupy the service. + rateBurst = 60 + ratePerSec = 10 +) + +// Uplink is the /v1/box handler. +type Uplink struct { + store *store.Store + registry *registry.Registry + key *ecdsa.PrivateKey + log *slog.Logger + now func() time.Time + + mu sync.Mutex + conns map[string]*boxConn + + // Dropped counts box messages of a type this service does not know. A + // newer box works with an older service by this rule. + Dropped atomic.Int64 +} + +// New builds the handler. key is the service key that signs hello_ok and +// purged; boxes carry its public half. +func New(st *store.Store, reg *registry.Registry, key *ecdsa.PrivateKey, log *slog.Logger) *Uplink { + if log == nil { + log = slog.Default() + } + return &Uplink{store: st, registry: reg, key: key, log: log, now: time.Now, conns: map[string]*boxConn{}} +} + +var upgrader = websocket.Upgrader{ + EnableCompression: false, + // A box sends no Origin. A browser always does, and has no business + // here. + CheckOrigin: func(r *http.Request) bool { return r.Header.Get("Origin") == "" }, +} + +func (u *Uplink) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.TLS == nil || r.TLS.Version != tls.VersionTLS13 { + http.Error(w, "TLS 1.3 only", http.StatusUpgradeRequired) + return + } + // The handshake is bound to this TLS session. A proxy that ended TLS + // in front of the service would hold another session with another + // value, and the box's check fails there. + ekm, err := r.TLS.ExportKeyingMaterial(cloudwire.EKMLabel, nil, cloudwire.EKMBytes) + if err != nil { + http.Error(w, "no exporter", http.StatusInternalServerError) + return + } + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer conn.Close() + conn.SetReadLimit(maxBoxFrame) + c := &boxConn{u: u, conn: conn, ekm: ekm, tokens: rateBurst, refilled: u.now()} + c.write = c.writeFrame + if err := c.run(); err != nil && !websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { + u.log.Debug("box uplink closed", "err", err) + } +} + +// Watch asks a home's box for one-second updates for a minute. The app +// session (C3) calls it while a phone is looking. +func (u *Uplink) Watch(home string) { + u.mu.Lock() + c := u.conns[home] + u.mu.Unlock() + if c != nil { + _ = c.send(watch{Hz: 1}) + } +} + +// boxConn is one box's connection. +type boxConn struct { + u *Uplink + conn *websocket.Conn + ekm []byte + // write sends one frame; tests replace it. + write func([]byte) error + writeMu sync.Mutex + + home string + pub *ecdsa.PublicKey + handle *store.Handle + purged bool + + tokens float64 + refilled time.Time +} + +func (c *boxConn) writeFrame(frame []byte) error { + c.writeMu.Lock() + defer c.writeMu.Unlock() + if err := c.conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { + return err + } + return c.conn.WriteMessage(websocket.BinaryMessage, frame) +} + +var errProtocol = errors.New("uplink: the box broke the protocol") + +func (c *boxConn) run() error { + nonce := make([]byte, cloudwire.NonceBytes) + if _, err := rand.Read(nonce); err != nil { + return err + } + if err := c.send(challenge{V: cloudwire.Version, Nonce: nonce}); err != nil { + return err + } + + _ = c.conn.SetReadDeadline(time.Now().Add(handshakeTimeout)) + kind, msg, err := c.conn.ReadMessage() + if err != nil { + return err + } + env, err := parseUplink(kind, msg) + if err != nil { + return err + } + hello, err := c.hello(env, nonce) + if err != nil { + return err + } + + switch err := c.u.registry.Claim(c.home, hello.PK, c.u.now()); { + case errors.Is(err, registry.ErrPurged): + // A box still waiting for its receipt. It may say bye, and get + // the receipt again; nothing else. + c.purged = true + case err != nil: + return err + } + if !c.purged { + if c.handle, err = c.u.store.For(store.Principal{Kind: store.Box, Home: c.home}); err != nil { + return err + } + gen := c.handle.Connected() + defer c.handle.Disconnected(gen) + c.u.mu.Lock() + if old := c.u.conns[c.home]; old != nil { + // One uplink per home: the newer one wins. + _ = old.conn.Close() + } + c.u.conns[c.home] = c + c.u.mu.Unlock() + defer func() { + c.u.mu.Lock() + if c.u.conns[c.home] == c { + delete(c.u.conns, c.home) + } + c.u.mu.Unlock() + }() + } + + if err := c.send(helloOK{BoxNonce: hello.BoxNonce, EKM: c.ekm, Have: map[string]string{}, RosterSeq: c.rosterSeq()}); err != nil { + return err + } + + done := make(chan struct{}) + defer close(done) + go c.ping(done) + _ = c.conn.SetReadDeadline(time.Now().Add(readTimeout)) + c.conn.SetPongHandler(func(string) error { + return c.conn.SetReadDeadline(time.Now().Add(readTimeout)) + }) + for { + kind, msg, err := c.conn.ReadMessage() + if err != nil { + return err + } + _ = c.conn.SetReadDeadline(time.Now().Add(readTimeout)) + if !c.allow() { + return fmt.Errorf("%w: faster than %d messages a second", errProtocol, ratePerSec) + } + env, err := parseUplink(kind, msg) + if err != nil { + return err + } + stop, err := c.apply(env) + if err != nil || stop { + return err + } + } +} + +// hello checks the box's opening message and learns the home from its key. +func (c *boxConn) hello(env cloudwire.Envelope, nonce []byte) (cloudwire.Hello, error) { + var h cloudwire.Hello + if env.T != cloudwire.MsgHello || cloudwire.Unmarshal(env.B, &h) != nil || h.V != cloudwire.Version { + return h, fmt.Errorf("%w: expected hello", errProtocol) + } + pub, err := cloudwire.ParsePublicKey(h.PK) + if err != nil { + return h, err + } + home := cloudwire.HomeID(h.PK) + if err := cloudwire.Verify(pub, home, env); err != nil { + return h, err + } + if !bytes.Equal(h.Nonce, nonce) || !bytes.Equal(h.EKM, c.ekm) || len(h.BoxNonce) != cloudwire.NonceBytes { + return h, fmt.Errorf("%w: hello is not bound to this challenge and TLS session", errProtocol) + } + c.home, c.pub = home, pub + return h, nil +} + +func (c *boxConn) rosterSeq() uint64 { + var seq uint64 + if c.handle != nil { + _, seq = c.handle.Roster() + } + if h, ok, err := c.u.registry.Get(c.home); err == nil && ok && h.RosterSeq > seq { + seq = h.RosterSeq + } + return seq +} + +func (c *boxConn) ping(done <-chan struct{}) { + t := time.NewTicker(pingEvery) + defer t.Stop() + for { + select { + case <-done: + return + case <-t.C: + if c.conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(writeTimeout)) != nil { + return + } + } + } +} + +// allow is a token bucket over the box's messages. +func (c *boxConn) allow() bool { + now := c.u.now() + c.tokens = min(rateBurst, c.tokens+now.Sub(c.refilled).Seconds()*ratePerSec) + c.refilled = now + if c.tokens < 1 { + return false + } + c.tokens-- + return true +} + +// parseUplink reads one box frame. Only binary lane 1 frames carry the +// protocol. +func parseUplink(kind int, msg []byte) (cloudwire.Envelope, error) { + if kind != websocket.BinaryMessage { + return cloudwire.Envelope{}, fmt.Errorf("%w: a text message", errProtocol) + } + payload, err := cloudwire.DecodeFrame(msg) + if err != nil { + return cloudwire.Envelope{}, fmt.Errorf("%w: %v", errProtocol, err) + } + env, err := cloudwire.DecodeEnvelope(payload) + if err != nil { + return cloudwire.Envelope{}, fmt.Errorf("%w: %v", errProtocol, err) + } + return env, nil +} + +// apply acts on one signed message from the box. stop ends the connection. +func (c *boxConn) apply(env cloudwire.Envelope) (stop bool, err error) { + if err := cloudwire.Verify(c.pub, c.home, env); err != nil { + return true, err + } + var h cloudwire.Header + if err := cloudwire.Unmarshal(env.B, &h); err != nil { + return true, err + } + if c.purged && env.T != cloudwire.MsgBye { + return true, fmt.Errorf("%w: %s from a purged home", errProtocol, env.T) + } + + switch env.T { + case cloudwire.MsgRoster: + return false, c.roster(env) + case cloudwire.MsgBye: + var b cloudwire.Bye + if err := cloudwire.Unmarshal(env.B, &b); err != nil { + return true, err + } + if b.Purge { + return true, c.purge() + } + return true, nil + case cloudwire.MsgHello: + return true, fmt.Errorf("%w: a second hello", errProtocol) + } + + if _, forwarded := cloudwire.ClassScope[env.T]; !forwarded { + c.u.Dropped.Add(1) + return false, nil + } + raw, err := env.Encode() + if err != nil { + return true, err + } + return false, c.handle.Put(env.T, raw, h.Seq, env.T == cloudwire.MsgLive && carriesTable(env.B)) +} + +// carriesTable reports whether a live body holds its source table. +func carriesTable(body []byte) bool { + var l struct { + Sources map[string]cbor.RawMessage `cbor:"sources"` + } + return cloudwire.Unmarshal(body, &l) == nil && len(l.Sources) > 0 +} + +// roster keeps a signed roster whose grant and scopes are within what FTW +// Cloud may hold. +func (c *boxConn) roster(env cloudwire.Envelope) error { + var r cloudwire.Roster + if err := cloudwire.Unmarshal(env.B, &r); err != nil { + return err + } + within := func(have, allowed []string) bool { + for _, s := range have { + if !slices.Contains(allowed, s) { + return false + } + } + return true + } + if len(r.Grant.ID) != cloudwire.GrantIDBytes || r.Grant.Controller == "" || r.TTLS <= 0 || + !within(r.Grant.Scopes, protocol.CloudScopes) || !within(r.Grant.Purposes, protocol.CloudPurposes) { + return fmt.Errorf("%w: a roster outside what FTW Cloud may hold", errProtocol) + } + for _, d := range r.Devices { + if len(d.Key) != 32 || d.ID == "" || !within(d.Scopes, protocol.CloudScopes) { + return fmt.Errorf("%w: a roster device the service cannot use", errProtocol) + } + } + if !c.handle.SetRoster(&r) { + return nil // older than what is held: ignored + } + raw, err := env.Encode() + if err != nil { + return err + } + return c.u.registry.SetRoster(c.home, r.RosterSeq, r.Grant.ID, raw, c.u.now()) +} + +// purge deletes the home and sends the signed receipt. For a home already +// purged, the receipt goes again. +func (c *boxConn) purge() error { + now := c.u.now() + var grantID []byte + if c.handle != nil { + if r, _ := c.handle.Roster(); r != nil { + grantID = r.Grant.ID + } + } + if h, ok, err := c.u.registry.Get(c.home); err == nil && ok && grantID == nil { + grantID = h.GrantID + } + if !c.purged { + if err := c.u.registry.Purge(c.home, grantID, now); err != nil { + return err + } + c.handle.Delete() + c.u.log.Info("home purged at its box's request") + } else if grantID == nil { + grantID = c.u.registry.TombstoneGrant(c.home) + } + return c.send(purged{AtMs: now.UnixMilli(), What: cloudwire.PurgedHome, GrantID: grantID}) +} diff --git a/internal/uplink/uplink_test.go b/internal/uplink/uplink_test.go new file mode 100644 index 0000000..bc6c62d --- /dev/null +++ b/internal/uplink/uplink_test.go @@ -0,0 +1,382 @@ +package uplink + +import ( + "bytes" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "io" + "log/slog" + "net/http/httptest" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw-cloud/internal/store" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +type service struct { + srv *httptest.Server + up *Uplink + store *store.Store + reg *registry.Registry + key *ecdsa.PrivateKey +} + +func newService(t *testing.T) *service { + t.Helper() + key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = reg.Close() }) + st := store.New(nil) + up := New(st, reg, key, slog.New(slog.NewTextHandler(io.Discard, nil))) + srv := httptest.NewUnstartedServer(up) + srv.TLS = &tls.Config{MinVersion: tls.VersionTLS13} + srv.StartTLS() + t.Cleanup(srv.Close) + return &service{srv: srv, up: up, store: st, reg: reg, key: key} +} + +// box is a box's half of the uplink, enough to drive the service. +type box struct { + t *testing.T + key *ecdsa.PrivateKey + pk []byte + home string + boot []byte + seq uint64 + conn *websocket.Conn +} + +func newBox(t *testing.T) *box { + key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + pk, _ := cloudwire.PublicKeyBytesOf(&key.PublicKey) + return &box{t: t, key: key, pk: pk, home: cloudwire.HomeID(pk), boot: bytes.Repeat([]byte{3}, 16)} +} + +func (b *box) header() cloudwire.Header { + b.seq++ + return cloudwire.Header{Seq: b.seq, Boot: b.boot, UptimeMs: int64(b.seq) * 1000} +} + +// connect runs the handshake. ekmOverride replaces the box's exporter value +// in hello, as a proxy relaying the box's hello would have to. +func (b *box) connect(s *service, ekmOverride []byte) (cloudwire.HelloOK, error) { + b.t.Helper() + pool := x509.NewCertPool() + pool.AddCert(s.srv.Certificate()) + d := websocket.Dialer{TLSClientConfig: &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS13}} + conn, _, err := d.Dial("wss://"+s.srv.Listener.Addr().String()+"/v1/box", nil) + if err != nil { + b.t.Fatal(err) + } + b.conn = conn + b.t.Cleanup(func() { _ = conn.Close() }) + cs := conn.NetConn().(*tls.Conn).ConnectionState() + ekm, _ := cs.ExportKeyingMaterial(cloudwire.EKMLabel, nil, 32) + + env := b.read() + var ch cloudwire.Challenge + if env.T != cloudwire.MsgChallenge || cloudwire.Unmarshal(env.B, &ch) != nil || len(env.Sig) != 0 { + b.t.Fatalf("first message %q", env.T) + } + if ekmOverride != nil { + ekm = ekmOverride + } + boxNonce := bytes.Repeat([]byte{4}, 32) + b.send(cloudwire.MsgHello, cloudwire.Hello{Header: b.header(), V: 1, PK: b.pk, Nonce: ch.Nonce, + EKM: ekm, BoxNonce: boxNonce, Build: "test", Clock: cloudwire.ClockNTP, RosterSeq: 1, + GrantID: bytes.Repeat([]byte{5}, 16)}) + + _ = conn.SetReadDeadline(time.Now().Add(3 * time.Second)) + _, msg, err := conn.ReadMessage() + if err != nil { + return cloudwire.HelloOK{}, err + } + env, _ = parseUplink(websocket.BinaryMessage, msg) + var ok cloudwire.HelloOK + if env.T != cloudwire.MsgHelloOK || cloudwire.Verify(&s.key.PublicKey, b.home, env) != nil || + cloudwire.Unmarshal(env.B, &ok) != nil || !bytes.Equal(ok.BoxNonce, boxNonce) || !bytes.Equal(ok.EKM, ekm) { + b.t.Fatalf("hello_ok did not check out") + } + return ok, nil +} + +func (b *box) read() cloudwire.Envelope { + b.t.Helper() + _ = b.conn.SetReadDeadline(time.Now().Add(3 * time.Second)) + kind, msg, err := b.conn.ReadMessage() + if err != nil { + b.t.Fatalf("read: %v", err) + } + env, err := parseUplink(kind, msg) + if err != nil { + b.t.Fatal(err) + } + return env +} + +// send seals and writes one message and returns its envelope bytes. +func (b *box) send(t string, body any) []byte { + return b.sendAs(b.key, t, body) +} + +func (b *box) sendAs(key *ecdsa.PrivateKey, t string, body any) []byte { + env, err := cloudwire.Seal(key, b.home, t, body) + if err != nil { + b.t.Fatal(err) + } + raw, _ := env.Encode() + frame, _ := cloudwire.EncodeFrame(raw) + if err := b.conn.WriteMessage(websocket.BinaryMessage, frame); err != nil { + b.t.Fatal(err) + } + return raw +} + +// closed waits for the service to hang up. +func (b *box) closed() bool { + _ = b.conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + for { + if _, _, err := b.conn.ReadMessage(); err != nil { + return !strings.Contains(err.Error(), "timeout") + } + } +} + +var ownerKey = bytes.Repeat([]byte{7}, 32) + +func (b *box) roster(scopes ...string) cloudwire.Roster { + return cloudwire.Roster{Header: b.header(), RosterSeq: 1, TTLS: 86_400, + Devices: []cloudwire.Device{{ID: "owner", Key: ownerKey, Role: "owner", Scopes: protocol.CloudScopes, Epoch: 1}}, + Grant: cloudwire.Grant{ID: bytes.Repeat([]byte{5}, 16), Scopes: scopes, + Purposes: []string{protocol.CloudPurposeServe}, Controller: cloudwire.ControllerSourceful}} +} + +// wait polls until cond holds, or fails. +func wait(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(10 * time.Millisecond) + } +} + +func snapshot(t *testing.T, s *service, home string) [][]byte { + t.Helper() + dev, err := s.store.For(store.Principal{Kind: store.Device, Home: home, DeviceKey: ownerKey}) + if err != nil { + return nil + } + snap, _ := dev.Snapshot() + return snap.Envelopes +} + +// The service holds the box's bytes, unchanged, and serves them by scope. +func TestTheServiceHoldsExactlyWhatTheBoxSent(t *testing.T) { + s := newService(t) + b := newBox(t) + if _, err := b.connect(s, nil); err != nil { + t.Fatal(err) + } + b.send(cloudwire.MsgRoster, b.roster(protocol.ScopeLiveRead)) + live := b.send(cloudwire.MsgLive, cloudwire.Live{Header: b.header(), Fields: map[string]int64{"2": 1200}, + Sources: map[string]cloudwire.Source{"x": {Kind: "driver", State: "live", StaleAfterMs: 10_000}}, SrcRev: 1}) + plan := b.send(cloudwire.MsgPlan, map[string]any{"seq": b.header().Seq, "decisionId": "d1"}) + hb := b.send(cloudwire.MsgHB, cloudwire.HB{Header: b.header(), Clock: cloudwire.ClockNTP, SrcRev: 1, LiveSeq: 3}) + + wait(t, "the hb", func() bool { return len(snapshot(t, s, b.home)) == 2 }) + got := snapshot(t, s, b.home) + if !bytes.Equal(got[0], live) || !bytes.Equal(got[1], hb) { + t.Fatal("the service changed the box's bytes") + } + for _, e := range got { + if bytes.Equal(e, plan) { + t.Fatal("plan was served without ftw.plan.read in the grant") + } + } + if h, ok, _ := s.reg.Get(b.home); !ok || h.RosterSeq != 1 || !bytes.Equal(h.PK, b.pk) { + t.Fatalf("registry %+v", h) + } +} + +// A hello that carries another TLS session's exporter value is a hello +// relayed through a proxy. The service closes, and no home exists. +func TestAHelloFromAnotherSessionIsRefused(t *testing.T) { + s := newService(t) + b := newBox(t) + if _, err := b.connect(s, bytes.Repeat([]byte{1}, 32)); err == nil { + t.Fatal("the service answered a hello bound to another session") + } + if _, ok, _ := s.reg.Get(b.home); ok { + t.Fatal("a refused hello claimed a home") + } +} + +func TestAMessageNotSignedByTheBoxCloses(t *testing.T) { + s := newService(t) + b := newBox(t) + _, _ = b.connect(s, nil) + b.send(cloudwire.MsgRoster, b.roster(protocol.ScopeLiveRead)) + other, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + b.sendAs(other, cloudwire.MsgLive, cloudwire.Live{Header: b.header(), Fields: map[string]int64{"2": 9999}}) + if !b.closed() { + t.Fatal("the service kept a connection that sent a forged message") + } + if got := snapshot(t, s, b.home); len(got) != 0 { + t.Fatal("a forged live was stored") + } +} + +func TestARosterBeyondFTWCloudIsRefused(t *testing.T) { + s := newService(t) + b := newBox(t) + _, _ = b.connect(s, nil) + b.send(cloudwire.MsgRoster, b.roster(protocol.ScopeLiveRead, protocol.ScopeDispatchWrite)) + if !b.closed() { + t.Fatal("a roster granting a write scope was kept") + } +} + +// bye with purge deletes the home and returns a receipt signed with the +// service key, which a box whose first receipt was lost can ask for again. +func TestPurgeDeletesTheHomeAndSignsAReceipt(t *testing.T) { + s := newService(t) + b := newBox(t) + _, _ = b.connect(s, nil) + r := b.roster(protocol.ScopeLiveRead) + b.send(cloudwire.MsgRoster, r) + b.send(cloudwire.MsgLive, cloudwire.Live{Header: b.header(), Fields: map[string]int64{"2": 1}}) + wait(t, "the live", func() bool { return len(snapshot(t, s, b.home)) == 1 }) + + b.send(cloudwire.MsgBye, cloudwire.Bye{Header: b.header(), Why: cloudwire.ByeRevoked, Purge: true}) + receipt := func() { + env := b.read() + var p cloudwire.Purged + if env.T != cloudwire.MsgPurged || cloudwire.Verify(&s.key.PublicKey, b.home, env) != nil || + cloudwire.Unmarshal(env.B, &p) != nil || p.What != cloudwire.PurgedHome || !bytes.Equal(p.GrantID, r.Grant.ID) { + t.Fatalf("receipt %q does not check out", env.T) + } + } + receipt() + if _, ok, _ := s.reg.Get(b.home); ok { + t.Fatal("the registry kept the home") + } + if _, err := s.store.For(store.Principal{Kind: store.Device, Home: b.home, DeviceKey: ownerKey}); err == nil { + t.Fatal("memory kept the home") + } + + _ = b.conn.Close() + if _, err := b.connect(s, nil); err != nil { + t.Fatal(err) + } + b.send(cloudwire.MsgBye, cloudwire.Bye{Header: b.header(), Why: cloudwire.ByeRevoked, Purge: true}) + receipt() + + // A purged home's box may say only bye. + _ = b.conn.Close() + _, _ = b.connect(s, nil) + b.send(cloudwire.MsgLive, cloudwire.Live{Header: b.header()}) + if !b.closed() { + t.Fatal("a purged home's box sent readings and stayed") + } +} + +func TestAnUnknownBoxMessageIsCountedNotFatal(t *testing.T) { + s := newService(t) + b := newBox(t) + _, _ = b.connect(s, nil) + b.send(cloudwire.MsgRoster, b.roster(protocol.ScopeLiveRead)) + b.send("stats.future", map[string]any{"seq": b.header().Seq}) + b.send(cloudwire.MsgLive, cloudwire.Live{Header: b.header(), Fields: map[string]int64{"2": 1}}) + wait(t, "the live after the stranger", func() bool { return len(snapshot(t, s, b.home)) == 1 }) + if s.up.Dropped.Load() != 1 { + t.Fatalf("dropped %d", s.up.Dropped.Load()) + } +} + +func TestASecondUplinkReplacesTheFirst(t *testing.T) { + s := newService(t) + first := newBox(t) + _, _ = first.connect(s, nil) + second := *first + second.conn = nil + if _, err := second.connect(s, nil); err != nil { + t.Fatal(err) + } + if !first.closed() { + t.Fatal("two uplinks for one home") + } +} + +// The service can say five things to a box, and they are the protocol's five. +func TestDownlinkIsTheProtocolsClosedSet(t *testing.T) { + s := newService(t) + all := []down{challenge{V: 1, Nonce: make([]byte, 32)}, helloOK{Have: map[string]string{}}, + watch{Hz: 1}, resync{What: []string{cloudwire.ResyncFull}}, purged{What: cloudwire.PurgedHome}} + var names []string + for _, m := range all { + frame, err := s.up.encodeDown("home", m) + if err != nil { + t.Fatal(err) + } + if len(frame) > cloudwire.MaxServiceFrame { + t.Fatalf("%d bytes", len(frame)) + } + env, err := parseUplink(websocket.BinaryMessage, frame) + if err != nil { + t.Fatal(err) + } + _, _, signed := m.down() + if signed != (cloudwire.Verify(&s.key.PublicKey, "home", env) == nil) { + t.Fatalf("%s: signed=%v", env.T, signed) + } + names = append(names, env.T) + } + if !slices.Equal(names, cloudwire.Downlink) { + t.Fatalf("downlink %v, protocol %v", names, cloudwire.Downlink) + } +} + +// Whatever a box with a valid key signs, the service stores it, ignores it +// or closes, and never panics. +func FuzzBoxMessage(f *testing.F) { + key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + pk, _ := cloudwire.PublicKeyBytesOf(&key.PublicKey) + home := cloudwire.HomeID(pk) + for _, t := range []string{cloudwire.MsgLive, cloudwire.MsgRoster, cloudwire.MsgBye, cloudwire.MsgHB, "x"} { + body, _ := cloudwire.Marshal(map[string]any{"seq": 1, "boot": []byte{1}, "purge": true, "rosterSeq": 1}) + f.Add(t, body) + } + f.Fuzz(func(t *testing.T, typ string, body []byte) { + reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db")) + if err != nil { + t.Fatal(err) + } + defer reg.Close() + st := store.New(nil) + u := New(st, reg, key, slog.New(slog.NewTextHandler(io.Discard, nil))) + handle, _ := st.For(store.Principal{Kind: store.Box, Home: home}) + _ = reg.Claim(home, pk, time.Now()) + c := &boxConn{u: u, home: home, pub: &key.PublicKey, handle: handle, write: func([]byte) error { return nil }} + env, err := cloudwire.SealBytes(key, home, typ, body) + if err != nil || typ == "" { + return + } + _, _ = c.apply(env) + }) +} diff --git a/scripts/check-imports.sh b/scripts/check-imports.sh new file mode 100755 index 0000000..fc0315a --- /dev/null +++ b/scripts/check-imports.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +# FTW Cloud may import the wire format from srcfl/ftw and nothing else of it. +# go/protocol imports nothing from the box (srcfl/ftw checks that), so this +# keeps every line of control, planner and storage code out of this build. + +set -euo pipefail + +reached=$(go list -deps -test -f '{{.ImportPath}}' ./... | awk ' + $1 ~ /^github\.com\/srcfl\/ftw\// && + $1 !~ /^github\.com\/srcfl\/ftw\/go\/protocol(\/|$)/ { print $1 } +' | sort -u) + +if [ -n "${reached}" ]; then + echo "ftw-cloud may import only github.com/srcfl/ftw/go/protocol. It reaches:" >&2 + printf '%s\n' "${reached}" >&2 + exit 1 +fi + +echo "ftw-cloud: only go/protocol from srcfl/ftw ($(go list -deps ./... | grep -c '^github.com/srcfl/ftw/') packages)"