mrhid6/vantage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 22 Wiki Activity

fix(server): harden per-org settings migration and sweeps

Browse Source 
Review follow-ups on e5363a6:

- MigrateSettingsOrg no longer guesses via the "default" slug. One org
  means stamp that org; zero orgs means synthesise Default; more than
  one means leave it alone and log, since guessing would hand one org
  another's SMTP config and ESO token.
- EnsureSettingsIndexes failure is now fatal. Without the unique index
  on org_id, GetSettings returns an arbitrary duplicate; without the one
  on the token hash, ResolveSecretsReadToken picks an arbitrary org.
- Name the token-hash index explicitly so it stops colliding with the
  legacy name DropOne targets, and exclude the empty string from the
  partial filter.
- Log retention: distinguish a missing run doc from a Mongo error, so a
  transient failure skips the directory rather than purging it at the
  30-day default.
- Offline sweep: fresh context per org, log-and-continue on a per-org
  error, plus a final pass for servers whose org no longer exists.
- ESO handler 401s on an empty token-derived org rather than querying
  org_id "".
This commit is contained in:
mrhid6
2026-07-22 09:41:51 +01:00
parent e5363a64ee
commit 4f512d01f1
6 changed files with 132 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/api/secrets.go
+5
View File
@@ -53,6 +53,11 @@ func esoGetGroup(c *gin.Context) {
group := c.Param("group")
// Org comes from the bearer token (set by secretsReadAuth), not a session.
orgID := c.GetString(ctxSecretsOrgKey)
if orgID == "" {
// Defence in depth: never query the store unscoped.
c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid token"})
return
}
values, err := services.GetSecretGroupDecrypted(orgID, group)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "store error"})
server/internal/services/migrate.go
+26 -6
View File
@@ -2,6 +2,7 @@ package services
import (
"context"
"log"
"time"
"github.com/google/uuid"
@@ -103,19 +104,38 @@ func MigrateSettingsOrg() error {
n, _ := db.Col("settings").CountDocuments(ctx, bson.M{"org_id": bson.M{"$exists": false}})
if n > 0 {
// The org-less settings doc belongs to whichever org already exists —
// migration 0001 only creates a "default" org when there was legacy
// data to backfill, so keying off that slug would invent a phantom org
// and move the real org's config onto it.
var org models.Org
err := db.Col("orgs").FindOne(ctx, bson.M{"slug": "default"}).Decode(&org)
orgCount, err := db.Col("orgs").CountDocuments(ctx, bson.M{})
if err != nil {
return err
}
switch orgCount {
case 1:
if err := db.Col("orgs").FindOne(ctx, bson.M{}).Decode(&org); err != nil {
return err
}
case 0:
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
}
default:
// Ambiguous: several orgs but an unstamped settings doc. Guessing
// would hand one org another's SMTP config and ESO token, so leave
// it for an operator to resolve.
log.Printf("settings org migration: %d unstamped settings doc(s) with %d orgs present; skipping", n, orgCount)
}
if _, err := db.Col("settings").UpdateMany(ctx,
bson.M{"org_id": bson.M{"$exists": false}},
bson.M{"$set": bson.M{"org_id": org.OrgID}},
); err != nil {
return err
if org.OrgID != "" {
if _, err := db.Col("settings").UpdateMany(ctx,
bson.M{"org_id": bson.M{"$exists": false}},
bson.M{"$set": bson.M{"org_id": org.OrgID}},
); err != nil {
return err
}
}
}
server/internal/services/servers.go
+70 -46
View File
@@ -6,6 +6,7 @@ import (
"crypto/sha256"
"encoding/hex"
"fmt"
"log"
"strings"
"time"
@@ -277,60 +278,83 @@ func StoreAvailableUpdates(serverID string, pkgs []models.PackageUpdate) error {
}
func MarkOfflineServers() error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
// No session here, so the sweep runs per-org and each org's threshold and
// alert config come from that org's own settings doc.
orgIDs, err := ListOrgIDs()
if err != nil {
return err
}
// No session here, so the sweep runs per-org and each org's threshold and
// alert config come from that org's own settings doc. Each org gets its own
// deadline so a slow org can't starve the ones after it, and a failure on
// one org is logged rather than aborting the whole sweep.
for _, orgID := range orgIDs {
settings, _ := GetSettings(orgID)
thresholdMinutes := 5
if settings != nil && settings.Alerts.OfflineThresholdMinutes > 0 {
thresholdMinutes = settings.Alerts.OfflineThresholdMinutes
if err := markOfflineForFilter(bson.M{"org_id": orgID}, orgID); err != nil {
log.Printf("offline sweep failed for org %s: %v", orgID, err)
}
cutoff := time.Now().Add(-time.Duration(thresholdMinutes) * time.Minute)
}
filter := bson.M{
"org_id": orgID,
"status": "active",
"last_seen": bson.M{"$lt": cutoff},
}
// Find servers about to transition to offline so we can alert on them.
cursor, err := db.Col("servers").Find(ctx, filter)
if err != nil {
return err
}
var goingOffline []models.Server
err = cursor.All(ctx, &goingOffline)
cursor.Close(ctx)
if err != nil {
return err
}
if len(goingOffline) == 0 {
continue
}
for _, s := range goingOffline {
LogEvent(s.OrgID, "server.offline", "system", s.ServerID, "", fmt.Sprintf("%s (%s) went offline", s.Hostname, s.IPAddress))
if settings != nil && settings.Alerts.Enabled && settings.Alerts.WebhookURL != "" {
go SendOfflineWebhook(settings.Alerts.WebhookURL, s.Hostname, s.ServerID, s.IPAddress)
}
if settings != nil && settings.Email.Enabled {
go SendOfflineEmail(settings.Email, s.Hostname, s.ServerID, s.IPAddress)
}
}
if _, err := db.Col("servers").UpdateMany(ctx, filter,
bson.M{"$set": bson.M{"status": "offline"}},
); err != nil {
return err
}
// Servers whose org_id matches no existing org (org deleted, or the doc
// predates the backfill) would otherwise never be swept, where the old
// global query caught them. Sweep them with the default threshold; there is
// no org settings doc to read, and no org to alert.
if err := markOfflineForFilter(bson.M{"org_id": bson.M{"$nin": orgIDs}}, ""); err != nil {
log.Printf("offline sweep failed for orphaned servers: %v", err)
}
return nil
}
// markOfflineForFilter transitions active-but-stale servers matching scope to
// offline. orgID selects whose settings supply the threshold and alert config;
// empty means defaults with no alerting (orphaned servers).
func markOfflineForFilter(scope bson.M, orgID string) error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
var settings *models.Settings
thresholdMinutes := 5
if orgID != "" {
settings, _ = GetSettings(orgID)
if settings != nil && settings.Alerts.OfflineThresholdMinutes > 0 {
thresholdMinutes = settings.Alerts.OfflineThresholdMinutes
}
}
cutoff := time.Now().Add(-time.Duration(thresholdMinutes) * time.Minute)
filter := bson.M{
"status": "active",
"last_seen": bson.M{"$lt": cutoff},
}
for k, v := range scope {
filter[k] = v
}
// Find servers about to transition to offline so we can alert on them.
cursor, err := db.Col("servers").Find(ctx, filter)
if err != nil {
return err
}
var goingOffline []models.Server
err = cursor.All(ctx, &goingOffline)
cursor.Close(ctx)
if err != nil {
return err
}
if len(goingOffline) == 0 {
return nil
}
for _, s := range goingOffline {
LogEvent(s.OrgID, "server.offline", "system", s.ServerID, "", fmt.Sprintf("%s (%s) went offline", s.Hostname, s.IPAddress))
if settings != nil && settings.Alerts.Enabled && settings.Alerts.WebhookURL != "" {
go SendOfflineWebhook(settings.Alerts.WebhookURL, s.Hostname, s.ServerID, s.IPAddress)
}
if settings != nil && settings.Email.Enabled {
go SendOfflineEmail(settings.Email, s.Hostname, s.ServerID, s.IPAddress)
}
}
_, err = db.Col("servers").UpdateMany(ctx, filter,
bson.M{"$set": bson.M{"status": "offline"}},
)
return err
}
server/internal/services/settings.go
+7 -4
View File
@@ -53,12 +53,15 @@ func EnsureSettingsIndexes() error {
}
// Partial so the many settings docs with no ESO token set don't collide on
// a missing field.
// a missing (or empty) field. Explicitly named so it does not share Mongo's
// default name with the legacy index dropped above, which would make every
// restart drop and rebuild the enforcing index.
_, err := db.Col("settings").Indexes().CreateOne(ctx, mongo.IndexModel{
Keys: bson.D{{Key: "secrets.read_token_hash", Value: 1}},
Options: options.Index().SetUnique(true).SetPartialFilterExpression(
bson.M{"secrets.read_token_hash": bson.M{"$type": "string"}},
),
Options: options.Index().SetUnique(true).SetName("settings_read_token_hash_unique").
SetPartialFilterExpression(bson.M{
"secrets.read_token_hash": bson.M{"$type": "string", "$gt": ""},
}),
})
return err
}
server/internal/services/steplogs.go
+20 -6
View File
@@ -2,6 +2,7 @@ package services
import (
"bytes"
"log"
"os"
"path/filepath"
"strings"
@@ -10,6 +11,7 @@ import (
"github.com/mrhid6/vantage/server/internal/db"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
)
// WorkflowLogDir returns the base directory for workflow step logs, creating it.
@@ -184,7 +186,14 @@ func sweepLogs() {
runID := e.Name()
dir := filepath.Join(base, runID)
orgID, finishedAt, found := runRetentionInfo(runID)
orgID, finishedAt, found, err := runRetentionInfo(runID)
if err != nil {
// A transient lookup failure is not evidence the run is gone —
// purging at the default retention here would delete logs an org
// had set to keep longer, or forever.
log.Printf("log sweep: retention lookup failed for run %s: %v", runID, err)
continue
}
if found && finishedAt == nil {
continue // still running / never finished — keep
}
@@ -220,16 +229,21 @@ func sweepLogs() {
const defaultRetentionDays = 30
// runRetentionInfo returns the owning org and finish time of a run, and whether
// the run doc still exists.
func runRetentionInfo(runID string) (string, *time.Time, bool) {
// the run doc still exists. A non-nil error means the lookup itself failed and
// says nothing about whether the run doc exists.
func runRetentionInfo(runID string) (string, *time.Time, bool, error) {
ctx, cancel := wfCtx()
defer cancel()
var run struct {
OrgID string `bson:"org_id"`
FinishedAt *time.Time `bson:"finished_at"`
}
if err := db.Col("workflow_runs").FindOne(ctx, bson.M{"run_id": runID}).Decode(&run); err != nil {
return "", nil, false
err := db.Col("workflow_runs").FindOne(ctx, bson.M{"run_id": runID}).Decode(&run)
if err == mongo.ErrNoDocuments {
return "", nil, false, nil
}
return run.OrgID, run.FinishedAt, true
if err != nil {
return "", nil, false, err
}
return run.OrgID, run.FinishedAt, true, nil
}