From 2a831e9783d6d711cb093123281f4cfb0da8ec75 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 29 Aug 2026 09:50:52 -0700 Subject: [PATCH 1/3] feat: let a linked self-hosted instance sign Google and Microsoft mailboxes in through Warmbly Cloud's own OAuth apps and send with cloud-brokered access tokens: the cloud runs the consent (pool_link_mailboxes.managed, brokered state in Redis, the existing /addresses/*/callback completes it and redirects to the instance's /cloud-oauth/done), keeps the refresh grant, mints short-lived tokens at /pool-link/instance/mailboxes/:id/token and refuses them for revoked links, removed, inactive or blocked mailboxes; the instance mirrors such mailboxes without a credential (cloud_link_mailboxes.managed), ships them to the worker as brokered so goog/msgraph init on a token source that pulls from /api/v1/internal/cloud-link/token/:id, lets the consumer ignore cloud warmup tokens for enrolled mailboxes, and can adopt mailboxes connected directly on the workspace; Add account shows the cloud path and the adoptable list, and the Warmbly Cloud guide documents the model --- cmd/backend/main.go | 7 +- cmd/consumer/main.go | 1 + cmd/worker/main.go | 7 + docs/content/docs/api/endpoints.mdx | 2 +- docs/content/docs/guides/warmbly-cloud.mdx | 27 +- internal/api/handler/cloudlink.go | 109 +++++++ internal/api/handler/email_oauth_callback.go | 16 +- internal/api/handler/poollink.go | 92 ++++++ internal/api/routes.go | 13 + internal/app/cloudlink/managed.go | 205 +++++++++++++ internal/app/cloudlink/service.go | 44 ++- internal/app/consumer/event_new_email.go | 8 + internal/app/consumer/service.go | 4 +- internal/app/email/broker.go | 109 +++++++ internal/app/email/loader.go | 17 ++ internal/app/email/service.go | 14 + internal/app/poollink/oauth.go | 289 ++++++++++++++++++ internal/app/poollink/service.go | 28 +- internal/app/worker/mailmanager/add_email.go | 10 + internal/app/worker/mailmanager/manager.go | 5 + internal/app/worker/service.go | 6 + internal/app/worker/wmail/wmail.go | 15 +- internal/client/goog/goog.go | 5 + internal/client/msgraph/msgraph.go | 5 + .../000109_pool_link_managed.down.sql | 4 + .../000109_pool_link_managed.up.sql | 11 + internal/models/poollink.go | 54 ++++ internal/models/worker.go | 4 + internal/repository/http_brokered_token.go | 112 +++++++ internal/repository/pg_cloudlink.go | 22 +- internal/repository/pg_email.go | 22 ++ internal/repository/pg_poollink.go | 52 +++- web/src/app/app/emails/page.tsx | 15 +- .../settings/warmbly-cloud/MailboxTable.tsx | 16 +- .../app/settings/warmbly-cloud/providers.ts | 6 +- web/src/app/cloud-oauth/done/page.tsx | 64 ++++ .../components/app/cloud/CloudLinkCard.tsx | 5 +- .../components/app/emails/CloudWarmupCard.tsx | 17 +- .../components/app/modals/AddEmailModal.tsx | 187 +++++++++++- .../lib/api/client/app/cloudlink/cloudLink.ts | 23 ++ .../api/hooks/app/cloudlink/useCloudLink.ts | 17 ++ .../lib/api/models/app/cloudlink/CloudLink.ts | 19 ++ web/src/main.tsx | 6 + 43 files changed, 1614 insertions(+), 80 deletions(-) create mode 100644 internal/app/cloudlink/managed.go create mode 100644 internal/app/email/broker.go create mode 100644 internal/app/poollink/oauth.go create mode 100644 internal/infrastructure/db/migrations/000109_pool_link_managed.down.sql create mode 100644 internal/infrastructure/db/migrations/000109_pool_link_managed.up.sql create mode 100644 internal/repository/http_brokered_token.go create mode 100644 web/src/app/cloud-oauth/done/page.tsx diff --git a/cmd/backend/main.go b/cmd/backend/main.go index 9f806e91..d8276c1a 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -1180,7 +1180,12 @@ func main() { if g, ok := featureGateService.(interface{ WirePoolLink(feature.PoolLinkReader) }); ok { g.WirePoolLink(poolLinkService) } - cloudLinkService = cloudlink.NewService(repository.NewCloudLinkRepository(primaryDB.Pool, credEncrypter), emailRepostory) + if w, ok := poolLinkService.(poollink.CacheWirer); ok { + w.WireCache(cache) + } + cloudLinkRepository := repository.NewCloudLinkRepository(primaryDB.Pool, credEncrypter) + emailService.WireCloudLink(cloudLinkRepository) + cloudLinkService = cloudlink.NewService(cloudLinkRepository, emailRepostory, emailService) rateLimitRepository := repository.NewRateLimitRepository(primaryDB) rateLimitService = ratelimit.NewService(cache, rateLimitRepository) diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index abb0501a..b7082fe6 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -373,6 +373,7 @@ func main() { EmailAccountErrorRepository: emailAccountErrorRepo, WarmupRepo: warmupRepo, PoolLinkRepo: repository.NewPoolLinkRepository(primaryDB.Pool), + CloudLinkRepo: repository.NewCloudLinkRepository(primaryDB.Pool, credEncrypter), WarmupContentRepo: repository.NewWarmupContentRepository(primaryDB.Pool), WarmupEngagementRepo: repository.NewWarmupEngagementRepository(primaryDB.Pool), WarmupService: warmupService, diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 37e579d2..efc23a48 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -109,6 +109,12 @@ func main() { if err != nil { log.Fatal(err) } + // Mailboxes managed by Warmbly Cloud send with access tokens the backend + // brokers; the refresh grant never reaches the worker. + tokenBroker, err := repository.NewHTTPBrokeredTokenClient(internalBaseURL, internalToken) + if err != nil { + log.Fatal(err) + } // Blob storage (S3 by default, filesystem when BLOB_PROVIDER=filesystem). s3Client, err := storage.NewFromEnv(ctx, awscfg, "main") @@ -172,6 +178,7 @@ func main() { EmailMessageMapRepository: emailMessageMapRepo, SyncContextRepository: syncContextRepo, OauthInbox: &oauthInbox, + TokenBroker: tokenBroker, } if err := workerService.Init(); err != nil { diff --git a/docs/content/docs/api/endpoints.mdx b/docs/content/docs/api/endpoints.mdx index 944afed7..7acc0d80 100644 --- a/docs/content/docs/api/endpoints.mdx +++ b/docs/content/docs/api/endpoints.mdx @@ -298,7 +298,7 @@ These never accept an API key. They depend on a human-bound session: billing flo - All of `/organization/*` (create, switch, members, invitations, transfer ownership, avatar, danger zone) - `GET /website-tracking/settings`, `PATCH /website-tracking/settings`, `POST /website-tracking/settings/rotate-key` (the [website tracking](/guides/website-tracking/) snippet's consent mode, location precision, allowed hosts and retention; JWT permission `MANAGE_SETTINGS`. The rotate is bodyless and safe to repeat, each call issues a new key) - All of `/subscription/*` (checkout, portal, cancel, change-plan, preview-change, enterprise-inquiry, discounts, referrals, etc.) -- All of `/pool-link/*` and `/cloud-link/*` (the self-hosted warmup pool link: approving an instance's code, listing and unlinking instances, and on a self-hosted instance the connect flow and mailbox enrollment). `POST /pool-link/codes` and `POST /pool-link/poll` are public and per-IP rate limited: they are the device-code handshake an instance uses before it has a token, and `/pool-link/instance/*` accepts only an instance token +- All of `/pool-link/*` and `/cloud-link/*` (the self-hosted warmup pool link: approving an instance's code, listing and unlinking instances, and on a self-hosted instance the connect flow and mailbox enrollment). `POST /pool-link/codes` and `POST /pool-link/poll` are public and per-IP rate limited: they are the device-code handshake an instance uses before it has a token, and `/pool-link/instance/*` accepts only an instance token. `/pool-link/instance/oauth/*`, `/pool-link/instance/mailboxes/:id/token`, `/pool-link/instance/workspace-mailboxes` and `/pool-link/instance/mailboxes/adopt` are the cloud-managed mailbox surface (Google and Microsoft sign-in on Warmbly's OAuth apps, brokered access tokens); their instance-side counterparts are `/cloud-link/oauth/*` and `/cloud-link/workspace-mailboxes/*` - All of `/admin/*` ### Referrals and discounts diff --git a/docs/content/docs/guides/warmbly-cloud.mdx b/docs/content/docs/guides/warmbly-cloud.mdx index 919ad5c9..25b33b4e 100644 --- a/docs/content/docs/guides/warmbly-cloud.mdx +++ b/docs/content/docs/guides/warmbly-cloud.mdx @@ -5,7 +5,7 @@ description: "Warm the mailboxes on your own Warmbly instance in the hosted warm A self-hosted Warmbly instance warms mailboxes only against the other mailboxes on that same instance, which for most installs is a handful. Linking the instance to Warmbly Cloud lets those mailboxes warm in the hosted pool instead: thousands of real, monitored mailboxes exchanging natural-looking mail, with replies and inbox engagement handled for you. -Only warmup moves to the cloud. Campaigns, contacts, the unibox, tracking and every stored message stay on your server exactly as before. +Only warmup moves to the cloud. Campaigns, contacts, the unibox, tracking and every stored message stay on your server exactly as before. Linked instances also get Google and Microsoft sign-in through Warmbly's own OAuth apps, so there is no client to register: see [Google and Microsoft mailboxes](#google-and-microsoft-mailboxes-sign-in-through-warmbly-cloud). A free Warmbly Cloud workspace holds up to 10 mailboxes, connected directly or through a linked instance, and warms them at no cost. Unlimited mailboxes cost $15 a month per workspace. Nothing else about self-hosting changes. @@ -20,7 +20,7 @@ Your instance keeps a single link to a Warmbly Cloud workspace. When you enroll - your instance stops its own warmup for that mailbox and keeps sending campaigns from it as usual - health, daily volume and 7 day totals show up in your instance's settings as the cloud reports them -Unenrolling a mailbox, or disconnecting the instance, deletes the credential on the cloud immediately. +Unenrolling an SMTP/IMAP mailbox, or disconnecting the instance, deletes its credential on the cloud immediately. A mailbox that was signed in through Warmbly Cloud stays in the cloud workspace when you remove it from the instance; disconnecting the instance removes its mirrors of such mailboxes, since they cannot send without the link. ## Linking your instance @@ -32,11 +32,25 @@ The link is offered in three places on a self-hosted instance: as the last step The whole flow takes about a minute. A code expires after 15 minutes if nobody approves it; start again from the instance. -## Which mailboxes can be enrolled +## Google and Microsoft mailboxes: sign in through Warmbly Cloud -Mailboxes connected with **SMTP/IMAP** (including Google Workspace and Microsoft 365 through an app password) can be enrolled. +On a linked instance, **Add account > Gmail / Google Workspace** and **Outlook / Microsoft 365** no longer need an OAuth client of your own. The sign-in window opens on Warmbly's verified Google and Microsoft apps, and when you approve: -Mailboxes connected through **Sign in with Google** or **Sign in with Microsoft** cannot be enrolled yet. Their refresh grant is bound to the OAuth app that issued it, which on a self-hosted instance is your own, so the cloud could never refresh the token. Reconnect such a mailbox with SMTP/IMAP to enroll it. +- the mailbox is created in your Warmbly Cloud workspace, where its sign-in lives, and starts warming in the pool right away +- your instance gets a mirror of the mailbox with no credential on it. Campaigns, replies and the unibox work from your server as before; to send and sync, the instance asks the cloud for a short-lived access token (about an hour, cached) using its link token +- the refresh grant never leaves the cloud. Removing the mailbox from either side, unlinking the instance, or the cloud blocking the mailbox stops the instance sending from it within the hour + +These mailboxes show the blue **Cloud** badge with "Signed in through Warmbly Cloud". The `BOX_GOOGLE_*` and `BOX_OUTLOOK_*` variables are still honoured when set, but an instance without them can now connect Google and Microsoft mailboxes as long as it is linked. + +### Mailboxes connected on the cloud + +A Google or Microsoft mailbox you connect directly on app.warmbly.com shows up in your instance's **Add account** dialog under **In your Warmbly Cloud workspace**. Press **Connect** to add it to the instance the same way: the cloud keeps the sign-in, the instance sends with brokered tokens. A mailbox can be linked to one instance at a time. + +### Which existing mailboxes can be enrolled + +Mailboxes connected with **SMTP/IMAP** (including Google Workspace and Microsoft 365 through an app password) can be enrolled as they are: the instance sends the credential to the cloud and keeps its own copy. + +A Google or Microsoft mailbox that was signed in with your instance's own OAuth app cannot be enrolled, because that grant can only be refreshed by the app that issued it. Remove it and add it again through Warmbly Cloud to warm it there. ## What your instance shows @@ -67,6 +81,8 @@ Upgrade from the link card on your instance, or from **Settings > Billing** on W Enrolled mailboxes are ordinary members of the hosted pool and are held to the same rules: verification tokens on every warmup message, spam placement and complaint tracking, and automatic quarantine or blocking when a mailbox starts hurting partners. A mailbox that gets quarantined on the cloud shows that state on your instance, stops being offered to partners, and re-enters on the same probation terms as any hosted mailbox. +For mailboxes signed in through Warmbly Cloud the enforcement reaches sending too. Every access token the instance draws is checked on the cloud: a mailbox the cloud has **blocked**, one that is no longer active there, one that was removed from the workspace, or an instance whose link was revoked gets no token, and the instance stops sending and syncing from that mailbox as soon as its cached token expires. + ## Unlinking On your instance, **Settings > Warmbly Cloud > Disconnect** removes every enrolled mailbox from the pool, deletes their credentials on the cloud and lets local warmup take over. On Warmbly Cloud, **Settings > Linked instances** lists the instances linked to a workspace and can unlink any of them from the other side. @@ -75,6 +91,7 @@ On your instance, **Settings > Warmbly Cloud > Disconnect** removes every enroll - Traffic is outbound only: the instance calls `WARMBLY_CLOUD_URL` (default `https://api.warmbly.com`) over HTTPS. Nothing needs to reach the instance from the internet - The instance token is stored sealed with `CREDENTIALS_ENCRYPTION_KEY`. Rotating that key invalidates the stored token; reconnect afterwards +- Brokered access tokens reach the worker through the backend's internal API (`/api/v1/internal/cloud-link/token/:id`), so workers keep needing only `ENCRYPTED_KEYS_BACKEND_URL` and `ENCRYPTED_KEYS_WORKER_TOKEN`. A worker that cannot reach the backend keeps sending from such mailboxes until its cached token expires, about an hour - The link and the enrollment list are properties of the instance, not of a workspace, so they are not part of a [workspace export](/guides/workspace-export-import/). Reconnect after moving a workspace to a new instance - `INSTANCE_NAME` sets the name shown on the approval page; the hostname is used when unset diff --git a/internal/api/handler/cloudlink.go b/internal/api/handler/cloudlink.go index 72260450..b7e68fe4 100644 --- a/internal/api/handler/cloudlink.go +++ b/internal/api/handler/cloudlink.go @@ -172,3 +172,112 @@ func (h *Handler) cloudLinkLifecycle(c *gin.Context, action string, audit models h.auditOrg(c, audit, models.AuditEntityCloudLink, &id, nil, nil) c.JSON(http.StatusOK, row) } + +// Cloud-managed mailboxes: sign in through Warmbly Cloud, adopt workspace mailboxes. + +func (h *Handler) CloudLinkOAuthStart(c *gin.Context) { + if !h.cloudLinkReady(c) { + return + } + orgID := middleware.GetOrganizationID(c) + userID, err := middleware.GetUserUUID(c) + if orgID == nil || err != nil { + errx.JSON(c, errx.ErrNoOrganization) + return + } + var req struct { + Provider models.InboxProvider `json:"provider"` + } + if err := c.ShouldBindJSON(&req); err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid request body")) + return + } + res, xerr := h.CloudLinkService.StartOAuth(c.Request.Context(), *orgID, userID, req.Provider) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.JSON(http.StatusCreated, res) +} + +func (h *Handler) CloudLinkOAuthFinish(c *gin.Context) { + if !h.cloudLinkReady(c) { + return + } + orgID := middleware.GetOrganizationID(c) + userID, err := middleware.GetUserUUID(c) + if orgID == nil || err != nil { + errx.JSON(c, errx.ErrNoOrganization) + return + } + var req struct { + Session string `json:"session"` + } + if err := c.ShouldBindJSON(&req); err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid request body")) + return + } + acc, xerr := h.CloudLinkService.FinishOAuth(c.Request.Context(), *orgID, userID, req.Session) + if xerr != nil { + errx.JSON(c, xerr) + return + } + h.auditOrg(c, models.AuditActionConnect, models.AuditEntityEmailAccount, &acc.ID, nil, map[string]string{"email": acc.Email, "via": "warmbly_cloud"}) + c.JSON(http.StatusCreated, acc) +} + +func (h *Handler) CloudLinkWorkspaceMailboxes(c *gin.Context) { + if !h.cloudLinkReady(c) { + return + } + list, xerr := h.CloudLinkService.ListWorkspaceMailboxes(c.Request.Context()) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.JSON(http.StatusOK, gin.H{"data": list}) +} + +func (h *Handler) CloudLinkAdopt(c *gin.Context) { + if !h.cloudLinkReady(c) { + return + } + id, orgID, ok := cloudLinkAccountID(c) + if !ok { + return + } + userID, err := middleware.GetUserUUID(c) + if err != nil { + errx.JSON(c, errx.ErrUnauthorized) + return + } + acc, xerr := h.CloudLinkService.Adopt(c.Request.Context(), *orgID, userID, id) + if xerr != nil { + errx.JSON(c, xerr) + return + } + h.auditOrg(c, models.AuditActionConnect, models.AuditEntityEmailAccount, &acc.ID, nil, map[string]string{"email": acc.Email, "via": "warmbly_cloud"}) + c.JSON(http.StatusCreated, acc) +} + +// InternalCloudLinkToken is the worker's credential for a managed mailbox. +// +// GET /api/v1/internal/cloud-link/token/:id -> 200 {"access_token","expires_at"} | 4xx with the cloud's reason +func (h *Handler) InternalCloudLinkToken(c *gin.Context) { + if h.CloudLinkService == nil { + errx.JSON(c, errx.ErrNotFound) + return + } + id, err := uuid.Parse(c.Param("id")) + if err != nil { + errx.JSON(c, errx.ErrUuid) + return + } + tok, xerr := h.CloudLinkService.AccessToken(c.Request.Context(), id) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.Header("Cache-Control", "no-store") + c.JSON(http.StatusOK, tok) +} diff --git a/internal/api/handler/email_oauth_callback.go b/internal/api/handler/email_oauth_callback.go index cdc90264..7ad1eac8 100644 --- a/internal/api/handler/email_oauth_callback.go +++ b/internal/api/handler/email_oauth_callback.go @@ -8,6 +8,7 @@ import ( "strings" "github.com/gin-gonic/gin" + "github.com/warmbly/warmbly/internal/app/poollink" "github.com/warmbly/warmbly/internal/config" ) @@ -92,18 +93,27 @@ func callbackTargetOrigin() string { } func (h *Handler) EmailOAuthCallbackGmail(c *gin.Context) { - renderOAuthCallback(c, "gmail") + h.renderOAuthCallback(c, "gmail") } func (h *Handler) EmailOAuthCallbackOutlook(c *gin.Context) { - renderOAuthCallback(c, "outlook") + h.renderOAuthCallback(c, "outlook") } -func renderOAuthCallback(c *gin.Context, provider string) { +func (h *Handler) renderOAuthCallback(c *gin.Context, provider string) { code := c.Query("code") state := c.Query("state") providerErr := c.Query("error") + // A brokered consent (linked instance) completes here and lands back on + // the instance; there is no opener on our origin to post to. + if h.PoolLinkService != nil && strings.HasPrefix(state, poollink.BrokerStatePrefix) { + if to := h.PoolLinkService.CompleteOAuthCallback(c.Request.Context(), provider, code, state, providerErr); to != "" { + c.Redirect(http.StatusFound, to) + return + } + } + data := callbackData{ Provider: provider, Code: code, diff --git a/internal/api/handler/poollink.go b/internal/api/handler/poollink.go index daa11d80..e10b5d43 100644 --- a/internal/api/handler/poollink.go +++ b/internal/api/handler/poollink.go @@ -301,3 +301,95 @@ func (h *Handler) PoolLinkUnenroll(c *gin.Context) { } c.Status(http.StatusNoContent) } + +// Cloud-managed mailboxes for a linked instance. + +func (h *Handler) PoolLinkOAuthStart(c *gin.Context) { + inst := middleware.GetPoolLinkInstance(c) + if inst == nil { + errx.JSON(c, errx.ErrUnauthorized) + return + } + var req models.PoolLinkOAuthStartRequest + if err := c.ShouldBindJSON(&req); err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid request body")) + return + } + res, xerr := h.PoolLinkService.StartOAuth(c.Request.Context(), inst, req) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.JSON(http.StatusCreated, res) +} + +func (h *Handler) PoolLinkOAuthFinish(c *gin.Context) { + inst := middleware.GetPoolLinkInstance(c) + if inst == nil { + errx.JSON(c, errx.ErrUnauthorized) + return + } + var req models.PoolLinkOAuthFinishRequest + if err := c.ShouldBindJSON(&req); err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid request body")) + return + } + state, xerr := h.PoolLinkService.FinishOAuth(c.Request.Context(), inst, req.Session) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.JSON(http.StatusCreated, state) +} + +func (h *Handler) PoolLinkAccessToken(c *gin.Context) { + inst := middleware.GetPoolLinkInstance(c) + if inst == nil { + errx.JSON(c, errx.ErrUnauthorized) + return + } + remoteID, ok := poolLinkRemoteID(c) + if !ok { + return + } + tok, xerr := h.PoolLinkService.AccessToken(c.Request.Context(), inst, remoteID) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.Header("Cache-Control", "no-store") + c.JSON(http.StatusOK, tok) +} + +func (h *Handler) PoolLinkWorkspaceMailboxes(c *gin.Context) { + inst := middleware.GetPoolLinkInstance(c) + if inst == nil { + errx.JSON(c, errx.ErrUnauthorized) + return + } + list, xerr := h.PoolLinkService.ListWorkspaceMailboxes(c.Request.Context(), inst) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.JSON(http.StatusOK, list) +} + +func (h *Handler) PoolLinkAdopt(c *gin.Context) { + inst := middleware.GetPoolLinkInstance(c) + if inst == nil { + errx.JSON(c, errx.ErrUnauthorized) + return + } + var req models.PoolLinkAdoptRequest + if err := c.ShouldBindJSON(&req); err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid request body")) + return + } + state, xerr := h.PoolLinkService.Adopt(c.Request.Context(), inst, req) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.JSON(http.StatusCreated, state) +} diff --git a/internal/api/routes.go b/internal/api/routes.go index b807c301..884bcac0 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -136,6 +136,9 @@ func Run( // something the mailbox sent?" (tasks, message map, unibox threads). internal.GET("/sync/own-conversation", h.InternalSyncOwnConversation) + // Brokered credential for a mailbox managed by Warmbly Cloud. + internal.GET("/cloud-link/token/:id", h.InternalCloudLinkToken) + // Worker bootstrap config + heartbeat. Workers POST their identity // on boot (worker_id + bind_ip + tag) and pull their runtime config // instead of carrying it all in the install-time env file. @@ -1123,6 +1126,12 @@ func Run( poolLinkInstance.GET("/mailboxes/:remoteId", h.PoolLinkGetMailbox) poolLinkInstance.PATCH("/mailboxes/:remoteId", h.PoolLinkPatchMailbox) poolLinkInstance.DELETE("/mailboxes/:remoteId", h.PoolLinkUnenroll) + // Cloud-managed mailboxes: consent on this deployment's OAuth app, brokered tokens. + poolLinkInstance.POST("/oauth/start", h.PoolLinkOAuthStart) + poolLinkInstance.POST("/oauth/finish", h.PoolLinkOAuthFinish) + poolLinkInstance.GET("/mailboxes/:remoteId/token", h.PoolLinkAccessToken) + poolLinkInstance.GET("/workspace-mailboxes", h.PoolLinkWorkspaceMailboxes) + poolLinkInstance.POST("/mailboxes/adopt", h.PoolLinkAdopt) } // Self-hosted side: Settings > Warmbly Cloud. @@ -1140,6 +1149,10 @@ func Run( cloudLink.DELETE("/mailboxes/:id/enroll", m.RequirePermission(models.PermManageEmails), h.CloudLinkUnenroll) cloudLink.POST("/mailboxes/:id/pause", m.RequirePermission(models.PermManageEmails), h.CloudLinkPause) cloudLink.POST("/mailboxes/:id/resume", m.RequirePermission(models.PermManageEmails), h.CloudLinkResume) + cloudLink.POST("/oauth/start", m.RequirePermission(models.PermManageEmails), h.CloudLinkOAuthStart) + cloudLink.POST("/oauth/finish", m.RequirePermission(models.PermManageEmails), h.CloudLinkOAuthFinish) + cloudLink.GET("/workspace-mailboxes", h.CloudLinkWorkspaceMailboxes) + cloudLink.POST("/workspace-mailboxes/:id/adopt", m.RequirePermission(models.PermManageEmails), h.CloudLinkAdopt) } subscriptions := jwtOnly.Group("/subscription") diff --git a/internal/app/cloudlink/managed.go b/internal/app/cloudlink/managed.go new file mode 100644 index 00000000..639fdad8 --- /dev/null +++ b/internal/app/cloudlink/managed.go @@ -0,0 +1,205 @@ +package cloudlink + +import ( + "context" + "net/http" + "strings" + "time" + + "github.com/google/uuid" + "github.com/rs/zerolog/log" + "github.com/warmbly/warmbly/internal/config" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +// Cloud-managed mailboxes: Google and Microsoft sign-in runs on Warmbly +// Cloud's OAuth app, the grant stays there, and this instance sends with +// access tokens it draws from the cloud. + +var ( + ErrNotManaged = errx.NewWithIdentifier(errx.NotFound, "cloud_link_not_managed", "This mailbox is not managed by Warmbly Cloud.") + ErrOAuthSession = errx.NewWithIdentifier(errx.NotFound, "cloud_link_oauth_session", "That sign-in session is unknown or has expired. Start again.") +) + +// OAuthReturnPath is the dashboard route the cloud sends the popup back to. +const OAuthReturnPath = "/cloud-oauth/done" + +// tokenCacheMax bounds how long a brokered token is reused before the cloud is asked again. +const tokenCacheMax = 10 * time.Minute + +type oauthSession struct { + OrgID uuid.UUID + UserID uuid.UUID + Provider models.InboxProvider + ExpiresAt time.Time +} + +type cachedToken struct { + token *models.PoolLinkAccessToken + expires time.Time +} + +func (s *service) StartOAuth(ctx context.Context, orgID, userID uuid.UUID, provider models.InboxProvider) (*models.CloudLinkOAuthStart, *errx.Error) { + l, xerr := s.link(ctx) + if xerr != nil { + return nil, xerr + } + if provider != models.InboxProviderGoogle && provider != models.InboxProviderOutlook { + return nil, errx.ErrEmailOnboardProvider + } + var res models.PoolLinkOAuthStartResponse + req := models.PoolLinkOAuthStartRequest{Provider: provider, ReturnURL: strings.TrimRight(config.AppBaseURL(), "/") + OAuthReturnPath} + if xerr := s.clientFor(l).do(ctx, http.MethodPost, "/instance/oauth/start", req, &res); xerr != nil { + return nil, xerr + } + s.mu.Lock() + s.sessions[res.Session] = oauthSession{OrgID: orgID, UserID: userID, Provider: provider, ExpiresAt: time.Now().Add(15 * time.Minute)} + for k, v := range s.sessions { + if time.Now().After(v.ExpiresAt) { + delete(s.sessions, k) + } + } + s.mu.Unlock() + return &models.CloudLinkOAuthStart{URL: res.URL, Session: res.Session}, nil +} + +func (s *service) FinishOAuth(ctx context.Context, orgID, userID uuid.UUID, session string) (*models.Email, *errx.Error) { + s.mu.Lock() + sess, ok := s.sessions[session] + s.mu.Unlock() + if !ok || sess.OrgID != orgID || time.Now().After(sess.ExpiresAt) { + return nil, ErrOAuthSession + } + l, xerr := s.link(ctx) + if xerr != nil { + return nil, xerr + } + var state models.PoolLinkMailboxState + if xerr := s.clientFor(l).do(ctx, http.MethodPost, "/instance/oauth/finish", models.PoolLinkOAuthFinishRequest{Session: session}, &state); xerr != nil { + return nil, xerr + } + s.mu.Lock() + delete(s.sessions, session) + s.mu.Unlock() + return s.mirror(ctx, l, orgID, userID, &state) +} + +// mirror creates the local, credential-free copy of a cloud-managed mailbox. +func (s *service) mirror(ctx context.Context, l *models.CloudLink, orgID, userID uuid.UUID, state *models.PoolLinkMailboxState) (*models.Email, *errx.Error) { + name := strings.TrimSpace(state.Name) + if name == "" { + name = state.Email + } + acc, xerr := s.emails.NewManagedAccount(ctx, userID.String(), models.NewOauthAccount{ + OrganizationID: &orgID, + Provider: models.InboxProvider(state.Provider), + Name: name, + Email: state.Email, + }) + if xerr != nil { + // The cloud side exists without a local twin: release it so the next attempt is clean. + if rerr := s.clientFor(l).do(ctx, http.MethodDelete, "/instance/mailboxes/"+state.RemoteID.String(), nil, nil); rerr != nil { + log.Error().Str("remote_id", state.RemoteID.String()).Str("code", rerr.Identifier).Msg("cloud link: local mirror failed and the cloud link could not be released") + } + return nil, xerr + } + if _, err := s.repo.Enroll(ctx, acc.ID, state.RemoteID, true); err != nil { + if s.emailSvc != nil { + _ = s.emailSvc.Delete(ctx, userID.String(), acc.ID.String()) + } + return nil, errx.InternalError() + } + if s.emailSvc != nil { + if err := s.emailSvc.LoadAccountOntoWorker(ctx, acc.ID); err != nil { + log.Warn().Err(err).Str("account_id", acc.ID.String()).Msg("cloud link: worker load of managed mailbox failed; reconciler will retry") + } + } + return acc, nil +} + +func (s *service) ListWorkspaceMailboxes(ctx context.Context) ([]models.PoolLinkWorkspaceMailbox, *errx.Error) { + l, xerr := s.link(ctx) + if xerr != nil { + return nil, xerr + } + var list []models.PoolLinkWorkspaceMailbox + if xerr := s.clientFor(l).do(ctx, http.MethodGet, "/instance/workspace-mailboxes", nil, &list); xerr != nil { + return nil, xerr + } + if list == nil { + list = []models.PoolLinkWorkspaceMailbox{} + } + return list, nil +} + +func (s *service) Adopt(ctx context.Context, orgID, userID, cloudAccountID uuid.UUID) (*models.Email, *errx.Error) { + l, xerr := s.link(ctx) + if xerr != nil { + return nil, xerr + } + var state models.PoolLinkMailboxState + req := models.PoolLinkAdoptRequest{RemoteID: uuid.New(), EmailAccountID: cloudAccountID} + if xerr := s.clientFor(l).do(ctx, http.MethodPost, "/instance/mailboxes/adopt", req, &state); xerr != nil { + return nil, xerr + } + return s.mirror(ctx, l, orgID, userID, &state) +} + +// AccessToken is what the worker (through the backend) sends and syncs with. +// Cached until two minutes before expiry so a busy mailbox does not hammer the cloud. +func (s *service) AccessToken(ctx context.Context, accountID uuid.UUID) (*models.PoolLinkAccessToken, *errx.Error) { + s.mu.Lock() + if c, ok := s.tokens[accountID]; ok && time.Now().Before(c.expires) { + s.mu.Unlock() + return c.token, nil + } + s.mu.Unlock() + + m, err := s.repo.GetByAccount(ctx, accountID) + if err != nil { + return nil, errx.InternalError() + } + if m == nil || !m.Managed { + return nil, ErrNotManaged + } + l, xerr := s.link(ctx) + if xerr != nil { + return nil, xerr + } + var tok models.PoolLinkAccessToken + if xerr := s.clientFor(l).do(ctx, http.MethodGet, "/instance/mailboxes/"+m.RemoteID.String()+"/token", nil, &tok); xerr != nil { + return nil, xerr + } + // Cap the cache so a revocation or block on the cloud bites within minutes, not an hour. + until := tok.ExpiresAt.Add(-2 * time.Minute) + if cap := time.Now().Add(tokenCacheMax); until.After(cap) { + until = cap + } + s.mu.Lock() + s.tokens[accountID] = cachedToken{token: &tok, expires: until} + s.mu.Unlock() + return &tok, nil +} + +func (s *service) forgetToken(accountID uuid.UUID) { + s.mu.Lock() + delete(s.tokens, accountID) + s.mu.Unlock() +} + +// removeManaged deletes the local mirror; the cloud keeps the mailbox in the workspace. +func (s *service) removeManaged(ctx context.Context, userID string, m *models.CloudLinkMailbox) *errx.Error { + if l, err := s.repo.Get(ctx); err == nil && l != nil { + if xerr := s.clientFor(l).do(ctx, http.MethodDelete, "/instance/mailboxes/"+m.RemoteID.String(), nil, nil); xerr != nil && xerr.Identifier != "pool_link_mailbox_not_found" { + return xerr + } + } + s.forgetToken(m.EmailAccountID) + if s.emailSvc != nil { + if xerr := s.emailSvc.Delete(ctx, userID, m.EmailAccountID.String()); xerr != nil && xerr != errx.ErrNotFound { + return xerr + } + } + return nil +} diff --git a/internal/app/cloudlink/service.go b/internal/app/cloudlink/service.go index dd7f77ae..a6b3413d 100644 --- a/internal/app/cloudlink/service.go +++ b/internal/app/cloudlink/service.go @@ -11,6 +11,7 @@ import ( "github.com/google/uuid" "github.com/rs/zerolog/log" + "github.com/warmbly/warmbly/internal/app/email" "github.com/warmbly/warmbly/internal/config" "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/models" @@ -98,20 +99,31 @@ type Service interface { Unenroll(ctx context.Context, orgID, accountID uuid.UUID) *errx.Error SetLifecycle(ctx context.Context, orgID, accountID uuid.UUID, action string) (*models.CloudLinkMailboxRow, *errx.Error) + // Cloud-managed mailboxes: consent through the cloud, tokens brokered from it (managed.go). + StartOAuth(ctx context.Context, orgID, userID uuid.UUID, provider models.InboxProvider) (*models.CloudLinkOAuthStart, *errx.Error) + FinishOAuth(ctx context.Context, orgID, userID uuid.UUID, session string) (*models.Email, *errx.Error) + ListWorkspaceMailboxes(ctx context.Context) ([]models.PoolLinkWorkspaceMailbox, *errx.Error) + Adopt(ctx context.Context, orgID, userID, cloudAccountID uuid.UUID) (*models.Email, *errx.Error) + // AccessToken is the worker's credential for a managed mailbox, via the internal API. + AccessToken(ctx context.Context, accountID uuid.UUID) (*models.PoolLinkAccessToken, *errx.Error) + // IsEnrolled is the local warmup scheduler's stand-down check; fails closed to false. IsEnrolled(ctx context.Context, accountID uuid.UUID) bool } type service struct { - repo repository.CloudLinkRepository - emails repository.EmailRepository + repo repository.CloudLinkRepository + emails repository.EmailRepository + emailSvc email.EmailService - mu sync.Mutex - pending *PendingConnect + mu sync.Mutex + pending *PendingConnect + sessions map[string]oauthSession + tokens map[uuid.UUID]cachedToken } -func NewService(repo repository.CloudLinkRepository, emails repository.EmailRepository) Service { - return &service{repo: repo, emails: emails} +func NewService(repo repository.CloudLinkRepository, emails repository.EmailRepository, emailSvc email.EmailService) Service { + return &service{repo: repo, emails: emails, emailSvc: emailSvc, sessions: map[string]oauthSession{}, tokens: map[uuid.UUID]cachedToken{}} } func (s *service) link(ctx context.Context) (*models.CloudLink, *errx.Error) { @@ -254,6 +266,18 @@ func (s *service) Disconnect(ctx context.Context) *errx.Error { if xerr := s.clientFor(l).do(ctx, http.MethodDelete, "/instance", nil, nil); xerr != nil { log.Warn().Str("code", xerr.Identifier).Msg("cloud link: remote disconnect failed; clearing local link anyway") } + // Managed mirrors have no credential of their own; they end with the link. + if rows, err := s.repo.List(ctx); err == nil { + for _, m := range rows { + if !m.Managed || s.emailSvc == nil { + continue + } + if acc, xerr := s.emails.GetByID(ctx, m.EmailAccountID); xerr == nil { + s.forgetToken(m.EmailAccountID) + _ = s.emailSvc.Delete(ctx, acc.UserID, acc.ID.String()) + } + } + } if err := s.repo.UnenrollAll(ctx); err != nil { return errx.InternalError() } @@ -302,6 +326,7 @@ func (s *service) ListMailboxes(ctx context.Context, orgID uuid.UUID) ([]models. at := e.EnrolledAt row.Enrolled = true row.EnrolledAt = &at + row.Managed = e.Managed row.Cloud = cloudByRemote[e.RemoteID] } rows = append(rows, row) @@ -365,7 +390,7 @@ func (s *service) Enroll(ctx context.Context, orgID, accountID uuid.UUID) (*mode if xerr := s.clientFor(l).do(ctx, http.MethodPost, "/instance/mailboxes", req, &state); xerr != nil { return nil, xerr } - if _, err := s.repo.Enroll(ctx, acc.ID, acc.ID); err != nil { + if _, err := s.repo.Enroll(ctx, acc.ID, acc.ID, false); err != nil { // Without the local row the mailbox would warm in both places; undo the cloud side. if xerr := s.clientFor(l).do(ctx, http.MethodDelete, "/instance/mailboxes/"+acc.ID.String(), nil, nil); xerr != nil { log.Error().Str("account_id", acc.ID.String()).Str("code", xerr.Identifier).Msg("cloud link: local enrollment failed and the cloud copy could not be removed; unenroll it from Settings") @@ -390,6 +415,9 @@ func (s *service) Unenroll(ctx context.Context, orgID, accountID uuid.UUID) *err if m == nil { return nil } + if m.Managed { + return s.removeManaged(ctx, acc.UserID, m) + } // Local row first, so a failed cloud call can be retried from a consistent // state instead of leaving the mailbox with no warmup anywhere. if err := s.repo.Unenroll(ctx, accountID); err != nil { @@ -397,7 +425,7 @@ func (s *service) Unenroll(ctx context.Context, orgID, accountID uuid.UUID) *err } if l, err := s.repo.Get(ctx); err == nil && l != nil { if xerr := s.clientFor(l).do(ctx, http.MethodDelete, "/instance/mailboxes/"+m.RemoteID.String(), nil, nil); xerr != nil && xerr.Identifier != "pool_link_mailbox_not_found" { - if _, rerr := s.repo.Enroll(ctx, accountID, m.RemoteID); rerr != nil { + if _, rerr := s.repo.Enroll(ctx, accountID, m.RemoteID, false); rerr != nil { log.Error().Str("account_id", accountID.String()).Msg("cloud link: cloud unenroll failed and the local row could not be restored") } return xerr diff --git a/internal/app/consumer/event_new_email.go b/internal/app/consumer/event_new_email.go index b64d4cb3..60602622 100644 --- a/internal/app/consumer/event_new_email.go +++ b/internal/app/consumer/event_new_email.go @@ -31,6 +31,14 @@ func (s *JobsService) HandleNewEmail(ctx context.Context, e *models.JobEventNewE if warmupToken == "" { warmupToken = extractHeaderValue(e.Message, "X-Warmbly-Token") } + // Warmup mail for a mailbox Warmbly Cloud warms carries the cloud's tokens, + // which this instance cannot verify; counting them as forgeries would + // poison the mailbox's local score, and filing them would flood the inbox. + if warmupToken != "" && s.CloudLinkRepo != nil { + if enrolled, lerr := s.CloudLinkRepo.IsEnrolled(ctx, e.Message.EmailID); lerr == nil && enrolled { + return nil + } + } if warmupToken != "" { handled, err := s.handleWarmupEmail(ctx, e, warmupToken) if err != nil { diff --git a/internal/app/consumer/service.go b/internal/app/consumer/service.go index 3e5db574..9e596294 100644 --- a/internal/app/consumer/service.go +++ b/internal/app/consumer/service.go @@ -32,7 +32,9 @@ type JobsService struct { EmailAccountErrorRepository repository.EmailAccountErrorRepository WarmupRepo repository.WarmupRepository // PoolLinkRepo marks warmup-only mailboxes of linked instances; nil when unused. - PoolLinkRepo repository.PoolLinkRepository + PoolLinkRepo repository.PoolLinkRepository + // CloudLinkRepo (self-hosted) marks mailboxes the cloud warms, whose warmup mail is not ours to verify. + CloudLinkRepo repository.CloudLinkRepository WarmupContentRepo repository.WarmupContentRepository WarmupEngagementRepo repository.WarmupEngagementRepository WarmupService warmupapp.Service diff --git a/internal/app/email/broker.go b/internal/app/email/broker.go new file mode 100644 index 00000000..53fdb15d --- /dev/null +++ b/internal/app/email/broker.go @@ -0,0 +1,109 @@ +package email + +import ( + "context" + "strings" + "time" + + "github.com/google/uuid" + "github.com/rs/zerolog/log" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/infrastructure/pubsub" + "github.com/warmbly/warmbly/internal/models" + "golang.org/x/oauth2" +) + +// Brokered OAuth: the cloud runs a consent on its own OAuth app for a linked +// instance, keeps the grant, and mints access tokens on request. + +func (s *emailService) OAuthAuthorizeURL(provider models.InboxProvider, state string) (string, *errx.Error) { + cfg, xerr := s.oauthConfigFor(provider) + if xerr != nil { + return "", xerr + } + return cfg.AuthCodeURL(state, oauth2.AccessTypeOffline, oauth2.ApprovalForce), nil +} + +func (s *emailService) OAuthConnectWithCode(ctx context.Context, userID string, orgID *uuid.UUID, provider models.InboxProvider, code string) (*models.Email, *errx.Error) { + if code = strings.TrimSpace(code); code == "" { + return nil, errx.ErrEmailOnboardCode + } + if xerr := s.guardInboxLimit(ctx, orgID); xerr != nil { + return nil, xerr + } + cfg, xerr := s.oauthConfigFor(provider) + if xerr != nil { + return nil, xerr + } + tok, err := cfg.Exchange(ctx, code) + if err != nil { + return nil, errx.ErrEmailOnboardExchange + } + owner, xerr := fetchInboxOwner(ctx, provider, tok.AccessToken) + if xerr != nil { + return nil, xerr + } + if exists, xerr := s.emailRepository.ExistsForUser(ctx, userID, owner.Email); xerr != nil { + return nil, xerr + } else if exists { + return nil, errx.ErrEmailOnboardAlreadyExists + } + name := strings.TrimSpace(owner.Name) + if name == "" { + name = deriveNameFromEmail(owner.Email) + } + if xerr := s.guardMailboxThrottle(ctx, orgID); xerr != nil { + return nil, xerr + } + acc, xerr := s.emailRepository.NewOauthAccount(ctx, userID, models.NewOauthAccount{ + OrganizationID: orgID, + Provider: provider, + Name: name, + Email: owner.Email, + AccessToken: tok.AccessToken, + RefreshToken: tok.RefreshToken, + ExpiresAt: tok.Expiry, + }) + if xerr != nil { + return nil, xerr + } + s.syncWarmupPoolMembership(ctx, acc) + s.publishAccountEvent(ctx, pubsub.EventAccountConnected, acc) + s.dispatchAccountConnected(ctx, orgID, acc) + s.loadAccountBestEffort(ctx, acc.ID) + return acc, nil +} + +// OAuthAccessToken returns a live access token, refreshing and re-sealing the +// stored grant when it is within two minutes of expiry. +func (s *emailService) OAuthAccessToken(ctx context.Context, accountID uuid.UUID) (*oauth2.Token, *errx.Error) { + acc, xerr := s.emailRepository.GetByID(ctx, accountID) + if xerr != nil { + return nil, xerr + } + creds, xerr := s.emailRepository.GetOAuthCredentials(ctx, accountID) + if xerr != nil { + return nil, xerr + } + current := &oauth2.Token{AccessToken: creds.AccessToken, RefreshToken: creds.RefreshToken, Expiry: creds.ExpiresAt} + if current.AccessToken != "" && time.Until(current.Expiry) > 2*time.Minute { + return current, nil + } + // The client config is only needed to refresh. + cfg, xerr := s.oauthConfigFor(models.InboxProvider(acc.Provider)) + if xerr != nil { + return nil, xerr + } + fresh, err := cfg.TokenSource(ctx, current).Token() + if err != nil { + log.Warn().Err(err).Str("account_id", accountID.String()).Msg("brokered token refresh failed") + return nil, errx.ErrEmailCredentials + } + if fresh.RefreshToken == "" { + fresh.RefreshToken = current.RefreshToken + } + if err := s.emailRepository.RefreshBoxToken(ctx, accountID, fresh.AccessToken, fresh.RefreshToken, fresh.Expiry); err != nil { + log.Warn().Err(err).Str("account_id", accountID.String()).Msg("brokered token persist failed") + } + return fresh, nil +} diff --git a/internal/app/email/loader.go b/internal/app/email/loader.go index 07e9210a..03ba5ae4 100644 --- a/internal/app/email/loader.go +++ b/internal/app/email/loader.go @@ -273,6 +273,23 @@ func (s *emailService) buildAddWorkerEmail(ctx context.Context, acc *models.Emai SaveToSent: &saveToSent, } + // A managed mailbox has no local credential: the worker draws access + // tokens from the backend, which brokers them from the cloud. + if s.cloudLink != nil { + if m, err := s.cloudLink.GetByAccount(ctx, acc.ID); err == nil && m != nil && m.Managed { + out.Brokered = true + switch provider { + case models.InboxProviderGoogle: + out.Google = &models.AddWorkerEmailGoogleData{LastHistoryID: s.lastHistoryFor(ctx, userID, acc.ID, acc.LastID)} + case models.InboxProviderOutlook: + out.Graph = &models.AddWorkerEmailGraphData{DeltaLinks: s.deltaLinksFor(ctx, userID, acc.ID)} + default: + return nil, nil + } + return out, nil + } + } + switch provider { case models.InboxProviderGoogle: creds, cerr := s.emailRepository.GetOAuthCredentials(ctx, acc.ID) diff --git a/internal/app/email/service.go b/internal/app/email/service.go index ad3fa5f3..e06e8e58 100644 --- a/internal/app/email/service.go +++ b/internal/app/email/service.go @@ -3,6 +3,7 @@ package email import ( "context" "github.com/warmbly/warmbly/internal/app/instancesettings" + "golang.org/x/oauth2" "time" "github.com/google/uuid" @@ -81,6 +82,13 @@ type EmailService interface { WireMailboxes(repo repository.MailboxRepository) WireSyncBudget(src SyncBudgetSource) WirePoolLink(repo repository.PoolLinkRepository) + // WireCloudLink marks managed mailboxes, which ship to the worker without a credential. + WireCloudLink(repo repository.CloudLinkRepository) + // Brokered OAuth (cloud side): consent on this deployment's OAuth app for a linked instance. + OAuthAuthorizeURL(provider models.InboxProvider, state string) (string, *errx.Error) + OAuthConnectWithCode(ctx context.Context, userID string, orgID *uuid.UUID, provider models.InboxProvider, code string) (*models.Email, *errx.Error) + // OAuthAccessToken is a live access token for an OAuth mailbox, refreshed when near expiry. + OAuthAccessToken(ctx context.Context, accountID uuid.UUID) (*oauth2.Token, *errx.Error) // LoadAccountOntoWorker assigns a worker if needed and ships the mailbox // to it (idempotent; the reconciler calls it too). LoadAccountOntoWorker(ctx context.Context, accountID uuid.UUID) error @@ -113,6 +121,8 @@ type emailService struct { syncBudget SyncBudgetSource // poolLink marks linked warmup-only mailboxes, which sync with no history. poolLink repository.PoolLinkRepository + // cloudLink marks managed mailboxes whose credential the cloud holds. + cloudLink repository.CloudLinkRepository // webhookService is optional. When non-nil, account lifecycle events // (email_account.connected, email_account.removed) are dispatched to // subscribed customer webhooks. @@ -174,6 +184,10 @@ func (s *emailService) WireSyncBudget(src SyncBudgetSource) { // WirePoolLink attaches the pool-link repository so linked mailboxes get the // warmup-only sync policy. +func (s *emailService) WireCloudLink(repo repository.CloudLinkRepository) { + s.cloudLink = repo +} + func (s *emailService) WirePoolLink(repo repository.PoolLinkRepository) { s.poolLink = repo } diff --git a/internal/app/poollink/oauth.go b/internal/app/poollink/oauth.go new file mode 100644 index 00000000..0f2973f3 --- /dev/null +++ b/internal/app/poollink/oauth.go @@ -0,0 +1,289 @@ +package poollink + +import ( + "context" + "errors" + "net/url" + "strings" + "time" + + "github.com/google/uuid" + "github.com/redis/go-redis/v9" + "github.com/rs/zerolog/log" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/infrastructure/cache" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/pkg/crypt" +) + +// Cloud-managed mailboxes: the consent runs on this deployment's OAuth app, +// the grant stays here, and the instance sends with brokered access tokens. + +var ( + ErrOAuthReturnURL = errx.NewWithIdentifier(errx.BadRequest, "pool_link_return_url", "The return URL must be an absolute http(s) URL on the linked instance.") + ErrOAuthPending = errx.NewWithIdentifier(errx.Conflict, "pool_link_oauth_pending", "The sign-in has not completed yet.") + ErrOAuthSession = errx.NewWithIdentifier(errx.NotFound, "pool_link_oauth_session", "That sign-in session is unknown or has expired. Start again.") + ErrMailboxNotManaged = errx.NewWithIdentifier(errx.Forbidden, "pool_link_not_managed", "This mailbox's credential is held by the instance, not by Warmbly Cloud.") + ErrMailboxBlocked = errx.NewWithIdentifier(errx.Forbidden, "pool_link_mailbox_blocked", "Warmbly Cloud has blocked this mailbox for hurting the pool. Sending from it is suspended until it is reviewed.") + ErrMailboxInactive = errx.NewWithIdentifier(errx.Forbidden, "pool_link_mailbox_inactive", "This mailbox is not active on Warmbly Cloud. Reconnect it to keep sending.") + ErrAlreadyAdopted = errx.NewWithIdentifier(errx.Conflict, "pool_link_already_adopted", "That mailbox is already linked to an instance.") + ErrNotAdoptable = errx.NewWithIdentifier(errx.Unprocessable, "pool_link_not_adoptable", "Only active Google and Microsoft mailboxes in this workspace can be linked.") +) + +// BrokerStatePrefix marks a consent state as brokered so the callback can route it. +const BrokerStatePrefix = "pl_" + +const brokerTTL = 10 * time.Minute + +type brokerState struct { + InstanceID uuid.UUID `json:"instance_id"` + Provider string `json:"provider"` + ReturnURL string `json:"return_url"` + Session string `json:"session"` +} + +type brokerResult struct { + InstanceID uuid.UUID `json:"instance_id"` + RemoteID uuid.UUID `json:"remote_id"` + Pending bool `json:"pending"` + ErrorCode string `json:"error_code,omitempty"` + ErrorText string `json:"error_text,omitempty"` +} + +// CacheWirer is how main attaches Redis, which holds consent state for the round trip. +type CacheWirer interface{ WireCache(*cache.Cache) } + +func (s *service) WireCache(c *cache.Cache) { s.cache = c } + +func brokerStateKey(state string) string { return "poollink:oauth:state:" + state } +func brokerSessionKey(session string) string { return "poollink:oauth:session:" + session } + +// returnURLAllowed: absolute http(s), and on the instance's own host when one is known. +func returnURLAllowed(raw, instanceURL string) bool { + u, err := url.Parse(raw) + if err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Host == "" { + return false + } + if iu, err := url.Parse(instanceURL); err == nil && iu.Host != "" && !strings.EqualFold(iu.Host, u.Host) { + return false + } + return true +} + +func (s *service) StartOAuth(ctx context.Context, inst *models.PoolLinkInstance, req models.PoolLinkOAuthStartRequest) (*models.PoolLinkOAuthStartResponse, *errx.Error) { + if s.cache == nil { + return nil, errx.InternalError() + } + if req.Provider != models.InboxProviderGoogle && req.Provider != models.InboxProviderOutlook { + return nil, errx.ErrEmailOnboardProvider + } + if !returnURLAllowed(req.ReturnURL, inst.URL) { + return nil, ErrOAuthReturnURL + } + plan, xerr := s.Plan(ctx, inst.OrganizationID) + if xerr != nil { + return nil, xerr + } + if plan.MailboxLimit != nil && plan.Enrolled >= *plan.MailboxLimit { + return nil, ErrMailboxLimit + } + nonce, err := crypt.Nonce() + if err != nil { + return nil, errx.InternalError() + } + session, err := crypt.Nonce() + if err != nil { + return nil, errx.InternalError() + } + state := BrokerStatePrefix + nonce + authURL, xerr := s.emailSvc.OAuthAuthorizeURL(req.Provider, state) + if xerr != nil { + return nil, xerr + } + st := brokerState{InstanceID: inst.ID, Provider: string(req.Provider), ReturnURL: req.ReturnURL, Session: session} + if err := s.cache.SetJSON(ctx, brokerStateKey(state), st, brokerTTL); err != nil { + return nil, errx.InternalError() + } + if err := s.cache.SetJSON(ctx, brokerSessionKey(session), brokerResult{InstanceID: inst.ID, Pending: true}, brokerTTL); err != nil { + return nil, errx.InternalError() + } + return &models.PoolLinkOAuthStartResponse{URL: authURL, Session: session}, nil +} + +// CompleteOAuthCallback finishes a brokered consent server-side and returns +// where to send the browser. Never errors: every outcome lands on the instance. +func (s *service) CompleteOAuthCallback(ctx context.Context, provider, code, state, providerErr string) string { + if s.cache == nil { + return "" + } + var st brokerState + if err := s.cache.GetJSON(ctx, brokerStateKey(state), &st); err != nil { + return "" + } + _ = s.cache.Del(ctx, brokerStateKey(state)).Err() + if st.Provider != provider { + return "" + } + res := brokerResult{InstanceID: st.InstanceID} + if providerErr != "" { + res.ErrorCode, res.ErrorText = providerErr, "The provider did not complete the sign-in." + } else if code == "" { + res.ErrorCode, res.ErrorText = "missing_code", "The provider returned no authorization code." + } else if remoteID, xerr := s.connectBrokered(ctx, st, code); xerr != nil { + res.ErrorCode, res.ErrorText = xerr.Identifier, xerr.Message + if res.ErrorCode == "" { + res.ErrorCode = "pool_link_oauth_failed" + } + } else { + res.RemoteID = remoteID + } + if err := s.cache.SetJSON(ctx, brokerSessionKey(st.Session), res, brokerTTL); err != nil { + log.Error().Err(err).Msg("pool link: could not store brokered consent result") + } + q := url.Values{"session": {st.Session}} + if res.ErrorCode != "" { + q.Set("status", "error") + q.Set("error", res.ErrorCode) + q.Set("message", res.ErrorText) + } else { + q.Set("status", "ok") + } + sep := "?" + if strings.Contains(st.ReturnURL, "?") { + sep = "&" + } + return st.ReturnURL + sep + q.Encode() +} + +func (s *service) connectBrokered(ctx context.Context, st brokerState, code string) (uuid.UUID, *errx.Error) { + inst, err := s.repo.GetInstance(ctx, st.InstanceID) + if err != nil { + return uuid.Nil, errx.InternalError() + } + if inst == nil || inst.RevokedAt != nil { + return uuid.Nil, ErrInstanceRevoked + } + userID, xerr := s.ownerUserID(ctx, inst) + if xerr != nil { + return uuid.Nil, xerr + } + orgID := inst.OrganizationID + acc, xerr := s.emailSvc.OAuthConnectWithCode(ctx, userID, &orgID, models.InboxProvider(st.Provider), code) + if xerr != nil { + return uuid.Nil, xerr + } + remoteID := uuid.New() + if err := s.repo.EnrollMailbox(ctx, &models.PoolLinkMailbox{InstanceID: inst.ID, RemoteID: remoteID, EmailAccountID: acc.ID, Managed: true}); err != nil { + _ = s.emailSvc.Delete(ctx, userID, acc.ID.String()) + return uuid.Nil, errx.InternalError() + } + s.startWarmup(ctx, userID, acc.ID) + return remoteID, nil +} + +// startWarmup brings a freshly linked mailbox into the pool; failures are retried by the reconciler. +func (s *service) startWarmup(ctx context.Context, userID string, accountID uuid.UUID) { + if _, xerr := s.emailSvc.SetWarmupLifecycle(ctx, userID, accountID.String(), "start"); xerr != nil { + log.Warn().Str("account_id", accountID.String()).Msg("pool link: warmup start failed after enrollment") + } + if err := s.emailSvc.LoadAccountOntoWorker(ctx, accountID); err != nil { + log.Warn().Err(err).Str("account_id", accountID.String()).Msg("pool link: worker load failed; reconciler will retry") + } + if s.scheduler != nil { + _ = s.scheduler.EnsureWarmupScheduled(ctx, accountID) + } +} + +func (s *service) FinishOAuth(ctx context.Context, inst *models.PoolLinkInstance, session string) (*models.PoolLinkMailboxState, *errx.Error) { + if s.cache == nil || strings.TrimSpace(session) == "" { + return nil, ErrOAuthSession + } + var res brokerResult + if err := s.cache.GetJSON(ctx, brokerSessionKey(session), &res); err != nil { + if errors.Is(err, redis.Nil) { + return nil, ErrOAuthSession + } + return nil, errx.InternalError() + } + if res.InstanceID != inst.ID { + return nil, ErrOAuthSession + } + if res.Pending { + return nil, ErrOAuthPending + } + _ = s.cache.Del(ctx, brokerSessionKey(session)).Err() + if res.ErrorCode != "" { + return nil, errx.NewWithIdentifier(errx.BadRequest, res.ErrorCode, res.ErrorText) + } + return s.GetMailbox(ctx, inst, res.RemoteID) +} + +// AccessToken mints a short-lived provider token for a managed mailbox. This +// is the enforcement point: a revoked link, a removed or inactive mailbox, or +// a hard-blocked one gets no token and the instance stops sending from it. +func (s *service) AccessToken(ctx context.Context, inst *models.PoolLinkInstance, remoteID uuid.UUID) (*models.PoolLinkAccessToken, *errx.Error) { + m, err := s.repo.GetMailboxByRemote(ctx, inst.ID, remoteID) + if err != nil { + return nil, errx.InternalError() + } + if m == nil { + return nil, ErrMailboxNotFound + } + if !m.Managed { + return nil, ErrMailboxNotManaged + } + acc, xerr := s.emails.GetByID(ctx, m.EmailAccountID) + if xerr != nil { + return nil, xerr + } + if acc.Status != "active" { + return nil, ErrMailboxInactive + } + if s.analytics != nil { + if status, xerr := s.analytics.GetAccountStatus(ctx, inst.OrganizationID, acc.ID); xerr == nil && status != nil && status.WarmupHealth != nil && status.WarmupHealth.State == "blocked" { + return nil, ErrMailboxBlocked + } + } + tok, xerr := s.emailSvc.OAuthAccessToken(ctx, acc.ID) + if xerr != nil { + return nil, xerr + } + _ = s.repo.TouchMailboxToken(ctx, inst.ID, remoteID) + return &models.PoolLinkAccessToken{AccessToken: tok.AccessToken, ExpiresAt: tok.Expiry, Provider: acc.Provider, Email: acc.Email}, nil +} + +func (s *service) ListWorkspaceMailboxes(ctx context.Context, inst *models.PoolLinkInstance) ([]models.PoolLinkWorkspaceMailbox, *errx.Error) { + list, err := s.repo.ListAdoptableMailboxes(ctx, inst.OrganizationID) + if err != nil { + return nil, errx.InternalError() + } + return list, nil +} + +// Adopt links a mailbox that was connected directly on the workspace. +func (s *service) Adopt(ctx context.Context, inst *models.PoolLinkInstance, req models.PoolLinkAdoptRequest) (*models.PoolLinkMailboxState, *errx.Error) { + if req.RemoteID == uuid.Nil || req.EmailAccountID == uuid.Nil { + return nil, ErrBadRequest + } + acc, xerr := s.emails.GetByID(ctx, req.EmailAccountID) + if xerr != nil { + return nil, xerr + } + if acc.OrganizationID == nil || *acc.OrganizationID != inst.OrganizationID || acc.Status != "active" || + (acc.Provider != string(models.InboxProviderGoogle) && acc.Provider != string(models.InboxProviderOutlook)) { + return nil, ErrNotAdoptable + } + if existing, err := s.repo.GetMailboxByAccount(ctx, acc.ID); err != nil { + return nil, errx.InternalError() + } else if existing != nil { + return nil, ErrAlreadyAdopted + } + if err := s.repo.EnrollMailbox(ctx, &models.PoolLinkMailbox{InstanceID: inst.ID, RemoteID: req.RemoteID, EmailAccountID: acc.ID, Managed: true}); err != nil { + return nil, errx.InternalError() + } + userID, xerr := s.ownerUserID(ctx, inst) + if xerr == nil { + s.startWarmup(ctx, userID, acc.ID) + } + return s.GetMailbox(ctx, inst, req.RemoteID) +} diff --git a/internal/app/poollink/service.go b/internal/app/poollink/service.go index 2803cb65..a5ac2a0c 100644 --- a/internal/app/poollink/service.go +++ b/internal/app/poollink/service.go @@ -19,6 +19,7 @@ import ( "github.com/warmbly/warmbly/internal/app/organization" "github.com/warmbly/warmbly/internal/config" "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/infrastructure/cache" "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/repository" ) @@ -64,6 +65,15 @@ type Service interface { PatchMailbox(ctx context.Context, inst *models.PoolLinkInstance, remoteID uuid.UUID, patch models.PoolLinkMailboxPatch) (*models.PoolLinkMailboxState, *errx.Error) Unenroll(ctx context.Context, inst *models.PoolLinkInstance, remoteID uuid.UUID) *errx.Error + // Cloud-managed mailboxes for linked instances (oauth.go). + StartOAuth(ctx context.Context, inst *models.PoolLinkInstance, req models.PoolLinkOAuthStartRequest) (*models.PoolLinkOAuthStartResponse, *errx.Error) + // CompleteOAuthCallback finishes a brokered consent; returns "" when the state is not brokered. + CompleteOAuthCallback(ctx context.Context, provider, code, state, providerErr string) string + FinishOAuth(ctx context.Context, inst *models.PoolLinkInstance, session string) (*models.PoolLinkMailboxState, *errx.Error) + AccessToken(ctx context.Context, inst *models.PoolLinkInstance, remoteID uuid.UUID) (*models.PoolLinkAccessToken, *errx.Error) + ListWorkspaceMailboxes(ctx context.Context, inst *models.PoolLinkInstance) ([]models.PoolLinkWorkspaceMailbox, *errx.Error) + Adopt(ctx context.Context, inst *models.PoolLinkInstance, req models.PoolLinkAdoptRequest) (*models.PoolLinkMailboxState, *errx.Error) + // IsLinkedMailbox is the consumer's hot-path warmup-only check. IsLinkedMailbox(ctx context.Context, accountID uuid.UUID) bool // HasActiveLink entitles a workspace to warm its linked mailboxes. @@ -82,6 +92,7 @@ type service struct { orgs organization.OrganizationService scheduler WarmupScheduler planRepo repository.PlanRepository + cache *cache.Cache } func NewService( @@ -539,9 +550,11 @@ func (s *service) state(ctx context.Context, inst *models.PoolLinkInstance, m *m RemoteID: m.RemoteID, EmailAccountID: acc.ID, Email: acc.Email, + Name: acc.Name, Provider: acc.Provider, Status: acc.Status, EnrolledAt: m.EnrolledAt, + Managed: m.Managed, AuthState: acc.AuthState, Settings: models.PoolLinkWarmupSettings{ Base: acc.WarmupBase, Max: acc.WarmupMax, Increase: acc.WarmupIncrease, ReplyRate: acc.WarmupReplyRate, @@ -629,12 +642,15 @@ func (s *service) Unenroll(ctx context.Context, inst *models.PoolLinkInstance, r if m == nil { return ErrMailboxNotFound } - userID, xerr := s.ownerUserID(ctx, inst) - if xerr != nil { - return xerr - } - if xerr := s.emailSvc.Delete(ctx, userID, m.EmailAccountID.String()); xerr != nil && xerr != errx.ErrNotFound { - return xerr + // A managed mailbox belongs to the workspace; only the link goes. + if !m.Managed { + userID, xerr := s.ownerUserID(ctx, inst) + if xerr != nil { + return xerr + } + if xerr := s.emailSvc.Delete(ctx, userID, m.EmailAccountID.String()); xerr != nil && xerr != errx.ErrNotFound { + return xerr + } } if err := s.repo.DeleteMailbox(ctx, inst.ID, remoteID); err != nil { return errx.InternalError() diff --git a/internal/app/worker/mailmanager/add_email.go b/internal/app/worker/mailmanager/add_email.go index ffe3da75..14db25d4 100644 --- a/internal/app/worker/mailmanager/add_email.go +++ b/internal/app/worker/mailmanager/add_email.go @@ -2,11 +2,15 @@ package mailmanager import ( "context" + "errors" "github.com/warmbly/warmbly/internal/app/worker/wmail" "github.com/warmbly/warmbly/internal/models" ) +// errBrokerUnavailable: a managed mailbox needs the backend's token broker. +var errBrokerUnavailable = errors.New("cloud-managed mailbox needs ENCRYPTED_KEYS_BACKEND_URL and ENCRYPTED_KEYS_WORKER_TOKEN for brokered tokens") + func (m *MailManager) AddWMail( ctx context.Context, data *models.AddWorkerEmail, @@ -17,6 +21,12 @@ func (m *MailManager) AddWMail( // Cfg is avro-excluded from the payload, so rebuild it from the worker's // local oauth config for token refresh (no-op for smtp_imap). data.Cfg = m.cfgFor(data.Type) + if data.Brokered { + if m.tokenBroker == nil { + return errBrokerUnavailable + } + data.TokenSource = m.tokenBroker.Source(data.ID) + } newMail, err := wmail.NewWMail( data, diff --git a/internal/app/worker/mailmanager/manager.go b/internal/app/worker/mailmanager/manager.go index ff74cf2b..19676744 100644 --- a/internal/app/worker/mailmanager/manager.go +++ b/internal/app/worker/mailmanager/manager.go @@ -24,8 +24,13 @@ type MailManager struct { syncContextRepository repository.SyncContextRepository cipherService cipher.CipherService oauthInbox *config.Oauth2Inbox + // tokenBroker serves mailboxes whose credential Warmbly Cloud holds; nil disables them. + tokenBroker repository.BrokeredTokenClient } +// WireTokenBroker enables cloud-managed mailboxes. +func (m *MailManager) WireTokenBroker(b repository.BrokeredTokenClient) { m.tokenBroker = b } + func NewMailManager( onEvent func(eventType models.JobEventType, key string, body any) error, cache *cache.Cache, diff --git a/internal/app/worker/service.go b/internal/app/worker/service.go index 993447cf..bd1f1a10 100644 --- a/internal/app/worker/service.go +++ b/internal/app/worker/service.go @@ -34,6 +34,9 @@ type WorkerService struct { // endpoint) the worker needs to refresh delegated tokens locally. Cfg is not // serialized in the AddWorkerEmail payload, so the worker rebuilds it here. OauthInbox *config.Oauth2Inbox + // TokenBroker serves mailboxes managed by Warmbly Cloud (brokered access + // tokens over the internal API). Optional: nil refuses such mailboxes. + TokenBroker repository.BrokeredTokenClient mailManager *mailmanager.MailManager @@ -66,6 +69,9 @@ func (s *WorkerService) Init() error { s.CipherService, s.OauthInbox, ) + if s.TokenBroker != nil { + s.mailManager.WireTokenBroker(s.TokenBroker) + } return nil } diff --git a/internal/app/worker/wmail/wmail.go b/internal/app/worker/wmail/wmail.go index 4cfba7eb..1bf74469 100644 --- a/internal/app/worker/wmail/wmail.go +++ b/internal/app/worker/wmail/wmail.go @@ -180,7 +180,13 @@ func NewWMail( LastHistoryID: data.Google.LastHistoryID, } - if err := mail.GoogleData.Client.Init(mailCtx, data.Google.Token, data.Cfg); err != nil { + if data.TokenSource != nil { + // Brokered: the cloud refreshes; there is nothing to persist here. + mail.GoogleData.Client.OnTokenRefresh = nil + if err := mail.GoogleData.Client.InitWithSource(mailCtx, data.TokenSource); err != nil { + return nil, err + } + } else if err := mail.GoogleData.Client.Init(mailCtx, data.Google.Token, data.Cfg); err != nil { return nil, err } case models.InboxProviderOutlook: @@ -215,7 +221,12 @@ func NewWMail( }, } - if err := mail.GraphData.Client.Init(mailCtx, token, data.Cfg); err != nil { + if data.TokenSource != nil { + mail.GraphData.Client.OnTokenRefresh = nil + if err := mail.GraphData.Client.InitWithSource(mailCtx, data.TokenSource); err != nil { + return nil, err + } + } else if err := mail.GraphData.Client.Init(mailCtx, token, data.Cfg); err != nil { return nil, err } case models.InboxProviderSMTPIMAP: diff --git a/internal/client/goog/goog.go b/internal/client/goog/goog.go index 14e05ad2..1f2c6457 100644 --- a/internal/client/goog/goog.go +++ b/internal/client/goog/goog.go @@ -41,7 +41,12 @@ func (c *Client) Init(ctx context.Context, token *oauth2.Token, cfg oauth2.Confi return c.OnTokenRefresh(context.Background(), token) }) } + return c.InitWithSource(ctx, ts) +} +// InitWithSource builds the client on a caller-owned token source (brokered +// tokens from Warmbly Cloud); nothing is persisted from it. +func (c *Client) InitWithSource(ctx context.Context, ts oauth2.TokenSource) *errx.MailError { httpClient := oauth2.NewClient(ctx, ts) var err error c.srv, err = gmail.NewService(ctx, option.WithHTTPClient(httpClient)) diff --git a/internal/client/msgraph/msgraph.go b/internal/client/msgraph/msgraph.go index d73aceeb..62cf81a3 100644 --- a/internal/client/msgraph/msgraph.go +++ b/internal/client/msgraph/msgraph.go @@ -78,7 +78,12 @@ func (c *Client) Init(ctx context.Context, token *oauth2.Token, cfg oauth2.Confi return c.OnTokenRefresh(context.Background(), t) }) } + return c.InitWithSource(ctx, ts) +} +// InitWithSource builds the client on a caller-owned token source (brokered +// tokens from Warmbly Cloud); nothing is persisted from it. +func (c *Client) InitWithSource(ctx context.Context, ts oauth2.TokenSource) *errx.MailError { c.hc = oauth2.NewClient(ctx, ts) if c.DeltaLinks == nil { c.DeltaLinks = map[string]string{} diff --git a/internal/infrastructure/db/migrations/000109_pool_link_managed.down.sql b/internal/infrastructure/db/migrations/000109_pool_link_managed.down.sql new file mode 100644 index 00000000..d08f3297 --- /dev/null +++ b/internal/infrastructure/db/migrations/000109_pool_link_managed.down.sql @@ -0,0 +1,4 @@ +ALTER TABLE cloud_link_mailboxes DROP COLUMN IF EXISTS managed; + +ALTER TABLE pool_link_mailboxes DROP COLUMN IF EXISTS last_token_at; +ALTER TABLE pool_link_mailboxes DROP COLUMN IF EXISTS managed; diff --git a/internal/infrastructure/db/migrations/000109_pool_link_managed.up.sql b/internal/infrastructure/db/migrations/000109_pool_link_managed.up.sql new file mode 100644 index 00000000..0da3a705 --- /dev/null +++ b/internal/infrastructure/db/migrations/000109_pool_link_managed.up.sql @@ -0,0 +1,11 @@ +-- Cloud-managed mailboxes for linked instances. +-- +-- managed = true means the cloud holds the mailbox's only credential (the +-- OAuth grant came through Warmbly's app) and the instance sends with +-- short-lived access tokens it fetches from the cloud. Unenrolling such a +-- mailbox removes the link, not the workspace mailbox. + +ALTER TABLE pool_link_mailboxes ADD COLUMN IF NOT EXISTS managed boolean NOT NULL DEFAULT false; +ALTER TABLE pool_link_mailboxes ADD COLUMN IF NOT EXISTS last_token_at timestamptz; + +ALTER TABLE cloud_link_mailboxes ADD COLUMN IF NOT EXISTS managed boolean NOT NULL DEFAULT false; diff --git a/internal/models/poollink.go b/internal/models/poollink.go index cc8ec6e1..337cec43 100644 --- a/internal/models/poollink.go +++ b/internal/models/poollink.go @@ -54,6 +54,9 @@ type PoolLinkMailbox struct { RemoteID uuid.UUID `json:"remote_id"` EmailAccountID uuid.UUID `json:"email_account_id"` EnrolledAt time.Time `json:"enrolled_at"` + // Managed: the cloud holds the only credential and the instance sends with brokered tokens. + Managed bool `json:"managed"` + LastTokenAt *time.Time `json:"last_token_at,omitempty"` } // PoolLinkStartRequest is what a self-hosted instance sends to begin linking. @@ -138,14 +141,56 @@ type PoolLinkEnrollRequest struct { Warmup PoolLinkWarmupSettings `json:"warmup"` } +// PoolLinkOAuthStartRequest asks the cloud for a Google or Microsoft consent URL on Warmbly's app. +type PoolLinkOAuthStartRequest struct { + Provider InboxProvider `json:"provider"` + ReturnURL string `json:"return_url"` +} + +// PoolLinkOAuthStartResponse: open URL in the browser; redeem Session once the popup returns. +type PoolLinkOAuthStartResponse struct { + URL string `json:"url"` + Session string `json:"session"` +} + +// PoolLinkOAuthFinishRequest redeems a completed consent for its mailbox. +type PoolLinkOAuthFinishRequest struct { + Session string `json:"session"` +} + +// PoolLinkAdoptRequest links a mailbox already in the workspace to the instance. +type PoolLinkAdoptRequest struct { + RemoteID uuid.UUID `json:"remote_id"` + EmailAccountID uuid.UUID `json:"email_account_id"` +} + +// PoolLinkAccessToken is a short-lived provider token minted for a managed mailbox; never a refresh token. +type PoolLinkAccessToken struct { + AccessToken string `json:"access_token"` + ExpiresAt time.Time `json:"expires_at"` + Provider string `json:"provider"` + Email string `json:"email"` +} + +// PoolLinkWorkspaceMailbox is a workspace mailbox an instance may adopt. +type PoolLinkWorkspaceMailbox struct { + ID uuid.UUID `json:"id"` + Email string `json:"email"` + Name string `json:"name"` + Provider string `json:"provider"` + Status string `json:"status"` +} + // PoolLinkMailboxState is the per-mailbox view shown in both dashboards. type PoolLinkMailboxState struct { RemoteID uuid.UUID `json:"remote_id"` EmailAccountID uuid.UUID `json:"email_account_id"` Email string `json:"email"` + Name string `json:"name"` Provider string `json:"provider"` Status string `json:"status"` EnrolledAt time.Time `json:"enrolled_at"` + Managed bool `json:"managed"` Warmup *WarmupStatusInfo `json:"warmup,omitempty"` Health *WarmupHealthInfo `json:"health,omitempty"` SentToday int `json:"sent_today"` @@ -183,6 +228,14 @@ type CloudLinkMailbox struct { EmailAccountID uuid.UUID `json:"email_account_id"` RemoteID uuid.UUID `json:"remote_id"` EnrolledAt time.Time `json:"enrolled_at"` + // Managed: no local credential; the worker sends with tokens brokered by the cloud. + Managed bool `json:"managed"` +} + +// CloudLinkOAuthStart is the instance dashboard's handle on a cloud-brokered consent. +type CloudLinkOAuthStart struct { + URL string `json:"url"` + Session string `json:"session"` } // CloudLinkStatus is the self-hosted dashboard's view of the link. @@ -206,5 +259,6 @@ type CloudLinkMailboxRow struct { Status string `json:"status"` Enrolled bool `json:"enrolled"` EnrolledAt *time.Time `json:"enrolled_at,omitempty"` + Managed bool `json:"managed"` Cloud *PoolLinkMailboxState `json:"cloud,omitempty"` } diff --git a/internal/models/worker.go b/internal/models/worker.go index bb2fc0dc..f89ceb69 100644 --- a/internal/models/worker.go +++ b/internal/models/worker.go @@ -231,8 +231,12 @@ type AddWorkerEmail struct { // Sync is the fair-use budget and resume state. Nil only from a publisher // older than the sync policy; the worker then applies compiled defaults. Sync *AddWorkerEmailSyncData `json:"sync" avro:"sync"` + // Brokered: no credential travels; the worker fetches access tokens from the backend. + Brokered bool `json:"brokered" avro:"brokered"` Cfg oauth2.Config `json:"-" avro:"-"` + // TokenSource is set by the worker for brokered mailboxes. + TokenSource oauth2.TokenSource `json:"-" avro:"-"` } // SavesSentCopy reports whether the worker should APPEND a copy of each sent diff --git a/internal/repository/http_brokered_token.go b/internal/repository/http_brokered_token.go new file mode 100644 index 00000000..14673f77 --- /dev/null +++ b/internal/repository/http_brokered_token.go @@ -0,0 +1,112 @@ +package repository + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/google/uuid" + "golang.org/x/oauth2" +) + +// BrokeredTokenClient is the worker's way to a credential for a mailbox that +// Warmbly Cloud manages: the backend brokers a short-lived access token, the +// refresh grant never reaches this process. +type BrokeredTokenClient interface { + Token(ctx context.Context, accountID uuid.UUID) (*oauth2.Token, error) + // Source adapts a mailbox to oauth2; the returned source caches until expiry. + Source(accountID uuid.UUID) oauth2.TokenSource +} + +type httpBrokeredTokenClient struct { + baseURL string + token string + client *http.Client +} + +// NewHTTPBrokeredTokenClient returns the proxy for GET {BaseURL}/api/v1/internal/cloud-link/token/:id. +func NewHTTPBrokeredTokenClient(baseURL, token string) (BrokeredTokenClient, error) { + if baseURL == "" { + return nil, errors.New("brokered_token.http: baseURL is required") + } + if token == "" { + return nil, errors.New("brokered_token.http: token is required") + } + return &httpBrokeredTokenClient{ + baseURL: strings.TrimRight(baseURL, "/"), + token: token, + client: &http.Client{Timeout: 25 * time.Second}, + }, nil +} + +// BrokeredTokenRefused is a definitive no from the cloud (revoked, blocked, +// removed): callers should treat it as an authentication failure, not a blip. +type BrokeredTokenRefused struct { + Code string + Message string +} + +func (e *BrokeredTokenRefused) Error() string { + return fmt.Sprintf("brokered token refused (%s): %s", e.Code, e.Message) +} + +func (r *httpBrokeredTokenClient) Token(ctx context.Context, accountID uuid.UUID) (*oauth2.Token, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.baseURL+"/api/v1/internal/cloud-link/token/"+accountID.String(), nil) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "Bearer "+r.token) + req.Header.Set("User-Agent", "warmbly-worker/brokered-token-http") + resp, err := r.client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10)) + if resp.StatusCode != http.StatusOK { + var env struct { + Code string `json:"code"` + Message string `json:"message"` + Error string `json:"error"` + } + _ = json.Unmarshal(body, &env) + if env.Message == "" { + env.Message = env.Error + } + if resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusUnauthorized { + return nil, &BrokeredTokenRefused{Code: env.Code, Message: env.Message} + } + return nil, fmt.Errorf("brokered_token.http: status %d: %s", resp.StatusCode, env.Message) + } + var out struct { + AccessToken string `json:"access_token"` + ExpiresAt time.Time `json:"expires_at"` + } + if err := json.Unmarshal(body, &out); err != nil { + return nil, err + } + if out.AccessToken == "" { + return nil, errors.New("brokered_token.http: empty access token") + } + return &oauth2.Token{AccessToken: out.AccessToken, TokenType: "Bearer", Expiry: out.ExpiresAt}, nil +} + +func (r *httpBrokeredTokenClient) Source(accountID uuid.UUID) oauth2.TokenSource { + return oauth2.ReuseTokenSource(nil, brokeredSource{client: r, accountID: accountID}) +} + +type brokeredSource struct { + client *httpBrokeredTokenClient + accountID uuid.UUID +} + +func (b brokeredSource) Token() (*oauth2.Token, error) { + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second) + defer cancel() + return b.client.Token(ctx, b.accountID) +} diff --git a/internal/repository/pg_cloudlink.go b/internal/repository/pg_cloudlink.go index f041fb1b..d0181730 100644 --- a/internal/repository/pg_cloudlink.go +++ b/internal/repository/pg_cloudlink.go @@ -20,7 +20,7 @@ type CloudLinkRepository interface { Delete(ctx context.Context) error SetSyncResult(ctx context.Context, at time.Time, lastError string) error - Enroll(ctx context.Context, accountID, remoteID uuid.UUID) (*models.CloudLinkMailbox, error) + Enroll(ctx context.Context, accountID, remoteID uuid.UUID, managed bool) (*models.CloudLinkMailbox, error) Unenroll(ctx context.Context, accountID uuid.UUID) error UnenrollAll(ctx context.Context) error GetByAccount(ctx context.Context, accountID uuid.UUID) (*models.CloudLinkMailbox, error) @@ -101,15 +101,15 @@ func (r *cloudLinkRepository) SetSyncResult(ctx context.Context, at time.Time, l return err } -func (r *cloudLinkRepository) Enroll(ctx context.Context, accountID, remoteID uuid.UUID) (*models.CloudLinkMailbox, error) { +func (r *cloudLinkRepository) Enroll(ctx context.Context, accountID, remoteID uuid.UUID, managed bool) (*models.CloudLinkMailbox, error) { query := ` - INSERT INTO cloud_link_mailboxes (email_account_id, remote_id) - VALUES ($1, $2) - ON CONFLICT (email_account_id) DO UPDATE SET remote_id = EXCLUDED.remote_id - RETURNING email_account_id, remote_id, enrolled_at + INSERT INTO cloud_link_mailboxes (email_account_id, remote_id, managed) + VALUES ($1, $2, $3) + ON CONFLICT (email_account_id) DO UPDATE SET remote_id = EXCLUDED.remote_id, managed = EXCLUDED.managed + RETURNING email_account_id, remote_id, enrolled_at, managed ` var m models.CloudLinkMailbox - if err := r.db.QueryRow(ctx, query, accountID, remoteID).Scan(&m.EmailAccountID, &m.RemoteID, &m.EnrolledAt); err != nil { + if err := r.db.QueryRow(ctx, query, accountID, remoteID, managed).Scan(&m.EmailAccountID, &m.RemoteID, &m.EnrolledAt, &m.Managed); err != nil { db.CaptureError(err, query, nil, "queryrow") return nil, err } @@ -130,9 +130,9 @@ func (r *cloudLinkRepository) UnenrollAll(ctx context.Context) error { } func (r *cloudLinkRepository) GetByAccount(ctx context.Context, accountID uuid.UUID) (*models.CloudLinkMailbox, error) { - query := `SELECT email_account_id, remote_id, enrolled_at FROM cloud_link_mailboxes WHERE email_account_id = $1` + query := `SELECT email_account_id, remote_id, enrolled_at, managed FROM cloud_link_mailboxes WHERE email_account_id = $1` var m models.CloudLinkMailbox - if err := r.db.QueryRow(ctx, query, accountID).Scan(&m.EmailAccountID, &m.RemoteID, &m.EnrolledAt); err != nil { + if err := r.db.QueryRow(ctx, query, accountID).Scan(&m.EmailAccountID, &m.RemoteID, &m.EnrolledAt, &m.Managed); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, nil } @@ -143,7 +143,7 @@ func (r *cloudLinkRepository) GetByAccount(ctx context.Context, accountID uuid.U } func (r *cloudLinkRepository) List(ctx context.Context) ([]models.CloudLinkMailbox, error) { - query := `SELECT email_account_id, remote_id, enrolled_at FROM cloud_link_mailboxes ORDER BY enrolled_at` + query := `SELECT email_account_id, remote_id, enrolled_at, managed FROM cloud_link_mailboxes ORDER BY enrolled_at` rows, err := r.db.Query(ctx, query) if err != nil { db.CaptureError(err, query, nil, "query") @@ -153,7 +153,7 @@ func (r *cloudLinkRepository) List(ctx context.Context) ([]models.CloudLinkMailb out := []models.CloudLinkMailbox{} for rows.Next() { var m models.CloudLinkMailbox - if err := rows.Scan(&m.EmailAccountID, &m.RemoteID, &m.EnrolledAt); err != nil { + if err := rows.Scan(&m.EmailAccountID, &m.RemoteID, &m.EnrolledAt, &m.Managed); err != nil { return nil, err } out = append(out, m) diff --git a/internal/repository/pg_email.go b/internal/repository/pg_email.go index f315e5c6..a4409234 100644 --- a/internal/repository/pg_email.go +++ b/internal/repository/pg_email.go @@ -128,6 +128,8 @@ type EmailRepository interface { Delete(ctx context.Context, userID, emailAccountID string, workerLoadRefund float64) *errx.Error NewOauthAccount(ctx context.Context, userID string, data models.NewOauthAccount) (*models.Email, *errx.Error) + // NewManagedAccount creates an OAuth mailbox whose credential lives on Warmbly Cloud, so no token row is written. + NewManagedAccount(ctx context.Context, userID string, data models.NewOauthAccount) (*models.Email, *errx.Error) NewSMTPIMAPAccount(ctx context.Context, userID string, data models.NewSMTPIMAPAccount) (*models.Email, *errx.Error) RefreshBoxToken(ctx context.Context, id uuid.UUID, accessToken, refreshToken string, expiresAt time.Time) error // ReplaceSMTPIMAPCredentials overwrites a mailbox's SMTP/IMAP credential @@ -426,6 +428,26 @@ func (r *emailRepository) NewOauthAccount(ctx context.Context, userID string, da }, nil } +func (r *emailRepository) NewManagedAccount(ctx context.Context, userID string, data models.NewOauthAccount) (*models.Email, *errx.Error) { + if data.Provider != models.InboxProviderGoogle && data.Provider != models.InboxProviderOutlook { + sentry.CaptureException(errors.New("managed account: unsupported provider")) + return nil, errx.InternalError() + } + sigplain := utils.GetSignaturePlain(data.Name) + sightml := utils.GetSignatureHTML(data.Name) + t := time.Now() + id := uuid.New() + query := ` + INSERT INTO email_accounts (id, user_id, organization_id, email, name, provider, signature_plain, signature_html, tracking_domain, last_synced_at, created_at, updated_at, warmup_tag) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $10, $10, $11) + ` + if _, err := r.DB.Exec(ctx, query, id, userID, data.OrganizationID, data.Email, data.Name, data.Provider, sigplain, sightml, "", t, ""); err != nil { + db.CaptureError(err, query, nil, "exec") + return nil, errx.InternalError() + } + return r.GetByID(ctx, id) +} + func (r *emailRepository) NewSMTPIMAPAccount(ctx context.Context, userID string, data models.NewSMTPIMAPAccount) (*models.Email, *errx.Error) { if r.Encrypt == nil { sentry.CaptureException(errNoCredentialEncrypter) diff --git a/internal/repository/pg_poollink.go b/internal/repository/pg_poollink.go index 62249c47..60bf3c14 100644 --- a/internal/repository/pg_poollink.go +++ b/internal/repository/pg_poollink.go @@ -38,6 +38,10 @@ type PoolLinkRepository interface { ListMailboxes(ctx context.Context, instanceID uuid.UUID) ([]models.PoolLinkMailbox, error) CountMailboxesForOrganization(ctx context.Context, orgID uuid.UUID) (int, error) DeleteMailbox(ctx context.Context, instanceID, remoteID uuid.UUID) error + // TouchMailboxToken records when a managed mailbox last drew an access token. + TouchMailboxToken(ctx context.Context, instanceID, remoteID uuid.UUID) error + // ListAdoptableMailboxes: the workspace's active OAuth mailboxes no instance holds yet. + ListAdoptableMailboxes(ctx context.Context, orgID uuid.UUID) ([]models.PoolLinkWorkspaceMailbox, error) } type poolLinkRepository struct { @@ -249,11 +253,11 @@ func (r *poolLinkRepository) HasActiveInstance(ctx context.Context, orgID uuid.U func (r *poolLinkRepository) EnrollMailbox(ctx context.Context, m *models.PoolLinkMailbox) error { query := ` - INSERT INTO pool_link_mailboxes (instance_id, remote_id, email_account_id) - VALUES ($1, $2, $3) + INSERT INTO pool_link_mailboxes (instance_id, remote_id, email_account_id, managed) + VALUES ($1, $2, $3, $4) RETURNING enrolled_at ` - if err := r.db.QueryRow(ctx, query, m.InstanceID, m.RemoteID, m.EmailAccountID).Scan(&m.EnrolledAt); err != nil { + if err := r.db.QueryRow(ctx, query, m.InstanceID, m.RemoteID, m.EmailAccountID, m.Managed).Scan(&m.EnrolledAt); err != nil { db.CaptureError(err, query, nil, "queryrow") return err } @@ -262,7 +266,7 @@ func (r *poolLinkRepository) EnrollMailbox(ctx context.Context, m *models.PoolLi func scanPoolLinkMailbox(row pgx.Row) (*models.PoolLinkMailbox, error) { var m models.PoolLinkMailbox - if err := row.Scan(&m.InstanceID, &m.RemoteID, &m.EmailAccountID, &m.EnrolledAt); err != nil { + if err := row.Scan(&m.InstanceID, &m.RemoteID, &m.EmailAccountID, &m.EnrolledAt, &m.Managed, &m.LastTokenAt); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, nil } @@ -272,7 +276,7 @@ func scanPoolLinkMailbox(row pgx.Row) (*models.PoolLinkMailbox, error) { } func (r *poolLinkRepository) GetMailboxByRemote(ctx context.Context, instanceID, remoteID uuid.UUID) (*models.PoolLinkMailbox, error) { - query := `SELECT instance_id, remote_id, email_account_id, enrolled_at FROM pool_link_mailboxes WHERE instance_id = $1 AND remote_id = $2` + query := `SELECT instance_id, remote_id, email_account_id, enrolled_at, managed, last_token_at FROM pool_link_mailboxes WHERE instance_id = $1 AND remote_id = $2` m, err := scanPoolLinkMailbox(r.db.QueryRow(ctx, query, instanceID, remoteID)) if err != nil { db.CaptureError(err, query, nil, "queryrow") @@ -282,7 +286,7 @@ func (r *poolLinkRepository) GetMailboxByRemote(ctx context.Context, instanceID, } func (r *poolLinkRepository) GetMailboxByAccount(ctx context.Context, accountID uuid.UUID) (*models.PoolLinkMailbox, error) { - query := `SELECT instance_id, remote_id, email_account_id, enrolled_at FROM pool_link_mailboxes WHERE email_account_id = $1` + query := `SELECT instance_id, remote_id, email_account_id, enrolled_at, managed, last_token_at FROM pool_link_mailboxes WHERE email_account_id = $1` m, err := scanPoolLinkMailbox(r.db.QueryRow(ctx, query, accountID)) if err != nil { db.CaptureError(err, query, []any{accountID}, "queryrow") @@ -292,7 +296,7 @@ func (r *poolLinkRepository) GetMailboxByAccount(ctx context.Context, accountID } func (r *poolLinkRepository) ListMailboxes(ctx context.Context, instanceID uuid.UUID) ([]models.PoolLinkMailbox, error) { - query := `SELECT instance_id, remote_id, email_account_id, enrolled_at FROM pool_link_mailboxes WHERE instance_id = $1 ORDER BY enrolled_at` + query := `SELECT instance_id, remote_id, email_account_id, enrolled_at, managed, last_token_at FROM pool_link_mailboxes WHERE instance_id = $1 ORDER BY enrolled_at` rows, err := r.db.Query(ctx, query, instanceID) if err != nil { db.CaptureError(err, query, []any{instanceID}, "query") @@ -302,7 +306,7 @@ func (r *poolLinkRepository) ListMailboxes(ctx context.Context, instanceID uuid. out := []models.PoolLinkMailbox{} for rows.Next() { var m models.PoolLinkMailbox - if err := rows.Scan(&m.InstanceID, &m.RemoteID, &m.EmailAccountID, &m.EnrolledAt); err != nil { + if err := rows.Scan(&m.InstanceID, &m.RemoteID, &m.EmailAccountID, &m.EnrolledAt, &m.Managed, &m.LastTokenAt); err != nil { return nil, err } out = append(out, m) @@ -332,3 +336,35 @@ func (r *poolLinkRepository) DeleteMailbox(ctx context.Context, instanceID, remo } return nil } + +func (r *poolLinkRepository) TouchMailboxToken(ctx context.Context, instanceID, remoteID uuid.UUID) error { + _, err := r.db.Exec(ctx, `UPDATE pool_link_mailboxes SET last_token_at = NOW() WHERE instance_id = $1 AND remote_id = $2`, instanceID, remoteID) + return err +} + +func (r *poolLinkRepository) ListAdoptableMailboxes(ctx context.Context, orgID uuid.UUID) ([]models.PoolLinkWorkspaceMailbox, error) { + query := ` + SELECT a.id, a.email, a.name, a.provider, a.status + FROM email_accounts a + WHERE a.organization_id = $1 + AND a.status = 'active' + AND a.provider IN ('gmail', 'outlook') + AND NOT EXISTS (SELECT 1 FROM pool_link_mailboxes m WHERE m.email_account_id = a.id) + ORDER BY a.created_at + ` + rows, err := r.db.Query(ctx, query, orgID) + if err != nil { + db.CaptureError(err, query, []any{orgID}, "query") + return nil, err + } + defer rows.Close() + out := []models.PoolLinkWorkspaceMailbox{} + for rows.Next() { + var m models.PoolLinkWorkspaceMailbox + if err := rows.Scan(&m.ID, &m.Email, &m.Name, &m.Provider, &m.Status); err != nil { + return nil, err + } + out = append(out, m) + } + return out, rows.Err() +} diff --git a/web/src/app/app/emails/page.tsx b/web/src/app/app/emails/page.tsx index 22b7ad58..9820ed18 100644 --- a/web/src/app/app/emails/page.tsx +++ b/web/src/app/app/emails/page.tsx @@ -622,7 +622,7 @@ function MailboxRow({ {box.email} {inCloud && ( Cloud @@ -712,13 +712,18 @@ function MailboxRow({ - confirm.show(`Stop warming ${box.email} in the Warmbly pool? The cloud deletes its credential right away.`, async () => { - await cloudRun(() => cloudUnenroll.mutateAsync(box.id), `${box.email} removed from the pool`); - }) + confirm.show( + cloud?.managed + ? `Remove ${box.email} from this instance? It stays in your Warmbly Cloud workspace, where its sign-in lives; campaigns here stop sending from it.` + : `Stop warming ${box.email} in the Warmbly pool? The cloud deletes its credential right away.`, + async () => { + await cloudRun(() => cloudUnenroll.mutateAsync(box.id), cloud?.managed ? `${box.email} removed from this instance` : `${box.email} removed from the pool`); + }, + ) } icon={} > - Remove from Warmbly Cloud + {cloud?.managed ? "Remove from this instance" : "Remove from Warmbly Cloud"} diff --git a/web/src/app/app/settings/warmbly-cloud/MailboxTable.tsx b/web/src/app/app/settings/warmbly-cloud/MailboxTable.tsx index 125456ed..b61e1d16 100644 --- a/web/src/app/app/settings/warmbly-cloud/MailboxTable.tsx +++ b/web/src/app/app/settings/warmbly-cloud/MailboxTable.tsx @@ -40,9 +40,14 @@ export default function MailboxTable() { const flip = (row: CloudLinkMailboxRow) => { if (row.enrolled) { - confirm.show(`Stop warming ${row.email} in the Warmbly pool? The cloud deletes its credential right away.`, async () => { - await run(row.id, () => unenroll.mutateAsync(row.id), `${row.email} removed from the pool`); - }); + confirm.show( + row.managed + ? `Remove ${row.email} from this instance? It stays in your Warmbly Cloud workspace, where its sign-in lives.` + : `Stop warming ${row.email} in the Warmbly pool? The cloud deletes its credential right away.`, + async () => { + await run(row.id, () => unenroll.mutateAsync(row.id), row.managed ? `${row.email} removed from this instance` : `${row.email} removed from the pool`); + }, + ); return; } void run(row.id, () => enroll.mutateAsync(row.id), `${row.email} is now warming in the pool`); @@ -75,7 +80,7 @@ export default function MailboxTable() { {list.map((row) => { - const supported = providerSupported(row.provider); + const supported = providerSupported(row.provider) || row.managed; const cloud = row.cloud; const paused = !!cloud?.warmup?.paused; return ( @@ -85,7 +90,8 @@ export default function MailboxTable() {

{row.email}

{providerLabel(row.provider)} - {!supported && " · connect with SMTP/IMAP to enroll"} + {row.managed && " · signed in through Warmbly Cloud"} + {!supported && " · signed in with this instance's own OAuth app; add it again through Warmbly Cloud to warm it"} {row.enrolled && !cloud && " · waiting for the cloud"} {cloud?.errors && cloud.errors.length > 0 && ( diff --git a/web/src/app/app/settings/warmbly-cloud/providers.ts b/web/src/app/app/settings/warmbly-cloud/providers.ts index 6249d4a7..5aecbb90 100644 --- a/web/src/app/app/settings/warmbly-cloud/providers.ts +++ b/web/src/app/app/settings/warmbly-cloud/providers.ts @@ -1,6 +1,6 @@ -// Only SMTP/IMAP mailboxes can be warmed by the cloud: a Google or Microsoft -// refresh grant is bound to the OAuth app that issued it, which is this -// instance's own, so the cloud could never refresh it. +// Which local mailboxes can be enrolled as they are. A Google or Microsoft +// grant issued by this instance's own OAuth app cannot be refreshed by the +// cloud; such mailboxes are instead signed in through Warmbly Cloud (managed). export function providerSupported(provider: string): boolean { return provider === "smtp_imap"; } diff --git a/web/src/app/cloud-oauth/done/page.tsx b/web/src/app/cloud-oauth/done/page.tsx new file mode 100644 index 00000000..12f4f652 --- /dev/null +++ b/web/src/app/cloud-oauth/done/page.tsx @@ -0,0 +1,64 @@ +// Landing page for the Google/Microsoft popup that Warmbly Cloud ran on a +// linked instance's behalf. Hands the session back to the opener (the Add +// account dialog) and closes; without an opener it explains what happened. + +import React from "react"; +import { Link } from "react-router-dom"; +import { CheckIcon, XIcon } from "lucide-react"; +import { Logo } from "@/components/svg"; + +export interface CloudOAuthDoneMessage { + type: "cloud_oauth_callback"; + session: string; + status: "ok" | "error"; + error?: string; + message?: string; +} + +export default function CloudOAuthDonePage() { + const params = new URLSearchParams(window.location.search); + const session = params.get("session") ?? ""; + const status = params.get("status") === "ok" ? "ok" : "error"; + const error = params.get("error") ?? ""; + const message = params.get("message") ?? ""; + const [delivered, setDelivered] = React.useState(false); + + React.useEffect(() => { + const payload: CloudOAuthDoneMessage = { type: "cloud_oauth_callback", session, status, error, message }; + let ok = false; + try { + if (window.opener) { + window.opener.postMessage(payload, window.location.origin); + ok = true; + } + } catch { + /* no opener */ + } + setDelivered(ok); + if (ok) window.setTimeout(() => window.close(), 400); + }, [session, status, error, message]); + + return ( +

+
+ + + {status === "ok" ? : } + +

{status === "ok" ? "Mailbox signed in" : "Sign-in did not complete"}

+

+ {delivered + ? "This window closes on its own." + : status === "ok" + ? "Go back to the Warmbly tab; the mailbox is being added there." + : message || error || "Try again from Add account."} +

+ {!delivered && ( + + Back to mailboxes + + )} +
+
+ ); +} diff --git a/web/src/components/app/cloud/CloudLinkCard.tsx b/web/src/components/app/cloud/CloudLinkCard.tsx index 6a853e31..b4c55658 100644 --- a/web/src/components/app/cloud/CloudLinkCard.tsx +++ b/web/src/components/app/cloud/CloudLinkCard.tsx @@ -129,7 +129,7 @@ export default function CloudLinkCard({

Warmbly warms your mailboxes for you.

-

Free for 10 mailboxes. Your data stays on this server; only warmup runs in the cloud.

+

Free for 10 mailboxes. Google and Microsoft sign-in without OAuth setup. Your data stays on this server; only warmup runs in the cloud.

{start.isPending && (

Getting a code @@ -153,7 +153,8 @@ export default function CloudLinkCard({

+ + ))} + ); } @@ -431,10 +578,12 @@ function PickProvider({ onPick }: { onPick: (v: View) => void }) { function OAuthPanel({ provider, busy, + viaCloud, onConnect, }: { provider: OAuthProvider; busy: boolean; + viaCloud: boolean; onConnect: () => void; }) { const label = provider === "gmail" ? "Google" : "Microsoft"; @@ -450,16 +599,26 @@ function OAuthPanel({ Connect with {label}
- We'll open a {label} window. Approve the scopes and you're done. + {viaCloud + ? `Warmbly Cloud opens the ${label} window on its own app. Approve and you're done.` + : `We'll open a ${label} window. Approve the scopes and you're done.`}
- + {viaCloud ? ( + + ) : ( + + )} { return await Request({ method: "GET", url: "/cloud-link", authorization: true }); @@ -45,3 +48,23 @@ export async function unenrollCloudLinkMailbox(id: string): Promise { export async function setCloudLinkMailboxLifecycle(id: string, action: "pause" | "resume"): Promise { return await Request({ method: "POST", url: `/cloud-link/mailboxes/${id}/${action}`, authorization: true }); } + +// Cloud-managed mailboxes: Google/Microsoft sign-in through Warmbly Cloud's own +// OAuth app, and adoption of mailboxes connected directly on the workspace. + +export async function startCloudOAuth(provider: "gmail" | "outlook"): Promise { + return await Request({ method: "POST", url: "/cloud-link/oauth/start", data: { provider }, authorization: true }); +} + +export async function finishCloudOAuth(session: string): Promise { + return await Request({ method: "POST", url: "/cloud-link/oauth/finish", data: { session }, authorization: true }); +} + +export async function listCloudWorkspaceMailboxes(): Promise { + const res = await Request<{ data: PoolLinkWorkspaceMailbox[] }>({ method: "GET", url: "/cloud-link/workspace-mailboxes", authorization: true }); + return res.data ?? []; +} + +export async function adoptCloudMailbox(id: string): Promise { + return await Request({ method: "POST", url: `/cloud-link/workspace-mailboxes/${id}/adopt`, authorization: true }); +} diff --git a/web/src/lib/api/hooks/app/cloudlink/useCloudLink.ts b/web/src/lib/api/hooks/app/cloudlink/useCloudLink.ts index 226c1657..9d594658 100644 --- a/web/src/lib/api/hooks/app/cloudlink/useCloudLink.ts +++ b/web/src/lib/api/hooks/app/cloudlink/useCloudLink.ts @@ -1,7 +1,9 @@ import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { + adoptCloudMailbox, disconnectCloudLink, enrollCloudLinkMailbox, + listCloudWorkspaceMailboxes, getCloudLinkStatus, listCloudLinkMailboxes, pollCloudLinkConnect, @@ -85,6 +87,21 @@ export function useCloudLinkMailboxLifecycle() { }); } +export function useCloudWorkspaceMailboxes(enabled = true) { + return useQuery({ queryKey: [...CLOUD_LINK_KEY, "workspace-mailboxes"], queryFn: listCloudWorkspaceMailboxes, enabled, staleTime: 10_000 }); +} + +export function useAdoptCloudMailbox() { + const qc = useQueryClient(); + return useMutation({ + mutationFn: (id: string) => adoptCloudMailbox(id), + onSuccess: () => { + void qc.invalidateQueries({ queryKey: CLOUD_LINK_KEY }); + void qc.invalidateQueries({ queryKey: ["emails"] }); + }, + }); +} + // Cloud side. export function usePoolLinkCode(code: string) { diff --git a/web/src/lib/api/models/app/cloudlink/CloudLink.ts b/web/src/lib/api/models/app/cloudlink/CloudLink.ts index c710e178..d4690171 100644 --- a/web/src/lib/api/models/app/cloudlink/CloudLink.ts +++ b/web/src/lib/api/models/app/cloudlink/CloudLink.ts @@ -97,9 +97,12 @@ export interface PoolLinkMailboxState { remote_id: string; email_account_id: string; email: string; + name: string; provider: string; status: string; enrolled_at: Date; + /** The cloud holds the only credential; the instance sends with brokered tokens. */ + managed: boolean; warmup?: PoolLinkWarmupStatus | null; health?: PoolLinkWarmupHealth | null; sent_today: number; @@ -151,5 +154,21 @@ export interface CloudLinkMailboxRow { status: string; enrolled: boolean; enrolled_at?: Date | null; + managed: boolean; cloud?: PoolLinkMailboxState | null; } + +/** A Google or Microsoft consent started through Warmbly Cloud. */ +export interface CloudLinkOAuthStart { + url: string; + session: string; +} + +/** A mailbox connected directly on the cloud workspace that this instance can adopt. */ +export interface PoolLinkWorkspaceMailbox { + id: string; + email: string; + name: string; + provider: string; + status: string; +} diff --git a/web/src/main.tsx b/web/src/main.tsx index 892740d9..c74ef23b 100644 --- a/web/src/main.tsx +++ b/web/src/main.tsx @@ -81,6 +81,7 @@ import OnboardingPage from './app/onboarding/page'; import SelectOrgPage from './app/select-org/page'; import InviteAcceptPage from './app/invite/page'; import ConnectPage from './app/connect/page'; +import CloudOAuthDonePage from './app/cloud-oauth/done/page'; import WarmblyCloudSettingsPage from './app/app/settings/warmbly-cloud/page'; import SetupPage from './app/setup/page'; import SSOCallbackPage from './app/auth/sso/page'; @@ -200,6 +201,11 @@ const router = createBrowserRouter([ path: "connect", element: , }, + { + // Where Warmbly Cloud sends the Google/Microsoft popup back to on a linked instance. + path: "cloud-oauth/done", + element: , + }, { // First-run claim link printed by the backend on an empty database. path: "setup", From 1e476482087fb1ea1f7fa9de6badb4834f68abda Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 29 Aug 2026 09:52:19 -0700 Subject: [PATCH 2/3] feat: show the 'warmup cannot run with one mailbox' pool-size notice only on self-hosted instances, since a hosted workspace warms against the shared pool and one mailbox already has hundreds of partners there --- web/src/app/app/emails/page.tsx | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/web/src/app/app/emails/page.tsx b/web/src/app/app/emails/page.tsx index 9820ed18..7228e1a1 100644 --- a/web/src/app/app/emails/page.tsx +++ b/web/src/app/app/emails/page.tsx @@ -331,14 +331,17 @@ export default function AddressesPage() { /> setCloudDialog(true)} mailboxCount={stats.total} /> {!emailsData.isLoading && p?.setAddEmail(true)} />} - p?.setAddEmail(true)} - onConnectCloud={cloud.selfHosted && !cloud.connected ? () => setCloudDialog(true) : undefined} - cloudConnected={cloud.connected} - /> + {/* Hosted, the pool is thousands of mailboxes: the pool-size advice is self-host only. */} + {cloud.selfHosted && ( + p?.setAddEmail(true)} + onConnectCloud={!cloud.connected ? () => setCloudDialog(true) : undefined} + cloudConnected={cloud.connected} + /> + )} setCloudDialog(false)} /> {emailsData.isLoading ? (
From 44b2c189068d481c18bc3bdb80aa62069cac5992 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 29 Aug 2026 10:07:36 -0700 Subject: [PATCH 3/3] feat: address review on cloud-managed mailboxes: the consumer now asks the cloud to vouch for a warmup token in a mailbox it warms (GET /pool-link/instance/mailboxes/:id/warmup-tokens/:token) and files anything unverified as ordinary mail instead of dropping on a sender-controlled header, disconnect keeps local mirrors and the link until the cloud confirms the instance is released so managed mailboxes cannot be stranded, and long narrative comments are cut to one line --- cmd/consumer/main.go | 3 +- internal/api/handler/email_oauth_callback.go | 3 +- internal/api/handler/poollink.go | 23 ++++++++++++++ internal/api/routes.go | 1 + internal/app/cloudlink/managed.go | 31 ++++++++++++++----- internal/app/cloudlink/service.go | 18 +++++++++-- internal/app/consumer/event_new_email.go | 10 +++--- internal/app/consumer/service.go | 12 +++++-- internal/app/email/broker.go | 6 ++-- internal/app/email/loader.go | 3 +- internal/app/poollink/oauth.go | 29 +++++++++++++---- internal/app/poollink/service.go | 1 + internal/repository/http_brokered_token.go | 7 ++--- .../components/app/modals/AddEmailModal.tsx | 6 ++-- 14 files changed, 112 insertions(+), 41 deletions(-) diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index b7082fe6..f4c05b14 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -2,6 +2,7 @@ package main import ( "context" + "github.com/warmbly/warmbly/internal/app/cloudlink" "log" "os" "os/signal" @@ -373,7 +374,7 @@ func main() { EmailAccountErrorRepository: emailAccountErrorRepo, WarmupRepo: warmupRepo, PoolLinkRepo: repository.NewPoolLinkRepository(primaryDB.Pool), - CloudLinkRepo: repository.NewCloudLinkRepository(primaryDB.Pool, credEncrypter), + CloudLink: cloudlink.NewService(repository.NewCloudLinkRepository(primaryDB.Pool, credEncrypter), emailRepo, nil), WarmupContentRepo: repository.NewWarmupContentRepository(primaryDB.Pool), WarmupEngagementRepo: repository.NewWarmupEngagementRepository(primaryDB.Pool), WarmupService: warmupService, diff --git a/internal/api/handler/email_oauth_callback.go b/internal/api/handler/email_oauth_callback.go index 7ad1eac8..1858ab1b 100644 --- a/internal/api/handler/email_oauth_callback.go +++ b/internal/api/handler/email_oauth_callback.go @@ -105,8 +105,7 @@ func (h *Handler) renderOAuthCallback(c *gin.Context, provider string) { state := c.Query("state") providerErr := c.Query("error") - // A brokered consent (linked instance) completes here and lands back on - // the instance; there is no opener on our origin to post to. + // A brokered consent has no opener on our origin; it completes here and redirects to the instance. if h.PoolLinkService != nil && strings.HasPrefix(state, poollink.BrokerStatePrefix) { if to := h.PoolLinkService.CompleteOAuthCallback(c.Request.Context(), provider, code, state, providerErr); to != "" { c.Redirect(http.StatusFound, to) diff --git a/internal/api/handler/poollink.go b/internal/api/handler/poollink.go index e10b5d43..80b8da57 100644 --- a/internal/api/handler/poollink.go +++ b/internal/api/handler/poollink.go @@ -393,3 +393,26 @@ func (h *Handler) PoolLinkAdopt(c *gin.Context) { } c.JSON(http.StatusCreated, state) } + +func (h *Handler) PoolLinkVerifyWarmupToken(c *gin.Context) { + inst := middleware.GetPoolLinkInstance(c) + if inst == nil { + errx.JSON(c, errx.ErrUnauthorized) + return + } + remoteID, ok := poolLinkRemoteID(c) + if !ok { + return + } + token, err := uuid.Parse(c.Param("token")) + if err != nil { + c.JSON(http.StatusOK, gin.H{"valid": false}) + return + } + valid, xerr := h.PoolLinkService.VerifyWarmupToken(c.Request.Context(), inst, remoteID, token) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.JSON(http.StatusOK, gin.H{"valid": valid}) +} diff --git a/internal/api/routes.go b/internal/api/routes.go index 884bcac0..be842f0d 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -1130,6 +1130,7 @@ func Run( poolLinkInstance.POST("/oauth/start", h.PoolLinkOAuthStart) poolLinkInstance.POST("/oauth/finish", h.PoolLinkOAuthFinish) poolLinkInstance.GET("/mailboxes/:remoteId/token", h.PoolLinkAccessToken) + poolLinkInstance.GET("/mailboxes/:remoteId/warmup-tokens/:token", h.PoolLinkVerifyWarmupToken) poolLinkInstance.GET("/workspace-mailboxes", h.PoolLinkWorkspaceMailboxes) poolLinkInstance.POST("/mailboxes/adopt", h.PoolLinkAdopt) } diff --git a/internal/app/cloudlink/managed.go b/internal/app/cloudlink/managed.go index 639fdad8..93cea0e9 100644 --- a/internal/app/cloudlink/managed.go +++ b/internal/app/cloudlink/managed.go @@ -3,6 +3,7 @@ package cloudlink import ( "context" "net/http" + "net/url" "strings" "time" @@ -13,9 +14,7 @@ import ( "github.com/warmbly/warmbly/internal/models" ) -// Cloud-managed mailboxes: Google and Microsoft sign-in runs on Warmbly -// Cloud's OAuth app, the grant stays there, and this instance sends with -// access tokens it draws from the cloud. +// Cloud-managed mailboxes: the grant lives on Warmbly Cloud; this instance sends with brokered access tokens. var ( ErrNotManaged = errx.NewWithIdentifier(errx.NotFound, "cloud_link_not_managed", "This mailbox is not managed by Warmbly Cloud.") @@ -98,7 +97,7 @@ func (s *service) mirror(ctx context.Context, l *models.CloudLink, orgID, userID Email: state.Email, }) if xerr != nil { - // The cloud side exists without a local twin: release it so the next attempt is clean. + // Release the cloud side so the next attempt is clean. if rerr := s.clientFor(l).do(ctx, http.MethodDelete, "/instance/mailboxes/"+state.RemoteID.String(), nil, nil); rerr != nil { log.Error().Str("remote_id", state.RemoteID.String()).Str("code", rerr.Identifier).Msg("cloud link: local mirror failed and the cloud link could not be released") } @@ -146,8 +145,7 @@ func (s *service) Adopt(ctx context.Context, orgID, userID, cloudAccountID uuid. return s.mirror(ctx, l, orgID, userID, &state) } -// AccessToken is what the worker (through the backend) sends and syncs with. -// Cached until two minutes before expiry so a busy mailbox does not hammer the cloud. +// AccessToken is the worker's credential for a managed mailbox, cached briefly. func (s *service) AccessToken(ctx context.Context, accountID uuid.UUID) (*models.PoolLinkAccessToken, *errx.Error) { s.mu.Lock() if c, ok := s.tokens[accountID]; ok && time.Now().Before(c.expires) { @@ -171,7 +169,7 @@ func (s *service) AccessToken(ctx context.Context, accountID uuid.UUID) (*models if xerr := s.clientFor(l).do(ctx, http.MethodGet, "/instance/mailboxes/"+m.RemoteID.String()+"/token", nil, &tok); xerr != nil { return nil, xerr } - // Cap the cache so a revocation or block on the cloud bites within minutes, not an hour. + // Capped so a cloud-side revocation bites within minutes. until := tok.ExpiresAt.Add(-2 * time.Minute) if cap := time.Now().Add(tokenCacheMax); until.After(cap) { until = cap @@ -203,3 +201,22 @@ func (s *service) removeManaged(ctx context.Context, userID string, m *models.Cl } return nil } + +// VerifyWarmupToken asks the cloud whether warmup mail in an enrolled mailbox is its own. +func (s *service) VerifyWarmupToken(ctx context.Context, accountID uuid.UUID, token string) (bool, error) { + m, err := s.repo.GetByAccount(ctx, accountID) + if err != nil || m == nil { + return false, err + } + l, xerr := s.link(ctx) + if xerr != nil { + return false, xerr + } + var out struct { + Valid bool `json:"valid"` + } + if xerr := s.clientFor(l).do(ctx, http.MethodGet, "/instance/mailboxes/"+m.RemoteID.String()+"/warmup-tokens/"+url.PathEscape(token), nil, &out); xerr != nil { + return false, xerr + } + return out.Valid, nil +} diff --git a/internal/app/cloudlink/service.go b/internal/app/cloudlink/service.go index a6b3413d..85a689d8 100644 --- a/internal/app/cloudlink/service.go +++ b/internal/app/cloudlink/service.go @@ -109,6 +109,8 @@ type Service interface { // IsEnrolled is the local warmup scheduler's stand-down check; fails closed to false. IsEnrolled(ctx context.Context, accountID uuid.UUID) bool + // VerifyWarmupToken is the consumer's check that warmup mail in an enrolled mailbox is the cloud's. + VerifyWarmupToken(ctx context.Context, accountID uuid.UUID, token string) (bool, error) } type service struct { @@ -257,14 +259,24 @@ func (s *service) clearPending(p *PendingConnect) { s.mu.Unlock() } +// linkAlreadyGone: the cloud no longer recognises the instance, so nothing is left to release there. +func linkAlreadyGone(xerr *errx.Error) bool { + switch xerr.Identifier { + case "pool_link_revoked", "pool_link_instance_not_found", "unauthorized": + return true + } + return false +} + func (s *service) Disconnect(ctx context.Context) *errx.Error { l, xerr := s.link(ctx) if xerr != nil { return xerr } - // Best effort: an unreachable or revoked cloud must not keep the instance chained to it. - if xerr := s.clientFor(l).do(ctx, http.MethodDelete, "/instance", nil, nil); xerr != nil { - log.Warn().Str("code", xerr.Identifier).Msg("cloud link: remote disconnect failed; clearing local link anyway") + // The cloud must confirm (or already have dropped) the link before local + // state goes, or managed mailboxes stay owned there with no way to retry. + if xerr := s.clientFor(l).do(ctx, http.MethodDelete, "/instance", nil, nil); xerr != nil && !linkAlreadyGone(xerr) { + return xerr } // Managed mirrors have no credential of their own; they end with the link. if rows, err := s.repo.List(ctx); err == nil { diff --git a/internal/app/consumer/event_new_email.go b/internal/app/consumer/event_new_email.go index 60602622..c984ee6d 100644 --- a/internal/app/consumer/event_new_email.go +++ b/internal/app/consumer/event_new_email.go @@ -31,13 +31,13 @@ func (s *JobsService) HandleNewEmail(ctx context.Context, e *models.JobEventNewE if warmupToken == "" { warmupToken = extractHeaderValue(e.Message, "X-Warmbly-Token") } - // Warmup mail for a mailbox Warmbly Cloud warms carries the cloud's tokens, - // which this instance cannot verify; counting them as forgeries would - // poison the mailbox's local score, and filing them would flood the inbox. - if warmupToken != "" && s.CloudLinkRepo != nil { - if enrolled, lerr := s.CloudLinkRepo.IsEnrolled(ctx, e.Message.EmailID); lerr == nil && enrolled { + // A mailbox Warmbly Cloud warms receives the cloud's tokens: the cloud + // vouches for those; anything else is ordinary mail this instance cannot score. + if warmupToken != "" && s.CloudLink != nil && s.CloudLink.IsEnrolled(ctx, e.Message.EmailID) { + if ok, err := s.CloudLink.VerifyWarmupToken(ctx, e.Message.EmailID, warmupToken); err == nil && ok { return nil } + warmupToken = "" } if warmupToken != "" { handled, err := s.handleWarmupEmail(ctx, e, warmupToken) diff --git a/internal/app/consumer/service.go b/internal/app/consumer/service.go index 9e596294..d0ed53c7 100644 --- a/internal/app/consumer/service.go +++ b/internal/app/consumer/service.go @@ -3,6 +3,8 @@ package jobs import ( "context" + "github.com/google/uuid" + "github.com/rs/zerolog/log" "github.com/warmbly/warmbly/internal/app/advanced" warmupapp "github.com/warmbly/warmbly/internal/app/warmup" @@ -17,6 +19,12 @@ import ( "github.com/warmbly/warmbly/internal/repository" ) +// CloudLinkVerifier is the self-hosted pool link as the consumer sees it. +type CloudLinkVerifier interface { + IsEnrolled(ctx context.Context, accountID uuid.UUID) bool + VerifyWarmupToken(ctx context.Context, accountID uuid.UUID, token string) (bool, error) +} + type JobsService struct { // Bus delivers the jobs.worker-events stream (Kafka or NATS). Bus eventbus.EventBus @@ -33,8 +41,8 @@ type JobsService struct { WarmupRepo repository.WarmupRepository // PoolLinkRepo marks warmup-only mailboxes of linked instances; nil when unused. PoolLinkRepo repository.PoolLinkRepository - // CloudLinkRepo (self-hosted) marks mailboxes the cloud warms, whose warmup mail is not ours to verify. - CloudLinkRepo repository.CloudLinkRepository + // CloudLink (self-hosted) verifies cloud warmup mail in mailboxes the cloud warms; nil when unused. + CloudLink CloudLinkVerifier WarmupContentRepo repository.WarmupContentRepository WarmupEngagementRepo repository.WarmupEngagementRepository WarmupService warmupapp.Service diff --git a/internal/app/email/broker.go b/internal/app/email/broker.go index 53fdb15d..76a9a9fb 100644 --- a/internal/app/email/broker.go +++ b/internal/app/email/broker.go @@ -13,8 +13,7 @@ import ( "golang.org/x/oauth2" ) -// Brokered OAuth: the cloud runs a consent on its own OAuth app for a linked -// instance, keeps the grant, and mints access tokens on request. +// Brokered OAuth: consent on this deployment's OAuth app for a linked instance; the grant stays here. func (s *emailService) OAuthAuthorizeURL(provider models.InboxProvider, state string) (string, *errx.Error) { cfg, xerr := s.oauthConfigFor(provider) @@ -74,8 +73,7 @@ func (s *emailService) OAuthConnectWithCode(ctx context.Context, userID string, return acc, nil } -// OAuthAccessToken returns a live access token, refreshing and re-sealing the -// stored grant when it is within two minutes of expiry. +// OAuthAccessToken refreshes and re-seals the grant when within two minutes of expiry. func (s *emailService) OAuthAccessToken(ctx context.Context, accountID uuid.UUID) (*oauth2.Token, *errx.Error) { acc, xerr := s.emailRepository.GetByID(ctx, accountID) if xerr != nil { diff --git a/internal/app/email/loader.go b/internal/app/email/loader.go index 03ba5ae4..180cb22b 100644 --- a/internal/app/email/loader.go +++ b/internal/app/email/loader.go @@ -273,8 +273,7 @@ func (s *emailService) buildAddWorkerEmail(ctx context.Context, acc *models.Emai SaveToSent: &saveToSent, } - // A managed mailbox has no local credential: the worker draws access - // tokens from the backend, which brokers them from the cloud. + // A managed mailbox has no local credential; the worker draws brokered tokens. if s.cloudLink != nil { if m, err := s.cloudLink.GetByAccount(ctx, acc.ID); err == nil && m != nil && m.Managed { out.Brokered = true diff --git a/internal/app/poollink/oauth.go b/internal/app/poollink/oauth.go index 0f2973f3..a3458d14 100644 --- a/internal/app/poollink/oauth.go +++ b/internal/app/poollink/oauth.go @@ -110,8 +110,7 @@ func (s *service) StartOAuth(ctx context.Context, inst *models.PoolLinkInstance, return &models.PoolLinkOAuthStartResponse{URL: authURL, Session: session}, nil } -// CompleteOAuthCallback finishes a brokered consent server-side and returns -// where to send the browser. Never errors: every outcome lands on the instance. +// CompleteOAuthCallback finishes a brokered consent; every outcome redirects to the instance. func (s *service) CompleteOAuthCallback(ctx context.Context, provider, code, state, providerErr string) string { if s.cache == nil { return "" @@ -181,7 +180,7 @@ func (s *service) connectBrokered(ctx context.Context, st brokerState, code stri return remoteID, nil } -// startWarmup brings a freshly linked mailbox into the pool; failures are retried by the reconciler. +// startWarmup: failures here are retried by the reconciler. func (s *service) startWarmup(ctx context.Context, userID string, accountID uuid.UUID) { if _, xerr := s.emailSvc.SetWarmupLifecycle(ctx, userID, accountID.String(), "start"); xerr != nil { log.Warn().Str("account_id", accountID.String()).Msg("pool link: warmup start failed after enrollment") @@ -218,9 +217,7 @@ func (s *service) FinishOAuth(ctx context.Context, inst *models.PoolLinkInstance return s.GetMailbox(ctx, inst, res.RemoteID) } -// AccessToken mints a short-lived provider token for a managed mailbox. This -// is the enforcement point: a revoked link, a removed or inactive mailbox, or -// a hard-blocked one gets no token and the instance stops sending from it. +// AccessToken is the enforcement point: a revoked link or a removed, inactive or blocked mailbox gets no token. func (s *service) AccessToken(ctx context.Context, inst *models.PoolLinkInstance, remoteID uuid.UUID) (*models.PoolLinkAccessToken, *errx.Error) { m, err := s.repo.GetMailboxByRemote(ctx, inst.ID, remoteID) if err != nil { @@ -287,3 +284,23 @@ func (s *service) Adopt(ctx context.Context, inst *models.PoolLinkInstance, req } return s.GetMailbox(ctx, inst, req.RemoteID) } + +// VerifyWarmupToken lets a linked instance tell the cloud's warmup mail apart +// from anything else arriving in a mailbox it warms. +func (s *service) VerifyWarmupToken(ctx context.Context, inst *models.PoolLinkInstance, remoteID, token uuid.UUID) (bool, *errx.Error) { + m, err := s.repo.GetMailboxByRemote(ctx, inst.ID, remoteID) + if err != nil { + return false, errx.InternalError() + } + if m == nil { + return false, ErrMailboxNotFound + } + if s.warmup == nil { + return false, nil + } + t, err := s.warmup.FindWarmupToken(ctx, token) + if err != nil { + return false, errx.InternalError() + } + return t != nil && t.RecipientAccountID == m.EmailAccountID, nil +} diff --git a/internal/app/poollink/service.go b/internal/app/poollink/service.go index a5ac2a0c..c95256cc 100644 --- a/internal/app/poollink/service.go +++ b/internal/app/poollink/service.go @@ -73,6 +73,7 @@ type Service interface { AccessToken(ctx context.Context, inst *models.PoolLinkInstance, remoteID uuid.UUID) (*models.PoolLinkAccessToken, *errx.Error) ListWorkspaceMailboxes(ctx context.Context, inst *models.PoolLinkInstance) ([]models.PoolLinkWorkspaceMailbox, *errx.Error) Adopt(ctx context.Context, inst *models.PoolLinkInstance, req models.PoolLinkAdoptRequest) (*models.PoolLinkMailboxState, *errx.Error) + VerifyWarmupToken(ctx context.Context, inst *models.PoolLinkInstance, remoteID, token uuid.UUID) (bool, *errx.Error) // IsLinkedMailbox is the consumer's hot-path warmup-only check. IsLinkedMailbox(ctx context.Context, accountID uuid.UUID) bool diff --git a/internal/repository/http_brokered_token.go b/internal/repository/http_brokered_token.go index 14673f77..0e788c44 100644 --- a/internal/repository/http_brokered_token.go +++ b/internal/repository/http_brokered_token.go @@ -14,9 +14,7 @@ import ( "golang.org/x/oauth2" ) -// BrokeredTokenClient is the worker's way to a credential for a mailbox that -// Warmbly Cloud manages: the backend brokers a short-lived access token, the -// refresh grant never reaches this process. +// BrokeredTokenClient fetches short-lived access tokens for cloud-managed mailboxes; no refresh grant reaches the worker. type BrokeredTokenClient interface { Token(ctx context.Context, accountID uuid.UUID) (*oauth2.Token, error) // Source adapts a mailbox to oauth2; the returned source caches until expiry. @@ -44,8 +42,7 @@ func NewHTTPBrokeredTokenClient(baseURL, token string) (BrokeredTokenClient, err }, nil } -// BrokeredTokenRefused is a definitive no from the cloud (revoked, blocked, -// removed): callers should treat it as an authentication failure, not a blip. +// BrokeredTokenRefused is a definitive no from the cloud, not a transient failure. type BrokeredTokenRefused struct { Code string Message string diff --git a/web/src/components/app/modals/AddEmailModal.tsx b/web/src/components/app/modals/AddEmailModal.tsx index 121c7a22..bb63f54d 100644 --- a/web/src/components/app/modals/AddEmailModal.tsx +++ b/web/src/components/app/modals/AddEmailModal.tsx @@ -13,10 +13,8 @@ // OAuth popup posts {type:"email_oauth_callback", code, state} back here // via window.postMessage; we then call OAuth-finish with the user's bearer. // -// On a self-hosted instance linked to Warmbly Cloud, Google and Microsoft -// sign-in runs on the cloud's OAuth app instead (no BOX_* setup): the popup -// comes back to /cloud-oauth/done, which posts {type:"cloud_oauth_callback", -// session} and we redeem the session with /cloud-link/oauth/finish. +// Linked to Warmbly Cloud: the consent runs on the cloud's app, the popup +// returns to /cloud-oauth/done, and we redeem its session via /cloud-link/oauth/finish. import React from "react"; import { AnimatePresence, motion } from "framer-motion";