From 36786723e376bd6ec8d9a4e9823ea4096312b723 Mon Sep 17 00:00:00 2001 From: QiuSW <105186638@qq.com> Date: Fri, 7 Aug 2026 18:23:24 +0800 Subject: [PATCH] feat(store): add Area admission and audit outbox [T-010] --- Sense/README.md | 10 +- Sense/internal/device/model.go | 33 +- Sense/internal/store/audit.go | 127 +++++++ Sense/internal/store/postgres.go | 214 +++++++++-- Sense/internal/store/postgres_test.go | 336 +++++++++++++++++- Sense/internal/store/repository.go | 22 ++ Sense/internal/store/sqlite_test.go | 3 +- deploy/postgres/005_area_policy.sql | 127 +++++++ .../postgres/006_device_operation_outbox.sql | 67 ++++ deploy/postgres/007_privileges_area_audit.sql | 18 + deploy/postgres/README.md | 11 +- deploy/postgres/tests/assertions.sql | 65 +++- docs/04-architecture.md | 18 +- docs/06-tasks.md | 3 +- docs/api.md | 14 +- docs/contracts/README.md | 14 +- docs/contracts/area-policy-v1.sql | 30 ++ .../sense-device-audit-v1.schema.json | 109 ++++++ docs/current-state.md | 13 +- docs/tasks/T-010.md | 11 +- scripts/test_postgres.ps1 | 21 +- tests/test_area_audit_contract.py | 154 ++++++++ tests/test_postgres_contract.py | 25 +- 23 files changed, 1346 insertions(+), 99 deletions(-) create mode 100644 Sense/internal/store/audit.go create mode 100644 deploy/postgres/005_area_policy.sql create mode 100644 deploy/postgres/006_device_operation_outbox.sql create mode 100644 deploy/postgres/007_privileges_area_audit.sql create mode 100644 docs/contracts/area-policy-v1.sql create mode 100644 docs/contracts/sense-device-audit-v1.schema.json create mode 100644 tests/test_area_audit_contract.py diff --git a/Sense/README.md b/Sense/README.md index 30fec3c..1647822 100644 --- a/Sense/README.md +++ b/Sense/README.md @@ -1,6 +1,6 @@ # Sense M1/M2 接入骨架 -本目录是 YoVision Sense 的 M1/M2 接入骨架。数据库保存期望态，ONVIF 和 MediaMTX 通过端口隔离；M1 默认使用 SQLite，T-009 增加 PostgreSQL 双 schema 生产路径。默认关闭真实 ONVIF，显式设置 `SENSE_ONVIF_MODE=standard` 后才启用标准 SOAP/WS-Security 适配器。T-006 的真实样机结论仅覆盖已批准的精确海康基线，不能据此宣称多品牌兼容。 +本目录是 YoVision Sense 的 M1/M2 接入骨架。数据库保存期望态，ONVIF 和 MediaMTX 通过端口隔离；M1 默认使用 SQLite，T-009/T-010 增加 PostgreSQL 双 schema、Area 准入和本地审计 Outbox 生产基础。默认关闭真实 ONVIF，显式设置 `SENSE_ONVIF_MODE=standard` 后才启用标准 SOAP/WS-Security 适配器。T-006 的真实样机结论仅覆盖已批准的精确海康基线，不能据此宣称多品牌兼容。 ## 常用命令 @@ -45,9 +45,11 @@ SENSE_CREDENTIAL__RTSP_PASSWORD MediaMTX `v1.19.3` 应作为独立二进制启动并只在可信网络开放 API。获取与 SHA-256 校验值见 `docs/03-tech-stack.md`。生成客户端使用固定版本工具和 vendored 官方 OpenAPI；`internal/mtx/generated/client.gen.go` 不可手改。 -## T-009 PostgreSQL 17.10 +## T-009/T-010 PostgreSQL 17.10 -初始化 SQL 位于 `deploy/postgres/`，由高权限部署步骤按文件名前缀执行；Sense 进程不会自动创建角色、schema 或 Bell 对象。`bell_app` 拥有 Site/配额和 `bell.site_quota_v1`，`sense_app` 只能读取该视图，不能读取或写入 Bell 源表。应用登录角色和密码由部署环境创建，不进入仓库。 +初始化与增量 SQL 位于 `deploy/postgres/`，由高权限部署步骤按文件名前缀执行；Sense 进程不会自动创建角色、schema 或 Bell 对象。`bell_app` 拥有 Site/Area、配额、`capture_policy` 及两个版本化视图，`sense_app` 只能读取 `bell.site_quota_v1` 和 `bell.area_policy_v1`，不能读取或写入 Bell 源表。应用登录角色和密码由部署环境创建，不进入仓库。 + +PostgreSQL 新建设备必须携带匹配 tenant/Site 的 `area_id`。具有 `video_capture` 能力的设备在创建和从 disabled 切到 enabled 时执行 Area 准入；`non_imaging_only` 拒绝成像设备但允许非成像设备。投影缺失、非法或版本回退只拒绝新变更，不关闭已有流。设备创建和期望态受理与 `sense.device_operation_outbox` 在同一事务写入；Outbox 不保存 endpoint、credential、完整 URI 或 MediaMTX 配置。Bell relay 尚未实现。 Windows 本机集成测试从仓库根目录执行： @@ -65,7 +67,7 @@ $env:SENSE_DB_DSN = '由部署环境私下设置' go run ./cmd/sense-api ``` -PostgreSQL 启动会检查 Sense migration 版本及当前角色的跨 schema 权限；权限过宽、配额视图不可读或 schema 未安装时 readiness 初始化失败。默认 SQLite 路径和 `cmd/sense-lab` 保持不变，T-009 不搬迁现有 SQLite 数据。 +PostgreSQL 启动会检查 Sense migration 版本及当前角色对两个 Bell 投影视图的最小权限；权限过宽、视图不可读或 schema 未安装时 readiness 初始化失败。默认 SQLite 路径和 `cmd/sense-lab` 保持不变，但 SQLite 不实现生产 Area/Outbox 一致性，后续公共控制 API 不得在 SQLite 上启用。 Windows 本地准备 MediaMTX（从仓库根目录执行）： diff --git a/Sense/internal/device/model.go b/Sense/internal/device/model.go index d371f21..5cc9266 100644 --- a/Sense/internal/device/model.go +++ b/Sense/internal/device/model.go @@ -75,21 +75,24 @@ func (s Site) Validate() error { } type Device struct { - ID string - TenantID string - SiteID string - SerialNumber string - Name string - Modality Modality - Capabilities []Capability - DesiredState DesiredState - ActualState ActualState - EndpointRef string - CredentialRef string - PathName string - Generation int64 - CreatedAt time.Time - UpdatedAt time.Time + ID string + TenantID string + SiteID string + AreaID string + SerialNumber string + Name string + Modality Modality + Capabilities []Capability + DesiredState DesiredState + ActualState ActualState + EndpointRef string + CredentialRef string + PathName string + Generation int64 + QuotaSourceVersion int64 + AreaPolicySourceVersion int64 + CreatedAt time.Time + UpdatedAt time.Time } func (d Device) Validate() error { diff --git a/Sense/internal/store/audit.go b/Sense/internal/store/audit.go new file mode 100644 index 0000000..630bfdc --- /dev/null +++ b/Sense/internal/store/audit.go @@ -0,0 +1,127 @@ +package store + +import ( + "context" + "crypto/rand" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "strings" + "time" +) + +type AuditActorType string + +const ( + AuditActorUser AuditActorType = "user" + AuditActorService AuditActorType = "service" + AuditActorSystem AuditActorType = "system" +) + +// AuditContext is supplied by a verified caller boundary. Repository callers +// without an HTTP principal use the explicit system/sense fallback. +type AuditContext struct { + ActorType AuditActorType + ActorID string + Reason string + TraceID string +} + +type auditContextKey struct{} + +func WithAuditContext(ctx context.Context, value AuditContext) context.Context { + return context.WithValue(ctx, auditContextKey{}, value) +} + +func auditFromContext(ctx context.Context) AuditContext { + value, ok := ctx.Value(auditContextKey{}).(AuditContext) + if !ok { + return AuditContext{ActorType: AuditActorSystem, ActorID: "sense"} + } + return value +} + +func validateAudit(value AuditContext) error { + if value.ActorType != AuditActorUser && value.ActorType != AuditActorService && value.ActorType != AuditActorSystem { + return errors.New("invalid audit actor type") + } + if strings.TrimSpace(value.ActorID) == "" || len(value.ActorID) > 200 { + return errors.New("invalid audit actor ID") + } + if len(value.Reason) > 500 { + return errors.New("audit reason is too long") + } + if len(value.TraceID) > 128 { + return errors.New("audit trace ID is too long") + } + return nil +} + +type postgresAuditEvent struct { + EventType string + TenantID string + SiteID string + DeviceID string + Generation int64 + QuotaSourceVersion int64 + AreaPolicySourceVersion int64 + Payload any + OccurredAt time.Time +} + +func insertPostgresAudit( + ctx context.Context, + tx *sql.Tx, + event postgresAuditEvent, +) error { + audit := auditFromContext(ctx) + if err := validateAudit(audit); err != nil { + return err + } + eventID, err := newAuditEventID() + if err != nil { + return errors.New("generate audit event ID") + } + payload, err := json.Marshal(event.Payload) + if err != nil { + return errors.New("encode audit payload") + } + if !json.Valid(payload) { + return errors.New("invalid audit payload") + } + occurredAt := event.OccurredAt.UTC() + if occurredAt.IsZero() { + occurredAt = time.Now().UTC() + } + _, err = tx.ExecContext(ctx, `INSERT INTO sense.device_operation_outbox( + event_id, event_type, tenant_id, site_id, device_id, + actor_type, actor_id, reason, trace_id, aggregate_generation, + quota_source_version, area_policy_source_version, payload, + occurred_at, available_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, NULLIF($8, ''), NULLIF($9, ''), + $10, $11, $12, $13, $14, $14)`, + eventID, event.EventType, event.TenantID, event.SiteID, event.DeviceID, + audit.ActorType, audit.ActorID, audit.Reason, audit.TraceID, + event.Generation, nullableVersion(event.QuotaSourceVersion), + nullableVersion(event.AreaPolicySourceVersion), payload, occurredAt) + if err != nil { + return errors.New("insert device operation audit") + } + return nil +} + +func newAuditEventID() (string, error) { + value := make([]byte, 16) + if _, err := rand.Read(value); err != nil { + return "", err + } + return "audit_" + hex.EncodeToString(value), nil +} + +func nullableVersion(value int64) any { + if value < 1 { + return nil + } + return value +} diff --git a/Sense/internal/store/postgres.go b/Sense/internal/store/postgres.go index ab41a50..a72e2e4 100644 --- a/Sense/internal/store/postgres.go +++ b/Sense/internal/store/postgres.go @@ -16,7 +16,7 @@ import ( ) // Postgres persists Sense state in the sense schema and consumes only Bell's -// versioned quota view. Migrations are deliberately installed out of process. +// versioned quota and Area views. Migrations are installed out of process. type Postgres struct { db *sql.DB } @@ -53,20 +53,29 @@ func (s *Postgres) Close() error { func (s *Postgres) verifySchemaAndPrivileges(ctx context.Context) error { var version sql.NullInt64 if err := s.db.QueryRowContext(ctx, - `SELECT MAX(version) FROM sense.schema_migrations`).Scan(&version); err != nil || !version.Valid || version.Int64 < 1 { - return errors.New("postgres sense schema migration v1 is required") + `SELECT MAX(version) FROM sense.schema_migrations`).Scan(&version); err != nil || !version.Valid || version.Int64 < 3 { + return errors.New("postgres sense schema migration v3 is required") } - var canReadView, canWriteView, canReadSource, canWriteSource bool + var canReadQuotaView, canWriteQuotaView, canReadSiteSource, canWriteSiteSource bool + var canReadAreaView, canWriteAreaView, canReadAreaSource, canWriteAreaSource bool if err := s.db.QueryRowContext(ctx, `SELECT has_table_privilege(current_user, 'bell.site_quota_v1', 'SELECT'), has_table_privilege(current_user, 'bell.site_quota_v1', 'INSERT,UPDATE,DELETE'), has_table_privilege(current_user, 'bell.sites', 'SELECT'), - has_table_privilege(current_user, 'bell.sites', 'INSERT,UPDATE,DELETE')`). - Scan(&canReadView, &canWriteView, &canReadSource, &canWriteSource); err != nil { - return errors.New("verify postgres quota privileges") + has_table_privilege(current_user, 'bell.sites', 'INSERT,UPDATE,DELETE'), + has_table_privilege(current_user, 'bell.area_policy_v1', 'SELECT'), + has_table_privilege(current_user, 'bell.area_policy_v1', 'INSERT,UPDATE,DELETE'), + has_table_privilege(current_user, 'bell.areas', 'SELECT'), + has_table_privilege(current_user, 'bell.areas', 'INSERT,UPDATE,DELETE')`). + Scan( + &canReadQuotaView, &canWriteQuotaView, &canReadSiteSource, &canWriteSiteSource, + &canReadAreaView, &canWriteAreaView, &canReadAreaSource, &canWriteAreaSource, + ); err != nil { + return errors.New("verify postgres Bell projection privileges") } - if !canReadView || canWriteView || canReadSource || canWriteSource { - return errors.New("postgres role violates Bell quota privilege boundary") + if !canReadQuotaView || canWriteQuotaView || canReadSiteSource || canWriteSiteSource || + !canReadAreaView || canWriteAreaView || canReadAreaSource || canWriteAreaSource { + return errors.New("postgres role violates Bell projection privilege boundary") } return nil } @@ -92,7 +101,14 @@ func (s *Postgres) CreateDevice(ctx context.Context, value device.Device) error return errors.New("begin postgres create device") } defer tx.Rollback() - var quotaVersion any + areaVersion, err := checkPostgresAreaPolicy( + ctx, tx, value.TenantID, value.SiteID, value.AreaID, + value.HasCapability(device.CapabilityVideoCapture), now, + ) + if err != nil { + return err + } + var quotaVersion int64 if value.ConsumesVideoChannel() { version, quotaErr := checkPostgresVideoQuota(ctx, tx, value.TenantID, value.SiteID, now) if quotaErr != nil { @@ -101,14 +117,15 @@ func (s *Postgres) CreateDevice(ctx context.Context, value device.Device) error quotaVersion = version } _, err = tx.ExecContext(ctx, `INSERT INTO sense.devices( - id, tenant_id, site_id, serial_number, name, modality, + id, tenant_id, site_id, area_id, serial_number, name, modality, desired_state, actual_state, endpoint_ref, credential_ref, - path_name, generation, quota_source_version, created_at, updated_at - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)`, - value.ID, value.TenantID, value.SiteID, value.SerialNumber, value.Name, + path_name, generation, quota_source_version, area_policy_source_version, + created_at, updated_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)`, + value.ID, value.TenantID, value.SiteID, value.AreaID, value.SerialNumber, value.Name, value.Modality, value.DesiredState, value.ActualState, value.EndpointRef, - value.CredentialRef, value.PathName, value.Generation, quotaVersion, - value.CreatedAt, value.UpdatedAt) + value.CredentialRef, value.PathName, value.Generation, nullableVersion(quotaVersion), + areaVersion, value.CreatedAt, value.UpdatedAt) if err != nil { return errors.New("insert postgres device") } @@ -123,12 +140,83 @@ func (s *Postgres) CreateDevice(ctx context.Context, value device.Device) error VALUES ($1, $2)`, value.ID, now); err != nil { return errors.New("insert postgres reconcile state") } + if err := insertPostgresAudit(ctx, tx, postgresAuditEvent{ + EventType: "device.created", TenantID: value.TenantID, SiteID: value.SiteID, + DeviceID: value.ID, Generation: value.Generation, + QuotaSourceVersion: quotaVersion, AreaPolicySourceVersion: areaVersion, + OccurredAt: now, + Payload: map[string]any{ + "kind": "device_created", "area_id": value.AreaID, + "modality": value.Modality, "capabilities": sortedCapabilities(value.Capabilities), + "desired_state": value.DesiredState, + }, + }); err != nil { + return err + } if err := tx.Commit(); err != nil { return errors.New("commit postgres create device") } return nil } +func checkPostgresAreaPolicy( + ctx context.Context, + tx *sql.Tx, + tenantID, siteID, areaID string, + imaging bool, + now time.Time, +) (int64, error) { + if strings.TrimSpace(areaID) == "" { + return 0, areaPolicyUnavailable() + } + // Area projection observation is serialized before the Site quota lock. + // No admission path acquires these locks in the opposite order. + if _, err := tx.ExecContext(ctx, + `SELECT pg_advisory_xact_lock(hashtext($1), hashtext('area:' || $2))`, tenantID, areaID); err != nil { + return 0, errors.New("lock postgres Area admission") + } + var capturePolicy string + var sourceVersion int64 + var sourceUpdatedAt time.Time + err := tx.QueryRowContext(ctx, `SELECT capture_policy, source_version, source_updated_at + FROM bell.area_policy_v1 + WHERE tenant_id = $1 AND site_id = $2 AND area_id = $3`, tenantID, siteID, areaID). + Scan(&capturePolicy, &sourceVersion, &sourceUpdatedAt) + if errors.Is(err, sql.ErrNoRows) { + return 0, areaPolicyUnavailable() + } + if err != nil { + return 0, areaPolicyUnavailable() + } + if (capturePolicy != "video_allowed" && capturePolicy != "non_imaging_only") || + sourceVersion < 1 || sourceUpdatedAt.IsZero() { + return 0, areaPolicyInvalid() + } + var previous sql.NullInt64 + err = tx.QueryRowContext(ctx, `SELECT source_version + FROM sense.area_policy_projection_state + WHERE tenant_id = $1 AND site_id = $2 AND area_id = $3`, tenantID, siteID, areaID). + Scan(&previous) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return 0, errors.New("read postgres Area projection state") + } + if previous.Valid && sourceVersion < previous.Int64 { + return 0, areaPolicyInvalid() + } + if imaging && capturePolicy == "non_imaging_only" { + return 0, areaPolicyDenied() + } + if _, err := tx.ExecContext(ctx, `INSERT INTO sense.area_policy_projection_state( + tenant_id, site_id, area_id, source_version, synced_at + ) VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (tenant_id, site_id, area_id) DO UPDATE SET + source_version = EXCLUDED.source_version, + synced_at = EXCLUDED.synced_at`, tenantID, siteID, areaID, sourceVersion, now); err != nil { + return 0, errors.New("record postgres Area projection state") + } + return sourceVersion, nil +} + func checkPostgresVideoQuota( ctx context.Context, tx *sql.Tx, @@ -199,10 +287,17 @@ func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired devic } defer tx.Rollback() var tenantID, siteID, endpointRef, pathName string + var areaID sql.NullString var current device.DesiredState - err = tx.QueryRowContext(ctx, `SELECT tenant_id, site_id, desired_state, endpoint_ref, path_name + var generation int64 + var storedQuotaVersion, storedAreaVersion sql.NullInt64 + err = tx.QueryRowContext(ctx, `SELECT tenant_id, site_id, area_id, desired_state, + endpoint_ref, path_name, generation, quota_source_version, area_policy_source_version FROM sense.devices WHERE id = $1 FOR UPDATE`, id). - Scan(&tenantID, &siteID, ¤t, &endpointRef, &pathName) + Scan( + &tenantID, &siteID, &areaID, ¤t, &endpointRef, &pathName, + &generation, &storedQuotaVersion, &storedAreaVersion, + ) if errors.Is(err, sql.ErrNoRows) { return ErrNotFound } @@ -210,9 +305,25 @@ func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired devic return errors.New("read postgres device desired state") } if current == desired { - return tx.Commit() + if err := insertPostgresAudit(ctx, tx, postgresAuditEvent{ + EventType: "device.desired_state.accepted", TenantID: tenantID, SiteID: siteID, + DeviceID: id, Generation: generation, + QuotaSourceVersion: storedQuotaVersion.Int64, + AreaPolicySourceVersion: storedAreaVersion.Int64, + OccurredAt: time.Now().UTC(), + Payload: map[string]any{ + "kind": "desired_state_accepted", "previous_desired_state": current, + "desired_state": desired, "changed": false, + }, + }); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return errors.New("commit postgres no-op desired-state audit") + } + return nil } - var quotaVersion any + var quotaVersion, areaVersion int64 if desired == device.DesiredEnabled { var hasVideo bool if err := tx.QueryRowContext(ctx, `SELECT EXISTS( @@ -225,6 +336,13 @@ func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired devic if strings.TrimSpace(endpointRef) == "" || strings.TrimSpace(pathName) == "" { return errors.New("enabled video devices require endpoint ref and path name") } + version, areaErr := checkPostgresAreaPolicy( + ctx, tx, tenantID, siteID, areaID.String, true, time.Now().UTC(), + ) + if areaErr != nil { + return areaErr + } + areaVersion = version version, quotaErr := checkPostgresVideoQuota(ctx, tx, tenantID, siteID, time.Now().UTC()) if quotaErr != nil { return quotaErr @@ -232,21 +350,41 @@ func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired devic quotaVersion = version } } - result, err := tx.ExecContext(ctx, `UPDATE sense.devices SET + now := time.Now().UTC() + var updatedQuotaVersion, updatedAreaVersion sql.NullInt64 + err = tx.QueryRowContext(ctx, `UPDATE sense.devices SET desired_state = $1, actual_state = 'pending', generation = generation + 1, - quota_source_version = COALESCE($2, quota_source_version), updated_at = $3 - WHERE id = $4`, desired, quotaVersion, time.Now().UTC(), id) + quota_source_version = COALESCE($2, quota_source_version), + area_policy_source_version = COALESCE($3, area_policy_source_version), + updated_at = $4 + WHERE id = $5 + RETURNING generation, quota_source_version, area_policy_source_version`, + desired, nullableVersion(quotaVersion), nullableVersion(areaVersion), now, id). + Scan(&generation, &updatedQuotaVersion, &updatedAreaVersion) + if errors.Is(err, sql.ErrNoRows) { + return ErrNotFound + } if err != nil { return errors.New("update postgres desired state") } - if affected, _ := result.RowsAffected(); affected != 1 { - return ErrNotFound - } if _, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET failure_count = 0, next_attempt_at = NULL, last_error_code = NULL, updated_at = $1 - WHERE device_id = $2`, time.Now().UTC(), id); err != nil { + WHERE device_id = $2`, now, id); err != nil { return errors.New("reset postgres reconcile state") } + if err := insertPostgresAudit(ctx, tx, postgresAuditEvent{ + EventType: "device.desired_state.accepted", TenantID: tenantID, SiteID: siteID, + DeviceID: id, Generation: generation, + QuotaSourceVersion: updatedQuotaVersion.Int64, + AreaPolicySourceVersion: updatedAreaVersion.Int64, + OccurredAt: now, + Payload: map[string]any{ + "kind": "desired_state_accepted", "previous_desired_state": current, + "desired_state": desired, "changed": true, + }, + }); err != nil { + return err + } if err := tx.Commit(); err != nil { return errors.New("commit postgres desired-state update") } @@ -288,18 +426,24 @@ func (s *Postgres) ListDueReconcile(ctx context.Context, now time.Time, limit in values := make([]ReconcileCandidate, 0) for rows.Next() { var candidate ReconcileCandidate + var areaID sql.NullString + var quotaVersion, areaVersion sql.NullInt64 var nextAttempt sql.NullTime if err := rows.Scan( &candidate.Device.ID, &candidate.Device.TenantID, &candidate.Device.SiteID, - &candidate.Device.SerialNumber, &candidate.Device.Name, &candidate.Device.Modality, + &areaID, &candidate.Device.SerialNumber, &candidate.Device.Name, &candidate.Device.Modality, &candidate.Device.DesiredState, &candidate.Device.ActualState, &candidate.Device.EndpointRef, &candidate.Device.CredentialRef, &candidate.Device.PathName, &candidate.Device.Generation, + "aVersion, &areaVersion, &candidate.Device.CreatedAt, &candidate.Device.UpdatedAt, &candidate.FailureCount, &nextAttempt, ); err != nil { return nil, errors.New("scan postgres due reconcile device") } + candidate.Device.AreaID = areaID.String + candidate.Device.QuotaSourceVersion = quotaVersion.Int64 + candidate.Device.AreaPolicySourceVersion = areaVersion.Int64 if nextAttempt.Valid { value := nextAttempt.Time candidate.NextAttempt = &value @@ -483,20 +627,26 @@ func (s *Postgres) ConvergenceSnapshot(ctx context.Context) (ConvergenceSnapshot return snapshot, nil } -const postgresDeviceColumns = `d.id, d.tenant_id, d.site_id, d.serial_number, d.name, d.modality, +const postgresDeviceColumns = `d.id, d.tenant_id, d.site_id, d.area_id, d.serial_number, d.name, d.modality, d.desired_state, d.actual_state, d.endpoint_ref, d.credential_ref, - d.path_name, d.generation, d.created_at, d.updated_at` + d.path_name, d.generation, d.quota_source_version, d.area_policy_source_version, + d.created_at, d.updated_at` const postgresDeviceSelect = `SELECT ` + postgresDeviceColumns + ` FROM sense.devices d` func scanPostgresDevice(row scanner) (device.Device, error) { var value device.Device + var areaID sql.NullString + var quotaVersion, areaVersion sql.NullInt64 err := row.Scan( - &value.ID, &value.TenantID, &value.SiteID, &value.SerialNumber, + &value.ID, &value.TenantID, &value.SiteID, &areaID, &value.SerialNumber, &value.Name, &value.Modality, &value.DesiredState, &value.ActualState, &value.EndpointRef, &value.CredentialRef, &value.PathName, - &value.Generation, &value.CreatedAt, &value.UpdatedAt, + &value.Generation, "aVersion, &areaVersion, &value.CreatedAt, &value.UpdatedAt, ) + value.AreaID = areaID.String + value.QuotaSourceVersion = quotaVersion.Int64 + value.AreaPolicySourceVersion = areaVersion.Int64 return value, err } diff --git a/Sense/internal/store/postgres_test.go b/Sense/internal/store/postgres_test.go index 110df1d..c2daaf3 100644 --- a/Sense/internal/store/postgres_test.go +++ b/Sense/internal/store/postgres_test.go @@ -48,6 +48,7 @@ func TestPostgresDefaultAndMaximumQuota(t *testing.T) { } radar := device.Device{ ID: "radar-default", TenantID: "tenant-default", SiteID: "site-default", + AreaID: "area-default", SerialNumber: "radar-default", Name: "Radar", Modality: device.ModalityRadar, Capabilities: []device.Capability{device.CapabilityTelemetry}, DesiredState: device.DesiredEnabled, ActualState: device.ActualPending, @@ -113,8 +114,8 @@ func TestPostgresProjectionFailureAndRollbackFailClosed(t *testing.T) { store, admin := openPostgresTestStore(t) ctx := context.Background() err := store.CreateDevice(ctx, videoDevice(1, "missing-tenant", "missing-site")) - if !errors.Is(err, ErrQuotaProjectionUnavailable) { - t.Fatalf("missing projection must fail closed, got %v", err) + if !errors.Is(err, ErrAreaPolicyUnavailable) { + t.Fatalf("missing Area projection must fail closed first, got %v", err) } insertBellSite(t, admin, "tenant", "site", 2) @@ -285,12 +286,12 @@ func TestPostgresOpenRejectsOverprivilegedRuntimeRole(t *testing.T) { _, admin := openPostgresTestStore(t) ctx := context.Background() if _, err := admin.ExecContext(ctx, - `GRANT UPDATE ON bell.site_quota_v1 TO yovision_t009_sense`); err != nil { + `GRANT UPDATE ON bell.site_quota_v1 TO yovision_t010_sense`); err != nil { t.Fatal(err) } defer func() { _, _ = admin.ExecContext(context.Background(), - `REVOKE UPDATE ON bell.site_quota_v1 FROM yovision_t009_sense`) + `REVOKE UPDATE ON bell.site_quota_v1 FROM yovision_t010_sense`) }() value, err := OpenPostgres(ctx, os.Getenv(postgresTestDSNEnv)) if value != nil { @@ -302,6 +303,316 @@ func TestPostgresOpenRejectsOverprivilegedRuntimeRole(t *testing.T) { } } +func TestPostgresAreaPolicyAllowsNonImagingAndDeniesImagingCreate(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + insertBellSite(t, admin, "tenant", "site", 2) + if _, err := admin.ExecContext(ctx, `UPDATE bell.areas SET capture_policy = 'non_imaging_only' + WHERE tenant_id = 'tenant' AND id = 'area-default'`); err != nil { + t.Fatal(err) + } + + blocked := videoDevice(1, "tenant", "site") + blocked.DesiredState = device.DesiredDisabled + err := store.CreateDevice(ctx, blocked) + if !errors.Is(err, ErrAreaPolicyDenied) { + t.Fatalf("disabled imaging create must still be denied, got %v", err) + } + + radar := device.Device{ + ID: "radar-001", TenantID: "tenant", SiteID: "site", AreaID: "area-default", + SerialNumber: "radar-001", Name: "Radar", Modality: device.ModalityRadar, + Capabilities: []device.Capability{device.CapabilityTelemetry}, + DesiredState: device.DesiredEnabled, ActualState: device.ActualPending, + } + if err := store.CreateDevice(ctx, radar); err != nil { + t.Fatalf("non-imaging device must be allowed: %v", err) + } + stored, err := store.GetDevice(ctx, radar.ID) + if err != nil { + t.Fatal(err) + } + if stored.AreaID != "area-default" || stored.AreaPolicySourceVersion != 2 { + t.Fatalf("Area projection evidence was not stored: %+v", stored) + } + var devices, audits int + if err := admin.QueryRowContext(ctx, `SELECT count(*) FROM sense.devices`).Scan(&devices); err != nil { + t.Fatal(err) + } + if err := admin.QueryRowContext(ctx, `SELECT count(*) FROM sense.device_operation_outbox`).Scan(&audits); err != nil { + t.Fatal(err) + } + if devices != 1 || audits != 1 { + t.Fatalf("denied create left partial state: devices=%d audits=%d", devices, audits) + } +} + +func TestPostgresAreaProjectionMissingAndRollbackFailClosed(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + insertBellSite(t, admin, "tenant", "site", 2) + + missing := videoDevice(1, "tenant", "site") + missing.AreaID = "missing-area" + if err := store.CreateDevice(ctx, missing); !errors.Is(err, ErrAreaPolicyUnavailable) { + t.Fatalf("missing Area projection must fail closed, got %v", err) + } + if _, err := store.db.ExecContext(ctx, `INSERT INTO sense.area_policy_projection_state( + tenant_id, site_id, area_id, source_version, synced_at + ) VALUES ('tenant', 'site', 'area-default', 99, clock_timestamp())`); err != nil { + t.Fatal(err) + } + if err := store.CreateDevice(ctx, videoDevice(2, "tenant", "site")); !errors.Is(err, ErrAreaPolicyInvalid) { + t.Fatalf("Area source-version rollback must fail closed, got %v", err) + } + var devices, audits int + if err := admin.QueryRowContext(ctx, `SELECT count(*) FROM sense.devices`).Scan(&devices); err != nil { + t.Fatal(err) + } + if err := admin.QueryRowContext(ctx, `SELECT count(*) FROM sense.device_operation_outbox`).Scan(&audits); err != nil { + t.Fatal(err) + } + if devices != 0 || audits != 0 { + t.Fatalf("failed Area admissions persisted state: devices=%d audits=%d", devices, audits) + } +} + +func TestPostgresConcurrentAreaObservationIsMonotonic(t *testing.T) { + store, admin := openPostgresTestStore(t) + insertBellSite(t, admin, "tenant", "site", 2) + ctx := context.Background() + start := make(chan struct{}) + errorsFound := make(chan error, 2) + var wait sync.WaitGroup + for index := 1; index <= 2; index++ { + wait.Add(1) + go func(index int) { + defer wait.Done() + value := videoDevice(index, "tenant", "site") + value.DesiredState = device.DesiredDisabled + <-start + errorsFound <- store.CreateDevice(ctx, value) + }(index) + } + close(start) + wait.Wait() + close(errorsFound) + for err := range errorsFound { + if err != nil { + t.Fatalf("concurrent Area observation failed: %v", err) + } + } + var sourceVersion int64 + if err := admin.QueryRowContext(ctx, `SELECT source_version + FROM sense.area_policy_projection_state + WHERE tenant_id = 'tenant' AND site_id = 'site' AND area_id = 'area-default'`). + Scan(&sourceVersion); err != nil { + t.Fatal(err) + } + if sourceVersion != 1 { + t.Fatalf("concurrent observation recorded version %d", sourceVersion) + } +} + +func TestPostgresAreaProjectionCannotCrossTenantBoundary(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + insertBellSite(t, admin, "tenant-a", "site", 2) + insertBellSite(t, admin, "tenant-b", "site", 2) + insertBellArea(t, admin, "tenant-b", "site", "private-area", "video_allowed") + value := videoDevice(1, "tenant-a", "site") + value.AreaID = "private-area" + if err := store.CreateDevice(ctx, value); !errors.Is(err, ErrAreaPolicyUnavailable) { + t.Fatalf("cross-tenant Area was not hidden as unavailable: %v", err) + } +} + +func TestPostgresEnableRechecksAreaWithoutStoppingExistingDevice(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + insertBellSite(t, admin, "tenant", "site", 2) + disabled := videoDevice(1, "tenant", "site") + disabled.DesiredState = device.DesiredDisabled + if err := store.CreateDevice(ctx, disabled); err != nil { + t.Fatal(err) + } + if err := store.CreateDevice(ctx, videoDevice(2, "tenant", "site")); err != nil { + t.Fatal(err) + } + if _, err := admin.ExecContext(ctx, `UPDATE bell.areas SET capture_policy = 'non_imaging_only' + WHERE tenant_id = 'tenant' AND id = 'area-default'`); err != nil { + t.Fatal(err) + } + if err := store.SetDesiredState(ctx, disabled.ID, device.DesiredEnabled); !errors.Is(err, ErrAreaPolicyDenied) { + t.Fatalf("enable under non-imaging policy must be denied, got %v", err) + } + stillDisabled, err := store.GetDevice(ctx, disabled.ID) + if err != nil { + t.Fatal(err) + } + stillEnabled, err := store.GetDevice(ctx, "camera-002") + if err != nil { + t.Fatal(err) + } + if stillDisabled.DesiredState != device.DesiredDisabled || + stillEnabled.DesiredState != device.DesiredEnabled { + t.Fatalf("policy change altered existing state: disabled=%s enabled=%s", + stillDisabled.DesiredState, stillEnabled.DesiredState) + } +} + +func TestPostgresAuditContextRedactionAndNoopDesiredState(t *testing.T) { + store, admin := openPostgresTestStore(t) + insertBellSite(t, admin, "tenant", "site", 2) + ctx := WithAuditContext(context.Background(), AuditContext{ + ActorType: AuditActorUser, ActorID: "operator-7", Reason: "approved change", TraceID: "trace-7", + }) + value := videoDevice(1, "tenant", "site") + if err := store.CreateDevice(ctx, value); err != nil { + t.Fatal(err) + } + if err := store.SetDesiredState(ctx, value.ID, device.DesiredEnabled); err != nil { + t.Fatal(err) + } + stored, err := store.GetDevice(ctx, value.ID) + if err != nil { + t.Fatal(err) + } + if stored.Generation != 1 { + t.Fatalf("no-op desired-state request changed generation to %d", stored.Generation) + } + rows, err := admin.QueryContext(ctx, `SELECT event_id, event_type, actor_type, actor_id, + COALESCE(reason, ''), COALESCE(trace_id, ''), payload::text + FROM sense.device_operation_outbox ORDER BY occurred_at, event_id`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var count int + for rows.Next() { + var eventID, eventType, actorType, actorID, reason, traceID, payload string + if err := rows.Scan(&eventID, &eventType, &actorType, &actorID, &reason, &traceID, &payload); err != nil { + t.Fatal(err) + } + count++ + if !strings.HasPrefix(eventID, "audit_") || len(eventID) != 38 { + t.Fatalf("invalid audit event ID %q", eventID) + } + if actorType != "user" || actorID != "operator-7" || reason != "approved change" || traceID != "trace-7" { + t.Fatalf("audit principal/context drift: %s/%s %s %s", actorType, actorID, reason, traceID) + } + for _, secret := range []string{value.EndpointRef, value.CredentialRef, value.PathName} { + if strings.Contains(payload, secret) { + t.Fatalf("audit payload leaked sensitive runtime data for %s", eventType) + } + } + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + if count != 2 { + t.Fatalf("expected create and no-op audit facts, got %d", count) + } +} + +func TestPostgresOutboxFailureRollsBackAdmission(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + insertBellSite(t, admin, "tenant", "site", 2) + if _, err := admin.ExecContext(ctx, `CREATE FUNCTION sense.t010_reject_outbox() + RETURNS trigger LANGUAGE plpgsql AS $function$ + BEGIN RAISE EXCEPTION 'synthetic outbox failure'; END + $function$; + CREATE TRIGGER t010_reject_outbox BEFORE INSERT ON sense.device_operation_outbox + FOR EACH ROW EXECUTE FUNCTION sense.t010_reject_outbox()`); err != nil { + t.Fatal(err) + } + defer func() { + _, _ = admin.ExecContext(context.Background(), + `DROP TRIGGER IF EXISTS t010_reject_outbox ON sense.device_operation_outbox; + DROP FUNCTION IF EXISTS sense.t010_reject_outbox()`) + }() + if err := store.CreateDevice(ctx, videoDevice(1, "tenant", "site")); err == nil { + t.Fatal("synthetic Outbox failure did not reject device creation") + } + var devices, projections, audits int + if err := admin.QueryRowContext(ctx, `SELECT count(*) FROM sense.devices`).Scan(&devices); err != nil { + t.Fatal(err) + } + if err := admin.QueryRowContext(ctx, `SELECT count(*) FROM sense.area_policy_projection_state`).Scan(&projections); err != nil { + t.Fatal(err) + } + if err := admin.QueryRowContext(ctx, `SELECT count(*) FROM sense.device_operation_outbox`).Scan(&audits); err != nil { + t.Fatal(err) + } + if devices != 0 || projections != 0 || audits != 0 { + t.Fatalf("Outbox failure left partial transaction: devices=%d projections=%d audits=%d", + devices, projections, audits) + } +} + +func TestPostgresOutboxFailureRollsBackDesiredState(t *testing.T) { + store, admin := openPostgresTestStore(t) + ctx := context.Background() + insertBellSite(t, admin, "tenant", "site", 2) + value := videoDevice(1, "tenant", "site") + value.DesiredState = device.DesiredDisabled + if err := store.CreateDevice(ctx, value); err != nil { + t.Fatal(err) + } + if _, err := admin.ExecContext(ctx, `CREATE FUNCTION sense.t010_reject_state_audit() + RETURNS trigger LANGUAGE plpgsql AS $function$ + BEGIN RAISE EXCEPTION 'synthetic state-audit failure'; END + $function$; + CREATE TRIGGER t010_reject_state_audit BEFORE INSERT ON sense.device_operation_outbox + FOR EACH ROW EXECUTE FUNCTION sense.t010_reject_state_audit()`); err != nil { + t.Fatal(err) + } + defer func() { + _, _ = admin.ExecContext(context.Background(), + `DROP TRIGGER IF EXISTS t010_reject_state_audit ON sense.device_operation_outbox; + DROP FUNCTION IF EXISTS sense.t010_reject_state_audit()`) + }() + if err := store.SetDesiredState(ctx, value.ID, device.DesiredEnabled); err == nil { + t.Fatal("synthetic Outbox failure did not reject desired-state change") + } + stored, err := store.GetDevice(ctx, value.ID) + if err != nil { + t.Fatal(err) + } + if stored.DesiredState != device.DesiredDisabled || stored.Generation != 1 { + t.Fatalf("Outbox failure committed desired state: %+v", stored) + } + var audits int + if err := admin.QueryRowContext(ctx, `SELECT count(*) FROM sense.device_operation_outbox`).Scan(&audits); err != nil { + t.Fatal(err) + } + if audits != 1 { + t.Fatalf("failed desired-state transaction changed Outbox count to %d", audits) + } +} + +func TestPostgresOpenRejectsAreaSourcePrivilege(t *testing.T) { + _, admin := openPostgresTestStore(t) + ctx := context.Background() + if _, err := admin.ExecContext(ctx, + `GRANT SELECT ON bell.areas TO yovision_t010_sense`); err != nil { + t.Fatal(err) + } + defer func() { + _, _ = admin.ExecContext(context.Background(), + `REVOKE SELECT ON bell.areas FROM yovision_t010_sense`) + }() + value, err := OpenPostgres(ctx, os.Getenv(postgresTestDSNEnv)) + if value != nil { + _ = value.Close() + t.Fatal("Area-source privilege was accepted") + } + if err == nil || !strings.Contains(err.Error(), "privilege boundary") { + t.Fatalf("expected Area privilege-boundary error, got %v", err) + } +} + func openPostgresTestStore(t *testing.T) (*Postgres, *sql.DB) { t.Helper() dsn := os.Getenv(postgresTestDSNEnv) @@ -318,10 +629,13 @@ func openPostgresTestStore(t *testing.T) (*Postgres, *sql.DB) { t.Fatal("connect PostgreSQL test administrator") } if _, err := admin.ExecContext(context.Background(), `TRUNCATE + sense.device_operation_outbox, sense.device_capabilities, sense.reconcile_state, sense.devices, sense.site_quota_projection_state, + sense.area_policy_projection_state, + bell.areas, bell.sites CASCADE`); err != nil { admin.Close() t.Fatal(err) @@ -350,4 +664,18 @@ func insertBellSite(t *testing.T, admin *sql.DB, tenantID, siteID string, quota if _, err := admin.ExecContext(context.Background(), query, arguments...); err != nil { t.Fatal(err) } + insertBellArea(t, admin, tenantID, siteID, "area-default", "video_allowed") +} + +func insertBellArea( + t *testing.T, + admin *sql.DB, + tenantID, siteID, areaID, capturePolicy string, +) { + t.Helper() + if _, err := admin.ExecContext(context.Background(), `INSERT INTO bell.areas( + tenant_id, site_id, id, name, capture_policy + ) VALUES ($1, $2, $3, $4, $5)`, tenantID, siteID, areaID, "Test Area", capturePolicy); err != nil { + t.Fatal(err) + } } diff --git a/Sense/internal/store/repository.go b/Sense/internal/store/repository.go index 3d3347e..50d3867 100644 --- a/Sense/internal/store/repository.go +++ b/Sense/internal/store/repository.go @@ -18,6 +18,9 @@ const ( var ( ErrQuotaProjectionUnavailable = errors.New("quota projection unavailable") ErrQuotaProjectionInvalid = errors.New("quota projection invalid") + ErrAreaPolicyUnavailable = errors.New("area policy unavailable") + ErrAreaPolicyInvalid = errors.New("area policy invalid") + ErrAreaPolicyDenied = errors.New("area policy denies imaging device") ) // Repository is the storage boundary used by the Sense process. SQLite stays @@ -61,3 +64,22 @@ func projectionUnavailable() error { func projectionInvalid() error { return "aProjectionError{kind: ErrQuotaProjectionInvalid} } + +type areaPolicyError struct { + kind error +} + +func (e *areaPolicyError) Error() string { return e.kind.Error() } +func (e *areaPolicyError) Unwrap() error { return e.kind } + +func areaPolicyUnavailable() error { + return &areaPolicyError{kind: ErrAreaPolicyUnavailable} +} + +func areaPolicyInvalid() error { + return &areaPolicyError{kind: ErrAreaPolicyInvalid} +} + +func areaPolicyDenied() error { + return &areaPolicyError{kind: ErrAreaPolicyDenied} +} diff --git a/Sense/internal/store/sqlite_test.go b/Sense/internal/store/sqlite_test.go index a8df3a9..30e6d03 100644 --- a/Sense/internal/store/sqlite_test.go +++ b/Sense/internal/store/sqlite_test.go @@ -195,7 +195,8 @@ func openTestStore(t *testing.T) *SQLite { func videoDevice(index int, tenantID, siteID string) device.Device { id := fmt.Sprintf("camera-%03d", index) return device.Device{ - ID: id, TenantID: tenantID, SiteID: siteID, SerialNumber: id, Name: id, + ID: id, TenantID: tenantID, SiteID: siteID, AreaID: "area-default", + SerialNumber: id, Name: id, Modality: device.ModalityVideo, Capabilities: []device.Capability{device.CapabilityVideoCapture, device.CapabilitySpatialRule}, DesiredState: device.DesiredEnabled, ActualState: device.ActualPending, diff --git a/deploy/postgres/005_area_policy.sql b/deploy/postgres/005_area_policy.sql new file mode 100644 index 0000000..b96149f --- /dev/null +++ b/deploy/postgres/005_area_policy.sql @@ -0,0 +1,127 @@ +-- Bell owns Area and capture-policy truth. Sense records only the highest +-- projection version it has observed and the version used for admission. + +CREATE TABLE IF NOT EXISTS bell.areas ( + tenant_id text NOT NULL, + site_id text NOT NULL, + id text NOT NULL, + name text NOT NULL, + capture_policy text NOT NULL DEFAULT 'video_allowed', + version bigint NOT NULL DEFAULT 1, + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + updated_at timestamptz NOT NULL DEFAULT clock_timestamp(), + deleted_at timestamptz, + PRIMARY KEY (tenant_id, id), + CONSTRAINT bell_areas_site_fk FOREIGN KEY (tenant_id, site_id) + REFERENCES bell.sites(tenant_id, id), + CONSTRAINT bell_areas_capture_policy CHECK ( + capture_policy IN ('video_allowed', 'non_imaging_only') + ), + CONSTRAINT bell_areas_version_positive CHECK (version >= 1), + CONSTRAINT bell_areas_identity_not_blank CHECK ( + btrim(tenant_id) <> '' AND btrim(site_id) <> '' + AND btrim(id) <> '' AND btrim(name) <> '' + ) +); +ALTER TABLE bell.areas OWNER TO bell_app; + +CREATE OR REPLACE FUNCTION bell.bump_area_version() +RETURNS trigger +LANGUAGE plpgsql +SECURITY INVOKER +SET search_path = pg_catalog, bell +AS $function$ +BEGIN + IF NEW.tenant_id IS DISTINCT FROM OLD.tenant_id + OR NEW.site_id IS DISTINCT FROM OLD.site_id + OR NEW.id IS DISTINCT FROM OLD.id THEN + RAISE EXCEPTION 'Bell Area identity and Site are immutable'; + END IF; + NEW.version := OLD.version + 1; + NEW.updated_at := clock_timestamp(); + RETURN NEW; +END +$function$; +ALTER FUNCTION bell.bump_area_version() OWNER TO bell_app; + +DROP TRIGGER IF EXISTS bell_areas_bump_version ON bell.areas; +CREATE TRIGGER bell_areas_bump_version +BEFORE UPDATE ON bell.areas +FOR EACH ROW EXECUTE FUNCTION bell.bump_area_version(); + +CREATE OR REPLACE VIEW bell.area_policy_v1 ( + tenant_id, + site_id, + area_id, + capture_policy, + source_version, + source_updated_at +) AS +SELECT + area.tenant_id, + area.site_id, + area.id, + area.capture_policy, + area.version, + area.updated_at +FROM bell.areas AS area +JOIN bell.sites AS site + ON site.tenant_id = area.tenant_id AND site.id = area.site_id +WHERE area.deleted_at IS NULL AND site.deleted_at IS NULL; +ALTER VIEW bell.area_policy_v1 OWNER TO bell_app; +COMMENT ON VIEW bell.area_policy_v1 IS + 'v1 read-only Area capture-policy projection owned by Bell and consumed by Sense'; + +ALTER TABLE sense.devices ADD COLUMN IF NOT EXISTS area_id text; +ALTER TABLE sense.devices ADD COLUMN IF NOT EXISTS area_policy_source_version bigint; + +DO $constraints$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conrelid = 'sense.devices'::regclass + AND conname = 'sense_devices_area_not_blank' + ) THEN + ALTER TABLE sense.devices ADD CONSTRAINT sense_devices_area_not_blank + CHECK (area_id IS NULL OR btrim(area_id) <> ''); + END IF; + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conrelid = 'sense.devices'::regclass + AND conname = 'sense_devices_area_version_positive' + ) THEN + ALTER TABLE sense.devices ADD CONSTRAINT sense_devices_area_version_positive + CHECK (area_policy_source_version IS NULL OR area_policy_source_version >= 1); + END IF; + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conrelid = 'sense.devices'::regclass + AND conname = 'sense_devices_tenant_site_id_unique' + ) THEN + ALTER TABLE sense.devices ADD CONSTRAINT sense_devices_tenant_site_id_unique + UNIQUE (tenant_id, site_id, id); + END IF; +END +$constraints$; + +CREATE TABLE IF NOT EXISTS sense.area_policy_projection_state ( + tenant_id text NOT NULL, + site_id text NOT NULL, + area_id text NOT NULL, + source_version bigint NOT NULL, + synced_at timestamptz NOT NULL, + PRIMARY KEY (tenant_id, site_id, area_id), + CONSTRAINT sense_area_projection_identity_not_blank CHECK ( + btrim(tenant_id) <> '' AND btrim(site_id) <> '' AND btrim(area_id) <> '' + ), + CONSTRAINT sense_area_projection_version_positive CHECK (source_version >= 1) +); +ALTER TABLE sense.area_policy_projection_state OWNER TO sense_app; + +CREATE INDEX IF NOT EXISTS sense_devices_area_idx + ON sense.devices(tenant_id, site_id, area_id); + +INSERT INTO bell.schema_migrations(version) VALUES (2) +ON CONFLICT (version) DO NOTHING; +INSERT INTO sense.schema_migrations(version) VALUES (2) +ON CONFLICT (version) DO NOTHING; diff --git a/deploy/postgres/006_device_operation_outbox.sql b/deploy/postgres/006_device_operation_outbox.sql new file mode 100644 index 0000000..6609d1f --- /dev/null +++ b/deploy/postgres/006_device_operation_outbox.sql @@ -0,0 +1,67 @@ +-- Local durable audit facts. Transport, signatures, acknowledgements and +-- retention belong to a later Bell relay contract. + +CREATE TABLE IF NOT EXISTS sense.device_operation_outbox ( + event_id text PRIMARY KEY, + event_type text NOT NULL, + tenant_id text NOT NULL, + site_id text NOT NULL, + device_id text NOT NULL, + actor_type text NOT NULL, + actor_id text NOT NULL, + reason text, + trace_id text, + aggregate_generation bigint NOT NULL, + quota_source_version bigint, + area_policy_source_version bigint, + payload jsonb NOT NULL, + occurred_at timestamptz NOT NULL, + available_at timestamptz NOT NULL, + attempt_count integer NOT NULL DEFAULT 0, + next_attempt_at timestamptz, + delivered_at timestamptz, + CONSTRAINT sense_outbox_device_fk FOREIGN KEY (tenant_id, site_id, device_id) + REFERENCES sense.devices(tenant_id, site_id, id), + CONSTRAINT sense_outbox_event_id_format CHECK (event_id ~ '^audit_[0-9a-f]{32}$'), + CONSTRAINT sense_outbox_event_type CHECK ( + event_type IN ('device.created', 'device.desired_state.accepted') + ), + CONSTRAINT sense_outbox_identity_not_blank CHECK ( + btrim(tenant_id) <> '' AND btrim(site_id) <> '' + AND btrim(device_id) <> '' AND btrim(actor_id) <> '' + ), + CONSTRAINT sense_outbox_actor_type CHECK ( + actor_type IN ('user', 'service', 'system') + ), + CONSTRAINT sense_outbox_generation_positive CHECK (aggregate_generation >= 1), + CONSTRAINT sense_outbox_quota_version_positive CHECK ( + quota_source_version IS NULL OR quota_source_version >= 1 + ), + CONSTRAINT sense_outbox_area_version_positive CHECK ( + area_policy_source_version IS NULL OR area_policy_source_version >= 1 + ), + CONSTRAINT sense_outbox_attempt_nonnegative CHECK (attempt_count >= 0), + CONSTRAINT sense_outbox_reason_length CHECK (reason IS NULL OR char_length(reason) <= 500), + CONSTRAINT sense_outbox_trace_length CHECK (trace_id IS NULL OR char_length(trace_id) <= 128), + CONSTRAINT sense_outbox_payload_object CHECK (jsonb_typeof(payload) = 'object'), + CONSTRAINT sense_outbox_payload_kind CHECK ( + (event_type = 'device.created' AND payload ->> 'kind' = 'device_created') + OR (event_type = 'device.desired_state.accepted' + AND payload ->> 'kind' = 'desired_state_accepted') + ), + CONSTRAINT sense_outbox_payload_redacted CHECK ( + NOT (payload ?| ARRAY[ + 'endpoint_ref', 'credential_ref', 'profile_token', 'path_name', + 'password', 'stream_uri', 'mediamtx_config' + ]) + ) +); +ALTER TABLE sense.device_operation_outbox OWNER TO sense_app; + +CREATE INDEX IF NOT EXISTS sense_outbox_delivery_idx + ON sense.device_operation_outbox(delivered_at, next_attempt_at, available_at, event_id); +CREATE INDEX IF NOT EXISTS sense_outbox_device_idx + ON sense.device_operation_outbox(tenant_id, site_id, device_id, occurred_at, event_id); + +INSERT INTO sense.schema_migrations(version) VALUES (3) +ON CONFLICT (version) DO NOTHING; diff --git a/deploy/postgres/007_privileges_area_audit.sql b/deploy/postgres/007_privileges_area_audit.sql new file mode 100644 index 0000000..654e72b --- /dev/null +++ b/deploy/postgres/007_privileges_area_audit.sql @@ -0,0 +1,18 @@ +-- Extend the T-009 least-privilege boundary to the Area projection and new +-- Sense-owned state. Reapplying this file is intentional and safe. + +REVOKE ALL ON TABLE bell.area_policy_v1 FROM PUBLIC; +REVOKE ALL ON TABLE bell.areas FROM PUBLIC; +REVOKE ALL ON FUNCTION bell.bump_area_version() FROM PUBLIC; + +GRANT USAGE ON SCHEMA bell TO sense_app; +GRANT SELECT ON TABLE bell.area_policy_v1 TO sense_app; +REVOKE ALL ON TABLE bell.areas FROM sense_app; +REVOKE ALL ON FUNCTION bell.bump_area_version() FROM sense_app; + +REVOKE ALL ON TABLE sense.area_policy_projection_state FROM PUBLIC; +REVOKE ALL ON TABLE sense.device_operation_outbox FROM PUBLIC; +GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE + sense.area_policy_projection_state, + sense.device_operation_outbox +TO sense_app; diff --git a/deploy/postgres/README.md b/deploy/postgres/README.md index 4cfca23..a678c4e 100644 --- a/deploy/postgres/README.md +++ b/deploy/postgres/README.md @@ -1,6 +1,6 @@ # YoVision PostgreSQL 初始化 -本目录实现 T-009 的 PostgreSQL `17.10` 初始 schema。SQL 按文件名前缀顺序执行：先创建 NOLOGIN 权限角色，再创建 Bell、Sense 对象，最后收窄跨 schema 权限。应用登录角色及其密码由部署环境或密钥系统创建，并分别加入 `bell_app`、`sense_app`；仓库不保存登录凭据。 +本目录实现 T-009/T-010 的 PostgreSQL `17.10` schema。SQL 必须按文件名前缀顺序执行：`001`～`004` 创建 NOLOGIN 权限角色、Bell/Sense 初始对象和配额权限；`005`～`007` 增量增加 Area 策略投影、设备操作审计 Outbox 和对应最小权限。全部 SQL 可重放。应用登录角色及其密码由部署环境或密钥系统创建，并分别加入 `bell_app`、`sense_app`；仓库不保存登录凭据。 生产/共享实例必须由管理员先备份并在 YoVision 专用数据库中执行。Sense 进程不会用高权限自动建库或建角色。示例只使用私有环境变量，不把实际 DSN 写入脚本或日志： @@ -15,11 +15,12 @@ Get-ChildItem deploy/postgres/[0-9][0-9][0-9]_*.sql | ## 权限边界 -- `bell_app` 拥有 `bell.sites`、版本 trigger 和 `bell.site_quota_v1`。 -- `sense_app` 拥有 `sense` schema，只获得 `bell` schema 的 `USAGE` 和配额视图的 `SELECT`。 -- `sense_app` 对 `bell.sites`、Bell migration 表和 trigger function 没有权限。 +- `bell_app` 拥有 `bell.sites`/`bell.areas`、版本 trigger、`bell.site_quota_v1` 和 `bell.area_policy_v1`。 +- `sense_app` 拥有 `sense` schema，只获得 `bell` schema 的 `USAGE` 和两个投影视图的 `SELECT`。 +- `sense_app` 对 Bell 源表、Bell migration 表和 trigger function 没有权限；启动检查发现权限过宽时拒绝运行。 +- `sense.device_operation_outbox` 是本地持久化审计事实，不是 Bell 全局审计真相；relay 的传输、签名、确认和留存尚未实现。 - PUBLIC 对两个业务 schema 的表和函数没有权限。 ## 恢复 -T-009 不迁移或删除现有 SQLite 数据，默认运行配置也仍为 SQLite。若 PostgreSQL 切换验证失败，停止 PostgreSQL 模式的 Sense 进程并恢复 `SENSE_DB_DRIVER=sqlite` 与原 SQLite DSN。初始 schema 不提供自动 down migration；需要移除时由管理员先备份，再只删除明确确认的 YoVision 专用数据库，禁止对共享实例做模糊或递归清理。 +T-009/T-010 不迁移或删除现有 SQLite 数据，默认运行配置也仍为 SQLite。SQLite 不具备生产 Area/Outbox 一致性，不能作为后续公共控制 API 的生产后端。若 PostgreSQL 切换验证失败，停止 PostgreSQL 模式的 Sense 进程并恢复原实验室 SQLite DSN；不要把此回退误报为生产语义等价。schema 不提供自动 down migration；需要移除时由管理员先备份，再只删除明确确认的 YoVision 专用数据库，禁止对共享实例做模糊或递归清理。 diff --git a/deploy/postgres/tests/assertions.sql b/deploy/postgres/tests/assertions.sql index 59d59cf..8079490 100644 --- a/deploy/postgres/tests/assertions.sql +++ b/deploy/postgres/tests/assertions.sql @@ -6,10 +6,12 @@ DECLARE default_quota integer; initial_version bigint; updated_version bigint; + area_initial_version bigint; + area_updated_version bigint; BEGIN IF current_setting('server_version_num')::integer < 170000 OR current_setting('server_version_num')::integer >= 180000 THEN - RAISE EXCEPTION 'T-009 requires PostgreSQL 17.x'; + RAISE EXCEPTION 'YoVision requires PostgreSQL 17.x'; END IF; SELECT array_agg(column_name ORDER BY ordinal_position) @@ -37,6 +39,43 @@ BEGIN RAISE EXCEPTION 'sense_app has forbidden Bell source-table privileges'; END IF; + SELECT array_agg(column_name ORDER BY ordinal_position) + INTO actual_columns + FROM information_schema.columns + WHERE table_schema = 'bell' AND table_name = 'area_policy_v1'; + IF actual_columns IS DISTINCT FROM ARRAY[ + 'tenant_id', 'site_id', 'area_id', 'capture_policy', + 'source_version', 'source_updated_at' + ] THEN + RAISE EXCEPTION 'area_policy_v1 signature drift: %', actual_columns; + END IF; + + IF NOT has_table_privilege('sense_app', 'bell.area_policy_v1', 'SELECT') THEN + RAISE EXCEPTION 'sense_app lacks SELECT on Area policy view'; + END IF; + IF has_table_privilege('sense_app', 'bell.area_policy_v1', 'INSERT') + OR has_table_privilege('sense_app', 'bell.area_policy_v1', 'UPDATE') + OR has_table_privilege('sense_app', 'bell.area_policy_v1', 'DELETE') + OR has_table_privilege('sense_app', 'bell.areas', 'SELECT') + OR has_table_privilege('sense_app', 'bell.areas', 'INSERT') + OR has_table_privilege('sense_app', 'bell.areas', 'UPDATE') + OR has_table_privilege('sense_app', 'bell.areas', 'DELETE') THEN + RAISE EXCEPTION 'sense_app violates the Area projection boundary'; + END IF; + IF has_function_privilege('sense_app', 'bell.bump_area_version()', 'EXECUTE') THEN + RAISE EXCEPTION 'sense_app can execute the Bell Area version trigger function'; + END IF; + IF NOT has_table_privilege('sense_app', 'sense.device_operation_outbox', 'SELECT') + OR NOT has_table_privilege('sense_app', 'sense.device_operation_outbox', 'INSERT') + OR NOT has_table_privilege('sense_app', 'sense.device_operation_outbox', 'UPDATE') + OR NOT has_table_privilege('sense_app', 'sense.device_operation_outbox', 'DELETE') THEN + RAISE EXCEPTION 'sense_app lacks access to its local audit Outbox'; + END IF; + IF (SELECT max(version) FROM bell.schema_migrations) <> 2 + OR (SELECT max(version) FROM sense.schema_migrations) <> 3 THEN + RAISE EXCEPTION 'schema migration version drift'; + END IF; + INSERT INTO bell.sites(tenant_id, id, name) VALUES ('assertion-tenant', 'assertion-site', 'Assertion Site') RETURNING max_video_channels, version INTO default_quota, initial_version; @@ -51,6 +90,28 @@ BEGIN RAISE EXCEPTION 'Bell site source_version is not monotonic'; END IF; + INSERT INTO bell.areas(tenant_id, site_id, id, name, capture_policy) + VALUES ( + 'assertion-tenant', 'assertion-site', 'assertion-area', + 'Assertion Area', 'video_allowed' + ) + RETURNING version INTO area_initial_version; + + UPDATE bell.areas SET capture_policy = 'non_imaging_only' + WHERE tenant_id = 'assertion-tenant' AND id = 'assertion-area' + RETURNING version INTO area_updated_version; + IF area_initial_version <> 1 OR area_updated_version <> area_initial_version + 1 THEN + RAISE EXCEPTION 'Bell Area source_version is not monotonic'; + END IF; + + BEGIN + INSERT INTO bell.areas(tenant_id, site_id, id, name, capture_policy) + VALUES ('assertion-tenant', 'assertion-site', 'invalid-area', 'Invalid', 'unknown'); + RAISE EXCEPTION 'unknown capture_policy unexpectedly accepted'; + EXCEPTION + WHEN check_violation THEN NULL; + END; + BEGIN INSERT INTO bell.sites(tenant_id, id, name, max_video_channels) VALUES ('assertion-tenant', 'invalid-site', 'Invalid', 129); @@ -59,6 +120,8 @@ BEGIN WHEN check_violation THEN NULL; END; + DELETE FROM bell.areas + WHERE tenant_id = 'assertion-tenant' AND id = 'assertion-area'; DELETE FROM bell.sites WHERE tenant_id = 'assertion-tenant' AND id = 'assertion-site'; END diff --git a/docs/04-architecture.md b/docs/04-architecture.md index 74be670..c0a13c1 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -45,7 +45,7 @@ Sense ── 视频流/触发信号 ──> Brain 主流程： -1. Bell 持有站点、Area、配额与 `capture_policy`；首期在同一 PostgreSQL 实例内发布版本化只读视图 `bell.site_quota_v1`。T-009 已实现 Bell 源表/视图、最小权限和 Sense PostgreSQL repository；Sense 在新增/启用设备的同一事务内以站点 advisory lock 校验配额并记录所用 `source_version`。未来分库必须发布新版本契约，不能静默改变 v1 语义。 +1. Bell 持有站点、Area、配额与 `capture_policy`；首期在同一 PostgreSQL 实例内发布 `bell.site_quota_v1` 和 `bell.area_policy_v1` 两个版本化只读视图。T-009/T-010 已实现 Bell 源表/视图、最小权限和 Sense PostgreSQL repository；Sense 按 Area→Site 的固定 advisory-lock 顺序执行策略与配额准入并记录所用版本。未来分库必须发布新版本契约，不能静默改变 v1 语义。 2. Sense 维护设备期望态，通过 MediaMTX API 和对账器收敛实际态。 3. Brain 消费视频与触发信号，产生符合 v0.1 的事件。 4. Bell 做 schema 与代码级断言，生成平台 ULID，保存不可变事件。 @@ -63,12 +63,12 @@ Sense ── 视频流/触发信号 ──> Brain 4. 平台事件 ULID 由 Bell 生成；Brain 只填 `source_event_id`。 5. 一个 PostgreSQL 实例，`sense`/`bell` schema 分离；Brain 无业务 schema。 6. 16/128 都不是单机保证；媒体与推理按独立分片横向扩展。 -7. Bell 拥有 `site.max_video_channels` 并拥有 `bell.site_quota_v1`；Sense 角色只获得视图 `SELECT`，在设备写路径执行准入，不能写 Bell schema 或读取 Bell 源表。 +7. Bell 拥有 Site/Area、`site.max_video_channels` 与 `capture_policy`，并拥有两个 v1 投影视图；Sense 角色只获得视图 `SELECT`，在设备写路径执行准入，不能写 Bell schema 或读取 Bell 源表。 8. 事件片段写入客户侧 MinIO/S3，常态录像留在客户 NVR；元数据/审计、人脸和训练样本使用独立生命周期。 9. 投递状态机只依赖 Bell provider 接口，不直接依赖某家短信或语音 SDK；生产前至少两条独立路径并能故障切换。 10. 设备领域模型使用 `modality + capabilities`，页面不以摄像头作为唯一根实体；未实现协议适配器明确为 `adapter_not_ready`，不得用模拟遥测伪装交付。 11. Tenant/Site/Area/RBAC、配额、`capture_policy` 与全局审计属于 Bell；Sense Control API v1 只管理 Device 期望态与收敛查询，Sense 只读消费版本化投影并在设备写路径执行，投影不可用时只阻断相关新变更，不静默切断已有链路。 -12. Sense 的设备操作审计先写本地持久化 outbox，再由幂等 relay 异步送入 Bell 全局审计；不得使用“先执行高风险操作、再尽力入队”的顺序。具体字段、签名、重放与留存契约必须由独立 API/契约任务冻结后实现。 +12. Sense 的设备操作审计先写本地持久化 Outbox，再由幂等 relay 异步送入 Bell 全局审计；不得使用“先执行高风险操作、再尽力入队”的顺序。T-010 已冻结脱敏本地事件并实现原子写入；relay 的 transport、签名、确认、重放窗口与留存仍须独立冻结。 ## 6. 容量架构 @@ -85,8 +85,8 @@ Sense ── 视频流/触发信号 ──> Brain - 对账器水平触发、幂等、指数退避、限制并发；部分失败不做跨系统回滚，只持续收敛。 - 孤儿删除必须有 10% 安全闸和人工可观察指标。 - `bell.site_quota_v1` 行缺失、数值越界、版本回退或读取失败只阻止视频设备新增/启用，不中断已有流；降低配额导致超限时不自动停用，后续准入返回稳定错误并产生运维信号。多 Sense 实例使用 PostgreSQL transaction-scoped advisory lock 串行化同 tenant/site 的计数与写入，不能用进程内锁替代。 -- Area 策略投影读取失败只阻止相关设备新增/启用；已有设备保持原状态并产生运维告警。策略变更与已有成像设备冲突时由 Bell 管理端显式处置。 -- 高风险设备操作在本地事务内同时写期望态与审计 outbox；异步 relay 可重试、幂等投递到 Bell。T-004 只验证交互归属，不定义或实现接口契约。 +- `bell.area_policy_v1` 缺失、非法、版本回退或读取失败时，PostgreSQL repository 拒绝相关新增/启用；`non_imaging_only` 允许非成像设备但拒绝具有 `video_capture` 的设备。已有设备保持原状态，策略冲突由 Bell 管理端显式迁移或取消。同库实时视图不以源记录年龄误判 freshness。 +- 设备创建和期望态受理在本地事务内同时写脱敏 `sense.device_operation_outbox`；Outbox 失败回滚业务写入，相同期望态不增加 generation 但仍审计。异步 relay 尚未实现。 - Brain 投递失败落本地队列重试，不阻塞实时推理主链路。 - Alert 先落库再投递，进程重启恢复未完成升级链。 - 值班排班发布前必须按 Site 时区校验班次空档、重叠、联系人停用和通道验证；排班以新版本和未来生效时间发布，不原地改写历史。交接班是进行中 Alert 的显式责任转移事件，不替代排班版本变更。 @@ -94,8 +94,8 @@ Sense ── 视频流/触发信号 ──> Brain ## 8. 数据与契约 -- Bell 核心实体：Tenant → Site → Area（含 `capture_policy`）以及 Role/Binding/Quota/Audit；Sense 核心实体：Device（含 `modality + capabilities`）→ StreamBinding/Zone，以及带 `source_version`/`synced_at` 的 SiteQuota/AreaPolicyProjection。两个 schema 以稳定逻辑 ID 关联，不跨 schema 写入；配额 v1 的跨 schema 列签名限定为 `tenant_id/site_id/max_video_channels/source_version/source_updated_at`。 -- Sense Control API v1 使用站点作用域路径、认证上下文 tenant、cursor 分页、幂等键与 ETag；敏感连接引用只写不读。正式签名和兼容规则以 [`contracts/`](contracts/) 为准。T-009 只实现 PostgreSQL migration/repository，不等于 handler、认证或 Bell 管理服务已实现。 +- Bell 核心实体：Tenant → Site → Area（含 `capture_policy`）以及 Role/Binding/Quota/Audit；Sense 核心实体：Device（含 `modality + capabilities`）→ StreamBinding/Zone，以及只记录已观察版本的 SiteQuota/AreaPolicyProjection。两个 schema 以稳定逻辑 ID 关联，不跨 schema 写入；配额 v1 为五列，Area v1 固定为 `tenant_id/site_id/area_id/capture_policy/source_version/source_updated_at` 六列。 +- Sense Control API v1 使用站点作用域路径、认证上下文 tenant、cursor 分页、幂等键与 ETag；敏感连接引用只写不读。正式签名和兼容规则以 [`contracts/`](contracts/) 为准。T-009/T-010 已实现 PostgreSQL 准入与审计事务基础，不等于 handler、认证、Bell 管理服务或 Outbox relay 已实现。 - 业务实体：Rule → Event → Alert → DeliveryAttempt/Ack；Event 与 Alert 不合并。 - Bell 通知域分为三个聚合：Contact/Team 保存身份、成员关系和已验证通道；OnCallSchedule/ScheduleVersion/ShiftException 保存时区、轮换与例外；EscalationPolicy/Step 通过 `person / team / on_call_schedule` 类型化 `target_ref` 引用目标。三者共享逻辑 ID，不复制手机号、班次或轮换字段。 - 每个 DeliveryAttempt 创建时解析当时生效的排班版本，并保存实际收件人、通道、`schedule_version` 和解析时间快照；之后联系人或排班修改不得回写既有投递事实。 @@ -111,10 +111,10 @@ Sense/cmd + Sense/internal/{device,onvif,mtx,reconcile,probe,trigger,tunnel,auth Brain/{pipeline,models,judge,emit,trigger,contracts} Bell/cmd + Bell/internal/{ingest,event,rule,alert,deliver,feedback,tenant,audit,store} Bell/{web,packs,contracts} -deploy/postgres/{001_roles.sql,002_bell.sql,003_sense.sql,004_privileges.sql,tests} +deploy/postgres/{001_roles.sql,...,007_privileges_area_audit.sql,tests} ``` -当前只有空目录占位；真实脚手架必须由对应任务创建。 +Sense 脚手架和 PostgreSQL `001`～`007` 已实现；Brain/Bell 应用目录仍是占位，真实服务必须由对应任务创建。 ## 10. 开发顺序 diff --git a/docs/06-tasks.md b/docs/06-tasks.md index c6da134..bf63f34 100644 --- a/docs/06-tasks.md +++ b/docs/06-tasks.md @@ -24,7 +24,8 @@ - T-008：冻结 Sense Control API v1 与 Bell `site_quota_v1` 只读投影契约；只建立接口和测试门禁，不等于 handler、Bell 表或 migration 已实现。 - T-009：使用 PostgreSQL 17.10 建立 `sense`/`bell` schema、Bell 配额源表/只读视图、最小权限和 Sense PostgreSQL repository；SQLite 保留为 M1 本地路径。 -- 后续任务按 T-008 契约实现认证 tenant 上下文、设备管理 handler、幂等/ETag/cursor 与批量操作；不得把 T-009 数据基础误报为 API 已交付。 +- T-010：建立 Bell Area/`capture_policy` 只读投影、Sense 成像准入与设备操作审计 Outbox 原子写入；relay 传输协议留给后续任务。 +- 后续 T-011 按 T-008 契约实现认证 tenant 上下文、设备管理 handler、幂等/ETag/cursor 与批量操作；公共控制 API 只在 PostgreSQL 路径启用，不得把 T-009/T-010 数据基础误报为 API 已交付。 - 对账器幂等/退避/并发/10% 安全闸。 - WireGuard 边缘隧道与断网恢复。 - 16 路批量开通、停用和容量基准。 diff --git a/docs/api.md b/docs/api.md index 24520b3..60ca343 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1,6 +1,6 @@ # API 与契约 -> Brain → Bell 事件契约 v0.1、Sense Control API v1 与 Bell 站点配额只读投影 v1 已冻结；其他 API 仍在设计阶段。不得把本文的“待定”自行具体化为公共契约。 +> Brain → Bell 事件契约 v0.1、Sense Control API v1、Bell 配额/Area 只读投影 v1 与 Sense 本地设备审计事件 v1 已冻结；其他 API 仍在设计阶段。不得把本文的“待定”自行具体化为公共契约。 ## 1. 已冻结：Brain → Bell 事件契约 @@ -22,12 +22,14 @@ | 调用方 → 提供方 | 用途 | 当前约束 | 状态 | | --- | --- | --- | --- | | Sense → Bell | 读取站点视频配额 | 同一 PostgreSQL 实例内只读 `bell.site_quota_v1`；默认 16、最大 128；失败时拒绝新增/启用但不影响已有流 | T-008 冻结，T-009 已实现数据路径 | +| Sense → Bell | 读取 Area 成像准入 | 只读 `bell.area_policy_v1`；`video_allowed | non_imaging_only`；缺失/非法/回退失败关闭但不影响已有设备 | T-010 已冻结并实现数据路径 | +| Sense → Bell | 汇入设备操作审计 | 本地 Outbox 事件已冻结并原子落库；transport、签名、确认与留存未冻结 | T-010 本地基础已实现，relay 待设计 | | Bell → Sense | 请求事件证据/pre-roll 切片 | 幂等、按租户授权、异步结果、不得暴露原始凭据 | 待 M3 设计 | | Bell → Brain | outcome/误报反馈 | 原事件不可变；反馈可重试、去重、审计 | 待 M3 设计 | | Sense → Brain | 流绑定与设备型触发 | 分片可路由，触发入口与流控制解耦 | 待 M2/M3 设计 | | Worker → 控制面 | 注册、心跳、容量 | `max_sources` 来自 profile/压测，不固定为 16 | 待 M3 设计 | -冻结签名和失败语义见 [`contracts/README.md`](contracts/README.md) 与 [`contracts/site-quota-v1.sql`](contracts/site-quota-v1.sql)。Bell 拥有源数据和视图，Sense 数据库角色只有 `SELECT`；未来分库必须发布新版本，不能在 v1 下把本地视图静默替换为网络调用。 +冻结签名和失败语义见 [`contracts/README.md`](contracts/README.md)、[`contracts/site-quota-v1.sql`](contracts/site-quota-v1.sql)、[`contracts/area-policy-v1.sql`](contracts/area-policy-v1.sql) 与 [`contracts/sense-device-audit-v1.schema.json`](contracts/sense-device-audit-v1.schema.json)。Bell 拥有投影源数据和视图，Sense 数据库角色只有 `SELECT`；未来分库必须发布新版本，不能在 v1 下把本地视图静默替换为网络调用。 ## 3. 已冻结：Sense Control API v1 @@ -43,7 +45,7 @@ - `endpoint_ref`、`credential_ref`、`profile_token` 只写不读；设备 ID 由服务端生成。普通响应和错误不得包含凭据、完整流 URI、token 或 MediaMTX 内部配置。 - v1 不提供删除设备；停用设备保留历史。写入受理只表示期望态已持久化，不能表示实际态已收敛。 -T-008 冻结契约；T-009 已实现 PostgreSQL `bell.sites`/配额视图、最小权限和 Sense repository，但尚未实现 Sense HTTP handler、认证中间件或 Bell 管理服务。 +T-008 冻结公共控制契约；T-009/T-010 已实现 PostgreSQL Site/Area 源表、两个只读视图、最小权限、Sense 准入 repository 和本地审计 Outbox，但尚未实现 Sense HTTP handler、认证中间件、Bell 管理服务或 Outbox relay。 ## 4. 待冻结的 Bell 公共 API @@ -68,7 +70,7 @@ Sense 使用 MediaMTX 官方 OpenAPI 生成客户端并加薄封装。业务代 | --- | --- | --- | | ONVIF adapter | `Probe(target)`、`SetSystemDateAndTime(target, time)` | 设备是外部来源；`target` 只含 endpoint ref 与不透明 credential ref。错误稳定映射为认证失败、超时、不可用、响应无效，不记录凭据或完整流地址 | | MediaMTX paths | `CreatePath`、`GetPath`、`EnsurePath`、`DeletePath`、`PathReady` | SQLite 设备台账持有期望态，MediaMTX 只持有运行配置；`EnsurePath` 相同 source 不写、不同 source patch、缺失时 add；当前调和器绝不枚举或删除孤儿 | -| Device repository | 设备、期望态、实际态、调和进度 | SQLite 是 M1 默认开发路径；PostgreSQL 是 M2 生产路径并只读 `bell.site_quota_v1`。调和失败次数与下次时间持久化，进程重启不清空退避；配额读取失败时拒绝新增或启用，不关闭已有流 | +| Device repository | 设备、期望态、实际态、调和进度 | SQLite 是 M1 默认开发路径；PostgreSQL 是 M2 生产路径并只读 Site/Area 两个 Bell 视图。调和退避持久化；配额/Area 失败时拒绝相关新增或启用，不关闭已有流；高风险写入与脱敏 Outbox 同事务 | MediaMTX 薄封装调用同版官方 OpenAPI 的 `/v3/config/paths/get|add|patch|delete/{name}` 与 `/v3/paths/get/{name}`。生成源、版本和 SHA-256 见 `docs/03-tech-stack.md`；业务包不得直接 import 生成包。 @@ -76,7 +78,7 @@ MediaMTX 薄封装调用同版官方 OpenAPI 的 `/v3/config/paths/get|add|patch - 设备类型由 `modality` 表达物理类别，由多值 `capabilities` 表达视频采集、音频、空间规则或遥测能力，避免把“摄像头”固化为唯一设备模型。 - 视频配额只统计 `desired_state=enabled` 且具有 `video_capture` capability 的设备；站点默认 16、可配置 1～128。禁用设备和非视频传感器不占视频路数。 -- SQLite 表使用 `sense_` 前缀；T-009 PostgreSQL 使用 `sense.devices`、`sense.device_capabilities`、`sense.reconcile_state` 和只记录已观察版本的 `sense.site_quota_projection_state`。PostgreSQL 不建立可写 Site 真相副本，站点与配额只来自 Bell 视图；SQLite 的 `sense_sites` 仅是 M1 本地兼容表，不是跨系统公共契约。 +- SQLite 表使用 `sense_` 前缀且只是 M1 实验室兼容路径；PostgreSQL 使用 `sense.devices`、能力/调和表、两个投影观察表和 `sense.device_operation_outbox`。PostgreSQL 不建立可写 Site/Area 真相副本；后续公共控制 API 只在 PostgreSQL 路径启用，不能把 SQLite 描述为 Area/Outbox 生产等价实现。 - 摄像头密码不进入设备普通字段。`credential_ref` 只保存外部密钥引用；ONVIF 返回的 stream URI 只在内存中传给 MediaMTX，不写入设备台账或日志。 ### 5.3 Sense 进程 HTTP 面 @@ -90,4 +92,4 @@ T-003 只实现了运维探针：`GET /healthz` 表示进程存活，`GET /ready 3. 同步生产者、消费者、契约测试和示例。 4. 记录迁移、回滚与版本废弃策略。 -事件 v0.1、Sense Control API v1 或站点配额投影 v1 的破坏性变化必须发布新版本，不能原地修改已被生产者/消费者使用的契约。 +事件 v0.1、Sense Control API v1、配额/Area 投影 v1 或本地设备审计 v1 的破坏性变化必须发布新版本，不能原地修改已被生产者/消费者使用的契约。 diff --git a/docs/contracts/README.md b/docs/contracts/README.md index c96ebd6..9a27735 100644 --- a/docs/contracts/README.md +++ b/docs/contracts/README.md @@ -1,6 +1,6 @@ -# Sense 控制面与站点配额契约 v1 +# Sense 控制面、准入投影与本地审计契约 v1 -> 冻结日期：2026-08-07。契约版本：`1.0.0`。Sense 是设备期望态的提供方；Bell 是 Tenant、Site、Area、RBAC 与配额的所有者。本文冻结接口，不表示 HTTP handler、Bell 表或 PostgreSQL migration 已实现。 +> 冻结日期：2026-08-07。契约版本：`1.0.0`。Sense 是设备期望态的提供方；Bell 是 Tenant、Site、Area、RBAC、配额与全局审计的所有者。T-009/T-010 已实现两个只读投影和本地审计事务基础；HTTP handler、Bell 管理服务和 Outbox relay 仍未实现。 ## 契约文件 @@ -8,6 +8,8 @@ | --- | --- | --- | --- | | [`sense-control-v1.openapi.json`](sense-control-v1.openapi.json) | Sense | Bell 管理面、受控集成方 | 设备查询、创建、修改、启停与批量操作 | | [`site-quota-v1.sql`](site-quota-v1.sql) | Bell | Sense | 单 PostgreSQL 实例内的站点视频配额只读投影 | +| [`area-policy-v1.sql`](area-policy-v1.sql) | Bell | Sense | Area 归属与 `capture_policy` 只读投影 | +| [`sense-device-audit-v1.schema.json`](sense-device-audit-v1.schema.json) | Sense | 本地 Outbox；未来 Bell relay | 脱敏设备操作审计事实，不包含传输协议 | OpenAPI 的 `/api/v1` 路径是公共控制面边界；`/healthz`、`/readyz` 仍是非业务运维探针。v1 不提供设备删除：停用设备使用期望态接口，保留设备、操作和审计历史。Site、Area、配额、RBAC 和审计聚合不由 Sense 提供 CRUD。 @@ -48,7 +50,13 @@ Bell migration 最终创建 `bell.site_quota_v1`，列顺序和含义固定如 配额只统计 `desired_state=enabled` 且 capabilities 含 `video_capture` 的设备。创建已启用视频设备或把视频设备启用时，Sense 必须在同一设备写路径读取并记录所用 `source_version`，同时验证 Area 策略投影。降低配额不会自动停用已有设备；若当前占用已超限，后续创建/启用返回 `409 quota_exceeded`。配额行缺失、越界、版本回退或投影不可读时返回 `503 quota_projection_unavailable`，只阻止相关创建/启用，读取、非准入属性修改和停用仍允许，已有流保持运行。 -Area/capture policy 的投影形态不在 T-008 中冻结；Sense v1 仍保留 `area_policy_unavailable` 与 `area_policy_denied` 稳定错误语义，后续契约不得放宽同写路径校验要求。 +T-010 冻结 `bell.area_policy_v1` 的列顺序为 `tenant_id/site_id/area_id/capture_policy/source_version/source_updated_at`，策略仅允许 `video_allowed | non_imaging_only`。Sense 对所有 PostgreSQL 新建设备验证 Area 归属；具有 `video_capture` 能力的设备在创建（包括 disabled 创建）和启用时检查策略。缺失、非法、版本回退或不可读映射为 `area_policy_unavailable`，`non_imaging_only` 拒绝成像变更并映射为 `area_policy_denied`。同库视图实时读取，不把长期未修改记录的 `source_updated_at` 年龄误判为过期。 + +## 本地设备操作审计 + +`sense-device-audit-v1.schema.json` 冻结本地审计事实的逻辑 envelope。当前事件只有 `device.created` 和 `device.desired_state.accepted`；主体类型为 `user | service | system`，投影版本与 generation 随事实保存。`data` 只包含 Area、模态、能力和状态变化等脱敏字段，禁止 endpoint、credential、profile token、path、密码、完整流 URI 或 MediaMTX 配置。 + +PostgreSQL repository 必须在设备创建/期望态事务内写 `sense.device_operation_outbox`；Outbox 失败回滚业务写入。相同期望态不增加 generation，但仍产生独立审计事实。Schema 不是 Bell relay 协议：传输端点、签名、批量确认、重放窗口和留存由后续任务冻结。 ## 兼容与废弃 diff --git a/docs/contracts/area-policy-v1.sql b/docs/contracts/area-policy-v1.sql new file mode 100644 index 0000000..181056a --- /dev/null +++ b/docs/contracts/area-policy-v1.sql @@ -0,0 +1,30 @@ +-- Contract-only reference for the Bell -> Sense Area admission projection. +-- The executable migration is deploy/postgres/005_area_policy.sql. + +CREATE OR REPLACE VIEW bell.area_policy_v1 ( + tenant_id, + site_id, + area_id, + capture_policy, + source_version, + source_updated_at +) AS +SELECT + area.tenant_id, + area.site_id, + area.id, + area.capture_policy, + area.version, + area.updated_at +FROM bell.areas AS area +JOIN bell.sites AS site + ON site.tenant_id = area.tenant_id AND site.id = area.site_id +WHERE area.deleted_at IS NULL AND site.deleted_at IS NULL; + +COMMENT ON VIEW bell.area_policy_v1 IS + 'v1 read-only Area capture-policy projection owned by Bell and consumed by Sense'; + +ALTER VIEW bell.area_policy_v1 OWNER TO bell_app; +REVOKE ALL PRIVILEGES ON TABLE bell.area_policy_v1 FROM PUBLIC; +REVOKE ALL PRIVILEGES ON TABLE bell.area_policy_v1 FROM sense_app; +GRANT SELECT ON TABLE bell.area_policy_v1 TO sense_app; diff --git a/docs/contracts/sense-device-audit-v1.schema.json b/docs/contracts/sense-device-audit-v1.schema.json new file mode 100644 index 0000000..ae1dfe0 --- /dev/null +++ b/docs/contracts/sense-device-audit-v1.schema.json @@ -0,0 +1,109 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://yovision.local/contracts/sense-device-audit-v1.schema.json", + "title": "Sense Device Audit Event v1", + "description": "Sense 本地持久化的设备操作审计事实；不定义 Bell relay 的传输、签名、确认或留存。", + "type": "object", + "additionalProperties": false, + "required": [ + "event_id", + "event_type", + "tenant_id", + "site_id", + "device_id", + "actor", + "reason", + "trace_id", + "aggregate_generation", + "projection_versions", + "data", + "occurred_at" + ], + "properties": { + "event_id": { + "type": "string", + "pattern": "^audit_[0-9a-f]{32}$" + }, + "event_type": { + "enum": ["device.created", "device.desired_state.accepted"] + }, + "tenant_id": {"$ref": "#/$defs/logicalId"}, + "site_id": {"$ref": "#/$defs/logicalId"}, + "device_id": {"$ref": "#/$defs/logicalId"}, + "actor": { + "type": "object", + "additionalProperties": false, + "required": ["type", "id"], + "properties": { + "type": {"enum": ["user", "service", "system"]}, + "id": {"type": "string", "minLength": 1, "maxLength": 200} + } + }, + "reason": {"type": ["string", "null"], "maxLength": 500}, + "trace_id": {"type": ["string", "null"], "maxLength": 128}, + "aggregate_generation": {"type": "integer", "minimum": 1}, + "projection_versions": { + "type": "object", + "additionalProperties": false, + "required": ["quota_source_version", "area_policy_source_version"], + "properties": { + "quota_source_version": {"type": ["integer", "null"], "minimum": 1}, + "area_policy_source_version": {"type": ["integer", "null"], "minimum": 1} + } + }, + "data": { + "oneOf": [ + {"$ref": "#/$defs/deviceCreatedData"}, + {"$ref": "#/$defs/desiredStateData"} + ] + }, + "occurred_at": {"type": "string", "format": "date-time"} + }, + "allOf": [ + { + "if": {"properties": {"event_type": {"const": "device.created"}}}, + "then": {"properties": {"data": {"$ref": "#/$defs/deviceCreatedData"}}} + }, + { + "if": {"properties": {"event_type": {"const": "device.desired_state.accepted"}}}, + "then": {"properties": {"data": {"$ref": "#/$defs/desiredStateData"}}} + } + ], + "$defs": { + "logicalId": { + "type": "string", + "minLength": 1, + "maxLength": 128, + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]*$" + }, + "desiredState": {"enum": ["disabled", "enabled"]}, + "deviceCreatedData": { + "type": "object", + "additionalProperties": false, + "required": ["kind", "area_id", "modality", "capabilities", "desired_state"], + "properties": { + "kind": {"const": "device_created"}, + "area_id": {"$ref": "#/$defs/logicalId"}, + "modality": {"enum": ["video", "radar", "contact", "button", "wearable", "other"]}, + "capabilities": { + "type": "array", + "maxItems": 16, + "uniqueItems": true, + "items": {"enum": ["video_capture", "audio_capture", "spatial_rule", "telemetry"]} + }, + "desired_state": {"$ref": "#/$defs/desiredState"} + } + }, + "desiredStateData": { + "type": "object", + "additionalProperties": false, + "required": ["kind", "previous_desired_state", "desired_state", "changed"], + "properties": { + "kind": {"const": "desired_state_accepted"}, + "previous_desired_state": {"$ref": "#/$defs/desiredState"}, + "desired_state": {"$ref": "#/$defs/desiredState"}, + "changed": {"type": "boolean"} + } + } + } +} diff --git a/docs/current-state.md b/docs/current-state.md index b4cdd24..f30ad23 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -4,8 +4,8 @@ ## 当前阶段 -- 阶段：M0 指定摄像头型号准入已完成；M1 的“一实机 + 四合成源”实验室软件闭环已通过；M2 已完成控制面契约和 PostgreSQL 数据基础，但五条独立真实上游和生产 SLA 尚未验收。 -- 生产代码：Sense 已包含可构建进程、SQLite/PostgreSQL repository、标准 ONVIF SOAP/WS-Security adapter、凭据引用、MediaMTX 生成客户端、对账、探活和 MediaMTX 重启重建；公共设备管理 API、认证和完整生产部署仍未实现。 +- 阶段：M0 指定摄像头型号准入已完成；M1 的“一实机 + 四合成源”实验室软件闭环已通过；M2 已完成控制面契约、Site/Area 准入与本地审计事务基础，但五条独立真实上游和生产 SLA 尚未验收。 +- 生产代码：Sense 已包含可构建进程、SQLite/PostgreSQL repository、Site/Area 准入、设备操作 Outbox、标准 ONVIF SOAP/WS-Security adapter、凭据引用、MediaMTX 生成客户端、对账、探活和 MediaMTX 重启重建；公共设备管理 API、认证、Outbox relay 和完整生产部署仍未实现。 - 默认容量：16 路；单站点本阶段上限 128 路，必须横向分片。 ## 仓库现实 @@ -13,13 +13,14 @@ - `Sense/` 已有 Go module 与 `cmd/sense-api`；`Brain/`、`Bell/` 仍只有目录占位，Bell 的首个数据库源表/视图 migration 位于根目录 `deploy/postgres/`。 - Sense 设备模型使用 `modality + capabilities`，SQLite 执行 v1 migration；视频配额默认 16、允许 1～128，17/128/129、新增/启用和“降低配额不关闭已有流”均有测试。 - T-009 冻结 PostgreSQL `17.10` 和 `pgx/v5 v5.10.0`，实现 `bell`/`sense` schema、NOLOGIN 权限角色、Bell Site 版本 trigger、`bell.site_quota_v1` 和 Sense PostgreSQL repository；同站点并发准入用事务级 advisory lock，配额缺失/越界/版本回退时失败关闭且不改变已有流。 -- Windows 隔离测试使用 `D:\pgsql17\bin` 启动随机回环端口临时集群，migration 可重放、权限断言和 8 组真实 PostgreSQL repository 集成场景通过后自动清理；现有 `D:\pgsql17\data` 和 5432 服务未被读取、停止或修改。 +- T-010 增量实现 `bell.areas`、`bell.area_policy_v1`、Area 版本观察和 `sense.device_operation_outbox`；`non_imaging_only` 拒绝成像设备创建/启用，失败不改变已有设备。设备创建/期望态受理与脱敏 Outbox 同事务，相同期望态不增加 generation 但仍审计。 +- Windows 隔离测试使用 `D:\pgsql17\bin` 启动随机回环端口临时集群，`001`～`007` migration 可重放、权限断言和 18 个真实 PostgreSQL 测试通过后自动清理；现有 `D:\pgsql17\data` 和 5432 服务未被读取、停止或修改。 - MediaMTX 固定为独立二进制 `v1.19.3`，官方 OpenAPI 已按 SHA-256 vendoring，并由固定 `oapi-codegen v2.8.0` 生成客户端；手写薄封装有 create/read/delete、幂等 ensure 与探活假 HTTP 测试。 - T-003 对账进度与指数退避持久化，覆盖取消和 SQLite 重启恢复；T-006 增加真实 ONVIF adapter、RTSP router、实验室播种/状态工具、故障代理和五路自动验收。当前仍不枚举/删除孤儿。 - T-006 正式使用 1 台准入实机和 4 个独立合成 publisher 连续观察 `1806.6 s` / 180 次采样，四类恢复均通过，最大与最终 `unconverged` 均为 0；详细证据见 `docs/research/sense-5-stream-integration.md`。 - `docs/raw/01`～`08` 已记录需求、分析、方案、客户场景、事件比对和三系统职责。 - `docs/raw/contracts/event-v0.1.schema.json` 已冻结，并有多份示例与语义说明。 -- `docs/contracts/sense-control-v1.openapi.json` 已冻结站点作用域的设备查询、创建、修改、启停与最多 128 项批量操作；`site-quota-v1.sql` 已冻结 Bell 所有、Sense 只读的配额视图签名。T-009 已实现数据库 migration/repository；HTTP handler、认证和 Bell 管理服务尚未实现。 +- `docs/contracts/sense-control-v1.openapi.json` 已冻结站点作用域的设备查询、创建、修改、启停与最多 128 项批量操作；配额/Area 只读视图和本地设备审计事件均已冻结并实现数据库基础。HTTP handler、认证、Bell 管理服务和 Outbox relay 尚未实现。 - harness coding 文档、上下文清单、Gitea Issue/PR 模板和治理脚本已接入。 - Gitea 已初始化 12 个协作标签；`status/waiting` 用于依赖或外部条件未满足的未领取任务，实时可领取状态必须从 Gitea 查询，不在本文复制。 - T-002 已关闭架构影响型问题：首期 S2 民办寄宿学校、客户侧私有部署、ONVIF/RTSP、NVIDIA 主路径、自研 Bell、Web/H5 与客户侧证据存储等边界已批准。 @@ -65,12 +66,12 @@ Sense 默认监听 `127.0.0.1:8080`，提供 `/healthz` 与 `/readyz` 运维探 - 人脸方向已延后至 M5 的 S4 成人园区候选试点；必要性/PIP 影响评估、单独同意与替代方式、合法底库来源和删除流程未完成，阻塞人脸能力上线。 - 短信/语音具体供应商未选；生产前必须选定两条独立投递路径并验证故障切换。 - Python/Savant 的精确版本、目标硬件和 Bell 前端栈尚未冻结；Sense M1 的 Go、SQLite driver、MediaMTX、生成器及生成运行时版本已在 T-003 冻结，PostgreSQL/pgx 版本已在 T-009 冻结。 -- 本机现有 PostgreSQL 5432 实例使用 SCRAM 且当前开发进程没有管理员密码；T-009 不绕过认证，自动验收使用隔离临时集群。向共享/生产实例安装 migration 前仍需管理员私下提供专用数据库、登录角色与备份方案。 +- 本机现有 PostgreSQL 5432 实例使用 SCRAM 且当前开发进程没有管理员密码；T-009/T-010 不绕过认证，自动验收使用隔离临时集群。向共享/生产实例安装 migration 前仍需管理员私下提供专用数据库、登录角色与备份方案。 - 代码知识图谱在无业务代码阶段可能为空；工具不可用时使用 `rg` 处理文档与配置。 ## 下一步 -下一步按 Gitea 流程建立 Sense Control API v1 实现任务：基于 T-009 PostgreSQL repository 实现认证 tenant 上下文、设备 handler、幂等收据、ETag/cursor 和批量操作。客户授权、借用或租赁条件具备后再执行 T-007 五条独立真实上游现场门禁。T-006 的合成结果不解除 T-007，也不形成容量或生产 SLA 承诺。实时领取状态仍以 Gitea 为准。 +下一步按 Gitea 流程建立 T-011 Sense Control API v1 实现任务：基于 T-009/T-010 PostgreSQL repository 实现认证 tenant 上下文、7 个设备/operation handler、幂等收据、ETag/cursor 和最多 128 项批量操作；公共 API 不在 SQLite 路径启用。客户授权、借用或租赁条件具备后再执行 T-007 五条独立真实上游现场门禁。T-006 的合成结果不解除 T-007，也不形成容量或生产 SLA 承诺。 ## 已知风险 diff --git a/docs/tasks/T-010.md b/docs/tasks/T-010.md index 9ad3e7c..9facdac 100644 --- a/docs/tasks/T-010.md +++ b/docs/tasks/T-010.md @@ -3,7 +3,7 @@ id: T-010 title: 建立 Area 准入投影与设备操作审计 Outbox 基础 phase: 2 deps: [T-009] -status: DOING +status: DONE created: 2026-08-07 issue: 35 context_ref: 0b40cf16c3e72f4670c33a497e7d533d663ba135 @@ -82,6 +82,15 @@ T-008 已要求 Sense Control API v1 的设备创建携带 `area_id`，并冻结 ## 执行记录 +### 2026-08-07 完成 Area 准入与本地审计事务基础 + +- 新增 `005`～`007` 增量 migration：Bell Area 源表、不可变 tenant/Site/ID 与单调版本 trigger、六列 `bell.area_policy_v1`，以及 Sense Area 观察状态、设备投影版本、脱敏设备操作 Outbox 和最小权限；`001`～`007` 连续执行两遍通过。 +- 冻结 `video_allowed | non_imaging_only` 和本地设备审计 JSON Schema。PostgreSQL 新建设备必须关联匹配 tenant/Site 的 Area；成像设备在创建（包括 disabled）和启用时检查策略，缺失/非法/回退失败关闭，策略变化不静默修改已有设备。 +- Repository 按 Area→Site 固定顺序使用 transaction-scoped advisory lock；设备创建/期望态受理与 Outbox 同事务。相同期望态不增加 generation 但生成独立审计事实；Outbox 故障会回滚设备创建或状态变化，payload 不含 endpoint、credential、path、密码、完整 URI 或 MediaMTX 配置。 +- 新增 9 个 T-010 PostgreSQL 集成场景，总计 18 个 `TestPostgres*`：策略允许/拒绝、缺失/版本回退、并发观察、跨 tenant 隔离、启用重检、既有设备保持、主体上下文/脱敏、创建与期望态两类 Outbox 回滚、过宽 Area 源权限拒绝。`./scripts/test_postgres.ps1 -PgRoot D:\pgsql17` 通过并清理随机回环端口临时集群，现有 `D:\pgsql17\data` 与 5432 listener 未被读取、停止或修改。 +- `./init.ps1` 通过；`python -m unittest discover -s tests -p "test_*.py"` 共 40 项通过，其中 T-010 静态契约 7 项；上下文/治理校验、`go -C Sense test ./...`、`go -C Sense vet ./...`、`go -C Sense build ./...` 和 `git diff --check` 全部通过。不需要摄像头、UI 或客户现场验收。 +- SQLite 继续作为 M1 实验室路径，不宣称具备生产 Area/Outbox 一致性；T-011 公共控制 API 必须只在 PostgreSQL 路径启用。Outbox relay 的 transport、签名、确认、重放窗口和留存仍是后续任务边界。 + ### 2026-08-07 领取任务 - dispatcher `ila` 将 Issue #35 分配给 `codex`；`context_ref` 为 `0b40cf16c3e72f4670c33a497e7d533d663ba135`，claim 为 `claims/T-010`，工作分支为 `agent/codex/T-010`。 diff --git a/scripts/test_postgres.ps1 b/scripts/test_postgres.ps1 index 5311f14..9460399 100644 --- a/scripts/test_postgres.ps1 +++ b/scripts/test_postgres.ps1 @@ -22,7 +22,7 @@ foreach ($required in @($initdb, $pgCtl, $psql, $createdb)) { $postgresExe = Join-Path $bin 'postgres.exe' $version = (& $postgresExe --version 2>&1 | Out-String).Trim() if ($LASTEXITCODE -ne 0 -or $version -notmatch 'PostgreSQL\) 17\.10$') { - throw 'T-009 integration requires the frozen PostgreSQL 17.10 binaries.' + throw 'PostgreSQL integration requires the frozen PostgreSQL 17.10 binaries.' } $existing5432 = @( @@ -40,7 +40,7 @@ $port = ([System.Net.IPEndPoint]$listener.LocalEndpoint).Port $listener.Stop() $systemTemp = [IO.Path]::GetFullPath([IO.Path]::GetTempPath()) -$clusterName = 'yovision-t009-pg-' + [guid]::NewGuid().ToString('N') +$clusterName = 'yovision-t010-pg-' + [guid]::NewGuid().ToString('N') $dataDir = Join-Path $systemTemp $clusterName $logPath = Join-Path $systemTemp ($clusterName + '.log') $started = $false @@ -69,19 +69,26 @@ try { $started = $true $adminRootDSN = "postgres://postgres@127.0.0.1:$port/postgres?sslmode=disable" - $databaseName = 'yovision_t009' + $databaseName = 'yovision_t010' $adminDatabaseDSN = "postgres://postgres@127.0.0.1:$port/${databaseName}?sslmode=disable" - $senseDSN = "postgres://yovision_t009_sense@127.0.0.1:$port/${databaseName}?sslmode=disable" + $senseDSN = "postgres://yovision_t010_sense@127.0.0.1:$port/${databaseName}?sslmode=disable" Invoke-Checked $psql '-X' '-v' 'ON_ERROR_STOP=1' '-d' $adminRootDSN '-f' (Join-Path $repoRoot 'deploy\postgres\001_roles.sql') Invoke-Checked $createdb '-h' '127.0.0.1' '-p' ([string]$port) '-U' 'postgres' $databaseName foreach ($pass in 1..2) { - foreach ($name in @('002_bell.sql', '003_sense.sql', '004_privileges.sql')) { + foreach ($name in @( + '002_bell.sql', + '003_sense.sql', + '004_privileges.sql', + '005_area_policy.sql', + '006_device_operation_outbox.sql', + '007_privileges_area_audit.sql' + )) { Invoke-Checked $psql '-X' '-v' 'ON_ERROR_STOP=1' '-d' $adminDatabaseDSN '-f' (Join-Path $repoRoot "deploy\postgres\$name") } } - Invoke-Checked $psql '-X' '-v' 'ON_ERROR_STOP=1' '-d' $adminRootDSN '-c' 'CREATE ROLE yovision_t009_sense LOGIN IN ROLE sense_app' + Invoke-Checked $psql '-X' '-v' 'ON_ERROR_STOP=1' '-d' $adminRootDSN '-c' 'CREATE ROLE yovision_t010_sense LOGIN IN ROLE sense_app' Invoke-Checked $psql '-X' '-v' 'ON_ERROR_STOP=1' '-d' $adminDatabaseDSN '-f' (Join-Path $repoRoot 'deploy\postgres\tests\assertions.sql') $env:YOVISION_TEST_POSTGRES_DSN = $senseDSN @@ -96,7 +103,7 @@ finally { if (-not $started -or $stopped) { $resolvedData = [IO.Path]::GetFullPath($dataDir) if (-not $resolvedData.StartsWith($systemTemp, [StringComparison]::OrdinalIgnoreCase) -or - [IO.Path]::GetFileName($resolvedData) -notlike 'yovision-t009-pg-*') { + [IO.Path]::GetFileName($resolvedData) -notlike 'yovision-t010-pg-*') { throw "Refusing to clean unexpected temporary path." } if (Test-Path -LiteralPath $resolvedData) { diff --git a/tests/test_area_audit_contract.py b/tests/test_area_audit_contract.py new file mode 100644 index 0000000..880e212 --- /dev/null +++ b/tests/test_area_audit_contract.py @@ -0,0 +1,154 @@ +"""Static contract and safety checks for T-010 Area admission and audit Outbox.""" + +from __future__ import annotations + +import json +import re +import unittest +from pathlib import Path +from typing import Any + + +ROOT = Path(__file__).resolve().parents[1] +MIGRATION_ROOT = ROOT / "deploy" / "postgres" +CONTRACT_ROOT = ROOT / "docs" / "contracts" + + +def text(path: Path) -> str: + return path.read_text(encoding="utf-8") + + +def normalized(value: str) -> str: + return re.sub(r"\s+", " ", value.lower()).strip() + + +def property_names(value: Any) -> set[str]: + names: set[str] = set() + if isinstance(value, dict): + properties = value.get("properties") + if isinstance(properties, dict): + names.update(str(name) for name in properties) + for child in value.values(): + names.update(property_names(child)) + elif isinstance(value, list): + for child in value: + names.update(property_names(child)) + return names + + +def area_privilege_findings(value: str) -> list[str]: + source = normalized(value) + findings: list[str] = [] + if "grant select on table bell.area_policy_v1 to sense_app" not in source: + findings.append("missing Area-view SELECT") + if "revoke all on table bell.areas from sense_app" not in source: + findings.append("missing Area-source revoke") + if re.search( + r"grant\s+(?:all(?:\s+privileges)?|insert|update|delete|truncate|references|trigger)" + r"(?:\s*,\s*(?:insert|update|delete|truncate|references|trigger))*" + r"\s+on(?:\s+table)?\s+bell\.", + source, + ): + findings.append("Bell write privilege granted") + if re.search(r"grant\s+select\s+on(?:\s+table)?\s+bell\.areas\s+to\s+sense_app", source): + findings.append("Bell Area source is readable") + return findings + + +class AreaAuditContractTests(unittest.TestCase): + def test_area_projection_contract_has_exact_signature_and_policy_enum(self) -> None: + contract = normalized(text(CONTRACT_ROOT / "area-policy-v1.sql")) + migration = normalized(text(MIGRATION_ROOT / "005_area_policy.sql")) + signature = ( + "create or replace view bell.area_policy_v1 ( tenant_id, site_id, area_id, " + "capture_policy, source_version, source_updated_at ) as" + ) + for value in (contract, migration): + self.assertIn(signature, value) + self.assertIn("alter view bell.area_policy_v1 owner to bell_app", value) + self.assertIn( + "capture_policy in ('video_allowed', 'non_imaging_only')", migration + ) + self.assertIn("new.version := old.version + 1", migration) + + def test_area_projection_is_not_a_second_writable_truth(self) -> None: + migration = normalized(text(MIGRATION_ROOT / "005_area_policy.sql")) + self.assertNotRegex(migration, r"create table(?: if not exists)? sense\.areas\b") + self.assertIn("create table if not exists sense.area_policy_projection_state", migration) + self.assertIn("area_policy_source_version bigint", migration) + self.assertIn("area_id text", migration) + + def test_area_privileges_are_select_only_and_negative_cases_fail(self) -> None: + original = text(MIGRATION_ROOT / "007_privileges_area_audit.sql") + self.assertEqual([], area_privilege_findings(original)) + self.assertTrue( + area_privilege_findings( + original + "\nGRANT UPDATE ON bell.area_policy_v1 TO sense_app;\n" + ) + ) + self.assertTrue( + area_privilege_findings(original + "\nGRANT SELECT ON bell.areas TO sense_app;\n") + ) + + def test_local_audit_schema_is_strict_and_contains_no_secret_fields(self) -> None: + schema = json.loads(text(CONTRACT_ROOT / "sense-device-audit-v1.schema.json")) + self.assertEqual("https://json-schema.org/draft/2020-12/schema", schema["$schema"]) + self.assertFalse(schema["additionalProperties"]) + self.assertEqual( + {"device.created", "device.desired_state.accepted"}, + set(schema["properties"]["event_type"]["enum"]), + ) + self.assertEqual( + {"user", "service", "system"}, + set(schema["properties"]["actor"]["properties"]["type"]["enum"]), + ) + forbidden = { + "endpoint_ref", + "credential_ref", + "profile_token", + "path_name", + "password", + "stream_uri", + "mediamtx_config", + } + self.assertEqual(set(), forbidden & property_names(schema)) + + def test_outbox_migration_has_delivery_state_and_integrity_constraints(self) -> None: + migration = normalized(text(MIGRATION_ROOT / "006_device_operation_outbox.sql")) + for marker in ( + "create table if not exists sense.device_operation_outbox", + "payload jsonb not null", + "attempt_count integer not null default 0", + "next_attempt_at timestamptz", + "delivered_at timestamptz", + "jsonb_typeof(payload) = 'object'", + "insert into sense.schema_migrations(version) values (3)", + ): + self.assertIn(marker, migration) + + def test_repository_writes_audit_before_commit_without_sensitive_payload(self) -> None: + source = text(ROOT / "Sense" / "internal" / "store" / "postgres.go") + create_start = source.index("func (s *Postgres) CreateDevice") + create_end = source.index("func checkPostgresAreaPolicy", create_start) + create_body = source[create_start:create_end] + state_start = source.index("func (s *Postgres) SetDesiredState") + state_end = source.index("func (s *Postgres) GetDevice", state_start) + state_body = source[state_start:state_end] + self.assertLess(create_body.index("insertPostgresAudit"), create_body.index("tx.Commit")) + self.assertIn("insertPostgresAudit", state_body) + self.assertLess(state_body.rindex("insertPostgresAudit"), state_body.rindex("tx.Commit")) + payload_fragments = "\n".join( + re.findall(r"Payload:\s*map\[string\]any\{(.*?)\n\s*\}", source, re.S) + ) + for forbidden in ("EndpointRef", "CredentialRef", "PathName"): + self.assertNotIn(forbidden, payload_fragments) + + def test_public_api_retains_stable_area_failure_codes(self) -> None: + spec = json.loads(text(CONTRACT_ROOT / "sense-control-v1.openapi.json")) + error_codes = set(spec["components"]["schemas"]["ErrorCode"]["enum"]) + self.assertIn("area_policy_unavailable", error_codes) + self.assertIn("area_policy_denied", error_codes) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_postgres_contract.py b/tests/test_postgres_contract.py index c301947..6b2dd8c 100644 --- a/tests/test_postgres_contract.py +++ b/tests/test_postgres_contract.py @@ -24,8 +24,12 @@ def privilege_findings(text: str) -> list[str]: findings: list[str] = [] if "grant select on table bell.site_quota_v1 to sense_app" not in value: findings.append("Sense SELECT grant on quota view is missing") + if "grant select on table bell.area_policy_v1 to sense_app" not in value: + findings.append("Sense SELECT grant on Area view is missing") if "revoke all on table bell.sites from sense_app" not in value: findings.append("Bell source table revoke is missing") + if "revoke all on table bell.areas from sense_app" not in value: + findings.append("Bell Area source table revoke is missing") if re.search( r"grant\s+(?:all(?:\s+privileges)?|insert|update|delete|truncate|references|trigger)" r"(?:\s*,\s*(?:insert|update|delete|truncate|references|trigger))*" @@ -42,7 +46,15 @@ class PostgresContractTests(unittest.TestCase): def test_migration_sequence_is_complete_and_stable(self) -> None: names = [path.name for path in sorted(MIGRATION_ROOT.glob("[0-9][0-9][0-9]_*.sql"))] self.assertEqual( - ["001_roles.sql", "002_bell.sql", "003_sense.sql", "004_privileges.sql"], + [ + "001_roles.sql", + "002_bell.sql", + "003_sense.sql", + "004_privileges.sql", + "005_area_policy.sql", + "006_device_operation_outbox.sql", + "007_privileges_area_audit.sql", + ], names, ) @@ -75,10 +87,15 @@ class PostgresContractTests(unittest.TestCase): self.assertIn("create table if not exists sense.site_quota_projection_state", text) def test_cross_schema_privileges_are_minimal(self) -> None: - self.assertEqual([], privilege_findings(migration_text("004_privileges.sql"))) + privileges = migration_text("004_privileges.sql") + migration_text( + "007_privileges_area_audit.sql" + ) + self.assertEqual([], privilege_findings(privileges)) def test_privilege_validator_rejects_broad_or_source_access(self) -> None: - original = migration_text("004_privileges.sql") + original = migration_text("004_privileges.sql") + migration_text( + "007_privileges_area_audit.sql" + ) changed = original + "\nGRANT ALL PRIVILEGES ON bell.sites TO sense_app;\n" self.assertTrue(privilege_findings(changed)) changed = original + "\nGRANT SELECT ON bell.sites TO sense_app;\n" @@ -90,7 +107,7 @@ class PostgresContractTests(unittest.TestCase): "initdb.exe", "pg_ctl.exe", "127.0.0.1", - "yovision-t009-pg-", + "yovision-t010-pg-", "YOVISION_TEST_POSTGRES_DSN", "Get-NetTCPConnection", ):