[T-010] 建立 Area 准入投影与设备操作审计 Outbox 基础 #37

Merged
ila merged 2 commits from agent/codex/T-010 into main 2026-08-07 18:24:47 +08:00
23 changed files with 1354 additions and 102 deletions
+6 -4
View File
@@ -1,6 +1,6 @@
# Sense M1/M2 接入骨架 # 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_<KEY>_RTSP_PASSWORD
MediaMTX `v1.19.3` 应作为独立二进制启动并只在可信网络开放 API。获取与 SHA-256 校验值见 `docs/03-tech-stack.md`。生成客户端使用固定版本工具和 vendored 官方 OpenAPI;`internal/mtx/generated/client.gen.go` 不可手改。 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 本机集成测试从仓库根目录执行: Windows 本机集成测试从仓库根目录执行:
@@ -65,7 +67,7 @@ $env:SENSE_DB_DSN = '由部署环境私下设置'
go run ./cmd/sense-api 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(从仓库根目录执行): Windows 本地准备 MediaMTX(从仓库根目录执行):
+3
View File
@@ -78,6 +78,7 @@ type Device struct {
ID string ID string
TenantID string TenantID string
SiteID string SiteID string
AreaID string
SerialNumber string SerialNumber string
Name string Name string
Modality Modality Modality Modality
@@ -88,6 +89,8 @@ type Device struct {
CredentialRef string CredentialRef string
PathName string PathName string
Generation int64 Generation int64
QuotaSourceVersion int64
AreaPolicySourceVersion int64
CreatedAt time.Time CreatedAt time.Time
UpdatedAt time.Time UpdatedAt time.Time
} }
+127
View File
@@ -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
}
+182 -32
View File
@@ -16,7 +16,7 @@ import (
) )
// Postgres persists Sense state in the sense schema and consumes only Bell's // 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 { type Postgres struct {
db *sql.DB db *sql.DB
} }
@@ -53,20 +53,29 @@ func (s *Postgres) Close() error {
func (s *Postgres) verifySchemaAndPrivileges(ctx context.Context) error { func (s *Postgres) verifySchemaAndPrivileges(ctx context.Context) error {
var version sql.NullInt64 var version sql.NullInt64
if err := s.db.QueryRowContext(ctx, if err := s.db.QueryRowContext(ctx,
`SELECT MAX(version) FROM sense.schema_migrations`).Scan(&version); err != nil || !version.Valid || version.Int64 < 1 { `SELECT MAX(version) FROM sense.schema_migrations`).Scan(&version); err != nil || !version.Valid || version.Int64 < 3 {
return errors.New("postgres sense schema migration v1 is required") 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 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', 'SELECT'),
has_table_privilege(current_user, 'bell.site_quota_v1', 'INSERT,UPDATE,DELETE'), 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', 'SELECT'),
has_table_privilege(current_user, 'bell.sites', 'INSERT,UPDATE,DELETE')`). has_table_privilege(current_user, 'bell.sites', 'INSERT,UPDATE,DELETE'),
Scan(&canReadView, &canWriteView, &canReadSource, &canWriteSource); err != nil { has_table_privilege(current_user, 'bell.area_policy_v1', 'SELECT'),
return errors.New("verify postgres quota privileges") 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 { if !canReadQuotaView || canWriteQuotaView || canReadSiteSource || canWriteSiteSource ||
return errors.New("postgres role violates Bell quota privilege boundary") !canReadAreaView || canWriteAreaView || canReadAreaSource || canWriteAreaSource {
return errors.New("postgres role violates Bell projection privilege boundary")
} }
return nil return nil
} }
@@ -92,7 +101,14 @@ func (s *Postgres) CreateDevice(ctx context.Context, value device.Device) error
return errors.New("begin postgres create device") return errors.New("begin postgres create device")
} }
defer tx.Rollback() 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() { if value.ConsumesVideoChannel() {
version, quotaErr := checkPostgresVideoQuota(ctx, tx, value.TenantID, value.SiteID, now) version, quotaErr := checkPostgresVideoQuota(ctx, tx, value.TenantID, value.SiteID, now)
if quotaErr != nil { if quotaErr != nil {
@@ -101,14 +117,15 @@ func (s *Postgres) CreateDevice(ctx context.Context, value device.Device) error
quotaVersion = version quotaVersion = version
} }
_, err = tx.ExecContext(ctx, `INSERT INTO sense.devices( _, 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, desired_state, actual_state, endpoint_ref, credential_ref,
path_name, generation, quota_source_version, created_at, updated_at path_name, generation, quota_source_version, area_policy_source_version,
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)`, created_at, updated_at
value.ID, value.TenantID, value.SiteID, value.SerialNumber, value.Name, ) 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.Modality, value.DesiredState, value.ActualState, value.EndpointRef,
value.CredentialRef, value.PathName, value.Generation, quotaVersion, value.CredentialRef, value.PathName, value.Generation, nullableVersion(quotaVersion),
value.CreatedAt, value.UpdatedAt) areaVersion, value.CreatedAt, value.UpdatedAt)
if err != nil { if err != nil {
return errors.New("insert postgres device") 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 { VALUES ($1, $2)`, value.ID, now); err != nil {
return errors.New("insert postgres reconcile state") 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 { if err := tx.Commit(); err != nil {
return errors.New("commit postgres create device") return errors.New("commit postgres create device")
} }
return nil 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( func checkPostgresVideoQuota(
ctx context.Context, ctx context.Context,
tx *sql.Tx, tx *sql.Tx,
@@ -199,10 +287,17 @@ func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired devic
} }
defer tx.Rollback() defer tx.Rollback()
var tenantID, siteID, endpointRef, pathName string var tenantID, siteID, endpointRef, pathName string
var areaID sql.NullString
var current device.DesiredState 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). FROM sense.devices WHERE id = $1 FOR UPDATE`, id).
Scan(&tenantID, &siteID, &current, &endpointRef, &pathName) Scan(
&tenantID, &siteID, &areaID, &current, &endpointRef, &pathName,
&generation, &storedQuotaVersion, &storedAreaVersion,
)
if errors.Is(err, sql.ErrNoRows) { if errors.Is(err, sql.ErrNoRows) {
return ErrNotFound 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") return errors.New("read postgres device desired state")
} }
if current == desired { 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
} }
var quotaVersion any if err := tx.Commit(); err != nil {
return errors.New("commit postgres no-op desired-state audit")
}
return nil
}
var quotaVersion, areaVersion int64
if desired == device.DesiredEnabled { if desired == device.DesiredEnabled {
var hasVideo bool var hasVideo bool
if err := tx.QueryRowContext(ctx, `SELECT EXISTS( 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) == "" { if strings.TrimSpace(endpointRef) == "" || strings.TrimSpace(pathName) == "" {
return errors.New("enabled video devices require endpoint ref and path name") 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()) version, quotaErr := checkPostgresVideoQuota(ctx, tx, tenantID, siteID, time.Now().UTC())
if quotaErr != nil { if quotaErr != nil {
return quotaErr return quotaErr
@@ -232,21 +350,41 @@ func (s *Postgres) SetDesiredState(ctx context.Context, id string, desired devic
quotaVersion = version 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, desired_state = $1, actual_state = 'pending', generation = generation + 1,
quota_source_version = COALESCE($2, quota_source_version), updated_at = $3 quota_source_version = COALESCE($2, quota_source_version),
WHERE id = $4`, desired, quotaVersion, time.Now().UTC(), id) 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 { if err != nil {
return errors.New("update postgres desired state") 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 if _, err := tx.ExecContext(ctx, `UPDATE sense.reconcile_state SET
failure_count = 0, next_attempt_at = NULL, last_error_code = NULL, updated_at = $1 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") 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 { if err := tx.Commit(); err != nil {
return errors.New("commit postgres desired-state update") 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) values := make([]ReconcileCandidate, 0)
for rows.Next() { for rows.Next() {
var candidate ReconcileCandidate var candidate ReconcileCandidate
var areaID sql.NullString
var quotaVersion, areaVersion sql.NullInt64
var nextAttempt sql.NullTime var nextAttempt sql.NullTime
if err := rows.Scan( if err := rows.Scan(
&candidate.Device.ID, &candidate.Device.TenantID, &candidate.Device.SiteID, &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.DesiredState, &candidate.Device.ActualState,
&candidate.Device.EndpointRef, &candidate.Device.CredentialRef, &candidate.Device.EndpointRef, &candidate.Device.CredentialRef,
&candidate.Device.PathName, &candidate.Device.Generation, &candidate.Device.PathName, &candidate.Device.Generation,
&quotaVersion, &areaVersion,
&candidate.Device.CreatedAt, &candidate.Device.UpdatedAt, &candidate.Device.CreatedAt, &candidate.Device.UpdatedAt,
&candidate.FailureCount, &nextAttempt, &candidate.FailureCount, &nextAttempt,
); err != nil { ); err != nil {
return nil, errors.New("scan postgres due reconcile device") 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 { if nextAttempt.Valid {
value := nextAttempt.Time value := nextAttempt.Time
candidate.NextAttempt = &value candidate.NextAttempt = &value
@@ -483,20 +627,26 @@ func (s *Postgres) ConvergenceSnapshot(ctx context.Context) (ConvergenceSnapshot
return snapshot, nil 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.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` const postgresDeviceSelect = `SELECT ` + postgresDeviceColumns + ` FROM sense.devices d`
func scanPostgresDevice(row scanner) (device.Device, error) { func scanPostgresDevice(row scanner) (device.Device, error) {
var value device.Device var value device.Device
var areaID sql.NullString
var quotaVersion, areaVersion sql.NullInt64
err := row.Scan( 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.Name, &value.Modality, &value.DesiredState, &value.ActualState,
&value.EndpointRef, &value.CredentialRef, &value.PathName, &value.EndpointRef, &value.CredentialRef, &value.PathName,
&value.Generation, &value.CreatedAt, &value.UpdatedAt, &value.Generation, &quotaVersion, &areaVersion, &value.CreatedAt, &value.UpdatedAt,
) )
value.AreaID = areaID.String
value.QuotaSourceVersion = quotaVersion.Int64
value.AreaPolicySourceVersion = areaVersion.Int64
return value, err return value, err
} }
+332 -4
View File
@@ -48,6 +48,7 @@ func TestPostgresDefaultAndMaximumQuota(t *testing.T) {
} }
radar := device.Device{ radar := device.Device{
ID: "radar-default", TenantID: "tenant-default", SiteID: "site-default", ID: "radar-default", TenantID: "tenant-default", SiteID: "site-default",
AreaID: "area-default",
SerialNumber: "radar-default", Name: "Radar", Modality: device.ModalityRadar, SerialNumber: "radar-default", Name: "Radar", Modality: device.ModalityRadar,
Capabilities: []device.Capability{device.CapabilityTelemetry}, Capabilities: []device.Capability{device.CapabilityTelemetry},
DesiredState: device.DesiredEnabled, ActualState: device.ActualPending, DesiredState: device.DesiredEnabled, ActualState: device.ActualPending,
@@ -113,8 +114,8 @@ func TestPostgresProjectionFailureAndRollbackFailClosed(t *testing.T) {
store, admin := openPostgresTestStore(t) store, admin := openPostgresTestStore(t)
ctx := context.Background() ctx := context.Background()
err := store.CreateDevice(ctx, videoDevice(1, "missing-tenant", "missing-site")) err := store.CreateDevice(ctx, videoDevice(1, "missing-tenant", "missing-site"))
if !errors.Is(err, ErrQuotaProjectionUnavailable) { if !errors.Is(err, ErrAreaPolicyUnavailable) {
t.Fatalf("missing projection must fail closed, got %v", err) t.Fatalf("missing Area projection must fail closed first, got %v", err)
} }
insertBellSite(t, admin, "tenant", "site", 2) insertBellSite(t, admin, "tenant", "site", 2)
@@ -285,12 +286,12 @@ func TestPostgresOpenRejectsOverprivilegedRuntimeRole(t *testing.T) {
_, admin := openPostgresTestStore(t) _, admin := openPostgresTestStore(t)
ctx := context.Background() ctx := context.Background()
if _, err := admin.ExecContext(ctx, 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) t.Fatal(err)
} }
defer func() { defer func() {
_, _ = admin.ExecContext(context.Background(), _, _ = 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)) value, err := OpenPostgres(ctx, os.Getenv(postgresTestDSNEnv))
if value != nil { 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) { func openPostgresTestStore(t *testing.T) (*Postgres, *sql.DB) {
t.Helper() t.Helper()
dsn := os.Getenv(postgresTestDSNEnv) dsn := os.Getenv(postgresTestDSNEnv)
@@ -318,10 +629,13 @@ func openPostgresTestStore(t *testing.T) (*Postgres, *sql.DB) {
t.Fatal("connect PostgreSQL test administrator") t.Fatal("connect PostgreSQL test administrator")
} }
if _, err := admin.ExecContext(context.Background(), `TRUNCATE if _, err := admin.ExecContext(context.Background(), `TRUNCATE
sense.device_operation_outbox,
sense.device_capabilities, sense.device_capabilities,
sense.reconcile_state, sense.reconcile_state,
sense.devices, sense.devices,
sense.site_quota_projection_state, sense.site_quota_projection_state,
sense.area_policy_projection_state,
bell.areas,
bell.sites CASCADE`); err != nil { bell.sites CASCADE`); err != nil {
admin.Close() admin.Close()
t.Fatal(err) 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 { if _, err := admin.ExecContext(context.Background(), query, arguments...); err != nil {
t.Fatal(err) 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)
}
} }
+22
View File
@@ -18,6 +18,9 @@ const (
var ( var (
ErrQuotaProjectionUnavailable = errors.New("quota projection unavailable") ErrQuotaProjectionUnavailable = errors.New("quota projection unavailable")
ErrQuotaProjectionInvalid = errors.New("quota projection invalid") 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 // Repository is the storage boundary used by the Sense process. SQLite stays
@@ -61,3 +64,22 @@ func projectionUnavailable() error {
func projectionInvalid() error { func projectionInvalid() error {
return &quotaProjectionError{kind: ErrQuotaProjectionInvalid} return &quotaProjectionError{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}
}
+2 -1
View File
@@ -195,7 +195,8 @@ func openTestStore(t *testing.T) *SQLite {
func videoDevice(index int, tenantID, siteID string) device.Device { func videoDevice(index int, tenantID, siteID string) device.Device {
id := fmt.Sprintf("camera-%03d", index) id := fmt.Sprintf("camera-%03d", index)
return device.Device{ 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, Modality: device.ModalityVideo,
Capabilities: []device.Capability{device.CapabilityVideoCapture, device.CapabilitySpatialRule}, Capabilities: []device.Capability{device.CapabilityVideoCapture, device.CapabilitySpatialRule},
DesiredState: device.DesiredEnabled, ActualState: device.ActualPending, DesiredState: device.DesiredEnabled, ActualState: device.ActualPending,
+127
View File
@@ -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;
@@ -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;
@@ -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;
+6 -5
View File
@@ -1,6 +1,6 @@
# YoVision PostgreSQL 初始化 # 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 写入脚本或日志: 生产/共享实例必须由管理员先备份并在 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`。 - `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` 拥有 `sense` schema,只获得 `bell` schema 的 `USAGE` 和两个投影视图的 `SELECT`。
- `sense_app` 对 `bell.sites`、Bell migration 表和 trigger function 没有权限。 - `sense_app` 对 Bell 源表、Bell migration 表和 trigger function 没有权限;启动检查发现权限过宽时拒绝运行。
- `sense.device_operation_outbox` 是本地持久化审计事实,不是 Bell 全局审计真相;relay 的传输、签名、确认和留存尚未实现。
- PUBLIC 对两个业务 schema 的表和函数没有权限。 - 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 专用数据库,禁止对共享实例做模糊或递归清理。
+64 -1
View File
@@ -6,10 +6,12 @@ DECLARE
default_quota integer; default_quota integer;
initial_version bigint; initial_version bigint;
updated_version bigint; updated_version bigint;
area_initial_version bigint;
area_updated_version bigint;
BEGIN BEGIN
IF current_setting('server_version_num')::integer < 170000 IF current_setting('server_version_num')::integer < 170000
OR current_setting('server_version_num')::integer >= 180000 THEN 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; END IF;
SELECT array_agg(column_name ORDER BY ordinal_position) SELECT array_agg(column_name ORDER BY ordinal_position)
@@ -37,6 +39,43 @@ BEGIN
RAISE EXCEPTION 'sense_app has forbidden Bell source-table privileges'; RAISE EXCEPTION 'sense_app has forbidden Bell source-table privileges';
END IF; 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) INSERT INTO bell.sites(tenant_id, id, name)
VALUES ('assertion-tenant', 'assertion-site', 'Assertion Site') VALUES ('assertion-tenant', 'assertion-site', 'Assertion Site')
RETURNING max_video_channels, version INTO default_quota, initial_version; RETURNING max_video_channels, version INTO default_quota, initial_version;
@@ -51,6 +90,28 @@ BEGIN
RAISE EXCEPTION 'Bell site source_version is not monotonic'; RAISE EXCEPTION 'Bell site source_version is not monotonic';
END IF; 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 BEGIN
INSERT INTO bell.sites(tenant_id, id, name, max_video_channels) INSERT INTO bell.sites(tenant_id, id, name, max_video_channels)
VALUES ('assertion-tenant', 'invalid-site', 'Invalid', 129); VALUES ('assertion-tenant', 'invalid-site', 'Invalid', 129);
@@ -59,6 +120,8 @@ BEGIN
WHEN check_violation THEN NULL; WHEN check_violation THEN NULL;
END; END;
DELETE FROM bell.areas
WHERE tenant_id = 'assertion-tenant' AND id = 'assertion-area';
DELETE FROM bell.sites DELETE FROM bell.sites
WHERE tenant_id = 'assertion-tenant' AND id = 'assertion-site'; WHERE tenant_id = 'assertion-tenant' AND id = 'assertion-site';
END END
+9 -9
View File
@@ -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 和对账器收敛实际态。 2. Sense 维护设备期望态,通过 MediaMTX API 和对账器收敛实际态。
3. Brain 消费视频与触发信号,产生符合 v0.1 的事件。 3. Brain 消费视频与触发信号,产生符合 v0.1 的事件。
4. Bell 做 schema 与代码级断言,生成平台 ULID,保存不可变事件。 4. Bell 做 schema 与代码级断言,生成平台 ULID,保存不可变事件。
@@ -63,12 +63,12 @@ Sense ── 视频流/触发信号 ──> Brain
4. 平台事件 ULID 由 Bell 生成;Brain 只填 `source_event_id`。 4. 平台事件 ULID 由 Bell 生成;Brain 只填 `source_event_id`。
5. 一个 PostgreSQL 实例,`sense`/`bell` schema 分离;Brain 无业务 schema。 5. 一个 PostgreSQL 实例,`sense`/`bell` schema 分离;Brain 无业务 schema。
6. 16/128 都不是单机保证;媒体与推理按独立分片横向扩展。 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;元数据/审计、人脸和训练样本使用独立生命周期。 8. 事件片段写入客户侧 MinIO/S3,常态录像留在客户 NVR;元数据/审计、人脸和训练样本使用独立生命周期。
9. 投递状态机只依赖 Bell provider 接口,不直接依赖某家短信或语音 SDK;生产前至少两条独立路径并能故障切换。 9. 投递状态机只依赖 Bell provider 接口,不直接依赖某家短信或语音 SDK;生产前至少两条独立路径并能故障切换。
10. 设备领域模型使用 `modality + capabilities`,页面不以摄像头作为唯一根实体;未实现协议适配器明确为 `adapter_not_ready`,不得用模拟遥测伪装交付。 10. 设备领域模型使用 `modality + capabilities`,页面不以摄像头作为唯一根实体;未实现协议适配器明确为 `adapter_not_ready`,不得用模拟遥测伪装交付。
11. Tenant/Site/Area/RBAC、配额、`capture_policy` 与全局审计属于 Bell;Sense Control API v1 只管理 Device 期望态与收敛查询,Sense 只读消费版本化投影并在设备写路径执行,投影不可用时只阻断相关新变更,不静默切断已有链路。 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. 容量架构 ## 6. 容量架构
@@ -85,8 +85,8 @@ Sense ── 视频流/触发信号 ──> Brain
- 对账器水平触发、幂等、指数退避、限制并发;部分失败不做跨系统回滚,只持续收敛。 - 对账器水平触发、幂等、指数退避、限制并发;部分失败不做跨系统回滚,只持续收敛。
- 孤儿删除必须有 10% 安全闸和人工可观察指标。 - 孤儿删除必须有 10% 安全闸和人工可观察指标。
- `bell.site_quota_v1` 行缺失、数值越界、版本回退或读取失败只阻止视频设备新增/启用,不中断已有流;降低配额导致超限时不自动停用,后续准入返回稳定错误并产生运维信号。多 Sense 实例使用 PostgreSQL transaction-scoped advisory lock 串行化同 tenant/site 的计数与写入,不能用进程内锁替代。 - `bell.site_quota_v1` 行缺失、数值越界、版本回退或读取失败只阻止视频设备新增/启用,不中断已有流;降低配额导致超限时不自动停用,后续准入返回稳定错误并产生运维信号。多 Sense 实例使用 PostgreSQL transaction-scoped advisory lock 串行化同 tenant/site 的计数与写入,不能用进程内锁替代。
- Area 策略投影读取失败只阻止相关设备新增/启用;已有设备保持原状态并产生运维告警。策略变更与已有成像设备冲突时由 Bell 管理端显式处置。 - `bell.area_policy_v1` 缺失、非法、版本回退或读取失败时,PostgreSQL repository 拒绝相关新增/启用;`non_imaging_only` 允许非成像设备但拒绝具有 `video_capture` 的设备。已有设备保持原状态,策略冲突由 Bell 管理端显式迁移或取消。同库实时视图不以源记录年龄误判 freshness。
- 高风险设备操作在本地事务内同时写期望态与审计 outbox;异步 relay 可重试、幂等投递到 Bell。T-004 只验证交互归属,不定义或实现接口契约。 - 设备创建和期望态受理在本地事务内同时写脱敏 `sense.device_operation_outbox`;Outbox 失败回滚业务写入,相同期望态不增加 generation 但仍审计。异步 relay 尚未实现。
- Brain 投递失败落本地队列重试,不阻塞实时推理主链路。 - Brain 投递失败落本地队列重试,不阻塞实时推理主链路。
- Alert 先落库再投递,进程重启恢复未完成升级链。 - Alert 先落库再投递,进程重启恢复未完成升级链。
- 值班排班发布前必须按 Site 时区校验班次空档、重叠、联系人停用和通道验证;排班以新版本和未来生效时间发布,不原地改写历史。交接班是进行中 Alert 的显式责任转移事件,不替代排班版本变更。 - 值班排班发布前必须按 Site 时区校验班次空档、重叠、联系人停用和通道验证;排班以新版本和未来生效时间发布,不原地改写历史。交接班是进行中 Alert 的显式责任转移事件,不替代排班版本变更。
@@ -94,8 +94,8 @@ Sense ── 视频流/触发信号 ──> Brain
## 8. 数据与契约 ## 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`。 - 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 只实现 PostgreSQL migration/repository,不等于 handler、认证或 Bell 管理服务已实现。 - 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 不合并。 - 业务实体:Rule → Event → Alert → DeliveryAttempt/Ack;Event 与 Alert 不合并。
- Bell 通知域分为三个聚合:Contact/Team 保存身份、成员关系和已验证通道;OnCallSchedule/ScheduleVersion/ShiftException 保存时区、轮换与例外;EscalationPolicy/Step 通过 `person / team / on_call_schedule` 类型化 `target_ref` 引用目标。三者共享逻辑 ID,不复制手机号、班次或轮换字段。 - Bell 通知域分为三个聚合:Contact/Team 保存身份、成员关系和已验证通道;OnCallSchedule/ScheduleVersion/ShiftException 保存时区、轮换与例外;EscalationPolicy/Step 通过 `person / team / on_call_schedule` 类型化 `target_ref` 引用目标。三者共享逻辑 ID,不复制手机号、班次或轮换字段。
- 每个 DeliveryAttempt 创建时解析当时生效的排班版本,并保存实际收件人、通道、`schedule_version` 和解析时间快照;之后联系人或排班修改不得回写既有投递事实。 - 每个 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} Brain/{pipeline,models,judge,emit,trigger,contracts}
Bell/cmd + Bell/internal/{ingest,event,rule,alert,deliver,feedback,tenant,audit,store} Bell/cmd + Bell/internal/{ingest,event,rule,alert,deliver,feedback,tenant,audit,store}
Bell/{web,packs,contracts} 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. 开发顺序 ## 10. 开发顺序
+2 -1
View File
@@ -24,7 +24,8 @@
- T-008:冻结 Sense Control API v1 与 Bell `site_quota_v1` 只读投影契约;只建立接口和测试门禁,不等于 handler、Bell 表或 migration 已实现。 - 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-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% 安全闸。 - 对账器幂等/退避/并发/10% 安全闸。
- WireGuard 边缘隧道与断网恢复。 - WireGuard 边缘隧道与断网恢复。
- 16 路批量开通、停用和容量基准。 - 16 路批量开通、停用和容量基准。
+8 -6
View File
@@ -1,6 +1,6 @@
# API 与契约 # 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 事件契约 ## 1. 已冻结:Brain → Bell 事件契约
@@ -22,12 +22,14 @@
| 调用方 → 提供方 | 用途 | 当前约束 | 状态 | | 调用方 → 提供方 | 用途 | 当前约束 | 状态 |
| --- | --- | --- | --- | | --- | --- | --- | --- |
| Sense → Bell | 读取站点视频配额 | 同一 PostgreSQL 实例内只读 `bell.site_quota_v1`;默认 16、最大 128;失败时拒绝新增/启用但不影响已有流 | T-008 冻结,T-009 已实现数据路径 | | 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 → Sense | 请求事件证据/pre-roll 切片 | 幂等、按租户授权、异步结果、不得暴露原始凭据 | 待 M3 设计 |
| Bell → Brain | outcome/误报反馈 | 原事件不可变;反馈可重试、去重、审计 | 待 M3 设计 | | Bell → Brain | outcome/误报反馈 | 原事件不可变;反馈可重试、去重、审计 | 待 M3 设计 |
| Sense → Brain | 流绑定与设备型触发 | 分片可路由,触发入口与流控制解耦 | 待 M2/M3 设计 | | Sense → Brain | 流绑定与设备型触发 | 分片可路由,触发入口与流控制解耦 | 待 M2/M3 设计 |
| Worker → 控制面 | 注册、心跳、容量 | `max_sources` 来自 profile/压测,不固定为 16 | 待 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 ## 3. 已冻结:Sense Control API v1
@@ -43,7 +45,7 @@
- `endpoint_ref`、`credential_ref`、`profile_token` 只写不读;设备 ID 由服务端生成。普通响应和错误不得包含凭据、完整流 URI、token 或 MediaMTX 内部配置。 - `endpoint_ref`、`credential_ref`、`profile_token` 只写不读;设备 ID 由服务端生成。普通响应和错误不得包含凭据、完整流 URI、token 或 MediaMTX 内部配置。
- v1 不提供删除设备;停用设备保留历史。写入受理只表示期望态已持久化,不能表示实际态已收敛。 - 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 ## 4. 待冻结的 Bell 公共 API
@@ -68,7 +70,7 @@ Sense 使用 MediaMTX 官方 OpenAPI 生成客户端并加薄封装。业务代
| --- | --- | --- | | --- | --- | --- |
| ONVIF adapter | `Probe(target)`、`SetSystemDateAndTime(target, time)` | 设备是外部来源;`target` 只含 endpoint ref 与不透明 credential ref。错误稳定映射为认证失败、超时、不可用、响应无效,不记录凭据或完整流地址 | | 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;当前调和器绝不枚举或删除孤儿 | | 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 生成包。 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` 表达视频采集、音频、空间规则或遥测能力,避免把“摄像头”固化为唯一设备模型。 - 设备类型由 `modality` 表达物理类别,由多值 `capabilities` 表达视频采集、音频、空间规则或遥测能力,避免把“摄像头”固化为唯一设备模型。
- 视频配额只统计 `desired_state=enabled` 且具有 `video_capture` capability 的设备;站点默认 16、可配置 1~128。禁用设备和非视频传感器不占视频路数。 - 视频配额只统计 `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,不写入设备台账或日志。 - 摄像头密码不进入设备普通字段。`credential_ref` 只保存外部密钥引用;ONVIF 返回的 stream URI 只在内存中传给 MediaMTX,不写入设备台账或日志。
### 5.3 Sense 进程 HTTP 面 ### 5.3 Sense 进程 HTTP 面
@@ -90,4 +92,4 @@ T-003 只实现了运维探针:`GET /healthz` 表示进程存活,`GET /ready
3. 同步生产者、消费者、契约测试和示例。 3. 同步生产者、消费者、契约测试和示例。
4. 记录迁移、回滚与版本废弃策略。 4. 记录迁移、回滚与版本废弃策略。
事件 v0.1、Sense Control API v1 或站点配额投影 v1 的破坏性变化必须发布新版本,不能原地修改已被生产者/消费者使用的契约。 事件 v0.1、Sense Control API v1、配额/Area 投影 v1 或本地设备审计 v1 的破坏性变化必须发布新版本,不能原地修改已被生产者/消费者使用的契约。
+11 -3
View File
@@ -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 管理面、受控集成方 | 设备查询、创建、修改、启停与批量操作 | | [`sense-control-v1.openapi.json`](sense-control-v1.openapi.json) | Sense | Bell 管理面、受控集成方 | 设备查询、创建、修改、启停与批量操作 |
| [`site-quota-v1.sql`](site-quota-v1.sql) | Bell | Sense | 单 PostgreSQL 实例内的站点视频配额只读投影 | | [`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。 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`,只阻止相关创建/启用,读取、非准入属性修改和停用仍允许,已有流保持运行。 配额只统计 `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 协议:传输端点、签名、批量确认、重放窗口和留存由后续任务冻结。
## 兼容与废弃 ## 兼容与废弃
+30
View File
@@ -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;
@@ -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"}
}
}
}
}
+7 -6
View File
@@ -4,8 +4,8 @@
## 当前阶段 ## 当前阶段
- 阶段:M0 指定摄像头型号准入已完成;M1 的“一实机 + 四合成源”实验室软件闭环已通过;M2 已完成控制面契约和 PostgreSQL 数据基础,但五条独立真实上游和生产 SLA 尚未验收。 - 阶段:M0 指定摄像头型号准入已完成;M1 的“一实机 + 四合成源”实验室软件闭环已通过;M2 已完成控制面契约、Site/Area 准入与本地审计事务基础,但五条独立真实上游和生产 SLA 尚未验收。
- 生产代码:Sense 已包含可构建进程、SQLite/PostgreSQL repository、标准 ONVIF SOAP/WS-Security adapter、凭据引用、MediaMTX 生成客户端、对账、探活和 MediaMTX 重启重建;公共设备管理 API、认证和完整生产部署仍未实现。 - 生产代码:Sense 已包含可构建进程、SQLite/PostgreSQL repository、Site/Area 准入、设备操作 Outbox、标准 ONVIF SOAP/WS-Security adapter、凭据引用、MediaMTX 生成客户端、对账、探活和 MediaMTX 重启重建;公共设备管理 API、认证、Outbox relay 和完整生产部署仍未实现。
- 默认容量:16 路;单站点本阶段上限 128 路,必须横向分片。 - 默认容量:16 路;单站点本阶段上限 128 路,必须横向分片。
## 仓库现实 ## 仓库现实
@@ -13,13 +13,14 @@
- `Sense/` 已有 Go module 与 `cmd/sense-api`;`Brain/`、`Bell/` 仍只有目录占位,Bell 的首个数据库源表/视图 migration 位于根目录 `deploy/postgres/`。 - `Sense/` 已有 Go module 与 `cmd/sense-api`;`Brain/`、`Bell/` 仍只有目录占位,Bell 的首个数据库源表/视图 migration 位于根目录 `deploy/postgres/`。
- Sense 设备模型使用 `modality + capabilities`,SQLite 执行 v1 migration;视频配额默认 16、允许 1~128,17/128/129、新增/启用和“降低配额不关闭已有流”均有测试。 - 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,配额缺失/越界/版本回退时失败关闭且不改变已有流。 - 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 测试。 - 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-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`。 - T-006 正式使用 1 台准入实机和 4 个独立合成 publisher 连续观察 `1806.6 s` / 180 次采样,四类恢复均通过,最大与最终 `unconverged` 均为 0;详细证据见 `docs/research/sense-5-stream-integration.md`。
- `docs/raw/01`~`08` 已记录需求、分析、方案、客户场景、事件比对和三系统职责。 - `docs/raw/01`~`08` 已记录需求、分析、方案、客户场景、事件比对和三系统职责。
- `docs/raw/contracts/event-v0.1.schema.json` 已冻结,并有多份示例与语义说明。 - `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 模板和治理脚本已接入。 - harness coding 文档、上下文清单、Gitea Issue/PR 模板和治理脚本已接入。
- Gitea 已初始化 12 个协作标签;`status/waiting` 用于依赖或外部条件未满足的未领取任务,实时可领取状态必须从 Gitea 查询,不在本文复制。 - Gitea 已初始化 12 个协作标签;`status/waiting` 用于依赖或外部条件未满足的未领取任务,实时可领取状态必须从 Gitea 查询,不在本文复制。
- T-002 已关闭架构影响型问题:首期 S2 民办寄宿学校、客户侧私有部署、ONVIF/RTSP、NVIDIA 主路径、自研 Bell、Web/H5 与客户侧证据存储等边界已批准。 - T-002 已关闭架构影响型问题:首期 S2 民办寄宿学校、客户侧私有部署、ONVIF/RTSP、NVIDIA 主路径、自研 Bell、Web/H5 与客户侧证据存储等边界已批准。
@@ -65,12 +66,12 @@ Sense 默认监听 `127.0.0.1:8080`,提供 `/healthz` 与 `/readyz` 运维探
- 人脸方向已延后至 M5 的 S4 成人园区候选试点;必要性/PIP 影响评估、单独同意与替代方式、合法底库来源和删除流程未完成,阻塞人脸能力上线。 - 人脸方向已延后至 M5 的 S4 成人园区候选试点;必要性/PIP 影响评估、单独同意与替代方式、合法底库来源和删除流程未完成,阻塞人脸能力上线。
- 短信/语音具体供应商未选;生产前必须选定两条独立投递路径并验证故障切换。 - 短信/语音具体供应商未选;生产前必须选定两条独立投递路径并验证故障切换。
- Python/Savant 的精确版本、目标硬件和 Bell 前端栈尚未冻结;Sense M1 的 Go、SQLite driver、MediaMTX、生成器及生成运行时版本已在 T-003 冻结,PostgreSQL/pgx 版本已在 T-009 冻结。 - 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` 处理文档与配置。 - 代码知识图谱在无业务代码阶段可能为空;工具不可用时使用 `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 承诺。
## 已知风险 ## 已知风险
+18 -4
View File
@@ -3,12 +3,12 @@ id: T-010
title: 建立 Area 准入投影与设备操作审计 Outbox 基础 title: 建立 Area 准入投影与设备操作审计 Outbox 基础
phase: 2 phase: 2
deps: [T-009] deps: [T-009]
status: TODO status: DONE
created: 2026-08-07 created: 2026-08-07
issue: 35 issue: 35
context_ref: null context_ref: 0b40cf16c3e72f4670c33a497e7d533d663ba135
claim_branch: null claim_branch: claims/T-010
work_branch: null work_branch: agent/codex/T-010
write_paths: write_paths:
- docs/tasks/T-010.md - docs/tasks/T-010.md
- docs/contracts/ - docs/contracts/
@@ -82,6 +82,20 @@ 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`。
- 已读回 Issue `status/doing`、结构化 CLAIM 与两个分支 SHA 一致;接受 frontmatter 中全部 `write_paths`,当前没有其他活跃任务占用这些路径。
### 2026-08-07 Gitea 映射 ### 2026-08-07 Gitea 映射
- 任务规格先合入默认分支,再创建唯一主 Issue #35;本提交只回填双向映射,映射合入并读回前不领取任务。 - 任务规格先合入默认分支,再创建唯一主 Issue #35;本提交只回填双向映射,映射合入并读回前不领取任务。
+14 -7
View File
@@ -22,7 +22,7 @@ foreach ($required in @($initdb, $pgCtl, $psql, $createdb)) {
$postgresExe = Join-Path $bin 'postgres.exe' $postgresExe = Join-Path $bin 'postgres.exe'
$version = (& $postgresExe --version 2>&1 | Out-String).Trim() $version = (& $postgresExe --version 2>&1 | Out-String).Trim()
if ($LASTEXITCODE -ne 0 -or $version -notmatch 'PostgreSQL\) 17\.10$') { 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 = @( $existing5432 = @(
@@ -40,7 +40,7 @@ $port = ([System.Net.IPEndPoint]$listener.LocalEndpoint).Port
$listener.Stop() $listener.Stop()
$systemTemp = [IO.Path]::GetFullPath([IO.Path]::GetTempPath()) $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 $dataDir = Join-Path $systemTemp $clusterName
$logPath = Join-Path $systemTemp ($clusterName + '.log') $logPath = Join-Path $systemTemp ($clusterName + '.log')
$started = $false $started = $false
@@ -69,19 +69,26 @@ try {
$started = $true $started = $true
$adminRootDSN = "postgres://postgres@127.0.0.1:$port/postgres?sslmode=disable" $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" $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 $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 Invoke-Checked $createdb '-h' '127.0.0.1' '-p' ([string]$port) '-U' 'postgres' $databaseName
foreach ($pass in 1..2) { 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' $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') 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 $env:YOVISION_TEST_POSTGRES_DSN = $senseDSN
@@ -96,7 +103,7 @@ finally {
if (-not $started -or $stopped) { if (-not $started -or $stopped) {
$resolvedData = [IO.Path]::GetFullPath($dataDir) $resolvedData = [IO.Path]::GetFullPath($dataDir)
if (-not $resolvedData.StartsWith($systemTemp, [StringComparison]::OrdinalIgnoreCase) -or 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." throw "Refusing to clean unexpected temporary path."
} }
if (Test-Path -LiteralPath $resolvedData) { if (Test-Path -LiteralPath $resolvedData) {
+154
View File
@@ -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()
+21 -4
View File
@@ -24,8 +24,12 @@ def privilege_findings(text: str) -> list[str]:
findings: list[str] = [] findings: list[str] = []
if "grant select on table bell.site_quota_v1 to sense_app" not in value: 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") 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: if "revoke all on table bell.sites from sense_app" not in value:
findings.append("Bell source table revoke is missing") 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( if re.search(
r"grant\s+(?:all(?:\s+privileges)?|insert|update|delete|truncate|references|trigger)" r"grant\s+(?:all(?:\s+privileges)?|insert|update|delete|truncate|references|trigger)"
r"(?:\s*,\s*(?: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: 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"))] names = [path.name for path in sorted(MIGRATION_ROOT.glob("[0-9][0-9][0-9]_*.sql"))]
self.assertEqual( 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, names,
) )
@@ -75,10 +87,15 @@ class PostgresContractTests(unittest.TestCase):
self.assertIn("create table if not exists sense.site_quota_projection_state", text) self.assertIn("create table if not exists sense.site_quota_projection_state", text)
def test_cross_schema_privileges_are_minimal(self) -> None: 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: 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" changed = original + "\nGRANT ALL PRIVILEGES ON bell.sites TO sense_app;\n"
self.assertTrue(privilege_findings(changed)) self.assertTrue(privilege_findings(changed))
changed = original + "\nGRANT SELECT ON bell.sites TO sense_app;\n" changed = original + "\nGRANT SELECT ON bell.sites TO sense_app;\n"
@@ -90,7 +107,7 @@ class PostgresContractTests(unittest.TestCase):
"initdb.exe", "initdb.exe",
"pg_ctl.exe", "pg_ctl.exe",
"127.0.0.1", "127.0.0.1",
"yovision-t009-pg-", "yovision-t010-pg-",
"YOVISION_TEST_POSTGRES_DSN", "YOVISION_TEST_POSTGRES_DSN",
"Get-NetTCPConnection", "Get-NetTCPConnection",
): ):