306 lines
11 KiB
Go
306 lines
11 KiB
Go
|
package onboarding
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"encoding/json"
|
||
|
|
"errors"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"data-terminal-cloud/internal/hardwareprovision"
|
||
|
|
"data-terminal-cloud/internal/outbox"
|
||
|
|
"github.com/google/uuid"
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
)
|
||
|
|
|
||
|
|
// Commit the execution fence before contacting the provider. A lost response or
|
||
|
|
// process crash is uncertain, never permission to register the same device again.
|
||
|
|
func (s *Service) processServerActivation(ctx context.Context, event outbox.Event) error {
|
||
|
|
if !s.options.ActivationEnabled {
|
||
|
|
return outbox.Retryable(errors.New("onboarding activation paused"))
|
||
|
|
}
|
||
|
|
var in struct {
|
||
|
|
DeviceID string `json:"device_id"`
|
||
|
|
ActivationID string `json:"activation_id"`
|
||
|
|
Generation int64 `json:"binding_generation"`
|
||
|
|
}
|
||
|
|
if json.Unmarshal(event.Payload, &in) != nil || !uuidValid(in.DeviceID) {
|
||
|
|
return errors.New("invalid activation event")
|
||
|
|
}
|
||
|
|
tx, err := s.pool.Begin(ctx)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
defer tx.Rollback(ctx)
|
||
|
|
d, err := readDevice(ctx, tx, in.DeviceID, true)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
j, err := readJob(ctx, tx, d.ID)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
bound, err := voiceBound(ctx, tx, d)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if !bound || !deviceAllowed(d) || j.ID != in.ActivationID || j.Generation != d.Generation || in.Generation != d.Generation {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
if j.Product != s.options.ProductKey || j.Instance != s.options.InstanceID {
|
||
|
|
return fail(409, "ACTIVATION_PRODUCT_CHANGED", "设备所属火山产品已变化，请核对")
|
||
|
|
}
|
||
|
|
if j.Registered != nil {
|
||
|
|
// Old device-side registrations may not have a server copy of the secret.
|
||
|
|
// Wait for migration; never re-register them to recover a missing secret.
|
||
|
|
if len(j.Secret) > 0 && j.RTCAppID != "" && oneOf(j.Status, "REGISTERED_OBSERVED", "READY") {
|
||
|
|
if err = s.ensureConfig(ctx, tx, d, &j); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
if err = s.expireAttempt(ctx, tx, d, &j); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
var busy bool
|
||
|
|
if err = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_activation_attempt WHERE activation_id=$1::uuid AND status IN ('CONTACT_POSSIBLE','UNKNOWN','SUCCESS_OBSERVED') AND stage IN ('SERVER_REGISTER','SDK_REGISTER'))`, j.ID).Scan(&busy); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if busy || !oneOf(j.Status, "NOT_STARTED", "RESERVED") {
|
||
|
|
return tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
if s.options.Registrar == nil {
|
||
|
|
return errors.New("server registrar unavailable")
|
||
|
|
}
|
||
|
|
attempt := uuid.NewString()
|
||
|
|
_, err = tx.Exec(ctx, `INSERT INTO device_activation_attempt(id,activation_id,executor,stage,binding_generation,config_version,status,permit_expires_at) VALUES($1::uuid,$2::uuid,'SERVER','SERVER_REGISTER',$3,$4,'CONTACT_POSSIBLE',$5)`, attempt, j.ID, d.Generation, j.ConfigVersion, s.now().Add(30*time.Second))
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_job SET status='CONTACT_POSSIBLE',current_attempt_id=$2::uuid,last_error_code=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, j.ID, attempt)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if err = tx.Commit(ctx); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
callCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
|
||
|
|
result, callErr := s.options.Registrar.Register(callCtx, hardwareprovision.RegisterInput{InstanceID: j.Instance, ProductKey: j.Product, DeviceName: j.Name})
|
||
|
|
cancel()
|
||
|
|
// Do not discard successful credentials just because the worker was cancelled.
|
||
|
|
saveCtx, saveCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
||
|
|
defer saveCancel()
|
||
|
|
return s.saveServerRegistration(saveCtx, d.ID, j.ID, attempt, result, callErr)
|
||
|
|
}
|
||
|
|
|
||
|
|
func voiceBound(ctx context.Context, tx pgx.Tx, d Device) (bool, error) {
|
||
|
|
var bound bool
|
||
|
|
err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM device_binding WHERE device_id=$1::uuid AND generation=$2 AND unbound_at IS NULL) OR EXISTS(SELECT 1 FROM device_account_binding WHERE device_id=$1::uuid AND generation=$2 AND released_at IS NULL)`, d.ID, d.Generation).Scan(&bound)
|
||
|
|
return bound, err
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) saveServerRegistration(ctx context.Context, device, job, attempt string, result hardwareprovision.RegisterResult, callErr error) error {
|
||
|
|
tx, err := s.pool.Begin(ctx)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
defer tx.Rollback(ctx)
|
||
|
|
d, err := readDevice(ctx, tx, device, true)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
j, err := readJob(ctx, tx, device)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if j.ID != job {
|
||
|
|
return errState
|
||
|
|
}
|
||
|
|
var status string
|
||
|
|
if err = tx.QueryRow(ctx, `SELECT status FROM device_activation_attempt WHERE id=$1::uuid AND activation_id=$2::uuid AND executor='SERVER' FOR UPDATE`, attempt, job).Scan(&status); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if !oneOf(status, "CONTACT_POSSIBLE", "UNKNOWN") {
|
||
|
|
return tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
good := callErr == nil && validVoiceSecret(result.DeviceSecret) && validRTCAppID(result.RTCAppID)
|
||
|
|
if !good {
|
||
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_attempt SET status='UNKNOWN',completed_at=now() WHERE id=$1::uuid`, attempt)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_job SET status='RECONCILE_REQUIRED',last_error_code='SERVER_RESULT_UNCERTAIN',version=version+1,updated_at=now() WHERE id=$1::uuid`, job)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
return tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
// A rebind/unbind while the HTTP call ran does not erase the physical device's
|
||
|
|
// newly acquired identity. Save it, but only deliver it to a current binding.
|
||
|
|
j.Secret = s.crypt.seal([]byte(result.DeviceSecret), "provider:"+job)
|
||
|
|
j.RTCAppID = result.RTCAppID
|
||
|
|
now := s.now()
|
||
|
|
j.Registered = &now
|
||
|
|
bound, err := voiceBound(ctx, tx, d)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
j.Status = "BLOCKED"
|
||
|
|
if bound && deviceAllowed(d) && j.Generation == d.Generation {
|
||
|
|
j.Status = "REGISTERED_OBSERVED"
|
||
|
|
}
|
||
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_attempt SET status='SUCCESS_OBSERVED',completed_at=now() WHERE id=$1::uuid`, attempt)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_job SET provider_secret_ciphertext=$2,provider_rtc_app_id=$3,sdk_registered_at=now(),engine_ready_at=NULL,config_ciphertext=NULL,config_digest=NULL,config_expires_at=NULL,config_ack_at=NULL,status=$4,last_error_code=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, job, j.Secret, j.RTCAppID, j.Status)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
j.Config = nil
|
||
|
|
j.Expires = nil
|
||
|
|
j.Ack = nil
|
||
|
|
j.Engine = nil
|
||
|
|
if j.Status == "REGISTERED_OBSERVED" {
|
||
|
|
if err = s.ensureConfig(ctx, tx, d, &j); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
|
||
|
|
type DeviceVoiceCredential struct {
|
||
|
|
InstanceID string `json:"instance_id"`
|
||
|
|
ProductKey string `json:"product_key"`
|
||
|
|
DeviceName string `json:"device_name"`
|
||
|
|
DeviceSecret string `json:"device_secret"`
|
||
|
|
RTCAppID string `json:"RTCAppID"`
|
||
|
|
}
|
||
|
|
|
||
|
|
func validVoiceSecret(v string) bool { return len(v) >= 16 && len(v) <= 256 }
|
||
|
|
func validRTCAppID(v string) bool { return len(v) > 0 && len(v) <= 128 }
|
||
|
|
|
||
|
|
// Does not expose credentials through mini-program/account endpoints.
|
||
|
|
func (s *Service) voiceFields(ctx context.Context, tx pgx.Tx, d Device) (map[string]any, error) {
|
||
|
|
out := map[string]any{"voice_status": "preparing", "voice": nil}
|
||
|
|
if !s.options.ActivationEnabled {
|
||
|
|
out["voice_status"] = "paused"
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
j, err := readJob(ctx, tx, d.ID)
|
||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
if j.Generation != d.Generation {
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
if j.Registered != nil && (len(j.Secret) == 0 || j.RTCAppID == "") {
|
||
|
|
out["voice_status"] = "migration_required"
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
if oneOf(j.Status, "BLOCKED", "RECONCILE_REQUIRED") {
|
||
|
|
out["voice_status"] = "reconcile_required"
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
if j.Registered == nil || len(j.Secret) == 0 || j.RTCAppID == "" {
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
secret, err := s.crypt.open(j.Secret, "provider:"+j.ID)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
if !validVoiceSecret(string(secret)) || !validRTCAppID(j.RTCAppID) {
|
||
|
|
return nil, errors.New("invalid stored provider credential")
|
||
|
|
}
|
||
|
|
out["voice_status"] = "ready"
|
||
|
|
out["voice"] = DeviceVoiceCredential{j.Instance, j.Product, j.Name, string(secret), j.RTCAppID}
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
// Callable immediately after binding, including before boot registration.
|
||
|
|
func (s *Service) VoiceConfig(ctx context.Context, d Device) (map[string]any, error) {
|
||
|
|
tx, err := s.pool.Begin(ctx)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
defer tx.Rollback(ctx)
|
||
|
|
current, err := readDevice(ctx, tx, d.ID, true)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
bound, err := voiceBound(ctx, tx, current)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
if !deviceAllowed(current) || !bound {
|
||
|
|
return nil, fail(409, "DEVICE_UNBOUND", "绑定错误：设备没有绑定人员")
|
||
|
|
}
|
||
|
|
out, err := s.voiceFields(ctx, tx, current)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
return out, tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
|
||
|
|
// One-time import of credentials held only by old, already-registered firmware.
|
||
|
|
// Refuses unknown/unregistered devices and never overwrites a different secret.
|
||
|
|
func (s *Service) ImportVoiceCredential(ctx context.Context, d Device, in DeviceVoiceCredential) (map[string]any, error) {
|
||
|
|
if !validVoiceSecret(in.DeviceSecret) || !validRTCAppID(in.RTCAppID) {
|
||
|
|
return nil, fail(400, "INVALID_VOICE_CREDENTIAL", "语音凭据格式错误")
|
||
|
|
}
|
||
|
|
tx, err := s.pool.Begin(ctx)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
defer tx.Rollback(ctx)
|
||
|
|
current, err := readDevice(ctx, tx, d.ID, true)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
bound, err := voiceBound(ctx, tx, current)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
if !bound || !deviceAllowed(current) {
|
||
|
|
return nil, errAccess
|
||
|
|
}
|
||
|
|
j, err := readJob(ctx, tx, d.ID)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
if j.Registered == nil || j.Generation != current.Generation || j.Instance != in.InstanceID || j.Product != in.ProductKey || j.Name != in.DeviceName {
|
||
|
|
return nil, fail(409, "VOICE_MIGRATION_NOT_ALLOWED", "仅可迁移本设备已完成注册的原凭据")
|
||
|
|
}
|
||
|
|
if len(j.Secret) > 0 {
|
||
|
|
old, e := s.crypt.open(j.Secret, "provider:"+j.ID)
|
||
|
|
if e != nil {
|
||
|
|
return nil, e
|
||
|
|
}
|
||
|
|
if !sameSecret(string(old), in.DeviceSecret) || (j.RTCAppID != "" && j.RTCAppID != in.RTCAppID) {
|
||
|
|
return nil, fail(409, "VOICE_CREDENTIAL_CONFLICT", "平台已有不同凭据，请核对")
|
||
|
|
}
|
||
|
|
if j.RTCAppID == in.RTCAppID {
|
||
|
|
return map[string]any{"accepted": true}, tx.Commit(ctx)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if !oneOf(j.Status, "REGISTERED_OBSERVED", "READY") {
|
||
|
|
return nil, errState
|
||
|
|
}
|
||
|
|
j.Secret = s.crypt.seal([]byte(in.DeviceSecret), "provider:"+j.ID)
|
||
|
|
j.RTCAppID = in.RTCAppID
|
||
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_job SET provider_secret_ciphertext=$2,provider_rtc_app_id=$3,config_ciphertext=NULL,config_expires_at=NULL,config_ack_at=NULL,engine_ready_at=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, j.ID, j.Secret, j.RTCAppID)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
j.Config = nil
|
||
|
|
j.Expires = nil
|
||
|
|
j.Ack = nil
|
||
|
|
if err = s.ensureConfig(ctx, tx, current, &j); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
return map[string]any{"accepted": true}, tx.Commit(ctx)
|
||
|
|
}
|