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.
272 lines
14 KiB
Go
272 lines
14 KiB
Go
package onboarding
|
|
|
|
import (
|
|
"context"
|
|
"data-terminal-cloud/internal/deviceactivation"
|
|
"data-terminal-cloud/internal/identity"
|
|
"data-terminal-cloud/internal/outbox"
|
|
"encoding/json"
|
|
"errors"
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
type RotationInput struct {
|
|
Reason string `json:"reason"`
|
|
ExpectedVersion int64 `json:"expected_key_version"`
|
|
}
|
|
|
|
func (s *Service) Rotate(ctx context.Context, p identity.Principal, tenant, device, idempotencyKey, requestID string, raw []byte, in RotationInput) (map[string]any, error) {
|
|
if !uuidValid(device) || !nonempty(in.Reason, 200) || in.ExpectedVersion < 1 || len(idempotencyKey) < 16 || len(idempotencyKey) > 160 {
|
|
return nil, fail(400, "VALIDATION_FAILED", "密钥轮换参数不正确")
|
|
}
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
a, err := operatorAccess(ctx, tx, p, tenant, true)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var internal string
|
|
err = tx.QueryRow(ctx, `SELECT id::text FROM device WHERE public_id=$1::uuid AND tenant_id=$2::uuid`, device, a.TenantID).Scan(&internal)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, errAccess
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
d, err := readDevice(ctx, tx, internal, true)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !deviceAllowed(d) {
|
|
return nil, errAccess
|
|
}
|
|
var job, stored string
|
|
err = tx.QueryRow(ctx, `SELECT id::text,request_digest FROM device_enrollment_job WHERE device_id=$1::uuid AND actor_id=$2::uuid AND request_key=$3`, d.ID, a.ID, idempotencyKey).Scan(&job, &stored)
|
|
if err == nil {
|
|
if stored != digest(raw) {
|
|
return nil, fail(409, "REQUEST_BODY_CONFLICT", "轮换请求号已用于其他内容")
|
|
}
|
|
return inventoryResult(d, job), tx.Commit(ctx)
|
|
}
|
|
if !errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, err
|
|
}
|
|
var active, latest int64
|
|
var pending bool
|
|
err = tx.QueryRow(ctx, `SELECT COALESCE(max(key_version) FILTER(WHERE status='ACTIVE'),0),max(key_version),bool_or(status='PENDING') FROM device_onboarding_key WHERE device_id=$1::uuid`, d.ID).Scan(&active, &latest, &pending)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if active != in.ExpectedVersion {
|
|
return nil, fail(409, "KEY_VERSION_CONFLICT", "当前凭据版本已变化")
|
|
}
|
|
if pending {
|
|
return nil, fail(409, "DELIVERY_PENDING", "已有凭据交付任务待完成")
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO device_onboarding_key(device_id,tenant_id,key_version,ciphertext,status) VALUES($1::uuid,$2::uuid,$3,$4,'PENDING')`, d.ID, d.TenantID, latest+1, s.crypt.seal(randomBytes(32), keyAAD(d.ID, latest+1)))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
err = tx.QueryRow(ctx, `INSERT INTO device_enrollment_job(tenant_id,device_id,actor_id,request_key,request_digest,batch_no,key_version) VALUES($1::uuid,$2::uuid,$3::uuid,$4,$5,'credential-rotation',$6) RETURNING id::text`, d.TenantID, d.ID, a.ID, idempotencyKey, digest(raw), latest+1).Scan(&job)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err = s.auditEvent(ctx, tx, d, "user", a.PublicID, "onboarding.key.rotation.prepare", requestID, map[string]any{"job_id": job, "key_version": latest + 1, "reason": in.Reason}); err != nil {
|
|
return nil, err
|
|
}
|
|
result := inventoryResult(d, job)
|
|
result["key_version"] = latest + 1
|
|
return result, tx.Commit(ctx)
|
|
}
|
|
func (s *Service) OpsList(ctx context.Context, p identity.Principal, tenant string) (map[string]any, error) {
|
|
a, err := operatorAccess(ctx, s.pool, p, tenant, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.pool.Query(ctx, `SELECT d.public_id::text,d.factory_device_id,d.model,d.binding_generation,d.onboarding_revision,d.onboarding_released_at IS NOT NULL,EXISTS(SELECT 1 FROM device_binding b WHERE b.device_id=d.id AND b.unbound_at IS NULL),COALESCE(j.id::text,''),COALESCE(j.status,'NOT_STARTED'),COALESCE(j.provider_device_name,''),COALESCE(j.last_error_code,''),COALESCE(j.version,0),COALESCE((SELECT a.id::text FROM device_activation_attempt a WHERE a.activation_id=j.id AND a.executor='DEVICE' AND a.status IN ('CONTACT_POSSIBLE','UNKNOWN') ORDER BY a.created_at DESC LIMIT 1),j.current_attempt_id::text,''),COALESCE((SELECT max(key_version) FROM device_onboarding_key k WHERE k.device_id=d.id AND k.status='PENDING'),0),COALESCE((SELECT max(key_version) FROM device_onboarding_key k WHERE k.device_id=d.id AND k.status='ACTIVE'),0),COALESCE((SELECT ej.id::text FROM device_enrollment_job ej WHERE ej.device_id=d.id ORDER BY created_at DESC LIMIT 1),'') FROM device d LEFT JOIN device_activation_job j ON j.device_id=d.id WHERE d.tenant_id=$1::uuid AND d.enrollment_mode='physical_confirm' ORDER BY d.created_at DESC LIMIT 200`, a.TenantID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []map[string]any{}
|
|
for rows.Next() {
|
|
var id, factory, model, job, status, name, code, enrollment, attempt string
|
|
var gen, rev, ver, key, pendingKey int64
|
|
var released, bound bool
|
|
if err = rows.Scan(&id, &factory, &model, &gen, &rev, &released, &bound, &job, &status, &name, &code, &ver, &attempt, &pendingKey, &key, &enrollment); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, map[string]any{"platform_device_id": id, "device_id": factory, "model": model, "binding_generation": gen, "device_revision": rev, "released": released, "bound": bound, "activation_id": job, "activation_state": status, "provider_device_name": name, "last_error_code": code, "activation_version": ver, "key_version": key, "pending_key_version": pendingKey, "current_attempt_id": attempt, "enrollment_job_id": enrollment})
|
|
}
|
|
return map[string]any{"items": items, "limit": 200}, rows.Err()
|
|
}
|
|
|
|
type ReconcileInput struct {
|
|
ExpectedVersion int64 `json:"expected_version"`
|
|
Conclusion string `json:"conclusion"`
|
|
EvidenceRef string `json:"evidence_ref"`
|
|
Reason string `json:"reason"`
|
|
AttemptID string `json:"attempt_id,omitempty"`
|
|
DeviceSecret string `json:"device_secret,omitempty"`
|
|
RTCAppID string `json:"rtc_app_id,omitempty"`
|
|
}
|
|
|
|
func (s *Service) Reconcile(ctx context.Context, p identity.Principal, tenant, activationID, requestID string, in ReconcileInput) (map[string]any, error) {
|
|
if !uuidV4(activationID) || in.ExpectedVersion < 1 || !nonempty(in.EvidenceRef, 500) || !nonempty(in.Reason, 200) || !oneOf(in.Conclusion, "OBSERVED_REGISTERED", "CONFIRMED_NOT_REGISTERED", "DEVICE_ATTEMPT_STOPPED") {
|
|
return nil, fail(400, "VALIDATION_FAILED", "对账结论、版本和证据必填")
|
|
}
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
a, err := operatorAccess(ctx, tx, p, tenant, true)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if a.Role != "ops_admin" {
|
|
return nil, errAccess
|
|
}
|
|
var device string
|
|
err = tx.QueryRow(ctx, `SELECT device_id::text FROM device_activation_job WHERE id=$1::uuid AND tenant_id=$2::uuid`, activationID, a.TenantID).Scan(&device)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, errAccess
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
d, err := readDevice(ctx, tx, device, true)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
j, err := readJob(ctx, tx, d.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if j.Version != in.ExpectedVersion {
|
|
return nil, fail(409, "ACTIVATION_VERSION_CONFLICT", "任务已变化，请刷新后核对")
|
|
}
|
|
if in.Conclusion == "DEVICE_ATTEMPT_STOPPED" {
|
|
if !uuidV4(in.AttemptID) {
|
|
return nil, fail(400, "VALIDATION_FAILED", "请指定设备尝试")
|
|
}
|
|
tag, err := tx.Exec(ctx, `UPDATE device_activation_attempt SET status='DEFINITE_FAILURE',completed_at=now(),evidence=jsonb_build_object('operator_evidence',$3::text) WHERE id=$1::uuid AND activation_id=$2::uuid AND executor='DEVICE' AND stage='CONVERSATION_TEST' AND status IN ('CONTACT_POSSIBLE','UNKNOWN') AND permit_expires_at<now()-interval '3 minutes'`, in.AttemptID, j.ID, in.EvidenceRef)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if tag.RowsAffected() != 1 {
|
|
return nil, fail(409, "ATTEMPT_STILL_POSSIBLE", "许可尚未超过安全等待期，或已确认结果")
|
|
}
|
|
} else {
|
|
if !oneOf(j.Status, "CONTACT_POSSIBLE", "RECONCILE_REQUIRED", "BLOCKED") || len(j.Secret) > 0 {
|
|
return nil, errState
|
|
}
|
|
var attempt string
|
|
var ended bool
|
|
err = tx.QueryRow(ctx, `SELECT id::text,permit_expires_at<now()-interval '3 minutes' FROM device_activation_attempt WHERE activation_id=$1::uuid AND (executor='SERVER' OR stage='SDK_REGISTER') ORDER BY created_at DESC LIMIT 1 FOR UPDATE`, j.ID).Scan(&attempt, &ended)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !ended {
|
|
return nil, fail(409, "ATTEMPT_STILL_POSSIBLE", "供应商调用尚未超过安全等待期")
|
|
}
|
|
if in.Conclusion == "OBSERVED_REGISTERED" {
|
|
if (in.DeviceSecret != "" && len(in.DeviceSecret) < 16) || len(in.DeviceSecret) > 4096 || len(in.RTCAppID) > 200 {
|
|
return nil, fail(400, "VALIDATION_FAILED", "请提供核验后的设备凭据")
|
|
}
|
|
if in.DeviceSecret != "" {
|
|
j.Secret = s.crypt.seal([]byte(in.DeviceSecret), "provider:"+j.ID)
|
|
}
|
|
observed := s.now()
|
|
j.Registered = &observed
|
|
j.RTCAppID = in.RTCAppID
|
|
status := "BLOCKED"
|
|
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 AND generation=$2)`, d.ID, d.Generation).Scan(&bound); err != nil {
|
|
return nil, err
|
|
}
|
|
if bound && deviceAllowed(d) && j.Generation == d.Generation {
|
|
status = "REGISTERED_OBSERVED"
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_job SET provider_secret_ciphertext=$2,provider_rtc_app_id=$3,status=$4,sdk_registered_at=COALESCE(sdk_registered_at,now()),engine_ready_at=now() WHERE id=$1::uuid`, j.ID, j.Secret, j.RTCAppID, status)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_attempt SET status='SUCCESS_OBSERVED',completed_at=now(),evidence=jsonb_build_object('operator_evidence',$2::text) WHERE id=$1::uuid`, attempt, in.EvidenceRef)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
j.Status = status
|
|
if status == "REGISTERED_OBSERVED" {
|
|
if d.Lifecycle == "BOUND" {
|
|
if err = requireLifecycle(ctx, tx, d, deviceactivation.Provisioning, requestID); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
if err = s.ensureConfig(ctx, tx, d, &j); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
} else {
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_attempt SET status='DEFINITE_FAILURE',completed_at=now(),evidence=jsonb_build_object('operator_evidence',$2::text) WHERE id=$1::uuid`, attempt, in.EvidenceRef)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_job SET status='NOT_STARTED',current_attempt_id=NULL WHERE id=$1::uuid`, j.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
dedup := "reconcile:" + requestID
|
|
payload, _ := json.Marshal(map[string]any{"device_id": d.ID, "activation_id": j.ID, "binding_generation": d.Generation})
|
|
_, err = s.outbox.Enqueue(ctx, tx, outbox.NewEvent{TenantID: &d.TenantID, AggregateType: "device", AggregateID: d.ID, EventType: ActivationEvent, DeduplicationKey: &dedup, Payload: payload})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
_, err = tx.Exec(ctx, `UPDATE device_activation_job SET version=version+1,reconciliation_evidence=$2,last_error_code=NULL,status=CASE WHEN $3='DEVICE_ATTEMPT_STOPPED' THEN 'REGISTERED_OBSERVED' ELSE status END,updated_at=now() WHERE id=$1::uuid`, j.ID, in.EvidenceRef, in.Conclusion)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err = bump(ctx, tx, d.ID); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = s.auditEvent(ctx, tx, d, "user", a.PublicID, "onboarding.activation.reconcile", requestID, map[string]any{"activation_id": j.ID, "conclusion": in.Conclusion, "evidence_ref": in.EvidenceRef, "reason": in.Reason}); err != nil {
|
|
return nil, err
|
|
}
|
|
st, err := s.state(ctx, tx, d)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return map[string]any{"state": st}, tx.Commit(ctx)
|
|
}
|
|
|
|
// This status route remains available after the temporary setup closes.
|
|
func (s *Service) AccountDevices(ctx context.Context, p identity.Principal) (map[string]any, error) {
|
|
rows, err := s.pool.Query(ctx, `SELECT d.public_id::text,d.factory_device_id,d.binding_generation,d.onboarding_revision,COALESCE(j.status,'NOT_STARTED'),h.name,sp.preferred_name FROM device d JOIN device_binding b ON b.device_id=d.id AND b.unbound_at IS NULL JOIN household h ON h.id=b.household_id AND h.status='active' JOIN subject_profile sp ON sp.id=b.subject_id JOIN household_member hm ON hm.household_id=h.id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.status='active' JOIN tenant t ON t.id=d.tenant_id AND t.status='active' LEFT JOIN device_activation_job j ON j.device_id=d.id WHERE ua.oidc_issuer=$1 AND ua.oidc_subject=$2 AND d.enrollment_mode='physical_confirm' ORDER BY d.registered_at DESC LIMIT 200`, p.Issuer, p.Subject)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []map[string]any{}
|
|
for rows.Next() {
|
|
var id, factory, status, household, name string
|
|
var gen, rev int64
|
|
if err = rows.Scan(&id, &factory, &gen, &rev, &status, &household, &name); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, map[string]any{"platform_device_id": id, "device_id": factory, "binding_generation": gen, "device_revision": rev, "activation_state": status, "household_name": household, "preferred_name": name})
|
|
}
|
|
return map[string]any{"items": items}, rows.Err()
|
|
}
|
|
|
|
// Recovered settings never reset ownership; the regular authorized unbind
|
|
// route must first close the binding, then firmware may clear local user data.
|
|
func requireLifecycle(ctx context.Context, tx pgx.Tx, d Device, target deviceactivation.State, requestID string) error {
|
|
_, err := deviceactivation.Apply(ctx, tx, deviceactivation.Input{DeviceID: d.ID, TenantID: d.TenantID, Target: target, Reason: "首次接入状态同步", Source: "onboarding", ActorType: "system", RequestID: requestID})
|
|
return err
|
|
}
|