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") + } +}