Skip to content

Commit ec3d6e3

Browse files
KbayeroKbayero
committed
feat(v12): finish resizable table migration and fix manual execution status, pipeline order persistence and duplicated AD users
1 parent 9a6c4cd commit ec3d6e3

47 files changed

Lines changed: 2344 additions & 1791 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎backend/modules/adaudit/repository/ad_user_pg.go‎

Lines changed: 128 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -25,30 +25,31 @@ func NewADUserRepository(db *gorm.DB) connectors.ADUserRepository {
2525
return &pgADUserRepository{db: db}
2626
}
2727

28-
func (r *pgADUserRepository) Upsert(ctx context.Context, users []domain.ADUser) error {
29-
if len(users) == 0 {
30-
return nil
31-
}
32-
33-
var windowsBatch []domain.ADUser
34-
var linuxResolved []domain.ADUser
35-
var linuxProvisional []domain.ADUser
36-
28+
func splitByIdentity(users []domain.ADUser) (windows, linuxKeyed, linuxByAccount []domain.ADUser) {
3729
for _, u := range users {
3830
switch u.Source {
3931
case "linux":
40-
if u.MachineID != nil && *u.MachineID != "" {
41-
linuxResolved = append(linuxResolved, u)
32+
if u.MachineID != nil && *u.MachineID != "" && u.UIDNumber != nil {
33+
linuxKeyed = append(linuxKeyed, u)
4234
} else {
43-
linuxProvisional = append(linuxProvisional, u)
35+
linuxByAccount = append(linuxByAccount, u)
4436
}
4537
default:
4638
if u.Source == "" {
4739
u.Source = "windows"
4840
}
49-
windowsBatch = append(windowsBatch, u)
41+
windows = append(windows, u)
5042
}
5143
}
44+
return windows, linuxKeyed, linuxByAccount
45+
}
46+
47+
func (r *pgADUserRepository) Upsert(ctx context.Context, users []domain.ADUser) error {
48+
if len(users) == 0 {
49+
return nil
50+
}
51+
52+
windowsBatch, linuxKeyed, linuxByAccount := splitByIdentity(users)
5253

5354
if len(windowsBatch) > 0 {
5455
if err := r.db.WithContext(ctx).Clauses(clause.OnConflict{
@@ -63,73 +64,142 @@ func (r *pgADUserRepository) Upsert(ctx context.Context, users []domain.ADUser)
6364
}
6465
}
6566

66-
if len(linuxResolved) > 0 {
67+
if len(linuxKeyed) > 0 {
68+
// An account first seen without its uid is already a row; give it the
69+
// uid instead of inserting a second one beside it.
70+
for i := range linuxKeyed {
71+
if err := r.adoptUID(ctx, &linuxKeyed[i]); err != nil {
72+
return err
73+
}
74+
}
6775
if err := r.db.WithContext(ctx).Clauses(clause.OnConflict{
6876
Columns: []clause.Column{{Name: "tenant_id"}, {Name: "machine_id"}, {Name: "uid_number"}},
6977
TargetWhere: clause.Where{Exprs: []clause.Expression{clause.Expr{SQL: "source = 'linux'"}}},
7078
DoUpdates: clause.AssignmentColumns([]string{
7179
"username", "hostname", "active",
7280
"account_created_at", "last_logon", "account_deleted_at", "last_seen",
7381
}),
74-
}).Create(&linuxResolved).Error; err != nil {
82+
}).Create(&linuxKeyed).Error; err != nil {
7583
return err
7684
}
7785
}
7886

79-
for i := range linuxProvisional {
80-
u := &linuxProvisional[i]
81-
if u.Hostname == nil || u.Username == nil {
82-
continue
83-
}
84-
var existing domain.ADUser
85-
err := r.db.WithContext(ctx).Where(
86-
"tenant_id = ? AND source = 'linux' AND hostname = ? AND username = ? AND machine_id IS NULL",
87-
u.TenantID, *u.Hostname, *u.Username,
88-
).First(&existing).Error
89-
if errors.Is(err, gorm.ErrRecordNotFound) {
90-
if err := r.db.WithContext(ctx).Create(u).Error; err != nil {
91-
return err
92-
}
93-
} else if err != nil {
87+
for i := range linuxByAccount {
88+
if err := r.upsertByAccount(ctx, &linuxByAccount[i]); err != nil {
9489
return err
95-
} else {
96-
updates := map[string]any{
97-
"active": u.Active,
98-
"account_created_at": u.AccountCreatedAt,
99-
"last_logon": u.LastLogon,
100-
"account_deleted_at": u.AccountDeletedAt,
101-
"last_seen": u.LastSeen,
102-
}
103-
if u.UIDNumber != nil && existing.UIDNumber == nil {
104-
updates["uid_number"] = *u.UIDNumber
105-
}
106-
if err := r.db.WithContext(ctx).Model(&existing).Updates(updates).Error; err != nil {
107-
return err
108-
}
10990
}
11091
}
11192

11293
return nil
11394
}
11495

96+
// adoptUID stamps the uid (and machine) onto the row an account already has
97+
// without one, so the keyed upsert that follows updates it.
98+
func (r *pgADUserRepository) adoptUID(ctx context.Context, u *domain.ADUser) error {
99+
if u.Hostname == nil || u.Username == nil {
100+
return nil
101+
}
102+
return r.db.WithContext(ctx).Exec(`
103+
UPDATE ad_user SET uid_number = ?, machine_id = ?
104+
WHERE id = (
105+
SELECT id FROM ad_user
106+
WHERE tenant_id = ? AND source = 'linux' AND hostname = ? AND username = ? AND uid_number IS NULL
107+
ORDER BY (machine_id IS NOT NULL) DESC, last_seen DESC NULLS LAST, id
108+
LIMIT 1
109+
)
110+
AND NOT EXISTS (
111+
SELECT 1 FROM ad_user x
112+
WHERE x.tenant_id = ? AND x.source = 'linux' AND x.machine_id = ? AND x.uid_number = ?
113+
)`,
114+
*u.UIDNumber, *u.MachineID, u.TenantID, *u.Hostname, *u.Username,
115+
u.TenantID, *u.MachineID, *u.UIDNumber,
116+
).Error
117+
}
118+
119+
// upsertByAccount records an observation of a Linux account by who it is —
120+
// host and user name — because the observation does not carry everything the
121+
// unique index needs. The account keeps one row however many times it is seen,
122+
// and whether or not its machine id has been resolved in the meantime.
123+
func (r *pgADUserRepository) upsertByAccount(ctx context.Context, u *domain.ADUser) error {
124+
if u.Hostname == nil || u.Username == nil {
125+
return nil
126+
}
127+
var existing domain.ADUser
128+
err := r.db.WithContext(ctx).Where(
129+
"tenant_id = ? AND source = 'linux' AND hostname = ? AND username = ?",
130+
u.TenantID, *u.Hostname, *u.Username,
131+
).Order("(uid_number IS NOT NULL) DESC, last_seen DESC NULLS LAST, id").First(&existing).Error
132+
if errors.Is(err, gorm.ErrRecordNotFound) {
133+
return r.db.WithContext(ctx).Create(u).Error
134+
}
135+
if err != nil {
136+
return err
137+
}
138+
139+
updates := map[string]any{
140+
"active": u.Active,
141+
"account_created_at": u.AccountCreatedAt,
142+
"last_logon": u.LastLogon,
143+
"account_deleted_at": u.AccountDeletedAt,
144+
"last_seen": u.LastSeen,
145+
}
146+
// Filling in what the row lacks is only safe while the row has no machine
147+
// id yet: with one, the (machine, uid) pair could already belong to another.
148+
if existing.MachineID == nil {
149+
if u.UIDNumber != nil && existing.UIDNumber == nil {
150+
updates["uid_number"] = *u.UIDNumber
151+
}
152+
if u.MachineID != nil && *u.MachineID != "" {
153+
updates["machine_id"] = *u.MachineID
154+
}
155+
}
156+
return r.db.WithContext(ctx).Model(&existing).Updates(updates).Error
157+
}
158+
159+
const foldProvisionalSQL = `
160+
WITH folded AS (
161+
DELETE FROM ad_user p
162+
USING ad_user r
163+
WHERE p.tenant_id = ? AND p.source = 'linux' AND p.hostname = ? AND p.machine_id IS NULL
164+
AND r.tenant_id = p.tenant_id AND r.source = 'linux' AND r.machine_id = ? AND r.username = p.username
165+
RETURNING r.id AS keep_id, p.last_seen, p.last_logon, p.account_created_at
166+
)
167+
UPDATE ad_user k SET
168+
last_seen = GREATEST(k.last_seen, f.last_seen),
169+
last_logon = GREATEST(k.last_logon, f.last_logon),
170+
account_created_at = LEAST(k.account_created_at, f.account_created_at)
171+
FROM (
172+
SELECT keep_id, MAX(last_seen) AS last_seen, MAX(last_logon) AS last_logon, MIN(account_created_at) AS account_created_at
173+
FROM folded GROUP BY keep_id
174+
) f
175+
WHERE k.id = f.keep_id`
176+
115177
func (r *pgADUserRepository) ResolveLinuxIdentity(ctx context.Context, tenantID, hostname, machineID string) (int64, error) {
116178
if tenantID == "" || hostname == "" || machineID == "" {
117179
return 0, nil
118180
}
119-
result := r.db.WithContext(ctx).Exec(`
120-
UPDATE ad_user SET machine_id = ?
121-
WHERE tenant_id = ?
122-
AND source = 'linux'
123-
AND hostname = ?
124-
AND machine_id IS NULL
125-
AND NOT EXISTS (
126-
SELECT 1 FROM ad_user r2
127-
WHERE r2.tenant_id = ad_user.tenant_id
128-
AND r2.source = 'linux'
129-
AND r2.machine_id = ?
130-
AND r2.uid_number = ad_user.uid_number
131-
)`, machineID, tenantID, hostname, machineID)
132-
return result.RowsAffected, result.Error
181+
var resolved int64
182+
err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
183+
if err := tx.Exec(foldProvisionalSQL, tenantID, hostname, machineID).Error; err != nil {
184+
return err
185+
}
186+
result := tx.Exec(`
187+
UPDATE ad_user SET machine_id = ?
188+
WHERE tenant_id = ?
189+
AND source = 'linux'
190+
AND hostname = ?
191+
AND machine_id IS NULL
192+
AND NOT EXISTS (
193+
SELECT 1 FROM ad_user r2
194+
WHERE r2.tenant_id = ad_user.tenant_id
195+
AND r2.source = 'linux'
196+
AND r2.machine_id = ?
197+
AND r2.uid_number = ad_user.uid_number
198+
)`, machineID, tenantID, hostname, machineID)
199+
resolved = result.RowsAffected
200+
return result.Error
201+
})
202+
return resolved, err
133203
}
134204

135205
func applyStatus(q *gorm.DB, status string) *gorm.DB {

‎backend/modules/adaudit/repository/ad_user_pg_test.go‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
package repository
2+
3+
import (
4+
"testing"
5+
6+
"github.com/utmstack/utmstack/backend/modules/adaudit/domain"
7+
)
8+
9+
func str(s string) *string { return &s }
10+
func uid(n uint32) *uint32 { return &n }
11+
12+
// The unique index on (tenant, machine, uid) cannot tell rows apart when the
13+
// uid is missing — NULLs never collide — so an account seen without a uid was
14+
// inserted again on every observation. Only a fully keyed account may take the
15+
// conflict-based path.
16+
func TestSplitByIdentity(t *testing.T) {
17+
keyed := domain.ADUser{Source: "linux", MachineID: str("m-1"), UIDNumber: uid(1000), Hostname: str("h"), Username: str("root")}
18+
machineWithoutUID := domain.ADUser{Source: "linux", MachineID: str("m-1"), Hostname: str("h"), Username: str("root")}
19+
uidWithoutMachine := domain.ADUser{Source: "linux", UIDNumber: uid(1000), Hostname: str("h"), Username: str("root")}
20+
blankMachine := domain.ADUser{Source: "linux", MachineID: str(""), UIDNumber: uid(1000), Hostname: str("h"), Username: str("root")}
21+
windows := domain.ADUser{Source: "windows", SID: str("S-1-5-21-1")}
22+
unlabelled := domain.ADUser{SID: str("S-1-5-21-2")}
23+
24+
win, linuxKeyed, byAccount := splitByIdentity([]domain.ADUser{keyed, machineWithoutUID, uidWithoutMachine, blankMachine, windows, unlabelled})
25+
26+
if len(linuxKeyed) != 1 || linuxKeyed[0].UIDNumber == nil || *linuxKeyed[0].UIDNumber != 1000 || *linuxKeyed[0].MachineID != "m-1" {
27+
t.Fatalf("only the account with machine and uid is keyed, got %+v", linuxKeyed)
28+
}
29+
if len(byAccount) != 3 {
30+
t.Fatalf("machine without uid, uid without machine and a blank machine are matched by account, got %d", len(byAccount))
31+
}
32+
if len(win) != 2 {
33+
t.Fatalf("windows accounts keep the SID path, got %d", len(win))
34+
}
35+
if win[1].Source != "windows" {
36+
t.Fatalf("an unlabelled account defaults to windows, got %q", win[1].Source)
37+
}
38+
}

‎backend/modules/eventprocessing/dto/pipeline.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ type PipelineResponse struct {
2323
Active bool `json:"active"`
2424
DataTypes []string `json:"dataTypes"`
2525
Order int32 `json:"order"`
26+
Position int32 `json:"position"`
2627
}
2728

2829
type PipelineFilters struct {

‎backend/modules/eventprocessing/usecase/pipeline.go‎

Lines changed: 45 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -257,39 +257,36 @@ func (u *pipelineUsecase) GetByRelPath(ctx context.Context, relPath string) (*dt
257257
}
258258

259259
func (u *pipelineUsecase) List(ctx context.Context, f dto.PipelineFilters) (*connectors.ListResult[dto.PipelineResponse], error) {
260-
all := u.store.List(authz.TenantIDFromContext(ctx))
261260
tenant := authz.TenantIDFromContext(ctx)
261+
all := u.store.List(tenant)
262262

263-
// Apply in-memory filters.
264-
out := make([]dto.PipelineResponse, 0, len(all))
263+
// The position is taken over everything the tenant runs, before any filter,
264+
// so a pipeline reads the same number whatever the view is narrowed to.
265+
visible := make([]dto.PipelineResponse, 0, len(all))
265266
for i := range all {
266-
e := &all[i]
267-
if !visiblePipeline(e, tenant) {
268-
continue
267+
if visiblePipeline(&all[i], tenant) {
268+
visible = append(visible, *toFilterResponse(&all[i]))
269269
}
270-
if f.IsActiveEq != nil && e.Active != *f.IsActiveEq {
270+
}
271+
sequencePipelines(visible, u.config.PipelineOrder(tenant))
272+
273+
out := make([]dto.PipelineResponse, 0, len(visible))
274+
for _, resp := range visible {
275+
if f.IsActiveEq != nil && resp.Active != *f.IsActiveEq {
271276
continue
272277
}
273-
if f.SystemEq != nil && e.System != *f.SystemEq {
278+
if f.SystemEq != nil && resp.System != *f.SystemEq {
274279
continue
275280
}
276-
if f.RelPathContains != nil && !strings.Contains(e.RelPath, *f.RelPathContains) {
281+
if f.RelPathContains != nil && !strings.Contains(resp.RelPath, *f.RelPathContains) {
277282
continue
278283
}
279-
resp := toFilterResponse(e)
280284
if f.DataTypeEq != nil && *f.DataTypeEq != "" && !hasDataType(resp.DataTypes, *f.DataTypeEq) {
281285
continue
282286
}
283-
out = append(out, *resp)
287+
out = append(out, resp)
284288
}
285289

286-
sort.Slice(out, func(i, j int) bool {
287-
if out[i].Order != out[j].Order {
288-
return out[i].Order < out[j].Order
289-
}
290-
return out[i].RelPath < out[j].RelPath
291-
})
292-
293290
total := int64(len(out))
294291

295292
// Pagination.
@@ -354,6 +351,36 @@ func (u *pipelineUsecase) SetOrder(ctx context.Context, order []string) error {
354351
return u.config.SetPipelineOrder(authz.TenantIDFromContext(ctx), order)
355352
}
356353

354+
// sequencePipelines puts pipelines in the order the engine runs them for a
355+
// tenant and numbers them by it (Position). The names the tenant listed come
356+
// first, in its order; the rest follow in the order their own files declare.
357+
// This is the engine's rule, and the listing has to agree with it or a saved
358+
// order shows up as unchanged the next time the page loads.
359+
func sequencePipelines(items []dto.PipelineResponse, saved []string) {
360+
rank := make(map[string]int, len(saved))
361+
for i, name := range saved {
362+
if _, seen := rank[name]; !seen {
363+
rank[name] = i
364+
}
365+
}
366+
sort.SliceStable(items, func(i, j int) bool {
367+
ri, listedI := rank[pipelineIdentity(items[i].RelPath)]
368+
rj, listedJ := rank[pipelineIdentity(items[j].RelPath)]
369+
switch {
370+
case listedI && listedJ:
371+
return ri < rj
372+
case listedI != listedJ:
373+
return listedI
374+
case items[i].Order != items[j].Order:
375+
return items[i].Order < items[j].Order
376+
}
377+
return items[i].RelPath < items[j].RelPath
378+
})
379+
for i := range items {
380+
items[i].Position = int32(i)
381+
}
382+
}
383+
357384
// pipelineIdentity is the name the engine matches on: the file's base name
358385
// without its extension, the same identity used in the disabled list.
359386
func pipelineIdentity(relPath string) string {

0 commit comments

Comments
 (0)