519 lines
23 KiB
Go
519 lines
23 KiB
Go
package onboarding
|
|
|
|
import (
|
|
"context"
|
|
"crypto/hmac"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/deviceactivation"
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
type RegisterInput struct {
|
|
ChallengeID string `json:"challenge_id"`
|
|
BootID string `json:"boot_id"`
|
|
Model string `json:"hardware_model"`
|
|
HardwareRevision string `json:"hardware_revision"`
|
|
FirmwareVersion string `json:"firmware_version"`
|
|
SDKVersion string `json:"sdk_version"`
|
|
}
|
|
|
|
func (s *Service) Register(ctx context.Context, d Device, requestID string, raw []byte, in RegisterInput) (map[string]any, error) {
|
|
if !uuidV4(in.ChallengeID) || !uuidV4(in.BootID) || in.Model != d.Model || in.HardwareRevision != d.HardwareRevision || len(in.FirmwareVersion) < 1 || len(in.FirmwareVersion) > 32 || len(in.SDKVersion) < 1 || len(in.SDKVersion) > 64 {
|
|
return nil, fail(400, "VALIDATION_FAILED", "注册字段与生产记录不一致")
|
|
}
|
|
return s.mutate(ctx, d, "device:"+d.ID, "register", requestID, raw, 24*time.Hour, func(tx pgx.Tx, current Device) (map[string]any, error) {
|
|
tag, err := tx.Exec(ctx, `UPDATE device_time_challenge SET consumed_by=$1::uuid WHERE id=$2::uuid AND device_id=$3::uuid AND key_version=$4 AND boot_id=$5::uuid AND expires_at>now() AND consumed_by IS NULL`, requestID, in.ChallengeID, d.ID, d.KeyVersion, in.BootID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return nil, fail(409, "CHALLENGE_INVALID", "请重新获取校时挑战")
|
|
}
|
|
if current.BootID != "" && current.BootID != in.BootID {
|
|
if _, err = tx.Exec(ctx, `UPDATE device_setup_session SET closed_at=COALESCE(closed_at,now()),close_reason='DEVICE_REBOOT' WHERE device_id=$1::uuid`, d.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
if _, err = tx.Exec(ctx, `UPDATE device_binding_claim SET status='CANCELLED',completed_at=now() WHERE device_id=$1::uuid AND status='PENDING'`, d.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
// A new boot cannot reuse a still-live pre-reboot execution permit:
|
|
// the previous SDK call may have reached the supplier before power loss.
|
|
pending, e := tx.Exec(ctx, `UPDATE device_activation_attempt a SET status='UNKNOWN' FROM device_activation_job j WHERE a.activation_id=j.id AND j.device_id=$1::uuid AND a.executor='DEVICE' AND a.status='CONTACT_POSSIBLE'`, d.ID)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if pending.RowsAffected() > 0 {
|
|
if _, err = tx.Exec(ctx, `UPDATE device_activation_job SET status='RECONCILE_REQUIRED',last_error_code='DEVICE_REBOOT_DURING_ATTEMPT',version=version+1,updated_at=now() WHERE device_id=$1::uuid`, d.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE device SET onboarding_boot_id=$2::uuid,onboarding_last_seen_at=now(),onboarding_revision=onboarding_revision+1 WHERE id=$1::uuid`, d.ID, in.BootID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE device_twin SET firmware_version=$2,board_revision=$3,rtc_sdk_version=$4,last_reported_at=now(),updated_at=now() WHERE device_id=$1::uuid`, d.ID, in.FirmwareVersion, in.HardwareRevision, in.SDKVersion)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
st, err := s.state(ctx, tx, d)
|
|
return map[string]any{"state": st, "contract_version": ContractVersion}, err
|
|
})
|
|
}
|
|
|
|
type SetupInput struct {
|
|
BootID string `json:"boot_id"`
|
|
LocalSetupID string `json:"local_setup_id"`
|
|
Transport string `json:"transport_profile"`
|
|
}
|
|
type setup struct {
|
|
ID, DeviceID, BootID, LocalID, Mode string
|
|
Generation int64
|
|
Expires time.Time
|
|
Closed *time.Time
|
|
}
|
|
|
|
func readSetup(ctx context.Context, tx pgx.Tx, id string, d Device, open bool) (setup, error) {
|
|
var v setup
|
|
if !uuidV4(id) {
|
|
return v, fail(400, "VALIDATION_FAILED", "设置会话编号不正确")
|
|
}
|
|
err := tx.QueryRow(ctx, `SELECT id::text,device_id::text,boot_id::text,local_setup_id::text,mode,binding_generation,expires_at,closed_at FROM device_setup_session WHERE id=$1::uuid AND device_id=$2::uuid AND tenant_id=$3::uuid FOR UPDATE`, id, d.ID, d.TenantID).Scan(&v.ID, &v.DeviceID, &v.BootID, &v.LocalID, &v.Mode, &v.Generation, &v.Expires, &v.Closed)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return v, errAccess
|
|
}
|
|
if err != nil {
|
|
return v, err
|
|
}
|
|
if v.BootID != d.BootID {
|
|
return v, fail(409, "BOOT_CHANGED", "设备已重启，请恢复新的设置会话")
|
|
}
|
|
if v.Generation != d.Generation {
|
|
return v, errState
|
|
}
|
|
if open && (v.Closed != nil || !v.Expires.After(time.Now())) {
|
|
return v, fail(410, "SETUP_EXPIRED", "设置入口已关闭，请从设备恢复")
|
|
}
|
|
return v, nil
|
|
}
|
|
func (s *Service) Setup(ctx context.Context, d Device, requestID string, raw []byte, in SetupInput) (map[string]any, error) {
|
|
if !uuidV4(in.BootID) || !uuidV4(in.LocalSetupID) || in.Transport != "LOCAL_WEB" {
|
|
return nil, fail(400, "VALIDATION_FAILED", "设置会话参数不正确")
|
|
}
|
|
return s.mutate(ctx, d, "device:"+d.ID, "setup", requestID, raw, 15*time.Minute, func(tx pgx.Tx, current Device) (map[string]any, error) {
|
|
if current.BootID != in.BootID {
|
|
return nil, fail(409, "BOOT_CHANGED", "请先完成本次启动注册")
|
|
}
|
|
st, err := s.state(ctx, tx, current)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
mode := "ENROLL"
|
|
if st.BindingState == "BOUND" {
|
|
mode = "RECOVERY"
|
|
}
|
|
if mode == "ENROLL" && !s.Enabled() {
|
|
return nil, fail(503, "ENROLLMENT_PAUSED", "首次绑定入口暂停")
|
|
}
|
|
var id string
|
|
var expires time.Time
|
|
err = tx.QueryRow(ctx, `SELECT id::text,expires_at FROM device_setup_session WHERE device_id=$1::uuid AND boot_id=$2::uuid AND local_setup_id=$3::uuid AND binding_generation=$4 AND mode=$5 AND closed_at IS NULL AND expires_at>now() ORDER BY created_at DESC LIMIT 1`, d.ID, in.BootID, in.LocalSetupID, current.Generation, mode).Scan(&id, &expires)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
expires = s.now().Add(15 * time.Minute)
|
|
err = tx.QueryRow(ctx, `INSERT INTO device_setup_session(device_id,tenant_id,boot_id,local_setup_id,binding_generation,mode,expires_at) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5,$6,$7) RETURNING id::text`, d.ID, d.TenantID, in.BootID, in.LocalSetupID, current.Generation, mode, expires).Scan(&id)
|
|
}
|
|
return map[string]any{"setup_id": id, "expires_at_ms": expires.UnixMilli(), "poll_after_ms": 2000, "state": st}, err
|
|
})
|
|
}
|
|
|
|
type HandoffInput struct {
|
|
SetupID string `json:"setup_id"`
|
|
LocalID string `json:"local_setup_id"`
|
|
Challenge string `json:"code_challenge"`
|
|
}
|
|
|
|
func (s *Service) Handoff(ctx context.Context, d Device, requestID string, raw []byte, in HandoffInput) (map[string]any, error) {
|
|
if !uuidV4(in.LocalID) || !validB64(in.Challenge, 32) {
|
|
return nil, fail(400, "VALIDATION_FAILED", "入口挑战不正确")
|
|
}
|
|
return s.mutate(ctx, d, "device:"+d.ID, "handoff", requestID, raw, 90*time.Second, func(tx pgx.Tx, current Device) (map[string]any, error) {
|
|
setup, err := readSetup(ctx, tx, in.SetupID, current, true)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if setup.Mode != "ENROLL" || setup.LocalID != in.LocalID {
|
|
return nil, errAccess
|
|
}
|
|
var bound bool
|
|
if err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_binding WHERE device_id=$1::uuid AND unbound_at IS NULL)`, d.ID).Scan(&bound); err != nil {
|
|
return nil, err
|
|
}
|
|
if bound {
|
|
return nil, errAccess
|
|
}
|
|
token := randomToken(32)
|
|
expires := s.now().Add(90 * time.Second)
|
|
if setup.Expires.Before(expires) {
|
|
expires = setup.Expires
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO device_handoff(token_hash,setup_id,code_challenge,expires_at) VALUES($1,$2::uuid,$3,$4)`, digest([]byte(token)), setup.ID, in.Challenge, expires)
|
|
return map[string]any{"handoff_token": token, "expires_at_ms": expires.UnixMilli(), "entry_url": s.options.EntryURL}, err
|
|
})
|
|
}
|
|
|
|
type PollInput struct {
|
|
SetupID string `json:"setup_id"`
|
|
BootID string `json:"boot_id"`
|
|
KnownRevision int64 `json:"known_revision"`
|
|
}
|
|
|
|
func (s *Service) Poll(ctx context.Context, d Device, in PollInput) (map[string]any, error) {
|
|
if !uuidV4(in.BootID) || in.KnownRevision < 0 {
|
|
return nil, fail(400, "VALIDATION_FAILED", "轮询参数不正确")
|
|
}
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
d, err = readDevice(ctx, tx, d.ID, true)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !deviceAllowed(d) {
|
|
return nil, errAccess
|
|
}
|
|
setup, err := readSetup(ctx, tx, in.SetupID, d, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if in.BootID != d.BootID {
|
|
return nil, errState
|
|
}
|
|
if err = s.expireClaims(ctx, tx, d.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
st, err := s.state(ctx, tx, d)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var command any
|
|
if setup.Closed == nil && setup.Expires.After(s.now()) && setup.Mode == "ENROLL" {
|
|
var id, hash, masked string
|
|
var encrypted []byte
|
|
var expires time.Time
|
|
var gen int64
|
|
err = tx.QueryRow(ctx, `SELECT c.id::text,c.challenge_ciphertext,c.binding_digest,c.expected_generation,c.expires_at,COALESCE(ua.display_name,'') FROM device_binding_claim c JOIN user_account ua ON ua.id=c.user_id WHERE c.device_id=$1::uuid AND c.setup_id=$2::uuid AND c.status='PENDING' AND c.expires_at>now()`, d.ID, setup.ID).Scan(&id, &encrypted, &hash, &gen, &expires, &masked)
|
|
if err == nil {
|
|
plain, e := s.crypt.open(encrypted, "claim:"+id)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
runes := []rune(masked)
|
|
if len(runes) > 4 {
|
|
masked = "尾号" + string(runes[len(runes)-4:])
|
|
} else {
|
|
masked = "当前登录账号"
|
|
}
|
|
command = map[string]any{"command_id": id, "type": "CONFIRM_CLAIM", "expires_at_ms": expires.UnixMilli(), "payload": map[string]any{"claim_id": id, "challenge": string(plain), "binding_digest": hash, "expected_binding_generation": gen, "masked_account": masked}}
|
|
} else if !errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, err
|
|
}
|
|
}
|
|
if command == nil && st.BindingState == "BOUND" {
|
|
command, err = s.activationCommand(ctx, tx, d)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
st, err = s.state(ctx, tx, d)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
interval := 2000
|
|
if st.BindingState == "BOUND" {
|
|
interval = 10000
|
|
}
|
|
if st.ActivationState == "READY" {
|
|
interval = 60000
|
|
}
|
|
return map[string]any{"state": st, "command": command, "poll_after_ms": interval}, tx.Commit(ctx)
|
|
}
|
|
func (s *Service) expireClaims(ctx context.Context, tx pgx.Tx, id string) error {
|
|
tag, err := tx.Exec(ctx, `UPDATE device_binding_claim SET status='EXPIRED',completed_at=now() WHERE device_id=$1::uuid AND status='PENDING' AND expires_at<=now()`, id)
|
|
if err == nil && tag.RowsAffected() > 0 {
|
|
err = bump(ctx, tx, id)
|
|
}
|
|
return err
|
|
}
|
|
|
|
type ConfirmationInput struct {
|
|
ClaimID string `json:"claim_id"`
|
|
Challenge string `json:"challenge"`
|
|
Digest string `json:"binding_digest"`
|
|
ExpectedGeneration int64 `json:"expected_binding_generation"`
|
|
PressEventID string `json:"press_event_id"`
|
|
}
|
|
|
|
func (s *Service) Confirm(ctx context.Context, d Device, requestID string, raw []byte, in ConfirmationInput) (map[string]any, error) {
|
|
if !uuidV4(in.ClaimID) || !uuidV4(in.PressEventID) || !validB64(in.Challenge, 32) || !validDigest(in.Digest) || in.ExpectedGeneration < 0 {
|
|
return nil, fail(400, "VALIDATION_FAILED", "设备确认字段不正确")
|
|
}
|
|
return s.mutate(ctx, d, "device:"+d.ID, "confirm", requestID, raw, 24*time.Hour, func(tx pgx.Tx, current Device) (map[string]any, error) {
|
|
var claim claimRecord
|
|
var err error
|
|
claim, err = readClaim(ctx, tx, in.ClaimID, current.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err = claimPermission(ctx, tx, claim); err != nil {
|
|
return nil, err
|
|
}
|
|
if !hmac.Equal([]byte(claim.ChallengeHash), []byte(digest([]byte(in.Challenge)))) || !hmac.Equal([]byte(claim.Digest), []byte(in.Digest)) || claim.Generation != in.ExpectedGeneration {
|
|
return nil, errState
|
|
}
|
|
if claim.Status == "CONFIRMED" {
|
|
if claim.PressID != in.PressEventID || current.Generation != claim.Generation+1 {
|
|
return nil, errState
|
|
}
|
|
var active bool
|
|
err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_binding WHERE id=$1::uuid AND unbound_at IS NULL AND generation=$2)`, claim.BindingID, current.Generation).Scan(&active)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !active {
|
|
return nil, errState
|
|
}
|
|
st, e := s.state(ctx, tx, current)
|
|
return map[string]any{"claim_id": claim.ID, "state": st, "already_applied": true}, e
|
|
}
|
|
if current.Generation != in.ExpectedGeneration {
|
|
return nil, errState
|
|
}
|
|
setup, err := readSetup(ctx, tx, claim.SetupID, current, true)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if setup.Mode != "ENROLL" {
|
|
return nil, errAccess
|
|
}
|
|
if claim.Status != "PENDING" || !claim.Expires.After(s.now()) {
|
|
return nil, fail(410, "CLAIM_EXPIRED", "绑定申请已过期或取消")
|
|
}
|
|
var active bool
|
|
err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_binding WHERE device_id=$1::uuid AND unbound_at IS NULL)`, d.ID).Scan(&active)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if active {
|
|
return nil, errState
|
|
}
|
|
if _, err = tx.Exec(ctx, `SELECT set_config('app.onboarding_claim',$1,true)`, claim.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
var bindingID string
|
|
err = tx.QueryRow(ctx, `INSERT INTO device_binding(tenant_id,device_id,household_id,subject_id,bound_by,reason) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5::uuid,'physical_confirm') RETURNING id::text`, d.TenantID, d.ID, claim.HouseholdID, claim.SubjectID, claim.UserID).Scan(&bindingID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE device_binding_claim SET status='CONFIRMED',press_event_id=$2::uuid,binding_id=$3::uuid,completed_at=now() WHERE id=$1::uuid`, claim.ID, in.PressEventID, bindingID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE device_setup_session SET binding_generation=$2 WHERE id=$1::uuid`, setup.ID, current.Generation+1)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if _, err = deviceactivation.Apply(ctx, tx, deviceactivation.Input{DeviceID: d.ID, TenantID: d.TenantID, Target: deviceactivation.Bound, Reason: "设备按键确认绑定", Source: "onboarding", ActorType: "device", ActorPublicID: &d.PublicID, RequestID: requestID}); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = s.enqueueActivation(ctx, tx, current, current.Generation+1, requestID); err != nil {
|
|
return nil, err
|
|
}
|
|
// Match the platform's basic voice mode. Optional storage, extraction and
|
|
// personalization stay off until their existing consent flows are used.
|
|
_, err = tx.Exec(ctx, `INSERT INTO consent_snapshot(tenant_id,subject_id,snapshot_version,effective_consents,source_consent_ids) SELECT $1::uuid,$2::uuid,1,'{"voice_chat":true,"transcript_storage":false,"raw_audio_storage":false,"story_extraction":false,"personalized_memory":false,"family_sharing":false}'::jsonb,'[]'::jsonb WHERE NOT EXISTS(SELECT 1 FROM consent_snapshot WHERE tenant_id=$1::uuid AND subject_id=$2::uuid)`, current.TenantID, claim.SubjectID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err = s.auditEvent(ctx, tx, d, "device", d.PublicID, "device.binding.confirm", requestID, map[string]any{"claim_id": claim.ID, "binding_generation": current.Generation + 1}); err != nil {
|
|
return nil, err
|
|
}
|
|
st, err := s.state(ctx, tx, current)
|
|
return map[string]any{"claim_id": claim.ID, "state": st, "already_applied": false}, err
|
|
})
|
|
}
|
|
|
|
type claimRecord struct {
|
|
ID, SetupID, DeviceID, TenantID, UserID, HouseholdID, SubjectID, ChallengeHash, Digest, Status, BindingID, PressID string
|
|
Generation int64
|
|
Expires time.Time
|
|
}
|
|
|
|
func readClaim(ctx context.Context, tx pgx.Tx, id, device string) (claimRecord, error) {
|
|
var c claimRecord
|
|
err := tx.QueryRow(ctx, `SELECT id::text,setup_id::text,device_id::text,tenant_id::text,user_id::text,household_id::text,subject_id::text,challenge_hash,binding_digest,status,COALESCE(binding_id::text,''),COALESCE(press_event_id::text,''),expected_generation,expires_at FROM device_binding_claim WHERE id=$1::uuid AND device_id=$2::uuid FOR UPDATE`, id, device).Scan(&c.ID, &c.SetupID, &c.DeviceID, &c.TenantID, &c.UserID, &c.HouseholdID, &c.SubjectID, &c.ChallengeHash, &c.Digest, &c.Status, &c.BindingID, &c.PressID, &c.Generation, &c.Expires)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return c, errAccess
|
|
}
|
|
return c, err
|
|
}
|
|
func claimPermission(ctx context.Context, tx pgx.Tx, c claimRecord) error {
|
|
var allowed bool
|
|
err := tx.QueryRow(ctx, `SELECT true FROM household_member hm JOIN user_account ua ON ua.id=hm.user_id AND ua.status='active' JOIN household h ON h.id=hm.household_id AND h.status='active' JOIN subject_profile sp ON sp.household_id=h.id AND sp.tenant_id=h.tenant_id WHERE ua.id=$1::uuid AND hm.household_id=$2::uuid AND sp.id=$3::uuid AND hm.tenant_id=$4::uuid AND hm.status='active' AND hm.role IN ('owner','admin','editor') AND sp.status='active' FOR SHARE OF ua,hm,h,sp`, c.UserID, c.HouseholdID, c.SubjectID, c.TenantID).Scan(&allowed)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return errAccess
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !allowed {
|
|
return errAccess
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type StateAckInput struct {
|
|
Revision int64 `json:"device_revision"`
|
|
Generation int64 `json:"binding_generation"`
|
|
LocalGeneration int64 `json:"local_generation"`
|
|
Hash string `json:"state_hash"`
|
|
}
|
|
|
|
func (s *Service) StateAck(ctx context.Context, d Device, requestID string, raw []byte, in StateAckInput) (map[string]any, error) {
|
|
if in.Revision < 1 || in.Generation < 0 || in.LocalGeneration < 0 || !validDigest(in.Hash) {
|
|
return nil, fail(400, "VALIDATION_FAILED", "状态回执不正确")
|
|
}
|
|
return s.mutate(ctx, d, "device:"+d.ID, "state-ack", requestID, raw, 24*time.Hour, func(tx pgx.Tx, current Device) (map[string]any, error) {
|
|
st, err := s.state(ctx, tx, current)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if st.Revision != in.Revision || st.Generation != in.Generation || st.Hash != in.Hash {
|
|
return nil, errState
|
|
}
|
|
var local int64
|
|
var boot string
|
|
err = tx.QueryRow(ctx, `SELECT local_generation,boot_id::text FROM device_state_ack WHERE device_id=$1::uuid`, d.ID).Scan(&local, &boot)
|
|
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, err
|
|
}
|
|
if err == nil && boot == current.BootID && local > in.LocalGeneration {
|
|
return nil, errState
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO device_state_ack(device_id,revision,binding_generation,state_hash,boot_id,local_generation) VALUES($1::uuid,$2,$3,$4,$5::uuid,$6) ON CONFLICT(device_id) DO UPDATE SET revision=EXCLUDED.revision,binding_generation=EXCLUDED.binding_generation,state_hash=EXCLUDED.state_hash,boot_id=EXCLUDED.boot_id,local_generation=EXCLUDED.local_generation,accepted_at=CASE WHEN device_state_ack.binding_generation=EXCLUDED.binding_generation AND device_state_ack.boot_id=EXCLUDED.boot_id THEN device_state_ack.accepted_at ELSE now() END`, d.ID, in.Revision, in.Generation, in.Hash, current.BootID, in.LocalGeneration)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
canClose, err := s.stateAckCanClose(ctx, tx, current)
|
|
return map[string]any{"accepted_revision": in.Revision, "setup_can_close": canClose}, err
|
|
})
|
|
}
|
|
|
|
type CloseInput struct {
|
|
SetupID string `json:"setup_id"`
|
|
Reason string `json:"reason"`
|
|
Revision int64 `json:"final_revision"`
|
|
}
|
|
|
|
func (s *Service) Close(ctx context.Context, d Device, requestID string, raw []byte, in CloseInput) (map[string]any, error) {
|
|
if !uuidV4(in.SetupID) || in.Revision < 1 || !oneOf(in.Reason, "SUCCESS", "TIMEOUT", "CANCELLED", "NETWORK_SWITCH") {
|
|
return nil, fail(400, "VALIDATION_FAILED", "收尾参数不正确")
|
|
}
|
|
return s.mutate(ctx, d, "device:"+d.ID, "close", requestID, raw, 24*time.Hour, func(tx pgx.Tx, current Device) (map[string]any, error) {
|
|
setup, err := readSetup(ctx, tx, in.SetupID, current, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if in.Revision > current.Revision {
|
|
return nil, errState
|
|
}
|
|
if in.Reason == "SUCCESS" {
|
|
_, allowed, e := s.closeDelay(ctx, tx, current, setup.ID)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if !allowed {
|
|
return nil, fail(409, "SETUP_CLOSE_WAIT", "请等待设备和页面回执的收尾时间")
|
|
}
|
|
var saved bool
|
|
err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_state_ack a JOIN device_binding b ON b.device_id=a.device_id AND b.unbound_at IS NULL WHERE a.device_id=$1::uuid AND a.binding_generation=$2 AND a.boot_id=$3::uuid)`, d.ID, current.Generation, current.BootID).Scan(&saved)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !saved {
|
|
return nil, fail(409, "STATE_ACK_REQUIRED", "设备保存后才能确认绑定收尾")
|
|
}
|
|
}
|
|
if _, err = tx.Exec(ctx, `UPDATE device_setup_session SET closed_at=COALESCE(closed_at,now()),close_reason=COALESCE(close_reason,$2) WHERE id=$1::uuid`, setup.ID, in.Reason); err != nil {
|
|
return nil, err
|
|
}
|
|
tag, err := tx.Exec(ctx, `UPDATE device_binding_claim SET status='CANCELLED',completed_at=now() WHERE setup_id=$1::uuid AND status='PENDING'`, setup.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if tag.RowsAffected() > 0 {
|
|
if err = bump(ctx, tx, d.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
st, err := s.state(ctx, tx, current)
|
|
return map[string]any{"close_allowed": true, "state": st}, err
|
|
})
|
|
}
|
|
func oneOf(s string, values ...string) bool {
|
|
for _, v := range values {
|
|
if s == v {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func bindingDigest(d Device, user, household, subject, claim string) string {
|
|
return digest([]byte(fmt.Sprintf("CLAIM1\n%s\n%s\n%s\n%s\n%s\n%d", d.FactoryID, user, household, subject, claim, d.Generation)))
|
|
}
|
|
func newClaim(ctx context.Context, s *Service, tx pgx.Tx, d Device, setupID, user, household, subject string) (map[string]any, error) {
|
|
if err := s.expireClaims(ctx, tx, d.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
var pending bool
|
|
err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_binding_claim WHERE device_id=$1::uuid AND status='PENDING')`, d.ID).Scan(&pending)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if pending {
|
|
return nil, fail(409, "CLAIM_BUSY", "设备正在等待另一笔确认，请稍后再试")
|
|
}
|
|
id := uuid.NewString()
|
|
challenge := randomToken(32)
|
|
hash := bindingDigest(d, user, household, subject, id)
|
|
expires := s.now().Add(120 * time.Second)
|
|
setup, err := readSetup(ctx, tx, setupID, d, true)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if setup.Expires.Before(expires) {
|
|
expires = setup.Expires
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO device_binding_claim(id,device_id,tenant_id,setup_id,user_id,household_id,subject_id,expected_generation,challenge_hash,challenge_ciphertext,binding_digest,expires_at) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5::uuid,$6::uuid,$7::uuid,$8,$9,$10,$11,$12)`, id, d.ID, d.TenantID, setupID, user, household, subject, d.Generation, digest([]byte(challenge)), s.crypt.seal([]byte(challenge), "claim:"+id), hash, expires)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err = bump(ctx, tx, d.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
st, err := s.state(ctx, tx, d)
|
|
return map[string]any{"claim_id": id, "expires_at_ms": expires.UnixMilli(), "state": st}, err
|
|
}
|
|
|
|
// Used by protocol tests without leaking transient challenges into logs.
|
|
func requestJSON(v any) []byte { b, _ := json.Marshal(v); return b }
|
|
func nonempty(v string, max int) bool { return len(strings.TrimSpace(v)) > 0 && len(v) <= max }
|