From 956ef6a1d3437b87b77ebee4aeeb34d2b147b630 Mon Sep 17 00:00:00 2001 From: anonsaber Date: Fri, 25 Sep 2026 20:11:35 +0800 Subject: [PATCH] feat(config): add global Redis-backed enable switch --- README.md | 3 ++ docs/deployment.md | 5 +++ src/notify.rs | 98 +++++++++++++++++++++++++++++++++++++++++++++- src/state.rs | 41 +++++++++++++++++++ web/config.js | 26 ++++++++++++ web/index.html | 1 + 6 files changed, 173 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 14d56d4..39c502a 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,9 @@ HostStack 使用根目录 `hoststack.yaml` 构建并运行 Rust 服务：执行 进行 30 秒间隔、5 秒超时的健康检查。请在 HostStack Secret 中注入 `REDIS_URL` 与 `CONFIG_ENCRYPTION_KEY`；密钥不写入 YAML、镜像或仓库文件。 +管理会话内的“启用业务处理”开关持久化在 Redis `config:enabled`；缺失或读取失败均按关闭处理。 +关闭时 Webhook、Push、Reconcile 和 Worker 返回 HTTP 503，不确认或丢弃上游事件；SPA、状态和管理配置 API 仍可用。 + --- ## 启动后怎么看它是否正常 diff --git a/docs/deployment.md b/docs/deployment.md index 20d7fc8..571344c 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -90,6 +90,11 @@ HTTP 端口并以 `/healthz` 做 interval 30 秒、timeout 5 秒的健康检查 `CONFIG_ENCRYPTION_KEY` 必须配置为 HostStack Secret；不得将密钥值写入 YAML、镜像、日志 或代码仓库。现有 Dockerfile 部署方式仍受支持。 +全局业务开关通过受保护的 `GET|PUT /api/enabled` 管理，持久化 Redis key 为 +`config:enabled`，默认关闭且读取失败 fail-closed。关闭时 `/webhook/tg`、`/push/jmap`、 +`/reconcile`、`/worker` 返回 503，不执行入队、ACK 或业务处理；SPA、`/api/status`、管理 +会话和配置 API 保持可用，打开后立即生效。 + > 迁移后生产环境仅需 `REDIS_URL` 与 `CONFIG_ENCRYPTION_KEY`；下表中的业务环境变量是历史阶段说明，不应再注入生产容器。业务密钥通过受保护的 `/api/bootstrap` 或管理员 PUT 写入 Redis，并在成功后热重建客户端。 （保留既有 Secrets 注入约定：禁入镜像、`secrecy` 包裹、日志屏蔽；新增运行模式相关变量） diff --git a/src/notify.rs b/src/notify.rs index 06ea600..4ec4c6c 100644 --- a/src/notify.rs +++ b/src/notify.rs @@ -83,6 +83,9 @@ async fn telegram_webhook( body: Bytes, ) -> Response { refresh_business_config(&app).await; + if !business_enabled(&app).await { + return (StatusCode::SERVICE_UNAVAILABLE, "service disabled").into_response(); + } let (auth, _, allowlist) = auth_snapshot(&app); if !header_value_matches( &headers, @@ -117,6 +120,9 @@ async fn telegram_webhook( async fn reconcile(State(app): State, headers: HeaderMap) -> Response { refresh_business_config(&app).await; + if !business_enabled(&app).await { + return (StatusCode::SERVICE_UNAVAILABLE, "service disabled").into_response(); + } let (auth, _, _) = auth_snapshot(&app); let valid = headers .get(header::AUTHORIZATION) @@ -146,6 +152,9 @@ async fn reconcile(State(app): State, headers: HeaderMap) -> Response async fn worker(State(app): State, headers: HeaderMap, body: Bytes) -> Response { refresh_business_config(&app).await; + if !business_enabled(&app).await { + return (StatusCode::SERVICE_UNAVAILABLE, "service disabled").into_response(); + } let (_, worker_token, _) = auth_snapshot(&app); let valid = headers .get(header::AUTHORIZATION) @@ -286,6 +295,33 @@ async fn put_config(State(app): State, headers: HeaderMap, body: Bytes } } +async fn get_enabled(State(app): State, headers: HeaderMap) -> Response { + if !config_authorized(&app, &headers).await { + return StatusCode::UNAUTHORIZED.into_response(); + } + axum::Json(serde_json::json!({"enabled": business_enabled(&app).await})).into_response() +} + +async fn put_enabled(State(app): State, headers: HeaderMap, body: Bytes) -> Response { + if !config_authorized(&app, &headers).await { + return StatusCode::UNAUTHORIZED.into_response(); + } + let Ok(payload) = serde_json::from_slice::(&body) else { + return StatusCode::BAD_REQUEST.into_response(); + }; + let Some(enabled) = payload.get("enabled").and_then(serde_json::Value::as_bool) else { + return StatusCode::UNPROCESSABLE_ENTITY.into_response(); + }; + if app.state.set_enabled(enabled).await.is_err() { + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + } + axum::Json(serde_json::json!({"enabled": enabled})).into_response() +} + +async fn business_enabled(app: &AppState) -> bool { + app.state.is_enabled().await.unwrap_or(false) +} + /// Replace the complete encrypted business configuration. Client construction is performed /// before persistence and swap, so malformed/unreachable settings leave the old worker live. async fn put_business_config( @@ -504,6 +540,9 @@ struct WorkerRequest { async fn jmap_push(State(app): State, body: Bytes) -> Response { refresh_business_config(&app).await; + if !business_enabled(&app).await { + return (StatusCode::SERVICE_UNAVAILABLE, "service disabled").into_response(); + } let (auth, _, _) = auth_snapshot(&app); // Stalwart's PushSubscription callback carries verificationCode in its // JSON object. Check it before any future queue/Redis side effect. @@ -730,6 +769,7 @@ fn router_with_worker_state_runtime_bootstrap_config .route("/reconcile", post(reconcile)) .route("/worker", post(worker)) .route("/api/config", get(get_config).put(put_config)) + .route("/api/enabled", get(get_enabled).put(put_enabled)) .route("/api/business-config", put(put_business_config)) .route("/api/bootstrap", post(bootstrap)) .route("/api/admin/session/revoke", post(revoke_admin_session)) @@ -768,7 +808,7 @@ mod tests { jmap_push_verification: SecretString::new("jmap-verification".into()), }, SecretString::new("worker-secret".into()), - MemoryState::default(), + MemoryState::enabled_for_tests(), HashSet::new(), ) } @@ -1038,6 +1078,62 @@ mod tests { assert_eq!(status, StatusCode::UNAUTHORIZED); } + #[tokio::test] + async fn disabled_gate_rejects_business_entry_without_side_effects() { + let app = router_with_state( + AuthSecrets { + reconcile_token: SecretString::new("r".into()), + telegram_webhook_secret: SecretString::new("t".into()), + jmap_push_verification: SecretString::new("p".into()), + }, + SecretString::new("w".into()), + MemoryState::default(), + HashSet::new(), + ); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/webhook/tg") + .header("x-telegram-bot-api-secret-token", "t") + .body(Body::from(r#"{"update_id":1}"#)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + } + + #[tokio::test] + async fn enabled_api_persists_toggle() { + let app = test_router(); + let response = app + .clone() + .oneshot( + Request::builder() + .method("PUT") + .uri("/api/enabled") + .header("authorization", "Bearer worker-secret") + .body(Body::from(r#"{"enabled":false}"#)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let response = app + .oneshot( + Request::builder() + .method("GET") + .uri("/api/enabled") + .header("authorization", "Bearer worker-secret") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + } + #[tokio::test] async fn outbound_config_api_is_authenticated_and_validated() { let app = test_router(); diff --git a/src/state.rs b/src/state.rs index d845cdf..10d9208 100644 --- a/src/state.rs +++ b/src/state.rs @@ -114,6 +114,9 @@ pub trait ReliableState: Send + Sync { async fn admin_session_valid(&self, digest: &str) -> Result; async fn revoke_admin_session(&self, digest: &str) -> Result<(), StateError>; async fn business_config_revision(&self) -> Result; + /// Global durable enable gate; missing/read failures are interpreted by callers as disabled. + async fn is_enabled(&self) -> Result; + async fn set_enabled(&self, enabled: bool) -> Result<(), StateError>; } pub struct RedisState { @@ -430,6 +433,25 @@ impl ReliableState for RedisState { .await?; Ok(value.unwrap_or(0)) } + + async fn is_enabled(&self) -> Result { + let mut connection = self.connection.clone(); + let value: Option = redis::cmd("GET") + .arg("config:enabled") + .query_async(&mut connection) + .await?; + Ok(value.unwrap_or(false)) + } + + async fn set_enabled(&self, enabled: bool) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + redis::cmd("SET") + .arg("config:enabled") + .arg(if enabled { "1" } else { "0" }) + .query_async::<()>(&mut connection) + .await?; + Ok(()) + } } fn stream_id_to_message(item: redis::streams::StreamId) -> Option { @@ -464,6 +486,15 @@ pub struct MemoryState { inner: Arc>, } +#[cfg(test)] +impl MemoryState { + pub fn enabled_for_tests() -> Self { + let state = Self::default(); + state.inner.lock().expect("new state lock").enabled = true; + state + } +} + #[derive(Default)] struct MemoryInner { dedup: HashSet, @@ -476,6 +507,7 @@ struct MemoryInner { business_config: Option, admin_session: Option<(String, i64)>, business_revision: u64, + enabled: bool, } #[async_trait] @@ -673,6 +705,15 @@ impl ReliableState for MemoryState { let inner = self.inner.lock().map_err(|_| StateError::Poisoned)?; Ok(inner.business_revision) } + + async fn is_enabled(&self) -> Result { + Ok(self.inner.lock().map_err(|_| StateError::Poisoned)?.enabled) + } + + async fn set_enabled(&self, enabled: bool) -> Result<(), StateError> { + self.inner.lock().map_err(|_| StateError::Poisoned)?.enabled = enabled; + Ok(()) + } } fn unix_now() -> i64 { diff --git a/web/config.js b/web/config.js index 314faf0..eff7a4d 100644 --- a/web/config.js +++ b/web/config.js @@ -44,6 +44,8 @@ const runtimeSaveButton = document.querySelector("#runtime-save-button"); const businessSaveState = document.querySelector("#business-save-state"); const runtimeSaveState = document.querySelector("#runtime-save-state"); + const serviceEnabled = document.querySelector("#service-enabled"); + const serviceEnabledState = document.querySelector("#service-enabled-state"); const sessionExpiry = document.querySelector("#session-expiry"); const llmEnabled = document.querySelector("#llm-enabled"); const llmApiKey = document.querySelector("#llm-api-key"); @@ -166,6 +168,12 @@ } } + async function loadEnabled() { + const state = await request("/api/enabled"); + serviceEnabled.checked = state.enabled === true; + serviceEnabledState.textContent = serviceEnabled.checked ? "业务入口已启用" : "业务入口已关闭（fail-closed）"; + } + function showSetupView(message, missing = []) { setupMode = true; setupMessage.textContent = message; @@ -516,6 +524,7 @@ setBusy(""); showNotice("success", "管理会话已创建。运行参数正在读取；业务配置需要完整重新填写。 "); await loadRuntimeConfig(); + await loadEnabled(); } catch (error) { bootstrapCredential = ""; setBusy(""); @@ -529,6 +538,23 @@ reloadButton.addEventListener("click", loadRuntimeConfig); + serviceEnabled.addEventListener("change", async () => { + if (!adminSession || busyAction) return; + const desired = serviceEnabled.checked; + serviceEnabled.disabled = true; + try { + await request("/api/enabled", "PUT", { enabled: desired }); + serviceEnabledState.textContent = desired ? "业务入口已启用" : "业务入口已关闭（fail-closed）"; + showNotice("success", desired ? "业务处理已启用。" : "业务处理已关闭，入口将返回未启用状态。 "); + } catch (error) { + serviceEnabled.checked = !desired; + serviceEnabledState.textContent = serviceEnabled.checked ? "业务入口已启用" : "业务入口已关闭（fail-closed）"; + showNotice("error", error.status === 401 ? "管理会话已失效，请重新授权。" : "无法保存业务开关状态。 "); + } finally { + serviceEnabled.disabled = false; + } + }); + logoutButton.addEventListener("click", async () => { if (!adminSession || busyAction) return; setBusy("logout"); diff --git a/web/index.html b/web/index.html index 2115dce..2cc4a8a 100644 --- a/web/index.html +++ b/web/index.html @@ -67,6 +67,7 @@
管理会话有效会话仅保存在当前页面内存
+