feat(config): add global Redis-backed enable switch
This commit is contained in:
parent
a5710331fa
commit
956ef6a1d3
|
|
@ -58,6 +58,9 @@ HostStack 使用根目录 `hoststack.yaml` 构建并运行 Rust 服务:执行
|
||||||
进行 30 秒间隔、5 秒超时的健康检查。请在 HostStack Secret 中注入 `REDIS_URL` 与
|
进行 30 秒间隔、5 秒超时的健康检查。请在 HostStack Secret 中注入 `REDIS_URL` 与
|
||||||
`CONFIG_ENCRYPTION_KEY`;密钥不写入 YAML、镜像或仓库文件。
|
`CONFIG_ENCRYPTION_KEY`;密钥不写入 YAML、镜像或仓库文件。
|
||||||
|
|
||||||
|
管理会话内的“启用业务处理”开关持久化在 Redis `config:enabled`;缺失或读取失败均按关闭处理。
|
||||||
|
关闭时 Webhook、Push、Reconcile 和 Worker 返回 HTTP 503,不确认或丢弃上游事件;SPA、状态和管理配置 API 仍可用。
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## 启动后怎么看它是否正常
|
## 启动后怎么看它是否正常
|
||||||
|
|
|
||||||
|
|
@ -90,6 +90,11 @@ HTTP 端口并以 `/healthz` 做 interval 30 秒、timeout 5 秒的健康检查
|
||||||
`CONFIG_ENCRYPTION_KEY` 必须配置为 HostStack Secret;不得将密钥值写入 YAML、镜像、日志
|
`CONFIG_ENCRYPTION_KEY` 必须配置为 HostStack Secret;不得将密钥值写入 YAML、镜像、日志
|
||||||
或代码仓库。现有 Dockerfile 部署方式仍受支持。
|
或代码仓库。现有 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,并在成功后热重建客户端。
|
> 迁移后生产环境仅需 `REDIS_URL` 与 `CONFIG_ENCRYPTION_KEY`;下表中的业务环境变量是历史阶段说明,不应再注入生产容器。业务密钥通过受保护的 `/api/bootstrap` 或管理员 PUT 写入 Redis,并在成功后热重建客户端。
|
||||||
|
|
||||||
(保留既有 Secrets 注入约定:禁入镜像、`secrecy` 包裹、日志屏蔽;新增运行模式相关变量)
|
(保留既有 Secrets 注入约定:禁入镜像、`secrecy` 包裹、日志屏蔽;新增运行模式相关变量)
|
||||||
|
|
|
||||||
|
|
@ -83,6 +83,9 @@ async fn telegram_webhook(
|
||||||
body: Bytes,
|
body: Bytes,
|
||||||
) -> Response {
|
) -> Response {
|
||||||
refresh_business_config(&app).await;
|
refresh_business_config(&app).await;
|
||||||
|
if !business_enabled(&app).await {
|
||||||
|
return (StatusCode::SERVICE_UNAVAILABLE, "service disabled").into_response();
|
||||||
|
}
|
||||||
let (auth, _, allowlist) = auth_snapshot(&app);
|
let (auth, _, allowlist) = auth_snapshot(&app);
|
||||||
if !header_value_matches(
|
if !header_value_matches(
|
||||||
&headers,
|
&headers,
|
||||||
|
|
@ -117,6 +120,9 @@ async fn telegram_webhook(
|
||||||
|
|
||||||
async fn reconcile(State(app): State<AppState>, headers: HeaderMap) -> Response {
|
async fn reconcile(State(app): State<AppState>, headers: HeaderMap) -> Response {
|
||||||
refresh_business_config(&app).await;
|
refresh_business_config(&app).await;
|
||||||
|
if !business_enabled(&app).await {
|
||||||
|
return (StatusCode::SERVICE_UNAVAILABLE, "service disabled").into_response();
|
||||||
|
}
|
||||||
let (auth, _, _) = auth_snapshot(&app);
|
let (auth, _, _) = auth_snapshot(&app);
|
||||||
let valid = headers
|
let valid = headers
|
||||||
.get(header::AUTHORIZATION)
|
.get(header::AUTHORIZATION)
|
||||||
|
|
@ -146,6 +152,9 @@ async fn reconcile(State(app): State<AppState>, headers: HeaderMap) -> Response
|
||||||
|
|
||||||
async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) -> Response {
|
async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) -> Response {
|
||||||
refresh_business_config(&app).await;
|
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 (_, worker_token, _) = auth_snapshot(&app);
|
||||||
let valid = headers
|
let valid = headers
|
||||||
.get(header::AUTHORIZATION)
|
.get(header::AUTHORIZATION)
|
||||||
|
|
@ -286,6 +295,33 @@ async fn put_config(State(app): State<AppState>, headers: HeaderMap, body: Bytes
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn get_enabled(State(app): State<AppState>, 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<AppState>, headers: HeaderMap, body: Bytes) -> Response {
|
||||||
|
if !config_authorized(&app, &headers).await {
|
||||||
|
return StatusCode::UNAUTHORIZED.into_response();
|
||||||
|
}
|
||||||
|
let Ok(payload) = serde_json::from_slice::<serde_json::Value>(&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
|
/// Replace the complete encrypted business configuration. Client construction is performed
|
||||||
/// before persistence and swap, so malformed/unreachable settings leave the old worker live.
|
/// before persistence and swap, so malformed/unreachable settings leave the old worker live.
|
||||||
async fn put_business_config(
|
async fn put_business_config(
|
||||||
|
|
@ -504,6 +540,9 @@ struct WorkerRequest {
|
||||||
|
|
||||||
async fn jmap_push(State(app): State<AppState>, body: Bytes) -> Response {
|
async fn jmap_push(State(app): State<AppState>, body: Bytes) -> Response {
|
||||||
refresh_business_config(&app).await;
|
refresh_business_config(&app).await;
|
||||||
|
if !business_enabled(&app).await {
|
||||||
|
return (StatusCode::SERVICE_UNAVAILABLE, "service disabled").into_response();
|
||||||
|
}
|
||||||
let (auth, _, _) = auth_snapshot(&app);
|
let (auth, _, _) = auth_snapshot(&app);
|
||||||
// Stalwart's PushSubscription callback carries verificationCode in its
|
// Stalwart's PushSubscription callback carries verificationCode in its
|
||||||
// JSON object. Check it before any future queue/Redis side effect.
|
// JSON object. Check it before any future queue/Redis side effect.
|
||||||
|
|
@ -730,6 +769,7 @@ fn router_with_worker_state_runtime_bootstrap_config<S: ReliableState + 'static>
|
||||||
.route("/reconcile", post(reconcile))
|
.route("/reconcile", post(reconcile))
|
||||||
.route("/worker", post(worker))
|
.route("/worker", post(worker))
|
||||||
.route("/api/config", get(get_config).put(put_config))
|
.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/business-config", put(put_business_config))
|
||||||
.route("/api/bootstrap", post(bootstrap))
|
.route("/api/bootstrap", post(bootstrap))
|
||||||
.route("/api/admin/session/revoke", post(revoke_admin_session))
|
.route("/api/admin/session/revoke", post(revoke_admin_session))
|
||||||
|
|
@ -768,7 +808,7 @@ mod tests {
|
||||||
jmap_push_verification: SecretString::new("jmap-verification".into()),
|
jmap_push_verification: SecretString::new("jmap-verification".into()),
|
||||||
},
|
},
|
||||||
SecretString::new("worker-secret".into()),
|
SecretString::new("worker-secret".into()),
|
||||||
MemoryState::default(),
|
MemoryState::enabled_for_tests(),
|
||||||
HashSet::new(),
|
HashSet::new(),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
@ -1038,6 +1078,62 @@ mod tests {
|
||||||
assert_eq!(status, StatusCode::UNAUTHORIZED);
|
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]
|
#[tokio::test]
|
||||||
async fn outbound_config_api_is_authenticated_and_validated() {
|
async fn outbound_config_api_is_authenticated_and_validated() {
|
||||||
let app = test_router();
|
let app = test_router();
|
||||||
|
|
|
||||||
41
src/state.rs
41
src/state.rs
|
|
@ -114,6 +114,9 @@ pub trait ReliableState: Send + Sync {
|
||||||
async fn admin_session_valid(&self, digest: &str) -> Result<bool, StateError>;
|
async fn admin_session_valid(&self, digest: &str) -> Result<bool, StateError>;
|
||||||
async fn revoke_admin_session(&self, digest: &str) -> Result<(), StateError>;
|
async fn revoke_admin_session(&self, digest: &str) -> Result<(), StateError>;
|
||||||
async fn business_config_revision(&self) -> Result<u64, StateError>;
|
async fn business_config_revision(&self) -> Result<u64, StateError>;
|
||||||
|
/// Global durable enable gate; missing/read failures are interpreted by callers as disabled.
|
||||||
|
async fn is_enabled(&self) -> Result<bool, StateError>;
|
||||||
|
async fn set_enabled(&self, enabled: bool) -> Result<(), StateError>;
|
||||||
}
|
}
|
||||||
|
|
||||||
pub struct RedisState {
|
pub struct RedisState {
|
||||||
|
|
@ -430,6 +433,25 @@ impl ReliableState for RedisState {
|
||||||
.await?;
|
.await?;
|
||||||
Ok(value.unwrap_or(0))
|
Ok(value.unwrap_or(0))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn is_enabled(&self) -> Result<bool, StateError> {
|
||||||
|
let mut connection = self.connection.clone();
|
||||||
|
let value: Option<bool> = 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<StreamMessage> {
|
fn stream_id_to_message(item: redis::streams::StreamId) -> Option<StreamMessage> {
|
||||||
|
|
@ -464,6 +486,15 @@ pub struct MemoryState {
|
||||||
inner: Arc<Mutex<MemoryInner>>,
|
inner: Arc<Mutex<MemoryInner>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[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)]
|
#[derive(Default)]
|
||||||
struct MemoryInner {
|
struct MemoryInner {
|
||||||
dedup: HashSet<String>,
|
dedup: HashSet<String>,
|
||||||
|
|
@ -476,6 +507,7 @@ struct MemoryInner {
|
||||||
business_config: Option<serde_json::Value>,
|
business_config: Option<serde_json::Value>,
|
||||||
admin_session: Option<(String, i64)>,
|
admin_session: Option<(String, i64)>,
|
||||||
business_revision: u64,
|
business_revision: u64,
|
||||||
|
enabled: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
|
|
@ -673,6 +705,15 @@ impl ReliableState for MemoryState {
|
||||||
let inner = self.inner.lock().map_err(|_| StateError::Poisoned)?;
|
let inner = self.inner.lock().map_err(|_| StateError::Poisoned)?;
|
||||||
Ok(inner.business_revision)
|
Ok(inner.business_revision)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn is_enabled(&self) -> Result<bool, StateError> {
|
||||||
|
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 {
|
fn unix_now() -> i64 {
|
||||||
|
|
|
||||||
|
|
@ -44,6 +44,8 @@
|
||||||
const runtimeSaveButton = document.querySelector("#runtime-save-button");
|
const runtimeSaveButton = document.querySelector("#runtime-save-button");
|
||||||
const businessSaveState = document.querySelector("#business-save-state");
|
const businessSaveState = document.querySelector("#business-save-state");
|
||||||
const runtimeSaveState = document.querySelector("#runtime-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 sessionExpiry = document.querySelector("#session-expiry");
|
||||||
const llmEnabled = document.querySelector("#llm-enabled");
|
const llmEnabled = document.querySelector("#llm-enabled");
|
||||||
const llmApiKey = document.querySelector("#llm-api-key");
|
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 = []) {
|
function showSetupView(message, missing = []) {
|
||||||
setupMode = true;
|
setupMode = true;
|
||||||
setupMessage.textContent = message;
|
setupMessage.textContent = message;
|
||||||
|
|
@ -516,6 +524,7 @@
|
||||||
setBusy("");
|
setBusy("");
|
||||||
showNotice("success", "管理会话已创建。运行参数正在读取;业务配置需要完整重新填写。 ");
|
showNotice("success", "管理会话已创建。运行参数正在读取;业务配置需要完整重新填写。 ");
|
||||||
await loadRuntimeConfig();
|
await loadRuntimeConfig();
|
||||||
|
await loadEnabled();
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
bootstrapCredential = "";
|
bootstrapCredential = "";
|
||||||
setBusy("");
|
setBusy("");
|
||||||
|
|
@ -529,6 +538,23 @@
|
||||||
|
|
||||||
reloadButton.addEventListener("click", loadRuntimeConfig);
|
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 () => {
|
logoutButton.addEventListener("click", async () => {
|
||||||
if (!adminSession || busyAction) return;
|
if (!adminSession || busyAction) return;
|
||||||
setBusy("logout");
|
setBusy("logout");
|
||||||
|
|
|
||||||
|
|
@ -67,6 +67,7 @@
|
||||||
<div class="panel config-toolbar">
|
<div class="panel config-toolbar">
|
||||||
<div class="session-state"><span class="status-pulse" aria-hidden="true"></span><div><strong>管理会话有效</strong><span id="session-expiry">会话仅保存在当前页面内存</span></div></div>
|
<div class="session-state"><span class="status-pulse" aria-hidden="true"></span><div><strong>管理会话有效</strong><span id="session-expiry">会话仅保存在当前页面内存</span></div></div>
|
||||||
<div class="toolbar-actions">
|
<div class="toolbar-actions">
|
||||||
|
<label class="toggle-card" for="service-enabled"><input id="service-enabled" type="checkbox" /><span><strong>启用业务处理</strong><small id="service-enabled-state">读取中…</small></span></label>
|
||||||
<button id="reload-button" class="button button-quiet" type="button">重新读取运行参数</button>
|
<button id="reload-button" class="button button-quiet" type="button">重新读取运行参数</button>
|
||||||
<button id="logout-button" class="button button-quiet" type="button">退出并清除会话</button>
|
<button id="logout-button" class="button button-quiet" type="button">退出并清除会话</button>
|
||||||
</div>
|
</div>
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue