439 lines
21 KiB
Go
439 lines
21 KiB
Go
package mysql
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"reflect"
|
|
"sort"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.yuntuo.hk/k2safe/orbit/internal/credentialvault"
|
|
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/schedule"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
|
)
|
|
|
|
func TestCredentialPermissionOperationValidation(t *testing.T) {
|
|
base := CredentialChangeInput{NamespaceID: testID(t), CredentialID: testID(t), ExpectedRevision: 1, Action: credentialPermissionsAction, Reason: "replace explicit actions", Actions: []string{"schedules:manage", "tasks:read"}}
|
|
if err := validateCredentialChangeInput(base); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, actions := range [][]string{nil, {}, {""}, {"*"}, {"schedules:*"}, {"unknown"}, {"tasks:read", "tasks:read"}, {" tasks:read"}} {
|
|
bad := base
|
|
bad.Actions = actions
|
|
if err := validateCredentialChangeInput(bad); fault.Code(err) != "INVALID_ARGUMENT" {
|
|
t.Fatalf("actions %q accepted: %v", actions, err)
|
|
}
|
|
}
|
|
for _, field := range []string{"validity", "grace"} {
|
|
bad := base
|
|
if field == "validity" {
|
|
bad.ValidFor = time.Hour
|
|
} else {
|
|
bad.GraceFor = time.Hour
|
|
}
|
|
if err := validateCredentialChangeInput(bad); fault.Code(err) != "INVALID_ARGUMENT" {
|
|
t.Fatalf("permission change accepted %s: %v", field, err)
|
|
}
|
|
}
|
|
op := CredentialOperation{Version: 3, OperationID: testID(t), NamespaceID: base.NamespaceID, CredentialID: base.CredentialID, ExpectedRevision: 1, Action: credentialPermissionsAction, Reason: base.Reason, Actions: base.Actions}
|
|
if err := ValidateCredentialOperation(op); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, mutate := range []func(*CredentialOperation){
|
|
func(v *CredentialOperation) { v.Version = 1 },
|
|
func(v *CredentialOperation) { v.Version = 2 },
|
|
func(v *CredentialOperation) { v.Actions = []string{"tasks:read", "schedules:manage"} },
|
|
func(v *CredentialOperation) { v.Actions = []string{"tasks:read", "tasks:read"} },
|
|
func(v *CredentialOperation) { v.Actions = nil },
|
|
func(v *CredentialOperation) { v.Token = "a token is forbidden" },
|
|
func(v *CredentialOperation) { v.ExpiresAt = timePtr(time.Now().Add(time.Hour)) },
|
|
func(v *CredentialOperation) { v.PreviousValidUntil = timePtr(time.Now().Add(time.Hour)) },
|
|
func(v *CredentialOperation) { v.Action = credentialExpiryAction },
|
|
} {
|
|
bad := op
|
|
mutate(&bad)
|
|
if err := ValidateCredentialOperation(bad); fault.Code(err) != "INVALID_ARGUMENT" {
|
|
t.Fatalf("invalid permission operation accepted: %v", err)
|
|
}
|
|
}
|
|
// 旧固定文件的精确字段、序列化原文及摘要保持不变；包括旧轮换的秘密字段。
|
|
type legacyOperation struct {
|
|
Version int `json:"version"`
|
|
OperationID core.ID `json:"operation_id"`
|
|
NamespaceID core.ID `json:"namespace_id"`
|
|
CredentialID core.ID `json:"credential_id"`
|
|
ExpectedRevision uint64 `json:"expected_revision"`
|
|
Action string `json:"action"`
|
|
Reason string `json:"reason"`
|
|
Token string `json:"token,omitempty"`
|
|
PreviousValidUntil *time.Time `json:"previous_valid_until,omitempty"`
|
|
ExpiresAt *time.Time `json:"expires_at,omitempty"`
|
|
}
|
|
for _, version := range []int{1, 2} {
|
|
for _, action := range []string{credentialRotateAction, credentialRevokeAction, credentialExpiryAction, credentialFinalizeAction} {
|
|
if version == 1 && action != credentialRotateAction && action != credentialRevokeAction {
|
|
continue
|
|
}
|
|
old := legacyOperation{Version: version, OperationID: testID(t), NamespaceID: base.NamespaceID, CredentialID: base.CredentialID, ExpectedRevision: 1, Action: action, Reason: "legacy frozen shape"}
|
|
if action == credentialRotateAction {
|
|
old.Token, _, _ = auth.Issue(base.NamespaceID, base.CredentialID)
|
|
old.ExpiresAt = timePtr(time.Now().Add(time.Hour))
|
|
}
|
|
raw, _ := json.Marshal(old)
|
|
var current CredentialOperation
|
|
if err := json.Unmarshal(raw, ¤t); err != nil || ValidateCredentialOperation(current) != nil {
|
|
t.Fatal("legacy operation no longer valid", err)
|
|
}
|
|
currentRaw, _ := json.Marshal(current)
|
|
oldDigest, _ := requestDigest(old)
|
|
currentDigest, _ := requestDigest(current)
|
|
if !bytes.Equal(raw, currentRaw) || oldDigest != currentDigest {
|
|
t.Fatal("legacy JSON or digest changed")
|
|
}
|
|
for _, actions := range [][]string{{}, {"tasks:read"}} {
|
|
current.Actions = actions
|
|
if err := ValidateCredentialOperation(current); fault.Code(err) != "INVALID_ARGUMENT" {
|
|
t.Fatal("legacy operation accepted actions", err)
|
|
}
|
|
bad := base
|
|
bad.Action, bad.Actions = action, actions
|
|
if err := validateCredentialChangeInput(bad); fault.Code(err) != "INVALID_ARGUMENT" {
|
|
t.Fatal("legacy preparation accepted actions", err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestIntegrationCredentialPermissions(t *testing.T) {
|
|
ctx, db := overlapDB(t)
|
|
ring, err := credentialvault.Read(strings.NewReader(`{"active":"test","keys":{"test":"` + base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{43}, 32)) + `"}}`))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var initial BootstrapResult
|
|
if err := Bootstrap(ctx, db, BootstrapInput{Application: "permissions-" + string(testID(t)), Namespace: "default", Environment: "test", Queue: "work", ValidFor: 48 * time.Hour}, func(v BootstrapResult) error { initial = v; return nil }, ring); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
oldToken, _ := auth.Parse(initial.Token)
|
|
tasks := NewTaskStore(db)
|
|
otherQueue, err := service.New(tasks).CreateQueue(ctx, oldToken, string(testID(t)), service.QueueInput{Name: "outside-scope", MaxConcurrency: 1, RatePerSecond: 1})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// 隔离测试夹具先收窄队列，证明添加动作不能隐式恢复命名空间通配范围。
|
|
queues, _ := json.Marshal([]string{string(initial.QueueID)})
|
|
if _, err := db.ExecContext(ctx, `UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',CAST(? AS JSON)) WHERE namespace_id=? AND id=?`, queues, initial.NamespaceID, initial.CredentialID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
schedules, err := NewScheduleStore(db, testID(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
scheduleAPI := schedule.New(schedules)
|
|
plan := schedule.CreateInput{Name: "permission-check", Paused: true, Definition: schedule.Definition{
|
|
Rule: schedule.Rule{Kind: "FIXED_RATE", Timezone: "UTC", StartAt: time.Now().UTC().Add(time.Hour), IntervalMS: 60000},
|
|
Policy: schedule.Policy{Misfire: "SKIP", Overlap: "PARALLEL", CatchupLimit: 1, CatchupWindowMS: 30000, GraceMS: 1000},
|
|
Task: service.SubmitInput{QueueID: initial.QueueID, Handler: "permission.check", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)},
|
|
}}
|
|
if _, err := scheduleAPI.Create(ctx, oldToken, string(testID(t)), plan); fault.Code(err) != "PERMISSION_DENIED" {
|
|
t.Fatal("schedule permission unexpectedly present before change", err)
|
|
}
|
|
prepare := func(revision uint64, actions []string) CredentialOperation {
|
|
t.Helper()
|
|
op, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: revision, Action: credentialPermissionsAction, Reason: "explicit complete action set", Actions: actions})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return op
|
|
}
|
|
rotate, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: 1, Action: credentialRotateAction, ValidFor: 24 * time.Hour, GraceFor: time.Hour, Reason: "test two retained generations"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err = ApplyCredentialOperation(ctx, db, rotate, ring); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
before, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// 快照只在测试内比较，绝不把摘要、密文或 Token 打到失败日志。
|
|
protected := func() string {
|
|
t.Helper()
|
|
var out string
|
|
if err := db.QueryRowContext(ctx, `SELECT CAST(JSON_OBJECT('namespace',HEX(namespace_id),'id',HEX(id),'name',name,'current',HEX(secret_digest),'previous',HEX(previous_secret_digest),'key',secret_key_id,'cipher',HEX(secret_ciphertext),'expiry',expires_at,'grace',previous_valid_until,'revoked',revoked_at,'queues',JSON_EXTRACT(permissions,'$.queues'),'created',created_at) AS CHAR) FROM orbit_credentials WHERE namespace_id=? AND id=?`, initial.NamespaceID, initial.CredentialID).Scan(&out); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return out
|
|
}
|
|
frozen := protected()
|
|
actions := []string{"tasks:submit", "schedules:read", "queues:read", "schedules:manage"}
|
|
op := prepare(2, actions)
|
|
actions[0] = "tasks:cancel" // Prepare 必须复制输入，不共享调用方数组。
|
|
if op.Version != 3 || op.Token != "" || !sort.StringsAreSorted(op.Actions) || op.Actions[len(op.Actions)-1] != "tasks:submit" {
|
|
t.Fatal("permission operation did not freeze a canonical secret-free set")
|
|
}
|
|
trigger := "permissions_audit_" + string(testID(t))
|
|
if _, err := db.ExecContext(ctx, "CREATE TRIGGER "+trigger+" BEFORE INSERT ON orbit_audit_events FOR EACH ROW BEGIN IF NEW.namespace_id=UNHEX('"+string(initial.NamespaceID)+"') AND NEW.action='credential.set_permissions' THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='permission audit failure test'; END IF; END"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer db.ExecContext(ctx, "DROP TRIGGER IF EXISTS "+trigger)
|
|
if _, err := ApplyCredentialOperation(ctx, db, op); err == nil {
|
|
t.Fatal("audit failure accepted")
|
|
}
|
|
failed, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID)
|
|
if err != nil || !reflect.DeepEqual(failed, before) || protected() != frozen {
|
|
t.Fatal("failed audit did not roll back credential state", err)
|
|
}
|
|
var rows int
|
|
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_idempotency_keys WHERE namespace_id=? AND id=?`, initial.NamespaceID, op.OperationID).Scan(&rows); err != nil || rows != 0 {
|
|
t.Fatal("failed audit kept operation identity", err)
|
|
}
|
|
if _, err := db.ExecContext(ctx, "DROP TRIGGER "+trigger); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
grant, err := ApplyCredentialOperation(ctx, db, op) // 无 vault 也不能清除既有密文。
|
|
if err != nil || grant.Revision != 3 || grant.Deduplicated || protected() != frozen {
|
|
t.Fatal("permission grant changed protected credential state", err)
|
|
}
|
|
newToken, _ := auth.Parse(rotate.Token)
|
|
assertActions := func(want auth.Permissions) {
|
|
t.Helper()
|
|
for _, token := range []auth.Token{oldToken, newToken} {
|
|
err := WithTx(ctx, db, func(tx *sql.Tx) error {
|
|
p, err := authenticate(ctx, tx, token)
|
|
if err == nil && !reflect.DeepEqual(p, want) {
|
|
t.Error("token authentication did not reload exact current permissions")
|
|
}
|
|
return err
|
|
})
|
|
if err != nil {
|
|
t.Fatal("preserved token no longer authenticates", err)
|
|
}
|
|
if _, err := tasks.GetQueue(ctx, token, initial.QueueID); err != nil {
|
|
t.Fatal("original queue unavailable", err)
|
|
}
|
|
if _, err := tasks.GetQueue(ctx, token, otherQueue.Queue.ID); fault.Code(err) != "NOT_FOUND" {
|
|
t.Fatal("permission change expanded queue scope", err)
|
|
}
|
|
}
|
|
}
|
|
after, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID)
|
|
if err != nil || !reflect.DeepEqual(after.Permissions.Actions, op.Actions) || !reflect.DeepEqual(after.Permissions.Queues, before.Permissions.Queues) {
|
|
t.Fatal("exact permissions status incorrect", err)
|
|
}
|
|
assertActions(after.Permissions)
|
|
created, err := scheduleAPI.Create(ctx, oldToken, string(testID(t)), plan)
|
|
if err != nil {
|
|
t.Fatal("new permission did not immediately allow real schedule creation", err)
|
|
}
|
|
if _, err := scheduleAPI.Get(ctx, newToken, created.Schedule.ID); err != nil {
|
|
t.Fatal("current token did not receive new schedule read permission", err)
|
|
}
|
|
forbiddenPlan := plan
|
|
forbiddenPlan.Name = "outside-queue"
|
|
forbiddenPlan.Definition.Task.QueueID = otherQueue.Queue.ID
|
|
if _, err := scheduleAPI.Create(ctx, newToken, string(testID(t)), forbiddenPlan); fault.Code(err) != "PERMISSION_DENIED" {
|
|
t.Fatal("new schedule action bypassed preserved queue scope", err)
|
|
}
|
|
var auditBefore, auditAfter []byte
|
|
if err := db.QueryRowContext(ctx, "SELECT `before`,`after` FROM orbit_audit_events WHERE namespace_id=? AND operation_id=?", initial.NamespaceID, op.OperationID).Scan(&auditBefore, &auditAfter); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for i, raw := range [][]byte{auditBefore, auditAfter} {
|
|
var audit struct {
|
|
Permissions auth.Permissions `json:"permissions"`
|
|
}
|
|
if err := json.Unmarshal(raw, &audit); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
want := before.Permissions
|
|
if i == 1 {
|
|
want = after.Permissions
|
|
}
|
|
if !reflect.DeepEqual(audit.Permissions, want) || bytes.Contains(raw, []byte("secret_")) || bytes.Contains(raw, []byte(rotate.Token)) {
|
|
t.Fatal("permissions audit missing exact scope or leaked secret")
|
|
}
|
|
}
|
|
if replay, err := ApplyCredentialOperation(ctx, db, op); err != nil || !replay.Deduplicated || replay.Revision != grant.Revision {
|
|
t.Fatal("permission replay changed receipt", err)
|
|
}
|
|
tampered := op
|
|
tampered.Actions = []string{"queues:read"}
|
|
if _, err := ApplyCredentialOperation(ctx, db, tampered); fault.Code(err) != "IDEMPOTENCY_CONFLICT" {
|
|
t.Fatal("changed frozen actions accepted", err)
|
|
}
|
|
wrongScope := op
|
|
wrongScope.NamespaceID = testID(t)
|
|
if _, err := ApplyCredentialOperation(ctx, db, wrongScope); fault.Code(err) != "NOT_FOUND" {
|
|
t.Fatal("cross namespace credential accepted", err)
|
|
}
|
|
remove := prepare(3, []string{"queues:read"})
|
|
if _, err := ApplyCredentialOperation(ctx, db, remove); err != nil || protected() != frozen {
|
|
t.Fatal("permission removal changed protected state", err)
|
|
}
|
|
assertActions(auth.Permissions{Actions: remove.Actions, Queues: before.Permissions.Queues})
|
|
for _, token := range []auth.Token{oldToken, newToken} {
|
|
if err := tasks.AuthorizeSubmit(ctx, token); fault.Code(err) != "PERMISSION_DENIED" {
|
|
t.Fatal("removed permission still usable", err)
|
|
}
|
|
if _, err := scheduleAPI.Get(ctx, token, created.Schedule.ID); fault.Code(err) != "PERMISSION_DENIED" {
|
|
t.Fatal("removed schedule permission still usable", err)
|
|
}
|
|
}
|
|
if _, err := ApplyCredentialOperation(ctx, db, op); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assertActions(auth.Permissions{Actions: remove.Actions, Queues: before.Permissions.Queues})
|
|
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_audit_events WHERE namespace_id=? AND resource_id=? AND action='credential.set_permissions'`, initial.NamespaceID, initial.CredentialID).Scan(&rows); err != nil || rows != 2 {
|
|
t.Fatal("replay duplicated permission audit", err)
|
|
}
|
|
var foreignKeys int
|
|
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&foreignKeys); err != nil || foreignKeys != 0 {
|
|
t.Fatal("schema acquired a foreign key", err)
|
|
}
|
|
}
|
|
|
|
func TestIntegrationCredentialPermissionsLegacyReplay(t *testing.T) {
|
|
ctx, db := overlapDB(t)
|
|
var initial BootstrapResult
|
|
if err := Bootstrap(ctx, db, BootstrapInput{Application: "permission-legacy-" + string(testID(t)), Namespace: "default", Environment: "test", Queue: "work"}, func(v BootstrapResult) error { initial = v; return nil }); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
old, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: 1, Action: credentialRotateAction, ValidFor: time.Hour, Reason: "legacy v1 replay"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
old.Version = 1
|
|
if _, err := ApplyCredentialOperation(ctx, db, old); err != nil {
|
|
t.Fatal("legacy v1 no longer applies", err)
|
|
}
|
|
permissions, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: 2, Action: credentialPermissionsAction, Actions: []string{"queues:read"}, Reason: "new permission revision"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := ApplyCredentialOperation(ctx, db, permissions); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if replay, err := ApplyCredentialOperation(ctx, db, old); err != nil || !replay.Deduplicated || replay.Revision != 2 {
|
|
t.Fatal("legacy receipt no longer replays after v3", err)
|
|
}
|
|
expiry, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: 3, Action: credentialExpiryAction, Reason: "legacy v2 still generated"})
|
|
if err != nil || expiry.Version != 2 {
|
|
t.Fatal("old action changed fixed format", err)
|
|
}
|
|
if _, err := ApplyCredentialOperation(ctx, db, expiry); err != nil {
|
|
t.Fatal("legacy v2 no longer applies", err)
|
|
}
|
|
status, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID)
|
|
if err != nil || status.Revision != 4 || status.ExpiresAt != nil || !reflect.DeepEqual(status.Permissions.Actions, permissions.Actions) {
|
|
t.Fatal("legacy changes overwrote new permissions", err)
|
|
}
|
|
}
|
|
|
|
// 竞争由专用事务持有真实凭据行锁，确认每个请求实际到达 InnoDB 等待后再放行。
|
|
func competeCredentialPermissions(t *testing.T, ctx context.Context, db *sql.DB, namespace, credential core.ID, operations ...func() error) []error {
|
|
t.Helper()
|
|
guard, err := db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer guard.Rollback()
|
|
var connection uint64
|
|
if err := guard.QueryRowContext(ctx, `SELECT CONNECTION_ID()`).Scan(&connection); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var revision uint64
|
|
if err := guard.QueryRowContext(ctx, `SELECT revision FROM orbit_credentials WHERE namespace_id=? AND id=? FOR UPDATE`, namespace, credential).Scan(&revision); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
results := make([]chan error, len(operations))
|
|
for i, operation := range operations {
|
|
results[i] = make(chan error, 1)
|
|
go func(ch chan error, run func() error) { ch <- run() }(results[i], operation)
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
for {
|
|
select {
|
|
case early := <-results[i]:
|
|
t.Fatalf("permission competitor returned before DB lock: %v", early)
|
|
default:
|
|
}
|
|
var waiting int
|
|
if err := db.QueryRowContext(ctx, `SELECT COUNT(DISTINCT w.REQUESTING_THREAD_ID) FROM performance_schema.data_lock_waits w JOIN performance_schema.threads b ON b.THREAD_ID=w.BLOCKING_THREAD_ID WHERE b.PROCESSLIST_ID=?`, connection).Scan(&waiting); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if waiting == i+1 {
|
|
break
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("only %d/%d requests reached credential lock", waiting, i+1)
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
}
|
|
if err := guard.Commit(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
errs := make([]error, len(operations))
|
|
for i, ch := range results {
|
|
select {
|
|
case errs[i] = <-ch:
|
|
case <-ctx.Done():
|
|
t.Fatal(ctx.Err())
|
|
}
|
|
}
|
|
return errs
|
|
}
|
|
|
|
func TestIntegrationCredentialPermissionsConcurrency(t *testing.T) {
|
|
ctx, db := overlapDB(t)
|
|
var initial BootstrapResult
|
|
if err := Bootstrap(ctx, db, BootstrapInput{Application: "permission-race-" + string(testID(t)), Namespace: "default", Environment: "test", Queue: "work"}, func(v BootstrapResult) error { initial = v; return nil }); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
prepare := func(revision uint64, action string) CredentialOperation {
|
|
t.Helper()
|
|
op, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: revision, Action: credentialPermissionsAction, Reason: "permission CAS race", Actions: []string{action}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return op
|
|
}
|
|
op := prepare(1, "queues:read")
|
|
var receipts [2]CredentialChangeReceipt
|
|
errs := competeCredentialPermissions(t, ctx, db, initial.NamespaceID, initial.CredentialID,
|
|
func() (err error) { receipts[0], err = ApplyCredentialOperation(ctx, db, op); return },
|
|
func() (err error) { receipts[1], err = ApplyCredentialOperation(ctx, db, op); return })
|
|
if errs[0] != nil || errs[1] != nil || receipts[0].Deduplicated == receipts[1].Deduplicated || receipts[0].Revision != 2 || receipts[1].Revision != 2 {
|
|
t.Fatalf("concurrent replay: %v", errs)
|
|
}
|
|
left, right := prepare(2, "tasks:read"), prepare(2, "schedules:read")
|
|
errs = competeCredentialPermissions(t, ctx, db, initial.NamespaceID, initial.CredentialID,
|
|
func() error { _, err := ApplyCredentialOperation(ctx, db, left); return err },
|
|
func() error { _, err := ApplyCredentialOperation(ctx, db, right); return err })
|
|
if !((errs[0] == nil && fault.Code(errs[1]) == "STATE_CONFLICT") || (errs[1] == nil && fault.Code(errs[0]) == "STATE_CONFLICT")) {
|
|
t.Fatalf("permission CAS did not select exactly one winner: %v", errs)
|
|
}
|
|
winner := left
|
|
if errs[1] == nil {
|
|
winner = right
|
|
}
|
|
status, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID)
|
|
if err != nil || status.Revision != 3 || !reflect.DeepEqual(status.Permissions.Actions, winner.Actions) {
|
|
t.Fatal("CAS did not persist winner", err)
|
|
}
|
|
var count int
|
|
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_audit_events WHERE namespace_id=? AND action='credential.set_permissions'`, initial.NamespaceID).Scan(&count); err != nil || count != 2 {
|
|
t.Fatal("losing CAS or replay created audit", err)
|
|
}
|
|
}
|