From 30eb0a8c85d8016eaa32528ab0e78f44f2ef6ef5 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 13 Oct 2026 08:14:11 +0200 Subject: [PATCH 1/5] feat: alerts through APNs, signed by the box (FTW Cloud C4) A box's signed event now reaches its phones through Apple, wrapped unchanged; the phone checks the signature, writes its own words and picks the buttons from the verified kind. The service chooses none of that. - internal/apns: HTTP/2 to Apple's provider API with an ES256 token from the team's .p8, renewed every 40 minutes and once on ExpiredProviderToken. 410, BadDeviceToken and DeviceTokenNotForTopic mark a token gone. Every alert has the same push type and priority and no collapse id. - internal/registry: a push table bound to roster entries (home, device id, epoch), cascading with the home on purge and expiry, and expiring 30 days after the box's last roster. - internal/push: registration (only devices whose effective scope reads events), roster pruning, fan-out with an hourly cap per home, and the dead man's switch: box.unreachable after ten minutes away, held back for an hour after bye{why:"update"}, never for a revoke. - uplink: OnConnect (as C3 has it), OnEvent, OnRoster and OnGone hooks. - ftwcloudd: -apns-dir; without a key nothing is sent. Development builds get -dev-push-listen to register a token by hand before the app session can. - deploy: apns.sh reads the key from Parameter Store into tmpfs at boot; the host role can already read it. apns.md is the owner's steps. Pins go/protocol at srcfl/ftw 8cca4203. Co-Authored-By: Claude Opus 5.5 --- README.md | 18 +- cmd/ftwcloudd/devpush_dev.go | 101 ++++++++++ cmd/ftwcloudd/devpush_other.go | 17 ++ cmd/ftwcloudd/main.go | 59 +++++- deploy/apns.md | 61 ++++++ deploy/apns.sh | 64 ++++++ deploy/compose.yml | 4 + go.mod | 2 +- go.sum | 2 + internal/apns/apns.go | 266 ++++++++++++++++++++++++ internal/apns/apns_test.go | 236 ++++++++++++++++++++++ internal/push/push.go | 314 +++++++++++++++++++++++++++++ internal/push/push_test.go | 357 +++++++++++++++++++++++++++++++++ internal/registry/push.go | 135 +++++++++++++ internal/registry/push_test.go | 111 ++++++++++ internal/registry/registry.go | 6 +- internal/uplink/hooks_test.go | 108 ++++++++++ internal/uplink/uplink.go | 35 +++- 18 files changed, 1888 insertions(+), 8 deletions(-) create mode 100644 cmd/ftwcloudd/devpush_dev.go create mode 100644 cmd/ftwcloudd/devpush_other.go create mode 100644 deploy/apns.md create mode 100755 deploy/apns.sh create mode 100644 internal/apns/apns.go create mode 100644 internal/apns/apns_test.go create mode 100644 internal/push/push.go create mode 100644 internal/push/push_test.go create mode 100644 internal/registry/push.go create mode 100644 internal/registry/push_test.go create mode 100644 internal/uplink/hooks_test.go diff --git a/README.md b/README.md index cbd5fca..4b4e07c 100644 --- a/README.md +++ b/README.md @@ -22,13 +22,21 @@ both in `srcfl/ftw`. Both are proposed; nothing here is deployed. - 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. +- **Alerts (C4).** A box's signed `event` goes to its phones through APNs, + wrapped unchanged: the phone checks the signature, writes its own words and + picks the buttons from the verified kind. Every alert has the same aps, + priority and push type and no collapse id. Ten minutes without the box's + socket sends `box.unreachable`, unless the box said `bye` for an update, + which holds it back for an hour. Tokens are bound to roster entries that + may read `ftw.events.read` and go when the roster drops them or Apple says + they are gone. Apple's key comes from Parameter Store + ([deploy/apns.md](deploy/apns.md)). 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 | @@ -83,8 +91,12 @@ 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. +the service holds so a test can compare it with the preview, and +`-dev-push-listen 127.0.0.1:18444`, where `POST /dev/push/register` with +`{home, deviceId, apnsToken (hex), environment}` registers a token for a +device in the home's roster, as the app session will. `-apns-dir` names a +directory holding `key.p8`, `key-id` and `team-id`; without it nothing is +sent. Release builds have neither dev route. ## Develop diff --git a/cmd/ftwcloudd/devpush_dev.go b/cmd/ftwcloudd/devpush_dev.go new file mode 100644 index 0000000..eb39653 --- /dev/null +++ b/cmd/ftwcloudd/devpush_dev.go @@ -0,0 +1,101 @@ +//go:build ftwclouddev + +package main + +import ( + "context" + "encoding/hex" + "encoding/json" + "errors" + "log/slog" + "net" + "net/http" + "slices" + "time" + + "github.com/srcfl/ftw-cloud/internal/push" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// serveDevPush listens on loopback for POST /dev/push/register, which +// registers an APNs token for a device in a home's signed roster, as the +// app's session will. It lets an end-to-end test run before the app session +// can. Release builds do not have it. +func serveDevPush(ctx context.Context, addr string, reg *registry.Registry, svc *push.Service, log *slog.Logger) error { + host, _, err := net.SplitHostPort(addr) + if err != nil || !net.ParseIP(host).IsLoopback() { + return errors.New("-dev-push-listen must be a loopback address") + } + mux := http.NewServeMux() + mux.HandleFunc("POST /dev/push/register", func(w http.ResponseWriter, r *http.Request) { + var req struct { + Home string `json:"home"` + DeviceID string `json:"deviceId"` + Token string `json:"apnsToken"` + Environment string `json:"environment"` + } + if json.NewDecoder(r.Body).Decode(&req) != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + token, err := hex.DecodeString(req.Token) + if err != nil { + http.Error(w, "apnsToken is hex", http.StatusBadRequest) + return + } + dev, scopes, ok := rosterDevice(reg, req.Home, req.DeviceID) + if !ok { + http.Error(w, "no such device in the home's roster", http.StatusNotFound) + return + } + if err := svc.Register(req.Home, dev, scopes, cloudwire.PushRegister{ + APNsToken: token, Environment: req.Environment, Kind: cloudwire.PushKindAlert}); err != nil { + http.Error(w, err.Error(), http.StatusForbidden) + return + } + w.WriteHeader(http.StatusNoContent) + }) + srv := &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second} + go func() { + <-ctx.Done() + _ = srv.Close() + }() + go func() { + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Warn("dev push route stopped", "err", err) + } + }() + log.Warn("development build: tokens can be registered by hand", "addr", addr) + return nil +} + +// rosterDevice finds a device in the roster the box last signed, with its +// effective scope. +func rosterDevice(reg *registry.Registry, home, id string) (cloudwire.Device, []string, bool) { + h, ok, err := reg.Get(home) + if err != nil || !ok || h.Roster == nil { + return cloudwire.Device{}, nil, false + } + env, err := cloudwire.DecodeEnvelope(h.Roster) + if err != nil { + return cloudwire.Device{}, nil, false + } + var r cloudwire.Roster + if cloudwire.Unmarshal(env.B, &r) != nil { + return cloudwire.Device{}, nil, false + } + for _, d := range r.Devices { + if d.ID != id { + continue + } + var scopes []string + for _, s := range d.Scopes { + if slices.Contains(r.Grant.Scopes, s) { + scopes = append(scopes, s) + } + } + return d, scopes, true + } + return cloudwire.Device{}, nil, false +} diff --git a/cmd/ftwcloudd/devpush_other.go b/cmd/ftwcloudd/devpush_other.go new file mode 100644 index 0000000..1a822d6 --- /dev/null +++ b/cmd/ftwcloudd/devpush_other.go @@ -0,0 +1,17 @@ +//go:build !ftwclouddev + +package main + +import ( + "context" + "errors" + "log/slog" + + "github.com/srcfl/ftw-cloud/internal/push" + "github.com/srcfl/ftw-cloud/internal/registry" +) + +// serveDevPush exists only in development builds. +func serveDevPush(context.Context, string, *registry.Registry, *push.Service, *slog.Logger) error { + return errors.New("-dev-push-listen needs a build with -tags ftwclouddev") +} diff --git a/cmd/ftwcloudd/main.go b/cmd/ftwcloudd/main.go index e758bd8..01ba3f0 100644 --- a/cmd/ftwcloudd/main.go +++ b/cmd/ftwcloudd/main.go @@ -25,9 +25,12 @@ import ( "os" "os/signal" "path/filepath" + "strings" "syscall" "time" + "github.com/srcfl/ftw-cloud/internal/apns" + "github.com/srcfl/ftw-cloud/internal/push" "github.com/srcfl/ftw-cloud/internal/registry" "github.com/srcfl/ftw-cloud/internal/server" "github.com/srcfl/ftw-cloud/internal/store" @@ -59,6 +62,9 @@ func serve(args []string) error { 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") + apnsDir := fs.String("apns-dir", "/run/secrets/apns", "directory with Apple's key.p8, key-id and team-id; without it no alert is sent") + apnsTopic := fs.String("apns-topic", "energy.ftw.app", "the app's bundle id") + devPush := fs.String("dev-push-listen", "", "loopback address for registering a token by hand (development builds only)") _ = fs.Parse(args) harden() @@ -74,7 +80,26 @@ func serve(args []string) error { defer reg.Close() st := store.New(nil) - srv := &server.Server{Store: st, Registry: reg, Uplink: uplink.New(st, reg, key, log), Log: log} + up := uplink.New(st, reg, key, log) + srv := &server.Server{Store: st, Registry: reg, Uplink: up, Log: log} + + // Alerts: the box's signed events to its phones through Apple, and + // box.unreachable when a box is away. Without a key the service keeps + // registrations and sends nothing. + var sender push.Sender + switch client, err := loadAPNs(*apnsDir, *apnsTopic); { + case err != nil: + return err + case client == nil: + log.Warn("APNs off: no key in " + *apnsDir) + default: + sender = client + } + pushSvc := push.New(reg, sender, log) + up.OnConnect = pushSvc.BoxConnected + up.OnEvent = pushSvc.Event + up.OnRoster = pushSvc.Roster + up.OnGone = pushSvc.BoxGone // 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 @@ -111,6 +136,12 @@ func serve(args []string) error { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() go srv.Upkeep(ctx, time.Minute) + go pushSvc.Run(ctx) + if *devPush != "" { + if err := serveDevPush(ctx, *devPush, reg, pushSvc, log); err != nil { + return err + } + } go func() { <-ctx.Done() shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second) @@ -128,6 +159,32 @@ func serve(args []string) error { func log0() *log.Logger { return log.New(io.Discard, "", 0) } +// loadAPNs reads Apple's key from dir: key.p8, key-id and team-id, as the +// host writes them from SSM Parameter Store at boot. No directory or no key +// is no sender; a key that is there and wrong is an error. +func loadAPNs(dir, topic string) (*apns.Client, error) { + p8, err := os.ReadFile(filepath.Join(dir, "key.p8")) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("apns key: %w", err) + } + read := func(name string) (string, error) { + b, err := os.ReadFile(filepath.Join(dir, name)) + return strings.TrimSpace(string(b)), err + } + keyID, err := read("key-id") + if err != nil { + return nil, fmt.Errorf("apns key id: %w", err) + } + teamID, err := read("team-id") + if err != nil { + return nil, fmt.Errorf("apns team id: %w", err) + } + return apns.New(apns.Config{Key: p8, KeyID: keyID, TeamID: teamID, Topic: topic}) +} + func loadServiceKey(path string) (*ecdsa.PrivateKey, error) { raw, err := os.ReadFile(path) if err != nil { diff --git a/deploy/apns.md b/deploy/apns.md new file mode 100644 index 0000000..85f6790 --- /dev/null +++ b/deploy/apns.md @@ -0,0 +1,61 @@ +# Apple's push key + +`ftwcloudd` sends alerts through APNs with a token key (`.p8`) from Sourceful's +Apple team. Only the owner handles the key: he makes it at Apple and puts it +in AWS Systems Manager Parameter Store himself. The host reads it from there +into tmpfs at every boot. It is never in the repository, the image, the volume +or a log. + +## 1. Make the key (owner, once) + +1. [developer.apple.com/account](https://developer.apple.com/account) → + Certificates, Identifiers & Profiles → **Keys** → **+**. +2. Name it `FTW Cloud APNs`, tick **Apple Push Notifications service (APNs)**, + and choose **Configure**: environment **Sandbox & Production**; key + restriction **Topic Specific** with `energy.ftw.app`, or **Team Scoped**. + Save, then **Continue** → **Register**. +3. **Download** `AuthKey_.p8`. Apple lets you download it once. Note the + 10-character **Key ID**. The Team ID is `L66M368PZ6`. +4. Under **Identifiers** → `energy.ftw.app`, check that **Push Notifications** + and **Time Sensitive Notifications** are on. Xcode turns them on the first + time it signs the app with the new entitlements. + +## 2. Put it in Parameter Store (owner, once) + +In a terminal signed in to the FTW Cloud account (976075258475), with the +file where Safari saved it: + +```bash +aws ssm put-parameter --region eu-north-1 --name /ftw-cloud/apns/key --type SecureString --value "file://$HOME/Downloads/AuthKey_KEYID.p8" +aws ssm put-parameter --region eu-north-1 --name /ftw-cloud/apns/key-id --type String --value KEYID +aws ssm put-parameter --region eu-north-1 --name /ftw-cloud/apns/team-id --type String --value L66M368PZ6 +``` + +Then keep the `.p8` in the password manager or delete it. A new key replaces +an old one with `--overwrite`; revoke the old one at Apple afterwards. + +## 3. Give it to the host (once per host) + +`deploy/apns.sh` installs `ftw-cloud-apns.service`, which reads the three +parameters into `/run/ftw-cloud/apns/` (tmpfs, readable only by the +container's user) before Docker starts. `compose.yml` mounts that directory +read-only. Run it through SSM, then redeploy so a new container mounts the +key: + +```bash +aws ssm send-command --region eu-north-1 --instance-ids i-0ecac4cbc6b5f8adc \ + --document-name AWS-RunShellScript \ + --parameters 'commands=["bash /opt/ftw-cloud/deploy/apns.sh","/usr/local/sbin/ftw-cloud-deploy"]' +``` + +The log then has no `APNs off` line. Without the parameters the service runs +and sends no alerts. + +## Access + +The host role needs nothing new. `AmazonSSMManagedInstanceCore` allows +`ssm:GetParameter` on every parameter, and the account's `aws/ssm` key allows +decrypting through SSM for every principal in the account. Checked on +2026-10-13 with `iam simulate-principal-policy` (`ssm:GetParameter` allowed) +and the key policy of `alias/aws/ssm` in eu-north-1. Secrets Manager would +need a policy change; Parameter Store does not. diff --git a/deploy/apns.sh b/deploy/apns.sh new file mode 100755 index 0000000..ee9041d --- /dev/null +++ b/deploy/apns.sh @@ -0,0 +1,64 @@ +#!/bin/bash +# Give ftwcloudd Apple's push key. Run as root on the FTW Cloud host through +# SSM, after the owner has put the key in Parameter Store (deploy/apns.md). +# Safe to re-run. +# +# The key rests only in Parameter Store, as a SecureString under the +# account's aws/ssm key. At every boot, before Docker starts, a unit reads it +# into tmpfs, where compose.yml mounts it into the container. It is never +# written to the volume or the image, and nothing prints it. +# +# The host role reads it with what it already has: AmazonSSMManagedInstanceCore +# allows ssm:GetParameter, and the aws/ssm key's policy allows decrypting +# through SSM for every principal in the account. No IAM change is needed. +set -euo pipefail + +cat > /usr/local/sbin/ftw-cloud-apns <<'FETCH' +#!/bin/bash +# Read Apple's push key from Parameter Store into tmpfs for ftwcloudd. +# Without the parameters ftwcloudd runs and sends no alerts. +set -euo pipefail +export AWS_PAGER="" +region=eu-north-1 +dir=/run/ftw-cloud/apns +umask 077 +install -d -m 0700 /run/ftw-cloud +rm -rf "$dir.new" +install -d -m 0700 "$dir.new" +get() { + aws ssm get-parameter --region "$region" --name "$1" --with-decryption \ + --query Parameter.Value --output text > "$dir.new/$2" +} +if get /ftw-cloud/apns/key key.p8 && get /ftw-cloud/apns/key-id key-id && get /ftw-cloud/apns/team-id team-id; then + chown -R 65532:65532 "$dir.new" + chmod 0400 "$dir.new"/* + chmod 0500 "$dir.new" + rm -rf "$dir" + mv "$dir.new" "$dir" + echo "APNs key in place" +else + rm -rf "$dir.new" + echo "no APNs key in Parameter Store; ftwcloudd will send no alerts" +fi +FETCH +chmod 0755 /usr/local/sbin/ftw-cloud-apns + +cat > /etc/systemd/system/ftw-cloud-apns.service <<'UNIT' +[Unit] +Description=FTW Cloud APNs key into tmpfs +Before=docker.service +Wants=network-online.target +After=network-online.target + +[Service] +Type=oneshot +RemainAfterExit=yes +ExecStart=/usr/local/sbin/ftw-cloud-apns + +[Install] +WantedBy=docker.service +UNIT + +systemctl daemon-reload +systemctl enable ftw-cloud-apns.service +systemctl restart ftw-cloud-apns.service diff --git a/deploy/compose.yml b/deploy/compose.yml index d632f35..fd359ef 100644 --- a/deploy/compose.yml +++ b/deploy/compose.yml @@ -20,9 +20,13 @@ services: - -state=/state - -service-key=/run/secrets/ftw-cloud-service.key - -acme-email=${FTW_CLOUD_ACME_EMAIL:-} + - -apns-dir=/run/secrets/apns volumes: # The registry and certificates. No readings are ever written here. - ftw-cloud-state:/state + # Apple's push key, read from Parameter Store into tmpfs at boot by + # deploy/apns.sh. Empty means no alerts, and the service runs. + - /run/ftw-cloud/apns:/run/secrets/apns:ro secrets: - ftw-cloud-service.key read_only: true diff --git a/go.mod b/go.mod index 2c89de0..7934f39 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ 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 + github.com/srcfl/ftw/go/protocol v0.0.0-20261013054921-8cca42036665 golang.org/x/crypto v0.57.0 golang.org/x/sys v0.48.0 modernc.org/sqlite v1.59.0 diff --git a/go.sum b/go.sum index 70f63f7..ebda1f8 100644 --- a/go.sum +++ b/go.sum @@ -18,6 +18,8 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94 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/srcfl/ftw/go/protocol v0.0.0-20261013054921-8cca42036665 h1:xN63Svp0XMecqe5N5daGFM8gsF7vo095Z+qk/1WhKeM= +github.com/srcfl/ftw/go/protocol v0.0.0-20261013054921-8cca42036665/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= diff --git a/internal/apns/apns.go b/internal/apns/apns.go new file mode 100644 index 0000000..d16b37a --- /dev/null +++ b/internal/apns/apns.go @@ -0,0 +1,266 @@ +// Package apns sends alerts to Apple Push Notification service: HTTP/2 to +// Apple's provider API, authenticated with a token (a JWT signed with ES256 +// by the team's .p8 key) that is renewed every 40 minutes. +// +// It knows nothing about homes or events. The caller hands it a device +// token and the payload bytes, and gets back what Apple said, including +// whether the token is gone for good. +package apns + +import ( + "bytes" + "context" + "crypto/ecdsa" + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "encoding/base64" + "encoding/hex" + "encoding/json" + "encoding/pem" + "errors" + "fmt" + "io" + "net/http" + "strconv" + "sync" + "time" +) + +// Apple's hosts. Sandbox serves development builds of the app; production +// serves TestFlight and the App Store. +const ( + ProductionURL = "https://api.push.apple.com" + SandboxURL = "https://api.sandbox.push.apple.com" +) + +// Environments, as a phone names them when it registers. +const ( + Sandbox = "sandbox" + Production = "production" +) + +// TokenLife is how long one provider token is used. Apple refuses a token +// older than an hour and one renewed more often than every 20 minutes. +const TokenLife = 40 * time.Minute + +// requestTimeout bounds one request. +const requestTimeout = 10 * time.Second + +// Config is one team key and the app it sends for. +type Config struct { + // Key is the .p8 file Apple issued: a PKCS #8 P-256 key in PEM. + Key []byte + // KeyID is the key's 10-character id; TeamID the team's. + KeyID string + TeamID string + // Topic is the app's bundle id. + Topic string + + // Seams for tests. Production leaves them zero. + SandboxURL string + ProductionURL string + HTTPClient *http.Client + Now func() time.Time +} + +// Client sends to Apple. It is safe for concurrent use and keeps one HTTP/2 +// connection per host. +type Client struct { + cfg Config + key *ecdsa.PrivateKey + http *http.Client + now func() time.Time + + mu sync.Mutex + token string + tokenAt time.Time +} + +// New parses the key and returns a client. +func New(cfg Config) (*Client, error) { + key, err := ParseKey(cfg.Key) + if err != nil { + return nil, err + } + if len(cfg.KeyID) != 10 || len(cfg.TeamID) != 10 || cfg.Topic == "" { + return nil, errors.New("apns: a 10-character key id, a 10-character team id and a topic are required") + } + if cfg.SandboxURL == "" { + cfg.SandboxURL = SandboxURL + } + if cfg.ProductionURL == "" { + cfg.ProductionURL = ProductionURL + } + if cfg.Now == nil { + cfg.Now = time.Now + } + client := cfg.HTTPClient + if client == nil { + client = &http.Client{Transport: &http.Transport{ + TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12}, + ForceAttemptHTTP2: true, + IdleConnTimeout: time.Hour, + }} + } + return &Client{cfg: cfg, key: key, http: client, now: cfg.Now}, nil +} + +// ParseKey reads a .p8: PEM around a PKCS #8 P-256 key. +func ParseKey(p8 []byte) (*ecdsa.PrivateKey, error) { + block, _ := pem.Decode(p8) + if block == nil || block.Type != "PRIVATE KEY" { + return nil, errors.New("apns: the key is not a PEM private key") + } + parsed, err := x509.ParsePKCS8PrivateKey(block.Bytes) + if err != nil { + return nil, fmt.Errorf("apns: %w", err) + } + key, ok := parsed.(*ecdsa.PrivateKey) + if !ok || key.Curve.Params().Name != "P-256" { + return nil, errors.New("apns: the key is not a P-256 key") + } + return key, nil +} + +// Notification is one push. +type Notification struct { + // Environment is Sandbox or Production. + Environment string + // DeviceToken is the token the phone got from iOS. + DeviceToken []byte + Payload []byte + // Expiration is when Apple may stop trying to deliver. Zero means + // deliver now or not at all. + Expiration time.Time +} + +// Result is Apple's answer. +type Result struct { + // Status is the HTTP status; 200 is delivered to Apple. + Status int + // Reason is Apple's reason string when Status is not 200. + Reason string + // Gone means the token no longer names a phone for this app: forget it. + Gone bool + // GoneSince is when Apple last knew the token was valid, for a 410. + GoneSince time.Time +} + +// OK reports whether Apple took the push. +func (r Result) OK() bool { return r.Status == http.StatusOK } + +// Send posts one alert. Every alert has the same priority and push type, +// and no collapse id, so nothing in the request says what kind it is. +func (c *Client) Send(ctx context.Context, n Notification) (Result, error) { + base := c.cfg.ProductionURL + switch n.Environment { + case Production: + case Sandbox: + base = c.cfg.SandboxURL + default: + return Result{}, fmt.Errorf("apns: unknown environment %q", n.Environment) + } + if len(n.DeviceToken) == 0 { + return Result{}, errors.New("apns: no device token") + } + res, err := c.post(ctx, base, n, false) + if err == nil && (res.Reason == "ExpiredProviderToken" || res.Reason == "InvalidProviderToken") { + // A token Apple no longer takes: make a new one and try once more. + res, err = c.post(ctx, base, n, true) + } + return res, err +} + +func (c *Client) post(ctx context.Context, base string, n Notification, renew bool) (Result, error) { + token, err := c.providerToken(renew) + if err != nil { + return Result{}, err + } + ctx, cancel := context.WithTimeout(ctx, requestTimeout) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, base+"/3/device/"+hex.EncodeToString(n.DeviceToken), + bytes.NewReader(n.Payload)) + if err != nil { + return Result{}, err + } + req.Header.Set("authorization", "bearer "+token) + req.Header.Set("apns-topic", c.cfg.Topic) + req.Header.Set("apns-push-type", "alert") + req.Header.Set("apns-priority", "10") + exp := int64(0) + if !n.Expiration.IsZero() { + exp = n.Expiration.Unix() + } + req.Header.Set("apns-expiration", strconv.FormatInt(exp, 10)) + req.Header.Set("content-type", "application/json") + + resp, err := c.http.Do(req) + if err != nil { + return Result{}, fmt.Errorf("apns: %w", err) + } + defer resp.Body.Close() + res := Result{Status: resp.StatusCode} + if resp.StatusCode == http.StatusOK { + _, _ = io.Copy(io.Discard, resp.Body) + return res, nil + } + var body struct { + Reason string `json:"reason"` + Timestamp int64 `json:"timestamp"` + } + _ = json.NewDecoder(io.LimitReader(resp.Body, 4096)).Decode(&body) + res.Reason = body.Reason + switch { + case resp.StatusCode == http.StatusGone: + res.Gone = true + if body.Timestamp > 0 { + res.GoneSince = time.UnixMilli(body.Timestamp) + } + case body.Reason == "BadDeviceToken" || body.Reason == "DeviceTokenNotForTopic": + res.Gone = true + } + return res, nil +} + +// providerToken is the JWT Apple authenticates the provider by, renewed +// after TokenLife or when renew asks. +func (c *Client) providerToken(renew bool) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + now := c.now() + if !renew && c.token != "" && now.Sub(c.tokenAt) < TokenLife { + return c.token, nil + } + token, err := signToken(c.key, c.cfg.KeyID, c.cfg.TeamID, now) + if err != nil { + return "", err + } + c.token, c.tokenAt = token, now + return token, nil +} + +// signToken makes Apple's provider token: {alg: ES256, kid} . {iss, iat}, +// signed with ES256 as r ‖ s. +func signToken(key *ecdsa.PrivateKey, keyID, teamID string, now time.Time) (string, error) { + enc := base64.RawURLEncoding + header, err := json.Marshal(map[string]string{"alg": "ES256", "kid": keyID}) + if err != nil { + return "", err + } + claims, err := json.Marshal(map[string]any{"iss": teamID, "iat": now.Unix()}) + if err != nil { + return "", err + } + signing := enc.EncodeToString(header) + "." + enc.EncodeToString(claims) + digest := sha256.Sum256([]byte(signing)) + r, s, err := ecdsa.Sign(rand.Reader, key, digest[:]) + if err != nil { + return "", err + } + sig := make([]byte, 64) + r.FillBytes(sig[:32]) + s.FillBytes(sig[32:]) + return signing + "." + enc.EncodeToString(sig), nil +} diff --git a/internal/apns/apns_test.go b/internal/apns/apns_test.go new file mode 100644 index 0000000..42fe07a --- /dev/null +++ b/internal/apns/apns_test.go @@ -0,0 +1,236 @@ +package apns + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/sha256" + "crypto/x509" + "encoding/base64" + "encoding/json" + "encoding/pem" + "io" + "math/big" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +func testP8(t *testing.T) ([]byte, *ecdsa.PrivateKey) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + der, err := x509.MarshalPKCS8PrivateKey(key) + if err != nil { + t.Fatal(err) + } + return pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: der}), key +} + +// fakeApple is Apple's provider API, over HTTP/2 with TLS. +type fakeApple struct { + t *testing.T + pub *ecdsa.PublicKey + srv *httptest.Server + + mu sync.Mutex + reqs []*http.Request + bodies [][]byte + tokens []string + respond func(r *http.Request) (int, string) +} + +func newFakeApple(t *testing.T, pub *ecdsa.PublicKey) *fakeApple { + f := &fakeApple{t: t, pub: pub} + f.srv = httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + f.mu.Lock() + f.reqs = append(f.reqs, r) + f.bodies = append(f.bodies, body) + f.tokens = append(f.tokens, strings.TrimPrefix(r.Header.Get("authorization"), "bearer ")) + respond := f.respond + f.mu.Unlock() + status, reason := http.StatusOK, "" + if respond != nil { + status, reason = respond(r) + } + w.Header().Set("apns-id", "00000000-0000-0000-0000-000000000000") + w.WriteHeader(status) + if reason != "" { + body := map[string]any{"reason": reason} + if status == http.StatusGone { + body["timestamp"] = int64(1_791_820_800_000) + } + _ = json.NewEncoder(w).Encode(body) + } + })) + f.srv.EnableHTTP2 = true + f.srv.StartTLS() + t.Cleanup(f.srv.Close) + return f +} + +func (f *fakeApple) client(t *testing.T, p8 []byte, now func() time.Time) *Client { + t.Helper() + c, err := New(Config{Key: p8, KeyID: "ABC123DEFG", TeamID: "L66M368PZ6", Topic: "energy.ftw.app", + SandboxURL: f.srv.URL + "/sandbox", ProductionURL: f.srv.URL, HTTPClient: f.srv.Client(), Now: now}) + if err != nil { + t.Fatal(err) + } + return c +} + +// checkToken verifies a provider token the way Apple does. +func (f *fakeApple) checkToken(token string) (header map[string]string, claims map[string]any) { + f.t.Helper() + parts := strings.Split(token, ".") + if len(parts) != 3 { + f.t.Fatalf("token %q", token) + } + dec := base64.RawURLEncoding + h, _ := dec.DecodeString(parts[0]) + c, _ := dec.DecodeString(parts[1]) + sig, _ := dec.DecodeString(parts[2]) + if json.Unmarshal(h, &header) != nil || json.Unmarshal(c, &claims) != nil || len(sig) != 64 { + f.t.Fatalf("token %q does not parse", token) + } + digest := sha256.Sum256([]byte(parts[0] + "." + parts[1])) + if !ecdsa.Verify(f.pub, digest[:], new(big.Int).SetBytes(sig[:32]), new(big.Int).SetBytes(sig[32:])) { + f.t.Fatal("token signature does not verify") + } + return header, claims +} + +func TestAnAlertGoesOverHTTP2WithTheSameHeadersForEveryKind(t *testing.T) { + p8, key := testP8(t) + apple := newFakeApple(t, &key.PublicKey) + now := time.Unix(1_791_820_800, 0) + c := apple.client(t, p8, func() time.Time { return now }) + + payload := []byte(`{"aps":{"mutable-content":1},"ftw":{"v":1}}`) + res, err := c.Send(context.Background(), Notification{Environment: Production, + DeviceToken: []byte{0xab, 0xcd}, Payload: payload, Expiration: now.Add(time.Hour)}) + if err != nil || !res.OK() { + t.Fatalf("send: %+v %v", res, err) + } + r := apple.reqs[0] + if r.ProtoMajor != 2 { + t.Fatalf("Apple was reached over HTTP/%d", r.ProtoMajor) + } + if r.Method != http.MethodPost || r.URL.Path != "/3/device/abcd" { + t.Fatalf("%s %s", r.Method, r.URL.Path) + } + for k, want := range map[string]string{ + "apns-topic": "energy.ftw.app", "apns-push-type": "alert", "apns-priority": "10", + "apns-expiration": "1791824400", "content-type": "application/json", + } { + if got := r.Header.Get(k); got != want { + t.Errorf("%s = %q, want %q", k, got, want) + } + } + if r.Header.Get("apns-collapse-id") != "" { + t.Error("a collapse id says what kind an alert is") + } + if string(apple.bodies[0]) != string(payload) { + t.Fatalf("payload %s", apple.bodies[0]) + } + header, claims := apple.checkToken(apple.tokens[0]) + if header["alg"] != "ES256" || header["kid"] != "ABC123DEFG" || + claims["iss"] != "L66M368PZ6" || claims["iat"] != float64(now.Unix()) { + t.Fatalf("token %v %v", header, claims) + } +} + +func TestTheProviderTokenIsRenewedEveryFortyMinutes(t *testing.T) { + p8, key := testP8(t) + apple := newFakeApple(t, &key.PublicKey) + now := time.Unix(1_791_820_800, 0) + c := apple.client(t, p8, func() time.Time { return now }) + send := func() { + t.Helper() + if _, err := c.Send(context.Background(), Notification{Environment: Sandbox, DeviceToken: []byte{1}, Payload: []byte("{}")}); err != nil { + t.Fatal(err) + } + } + send() + now = now.Add(39 * time.Minute) + send() + now = now.Add(2 * time.Minute) + send() + if apple.tokens[0] != apple.tokens[1] { + t.Fatal("renewed inside 40 minutes") + } + if apple.tokens[2] == apple.tokens[1] { + t.Fatal("still the same token after 41 minutes") + } + if !strings.HasPrefix(apple.reqs[0].URL.Path, "/sandbox/3/device/") { + t.Fatalf("sandbox went to %s", apple.reqs[0].URL.Path) + } +} + +func TestAnExpiredProviderTokenIsRenewedAndTriedOnce(t *testing.T) { + p8, key := testP8(t) + apple := newFakeApple(t, &key.PublicKey) + calls := 0 + apple.respond = func(*http.Request) (int, string) { + calls++ + if calls == 1 { + return http.StatusForbidden, "ExpiredProviderToken" + } + return http.StatusOK, "" + } + c := apple.client(t, p8, nil) + res, err := c.Send(context.Background(), Notification{Environment: Production, DeviceToken: []byte{1}, Payload: []byte("{}")}) + if err != nil || !res.OK() || calls != 2 { + t.Fatalf("%+v %v after %d calls", res, err, calls) + } +} + +func TestAGoneTokenSaysSo(t *testing.T) { + p8, key := testP8(t) + apple := newFakeApple(t, &key.PublicKey) + for _, c := range []struct { + status int + reason string + gone bool + }{ + {http.StatusGone, "Unregistered", true}, + {http.StatusBadRequest, "BadDeviceToken", true}, + {http.StatusBadRequest, "DeviceTokenNotForTopic", true}, + {http.StatusTooManyRequests, "TooManyRequests", false}, + {http.StatusServiceUnavailable, "ServiceUnavailable", false}, + } { + apple.respond = func(*http.Request) (int, string) { return c.status, c.reason } + res, err := apple.client(t, p8, nil).Send(context.Background(), + Notification{Environment: Production, DeviceToken: []byte{1}, Payload: []byte("{}")}) + if err != nil || res.OK() || res.Gone != c.gone || res.Reason != c.reason { + t.Fatalf("%d %s: %+v %v", c.status, c.reason, res, err) + } + if c.status == http.StatusGone && res.GoneSince.UnixMilli() != 1_791_820_800_000 { + t.Fatalf("gone since %v", res.GoneSince) + } + } +} + +func TestOnlyAnAppleKeyIsTaken(t *testing.T) { + if _, err := New(Config{Key: []byte("not a key"), KeyID: "ABC123DEFG", TeamID: "L66M368PZ6", Topic: "x"}); err == nil { + t.Fatal("took garbage") + } + p8, _ := testP8(t) + if _, err := New(Config{Key: p8, KeyID: "short", TeamID: "L66M368PZ6", Topic: "x"}); err == nil { + t.Fatal("took a key id Apple never issues") + } + c, err := New(Config{Key: p8, KeyID: "ABC123DEFG", TeamID: "L66M368PZ6", Topic: "x"}) + if err != nil { + t.Fatal(err) + } + if _, err := c.Send(context.Background(), Notification{Environment: "staging", DeviceToken: []byte{1}}); err == nil { + t.Fatal("sent to an environment that does not exist") + } +} diff --git a/internal/push/push.go b/internal/push/push.go new file mode 100644 index 0000000..f27e3cf --- /dev/null +++ b/internal/push/push.go @@ -0,0 +1,314 @@ +// Package push tells a home's phones what its box said while they were not +// looking, through Apple. docs/protocol-cloud.md in srcfl/ftw, "Push", is the +// wire format; go/protocol/cloudwire builds the payload. +// +// The service chooses nothing a phone shows. An alert carries the box's +// signed event exactly as it arrived; the phone checks the signature, writes +// its own words and picks the buttons from the verified kind. Every alert +// has the same aps, priority and push type, and no collapse id, so the +// request says nothing a kind would. The one alert the service writes is +// box.unreachable, after the box has been away for ten minutes without +// saying it was restarting on purpose. It carries no arguments and no +// buttons. +// +// Tokens are bound to roster entries. Only a device whose role and the +// owner's grant both allow ftw.events.read may register, and a roster that +// drops the device, moves its epoch or narrows its scope drops its tokens. +// Nothing here logs a token or a home id. +package push + +import ( + "context" + "errors" + "log/slog" + "slices" + "sync" + "time" + + "github.com/srcfl/ftw-cloud/internal/apns" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +const ( + // MaxAlertsPerHour bounds what one home can make its phones buzz. A box + // raises a few notifications a day; past this, something is wrong and + // the rest are dropped. + MaxAlertsPerHour = 20 + // eventExpiry is how long Apple may keep trying to deliver an event. + // Later than that the phone would show only the generic line. + eventExpiry = cloudwire.AlertMaxAgeMs * time.Millisecond + // unreachableExpiry is the same for box.unreachable, which stays true + // while the box is away. + unreachableExpiry = time.Hour + // queueSize bounds the alerts waiting to be sent. + queueSize = 256 + // expireEvery is how often registrations of long-silent homes go. + expireEvery = time.Hour +) + +// Sender is Apple. apns.Client is one. +type Sender interface { + Send(ctx context.Context, n apns.Notification) (apns.Result, error) +} + +// ErrNotForThisDevice refuses a registration the device may not make: a +// viewer, a grant without events, or a kind that does not exist yet. The +// app session goes on. +var ErrNotForThisDevice = errors.New("push: not a registration this device may make") + +// Service keeps registrations and sends alerts. +type Service struct { + reg *registry.Registry + sender Sender + log *slog.Logger + now func() time.Time + // after runs f after d and returns a function that stops it. Tests + // replace it. + after func(d time.Duration, f func()) (stop func() bool) + + work chan func(context.Context) + + mu sync.Mutex + homes map[string]*homeState +} + +// homeState is one home's box connection, as the dead man's switch sees it, +// and its recent alerts. +type homeState struct { + online bool + // gen moves on every connect and disconnect, so a timer set for one + // outage cannot fire into the next. + gen uint64 + stop func() bool + told bool + sent []time.Time +} + +// New builds the service. A nil sender keeps registrations and sends +// nothing, for a service without an APNs key. +func New(reg *registry.Registry, sender Sender, log *slog.Logger) *Service { + if log == nil { + log = slog.Default() + } + return &Service{ + reg: reg, sender: sender, log: log, now: time.Now, + after: func(d time.Duration, f func()) func() bool { return time.AfterFunc(d, f).Stop }, + work: make(chan func(context.Context), queueSize), + homes: map[string]*homeState{}, + } +} + +// Run sends queued alerts and drops long-silent registrations until ctx +// ends. +func (s *Service) Run(ctx context.Context) { + t := time.NewTicker(expireEvery) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case job := <-s.work: + job(ctx) + case <-t.C: + if n, err := s.reg.PushExpire(s.now()); err != nil { + s.log.Warn("push upkeep failed", "err", err) + } else if n > 0 { + s.log.Info("push registrations expired with their silent homes", "count", n) + } + } + } +} + +// enqueue hands a job to Run without ever blocking the caller, which is a +// box's uplink. A full queue drops the job. +func (s *Service) enqueue(job func(context.Context)) { + select { + case s.work <- job: + default: + s.log.Warn("push queue full; an alert was dropped") + } +} + +// -------------------------------------------------------------------------- +// Registrations, from a phone's app session +// -------------------------------------------------------------------------- + +// Register keeps a device's APNs token. dev is the roster entry the phone's +// Noise key matched and scopes its effective scope, both from store.For. +func (s *Service) Register(home string, dev cloudwire.Device, scopes []string, r cloudwire.PushRegister) error { + if r.Kind != cloudwire.PushKindAlert || !slices.Contains(scopes, protocol.ScopeEventsRead) || + len(r.APNsToken) == 0 || len(r.APNsToken) > cloudwire.APNsTokenMaxBytes || + (r.Environment != cloudwire.PushSandbox && r.Environment != cloudwire.PushProduction) { + return ErrNotForThisDevice + } + return s.reg.PushRegister(registry.PushReg{Home: home, DeviceID: dev.ID, Epoch: dev.Epoch, + Token: r.APNsToken, Environment: r.Environment, Kind: r.Kind, At: s.now()}) +} + +// Unregister forgets one of the device's tokens. +func (s *Service) Unregister(home string, dev cloudwire.Device, r cloudwire.PushUnregister) error { + return s.reg.PushUnregister(home, dev.ID, r.APNsToken) +} + +// -------------------------------------------------------------------------- +// From a box's uplink. None of these blocks. +// -------------------------------------------------------------------------- + +// Roster drops the tokens of devices the box's roster no longer lets read +// events: gone, paired again under a new epoch, or narrowed. +func (s *Service) Roster(home string, r *cloudwire.Roster) { + if r == nil { + return + } + allowed := map[string]uint64{} + for _, d := range r.Devices { + if slices.Contains(d.Scopes, protocol.ScopeEventsRead) && slices.Contains(r.Grant.Scopes, protocol.ScopeEventsRead) { + allowed[d.ID] = d.Epoch + } + } + if err := s.reg.PushPrune(home, func(id string, epoch uint64) bool { + e, ok := allowed[id] + return ok && e == epoch + }); err != nil { + s.log.Warn("could not drop the tokens a roster dropped", "err", err) + } +} + +// Event sends a box-signed event envelope, as it arrived, to the home's +// registered phones. +func (s *Service) Event(home string, envelope []byte) { + payload, err := cloudwire.AlertPayload(cloudwire.Alert{Home: home, Event: envelope}) + if err != nil { + return + } + s.enqueue(func(ctx context.Context) { s.send(ctx, home, payload, eventExpiry) }) +} + +// BoxConnected stops the dead man's switch for a home whose box is back. +func (s *Service) BoxConnected(home string) { + s.mu.Lock() + defer s.mu.Unlock() + st := s.state(home) + st.gen++ + st.online, st.told = true, false + if st.stop != nil { + st.stop() + st.stop = nil + } +} + +// BoxGone starts the dead man's switch. why is the box's bye reason, empty +// when the socket just dropped. A bye for an update holds box.unreachable +// back until the box has been away for cloudwire.UpdateGraceMs; a revoke +// ends it, since the home is gone. +func (s *Service) BoxGone(home, why string) { + s.mu.Lock() + defer s.mu.Unlock() + if why == cloudwire.ByeRevoked { + if st := s.homes[home]; st != nil && st.stop != nil { + st.stop() + } + delete(s.homes, home) + return + } + st := s.state(home) + st.gen++ + st.online = false + if st.stop != nil { + st.stop() + } + wait := time.Duration(cloudwire.UnreachableAfterMs) * time.Millisecond + if why == cloudwire.ByeUpdate { + wait = time.Duration(cloudwire.UpdateGraceMs) * time.Millisecond + } + gen := st.gen + st.stop = s.after(wait, func() { s.unreachable(home, gen) }) +} + +func (s *Service) unreachable(home string, gen uint64) { + s.mu.Lock() + st := s.homes[home] + if st == nil || st.gen != gen || st.online || st.told { + s.mu.Unlock() + return + } + st.told, st.stop = true, nil + s.mu.Unlock() + payload, err := cloudwire.AlertPayload(cloudwire.Alert{Home: home, Kind: protocol.PushBoxUnreachable}) + if err != nil { + return + } + s.enqueue(func(ctx context.Context) { s.send(ctx, home, payload, unreachableExpiry) }) +} + +func (s *Service) state(home string) *homeState { + st := s.homes[home] + if st == nil { + st = &homeState{} + s.homes[home] = st + } + return st +} + +// allow counts one alert against the home's hourly cap. +func (s *Service) allow(home string) bool { + s.mu.Lock() + defer s.mu.Unlock() + st := s.state(home) + now := s.now() + keep := st.sent[:0] + for _, at := range st.sent { + if now.Sub(at) < time.Hour { + keep = append(keep, at) + } + } + st.sent = keep + if len(st.sent) >= MaxAlertsPerHour { + return false + } + st.sent = append(st.sent, now) + return true +} + +// send posts one payload to every phone of the home registered for alerts, +// and forgets the tokens Apple says are gone. +func (s *Service) send(ctx context.Context, home string, payload []byte, expiry time.Duration) { + if s.sender == nil { + return + } + regs, err := s.reg.PushRegs(home, cloudwire.PushKindAlert) + if err != nil { + s.log.Warn("could not read push registrations", "err", err) + return + } + if len(regs) == 0 { + return + } + if !s.allow(home) { + s.log.Warn("a home passed its alert cap; alert dropped", "cap_per_hour", MaxAlertsPerHour) + return + } + var sent, gone, failed int + exp := s.now().Add(expiry) + for _, r := range regs { + res, err := s.sender.Send(ctx, apns.Notification{Environment: r.Environment, DeviceToken: r.Token, + Payload: payload, Expiration: exp}) + switch { + case err != nil: + failed++ + case res.OK(): + sent++ + case res.Gone: + gone++ + if err := s.reg.PushGone(home, r.Token, res.GoneSince); err != nil { + s.log.Warn("could not forget a gone token", "err", err) + } + default: + failed++ + s.log.Warn("APNs refused an alert", "status", res.Status, "reason", res.Reason) + } + } + s.log.Info("alert sent", "phones", sent, "gone", gone, "failed", failed) +} diff --git a/internal/push/push_test.go b/internal/push/push_test.go new file mode 100644 index 0000000..78584ff --- /dev/null +++ b/internal/push/push_test.go @@ -0,0 +1,357 @@ +package push + +import ( + "bytes" + "context" + "encoding/json" + "io" + "log/slog" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/srcfl/ftw-cloud/internal/apns" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +type fakeApple struct { + mu sync.Mutex + sent []apns.Notification + gone map[string]bool +} + +func (f *fakeApple) Send(_ context.Context, n apns.Notification) (apns.Result, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.sent = append(f.sent, n) + if f.gone[string(n.DeviceToken)] { + return apns.Result{Status: 410, Reason: "Unregistered", Gone: true}, nil + } + return apns.Result{Status: 200}, nil +} + +func (f *fakeApple) notifications() []apns.Notification { + f.mu.Lock() + defer f.mu.Unlock() + return append([]apns.Notification(nil), f.sent...) +} + +// timers is a clock the test moves by hand. +type timers struct { + mu sync.Mutex + now time.Time + due []*timer +} + +type timer struct { + at time.Time + f func() + stopped bool +} + +func (c *timers) after(d time.Duration, f func()) func() bool { + c.mu.Lock() + defer c.mu.Unlock() + t := &timer{at: c.now.Add(d), f: f} + c.due = append(c.due, t) + return func() bool { + c.mu.Lock() + defer c.mu.Unlock() + was := !t.stopped + t.stopped = true + return was + } +} + +func (c *timers) advance(d time.Duration) { + c.mu.Lock() + c.now = c.now.Add(d) + var fire []func() + for _, t := range c.due { + if !t.stopped && !t.at.After(c.now) { + t.stopped = true + fire = append(fire, t.f) + } + } + c.mu.Unlock() + for _, f := range fire { + f() + } +} + +func (c *timers) time() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +type rig struct { + svc *Service + reg *registry.Registry + apple *fakeApple + clock *timers + home string + owner cloudwire.Device + viewer cloudwire.Device + all []string +} + +func newRig(t *testing.T) *rig { + t.Helper() + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = reg.Close() }) + r := &rig{reg: reg, apple: &fakeApple{gone: map[string]bool{}}, + clock: &timers{now: time.Unix(1_791_820_800, 0)}, home: "fjzyr67buzpyncqmjliwgezvde"} + r.svc = New(reg, r.apple, slog.New(slog.NewTextHandler(io.Discard, nil))) + r.svc.now, r.svc.after = r.clock.time, r.clock.after + if err := reg.Claim(r.home, []byte("pk"), r.clock.time()); err != nil { + t.Fatal(err) + } + r.all = []string{protocol.ScopeLiveRead, protocol.ScopeEventsRead} + r.owner = cloudwire.Device{ID: "owner-phone", Key: bytes.Repeat([]byte{5}, 32), Role: "owner", Scopes: r.all, Epoch: 1} + r.viewer = cloudwire.Device{ID: "viewer-phone", Key: bytes.Repeat([]byte{6}, 32), Role: "viewer", + Scopes: []string{protocol.ScopeLiveRead}, Epoch: 1} + return r +} + +// drain runs what the uplink queued, as Run would. +func (r *rig) drain() { + for { + select { + case job := <-r.svc.work: + job(context.Background()) + default: + return + } + } +} + +func (r *rig) register(t *testing.T, dev cloudwire.Device, scopes []string, token string) error { + t.Helper() + return r.svc.Register(r.home, dev, scopes, cloudwire.PushRegister{ + APNsToken: []byte(token), Environment: cloudwire.PushSandbox, Kind: cloudwire.PushKindAlert}) +} + +func TestAnEventGoesToRegisteredPhonesExactlyAsTheBoxSignedIt(t *testing.T) { + r := newRig(t) + if err := r.register(t, r.owner, r.all, "owner-token"); err != nil { + t.Fatal(err) + } + envelope := []byte{0xa3, 0x61, 0x62, 0x41, 0x00, 0x61, 0x74, 0x65, 'e', 'v', 'e', 'n', 't', 0x63, 's', 'i', 'g', 0x40} + r.svc.Event(r.home, envelope) + r.drain() + sent := r.apple.notifications() + if len(sent) != 1 { + t.Fatalf("sent %d", len(sent)) + } + n := sent[0] + want, _ := cloudwire.AlertPayload(cloudwire.Alert{Home: r.home, Event: envelope}) + if !bytes.Equal(n.Payload, want) || string(n.DeviceToken) != "owner-token" || n.Environment != apns.Sandbox { + t.Fatalf("notification %+v", n) + } + var p struct { + FTW cloudwire.Alert `json:"ftw"` + } + if json.Unmarshal(n.Payload, &p) != nil || !bytes.Equal(p.FTW.Event, envelope) { + t.Fatal("the box's envelope did not arrive unchanged") + } + if !n.Expiration.Equal(r.clock.time().Add(10 * time.Minute)) { + t.Fatalf("expires %v", n.Expiration) + } +} + +func TestOnlyADeviceThatMayReadEventsRegisters(t *testing.T) { + r := newRig(t) + if err := r.register(t, r.viewer, r.viewer.Scopes, "viewer-token"); err != ErrNotForThisDevice { + t.Fatalf("a viewer registered: %v", err) + } + // An owner under a grant that leaves events out has no events scope + // in its effective scope. + if err := r.register(t, r.owner, []string{protocol.ScopeLiveRead}, "owner-token"); err != ErrNotForThisDevice { + t.Fatalf("an owner without events registered: %v", err) + } + bad := []cloudwire.PushRegister{ + {APNsToken: []byte("t"), Environment: "staging", Kind: cloudwire.PushKindAlert}, + {APNsToken: []byte("t"), Environment: cloudwire.PushSandbox, Kind: "liveactivity"}, + {APNsToken: nil, Environment: cloudwire.PushSandbox, Kind: cloudwire.PushKindAlert}, + {APNsToken: make([]byte, 101), Environment: cloudwire.PushSandbox, Kind: cloudwire.PushKindAlert}, + } + for _, b := range bad { + if err := r.svc.Register(r.home, r.owner, r.all, b); err != ErrNotForThisDevice { + t.Fatalf("registered %+v: %v", b, err) + } + } + if regs, _ := r.reg.PushRegs(r.home, cloudwire.PushKindAlert); len(regs) != 0 { + t.Fatalf("kept %d", len(regs)) + } +} + +func TestARosterThatDropsOrNarrowsADeviceDropsItsTokens(t *testing.T) { + r := newRig(t) + other := r.owner + other.ID, other.Key = "second-owner", bytes.Repeat([]byte{7}, 32) + _ = r.register(t, r.owner, r.all, "a") + _ = r.register(t, other, r.all, "b") + grant := cloudwire.Grant{ID: bytes.Repeat([]byte{4}, 16), Scopes: r.all} + + // The second owner became a viewer: no events in its scopes. + narrowed := other + narrowed.Scopes, narrowed.Role = []string{protocol.ScopeLiveRead}, "viewer" + r.svc.Roster(r.home, &cloudwire.Roster{Devices: []cloudwire.Device{r.owner, narrowed}, Grant: grant}) + regs, _ := r.reg.PushRegs(r.home, cloudwire.PushKindAlert) + if len(regs) != 1 || regs[0].DeviceID != r.owner.ID { + t.Fatalf("after narrowing: %+v", regs) + } + // The owner paired again: a new epoch. + again := r.owner + again.Epoch = 2 + r.svc.Roster(r.home, &cloudwire.Roster{Devices: []cloudwire.Device{again}, Grant: grant}) + if regs, _ := r.reg.PushRegs(r.home, cloudwire.PushKindAlert); len(regs) != 0 { + t.Fatalf("after a new epoch: %+v", regs) + } + // And a grant that drops events drops every token. + _ = r.register(t, again, r.all, "c") + grant.Scopes = []string{protocol.ScopeLiveRead} + r.svc.Roster(r.home, &cloudwire.Roster{Devices: []cloudwire.Device{again}, Grant: grant}) + if regs, _ := r.reg.PushRegs(r.home, cloudwire.PushKindAlert); len(regs) != 0 { + t.Fatalf("after the grant dropped events: %+v", regs) + } +} + +func TestATokenAppleSaysIsGoneIsForgotten(t *testing.T) { + r := newRig(t) + _ = r.register(t, r.owner, r.all, "dead") + r.apple.gone["dead"] = true + r.svc.Event(r.home, []byte{0xa0}) + r.drain() + if regs, _ := r.reg.PushRegs(r.home, cloudwire.PushKindAlert); len(regs) != 0 { + t.Fatal("kept a token Apple called gone") + } +} + +func unreachableSent(t *testing.T, r *rig) int { + t.Helper() + r.drain() + n := 0 + for _, x := range r.apple.notifications() { + var p struct { + FTW cloudwire.Alert `json:"ftw"` + } + if json.Unmarshal(x.Payload, &p) == nil && p.FTW.Kind == protocol.PushBoxUnreachable { + if len(p.FTW.Event) != 0 || p.FTW.Home != r.home { + t.Fatalf("unreachable payload %s", x.Payload) + } + n++ + } + } + return n +} + +func TestBoxUnreachableAfterTenMinutesOncePerOutage(t *testing.T) { + r := newRig(t) + _ = r.register(t, r.owner, r.all, "tok") + r.svc.BoxConnected(r.home) + r.svc.BoxGone(r.home, "") + r.clock.advance(599 * time.Second) + if n := unreachableSent(t, r); n != 0 { + t.Fatal("before ten minutes") + } + r.clock.advance(2 * time.Second) + if n := unreachableSent(t, r); n != 1 { + t.Fatalf("sent %d at ten minutes", n) + } + r.clock.advance(time.Hour) + if n := unreachableSent(t, r); n != 1 { + t.Fatalf("sent %d over one outage", n) + } + // Back and away again is a new outage. + r.svc.BoxConnected(r.home) + r.svc.BoxGone(r.home, cloudwire.ByeShutdown) + r.clock.advance(601 * time.Second) + if n := unreachableSent(t, r); n != 2 { + t.Fatalf("sent %d over two outages", n) + } +} + +func TestABoxBackInTimeSaysNothing(t *testing.T) { + r := newRig(t) + _ = r.register(t, r.owner, r.all, "tok") + r.svc.BoxConnected(r.home) + r.svc.BoxGone(r.home, "") + r.clock.advance(5 * time.Minute) + r.svc.BoxConnected(r.home) + r.clock.advance(time.Hour) + if n := unreachableSent(t, r); n != 0 { + t.Fatalf("sent %d for a box that came back", n) + } +} + +// A planned update sends nothing, unless the box is still away after an +// hour. +func TestAPlannedUpdateHoldsItBackForAnHour(t *testing.T) { + r := newRig(t) + _ = r.register(t, r.owner, r.all, "tok") + r.svc.BoxConnected(r.home) + r.svc.BoxGone(r.home, cloudwire.ByeUpdate) + r.clock.advance(20 * time.Minute) + if n := unreachableSent(t, r); n != 0 { + t.Fatal("an update was reported as out of reach") + } + r.svc.BoxConnected(r.home) + r.clock.advance(2 * time.Hour) + if n := unreachableSent(t, r); n != 0 { + t.Fatal("an update that came back was reported") + } + + r.svc.BoxGone(r.home, cloudwire.ByeUpdate) + r.clock.advance(time.Hour + time.Second) + if n := unreachableSent(t, r); n != 1 { + t.Fatalf("an update still away after an hour: %d", n) + } +} + +func TestARevokedHomeHasNoDeadMan(t *testing.T) { + r := newRig(t) + _ = r.register(t, r.owner, r.all, "tok") + r.svc.BoxConnected(r.home) + r.svc.BoxGone(r.home, cloudwire.ByeRevoked) + r.clock.advance(2 * time.Hour) + if n := unreachableSent(t, r); n != 0 { + t.Fatal("a revoked home was reported out of reach") + } +} + +func TestAHomeCannotMakeItsPhonesBuzzWithoutEnd(t *testing.T) { + r := newRig(t) + _ = r.register(t, r.owner, r.all, "tok") + for range MaxAlertsPerHour + 5 { + r.svc.Event(r.home, []byte{0xa0}) + } + r.drain() + if n := len(r.apple.notifications()); n != MaxAlertsPerHour { + t.Fatalf("sent %d in an hour", n) + } + r.clock.advance(time.Hour) + r.svc.Event(r.home, []byte{0xa0}) + r.drain() + if n := len(r.apple.notifications()); n != MaxAlertsPerHour+1 { + t.Fatalf("the cap did not lift after an hour: %d", n) + } +} + +func TestWithoutAKeyNothingIsSent(t *testing.T) { + r := newRig(t) + r.svc.sender = nil + _ = r.register(t, r.owner, r.all, "tok") + r.svc.Event(r.home, []byte{0xa0}) + r.drain() + if regs, _ := r.reg.PushRegs(r.home, cloudwire.PushKindAlert); len(regs) != 1 { + t.Fatal("registrations need no key") + } +} diff --git a/internal/registry/push.go b/internal/registry/push.go new file mode 100644 index 0000000..1bc25ec --- /dev/null +++ b/internal/registry/push.go @@ -0,0 +1,135 @@ +package registry + +import ( + "database/sql" + "time" +) + +// APNs registrations. Each row binds one device token to one roster entry +// (device id and epoch) of one home. A row goes when its home goes (the +// foreign key cascades on purge and expiry), when a roster drops the device +// or moves its epoch, when the phone unregisters, when Apple says the token +// is gone, and after PushKeptWithoutBox without a roster from the box. + +// PushKeptWithoutBox is how long registrations outlive the box's last +// roster. The box renews its roster every six hours while it is connected. +const PushKeptWithoutBox = 30 * 24 * time.Hour + +// PushReg is one registration. +type PushReg struct { + Home string + DeviceID string + Epoch uint64 + Token []byte + Environment string + Kind string + At time.Time +} + +func createPushTable(db *sql.DB) error { + _, err := db.Exec(` +CREATE TABLE IF NOT EXISTS push ( + home TEXT NOT NULL REFERENCES homes(home) ON DELETE CASCADE, + device_id TEXT NOT NULL, + epoch INTEGER NOT NULL, + token BLOB NOT NULL, + environment TEXT NOT NULL, + kind TEXT NOT NULL, + registered_ms INTEGER NOT NULL, + PRIMARY KEY (home, token, kind) +);`) + return err +} + +// PushRegister keeps a device's token. A token already registered for the +// home moves to this device: a phone that paired again keeps its token. +func (r *Registry) PushRegister(reg PushReg) error { + _, err := r.db.Exec(`INSERT INTO push (home, device_id, epoch, token, environment, kind, registered_ms) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (home, token, kind) DO UPDATE SET device_id = excluded.device_id, epoch = excluded.epoch, + environment = excluded.environment, registered_ms = excluded.registered_ms`, + reg.Home, reg.DeviceID, int64(reg.Epoch), reg.Token, reg.Environment, reg.Kind, reg.At.UnixMilli()) + return err +} + +// PushUnregister forgets one of a device's tokens. +func (r *Registry) PushUnregister(home, deviceID string, token []byte) error { + _, err := r.db.Exec(`DELETE FROM push WHERE home = ? AND device_id = ? AND token = ?`, home, deviceID, token) + return err +} + +// PushRegs returns a home's registrations of one kind. +func (r *Registry) PushRegs(home, kind string) ([]PushReg, error) { + rows, err := r.db.Query(`SELECT device_id, epoch, token, environment, registered_ms FROM push + WHERE home = ? AND kind = ? ORDER BY registered_ms`, home, kind) + if err != nil { + return nil, err + } + defer rows.Close() + var out []PushReg + for rows.Next() { + reg := PushReg{Home: home, Kind: kind} + var epoch, ms int64 + if err := rows.Scan(®.DeviceID, &epoch, ®.Token, ®.Environment, &ms); err != nil { + return nil, err + } + reg.Epoch, reg.At = uint64(epoch), time.UnixMilli(ms) + out = append(out, reg) + } + return out, rows.Err() +} + +// PushGone forgets a token Apple says is gone, unless the phone registered +// it again after Apple last knew it valid. +func (r *Registry) PushGone(home string, token []byte, since time.Time) error { + if since.IsZero() { + _, err := r.db.Exec(`DELETE FROM push WHERE home = ? AND token = ?`, home, token) + return err + } + _, err := r.db.Exec(`DELETE FROM push WHERE home = ? AND token = ? AND registered_ms <= ?`, + home, token, since.UnixMilli()) + return err +} + +// PushPrune keeps only the registrations whose device keep allows at its +// epoch, as a new roster says. +func (r *Registry) PushPrune(home string, keep func(deviceID string, epoch uint64) bool) error { + rows, err := r.db.Query(`SELECT DISTINCT device_id, epoch FROM push WHERE home = ?`, home) + if err != nil { + return err + } + type entry struct { + id string + epoch int64 + } + var gone []entry + for rows.Next() { + var e entry + if err := rows.Scan(&e.id, &e.epoch); err != nil { + _ = rows.Close() + return err + } + if !keep(e.id, uint64(e.epoch)) { + gone = append(gone, e) + } + } + _ = rows.Close() + for _, e := range gone { + if _, err := r.db.Exec(`DELETE FROM push WHERE home = ? AND device_id = ? AND epoch = ?`, home, e.id, e.epoch); err != nil { + return err + } + } + return nil +} + +// PushExpire forgets the registrations of homes whose box has sent no +// roster for PushKeptWithoutBox, and returns how many went. +func (r *Registry) PushExpire(now time.Time) (int64, error) { + cutoff := now.Add(-PushKeptWithoutBox).UnixMilli() + res, err := r.db.Exec(`DELETE FROM push WHERE home IN (SELECT home FROM homes + WHERE (roster_ms > 0 AND roster_ms < ?) OR (roster_ms = 0 AND created_ms < ?))`, cutoff, cutoff) + if err != nil { + return 0, err + } + return res.RowsAffected() +} diff --git a/internal/registry/push_test.go b/internal/registry/push_test.go new file mode 100644 index 0000000..d2931d7 --- /dev/null +++ b/internal/registry/push_test.go @@ -0,0 +1,111 @@ +package registry + +import ( + "testing" + "time" +) + +func reg(home, device string, epoch uint64, token string, at time.Time) PushReg { + return PushReg{Home: home, DeviceID: device, Epoch: epoch, Token: []byte(token), + Environment: "sandbox", Kind: "alert", At: at} +} + +func TestARegistrationBelongsToARosterEntryAndGoesWithIt(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) + } + for _, x := range []PushReg{reg("h1", "phone-a", 1, "tok-a", now), reg("h1", "phone-b", 2, "tok-b", now)} { + if err := r.PushRegister(x); err != nil { + t.Fatal(err) + } + } + got, err := r.PushRegs("h1", "alert") + if err != nil || len(got) != 2 { + t.Fatalf("%v %v", got, err) + } + + // The roster drops phone-b and moves phone-a's epoch: both go. + if err := r.PushPrune("h1", func(id string, epoch uint64) bool { return id == "phone-a" && epoch == 2 }); err != nil { + t.Fatal(err) + } + if got, _ := r.PushRegs("h1", "alert"); len(got) != 0 { + t.Fatalf("kept %v", got) + } +} + +func TestATokenMovesToTheDeviceThatRegistersIt(t *testing.T) { + r := open(t) + now := time.Unix(1_791_820_800, 0) + _ = r.Claim("h1", []byte("pk"), now) + _ = r.PushRegister(reg("h1", "phone-a", 1, "tok", now)) + _ = r.PushRegister(reg("h1", "phone-a2", 3, "tok", now.Add(time.Minute))) + got, _ := r.PushRegs("h1", "alert") + if len(got) != 1 || got[0].DeviceID != "phone-a2" || got[0].Epoch != 3 { + t.Fatalf("%+v", got) + } + if err := r.PushUnregister("h1", "phone-a", []byte("tok")); err != nil { + t.Fatal(err) + } + if got, _ := r.PushRegs("h1", "alert"); len(got) != 1 { + t.Fatal("another device unregistered a token that is not its own") + } + _ = r.PushUnregister("h1", "phone-a2", []byte("tok")) + if got, _ := r.PushRegs("h1", "alert"); len(got) != 0 { + t.Fatal("unregister kept the token") + } +} + +func TestAGoneTokenGoesUnlessRegisteredSince(t *testing.T) { + r := open(t) + now := time.Unix(1_791_820_800, 0) + _ = r.Claim("h1", []byte("pk"), now) + _ = r.PushRegister(reg("h1", "phone-a", 1, "tok", now)) + if err := r.PushGone("h1", []byte("tok"), now.Add(-time.Hour)); err != nil { + t.Fatal(err) + } + if got, _ := r.PushRegs("h1", "alert"); len(got) != 1 { + t.Fatal("a token registered after Apple last knew it was dropped") + } + _ = r.PushGone("h1", []byte("tok"), now.Add(time.Hour)) + if got, _ := r.PushRegs("h1", "alert"); len(got) != 0 { + t.Fatal("a gone token was kept") + } +} + +func TestAPurgedOrSilentHomeTakesItsRegistrationsWithIt(t *testing.T) { + r := open(t) + now := time.Unix(1_791_820_800, 0) + for _, h := range []string{"h1", "h2", "h3"} { + _ = r.Claim(h, []byte("pk-"+h), now) + _ = r.SetRoster(h, 1, []byte("g"), []byte("roster"), now) + _ = r.PushRegister(reg(h, "phone", 1, "tok-"+h, now)) + } + if err := r.Purge("h1", []byte("g"), now); err != nil { + t.Fatal(err) + } + if got, _ := r.PushRegs("h1", "alert"); len(got) != 0 { + t.Fatal("a purged home kept its registrations") + } + // h3's box keeps sending its roster; h2's went quiet a month ago. + later := now.Add(PushKeptWithoutBox + time.Hour) + _ = r.SetRoster("h3", 2, []byte("g"), []byte("roster"), later) + n, err := r.PushExpire(later) + if err != nil || n != 1 { + t.Fatalf("expired %d, %v", n, err) + } + if got, _ := r.PushRegs("h2", "alert"); len(got) != 0 { + t.Fatal("a home silent for a month kept its registrations") + } + if got, _ := r.PushRegs("h3", "alert"); len(got) != 1 { + t.Fatal("a home whose box is there lost its registrations") + } +} + +func TestARegistrationNeedsAHome(t *testing.T) { + r := open(t) + if err := r.PushRegister(reg("nowhere", "phone", 1, "tok", time.Now())); err == nil { + t.Fatal("registered a token for a home that does not exist") + } +} diff --git a/internal/registry/registry.go b/internal/registry/registry.go index 6808993..a052e25 100644 --- a/internal/registry/registry.go +++ b/internal/registry/registry.go @@ -44,7 +44,7 @@ type Home struct { // 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)") + db, err := sql.Open("sqlite", "file:"+path+"?_pragma=journal_mode(WAL)&_pragma=secure_delete(ON)&_pragma=busy_timeout(5000)&_pragma=foreign_keys(1)") if err != nil { return nil, err } @@ -68,6 +68,10 @@ CREATE TABLE IF NOT EXISTS tombstones ( _ = db.Close() return nil, fmt.Errorf("registry: schema: %w", err) } + if err := createPushTable(db); err != nil { + _ = db.Close() + return nil, fmt.Errorf("registry: schema: %w", err) + } return &Registry{db: db}, nil } diff --git a/internal/uplink/hooks_test.go b/internal/uplink/hooks_test.go new file mode 100644 index 0000000..8ffe84d --- /dev/null +++ b/internal/uplink/hooks_test.go @@ -0,0 +1,108 @@ +package uplink + +import ( + "bytes" + "sync" + "testing" + "time" + + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// heard records what the uplink tells push. +type heard struct { + mu sync.Mutex + events [][]byte + rosters []uint64 + gone []string + conns int +} + +func listen(s *service) *heard { + h := &heard{} + s.up.OnConnect = func(string) { h.mu.Lock(); h.conns++; h.mu.Unlock() } + s.up.OnEvent = func(home string, env []byte) { + h.mu.Lock() + h.events = append(h.events, env) + h.mu.Unlock() + } + s.up.OnRoster = func(home string, r *cloudwire.Roster) { + h.mu.Lock() + h.rosters = append(h.rosters, r.RosterSeq) + h.mu.Unlock() + } + s.up.OnGone = func(home, why string) { + h.mu.Lock() + h.gone = append(h.gone, why) + h.mu.Unlock() + } + return h +} + +func (h *heard) count(of func(*heard) int) func() int { + return func() int { + h.mu.Lock() + defer h.mu.Unlock() + return of(h) + } +} + +// A signed event reaches push as the envelope the box sent, and only a +// signed one. +func TestAnEventReachesPushUnchanged(t *testing.T) { + s := newService(t) + h := listen(s) + b := newBox(t) + _, _ = b.connect(s, nil) + b.send(cloudwire.MsgRoster, b.roster(protocol.ScopeLiveRead, protocol.ScopeEventsRead)) + sent := b.send(cloudwire.MsgEvent, cloudwire.Event{Header: b.header(), EventID: "e1", + Kind: protocol.PushChargingInterrupted, Severity: cloudwire.SeverityWarning, Args: map[string]string{}}) + events := h.count(func(h *heard) int { return len(h.events) }) + wait(t, "the event", func() bool { return events() == 1 }) + if !bytes.Equal(h.events[0], sent) { + t.Fatal("push got other bytes than the box sent") + } + if rosters := h.count(func(h *heard) int { return len(h.rosters) }); rosters() != 1 { + t.Fatal("push did not hear the roster") + } +} + +// The end of an uplink tells push why: the box's bye, or nothing when the +// socket dropped. A connection a newer one replaced says nothing. +func TestTheEndOfAnUplinkSaysWhy(t *testing.T) { + s := newService(t) + h := listen(s) + gone := h.count(func(h *heard) int { return len(h.gone) }) + conns := h.count(func(h *heard) int { return h.conns }) + + b := newBox(t) + _, _ = b.connect(s, nil) + b.send(cloudwire.MsgRoster, b.roster(protocol.ScopeLiveRead)) + wait(t, "the connect", func() bool { return conns() == 1 }) + b.send(cloudwire.MsgBye, cloudwire.Bye{Header: b.header(), Why: cloudwire.ByeUpdate}) + wait(t, "the bye", func() bool { return gone() == 1 }) + if h.gone[0] != cloudwire.ByeUpdate { + t.Fatalf("gone with %q", h.gone[0]) + } + + _, _ = b.connect(s, nil) + second := *b + second.conn = nil + _, _ = second.connect(s, nil) + if !b.closed() { + t.Fatal("two uplinks for one home") + } + time.Sleep(100 * time.Millisecond) + if gone() != 1 { + t.Fatal("a replaced uplink told push the box was gone") + } + _ = second.conn.Close() + wait(t, "the drop", func() bool { return gone() == 2 }) + if h.gone[1] != "" { + t.Fatalf("a dropped socket gave %q", h.gone[1]) + } + if conns() != 3 { + t.Fatalf("connects %d", conns()) + } +} diff --git a/internal/uplink/uplink.go b/internal/uplink/uplink.go index da1e4f6..a54a7d5 100644 --- a/internal/uplink/uplink.go +++ b/internal/uplink/uplink.go @@ -62,6 +62,18 @@ type Uplink struct { // 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 + + // OnConnect runs after a box's hello_ok, so the app sessions can ask a + // box that just came back for one-second updates. Set before serving. + OnConnect func(home string) + // OnEvent gets each event the box signed, as the envelope it arrived + // in; OnRoster each roster the service keeps; OnGone the end of the + // home's current uplink, with the reason the box gave in its bye, or + // "" when the socket just dropped. Push uses them. Each runs on the + // box's goroutine and must not block. Set before serving. + OnEvent func(home string, envelope []byte) + OnRoster func(home string, r *cloudwire.Roster) + OnGone func(home, why string) } // New builds the handler. key is the service key that signs hello_ok and @@ -130,6 +142,8 @@ type boxConn struct { pub *ecdsa.PublicKey handle *store.Handle purged bool + // bye is the reason the box gave for leaving, if it gave one. + bye string tokens float64 refilled time.Time @@ -192,16 +206,23 @@ func (c *boxConn) run() error { c.u.mu.Unlock() defer func() { c.u.mu.Lock() - if c.u.conns[c.home] == c { + current := c.u.conns[c.home] == c + if current { delete(c.u.conns, c.home) } c.u.mu.Unlock() + if current && c.u.OnGone != nil { + c.u.OnGone(c.home, c.bye) + } }() } if err := c.send(helloOK{BoxNonce: hello.BoxNonce, EKM: c.ekm, Have: map[string]string{}, RosterSeq: c.rosterSeq()}); err != nil { return err } + if !c.purged && c.u.OnConnect != nil { + go c.u.OnConnect(c.home) + } done := make(chan struct{}) defer close(done) @@ -327,6 +348,7 @@ func (c *boxConn) apply(env cloudwire.Envelope) (stop bool, err error) { if err := cloudwire.Unmarshal(env.B, &b); err != nil { return true, err } + c.bye = b.Why if b.Purge { return true, c.purge() } @@ -343,7 +365,13 @@ func (c *boxConn) apply(env cloudwire.Envelope) (stop bool, err error) { if err != nil { return true, err } - return false, c.handle.Put(env.T, raw, h.Seq, env.T == cloudwire.MsgLive && carriesTable(env.B)) + if err := c.handle.Put(env.T, raw, h.Seq, env.T == cloudwire.MsgLive && carriesTable(env.B)); err != nil { + return true, err + } + if env.T == cloudwire.MsgEvent && c.u.OnEvent != nil { + c.u.OnEvent(c.home, raw) + } + return false, nil } // carriesTable reports whether a live body holds its source table. @@ -381,6 +409,9 @@ func (c *boxConn) roster(env cloudwire.Envelope) error { if !c.handle.SetRoster(&r) { return nil // older than what is held: ignored } + if c.u.OnRoster != nil { + c.u.OnRoster(c.home, &r) + } raw, err := env.Encode() if err != nil { return err From bf797744e867f23e6da2ab0d3f4550be9fc6da9c Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 13 Oct 2026 08:43:12 +0200 Subject: [PATCH 2/5] feat(dev): capture alerts to files in development builds -dev-apns-capture writes each alert as JSON instead of sending it to Apple, so an end-to-end test can hand it to a simulator with xcrun simctl push before an APNs key exists. Release builds refuse the flag. Co-Authored-By: Claude Opus 5.5 --- cmd/ftwcloudd/devpush_dev.go | 37 ++++++++++++++++++++++++++++++++++ cmd/ftwcloudd/devpush_other.go | 5 +++++ cmd/ftwcloudd/main.go | 5 +++++ 3 files changed, 47 insertions(+) diff --git a/cmd/ftwcloudd/devpush_dev.go b/cmd/ftwcloudd/devpush_dev.go index eb39653..96317c1 100644 --- a/cmd/ftwcloudd/devpush_dev.go +++ b/cmd/ftwcloudd/devpush_dev.go @@ -7,12 +7,17 @@ import ( "encoding/hex" "encoding/json" "errors" + "fmt" "log/slog" "net" "net/http" + "os" + "path/filepath" "slices" + "sync" "time" + "github.com/srcfl/ftw-cloud/internal/apns" "github.com/srcfl/ftw-cloud/internal/push" "github.com/srcfl/ftw-cloud/internal/registry" "github.com/srcfl/ftw/go/protocol/cloudwire" @@ -70,6 +75,38 @@ func serveDevPush(ctx context.Context, addr string, reg *registry.Registry, svc return nil } +// captureSender writes each alert to dir as .json, {environment, +// deviceToken, payload}, and tells the service Apple took it. A test hands +// the payload to a simulator with xcrun simctl push. +func captureSender(dir string) (push.Sender, error) { + if err := os.MkdirAll(dir, 0o700); err != nil { + return nil, err + } + return &capture{dir: dir}, nil +} + +type capture struct { + mu sync.Mutex + dir string + n int +} + +func (c *capture) Send(_ context.Context, n apns.Notification) (apns.Result, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.n++ + raw, err := json.Marshal(map[string]any{"environment": n.Environment, + "deviceToken": hex.EncodeToString(n.DeviceToken), "payload": json.RawMessage(n.Payload)}) + if err != nil { + return apns.Result{}, err + } + name := filepath.Join(c.dir, fmt.Sprintf("%04d.json", c.n)) + if err := os.WriteFile(name+".tmp", raw, 0o600); err != nil { + return apns.Result{}, err + } + return apns.Result{Status: http.StatusOK}, os.Rename(name+".tmp", name) +} + // rosterDevice finds a device in the roster the box last signed, with its // effective scope. func rosterDevice(reg *registry.Registry, home, id string) (cloudwire.Device, []string, bool) { diff --git a/cmd/ftwcloudd/devpush_other.go b/cmd/ftwcloudd/devpush_other.go index 1a822d6..9b7442f 100644 --- a/cmd/ftwcloudd/devpush_other.go +++ b/cmd/ftwcloudd/devpush_other.go @@ -11,6 +11,11 @@ import ( "github.com/srcfl/ftw-cloud/internal/registry" ) +// captureSender exists only in development builds. +func captureSender(string) (push.Sender, error) { + return nil, errors.New("-dev-apns-capture needs a build with -tags ftwclouddev") +} + // serveDevPush exists only in development builds. func serveDevPush(context.Context, string, *registry.Registry, *push.Service, *slog.Logger) error { return errors.New("-dev-push-listen needs a build with -tags ftwclouddev") diff --git a/cmd/ftwcloudd/main.go b/cmd/ftwcloudd/main.go index 01ba3f0..6f0b86a 100644 --- a/cmd/ftwcloudd/main.go +++ b/cmd/ftwcloudd/main.go @@ -65,6 +65,7 @@ func serve(args []string) error { apnsDir := fs.String("apns-dir", "/run/secrets/apns", "directory with Apple's key.p8, key-id and team-id; without it no alert is sent") apnsTopic := fs.String("apns-topic", "energy.ftw.app", "the app's bundle id") devPush := fs.String("dev-push-listen", "", "loopback address for registering a token by hand (development builds only)") + devCapture := fs.String("dev-apns-capture", "", "write each alert to this directory instead of sending it to Apple (development builds only)") _ = fs.Parse(args) harden() @@ -90,6 +91,10 @@ func serve(args []string) error { switch client, err := loadAPNs(*apnsDir, *apnsTopic); { case err != nil: return err + case *devCapture != "": + if sender, err = captureSender(*devCapture); err != nil { + return err + } case client == nil: log.Warn("APNs off: no key in " + *apnsDir) default: From a9dcf00e26c71def05188b9804797ad3578b7ecd Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 13 Oct 2026 08:54:34 +0200 Subject: [PATCH 3/5] chore(push): say when a box goes away and when it is out of reach Neither line names the home. Co-Authored-By: Claude Opus 5.5 --- internal/push/push.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/internal/push/push.go b/internal/push/push.go index f27e3cf..18642c3 100644 --- a/internal/push/push.go +++ b/internal/push/push.go @@ -223,6 +223,7 @@ func (s *Service) BoxGone(home, why string) { if why == cloudwire.ByeUpdate { wait = time.Duration(cloudwire.UpdateGraceMs) * time.Millisecond } + s.log.Info("a box went away", "bye", why, "out_of_reach_after", wait) gen := st.gen st.stop = s.after(wait, func() { s.unreachable(home, gen) }) } @@ -236,6 +237,7 @@ func (s *Service) unreachable(home string, gen uint64) { } st.told, st.stop = true, nil s.mu.Unlock() + s.log.Info("a box is out of reach; telling its phones") payload, err := cloudwire.AlertPayload(cloudwire.Alert{Home: home, Kind: protocol.PushBoxUnreachable}) if err != nil { return From b70cf44611cbbfd8f2a042d41a3df8b9b24b5825 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 13 Oct 2026 09:08:50 +0200 Subject: [PATCH 4/5] deploy(apns): retry while the network comes up, stop at once without a key Like the service key's fetch in #3: at boot the network and the instance's credentials may lag, so a read is tried for a minute. A parameter that does not exist ends it at once, so a host without a key does not hold Docker back. Co-Authored-By: Claude Opus 5.5 --- deploy/apns.md | 5 +++++ deploy/apns.sh | 40 +++++++++++++++++++++++++--------------- 2 files changed, 30 insertions(+), 15 deletions(-) diff --git a/deploy/apns.md b/deploy/apns.md index 85f6790..2ec8968 100644 --- a/deploy/apns.md +++ b/deploy/apns.md @@ -59,3 +59,8 @@ decrypting through SSM for every principal in the account. Checked on 2026-10-13 with `iam simulate-principal-policy` (`ssm:GetParameter` allowed) and the key policy of `alias/aws/ssm` in eu-north-1. Secrets Manager would need a policy change; Parameter Store does not. + +srcfl/ftw-cloud#3 moves the service key to Secrets Manager with an inline +policy for that one secret. The APNs key could follow it there: one more ARN +in `deploy/host-policy.json`, applied by the owner, and `get-secret-value` +in `apns.sh`. Nothing else changes. diff --git a/deploy/apns.sh b/deploy/apns.sh index ee9041d..ea8b554 100755 --- a/deploy/apns.sh +++ b/deploy/apns.sh @@ -19,27 +19,37 @@ cat > /usr/local/sbin/ftw-cloud-apns <<'FETCH' # Without the parameters ftwcloudd runs and sends no alerts. set -euo pipefail export AWS_PAGER="" -region=eu-north-1 -dir=/run/ftw-cloud/apns umask 077 +dir=/run/ftw-cloud/apns install -d -m 0700 /run/ftw-cloud -rm -rf "$dir.new" -install -d -m 0700 "$dir.new" +new=$(mktemp -d /run/ftw-cloud/apns.XXXXXX) +trap 'rm -rf "$new"' EXIT +# get NAME FILE: 0 when read, 1 when the parameter does not exist. At boot +# the network and the instance's credentials may lag a little. get() { - aws ssm get-parameter --region "$region" --name "$1" --with-decryption \ - --query Parameter.Value --output text > "$dir.new/$2" + local err + for try in $(seq 30); do + if err=$(aws ssm get-parameter --region eu-north-1 --name "$1" --with-decryption \ + --query Parameter.Value --output text 2>&1 >"$new/$2"); then + return 0 + fi + case "$err" in *ParameterNotFound*) return 1 ;; esac + [ "$try" -lt 30 ] || { echo "$err" >&2; exit 1; } + sleep 2 + done } -if get /ftw-cloud/apns/key key.p8 && get /ftw-cloud/apns/key-id key-id && get /ftw-cloud/apns/team-id team-id; then - chown -R 65532:65532 "$dir.new" - chmod 0400 "$dir.new"/* - chmod 0500 "$dir.new" - rm -rf "$dir" - mv "$dir.new" "$dir" - echo "APNs key in place" -else - rm -rf "$dir.new" +if ! get /ftw-cloud/apns/key key.p8 || ! get /ftw-cloud/apns/key-id key-id || ! get /ftw-cloud/apns/team-id team-id; then echo "no APNs key in Parameter Store; ftwcloudd will send no alerts" + exit 0 fi +grep -q '^-----BEGIN PRIVATE KEY-----$' "$new/key.p8" +chown -R 65532:65532 "$new" +chmod 0400 "$new"/* +chmod 0500 "$new" +rm -rf "$dir" +mv "$new" "$dir" +trap - EXIT +echo "APNs key in place" FETCH chmod 0755 /usr/local/sbin/ftw-cloud-apns From cd95664c5ab26959c119d1c393cfad1a26365311 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 13 Oct 2026 09:17:47 +0200 Subject: [PATCH 5/5] feat(dev): send a raw payload through APNs in development builds POST /dev/push/raw on the loopback dev listener sends a payload unchanged to a home's registered phones, so the end-to-end test can put a forged alert through Apple's sandbox and watch the phone's extension refuse it. Co-Authored-By: Claude Opus 5.5 --- README.md | 5 ++++- cmd/ftwcloudd/devpush_dev.go | 41 +++++++++++++++++++++++++++++----- cmd/ftwcloudd/devpush_other.go | 2 +- cmd/ftwcloudd/main.go | 2 +- 4 files changed, 42 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 4b4e07c..d4c9334 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,10 @@ curl http://127.0.0.1:8080/api/cloud/preview the service holds so a test can compare it with the preview, and `-dev-push-listen 127.0.0.1:18444`, where `POST /dev/push/register` with `{home, deviceId, apnsToken (hex), environment}` registers a token for a -device in the home's roster, as the app session will. `-apns-dir` names a +device in the home's roster, as the app session will; `POST /dev/push/raw` +with `{home, payload}` sends a payload as it is, so a test can put a forged +one through Apple. `-dev-apns-capture ` writes alerts to files instead +of sending them. `-apns-dir` names a directory holding `key.p8`, `key-id` and `team-id`; without it nothing is sent. Release builds have neither dev route. diff --git a/cmd/ftwcloudd/devpush_dev.go b/cmd/ftwcloudd/devpush_dev.go index 96317c1..f97ff97 100644 --- a/cmd/ftwcloudd/devpush_dev.go +++ b/cmd/ftwcloudd/devpush_dev.go @@ -23,11 +23,16 @@ import ( "github.com/srcfl/ftw/go/protocol/cloudwire" ) -// serveDevPush listens on loopback for POST /dev/push/register, which -// registers an APNs token for a device in a home's signed roster, as the -// app's session will. It lets an end-to-end test run before the app session -// can. Release builds do not have it. -func serveDevPush(ctx context.Context, addr string, reg *registry.Registry, svc *push.Service, log *slog.Logger) error { +// serveDevPush listens on loopback for two routes. Release builds have +// neither. +// +// - POST /dev/push/register registers an APNs token for a device in a +// home's signed roster, as the app's session will, so an end-to-end test +// can run before the app session can. +// - POST /dev/push/raw {home, payload} sends a payload as it is to the +// home's registered phones, so a test can put a forged one through Apple +// and see the phone refuse it. +func serveDevPush(ctx context.Context, addr string, reg *registry.Registry, svc *push.Service, sender push.Sender, log *slog.Logger) error { host, _, err := net.SplitHostPort(addr) if err != nil || !net.ParseIP(host).IsLoopback() { return errors.New("-dev-push-listen must be a loopback address") @@ -61,6 +66,32 @@ func serveDevPush(ctx context.Context, addr string, reg *registry.Registry, svc } w.WriteHeader(http.StatusNoContent) }) + mux.HandleFunc("POST /dev/push/raw", func(w http.ResponseWriter, r *http.Request) { + var req struct { + Home string `json:"home"` + Payload json.RawMessage `json:"payload"` + } + if json.NewDecoder(r.Body).Decode(&req) != nil || len(req.Payload) == 0 { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + if sender == nil { + http.Error(w, "no APNs key", http.StatusServiceUnavailable) + return + } + regs, err := reg.PushRegs(req.Home, cloudwire.PushKindAlert) + if err != nil || len(regs) == 0 { + http.Error(w, "no phone registered for that home", http.StatusNotFound) + return + } + var out []string + for _, x := range regs { + res, err := sender.Send(r.Context(), apns.Notification{Environment: x.Environment, DeviceToken: x.Token, + Payload: req.Payload, Expiration: time.Now().Add(10 * time.Minute)}) + out = append(out, fmt.Sprintf("%d %s %v", res.Status, res.Reason, err)) + } + _ = json.NewEncoder(w).Encode(out) + }) srv := &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second} go func() { <-ctx.Done() diff --git a/cmd/ftwcloudd/devpush_other.go b/cmd/ftwcloudd/devpush_other.go index 9b7442f..68afd55 100644 --- a/cmd/ftwcloudd/devpush_other.go +++ b/cmd/ftwcloudd/devpush_other.go @@ -17,6 +17,6 @@ func captureSender(string) (push.Sender, error) { } // serveDevPush exists only in development builds. -func serveDevPush(context.Context, string, *registry.Registry, *push.Service, *slog.Logger) error { +func serveDevPush(context.Context, string, *registry.Registry, *push.Service, push.Sender, *slog.Logger) error { return errors.New("-dev-push-listen needs a build with -tags ftwclouddev") } diff --git a/cmd/ftwcloudd/main.go b/cmd/ftwcloudd/main.go index 6f0b86a..b8015a3 100644 --- a/cmd/ftwcloudd/main.go +++ b/cmd/ftwcloudd/main.go @@ -143,7 +143,7 @@ func serve(args []string) error { go srv.Upkeep(ctx, time.Minute) go pushSvc.Run(ctx) if *devPush != "" { - if err := serveDevPush(ctx, *devPush, reg, pushSvc, log); err != nil { + if err := serveDevPush(ctx, *devPush, reg, pushSvc, sender, log); err != nil { return err } }