Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,9 @@ maintains it.
- **No readings on disk.** Memory holds readings, and only while the roster's
grant serves a copy; SQLite holds only the home id, the box key, the signed
roster, owners' suspensions, tombstones, APNs registrations, the hashes of
extension tokens and when each away box's alert is due. Nothing logs
extension tokens when each away box's alert is due, browser notice sent markers and
anonymous daily fleet totals. Browser endpoints and box-encrypted offline
notices live only in the signed roster. Nothing logs
addresses.
- **A phone can only narrow, or talk to its own box.** On `/v1/app` it can
say whether it is looking, an owner can suspend a roster device, and a
Expand Down
16 changes: 16 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -364,3 +364,19 @@ with `go list -deps`, so no control code can reach this build.
`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).

Anonymous box reports use `POST /v1/fleet`. It accepts only the seven
`ftw.fleet/1` fields and reduces each report into bounded daily counts for
90 days. It stores no raw report, address or home id. An operator can read
`/fleet/stats` only on the loopback metrics listener, with
`docker exec ftw-cloud-ftwcloudd-1 ftwcloudd fleet`. These counts measure
reports, not unique boxes.

Browser offline notices use the box's optional signed roster rows. The box
encrypts each notice and signs a short-lived VAPID header; the service holds
no browser encryption key or box signing key. It uses the APNs outage timer,
including update grace and deploy drain, and checks the current device epoch,
events scope and suspension before sending. A readable-copy grant is not
needed. The roster holds the delivery endpoints and ciphertext; a separate
per-home row holds only the current outage's sent ids. Purge removes both.
The sender refuses redirects, proxies and local destinations.
33 changes: 27 additions & 6 deletions cmd/ftwcloudd/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ import (

"github.com/srcfl/ftw-cloud/internal/apns"
"github.com/srcfl/ftw-cloud/internal/appsession"
"github.com/srcfl/ftw-cloud/internal/fleet"
"github.com/srcfl/ftw-cloud/internal/keys"
"github.com/srcfl/ftw-cloud/internal/mcp"
"github.com/srcfl/ftw-cloud/internal/metrics"
Expand All @@ -49,6 +50,7 @@ import (
"github.com/srcfl/ftw-cloud/internal/store"
"github.com/srcfl/ftw-cloud/internal/token"
"github.com/srcfl/ftw-cloud/internal/uplink"
"github.com/srcfl/ftw-cloud/internal/webpush"
"github.com/srcfl/ftw/go/protocol/appwire"
"github.com/srcfl/ftw/go/protocol/cloudwire"
"golang.org/x/crypto/acme/autocert"
Expand All @@ -66,7 +68,7 @@ func main() {
}
return
}
if len(os.Args) > 1 && os.Args[1] == "metrics" {
if len(os.Args) > 1 && (os.Args[1] == "metrics" || os.Args[1] == "fleet") {
if err := printMetrics(os.Args[2:]); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
Expand Down Expand Up @@ -146,9 +148,13 @@ func serve(args []string) error {
} else if n > 0 {
log.Info("rosters restored from the registry", "homes", n, "took", time.Since(restoreStart).Round(time.Millisecond))
}
fleetSvc, err := fleet.New(reg)
if err != nil {
return err
}
tokens := token.New(st, reg, log)
tokens.Shards = shards
srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Log: log}
srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Fleet: fleetSvc, Log: log}

// Assistants: the authorization server, with agent keys in ftwkeysd.
agentKeys := keys.Dial(*keysSocket)
Expand Down Expand Up @@ -183,7 +189,13 @@ func serve(args []string) error {
default:
sender = client
}
webPushSvc, err := webpush.New(reg)
if err != nil {
return err
}
pushSvc := push.New(reg, sender, log)
pushSvc.BrowserUnreachable = webPushSvc.Send
Comment thread
frahlg marked this conversation as resolved.
pushSvc.BrowserHomes = webPushSvc.Homes
pushSvc.WatchApp = *apnsWatch
pushSvc.Owns = shards.Ours
up.OnHeld = pushSvc.Held
Expand Down Expand Up @@ -301,7 +313,7 @@ func serve(args []string) error {
}
}
if *metricsAddr != "" {
if err := serveMetrics(ctx, *metricsAddr); err != nil {
if err := serveMetrics(ctx, *metricsAddr, fleetSvc.Stats); err != nil {
return err
}
}
Expand Down Expand Up @@ -402,7 +414,7 @@ func restartSpread(sockets, rate int, floor time.Duration) time.Duration {
const defaultMetricsAddr = "127.0.0.1:9464"

// serveMetrics answers /metrics on a loopback address until ctx ends.
func serveMetrics(ctx context.Context, addr string) error {
func serveMetrics(ctx context.Context, addr string, fleetStats ...http.HandlerFunc) error {
host, _, err := net.SplitHostPort(addr)
if err != nil {
return fmt.Errorf("metrics-listen: %w", err)
Expand All @@ -414,7 +426,12 @@ func serveMetrics(ctx context.Context, addr string) error {
if err != nil {
return err
}
m := &http.Server{Handler: metrics.Default, ReadHeaderTimeout: 5 * time.Second, ErrorLog: log0()}
mux := http.NewServeMux()
mux.Handle("/metrics", metrics.Default)
if len(fleetStats) > 0 {
mux.HandleFunc("/fleet/stats", fleetStats[0])
}
m := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second, ErrorLog: log0()}
go func() { _ = m.Serve(ln) }()
go func() {
<-ctx.Done()
Expand All @@ -429,7 +446,11 @@ func printMetrics(args []string) error {
addr := fs.String("addr", defaultMetricsAddr, "where the service answers /metrics")
_ = fs.Parse(args)
c := &http.Client{Timeout: 5 * time.Second}
resp, err := c.Get("http://" + *addr + "/metrics")
path := "/metrics"
if len(os.Args) > 1 && os.Args[1] == "fleet" {
path = "/fleet/stats"
}
resp, err := c.Get("http://" + *addr + path)
if err != nil {
return err
}
Expand Down
2 changes: 1 addition & 1 deletion go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ require (
github.com/fxamacker/cbor/v2 v2.9.4
github.com/gorilla/websocket v1.5.3
github.com/quic-go/quic-go v0.63.0
github.com/srcfl/ftw/go/protocol v0.0.0-20261016102129-7fb84d6ed573
github.com/srcfl/ftw/go/protocol v0.0.0-20261017070253-cbdd8a587105
golang.org/x/crypto v0.57.0
golang.org/x/net v0.58.0
golang.org/x/sys v0.48.0
Expand Down
2 changes: 2 additions & 0 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,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-20261016102129-7fb84d6ed573 h1:1OOC4RdeERT9r6cLMTlcmTjg8+ML2kU6hCvhx/x+8xI=
github.com/srcfl/ftw/go/protocol v0.0.0-20261016102129-7fb84d6ed573/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU=
github.com/srcfl/ftw/go/protocol v0.0.0-20261017070253-cbdd8a587105 h1:56AF3FQxXwfTVacD+AtJQM96TdzntTeQy1f4yiN54ic=
github.com/srcfl/ftw/go/protocol v0.0.0-20261017070253-cbdd8a587105/go.mod h1:VEjr/o0yZijJ1nApuJK+nak6FxJ/fubAUBNOunqNrUU=
github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg=
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
Expand Down
98 changes: 98 additions & 0 deletions internal/fleet/fleet.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
// Package fleet reduces anonymous box reports to bounded daily counts.
package fleet

import (
"encoding/json"
"io"
"net/http"
"sync"
"time"

"github.com/srcfl/ftw-cloud/internal/registry"
"github.com/srcfl/ftw/go/protocol/fleetwire"
)

type Service struct {
reg *registry.Registry
now func() time.Time
mu sync.Mutex
at time.Time
tokens float64
}

func New(reg *registry.Registry) (*Service, error) {
if err := reg.FleetInit(); err != nil {
return nil, err
}
if err := reg.FleetPrune(time.Now()); err != nil {
return nil, err
}
return &Service{reg: reg, now: time.Now, tokens: 120}, nil
}

// ServeHTTP accepts reports only. Public requests never read totals.
func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Cache-Control", "no-store")
if r.Method != http.MethodPost {
w.Header().Set("Allow", http.MethodPost)
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
if !s.allow() {
w.WriteHeader(http.StatusTooManyRequests)
return
}
data, err := io.ReadAll(http.MaxBytesReader(w, r.Body, fleetwire.MaxBodyBytes))
if err != nil {
w.WriteHeader(http.StatusRequestEntityTooLarge)
return
}
report, err := fleetwire.Decode(data)
if err != nil {
w.WriteHeader(http.StatusBadRequest)
return
}
ok, err := s.reg.FleetPut(report, s.now())
switch {
case err != nil:
w.WriteHeader(http.StatusServiceUnavailable)
case !ok:
w.WriteHeader(http.StatusTooManyRequests)
default:
w.WriteHeader(http.StatusNoContent)
}
}

// Stats belongs only on the operator's loopback listener.
func (s *Service) Stats(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Cache-Control", "no-store")
if r.Method != http.MethodGet {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
view, err := s.reg.FleetView(s.now())
if err != nil {
w.WriteHeader(http.StatusServiceUnavailable)
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(view)
}

// One shared budget stores no addresses and admits five reports a second.
func (s *Service) allow() bool {
s.mu.Lock()
defer s.mu.Unlock()
now := s.now()
if !s.at.IsZero() {
s.tokens = min(120, s.tokens+max(0, now.Sub(s.at).Seconds())*5)
}
s.at = now
if s.tokens < 1 {
return false
}
s.tokens--
return true
}

func (s *Service) Prune(now time.Time) error { return s.reg.FleetPrune(now) }
92 changes: 92 additions & 0 deletions internal/fleet/fleet_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
package fleet

import (
"bytes"
"encoding/json"
"github.com/srcfl/ftw-cloud/internal/registry"
"github.com/srcfl/ftw/go/protocol/fleetwire"
"net/http"
"net/http/httptest"
"path/filepath"
"sync"
"testing"
"time"
)

func TestReportsBecomeOnlyBoundedDailyTotalsAcrossARestart(t *testing.T) {
path := filepath.Join(t.TempDir(), "registry.db")
reg, err := registry.Open(path)
if err != nil {
t.Fatal(err)
}
svc, err := New(reg)
if err != nil {
t.Fatal(err)
}
now := time.Now().UTC().Truncate(24 * time.Hour)
svc.now = func() time.Time { return now }
report := fleetwire.Report{Schema: fleetwire.Schema, Version: "v0.146.0", Channel: "stable", Drivers: []string{}, Battery: "none", Zone: "SE4", Age: "unknown"}
raw, _ := json.Marshal(report)
var wg sync.WaitGroup
for range 100 {
wg.Go(func() {
w := httptest.NewRecorder()
r := httptest.NewRequest("POST", "/v1/fleet", bytes.NewReader(raw))
r.RemoteAddr = "198.51.100.8:443"
svc.ServeHTTP(w, r)
if w.Code != 204 {
t.Errorf("status %d", w.Code)
}
})
}
wg.Wait()
if err := reg.Close(); err != nil {
t.Fatal(err)
}
reg, err = registry.Open(path)
if err != nil {
t.Fatal(err)
}
defer reg.Close()
svc, err = New(reg)
if err != nil {
t.Fatal(err)
}
view, err := reg.FleetView(now)
if err != nil || len(view.Days) != 1 || view.Days[0].Reports != 100 {
t.Fatal(view, err)
}
for i := 1; i < 92; i++ {
if ok, err := reg.FleetPut(report, now.AddDate(0, 0, i)); err != nil || !ok {
t.Fatal(err)
}
}
view, _ = reg.FleetView(now.AddDate(0, 0, 91))
if len(view.Days) != 90 {
t.Fatal(len(view.Days))
}
// A long gap expires old populated dates, even without another report.
if err := reg.FleetPrune(now.AddDate(1, 0, 0)); err != nil {
t.Fatal(err)
}
view, _ = reg.FleetView(now.AddDate(1, 0, 0))
if len(view.Days) != 0 {
t.Fatal("old totals survived", view)
}
for _, tc := range []struct {
method, body string
code int
}{{"GET", "", 405}, {"POST", `{"home":"x"}`, 400}, {"POST", string(make([]byte, 4097)), 413}} {
w := httptest.NewRecorder()
svc.ServeHTTP(w, httptest.NewRequest(tc.method, "/v1/fleet", bytes.NewBufferString(tc.body)))
if w.Code != tc.code {
t.Fatal(w.Code, tc)
}
}
w := httptest.NewRecorder()
svc.Stats(w, httptest.NewRequest(http.MethodGet, "/fleet/stats", nil))
if w.Code != 200 || bytes.Contains(w.Body.Bytes(), []byte("198.51.100")) {
t.Fatal(w.Code, w.Body.String())
}
// The only extra table holds date and aggregate JSON, not source addresses.
}
Loading
Loading