83 lines
5.6 KiB
Go
83 lines
5.6 KiB
Go
package onboarding
|
|
|
|
import (
|
|
"context"
|
|
"crypto/hmac"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/outbox"
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
const ActivationEvent = "device.onboarding.activation.requested"
|
|
|
|
type activationJob struct {
|
|
ID, DeviceID, Status, Name, Instance, Product, Bot, Transport, MinimumSDK, RTCAppID string
|
|
Generation, ConfigVersion, Version int64
|
|
Secret, Config []byte
|
|
Digest, AttemptID string
|
|
Expires, Ack, Engine, Registered *time.Time
|
|
ReadyGeneration *int64
|
|
}
|
|
|
|
func readJob(ctx context.Context, tx pgx.Tx, device string) (activationJob, error) {
|
|
var j activationJob
|
|
err := tx.QueryRow(ctx, `SELECT id::text,device_id::text,status,provider_device_name,binding_generation,config_version,version,COALESCE(provider_instance_id,''),COALESCE(provider_product_key,''),COALESCE(provider_bot_id,''),COALESCE(provider_transport,''),COALESCE(provider_minimum_sdk,''),COALESCE(provider_rtc_app_id,''),provider_secret_ciphertext,config_ciphertext,COALESCE(config_digest,''),COALESCE(current_attempt_id::text,''),config_expires_at,config_ack_at,engine_ready_at,ready_binding_generation,sdk_registered_at FROM device_activation_job WHERE device_id=$1::uuid FOR UPDATE`, device).Scan(&j.ID, &j.DeviceID, &j.Status, &j.Name, &j.Generation, &j.ConfigVersion, &j.Version, &j.Instance, &j.Product, &j.Bot, &j.Transport, &j.MinimumSDK, &j.RTCAppID, &j.Secret, &j.Config, &j.Digest, &j.AttemptID, &j.Expires, &j.Ack, &j.Engine, &j.ReadyGeneration, &j.Registered)
|
|
return j, err
|
|
}
|
|
func (s *Service) enqueueActivation(ctx context.Context, tx pgx.Tx, d Device, generation int64, requestID string) error {
|
|
_, err := tx.Exec(ctx, `INSERT INTO device_activation_job(device_id,tenant_id,provider_device_name,binding_generation,provider_instance_id,provider_product_key,provider_bot_id,provider_transport,provider_minimum_sdk) VALUES($1::uuid,$2::uuid,$3,$4,$5,$6,$7,$8,$9) ON CONFLICT(device_id) DO NOTHING`, d.ID, d.TenantID, d.FactoryID, generation, s.options.InstanceID, s.options.ProductKey, s.options.BotID, s.options.Transport, s.options.MinimumSDKVersion)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
j, err := readJob(ctx, tx, d.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if j.Generation != generation {
|
|
status := "NOT_STARTED"
|
|
if j.Status == "RECONCILE_REQUIRED" || j.Status == "CONTACT_POSSIBLE" {
|
|
status = "RECONCILE_REQUIRED"
|
|
} else if j.Registered != nil {
|
|
status = "REGISTERED_OBSERVED"
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_job SET binding_generation=$2,config_version=config_version+1,config_ciphertext=NULL,config_digest=NULL,config_expires_at=NULL,config_ack_at=NULL,engine_ready_at=NULL,status=$3,current_attempt_id=NULL,version=version+1,updated_at=now() WHERE id=$1::uuid`, j.ID, generation, status)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_attempt SET status='DEFINITE_FAILURE',completed_at=now(),evidence='{"reason":"BINDING_CHANGED"}' WHERE activation_id=$1::uuid AND executor='DEVICE' AND status IN ('CONTACT_POSSIBLE','UNKNOWN')`, j.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err = bump(ctx, tx, d.ID); err != nil {
|
|
return err
|
|
}
|
|
dedup := fmt.Sprintf("%s:%d", d.ID, generation)
|
|
payload, _ := json.Marshal(map[string]any{"device_id": d.ID, "activation_id": j.ID, "binding_generation": generation})
|
|
_, err = s.outbox.Enqueue(ctx, tx, outbox.NewEvent{TenantID: &d.TenantID, AggregateType: "device", AggregateID: d.ID, EventType: ActivationEvent, DeduplicationKey: &dedup, Payload: payload})
|
|
return err
|
|
}
|
|
|
|
// Keep the established credential encryption format for existing subtitle and
|
|
// session services. The onboarding job remains the authority for READY.
|
|
func (s *Service) storeRuntimeCredential(ctx context.Context, tx pgx.Tx, d Device, j activationJob, secret []byte) error {
|
|
if s.providerCrypt.aead == nil {
|
|
return errors.New("provider credential master key unavailable")
|
|
}
|
|
ref := "volc-hardware://" + d.ID
|
|
aad := fmt.Sprintf("hardware-provider-credential-v1\n%s\n%d\n%s\n%s\n%s", ref, 1, j.Instance, j.Product, j.Name)
|
|
encrypted := s.providerCrypt.seal(secret, aad)
|
|
n := s.providerCrypt.aead.NonceSize()
|
|
_, err := tx.Exec(ctx, `INSERT INTO hardware_agent_provider_credential(tenant_id,device_id,credential_ref,instance_id,product_key,provider_device_name,hardware_id,encrypted_secret,nonce) VALUES($1::uuid,$2::uuid,$3,$4,$5,$6,$7,$8,$9) ON CONFLICT(device_id) DO UPDATE SET encrypted_secret=EXCLUDED.encrypted_secret,nonce=EXCLUDED.nonce,status='ACTIVE',updated_at=now() WHERE hardware_agent_provider_credential.credential_ref=EXCLUDED.credential_ref AND hardware_agent_provider_credential.provider_device_name=EXCLUDED.provider_device_name`, d.TenantID, d.ID, ref, j.Instance, j.Product, j.Name, d.FactoryID, encrypted[n:], encrypted[:n])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO hardware_agent_device_binding(tenant_id,device_id,provider,product_key,bot_id,transport,minimum_sdk_version,status) VALUES($1::uuid,$2::uuid,'volcengine',$3,$4,$5,$6,'ACTIVE') ON CONFLICT(device_id) DO UPDATE SET product_key=EXCLUDED.product_key,bot_id=EXCLUDED.bot_id,transport=EXCLUDED.transport,minimum_sdk_version=EXCLUDED.minimum_sdk_version,status='ACTIVE',updated_at=now()`, d.TenantID, d.ID, j.Product, j.Bot, j.Transport, j.MinimumSDK)
|
|
return err
|
|
}
|
|
func sameSecret(a, b string) bool { return hmac.Equal([]byte(a), []byte(b)) }
|