fix: P0/P1 安全与质量缺陷修复 — 10 项 QA 审查问题解决

P0 安全修复:
- tenant_rls: SQL 拼接改为参数化查询防止注入
- follow_up_service: UUID SQL 拼接改为参数化原生查询
- RLS 策略: 新迁移移除空字符串绕过条件
- SSE 消息推送: token 键名 'token' → 'access_token' 修复
- rate_limit: 登录端点 Redis 不可达时 fail-close

P1 质量修复:
- 小程序缓存清理: preservedKeys 补全认证键名
- 小程序 token 刷新: 失败时清除所有认证数据
- 小程序 401: redirectTo → reLaunch 兼容 tabBar
- 集成测试: 信号量限制并行数据库创建(4个)
- change_password: 乐观锁 version 硬编码 → 动态递增

测试: 516 全部通过 (含 153 集成测试)

crates/erp-auth/src/service/auth_service.rs

@@ -321,10 +321,11 @@ impl AuthService {
 
         // 3. Hash new password and update credential
         let new_hash = password::hash_password(new_password)?;
+        let current_version = cred.version;
         let mut cred_active: user_credential::ActiveModel = cred.into();
         cred_active.credential_data = Set(Some(serde_json::json!({ "hash": new_hash })));
         cred_active.updated_at = Set(Utc::now());
-        cred_active.version = Set(2);
+        cred_active.version = Set(current_version + 1);
         cred_active
             .update(db)
             .await

crates/erp-health/src/service/follow_up_service.rs

@@ -5,7 +5,7 @@ use erp_core::audit::AuditLog;
 use erp_core::audit_service;
 use erp_core::events::DomainEvent;
 use sea_orm::entity::prelude::*;
-use sea_orm::{ActiveValue::Set, QueryOrder, QuerySelect, TransactionTrait};
+use sea_orm::{ActiveValue::Set, DatabaseBackend, QueryOrder, QuerySelect, Statement, TransactionTrait};
 use uuid::Uuid;
 
 use erp_core::error::check_version;
@@ -71,13 +71,16 @@ pub async fn list_tasks(
     // 批量查询 assigned_to_name（从 users 表）
     let assigned_ids: HashSet<Uuid> = models.iter().filter_map(|m| m.assigned_to).collect();
     let assigned_names: HashMap<Uuid, String> = if !assigned_ids.is_empty() {
-        let ids_csv = assigned_ids.iter().map(|id| format!("'{}'", id)).collect::<Vec<_>>().join(",");
+        let params: Vec<sea_orm::Value> = assigned_ids.iter().map(|id| (*id).into()).collect();
+        let placeholders: Vec<String> = (1..=params.len()).map(|i| format!("${}", i)).collect();
+        let mut values = params;
+        values.push(tenant_id.into());
         let sql = format!(
-            "SELECT id, COALESCE(display_name, username) AS name FROM users WHERE id IN ({}) AND tenant_id = '{}'",
-            ids_csv, tenant_id
+            "SELECT id, COALESCE(display_name, username) AS name FROM users WHERE id IN ({}) AND tenant_id = ${}",
+            placeholders.join(","), values.len()
         );
-        let rows = state.db.query_all(sea_orm::Statement::from_string(
-            sea_orm::DatabaseBackend::Postgres, sql,
+        let rows = state.db.query_all(Statement::from_sql_and_values(
+            DatabaseBackend::Postgres, sql, values,
         )).await?;
         rows.into_iter()
             .filter_map(|row| {

crates/erp-server/migration/src/lib.rs

@@ -87,6 +87,7 @@ mod m20260427_000084_domain_events_cleanup;
 mod m20260427_000085_processed_events;
 mod m20260427_000086_enable_rls_all_tables;
 mod m20260427_000087_audit_logs_hash_chain;
+mod m20260428_000088_rls_policy_strict;
 
 pub struct Migrator;
 
@@ -181,6 +182,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20260427_000085_processed_events::Migration),
             Box::new(m20260427_000086_enable_rls_all_tables::Migration),
             Box::new(m20260427_000087_audit_logs_hash_chain::Migration),
+            Box::new(m20260428_000088_rls_policy_strict::Migration),
         ]
     }
 }

crates/erp-server/migration/src/m20260428_000088_rls_policy_strict.rs

@@ -0,0 +1,80 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        let conn = manager.get_connection();
+
+        // 替换所有表的 RLS 策略：移除空字符串绕过条件
+        // 原策略允许 current_setting(...) = '' 时通过（绕过 RLS），现在要求变量已设置且匹配
+        conn.execute_unprepared(
+            r#"
+            DO $$
+            DECLARE
+                tbl TEXT;
+            BEGIN
+                FOR tbl IN
+                    SELECT c.table_name FROM information_schema.columns c
+                    JOIN information_schema.tables t
+                      ON c.table_name = t.table_name AND c.table_schema = t.table_schema
+                    WHERE c.column_name = 'tenant_id'
+                      AND c.table_schema = 'public'
+                      AND t.table_type = 'BASE TABLE'
+                    ORDER BY c.table_name
+                LOOP
+                    EXECUTE format('DROP POLICY IF EXISTS tenant_isolation ON %I', tbl);
+                    EXECUTE format(
+                        'CREATE POLICY tenant_isolation ON %I USING (
+                            current_setting(''app.current_tenant_id'', true) != ''''
+                            AND tenant_id = current_setting(''app.current_tenant_id'', true)::uuid
+                        )',
+                        tbl
+                    );
+                END LOOP;
+            END;
+            $$;
+            "#,
+        ).await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        let conn = manager.get_connection();
+
+        // 回滚：恢复允许空字符串绕过的原策略
+        conn.execute_unprepared(
+            r#"
+            DO $$
+            DECLARE
+                tbl TEXT;
+            BEGIN
+                FOR tbl IN
+                    SELECT c.table_name FROM information_schema.columns c
+                    JOIN information_schema.tables t
+                      ON c.table_name = t.table_name AND c.table_schema = t.table_schema
+                    WHERE c.column_name = 'tenant_id'
+                      AND c.table_schema = 'public'
+                      AND t.table_type = 'BASE TABLE'
+                    ORDER BY c.table_name
+                LOOP
+                    EXECUTE format('DROP POLICY IF EXISTS tenant_isolation ON %I', tbl);
+                    EXECUTE format(
+                        'CREATE POLICY tenant_isolation ON %I USING (
+                            current_setting(''app.current_tenant_id'', true) = ''''
+                            OR tenant_id = current_setting(''app.current_tenant_id'', true)::uuid
+                        )',
+                        tbl
+                    );
+                END LOOP;
+            END;
+            $$;
+            "#,
+        ).await?;
+
+        Ok(())
+    }
+}

crates/erp-server/src/middleware/rate_limit.rs

@@ -180,10 +180,13 @@ pub async fn account_lockout_middleware(
 ) -> Response {
     let avail = redis_avail();
 
-    // Redis 不可达时 fail-open：放行请求
+    // Redis 不可达时 fail-close：拒绝登录请求（安全优先）
     if !avail.should_try().await {
-        tracing::warn!("Redis 不可达，fail-open 账户锁定检查放行");
-        return next.run(req).await;
+        tracing::error!("Redis 不可达，fail-close 拒绝登录请求");
+        return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(RateLimitResponse {
+            error: "service_unavailable".to_string(),
+            message: "安全服务暂不可用，请稍后重试".to_string(),
+        })).into_response();
     }
 
     // 获取 Redis 连接
@@ -193,9 +196,12 @@ pub async fn account_lockout_middleware(
             c
         }
         Err(e) => {
-            tracing::warn!(error = %e, "Redis 连接失败，fail-open 账户锁定检查放行");
+            tracing::error!(error = %e, "Redis 连接失败，fail-close 拒绝登录请求");
             avail.mark_failed().await;
-            return next.run(req).await;
+            return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(RateLimitResponse {
+                error: "service_unavailable".to_string(),
+                message: "安全服务暂不可用，请稍后重试".to_string(),
+            })).into_response();
         }
     };
 

crates/erp-server/src/middleware/tenant_rls.rs

@@ -3,7 +3,7 @@ use axum::http::Request;
 use axum::middleware::Next;
 use axum::response::Response;
 use erp_core::types::TenantContext;
-use sea_orm::ConnectionTrait;
+use sea_orm::{ConnectionTrait, DatabaseBackend, Statement};
 
 /// Tenant RLS 中间件。
 ///
@@ -21,11 +21,12 @@ pub async fn tenant_rls_middleware(
     let tenant_id = req.extensions().get::<TenantContext>().map(|ctx| ctx.tenant_id);
 
     if let Some(tid) = tenant_id {
-        // SET app.current_tenant_id — RLS 策略读取此值
+        // SET app.current_tenant_id — RLS 策略读取此值（参数化查询防止注入）
         if let Err(e) = db
-            .execute_unprepared(&format!(
-                "SET app.current_tenant_id = '{}'",
-                tid
+            .execute(Statement::from_sql_and_values(
+                DatabaseBackend::Postgres,
+                "SET app.current_tenant_id = $1",
+                [tid.into()],
             ))
             .await
         {

crates/erp-server/tests/integration/test_db.rs

@@ -1,7 +1,15 @@
 use sea_orm::{Database, ConnectionTrait, Statement, DatabaseBackend};
+use std::sync::Arc;
 
 use erp_server_migration::MigratorTrait;
 
+/// 全局信号量：限制同时创建数据库的测试数量，避免 PostgreSQL 连接耗尽
+static DB_SEMAPHORE: std::sync::OnceLock<Arc<tokio::sync::Semaphore>> = std::sync::OnceLock::new();
+
+fn db_semaphore() -> &'static Arc<tokio::sync::Semaphore> {
+    DB_SEMAPHORE.get_or_init(|| Arc::new(tokio::sync::Semaphore::new(4)))
+}
+
 /// 测试数据库 — 使用本地 PostgreSQL 创建隔离测试库
 ///
 /// 连接本地 PostgreSQL（wiki/infrastructure.md 配置），为每个测试创建独立的测试数据库。
@@ -9,10 +17,13 @@ use erp_server_migration::MigratorTrait;
 pub struct TestDb {
     db: Option<sea_orm::DatabaseConnection>,
     db_name: String,
+    _permit: Option<tokio::sync::OwnedSemaphorePermit>,
 }
 
 impl TestDb {
     pub async fn new() -> Self {
+        let permit = db_semaphore().clone().acquire_owned().await.expect("信号量获取失败");
+
         let db_name = format!("erp_test_{}", uuid::Uuid::now_v7().simple());
 
         let admin_url = std::env::var("TEST_DB_URL")
@@ -47,7 +58,7 @@ impl TestDb {
             .await
             .expect("执行数据库迁移失败");
 
-        Self { db: Some(db), db_name }
+        Self { db: Some(db), db_name, _permit: Some(permit) }
     }
 
     /// 获取数据库连接引用