From 8d8315bf45e70877287d49c32d7c4dc313525770 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Sat, 10 Oct 2026 15:05:05 +0800 Subject: [PATCH] Check and record quotas for sends and local delivery Mail between local addresses never passes through fmsgd, so it was neither checked against the recipients' receive limits nor recorded, and nothing checked or recorded send limits. Local delivery now makes fmsgd's checks for each local recipient: one more message of the message's stored size (body plus attachments) must fit its messages per day, bytes per day and total bytes received. A recipient over a limit isn't delivered and gets response code 101 (user full), as fmsgd records it, alongside 100 and 102 as before. Each delivered recipient is recorded with POST /fmsgid/recv. Send and react check the sender's per-message size, messages per day, bytes per day and total bytes sent before sending, refusing with 413, 429 or 422 and a code naming the limit, and record the message once with POST /fmsgid/send. A reaction is a message like any other. Add-to doesn't send a new message, so it isn't counted as a send, but its new local recipients are checked and counted. The checks fetch fmsgid's detail fresh each time instead of using the 30s lookup cache, since usage changes with every message. Failing to record usage is logged and doesn't fail the request. Co-Authored-By: Claude Opus 5.5 --- README.md | 60 +++- .../handlers/finalization_integration_test.go | 14 +- internal/handlers/messages.go | 65 +++-- internal/handlers/quota.go | 141 +++++++++ internal/handlers/quota_integration_test.go | 272 ++++++++++++++++++ internal/handlers/quota_test.go | 93 ++++++ internal/handlers/reactions.go | 7 + internal/middleware/fmsgid_usage.go | 113 ++++++++ internal/middleware/fmsgid_usage_test.go | 111 +++++++ 9 files changed, 851 insertions(+), 25 deletions(-) create mode 100644 internal/handlers/quota.go create mode 100644 internal/handlers/quota_integration_test.go create mode 100644 internal/handlers/quota_test.go create mode 100644 internal/middleware/fmsgid_usage.go create mode 100644 internal/middleware/fmsgid_usage_test.go diff --git a/README.md b/README.md index 1d2c2be..e35972c 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,7 @@ HTTP API providing user/client message handling for an fmsg host. Exposes CRUD o - [Running](#running) - [TLS mode (production)](#tls-mode-production) - [Plain HTTP mode (development / reverse proxy)](#plain-http-mode-development--reverse-proxy) +- [Quotas](#quotas) - [API Routes](#api-routes) - [Token and sub-account routes](#post-fmsgtoken) - [WebSocket route](#get-fmsgws) @@ -289,6 +290,38 @@ transfers are not dropped prematurely. These timeouts do not apply to `/fmsg/ws` connections: once upgraded, a WebSocket connection is hijacked from the HTTP server and kept alive by its own ping/pong heartbeat. +## Quotas + +fmsgid holds each address's limits and usage (`-1` is unlimited; an address in +a quota pool has its pool owner's limits and the whole pool's usage). fmsgd +checks and records mail it receives from other hosts. This service checks and +records the rest: the mail it sends, and the mail it delivers between local +addresses, which never passes through fmsgd. + +- **Sending.** Before a message is sent (`POST /fmsg/:id/send`, and the + reaction message `POST /fmsg/:id/react` sends), the sender's current limits + and usage are fetched from fmsgid and the message refused if it is bigger + than the per-message send size (`413`), or would exceed the messages sent per + day or bytes sent per day (`429`) or total bytes sent (`422`). Once sent, the + message is recorded once with `POST /fmsgid/send`, however many recipients it + has. Adding recipients with add-to doesn't send a new message, so it isn't + counted against send limits; each new local recipient is checked and counted + when it receives the message. +- **Local delivery.** Each local recipient gets the checks fmsgd makes: unknown + (`100`), not accepting new messages (`102`), or over its messages received per + day, bytes received per day or total bytes received for one more message of + this size (`101`, user full). Those recipients aren't delivered, and their + code is recorded as their `response_code` in `to_delivery`/`add_to`. Each + recipient that is delivered is recorded with `POST /fmsgid/recv`. + +Sizes are the stored size: the body plus every attachment. Usage is +timestamped when this service sent or delivered the message. The checks use +fmsgid's current usage, never the lookup cached for authentication, which only +holds whether the address exists and accepts new messages. If fmsgid can't be +reached a send is refused with `503`, and a local recipient is left pending. +Failing to record usage is logged and never fails the request, since the +message has already been sent. + ## API Routes All routes are prefixed with `/fmsg`. `POST /fmsg/token` accepts an API key and @@ -755,6 +788,14 @@ third-party delivery's outcome, so the reply is attempted and the wire's originating domain always passes (it retains its outgoing messages). Local recipients are unaffected. +The sender's send limits are then checked against the message's size, the +body plus its attachments (see [Quotas](#quotas)). A refusal leaves the +message a draft, and carries a `code` naming the limit: +`{"error": "daily message limit reached", "code": "send_count_per_1d_limit"}`. + +Local recipients are delivered at once, each subject to its receive limits; +a recipient over one is not delivered and its `response_code` is `101`. + **Response:** `200 OK` with `{"id": , "time": , "sha256": "<64 hex characters>"}`. **Errors:** @@ -762,10 +803,14 @@ recipients are unaffected. | Status | Condition | | ------ | --------- | | `400` | Message is not sendable (no recipients, invalid recipient address, no type, unsupported version) | -| `403` | Not the owner | +| `403` | Not the owner, or the sender is no longer in fmsgid or accepting new messages | | `404` | Message not found | | `409` | Message already sent | | `409` | Reply can never be accepted by one or more remote recipient domains (parent never addressed there, or its delivery there failed) | +| `413` | Message is bigger than the sender's per-message send size (`send_size_per_msg_limit`) | +| `422` | Sender's total sent bytes would exceed its limit, `storage limit reached` (`send_size_total_limit`) | +| `429` | Sender's messages per day (`send_count_per_1d_limit`) or bytes per day (`send_size_per_1d_limit`) would exceed its limit | +| `503` | fmsgid unavailable, so the send limits can't be checked | ### POST `/fmsg/:id/read` @@ -795,6 +840,11 @@ This endpoint records the add-to as a new `msg_add_to_batch` row (capturing the New addresses must be distinct among themselves (case-insensitive). +On a sent message, new local recipients are delivered at once, each subject to +its receive limits (a recipient over one gets `response_code` `101`). Adding +recipients isn't counted against the caller's send limits (see +[Quotas](#quotas)). + **Response:** `200 OK` with `{"id": , "added": , "batch_id": , "sha256": "<64 hex characters>"}`. On a draft, the batch hash is `null` until send finalizes it. **Errors:** @@ -816,8 +866,10 @@ in `to` or `add_to` — and the message must be sent and not terminal. The reaction is sent immediately as a reaction message: a reply to `:id` with `no_reply` and `terminal` set, type `text/plain;charset=UTF-8`, body the emoji, addressed to every other participant of the message. Delivery then proceeds -exactly as for any sent message. A reaction message cannot be replied to, -reacted to, or have recipients added. +exactly as for any sent message, and like any message it counts against the +reactor's send limits and each local recipient's receive limits (see +[Quotas](#quotas)). A reaction message cannot be replied to, reacted to, or +have recipients added. **Request body (JSON):** @@ -839,6 +891,8 @@ the caller has no reaction returns `200 OK` with `null` values. | `403` | Authenticated identity is not a participant | | `404` | Message not found | | `409` | Message is a draft, is terminal, has no other participants, or a remote recipient host cannot accept the reaction because it does not hold the message | +| `413`, `422`, `429` | The reaction would exceed one of the reactor's send limits, as for send | +| `503` | fmsgid unavailable, so the send limits can't be checked | ### GET `/fmsg/:id/data` diff --git a/internal/handlers/finalization_integration_test.go b/internal/handlers/finalization_integration_test.go index 30d3d4d..64bd01b 100644 --- a/internal/handlers/finalization_integration_test.go +++ b/internal/handlers/finalization_integration_test.go @@ -30,6 +30,15 @@ type finalizationAPI struct { } func newFinalizationAPI(t *testing.T) *finalizationAPI { + t.Helper() + return newFinalizationAPIWithID(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"acceptingNew":true}`) + })) +} + +// newFinalizationAPIWithID is newFinalizationAPI with fmsgid served by fmsgID. +func newFinalizationAPIWithID(t *testing.T, fmsgID http.Handler) *finalizationAPI { t.Helper() dsn := os.Getenv("FMSG_TEST_DATABASE_URL") if dsn == "" { @@ -66,10 +75,7 @@ func newFinalizationAPI(t *testing.T) *finalizationAPI { if _, err = pool.Exec(ctx, string(sql)); err != nil { t.Fatal(err) } - id := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - fmt.Fprint(w, `{"acceptingNew":true}`) - })) + id := httptest.NewServer(fmsgID) t.Cleanup(id.Close) h := NewMessageHandler(&db.DB{Pool: pool}, t.TempDir(), 1<<20, 2<<20, 256, nil, id.URL, "example.com") a := NewAttachmentHandler(h.DB, h.DataDir, 1<<20, 2<<20) diff --git a/internal/handlers/messages.go b/internal/handlers/messages.go index c660add..090e9e9 100644 --- a/internal/handlers/messages.go +++ b/internal/handlers/messages.go @@ -112,40 +112,56 @@ func formatDeliveredISO(t *float64) *string { // skips the local domain (no network hop is needed), so it never sets // time_delivered/response_code for same-domain recipients — fmsg-webapi has // to resolve that status itself via fmsgid, the same source fmsgd would have -// consulted. Errors talking to fmsgid are logged and left pending (no status -// written) rather than treated as a delivery failure. +// consulted, with the same checks: unknown (100), not accepting (102), or +// over a receive limit (101, user full) for one more message of the message's +// stored size. A delivered recipient's usage is recorded with fmsgid, at the +// delivery time. fmsgid is asked afresh for every recipient (not the cached +// lookup), since usage changes with every message. Errors talking to fmsgid +// are logged and left pending (no status written) rather than treated as a +// delivery failure; a failure to record usage is only logged. func (h *MessageHandler) resolveLocalDelivery(ctx context.Context, table string, msgID int64, localDomain string, addrs []string) { now := float64(time.Now().UnixMicro()) / 1e6 query := "UPDATE " + table + " SET time_delivered = $1, response_code = $2 WHERE msg_id = $3 AND addr = $4 AND time_delivered IS NULL AND response_code IS NULL" + size := int64(-1) // the message's stored size, loaded once it is needed for _, addr := range addrs { _, domain := parseAddr(addr) if !strings.EqualFold(domain, localDomain) { continue // remote — fmsgd handles this } - code, accepting, err := middleware.CheckFmsgID(h.IDURL, addr) + detail, status, err := middleware.FetchFmsgIDDetail(h.IDURL, addr) if err != nil { log.Printf("resolve local delivery: fmsgid check %s: %v", addr, err) continue } + if size < 0 && status == http.StatusOK { + if size, err = h.messageStoredSize(ctx, msgID); err != nil { + log.Printf("resolve local delivery: size of msg %d: %v", msgID, err) + size = -1 + continue + } + } + code, ok := localDeliveryCode(status, detail, size) + if !ok { + log.Printf("resolve local delivery: unexpected fmsgid status %d for %s", status, addr) + continue + } var delivered *float64 - var responseCode *int - switch { - case code == http.StatusNotFound: - rc := 100 // user unknown - responseCode = &rc - case code == http.StatusOK && !accepting: - rc := 102 // user not accepting - responseCode = &rc - case code == http.StatusOK && accepting: - rc := 200 // accept - responseCode = &rc + if code == responseCodeAccept { delivered = &now - default: - log.Printf("resolve local delivery: unexpected fmsgid status %d for %s", code, addr) - continue + } else if code == responseCodeUserFull { + log.Printf("resolve local delivery: msg %d to %s refused: over a receive limit", msgID, addr) } - if _, err := h.q().Exec(ctx, query, delivered, responseCode, msgID, addr); err != nil { + tag, err := h.q().Exec(ctx, query, delivered, code, msgID, addr) + if err != nil { log.Printf("resolve local delivery: update %s msg %d addr %s: %v", table, msgID, addr, err) + continue + } + // Record usage only for the delivery made here, never for a row + // another request had already resolved. + if delivered != nil && tag.RowsAffected() > 0 { + if err := middleware.RecordFmsgIDRecv(h.IDURL, addr, now, size); err != nil { + log.Printf("resolve local delivery: record recv usage %s msg %d: %v", addr, msgID, err) + } } } } @@ -961,6 +977,18 @@ func (h *MessageHandler) Send(c *gin.Context) { } } + // The sender's send limits, for the message's stored size (body plus + // attachments). The message is counted once, however many recipients. + size, err := h.messageStoredSize(ctx, msgID) + if err != nil { + log.Printf("send message %d: size: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve message"}) + return + } + if !h.checkSendQuota(c, existing.From, size) { + return + } + now := float64(time.Now().UnixMicro()) / 1e6 tx, ok := h.query.(pgx.Tx) if !ok { @@ -981,6 +1009,7 @@ func (h *MessageHandler) Send(c *gin.Context) { afterCommit(c, func() { plain := *h plain.query = nil + plain.recordSend(existing.From, now, size) plain.resolveLocalDelivery(ctx, "msg_to", msgID, h.LocalDomain, existing.To) for _, b := range existing.AddTo { plain.resolveLocalDelivery(ctx, "msg_add_to", msgID, h.LocalDomain, b.To) diff --git a/internal/handlers/quota.go b/internal/handlers/quota.go new file mode 100644 index 0000000..c05355d --- /dev/null +++ b/internal/handlers/quota.go @@ -0,0 +1,141 @@ +package handlers + +import ( + "context" + "fmt" + "log" + "net/http" + + "github.com/gin-gonic/gin" + "github.com/markmnl/fmsg-webapi/internal/middleware" +) + +// Quotas. fmsgid holds each address's limits and usage; this service checks +// and records the messages that don't pass through fmsgd's receive path: +// +// - Sending (send and react): the sender's send limits are checked before +// the message is sent, and the message is recorded once with POST +// /fmsgid/send after it is sent, whatever its number of recipients. +// - Local delivery (see resolveLocalDelivery): each local recipient's +// receive limits are checked as fmsgd checks them, a recipient over a +// limit is recorded with response code 101 (user full) and not delivered, +// and each delivered recipient is recorded with POST /fmsgid/recv. +// +// Every size is the stored size: the message body plus its attachments. +// Usage is timestamped when this service sent or delivered the message. +// Recording failures are logged and never fail the request, because the +// message has already been sent. + +// Per-recipient response codes (fmsg specification) that local delivery +// records in msg_to/msg_add_to.response_code, as fmsgd does for recipients of +// the messages it receives. +const ( + responseCodeUserUnknown = 100 + responseCodeUserFull = 101 + responseCodeUserNotAccepting = 102 + responseCodeAccept = 200 +) + +// recvLimitExceeded reports whether one more message of size bytes would take +// an address over any of its receive limits (-1 is unlimited). These are the +// checks fmsgd makes for a recipient before accepting a message. +func recvLimitExceeded(d *middleware.FmsgIDDetail, size int64) bool { + return (d.LimitRecvCountPer1d > -1 && d.RecvCountPer1d+1 > d.LimitRecvCountPer1d) || + (d.LimitRecvSizePer1d > -1 && d.RecvSizePer1d+size > d.LimitRecvSizePer1d) || + (d.LimitRecvSizeTotal > -1 && d.RecvSizeTotal+size > d.LimitRecvSizeTotal) +} + +// localDeliveryCode is the response code local delivery records for a +// recipient, given fmsgid's status and detail for it and the message's stored +// size. ok is false for an unexpected fmsgid status, which leaves the +// recipient pending. +func localDeliveryCode(status int, d *middleware.FmsgIDDetail, size int64) (code int, ok bool) { + switch { + case status == http.StatusNotFound: + return responseCodeUserUnknown, true + case status != http.StatusOK || d == nil: + return 0, false + case !d.AcceptingNew: + return responseCodeUserNotAccepting, true + case recvLimitExceeded(d, size): + return responseCodeUserFull, true + default: + return responseCodeAccept, true + } +} + +// sendLimitError is the refusal for a message over one of its sender's send +// limits. +type sendLimitError struct { + status int + code string + message string +} + +// checkSendLimits checks one more message of size bytes against a sender's +// send limits (-1 is unlimited), and returns the refusal for the first limit +// it would exceed, or nil. A message too big to ever send is 413, a daily +// limit is 429 (it frees up as the day's messages age out), and the sent-mail +// storage limit is 422. +func checkSendLimits(d *middleware.FmsgIDDetail, size int64) *sendLimitError { + switch { + case d.LimitSendSizePerMsg > -1 && size > d.LimitSendSizePerMsg: + return &sendLimitError{http.StatusRequestEntityTooLarge, "send_size_per_msg_limit", + fmt.Sprintf("message size %d exceeds the sending limit of %d bytes per message", size, d.LimitSendSizePerMsg)} + case d.LimitSendCountPer1d > -1 && d.SendCountPer1d+1 > d.LimitSendCountPer1d: + return &sendLimitError{http.StatusTooManyRequests, "send_count_per_1d_limit", "daily message limit reached"} + case d.LimitSendSizePer1d > -1 && d.SendSizePer1d+size > d.LimitSendSizePer1d: + return &sendLimitError{http.StatusTooManyRequests, "send_size_per_1d_limit", "daily sending size limit reached"} + case d.LimitSendSizeTotal > -1 && d.SendSizeTotal+size > d.LimitSendSizeTotal: + return &sendLimitError{http.StatusUnprocessableEntity, "send_size_total_limit", "storage limit reached"} + } + return nil +} + +// checkSendQuota checks that sender may send one more message of size bytes, +// with fmsgid's current usage (never the cached lookup). Otherwise it answers +// with the refusal and returns false. +func (h *MessageHandler) checkSendQuota(c *gin.Context, sender string, size int64) bool { + detail, status, err := middleware.FetchFmsgIDDetail(h.IDURL, sender) + if err != nil { + log.Printf("send quota: fmsgid detail %s: %v", sender, err) + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "identity service unavailable"}) + return false + } + switch { + case status == http.StatusNotFound: + c.JSON(http.StatusForbidden, gin.H{"error": fmt.Sprintf("User %s not found", sender)}) + return false + case status != http.StatusOK: + log.Printf("send quota: unexpected fmsgid status %d for %s", status, sender) + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "identity service unavailable"}) + return false + case !detail.AcceptingNew: + c.JSON(http.StatusForbidden, gin.H{"error": fmt.Sprintf("User %s not authorised to send new messages", sender)}) + return false + } + if refusal := checkSendLimits(detail, size); refusal != nil { + log.Printf("send quota: %s refused: %s", sender, refusal.code) + c.JSON(refusal.status, gin.H{"error": refusal.message, "code": refusal.code}) + return false + } + return true +} + +// recordSend records one sent message of size bytes for sender at ts. A +// failure is logged: the message has been sent. +func (h *MessageHandler) recordSend(sender string, ts float64, size int64) { + if err := middleware.RecordFmsgIDSend(h.IDURL, sender, ts, size); err != nil { + log.Printf("record send usage: %s: %v", sender, err) + } +} + +// messageStoredSize is a message's stored size: its body plus its +// attachments. +func (h *MessageHandler) messageStoredSize(ctx context.Context, msgID int64) (int64, error) { + var size int64 + err := h.q().QueryRow(ctx, + `SELECT m.size + COALESCE((SELECT sum(a.filesize) FROM msg_attachment a WHERE a.msg_id = m.id), 0) + FROM msg m WHERE m.id = $1`, msgID).Scan(&size) + return size, err +} diff --git a/internal/handlers/quota_integration_test.go b/internal/handlers/quota_integration_test.go new file mode 100644 index 0000000..35bdf91 --- /dev/null +++ b/internal/handlers/quota_integration_test.go @@ -0,0 +1,272 @@ +package handlers + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "sync" + "testing" + "time" +) + +// quotaFmsgID serves fmsgid's address detail from per-address overrides +// (unlimited and accepting by default; "@nobody@…" is unknown) and records the +// usage posted to it. +type quotaFmsgID struct { + mu sync.Mutex + details map[string]map[string]any + down bool + usage []postedUsage +} + +type postedUsage struct { + Kind string // "recv" or "send" + Address string `json:"address"` + Ts float64 `json:"ts"` + Size int64 `json:"size"` +} + +func newQuotaFmsgID() *quotaFmsgID { return "aFmsgID{details: map[string]map[string]any{}} } + +func (f *quotaFmsgID) set(addr string, fields map[string]any) { + f.mu.Lock() + defer f.mu.Unlock() + f.details[strings.ToLower(addr)] = fields +} + +func (f *quotaFmsgID) posted(kind string) []postedUsage { + f.mu.Lock() + defer f.mu.Unlock() + var out []postedUsage + for _, u := range f.usage { + if u.Kind == kind { + out = append(out, u) + } + } + return out +} + +func (f *quotaFmsgID) ServeHTTP(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + defer f.mu.Unlock() + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == http.MethodPost && (r.URL.Path == "/fmsgid/recv" || r.URL.Path == "/fmsgid/send"): + u := postedUsage{Kind: strings.TrimPrefix(r.URL.Path, "/fmsgid/")} + if err := json.NewDecoder(r.Body).Decode(&u); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + f.usage = append(f.usage, u) + case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/fmsgid/"): + if f.down { + w.WriteHeader(http.StatusInternalServerError) + return + } + addr := strings.ToLower(strings.TrimPrefix(r.URL.Path, "/fmsgid/")) + if strings.HasPrefix(addr, "@nobody@") { + w.WriteHeader(http.StatusNotFound) + return + } + detail := map[string]any{"address": addr, "acceptingNew": true} + for k, v := range f.details[addr] { + detail[k] = v + } + _ = json.NewEncoder(w).Encode(detail) + default: + w.WriteHeader(http.StatusNotFound) + } +} + +// sendDraft creates alice's draft to `to` with body, and an attachment when +// attach is set, returning its id. +func (a *finalizationAPI) sendDraft(t *testing.T, to []string, body string, attach bool) string { + t.Helper() + v := jsonResponse(t, a.request("@alice@example.com", "POST", "/fmsg", map[string]any{ + "version": 1, "from": "@alice@example.com", "to": to, "type": "text/plain;charset=UTF-8", "data": body, + }), 201) + id := fmt.Sprintf("%.0f", v["id"]) + if attach { + a.upload(t, id) // "note.txt", 16 bytes + } + return id +} + +type recipientRow struct { + delivered bool + code *int +} + +func (a *finalizationAPI) recipientRows(t *testing.T, table, id string) map[string]recipientRow { + t.Helper() + rows, err := a.pool.Query(context.Background(), + "SELECT addr, time_delivered IS NOT NULL, response_code FROM "+table+" WHERE msg_id = $1", mustID(t, id)) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + out := map[string]recipientRow{} + for rows.Next() { + var addr string + var r recipientRow + var code *int16 + if err := rows.Scan(&addr, &r.delivered, &code); err != nil { + t.Fatal(err) + } + if code != nil { + c := int(*code) + r.code = &c + } + out[addr] = r + } + return out +} + +func wantRow(t *testing.T, rows map[string]recipientRow, addr string, delivered bool, code int) { + t.Helper() + r, ok := rows[addr] + if !ok || r.delivered != delivered || r.code == nil || *r.code != code { + t.Fatalf("%s: got %+v (code %v), want delivered=%v code=%d", addr, r, r.code, delivered, code) + } +} + +func TestLocalDeliveryChecksAndRecordsReceiveQuota(t *testing.T) { + id := newQuotaFmsgID() + a := newFinalizationAPIWithID(t, id) + bob, carol, dave := "@bob@example.com", "@carol@example.com", "@dave@example.com" + id.set(carol, map[string]any{"limitRecvCountPer1d": 3, "recvCountPer1d": 3}) + // dave's storage has room for the body but not the attachment + id.set(dave, map[string]any{"limitRecvSizeTotal": 100, "recvSizeTotal": 80}) + + start := float64(time.Now().Unix()) + msg := a.sendDraft(t, []string{bob, carol, dave, "@nobody@example.com", "@remote@other.example"}, "hello", true) + const size = 5 + 16 // body + attachment + jsonResponse(t, a.request("@alice@example.com", "POST", "/fmsg/"+msg+"/send", nil), 200) + + rows := a.recipientRows(t, "msg_to", msg) + wantRow(t, rows, bob, true, responseCodeAccept) + wantRow(t, rows, carol, false, responseCodeUserFull) + wantRow(t, rows, dave, false, responseCodeUserFull) + wantRow(t, rows, "@nobody@example.com", false, responseCodeUserUnknown) + if r := rows["@remote@other.example"]; r.delivered || r.code != nil { + t.Fatalf("remote recipient resolved locally: %+v", r) + } + + // The sender sees each local recipient's outcome. + got := jsonResponse(t, a.request("@alice@example.com", "GET", "/fmsg/"+msg, nil), 200) + codes := map[string]any{} + for _, d := range got["to_delivery"].([]any) { + d := d.(map[string]any) + codes[d["addr"].(string)] = d["response_code"] + } + if codes[carol] != float64(101) || codes[bob] != float64(200) { + t.Fatalf("to_delivery codes: %v", codes) + } + + recv := id.posted("recv") + if len(recv) != 1 || recv[0].Address != bob || recv[0].Size != size || recv[0].Ts < start { + t.Fatalf("recv usage: %+v, want one for %s of %d bytes", recv, bob, size) + } + send := id.posted("send") + if len(send) != 1 || send[0].Address != "@alice@example.com" || send[0].Size != size || send[0].Ts < start { + t.Fatalf("send usage: %+v, want one for alice of %d bytes", send, size) + } + + // add-to: the new local recipients' receive limits are checked and their + // usage recorded; adding recipients sends no new message. + erin, frank := "@erin@example.com", "@frank@example.com" + id.set(frank, map[string]any{"limitRecvSizePer1d": 10}) + jsonResponse(t, a.request("@alice@example.com", "POST", "/fmsg/"+msg+"/add-to", map[string]any{"add_to": []string{erin, frank}}), 200) + rows = a.recipientRows(t, "msg_add_to", msg) + wantRow(t, rows, erin, true, responseCodeAccept) + wantRow(t, rows, frank, false, responseCodeUserFull) + if recv := id.posted("recv"); len(recv) != 2 || recv[1].Address != erin || recv[1].Size != size { + t.Fatalf("recv usage after add-to: %+v", recv) + } + if send := id.posted("send"); len(send) != 1 { + t.Fatalf("add-to recorded send usage: %+v", send) + } + + // A reaction is a message: sender and local recipients are counted. + jsonResponse(t, a.request(bob, "POST", "/fmsg/"+msg+"/react", map[string]any{"emoji": "👍"}), 201) + send = id.posted("send") + if len(send) != 2 || send[1].Address != bob || send[1].Size != int64(len("👍")) { + t.Fatalf("reaction send usage: %+v", send) + } + // carol is still full, and nobody unknown; alice, dave, erin and frank + // each have room for the reaction. + recv = id.posted("recv")[2:] + gotRecv := map[string]int64{} + for _, u := range recv { + gotRecv[u.Address] = u.Size + } + want := map[string]int64{"@alice@example.com": 4, dave: 4, erin: 4, frank: 4} + if fmt.Sprint(gotRecv) != fmt.Sprint(want) || len(recv) != len(want) { + t.Fatalf("reaction recv usage: %+v, want %v", recv, want) + } +} + +func TestSendRefusedOverSendQuota(t *testing.T) { + id := newQuotaFmsgID() + a := newFinalizationAPIWithID(t, id) + alice, bob := "@alice@example.com", "@bob@example.com" + tests := []struct { + name string + limits map[string]any + status int + code string + error string + }{ + {"daily count", map[string]any{"limitSendCountPer1d": 10, "sendCountPer1d": 10}, http.StatusTooManyRequests, "send_count_per_1d_limit", "daily message limit reached"}, + {"daily size", map[string]any{"limitSendSizePer1d": 30, "sendSizePer1d": 10}, http.StatusTooManyRequests, "send_size_per_1d_limit", "daily sending size limit reached"}, + {"storage", map[string]any{"limitSendSizeTotal": 1000, "sendSizeTotal": 990}, http.StatusUnprocessableEntity, "send_size_total_limit", "storage limit reached"}, + {"per message, attachments included", map[string]any{"limitSendSizePerMsg": 20}, http.StatusRequestEntityTooLarge, "send_size_per_msg_limit", "exceeds the sending limit"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + id.set(alice, tt.limits) + msg := a.sendDraft(t, []string{bob}, "hello", true) // 21 bytes + w := a.request(alice, "POST", "/fmsg/"+msg+"/send", nil) + v := jsonResponse(t, w, tt.status) + if v["code"] != tt.code || !strings.Contains(v["error"].(string), tt.error) { + t.Fatalf("refusal: %v", v) + } + draft := jsonResponse(t, a.request(alice, "GET", "/fmsg/"+msg, nil), 200) + if draft["time"] != nil { + t.Fatal("refused message was sent") + } + if rows := a.recipientRows(t, "msg_to", msg); rows[bob].delivered || rows[bob].code != nil { + t.Fatalf("refused message was delivered: %+v", rows[bob]) + } + }) + } + if send, recv := id.posted("send"), id.posted("recv"); len(send) != 0 || len(recv) != 0 { + t.Fatalf("refused sends recorded usage: send=%v recv=%v", send, recv) + } + + // Within the limits, the same message sends. + id.set(alice, map[string]any{"limitSendCountPer1d": 10, "sendCountPer1d": 9, "limitSendSizePerMsg": 21}) + msg := a.sendDraft(t, []string{bob}, "hello", true) + jsonResponse(t, a.request(alice, "POST", "/fmsg/"+msg+"/send", nil), 200) + + // A reaction over the reactor's daily count is refused and not sent. + id.set(bob, map[string]any{"limitSendCountPer1d": 0}) + if w := a.request(bob, "POST", "/fmsg/"+msg+"/react", map[string]any{"emoji": "👍"}); w.Code != http.StatusTooManyRequests { + t.Fatalf("reaction over quota: %d %s", w.Code, w.Body) + } + var reactions int + if err := a.pool.QueryRow(context.Background(), "SELECT count(*) FROM msg WHERE pid = $1", mustID(t, msg)).Scan(&reactions); err != nil || reactions != 0 { + t.Fatalf("refused reaction stored: %d %v", reactions, err) + } + + // fmsgid unavailable: refuse rather than send unchecked. + id.mu.Lock() + id.down = true + id.mu.Unlock() + msg = a.sendDraft(t, []string{bob}, "hello", false) + if w := a.request(alice, "POST", "/fmsg/"+msg+"/send", nil); w.Code != http.StatusServiceUnavailable { + t.Fatalf("send with fmsgid down: %d %s", w.Code, w.Body) + } +} diff --git a/internal/handlers/quota_test.go b/internal/handlers/quota_test.go new file mode 100644 index 0000000..e7ad1fd --- /dev/null +++ b/internal/handlers/quota_test.go @@ -0,0 +1,93 @@ +package handlers + +import ( + "net/http" + "testing" + + "github.com/markmnl/fmsg-webapi/internal/middleware" +) + +// unlimited is a detail with every limit unlimited and plenty of usage. +func unlimited() middleware.FmsgIDDetail { + return middleware.FmsgIDDetail{ + AcceptingNew: true, + LimitRecvSizeTotal: -1, LimitRecvSizePer1d: -1, LimitRecvCountPer1d: -1, + LimitSendSizeTotal: -1, LimitSendSizePerMsg: -1, LimitSendSizePer1d: -1, LimitSendCountPer1d: -1, + RecvSizeTotal: 1 << 40, RecvSizePer1d: 1 << 40, RecvCountPer1d: 1 << 40, + SendSizeTotal: 1 << 40, SendSizePer1d: 1 << 40, SendCountPer1d: 1 << 40, + } +} + +func TestLocalDeliveryCode(t *testing.T) { + with := func(f func(*middleware.FmsgIDDetail)) *middleware.FmsgIDDetail { + d := unlimited() + f(&d) + return &d + } + tests := []struct { + name string + status int + detail *middleware.FmsgIDDetail + size int64 + want int + ok bool + }{ + {"unknown", http.StatusNotFound, nil, 10, responseCodeUserUnknown, true}, + {"unexpected status", http.StatusInternalServerError, nil, 10, 0, false}, + {"not accepting", http.StatusOK, with(func(d *middleware.FmsgIDDetail) { d.AcceptingNew = false }), 10, responseCodeUserNotAccepting, true}, + {"unlimited", http.StatusOK, with(func(*middleware.FmsgIDDetail) {}), 1 << 30, responseCodeAccept, true}, + {"count fits", http.StatusOK, with(func(d *middleware.FmsgIDDetail) { d.LimitRecvCountPer1d, d.RecvCountPer1d = 10, 9 }), 10, responseCodeAccept, true}, + {"count full", http.StatusOK, with(func(d *middleware.FmsgIDDetail) { d.LimitRecvCountPer1d, d.RecvCountPer1d = 10, 10 }), 10, responseCodeUserFull, true}, + {"day size fits exactly", http.StatusOK, with(func(d *middleware.FmsgIDDetail) { d.LimitRecvSizePer1d, d.RecvSizePer1d = 100, 90 }), 10, responseCodeAccept, true}, + {"day size over", http.StatusOK, with(func(d *middleware.FmsgIDDetail) { d.LimitRecvSizePer1d, d.RecvSizePer1d = 100, 91 }), 10, responseCodeUserFull, true}, + {"storage over", http.StatusOK, with(func(d *middleware.FmsgIDDetail) { d.LimitRecvSizeTotal, d.RecvSizeTotal = 1000, 995 }), 10, responseCodeUserFull, true}, + {"zero limit", http.StatusOK, with(func(d *middleware.FmsgIDDetail) { d.LimitRecvCountPer1d, d.RecvCountPer1d = 0, 0 }), 0, responseCodeUserFull, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := localDeliveryCode(tt.status, tt.detail, tt.size) + if got != tt.want || ok != tt.ok { + t.Fatalf("localDeliveryCode = %d, %v; want %d, %v", got, ok, tt.want, tt.ok) + } + }) + } +} + +func TestCheckSendLimits(t *testing.T) { + tests := []struct { + name string + set func(*middleware.FmsgIDDetail) + size int64 + status int // 0 when the message may be sent + code string + }{ + {"unlimited", func(*middleware.FmsgIDDetail) {}, 1 << 30, 0, ""}, + {"per message fits", func(d *middleware.FmsgIDDetail) { d.LimitSendSizePerMsg = 100 }, 100, 0, ""}, + {"per message over", func(d *middleware.FmsgIDDetail) { d.LimitSendSizePerMsg = 100 }, 101, http.StatusRequestEntityTooLarge, "send_size_per_msg_limit"}, + {"count fits", func(d *middleware.FmsgIDDetail) { d.LimitSendCountPer1d, d.SendCountPer1d = 5, 4 }, 1, 0, ""}, + {"count reached", func(d *middleware.FmsgIDDetail) { d.LimitSendCountPer1d, d.SendCountPer1d = 5, 5 }, 1, http.StatusTooManyRequests, "send_count_per_1d_limit"}, + {"day size over", func(d *middleware.FmsgIDDetail) { d.LimitSendSizePer1d, d.SendSizePer1d = 1000, 950 }, 51, http.StatusTooManyRequests, "send_size_per_1d_limit"}, + {"storage over", func(d *middleware.FmsgIDDetail) { d.LimitSendSizeTotal, d.SendSizeTotal = 1000, 950 }, 51, http.StatusUnprocessableEntity, "send_size_total_limit"}, + {"storage fits exactly", func(d *middleware.FmsgIDDetail) { d.LimitSendSizeTotal, d.SendSizeTotal = 1000, 950 }, 50, 0, ""}, + {"per message checked first", func(d *middleware.FmsgIDDetail) { + d.LimitSendSizePerMsg = 10 + d.LimitSendCountPer1d, d.SendCountPer1d = 1, 1 + }, 11, http.StatusRequestEntityTooLarge, "send_size_per_msg_limit"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + d := unlimited() + tt.set(&d) + got := checkSendLimits(&d, tt.size) + if tt.status == 0 { + if got != nil { + t.Fatalf("refused: %+v", *got) + } + return + } + if got == nil || got.status != tt.status || got.code != tt.code || got.message == "" { + t.Fatalf("checkSendLimits = %+v, want %d %s", got, tt.status, tt.code) + } + }) + } +} diff --git a/internal/handlers/reactions.go b/internal/handlers/reactions.go index c7b2b73..ea1b2d0 100644 --- a/internal/handlers/reactions.go +++ b/internal/handlers/reactions.go @@ -398,6 +398,12 @@ func (h *MessageHandler) React(c *gin.Context) { c.JSON(500, gin.H{"error": "reaction parent has no finalized identity"}) return } + // A reaction is a message like any other: it counts against the + // sender's send limits, and its recipients' receive limits. + size := int64(len(want)) + if !h.checkSendQuota(c, identity, size) { + return + } now := float64(time.Now().UnixMicro()) / 1e6 tx, err := h.q().Begin(ctx) if err != nil { @@ -456,6 +462,7 @@ func (h *MessageHandler) React(c *gin.Context) { afterCommit(c, func() { plain := *h plain.query = nil + plain.recordSend(identity, now, size) plain.resolveLocalDelivery(ctx, "msg_to", reactionID, h.LocalDomain, recipients) }) diff --git a/internal/middleware/fmsgid_usage.go b/internal/middleware/fmsgid_usage.go new file mode 100644 index 0000000..3e89312 --- /dev/null +++ b/internal/middleware/fmsgid_usage.go @@ -0,0 +1,113 @@ +package middleware + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + "time" +) + +// FmsgIDDetail is the part of fmsgid's GET /fmsgid/:address answer used to +// check quotas: whether the address accepts new messages, its limits (-1 is +// unlimited) and its usage. For an address in a quota pool, fmsgid answers +// with the pool owner's limits and the usage of the whole pool. +type FmsgIDDetail struct { + AcceptingNew bool `json:"acceptingNew"` + LimitRecvSizeTotal int64 `json:"limitRecvSizeTotal"` + LimitRecvSizePer1d int64 `json:"limitRecvSizePer1d"` + LimitRecvCountPer1d int64 `json:"limitRecvCountPer1d"` + LimitSendSizeTotal int64 `json:"limitSendSizeTotal"` + LimitSendSizePerMsg int64 `json:"limitSendSizePerMsg"` + LimitSendSizePer1d int64 `json:"limitSendSizePer1d"` + LimitSendCountPer1d int64 `json:"limitSendCountPer1d"` + RecvSizeTotal int64 `json:"recvSizeTotal"` + RecvSizePer1d int64 `json:"recvSizePer1d"` + RecvCountPer1d int64 `json:"recvCountPer1d"` + SendSizeTotal int64 `json:"sendSizeTotal"` + SendSizePer1d int64 `json:"sendSizePer1d"` + SendCountPer1d int64 `json:"sendCountPer1d"` +} + +// FetchFmsgIDDetail fetches addr's detail from fmsgid, bypassing the lookup +// cache CheckFmsgID uses: usage changes with every message, so quota checks +// need it fresh. It returns fmsgid's HTTP status, and the detail only when +// that is 200. A limit fmsgid leaves out is unlimited. A 200 answer also +// refreshes the cached lookup CheckFmsgID uses. +func FetchFmsgIDDetail(idURL, addr string) (*FmsgIDDetail, int, error) { + lookupURL := fmsgIDAddressURL(idURL, addr) + req, err := http.NewRequest(http.MethodGet, lookupURL, nil) + if err != nil { + return nil, 0, err + } + resp, err := doFmsgID(req) + if err != nil { + return nil, 0, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, resp.StatusCode, nil + } + + var raw struct { + FmsgIDDetail + AcceptingNew *bool `json:"acceptingNew"` + } + raw.LimitRecvSizeTotal, raw.LimitRecvSizePer1d, raw.LimitRecvCountPer1d = -1, -1, -1 + raw.LimitSendSizeTotal, raw.LimitSendSizePerMsg, raw.LimitSendSizePer1d, raw.LimitSendCountPer1d = -1, -1, -1, -1 + if err := decodeJSON(resp.Body, &raw); err != nil { + return nil, 0, fmt.Errorf("decoding fmsgid address: %w", err) + } + if raw.AcceptingNew == nil { + return nil, 0, errors.New("fmsgid address response is missing acceptingNew") + } + detail := raw.FmsgIDDetail + detail.AcceptingNew = *raw.AcceptingNew + + fmsgIDCache.Store(lookupURL, fmsgIDEntry{ + expires: time.Now().Add(fmsgIDCacheTTL), + code: http.StatusOK, + acceptingNew: detail.AcceptingNew, + }) + return &detail, http.StatusOK, nil +} + +// RecordFmsgIDRecv records with fmsgid that addr received a message of size +// bytes at ts (seconds since the Unix epoch). +func RecordFmsgIDRecv(idURL, addr string, ts float64, size int64) error { + return postFmsgIDUsage(idURL, "recv", addr, ts, size) +} + +// RecordFmsgIDSend records with fmsgid that addr sent a message of size bytes +// at ts (seconds since the Unix epoch). +func RecordFmsgIDSend(idURL, addr string, ts float64, size int64) error { + return postFmsgIDUsage(idURL, "send", addr, ts, size) +} + +func postFmsgIDUsage(idURL, part, addr string, ts float64, size int64) error { + payload, err := json.Marshal(map[string]any{"address": addr, "ts": ts, "size": size}) + if err != nil { + return err + } + req, err := http.NewRequest(http.MethodPost, strings.TrimRight(idURL, "/")+"/fmsgid/"+part, bytes.NewReader(payload)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := doFmsgID(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("POST /fmsgid/%s returned %d", part, resp.StatusCode) + } + return nil +} + +// doFmsgID sends req to fmsgid with the bounded-timeout client. +func doFmsgID(req *http.Request) (*http.Response, error) { + return fmsgIDClient.Do(req) //nolint:gosec // URLs are built from trusted config +} diff --git a/internal/middleware/fmsgid_usage_test.go b/internal/middleware/fmsgid_usage_test.go new file mode 100644 index 0000000..2e4bbc6 --- /dev/null +++ b/internal/middleware/fmsgid_usage_test.go @@ -0,0 +1,111 @@ +package middleware + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" +) + +func TestFetchFmsgIDDetail(t *testing.T) { + const addr = "@detail@example.com" + var gets atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || r.URL.Path != "/fmsgid/"+addr { + w.WriteHeader(http.StatusNotFound) + return + } + gets.Add(1) + _, _ = w.Write([]byte(`{"address":"` + addr + `","acceptingNew":true,"limitRecvSizeTotal":1000,"limitRecvCountPer1d":10,` + + `"limitSendCountPer1d":5,"recvSizeTotal":100,"recvCountPer1d":3,"sendCountPer1d":4,"sendSizeTotal":77}`)) + })) + defer server.Close() + + for i := 1; i <= 2; i++ { + detail, status, err := FetchFmsgIDDetail(server.URL, addr) + if err != nil || status != http.StatusOK { + t.Fatalf("fetch: status=%d err=%v", status, err) + } + if int(gets.Load()) != i { + t.Fatalf("fetch %d made %d requests: the detail must not come from the cache", i, gets.Load()) + } + want := FmsgIDDetail{ + AcceptingNew: true, + LimitRecvSizeTotal: 1000, LimitRecvCountPer1d: 10, LimitSendCountPer1d: 5, + // limits fmsgid leaves out are unlimited + LimitRecvSizePer1d: -1, LimitSendSizeTotal: -1, LimitSendSizePerMsg: -1, LimitSendSizePer1d: -1, + RecvSizeTotal: 100, RecvCountPer1d: 3, SendCountPer1d: 4, SendSizeTotal: 77, + } + if *detail != want { + t.Fatalf("detail = %+v, want %+v", *detail, want) + } + } + + // A fresh detail also refreshes the lookup cache CheckFmsgID uses. + before := gets.Load() + if code, accepting, err := CheckFmsgID(server.URL, addr); err != nil || code != http.StatusOK || !accepting { + t.Fatalf("cached check: code=%d accepting=%v err=%v", code, accepting, err) + } + if gets.Load() != before { + t.Fatal("CheckFmsgID did not use the lookup cache refreshed by FetchFmsgIDDetail") + } + + if detail, status, err := FetchFmsgIDDetail(server.URL, "@missing@example.com"); err != nil || status != http.StatusNotFound || detail != nil { + t.Fatalf("missing address: detail=%v status=%d err=%v", detail, status, err) + } +} + +func TestFetchFmsgIDDetailRejectsInvalidResponse(t *testing.T) { + for _, body := range []string{"", "", `{}`, `{"acceptingNew":null}`, `{"acceptingNew":true,"recvSizeTotal":"x"}`} { + t.Run(body, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(body)) + })) + defer server.Close() + if detail, _, err := FetchFmsgIDDetail(server.URL, "@alice@example.com"); err == nil || detail != nil { + t.Fatalf("invalid response accepted: detail=%v err=%v", detail, err) + } + }) + } +} + +func TestRecordFmsgIDUsage(t *testing.T) { + type usage struct { + Address string `json:"address"` + Ts float64 `json:"ts"` + Size int64 `json:"size"` + } + got := map[string]usage{} + status := http.StatusOK + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var u usage + if r.Method != http.MethodPost || r.Header.Get("Content-Type") != "application/json" { + t.Errorf("unexpected %s %s %q", r.Method, r.URL.Path, r.Header.Get("Content-Type")) + } + if err := json.NewDecoder(r.Body).Decode(&u); err != nil { + t.Errorf("decode: %v", err) + } + got[r.URL.Path] = u + w.WriteHeader(status) + })) + defer server.Close() + + if err := RecordFmsgIDRecv(server.URL+"/", "@bob@example.com", 1700000000.5, 1234); err != nil { + t.Fatal(err) + } + if err := RecordFmsgIDSend(server.URL, "@alice@example.com", 1700000001, 99); err != nil { + t.Fatal(err) + } + if u := got["/fmsgid/recv"]; u != (usage{"@bob@example.com", 1700000000.5, 1234}) { + t.Fatalf("recv usage = %+v", u) + } + if u := got["/fmsgid/send"]; u != (usage{"@alice@example.com", 1700000001, 99}) { + t.Fatalf("send usage = %+v", u) + } + + status = http.StatusInternalServerError + if err := RecordFmsgIDSend(server.URL, "@alice@example.com", 1, 1); err == nil { + t.Fatal("a failed record was not reported") + } +}