Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
406 lines
16 KiB
Go
406 lines
16 KiB
Go
package onboarding
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/hardwareprovision"
|
|
"data-terminal-cloud/internal/identity"
|
|
"data-terminal-cloud/internal/outbox"
|
|
"data-terminal-cloud/internal/saas"
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
type testRegistrar struct {
|
|
calls atomic.Int32
|
|
err error
|
|
started chan struct{}
|
|
release chan struct{}
|
|
}
|
|
|
|
func (r *testRegistrar) Register(ctx context.Context, in hardwareprovision.RegisterInput) (hardwareprovision.RegisterResult, error) {
|
|
r.calls.Add(1)
|
|
if r.started != nil {
|
|
close(r.started)
|
|
select {
|
|
case <-r.release:
|
|
case <-ctx.Done():
|
|
return hardwareprovision.RegisterResult{}, ctx.Err()
|
|
}
|
|
}
|
|
return hardwareprovision.RegisterResult{DeviceSecret: "fixture-device-secret-at-least-32-bytes", RTCAppID: "test-app"}, r.err
|
|
}
|
|
|
|
type fixture struct {
|
|
t *testing.T
|
|
s *Service
|
|
ctx context.Context
|
|
tenant, tenantPublic, operatorID string
|
|
p, user identity.Principal
|
|
d Device
|
|
web WebSession
|
|
setup, elder, claim string
|
|
confirm ConfirmationInput
|
|
registrar *testRegistrar
|
|
}
|
|
|
|
func newFixture(t *testing.T) *fixture {
|
|
t.Helper()
|
|
dsn := os.Getenv("TEST_DATABASE_URL")
|
|
if dsn == "" {
|
|
t.Skip("TEST_DATABASE_URL required")
|
|
}
|
|
ctx := context.Background()
|
|
pool, err := pgxpool.New(ctx, dsn)
|
|
must(t, err)
|
|
t.Cleanup(pool.Close)
|
|
addr := os.Getenv("TEST_REDIS_ADDR")
|
|
if addr == "" {
|
|
addr = "127.0.0.1:56379"
|
|
}
|
|
cache := redis.NewClient(&redis.Options{Addr: addr})
|
|
must(t, cache.Ping(ctx).Err())
|
|
t.Cleanup(func() { _ = cache.Close() })
|
|
f := &fixture{t: t, ctx: ctx, tenant: uuid.NewString(), tenantPublic: uuid.NewString(), operatorID: uuid.NewString(), p: identity.Principal{Issuer: "urn:onboarding:test", Subject: uuid.NewString()}, user: identity.Principal{Issuer: "urn:onboarding:test", Subject: uuid.NewString(), DisplayName: "测试用户1234"}, registrar: &testRegistrar{}}
|
|
f.s, err = NewService(pool, cache, Options{Enabled: true, MasterKey: strings.Repeat("15", 32), ProviderMasterKey: strings.Repeat("25", 32), EntryURL: "https://platform.example/onboarding/entry", SelfSignupTenantIDs: []string{f.tenantPublic}, ActivationEnabled: true, Registrar: f.registrar, InstanceID: "test-instance", ProductKey: "test-product", ProductSecret: "fixture-product-secret-at-least-32-bytes", BotID: "test-bot", Transport: "high_quality", MinimumSDKVersion: "1.0.0"})
|
|
must(t, err)
|
|
_, err = pool.Exec(ctx, `INSERT INTO tenant(id,public_id,name) VALUES($1::uuid,$2::uuid,'onboarding integration')`, f.tenant, f.tenantPublic)
|
|
f.s.options.LegacyDeviceRegistration = true // Existing tests exercise old-device migration contracts.
|
|
must(t, err)
|
|
_, err = pool.Exec(ctx, `INSERT INTO user_account(id,tenant_id,oidc_issuer,oidc_subject,display_name,platform_role,account_kind) VALUES($1::uuid,$2::uuid,$3,$4,'生产管理员','ops_admin','operator')`, f.operatorID, f.tenant, f.p.Issuer, f.p.Subject)
|
|
must(t, err)
|
|
mac := strings.ToUpper(strings.ReplaceAll(uuid.NewString(), "-", ""))[:12]
|
|
in := InventoryInput{FactoryMAC: mac, Model: "KP01", HardwareRevision: "A1", BatchNo: "test"}
|
|
raw := requestJSON(in)
|
|
key := uuid.NewString()
|
|
inv, err := f.s.CreateInventory(ctx, f.p, f.tenantPublic, uuid.NewString(), key, raw, in)
|
|
must(t, err)
|
|
again, err := f.s.CreateInventory(ctx, f.p, f.tenantPublic, uuid.NewString(), key, raw, in)
|
|
must(t, err)
|
|
if inv["platform_device_id"] != again["platform_device_id"] {
|
|
t.Fatal("inventory identity changed")
|
|
}
|
|
public := inv["platform_device_id"].(string)
|
|
job := inv["enrollment_job_id"].(string)
|
|
if inv["released"] != true {
|
|
t.Fatal("new inventory did not become immediately claimable")
|
|
}
|
|
var lifecycle, keyStatus, jobStatus string
|
|
must(t, f.s.pool.QueryRow(ctx, `SELECT d.activation_state,k.status,ej.status FROM device d JOIN device_onboarding_key k ON k.device_id=d.id AND k.key_version=1 JOIN device_enrollment_job ej ON ej.device_id=d.id AND ej.id=$2::uuid WHERE d.public_id=$1::uuid`, public, job).Scan(&lifecycle, &keyStatus, &jobStatus))
|
|
if lifecycle != "CLAIMABLE" || keyStatus != "ACTIVE" || jobStatus != "RELEASED" {
|
|
t.Fatalf("direct inventory state lifecycle=%s key=%s job=%s", lifecycle, keyStatus, jobStatus)
|
|
}
|
|
f.d, err = f.s.findFactoryDevice(ctx, "KP01-"+mac, 1)
|
|
must(t, err)
|
|
boot := uuid.NewString()
|
|
challenge, err := f.s.TimeChallenge(ctx, TimeChallengeInput{DeviceID: f.d.FactoryID, KeyVersion: 1, BootID: boot, ClientNonce: randomToken(16)})
|
|
must(t, err)
|
|
decoded, err := base64.RawURLEncoding.DecodeString(challenge["payload_b64"].(string))
|
|
must(t, err)
|
|
var packet map[string]any
|
|
must(t, json.Unmarshal(decoded, &packet))
|
|
reg := RegisterInput{ChallengeID: packet["challenge_id"].(string), BootID: boot, Model: f.d.Model, HardwareRevision: f.d.HardwareRevision, FirmwareVersion: "1.2.0", SDKVersion: "1.0.0"}
|
|
requestID := uuid.NewString()
|
|
_, err = f.s.Register(ctx, f.d, requestID, requestJSON(reg), reg)
|
|
must(t, err)
|
|
f.refresh()
|
|
_, err = f.s.Register(ctx, f.d, requestID, requestJSON(reg), reg)
|
|
must(t, err)
|
|
setupIn := SetupInput{BootID: boot, LocalSetupID: uuid.NewString(), Transport: "LOCAL_WEB"}
|
|
setup, err := f.s.Setup(ctx, f.d, uuid.NewString(), requestJSON(setupIn), setupIn)
|
|
must(t, err)
|
|
f.setup = setup["setup_id"].(string)
|
|
verifier := randomToken(32)
|
|
sum := sha256.Sum256([]byte(verifier))
|
|
handoffIn := HandoffInput{SetupID: f.setup, LocalID: setupIn.LocalSetupID, Challenge: encode64(sum[:])}
|
|
handoff, err := f.s.Handoff(ctx, f.d, uuid.NewString(), requestJSON(handoffIn), handoffIn)
|
|
must(t, err)
|
|
credentials, err := f.s.Exchange(ctx, ExchangeInput{Token: handoff["handoff_token"].(string), Verifier: verifier})
|
|
must(t, err)
|
|
_, err = f.s.Exchange(ctx, ExchangeInput{Token: handoff["handoff_token"].(string), Verifier: verifier})
|
|
if err == nil {
|
|
t.Fatal("handoff redeemed twice")
|
|
}
|
|
ws, err := f.s.webSession(ctx, pool, digest([]byte(credentials.Token)))
|
|
must(t, err)
|
|
credentials, err = f.s.AuthLink(ctx, ws, f.user)
|
|
must(t, err)
|
|
f.web, err = f.s.webSession(ctx, pool, digest([]byte(credentials.Token)))
|
|
must(t, err)
|
|
profile := ProfileInput{HouseholdName: "测试家庭", Name: "奶奶", InitialConsents: saas.InitialConsentSelection{Types: []string{saas.ConsentTranscriptStorage, saas.ConsentPersonalMemory}, Confirmed: true}}
|
|
result, err := f.s.CreateProfile(ctx, f.web, uuid.NewString(), requestJSON(profile), profile)
|
|
must(t, err)
|
|
f.elder = result["elder_id"].(string)
|
|
return f
|
|
}
|
|
func must(t *testing.T, err error) {
|
|
t.Helper()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
func (f *fixture) refresh() {
|
|
f.t.Helper()
|
|
var err error
|
|
f.d, err = f.s.findFactoryDevice(f.ctx, f.d.FactoryID, 1)
|
|
must(f.t, err)
|
|
if f.web.Hash != "" {
|
|
f.web, err = f.s.webSession(f.ctx, f.s.pool, f.web.Hash)
|
|
must(f.t, err)
|
|
}
|
|
}
|
|
func (f *fixture) pending() {
|
|
f.t.Helper()
|
|
in := ClaimInput{ElderID: f.elder, Generation: f.d.Generation}
|
|
result, err := f.s.Claim(f.ctx, f.web, uuid.NewString(), requestJSON(in), in)
|
|
must(f.t, err)
|
|
f.claim = result["claim_id"].(string)
|
|
poll, err := f.s.Poll(f.ctx, f.d, PollInput{SetupID: f.setup, BootID: f.d.BootID})
|
|
must(f.t, err)
|
|
payload := poll["command"].(map[string]any)["payload"].(map[string]any)
|
|
f.confirm = ConfirmationInput{ClaimID: f.claim, Challenge: payload["challenge"].(string), Digest: payload["binding_digest"].(string), ExpectedGeneration: f.d.Generation, PressEventID: uuid.NewString()}
|
|
}
|
|
func (f *fixture) bound() {
|
|
f.t.Helper()
|
|
f.pending()
|
|
_, err := f.s.Confirm(f.ctx, f.d, uuid.NewString(), requestJSON(f.confirm), f.confirm)
|
|
must(f.t, err)
|
|
f.refresh()
|
|
}
|
|
func (f *fixture) event() outbox.Event {
|
|
f.t.Helper()
|
|
var e outbox.Event
|
|
err := f.s.pool.QueryRow(f.ctx, `SELECT id::text,payload FROM outbox_event WHERE aggregate_id=$1::uuid AND event_type=$2 ORDER BY created_at DESC LIMIT 1`, f.d.ID, ActivationEvent).Scan(&e.ID, &e.Payload)
|
|
must(f.t, err)
|
|
return e
|
|
}
|
|
|
|
func TestIntegrationPhysicalConfirmationAndReplay(t *testing.T) {
|
|
f := newFixture(t)
|
|
f.pending()
|
|
var count int
|
|
must(t, f.s.pool.QueryRow(f.ctx, `SELECT count(*) FROM device_binding WHERE device_id=$1::uuid`, f.d.ID).Scan(&count))
|
|
if count != 0 {
|
|
t.Fatal("claim established ownership before physical confirmation")
|
|
}
|
|
id := uuid.NewString()
|
|
_, err := f.s.Confirm(f.ctx, f.d, id, requestJSON(f.confirm), f.confirm)
|
|
must(t, err)
|
|
f.refresh()
|
|
if f.d.Generation != 1 {
|
|
t.Fatal("wrong binding generation")
|
|
}
|
|
_, err = f.s.Confirm(f.ctx, f.d, id, requestJSON(f.confirm), f.confirm)
|
|
must(t, err)
|
|
status, err := f.s.WebStatus(f.ctx, f.web)
|
|
must(t, err)
|
|
if status["binding_id"] == nil {
|
|
t.Fatal("binding result not recoverable")
|
|
}
|
|
_, err = f.s.pool.Exec(f.ctx, `UPDATE household_member SET status='removed' WHERE user_id=$1::uuid`, f.web.UserID)
|
|
must(t, err)
|
|
_, err = f.s.Confirm(f.ctx, f.d, id, requestJSON(f.confirm), f.confirm)
|
|
if !errors.Is(err, errAccess) {
|
|
t.Fatalf("replay bypassed revoked permission: %v", err)
|
|
}
|
|
_, err = f.s.pool.Exec(f.ctx, `UPDATE device_binding SET unbound_at=now() WHERE device_id=$1::uuid AND unbound_at IS NULL`, f.d.ID)
|
|
must(t, err)
|
|
current, err := readDevice(f.ctx, f.s.pool, f.d.ID, false)
|
|
must(t, err)
|
|
if current.Generation != 2 {
|
|
t.Fatal("unbind did not increment generation")
|
|
}
|
|
_, err = f.s.Confirm(f.ctx, current, id, requestJSON(f.confirm), f.confirm)
|
|
if err == nil {
|
|
t.Fatal("old confirmation revived unbound owner")
|
|
}
|
|
}
|
|
func TestIntegrationConcurrentClaimsAndCancel(t *testing.T) {
|
|
f := newFixture(t)
|
|
in := ClaimInput{ElderID: f.elder, Generation: 0}
|
|
var winners atomic.Int32
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < 2; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
_, err := f.s.Claim(f.ctx, f.web, uuid.NewString(), requestJSON(in), in)
|
|
if err == nil {
|
|
winners.Add(1)
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
if winners.Load() != 1 {
|
|
t.Fatalf("claim winners=%d", winners.Load())
|
|
}
|
|
var id string
|
|
must(t, f.s.pool.QueryRow(f.ctx, `SELECT id::text FROM device_binding_claim WHERE device_id=$1::uuid AND status='PENDING'`, f.d.ID).Scan(&id))
|
|
cancel := CancelInput{ClaimID: id, Generation: 0}
|
|
_, err := f.s.Cancel(f.ctx, f.web, uuid.NewString(), requestJSON(cancel), cancel)
|
|
must(t, err)
|
|
var challenge, hash string
|
|
var encrypted []byte
|
|
must(t, f.s.pool.QueryRow(f.ctx, `SELECT challenge_ciphertext,binding_digest FROM device_binding_claim WHERE id=$1::uuid`, id).Scan(&encrypted, &hash))
|
|
plain, err := f.s.crypt.open(encrypted, "claim:"+id)
|
|
must(t, err)
|
|
challenge = string(plain)
|
|
confirm := ConfirmationInput{ClaimID: id, Challenge: challenge, Digest: hash, ExpectedGeneration: 0, PressEventID: uuid.NewString()}
|
|
_, err = f.s.Confirm(f.ctx, f.d, uuid.NewString(), requestJSON(confirm), confirm)
|
|
if err == nil {
|
|
t.Fatal("cancelled claim confirmed")
|
|
}
|
|
}
|
|
func (f *fixture) job() activationJob {
|
|
f.t.Helper()
|
|
tx, err := f.s.pool.Begin(f.ctx)
|
|
must(f.t, err)
|
|
defer tx.Rollback(f.ctx)
|
|
j, err := readJob(f.ctx, tx, f.d.ID)
|
|
must(f.t, err)
|
|
return j
|
|
}
|
|
func (f *fixture) registerSDK() activationJob {
|
|
f.t.Helper()
|
|
must(f.t, f.s.ProcessActivation(f.ctx, f.event()))
|
|
f.refresh()
|
|
j := f.job()
|
|
attempt := AttemptInput{ActivationID: j.ID, ConfigVersion: j.ConfigVersion, AttemptID: uuid.NewString()}
|
|
_, err := f.s.Attempt(f.ctx, f.d, uuid.NewString(), requestJSON(attempt), attempt)
|
|
must(f.t, err)
|
|
result := ResultInput{ActivationID: j.ID, AttemptID: attempt.AttemptID, Stage: "SDK_REGISTER", Outcome: "SUCCESS_OBSERVED"}
|
|
_, err = f.s.ActivationResult(f.ctx, f.d, uuid.NewString(), requestJSON(result), result)
|
|
must(f.t, err)
|
|
return f.job()
|
|
}
|
|
func TestIntegrationDeviceUnknownDoesNotRegisterAgain(t *testing.T) {
|
|
f := newFixture(t)
|
|
f.bound()
|
|
must(t, f.s.ProcessActivation(f.ctx, f.event()))
|
|
f.refresh()
|
|
j := f.job()
|
|
attempt := AttemptInput{ActivationID: j.ID, ConfigVersion: j.ConfigVersion, AttemptID: uuid.NewString()}
|
|
permit, err := f.s.Attempt(f.ctx, f.d, uuid.NewString(), requestJSON(attempt), attempt)
|
|
must(t, err)
|
|
if !permit["allowed"].(bool) || time.UnixMilli(permit["permit_expires_at_ms"].(int64)).Sub(time.Now()) > 30*time.Second {
|
|
t.Fatal("permit contract")
|
|
}
|
|
result := ResultInput{ActivationID: j.ID, AttemptID: attempt.AttemptID, Stage: "SDK_REGISTER", SDKResult: -1, Outcome: "UNKNOWN"}
|
|
response, err := f.s.ActivationResult(f.ctx, f.d, uuid.NewString(), requestJSON(result), result)
|
|
must(t, err)
|
|
if response["reconcile_required"] != true {
|
|
t.Fatal("unknown must require reconciliation")
|
|
}
|
|
attempt.AttemptID = uuid.NewString()
|
|
_, err = f.s.Attempt(f.ctx, f.d, uuid.NewString(), requestJSON(attempt), attempt)
|
|
if err == nil {
|
|
t.Fatal("unknown issued another execution")
|
|
}
|
|
must(t, f.s.ProcessActivation(f.ctx, f.event()))
|
|
must(t, f.s.ProcessActivation(f.ctx, f.event()))
|
|
if f.registrar.calls.Load() != 0 {
|
|
t.Fatal("SaaS invoked supplier registration")
|
|
}
|
|
var bound bool
|
|
must(t, f.s.pool.QueryRow(f.ctx, `SELECT EXISTS(SELECT 1 FROM device_binding WHERE device_id=$1::uuid AND unbound_at IS NULL)`, f.d.ID).Scan(&bound))
|
|
if !bound {
|
|
t.Fatal("unknown removed binding")
|
|
}
|
|
}
|
|
func TestIntegrationDeviceConfigAndRegistration(t *testing.T) {
|
|
f := newFixture(t)
|
|
f.bound()
|
|
must(t, f.s.ProcessActivation(f.ctx, f.event()))
|
|
f.refresh()
|
|
response, err := f.s.Poll(f.ctx, f.d, PollInput{SetupID: f.setup, BootID: f.d.BootID})
|
|
must(t, err)
|
|
payload := response["command"].(map[string]any)["payload"].(map[string]any)
|
|
if len(payload) != 2 {
|
|
t.Fatal("extra activation command fields")
|
|
}
|
|
raw, err := base64.RawURLEncoding.DecodeString(payload["config_b64"].(string))
|
|
must(t, err)
|
|
var config map[string]any
|
|
must(t, json.Unmarshal(raw, &config))
|
|
if len(config) != 8 || config["contract_version"] != "ONBOARDING-1.1" {
|
|
t.Fatal("wrong activation config")
|
|
}
|
|
iot := config["iot"].(map[string]any)
|
|
if len(iot) != 4 || iot["product_secret"] != f.s.options.ProductSecret || iot["device_name"] != f.d.FactoryID {
|
|
t.Fatal("wrong SDK registration config")
|
|
}
|
|
if !sameSecret(domainMAC(f.d.Key, "CFG1\n"+payload["config_b64"].(string)), payload["config_mac"].(string)) {
|
|
t.Fatal("config MAC")
|
|
}
|
|
j := f.job()
|
|
ack := ActivationAckInput{ActivationID: j.ID, ConfigVersion: j.ConfigVersion, LocalGeneration: 1}
|
|
_, err = f.s.ActivationAck(f.ctx, f.d, uuid.NewString(), requestJSON(ack), ack)
|
|
if err == nil {
|
|
t.Fatal("ACK before registration was accepted")
|
|
}
|
|
f.registerSDK()
|
|
_, err = f.s.ActivationAck(f.ctx, f.d, uuid.NewString(), requestJSON(ack), ack)
|
|
must(t, err)
|
|
if f.job().Status != "REGISTERED_OBSERVED" {
|
|
t.Fatal("registration falsely implies first call")
|
|
}
|
|
if f.registrar.calls.Load() != 0 {
|
|
t.Fatal("server performed SDK registration")
|
|
}
|
|
}
|
|
func TestIntegrationDEVREQReplayAndTLS(t *testing.T) {
|
|
f := newFixture(t)
|
|
in := PollInput{SetupID: f.setup, BootID: f.d.BootID}
|
|
raw := requestJSON(in)
|
|
path := "/api/v1/device/poll"
|
|
id := uuid.NewString()
|
|
nonce := randomToken(16)
|
|
timestamp := time.Now().UnixMilli()
|
|
request := func() *http.Request {
|
|
r := httptest.NewRequest("POST", "https://platform.example"+path, bytes.NewReader(raw))
|
|
r.Header.Set("Content-Type", "application/json")
|
|
r.Header.Set("X-Device-ID", f.d.FactoryID)
|
|
r.Header.Set("X-Key-Version", "1")
|
|
r.Header.Set("X-Timestamp-Ms", strconv.FormatInt(timestamp, 10))
|
|
r.Header.Set("X-Nonce", nonce)
|
|
r.Header.Set("X-Request-ID", id)
|
|
r.Header.Set("Authorization", "Device-HMAC-SHA256 "+RequestMAC(f.d.Key, "POST", path, f.d.FactoryID, 1, timestamp, nonce, id, raw))
|
|
return r
|
|
}
|
|
mux := http.NewServeMux()
|
|
f.s.RegisterRoutes(mux)
|
|
w := httptest.NewRecorder()
|
|
mux.ServeHTTP(w, request())
|
|
if w.Code != 200 {
|
|
t.Fatal(w.Code, w.Body.String())
|
|
}
|
|
w = httptest.NewRecorder()
|
|
mux.ServeHTTP(w, request())
|
|
if w.Code != 401 {
|
|
t.Fatal("nonce replay was accepted", w.Code)
|
|
}
|
|
r := request()
|
|
r.TLS = nil
|
|
r.Header.Set("X-Forwarded-Proto", "https")
|
|
w = httptest.NewRecorder()
|
|
mux.ServeHTTP(w, r)
|
|
if w.Code != 400 {
|
|
t.Fatal("untrusted forwarded HTTPS accepted")
|
|
}
|
|
}
|