284 lines
11 KiB
Go
284 lines
11 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"os"
|
||
"strings"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
|
||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||
"git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver"
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
||
"git.yuntuo.hk/k2safe/orbit/pulse/worker"
|
||
"github.com/coder/websocket"
|
||
"github.com/coder/websocket/wsjson"
|
||
)
|
||
|
||
func overlapDB(t *testing.T) (context.Context, *sql.DB) {
|
||
t.Helper()
|
||
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
||
if dsn == "" {
|
||
t.Skip("requires disposable MySQL")
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||
t.Cleanup(cancel)
|
||
db, err := Open(ctx, dsn, 12, 3)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() { db.Close() })
|
||
if err = Migrate(ctx, db); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return ctx, db
|
||
}
|
||
|
||
func TestIntegrationCredentialOverlap(t *testing.T) {
|
||
ctx, db := overlapDB(t)
|
||
f := newExecutionFixture(t, ctx, db)
|
||
old := f.token
|
||
// 先使用有限有效期，再原地改长期；在途授权、秘密及绑定均不改变。
|
||
change := func(action string, revision uint64, grace time.Duration) CredentialOperation {
|
||
t.Helper()
|
||
op, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: old.NamespaceID, CredentialID: old.CredentialID, ExpectedRevision: revision, Action: action, GraceFor: grace, Reason: "credential overlap regression"})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return op
|
||
}
|
||
f.exec(`UPDATE orbit_credentials SET expires_at=DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 30 DAY) WHERE namespace_id=? AND id=?`, old.NamespaceID, old.CredentialID)
|
||
task := f.submit(1)
|
||
_, ref := f.claim()
|
||
if _, err := f.a.Start(ctx, old, f.binding, testID(t), ref); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
expiry := change(credentialExpiryAction, 1, 0)
|
||
receipt, err := ApplyCredentialOperation(ctx, db, expiry)
|
||
if err != nil || receipt.ExpiresAt != nil || expiry.Token != "" {
|
||
t.Fatalf("expiry update: %v", err)
|
||
}
|
||
if err := f.a.Authorize(ctx, old); err != nil {
|
||
t.Fatal("original token lost after expiry update", err)
|
||
}
|
||
rotation := change(credentialRotateAction, 2, 7*24*time.Hour)
|
||
var wg sync.WaitGroup
|
||
receipts := make([]CredentialChangeReceipt, 2)
|
||
errs := make([]error, 2)
|
||
for i := range receipts {
|
||
wg.Add(1)
|
||
go func(i int) { defer wg.Done(); receipts[i], errs[i] = ApplyCredentialOperation(ctx, db, rotation) }(i)
|
||
}
|
||
wg.Wait()
|
||
if errs[0] != nil || errs[1] != nil || receipts[0].Deduplicated == receipts[1].Deduplicated {
|
||
t.Fatalf("concurrent replay: %v", errs)
|
||
}
|
||
current, _ := auth.Parse(rotation.Token)
|
||
for _, token := range []auth.Token{old, current} {
|
||
if err := f.a.Authorize(ctx, token); err != nil {
|
||
t.Fatal("overlap auth rejected", err)
|
||
}
|
||
}
|
||
status, err := GetCredentialStatus(ctx, db, old.NamespaceID, old.CredentialID)
|
||
if err != nil || status.ExpiresAt != nil || status.PreviousValidUntil == nil {
|
||
t.Fatal("missing permanent/overlap status", err)
|
||
}
|
||
originalUntil := *status.PreviousValidUntil
|
||
// 过渡期间第二次轮换必须冲突，而不能挤掉仍在工作的上一代。
|
||
blocked := change(credentialRotateAction, 3, time.Hour)
|
||
if _, err := ApplyCredentialOperation(ctx, db, blocked); fault.Code(err) != "STATE_CONFLICT" {
|
||
t.Fatal("active predecessor overwritten", err)
|
||
}
|
||
renewed, err := f.a.Renew(ctx, old, f.binding, testID(t), ref)
|
||
if err != nil || !renewed.Lease.Active {
|
||
t.Fatal("in-flight old-token renew failed", err)
|
||
}
|
||
if _, err := f.a.Finish(ctx, old, f.binding, testID(t), worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`{"ok":true}`)}); err != nil {
|
||
t.Fatal("in-flight finish rejected", err)
|
||
}
|
||
if f.get(task.ID).State != pulse.TaskSucceeded || f.used() != 0 {
|
||
t.Fatal("task/permit changed by rotation")
|
||
}
|
||
final := change(credentialFinalizeAction, 3, 0)
|
||
if _, err := ApplyCredentialOperation(ctx, db, final); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := f.a.Authorize(ctx, old); fault.Code(err) != "UNAUTHENTICATED" {
|
||
t.Fatal("retired token accepted", err)
|
||
}
|
||
if err := f.a.Authorize(ctx, current); err != nil {
|
||
t.Fatal("current token rejected", err)
|
||
}
|
||
replay, err := ApplyCredentialOperation(ctx, db, rotation)
|
||
if err != nil || !replay.Deduplicated || !replay.PreviousValidUntil.Equal(originalUntil) {
|
||
t.Fatal("old receipt drifted", err)
|
||
}
|
||
if err := f.a.Authorize(ctx, old); err == nil {
|
||
t.Fatal("replay resurrected predecessor")
|
||
}
|
||
// 整张凭据撤销必须同时使两代令牌失效。
|
||
next := change(credentialRotateAction, 4, time.Hour)
|
||
if _, err := ApplyCredentialOperation(ctx, db, next); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
newest, _ := auth.Parse(next.Token)
|
||
revoke := change(credentialRevokeAction, 5, 0)
|
||
if _, err := ApplyCredentialOperation(ctx, db, revoke); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
for _, token := range []auth.Token{old, current, newest} {
|
||
if err := f.a.Authorize(ctx, token); fault.Code(err) != "UNAUTHENTICATED" {
|
||
t.Fatal("revocation missed a generation", err)
|
||
}
|
||
}
|
||
var audits string
|
||
if err := db.QueryRowContext(ctx, `SELECT CAST(JSON_ARRAYAGG(JSON_OBJECT('before',`+"`before`"+`,'after',`+"`after`"+`)) AS CHAR) FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='credential'`, old.NamespaceID).Scan(&audits); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if strings.Contains(audits, rotation.Token) || strings.Contains(audits, "secret_digest") {
|
||
t.Fatal("audit leaked secret")
|
||
}
|
||
}
|
||
|
||
func TestIntegrationCredentialOverlapDeadline(t *testing.T) {
|
||
ctx, db := overlapDB(t)
|
||
f := newExecutionFixture(t, ctx, db)
|
||
old := f.token
|
||
f.exec(`UPDATE orbit_credentials SET expires_at=DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR) WHERE namespace_id=? AND id=?`, old.NamespaceID, old.CredentialID)
|
||
before, err := GetCredentialStatus(ctx, db, old.NamespaceID, old.CredentialID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
op, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: old.NamespaceID, CredentialID: old.CredentialID, ExpectedRevision: 1, Action: credentialRotateAction, GraceFor: 24 * time.Hour, Reason: "bounded grace"})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
out, err := ApplyCredentialOperation(ctx, db, op)
|
||
if err != nil || !out.PreviousValidUntil.Equal(*before.ExpiresAt) {
|
||
t.Fatal("grace extended original expiry", err)
|
||
}
|
||
current, _ := auth.Parse(op.Token)
|
||
// 新令牌过期不会改变独立的旧令牌截止；旧令牌也不能越过自己的截止。
|
||
f.exec(`UPDATE orbit_credentials SET expires_at=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 SECOND) WHERE namespace_id=? AND id=?`, old.NamespaceID, old.CredentialID)
|
||
if f.a.Authorize(ctx, old) != nil || fault.Code(f.a.Authorize(ctx, current)) != "UNAUTHENTICATED" {
|
||
t.Fatal("expiry of two generations conflated")
|
||
}
|
||
f.exec(`UPDATE orbit_credentials SET expires_at=NULL,previous_valid_until=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 MICROSECOND) WHERE namespace_id=? AND id=?`, old.NamespaceID, old.CredentialID)
|
||
if f.a.Authorize(ctx, current) != nil || fault.Code(f.a.Authorize(ctx, old)) != "UNAUTHENTICATED" {
|
||
t.Fatal("expired old token remained usable")
|
||
}
|
||
// 过期的旧令牌不需定时清理才能失效；下一轮可以重用单一旧代槽。
|
||
next, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: old.NamespaceID, CredentialID: old.CredentialID, ExpectedRevision: 2, Action: credentialRotateAction, GraceFor: time.Hour, Reason: "next rotation"})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := ApplyCredentialOperation(ctx, db, next); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if f.a.Authorize(ctx, current) != nil || fault.Code(f.a.Authorize(ctx, old)) != "UNAUTHENTICATED" {
|
||
t.Fatal("wrong predecessor reused")
|
||
}
|
||
}
|
||
|
||
// 两个独立 HTTP/WS 节点共享真实 MySQL，验证旧长连接命令和跨节点重连。
|
||
func TestIntegrationCredentialOverlapWebSocket(t *testing.T) {
|
||
ctx, db := overlapDB(t)
|
||
f := newExecutionFixture(t, ctx, db)
|
||
serve := func(service *worker.Service) *httptest.Server {
|
||
hub := httpserver.NewWorkerHub(ctx, service)
|
||
server := httpserver.New("", nil, f.tasks)
|
||
httpserver.RegisterWorkers(server, hub)
|
||
ts := httptest.NewServer(server.Handler)
|
||
t.Cleanup(func() {
|
||
shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
if err := hub.Shutdown(shutdown); err != nil {
|
||
t.Error(err)
|
||
}
|
||
ts.Close()
|
||
})
|
||
return ts
|
||
}
|
||
a, b := serve(f.a), serve(f.b)
|
||
hello := f.hello
|
||
hello.InstanceID = testID(t)
|
||
hello.Capabilities = []worker.Capability{{Handler: "rotation.idle", HandlerVersion: "v1", PayloadVersion: "v1"}}
|
||
type frame struct {
|
||
Type string `json:"type"`
|
||
SessionID core.ID `json:"session_id"`
|
||
Error *fault.Error `json:"error"`
|
||
}
|
||
connect := func(url, token string) (*websocket.Conn, core.ID) {
|
||
t.Helper()
|
||
c, _, err := websocket.Dial(ctx, url+"/v1/worker/connect", &websocket.DialOptions{HTTPHeader: http.Header{"Authorization": []string{"Bearer " + token}}, Subprotocols: []string{"pulse.worker.v1"}})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Cleanup(func() { c.CloseNow() })
|
||
if err := wsjson.Write(ctx, c, map[string]any{"protocol_version": "1", "type": "worker.hello", "message_id": testID(t), "body": hello}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var out frame
|
||
if err := wsjson.Read(ctx, c, &out); err != nil || out.Type != "worker.welcome" {
|
||
t.Fatalf("WS welcome: %v, %s", err, out.Type)
|
||
}
|
||
return c, out.SessionID
|
||
}
|
||
capacity := func(c *websocket.Conn, session core.ID, seq int, want string) {
|
||
t.Helper()
|
||
body := worker.CapacityInput{Sequence: uint64(seq), Capacity: 0}
|
||
if err := wsjson.Write(ctx, c, map[string]any{"protocol_version": "1", "type": "worker.capacity", "message_id": testID(t), "session_id": session, "body": body}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var out frame
|
||
if err := wsjson.Read(ctx, c, &out); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
code := ""
|
||
if out.Error != nil {
|
||
code = out.Error.Code
|
||
}
|
||
if code != want {
|
||
t.Fatalf("WS auth: %s, want %s", code, want)
|
||
}
|
||
}
|
||
oldConn, session := connect(a.URL, f.resources.Token)
|
||
op, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: f.token.NamespaceID, CredentialID: f.token.CredentialID, ExpectedRevision: 1, Action: credentialRotateAction, GraceFor: time.Hour, Reason: "rolling switch"})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := ApplyCredentialOperation(ctx, db, op); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
capacity(oldConn, session, 1, "")
|
||
oldConn.CloseNow()
|
||
reconnected, oldSession := connect(b.URL, f.resources.Token)
|
||
capacity(reconnected, oldSession, 1, "")
|
||
hello.InstanceID = testID(t)
|
||
current, newSession := connect(a.URL, op.Token)
|
||
capacity(current, newSession, 1, "")
|
||
final, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: f.token.NamespaceID, CredentialID: f.token.CredentialID, ExpectedRevision: 2, Action: credentialFinalizeAction, Reason: "all services switched"})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := ApplyCredentialOperation(ctx, db, final); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
capacity(reconnected, oldSession, 2, "UNAUTHENTICATED")
|
||
capacity(current, newSession, 2, "")
|
||
c, response, err := websocket.Dial(ctx, b.URL+"/v1/worker/connect", &websocket.DialOptions{HTTPHeader: http.Header{"Authorization": []string{"Bearer " + f.resources.Token}}, Subprotocols: []string{"pulse.worker.v1"}})
|
||
if c != nil {
|
||
c.CloseNow()
|
||
}
|
||
if err == nil || response == nil || response.StatusCode != 401 {
|
||
t.Fatal("retired WS handshake accepted")
|
||
}
|
||
}
|