diff --git a/backend/migrations/000002_linux_user_auditor.down.sql b/backend/migrations/000002_linux_user_auditor.down.sql new file mode 100644 index 000000000..1b670dea2 --- /dev/null +++ b/backend/migrations/000002_linux_user_auditor.down.sql @@ -0,0 +1,19 @@ +-- 000002_linux_user_auditor.down.sql +-- +-- Reverse the partial-index topology and restore the full-table unique index. +-- The additive columns (source, machine_id, uid_number, hostname, username) are +-- intentionally NOT dropped — leaving them in place is harmless for the old code +-- path and preserves data. If a full schema rollback is required, drop the +-- columns manually after this migration runs. +-- +-- Ordering caveat (documented in the runbook): +-- Rolling back the backend code without running this .down.sql leaves the old +-- Windows ON CONFLICT (tenant_id, sid) statement targeting an index that no +-- longer exists. The Windows upsert path will fail. ALWAYS run this .down.sql +-- BEFORE rolling back the code. + +DROP INDEX IF EXISTS idx_aduser_linux; +DROP INDEX IF EXISTS idx_aduser_windows; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_ad_user_tenant_sid + ON ad_user (tenant_id, sid); diff --git a/backend/migrations/000002_linux_user_auditor.up.sql b/backend/migrations/000002_linux_user_auditor.up.sql new file mode 100644 index 000000000..0b36da88c --- /dev/null +++ b/backend/migrations/000002_linux_user_auditor.up.sql @@ -0,0 +1,44 @@ +-- 000002_linux_user_auditor.up.sql +-- +-- Purpose: +-- Extend the ad_user table so it can host both Windows (AD) users and Linux users +-- in a single inventory. Windows uniqueness stays on (tenant_id, sid); Linux +-- uniqueness moves to (tenant_id, machine_id, uid_number). Both are enforced by +-- Postgres partial unique indexes — the two universes cannot collide with each +-- other, and provisional Linux rows (machine_id IS NULL) coexist because Postgres +-- treats NULLs as distinct in unique indexes. +-- +-- Ordering: +-- AutoMigrate (GORM) runs BEFORE this migration and adds the columns declared on +-- domain.ADUser (source, machine_id, uid_number, hostname, username) with their +-- defaults. This SQL file swaps the old full-table unique index for the two +-- source-discriminated partial indexes. +-- +-- Safety: +-- ADD COLUMN IF NOT EXISTS is a belt-and-suspenders guard against environments +-- where AutoMigrate ran partially or was skipped. +-- + +-- Safety net: ensure the source column exists with the correct default before the +-- partial indexes reference it. AutoMigrate should have already added it, but this +-- is idempotent and covers partial-migration recovery paths. +ALTER TABLE ad_user + ADD COLUMN IF NOT EXISTS source VARCHAR(16) NOT NULL DEFAULT 'windows'; + +-- Drop the old full-table unique index. Windows uniqueness moves to a partial +-- variant that only enforces on rows where source = 'windows'. +DROP INDEX IF EXISTS idx_ad_user_tenant_sid; + +-- Windows partial unique index. +CREATE UNIQUE INDEX IF NOT EXISTS idx_aduser_windows + ON ad_user (tenant_id, sid) + WHERE source = 'windows'; + +-- Linux partial unique index. Rows where machine_id IS NULL are provisional and +-- Postgres treats their NULL as distinct — so multiple provisional rows for the +-- same (tenant_id, hostname, username) with machine_id = NULL are permitted at the +-- index level. Application-level de-duplication (see repository.Upsert) prevents +-- them in practice. +CREATE UNIQUE INDEX IF NOT EXISTS idx_aduser_linux + ON ad_user (tenant_id, machine_id, uid_number) + WHERE source = 'linux'; diff --git a/backend/modules/adaudit/connectors/connectors.go b/backend/modules/adaudit/connectors/connectors.go index d78bac957..42ea5f44f 100644 --- a/backend/modules/adaudit/connectors/connectors.go +++ b/backend/modules/adaudit/connectors/connectors.go @@ -11,13 +11,15 @@ import ( type ADUserRepository interface { Upsert(ctx context.Context, users []domain.ADUser) error List(ctx context.Context, f dto.ADUserFilter) ([]domain.ADUser, int64, error) - All(ctx context.Context) ([]domain.ADUser, error) + All(ctx context.Context, source string) ([]domain.ADUser, error) Stats(ctx context.Context, tenantID string) (*dto.ADUserStats, error) + ResolveLinuxIdentity(ctx context.Context, tenantID, hostname, machineID string) (int64, error) } type ADUserUsecase interface { Ingest(ctx context.Context, req dto.IngestRequest) (int, error) List(ctx context.Context, f dto.ADUserFilter) (*database.List[domain.ADUser], error) - All(ctx context.Context) ([]domain.ADUser, error) + All(ctx context.Context, source string) ([]domain.ADUser, error) Stats(ctx context.Context, tenantID string) (*dto.ADUserStats, error) + ResolveLinuxIdentity(ctx context.Context, req dto.ResolveLinuxIdentityRequest) (int64, error) } diff --git a/backend/modules/adaudit/domain/ad_user.go b/backend/modules/adaudit/domain/ad_user.go index d5fda0fdb..425e916c9 100644 --- a/backend/modules/adaudit/domain/ad_user.go +++ b/backend/modules/adaudit/domain/ad_user.go @@ -4,10 +4,15 @@ import "time" type ADUser struct { ID uint64 `gorm:"primaryKey;autoIncrement" json:"id"` - TenantID string `gorm:"column:tenant_id;size:64;not null;uniqueIndex:idx_ad_user_tenant_sid" json:"tenantId"` - SID string `gorm:"column:sid;size:128;not null;uniqueIndex:idx_ad_user_tenant_sid" json:"sid"` - SamAccountName string `gorm:"column:sam_account_name;size:255" json:"samAccountName"` - Domain string `gorm:"column:domain;size:255" json:"domain"` + TenantID string `gorm:"column:tenant_id;size:64;not null" json:"tenantId"` + Source string `gorm:"column:source;size:16;not null;default:'windows'" json:"source"` + SID *string `gorm:"column:sid;size:128" json:"sid,omitempty"` + SamAccountName string `gorm:"column:sam_account_name;size:255" json:"samAccountName,omitempty"` + Domain string `gorm:"column:domain;size:255" json:"domain,omitempty"` + MachineID *string `gorm:"column:machine_id;size:64" json:"machineId,omitempty"` + UIDNumber *string `gorm:"column:uid_number;size:32" json:"uidNumber,omitempty"` + Hostname *string `gorm:"column:hostname;size:255" json:"hostname,omitempty"` + Username *string `gorm:"column:username;size:255" json:"username,omitempty"` Active bool `gorm:"column:active;not null;default:true" json:"active"` AccountCreatedAt *time.Time `gorm:"column:account_created_at" json:"accountCreatedAt,omitempty"` LastLogon *time.Time `gorm:"column:last_logon" json:"lastLogon,omitempty"` diff --git a/backend/modules/adaudit/dto/ad_user.go b/backend/modules/adaudit/dto/ad_user.go index d8057f8c4..c56b0bfb2 100644 --- a/backend/modules/adaudit/dto/ad_user.go +++ b/backend/modules/adaudit/dto/ad_user.go @@ -8,9 +8,14 @@ import ( type IngestUser struct { TenantID string `json:"tenantId"` - SID string `json:"sid" binding:"required"` - SamAccountName string `json:"samAccountName"` - Domain string `json:"domain"` + Source string `json:"source,omitempty"` // "windows"|"linux"; defaults to "windows" if absent + SID string `json:"sid,omitempty"` // required when source=windows; enforced in usecase + SamAccountName string `json:"samAccountName,omitempty"` + Domain string `json:"domain,omitempty"` + MachineID *string `json:"machineId,omitempty"` + UIDNumber *string `json:"uidNumber,omitempty"` + Hostname *string `json:"hostname,omitempty"` + Username *string `json:"username,omitempty"` Active *bool `json:"active"` AccountCreatedAt *time.Time `json:"accountCreatedAt"` LastLogon *time.Time `json:"lastLogon"` @@ -25,6 +30,7 @@ type IngestRequest struct { type ADUserFilter struct { Search string `form:"search"` // substring on samAccountName/sid TenantID string `form:"tenantId"` // exact + Source string `form:"source"` // "windows"|"linux"|"" (all) Active *bool `form:"active"` Status string `form:"status"` // active|disabled|deleted|stale|service — overrides Active when set Sort string `form:"sort"` // recent (last_seen desc) | name (default, samAccountName asc) @@ -41,6 +47,12 @@ type DomainCount struct { Count int64 `json:"count"` } +// SourceCount is the by-source breakdown returned by GET /ad-audit/stats. +type SourceCount struct { + Windows int64 `json:"windows"` + Linux int64 `json:"linux"` +} + // ADUserStats is the inventory roll-up the UI overview renders. Counts honor the // optional tenant scope; Tenants is always the global distinct list so the // tenant picker stays stable regardless of the active scope. @@ -52,6 +64,18 @@ type ADUserStats struct { Stale int64 `json:"stale"` Service int64 `json:"service"` Seen24h int64 `json:"seen_24h"` + BySource SourceCount `json:"by_source"` ByDomain []DomainCount `json:"by_domain"` Tenants []string `json:"tenants"` } + +// ResolveLinuxIdentityRequest is the payload the ad-audit plugin sends when it +// learns the machine-id for a host that already has provisional Linux user rows. +// The backend updates all matching provisional rows to set machine_id: if a +// resolved row already exists for (tenant_id, machine_id, uid_number), +// the provisional row is left untouched. +type ResolveLinuxIdentityRequest struct { + TenantID string `json:"tenant_id" binding:"required"` + Hostname string `json:"hostname" binding:"required"` + MachineID string `json:"machine_id" binding:"required"` +} diff --git a/backend/modules/adaudit/handler/ad_user.go b/backend/modules/adaudit/handler/ad_user.go index 633feaa04..fe1fe778c 100644 --- a/backend/modules/adaudit/handler/ad_user.go +++ b/backend/modules/adaudit/handler/ad_user.go @@ -51,7 +51,8 @@ func (h *ADUserHandler) Ingest(c *gin.Context) { // @Tags AD Audit // @Security BearerAuth // @Produce json -// @Param search query string false "Substring on samAccountName/sid" +// @Param search query string false "Substring on samAccountName/sid/username" +// @Param source query string false "Filter by source: windows | linux (omit for all)" // @Param tenantId query string false "Filter by tenant" // @Param active query bool false "Filter by active" // @Param status query string false "Lifecycle bucket: active|disabled|deleted|stale|service (overrides active)" @@ -60,6 +61,7 @@ func (h *ADUserHandler) Ingest(c *gin.Context) { // @Param size query int false "Page size" // @Success 200 {array} domain.ADUser // @Header 200 {string} X-Total-Count "Total records" +// @Failure 400 {object} map[string]string // @Failure 500 {object} map[string]string // @Router /ad-audit/users [get] func (h *ADUserHandler) List(c *gin.Context) { @@ -68,6 +70,10 @@ func (h *ADUserHandler) List(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } + if f.Source != "" && f.Source != "windows" && f.Source != "linux" { + c.JSON(http.StatusBadRequest, gin.H{"error": "source must be 'windows', 'linux', or omitted"}) + return + } res, err := h.uc.List(c.Request.Context(), f) if err != nil { _ = catcher.Error("adaudit: list failed", err, nil) @@ -108,13 +114,21 @@ func (h *ADUserHandler) Stats(c *gin.Context) { // // @Summary Export all AD users (internal) // @Description Internal endpoint the ad-audit plugin calls at startup to seed its in-memory cache. +// @Description Accepts an optional `source` filter so the plugin can seed its Windows and Linux // @Tags AD Audit // @Produce json -// @Success 200 {array} domain.ADUser -// @Failure 500 {object} map[string]string +// @Param source query string false "Filter by source: windows | linux (omit for all)" +// @Success 200 {array} domain.ADUser +// @Failure 400 {object} map[string]string +// @Failure 500 {object} map[string]string // @Router /ad-audit/users/sync [get] func (h *ADUserHandler) Sync(c *gin.Context) { - users, err := h.uc.All(c.Request.Context()) + source := c.Query("source") + if source != "" && source != "windows" && source != "linux" { + c.JSON(http.StatusBadRequest, gin.H{"error": "source must be 'windows', 'linux', or omitted"}) + return + } + users, err := h.uc.All(c.Request.Context(), source) if err != nil { _ = catcher.Error("adaudit: sync failed", err, nil) c.JSON(http.StatusInternalServerError, gin.H{"error": "could not export users"}) @@ -122,3 +136,32 @@ func (h *ADUserHandler) Sync(c *gin.Context) { } c.JSON(http.StatusOK, users) } + +// Resolve godoc +// +// @Summary Resolve provisional Linux user rows (internal) +// @Description Internal endpoint the ad-audit plugin calls once it learns the +// @Description machine-id for a host that already has provisional Linux user rows +// @Description (machine_id IS NULL). +// @Tags AD Audit +// @Accept json +// @Produce json +// @Param input body dto.ResolveLinuxIdentityRequest true "Resolution payload" +// @Success 200 {object} map[string]int64 +// @Failure 400 {object} map[string]string +// @Failure 500 {object} map[string]string +// @Router /ad-audit/users/resolve [post] +func (h *ADUserHandler) Resolve(c *gin.Context) { + var req dto.ResolveLinuxIdentityRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + n, err := h.uc.ResolveLinuxIdentity(c.Request.Context(), req) + if err != nil { + _ = catcher.Error("adaudit: resolve failed", err, nil) + c.JSON(http.StatusInternalServerError, gin.H{"error": "could not resolve provisional rows"}) + return + } + c.JSON(http.StatusOK, gin.H{"resolved": n}) +} diff --git a/backend/modules/adaudit/repository/ad_user_pg.go b/backend/modules/adaudit/repository/ad_user_pg.go index 7702912dc..298b90dca 100644 --- a/backend/modules/adaudit/repository/ad_user_pg.go +++ b/backend/modules/adaudit/repository/ad_user_pg.go @@ -2,6 +2,7 @@ package repository import ( "context" + "errors" "github.com/utmstack/utmstack/backend/modules/adaudit/connectors" "github.com/utmstack/utmstack/backend/modules/adaudit/domain" @@ -10,6 +11,11 @@ import ( "gorm.io/gorm/clause" ) +type sourceRow struct { + Source string + Count int64 +} + type pgADUserRepository struct{ db *gorm.DB } func NewADUserRepository(db *gorm.DB) connectors.ADUserRepository { @@ -20,13 +26,107 @@ func (r *pgADUserRepository) Upsert(ctx context.Context, users []domain.ADUser) if len(users) == 0 { return nil } - return r.db.WithContext(ctx).Clauses(clause.OnConflict{ - Columns: []clause.Column{{Name: "tenant_id"}, {Name: "sid"}}, - DoUpdates: clause.AssignmentColumns([]string{ - "sam_account_name", "domain", "active", - "account_created_at", "last_logon", "account_deleted_at", "last_seen", - }), - }).Create(&users).Error + + var windowsBatch []domain.ADUser + var linuxResolved []domain.ADUser + var linuxProvisional []domain.ADUser + + for _, u := range users { + switch u.Source { + case "linux": + if u.MachineID != nil && *u.MachineID != "" { + linuxResolved = append(linuxResolved, u) + } else { + linuxProvisional = append(linuxProvisional, u) + } + default: + if u.Source == "" { + u.Source = "windows" + } + windowsBatch = append(windowsBatch, u) + } + } + + if len(windowsBatch) > 0 { + if err := r.db.WithContext(ctx).Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "tenant_id"}, {Name: "sid"}}, + TargetWhere: clause.Where{Exprs: []clause.Expression{clause.Expr{SQL: "source = 'windows'"}}}, + DoUpdates: clause.AssignmentColumns([]string{ + "sam_account_name", "domain", "active", + "account_created_at", "last_logon", "account_deleted_at", "last_seen", + }), + }).Create(&windowsBatch).Error; err != nil { + return err + } + } + + if len(linuxResolved) > 0 { + if err := r.db.WithContext(ctx).Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "tenant_id"}, {Name: "machine_id"}, {Name: "uid_number"}}, + TargetWhere: clause.Where{Exprs: []clause.Expression{clause.Expr{SQL: "source = 'linux'"}}}, + DoUpdates: clause.AssignmentColumns([]string{ + "username", "hostname", "active", + "account_created_at", "last_logon", "account_deleted_at", "last_seen", + }), + }).Create(&linuxResolved).Error; err != nil { + return err + } + } + + for i := range linuxProvisional { + u := &linuxProvisional[i] + if u.Hostname == nil || u.Username == nil { + continue + } + var existing domain.ADUser + err := r.db.WithContext(ctx).Where( + "tenant_id = ? AND source = 'linux' AND hostname = ? AND username = ? AND machine_id IS NULL", + u.TenantID, *u.Hostname, *u.Username, + ).First(&existing).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + if err := r.db.WithContext(ctx).Create(u).Error; err != nil { + return err + } + } else if err != nil { + return err + } else { + updates := map[string]any{ + "active": u.Active, + "account_created_at": u.AccountCreatedAt, + "last_logon": u.LastLogon, + "account_deleted_at": u.AccountDeletedAt, + "last_seen": u.LastSeen, + } + if u.UIDNumber != nil && existing.UIDNumber == nil { + updates["uid_number"] = *u.UIDNumber + } + if err := r.db.WithContext(ctx).Model(&existing).Updates(updates).Error; err != nil { + return err + } + } + } + + return nil +} + +func (r *pgADUserRepository) ResolveLinuxIdentity(ctx context.Context, tenantID, hostname, machineID string) (int64, error) { + if tenantID == "" || hostname == "" || machineID == "" { + return 0, nil + } + result := r.db.WithContext(ctx).Exec(` + UPDATE ad_user SET machine_id = ? + WHERE tenant_id = ? + AND source = 'linux' + AND hostname = ? + AND machine_id IS NULL + AND NOT EXISTS ( + SELECT 1 FROM ad_user r2 + WHERE r2.tenant_id = ad_user.tenant_id + AND r2.source = 'linux' + AND r2.machine_id = ? + AND r2.uid_number = ad_user.uid_number + )`, machineID, tenantID, hostname, machineID) + return result.RowsAffected, result.Error } // applyStatus narrows a query to a derived lifecycle bucket. These derivations @@ -43,7 +143,7 @@ func applyStatus(q *gorm.DB, status string) *gorm.DB { case "stale": return q.Where("active = true AND account_deleted_at IS NULL AND last_logon IS NOT NULL AND last_logon < NOW() - INTERVAL '30 days'") case "service": - return q.Where("account_deleted_at IS NULL AND sam_account_name ILIKE 'svc%'") + return q.Where("account_deleted_at IS NULL AND sam_account_name ILIKE 'svc%' AND source = 'windows'") } return q } @@ -53,6 +153,9 @@ func (r *pgADUserRepository) List(ctx context.Context, f dto.ADUserFilter) ([]do if f.TenantID != "" { q = q.Where("tenant_id = ?", f.TenantID) } + if f.Source != "" { + q = q.Where("source = ?", f.Source) + } if f.Status != "" { // Status is a richer, derived filter; when present it supersedes the raw // `active` flag so the two can't contradict each other. @@ -62,7 +165,7 @@ func (r *pgADUserRepository) List(ctx context.Context, f dto.ADUserFilter) ([]do } if f.Search != "" { like := "%" + f.Search + "%" - q = q.Where("sam_account_name ILIKE ? OR sid ILIKE ?", like, like) + q = q.Where("sam_account_name ILIKE ? OR sid ILIKE ? OR username ILIKE ?", like, like, like) } var total int64 @@ -113,6 +216,20 @@ func (r *pgADUserRepository) Stats(ctx context.Context, tenantID string) (*dto.A if err := base().Where("last_seen IS NOT NULL AND last_seen > NOW() - INTERVAL '24 hours'").Count(&s.Seen24h).Error; err != nil { return nil, err } + + var srcRows []sourceRow + if err := base().Select("source, COUNT(*) AS count").Group("source").Scan(&srcRows).Error; err != nil { + return nil, err + } + for _, row := range srcRows { + switch row.Source { + case "windows": + s.BySource.Windows = row.Count + case "linux": + s.BySource.Linux = row.Count + } + } + if err := base().Select("domain, COUNT(*) AS count").Group("domain").Order("count DESC").Limit(8).Scan(&s.ByDomain).Error; err != nil { return nil, err } @@ -123,9 +240,13 @@ func (r *pgADUserRepository) Stats(ctx context.Context, tenantID string) (*dto.A return &s, nil } -func (r *pgADUserRepository) All(ctx context.Context) ([]domain.ADUser, error) { +func (r *pgADUserRepository) All(ctx context.Context, source string) ([]domain.ADUser, error) { var items []domain.ADUser - if err := r.db.WithContext(ctx).Find(&items).Error; err != nil { + q := r.db.WithContext(ctx) + if source != "" { + q = q.Where("source = ?", source) + } + if err := q.Find(&items).Error; err != nil { return nil, err } return items, nil diff --git a/backend/modules/adaudit/routes.go b/backend/modules/adaudit/routes.go index f15db23e4..83545215e 100644 --- a/backend/modules/adaudit/routes.go +++ b/backend/modules/adaudit/routes.go @@ -9,9 +9,11 @@ func RegisterRoutes(api *gin.RouterGroup, m *Module, userAuth gin.HandlerFunc) { h := m.Handler() g := api.Group("/ad-audit", userAuth) - // Internal: the ad-audit plugin pushes changed users and seeds its cache. + // Internal: the ad-audit plugin pushes changed users, seeds its cache, + // and resolves provisional Linux rows when it learns a host's machine-id. g.POST("/users", middleware.RequireInternal(), h.Ingest) g.GET("/users/sync", middleware.RequireInternal(), h.Sync) + g.POST("/users/resolve", middleware.RequireInternal(), h.Resolve) // UI: list the AD user inventory + inventory stats for the overview. g.GET("/users", middleware.RequirePermission("adaudit.read"), h.List) diff --git a/backend/modules/adaudit/usecase/ad_user.go b/backend/modules/adaudit/usecase/ad_user.go index 1ee476992..fc628ce21 100644 --- a/backend/modules/adaudit/usecase/ad_user.go +++ b/backend/modules/adaudit/usecase/ad_user.go @@ -21,24 +21,54 @@ func NewADUserUsecase(repo connectors.ADUserRepository) connectors.ADUserUsecase func (u *adUserUsecase) Ingest(ctx context.Context, req dto.IngestRequest) (int, error) { users := make([]domain.ADUser, 0, len(req.Users)) for _, in := range req.Users { - if strings.TrimSpace(in.SID) == "" { + source := in.Source + if source == "" { + source = "windows" + } + + switch source { + case "windows": + if strings.TrimSpace(in.SID) == "" { + continue + } + case "linux": + hasResolved := in.MachineID != nil && strings.TrimSpace(*in.MachineID) != "" && + in.UIDNumber != nil && strings.TrimSpace(*in.UIDNumber) != "" + hasProvisional := in.Hostname != nil && strings.TrimSpace(*in.Hostname) != "" && + in.Username != nil && strings.TrimSpace(*in.Username) != "" + if !hasResolved && !hasProvisional { + continue + } + default: continue } + active := true if in.Active != nil { active = *in.Active } - users = append(users, domain.ADUser{ + + ad := domain.ADUser{ TenantID: in.TenantID, - SID: in.SID, + Source: source, SamAccountName: in.SamAccountName, Domain: in.Domain, + MachineID: in.MachineID, + UIDNumber: in.UIDNumber, + Hostname: in.Hostname, + Username: in.Username, Active: active, AccountCreatedAt: in.AccountCreatedAt, LastLogon: in.LastLogon, AccountDeletedAt: in.AccountDeletedAt, LastSeen: in.LastSeen, - }) + } + if source == "windows" { + sid := strings.TrimSpace(in.SID) + ad.SID = &sid + } + + users = append(users, ad) } if err := u.repo.Upsert(ctx, users); err != nil { return 0, err @@ -54,10 +84,14 @@ func (u *adUserUsecase) List(ctx context.Context, f dto.ADUserFilter) (*database return &database.List[domain.ADUser]{Items: items, Total: total}, nil } -func (u *adUserUsecase) All(ctx context.Context) ([]domain.ADUser, error) { - return u.repo.All(ctx) +func (u *adUserUsecase) All(ctx context.Context, source string) ([]domain.ADUser, error) { + return u.repo.All(ctx, source) } func (u *adUserUsecase) Stats(ctx context.Context, tenantID string) (*dto.ADUserStats, error) { return u.repo.Stats(ctx, tenantID) } + +func (u *adUserUsecase) ResolveLinuxIdentity(ctx context.Context, req dto.ResolveLinuxIdentityRequest) (int64, error) { + return u.repo.ResolveLinuxIdentity(ctx, req.TenantID, req.Hostname, req.MachineID) +} diff --git a/plugins/ad-audit/main.go b/plugins/ad-audit/main.go index 0cb22c708..c0212696b 100644 --- a/plugins/ad-audit/main.go +++ b/plugins/ad-audit/main.go @@ -9,6 +9,8 @@ import ( "net/http" "os" "os/signal" + "regexp" + "sort" "strconv" "strings" "sync" @@ -22,10 +24,6 @@ import ( const pluginName = "com.utmstack.ad-audit" func main() { - if plugins.GetCfg("plugin_"+pluginName).Env.Mode != "manager" { - return - } - loadBackendConfig() seedCacheFromBackend() @@ -46,18 +44,31 @@ const ( evtLogon = "4624" // An account was successfully logged on wineventlogDataType = "wineventlog" + linuxDataType = "linux" fEventCode = "eventCode" fTargetUser = "eventDataTargetUserName" fTargetSID = "eventDataTargetSid" fTargetSID2 = "eventDataTargetUserSid" fTargetDom = "eventDataTargetDomainName" + + auditdDedupTTL = 5 * time.Minute + auditdDedupCap = 10_000 + auditdDedupEvictN = 1000 ) func analyze(event *plugins.Event, _ plugins.Analysis_AnalyzeServer) error { - if event.GetDataType() != wineventlogDataType { + switch event.GetDataType() { + case wineventlogDataType: + return handleWindows(event) + case linuxDataType: + return handleLinux(event) + default: return io.EOF } +} + +func handleWindows(event *plugins.Event) error { code := logEventCode(event) switch code { case evtUserCreated, evtUserDeleted, evtLogon: @@ -82,6 +93,315 @@ func analyze(event *plugins.Event, _ plugins.Analysis_AnalyzeServer) error { return io.EOF } +func handleLinux(event *plugins.Event) error { + syslogID := logStr(event, "syslogIdentifier") + switch syslogID { + case "useradd", "userdel": + return handleLinuxJournald(event, syslogID) + } + action := strings.TrimSpace(event.GetAction()) + if action == "" { + action = logStr(event, "action") + } + switch action { + case "USER_LOGIN", "USER_START": + return handleLinuxAuditd(event, action) + } + return io.EOF +} + +var journaldUserAddRe = regexp.MustCompile(`^new user:\s*name=([^,]+),\s*UID=(\d+)`) + +var journaldUserDelRe = regexp.MustCompile(`^delete user\s+'([^']+)'`) + +func handleLinuxJournald(event *plugins.Event, syslogID string) error { + message := logStr(event, "message") + if message == "" { + return io.EOF + } + tenantID := event.GetTenantId() + hostname := strings.TrimSpace(event.GetDataSource()) + machineID := logStr(event, "machineId") + ts := parseTime(event.GetTimestamp()) + + if machineID != "" && hostname != "" { + key := tenantID + ":" + hostname + cacheMu.Lock() + if prev, ok := hostnameMachineID[key]; !ok || prev != machineID { + hostnameMachineID[key] = machineID + cacheMu.Unlock() + enqueueResolveIntent(tenantID, hostname, machineID) + } else { + cacheMu.Unlock() + } + } + + switch syslogID { + case "useradd": + m := journaldUserAddRe.FindStringSubmatch(message) + if len(m) < 3 { + return io.EOF + } + username := strings.TrimSpace(m[1]) + uid := strings.TrimSpace(m[2]) + if !isHumanUID(uid) || username == "" { + return io.EOF + } + applyLinuxEvent(tenantID, machineID, hostname, uid, username, "create", ts) + + case "userdel": + m := journaldUserDelRe.FindStringSubmatch(message) + if len(m) < 2 { + return io.EOF + } + username := strings.TrimSpace(m[1]) + if username == "" { + return io.EOF + } + applyLinuxEvent(tenantID, machineID, hostname, "", username, "delete", ts) + } + return io.EOF +} + +func handleLinuxAuditd(event *plugins.Event, action string) error { + tenantID := event.GetTenantId() + hostname := strings.TrimSpace(event.GetDataSource()) + ts := parseTime(event.GetTimestamp()) + + sequence := logNumStr(event, "sequence") + if dedupCheckAndMark(hostname, sequence) { + return io.EOF + } + + var subobj string + switch action { + case "USER_LOGIN": + subobj = "userlogin" + case "USER_START": + subobj = "userstart" + default: + return io.EOF + } + + result := logStrPath(event, subobj, "result") + if result != "success" { + return io.EOF + } + + auid := logStrPath(event, subobj, "auid") + if auid == "unset" { + return io.EOF + } + + acct := logStrPath(event, subobj, "acct") + if acct == "(invalid user)" { + return io.EOF + } + if isSystemAccount(acct) { + return io.EOF + } + + id := logStrPath(event, subobj, "id") + + var machineID string + if hostname != "" { + cacheMu.Lock() + machineID = hostnameMachineID[tenantID+":"+hostname] + cacheMu.Unlock() + } + + switch action { + case "USER_LOGIN": + if !isHumanUID(id) { + return io.EOF + } + username := acct + if username == "" { + username = lookupLinuxUsername(tenantID, machineID, hostname, id) + } + if username == "" { + return io.EOF + } + applyLinuxEvent(tenantID, machineID, hostname, id, username, "login", ts) + + case "USER_START": + if acct == "" { + return io.EOF + } + uidForApply := "" + if isHumanUID(id) { + uidForApply = id + } + applyLinuxEvent(tenantID, machineID, hostname, uidForApply, acct, "session", ts) + } + return io.EOF +} + +func dedupCheckAndMark(hostname, sequence string) bool { + if hostname == "" || sequence == "" { + return false + } + key := hostname + ":" + sequence + + dedupMu.Lock() + defer dedupMu.Unlock() + + now := time.Now() + if seen, ok := auditdDedup[key]; ok { + if now.Sub(seen) < auditdDedupTTL { + return true + } + } + + auditdDedup[key] = now + + if len(auditdDedup) > auditdDedupCap { + for k, ts := range auditdDedup { + if now.Sub(ts) >= auditdDedupTTL { + delete(auditdDedup, k) + } + } + if len(auditdDedup) > auditdDedupCap { + evictOldestN(auditdDedup, auditdDedupEvictN) + } + } + return false +} + +func evictOldestN(m map[string]time.Time, n int) { + type kt struct { + k string + t time.Time + } + all := make([]kt, 0, len(m)) + for k, t := range m { + all = append(all, kt{k, t}) + } + sort.Slice(all, func(i, j int) bool { return all[i].t.Before(all[j].t) }) + if n > len(all) { + n = len(all) + } + for i := 0; i < n; i++ { + delete(m, all[i].k) + } +} + +func logStrPath(event *plugins.Event, path ...string) string { + log := event.GetLog() + if log == nil || len(path) == 0 { + return "" + } + first := log[path[0]] + if first == nil { + return "" + } + v := first.AsInterface() + for i := 1; i < len(path); i++ { + m, ok := v.(map[string]interface{}) + if !ok { + return "" + } + v = m[path[i]] + } + if s, ok := v.(string); ok { + return strings.TrimSpace(s) + } + return "" +} + +func lookupLinuxUsername(tenantID, machineID, hostname, uid string) string { + cacheMu.Lock() + defer cacheMu.Unlock() + if machineID != "" && uid != "" { + key := tenantID + ":linux:resolved:" + machineID + ":" + uid + if cu, ok := linuxCache[key]; ok { + return cu.Username + } + } + prefix := tenantID + ":linux:provisional:" + hostname + ":" + for k, cu := range linuxCache { + if strings.HasPrefix(k, prefix) && cu.UIDNumber == uid { + return cu.Username + } + } + return "" +} + +func isHumanUID(uid string) bool { + if uid == "" || uid == "unset" { + return false + } + n, err := strconv.Atoi(uid) + if err != nil { + return false + } + return n >= 1000 +} + +func applyLinuxEvent(tenantID, machineID, hostname, uid, username, eventType string, et *time.Time) { + if username == "" && uid == "" { + return + } + + cacheMu.Lock() + defer cacheMu.Unlock() + + var key string + if machineID != "" && uid != "" { + key = tenantID + ":linux:resolved:" + machineID + ":" + uid + } else if hostname != "" && username != "" { + key = tenantID + ":linux:provisional:" + hostname + ":" + username + } else { + return + } + + cu := linuxCache[key] + if cu == nil { + cu = &cachedLinuxUser{ + TenantID: tenantID, + MachineID: machineID, + Hostname: hostname, + UIDNumber: uid, + Username: username, + Active: true, + } + linuxCache[key] = cu + } + before := *cu + + if machineID != "" && cu.MachineID == "" { + cu.MachineID = machineID + } + if uid != "" && cu.UIDNumber == "" { + cu.UIDNumber = uid + } + if username != "" && cu.Username == "" { + cu.Username = username + } + if hostname != "" && cu.Hostname == "" { + cu.Hostname = hostname + } + + cu.LastSeen = laterTime(cu.LastSeen, et) + switch eventType { + case "create": + cu.AccountCreatedAt = firstSet(cu.AccountCreatedAt, et) + cu.Active = true + case "delete": + cu.Active = false + cu.AccountDeletedAt = et + case "login": + cu.LastLogon = laterTime(cu.LastLogon, et) + cu.Active = true + case "session": + cu.Active = true + } + + if *cu != before { + linuxDirty[key] = struct{}{} + } +} + func logStr(event *plugins.Event, key string) string { log := event.GetLog() if log == nil || log[key] == nil { @@ -90,6 +410,21 @@ func logStr(event *plugins.Event, key string) string { return strings.TrimSpace(log[key].GetStringValue()) } +func logNumStr(event *plugins.Event, key string) string { + log := event.GetLog() + if log == nil || log[key] == nil { + return "" + } + switch v := log[key].AsInterface().(type) { + case string: + return strings.TrimSpace(v) + case float64: + return strconv.FormatFloat(v, 'f', -1, 64) + default: + return "" + } +} + func logEventCode(event *plugins.Event) string { log := event.GetLog() if log == nil || log[fEventCode] == nil { @@ -149,7 +484,7 @@ func isSystemSID(sid string) bool { // ── In-memory cache ─────────────────────────────────────────────────────────── -type cachedUser struct { +type cachedWindowsUser struct { TenantID string SID string SamAccountName string @@ -161,10 +496,40 @@ type cachedUser struct { LastSeen *time.Time } +type cachedLinuxUser struct { + TenantID string + MachineID string + Hostname string + UIDNumber string + Username string + Active bool + AccountCreatedAt *time.Time + LastLogon *time.Time + AccountDeletedAt *time.Time + LastSeen *time.Time +} + +type resolveIntent struct { + TenantID string + Hostname string + MachineID string +} + var ( - cacheMu sync.Mutex - cache = map[string]*cachedUser{} // "tenant:sid" -> user - dirty = map[string]struct{}{} // ids changed since the last flush + cacheMu sync.Mutex + windowsCache = map[string]*cachedWindowsUser{} + windowsDirty = map[string]struct{}{} + + linuxCache = map[string]*cachedLinuxUser{} + linuxDirty = map[string]struct{}{} + + hostnameMachineID = map[string]string{} + + resolveMu sync.Mutex + resolveQueue []resolveIntent + + dedupMu sync.Mutex + auditdDedup = map[string]time.Time{} ) func applyEvent(tenantID, sid, name, domain, code string, et *time.Time) { @@ -173,10 +538,10 @@ func applyEvent(tenantID, sid, name, domain, code string, et *time.Time) { cacheMu.Lock() defer cacheMu.Unlock() - cu := cache[id] + cu := windowsCache[id] if cu == nil { - cu = &cachedUser{TenantID: tenantID, SID: sid, Active: true} - cache[id] = cu + cu = &cachedWindowsUser{TenantID: tenantID, SID: sid, Active: true} + windowsCache[id] = cu } before := *cu @@ -200,7 +565,7 @@ func applyEvent(tenantID, sid, name, domain, code string, et *time.Time) { } if *cu != before { - dirty[id] = struct{}{} + windowsDirty[id] = struct{}{} } } @@ -246,9 +611,14 @@ var ( type ingestUser struct { TenantID string `json:"tenantId"` - SID string `json:"sid"` - SamAccountName string `json:"samAccountName"` - Domain string `json:"domain"` + Source string `json:"source,omitempty"` + SID string `json:"sid,omitempty"` + SamAccountName string `json:"samAccountName,omitempty"` + Domain string `json:"domain,omitempty"` + MachineID *string `json:"machineId,omitempty"` + UIDNumber *string `json:"uidNumber,omitempty"` + Hostname *string `json:"hostname,omitempty"` + Username *string `json:"username,omitempty"` Active *bool `json:"active"` AccountCreatedAt *time.Time `json:"accountCreatedAt"` LastLogon *time.Time `json:"lastLogon"` @@ -256,15 +626,52 @@ type ingestUser struct { LastSeen *time.Time `json:"lastSeen"` } -func (cu *cachedUser) toIngest() ingestUser { +func (cu *cachedWindowsUser) toIngest() ingestUser { active := cu.Active return ingestUser{ - TenantID: cu.TenantID, SID: cu.SID, SamAccountName: cu.SamAccountName, Domain: cu.Domain, - Active: &active, AccountCreatedAt: cu.AccountCreatedAt, LastLogon: cu.LastLogon, - AccountDeletedAt: cu.AccountDeletedAt, LastSeen: cu.LastSeen, + TenantID: cu.TenantID, + Source: "windows", + SID: cu.SID, + SamAccountName: cu.SamAccountName, + Domain: cu.Domain, + Active: &active, + AccountCreatedAt: cu.AccountCreatedAt, + LastLogon: cu.LastLogon, + AccountDeletedAt: cu.AccountDeletedAt, + LastSeen: cu.LastSeen, } } +func (cu *cachedLinuxUser) toIngest() ingestUser { + active := cu.Active + u := ingestUser{ + Source: "linux", + TenantID: cu.TenantID, + Active: &active, + AccountCreatedAt: cu.AccountCreatedAt, + LastLogon: cu.LastLogon, + AccountDeletedAt: cu.AccountDeletedAt, + LastSeen: cu.LastSeen, + } + if cu.MachineID != "" { + m := cu.MachineID + u.MachineID = &m + } + if cu.UIDNumber != "" { + n := cu.UIDNumber + u.UIDNumber = &n + } + if cu.Hostname != "" { + h := cu.Hostname + u.Hostname = &h + } + if cu.Username != "" { + n := cu.Username + u.Username = &n + } + return u +} + func loadBackendConfig() { cfg := plugins.PluginCfg("com.utmstack") backendURL = strings.TrimRight(cfg.Get("backend").String(), "/") @@ -284,36 +691,105 @@ func request(method, path string, body []byte) (*http.Response, error) { } func seedCacheFromBackend() { + seedWindows() + seedLinux() +} + +func seedWindows() { if backendURL == "" { return } - resp, err := request(http.MethodGet, "/api/v1/ad-audit/users/sync", nil) + resp, err := request(http.MethodGet, "/api/v1/ad-audit/users/sync?source=windows", nil) if err != nil { - _ = catcher.Error("ad-audit: seed request failed", err, nil) + _ = catcher.Error("ad-audit: seed windows request failed", err, nil) return } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - _ = catcher.Error("ad-audit: seed returned non-200", fmt.Errorf("status %d", resp.StatusCode), nil) + _ = catcher.Error("ad-audit: seed windows returned non-200", fmt.Errorf("status %d", resp.StatusCode), nil) return } - var users []ingestUser if err := json.NewDecoder(resp.Body).Decode(&users); err != nil { - _ = catcher.Error("ad-audit: decoding seed failed", err, nil) + _ = catcher.Error("ad-audit: decoding windows seed failed", err, nil) return } + cacheMu.Lock() + for _, u := range users { + windowsCache[u.TenantID+":"+u.SID] = &cachedWindowsUser{ + TenantID: u.TenantID, + SID: u.SID, + SamAccountName: u.SamAccountName, + Domain: u.Domain, + Active: u.Active == nil || *u.Active, + AccountCreatedAt: u.AccountCreatedAt, + LastLogon: u.LastLogon, + AccountDeletedAt: u.AccountDeletedAt, + LastSeen: u.LastSeen, + } + } + cacheMu.Unlock() + catcher.Info(fmt.Sprintf("ad-audit: seeded windows cache with %d users", len(users)), nil) +} +func seedLinux() { + if backendURL == "" { + return + } + resp, err := request(http.MethodGet, "/api/v1/ad-audit/users/sync?source=linux", nil) + if err != nil { + _ = catcher.Error("ad-audit: seed linux request failed", err, nil) + return + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + _ = catcher.Error("ad-audit: seed linux returned non-200", fmt.Errorf("status %d", resp.StatusCode), nil) + return + } + var users []ingestUser + if err := json.NewDecoder(resp.Body).Decode(&users); err != nil { + _ = catcher.Error("ad-audit: decoding linux seed failed", err, nil) + return + } cacheMu.Lock() for _, u := range users { - cache[u.TenantID+":"+u.SID] = &cachedUser{ - TenantID: u.TenantID, SID: u.SID, SamAccountName: u.SamAccountName, Domain: u.Domain, - Active: u.Active == nil || *u.Active, AccountCreatedAt: u.AccountCreatedAt, - LastLogon: u.LastLogon, AccountDeletedAt: u.AccountDeletedAt, LastSeen: u.LastSeen, + cu := &cachedLinuxUser{ + TenantID: u.TenantID, + Active: u.Active == nil || *u.Active, + AccountCreatedAt: u.AccountCreatedAt, + LastLogon: u.LastLogon, + AccountDeletedAt: u.AccountDeletedAt, + LastSeen: u.LastSeen, + } + if u.MachineID != nil { + cu.MachineID = *u.MachineID + } + if u.UIDNumber != nil { + cu.UIDNumber = *u.UIDNumber + } + if u.Hostname != nil { + cu.Hostname = *u.Hostname + } + if u.Username != nil { + cu.Username = *u.Username + } + + if cu.MachineID != "" && cu.Hostname != "" { + hostnameMachineID[cu.TenantID+":"+cu.Hostname] = cu.MachineID } + + var key string + if cu.MachineID != "" && cu.UIDNumber != "" { + key = cu.TenantID + ":linux:resolved:" + cu.MachineID + ":" + cu.UIDNumber + } else if cu.Hostname != "" && cu.Username != "" { + key = cu.TenantID + ":linux:provisional:" + cu.Hostname + ":" + cu.Username + } else { + continue + } + linuxCache[key] = cu } cacheMu.Unlock() - catcher.Info(fmt.Sprintf("ad-audit: seeded cache with %d users", len(users)), nil) + catcher.Info(fmt.Sprintf("ad-audit: seeded linux cache with %d users", len(users)), nil) } func flushLoop(ctx context.Context, interval time.Duration) { @@ -332,25 +808,71 @@ func flushLoop(ctx context.Context, interval time.Duration) { func flush() { cacheMu.Lock() - batch := make([]ingestUser, 0, len(dirty)) - ids := make([]string, 0, len(dirty)) - for id := range dirty { - if cu := cache[id]; cu != nil { + batch := make([]ingestUser, 0, len(windowsDirty)+len(linuxDirty)) + windowsIDs := make([]string, 0, len(windowsDirty)) + linuxIDs := make([]string, 0, len(linuxDirty)) + + for id := range windowsDirty { + if cu := windowsCache[id]; cu != nil { + batch = append(batch, cu.toIngest()) + windowsIDs = append(windowsIDs, id) + } + } + for id := range linuxDirty { + if cu := linuxCache[id]; cu != nil { batch = append(batch, cu.toIngest()) - ids = append(ids, id) + linuxIDs = append(linuxIDs, id) } } - dirty = map[string]struct{}{} // cleared; events during the POST re-dirty independently + windowsDirty = map[string]struct{}{} + linuxDirty = map[string]struct{}{} cacheMu.Unlock() - if len(batch) == 0 { - return + if len(batch) > 0 { + if err := postBatch(batch); err != nil { + _ = catcher.Error("ad-audit: flush failed; re-queuing", err, map[string]any{"count": len(batch)}) + cacheMu.Lock() + for _, id := range windowsIDs { + windowsDirty[id] = struct{}{} + } + for _, id := range linuxIDs { + linuxDirty[id] = struct{}{} + } + cacheMu.Unlock() + } } - if err := postBatch(batch); err != nil { - _ = catcher.Error("ad-audit: flush failed; re-queuing", err, map[string]any{"count": len(batch)}) + + resolveMu.Lock() + intents := resolveQueue + resolveQueue = nil + resolveMu.Unlock() + + for _, intent := range intents { + if err := resolveLinuxIdentity(intent.TenantID, intent.Hostname, intent.MachineID); err != nil { + _ = catcher.Error("ad-audit: resolve failed", err, map[string]any{ + "hostname": intent.Hostname, + "machineID": intent.MachineID, + }) + continue + } cacheMu.Lock() - for _, id := range ids { - dirty[id] = struct{}{} + provisionalPrefix := intent.TenantID + ":linux:provisional:" + intent.Hostname + ":" + for k, cu := range linuxCache { + if !strings.HasPrefix(k, provisionalPrefix) { + continue + } + if cu.UIDNumber == "" { + continue + } + newKey := intent.TenantID + ":linux:resolved:" + intent.MachineID + ":" + cu.UIDNumber + if _, exists := linuxCache[newKey]; exists { + delete(linuxCache, k) + continue + } + cu.MachineID = intent.MachineID + linuxCache[newKey] = cu + delete(linuxCache, k) + linuxDirty[newKey] = struct{}{} } cacheMu.Unlock() } @@ -371,3 +893,42 @@ func postBatch(users []ingestUser) error { } return nil } + +func enqueueResolveIntent(tenantID, hostname, machineID string) { + resolveMu.Lock() + resolveQueue = append(resolveQueue, resolveIntent{ + TenantID: tenantID, + Hostname: hostname, + MachineID: machineID, + }) + resolveMu.Unlock() +} + +var resolveLinuxIdentityFn = resolveLinuxIdentityDefault + +func resolveLinuxIdentity(tenantID, hostname, machineID string) error { + return resolveLinuxIdentityFn(tenantID, hostname, machineID) +} + +func resolveLinuxIdentityDefault(tenantID, hostname, machineID string) error { + if backendURL == "" || tenantID == "" || hostname == "" || machineID == "" { + return nil + } + body, err := json.Marshal(map[string]string{ + "tenant_id": tenantID, + "hostname": hostname, + "machine_id": machineID, + }) + if err != nil { + return err + } + resp, err := request(http.MethodPost, "/api/v1/ad-audit/users/resolve", body) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("resolve returned status %d", resp.StatusCode) + } + return nil +}