fix: repair migration collection names and cross-cutting scoping gaps
Findings from the final whole-branch review. - scopedCollections named "audit" and "channels", but the code writes to audit_logs and notification_channels. On upgrade from single-tenant, legacy audit events and channels would never get org_id, becoming invisible to org-filtered reads while channels silently stopped firing — and the detection loop counted the wrong names, so the 0001 marker could be written having migrated nothing. Names fixed, plus migration 0003 so an incorrectly-migrated instance converges with a fresh one. - EnsureAuthIndexes failure is now fatal. GetUserByEmail is unscoped and the OIDC cross-org guard compares against whichever duplicate Mongo returns first, so users.email uniqueness is a security invariant, and a legacy collection with duplicate emails is the realistic upgrade case. - Evict the per-org OIDC provider cache on save; rotating away from a compromised IdP previously had no effect until restart. - Build the oauth2 config per request instead of mutating a shared cached pointer outside the mutex, which raced on RedirectURL between concurrent logins for the same org. - Stamp org_id on console_sessions, incidents and monitor_rollups, the last collections with no tenant column. 0003 derives their org from the owning server/monitor rather than defaulting, so one org's console history and incident timeline cannot merge into another's. - Seed default steps when an org is created, not only at boot. - Reject an empty session OrgID at the middleware. - Derive the app root label from APP_ROOT_LABEL instead of hardcoding "vantage", which silently disabled the host guard off that domain. - Stop caching negative slug lookups, so a new org's subdomain resolves immediately.
This commit is contained in:
@@ -36,7 +36,7 @@ func consoleConnect(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
sess, err := services.CreateConsoleSession(body.ServerID, body.Protocol, body.KeyID, actorFromCtx(c), c.ClientIP())
|
||||
sess, err := services.CreateConsoleSession(auth.OrgID(c), body.ServerID, body.Protocol, body.KeyID, actorFromCtx(c), c.ClientIP())
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
@@ -48,14 +48,14 @@ func consoleConnect(c *gin.Context) {
|
||||
}
|
||||
|
||||
if (body.Protocol == "rdp" || body.Protocol == "vnc") && (body.RDPUsername != "" || body.RDPPassword != "") {
|
||||
if err := services.StashConsoleRDPCreds(sess.SessionID, body.RDPUsername, body.RDPPassword); err != nil {
|
||||
if err := services.StashConsoleRDPCreds(auth.OrgID(c), sess.SessionID, body.RDPUsername, body.RDPPassword); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if body.Protocol == "ssh" {
|
||||
if err := services.SetConsoleSSHUser(sess.SessionID, body.SSHUsername); err != nil {
|
||||
if err := services.SetConsoleSSHUser(auth.OrgID(c), sess.SessionID, body.SSHUsername); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
@@ -89,7 +89,8 @@ func consoleTunnel(c *gin.Context) {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid token"})
|
||||
return
|
||||
}
|
||||
sess, err := services.GetConsoleSession(sessionID)
|
||||
orgID := auth.OrgID(c)
|
||||
sess, err := services.GetConsoleSession(orgID, sessionID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "session not found"})
|
||||
return
|
||||
@@ -103,7 +104,7 @@ func consoleTunnel(c *gin.Context) {
|
||||
}
|
||||
|
||||
// Single-use: atomically spend the token so a replay within its TTL is rejected.
|
||||
if err := services.ConsumeSessionToken(sessionID); err != nil {
|
||||
if err := services.ConsumeSessionToken(orgID, sessionID); err != nil {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"error": "token already used"})
|
||||
return
|
||||
}
|
||||
@@ -127,7 +128,7 @@ func consoleTunnel(c *gin.Context) {
|
||||
|
||||
var rdpUser, rdpPass string
|
||||
if sess.Protocol == "rdp" || sess.Protocol == "vnc" {
|
||||
rdpUser, rdpPass, err = services.ConsumeConsoleRDPCreds(sessionID)
|
||||
rdpUser, rdpPass, err = services.ConsumeConsoleRDPCreds(orgID, sessionID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "could not load credentials"})
|
||||
return
|
||||
@@ -172,7 +173,7 @@ func consoleTunnel(c *gin.Context) {
|
||||
|
||||
wsServer := guac.NewWebsocketServer(connect)
|
||||
wsServer.OnDisconnect = func(id string, r *http.Request, t guac.Tunnel) {
|
||||
_ = services.EndConsoleSession(sessionID)
|
||||
_ = services.EndConsoleSession(orgID, sessionID)
|
||||
}
|
||||
wsServer.ServeHTTP(c.Writer, c.Request)
|
||||
}
|
||||
|
||||
@@ -130,7 +130,7 @@ func getMonitorIncidents(c *gin.Context) {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "monitor not found"})
|
||||
return
|
||||
}
|
||||
incidents, err := services.ListIncidents(c.Param("id"), 50)
|
||||
incidents, err := services.ListIncidents(auth.OrgID(c), c.Param("id"), 50)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
@@ -149,7 +149,7 @@ func getMonitorUptime(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
since := time.Now().Add(-30 * 24 * time.Hour)
|
||||
rollups, err := services.UptimeRollups(c.Param("id"), since)
|
||||
rollups, err := services.UptimeRollups(auth.OrgID(c), c.Param("id"), since)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
|
||||
@@ -155,5 +155,8 @@ func putOrgOIDC(c *gin.Context) {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
// Drop the cached provider so a rotated issuer takes effect immediately —
|
||||
// an admin moving off a compromised IdP must not keep authenticating there.
|
||||
auth.EvictOIDCProvider(auth.OrgID(c))
|
||||
c.JSON(http.StatusOK, gin.H{"saved": true})
|
||||
}
|
||||
|
||||
@@ -62,6 +62,13 @@ func Middleware() gin.HandlerFunc {
|
||||
return
|
||||
}
|
||||
|
||||
// An org-less session would turn every downstream scope into
|
||||
// {"org_id": ""} — fail closed rather than query across tenants.
|
||||
if sess.OrgID == "" {
|
||||
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "session has no organization"})
|
||||
return
|
||||
}
|
||||
|
||||
c.Set(ctxSessionKey, sess)
|
||||
|
||||
if hostOrg, ok := OrgFromHost(c); ok && hostOrg.OrgID != sess.OrgID {
|
||||
|
||||
@@ -13,16 +13,20 @@ import (
|
||||
"golang.org/x/oauth2"
|
||||
)
|
||||
|
||||
type orgProvider struct {
|
||||
provider *oidc.Provider
|
||||
oauth *oauth2.Config
|
||||
}
|
||||
|
||||
var (
|
||||
provMu sync.Mutex
|
||||
provCache = map[string]*orgProvider{}
|
||||
provCache = map[string]*oidc.Provider{}
|
||||
)
|
||||
|
||||
// EvictOIDCProvider drops an org's cached provider so the next login rediscovers
|
||||
// it from the (possibly changed) issuer. Called by the API layer after the org's
|
||||
// OIDC config is saved — services cannot import auth, so the handler wires it.
|
||||
func EvictOIDCProvider(orgID string) {
|
||||
provMu.Lock()
|
||||
delete(provCache, orgID)
|
||||
provMu.Unlock()
|
||||
}
|
||||
|
||||
func redirectURL(c *gin.Context) string {
|
||||
scheme := "https"
|
||||
if c.Request.TLS == nil && c.GetHeader("X-Forwarded-Proto") != "https" {
|
||||
@@ -31,34 +35,36 @@ func redirectURL(c *gin.Context) string {
|
||||
return fmt.Sprintf("%s://%s/auth/oidc/callback", scheme, c.Request.Host)
|
||||
}
|
||||
|
||||
func providerForOrg(ctx context.Context, c *gin.Context, orgID string) (*orgProvider, error) {
|
||||
// providerForOrg returns the org's (cached) OIDC provider plus a request-local
|
||||
// oauth2 config. The config is never stored on the cached entry: RedirectURL is
|
||||
// derived from this request's Host, so sharing it would let one in-flight login
|
||||
// overwrite another's redirect URI.
|
||||
func providerForOrg(ctx context.Context, c *gin.Context, orgID string) (*oidc.Provider, *oauth2.Config, error) {
|
||||
cfg, err := services.GetOrgOIDC(orgID)
|
||||
if err != nil || !cfg.Enabled {
|
||||
return nil, fmt.Errorf("org SSO not configured")
|
||||
return nil, nil, fmt.Errorf("org SSO not configured")
|
||||
}
|
||||
secret, err := services.GetOrgOIDCSecret(orgID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, nil, err
|
||||
}
|
||||
provMu.Lock()
|
||||
op := provCache[orgID]
|
||||
p := provCache[orgID]
|
||||
provMu.Unlock()
|
||||
if op == nil || op.provider == nil {
|
||||
p, err := oidc.NewProvider(ctx, cfg.Issuer)
|
||||
if p == nil {
|
||||
p, err = oidc.NewProvider(ctx, cfg.Issuer)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, nil, err
|
||||
}
|
||||
op = &orgProvider{provider: p}
|
||||
provMu.Lock()
|
||||
provCache[orgID] = op
|
||||
provCache[orgID] = p
|
||||
provMu.Unlock()
|
||||
}
|
||||
op.oauth = &oauth2.Config{
|
||||
return p, &oauth2.Config{
|
||||
ClientID: cfg.ClientID, ClientSecret: secret,
|
||||
RedirectURL: redirectURL(c), Endpoint: op.provider.Endpoint(),
|
||||
RedirectURL: redirectURL(c), Endpoint: p.Endpoint(),
|
||||
Scopes: []string{oidc.ScopeOpenID, "profile", "email"},
|
||||
}
|
||||
return op, nil
|
||||
}, nil
|
||||
}
|
||||
|
||||
func HandleOIDCStart(c *gin.Context) {
|
||||
@@ -68,7 +74,7 @@ func HandleOIDCStart(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
ctx := c.Request.Context()
|
||||
op, err := providerForOrg(ctx, c, org.OrgID)
|
||||
_, oauthCfg, err := providerForOrg(ctx, c, org.OrgID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
@@ -82,7 +88,7 @@ func HandleOIDCStart(c *gin.Context) {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "state save failed"})
|
||||
return
|
||||
}
|
||||
c.Redirect(http.StatusFound, op.oauth.AuthCodeURL(state))
|
||||
c.Redirect(http.StatusFound, oauthCfg.AuthCodeURL(state))
|
||||
}
|
||||
|
||||
func HandleOIDCCallback(c *gin.Context) {
|
||||
@@ -92,12 +98,12 @@ func HandleOIDCCallback(c *gin.Context) {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid state"})
|
||||
return
|
||||
}
|
||||
op, err := providerForOrg(ctx, c, orgID)
|
||||
provider, oauthCfg, err := providerForOrg(ctx, c, orgID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
token, err := op.oauth.Exchange(ctx, c.Query("code"))
|
||||
token, err := oauthCfg.Exchange(ctx, c.Query("code"))
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "token exchange failed"})
|
||||
return
|
||||
@@ -107,7 +113,7 @@ func HandleOIDCCallback(c *gin.Context) {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "missing id_token"})
|
||||
return
|
||||
}
|
||||
idToken, err := op.provider.Verifier(&oidc.Config{ClientID: op.oauth.ClientID}).Verify(ctx, rawIDToken)
|
||||
idToken, err := provider.Verifier(&oidc.Config{ClientID: oauthCfg.ClientID}).Verify(ctx, rawIDToken)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "token verification failed"})
|
||||
return
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package auth
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -22,6 +23,16 @@ var (
|
||||
|
||||
const orgCacheTTL = 60 * time.Second
|
||||
|
||||
// appRootLabel is the DNS label the app is deployed under, i.e. the "vantage"
|
||||
// in <slug>.vantage.<tld>. Deployments on another root must set APP_ROOT_LABEL
|
||||
// or every host resolves to no org, disabling the host/session mismatch guard.
|
||||
func appRootLabel() string {
|
||||
if v := os.Getenv("APP_ROOT_LABEL"); v != "" {
|
||||
return strings.ToLower(v)
|
||||
}
|
||||
return "vantage"
|
||||
}
|
||||
|
||||
// hostSlug extracts the leftmost DNS label if the host is a subdomain of the
|
||||
// app root. Returns "" for the apex or an unknown host shape.
|
||||
func hostSlug(host string) string {
|
||||
@@ -29,15 +40,16 @@ func hostSlug(host string) string {
|
||||
if i := strings.IndexByte(host, ':'); i >= 0 {
|
||||
host = host[:i]
|
||||
}
|
||||
// Expect <slug>.vantage.<...>; apex is vantage.<...>
|
||||
root := appRootLabel()
|
||||
// Expect <slug>.<root>.<...>; apex is <root>.<...>
|
||||
parts := strings.Split(host, ".")
|
||||
if len(parts) < 3 {
|
||||
return ""
|
||||
}
|
||||
if parts[1] != "vantage" {
|
||||
if parts[1] != root {
|
||||
return ""
|
||||
}
|
||||
if parts[0] == "vantage" || parts[0] == "www" {
|
||||
if parts[0] == root || parts[0] == "www" {
|
||||
return ""
|
||||
}
|
||||
return parts[0]
|
||||
@@ -56,11 +68,13 @@ func OrgFromHost(c *gin.Context) (*models.Org, bool) {
|
||||
orgCacheMu.Unlock()
|
||||
|
||||
org, err := services.GetOrgBySlug(slug)
|
||||
if err != nil {
|
||||
org = nil
|
||||
if err != nil || org == nil {
|
||||
// Never cache a miss: a just-bootstrapped org would otherwise 404 on its
|
||||
// own subdomain for the rest of the TTL. Misses are cheap and rare.
|
||||
return nil, false
|
||||
}
|
||||
orgCacheMu.Lock()
|
||||
orgCache[slug] = cachedOrg{org: org, at: time.Now()}
|
||||
orgCacheMu.Unlock()
|
||||
return org, org != nil
|
||||
return org, true
|
||||
}
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
|
||||
type ConsoleSession struct {
|
||||
ID bson.ObjectID `bson:"_id,omitempty" json:"_id,omitempty"`
|
||||
OrgID string `bson:"org_id" json:"org_id"`
|
||||
SessionID string `bson:"session_id" json:"session_id"`
|
||||
ServerID string `bson:"server_id" json:"server_id"`
|
||||
Protocol string `bson:"protocol" json:"protocol"` // ssh | rdp | vnc
|
||||
|
||||
@@ -63,6 +63,7 @@ type Monitor struct {
|
||||
}
|
||||
|
||||
type Incident struct {
|
||||
OrgID string `bson:"org_id" json:"org_id"`
|
||||
IncidentID string `bson:"incident_id" json:"incident_id"`
|
||||
MonitorID string `bson:"monitor_id" json:"monitor_id"`
|
||||
StartedAt time.Time `bson:"started_at" json:"started_at"`
|
||||
@@ -71,6 +72,7 @@ type Incident struct {
|
||||
}
|
||||
|
||||
type Rollup struct {
|
||||
OrgID string `bson:"org_id" json:"org_id"`
|
||||
MonitorID string `bson:"monitor_id" json:"monitor_id"`
|
||||
PeriodStart time.Time `bson:"period_start" json:"period_start"` // hour bucket
|
||||
Checks int `bson:"checks" json:"checks"`
|
||||
|
||||
@@ -129,11 +129,12 @@ func BuildGuacParams(srv *models.Server, protocol, sshUser, privateKey, passphra
|
||||
}
|
||||
}
|
||||
|
||||
func CreateConsoleSession(serverID, protocol, keyID, user, clientIP string) (*models.ConsoleSession, error) {
|
||||
func CreateConsoleSession(orgID, serverID, protocol, keyID, user, clientIP string) (*models.ConsoleSession, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
s := &models.ConsoleSession{
|
||||
OrgID: orgID,
|
||||
SessionID: uuid.NewString(),
|
||||
ServerID: serverID,
|
||||
Protocol: protocol,
|
||||
@@ -148,11 +149,11 @@ func CreateConsoleSession(serverID, protocol, keyID, user, clientIP string) (*mo
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func GetConsoleSession(sessionID string) (*models.ConsoleSession, error) {
|
||||
func GetConsoleSession(orgID, sessionID string) (*models.ConsoleSession, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
var s models.ConsoleSession
|
||||
if err := db.Col("console_sessions").FindOne(ctx, bson.M{"session_id": sessionID}).Decode(&s); err != nil {
|
||||
if err := db.Col("console_sessions").FindOne(ctx, bson.M{"session_id": sessionID, "org_id": orgID}).Decode(&s); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &s, nil
|
||||
@@ -160,7 +161,7 @@ func GetConsoleSession(sessionID string) (*models.ConsoleSession, error) {
|
||||
|
||||
// StashConsoleRDPCreds encrypts and stores single-use RDP credentials on the
|
||||
// session document. They are consumed (and cleared) when the tunnel opens.
|
||||
func StashConsoleRDPCreds(sessionID, username, password string) error {
|
||||
func StashConsoleRDPCreds(orgID, sessionID, username, password string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
u, err := encryptString(username)
|
||||
@@ -172,7 +173,7 @@ func StashConsoleRDPCreds(sessionID, username, password string) error {
|
||||
return err
|
||||
}
|
||||
_, err = db.Col("console_sessions").UpdateOne(ctx,
|
||||
bson.M{"session_id": sessionID},
|
||||
bson.M{"session_id": sessionID, "org_id": orgID},
|
||||
bson.M{"$set": bson.M{"rdp_user_enc": u, "rdp_pass_enc": p}},
|
||||
)
|
||||
return err
|
||||
@@ -181,8 +182,8 @@ func StashConsoleRDPCreds(sessionID, username, password string) error {
|
||||
// ConsumeConsoleRDPCreds decrypts and returns the stored RDP credentials, then
|
||||
// clears them from the session document (single-use). Returns empty strings if
|
||||
// none were stored.
|
||||
func ConsumeConsoleRDPCreds(sessionID string) (username, password string, err error) {
|
||||
s, err := GetConsoleSession(sessionID)
|
||||
func ConsumeConsoleRDPCreds(orgID, sessionID string) (username, password string, err error) {
|
||||
s, err := GetConsoleSession(orgID, sessionID)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
@@ -202,18 +203,18 @@ func ConsumeConsoleRDPCreds(sessionID string) (username, password string, err er
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
_, _ = db.Col("console_sessions").UpdateOne(ctx,
|
||||
bson.M{"session_id": sessionID},
|
||||
bson.M{"session_id": sessionID, "org_id": orgID},
|
||||
bson.M{"$unset": bson.M{"rdp_user_enc": "", "rdp_pass_enc": ""}},
|
||||
)
|
||||
return username, password, nil
|
||||
}
|
||||
|
||||
// SetConsoleSSHUser persists the SSH username to use on the session doc.
|
||||
func SetConsoleSSHUser(sessionID, username string) error {
|
||||
func SetConsoleSSHUser(orgID, sessionID, username string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
_, err := db.Col("console_sessions").UpdateOne(ctx,
|
||||
bson.M{"session_id": sessionID},
|
||||
bson.M{"session_id": sessionID, "org_id": orgID},
|
||||
bson.M{"$set": bson.M{"ssh_username": username}})
|
||||
return err
|
||||
}
|
||||
@@ -221,12 +222,12 @@ func SetConsoleSSHUser(sessionID, username string) error {
|
||||
// ConsumeSessionToken atomically marks a session's one-time token as spent.
|
||||
// It returns an error if the token was already consumed (replay) or the session
|
||||
// does not exist, so the tunnel can be opened at most once per issued token.
|
||||
func ConsumeSessionToken(sessionID string) error {
|
||||
func ConsumeSessionToken(orgID, sessionID string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
now := time.Now()
|
||||
res, err := db.Col("console_sessions").UpdateOne(ctx,
|
||||
bson.M{"session_id": sessionID, "token_consumed_at": nil},
|
||||
bson.M{"session_id": sessionID, "org_id": orgID, "token_consumed_at": nil},
|
||||
bson.M{"$set": bson.M{"token_consumed_at": now}},
|
||||
)
|
||||
if err != nil {
|
||||
@@ -238,12 +239,12 @@ func ConsumeSessionToken(sessionID string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func EndConsoleSession(sessionID string) error {
|
||||
func EndConsoleSession(orgID, sessionID string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
now := time.Now()
|
||||
_, err := db.Col("console_sessions").UpdateOne(ctx,
|
||||
bson.M{"session_id": sessionID, "ended_at": nil},
|
||||
bson.M{"session_id": sessionID, "org_id": orgID, "ended_at": nil},
|
||||
bson.M{"$set": bson.M{"ended_at": now}},
|
||||
)
|
||||
return err
|
||||
|
||||
@@ -16,7 +16,8 @@ import (
|
||||
var scopedCollections = []string{
|
||||
"servers", "keys", "assignments", "secrets",
|
||||
"workflows", "workflow_steps", "workflow_runs",
|
||||
"audit", "monitors", "channels",
|
||||
"audit_logs", "monitors", "notification_channels",
|
||||
"console_sessions", "incidents", "monitor_rollups",
|
||||
}
|
||||
|
||||
// EnsureAuthIndexes creates unique indexes for the new auth collections.
|
||||
@@ -45,6 +46,20 @@ func EnsureAuthIndexes() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// defaultBackfillOrg resolves the org that org-less legacy documents belong to:
|
||||
// the "default" org, created if absent. Shared by 0001 and 0003 so an instance
|
||||
// that ran either one converges on the same org.
|
||||
func defaultBackfillOrg(ctx context.Context) (*models.Org, error) {
|
||||
var org models.Org
|
||||
if err := db.Col("orgs").FindOne(ctx, bson.M{"slug": "default"}).Decode(&org); err != nil {
|
||||
org = models.Org{OrgID: uuid.NewString(), Name: "Default", Slug: "default", CreatedAt: time.Now()}
|
||||
if _, err := db.Col("orgs").InsertOne(ctx, org); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return &org, nil
|
||||
}
|
||||
|
||||
// RunMigrations backfills a default org onto pre-existing documents. Idempotent
|
||||
// via a marker in the migrations collection.
|
||||
func RunMigrations() error {
|
||||
@@ -67,13 +82,9 @@ func RunMigrations() error {
|
||||
}
|
||||
|
||||
if needs {
|
||||
var org models.Org
|
||||
err := db.Col("orgs").FindOne(ctx, bson.M{"slug": "default"}).Decode(&org)
|
||||
org, err := defaultBackfillOrg(ctx)
|
||||
if err != nil {
|
||||
org = models.Org{OrgID: uuid.NewString(), Name: "Default", Slug: "default", CreatedAt: time.Now()}
|
||||
if _, err := db.Col("orgs").InsertOne(ctx, org); err != nil {
|
||||
return err
|
||||
}
|
||||
return err
|
||||
}
|
||||
for _, col := range scopedCollections {
|
||||
if _, err := db.Col(col).UpdateMany(ctx,
|
||||
@@ -89,6 +100,100 @@ func RunMigrations() error {
|
||||
return err
|
||||
}
|
||||
|
||||
// MigrateMissedOrgScopes repairs collections that migration 0001 could not
|
||||
// reach. 0001 originally listed "audit" and "channels", but the real collections
|
||||
// are audit_logs and notification_channels, so on any instance that ran that
|
||||
// version those documents were left without org_id — invisible to org-filtered
|
||||
// reads, and in the channels' case silently non-firing. The 0001 marker is
|
||||
// already written there, so renaming alone does not repair them; this migration
|
||||
// converges both the never-migrated and the incorrectly-migrated case.
|
||||
//
|
||||
// It also stamps console_sessions, incidents and monitor_rollups, which gained
|
||||
// an org_id only after 0001 shipped. Those carry an owning monitor/server whose
|
||||
// org is authoritative, so they are derived rather than defaulted. Idempotent
|
||||
// via a marker in the migrations collection.
|
||||
func MigrateMissedOrgScopes() error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
const marker = "0003_missed_org_scopes"
|
||||
if n, _ := db.Col("migrations").CountDocuments(ctx, bson.M{"_id": marker}); n > 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Same org resolution 0001 uses, for the collections it meant to cover.
|
||||
missed := []string{"audit_logs", "notification_channels"}
|
||||
needs := false
|
||||
for _, col := range missed {
|
||||
n, _ := db.Col(col).CountDocuments(ctx, bson.M{"org_id": bson.M{"$exists": false}})
|
||||
if n > 0 {
|
||||
needs = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if needs {
|
||||
org, err := defaultBackfillOrg(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, col := range missed {
|
||||
if _, err := db.Col(col).UpdateMany(ctx,
|
||||
bson.M{"org_id": bson.M{"$exists": false}},
|
||||
bson.M{"$set": bson.M{"org_id": org.OrgID}},
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Derived from the owning record — defaulting these would hand one org
|
||||
// another org's console history and incident timeline.
|
||||
if err := backfillOrgFromOwner(ctx, "console_sessions", "server_id", "servers", "server_id"); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := backfillOrgFromOwner(ctx, "incidents", "monitor_id", "monitors", "monitor_id"); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := backfillOrgFromOwner(ctx, "monitor_rollups", "monitor_id", "monitors", "monitor_id"); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
_, err := db.Col("migrations").InsertOne(ctx, bson.M{"_id": marker, "applied_at": time.Now()})
|
||||
return err
|
||||
}
|
||||
|
||||
// backfillOrgFromOwner stamps org_id on every doc in col that lacks one, taking
|
||||
// the org from the record in ownerCol it points at. Orphans (owner already
|
||||
// deleted) are left alone; they are unreachable either way.
|
||||
func backfillOrgFromOwner(ctx context.Context, col, localField, ownerCol, ownerField string) error {
|
||||
var ids []string
|
||||
if err := db.Col(col).Distinct(ctx, localField,
|
||||
bson.M{"org_id": bson.M{"$exists": false}}).Decode(&ids); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, id := range ids {
|
||||
if id == "" {
|
||||
continue
|
||||
}
|
||||
var owner struct {
|
||||
OrgID string `bson:"org_id"`
|
||||
}
|
||||
if err := db.Col(ownerCol).FindOne(ctx, bson.M{ownerField: id}).Decode(&owner); err != nil {
|
||||
continue
|
||||
}
|
||||
if owner.OrgID == "" {
|
||||
continue
|
||||
}
|
||||
if _, err := db.Col(col).UpdateMany(ctx,
|
||||
bson.M{localField: id, "org_id": bson.M{"$exists": false}},
|
||||
bson.M{"$set": bson.M{"org_id": owner.OrgID}},
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// MigrateSettingsOrg stamps the legacy global settings singleton with the
|
||||
// default org's ID. Without it an upgrade would orphan the existing SMTP
|
||||
// config, alert config, retention setting, and ESO read token. Idempotent via
|
||||
|
||||
@@ -205,23 +205,22 @@ func DeleteMonitor(orgID, monitorID string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Only cascade when the org-scoped delete actually removed a monitor —
|
||||
// incidents/rollups carry no org_id of their own.
|
||||
// Only cascade when the org-scoped delete actually removed a monitor.
|
||||
if res.DeletedCount == 0 {
|
||||
return nil
|
||||
}
|
||||
db.Col("incidents").DeleteMany(ctx, bson.M{"monitor_id": monitorID})
|
||||
db.Col("monitor_rollups").DeleteMany(ctx, bson.M{"monitor_id": monitorID})
|
||||
db.Col("incidents").DeleteMany(ctx, bson.M{"monitor_id": monitorID, "org_id": orgID})
|
||||
db.Col("monitor_rollups").DeleteMany(ctx, bson.M{"monitor_id": monitorID, "org_id": orgID})
|
||||
return nil
|
||||
}
|
||||
|
||||
func ListIncidents(monitorID string, limit int64) ([]models.Incident, error) {
|
||||
func ListIncidents(orgID, monitorID string, limit int64) ([]models.Incident, error) {
|
||||
ctx, cancel := monCtx()
|
||||
defer cancel()
|
||||
if limit <= 0 {
|
||||
limit = 50
|
||||
}
|
||||
cur, err := db.Col("incidents").Find(ctx, bson.M{"monitor_id": monitorID},
|
||||
cur, err := db.Col("incidents").Find(ctx, bson.M{"monitor_id": monitorID, "org_id": orgID},
|
||||
options.Find().SetSort(bson.M{"started_at": -1}).SetLimit(limit))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -234,11 +233,11 @@ func ListIncidents(monitorID string, limit int64) ([]models.Incident, error) {
|
||||
}
|
||||
|
||||
// UptimeRollups returns hourly rollups for a monitor since the cutoff, oldest first.
|
||||
func UptimeRollups(monitorID string, since time.Time) ([]models.Rollup, error) {
|
||||
func UptimeRollups(orgID, monitorID string, since time.Time) ([]models.Rollup, error) {
|
||||
ctx, cancel := monCtx()
|
||||
defer cancel()
|
||||
cur, err := db.Col("monitor_rollups").Find(ctx,
|
||||
bson.M{"monitor_id": monitorID, "period_start": bson.M{"$gte": since}},
|
||||
bson.M{"monitor_id": monitorID, "org_id": orgID, "period_start": bson.M{"$gte": since}},
|
||||
options.Find().SetSort(bson.M{"period_start": 1}))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -339,9 +338,14 @@ func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
|
||||
if res.Up {
|
||||
up = 1
|
||||
}
|
||||
// org_id via $setOnInsert rather than the filter: a legacy bucket written
|
||||
// before rollups were tenanted must keep accumulating, not fork in two.
|
||||
db.Col("monitor_rollups").UpdateOne(ctx,
|
||||
bson.M{"monitor_id": monitorID, "period_start": bucket},
|
||||
bson.M{"$inc": bson.M{"checks": 1, "up_count": up, "sum_latency": int64(res.LatencyMs)}},
|
||||
bson.M{
|
||||
"$inc": bson.M{"checks": 1, "up_count": up, "sum_latency": int64(res.LatencyMs)},
|
||||
"$setOnInsert": bson.M{"org_id": m.OrgID},
|
||||
},
|
||||
options.UpdateOne().SetUpsert(true))
|
||||
|
||||
// Transition handling.
|
||||
@@ -349,6 +353,7 @@ func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
|
||||
switch newStatus {
|
||||
case models.StatusDown:
|
||||
inc := models.Incident{
|
||||
OrgID: m.OrgID,
|
||||
IncidentID: uuid.NewString(),
|
||||
MonitorID: monitorID,
|
||||
StartedAt: now,
|
||||
@@ -359,7 +364,7 @@ func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
|
||||
case models.StatusUp:
|
||||
if prev == models.StatusDown {
|
||||
db.Col("incidents").UpdateOne(ctx,
|
||||
bson.M{"monitor_id": monitorID, "resolved_at": nil},
|
||||
bson.M{"monitor_id": monitorID, "org_id": m.OrgID, "resolved_at": nil},
|
||||
bson.M{"$set": bson.M{"resolved_at": now}})
|
||||
notifyTransition(m, newStatus, res.Message)
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package services
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
@@ -95,5 +96,14 @@ func CreateOrg(name string) (*models.Org, error) {
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Boot-time seeding only covers orgs that already existed, so an org created
|
||||
// at runtime would have an empty step library until the next restart. Not
|
||||
// fatal: the org is usable without it and seeding is retried on boot.
|
||||
if created, updated, err := SeedDefaultSteps(o.OrgID); err != nil {
|
||||
log.Printf("warning: failed to seed default steps for new org %s: %v", o.OrgID, err)
|
||||
} else {
|
||||
log.Printf("default steps seeded for new org %s: %d created, %d updated", o.OrgID, created, updated)
|
||||
}
|
||||
return o, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user