diff --git a/README.md b/README.md index 87e47a3..f3cbeff 100644 --- a/README.md +++ b/README.md @@ -190,6 +190,10 @@ Mail flow: 4. **Third-party clients**: Connect with SMTP 465/587, IMAP 993, or POP3 995; in production, configure certificates that match `LANQIN_PUBLIC_HOSTNAME`. 5. **External mailbox access**: Users can add external IMAP accounts in personal mailbox management. Local-storage mode syncs mail into the database; remote-direct mode reads from the remote server each time and does not write into local mail tables. +## Open API + +External integrations should use the versioned `/api/open/v1` endpoints with scoped API Tokens. See the [API guide](docs/API.md) and the machine-readable [OpenAPI 3.1 contract](docs/openapi.json). Sending supports idempotency keys; final delivery events can be ingested through a signed endpoint and all status changes can be pushed through the reliable signed webhook outbox. + ## Development and Verification ```bash diff --git a/README.zh-CN.md b/README.zh-CN.md index 4a92f96..d1edf85 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -190,6 +190,10 @@ docker compose -f docker-compose.yml -f docker-compose.build.yml up -d --build 4. **第三方客户端**：可通过 SMTP 465/587、IMAP 993、POP3 995 连接；生产环境请配置匹配 `LANQIN_PUBLIC_HOSTNAME` 的证书。 5. **外部邮箱接入**：个人邮箱管理可添加外部 IMAP 账号。本地存储模式会同步入库；远端直连模式每次读取远端，不写入本地邮件表。 +## 开放 API + +外部系统应使用版本化的 `/api/open/v1` 接口和带 scope 的 API Token。详细说明见 [API 文档](docs/API.md)，机器可读契约见 [OpenAPI 3.1](docs/openapi.json)。发信支持幂等键；最终投递事件可通过签名入口写入，全部状态变化也可通过可靠的签名 webhook outbox 主动推送。 + ## 开发与验证 ```bash diff --git a/apps/api/internal/app/admin_handlers.go b/apps/api/internal/app/admin_handlers.go index 1f09a31..2845a17 100644 --- a/apps/api/internal/app/admin_handlers.go +++ b/apps/api/internal/app/admin_handlers.go @@ -964,7 +964,7 @@ func (a *App) handleCreateAlias(w http.ResponseWriter, r *http.Request) { source = normalizeLocalPart(source) + "@" + domain.Name } destination := normalizeEmail(req.Destination) - if source == "" || destination == "" || !strings.Contains(destination, "@") { + if source == "" || !strings.HasSuffix(source, "@"+domain.Name) || destination == "" || !strings.Contains(destination, "@") { badRequest(w, errors.New("invalid alias")) return } @@ -1009,7 +1009,7 @@ func (a *App) handleUpdateAlias(w http.ResponseWriter, r *http.Request) { source = normalizeLocalPart(source) + "@" + domain.Name } destination := normalizeEmail(req.Destination) - if source == "" || destination == "" || !strings.Contains(destination, "@") { + if source == "" || !strings.HasSuffix(source, "@"+domain.Name) || destination == "" || !strings.Contains(destination, "@") { badRequest(w, errors.New("invalid alias")) return } diff --git a/apps/api/internal/app/api_token_handlers.go b/apps/api/internal/app/api_token_handlers.go index 265abc9..b15f28b 100644 --- a/apps/api/internal/app/api_token_handlers.go +++ b/apps/api/internal/app/api_token_handlers.go @@ -3,7 +3,9 @@ package app import ( "context" "database/sql" + "encoding/json" "errors" + "fmt" "net/http" "strings" "time" @@ -13,9 +15,24 @@ import ( const defaultAPITokenTTL = 90 * 24 * time.Hour +var validAPITokenScopes = map[string]bool{ + "*": true, + "domains:read": true, + "domains:write": true, + "mailboxes:read": true, + "mailboxes:write": true, + "messages:read": true, + "messages:send": true, + "messages:manage": true, + "aliases:read": true, + "aliases:write": true, + "dns:read": true, + "dns:check": true, +} + func (a *App) handleListAPITokens(w http.ResponseWriter, r *http.Request) { user := currentUser(r) - rows, err := a.db.QueryContext(r.Context(), `SELECT id,name,last_used_at,expires_at,disabled,created_at,updated_at + rows, err := a.db.QueryContext(r.Context(), `SELECT id,name,last_used_at,expires_at,disabled,scopes_json,created_at,updated_at FROM api_tokens WHERE user_id=? ORDER BY created_at DESC`, user.ID) if err != nil { respondError(w, http.StatusInternalServerError, "failed to list api tokens") @@ -41,8 +58,9 @@ func (a *App) handleListAPITokens(w http.ResponseWriter, r *http.Request) { func (a *App) handleCreateAPIToken(w http.ResponseWriter, r *http.Request) { user := currentUser(r) var req struct { - Name string `json:"name"` - ExpiresAt string `json:"expiresAt"` + Name string `json:"name"` + ExpiresAt string `json:"expiresAt"` + Scopes json.RawMessage `json:"scopes"` } if err := decodeJSON(r, &req); err != nil { badRequest(w, err) @@ -66,6 +84,20 @@ func (a *App) handleCreateAPIToken(w http.ResponseWriter, r *http.Request) { defaultExpiry := a.now().UTC().Add(defaultAPITokenTTL) expiresAt = &defaultExpiry } + var requestedScopes []string + if len(req.Scopes) > 0 { + if string(req.Scopes) == "null" || json.Unmarshal(req.Scopes, &requestedScopes) != nil { + badRequest(w, errors.New("scopes must be an array of strings")) + return + } + } else { + requestedScopes = nil + } + scopes, err := normalizeAPITokenScopes(requestedScopes) + if err != nil { + badRequest(w, err) + return + } id := newID("apt") token := "lq_" + randomToken() now := a.now().UTC().Format(time.RFC3339Nano) @@ -73,8 +105,8 @@ func (a *App) handleCreateAPIToken(w http.ResponseWriter, r *http.Request) { if expiresAt != nil { expiresValue = expiresAt.UTC().Format(time.RFC3339Nano) } - if _, err := a.db.ExecContext(r.Context(), `INSERT INTO api_tokens(id,user_id,name,token_hash,expires_at,disabled,created_at,updated_at) - VALUES(?,?,?,?,?,?,?,?)`, id, user.ID, name, hashToken(token), expiresValue, 0, now, now); err != nil { + if _, err := a.db.ExecContext(r.Context(), `INSERT INTO api_tokens(id,user_id,name,token_hash,expires_at,disabled,scopes_json,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,?,?)`, id, user.ID, name, hashToken(token), expiresValue, 0, jsonEncode(scopes), now, now); err != nil { respondError(w, http.StatusInternalServerError, "failed to create api token") return } @@ -94,9 +126,10 @@ func (a *App) handleUpdateAPIToken(w http.ResponseWriter, r *http.Request) { return } var req struct { - Name *string `json:"name"` - ExpiresAt *string `json:"expiresAt"` - Disabled *bool `json:"disabled"` + Name *string `json:"name"` + ExpiresAt *string `json:"expiresAt"` + Disabled *bool `json:"disabled"` + Scopes *[]string `json:"scopes"` } if err := decodeJSON(r, &req); err != nil { badRequest(w, err) @@ -142,8 +175,16 @@ func (a *App) handleUpdateAPIToken(w http.ResponseWriter, r *http.Request) { if req.Disabled != nil { disabled = *req.Disabled } - res, err := a.db.ExecContext(r.Context(), `UPDATE api_tokens SET name=?,expires_at=?,disabled=?,updated_at=? WHERE id=? AND user_id=?`, - name, expiresValue, boolInt(disabled), a.now().UTC().Format(time.RFC3339Nano), id, user.ID) + scopes := current.Scopes + if req.Scopes != nil { + scopes, err = normalizeAPITokenScopes(*req.Scopes) + if err != nil { + badRequest(w, err) + return + } + } + res, err := a.db.ExecContext(r.Context(), `UPDATE api_tokens SET name=?,expires_at=?,disabled=?,scopes_json=?,updated_at=? WHERE id=? AND user_id=?`, + name, expiresValue, boolInt(disabled), jsonEncode(scopes), a.now().UTC().Format(time.RFC3339Nano), id, user.ID) if err != nil { respondError(w, http.StatusInternalServerError, "failed to update api token") return @@ -175,7 +216,7 @@ func (a *App) handleDeleteAPIToken(w http.ResponseWriter, r *http.Request) { } func (a *App) apiTokenByID(ctx context.Context, userID, id string) (APIToken, error) { - row := a.db.QueryRowContext(ctx, `SELECT id,name,last_used_at,expires_at,disabled,created_at,updated_at + row := a.db.QueryRowContext(ctx, `SELECT id,name,last_used_at,expires_at,disabled,scopes_json,created_at,updated_at FROM api_tokens WHERE id=? AND user_id=?`, id, userID) return scanAPIToken(row) } @@ -186,18 +227,44 @@ func scanAPIToken(row apiTokenScanner) (APIToken, error) { var item APIToken var lastUsed, expires sql.NullString var disabled int - var created, updated string - if err := row.Scan(&item.ID, &item.Name, &lastUsed, &expires, &disabled, &created, &updated); err != nil { + var scopesJSON, created, updated string + if err := row.Scan(&item.ID, &item.Name, &lastUsed, &expires, &disabled, &scopesJSON, &created, &updated); err != nil { return item, err } item.LastUsedAt = nullableTime(lastUsed) item.ExpiresAt = nullableTime(expires) item.Disabled = intBool(disabled) + item.Scopes = jsonDecodeSlice(scopesJSON) item.CreatedAt = parseTime(created) item.UpdatedAt = parseTime(updated) return item, nil } +func normalizeAPITokenScopes(scopes []string) ([]string, error) { + if scopes == nil { + return []string{"*"}, nil + } + if len(scopes) == 0 { + return nil, errors.New("at least one api token scope is required") + } + seen := map[string]bool{} + out := make([]string, 0, len(scopes)) + for _, scope := range scopes { + scope = strings.ToLower(strings.TrimSpace(scope)) + if !validAPITokenScopes[scope] { + return nil, fmt.Errorf("invalid api token scope: %s", scope) + } + if !seen[scope] { + seen[scope] = true + out = append(out, scope) + } + } + if seen["*"] && len(out) != 1 { + return nil, errors.New("wildcard scope cannot be combined with other scopes") + } + return out, nil +} + func parseOptionalFutureTime(value string, now time.Time) (*time.Time, error) { value = strings.TrimSpace(value) if value == "" { diff --git a/apps/api/internal/app/app.go b/apps/api/internal/app/app.go index 3c82da0..260587e 100644 --- a/apps/api/internal/app/app.go +++ b/apps/api/internal/app/app.go @@ -82,6 +82,7 @@ func New(cfg Config, logger *slog.Logger) (*App, error) { a.startWorker(func() { a.sendQueueWorker(workerCtx) }) a.startWorker(func() { a.externalIMAPWorker(workerCtx) }) a.startWorker(func() { a.smtpEventsCleanupWorker(workerCtx) }) + a.startWorker(func() { a.statusWebhookWorker(workerCtx) }) return a, nil } @@ -171,9 +172,20 @@ func (a *App) migrate(ctx context.Context) error { last_used_at TEXT, expires_at TEXT NOT NULL, disabled INTEGER NOT NULL DEFAULT 0, + scopes_json TEXT NOT NULL DEFAULT '["*"]', created_at TEXT NOT NULL, updated_at TEXT NOT NULL )`, + `CREATE TABLE IF NOT EXISTS send_idempotency_keys ( + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + idempotency_key TEXT NOT NULL, + request_hash TEXT NOT NULL, + sent_message_id TEXT NOT NULL DEFAULT '', + queue_id TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL, + PRIMARY KEY(user_id, idempotency_key) + )`, + `CREATE INDEX IF NOT EXISTS idx_send_idempotency_created ON send_idempotency_keys(created_at)`, `CREATE INDEX IF NOT EXISTS idx_api_tokens_user ON api_tokens(user_id, created_at DESC)`, `CREATE INDEX IF NOT EXISTS idx_api_tokens_hash ON api_tokens(token_hash)`, `CREATE TABLE IF NOT EXISTS system_settings ( @@ -327,6 +339,45 @@ func (a *App) migrate(ctx context.Context) error { created_at TEXT NOT NULL )`, `CREATE INDEX IF NOT EXISTS idx_send_audit_events_created ON send_audit_events(created_at)`, + `CREATE TABLE IF NOT EXISTS delivery_events ( + id TEXT PRIMARY KEY, + external_id TEXT NOT NULL, + provider TEXT NOT NULL, + queue_id TEXT NOT NULL DEFAULT '', + sent_message_id TEXT NOT NULL DEFAULT '', + rfc_message_id TEXT NOT NULL DEFAULT '', + recipient TEXT NOT NULL, + status TEXT NOT NULL, + reason TEXT NOT NULL DEFAULT '', + occurred_at TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(provider, external_id) + )`, + `CREATE INDEX IF NOT EXISTS idx_delivery_events_message ON delivery_events(sent_message_id, occurred_at, id)`, + `CREATE INDEX IF NOT EXISTS idx_delivery_events_rfc_message ON delivery_events(rfc_message_id, occurred_at, id)`, + `CREATE TABLE IF NOT EXISTS status_webhook_outbox ( + id TEXT PRIMARY KEY, + event_key TEXT NOT NULL UNIQUE, + event_type TEXT NOT NULL, + mailbox_id TEXT NOT NULL DEFAULT '', + payload_json TEXT NOT NULL, + attempt_count INTEGER NOT NULL DEFAULT 0, + next_attempt_at TEXT NOT NULL, + last_error TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + delivered_at TEXT + )`, + `CREATE INDEX IF NOT EXISTS idx_status_webhook_outbox_due ON status_webhook_outbox(delivered_at,next_attempt_at,created_at)`, + `CREATE INDEX IF NOT EXISTS idx_status_webhook_outbox_mailbox ON status_webhook_outbox(mailbox_id,created_at)`, + `CREATE TRIGGER IF NOT EXISTS trg_mailbox_delete_status_webhook_outbox + AFTER DELETE ON mailboxes BEGIN + DELETE FROM status_webhook_outbox WHERE mailbox_id=OLD.id; + END`, + `CREATE TRIGGER IF NOT EXISTS trg_send_queue_delete_delivery_events + AFTER DELETE ON send_queue BEGIN + DELETE FROM delivery_events WHERE queue_id=OLD.id; + END`, `CREATE TABLE IF NOT EXISTS attachments ( id TEXT PRIMARY KEY, message_id TEXT NOT NULL REFERENCES messages(id) ON DELETE CASCADE, @@ -546,12 +597,45 @@ func (a *App) migrate(ctx context.Context) error { if err := a.migrateExternalIMAP(ctx); err != nil { return err } + if err := a.migrateAPITokenScopes(ctx); err != nil { + return err + } if err := a.ensureDefaultPermissionGroups(ctx); err != nil { return err } return nil } +func (a *App) migrateAPITokenScopes(ctx context.Context) error { + rows, err := a.db.QueryContext(ctx, `PRAGMA table_info(api_tokens)`) + if err != nil { + return err + } + hasScopes := false + for rows.Next() { + var cid int + var name, typ string + var notnull int + var dflt any + var pk int + if err := rows.Scan(&cid, &name, &typ, ¬null, &dflt, &pk); err != nil { + rows.Close() + return err + } + if name == "scopes_json" { + hasScopes = true + } + } + if err := rows.Close(); err != nil { + return err + } + if hasScopes { + return nil + } + _, err = a.db.ExecContext(ctx, `ALTER TABLE api_tokens ADD COLUMN scopes_json TEXT NOT NULL DEFAULT '["*"]'`) + return err +} + func (a *App) migrateMessageAuthentication(ctx context.Context) error { rows, err := a.db.QueryContext(ctx, `PRAGMA table_info(messages)`) if err != nil { @@ -1196,6 +1280,22 @@ func (a *App) createMailbox(ctx context.Context, userID, domainID, localPart, di } func (a *App) createMailboxWithPasswordHash(ctx context.Context, userID, domainID, localPart, displayName, passwordHash string, quotaMB int, status string) (string, error) { + tx, err := a.db.BeginTx(ctx, nil) + if err != nil { + return "", err + } + defer tx.Rollback() + id, err := a.createMailboxWithPasswordHashTx(ctx, tx, userID, domainID, localPart, displayName, passwordHash, quotaMB, status) + if err != nil { + return "", err + } + if err := tx.Commit(); err != nil { + return "", err + } + return id, nil +} + +func (a *App) createMailboxWithPasswordHashTx(ctx context.Context, tx *sql.Tx, userID, domainID, localPart, displayName, passwordHash string, quotaMB int, status string) (string, error) { localPart = normalizeLocalPart(localPart) if localPart == "" { return "", errors.New("invalid local part") @@ -1207,7 +1307,7 @@ func (a *App) createMailboxWithPasswordHash(ctx context.Context, userID, domainI status = "active" } var domain string - if err := a.db.QueryRowContext(ctx, `SELECT name FROM domains WHERE id=?`, domainID).Scan(&domain); err != nil { + if err := tx.QueryRowContext(ctx, `SELECT name FROM domains WHERE id=?`, domainID).Scan(&domain); err != nil { return "", err } address := localPart + "@" + domain @@ -1215,15 +1315,9 @@ func (a *App) createMailboxWithPasswordHash(ctx context.Context, userID, domainI displayName = address } - tx, err := a.db.BeginTx(ctx, nil) - if err != nil { - return "", err - } - defer tx.Rollback() - id := newID("mbx") now := a.now().UTC().Format(time.RFC3339Nano) - _, err = tx.ExecContext(ctx, `INSERT INTO mailboxes(id,user_id,domain_id,local_part,address,display_name,password_hash,quota_mb,status,created_at,updated_at) + _, err := tx.ExecContext(ctx, `INSERT INTO mailboxes(id,user_id,domain_id,local_part,address,display_name,password_hash,quota_mb,status,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)`, id, userID, domainID, localPart, address, displayName, passwordHash, quotaMB, status, now, now) if err != nil { return "", err @@ -1234,9 +1328,6 @@ func (a *App) createMailboxWithPasswordHash(ctx context.Context, userID, domainI return "", err } } - if err := tx.Commit(); err != nil { - return "", err - } return id, nil } diff --git a/apps/api/internal/app/app_test.go b/apps/api/internal/app/app_test.go index f4a9df1..98d4c07 100644 --- a/apps/api/internal/app/app_test.go +++ b/apps/api/internal/app/app_test.go @@ -4,12 +4,16 @@ import ( "bufio" "bytes" "context" + "crypto/hmac" "crypto/rand" "crypto/rsa" + "crypto/sha256" "crypto/tls" "crypto/x509" "crypto/x509/pkix" + "database/sql" "encoding/base64" + "encoding/hex" "encoding/json" "encoding/pem" "errors" @@ -24,6 +28,7 @@ import ( "net/url" "os" "path/filepath" + "strconv" "strings" "testing" "time" @@ -222,6 +227,10 @@ type testClient struct { } func (c *testClient) do(method, path string, body any, out any) int { + return c.doWithHeaders(method, path, body, nil, out) +} + +func (c *testClient) doWithHeaders(method, path string, body any, headers map[string]string, out any) int { c.t.Helper() var reader io.Reader if body != nil { @@ -241,6 +250,9 @@ func (c *testClient) do(method, path string, body any, out any) int { if c.bearer != "" { req.Header.Set("Authorization", "Bearer "+c.bearer) } + for key, value := range headers { + req.Header.Set(key, value) + } resp, err := http.DefaultClient.Do(req) if err != nil { c.t.Fatal(err) @@ -284,12 +296,20 @@ func createTestMailbox(t *testing.T, admin *testClient, domainID, localPart, dis } func createTestAPIToken(t *testing.T, client *testClient, name string) string { + return createTestAPITokenWithScopes(t, client, name, nil) +} + +func createTestAPITokenWithScopes(t *testing.T, client *testClient, name string, scopes []string) string { t.Helper() var resp struct { Token string `json:"token"` Item APIToken `json:"item"` } - if code := client.do("POST", "/api/me/api-tokens", map[string]string{"name": name}, &resp); code != http.StatusCreated { + payload := map[string]any{"name": name} + if scopes != nil { + payload["scopes"] = scopes + } + if code := client.do("POST", "/api/me/api-tokens", payload, &resp); code != http.StatusCreated { t.Fatalf("create api token code=%d resp=%+v", code, resp) } if resp.Token == "" || resp.Item.ID == "" || resp.Item.Name != name { @@ -1684,6 +1704,21 @@ func TestAPITokenManagementStoresHashAndRevokes(t *testing.T) { if code := admin.do("POST", "/api/me/api-tokens", map[string]string{"name": "integration-test"}, &created); code != http.StatusCreated { t.Fatalf("create api token code=%d resp=%+v", code, created) } + nullScopes := bytes.NewBufferString(`{"name":"null-scopes","scopes":null}`) + req, err := http.NewRequest(http.MethodPost, ts.URL+"/api/me/api-tokens", nullScopes) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + req.AddCookie(admin.cookie) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("null scopes create code=%d", resp.StatusCode) + } if !strings.HasPrefix(created.Token, "lq_") || created.Item.ID == "" || created.Item.Name != "integration-test" || created.Item.ExpiresAt == nil { t.Fatalf("created token response=%+v", created) } @@ -1908,7 +1943,7 @@ func TestOpenAPISendStatusAndMailboxMessages(t *testing.T) { if code := senderOpen.do("GET", "/api/open/send/"+sent.ID, nil, &status); code != http.StatusOK { t.Fatalf("open api send status code=%d status=%+v", code, status) } - if status.ID != sent.QueueID || status.MessageID != sent.MessageID || status.Status != sendQueueStatusQueued { + if status.ID != sent.MessageID || status.QueueID != sent.QueueID || status.MessageID != sent.MessageID || status.Status != sendQueueStatusQueued { t.Fatalf("status=%+v sent=%+v", status, sent) } @@ -1936,6 +1971,327 @@ func TestOpenAPISendStatusAndMailboxMessages(t *testing.T) { } } +func TestOpenAPIV1ScopesIdempotencyAndDeliveryEvents(t *testing.T) { + a := newTestApp(t) + stopTestWorkers(a) + a.cfg.SMTPHost = "127.0.0.1" + a.cfg.SMTPPort = "25" + a.cfg.DeliveryWebhookSecret = "delivery-test-secret" + ts := httptest.NewServer(a.Router()) + defer ts.Close() + + admin := &testClient{t: t, server: ts} + var login map[string]any + if code := admin.do("POST", "/api/auth/login", map[string]string{"email": "admin@lanqin.local", "password": "ChangeMe123!"}, &login); code != http.StatusOK { + t.Fatalf("admin login code=%d", code) + } + domainID := mustDefaultDomainID(t, a) + sender := createTestMailbox(t, admin, domainID, "v1-sender", "V1 Sender", "Password123!", nil) + recipient := createTestMailbox(t, admin, domainID, "v1-recipient", "V1 Recipient", "Password123!", nil) + + adminReadToken := createTestAPITokenWithScopes(t, admin, "domain-reader", []string{"domains:read"}) + adminRead := &testClient{t: t, server: ts, bearer: adminReadToken} + if code := adminRead.do("GET", "/api/open/v1/domains", nil, &map[string]any{}); code != http.StatusOK { + t.Fatalf("v1 scoped domain list code=%d", code) + } + if code := adminRead.do("POST", "/api/open/v1/domains", map[string]string{"name": "scope-denied.example"}, &map[string]any{}); code != http.StatusForbidden { + t.Fatalf("read-only token domain create code=%d", code) + } + + senderClient := &testClient{t: t, server: ts} + if code := senderClient.do("POST", "/api/auth/login", map[string]string{"email": sender.Address, "password": "Password123!"}, &login); code != http.StatusOK { + t.Fatalf("sender login code=%d", code) + } + sendToken := createTestAPITokenWithScopes(t, senderClient, "send-only", []string{"messages:send"}) + sendClient := &testClient{t: t, server: ts, bearer: sendToken} + payload := map[string]any{"mailboxId": sender.ID, "to": []string{recipient.Address}, "subject": "idempotent send", "text": "one delivery"} + headers := map[string]string{"Idempotency-Key": "invoice-42"} + var first openAPISendStatus + if code := sendClient.doWithHeaders("POST", "/api/open/v1/send", payload, headers, &first); code != http.StatusCreated { + t.Fatalf("first idempotent send code=%d body=%+v", code, first) + } + if first.ID == "" || first.ID != first.MessageID || first.QueueID == "" || first.ID == first.QueueID { + t.Fatalf("stable send identifiers=%+v", first) + } + var replay openAPISendStatus + if code := sendClient.doWithHeaders("POST", "/api/open/v1/send", payload, headers, &replay); code != http.StatusOK { + t.Fatalf("idempotent replay code=%d body=%+v", code, replay) + } + if replay.ID != first.ID || replay.QueueID != first.QueueID { + t.Fatalf("replay=%+v first=%+v", replay, first) + } + var queueCount int + if err := a.db.QueryRow(`SELECT COUNT(*) FROM send_queue WHERE source=? AND sent_message_id=?`, sendSourceOpenAPI, first.MessageID).Scan(&queueCount); err != nil || queueCount != 1 { + t.Fatalf("idempotent queue count=%d err=%v", queueCount, err) + } + changed := map[string]any{"mailboxId": sender.ID, "to": []string{recipient.Address}, "subject": "changed", "text": "different"} + if code := sendClient.doWithHeaders("POST", "/api/open/v1/send", changed, headers, &map[string]any{}); code != http.StatusConflict { + t.Fatalf("changed idempotency payload code=%d", code) + } + if code := sendClient.do("GET", "/api/open/v1/send/"+first.ID, nil, &map[string]any{}); code != http.StatusForbidden { + t.Fatalf("send-only token read code=%d", code) + } + + readToken := createTestAPITokenWithScopes(t, senderClient, "read-only", []string{"messages:read"}) + readClient := &testClient{t: t, server: ts, bearer: readToken} + var queued openAPISendStatus + if code := readClient.do("GET", "/api/open/v1/send/"+first.ID, nil, &queued); code != http.StatusOK || queued.Status != sendQueueStatusQueued { + t.Fatalf("read status code=%d body=%+v", code, queued) + } + if _, err := a.db.Exec(`UPDATE send_queue SET status=?,updated_at=? WHERE id=?`, sendQueueStatusSending, a.now().UTC().Format(time.RFC3339Nano), first.QueueID); err != nil { + t.Fatal(err) + } + var sending openAPISendStatus + if code := readClient.do("GET", "/api/open/v1/send/"+first.ID, nil, &sending); code != http.StatusOK || sending.Status != sendQueueStatusSending { + t.Fatalf("sending status code=%d body=%+v", code, sending) + } + if _, err := a.db.Exec(`UPDATE send_queue SET status=?,delivered_at=?,updated_at=? WHERE id=?`, sendQueueStatusDelivered, a.now().UTC().Format(time.RFC3339Nano), a.now().UTC().Format(time.RFC3339Nano), first.QueueID); err != nil { + t.Fatal(err) + } + var relayed openAPISendStatus + if code := readClient.do("GET", "/api/open/v1/send/"+first.ID, nil, &relayed); code != http.StatusOK || relayed.Status != "relayed" || relayed.QueueStatus != sendQueueStatusDelivered { + t.Fatalf("relayed status code=%d body=%+v", code, relayed) + } + + eventPayload := struct { + Events []deliveryWebhookEvent `json:"events"` + }{Events: []deliveryWebhookEvent{{ID: "provider-event-1", Provider: "test-provider", MessageID: first.MessageID, Recipient: recipient.Address, Status: "bounced", Reason: "550 mailbox unavailable", OccurredAt: a.now().UTC().Format(time.RFC3339Nano)}}} + body, _ := json.Marshal(eventPayload) + timestamp := strconv.FormatInt(a.now().UTC().Unix(), 10) + mac := hmac.New(sha256.New, []byte(a.cfg.DeliveryWebhookSecret)) + _, _ = mac.Write([]byte(timestamp + ".")) + _, _ = mac.Write(body) + webhookHeaders := map[string]string{"X-LanQin-Timestamp": timestamp, "X-LanQin-Signature": "sha256=" + hex.EncodeToString(mac.Sum(nil))} + badSignatureHeaders := map[string]string{"X-LanQin-Timestamp": timestamp, "X-LanQin-Signature": "sha256=" + strings.Repeat("0", 64)} + if code := admin.doWithHeaders("POST", "/api/open/v1/delivery-events", eventPayload, badSignatureHeaders, &map[string]any{}); code != http.StatusUnauthorized { + t.Fatalf("invalid delivery webhook signature code=%d", code) + } + oldTimestamp := strconv.FormatInt(a.now().UTC().Add(-10*time.Minute).Unix(), 10) + oldMAC := hmac.New(sha256.New, []byte(a.cfg.DeliveryWebhookSecret)) + _, _ = oldMAC.Write([]byte(oldTimestamp + ".")) + _, _ = oldMAC.Write(body) + oldHeaders := map[string]string{"X-LanQin-Timestamp": oldTimestamp, "X-LanQin-Signature": "sha256=" + hex.EncodeToString(oldMAC.Sum(nil))} + if code := admin.doWithHeaders("POST", "/api/open/v1/delivery-events", eventPayload, oldHeaders, &map[string]any{}); code != http.StatusUnauthorized { + t.Fatalf("expired delivery webhook signature code=%d", code) + } + var webhookResult struct { + Accepted int `json:"accepted"` + Duplicates int `json:"duplicates"` + } + if code := admin.doWithHeaders("POST", "/api/open/v1/delivery-events", eventPayload, webhookHeaders, &webhookResult); code != http.StatusOK || webhookResult.Accepted != 1 { + t.Fatalf("delivery webhook code=%d body=%+v", code, webhookResult) + } + if code := admin.doWithHeaders("POST", "/api/open/v1/delivery-events", eventPayload, webhookHeaders, &webhookResult); code != http.StatusOK || webhookResult.Duplicates != 1 { + t.Fatalf("delivery webhook duplicate code=%d body=%+v", code, webhookResult) + } + var bounced openAPISendStatus + if code := readClient.do("GET", "/api/open/v1/send/"+first.ID, nil, &bounced); code != http.StatusOK || bounced.Status != "bounced" || len(bounced.RecipientStatuses) != 1 { + t.Fatalf("bounced status code=%d body=%+v", code, bounced) + } + var events struct { + DeliveryEvents []DeliveryEvent `json:"deliveryEvents"` + } + if code := readClient.do("GET", "/api/open/v1/send/"+first.ID+"/events", nil, &events); code != http.StatusOK || len(events.DeliveryEvents) != 1 { + t.Fatalf("delivery events code=%d body=%+v", code, events) + } + + manageToken := createTestAPITokenWithScopes(t, senderClient, "send-manager", []string{"messages:read", "messages:manage"}) + manageClient := &testClient{t: t, server: ts, bearer: manageToken} + if _, err := a.db.Exec(`UPDATE send_queue SET status=?,attempt_count=max_attempts,last_error='test failure',updated_at=? WHERE id=?`, sendQueueStatusFailed, a.now().UTC().Format(time.RFC3339Nano), first.QueueID); err != nil { + t.Fatal(err) + } + var failed openAPISendStatus + if code := manageClient.do("GET", "/api/open/v1/send/"+first.ID, nil, &failed); code != http.StatusOK || failed.QueueStatus != sendQueueStatusFailed { + t.Fatalf("failed status code=%d body=%+v", code, failed) + } + var retried openAPISendStatus + if code := manageClient.do("POST", "/api/open/v1/send/"+first.ID+"/retry", nil, &retried); code != http.StatusOK || retried.QueueStatus != sendQueueStatusQueued { + t.Fatalf("retry code=%d body=%+v", code, retried) + } + var canceled openAPISendStatus + if code := manageClient.do("POST", "/api/open/v1/send/"+first.ID+"/cancel", nil, &canceled); code != http.StatusOK || canceled.QueueStatus != sendQueueStatusCanceled { + t.Fatalf("cancel code=%d body=%+v", code, canceled) + } +} + +func TestOpenAPIPaginationAndMailboxCreateRollback(t *testing.T) { + a := newTestApp(t) + ts := httptest.NewServer(a.Router()) + defer ts.Close() + admin := &testClient{t: t, server: ts} + var login map[string]any + if code := admin.do("POST", "/api/auth/login", map[string]string{"email": "admin@lanqin.local", "password": "ChangeMe123!"}, &login); code != http.StatusOK { + t.Fatalf("login code=%d", code) + } + token := createTestAPITokenWithScopes(t, admin, "admin-v1", []string{"domains:read", "mailboxes:write"}) + openAdmin := &testClient{t: t, server: ts, bearer: token} + createTestDomain(t, admin, "pagination-one.test") + createTestDomain(t, admin, "pagination-two.test") + var firstPage struct { + Items []Domain `json:"items"` + NextCursor string `json:"nextCursor"` + } + if code := openAdmin.do("GET", "/api/open/v1/domains?limit=1", nil, &firstPage); code != http.StatusOK || len(firstPage.Items) != 1 || firstPage.NextCursor == "" { + t.Fatalf("first domain page code=%d body=%+v", code, firstPage) + } + var secondPage struct { + Items []Domain `json:"items"` + } + if code := openAdmin.do("GET", "/api/open/v1/domains?limit=1&cursor="+url.QueryEscape(firstPage.NextCursor), nil, &secondPage); code != http.StatusOK || len(secondPage.Items) != 1 || secondPage.Items[0].ID == firstPage.Items[0].ID { + t.Fatalf("second domain page code=%d body=%+v", code, secondPage) + } + + domainID := mustDefaultDomainID(t, a) + createTestMailbox(t, admin, domainID, "rollback-address", "Existing", "Password123!", nil) + payload := map[string]any{"domainId": domainID, "localPart": "rollback-address", "displayName": "Should Rollback", "password": "Password123!", "ownerEmail": "orphan-owner@example.test"} + if code := openAdmin.do("POST", "/api/open/v1/mailboxes", payload, &map[string]any{}); code != http.StatusBadRequest { + t.Fatalf("duplicate mailbox create code=%d", code) + } + var orphanCount int + if err := a.db.QueryRow(`SELECT COUNT(*) FROM users WHERE email=?`, "orphan-owner@example.test").Scan(&orphanCount); err != nil || orphanCount != 0 { + t.Fatalf("orphan user count=%d err=%v", orphanCount, err) + } +} + +func TestOpenAPIContractCoversV1Routes(t *testing.T) { + path := filepath.Join("..", "..", "..", "..", "docs", "openapi.json") + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var document struct { + OpenAPI string `json:"openapi"` + Paths map[string]map[string]json.RawMessage `json:"paths"` + Components struct { + SecuritySchemes map[string]json.RawMessage `json:"securitySchemes"` + } `json:"components"` + } + if err := json.Unmarshal(data, &document); err != nil { + t.Fatalf("parse openapi contract: %v", err) + } + if document.OpenAPI != "3.1.0" || document.Components.SecuritySchemes["bearerAuth"] == nil { + t.Fatalf("invalid openapi metadata: version=%q security=%v", document.OpenAPI, document.Components.SecuritySchemes) + } + routes := map[string][]string{ + "/domains": {"get", "post"}, "/domains/{id}": {"get", "post", "delete"}, + "/domains/{id}/dns-records": {"get"}, "/domains/{id}/dns-check": {"post"}, + "/mailboxes": {"get", "post"}, "/mailboxes/{id}": {"get", "post", "delete"}, + "/mailboxes/{id}/password": {"post"}, "/mailboxes/{id}/messages": {"get"}, + "/messages/{id}": {"get"}, "/attachments/{id}": {"get"}, + "/send": {"get", "post"}, "/send/{id}": {"get"}, "/send/{id}/events": {"get"}, + "/send/{id}/retry": {"post"}, "/send/{id}/cancel": {"post"}, + "/aliases": {"get", "post"}, "/aliases/{id}": {"get", "post", "delete"}, + "/delivery-events": {"post"}, + } + for route, methods := range routes { + pathItem := document.Paths[route] + if pathItem == nil { + t.Errorf("openapi missing path %s", route) + continue + } + for _, method := range methods { + if pathItem[method] == nil { + t.Errorf("openapi missing operation %s %s", strings.ToUpper(method), route) + } + } + if strings.Contains(route, "{id}") { + var raw map[string]any + if err := json.Unmarshal(data, &raw); err != nil { + t.Fatal(err) + } + paths, _ := raw["paths"].(map[string]any) + item, _ := paths[route].(map[string]any) + parameters, _ := item["parameters"].([]any) + foundID := false + for _, value := range parameters { + parameter, _ := value.(map[string]any) + if parameter["$ref"] == "#/components/parameters/ResourceId" || parameter["name"] == "id" { + foundID = true + } + } + if !foundID { + t.Errorf("openapi path %s does not declare id parameter", route) + } + } + } +} + +func TestStatusWebhookOutboxDeliveryRetryAndSSRFProtection(t *testing.T) { + a := newTestApp(t) + stopTestWorkers(a) + accept := false + requests := 0 + receiver := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests++ + body, err := io.ReadAll(r.Body) + if err != nil { + t.Error(err) + w.WriteHeader(http.StatusBadRequest) + return + } + timestamp := r.Header.Get("X-LanQin-Timestamp") + mac := hmac.New(sha256.New, []byte("outbound-test-secret")) + _, _ = mac.Write([]byte(timestamp + ".")) + _, _ = mac.Write(body) + if r.Header.Get("X-LanQin-Webhook-Id") == "" || r.Header.Get("X-LanQin-Signature") != "sha256="+hex.EncodeToString(mac.Sum(nil)) { + t.Error("invalid outbound webhook signature headers") + } + var envelope statusWebhookEnvelope + if err := json.Unmarshal(body, &envelope); err != nil || envelope.Type != "send.failed" { + t.Errorf("invalid outbound webhook payload: err=%v payload=%s", err, body) + } + if !accept { + w.WriteHeader(http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusNoContent) + })) + defer receiver.Close() + a.cfg.StatusWebhookURL = receiver.URL + a.cfg.StatusWebhookSecret = "outbound-test-secret" + a.cfg.StatusWebhookAllowPrivateHosts = true + + user, mb := defaultAdminUserAndMailbox(t, a) + a.recordSendAudit(context.Background(), sendAuditFailed, sendQueueStatusFailed, sendAuditInput{QueueID: "snd_test", UserID: user.ID, MailboxID: mb.ID, SentMessageID: "mail_test", Source: sendSourceOpenAPI, MailFrom: mb.Address, Recipients: []string{"recipient@example.test"}, Error: "test failure"}) + var outboxID string + if err := a.db.QueryRow(`SELECT id FROM status_webhook_outbox WHERE event_type='send.failed'`).Scan(&outboxID); err != nil { + t.Fatal(err) + } + if err := a.processDueStatusWebhooks(context.Background()); err != nil { + t.Fatal(err) + } + var attempts int + var deliveredAt sql.NullString + if err := a.db.QueryRow(`SELECT attempt_count,delivered_at FROM status_webhook_outbox WHERE id=?`, outboxID).Scan(&attempts, &deliveredAt); err != nil || attempts != 1 || deliveredAt.Valid { + t.Fatalf("failed delivery outbox attempts=%d delivered=%v err=%v", attempts, deliveredAt, err) + } + accept = true + if _, err := a.db.Exec(`UPDATE status_webhook_outbox SET next_attempt_at=? WHERE id=?`, a.now().UTC().Format(time.RFC3339Nano), outboxID); err != nil { + t.Fatal(err) + } + if err := a.processDueStatusWebhooks(context.Background()); err != nil { + t.Fatal(err) + } + if err := a.db.QueryRow(`SELECT attempt_count,delivered_at FROM status_webhook_outbox WHERE id=?`, outboxID).Scan(&attempts, &deliveredAt); err != nil || attempts != 2 || !deliveredAt.Valid || requests != 2 { + t.Fatalf("successful retry attempts=%d delivered=%v requests=%d err=%v", attempts, deliveredAt, requests, err) + } + if _, err := a.db.Exec(`DELETE FROM mailboxes WHERE id=?`, mb.ID); err != nil { + t.Fatal(err) + } + var remaining int + if err := a.db.QueryRow(`SELECT COUNT(*) FROM status_webhook_outbox WHERE id=?`, outboxID).Scan(&remaining); err != nil || remaining != 0 { + t.Fatalf("mailbox deletion should remove webhook outbox, remaining=%d err=%v", remaining, err) + } + + privateTLS := httptest.NewTLSServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + defer privateTLS.Close() + a.cfg.StatusWebhookURL = privateTLS.URL + a.cfg.StatusWebhookAllowPrivateHosts = false + if _, err := a.validatedStatusWebhookURL(context.Background()); err == nil || !strings.Contains(err.Error(), "private or local") { + t.Fatalf("private webhook target should be rejected, err=%v", err) + } +} + func TestSendQueueRecoversStaleSendingItems(t *testing.T) { a := newTestApp(t) stopTestWorkers(a) diff --git a/apps/api/internal/app/config.go b/apps/api/internal/app/config.go index a96cf9b..b5af951 100644 --- a/apps/api/internal/app/config.go +++ b/apps/api/internal/app/config.go @@ -51,6 +51,10 @@ type Config struct { ExternalIMAPOutlookClientSecret string MailTranslateEnabled bool MailTranslateMaxChars int + DeliveryWebhookSecret string + StatusWebhookURL string + StatusWebhookSecret string + StatusWebhookAllowPrivateHosts bool } func LoadConfig() Config { @@ -99,6 +103,10 @@ func LoadConfig() Config { ExternalIMAPOutlookClientSecret: getenv("LANQIN_EXTERNAL_IMAP_OUTLOOK_CLIENT_SECRET", ""), MailTranslateEnabled: getenvBool("LANQIN_MAIL_TRANSLATE_ENABLED", true), MailTranslateMaxChars: getenvInt("LANQIN_MAIL_TRANSLATE_MAX_CHARS", 8000), + DeliveryWebhookSecret: getenv("LANQIN_DELIVERY_WEBHOOK_SECRET", ""), + StatusWebhookURL: getenv("LANQIN_STATUS_WEBHOOK_URL", ""), + StatusWebhookSecret: getenv("LANQIN_STATUS_WEBHOOK_SECRET", ""), + StatusWebhookAllowPrivateHosts: getenvBool("LANQIN_STATUS_WEBHOOK_ALLOW_PRIVATE_HOSTS", false), } } diff --git a/apps/api/internal/app/open_api_extended.go b/apps/api/internal/app/open_api_extended.go new file mode 100644 index 0000000..ca835fd --- /dev/null +++ b/apps/api/internal/app/open_api_extended.go @@ -0,0 +1,414 @@ +package app + +import ( + "crypto/hmac" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "io" + "net/http" + "strconv" + "strings" + "time" + + "github.com/go-chi/chi/v5" +) + +const deliveryWebhookMaxAge = 5 * time.Minute + +type deliveryWebhookEvent struct { + ID string `json:"id"` + Provider string `json:"provider"` + QueueID string `json:"queueId"` + MessageID string `json:"messageId"` + RFCMessageID string `json:"rfcMessageId"` + Recipient string `json:"recipient"` + Status string `json:"status"` + Reason string `json:"reason"` + OccurredAt string `json:"occurredAt"` +} + +func (a *App) handleOpenAPIDeliveryWebhook(w http.ResponseWriter, r *http.Request) { + secret := strings.TrimSpace(a.cfg.DeliveryWebhookSecret) + if secret == "" { + respondError(w, http.StatusServiceUnavailable, "delivery webhook is not configured") + return + } + timestamp := strings.TrimSpace(r.Header.Get("X-LanQin-Timestamp")) + signature := strings.TrimPrefix(strings.TrimSpace(r.Header.Get("X-LanQin-Signature")), "sha256=") + unix, err := strconv.ParseInt(timestamp, 10, 64) + if err != nil || signature == "" { + respondError(w, http.StatusUnauthorized, "invalid webhook signature") + return + } + signedAt := time.Unix(unix, 0) + if delta := a.now().UTC().Sub(signedAt); delta < -deliveryWebhookMaxAge || delta > deliveryWebhookMaxAge { + respondError(w, http.StatusUnauthorized, "webhook timestamp is outside the allowed window") + return + } + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 1<<20)) + if err != nil { + badRequest(w, errors.New("invalid webhook body")) + return + } + mac := hmac.New(sha256.New, []byte(secret)) + _, _ = mac.Write([]byte(timestamp + ".")) + _, _ = mac.Write(body) + expected, err := hex.DecodeString(signature) + if err != nil || !hmac.Equal(mac.Sum(nil), expected) { + respondError(w, http.StatusUnauthorized, "invalid webhook signature") + return + } + var payload struct { + Events []deliveryWebhookEvent `json:"events"` + } + dec := json.NewDecoder(strings.NewReader(string(body))) + dec.DisallowUnknownFields() + if err := dec.Decode(&payload); err != nil || len(payload.Events) == 0 || len(payload.Events) > 100 { + badRequest(w, errors.New("events must contain between 1 and 100 items")) + return + } + accepted := 0 + tx, err := a.db.BeginTx(r.Context(), nil) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to start delivery event transaction") + return + } + defer tx.Rollback() + for _, event := range payload.Events { + inserted, err := a.storeDeliveryEvent(r, tx, event) + if err != nil { + badRequest(w, err) + return + } + if inserted { + accepted++ + } + } + if err := tx.Commit(); err != nil { + respondError(w, http.StatusInternalServerError, "failed to store delivery events") + return + } + respondJSON(w, http.StatusOK, map[string]any{"ok": true, "accepted": accepted, "duplicates": len(payload.Events) - accepted}) +} + +func (a *App) storeDeliveryEvent(r *http.Request, tx *sql.Tx, event deliveryWebhookEvent) (bool, error) { + event.ID = strings.TrimSpace(event.ID) + event.Provider = strings.ToLower(strings.TrimSpace(event.Provider)) + event.QueueID = strings.TrimSpace(event.QueueID) + event.MessageID = strings.TrimSpace(event.MessageID) + event.RFCMessageID = strings.TrimSpace(event.RFCMessageID) + event.Recipient = normalizeEmail(event.Recipient) + event.Status = strings.ToLower(strings.TrimSpace(event.Status)) + if event.ID == "" || len(event.ID) > 200 || event.Provider == "" || len(event.Provider) > 80 || event.Recipient == "" || len(event.Recipient) > 320 || len(event.Reason) > 2000 || !validDeliveryEventStatus(event.Status) { + return false, errors.New("invalid delivery event") + } + if event.QueueID == "" && event.MessageID == "" && event.RFCMessageID == "" { + return false, errors.New("queueId, messageId, or rfcMessageId is required") + } + occurredAt, err := time.Parse(time.RFC3339Nano, event.OccurredAt) + if err != nil { + return false, errors.New("occurredAt must be an RFC3339 timestamp") + } + var queueID, sentMessageID, rfcMessageID string + err = tx.QueryRowContext(r.Context(), `SELECT id,sent_message_id,message_id FROM send_queue + WHERE (?<>'' AND id=?) OR (?<>'' AND sent_message_id=?) OR (?<>'' AND message_id=?) + ORDER BY created_at DESC LIMIT 1`, event.QueueID, event.QueueID, event.MessageID, event.MessageID, event.RFCMessageID, event.RFCMessageID).Scan(&queueID, &sentMessageID, &rfcMessageID) + if err != nil { + return false, errors.New("send item not found") + } + if (event.QueueID != "" && event.QueueID != queueID) || (event.MessageID != "" && event.MessageID != sentMessageID) || (event.RFCMessageID != "" && event.RFCMessageID != rfcMessageID) { + return false, errors.New("delivery event identifiers do not refer to the same send item") + } + var recipientsJSON string + if err := tx.QueryRowContext(r.Context(), `SELECT recipients_json FROM send_queue WHERE id=?`, queueID).Scan(&recipientsJSON); err != nil { + return false, errors.New("send item not found") + } + foundRecipient := false + for _, recipient := range jsonDecodeSlice(recipientsJSON) { + if normalizeEmail(recipient) == event.Recipient { + foundRecipient = true + break + } + } + if !foundRecipient { + return false, errors.New("delivery event recipient does not belong to the send item") + } + id := newID("dev") + createdAt := a.now().UTC() + reason := strings.TrimSpace(event.Reason) + res, err := tx.ExecContext(r.Context(), `INSERT OR IGNORE INTO delivery_events(id,external_id,provider,queue_id,sent_message_id,rfc_message_id,recipient,status,reason,occurred_at,created_at) + VALUES(?,?,?,?,?,?,?,?,?,?,?)`, id, event.ID, event.Provider, queueID, sentMessageID, rfcMessageID, event.Recipient, event.Status, reason, occurredAt.UTC().Format(time.RFC3339Nano), createdAt.Format(time.RFC3339Nano)) + if err != nil { + return false, err + } + n, _ := res.RowsAffected() + if n > 0 { + item := DeliveryEvent{ID: id, ExternalID: event.ID, Provider: event.Provider, QueueID: queueID, MessageID: sentMessageID, RFCMessageID: rfcMessageID, Recipient: event.Recipient, Status: event.Status, Reason: reason, OccurredAt: occurredAt.UTC(), CreatedAt: createdAt} + var mailboxID string + if err := tx.QueryRowContext(r.Context(), `SELECT mailbox_id FROM send_queue WHERE id=?`, queueID).Scan(&mailboxID); err != nil { + return false, err + } + if err := a.enqueueStatusWebhook(r.Context(), tx, "delivery:"+event.Provider+":"+event.ID, "delivery."+event.Status, mailboxID, item); err != nil { + return false, err + } + } + return n > 0, nil +} + +func validDeliveryEventStatus(status string) bool { + switch status { + case "delivered", "bounced", "complained", "rejected", "deferred": + return true + default: + return false + } +} + +func (a *App) handleOpenAPIListSends(w http.ResponseWriter, r *http.Request) { + user := currentUser(r) + limit := parseOpenAPILimit(r, 30, 100) + where := "mb.user_id=?" + args := []any{user.ID} + if mailboxID := strings.TrimSpace(r.URL.Query().Get("mailboxId")); mailboxID != "" { + where += " AND sq.mailbox_id=?" + args = append(args, mailboxID) + } + if status := strings.TrimSpace(r.URL.Query().Get("status")); status != "" { + if !validSendQueueStatus(status) { + badRequest(w, errors.New("invalid send queue status")) + return + } + where += " AND sq.status=?" + args = append(args, status) + } + cursorCreatedAt, cursorID, _, err := parseSendQueueCursor(r.URL.Query().Get("cursor")) + if err != nil { + badRequest(w, err) + return + } + if cursorCreatedAt != "" { + where += " AND (sq.created_at limit { + items = items[:limit] + last := items[len(items)-1] + next = encodeSendQueueCursor(last.CreatedAt, last.QueueID) + } + respondJSON(w, http.StatusOK, map[string]any{"items": items, "nextCursor": next}) +} + +func (a *App) handleOpenAPISendEvents(w http.ResponseWriter, r *http.Request) { + item, err := a.resolveOpenAPISendQueue(r, chi.URLParam(r, "id")) + if err != nil { + respondError(w, http.StatusNotFound, "send item not found") + return + } + audit, err := a.sendAuditEvents(r, item.ID) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to load send events") + return + } + delivery, err := a.deliveryEvents(r, item.SentMessageID) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to load delivery events") + return + } + respondJSON(w, http.StatusOK, map[string]any{"auditEvents": audit, "deliveryEvents": delivery}) +} + +func (a *App) handleOpenAPIRetrySend(w http.ResponseWriter, r *http.Request) { + item, err := a.resolveOpenAPISendQueue(r, chi.URLParam(r, "id")) + if err != nil { + respondError(w, http.StatusNotFound, "send item not found") + return + } + if item.Status != sendQueueStatusFailed { + badRequest(w, errors.New("send item is not failed")) + return + } + now := a.now().UTC().Format(time.RFC3339Nano) + res, err := a.db.ExecContext(r.Context(), `UPDATE send_queue SET status=?,attempt_count=0,next_attempt_at=?,last_error='',updated_at=?,delivered_at=NULL WHERE id=? AND status=?`, sendQueueStatusQueued, now, now, item.ID, sendQueueStatusFailed) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to retry send item") + return + } + if affected, _ := res.RowsAffected(); affected == 0 { + respondError(w, http.StatusConflict, "send item status changed") + return + } + a.recordSendAudit(r.Context(), sendAuditRetry, sendQueueStatusQueued, sendAuditInputFromEntry(item, currentUser(r).ID, "")) + updated, _ := a.loadSendQueueEntryForUser(r.Context(), item.ID, currentUser(r).ID) + respondJSON(w, http.StatusOK, openAPISendStatusFromQueue(updated, updated.MailFrom)) +} + +func (a *App) handleOpenAPICancelSend(w http.ResponseWriter, r *http.Request) { + item, err := a.resolveOpenAPISendQueue(r, chi.URLParam(r, "id")) + if err != nil { + respondError(w, http.StatusNotFound, "send item not found") + return + } + if item.Status != sendQueueStatusQueued && item.Status != sendQueueStatusFailed { + badRequest(w, errors.New("send item cannot be canceled")) + return + } + now := a.now().UTC().Format(time.RFC3339Nano) + res, err := a.db.ExecContext(r.Context(), `UPDATE send_queue SET status=?,last_error='',updated_at=? WHERE id=? AND status IN (?,?)`, sendQueueStatusCanceled, now, item.ID, sendQueueStatusQueued, sendQueueStatusFailed) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to cancel send item") + return + } + if affected, _ := res.RowsAffected(); affected == 0 { + respondError(w, http.StatusConflict, "send item status changed") + return + } + a.recordSendAudit(r.Context(), sendAuditCanceled, sendQueueStatusCanceled, sendAuditInputFromEntry(item, currentUser(r).ID, "")) + updated, _ := a.loadSendQueueEntryForUser(r.Context(), item.ID, currentUser(r).ID) + respondJSON(w, http.StatusOK, openAPISendStatusFromQueue(updated, updated.MailFrom)) +} + +func sendAuditInputFromEntry(item SendQueueEntry, userID, errorText string) sendAuditInput { + return sendAuditInput{QueueID: item.ID, UserID: userID, MailboxID: item.MailboxID, SentMessageID: item.SentMessageID, Source: item.Source, MailFrom: item.MailFrom, HeaderFrom: item.HeaderFrom, Recipients: item.Recipients, Error: errorText} +} + +func (a *App) resolveOpenAPISendQueue(r *http.Request, id string) (SendQueueEntry, error) { + user := currentUser(r) + if item, err := a.loadSendQueueEntryForUser(r.Context(), strings.TrimSpace(id), user.ID); err == nil { + return item, nil + } + return a.loadLatestSendQueueForMessage(r.Context(), strings.TrimSpace(id), user.ID) +} + +func (a *App) handleOpenAPIMessage(w http.ResponseWriter, r *http.Request) { + msg, err := a.loadMessageForRequest(r, chi.URLParam(r, "id"), true) + if err != nil { + respondError(w, http.StatusNotFound, "message not found") + return + } + respondJSON(w, http.StatusOK, msg) +} + +func (a *App) handleOpenAPIListAliases(w http.ResponseWriter, r *http.Request) { + limit := parseOpenAPILimit(r, 50, 100) + sortValue, cursorID, err := parseOpenAPIListCursor(r.URL.Query().Get("cursor")) + if err != nil { + badRequest(w, err) + return + } + rows, err := a.db.QueryContext(r.Context(), `SELECT id,domain_id,source,destination,enabled,created_at FROM aliases + WHERE (?='' OR source>? OR (source=? AND id>?)) ORDER BY source,id LIMIT ?`, sortValue, sortValue, sortValue, cursorID, limit+1) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to list aliases") + return + } + defer rows.Close() + items := []Alias{} + for rows.Next() { + item, err := scanOpenAPIAlias(rows) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to scan aliases") + return + } + items = append(items, item) + } + if err := rows.Err(); err != nil { + respondError(w, http.StatusInternalServerError, "failed to list aliases") + return + } + next := "" + if len(items) > limit { + items = items[:limit] + last := items[len(items)-1] + next = encodeOpenAPIListCursor(last.Source, last.ID) + } + respondJSON(w, http.StatusOK, map[string]any{"items": items, "nextCursor": next}) +} + +func (a *App) handleOpenAPIGetAlias(w http.ResponseWriter, r *http.Request) { + item, err := scanOpenAPIAlias(a.db.QueryRowContext(r.Context(), `SELECT id,domain_id,source,destination,enabled,created_at FROM aliases WHERE id=?`, chi.URLParam(r, "id"))) + if err != nil { + respondError(w, http.StatusNotFound, "alias not found") + return + } + respondJSON(w, http.StatusOK, item) +} + +type aliasScanner interface{ Scan(...any) error } + +func scanOpenAPIAlias(row aliasScanner) (Alias, error) { + var item Alias + var enabled int + var created string + err := row.Scan(&item.ID, &item.DomainID, &item.Source, &item.Destination, &enabled, &created) + item.Enabled = intBool(enabled) + item.CreatedAt = parseTime(created) + return item, err +} + +func (a *App) sendAuditEvents(r *http.Request, queueID string) ([]SendAuditEvent, error) { + rows, err := a.db.QueryContext(r.Context(), `SELECT id,queue_id,mailbox_id,sent_message_id,source,event,status,mail_from,header_from,recipients_json,error,created_at FROM send_audit_events WHERE queue_id=? ORDER BY created_at,id`, queueID) + if err != nil { + return nil, err + } + defer rows.Close() + items := []SendAuditEvent{} + for rows.Next() { + var item SendAuditEvent + var recipientsJSON, createdAt string + if err := rows.Scan(&item.ID, &item.QueueID, &item.MailboxID, &item.SentMessageID, &item.Source, &item.Event, &item.Status, &item.MailFrom, &item.HeaderFrom, &recipientsJSON, &item.Error, &createdAt); err != nil { + return nil, err + } + item.Recipients = jsonDecodeSlice(recipientsJSON) + item.CreatedAt = parseTime(createdAt) + items = append(items, item) + } + return items, rows.Err() +} + +func (a *App) deliveryEvents(r *http.Request, sentMessageID string) ([]DeliveryEvent, error) { + rows, err := a.db.QueryContext(r.Context(), `SELECT id,external_id,provider,queue_id,sent_message_id,rfc_message_id,recipient,status,reason,occurred_at,created_at FROM delivery_events WHERE sent_message_id=? ORDER BY occurred_at,id`, sentMessageID) + if err != nil { + return nil, err + } + defer rows.Close() + items := []DeliveryEvent{} + for rows.Next() { + var item DeliveryEvent + var occurredAt, createdAt string + if err := rows.Scan(&item.ID, &item.ExternalID, &item.Provider, &item.QueueID, &item.MessageID, &item.RFCMessageID, &item.Recipient, &item.Status, &item.Reason, &occurredAt, &createdAt); err != nil { + return nil, err + } + item.OccurredAt = parseTime(occurredAt) + item.CreatedAt = parseTime(createdAt) + items = append(items, item) + } + return items, rows.Err() +} diff --git a/apps/api/internal/app/open_api_handlers.go b/apps/api/internal/app/open_api_handlers.go index 0eebf77..5afabc2 100644 --- a/apps/api/internal/app/open_api_handlers.go +++ b/apps/api/internal/app/open_api_handlers.go @@ -2,7 +2,11 @@ package app import ( "context" + "crypto/sha256" "database/sql" + "encoding/base64" + "encoding/hex" + "encoding/json" "errors" "net/http" "strconv" @@ -14,7 +18,14 @@ import ( ) func (a *App) handleOpenAPIListDomains(w http.ResponseWriter, r *http.Request) { - rows, err := a.db.QueryContext(r.Context(), `SELECT id,name,status,dkim_selector,dkim_public_key,dns_status,dns_checked_at,created_at FROM domains ORDER BY name`) + limit := parseOpenAPILimit(r, 50, 100) + sortValue, cursorID, err := parseOpenAPIListCursor(r.URL.Query().Get("cursor")) + if err != nil { + badRequest(w, err) + return + } + rows, err := a.db.QueryContext(r.Context(), `SELECT id,name,status,dkim_selector,dkim_public_key,dns_status,dns_checked_at,created_at FROM domains + WHERE (?='' OR name>? OR (name=? AND id>?)) ORDER BY name,id LIMIT ?`, sortValue, sortValue, sortValue, cursorID, limit+1) if err != nil { respondError(w, http.StatusInternalServerError, "failed to list domains") return @@ -33,7 +44,13 @@ func (a *App) handleOpenAPIListDomains(w http.ResponseWriter, r *http.Request) { respondError(w, http.StatusInternalServerError, "failed to list domains") return } - respondJSON(w, http.StatusOK, map[string]any{"items": items}) + next := "" + if len(items) > limit { + items = items[:limit] + last := items[len(items)-1] + next = encodeOpenAPIListCursor(last.Name, last.ID) + } + respondJSON(w, http.StatusOK, map[string]any{"items": items, "nextCursor": next}) } func (a *App) handleOpenAPICreateDomain(w http.ResponseWriter, r *http.Request) { @@ -121,8 +138,15 @@ func (a *App) handleOpenAPIDeleteDomain(w http.ResponseWriter, r *http.Request) } func (a *App) handleOpenAPIListMailboxes(w http.ResponseWriter, r *http.Request) { + limit := parseOpenAPILimit(r, 50, 100) + sortValue, cursorID, err := parseOpenAPIListCursor(r.URL.Query().Get("cursor")) + if err != nil { + badRequest(w, err) + return + } rows, err := a.db.QueryContext(r.Context(), `SELECT mb.id,mb.user_id,u.email,mb.domain_id,mb.local_part,mb.address,mb.display_name,mb.quota_mb,mb.status,mb.created_at - FROM mailboxes mb JOIN users u ON u.id=mb.user_id ORDER BY mb.address`) + FROM mailboxes mb JOIN users u ON u.id=mb.user_id + WHERE (?='' OR mb.address>? OR (mb.address=? AND mb.id>?)) ORDER BY mb.address,mb.id LIMIT ?`, sortValue, sortValue, sortValue, cursorID, limit+1) if err != nil { respondError(w, http.StatusInternalServerError, "failed to list mailboxes") return @@ -141,7 +165,13 @@ func (a *App) handleOpenAPIListMailboxes(w http.ResponseWriter, r *http.Request) respondError(w, http.StatusInternalServerError, "failed to list mailboxes") return } - respondJSON(w, http.StatusOK, map[string]any{"items": items}) + next := "" + if len(items) > limit { + items = items[:limit] + last := items[len(items)-1] + next = encodeOpenAPIListCursor(last.Address, last.ID) + } + respondJSON(w, http.StatusOK, map[string]any{"items": items, "nextCursor": next}) } func (a *App) handleOpenAPICreateMailbox(w http.ResponseWriter, r *http.Request) { @@ -185,16 +215,31 @@ func (a *App) handleOpenAPICreateMailbox(w http.ResponseWriter, r *http.Request) if displayName == "" { displayName = address } - userID, err := a.resolveMailboxOwner(r, req.UserID, req.OwnerEmail, address, displayName, req.Password) + passwordHash, err := bcrypt.GenerateFromPassword([]byte(req.Password), bcrypt.DefaultCost) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to hash password") + return + } + tx, err := a.db.BeginTx(r.Context(), nil) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to start transaction") + return + } + defer tx.Rollback() + userID, err := a.resolveMailboxOwnerTx(r.Context(), tx, req.UserID, req.OwnerEmail, address, displayName, string(passwordHash)) if err != nil { respondMailboxOwnerError(w, err) return } - mailboxID, err := a.createMailbox(r.Context(), userID, req.DomainID, localPart, displayName, req.Password, req.QuotaMB, "active") + mailboxID, err := a.createMailboxWithPasswordHashTx(r.Context(), tx, userID, req.DomainID, localPart, displayName, string(passwordHash), req.QuotaMB, "active") if err != nil { badRequest(w, err) return } + if err := tx.Commit(); err != nil { + respondError(w, http.StatusInternalServerError, "failed to create mailbox") + return + } mailbox, err := a.mailboxByID(r.Context(), mailboxID) if err != nil { respondError(w, http.StatusInternalServerError, "failed to load mailbox") @@ -318,6 +363,52 @@ func (a *App) handleOpenAPIDeleteMailbox(w http.ResponseWriter, r *http.Request) respondJSON(w, http.StatusOK, map[string]any{"ok": true}) } +func (a *App) handleOpenAPIResetMailboxPassword(w http.ResponseWriter, r *http.Request) { + var req struct { + Password string `json:"password"` + } + if err := decodeJSON(r, &req); err != nil { + badRequest(w, err) + return + } + if len(req.Password) < 8 { + badRequest(w, errors.New("password must be at least 8 characters")) + return + } + var userID string + if err := a.db.QueryRowContext(r.Context(), `SELECT user_id FROM mailboxes WHERE id=?`, chi.URLParam(r, "id")).Scan(&userID); err != nil { + respondError(w, http.StatusNotFound, "mailbox not found") + return + } + hash, err := bcrypt.GenerateFromPassword([]byte(req.Password), bcrypt.DefaultCost) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to hash password") + return + } + now := a.now().UTC().Format(time.RFC3339Nano) + tx, err := a.db.BeginTx(r.Context(), nil) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to start transaction") + return + } + defer tx.Rollback() + if _, err := tx.ExecContext(r.Context(), `UPDATE users SET password_hash=?,updated_at=? WHERE id=?`, string(hash), now, userID); err != nil { + respondError(w, http.StatusInternalServerError, "failed to reset password") + return + } + res, err := tx.ExecContext(r.Context(), `UPDATE mailboxes SET password_hash=?,updated_at=? WHERE user_id=?`, string(hash), now, userID) + if err != nil { + respondError(w, http.StatusInternalServerError, "failed to reset mailbox passwords") + return + } + if err := tx.Commit(); err != nil { + respondError(w, http.StatusInternalServerError, "failed to save password") + return + } + affected, _ := res.RowsAffected() + respondJSON(w, http.StatusOK, map[string]any{"ok": true, "affectedMailboxes": affected}) +} + func (a *App) handleOpenAPISendMail(w http.ResponseWriter, r *http.Request) { var req mailComposeInput if err := decodeJSON(r, &req); err != nil { @@ -329,8 +420,31 @@ func (a *App) handleOpenAPISendMail(w http.ResponseWriter, r *http.Request) { respondError(w, http.StatusNotFound, "mailbox not found") return } + idempotencyKey := strings.TrimSpace(r.Header.Get("Idempotency-Key")) + requestJSON, _ := json.Marshal(req) + requestSum := sha256.Sum256(requestJSON) + requestHash := hex.EncodeToString(requestSum[:]) + if idempotencyKey != "" { + if len(idempotencyKey) > 128 || strings.ContainsAny(idempotencyKey, "\r\n") { + badRequest(w, errors.New("invalid Idempotency-Key")) + return + } + status, replayed, err := a.reserveOpenAPISendIdempotency(r.Context(), currentUser(r).ID, idempotencyKey, requestHash) + if err != nil { + respondError(w, http.StatusConflict, err.Error()) + return + } + if replayed { + w.Header().Set("Idempotency-Replayed", "true") + respondJSON(w, http.StatusOK, status) + return + } + } msg, err := a.sendMailWithSource(r.Context(), currentUser(r), mb, req, sendSourceOpenAPI) if err != nil { + if idempotencyKey != "" { + _, _ = a.db.ExecContext(r.Context(), `DELETE FROM send_idempotency_keys WHERE user_id=? AND idempotency_key=? AND sent_message_id=''`, currentUser(r).ID, idempotencyKey) + } respondSendError(w, err) return } @@ -345,6 +459,10 @@ func (a *App) handleOpenAPISendMail(w http.ResponseWriter, r *http.Request) { status = openAPISendStatusFromQueue(item, mb.Address) } } + a.applyDeliveryStatus(r.Context(), &status) + if idempotencyKey != "" { + _, _ = a.db.ExecContext(r.Context(), `UPDATE send_idempotency_keys SET sent_message_id=?,queue_id=? WHERE user_id=? AND idempotency_key=?`, status.MessageID, status.QueueID, currentUser(r).ID, idempotencyKey) + } respondJSON(w, http.StatusCreated, status) } @@ -360,7 +478,9 @@ func (a *App) handleOpenAPISendStatus(w http.ResponseWriter, r *http.Request) { if mb, mbErr := a.mailboxByID(r.Context(), item.MailboxID); mbErr == nil { mailboxAddress = mb.Address } - respondJSON(w, http.StatusOK, openAPISendStatusFromQueue(item, mailboxAddress)) + status := openAPISendStatusFromQueue(item, mailboxAddress) + a.applyDeliveryStatus(r.Context(), &status) + respondJSON(w, http.StatusOK, status) return } msg, err := a.loadOpenAPISentMessageForUser(r.Context(), id, user.ID) @@ -383,7 +503,11 @@ func (a *App) handleOpenAPIMailboxMessages(w http.ResponseWriter, r *http.Reques return } limit := parseOpenAPILimit(r, 30, 100) - offset := parseOpenAPIOffset(r) + cursorReceivedAt, cursorID, offset, err := parseOpenAPIMessageCursor(r.URL.Query().Get("cursor")) + if err != nil { + badRequest(w, err) + return + } folder := strings.TrimSpace(r.URL.Query().Get("folder")) if folder == "" { folder = "Inbox" @@ -399,11 +523,20 @@ func (a *App) handleOpenAPIMailboxMessages(w http.ResponseWriter, r *http.Reques like := "%" + q + "%" args = append(args, like, like, like, like, like, like) } - args = append(args, limit+1, offset) - rows, err := a.db.QueryContext(r.Context(), `SELECT m.id,m.mailbox_id,m.folder_id,f.name,m.message_uid,m.imap_uid,m.imap_modseq,m.message_id,m.subject,m.from_addr,COALESCE(m.from_name,''),m.to_addrs,m.cc_addrs,m.bcc_addrs,m.sent_at,m.received_at,m.snippet,m.is_read,m.is_starred,m.has_attachments,m.size_bytes + if cursorReceivedAt != "" { + where += " AND (m.received_at 0 { + query += " OFFSET ?" + args = append(args, offset) + } + rows, err := a.db.QueryContext(r.Context(), query, args...) if err != nil { respondError(w, http.StatusInternalServerError, "failed to load messages") return @@ -425,7 +558,8 @@ func (a *App) handleOpenAPIMailboxMessages(w http.ResponseWriter, r *http.Reques nextCursor := "" if len(items) > limit { items = items[:limit] - nextCursor = strconv.Itoa(offset + limit) + last := items[len(items)-1] + nextCursor = encodeOpenAPIMessageCursor(last.ReceivedAt, last.ID) } respondJSON(w, http.StatusOK, map[string]any{"items": items, "nextCursor": nextCursor}) } @@ -459,29 +593,44 @@ func scanMailbox(row mailboxScanner) (Mailbox, error) { } type openAPISendStatus struct { - ID string `json:"id"` - QueueID string `json:"queueId,omitempty"` - Status string `json:"status"` - MessageID string `json:"messageId"` - RFCMessageID string `json:"rfcMessageId"` - MailboxID string `json:"mailboxId"` - MailboxAddress string `json:"mailboxAddress,omitempty"` - Subject string `json:"subject,omitempty"` - Recipients []string `json:"recipients,omitempty"` - AttemptCount int `json:"attemptCount,omitempty"` - MaxAttempts int `json:"maxAttempts,omitempty"` - NextAttemptAt *time.Time `json:"nextAttemptAt,omitempty"` - LastError string `json:"lastError,omitempty"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt *time.Time `json:"updatedAt,omitempty"` - DeliveredAt *time.Time `json:"deliveredAt,omitempty"` + ID string `json:"id"` + QueueID string `json:"queueId,omitempty"` + Status string `json:"status"` + QueueStatus string `json:"queueStatus,omitempty"` + MessageID string `json:"messageId"` + RFCMessageID string `json:"rfcMessageId"` + MailboxID string `json:"mailboxId"` + MailboxAddress string `json:"mailboxAddress,omitempty"` + Subject string `json:"subject,omitempty"` + Recipients []string `json:"recipients,omitempty"` + AttemptCount int `json:"attemptCount,omitempty"` + MaxAttempts int `json:"maxAttempts,omitempty"` + NextAttemptAt *time.Time `json:"nextAttemptAt,omitempty"` + LastError string `json:"lastError,omitempty"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt *time.Time `json:"updatedAt,omitempty"` + DeliveredAt *time.Time `json:"deliveredAt,omitempty"` + RecipientStatuses []openAPIRecipientStatus `json:"recipientStatuses,omitempty"` +} + +type openAPIRecipientStatus struct { + Recipient string `json:"recipient"` + Status string `json:"status"` + Reason string `json:"reason,omitempty"` + Provider string `json:"provider,omitempty"` + OccurredAt time.Time `json:"occurredAt"` } func openAPISendStatusFromQueue(item SendQueueEntry, mailboxAddress string) openAPISendStatus { + status := item.Status + if status == sendQueueStatusDelivered { + status = "relayed" + } return openAPISendStatus{ - ID: item.ID, + ID: firstNonEmpty(item.SentMessageID, item.ID), QueueID: item.ID, - Status: item.Status, + Status: status, + QueueStatus: item.Status, MessageID: item.SentMessageID, RFCMessageID: item.MessageID, MailboxID: item.MailboxID, @@ -498,6 +647,139 @@ func openAPISendStatusFromQueue(item SendQueueEntry, mailboxAddress string) open } } +func (a *App) reserveOpenAPISendIdempotency(ctx context.Context, userID, key, requestHash string) (openAPISendStatus, bool, error) { + _, _ = a.db.ExecContext(ctx, `DELETE FROM send_idempotency_keys WHERE created_at 0 { + return openAPISendStatus{}, false, nil + } + var storedHash, sentMessageID, queueID string + if err := a.db.QueryRowContext(ctx, `SELECT request_hash,sent_message_id,queue_id FROM send_idempotency_keys WHERE user_id=? AND idempotency_key=?`, userID, key).Scan(&storedHash, &sentMessageID, &queueID); err != nil { + return openAPISendStatus{}, false, err + } + if storedHash != requestHash { + return openAPISendStatus{}, false, errors.New("Idempotency-Key was already used with a different request") + } + if sentMessageID == "" { + return openAPISendStatus{}, false, errors.New("a request with this Idempotency-Key is still processing") + } + item, err := a.loadSendQueueEntryForUser(ctx, queueID, userID) + if err != nil { + return openAPISendStatus{}, false, err + } + status := openAPISendStatusFromQueue(item, item.MailFrom) + a.applyDeliveryStatus(ctx, &status) + return status, true, nil +} + +func (a *App) applyDeliveryStatus(ctx context.Context, status *openAPISendStatus) { + rows, err := a.db.QueryContext(ctx, `SELECT recipient,status,reason,provider,occurred_at FROM delivery_events + WHERE sent_message_id=? ORDER BY occurred_at DESC,id DESC`, status.MessageID) + if err != nil { + return + } + defer rows.Close() + seen := map[string]bool{} + counts := map[string]int{} + for rows.Next() { + var item openAPIRecipientStatus + var occurredAt string + if rows.Scan(&item.Recipient, &item.Status, &item.Reason, &item.Provider, &occurredAt) != nil || seen[item.Recipient] { + continue + } + seen[item.Recipient] = true + item.OccurredAt = parseTime(occurredAt) + status.RecipientStatuses = append(status.RecipientStatuses, item) + counts[item.Status]++ + } + if len(status.RecipientStatuses) == 0 { + return + } + if len(status.RecipientStatuses) < len(status.Recipients) || len(counts) > 1 { + status.Status = "partial" + return + } + for _, value := range []string{"complained", "bounced", "rejected", "deferred", "delivered"} { + if counts[value] > 0 { + status.Status = value + return + } + } +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if strings.TrimSpace(value) != "" { + return value + } + } + return "" +} + +type openAPIMessageCursor struct { + ReceivedAt string `json:"receivedAt"` + ID string `json:"id"` +} + +type openAPIListCursor struct { + Sort string `json:"sort"` + ID string `json:"id"` +} + +func encodeOpenAPIMessageCursor(receivedAt time.Time, id string) string { + payload, _ := json.Marshal(openAPIMessageCursor{ReceivedAt: receivedAt.UTC().Format(time.RFC3339Nano), ID: id}) + return base64.RawURLEncoding.EncodeToString(payload) +} + +func parseOpenAPIMessageCursor(raw string) (receivedAt, id string, offset int, err error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return "", "", 0, nil + } + if n, convErr := strconv.Atoi(raw); convErr == nil { + if n < 0 { + return "", "", 0, errors.New("invalid cursor") + } + return "", "", n, nil + } + data, err := base64.RawURLEncoding.DecodeString(raw) + if err != nil { + return "", "", 0, errors.New("invalid cursor") + } + var cursor openAPIMessageCursor + if err := json.Unmarshal(data, &cursor); err != nil || cursor.ReceivedAt == "" || cursor.ID == "" { + return "", "", 0, errors.New("invalid cursor") + } + if _, err := time.Parse(time.RFC3339Nano, cursor.ReceivedAt); err != nil { + return "", "", 0, errors.New("invalid cursor") + } + return cursor.ReceivedAt, cursor.ID, 0, nil +} + +func encodeOpenAPIListCursor(sortValue, id string) string { + payload, _ := json.Marshal(openAPIListCursor{Sort: sortValue, ID: id}) + return base64.RawURLEncoding.EncodeToString(payload) +} + +func parseOpenAPIListCursor(raw string) (sortValue, id string, err error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return "", "", nil + } + data, err := base64.RawURLEncoding.DecodeString(raw) + if err != nil { + return "", "", errors.New("invalid cursor") + } + var cursor openAPIListCursor + if err := json.Unmarshal(data, &cursor); err != nil || cursor.Sort == "" || cursor.ID == "" { + return "", "", errors.New("invalid cursor") + } + return cursor.Sort, cursor.ID, nil +} + func openAPISendStatusFromMessage(msg *MailMessage, mailboxAddress string) openAPISendStatus { recipients := append(append([]string{}, msg.To...), msg.CC...) recipients = append(recipients, msg.BCC...) @@ -521,12 +803,19 @@ func timePtr(t time.Time) *time.Time { return &t } -func (a *App) resolveMailboxOwner(r *http.Request, userID, ownerEmail, address, displayName, password string) (string, error) { +func (a *App) resolveMailboxOwnerTx(ctx context.Context, tx *sql.Tx, userID, ownerEmail, address, displayName, passwordHash string) (string, error) { userID = strings.TrimSpace(userID) if userID != "" { - if err := a.ensureActiveUserExists(r.Context(), userID); err != nil { + var disabled int + if err := tx.QueryRowContext(ctx, `SELECT disabled FROM users WHERE id=?`, userID).Scan(&disabled); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return "", errNotFound + } return "", err } + if intBool(disabled) { + return "", errors.New("owner user is disabled") + } return userID, nil } email := normalizeEmail(ownerEmail) @@ -537,32 +826,21 @@ func (a *App) resolveMailboxOwner(r *http.Request, userID, ownerEmail, address, return "", errors.New("invalid owner email") } var existing string - err := a.db.QueryRowContext(r.Context(), `SELECT id FROM users WHERE email=? AND disabled=0`, email).Scan(&existing) + err := tx.QueryRowContext(ctx, `SELECT id FROM users WHERE email=? AND disabled=0`, email).Scan(&existing) if err == nil { return existing, nil } if !errors.Is(err, sql.ErrNoRows) { return "", err } - return a.createMailboxOwnerUser(r, email, displayName, password) -} - -func (a *App) createMailboxOwnerUser(r *http.Request, email, displayName, password string) (string, error) { - hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost) - if err != nil { - return "", err - } - userID := newID("usr") + userID = newID("usr") now := a.now().UTC().Format(time.RFC3339Nano) if displayName == "" { displayName = email } - _, err = a.db.ExecContext(r.Context(), `INSERT INTO users(id,email,display_name,role,password_hash,disabled,created_at,updated_at) - VALUES(?,?,?,?,?,?,?,?)`, userID, email, displayName, "user", string(hash), 0, now, now) - if err != nil { - return "", err - } - return userID, nil + _, err = tx.ExecContext(ctx, `INSERT INTO users(id,email,display_name,role,password_hash,disabled,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,?)`, userID, email, displayName, "user", passwordHash, 0, now, now) + return userID, err } func (a *App) ensureActiveUserExists(ctx context.Context, userID string) error { diff --git a/apps/api/internal/app/router_auth.go b/apps/api/internal/app/router_auth.go index 1ccf9f4..680bd39 100644 --- a/apps/api/internal/app/router_auth.go +++ b/apps/api/internal/app/router_auth.go @@ -15,6 +15,7 @@ import ( type contextKey string const userContextKey contextKey = "user" +const apiTokenScopesContextKey contextKey = "api_token_scopes" func (a *App) Router() http.Handler { r := chi.NewRouter() @@ -75,22 +76,9 @@ func (a *App) Router() http.Handler { r.With(a.requireExternalIMAPEnabled).Get("/external-imap-oauth/{provider}/callback", a.handleExternalIMAPOAuthCallback) r.With(a.requireAuth).Get("/events", a.handleEvents) - r.Group(func(r chi.Router) { - r.Use(a.requireAPIToken) - r.With(a.requireAdminAccess, a.requireAnyPermission(PermissionDomainsView, PermissionDNSView, PermissionMailboxesView, PermissionAliasesView, PermissionSettingsView, PermissionTemplatesView)).Get("/open/domains", a.handleOpenAPIListDomains) - r.With(a.requireAdminAccess, a.requirePermission(PermissionDomainsCreate)).Post("/open/domains", a.handleOpenAPICreateDomain) - r.With(a.requireAdminAccess, a.requireAnyPermission(PermissionDomainsView, PermissionDNSView, PermissionMailboxesView, PermissionAliasesView, PermissionSettingsView, PermissionTemplatesView)).Get("/open/domains/{id}", a.handleOpenAPIGetDomain) - r.With(a.requireAdminAccess, a.requirePermission(PermissionDomainsUpdate)).Post("/open/domains/{id}", a.handleOpenAPIUpdateDomain) - r.With(a.requireAdminAccess, a.requirePermission(PermissionDomainsDelete)).Delete("/open/domains/{id}", a.handleOpenAPIDeleteDomain) - r.With(a.requireAdminAccess, a.requireAnyPermission(PermissionMailboxesView, PermissionMessagesView)).Get("/open/mailboxes", a.handleOpenAPIListMailboxes) - r.With(a.requireAdminAccess, a.requirePermission(PermissionMailboxesCreate)).Post("/open/mailboxes", a.handleOpenAPICreateMailbox) - r.With(a.requireAdminAccess, a.requireAnyPermission(PermissionMailboxesView, PermissionMessagesView)).Get("/open/mailboxes/{id}", a.handleOpenAPIGetMailbox) - r.With(a.requireAdminAccess, a.requirePermission(PermissionMailboxesUpdate)).Post("/open/mailboxes/{id}", a.handleOpenAPIUpdateMailbox) - r.With(a.requireAdminAccess, a.requirePermission(PermissionMailboxesDelete)).Delete("/open/mailboxes/{id}", a.handleOpenAPIDeleteMailbox) - r.With(a.requirePermission(PermissionMailSend)).Post("/open/send", a.handleOpenAPISendMail) - r.With(a.requirePermission(PermissionMailRead)).Get("/open/send/{id}", a.handleOpenAPISendStatus) - r.With(a.requirePermission(PermissionMailRead)).Get("/open/mailboxes/{id}/messages", a.handleOpenAPIMailboxMessages) - }) + r.Post("/open/v1/delivery-events", a.handleOpenAPIDeliveryWebhook) + r.Route("/open", func(r chi.Router) { a.registerOpenAPIRoutes(r) }) + r.Route("/open/v1", func(r chi.Router) { a.registerOpenAPIRoutes(r) }) r.Group(func(r chi.Router) { r.Use(a.requireAuth) @@ -179,6 +167,37 @@ func (a *App) Router() http.Handler { return r } +func (a *App) registerOpenAPIRoutes(r chi.Router) { + r.Use(a.requireAPIToken) + r.With(a.requireAPITokenScope("domains:read"), a.requireAdminAccess, a.requireAnyPermission(PermissionDomainsView, PermissionDNSView, PermissionMailboxesView, PermissionAliasesView, PermissionSettingsView, PermissionTemplatesView)).Get("/domains", a.handleOpenAPIListDomains) + r.With(a.requireAPITokenScope("domains:write"), a.requireAdminAccess, a.requirePermission(PermissionDomainsCreate)).Post("/domains", a.handleOpenAPICreateDomain) + r.With(a.requireAPITokenScope("domains:read"), a.requireAdminAccess, a.requireAnyPermission(PermissionDomainsView, PermissionDNSView, PermissionMailboxesView, PermissionAliasesView, PermissionSettingsView, PermissionTemplatesView)).Get("/domains/{id}", a.handleOpenAPIGetDomain) + r.With(a.requireAPITokenScope("domains:write"), a.requireAdminAccess, a.requirePermission(PermissionDomainsUpdate)).Post("/domains/{id}", a.handleOpenAPIUpdateDomain) + r.With(a.requireAPITokenScope("domains:write"), a.requireAdminAccess, a.requirePermission(PermissionDomainsDelete)).Delete("/domains/{id}", a.handleOpenAPIDeleteDomain) + r.With(a.requireAPITokenScope("dns:read"), a.requireAdminAccess, a.requirePermission(PermissionDNSView)).Get("/domains/{id}/dns-records", a.handleDNSRecords) + r.With(a.requireAPITokenScope("dns:check"), a.requireAdminAccess, a.requirePermission(PermissionDNSCheck)).Post("/domains/{id}/dns-check", a.handleDNSCheck) + r.With(a.requireAPITokenScope("mailboxes:read"), a.requireAdminAccess, a.requireAnyPermission(PermissionMailboxesView, PermissionMessagesView)).Get("/mailboxes", a.handleOpenAPIListMailboxes) + r.With(a.requireAPITokenScope("mailboxes:write"), a.requireAdminAccess, a.requirePermission(PermissionMailboxesCreate)).Post("/mailboxes", a.handleOpenAPICreateMailbox) + r.With(a.requireAPITokenScope("mailboxes:read"), a.requireAdminAccess, a.requireAnyPermission(PermissionMailboxesView, PermissionMessagesView)).Get("/mailboxes/{id}", a.handleOpenAPIGetMailbox) + r.With(a.requireAPITokenScope("mailboxes:write"), a.requireAdminAccess, a.requirePermission(PermissionMailboxesUpdate)).Post("/mailboxes/{id}", a.handleOpenAPIUpdateMailbox) + r.With(a.requireAPITokenScope("mailboxes:write"), a.requireAdminAccess, a.requirePermission(PermissionUsersResetPassword)).Post("/mailboxes/{id}/password", a.handleOpenAPIResetMailboxPassword) + r.With(a.requireAPITokenScope("mailboxes:write"), a.requireAdminAccess, a.requirePermission(PermissionMailboxesDelete)).Delete("/mailboxes/{id}", a.handleOpenAPIDeleteMailbox) + r.With(a.requireAPITokenScope("messages:send"), a.requirePermission(PermissionMailSend)).Post("/send", a.handleOpenAPISendMail) + r.With(a.requireAPITokenScope("messages:read"), a.requirePermission(PermissionMailRead)).Get("/send", a.handleOpenAPIListSends) + r.With(a.requireAPITokenScope("messages:read"), a.requirePermission(PermissionMailRead)).Get("/send/{id}", a.handleOpenAPISendStatus) + r.With(a.requireAPITokenScope("messages:read"), a.requirePermission(PermissionMailRead)).Get("/send/{id}/events", a.handleOpenAPISendEvents) + r.With(a.requireAPITokenScope("messages:manage"), a.requirePermission(PermissionMailSend)).Post("/send/{id}/retry", a.handleOpenAPIRetrySend) + r.With(a.requireAPITokenScope("messages:manage"), a.requirePermission(PermissionMailSend)).Post("/send/{id}/cancel", a.handleOpenAPICancelSend) + r.With(a.requireAPITokenScope("messages:read"), a.requirePermission(PermissionMailRead)).Get("/mailboxes/{id}/messages", a.handleOpenAPIMailboxMessages) + r.With(a.requireAPITokenScope("messages:read"), a.requirePermission(PermissionMailRead)).Get("/messages/{id}", a.handleOpenAPIMessage) + r.With(a.requireAPITokenScope("messages:read"), a.requirePermission(PermissionMailAttachments)).Get("/attachments/{id}", a.handleAttachment) + r.With(a.requireAPITokenScope("aliases:read"), a.requireAdminAccess, a.requirePermission(PermissionAliasesView)).Get("/aliases", a.handleOpenAPIListAliases) + r.With(a.requireAPITokenScope("aliases:write"), a.requireAdminAccess, a.requirePermission(PermissionAliasesCreate)).Post("/aliases", a.handleCreateAlias) + r.With(a.requireAPITokenScope("aliases:read"), a.requireAdminAccess, a.requirePermission(PermissionAliasesView)).Get("/aliases/{id}", a.handleOpenAPIGetAlias) + r.With(a.requireAPITokenScope("aliases:write"), a.requireAdminAccess, a.requirePermission(PermissionAliasesUpdate)).Post("/aliases/{id}", a.handleUpdateAlias) + r.With(a.requireAPITokenScope("aliases:write"), a.requireAdminAccess, a.requirePermission(PermissionAliasesDelete)).Delete("/aliases/{id}", a.handleDeleteAlias) +} + func (a *App) corsMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { origin := r.Header.Get("Origin") @@ -186,7 +205,7 @@ func (a *App) corsMiddleware(next http.Handler) http.Handler { w.Header().Set("Access-Control-Allow-Origin", origin) w.Header().Set("Vary", "Origin") w.Header().Set("Access-Control-Allow-Credentials", "true") - w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization") + w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, Idempotency-Key") w.Header().Set("Access-Control-Allow-Methods", "GET,POST,DELETE,OPTIONS") } if r.Method == http.MethodOptions { @@ -210,15 +229,30 @@ func (a *App) requireAuth(next http.Handler) http.Handler { func (a *App) requireAPIToken(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - user, err := a.authenticateAPIToken(r) + user, scopes, err := a.authenticateAPIToken(r) if err != nil { respondError(w, http.StatusUnauthorized, "api token required") return } - next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), userContextKey, user))) + ctx := context.WithValue(r.Context(), userContextKey, user) + ctx = context.WithValue(ctx, apiTokenScopesContextKey, scopes) + next.ServeHTTP(w, r.WithContext(ctx)) }) } +func (a *App) requireAPITokenScope(scope string) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + scopes, _ := r.Context().Value(apiTokenScopesContextKey).(map[string]bool) + if !scopes["*"] && !scopes[scope] { + respondError(w, http.StatusForbidden, "api token scope required: "+scope) + return + } + next.ServeHTTP(w, r) + }) + } +} + func currentUser(r *http.Request) *User { user, _ := r.Context().Value(userContextKey).(*User) return user @@ -250,33 +284,37 @@ func (a *App) authenticateRequest(r *http.Request) (*User, error) { return &u, nil } -func (a *App) authenticateAPIToken(r *http.Request) (*User, error) { +func (a *App) authenticateAPIToken(r *http.Request) (*User, map[string]bool, error) { token := bearerToken(r) if token == "" { - return nil, errors.New("no api token") + return nil, nil, errors.New("no api token") } now := a.now().UTC().Format(time.RFC3339Nano) - row := a.db.QueryRowContext(r.Context(), `SELECT at.id,u.id,u.email,u.display_name,u.role,u.disabled,u.two_factor_enabled,u.created_at + row := a.db.QueryRowContext(r.Context(), `SELECT at.id,at.scopes_json,u.id,u.email,u.display_name,u.role,u.disabled,u.two_factor_enabled,u.created_at FROM api_tokens at JOIN users u ON u.id=at.user_id WHERE at.token_hash=? AND at.disabled=0 AND at.expires_at > ?`, hashToken(token), now) - var tokenID string + var tokenID, scopesJSON string var u User var disabled, twoFactorEnabled int var created string - if err := row.Scan(&tokenID, &u.ID, &u.Email, &u.DisplayName, &u.Role, &disabled, &twoFactorEnabled, &created); err != nil { - return nil, err + if err := row.Scan(&tokenID, &scopesJSON, &u.ID, &u.Email, &u.DisplayName, &u.Role, &disabled, &twoFactorEnabled, &created); err != nil { + return nil, nil, err } u.Disabled = intBool(disabled) u.TwoFactorEnabled = intBool(twoFactorEnabled) u.CreatedAt = parseTime(created) if u.Disabled { - return nil, errors.New("disabled") + return nil, nil, errors.New("disabled") } if err := a.attachUserAuthorization(r.Context(), &u); err != nil { - return nil, err + return nil, nil, err } _, _ = a.db.ExecContext(r.Context(), `UPDATE api_tokens SET last_used_at=? WHERE id=?`, now, tokenID) - return &u, nil + scopes := map[string]bool{} + for _, scope := range jsonDecodeSlice(scopesJSON) { + scopes[scope] = true + } + return &u, scopes, nil } func bearerToken(r *http.Request) string { diff --git a/apps/api/internal/app/send_queue.go b/apps/api/internal/app/send_queue.go index c0dd55a..722d109 100644 --- a/apps/api/internal/app/send_queue.go +++ b/apps/api/internal/app/send_queue.go @@ -365,10 +365,26 @@ func (a *App) recordSendAudit(ctx context.Context, event, status string, in send if source == "" { source = "unknown" } - _, err := a.db.ExecContext(ctx, `INSERT INTO send_audit_events(id,queue_id,user_id,mailbox_id,sent_message_id,source,event,status,mail_from,header_from,recipients_json,error,created_at) - VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`, newID("audit"), in.QueueID, in.UserID, in.MailboxID, in.SentMessageID, source, event, status, normalizeEmail(in.MailFrom), normalizeEmail(in.HeaderFrom), jsonEncode(dedupeEmails(in.Recipients)), in.Error, a.now().UTC().Format(time.RFC3339Nano)) + id := newID("audit") + createdAt := a.now().UTC() + item := SendAuditEvent{ID: id, QueueID: in.QueueID, MailboxID: in.MailboxID, SentMessageID: in.SentMessageID, Source: source, Event: event, Status: status, MailFrom: normalizeEmail(in.MailFrom), HeaderFrom: normalizeEmail(in.HeaderFrom), Recipients: dedupeEmails(in.Recipients), Error: in.Error, CreatedAt: createdAt} + tx, err := a.db.BeginTx(ctx, nil) if err != nil { + a.log.Warn("failed to start send audit transaction", "event", event, "error", err) + return + } + defer tx.Rollback() + if _, err := tx.ExecContext(ctx, `INSERT INTO send_audit_events(id,queue_id,user_id,mailbox_id,sent_message_id,source,event,status,mail_from,header_from,recipients_json,error,created_at) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`, id, in.QueueID, in.UserID, in.MailboxID, in.SentMessageID, source, event, status, item.MailFrom, item.HeaderFrom, jsonEncode(item.Recipients), in.Error, createdAt.Format(time.RFC3339Nano)); err != nil { a.log.Warn("failed to record send audit", "event", event, "error", err) + return + } + if err := a.enqueueStatusWebhook(ctx, tx, "audit:"+id, "send."+event, in.MailboxID, item); err != nil { + a.log.Warn("failed to enqueue send status webhook", "event", event, "error", err) + return + } + if err := tx.Commit(); err != nil { + a.log.Warn("failed to commit send audit", "event", event, "error", err) } } diff --git a/apps/api/internal/app/status_webhook.go b/apps/api/internal/app/status_webhook.go new file mode 100644 index 0000000..562eec1 --- /dev/null +++ b/apps/api/internal/app/status_webhook.go @@ -0,0 +1,216 @@ +package app + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +const statusWebhookMaxAttempts = 10 + +type statusWebhookEnvelope struct { + ID string `json:"id"` + Type string `json:"type"` + CreatedAt string `json:"createdAt"` + Data any `json:"data"` +} + +func (a *App) enqueueStatusWebhook(ctx context.Context, db dbExecutor, eventKey, eventType, mailboxID string, data any) error { + if strings.TrimSpace(a.cfg.StatusWebhookURL) == "" { + return nil + } + now := a.now().UTC() + id := newID("whk") + payload := jsonEncode(statusWebhookEnvelope{ID: id, Type: eventType, CreatedAt: now.Format(time.RFC3339Nano), Data: data}) + _, err := db.ExecContext(ctx, `INSERT OR IGNORE INTO status_webhook_outbox(id,event_key,event_type,mailbox_id,payload_json,next_attempt_at,created_at,updated_at) + VALUES(?,?,?,?,?,?,?,?)`, id, eventKey, eventType, mailboxID, payload, now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano)) + return err +} + +func (a *App) statusWebhookWorker(ctx context.Context) { + if strings.TrimSpace(a.cfg.StatusWebhookURL) == "" { + return + } + a.log.Info("status webhook worker started") + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + for { + if err := a.processDueStatusWebhooks(ctx); err != nil && !errors.Is(err, context.Canceled) { + a.log.Warn("status webhook worker failed", "error", err) + } + select { + case <-ctx.Done(): + a.log.Info("status webhook worker stopped") + return + case <-ticker.C: + } + } +} + +func (a *App) processDueStatusWebhooks(ctx context.Context) error { + if strings.TrimSpace(a.cfg.StatusWebhookURL) == "" { + return nil + } + _, _ = a.db.ExecContext(ctx, `DELETE FROM status_webhook_outbox + WHERE updated_at=?)`, a.now().UTC().Add(-30*24*time.Hour).Format(time.RFC3339Nano), statusWebhookMaxAttempts) + rows, err := a.db.QueryContext(ctx, `SELECT id,payload_json,attempt_count FROM status_webhook_outbox + WHERE delivered_at IS NULL AND attempt_count= 300 { + return fmt.Errorf("status webhook returned %d", resp.StatusCode) + } + return nil +} + +func (a *App) validatedStatusWebhookURL(ctx context.Context) (*url.URL, error) { + if strings.TrimSpace(a.cfg.StatusWebhookSecret) == "" { + return nil, errors.New("LANQIN_STATUS_WEBHOOK_SECRET is required") + } + target, err := url.Parse(strings.TrimSpace(a.cfg.StatusWebhookURL)) + if err != nil || target.Hostname() == "" || target.User != nil || target.Fragment != "" { + return nil, errors.New("invalid status webhook URL") + } + if target.Scheme != "https" && !(a.cfg.StatusWebhookAllowPrivateHosts && target.Scheme == "http") { + return nil, errors.New("status webhook URL must use HTTPS") + } + if !a.cfg.StatusWebhookAllowPrivateHosts { + if err := validatePublicWebhookHost(ctx, target.Hostname()); err != nil { + return nil, err + } + } + return target, nil +} + +func (a *App) statusWebhookDialContext(ctx context.Context, network, address string) (net.Conn, error) { + host, port, err := net.SplitHostPort(address) + if err != nil { + return nil, err + } + if a.cfg.StatusWebhookAllowPrivateHosts { + return (&net.Dialer{Timeout: 5 * time.Second}).DialContext(ctx, network, address) + } + ips, err := net.DefaultResolver.LookupIP(ctx, "ip", host) + if err != nil { + return nil, err + } + for _, ip := range ips { + if !isPublicStatusWebhookIP(ip) { + return nil, errors.New("private or local status webhook hosts are not allowed") + } + } + dialer := &net.Dialer{Timeout: 5 * time.Second} + var lastErr error + for _, ip := range ips { + conn, err := dialer.DialContext(ctx, network, net.JoinHostPort(ip.String(), port)) + if err == nil { + return conn, nil + } + lastErr = err + } + if lastErr == nil { + lastErr = errors.New("status webhook host resolved without usable addresses") + } + return nil, lastErr +} + +func validatePublicWebhookHost(ctx context.Context, host string) error { + if strings.EqualFold(host, "localhost") { + return errors.New("localhost status webhook hosts are not allowed") + } + ips, err := net.DefaultResolver.LookupIP(ctx, "ip", host) + if err != nil { + return fmt.Errorf("failed to resolve status webhook host: %w", err) + } + for _, ip := range ips { + if !isPublicStatusWebhookIP(ip) { + return errors.New("private or local status webhook hosts are not allowed") + } + } + return nil +} + +func isPublicStatusWebhookIP(ip net.IP) bool { + if ip == nil { + return false + } + return ip.IsGlobalUnicast() && !ip.IsLoopback() && !ip.IsPrivate() && !ip.IsLinkLocalUnicast() && !ip.IsLinkLocalMulticast() && !ip.IsMulticast() && !ip.IsUnspecified() +} + +func truncateWebhookError(value string) string { + value = strings.TrimSpace(value) + if len(value) > 1000 { + return value[:1000] + } + return value +} diff --git a/apps/api/internal/app/types.go b/apps/api/internal/app/types.go index e21c4b2..a0cf9fb 100644 --- a/apps/api/internal/app/types.go +++ b/apps/api/internal/app/types.go @@ -29,10 +29,25 @@ type APIToken struct { LastUsedAt *time.Time `json:"lastUsedAt,omitempty"` ExpiresAt *time.Time `json:"expiresAt,omitempty"` Disabled bool `json:"disabled"` + Scopes []string `json:"scopes"` CreatedAt time.Time `json:"createdAt"` UpdatedAt time.Time `json:"updatedAt"` } +type DeliveryEvent struct { + ID string `json:"id"` + ExternalID string `json:"externalId"` + Provider string `json:"provider"` + QueueID string `json:"queueId"` + MessageID string `json:"messageId"` + RFCMessageID string `json:"rfcMessageId"` + Recipient string `json:"recipient"` + Status string `json:"status"` + Reason string `json:"reason,omitempty"` + OccurredAt time.Time `json:"occurredAt"` + CreatedAt time.Time `json:"createdAt"` +} + type Domain struct { ID string `json:"id"` Name string `json:"name"` diff --git a/apps/web/src/lib/api-types.ts b/apps/web/src/lib/api-types.ts index 1efcb8b..ea79935 100644 --- a/apps/web/src/lib/api-types.ts +++ b/apps/web/src/lib/api-types.ts @@ -51,7 +51,7 @@ export type PermissionLimits = { maxAttachmentMb: number; smtpDailyLimit: number export type PermissionGroupSummary = { id: string; name: string } export type PermissionGroup = { id: string; name: string; description: string; permissions: PermissionKey[]; limits: PermissionLimits; system: boolean; userCount: number; createdAt: string; updatedAt: string } export type User = { id: string; email: string; displayName: string; role: "admin" | "user"; disabled: boolean; protected: boolean; twoFactorEnabled: boolean; permissions: PermissionKey[]; limits: PermissionLimits; permissionGroupIds: string[]; permissionGroups: PermissionGroupSummary[]; createdAt: string } -export type APIToken = { id: string; name: string; lastUsedAt?: string; expiresAt?: string; disabled: boolean; createdAt: string; updatedAt: string } +export type APIToken = { id: string; name: string; lastUsedAt?: string; expiresAt?: string; disabled: boolean; scopes: string[]; createdAt: string; updatedAt: string } export type AdminUser = User & { mailboxCount: number; mailboxes?: string[] } export type AdminOverview = { users: number; activeUsers: number; domains: number; mailboxes: number; activeMailboxes: number; aliases: number; messages: number; unreadMessages: number; storageBytes: number } export type Domain = { id: string; name: string; status: string; dkimSelector: string; dkimPublicKey?: string; dnsStatus: string; dnsCheckedAt?: string; createdAt: string } diff --git a/apps/web/src/lib/api.ts b/apps/web/src/lib/api.ts index 8c964b8..0aaaa01 100644 --- a/apps/web/src/lib/api.ts +++ b/apps/web/src/lib/api.ts @@ -40,8 +40,8 @@ export const api = { updateProfile: (payload: { displayName: string }) => request<{ user: User }>("/api/me/profile", { method: "POST", body: JSON.stringify(payload) }), changePassword: (payload: { currentPassword: string; newPassword: string }) => request<{ ok: boolean }>("/api/me/password", { method: "POST", body: JSON.stringify(payload) }), apiTokens: () => request>("/api/me/api-tokens"), - createApiToken: (payload: { name: string; expiresAt?: string }) => request<{ token: string; item: APIToken }>("/api/me/api-tokens", { method: "POST", body: JSON.stringify(payload) }), - updateApiToken: (id: string, payload: { name?: string; expiresAt?: string; disabled?: boolean }) => request(`/api/me/api-tokens/${id}`, { method: "POST", body: JSON.stringify(payload) }), + createApiToken: (payload: { name: string; expiresAt?: string; scopes: string[] }) => request<{ token: string; item: APIToken }>("/api/me/api-tokens", { method: "POST", body: JSON.stringify(payload) }), + updateApiToken: (id: string, payload: { name?: string; expiresAt?: string; disabled?: boolean; scopes?: string[] }) => request(`/api/me/api-tokens/${id}`, { method: "POST", body: JSON.stringify(payload) }), deleteApiToken: (id: string) => request<{ ok: boolean }>(`/api/me/api-tokens/${id}`, { method: "DELETE" }), setupTwoFactor: () => request<{ secret: string; otpauthUrl: string }>("/api/me/2fa/setup", { method: "POST" }), enableTwoFactor: (code: string) => request<{ user: User }>("/api/me/2fa/enable", { method: "POST", body: JSON.stringify({ code }) }), diff --git a/apps/web/src/pages/profile.tsx b/apps/web/src/pages/profile.tsx index 25a3d18..8d54cda 100644 --- a/apps/web/src/pages/profile.tsx +++ b/apps/web/src/pages/profile.tsx @@ -148,12 +148,12 @@ export function ProfilePage() { onError: (error) => toast({ title: "关闭失败", description: error.message }), }) const createApiToken = useMutation({ - mutationFn: (payload: { name: string; expiresAt?: string }) => api.createApiToken(payload), + mutationFn: (payload: { name: string; expiresAt?: string; scopes: string[] }) => api.createApiToken(payload), onSuccess: (res) => { qc.invalidateQueries({ queryKey: ["api-tokens"] }); toast({ title: "API Token 已创建" }); return res }, onError: (error) => toast({ title: "创建失败", description: error.message }), }) const updateApiToken = useMutation({ - mutationFn: ({ id, payload }: { id: string; payload: { name?: string; expiresAt?: string; disabled?: boolean } }) => api.updateApiToken(id, payload), + mutationFn: ({ id, payload }: { id: string; payload: { name?: string; expiresAt?: string; disabled?: boolean; scopes?: string[] } }) => api.updateApiToken(id, payload), onSuccess: () => { qc.invalidateQueries({ queryKey: ["api-tokens"] }); toast({ title: "API Token 已更新" }) }, onError: (error) => toast({ title: "更新失败", description: error.message }), }) @@ -1005,9 +1005,18 @@ function ClientConfigRow({ label, value, security, onCopy }: { label: string; va ) } -function ApiTokensSection({ items, loading, pending, onCreate, onUpdate, onDelete, onCopy }: { items: APIToken[]; loading: boolean; pending: boolean; onCreate: (payload: { name: string; expiresAt?: string }) => Promise<{ token: string; item: APIToken }>; onUpdate: (id: string, payload: { name?: string; expiresAt?: string; disabled?: boolean }) => void; onDelete: (id: string) => void; onCopy: (text: string) => void }) { +const apiTokenScopeOptions = [ + ["messages:send", "发送邮件"], ["messages:read", "读取邮件与投递状态"], ["messages:manage", "重试或取消发送"], + ["domains:read", "查看域名"], ["domains:write", "管理域名"], ["mailboxes:read", "查看邮箱"], ["mailboxes:write", "管理邮箱"], + ["dns:read", "查看 DNS"], ["dns:check", "执行 DNS 检测"], ["aliases:read", "查看别名"], ["aliases:write", "管理别名"], +] as const + +function ApiTokensSection({ items, loading, pending, onCreate, onUpdate, onDelete, onCopy }: { items: APIToken[]; loading: boolean; pending: boolean; onCreate: (payload: { name: string; expiresAt?: string; scopes: string[] }) => Promise<{ token: string; item: APIToken }>; onUpdate: (id: string, payload: { name?: string; expiresAt?: string; disabled?: boolean; scopes?: string[] }) => void; onDelete: (id: string) => void; onCopy: (text: string) => void }) { const [createdToken, setCreatedToken] = React.useState("") const [pendingConfirm, setPendingConfirm] = React.useState(null) + const [scopes, setScopes] = React.useState(["messages:send", "messages:read"]) + const [editingToken, setEditingToken] = React.useState(null) + const [editingScopes, setEditingScopes] = React.useState([]) const defaultExpiresAt = React.useMemo(() => dateInputValue(new Date(Date.now() + 90 * 24 * 60 * 60 * 1000)), []) async function submit(event: React.FormEvent) { @@ -1016,9 +1025,10 @@ function ApiTokensSection({ items, loading, pending, onCreate, onUpdate, onDelet const form = new FormData(target) const expiresAt = dateInputToISOString(String(form.get("expiresAt") || "")) try { - const res = await onCreate({ name: String(form.get("name") || ""), expiresAt }) + const res = await onCreate({ name: String(form.get("name") || ""), expiresAt, scopes }) setCreatedToken(res.token) target.reset() + setScopes(["messages:send", "messages:read"]) } catch { // Mutation-level error handling already shows the toast. } @@ -1047,14 +1057,22 @@ function ApiTokensSection({ items, loading, pending, onCreate, onUpdate, onDelet )} -
- - + +
+ + + +
+ +
+ {apiTokenScopeOptions.map(([value, label]) => ( + + ))} +
- - - - @@ -1076,8 +1094,12 @@ function ApiTokensSection({ items, loading, pending, onCreate, onUpdate, onDelet 过期：{item.expiresAt ? formatDateTime(item.expiresAt) : "未设置"} 最后使用：{item.lastUsedAt ? formatDateTime(item.lastUsedAt) : "从未使用"} +
+ {(item.scopes || ["*"]).map((scope) => {scope})} +
+
@@ -1088,6 +1110,24 @@ function ApiTokensSection({ items, loading, pending, onCreate, onUpdate, onDelet + { if (!open) setEditingToken(null) }}> + + 编辑 Token 权限 +
+ {apiTokenScopeOptions.map(([value, label]) => ( + + ))} +
+ + + + + + + { if (!open) setPendingConfirm(null) }} onConfirm={() => pendingConfirm?.onConfirm()} /> ) diff --git a/deploy/.env.example b/deploy/.env.example index 28d8ac3..04d082a 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -97,6 +97,16 @@ LANQIN_SMTP_PASSWORD= # 外部 SMTP 要求 STARTTLS / TLS 时改 true；本机 Postfix 默认 false。 LANQIN_SMTP_REQUIRE_TLS=false +# 开放 API 最终投递事件回调的 HMAC-SHA256 密钥。生产环境请使用高强度随机值。 +# 未配置时 /api/open/v1/delivery-events 返回 503。 +LANQIN_DELIVERY_WEBHOOK_SECRET= + +# 可选：把发送队列与最终投递状态主动推送给外部系统。URL 默认必须为公网 HTTPS。 +LANQIN_STATUS_WEBHOOK_URL= +LANQIN_STATUS_WEBHOOK_SECRET= +# 仅可信内网或本地测试可开启；开启后也允许 HTTP 与私网目标。 +LANQIN_STATUS_WEBHOOK_ALLOW_PRIVATE_HOSTS=false + # 第三方客户端 SMTP 提交，由 LanQin API 监听 587/465；启用前必须配置可读 TLS 证书。 LANQIN_SUBMISSION_ADDR= LANQIN_SUBMISSION_TLS_ADDR= diff --git a/deploy/README.md b/deploy/README.md index 34c3abf..0c7311c 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -174,6 +174,24 @@ LANQIN_SMTP_PORT=25 LANQIN_SMTP_REQUIRE_TLS=false ``` +如需把上游服务商或 DSN 处理器的最终送达、退信、投诉、拒收事件写回开放 API，请设置： + +```env +LANQIN_DELIVERY_WEBHOOK_SECRET=replace-with-a-long-random-secret +``` + +回调地址、签名算法和事件格式见仓库中的 `docs/API.md` 与 `docs/openapi.json`。该接口未配置密钥时返回 `503`。 + +如需把状态变化主动推送到集成方，可额外设置： + +```env +LANQIN_STATUS_WEBHOOK_URL=https://integration.example.com/hooks/lanqin +LANQIN_STATUS_WEBHOOK_SECRET=replace-with-another-long-random-secret +LANQIN_STATUS_WEBHOOK_ALLOW_PRIVATE_HOSTS=false +``` + +事件先写入 SQLite outbox，再由后台 worker 投递；非 2xx 响应会按退避策略重试，最多 10 次。默认只允许公网 HTTPS，禁止重定向、URL 用户信息和私网/本机目标。只有可信内网或本地测试才应开启 `LANQIN_STATUS_WEBHOOK_ALLOW_PRIVATE_HOSTS`。 + Split stack 使用 `docker-compose.stack.yml` 时，API 容器默认会把 `LANQIN_SMTP_HOST` 覆盖为 `postfix`，让 Webmail 和 SMTP 提交都 relay 到 Postfix service。只有改用外部 SMTP 时才需要在 `.env` 明确填写 `LANQIN_STACK_SMTP_HOST` / `LANQIN_STACK_SMTP_PORT`。 如果发送队列里出现 relay 失败，通常是 Postfix 会话被中断或外部 SMTP 配置错误。优先检查： diff --git a/docs/API.md b/docs/API.md index 187b912..23fd787 100644 --- a/docs/API.md +++ b/docs/API.md @@ -1,8 +1,12 @@ # LanQin Email API -LanQin Email exposes integration-oriented APIs under `/api/open`. +LanQin Email exposes versioned integration APIs under `/api/open/v1`. The original `/api/open` paths remain compatibility aliases. -这些接口用于外部系统集成，统一放在 `/api/open` 下。它们不是匿名公开接口，只接受 API Token，不接受浏览器登录 Session Cookie。 +这些接口用于外部系统集成，稳定版本入口为 `/api/open/v1`。原 `/api/open` 路径继续作为兼容别名。它们不是匿名公开接口，只接受 API Token，不接受浏览器登录 Session Cookie。 + +Machine-readable OpenAPI 3.1 contract: [`docs/openapi.json`](./openapi.json). + +机器可读的 OpenAPI 3.1 契约见 [`docs/openapi.json`](./openapi.json)。 ## Base URL @@ -28,6 +32,7 @@ The API uses standard HTTP status codes: | `401 Unauthorized` | Missing or invalid API token / 缺少或无效的 API Token | | `403 Forbidden` | Token lacks required permissions / Token 缺少所需权限 | | `404 Not Found` | Resource does not exist / 资源不存在 | +| `409 Conflict` | Idempotency key conflict or concurrent status change / 幂等键冲突或状态并发变化 | | `429 Too Many Requests` | Rate limit exceeded / 超过频率限制 | | `500 Internal Server Error` | Server error / 服务器错误 | @@ -101,27 +106,32 @@ Tokens created without a custom expiration default to 90 days. You can disable o 如果没有自定义到期时间，Token 默认 90 天后过期。你可以在同一个个人中心页面中禁用或撤销 Token。 +Each token has independent scopes. Scopes only reduce the permissions of the owning user; they never grant permissions the user does not already have. Existing tokens created before scope support are migrated to `*` for compatibility. + +每个 Token 都有独立 scope。scope 只会收缩 Token 所属用户已有的权限，不会授予用户原本没有的权限。scope 功能上线前创建的 Token 会迁移为 `*`，以保持兼容。 + +| Scope | Purpose | +|---|---| +| `domains:read` / `domains:write` | View or manage sending domains | +| `mailboxes:read` / `mailboxes:write` | View or manage mailboxes; password reset is a write operation | +| `messages:read` / `messages:send` / `messages:manage` | Read messages/status, send, or retry/cancel | +| `aliases:read` / `aliases:write` | View or manage aliases | +| `dns:read` / `dns:check` | View required records or execute DNS checks | +| `*` | Compatibility wildcard; avoid for new integrations | + ## Permissions All Open API endpoints require an API token with appropriate permissions and role requirements: 所有 Open API 接口都需要具备相应权限和角色的 API Token： -| Endpoint 接口 | Required Permission 所需权限 | Required Role 所需角色 | -|----------|-------------------|---------------| -| `GET /api/open/domains` | any of `admin.domains.view`, `admin.dns.view`, `admin.mailboxes.view`, `admin.aliases.view`, `admin.settings.view`, `admin.templates.view` | admin | -| `POST /api/open/domains` | `admin.domains.create` | admin | -| `GET /api/open/domains/{id}` | any of `admin.domains.view`, `admin.dns.view`, `admin.mailboxes.view`, `admin.aliases.view`, `admin.settings.view`, `admin.templates.view` | admin | -| `POST /api/open/domains/{id}` | `admin.domains.update` | admin | -| `DELETE /api/open/domains/{id}` | `admin.domains.delete` | admin | -| `GET /api/open/mailboxes` | `admin.mailboxes.view` or `admin.messages.view` | admin | -| `POST /api/open/mailboxes` | `admin.mailboxes.create` | admin | -| `GET /api/open/mailboxes/{id}` | `admin.mailboxes.view` or `admin.messages.view` | admin | -| `POST /api/open/mailboxes/{id}` | `admin.mailboxes.update` | admin | -| `DELETE /api/open/mailboxes/{id}` | `admin.mailboxes.delete` | admin | -| `POST /api/open/send` | `mail.messages.send` | user or admin | -| `GET /api/open/send/{id}` | `mail.messages.read` | user or admin | -| `GET /api/open/mailboxes/{id}/messages` | `mail.messages.read` | user or admin | +| Endpoint group | Required scope | Role | +|---|---|---| +| Domains | `domains:read` or `domains:write` | admin | +| Mailboxes | `mailboxes:read` or `mailboxes:write` | admin | +| DNS | `dns:read` or `dns:check` | admin | +| Aliases | `aliases:read` or `aliases:write` | admin | +| Send / status / messages | `messages:send`, `messages:read`, or `messages:manage` | user or admin | **Notes:** - Admin endpoints check for `requireAdminAccess` (role must be `admin`). @@ -138,7 +148,7 @@ All Open API endpoints require an API token with appropriate permissions and rol ### List domains ```http -GET /api/open/domains +GET /api/open/v1/domains Authorization: Bearer lq_xxx ``` @@ -182,7 +192,7 @@ Authorization: Bearer lq_xxx ### Create domain ```http -POST /api/open/domains +POST /api/open/v1/domains Authorization: Bearer lq_xxx Content-Type: application/json @@ -221,7 +231,7 @@ Content-Type: application/json ### Get domain ```http -GET /api/open/domains/{id} +GET /api/open/v1/domains/{id} Authorization: Bearer lq_xxx ``` @@ -234,7 +244,7 @@ Authorization: Bearer lq_xxx ### Update domain status ```http -POST /api/open/domains/{id} +POST /api/open/v1/domains/{id} Authorization: Bearer lq_xxx Content-Type: application/json @@ -262,7 +272,7 @@ Content-Type: application/json ### Delete domain ```http -DELETE /api/open/domains/{id} +DELETE /api/open/v1/domains/{id} Authorization: Bearer lq_xxx ``` @@ -289,7 +299,7 @@ Authorization: Bearer lq_xxx ### List mailboxes ```http -GET /api/open/mailboxes +GET /api/open/v1/mailboxes Authorization: Bearer lq_xxx ``` @@ -323,7 +333,7 @@ Authorization: Bearer lq_xxx ### Create mailbox ```http -POST /api/open/mailboxes +POST /api/open/v1/mailboxes Authorization: Bearer lq_xxx Content-Type: application/json @@ -370,7 +380,7 @@ Content-Type: application/json ### Get mailbox ```http -GET /api/open/mailboxes/{id} +GET /api/open/v1/mailboxes/{id} Authorization: Bearer lq_xxx ``` @@ -383,7 +393,7 @@ Authorization: Bearer lq_xxx ### Update mailbox ```http -POST /api/open/mailboxes/{id} +POST /api/open/v1/mailboxes/{id} Authorization: Bearer lq_xxx Content-Type: application/json @@ -408,7 +418,7 @@ All fields are optional. Omitted (or empty / non-positive) fields keep their cur ### Delete mailbox ```http -DELETE /api/open/mailboxes/{id} +DELETE /api/open/v1/mailboxes/{id} Authorization: Bearer lq_xxx ``` @@ -433,8 +443,9 @@ Authorization: Bearer lq_xxx ## Send Mail ```http -POST /api/open/send +POST /api/open/v1/send Authorization: Bearer lq_xxx +Idempotency-Key: invoice-2026-0001 Content-Type: application/json { @@ -505,7 +516,7 @@ Total attachment size is limited by the sender's permission group (`maxAttachmen | Field 字段 | Description 说明 | |-------|-------------| -| `id` | Send identifier; use it with `GET /api/open/send/{id}` / 发信标识，可配合 `GET /api/open/send/{id}` 使用 | +| `id` | Send identifier; use it with `GET /api/open/v1/send/{id}` / 发信标识，可配合 `GET /api/open/v1/send/{id}` 使用 | | `queueId` | SMTP queue item id. Omitted when the message was only `accepted` / SMTP 队列项 ID；仅 `accepted` 时不返回 | | `status` | Delivery status, see values below / 投递状态，见下方取值 | | `messageId` | Internal stored message id / 内部存储的消息 ID | @@ -520,6 +531,10 @@ When SMTP delivery is not configured, the message can be stored as accepted with 如果没有配置 SMTP 投递，邮件可能只会进入 `accepted` 状态，不会产生 `queueId`。 +`id` is always the stable stored send id (`mail_*`). `queueId` is the queue item (`snd_*`) and may be absent. A repeated request with the same `Idempotency-Key` and identical body returns the original send with `200` and `Idempotency-Replayed: true`; reusing the key with a different body returns `409`. Keys are retained for 24 hours. + +`id` 始终是稳定的发送邮件 ID（`mail_*`）；`queueId` 是队列项 ID（`snd_*`），可能不存在。相同 `Idempotency-Key` 与相同请求体重试时返回原发送结果、状态码 `200`，并带 `Idempotency-Replayed: true`；相同 key 配不同请求体返回 `409`。key 保留 24 小时。 + Current status values: 当前状态取值： @@ -527,18 +542,22 @@ Current status values: - `accepted`: message was accepted and stored, but no SMTP queue item exists. - `queued`: queued for SMTP delivery. - `sending`: currently being delivered. -- `delivered`: SMTP delivery succeeded. +- `relayed`: the configured upstream SMTP server accepted the message; this is not final recipient delivery. - `failed`: delivery failed and may be retried. - `canceled`: delivery was canceled. +- `delivered`, `bounced`, `complained`, `rejected`, `deferred`: final per-recipient provider/DSN event. +- `partial`: final events currently differ between recipients or only cover part of the recipient list.
- `accepted`：邮件已被接受并存储，但没有 SMTP 队列项。 - `queued`：已进入 SMTP 投递队列。 - `sending`：正在投递中。 -- `delivered`：SMTP 投递成功。 +- `relayed`：配置的上游 SMTP 已接受邮件，但这不代表最终收件成功。 - `failed`：投递失败，可能会重试。 - `canceled`：投递已取消。 +- `delivered`、`bounced`、`complained`、`rejected`、`deferred`：每个收件人的最终供应商或 DSN 事件。 +- `partial`：不同收件人的最终状态不同，或当前只收到了部分收件人的事件。 **Error cases:** @@ -550,31 +569,31 @@ Current status values: | `429` | SMTP send rate limit exceeded / 超过 SMTP 发信频率限制 | | `507` | Mailbox quota exceeded / 邮箱配额已满 | -Bounce, complaint, rejection, and provider-specific delivery events require future webhook or delivery-event integration. +Final delivery events are exposed in `recipientStatuses` and through `GET /api/open/v1/send/{id}/events`. -退信、投诉、拒收等更细状态需要后续接入投递事件或 webhook 后才能完整提供。 +最终投递事件会出现在 `recipientStatuses`，完整时间线可通过 `GET /api/open/v1/send/{id}/events` 获取。 ## Send Status ```http -GET /api/open/send/{id} +GET /api/open/v1/send/{id} Authorization: Bearer lq_xxx ``` **Status:** `200 OK` or `404 Not Found` -`id` can be the value returned by `POST /api/open/send`. If a queue item exists, it can also be the queue id. +`id` can be the value returned by `POST /api/open/v1/send`. If a queue item exists, it can also be the queue id. `id` 可以使用发信接口返回的 `id`；如果存在队列项，也可以使用 `queueId`。 -**Response:** Same shape as the `POST /api/open/send` response. Only messages belonging to the token user's mailboxes are returned; otherwise `404`. +**Response:** Same shape as the `POST /api/open/v1/send` response. Only messages belonging to the token user's mailboxes are returned; otherwise `404`. -**响应：** 结构与 `POST /api/open/send` 的响应相同。只会返回属于 Token 拥有者邮箱的邮件，否则返回 `404`。 +**响应：** 结构与 `POST /api/open/v1/send` 的响应相同。只会返回属于 Token 拥有者邮箱的邮件，否则返回 `404`。 ## Received Messages ```http -GET /api/open/mailboxes/{id}/messages?folder=Inbox&limit=30&cursor=0&q=keyword +GET /api/open/v1/mailboxes/{id}/messages?folder=Inbox&limit=30&cursor=opaque&q=keyword Authorization: Bearer lq_xxx ``` @@ -586,7 +605,7 @@ Query parameters: - `folder`: folder name. Defaults to `Inbox`; use `all` for all folders. - `limit`: page size, defaults to `30`, maximum `100`. -- `cursor`: numeric offset. Pass back the `nextCursor` value from the previous response to fetch the next page. +- `cursor`: opaque stable cursor. Pass back `nextCursor` unchanged. Numeric offsets remain accepted for compatibility. - `q`: optional search keyword. Matches subject, from, to, snippet, and body text.
@@ -623,8 +642,44 @@ Response: `nextCursor` is empty when there are no more pages. Otherwise it contains the offset to pass as `cursor` for the next request. -当没有更多分页时，`nextCursor` 为空字符串；否则它是下次请求应作为 `cursor` 传入的偏移量。 +当没有更多分页时，`nextCursor` 为空字符串；否则应将它原样作为下一次请求的 `cursor` 传入。 -Users can only read messages from their own active mailboxes. +Users can only read messages from their own active mailboxes. Fetch message bodies and attachment metadata with `GET /api/open/v1/messages/{id}`; download an owned attachment with `GET /api/open/v1/attachments/{id}`. 用户只能读取自己拥有的 active 邮箱。 + +## Additional V1 Endpoints / 其他 V1 接口 + +- `GET /api/open/v1/send`: paginated send records. +- `GET /api/open/v1/send/{id}/events`: queue audit and final delivery events. +- `POST /api/open/v1/send/{id}/retry`: retry a failed queue item. +- `POST /api/open/v1/send/{id}/cancel`: cancel a queued or failed item. +- `POST /api/open/v1/mailboxes/{id}/password`: reset the owner user's password and all mailbox passwords owned by that user. +- `GET /api/open/v1/domains/{id}/dns-records` and `POST .../dns-check`: DNS configuration and check. +- `/api/open/v1/aliases`: alias CRUD. + +Domain names and mailbox addresses are immutable. Renaming them requires a storage/identity migration and is intentionally not exposed as a normal update operation. + +域名名称和邮箱地址不可直接修改。重命名需要迁移存储路径及身份信息，因此不作为普通更新操作开放。 + +## Delivery Event Webhook / 投递事件回调 + +Configure `LANQIN_DELIVERY_WEBHOOK_SECRET`, then post up to 100 events to `POST /api/open/v1/delivery-events`. This endpoint does not accept an API Token. Set the Unix timestamp in `X-LanQin-Timestamp`, compute `HMAC-SHA256(secret, timestamp + "." + rawBody)`, and send the lowercase hexadecimal digest as `X-LanQin-Signature: sha256=`. Timestamps outside five minutes are rejected. `(provider, event id)` is idempotent. + +配置 `LANQIN_DELIVERY_WEBHOOK_SECRET` 后，可向 `POST /api/open/v1/delivery-events` 一次提交最多 100 条事件。该接口不接受 API Token。将 Unix 时间戳放入 `X-LanQin-Timestamp`，计算 `HMAC-SHA256(secret, timestamp + "." + 原始请求体)`，再以 `X-LanQin-Signature: sha256=<小写十六进制>` 发送。超过五分钟的时间戳会被拒绝；`(provider, event id)` 具备幂等性。 + +Accepted event statuses: `delivered`, `bounced`, `complained`, `rejected`, `deferred`. Every event must identify an existing send using `queueId`, `messageId`, or `rfcMessageId`, and its recipient must belong to that send. + +## Outbound Status Webhook / 主动状态推送 + +Set `LANQIN_STATUS_WEBHOOK_URL` and `LANQIN_STATUS_WEBHOOK_SECRET` to receive status changes proactively. Events are persisted in a SQLite outbox before delivery. Non-2xx responses are retried with backoff up to 10 attempts. Delivered and retry-exhausted records are removed after 30 days. + +设置 `LANQIN_STATUS_WEBHOOK_URL` 和 `LANQIN_STATUS_WEBHOOK_SECRET` 后，可主动接收状态变化。事件会先持久化到 SQLite outbox，非 2xx 响应会按退避策略重试，最多 10 次；已送达和重试耗尽的记录会在 30 天后清理。 + +Outbound requests include `X-LanQin-Webhook-Id`, `X-LanQin-Timestamp`, and `X-LanQin-Signature`. Signature calculation is the same HMAC-SHA256 construction used by the inbound delivery-event endpoint: `HMAC(secret, timestamp + "." + rawBody)`. Event types include `send.accepted`, `send.queued`, `send.retry`, `send.delivered` (upstream SMTP accepted), `send.failed`, `send.canceled`, and `delivery.`. + +出站请求包含 `X-LanQin-Webhook-Id`、`X-LanQin-Timestamp` 和 `X-LanQin-Signature`。签名算法与入站投递事件相同：`HMAC(secret, timestamp + "." + 原始请求体)`。事件类型包括 `send.accepted`、`send.queued`、`send.retry`、`send.delivered`（上游 SMTP 接受）、`send.failed`、`send.canceled` 和 `delivery.<最终状态>`。 + +The target must be a public HTTPS URL by default. Redirects, URL credentials, loopback, private, link-local, and unspecified addresses are rejected. `LANQIN_STATUS_WEBHOOK_ALLOW_PRIVATE_HOSTS=true` relaxes this for explicitly trusted private deployments and also permits HTTP. + +目标地址默认必须是公网 HTTPS。重定向、URL 用户信息、loopback、私网、链路本地和未指定地址都会被拒绝。只有明确可信的私有部署才应设置 `LANQIN_STATUS_WEBHOOK_ALLOW_PRIVATE_HOSTS=true`；开启后也允许 HTTP。 diff --git a/docs/openapi.json b/docs/openapi.json new file mode 100644 index 0000000..0078644 --- /dev/null +++ b/docs/openapi.json @@ -0,0 +1,99 @@ +{ + "openapi": "3.1.0", + "info": { + "title": "LanQin Email Open API", + "version": "1.0.0", + "description": "Versioned integration API. The unversioned /api/open routes are compatibility aliases for /api/open/v1." + }, + "servers": [{ "url": "/api/open/v1" }], + "security": [{ "bearerAuth": [] }], + "paths": { + "/domains": { "get": { "parameters": [{ "$ref": "#/components/parameters/Limit" }, { "$ref": "#/components/parameters/Cursor" }], "responses": { "200": { "description": "Paginated domains" } } }, "post": { "responses": { "201": { "description": "Domain created" } } } }, + "/domains/{id}": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "get": { "responses": { "200": { "description": "Domain" }, "404": { "$ref": "#/components/responses/NotFound" } } }, "post": { "responses": { "200": { "description": "Domain updated" } } }, "delete": { "responses": { "200": { "description": "Domain deleted" }, "404": { "$ref": "#/components/responses/NotFound" } } } }, + "/domains/{id}/dns-records": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "get": { "responses": { "200": { "description": "Required DNS records" }, "404": { "$ref": "#/components/responses/NotFound" } } } }, + "/domains/{id}/dns-check": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "post": { "responses": { "200": { "description": "DNS check result" } } } }, + "/mailboxes": { "get": { "parameters": [{ "$ref": "#/components/parameters/Limit" }, { "$ref": "#/components/parameters/Cursor" }], "responses": { "200": { "description": "Paginated mailboxes" } } }, "post": { "responses": { "201": { "description": "Mailbox created" } } } }, + "/mailboxes/{id}": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "get": { "responses": { "200": { "description": "Mailbox" }, "404": { "$ref": "#/components/responses/NotFound" } } }, "post": { "responses": { "200": { "description": "Mailbox updated" } } }, "delete": { "responses": { "200": { "description": "Mailbox deleted" }, "404": { "$ref": "#/components/responses/NotFound" } } } }, + "/mailboxes/{id}/password": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "post": { "responses": { "200": { "description": "Owner and mailbox passwords reset" } } } }, + "/mailboxes/{id}/messages": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }, { "$ref": "#/components/parameters/Limit" }, { "$ref": "#/components/parameters/Cursor" }], "get": { "responses": { "200": { "description": "Paginated messages" } } } }, + "/messages/{id}": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "get": { "responses": { "200": { "description": "Message detail" }, "404": { "$ref": "#/components/responses/NotFound" } } } }, + "/attachments/{id}": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "get": { "responses": { "200": { "description": "Attachment bytes" }, "404": { "$ref": "#/components/responses/NotFound" } } } }, + "/send": { + "get": { "parameters": [{ "$ref": "#/components/parameters/Limit" }, { "$ref": "#/components/parameters/Cursor" }], "responses": { "200": { "description": "Paginated sends" } } }, + "post": { + "parameters": [{ "$ref": "#/components/parameters/IdempotencyKey" }], + "requestBody": { "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/SendRequest" } } } }, + "responses": { "200": { "description": "Idempotent replay", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/SendStatus" } } } }, "201": { "description": "Queued", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/SendStatus" } } } }, "409": { "$ref": "#/components/responses/Conflict" } } + } + }, + "/send/{id}": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "get": { "responses": { "200": { "description": "Send status", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/SendStatus" } } } }, "404": { "$ref": "#/components/responses/NotFound" } } } }, + "/send/{id}/events": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "get": { "responses": { "200": { "description": "Queue audit and delivery events" }, "404": { "$ref": "#/components/responses/NotFound" } } } }, + "/send/{id}/retry": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "post": { "responses": { "200": { "description": "Send requeued" }, "409": { "$ref": "#/components/responses/Conflict" } } } }, + "/send/{id}/cancel": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "post": { "responses": { "200": { "description": "Send canceled" }, "409": { "$ref": "#/components/responses/Conflict" } } } }, + "/aliases": { "get": { "parameters": [{ "$ref": "#/components/parameters/Limit" }, { "$ref": "#/components/parameters/Cursor" }], "responses": { "200": { "description": "Paginated aliases" } } }, "post": { "responses": { "201": { "description": "Alias created" } } } }, + "/aliases/{id}": { "parameters": [{ "$ref": "#/components/parameters/ResourceId" }], "get": { "responses": { "200": { "description": "Alias" }, "404": { "$ref": "#/components/responses/NotFound" } } }, "post": { "responses": { "200": { "description": "Alias updated" } } }, "delete": { "responses": { "200": { "description": "Alias deleted" }, "404": { "$ref": "#/components/responses/NotFound" } } } }, + "/delivery-events": { + "post": { + "security": [], + "description": "HMAC-SHA256 signed delivery event callback. Sign timestamp + '.' + raw body.", + "parameters": [ + { "name": "X-LanQin-Timestamp", "in": "header", "required": true, "schema": { "type": "string" } }, + { "name": "X-LanQin-Signature", "in": "header", "required": true, "schema": { "type": "string", "pattern": "^sha256=[a-f0-9]{64}$" } } + ], + "requestBody": { "required": true, "content": { "application/json": { "schema": { "type": "object", "required": ["events"], "properties": { "events": { "type": "array", "minItems": 1, "maxItems": 100, "items": { "$ref": "#/components/schemas/DeliveryEventInput" } } } } } } }, + "responses": { "200": { "description": "Events stored or deduplicated" }, "401": { "$ref": "#/components/responses/Unauthorized" } } + } + } + }, + "components": { + "securitySchemes": { "bearerAuth": { "type": "http", "scheme": "bearer", "bearerFormat": "lq_* API Token" } }, + "parameters": { + "IdempotencyKey": { "name": "Idempotency-Key", "in": "header", "required": false, "description": "Up to 128 characters; retained for 24 hours.", "schema": { "type": "string", "maxLength": 128 } }, + "ResourceId": { "name": "id", "in": "path", "required": true, "schema": { "type": "string" } }, + "Limit": { "name": "limit", "in": "query", "required": false, "schema": { "type": "integer", "minimum": 1, "maximum": 100 } }, + "Cursor": { "name": "cursor", "in": "query", "required": false, "schema": { "type": "string" } } + }, + "responses": { + "Unauthorized": { "description": "Missing or invalid authentication" }, + "NotFound": { "description": "Resource not found or not owned by the token user" }, + "Conflict": { "description": "Idempotency key conflict or concurrent state change" } + }, + "schemas": { + "SendRequest": { + "type": "object", "required": ["mailboxId"], + "properties": { + "mailboxId": { "type": "string" }, "from": { "type": "string" }, "fromName": { "type": "string" }, + "to": { "type": "array", "items": { "type": "string", "format": "email" } }, + "cc": { "type": "array", "items": { "type": "string", "format": "email" } }, + "bcc": { "type": "array", "items": { "type": "string", "format": "email" } }, + "subject": { "type": "string" }, "text": { "type": "string" }, "html": { "type": "string" }, + "attachments": { "type": "array", "items": { "$ref": "#/components/schemas/AttachmentInput" } } + } + }, + "AttachmentInput": { "type": "object", "required": ["filename", "contentBase64"], "properties": { "filename": { "type": "string" }, "contentType": { "type": "string" }, "contentBase64": { "type": "string", "contentEncoding": "base64" } } }, + "SendStatus": { + "type": "object", "required": ["id", "status", "messageId", "rfcMessageId", "mailboxId", "createdAt"], + "properties": { + "id": { "type": "string", "description": "Stable sent message id (mail_*)." }, + "queueId": { "type": "string", "description": "Internal queue id (snd_*)." }, + "status": { "type": "string", "enum": ["accepted", "queued", "sending", "relayed", "failed", "canceled", "delivered", "bounced", "complained", "rejected", "deferred", "partial"] }, + "queueStatus": { "type": "string", "enum": ["queued", "sending", "delivered", "failed", "canceled"] }, + "messageId": { "type": "string" }, "rfcMessageId": { "type": "string" }, "mailboxId": { "type": "string" }, + "mailboxAddress": { "type": "string" }, "subject": { "type": "string" }, "recipients": { "type": "array", "items": { "type": "string" } }, + "recipientStatuses": { "type": "array", "items": { "$ref": "#/components/schemas/RecipientStatus" } }, + "createdAt": { "type": "string", "format": "date-time" } + } + }, + "RecipientStatus": { "type": "object", "required": ["recipient", "status", "occurredAt"], "properties": { "recipient": { "type": "string" }, "status": { "type": "string" }, "reason": { "type": "string" }, "provider": { "type": "string" }, "occurredAt": { "type": "string", "format": "date-time" } } }, + "DeliveryEventInput": { + "type": "object", "required": ["id", "provider", "recipient", "status", "occurredAt"], + "properties": { + "id": { "type": "string" }, "provider": { "type": "string" }, "queueId": { "type": "string" }, "messageId": { "type": "string" }, "rfcMessageId": { "type": "string" }, + "recipient": { "type": "string", "format": "email" }, "status": { "type": "string", "enum": ["delivered", "bounced", "complained", "rejected", "deferred"] }, + "reason": { "type": "string" }, "occurredAt": { "type": "string", "format": "date-time" } + } + }, + "Error": { "type": "object", "required": ["error"], "properties": { "error": { "type": "string" } } } + } + } +}