This commit is contained in:
@@ -41,7 +41,7 @@ func GetChannel(orgID, channelID string) (*models.NotificationChannel, error) {
|
||||
return &ch, nil
|
||||
}
|
||||
|
||||
// GetChannels loads multiple channels by ID within an org, skipping any not found.
|
||||
|
||||
func GetChannels(orgID string, channelIDs []string) ([]models.NotificationChannel, error) {
|
||||
if len(channelIDs) == 0 {
|
||||
return nil, nil
|
||||
@@ -59,8 +59,8 @@ func GetChannels(orgID string, channelIDs []string) ([]models.NotificationChanne
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// validateChannelIDs rejects any channel that does not belong to the org.
|
||||
// Channel IDs arrive from the client as data on monitor writes.
|
||||
|
||||
|
||||
func validateChannelIDs(orgID string, channelIDs []string) error {
|
||||
for _, id := range channelIDs {
|
||||
ch, err := GetChannel(orgID, id)
|
||||
@@ -103,7 +103,7 @@ func DeleteChannel(orgID, channelID string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// TestChannel sends a synthetic alert to verify configuration.
|
||||
|
||||
func TestChannel(orgID, channelID string) error {
|
||||
ch, err := GetChannel(orgID, channelID)
|
||||
if err != nil {
|
||||
|
||||
@@ -17,7 +17,7 @@ import (
|
||||
)
|
||||
|
||||
func sessionHMACKey() ([]byte, error) {
|
||||
// Reuse the AES key material as the HMAC secret. Distinct domain via prefix.
|
||||
|
||||
k, err := encryptionKey()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -29,7 +29,7 @@ func sessionHMACKey() ([]byte, error) {
|
||||
|
||||
func b64(b []byte) string { return base64.RawURLEncoding.EncodeToString(b) }
|
||||
|
||||
// SignSessionToken returns a signed, expiring token binding a session id.
|
||||
|
||||
func SignSessionToken(sessionID string, ttl time.Duration) (string, error) {
|
||||
key, err := sessionHMACKey()
|
||||
if err != nil {
|
||||
@@ -42,7 +42,7 @@ func SignSessionToken(sessionID string, ttl time.Duration) (string, error) {
|
||||
return payload + "." + b64(mac.Sum(nil)), nil
|
||||
}
|
||||
|
||||
// VerifySessionToken checks signature + expiry and returns the session id.
|
||||
|
||||
func VerifySessionToken(token string) (string, error) {
|
||||
parts := strings.Split(token, ".")
|
||||
if len(parts) != 3 {
|
||||
@@ -86,10 +86,10 @@ func portOr(v, def int) string {
|
||||
return strconv.Itoa(v)
|
||||
}
|
||||
|
||||
// BuildGuacParams assembles the guacd connection parameter map for a protocol.
|
||||
// privateKey/passphrase are the decrypted SSH private key and its optional
|
||||
// passphrase (ssh only); rdpUser/rdpPass are used for rdp, and rdpPass carries
|
||||
// the password for vnc. None of these values are persisted or logged by the caller.
|
||||
|
||||
|
||||
|
||||
|
||||
func BuildGuacParams(srv *models.Server, protocol, sshUser, privateKey, passphrase, rdpUser, rdpPass string) (*GuacParams, error) {
|
||||
host := srv.IPAddress
|
||||
switch protocol {
|
||||
@@ -159,8 +159,8 @@ func GetConsoleSession(orgID, sessionID string) (*models.ConsoleSession, error)
|
||||
return &s, nil
|
||||
}
|
||||
|
||||
// StashConsoleRDPCreds encrypts and stores single-use RDP credentials on the
|
||||
// session document. They are consumed (and cleared) when the tunnel opens.
|
||||
|
||||
|
||||
func StashConsoleRDPCreds(orgID, sessionID, username, password string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -179,9 +179,9 @@ func StashConsoleRDPCreds(orgID, sessionID, username, password string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// 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(orgID, sessionID string) (username, password string, err error) {
|
||||
s, err := GetConsoleSession(orgID, sessionID)
|
||||
if err != nil {
|
||||
@@ -209,7 +209,7 @@ func ConsumeConsoleRDPCreds(orgID, sessionID string) (username, password string,
|
||||
return username, password, nil
|
||||
}
|
||||
|
||||
// SetConsoleSSHUser persists the SSH username to use on the session doc.
|
||||
|
||||
func SetConsoleSSHUser(orgID, sessionID, username string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -219,9 +219,9 @@ func SetConsoleSSHUser(orgID, sessionID, username string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// 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(orgID, sessionID string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -22,8 +22,8 @@ func encryptionKey() ([]byte, error) {
|
||||
return key, nil
|
||||
}
|
||||
|
||||
// encryptString encrypts a plaintext value with AES-256-GCM using the
|
||||
// shared KEY_ENCRYPTION_KEY, returning hex(nonce + ciphertext).
|
||||
|
||||
|
||||
func encryptString(plaintext string) (string, error) {
|
||||
key, err := encryptionKey()
|
||||
if err != nil {
|
||||
@@ -45,7 +45,7 @@ func encryptString(plaintext string) (string, error) {
|
||||
return hex.EncodeToString(sealed), nil
|
||||
}
|
||||
|
||||
// decryptString reverses encryptString.
|
||||
|
||||
func decryptString(ciphertextHex string) (string, error) {
|
||||
key, err := encryptionKey()
|
||||
if err != nil {
|
||||
|
||||
@@ -12,7 +12,7 @@ import (
|
||||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||||
)
|
||||
|
||||
// DefaultStepsDir returns the directory holding default step JSON files.
|
||||
|
||||
func DefaultStepsDir() string {
|
||||
dir := os.Getenv("VANTAGE_DEFAULT_STEPS_DIR")
|
||||
if dir == "" {
|
||||
@@ -22,9 +22,9 @@ func DefaultStepsDir() string {
|
||||
return dir
|
||||
}
|
||||
|
||||
// readDefaultStepFiles parses every *.json in the defaults dir into
|
||||
// source=default library steps (with slug set). Non-json and invalid files are
|
||||
// skipped silently; a slug is derived from the step name.
|
||||
|
||||
|
||||
|
||||
func readDefaultStepFiles() ([]models.WorkflowStep, error) {
|
||||
matches, err := filepath.Glob(filepath.Join(DefaultStepsDir(), "*.json"))
|
||||
if err != nil {
|
||||
@@ -50,8 +50,8 @@ func readDefaultStepFiles() ([]models.WorkflowStep, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// SeedDefaultSteps upserts default steps from disk keyed on {slug, source}.
|
||||
// Re-sync overwrites default-step content; user steps are never touched.
|
||||
|
||||
|
||||
func SeedDefaultSteps(orgID string) (created, updated int, err error) {
|
||||
steps, err := readDefaultStepFiles()
|
||||
if err != nil {
|
||||
|
||||
@@ -17,13 +17,13 @@ type commandDispatcher struct {
|
||||
channels map[string]chan *pb.ServerCommand
|
||||
}
|
||||
|
||||
// Dispatcher is the singleton command dispatcher used by both the gRPC server
|
||||
// and the REST API to push commands to connected agents.
|
||||
|
||||
|
||||
var Dispatcher = &commandDispatcher{
|
||||
channels: make(map[string]chan *pb.ServerCommand),
|
||||
}
|
||||
|
||||
// Connect registers an agent's command channel. Returns the channel to drain.
|
||||
|
||||
func (d *commandDispatcher) Connect(serverID string) chan *pb.ServerCommand {
|
||||
ch := make(chan *pb.ServerCommand, 16)
|
||||
d.mu.Lock()
|
||||
@@ -32,14 +32,14 @@ func (d *commandDispatcher) Connect(serverID string) chan *pb.ServerCommand {
|
||||
return ch
|
||||
}
|
||||
|
||||
// Disconnect removes the agent's channel on stream close.
|
||||
|
||||
func (d *commandDispatcher) Disconnect(serverID string) {
|
||||
d.mu.Lock()
|
||||
delete(d.channels, serverID)
|
||||
d.mu.Unlock()
|
||||
}
|
||||
|
||||
// IsConnected reports whether an agent is currently holding a CommandStream.
|
||||
|
||||
func (d *commandDispatcher) IsConnected(serverID string) bool {
|
||||
d.mu.RLock()
|
||||
_, ok := d.channels[serverID]
|
||||
@@ -62,15 +62,15 @@ func (d *commandDispatcher) dispatch(serverID string, cmd *pb.ServerCommand) err
|
||||
}
|
||||
}
|
||||
|
||||
// DispatchRunStep pushes a RunStepCmd to a server's agent. Caller must have
|
||||
// registered StepResults.Await(commandID) first.
|
||||
|
||||
|
||||
func DispatchRunStep(serverID, commandID string, cmd *pb.RunStepCmd) error {
|
||||
return Dispatcher.dispatch(serverID, &pb.ServerCommand{CommandId: commandID, RunStep: cmd})
|
||||
}
|
||||
|
||||
// DispatchCleanupWorkspace tells a server's agent to remove a run's working
|
||||
// directory. Best-effort and fire-and-forget: if the agent is gone the temp dir
|
||||
// is reclaimed by the OS on reboot anyway.
|
||||
|
||||
|
||||
|
||||
func DispatchCleanupWorkspace(serverID, workspaceID string) {
|
||||
if !Dispatcher.IsConnected(serverID) {
|
||||
return
|
||||
@@ -81,7 +81,7 @@ func DispatchCleanupWorkspace(serverID, workspaceID string) {
|
||||
})
|
||||
}
|
||||
|
||||
// KeyGenParams carries all options for a generate-key command.
|
||||
|
||||
type KeyGenParams struct {
|
||||
Label string
|
||||
KeyType string
|
||||
@@ -90,15 +90,15 @@ type KeyGenParams struct {
|
||||
Comment string
|
||||
}
|
||||
|
||||
// GetLatestAgentVersion queries the Gitea API for the latest agent/v* release tag
|
||||
// and returns just the version number (e.g. "1.2.3").
|
||||
|
||||
|
||||
func GetLatestAgentVersion() (string, error) {
|
||||
giteaHost := os.Getenv("GITEA_HOST")
|
||||
if giteaHost == "" {
|
||||
giteaHost = "gitea.example.com"
|
||||
}
|
||||
url := fmt.Sprintf("https://%s/api/v1/repos/mrhid6/vantage/releases?limit=20", giteaHost)
|
||||
resp, err := http.Get(url) //nolint:gosec
|
||||
resp, err := http.Get(url)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("fetch releases: %w", err)
|
||||
}
|
||||
@@ -122,8 +122,8 @@ func GetLatestAgentVersion() (string, error) {
|
||||
return "", fmt.Errorf("no agent release found")
|
||||
}
|
||||
|
||||
// DispatchUpdateAgent sends an update command to the named server's agent.
|
||||
// It fetches the latest version from Gitea and includes the download base URL.
|
||||
|
||||
|
||||
func DispatchUpdateAgent(serverID string) (string, error) {
|
||||
if !Dispatcher.IsConnected(serverID) {
|
||||
return "", fmt.Errorf("agent is not connected to the command stream")
|
||||
@@ -153,7 +153,7 @@ func DispatchUpdateAgent(serverID string) (string, error) {
|
||||
return version, nil
|
||||
}
|
||||
|
||||
// DispatchApplyUpdates sends an apply-updates command to the named server's agent.
|
||||
|
||||
func DispatchApplyUpdates(serverID string) error {
|
||||
if !Dispatcher.IsConnected(serverID) {
|
||||
return fmt.Errorf("agent is not connected to the command stream")
|
||||
@@ -165,8 +165,8 @@ func DispatchApplyUpdates(serverID string) error {
|
||||
return Dispatcher.dispatch(serverID, cmd)
|
||||
}
|
||||
|
||||
// DispatchDeleteKey sends a delete-key command to the named server's agent.
|
||||
// It is best-effort: if the agent is offline the local files will remain until next connection.
|
||||
|
||||
|
||||
func DispatchDeleteKey(serverID, label string) {
|
||||
if !Dispatcher.IsConnected(serverID) {
|
||||
return
|
||||
@@ -176,13 +176,13 @@ func DispatchDeleteKey(serverID, label string) {
|
||||
DeleteKey: &pb.DeleteKeyCmd{Label: label},
|
||||
}
|
||||
if err := Dispatcher.dispatch(serverID, cmd); err != nil {
|
||||
// Non-fatal: agent will clean up files on next manual intervention or reinstall.
|
||||
|
||||
_ = err
|
||||
}
|
||||
}
|
||||
|
||||
// DispatchGenerateKey sends a generate-key command to the named server's agent.
|
||||
// Returns the command ID that can be used to correlate the agent's result.
|
||||
|
||||
|
||||
func DispatchGenerateKey(serverID string, p KeyGenParams) (string, error) {
|
||||
if !Dispatcher.IsConnected(serverID) {
|
||||
return "", fmt.Errorf("agent is not connected to the command stream")
|
||||
|
||||
@@ -9,8 +9,8 @@ import (
|
||||
"go.mongodb.org/mongo-driver/v2/bson"
|
||||
)
|
||||
|
||||
// StoreInventory upserts the latest inventory snapshot onto the server document.
|
||||
// Metrics fields update every call; static fields only when r.IncludeStatic.
|
||||
|
||||
|
||||
func StoreInventory(serverID string, r *pb.InventoryReport) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -99,9 +99,6 @@ func GetPrivateKey(orgID, keyID string) (string, error) {
|
||||
return decryptPrivateKey(key.PrivateKeyEncrypted)
|
||||
}
|
||||
|
||||
// GetPassphrase returns the decrypted passphrase for a key, or an empty string
|
||||
// if the key has none stored. Agent-path (keyed by unique key_id from an
|
||||
// assignment lookup) — no org filter.
|
||||
func GetPassphrase(keyID string) (string, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -175,8 +172,6 @@ func AssignKey(orgID, keyID, serverID string) (*models.Assignment, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Both sides must belong to the caller's org — the IDs arrive from the
|
||||
// client as data and are consumed by unscoped agent-path queries later.
|
||||
if _, err := GetKey(orgID, keyID); err != nil {
|
||||
return nil, fmt.Errorf("key not found")
|
||||
}
|
||||
@@ -184,7 +179,7 @@ func AssignKey(orgID, keyID, serverID string) (*models.Assignment, error) {
|
||||
return nil, fmt.Errorf("server not found")
|
||||
}
|
||||
|
||||
// Check if already assigned and active
|
||||
|
||||
var existing models.Assignment
|
||||
err := db.Col("assignments").FindOne(ctx, bson.M{
|
||||
"org_id": orgID,
|
||||
|
||||
@@ -21,7 +21,6 @@ var scopedCollections = []string{
|
||||
"console_sessions", "incidents", "monitor_rollups",
|
||||
}
|
||||
|
||||
// EnsureAuthIndexes creates unique indexes for the new auth collections.
|
||||
func EnsureAuthIndexes() error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
@@ -47,18 +46,12 @@ 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
|
||||
err := db.Col("orgs").FindOne(ctx, bson.M{"slug": "default"}).Decode(&org)
|
||||
switch {
|
||||
case err == nil:
|
||||
case errors.Is(err, mongo.ErrNoDocuments):
|
||||
// Only a genuine absence justifies an insert. Treating a timeout or a
|
||||
// decode failure as "absent" would race the fatal unique orgs.slug index
|
||||
// and turn a transient blip into a boot crash.
|
||||
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
|
||||
@@ -69,8 +62,6 @@ func defaultBackfillOrg(ctx context.Context) (*models.Org, error) {
|
||||
return &org, nil
|
||||
}
|
||||
|
||||
// RunMigrations backfills a default org onto pre-existing documents. Idempotent
|
||||
// via a marker in the migrations collection.
|
||||
func RunMigrations() error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
@@ -80,7 +71,7 @@ func RunMigrations() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Only backfill if there is legacy data lacking org_id.
|
||||
|
||||
needs := false
|
||||
for _, col := range scopedCollections {
|
||||
n, _ := db.Col(col).CountDocuments(ctx, bson.M{"org_id": bson.M{"$exists": false}})
|
||||
@@ -109,18 +100,6 @@ 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()
|
||||
@@ -130,7 +109,7 @@ func MigrateMissedOrgScopes() error {
|
||||
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 {
|
||||
@@ -155,8 +134,6 @@ func MigrateMissedOrgScopes() error {
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
@@ -171,13 +148,8 @@ func MigrateMissedOrgScopes() error {
|
||||
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 {
|
||||
// Decoded loosely: a single null or non-string value in the collection would
|
||||
// fail a []string decode and abort the migration — and therefore boot — over
|
||||
// one unusable document. Skip what we cannot use instead.
|
||||
|
||||
var raw []bson.RawValue
|
||||
if err := db.Col(col).Distinct(ctx, localField,
|
||||
bson.M{"org_id": bson.M{"$exists": false}}).Decode(&raw); err != nil {
|
||||
@@ -207,10 +179,6 @@ func backfillOrgFromOwner(ctx context.Context, col, localField, ownerCol, ownerF
|
||||
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
|
||||
// a marker in the migrations collection.
|
||||
func MigrateSettingsOrg() error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
@@ -222,10 +190,6 @@ 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
|
||||
orgCount, err := db.Col("orgs").CountDocuments(ctx, bson.M{})
|
||||
if err != nil {
|
||||
@@ -242,12 +206,6 @@ func MigrateSettingsOrg() error {
|
||||
return err
|
||||
}
|
||||
default:
|
||||
// Ambiguous: several orgs but an unstamped settings doc. Guessing
|
||||
// would hand one org another's SMTP config and ESO token. Continuing
|
||||
// is not an option either: the unique settings.org_id index built
|
||||
// straight after this indexes every unstamped doc as null, so two or
|
||||
// more of them collide and boot fails there instead — with a far less
|
||||
// useful message. Stop here, where we can name the remedy.
|
||||
return fmt.Errorf(
|
||||
"settings org migration: %d settings document(s) have no org_id but %d orgs exist; "+
|
||||
"cannot infer the owner. Set org_id manually on each settings document "+
|
||||
|
||||
@@ -21,7 +21,7 @@ func monCtx() (context.Context, context.CancelFunc) {
|
||||
return context.WithTimeout(context.Background(), 5*time.Second)
|
||||
}
|
||||
|
||||
// SpecFor maps a monitor onto a checker.Spec.
|
||||
|
||||
func SpecFor(m *models.Monitor) checker.Spec {
|
||||
return checker.Spec{
|
||||
Type: m.Type,
|
||||
@@ -51,12 +51,6 @@ func ListMonitors(orgID string) ([]models.Monitor, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ListMonitorsForRunner returns enabled monitors whose Runner matches runner,
|
||||
// scoped to orgID. Runner is client-supplied at write time, so an agent fetching
|
||||
// its own work must scope by the org of its authenticated server record —
|
||||
// otherwise another org could point a monitor at that server_id and have it run
|
||||
// their checks. An empty orgID is rejected: it would silently widen the query to
|
||||
// every org.
|
||||
func ListMonitorsForRunner(orgID, runner string) ([]models.Monitor, error) {
|
||||
if orgID == "" {
|
||||
return nil, errors.New("org id required")
|
||||
@@ -64,16 +58,10 @@ func ListMonitorsForRunner(orgID, runner string) ([]models.Monitor, error) {
|
||||
return listMonitorsForRunner(orgID, runner)
|
||||
}
|
||||
|
||||
// ListServerScheduledMonitors returns every enabled server-run monitor across
|
||||
// all orgs. This is the in-process scheduler's entry point (mirrors the cross-org
|
||||
// MarkOfflineServers sweep) and must never be called from a request-driven path —
|
||||
// it performs no org scoping at all.
|
||||
func ListServerScheduledMonitors() ([]models.Monitor, error) {
|
||||
return listMonitorsForRunner("", models.RunnerServer)
|
||||
}
|
||||
|
||||
// listMonitorsForRunner is the shared query. An empty orgID means no org filter
|
||||
// and is only reachable via ListServerScheduledMonitors.
|
||||
func listMonitorsForRunner(orgID, runner string) ([]models.Monitor, error) {
|
||||
ctx, cancel := monCtx()
|
||||
defer cancel()
|
||||
@@ -92,7 +80,7 @@ func listMonitorsForRunner(orgID, runner string) ([]models.Monitor, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// GetMonitor looks up a monitor scoped to an org (handler/session use).
|
||||
|
||||
func GetMonitor(orgID, monitorID string) (*models.Monitor, error) {
|
||||
ctx, cancel := monCtx()
|
||||
defer cancel()
|
||||
@@ -107,8 +95,6 @@ func GetMonitor(orgID, monitorID string) (*models.Monitor, error) {
|
||||
return &m, nil
|
||||
}
|
||||
|
||||
// getMonitorByID looks up a monitor by its unique monitor_id with no org
|
||||
// filter. For agent/scheduler use only (IngestResult), which has no session.
|
||||
func getMonitorByID(monitorID string) (*models.Monitor, error) {
|
||||
ctx, cancel := monCtx()
|
||||
defer cancel()
|
||||
@@ -123,10 +109,6 @@ func getMonitorByID(monitorID string) (*models.Monitor, error) {
|
||||
return &m, nil
|
||||
}
|
||||
|
||||
// validateRunner rejects a runner that is neither the reserved server-scheduler
|
||||
// value nor a server in the org. The value is client-supplied and is later
|
||||
// consumed by an agent's own monitor fetch, so ownership has to be proven at
|
||||
// the write boundary.
|
||||
func validateRunner(orgID, runner string) error {
|
||||
if runner == "" || runner == models.RunnerServer {
|
||||
return nil
|
||||
@@ -168,8 +150,8 @@ func CreateMonitor(orgID string, m *models.Monitor) (*models.Monitor, error) {
|
||||
func UpdateMonitor(orgID, monitorID string, upd bson.M) error {
|
||||
ctx, cancel := monCtx()
|
||||
defer cancel()
|
||||
// A present-but-wrong-type value is a hard error: silently skipping the
|
||||
// check would still let the unvalidated value through to the $set.
|
||||
|
||||
|
||||
if raw, present := upd["channel_ids"]; present {
|
||||
ids, ok := raw.([]string)
|
||||
if !ok {
|
||||
@@ -187,9 +169,9 @@ func UpdateMonitor(orgID, monitorID string, upd bson.M) error {
|
||||
if err := validateRunner(orgID, runner); err != nil {
|
||||
return err
|
||||
}
|
||||
// Match CreateMonitor: an empty runner means the server scheduler.
|
||||
// Storing "" would match no runner at all and silently stop the
|
||||
// monitor being checked.
|
||||
|
||||
|
||||
|
||||
if runner == "" {
|
||||
upd["runner"] = models.RunnerServer
|
||||
}
|
||||
@@ -205,7 +187,7 @@ func DeleteMonitor(orgID, monitorID string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Only cascade when the org-scoped delete actually removed a monitor.
|
||||
|
||||
if res.DeletedCount == 0 {
|
||||
return nil
|
||||
}
|
||||
@@ -232,7 +214,7 @@ func ListIncidents(orgID, monitorID string, limit int64) ([]models.Incident, err
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// UptimeRollups returns hourly rollups for a monitor since the cutoff, oldest first.
|
||||
|
||||
func UptimeRollups(orgID, monitorID string, since time.Time) ([]models.Rollup, error) {
|
||||
ctx, cancel := monCtx()
|
||||
defer cancel()
|
||||
@@ -249,16 +231,6 @@ func UptimeRollups(orgID, monitorID string, since time.Time) ([]models.Rollup, e
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// IngestResult applies a check result to a monitor: updates state, opens/resolves
|
||||
// incidents on up<->down transitions, rolls up the hourly bucket, and fires
|
||||
// notifications on transition. Both the server scheduler and agent-reported
|
||||
// results funnel through here.
|
||||
//
|
||||
// monitorID is client-supplied on the agent path, so the caller passes the org
|
||||
// and runner it is authenticated as: orgID is the reporting agent's server org
|
||||
// and runner is its server_id. A result is only applied to a monitor owned by
|
||||
// that org and assigned to that runner. An empty orgID is rejected — it would
|
||||
// skip the ownership check entirely.
|
||||
func IngestResult(orgID, runner, monitorID string, res checker.Result) error {
|
||||
if orgID == "" {
|
||||
return errors.New("org id required")
|
||||
@@ -266,16 +238,10 @@ func IngestResult(orgID, runner, monitorID string, res checker.Result) error {
|
||||
return ingestResult(orgID, runner, monitorID, res)
|
||||
}
|
||||
|
||||
// IngestServerScheduledResult applies a result produced by the in-process server
|
||||
// scheduler, which has no org context of its own. This is the scheduler's entry
|
||||
// point and must never be called from a request-driven path — it skips the org
|
||||
// ownership check.
|
||||
func IngestServerScheduledResult(monitorID string, res checker.Result) error {
|
||||
return ingestResult("", models.RunnerServer, monitorID, res)
|
||||
}
|
||||
|
||||
// ingestResult is the shared implementation. An empty orgID skips the org
|
||||
// ownership check and is only reachable via IngestServerScheduledResult.
|
||||
func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
|
||||
ctx, cancel := monCtx()
|
||||
defer cancel()
|
||||
@@ -284,9 +250,6 @@ func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Report not-found the same way as a cross-org hit, so probing an unknown
|
||||
// monitor_id is no quieter than probing a foreign one and stale monitors
|
||||
// stay visible to operators.
|
||||
if m == nil {
|
||||
return fmt.Errorf("monitor %s not found", monitorID)
|
||||
}
|
||||
@@ -332,14 +295,14 @@ func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// Hourly rollup.
|
||||
|
||||
bucket := now.Truncate(time.Hour)
|
||||
up := 0
|
||||
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{
|
||||
@@ -348,7 +311,7 @@ func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
|
||||
},
|
||||
options.UpdateOne().SetUpsert(true))
|
||||
|
||||
// Transition handling.
|
||||
|
||||
if newStatus != prev {
|
||||
switch newStatus {
|
||||
case models.StatusDown:
|
||||
@@ -373,9 +336,9 @@ func ingestResult(orgID, runner, monitorID string, res checker.Result) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// notifyTransition dispatches notifications on an up<->down transition to each
|
||||
// enabled channel bound to the monitor. Deliveries run in the background;
|
||||
// failures are logged, not fatal.
|
||||
|
||||
|
||||
|
||||
func notifyTransition(m *models.Monitor, newStatus, message string) {
|
||||
if len(m.ChannelIDs) == 0 {
|
||||
return
|
||||
|
||||
@@ -29,8 +29,8 @@ func GetOrgOIDCSecret(orgID string) (string, error) {
|
||||
return decryptString(o.ClientSecretEnc)
|
||||
}
|
||||
|
||||
// SaveOrgOIDC upserts the org's provider config. An empty clientSecret keeps the
|
||||
// stored secret (so the UI need not resend it).
|
||||
|
||||
|
||||
func SaveOrgOIDC(orgID, issuer, clientID, clientSecret string, enabled bool) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -40,8 +40,8 @@ func GetOrgBySlug(slug string) (*models.Org, error) {
|
||||
return &o, nil
|
||||
}
|
||||
|
||||
// ListOrgIDs returns the org_id of every organization. Used by startup tasks
|
||||
// (e.g. seeding default workflow steps) that must run once per org.
|
||||
|
||||
|
||||
func ListOrgIDs() ([]string, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -61,17 +61,15 @@ func ListOrgIDs() ([]string, error) {
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
// CountOrgs returns the number of organizations on the instance. Used by
|
||||
// first-run bootstrap to tell "empty instance" from "upgraded single-tenant
|
||||
// instance whose data already sits under a migration-created org".
|
||||
|
||||
|
||||
|
||||
func CountOrgs() (int64, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
return db.Col("orgs").CountDocuments(ctx, bson.M{})
|
||||
}
|
||||
|
||||
// FirstOrg returns the sole/earliest org. Callers must have established that
|
||||
// exactly one exists before treating it as authoritative.
|
||||
func FirstOrg() (*models.Org, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -82,15 +80,6 @@ func FirstOrg() (*models.Org, error) {
|
||||
return &o, nil
|
||||
}
|
||||
|
||||
// AdoptOrg renames an existing org to name, re-slugging it when the new slug is
|
||||
// clean to take. It exists for the upgrade path: migration 0001 stamps every
|
||||
// legacy document with the "default" org's ID, so bootstrap must claim that org
|
||||
// rather than mint a second one — otherwise the operator signs in to an empty
|
||||
// instance while all their servers and keys stay behind under "default".
|
||||
//
|
||||
// The slug is only changed when the derived one is usable and free; anything
|
||||
// else keeps the current slug, including the reserved "default", which stays
|
||||
// valid because it is pre-existing rather than newly chosen.
|
||||
func AdoptOrg(orgID, name string) (*models.Org, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -135,7 +124,7 @@ func CreateOrg(name string) (*models.Org, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Resolve slug collision by suffixing -2, -3, ...
|
||||
|
||||
slug := base
|
||||
for i := 2; ; i++ {
|
||||
n, err := db.Col("orgs").CountDocuments(ctx, bson.M{"slug": slug})
|
||||
@@ -156,9 +145,9 @@ 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 {
|
||||
|
||||
@@ -14,9 +14,6 @@ import (
|
||||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||||
)
|
||||
|
||||
// EnsureSecretIndexes creates the unique compound index on (org_id, group, key).
|
||||
// The pre-multi-tenant index was on (group, key) alone, which made a second org
|
||||
// collide on the same group/key — drop it if a live DB still carries it.
|
||||
func EnsureSecretIndexes() error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
@@ -32,12 +29,6 @@ func EnsureSecretIndexes() error {
|
||||
return err
|
||||
}
|
||||
|
||||
// isIndexNotFound reports whether err is Mongo's IndexNotFound (27), returned
|
||||
// when dropping an index that was never created, or NamespaceNotFound (26),
|
||||
// returned when the collection itself does not exist yet. Both mean "there is
|
||||
// no legacy index to drop" — on a fresh install nothing has written to these
|
||||
// collections, so the drop must be tolerated or the index creation that follows
|
||||
// it never runs and a brand-new deployment crash-loops at startup.
|
||||
func isIndexNotFound(err error) bool {
|
||||
var ce mongo.CommandError
|
||||
if errors.As(err, &ce) {
|
||||
@@ -47,8 +38,6 @@ func isIndexNotFound(err error) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// ListSecretGroups returns a summary of every group with its key count and
|
||||
// most recent update time.
|
||||
func ListSecretGroups(orgID string) ([]models.GroupSummary, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -89,8 +78,6 @@ func ListSecretGroups(orgID string) ([]models.GroupSummary, error) {
|
||||
return groups, nil
|
||||
}
|
||||
|
||||
// GetSecretGroup returns the keys within a group, sorted by key name, without
|
||||
// decrypted values.
|
||||
func GetSecretGroup(orgID, group string) ([]models.Secret, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -109,9 +96,6 @@ func GetSecretGroup(orgID, group string) ([]models.Secret, error) {
|
||||
return docs, nil
|
||||
}
|
||||
|
||||
// GetSecretGroupDecrypted returns a flat map of key → plaintext value for a
|
||||
// group. Also used by the ESO read endpoint, which resolves its org from the
|
||||
// per-org bearer token rather than from a session.
|
||||
func GetSecretGroupDecrypted(orgID, group string) (map[string]string, error) {
|
||||
docs, err := GetSecretGroup(orgID, group)
|
||||
if err != nil {
|
||||
@@ -128,7 +112,6 @@ func GetSecretGroupDecrypted(orgID, group string) (map[string]string, error) {
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// RevealSecret returns the decrypted value of a single key.
|
||||
func RevealSecret(orgID, group, key string) (string, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -144,7 +127,6 @@ func RevealSecret(orgID, group, key string) (string, error) {
|
||||
return decryptString(doc.EncryptedValue)
|
||||
}
|
||||
|
||||
// UpsertSecrets encrypts and writes each key/value pair into the group.
|
||||
func UpsertSecrets(orgID, group string, values map[string]string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
@@ -170,7 +152,7 @@ func UpsertSecrets(orgID, group string, values map[string]string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// SortedKeys returns the map keys sorted — handy for stable audit messages.
|
||||
|
||||
func SortedKeys(m map[string]string) []string {
|
||||
keys := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
@@ -180,7 +162,7 @@ func SortedKeys(m map[string]string) []string {
|
||||
return keys
|
||||
}
|
||||
|
||||
// DeleteSecret removes a single key from a group.
|
||||
|
||||
func DeleteSecret(orgID, group, key string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -189,7 +171,7 @@ func DeleteSecret(orgID, group, key string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// DeleteSecretGroup removes an entire group and all its keys.
|
||||
|
||||
func DeleteSecretGroup(orgID, group string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -54,7 +54,7 @@ func CreateServer(orgID string) (*models.Server, string, error) {
|
||||
return s, token, nil
|
||||
}
|
||||
|
||||
// GetServer looks up a server scoped to an org (handler/session use).
|
||||
|
||||
func GetServer(orgID, serverID string) (*models.Server, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -67,8 +67,8 @@ func GetServer(orgID, serverID string) (*models.Server, error) {
|
||||
return &s, nil
|
||||
}
|
||||
|
||||
// getServerByID looks up a server by its unique server_id with no org filter.
|
||||
// For agent/internal use only (e.g. workflow runner resolving org from a run).
|
||||
|
||||
|
||||
func getServerByID(serverID string) (*models.Server, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -96,9 +96,9 @@ func GetServerByPreRegToken(token string) (*models.Server, error) {
|
||||
return &s, nil
|
||||
}
|
||||
|
||||
// OSTypeFromInfo derives a coarse os_type ("windows" or "linux") from the
|
||||
// agent-reported os_info string, which is formatted "<GOOS> <GOARCH>".
|
||||
// Anything that is not explicitly windows defaults to linux.
|
||||
|
||||
|
||||
|
||||
func OSTypeFromInfo(osInfo string) string {
|
||||
if strings.HasPrefix(strings.ToLower(osInfo), "windows") {
|
||||
return "windows"
|
||||
@@ -106,8 +106,8 @@ func OSTypeFromInfo(osInfo string) string {
|
||||
return "linux"
|
||||
}
|
||||
|
||||
// defaultConsoleFields returns the initial console configuration for a newly
|
||||
// registered server based on its os_type.
|
||||
|
||||
|
||||
func defaultConsoleFields(osType string) (protocols []string, sshPort, rdpPort int) {
|
||||
if osType == "windows" {
|
||||
return []string{"rdp"}, 22, 3389
|
||||
@@ -181,19 +181,19 @@ func ValidateAgentToken(serverID, agentToken string) (*models.Server, error) {
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid agent token")
|
||||
}
|
||||
// Defence in depth: every agent-path caller scopes its work by this OrgID,
|
||||
// so a blank one would widen those queries instead of narrowing them.
|
||||
|
||||
|
||||
if s.OrgID == "" {
|
||||
return nil, fmt.Errorf("server %s has no org", serverID)
|
||||
}
|
||||
return &s, nil
|
||||
}
|
||||
|
||||
// BackfillConsoleConfig sets default console_protocols/ports for a server that
|
||||
// predates the console feature (or was updated without re-registering). Servers
|
||||
// register only once via a single-use pre_reg_token, so Register() never runs
|
||||
// again to populate these fields — this runs on every sync as a cheap no-op
|
||||
// once the fields are present.
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
func BackfillConsoleConfig(srv *models.Server) error {
|
||||
if srv == nil || len(srv.ConsoleProtocols) > 0 {
|
||||
return nil
|
||||
@@ -262,7 +262,7 @@ func DeleteServer(orgID, serverID string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Also remove assignments
|
||||
|
||||
_, err = db.Col("assignments").DeleteMany(ctx, bson.M{"server_id": serverID, "org_id": orgID})
|
||||
return err
|
||||
}
|
||||
@@ -288,29 +288,29 @@ func MarkOfflineServers() error {
|
||||
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 {
|
||||
if err := markOfflineForFilter(bson.M{"org_id": orgID}, orgID); err != nil {
|
||||
log.Printf("offline sweep failed for org %s: %v", orgID, 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()
|
||||
@@ -333,7 +333,7 @@ func markOfflineForFilter(scope bson.M, orgID string) error {
|
||||
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
|
||||
|
||||
@@ -34,9 +34,9 @@ var defaultSettings = models.Settings{
|
||||
},
|
||||
}
|
||||
|
||||
// EnsureSettingsIndexes creates the per-org uniqueness constraints on settings.
|
||||
// Pre-multi-tenant deployments had a single global settings doc and no indexes;
|
||||
// drop any legacy index if a live DB still carries one.
|
||||
|
||||
|
||||
|
||||
func EnsureSettingsIndexes() error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
@@ -52,10 +52,10 @@ func EnsureSettingsIndexes() error {
|
||||
return err
|
||||
}
|
||||
|
||||
// Partial so the many settings docs with no ESO token set don't collide on
|
||||
// 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).SetName("settings_read_token_hash_unique").
|
||||
@@ -89,8 +89,8 @@ func hashToken(token string) string {
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// RotateSecretsReadToken generates a new ESO read token, stores its SHA-256
|
||||
// hash, and returns the plaintext token exactly once.
|
||||
|
||||
|
||||
func RotateSecretsReadToken(orgID string) (string, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -118,9 +118,9 @@ func RotateSecretsReadToken(orgID string) (string, error) {
|
||||
return token, nil
|
||||
}
|
||||
|
||||
// ResolveSecretsReadToken looks the presented token's hash up directly and
|
||||
// returns the owning org. This is the ESO machine-to-machine path: the org is
|
||||
// carried by the token itself, since there is no session to scope it.
|
||||
|
||||
|
||||
|
||||
func ResolveSecretsReadToken(token string) (string, bool) {
|
||||
if token == "" {
|
||||
return "", false
|
||||
@@ -167,8 +167,8 @@ func SaveSettings(orgID string, alerts models.AlertSettings, email models.EmailS
|
||||
return err
|
||||
}
|
||||
|
||||
// GetWorkflowLogRetentionDays returns the log retention in days: 30 when unset,
|
||||
// 0 for keep-forever, or the configured value.
|
||||
|
||||
|
||||
func GetWorkflowLogRetentionDays(orgID string) (int, error) {
|
||||
s, err := GetSettings(orgID)
|
||||
if err != nil {
|
||||
@@ -241,7 +241,7 @@ func SendOfflineEmail(cfg models.EmailSettings, hostname, serverID, ipAddress st
|
||||
}
|
||||
}
|
||||
|
||||
// sendMailTLS dials with implicit TLS (port 465) instead of STARTTLS.
|
||||
|
||||
func sendMailTLS(addr, host string, auth smtp.Auth, from string, to []string, msg []byte) error {
|
||||
conn, err := tls.Dial("tcp", addr, &tls.Config{ServerName: host})
|
||||
if err != nil {
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
|
||||
const StepDocKind = "vantage.step/v1"
|
||||
|
||||
// StepDoc is the portable, id-free representation of a step.
|
||||
|
||||
type StepDoc struct {
|
||||
Kind string `json:"kind"`
|
||||
Name string `json:"name"`
|
||||
@@ -21,7 +21,7 @@ type StepDoc struct {
|
||||
SecretRefs []string `json:"secret_refs"`
|
||||
}
|
||||
|
||||
// ExportStepDoc builds a portable doc from a library step (ids/source stripped).
|
||||
|
||||
func ExportStepDoc(s models.WorkflowStep) StepDoc {
|
||||
return StepDoc{
|
||||
Kind: StepDocKind,
|
||||
@@ -35,8 +35,8 @@ func ExportStepDoc(s models.WorkflowStep) StepDoc {
|
||||
}
|
||||
}
|
||||
|
||||
// ParseStepDoc validates a v1 doc and returns a normalized (id-free) step with
|
||||
// declared_outputs recomputed from the script.
|
||||
|
||||
|
||||
func ParseStepDoc(b []byte) (models.WorkflowStep, error) {
|
||||
var d StepDoc
|
||||
if err := json.Unmarshal(b, &d); err != nil {
|
||||
@@ -65,7 +65,7 @@ func ParseStepDoc(b []byte) (models.WorkflowStep, error) {
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ImportStepToLibrary parses a doc and persists it as a new user library step.
|
||||
|
||||
func ImportStepToLibrary(orgID string, b []byte) (*models.WorkflowStep, error) {
|
||||
s, err := ParseStepDoc(b)
|
||||
if err != nil {
|
||||
@@ -74,7 +74,7 @@ func ImportStepToLibrary(orgID string, b []byte) (*models.WorkflowStep, error) {
|
||||
return CreateStep(orgID, s)
|
||||
}
|
||||
|
||||
// ExportStep loads a library step and marshals it to a portable doc.
|
||||
|
||||
func ExportStep(orgID, stepID string) ([]byte, error) {
|
||||
ctx, cancel := wfCtx()
|
||||
defer cancel()
|
||||
|
||||
@@ -14,7 +14,7 @@ import (
|
||||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||||
)
|
||||
|
||||
// WorkflowLogDir returns the base directory for workflow step logs, creating it.
|
||||
|
||||
func WorkflowLogDir() string {
|
||||
dir := os.Getenv("VANTAGE_WORKFLOW_LOG_DIR")
|
||||
if dir == "" {
|
||||
@@ -24,19 +24,19 @@ func WorkflowLogDir() string {
|
||||
return dir
|
||||
}
|
||||
|
||||
// ServerRunLogPath is the per-server-run log file path.
|
||||
|
||||
func ServerRunLogPath(runID, serverID string) string {
|
||||
return filepath.Join(WorkflowLogDir(), runID, serverID+".log")
|
||||
}
|
||||
|
||||
// logTS is the UTC timestamp prefix stamped on every log line. Stored in UTC
|
||||
// (RFC3339, millisecond precision); the UI renders it in the viewer's timezone.
|
||||
|
||||
|
||||
func logTS() string {
|
||||
return time.Now().UTC().Format("2006-01-02T15:04:05.000") + "Z"
|
||||
}
|
||||
|
||||
// AppendMarker writes a timestamped event line to the server-run log and returns
|
||||
// the byte offset at which the write began (used as a step's log_offset).
|
||||
|
||||
|
||||
func AppendMarker(runID, serverID, text string) (int64, error) {
|
||||
path := ServerRunLogPath(runID, serverID)
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
|
||||
@@ -47,19 +47,19 @@ func AppendMarker(runID, serverID, text string) (int64, error) {
|
||||
return 0, err
|
||||
}
|
||||
defer f.Close()
|
||||
off, _ := f.Seek(0, 2) // current end = offset before write
|
||||
off, _ := f.Seek(0, 2)
|
||||
if _, err := f.WriteString("[" + logTS() + "] " + text + "\n"); err != nil {
|
||||
return off, err
|
||||
}
|
||||
return off, nil
|
||||
}
|
||||
|
||||
// ---- streamed chunk writer, boundary-safe secret masking ----
|
||||
|
||||
|
||||
type stepLogWriter struct {
|
||||
mu sync.Mutex
|
||||
f *os.File
|
||||
carry []byte // bytes of an as-yet-unterminated line
|
||||
carry []byte
|
||||
secrets []string
|
||||
}
|
||||
|
||||
@@ -70,7 +70,7 @@ type stepLogRegistry struct {
|
||||
|
||||
var StepLogs = &stepLogRegistry{writers: make(map[string]*stepLogWriter)}
|
||||
|
||||
// Open opens (append) the server-run file for a step's streamed chunks.
|
||||
|
||||
func (r *stepLogRegistry) Open(commandID, path string, secrets []string) error {
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
|
||||
return err
|
||||
@@ -92,10 +92,10 @@ func (r *stepLogRegistry) get(commandID string) *stepLogWriter {
|
||||
return r.writers[commandID]
|
||||
}
|
||||
|
||||
// Append buffers chunks into whole lines, then writes each complete line with a
|
||||
// UTC timestamp prefix and secret masking applied. Buffering by line means a
|
||||
// secret split across a chunk boundary is always masked (the whole line is
|
||||
// assembled first) and every line carries its own timestamp.
|
||||
|
||||
|
||||
|
||||
|
||||
func (r *stepLogRegistry) Append(commandID string, data []byte) {
|
||||
w := r.get(commandID)
|
||||
if w == nil {
|
||||
@@ -115,7 +115,7 @@ func (r *stepLogRegistry) Append(commandID string, data []byte) {
|
||||
w.carry = append([]byte{}, buf...)
|
||||
}
|
||||
|
||||
// writeLine emits one masked, timestamped log line. Caller holds w.mu.
|
||||
|
||||
func (w *stepLogWriter) writeLine(line []byte) {
|
||||
masked := maskBytes(line, w.secrets)
|
||||
_, _ = w.f.WriteString("[" + logTS() + "] ")
|
||||
@@ -123,7 +123,7 @@ func (w *stepLogWriter) writeLine(line []byte) {
|
||||
_, _ = w.f.WriteString("\n")
|
||||
}
|
||||
|
||||
// Close flushes any trailing partial line and closes the file.
|
||||
|
||||
func (r *stepLogRegistry) Close(commandID string) {
|
||||
r.mu.Lock()
|
||||
w := r.writers[commandID]
|
||||
@@ -152,9 +152,9 @@ func maskBytes(b []byte, secrets []string) []byte {
|
||||
return []byte(s)
|
||||
}
|
||||
|
||||
// ---- retention sweeper ----
|
||||
|
||||
// StartLogSweeper sweeps expired run-log dirs hourly (and once now).
|
||||
|
||||
|
||||
func StartLogSweeper() {
|
||||
go func() {
|
||||
sweepLogs()
|
||||
@@ -166,10 +166,10 @@ func StartLogSweeper() {
|
||||
}()
|
||||
}
|
||||
|
||||
// sweepLogs walks the run-log dirs on disk. Log dirs are keyed by run ID, not
|
||||
// by org, and this runs with no session — so retention is resolved per run from
|
||||
// the owning org of that run's doc, with the per-org values cached for the
|
||||
// sweep. Runs whose doc is gone fall back to the default retention.
|
||||
|
||||
|
||||
|
||||
|
||||
func sweepLogs() {
|
||||
base := WorkflowLogDir()
|
||||
entries, err := os.ReadDir(base)
|
||||
@@ -188,14 +188,14 @@ func sweepLogs() {
|
||||
|
||||
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
|
||||
continue
|
||||
}
|
||||
|
||||
days, ok := cache[orgID]
|
||||
@@ -209,7 +209,7 @@ func sweepLogs() {
|
||||
cache[orgID] = days
|
||||
}
|
||||
if days <= 0 {
|
||||
continue // keep forever
|
||||
continue
|
||||
}
|
||||
cutoff := now.AddDate(0, 0, -days)
|
||||
|
||||
@@ -219,7 +219,7 @@ func sweepLogs() {
|
||||
}
|
||||
continue
|
||||
}
|
||||
// run doc gone: use dir mtime
|
||||
|
||||
if fi, e := os.Stat(dir); e == nil && fi.ModTime().Before(cutoff) {
|
||||
_ = os.RemoveAll(dir)
|
||||
}
|
||||
@@ -228,9 +228,9 @@ func sweepLogs() {
|
||||
|
||||
const defaultRetentionDays = 30
|
||||
|
||||
// runRetentionInfo returns the owning org and finish time of a run, and whether
|
||||
// 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()
|
||||
|
||||
@@ -11,12 +11,12 @@ type stepResultRegistry struct {
|
||||
pending map[string]chan *pb.StepResult
|
||||
}
|
||||
|
||||
// StepResults correlates agent StepResult replies back to the workflow runner
|
||||
// goroutine that dispatched the matching RunStepCmd, keyed by command_id.
|
||||
|
||||
|
||||
var StepResults = &stepResultRegistry{pending: make(map[string]chan *pb.StepResult)}
|
||||
|
||||
// Await registers interest in a command's result BEFORE the command is
|
||||
// dispatched, and returns a buffered channel that receives the single result.
|
||||
|
||||
|
||||
func (r *stepResultRegistry) Await(commandID string) <-chan *pb.StepResult {
|
||||
ch := make(chan *pb.StepResult, 1)
|
||||
r.mu.Lock()
|
||||
@@ -25,14 +25,14 @@ func (r *stepResultRegistry) Await(commandID string) <-chan *pb.StepResult {
|
||||
return ch
|
||||
}
|
||||
|
||||
// Cancel removes a pending waiter (call on timeout to avoid leaks).
|
||||
|
||||
func (r *stepResultRegistry) Cancel(commandID string) {
|
||||
r.mu.Lock()
|
||||
delete(r.pending, commandID)
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
// Deliver routes an incoming StepResult to its waiter, if any.
|
||||
|
||||
func (r *stepResultRegistry) Deliver(res *pb.StepResult) {
|
||||
if res == nil {
|
||||
return
|
||||
|
||||
@@ -5,12 +5,12 @@ import (
|
||||
"strings"
|
||||
)
|
||||
|
||||
// keyAssign matches an env-var assignment target: KEY= (captures KEY).
|
||||
|
||||
var keyAssign = regexp.MustCompile(`([A-Za-z_][A-Za-z0-9_]*)=`)
|
||||
|
||||
// DeriveOutputs scans a step script and returns the output keys it writes to
|
||||
// $WORKFLOW_ENV. Best-effort: only lines that reference WORKFLOW_ENV are
|
||||
// considered. Deduplicated, first-seen order preserved.
|
||||
|
||||
|
||||
|
||||
func DeriveOutputs(script string) []string {
|
||||
out := []string{}
|
||||
seen := map[string]bool{}
|
||||
@@ -20,7 +20,7 @@ func DeriveOutputs(script string) []string {
|
||||
}
|
||||
for _, m := range keyAssign.FindAllStringSubmatch(line, -1) {
|
||||
key := m[1]
|
||||
// Skip the sentinel itself (e.g. "WORKFLOW_ENV=..." assignments).
|
||||
|
||||
if key == "WORKFLOW_ENV" || key == "env" {
|
||||
continue
|
||||
}
|
||||
@@ -36,7 +36,7 @@ func DeriveOutputs(script string) []string {
|
||||
|
||||
var slugStrip = regexp.MustCompile(`[^a-z0-9]+`)
|
||||
|
||||
// Slugify converts a step name into a stable kebab-case slug.
|
||||
|
||||
func Slugify(name string) string {
|
||||
s := strings.ToLower(name)
|
||||
s = slugStrip.ReplaceAllString(s, "-")
|
||||
|
||||
@@ -13,8 +13,8 @@ func BuildAuthorizedKeys(serverID string) ([]string, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Agent path — no session, so the org comes from the server record itself
|
||||
// and both follow-up queries are scoped to it.
|
||||
|
||||
|
||||
srv, err := getServerByID(serverID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -15,13 +15,13 @@ import (
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
)
|
||||
|
||||
// ErrLastOwner is returned when an operation would leave an org with no owner,
|
||||
// which would lock every remaining member out of org administration.
|
||||
|
||||
|
||||
var ErrLastOwner = errors.New("this is the organization's last owner — promote another member to owner first")
|
||||
|
||||
// CountUsers counts users across the whole instance. It answers "is this a
|
||||
// brand new deployment", so it is deliberately unscoped; anything that asks
|
||||
// about a single tenant must use CountOrgUsers.
|
||||
|
||||
|
||||
|
||||
func CountUsers() (int64, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -34,8 +34,8 @@ func CountOrgUsers(orgID string) (int64, error) {
|
||||
return db.Col("users").CountDocuments(ctx, bson.M{"org_id": orgID})
|
||||
}
|
||||
|
||||
// countOtherOwners counts owner-role users in the org excluding exceptUserID,
|
||||
// i.e. how many owners would remain if that user were removed or demoted.
|
||||
|
||||
|
||||
func countOtherOwners(orgID, exceptUserID string) (int64, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
@@ -142,7 +142,7 @@ func UpdateUserRole(orgID, userID, role string) error {
|
||||
if err != nil {
|
||||
return fmt.Errorf("user not found")
|
||||
}
|
||||
// Demoting the final owner would leave nobody able to administer the org.
|
||||
|
||||
if target.Role == models.RoleOwner && role != models.RoleOwner {
|
||||
others, err := countOtherOwners(orgID, userID)
|
||||
if err != nil {
|
||||
|
||||
@@ -6,7 +6,7 @@ import (
|
||||
"github.com/mrhid6/vantage/server/internal/models"
|
||||
)
|
||||
|
||||
// ValidateWorkflow checks each step ref sets exactly one of step_id / inline.
|
||||
|
||||
func ValidateWorkflow(w models.Workflow) error {
|
||||
for i, ref := range w.Steps {
|
||||
hasLib := ref.StepID != ""
|
||||
|
||||
@@ -17,8 +17,8 @@ import (
|
||||
|
||||
const stepDispatchGrace = 15 * time.Second
|
||||
|
||||
// TriggerWorkflow snapshots the workflow, creates a run doc, and starts a
|
||||
// background goroutine per target server (parallel fan-out). Returns run_id.
|
||||
|
||||
|
||||
func TriggerWorkflow(orgID, workflowID, actor string) (string, error) {
|
||||
wf, err := GetWorkflow(orgID, workflowID)
|
||||
if err != nil {
|
||||
@@ -30,13 +30,13 @@ func TriggerWorkflow(orgID, workflowID, actor string) (string, error) {
|
||||
if len(wf.Steps) == 0 {
|
||||
return "", fmt.Errorf("workflow has no steps")
|
||||
}
|
||||
// Re-check ownership at trigger time — targets may predate validation or a
|
||||
// server may have been removed since the workflow was saved.
|
||||
|
||||
|
||||
if err := validateTargetServers(orgID, wf.TargetServerIDs); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
// Reject a concurrent run of the same workflow.
|
||||
|
||||
ctx, cancel := wfCtx()
|
||||
running := db.Col("workflow_runs").FindOne(ctx, bson.M{"org_id": orgID, "workflow_id": workflowID, "status": "running"})
|
||||
cancel()
|
||||
@@ -82,8 +82,8 @@ func TriggerWorkflow(orgID, workflowID, actor string) (string, error) {
|
||||
return run.RunID, nil
|
||||
}
|
||||
|
||||
// resolveSteps freezes each workflow step ref into a ResolvedStep by loading the
|
||||
// library step and applying overrides.
|
||||
|
||||
|
||||
func resolveSteps(orgID string, wf *models.Workflow) ([]models.ResolvedStep, error) {
|
||||
ctx, cancel := wfCtx()
|
||||
defer cancel()
|
||||
@@ -133,7 +133,7 @@ func resolveSteps(orgID string, wf *models.Workflow) ([]models.ResolvedStep, err
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// resolveInlineStep freezes an ad-hoc (inline) step ref into a ResolvedStep.
|
||||
|
||||
func resolveInlineStep(ref models.WorkflowStepRef) models.ResolvedStep {
|
||||
in := ref.Inline
|
||||
inputs := map[string]string{}
|
||||
@@ -162,7 +162,7 @@ func resolveInlineStep(ref models.WorkflowStepRef) models.ResolvedStep {
|
||||
}
|
||||
}
|
||||
|
||||
// executeRun fans out one goroutine per server run and waits for all to finish.
|
||||
|
||||
func executeRun(runID string) {
|
||||
run, err := getRunByID(runID)
|
||||
if err != nil {
|
||||
@@ -179,7 +179,7 @@ func executeRun(runID string) {
|
||||
<-done
|
||||
}
|
||||
|
||||
// Aggregate status.
|
||||
|
||||
final, _ := getRunByID(runID)
|
||||
status := "success"
|
||||
for _, sr := range final.ServerRuns {
|
||||
@@ -194,8 +194,8 @@ func executeRun(runID string) {
|
||||
bson.M{"$set": bson.M{"status": status, "finished_at": now}})
|
||||
}
|
||||
|
||||
// runServer executes the resolved steps sequentially on one server, threading
|
||||
// output env forward and applying per-step failure policy.
|
||||
|
||||
|
||||
func runServer(orgID, runID string, srvIdx int, steps []models.ResolvedStep, serverID string) {
|
||||
now := time.Now()
|
||||
setServerRun(runID, srvIdx, bson.M{"server_runs.$.status": "running", "server_runs.$.started_at": now})
|
||||
@@ -223,14 +223,14 @@ func runServer(orgID, runID string, srvIdx int, steps []models.ResolvedStep, ser
|
||||
maxAttempts = step.MaxRetries + 1
|
||||
}
|
||||
|
||||
// Merge secrets into command env (kept out of persisted logs).
|
||||
|
||||
secretVals := resolveSecrets(orgID, step.SecretRefs)
|
||||
for k, v := range secretVals {
|
||||
allSecrets[k] = v
|
||||
}
|
||||
// Input values may template earlier step outputs and secrets, e.g.
|
||||
// URL="http://example.com/$VersionNumber". Expand against runEnv (outputs
|
||||
// threaded from prior steps) and this step's secrets before dispatch.
|
||||
|
||||
|
||||
|
||||
subst := map[string]string{}
|
||||
for k, v := range runEnv {
|
||||
subst[k] = v
|
||||
@@ -249,8 +249,8 @@ func runServer(orgID, runID string, srvIdx int, steps []models.ResolvedStep, ser
|
||||
cmdEnv[k] = v
|
||||
}
|
||||
|
||||
// Write the step marker to the server-run log and remember the offset so
|
||||
// the UI can slice this step's output later.
|
||||
|
||||
|
||||
marker := fmt.Sprintf("===== step %d/%d: %s (%s) =====", step.Order+1, len(steps), step.Name, step.Interpreter)
|
||||
offset, _ := AppendMarker(runID, serverID, marker)
|
||||
logPath := ServerRunLogPath(runID, serverID)
|
||||
@@ -262,8 +262,8 @@ func runServer(orgID, runID string, srvIdx int, steps []models.ResolvedStep, ser
|
||||
if attempts > 1 {
|
||||
_, _ = AppendMarker(runID, serverID, fmt.Sprintf("retry %d/%d after failure", attempts-1, maxAttempts-1))
|
||||
}
|
||||
// Open a fresh writer per attempt; the agent's eof closes it, and the
|
||||
// defensive Close below covers a missing result.
|
||||
|
||||
|
||||
_ = StepLogs.Open(commandID, logPath, secretsSlice)
|
||||
res = dispatchAndWait(serverID, commandID, &pb.RunStepCmd{
|
||||
Interpreter: step.Interpreter,
|
||||
@@ -272,18 +272,18 @@ func runServer(orgID, runID string, srvIdx int, steps []models.ResolvedStep, ser
|
||||
TimeoutSeconds: 0,
|
||||
WorkspaceId: runID,
|
||||
})
|
||||
StepLogs.Close(commandID) // idempotent; no-op if eof already closed it
|
||||
StepLogs.Close(commandID)
|
||||
if res != nil && res.ExitCode == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
exit := 1
|
||||
outEnv := map[string]string{} // masked copy, safe to persist
|
||||
outEnv := map[string]string{}
|
||||
if res != nil {
|
||||
exit = res.ExitCode
|
||||
for k, v := range res.OutputEnv {
|
||||
runEnv[k] = v // real, unmasked value threads forward to later steps
|
||||
runEnv[k] = v
|
||||
outEnv[k] = maskSecrets(v, allSecrets)
|
||||
}
|
||||
} else {
|
||||
@@ -304,7 +304,7 @@ func runServer(orgID, runID string, srvIdx int, steps []models.ResolvedStep, ser
|
||||
switch step.OnFailure {
|
||||
case "continue":
|
||||
_, _ = AppendMarker(runID, serverID, "on_failure=continue — proceeding to next step")
|
||||
default: // "stop" or exhausted "retry"
|
||||
default:
|
||||
serverFailed = true
|
||||
}
|
||||
if serverFailed {
|
||||
@@ -315,8 +315,8 @@ func runServer(orgID, runID string, srvIdx int, steps []models.ResolvedStep, ser
|
||||
}
|
||||
}
|
||||
|
||||
// Tell the agent to remove the run's working directory now that its steps are
|
||||
// done (success or failure). Best-effort; the OS reclaims temp dirs anyway.
|
||||
|
||||
|
||||
DispatchCleanupWorkspace(serverID, runID)
|
||||
|
||||
fin := time.Now()
|
||||
@@ -326,8 +326,8 @@ func runServer(orgID, runID string, srvIdx int, steps []models.ResolvedStep, ser
|
||||
}
|
||||
_, _ = AppendMarker(runID, serverID, fmt.Sprintf("run %s in %s — workspace removed",
|
||||
status, fin.Sub(now).Round(time.Millisecond)))
|
||||
// Persist only a masked copy of runEnv; the real (unmasked) runEnv was already
|
||||
// used above to build cmdEnv for each step and must never be written to the DB.
|
||||
|
||||
|
||||
maskedRunEnv := make(map[string]string, len(runEnv))
|
||||
for k, v := range runEnv {
|
||||
maskedRunEnv[k] = maskSecrets(v, allSecrets)
|
||||
@@ -339,8 +339,8 @@ func runServer(orgID, runID string, srvIdx int, steps []models.ResolvedStep, ser
|
||||
})
|
||||
}
|
||||
|
||||
// dispatchAndWait registers a waiter, dispatches the step, and blocks for the
|
||||
// result or a timeout.
|
||||
|
||||
|
||||
func dispatchAndWait(serverID, commandID string, cmd *pb.RunStepCmd) *pb.StepResult {
|
||||
ch := StepResults.Await(commandID)
|
||||
if err := DispatchRunStep(serverID, commandID, cmd); err != nil {
|
||||
@@ -360,9 +360,9 @@ func dispatchAndWait(serverID, commandID string, cmd *pb.RunStepCmd) *pb.StepRes
|
||||
}
|
||||
}
|
||||
|
||||
// expandVars substitutes $VAR and ${VAR} references in an input value from the
|
||||
// given lookup (prior step outputs and secrets). Unknown references expand to
|
||||
// empty, matching shell behaviour; a literal "$" is written as "$$".
|
||||
|
||||
|
||||
|
||||
func expandVars(v string, lookup map[string]string) string {
|
||||
return os.Expand(v, func(name string) string {
|
||||
if name == "$" {
|
||||
@@ -375,7 +375,7 @@ func expandVars(v string, lookup map[string]string) string {
|
||||
func resolveSecrets(orgID string, refs []string) map[string]string {
|
||||
out := map[string]string{}
|
||||
for _, ref := range refs {
|
||||
// ref format "group/KEY"; resolve via RevealSecret.
|
||||
|
||||
parts := strings.SplitN(ref, "/", 2)
|
||||
if len(parts) != 2 {
|
||||
continue
|
||||
@@ -397,7 +397,7 @@ func maskSecrets(s string, secrets map[string]string) string {
|
||||
return s
|
||||
}
|
||||
|
||||
// ---- run doc mutation helpers ----
|
||||
|
||||
|
||||
func setServerRun(runID string, srvIdx int, set bson.M) {
|
||||
ctx, cancel := wfCtx()
|
||||
@@ -407,7 +407,7 @@ func setServerRun(runID string, srvIdx int, set bson.M) {
|
||||
bson.M{"$set": set})
|
||||
}
|
||||
|
||||
// serverIDAt returns the server_id at an index (positional operator needs a match).
|
||||
|
||||
func serverIDAt(runID string, srvIdx int) string {
|
||||
r, err := getRunByID(runID)
|
||||
if err != nil || srvIdx >= len(r.ServerRuns) {
|
||||
@@ -436,7 +436,7 @@ func finishStep(runID, serverID string, order int, status string, attempts, exit
|
||||
})
|
||||
}
|
||||
|
||||
// secretValues returns just the values of a secret map, for masking log output.
|
||||
|
||||
func secretValues(m map[string]string) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for _, v := range m {
|
||||
@@ -471,11 +471,11 @@ func updateStep(runID, serverID string, order int, set bson.M) {
|
||||
)
|
||||
}
|
||||
|
||||
// ---- reads ----
|
||||
|
||||
// getRunByID looks up a run by its unique run_id with no org filter. For
|
||||
// agent/internal run-execution use only (executeRun/runServer, etc.), which
|
||||
// don't have a session and instead resolve org from the run doc itself.
|
||||
|
||||
|
||||
|
||||
|
||||
func getRunByID(runID string) (*models.WorkflowRun, error) {
|
||||
ctx, cancel := wfCtx()
|
||||
defer cancel()
|
||||
@@ -487,7 +487,7 @@ func getRunByID(runID string) (*models.WorkflowRun, error) {
|
||||
return &r, err
|
||||
}
|
||||
|
||||
// GetRun looks up a run scoped to an org (handler/session use).
|
||||
|
||||
func GetRun(orgID, runID string) (*models.WorkflowRun, error) {
|
||||
ctx, cancel := wfCtx()
|
||||
defer cancel()
|
||||
|
||||
@@ -25,8 +25,8 @@ func EnsureWorkflowIndexes() error {
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
// The pre-multi-tenant index was on slug alone, so seeding defaults for a
|
||||
// second org collided — drop it if a live DB still carries it.
|
||||
|
||||
|
||||
if err := db.Col("workflow_steps").Indexes().DropOne(ctx, "slug_1"); err != nil && !isIndexNotFound(err) {
|
||||
return err
|
||||
}
|
||||
@@ -48,7 +48,7 @@ func EnsureWorkflowIndexes() error {
|
||||
return err
|
||||
}
|
||||
|
||||
// ---- Steps ----
|
||||
|
||||
|
||||
func ListSteps(orgID string) ([]models.WorkflowStep, error) {
|
||||
ctx, cancel := wfCtx()
|
||||
@@ -66,8 +66,8 @@ func ListSteps(orgID string) ([]models.WorkflowStep, error) {
|
||||
return steps, nil
|
||||
}
|
||||
|
||||
// StepUsageCounts returns, per library step_id, the number of distinct
|
||||
// workflows that reference it. Inline steps have no step_id and are ignored.
|
||||
|
||||
|
||||
func StepUsageCounts(orgID string) (map[string]int, error) {
|
||||
ctx, cancel := wfCtx()
|
||||
defer cancel()
|
||||
@@ -139,7 +139,7 @@ func DeleteStep(orgID, stepID string) error {
|
||||
if _, err := db.Col("workflow_steps").DeleteOne(ctx, bson.M{"step_id": stepID, "org_id": orgID}); err != nil {
|
||||
return err
|
||||
}
|
||||
// Cascade: remove this step from every workflow that references it, re-sequencing orders.
|
||||
|
||||
cur, err := db.Col("workflows").Find(ctx, bson.M{"steps.step_id": stepID, "org_id": orgID})
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -179,7 +179,7 @@ func getStep(ctx context.Context, orgID, stepID string) (*models.WorkflowStep, e
|
||||
return &s, err
|
||||
}
|
||||
|
||||
// ---- Workflows ----
|
||||
|
||||
|
||||
func ListWorkflows(orgID string) ([]models.Workflow, error) {
|
||||
ctx, cancel := wfCtx()
|
||||
@@ -253,9 +253,9 @@ func UpdateWorkflow(orgID, id string, w models.Workflow) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// validateTargetServers rejects any target server that does not belong to the
|
||||
// org. The IDs are client-supplied and are later consumed by the runner's
|
||||
// unscoped lookups, so ownership has to be proven at the write boundary.
|
||||
|
||||
|
||||
|
||||
func validateTargetServers(orgID string, serverIDs []string) error {
|
||||
for _, sid := range serverIDs {
|
||||
if _, err := GetServer(orgID, sid); err != nil {
|
||||
@@ -265,8 +265,8 @@ func validateTargetServers(orgID string, serverIDs []string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// normalizeInlineSteps derives outputs for inline steps and strips fields that
|
||||
// only belong to library steps.
|
||||
|
||||
|
||||
func normalizeInlineSteps(w *models.Workflow) {
|
||||
for i := range w.Steps {
|
||||
in := w.Steps[i].Inline
|
||||
|
||||
Reference in New Issue
Block a user