diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..02fc553 --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,42 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + verify: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + with: + toolchain: '1.92' + components: rustfmt, clippy + + - name: Cache Rust build + uses: Swatinem/rust-cache@v2 + + - name: Check Rust formatting + run: cargo fmt --all -- --check + + - name: Run Clippy + run: cargo clippy --all-targets -- -D warnings + + - name: Run Rust tests + run: cargo test --all-targets + + - name: Install Node.js + uses: actions/setup-node@v4 + with: + node-version: '22' + cache: npm + cache-dependency-path: frontend/package-lock.json + + - name: Build frontend + working-directory: frontend + run: npm ci && npm run build diff --git a/docker/.env.production.example b/docker/.env.production.example index 8709e74..851d372 100644 --- a/docker/.env.production.example +++ b/docker/.env.production.example @@ -7,7 +7,7 @@ DEBIAN_MIRROR=http://deb.debian.org/debian CARGO_REGISTRY_MIRROR= # Public listener and URL -IMAGEFORGE_BIND_ADDRESS=0.0.0.0 +IMAGEFORGE_BIND_ADDRESS=127.0.0.1 IMAGEFORGE_PORT=8080 PUBLIC_BASE_URL=http://192.0.2.10:8080 diff --git a/docker/docker-compose.dev.yml b/docker/docker-compose.dev.yml index 1a297dc..7d3c952 100644 --- a/docker/docker-compose.dev.yml +++ b/docker/docker-compose.dev.yml @@ -1,5 +1,3 @@ -version: '3.8' - services: postgres: image: postgres:16-alpine @@ -8,14 +6,14 @@ services: POSTGRES_PASSWORD: devpassword POSTGRES_DB: imageforge ports: - - "5432:5432" + - "127.0.0.1:5432:5432" volumes: - postgres_data:/var/lib/postgresql/data redis: image: redis:7-alpine ports: - - "6379:6379" + - "127.0.0.1:6379:6379" volumes: - redis_data:/data diff --git a/docker/docker-compose.prod.yml b/docker/docker-compose.prod.yml index 2f236c4..994646e 100644 --- a/docker/docker-compose.prod.yml +++ b/docker/docker-compose.prod.yml @@ -83,7 +83,7 @@ services: MAIL_SMTP_PORT: "${MAIL_SMTP_PORT:-}" MAIL_SMTP_ENCRYPTION: "${MAIL_SMTP_ENCRYPTION:-}" ports: - - "${IMAGEFORGE_BIND_ADDRESS:-0.0.0.0}:${IMAGEFORGE_PORT:-8080}:8080" + - "${IMAGEFORGE_BIND_ADDRESS:-127.0.0.1}:${IMAGEFORGE_PORT:-8080}:8080" volumes: - uploads:/app/uploads tmpfs: diff --git a/docker/nginx.conf b/docker/nginx.conf index 82fd96d..8ba5a42 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -22,6 +22,12 @@ http { listen 80; server_name _; + add_header X-Content-Type-Options "nosniff" always; + add_header X-Frame-Options "SAMEORIGIN" always; + add_header Referrer-Policy "strict-origin-when-cross-origin" always; + add_header Permissions-Policy "camera=(), microphone=(), geolocation=()" always; + add_header Content-Security-Policy "default-src 'self'; base-uri 'self'; object-src 'none'; frame-ancestors 'self'; form-action 'self'; img-src 'self' data: blob: https:; font-src 'self' data:; style-src 'self' 'unsafe-inline'; script-src 'self'; connect-src 'self' https:" always; + root /usr/share/nginx/html; location /api/ { @@ -44,6 +50,11 @@ http { proxy_pass http://imageforge_api; } + # Prometheus should scrape the API container directly on the private network. + location = /metrics { + return 404; + } + # SPA fallback location / { try_files $uri $uri/ /index.html; diff --git a/docs/deployment.md b/docs/deployment.md index 3c63ace..d320333 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -44,6 +44,7 @@ API 健康后 Worker 才会启动，避免两个进程在首次部署时同时 ```bash docker compose --env-file .env.production -f docker/docker-compose.prod.yml ps curl --fail http://127.0.0.1:8080/health +curl --fail http://127.0.0.1:8080/metrics ``` 预期健康响应： @@ -66,7 +67,7 @@ docker compose --env-file .env.production -f docker/docker-compose.prod.yml up - ### 反向代理 -直接通过服务器地址访问时保持 `TRUST_PROXY_HEADERS=false`。只有当 8080 端口不对客户端开放、所有请求都经过可信反向代理时，才设置为 `true`，并由代理覆盖 `X-Forwarded-For` 与 `X-Forwarded-Proto`。 +生产 Compose 默认把 API 端口绑定到 `127.0.0.1`。若确实需要绕过反向代理直接通过服务器地址访问，显式设置 `IMAGEFORGE_BIND_ADDRESS=0.0.0.0`，同时保持 `TRUST_PROXY_HEADERS=false` 并配置主机防火墙。只有当 API 端口不对客户端开放、所有请求都经过可信反向代理时，才设置 `TRUST_PROXY_HEADERS=true`，并由代理覆盖 `X-Forwarded-For` 与 `X-Forwarded-Proto`。 代理至少需要： @@ -82,6 +83,19 @@ location / { } ``` +`/metrics` 包含运行状态与队列数据，不应通过公开域名暴露。生产 Nginx 应单独拒绝该路径，Prometheus 直接抓取只绑定回环地址的 API 端口： + +```nginx +location = /metrics { + allow 127.0.0.1; + allow ::1; + deny all; + proxy_pass http://127.0.0.1:8080; +} +``` + +应用会为所有响应设置 CSP、`X-Content-Type-Options`、`X-Frame-Options`、`Referrer-Policy` 和 `Permissions-Policy`。TLS 网关还应设置 `Strict-Transport-Security`。 + ### 日志与备份 ```bash diff --git a/docs/observability.md b/docs/observability.md index 6c22cc5..dc84f3f 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -1,16 +1,15 @@ -# 可观测性设计（日志/指标/追踪）- ImageForge +# 可观测性与告警 - ImageForge -目标：让“压缩效果、性能瓶颈、队列健康、计费正确性、滥用风险”都能被观测与告警，便于商用运营。 +目标：让“压缩效果、性能瓶颈、队列健康、计费正确性、滥用风险”都能被观测与告警，便于商用运营。下列请求标识和基础 Prometheus 指标已经实现；OpenTelemetry 和业务仪表板仍属于后续增强项。 --- ## 1. 统一规范 ### 1.1 请求标识 -- 每个 HTTP 请求生成 `request_id`（或从网关透传），写入： - - 响应头：`X-Request-Id` - - 日志字段：`request_id` - - Trace：`trace_id/span_id`（如启用 OpenTelemetry） +- API 会生成 `req_`，也会接受由可信网关透传的安全 `X-Request-Id`。 +- 请求 ID 会写入全部响应的 `X-Request-Id`、成功/失败请求日志和 JSON 错误体。 +- 传入值仅允许 1-128 个 ASCII 字母、数字、点、下划线、冒号和连字符，避免日志注入。 ### 1.2 日志格式 - 结构化日志（JSON）优先，便于 Loki/ELK 聚合。 @@ -25,40 +24,49 @@ --- -## 2. 指标（Prometheus） +## 2. 指标（Prometheus，已实现） + +API 在 `/metrics` 暴露 Prometheus 文本格式。生产环境只应从宿主机或监控私网抓取，不要通过公开域名开放该路径： + +```bash +curl --fail http://127.0.0.1:18180/metrics +``` + +压缩、S3 回退和死信累计值存放在 Redis Hash `metrics:imageforge`，因此 API 与独立 Worker 的事件会汇总到同一组指标。HTTP 请求与错误指标是 API 进程级指标，重启后归零。 ### 2.1 API 服务指标 请求类： -- `http_requests_total{route,method,status}` -- `http_request_duration_seconds_bucket{route,method}` +- `imageforge_http_requests_total{method,status_class}` +- `imageforge_http_request_duration_seconds_bucket` -鉴权与风控： -- `auth_fail_total{reason}` -- `rate_limited_total{scope}`（anonymous/user/api_key） -- `quota_exceeded_total{plan}` +错误与风控： +- `imageforge_errors_total{code}`，包含 `RATE_LIMITED`、`QUOTA_EXCEEDED` 等业务错误码 -计费链路： -- `billing_webhook_total{provider,event_type,result}` -- `subscription_state_total{state}` -- `invoice_total{status}` +依赖与队列： +- `imageforge_dependency_up{dependency="database|redis"}` +- `imageforge_active_tasks` +- `imageforge_queue_messages{state="stream|pending|dead_letter"}` ### 2.2 Worker 指标 队列与吞吐： -- `jobs_received_total` -- `jobs_inflight` -- `jobs_completed_total{result}` -- `job_duration_seconds_bucket{format,level}` +- `imageforge_compressions_total{result}` +- `imageforge_compression_duration_seconds_sum/count` 压缩效果： -- `bytes_in_total`、`bytes_out_total`、`bytes_saved_total` -- `compression_ratio_bucket{format,level}` +- `imageforge_compression_bytes_total{direction="input|output"}` 资源与异常： -- `decode_failed_total{reason}` -- `pixel_limit_hit_total` +- `imageforge_storage_fallbacks_total` +- `imageforge_dead_letters_total` -### 2.3 Redis/队列指标（可选） -- Streams 消费延迟、pending 数量、dead-letter 数量（如实现）。 +Prometheus 抓取示例： + +```yaml +scrape_configs: + - job_name: imageforge + static_configs: + - targets: ['127.0.0.1:18180'] +``` --- diff --git a/docs/security.md b/docs/security.md index aaa1ec7..a3d8286 100644 --- a/docs/security.md +++ b/docs/security.md @@ -55,6 +55,12 @@ - 支持禁用/轮换；可选 IP 白名单（Business/V1+）。 - 每次请求记录 `last_used_at/last_used_ip/user_agent`（审计）。 +### 3.4 运行时策略 +- 管理后台 `system_config` 中的 `features`、`rate_limits`、`file_limits` 会在最多 5 秒缓存后生效，无需重启服务。 +- 功能开关可控制注册、匿名上传和 API Key 的创建及使用；环境变量 `ALLOW_ANONYMOUS_UPLOAD=false` 是不可被后台重新开启的上层硬限制。 +- 限速配置覆盖匿名、登录用户、API Key、登录、注册、邮件验证和密码重置入口；API Key 自身限制与全局限制取较小值。 +- 配置更新会校验数值边界并写入 `audit_logs`，审计记录不包含密钥明文。 + --- ## 4. 上传与图片处理安全 diff --git a/frontend/index.html b/frontend/index.html index 2cbe70b..c51f0c7 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -2,7 +2,7 @@ - + ImageForge - 图片压缩 diff --git a/frontend/public/favicon.svg b/frontend/public/favicon.svg new file mode 100644 index 0000000..ef3e4e3 --- /dev/null +++ b/frontend/public/favicon.svg @@ -0,0 +1,12 @@ + + + + + + + + + + + + diff --git a/frontend/public/vite.svg b/frontend/public/vite.svg deleted file mode 100644 index e7b8dfb..0000000 --- a/frontend/public/vite.svg +++ /dev/null @@ -1 +0,0 @@ - \ No newline at end of file diff --git a/frontend/src/components/HelloWorld.vue b/frontend/src/components/HelloWorld.vue deleted file mode 100644 index b58e52b..0000000 --- a/frontend/src/components/HelloWorld.vue +++ /dev/null @@ -1,41 +0,0 @@ - - - - - diff --git a/migrations/009_runtime_policy_defaults.sql b/migrations/009_runtime_policy_defaults.sql new file mode 100644 index 0000000..501254a --- /dev/null +++ b/migrations/009_runtime_policy_defaults.sql @@ -0,0 +1,18 @@ +UPDATE system_config +SET value = '{ + "anonymous_per_minute": 10, + "anonymous_units_per_day": 10, + "user_per_minute": 60, + "api_key_per_minute": 100, + "login_ip_per_5_minutes": 30, + "login_identity_per_5_minutes": 10, + "register_ip_per_hour": 10, + "verification_email_per_minute": 1, + "email_verify_ip_per_15_minutes": 20, + "forgot_password_ip_per_15_minutes": 5, + "forgot_password_email_per_15_minutes": 3, + "password_reset_ip_per_15_minutes": 10, + "password_reset_token_per_15_minutes": 5 +}'::jsonb || value, + updated_at = NOW() +WHERE key = 'rate_limits'; diff --git a/src/api/admin.rs b/src/api/admin.rs index 8d9b96d..1ed5714 100644 --- a/src/api/admin.rs +++ b/src/api/admin.rs @@ -1298,6 +1298,7 @@ async fn update_stripe_config( Some(admin_id), ) .await?; + audit_config_action(&state, admin_id, "stripe", ip).await?; Ok(Json(Envelope { success: true, @@ -1363,6 +1364,7 @@ async fn update_auth_config( Some(admin_id), ) .await?; + audit_config_action(&state, admin_id, "auth", ip).await?; Ok(Json(Envelope { success: true, @@ -1520,6 +1522,7 @@ async fn update_mail_config( Some(admin_id), ) .await?; + audit_config_action(&state, admin_id, "mail", ip).await?; Ok(Json(Envelope { success: true, @@ -1652,6 +1655,10 @@ async fn update_config( if key.is_empty() { return Err(AppError::new(ErrorCode::InvalidRequest, "key 不能为空")); } + if key.len() > 100 { + return Err(AppError::new(ErrorCode::InvalidRequest, "key 过长")); + } + settings::validate_runtime_config_value(key, &req.value)?; let row = sqlx::query_as::<_, ConfigRow>( r#" @@ -1673,12 +1680,38 @@ async fn update_config( .await .map_err(|err| AppError::new(ErrorCode::Internal, "更新配置失败").with_source(err))?; + if matches!(key, "auth" | "features" | "rate_limits" | "file_limits") { + state.runtime_policy_cache.invalidate().await; + } + audit_config_action(&state, admin_id, key, ip).await?; + Ok(Json(Envelope { success: true, data: row, })) } +async fn audit_config_action( + state: &AppState, + admin_id: Uuid, + key: &str, + ip: IpAddr, +) -> Result<(), AppError> { + sqlx::query( + r#" + INSERT INTO audit_logs (user_id, action, resource_type, details, ip_address) + VALUES ($1, 'system_config_update', 'system_config', $2, $3::inet) + "#, + ) + .bind(admin_id) + .bind(serde_json::json!({ "key": key })) + .bind(ip.to_string()) + .execute(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "写入配置审计日志失败").with_source(err))?; + Ok(()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/api/auth.rs b/src/api/auth.rs index 1211210..29654d7 100644 --- a/src/api/auth.rs +++ b/src/api/auth.rs @@ -81,11 +81,18 @@ async fn register( Json(req): Json, ) -> Result>, AppError> { let ip = context::client_ip(&headers, addr.ip()); + let policy = settings::runtime_policy(&state).await?; + if !policy.features.registration_enabled { + return Err(AppError::new( + ErrorCode::Forbidden, + "用户注册功能当前已关闭", + )); + } rate_limit::enforce( &state, "auth_register_ip", &ip.to_string(), - 10, + policy.rate_limits.register_ip_per_hour, 60 * 60, "注册请求过于频繁，请稍后再试", ) @@ -96,7 +103,7 @@ async fn register( credentials::validate_password(&req.password)?; let password_hash = credentials::hash_password(&req.password).await?; - let verification_required = settings::email_verification_required(&state).await?; + let verification_required = policy.auth.email_verification_required; let verified_at = (!verification_required).then(Utc::now); let user = sqlx::query_as::<_, UserRow>( @@ -199,11 +206,12 @@ async fn login( } let ip = context::client_ip(&headers, addr.ip()); + let policy = settings::runtime_policy(&state).await?; rate_limit::enforce( &state, "auth_login_ip", &ip.to_string(), - 30, + policy.rate_limits.login_ip_per_5_minutes, 5 * 60, "登录请求过于频繁，请稍后再试", ) @@ -212,7 +220,7 @@ async fn login( &state, "auth_login_identity", &identity.to_lowercase(), - 10, + policy.rate_limits.login_identity_per_5_minutes, 5 * 60, "该账号登录尝试过于频繁，请稍后再试", ) @@ -269,7 +277,7 @@ async fn login( if !credentials::verify_password(&req.password, &user.password_hash).await? { return Err(AppError::new(ErrorCode::Unauthorized, "账号或密码错误")); } - let verification_required = settings::email_verification_required(&state).await?; + let verification_required = policy.auth.email_verification_required; let (token, expires_at) = auth::issue_jwt( &state.config.jwt_secret, @@ -305,8 +313,9 @@ async fn send_verification( headers: HeaderMap, ) -> Result>, AppError> { let claims = auth::require_jwt(&state.config.jwt_secret, &headers)?; + let policy = settings::runtime_policy(&state).await?; - if !settings::email_verification_required(&state).await? { + if !policy.auth.email_verification_required { return Ok(Json(Envelope { success: true, data: MessageResponse { @@ -319,7 +328,7 @@ async fn send_verification( &state, "auth_send_verification_user", &claims.sub.to_string(), - 1, + policy.rate_limits.verification_email_per_minute, 60, "发送过于频繁，请稍后再试", ) @@ -414,11 +423,12 @@ async fn verify_email( } let ip = context::client_ip(&headers, addr.ip()); + let policy = settings::runtime_policy(&state).await?; rate_limit::enforce( &state, "auth_verify_email_ip", &ip.to_string(), - 20, + policy.rate_limits.email_verify_ip_per_15_minutes, 15 * 60, "验证请求过于频繁，请稍后再试", ) @@ -484,11 +494,12 @@ async fn forgot_password( credentials::validate_email(&req.email)?; let ip = context::client_ip(&headers, addr.ip()); + let policy = settings::runtime_policy(&state).await?; rate_limit::enforce( &state, "auth_forgot_ip", &ip.to_string(), - 5, + policy.rate_limits.forgot_password_ip_per_15_minutes, 15 * 60, "找回密码请求过于频繁，请稍后再试", ) @@ -497,7 +508,7 @@ async fn forgot_password( &state, "auth_forgot_email", &req.email.to_lowercase(), - 3, + policy.rate_limits.forgot_password_email_per_15_minutes, 15 * 60, "找回密码请求过于频繁，请稍后再试", ) @@ -575,11 +586,12 @@ async fn reset_password( credentials::validate_password(&req.new_password)?; let ip = context::client_ip(&headers, addr.ip()); + let policy = settings::runtime_policy(&state).await?; rate_limit::enforce( &state, "auth_reset_ip", &ip.to_string(), - 10, + policy.rate_limits.password_reset_ip_per_15_minutes, 15 * 60, "重置密码请求过于频繁，请稍后再试", ) @@ -588,7 +600,7 @@ async fn reset_password( &state, "auth_reset_token", &req.token, - 5, + policy.rate_limits.password_reset_token_per_15_minutes, 15 * 60, "该重置链接尝试次数过多，请重新申请", ) diff --git a/src/api/compress.rs b/src/api/compress.rs index 280faef..7bc4996 100644 --- a/src/api/compress.rs +++ b/src/api/compress.rs @@ -111,6 +111,7 @@ async fn compress_json( let ip = context::client_ip(&headers, addr.ip()); let (jar, principal) = context::authenticate(&state, jar, &headers, ip).await?; context::require_api_permission(&principal, &["compress"])?; + context::enforce_anonymous_upload_rate(&state, &principal, ip).await?; let admission = prepare_single_admission(&state, &principal, ip, true).await?; let mut req = parse_single_file_request( diff --git a/src/api/context.rs b/src/api/context.rs index 8aacb84..157ab6a 100644 --- a/src/api/context.rs +++ b/src/api/context.rs @@ -1,5 +1,6 @@ use crate::auth; use crate::error::{AppError, ErrorCode}; +use crate::services::{rate_limit, settings}; use crate::state::AppState; use axum::http::HeaderMap; @@ -101,7 +102,8 @@ pub async fn authenticate( return Ok((jar, principal)); } - if !state.config.allow_anonymous_upload { + let policy = settings::runtime_policy(state).await?; + if !policy.features.anonymous_upload_enabled { return Err(AppError::new(ErrorCode::Unauthorized, "未登录")); } @@ -115,6 +117,39 @@ pub async fn authenticate( Ok((jar, Principal::Anonymous { session_id })) } +pub async fn enforce_anonymous_upload_rate( + state: &AppState, + principal: &Principal, + ip: IpAddr, +) -> Result<(), AppError> { + let Principal::Anonymous { session_id } = principal else { + return Ok(()); + }; + + let limit = settings::runtime_policy(state) + .await? + .rate_limits + .anonymous_per_minute; + rate_limit::enforce( + state, + "anonymous_upload_session", + session_id, + limit, + 60, + "匿名上传请求过于频繁，请稍后再试", + ) + .await?; + rate_limit::enforce( + state, + "anonymous_upload_ip", + &ip.to_string(), + limit, + 60, + "匿名上传请求过于频繁，请稍后再试", + ) + .await +} + async fn try_jwt(state: &AppState, headers: &HeaderMap) -> Result, AppError> { let auth_header = headers .get(axum::http::header::AUTHORIZATION) @@ -134,11 +169,13 @@ async fn try_jwt(state: &AppState, headers: &HeaderMap) -> Result>, token_version: i32, + rate_limit_override: Option, } let user = sqlx::query_as::<_, UserAuthRow>( r#" - SELECT id, role::text AS role, is_active, email_verified_at, token_version + SELECT id, role::text AS role, is_active, email_verified_at, token_version, + rate_limit_override FROM users WHERE id = $1 "#, @@ -159,13 +196,28 @@ async fn try_jwt(state: &AppState, headers: &HeaderMap) -> Result 0) + .map(|limit| limit as u32) + .unwrap_or(policy.rate_limits.user_per_minute) + .clamp(1, 100_000); + rate_limit::enforce( + state, + "user", + &user.id.to_string(), + request_limit, + 60, + "账号请求频率已超过限制", + ) + .await?; Ok(Some(Principal::User { user_id: user.id, role: user.role, - email_verified: user.email_verified_at.is_some() || !verification_required, + email_verified: user.email_verified_at.is_some() + || !policy.auth.email_verification_required, })) } @@ -185,6 +237,14 @@ async fn try_api_key( return Ok(None); }; + let policy = settings::runtime_policy(state).await?; + if !policy.features.api_key_enabled { + return Err(AppError::new( + ErrorCode::Forbidden, + "API Key 功能当前已关闭", + )); + } + let key_prefix = full_key .get(0..16) .ok_or_else(|| AppError::new(ErrorCode::Unauthorized, "API Key 格式错误"))?; @@ -234,11 +294,11 @@ async fn try_api_key( return Err(AppError::new(ErrorCode::Unauthorized, "API Key 无效")); } - crate::services::rate_limit::enforce( + rate_limit::enforce( state, "api_key", &row.id.to_string(), - row.rate_limit.clamp(1, 100_000) as u32, + (row.rate_limit.clamp(1, 100_000) as u32).min(policy.rate_limits.api_key_per_minute), 60, "API Key 请求频率已超过限制", ) @@ -264,14 +324,11 @@ async fn try_api_key( .execute(&state.db) .await; - let verification_required = - crate::services::settings::email_verification_required(state).await?; - Ok(Some(Principal::ApiKey { user_id: row.user_id, api_key_id: row.id, role: row.user_role, - email_verified: row.email_verified_at.is_some() || !verification_required, + email_verified: row.email_verified_at.is_some() || !policy.auth.email_verification_required, permissions, })) } diff --git a/src/api/metrics.rs b/src/api/metrics.rs new file mode 100644 index 0000000..d583777 --- /dev/null +++ b/src/api/metrics.rs @@ -0,0 +1,155 @@ +use crate::services::metrics::{ + self, CLUSTER_METRICS_KEY, DEAD_STREAM_KEY, QUEUE_GROUP_NAME, QUEUE_STREAM_KEY, +}; +use crate::state::AppState; + +use axum::extract::State; +use axum::http::header::{CACHE_CONTROL, CONTENT_TYPE}; +use axum::response::IntoResponse; +use redis::streams::StreamPendingReply; +use redis::AsyncCommands; +use std::collections::HashMap; +use std::fmt::Write; +use std::time::Duration; + +const SCRAPE_TIMEOUT: Duration = Duration::from_secs(2); + +pub async fn metrics(State(state): State) -> impl IntoResponse { + let database = tokio::time::timeout( + SCRAPE_TIMEOUT, + sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM tasks WHERE status IN ('pending', 'processing')", + ) + .fetch_one(&state.db), + ); + let redis = tokio::time::timeout(SCRAPE_TIMEOUT, redis_queue_stats(state.redis.clone())); + let (database, redis) = tokio::join!(database, redis); + + let (database_up, active_tasks) = match database { + Ok(Ok(value)) => (1, value), + _ => (0, 0), + }; + let (redis_up, queue_length, pending, dead_length, cluster) = match redis { + Ok(Ok((queue_length, pending, dead_length, cluster))) => { + (1, queue_length, pending, dead_length, cluster) + } + _ => (0, 0, 0, 0, HashMap::new()), + }; + + let mut output = metrics::render(); + output + .push_str("# HELP imageforge_dependency_up Whether a required dependency is reachable.\n"); + output.push_str("# TYPE imageforge_dependency_up gauge\n"); + let _ = writeln!( + output, + "imageforge_dependency_up{{dependency=\"database\"}} {database_up}" + ); + let _ = writeln!( + output, + "imageforge_dependency_up{{dependency=\"redis\"}} {redis_up}" + ); + output.push_str("# HELP imageforge_active_tasks Current pending or processing tasks.\n"); + output.push_str("# TYPE imageforge_active_tasks gauge\n"); + let _ = writeln!(output, "imageforge_active_tasks {active_tasks}"); + output.push_str("# HELP imageforge_queue_messages Current Redis stream message counts.\n"); + output.push_str("# TYPE imageforge_queue_messages gauge\n"); + let _ = writeln!( + output, + "imageforge_queue_messages{{state=\"stream\"}} {queue_length}" + ); + let _ = writeln!( + output, + "imageforge_queue_messages{{state=\"pending\"}} {pending}" + ); + let _ = writeln!( + output, + "imageforge_queue_messages{{state=\"dead_letter\"}} {dead_length}" + ); + if redis_up == 1 { + render_cluster_counters(&mut output, &cluster); + } + + ( + [ + (CONTENT_TYPE, "text/plain; version=0.0.4; charset=utf-8"), + (CACHE_CONTROL, "no-store"), + ], + output, + ) +} + +async fn redis_queue_stats( + mut connection: redis::aio::ConnectionManager, +) -> Result<(i64, usize, i64, HashMap), redis::RedisError> { + let queue_length: i64 = connection.xlen(QUEUE_STREAM_KEY).await?; + let pending = match connection + .xpending::<_, _, StreamPendingReply>(QUEUE_STREAM_KEY, QUEUE_GROUP_NAME) + .await + { + Ok(reply) => reply.count(), + Err(err) if err.to_string().contains("NOGROUP") => 0, + Err(err) => return Err(err), + }; + let dead_length: i64 = connection.xlen(DEAD_STREAM_KEY).await?; + let cluster: HashMap = connection.hgetall(CLUSTER_METRICS_KEY).await?; + Ok((queue_length, pending, dead_length, cluster)) +} + +fn render_cluster_counters(output: &mut String, counters: &HashMap) { + let value = |name: &str| counters.get(name).copied().unwrap_or(0).max(0); + let success = value("compression_success"); + let failed = value("compression_failed"); + let duration_seconds = value("compression_duration_micros") as f64 / 1_000_000.0; + + output.push_str("# HELP imageforge_compressions_total Compression attempts by result.\n"); + output.push_str("# TYPE imageforge_compressions_total counter\n"); + let _ = writeln!( + output, + "imageforge_compressions_total{{result=\"success\"}} {success}" + ); + let _ = writeln!( + output, + "imageforge_compressions_total{{result=\"failed\"}} {failed}" + ); + output.push_str( + "# HELP imageforge_compression_bytes_total Image bytes processed by direction.\n", + ); + output.push_str("# TYPE imageforge_compression_bytes_total counter\n"); + let _ = writeln!( + output, + "imageforge_compression_bytes_total{{direction=\"input\"}} {}", + value("compression_bytes_in") + ); + let _ = writeln!( + output, + "imageforge_compression_bytes_total{{direction=\"output\"}} {}", + value("compression_bytes_out") + ); + output.push_str("# HELP imageforge_compression_duration_seconds Total compression time.\n"); + output.push_str("# TYPE imageforge_compression_duration_seconds summary\n"); + let _ = writeln!( + output, + "imageforge_compression_duration_seconds_sum {duration_seconds}" + ); + let _ = writeln!( + output, + "imageforge_compression_duration_seconds_count {}", + success + failed + ); + output.push_str( + "# HELP imageforge_storage_fallbacks_total S3 writes that fell back to local storage.\n", + ); + output.push_str("# TYPE imageforge_storage_fallbacks_total counter\n"); + let _ = writeln!( + output, + "imageforge_storage_fallbacks_total {}", + value("storage_fallbacks") + ); + output.push_str("# HELP imageforge_dead_letters_total Jobs moved to the dead-letter stream.\n"); + output.push_str("# TYPE imageforge_dead_letters_total counter\n"); + let _ = writeln!( + output, + "imageforge_dead_letters_total {}", + value("dead_letters") + ); +} diff --git a/src/api/mod.rs b/src/api/mod.rs index 10b06b8..756ae02 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -7,8 +7,10 @@ mod context; mod downloads; mod envelope; mod health; +mod metrics; mod multipart; mod redemption; +pub(crate) mod request_context; mod response; mod tasks; mod user; @@ -21,7 +23,6 @@ use axum::extract::DefaultBodyLimit; use axum::Router; use std::net::SocketAddr; use tower_http::services::{ServeDir, ServeFile}; -use tower_http::trace::TraceLayer; pub async fn run(state: AppState) -> Result<(), AppError> { let addr = format!("{}:{}", state.config.host, state.config.port); @@ -36,10 +37,11 @@ pub async fn run(state: AppState) -> Result<(), AppError> { let app = Router::new() .route("/health", axum::routing::get(health::health)) + .route("/metrics", axum::routing::get(metrics::metrics)) .nest("/downloads", downloads::router()) .nest("/api/v1", v1) .fallback_service(static_service) - .layer(TraceLayer::new_for_http()) + .layer(axum::middleware::from_fn(request_context::middleware)) .with_state(state); let listener = tokio::net::TcpListener::bind(&addr) diff --git a/src/api/request_context.rs b/src/api/request_context.rs new file mode 100644 index 0000000..6cc4a32 --- /dev/null +++ b/src/api/request_context.rs @@ -0,0 +1,125 @@ +use crate::services::metrics; + +use axum::extract::Request; +use axum::http::HeaderValue; +use axum::middleware::Next; +use axum::response::Response; +use std::time::Instant; +use tracing::Instrument; +use uuid::Uuid; + +tokio::task_local! { + static REQUEST_ID: String; +} + +pub(crate) fn current_request_id() -> Option { + REQUEST_ID.try_with(Clone::clone).ok() +} + +pub(crate) async fn middleware(mut request: Request, next: Next) -> Response { + let request_id = request + .headers() + .get("x-request-id") + .and_then(|value| value.to_str().ok()) + .filter(|value| valid_request_id(value)) + .map(ToOwned::to_owned) + .unwrap_or_else(new_request_id); + let request_id_header = HeaderValue::from_str(&request_id) + .unwrap_or_else(|_| HeaderValue::from_static("invalid-request-id")); + request + .headers_mut() + .insert("x-request-id", request_id_header.clone()); + + let method = request.method().clone(); + let path = request.uri().path().to_string(); + let started = Instant::now(); + let span = tracing::info_span!( + "http_request", + request_id = %request_id, + method = %method, + path = %path, + ); + + let mut response = REQUEST_ID + .scope(request_id.clone(), next.run(request).instrument(span)) + .await; + let status = response.status(); + let elapsed = started.elapsed(); + + if path != "/metrics" { + metrics::record_http(method.as_str(), status.as_u16(), elapsed); + } + if path == "/health" || path == "/metrics" { + tracing::debug!( + request_id = %request_id, + method = %method, + path = %path, + status = status.as_u16(), + latency_ms = elapsed.as_millis(), + "HTTP request completed" + ); + } else { + tracing::info!( + request_id = %request_id, + method = %method, + path = %path, + status = status.as_u16(), + latency_ms = elapsed.as_millis(), + "HTTP request completed" + ); + } + + let headers = response.headers_mut(); + headers.insert("x-request-id", request_id_header); + headers.insert( + "x-content-type-options", + HeaderValue::from_static("nosniff"), + ); + headers.insert("x-frame-options", HeaderValue::from_static("SAMEORIGIN")); + headers.insert( + "referrer-policy", + HeaderValue::from_static("strict-origin-when-cross-origin"), + ); + headers.insert( + "permissions-policy", + HeaderValue::from_static("camera=(), microphone=(), geolocation=()"), + ); + headers.insert( + "content-security-policy", + HeaderValue::from_static( + "default-src 'self'; base-uri 'self'; object-src 'none'; frame-ancestors 'self'; form-action 'self'; img-src 'self' data: blob: https:; font-src 'self' data:; style-src 'self' 'unsafe-inline'; script-src 'self'; connect-src 'self' https:", + ), + ); + + response +} + +fn new_request_id() -> String { + format!("req_{}", Uuid::new_v4()) +} + +fn valid_request_id(value: &str) -> bool { + !value.is_empty() + && value.len() <= 128 + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b':' | b'-')) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn accepts_safe_gateway_request_ids() { + assert!(valid_request_id("req_1234-abcd.trace:01")); + } + + #[test] + fn rejects_control_characters_and_oversized_ids() { + assert!(!valid_request_id("")); + assert!(!valid_request_id("request id")); + assert!(!valid_request_id("request\nspoof")); + assert!(!valid_request_id(&"a".repeat(129))); + } +} diff --git a/src/api/tasks.rs b/src/api/tasks.rs index afbfef1..9c56479 100644 --- a/src/api/tasks.rs +++ b/src/api/tasks.rs @@ -93,6 +93,7 @@ async fn create_batch_task( let ip = context::client_ip(&headers, addr.ip()); let (jar, principal) = context::authenticate(&state, jar, &headers, ip).await?; context::require_api_permission(&principal, &["compress", "batch_compress"])?; + context::enforce_anonymous_upload_rate(&state, &principal, ip).await?; let admission = prepare_batch_admission(&state, &principal).await?; let idempotency_key = headers diff --git a/src/api/user.rs b/src/api/user.rs index 50f2a03..692ce4a 100644 --- a/src/api/user.rs +++ b/src/api/user.rs @@ -697,6 +697,16 @@ async fn create_api_key( if !email_verified { return Err(AppError::new(ErrorCode::EmailNotVerified, "请先验证邮箱")); } + if !settings::runtime_policy(&state) + .await? + .features + .api_key_enabled + { + return Err(AppError::new( + ErrorCode::Forbidden, + "API Key 功能当前已关闭", + )); + } let billing = billing::get_user_billing(&state, user_id).await?; if !billing.plan.feature_api_enabled { @@ -796,6 +806,16 @@ async fn rotate_api_key( if !email_verified { return Err(AppError::new(ErrorCode::EmailNotVerified, "请先验证邮箱")); } + if !settings::runtime_policy(&state) + .await? + .features + .api_key_enabled + { + return Err(AppError::new( + ErrorCode::Forbidden, + "API Key 功能当前已关闭", + )); + } let (full_key, key_prefix) = generate_api_key(); let key_hash = context::api_key_hash(&full_key, &state.config.api_key_pepper)?; diff --git a/src/error.rs b/src/error.rs index de7ebc4..1385a2f 100644 --- a/src/error.rs +++ b/src/error.rs @@ -5,7 +5,6 @@ use axum::{ }; use serde::Serialize; use std::fmt::{Display, Formatter}; -use uuid::Uuid; #[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)] #[serde(rename_all = "SCREAMING_SNAKE_CASE")] @@ -98,7 +97,8 @@ struct ErrorPayload { impl IntoResponse for AppError { fn into_response(self) -> axum::response::Response { - let request_id = format!("req_{}", Uuid::new_v4()); + let request_id = crate::api::request_context::current_request_id() + .unwrap_or_else(|| format!("req_{}", uuid::Uuid::new_v4())); let status = match self.code { ErrorCode::InvalidRequest => StatusCode::BAD_REQUEST, @@ -144,6 +144,8 @@ impl IntoResponse for AppError { }, }; + crate::services::metrics::record_error(self.code); + let mut response = (status, Json(body)).into_response(); if let Ok(value) = HeaderValue::from_str(&request_id) { response.headers_mut().insert("x-request-id", value); diff --git a/src/main.rs b/src/main.rs index 92182ef..dc3332e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -43,6 +43,7 @@ async fn main() -> Result<(), AppError> { redis, mailer: std::sync::Arc::new(mailer), image_processing_semaphore, + runtime_policy_cache: crate::services::settings::RuntimePolicyCache::new(), }; match state.config.role.as_str() { diff --git a/src/services/compress.rs b/src/services/compress.rs index 005cda1..21cc50f 100644 --- a/src/services/compress.rs +++ b/src/services/compress.rs @@ -16,6 +16,7 @@ use img_parts::{Bytes as ImgBytes, DynImage, ImageEXIF, ImageICC}; use oxipng::StripChunks; use rgb::FromSlice; use std::io::Cursor; +use std::time::Instant; const TARGET_MIN_LONG_EDGE: u32 = 640; const TARGET_MIN_SCALE: f64 = 0.55; @@ -287,7 +288,12 @@ pub async fn compress_image_bytes( max_height: Option, preserve_metadata: bool, ) -> Result, AppError> { - let max_image_pixels = state.config.max_image_pixels; + let started = Instant::now(); + let bytes_in = input.len() as u64; + let max_image_pixels = crate::services::settings::runtime_policy(state) + .await? + .file_limits + .max_image_pixels; let permit = state .image_processing_semaphore .clone() @@ -297,7 +303,7 @@ pub async fn compress_image_bytes( AppError::new(ErrorCode::Internal, "图片处理并发控制器已关闭").with_source(err) })?; - tokio::task::spawn_blocking(move || { + let result = match tokio::task::spawn_blocking(move || { let _permit = permit; compress_image_bytes_sync( input, @@ -313,9 +319,19 @@ pub async fn compress_image_bytes( ) }) .await - .map_err(|err| { - AppError::new(ErrorCode::CompressionFailed, "图片处理任务异常退出").with_source(err) - })? + { + Ok(result) => result, + Err(err) => Err( + AppError::new(ErrorCode::CompressionFailed, "图片处理任务异常退出").with_source(err), + ), + }; + crate::services::metrics::record_compression( + state, + started.elapsed(), + bytes_in, + result.as_ref().ok().map(|bytes| bytes.len() as u64), + ); + result } #[allow(clippy::too_many_arguments)] diff --git a/src/services/metrics.rs b/src/services/metrics.rs new file mode 100644 index 0000000..d21f9af --- /dev/null +++ b/src/services/metrics.rs @@ -0,0 +1,273 @@ +use crate::error::ErrorCode; +use crate::state::AppState; + +use std::fmt::Write; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::OnceLock; +use std::time::Duration; + +pub const QUEUE_STREAM_KEY: &str = "stream:compress_jobs"; +pub const QUEUE_GROUP_NAME: &str = "compress_workers"; +pub const DEAD_STREAM_KEY: &str = "stream:compress_jobs:dead"; +pub const CLUSTER_METRICS_KEY: &str = "metrics:imageforge"; + +const METHODS: [&str; 5] = ["GET", "POST", "PUT", "DELETE", "OTHER"]; +const STATUS_CLASSES: [&str; 5] = ["2xx", "3xx", "4xx", "5xx", "other"]; +const ERROR_CODES: [&str; 17] = [ + "INVALID_REQUEST", + "INVALID_IMAGE", + "UNSUPPORTED_FORMAT", + "TOO_MANY_PIXELS", + "FILE_TOO_LARGE", + "INVALID_TOKEN", + "UNAUTHORIZED", + "FORBIDDEN", + "NOT_FOUND", + "IDEMPOTENCY_CONFLICT", + "RATE_LIMITED", + "QUOTA_EXCEEDED", + "EMAIL_NOT_VERIFIED", + "COMPRESSION_FAILED", + "STORAGE_UNAVAILABLE", + "MAIL_SEND_FAILED", + "INTERNAL", +]; +const DURATION_BUCKETS: [f64; 9] = [0.01, 0.05, 0.1, 0.3, 1.0, 3.0, 10.0, 30.0, f64::INFINITY]; + +struct Histogram { + buckets: [AtomicU64; DURATION_BUCKETS.len()], + count: AtomicU64, + sum_micros: AtomicU64, +} + +impl Histogram { + fn new() -> Self { + Self { + buckets: std::array::from_fn(|_| AtomicU64::new(0)), + count: AtomicU64::new(0), + sum_micros: AtomicU64::new(0), + } + } + + fn observe(&self, duration: Duration) { + let seconds = duration.as_secs_f64(); + for (index, upper_bound) in DURATION_BUCKETS.iter().enumerate() { + if seconds <= *upper_bound { + self.buckets[index].fetch_add(1, Ordering::Relaxed); + } + } + self.count.fetch_add(1, Ordering::Relaxed); + self.sum_micros.fetch_add( + duration.as_micros().min(u64::MAX as u128) as u64, + Ordering::Relaxed, + ); + } + + fn render(&self, output: &mut String, name: &str) { + for (index, upper_bound) in DURATION_BUCKETS.iter().enumerate() { + let label = if upper_bound.is_infinite() { + "+Inf".to_string() + } else { + upper_bound.to_string() + }; + let _ = writeln!( + output, + "{name}_bucket{{le=\"{label}\"}} {}", + self.buckets[index].load(Ordering::Relaxed) + ); + } + let _ = writeln!( + output, + "{name}_sum {}", + self.sum_micros.load(Ordering::Relaxed) as f64 / 1_000_000.0 + ); + let _ = writeln!( + output, + "{name}_count {}", + self.count.load(Ordering::Relaxed) + ); + } +} + +struct Metrics { + http_requests: [[AtomicU64; STATUS_CLASSES.len()]; METHODS.len()], + http_duration: Histogram, + errors: [AtomicU64; ERROR_CODES.len()], +} + +impl Metrics { + fn new() -> Self { + Self { + http_requests: std::array::from_fn(|_| std::array::from_fn(|_| AtomicU64::new(0))), + http_duration: Histogram::new(), + errors: std::array::from_fn(|_| AtomicU64::new(0)), + } + } +} + +fn registry() -> &'static Metrics { + static METRICS: OnceLock = OnceLock::new(); + METRICS.get_or_init(Metrics::new) +} + +pub fn record_http(method: &str, status: u16, duration: Duration) { + let method_index = match method { + "GET" => 0, + "POST" => 1, + "PUT" => 2, + "DELETE" => 3, + _ => 4, + }; + let status_index = match status { + 200..=299 => 0, + 300..=399 => 1, + 400..=499 => 2, + 500..=599 => 3, + _ => 4, + }; + let metrics = registry(); + metrics.http_requests[method_index][status_index].fetch_add(1, Ordering::Relaxed); + metrics.http_duration.observe(duration); +} + +pub fn record_error(code: ErrorCode) { + registry().errors[error_index(code)].fetch_add(1, Ordering::Relaxed); +} + +pub fn record_compression( + state: &AppState, + duration: Duration, + bytes_in: u64, + bytes_out: Option, +) { + let mut increments = vec![ + ("compression_bytes_in", bytes_in), + ( + "compression_duration_micros", + duration.as_micros().min(u64::MAX as u128) as u64, + ), + ]; + if let Some(bytes_out) = bytes_out { + increments.push(("compression_success", 1)); + increments.push(("compression_bytes_out", bytes_out)); + } else { + increments.push(("compression_failed", 1)); + } + persist_cluster_increments(state, increments); +} + +pub fn record_storage_fallback(state: &AppState) { + persist_cluster_increments(state, vec![("storage_fallbacks", 1)]); +} + +pub fn record_dead_letter(state: &AppState) { + persist_cluster_increments(state, vec![("dead_letters", 1)]); +} + +pub fn render() -> String { + let metrics = registry(); + let mut output = String::with_capacity(8 * 1024); + + output.push_str( + "# HELP imageforge_http_requests_total HTTP requests handled by method and status class.\n", + ); + output.push_str("# TYPE imageforge_http_requests_total counter\n"); + for (method_index, method) in METHODS.iter().enumerate() { + for (status_index, status_class) in STATUS_CLASSES.iter().enumerate() { + let _ = writeln!( + output, + "imageforge_http_requests_total{{method=\"{method}\",status_class=\"{status_class}\"}} {}", + metrics.http_requests[method_index][status_index].load(Ordering::Relaxed) + ); + } + } + output.push_str("# HELP imageforge_http_request_duration_seconds HTTP request duration.\n"); + output.push_str("# TYPE imageforge_http_request_duration_seconds histogram\n"); + metrics + .http_duration + .render(&mut output, "imageforge_http_request_duration_seconds"); + + output.push_str("# HELP imageforge_errors_total Application errors by code.\n"); + output.push_str("# TYPE imageforge_errors_total counter\n"); + for (index, code) in ERROR_CODES.iter().enumerate() { + let _ = writeln!( + output, + "imageforge_errors_total{{code=\"{code}\"}} {}", + metrics.errors[index].load(Ordering::Relaxed) + ); + } + + output +} + +fn persist_cluster_increments(state: &AppState, increments: Vec<(&'static str, u64)>) { + let mut connection = state.redis.clone(); + tokio::spawn(async move { + let mut pipeline = redis::pipe(); + for (field, amount) in increments { + pipeline + .cmd("HINCRBY") + .arg(CLUSTER_METRICS_KEY) + .arg(field) + .arg(amount.min(i64::MAX as u64) as i64) + .ignore(); + } + if let Err(err) = pipeline.query_async::<_, ()>(&mut connection).await { + tracing::debug!(error = %err, "failed to persist cluster metric"); + } + }); +} + +fn error_index(code: ErrorCode) -> usize { + match code { + ErrorCode::InvalidRequest => 0, + ErrorCode::InvalidImage => 1, + ErrorCode::UnsupportedFormat => 2, + ErrorCode::TooManyPixels => 3, + ErrorCode::FileTooLarge => 4, + ErrorCode::InvalidToken => 5, + ErrorCode::Unauthorized => 6, + ErrorCode::Forbidden => 7, + ErrorCode::NotFound => 8, + ErrorCode::IdempotencyConflict => 9, + ErrorCode::RateLimited => 10, + ErrorCode::QuotaExceeded => 11, + ErrorCode::EmailNotVerified => 12, + ErrorCode::CompressionFailed => 13, + ErrorCode::StorageUnavailable => 14, + ErrorCode::MailSendFailed => 15, + ErrorCode::Internal => 16, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn error_codes_and_slots_stay_aligned() { + let codes = [ + ErrorCode::InvalidRequest, + ErrorCode::InvalidImage, + ErrorCode::UnsupportedFormat, + ErrorCode::TooManyPixels, + ErrorCode::FileTooLarge, + ErrorCode::InvalidToken, + ErrorCode::Unauthorized, + ErrorCode::Forbidden, + ErrorCode::NotFound, + ErrorCode::IdempotencyConflict, + ErrorCode::RateLimited, + ErrorCode::QuotaExceeded, + ErrorCode::EmailNotVerified, + ErrorCode::CompressionFailed, + ErrorCode::StorageUnavailable, + ErrorCode::MailSendFailed, + ErrorCode::Internal, + ]; + for (index, code) in codes.into_iter().enumerate() { + assert_eq!(error_index(code), index); + assert_eq!(ERROR_CODES[index], code.as_str()); + } + } +} diff --git a/src/services/mod.rs b/src/services/mod.rs index 11cd28e..eeb1c15 100644 --- a/src/services/mod.rs +++ b/src/services/mod.rs @@ -5,6 +5,7 @@ pub mod credentials; pub mod filename; pub mod idempotency; pub mod mail; +pub mod metrics; pub mod quota; pub mod rate_limit; pub mod settings; diff --git a/src/services/quota.rs b/src/services/quota.rs index cc57381..db6b18a 100644 --- a/src/services/quota.rs +++ b/src/services/quota.rs @@ -259,7 +259,10 @@ pub async fn consume_anonymous_units( let mut conn = state.redis.clone(); - let limit = state.config.anon_daily_units as i64; + let limit = crate::services::settings::runtime_policy(state) + .await? + .rate_limits + .anonymous_units_per_day as i64; let ttl_seconds = 48 * 60 * 60; let inc = units as i64; @@ -299,7 +302,7 @@ pub async fn consume_anonymous_units( if new_value < 0 { return Err(AppError::new( ErrorCode::QuotaExceeded, - "匿名试用次数已用完（每日 10 次）", + format!("匿名试用次数已用完（每日 {limit} 次）"), )); } diff --git a/src/services/settings.rs b/src/services/settings.rs index ed1efce..99d3fa4 100644 --- a/src/services/settings.rs +++ b/src/services/settings.rs @@ -9,6 +9,13 @@ use rand::RngCore; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; +use sqlx::FromRow; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::RwLock; +use tokio::time::Instant; + +const RUNTIME_POLICY_CACHE_TTL: Duration = Duration::from_secs(5); #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MailCustomSmtp { @@ -38,6 +45,151 @@ fn default_true() -> bool { true } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FeaturesConfigStored { + #[serde(default = "default_true")] + pub registration_enabled: bool, + #[serde(default = "default_true")] + pub api_key_enabled: bool, + #[serde(default = "default_true")] + pub anonymous_upload_enabled: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RateLimitsConfigStored { + #[serde(default = "default_anonymous_per_minute")] + pub anonymous_per_minute: u32, + #[serde(default = "default_anonymous_units_per_day")] + pub anonymous_units_per_day: u32, + #[serde(default = "default_user_per_minute")] + pub user_per_minute: u32, + #[serde(default = "default_api_key_per_minute")] + pub api_key_per_minute: u32, + #[serde(default = "default_login_ip_per_5_minutes")] + pub login_ip_per_5_minutes: u32, + #[serde(default = "default_login_identity_per_5_minutes")] + pub login_identity_per_5_minutes: u32, + #[serde(default = "default_register_ip_per_hour")] + pub register_ip_per_hour: u32, + #[serde(default = "default_verification_email_per_minute")] + pub verification_email_per_minute: u32, + #[serde(default = "default_email_verify_ip_per_15_minutes")] + pub email_verify_ip_per_15_minutes: u32, + #[serde(default = "default_forgot_password_ip_per_15_minutes")] + pub forgot_password_ip_per_15_minutes: u32, + #[serde(default = "default_forgot_password_email_per_15_minutes")] + pub forgot_password_email_per_15_minutes: u32, + #[serde(default = "default_password_reset_ip_per_15_minutes")] + pub password_reset_ip_per_15_minutes: u32, + #[serde(default = "default_password_reset_token_per_15_minutes")] + pub password_reset_token_per_15_minutes: u32, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FileLimitsConfigStored { + #[serde(default = "default_max_image_pixels")] + pub max_image_pixels: u64, +} + +#[derive(Debug, Clone)] +pub struct RuntimePolicy { + pub auth: AuthConfigStored, + pub features: FeaturesConfigStored, + pub rate_limits: RateLimitsConfigStored, + pub file_limits: FileLimitsConfigStored, +} + +#[derive(Debug, Clone)] +struct CachedRuntimePolicy { + loaded_at: Instant, + policy: RuntimePolicy, +} + +#[derive(Debug, Clone, Default)] +pub struct RuntimePolicyCache { + inner: Arc>>, +} + +impl RuntimePolicyCache { + pub fn new() -> Self { + Self::default() + } + + async fn get(&self) -> Option { + let cache = self.inner.read().await; + cache.as_ref().and_then(|cached| { + (cached.loaded_at.elapsed() < RUNTIME_POLICY_CACHE_TTL).then(|| cached.policy.clone()) + }) + } + + async fn set(&self, policy: RuntimePolicy) { + *self.inner.write().await = Some(CachedRuntimePolicy { + loaded_at: Instant::now(), + policy, + }); + } + + pub async fn invalidate(&self) { + *self.inner.write().await = None; + } +} + +fn default_anonymous_per_minute() -> u32 { + 10 +} + +fn default_anonymous_units_per_day() -> u32 { + 10 +} + +fn default_user_per_minute() -> u32 { + 60 +} + +fn default_api_key_per_minute() -> u32 { + 100 +} + +fn default_login_ip_per_5_minutes() -> u32 { + 30 +} + +fn default_login_identity_per_5_minutes() -> u32 { + 10 +} + +fn default_register_ip_per_hour() -> u32 { + 10 +} + +fn default_verification_email_per_minute() -> u32 { + 1 +} + +fn default_email_verify_ip_per_15_minutes() -> u32 { + 20 +} + +fn default_forgot_password_ip_per_15_minutes() -> u32 { + 5 +} + +fn default_forgot_password_email_per_15_minutes() -> u32 { + 3 +} + +fn default_password_reset_ip_per_15_minutes() -> u32 { + 10 +} + +fn default_password_reset_token_per_15_minutes() -> u32 { + 5 +} + +fn default_max_image_pixels() -> u64 { + 40_000_000 +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StripeConfigStored { pub secret_key_encrypted: Option, @@ -73,6 +225,144 @@ pub async fn load_system_config( Ok(Some(parsed)) } +pub async fn runtime_policy(state: &AppState) -> Result { + if let Some(policy) = state.runtime_policy_cache.get().await { + return Ok(policy); + } + + #[derive(Debug, FromRow)] + struct ConfigValueRow { + key: String, + value: serde_json::Value, + } + + let rows = sqlx::query_as::<_, ConfigValueRow>( + r#" + SELECT key, value + FROM system_config + WHERE key = ANY($1) + "#, + ) + .bind(["auth", "features", "rate_limits", "file_limits"]) + .fetch_all(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询运行策略失败").with_source(err))?; + + let mut auth = AuthConfigStored { + email_verification_required: true, + }; + let mut features = FeaturesConfigStored { + registration_enabled: true, + api_key_enabled: true, + anonymous_upload_enabled: true, + }; + let mut rate_limits = RateLimitsConfigStored { + anonymous_units_per_day: state.config.anon_daily_units, + ..RateLimitsConfigStored::default() + }; + let mut file_limits = FileLimitsConfigStored { + max_image_pixels: state.config.max_image_pixels, + }; + + for row in rows { + match row.key.as_str() { + "auth" => auth = parse_config_value(row.value, "认证配置")?, + "features" => features = parse_config_value(row.value, "功能开关")?, + "rate_limits" => rate_limits = parse_config_value(row.value, "限速配置")?, + "file_limits" => file_limits = parse_config_value(row.value, "文件限制")?, + _ => {} + } + } + + features.anonymous_upload_enabled &= state.config.allow_anonymous_upload; + validate_rate_limits(&rate_limits)?; + validate_file_limits(&file_limits)?; + + let policy = RuntimePolicy { + auth, + features, + rate_limits, + file_limits, + }; + state.runtime_policy_cache.set(policy.clone()).await; + Ok(policy) +} + +fn parse_config_value( + value: serde_json::Value, + label: &str, +) -> Result { + serde_json::from_value(value).map_err(|err| { + AppError::new(ErrorCode::Internal, format!("{label}格式错误")).with_source(err) + }) +} + +pub fn validate_runtime_config_value(key: &str, value: &serde_json::Value) -> Result<(), AppError> { + match key { + "auth" => { + serde_json::from_value::(value.clone()).map_err(|err| { + AppError::new(ErrorCode::InvalidRequest, "认证配置格式错误").with_source(err) + })?; + } + "features" => { + serde_json::from_value::(value.clone()).map_err(|err| { + AppError::new(ErrorCode::InvalidRequest, "功能开关格式错误").with_source(err) + })?; + } + "rate_limits" => { + let config = + serde_json::from_value::(value.clone()).map_err(|err| { + AppError::new(ErrorCode::InvalidRequest, "限速配置格式错误").with_source(err) + })?; + validate_rate_limits(&config)?; + } + "file_limits" => { + let config = + serde_json::from_value::(value.clone()).map_err(|err| { + AppError::new(ErrorCode::InvalidRequest, "文件限制格式错误").with_source(err) + })?; + validate_file_limits(&config)?; + } + _ => {} + } + Ok(()) +} + +fn validate_rate_limits(config: &RateLimitsConfigStored) -> Result<(), AppError> { + let values = [ + config.anonymous_per_minute, + config.anonymous_units_per_day, + config.user_per_minute, + config.api_key_per_minute, + config.login_ip_per_5_minutes, + config.login_identity_per_5_minutes, + config.register_ip_per_hour, + config.verification_email_per_minute, + config.email_verify_ip_per_15_minutes, + config.forgot_password_ip_per_15_minutes, + config.forgot_password_email_per_15_minutes, + config.password_reset_ip_per_15_minutes, + config.password_reset_token_per_15_minutes, + ]; + if values.iter().any(|value| !(1..=100_000).contains(value)) { + return Err(AppError::new( + ErrorCode::InvalidRequest, + "限速值必须在 1 到 100000 之间", + )); + } + Ok(()) +} + +fn validate_file_limits(config: &FileLimitsConfigStored) -> Result<(), AppError> { + if !(1_000_000..=200_000_000).contains(&config.max_image_pixels) { + return Err(AppError::new( + ErrorCode::InvalidRequest, + "max_image_pixels 必须在 1000000 到 200000000 之间", + )); + } + Ok(()) +} + pub async fn upsert_system_config( state: &AppState, key: &str, @@ -99,6 +389,10 @@ pub async fn upsert_system_config( .await .map_err(|err| AppError::new(ErrorCode::Internal, "更新系统配置失败").with_source(err))?; + if matches!(key, "auth" | "features" | "rate_limits" | "file_limits") { + state.runtime_policy_cache.invalidate().await; + } + Ok(()) } @@ -128,10 +422,10 @@ pub async fn load_mail_settings(state: &AppState) -> Result } pub async fn email_verification_required(state: &AppState) -> Result { - Ok(load_system_config::(state, "auth") + Ok(runtime_policy(state) .await? - .map(|config| config.email_verification_required) - .unwrap_or(true)) + .auth + .email_verification_required) } pub async fn load_stripe_secrets(state: &AppState) -> Result, AppError> { @@ -243,6 +537,26 @@ pub async fn get_stripe_webhook_secret(state: &AppState) -> Result Self { + Self { + anonymous_per_minute: default_anonymous_per_minute(), + anonymous_units_per_day: default_anonymous_units_per_day(), + user_per_minute: default_user_per_minute(), + api_key_per_minute: default_api_key_per_minute(), + login_ip_per_5_minutes: default_login_ip_per_5_minutes(), + login_identity_per_5_minutes: default_login_identity_per_5_minutes(), + register_ip_per_hour: default_register_ip_per_hour(), + verification_email_per_minute: default_verification_email_per_minute(), + email_verify_ip_per_15_minutes: default_email_verify_ip_per_15_minutes(), + forgot_password_ip_per_15_minutes: default_forgot_password_ip_per_15_minutes(), + forgot_password_email_per_15_minutes: default_forgot_password_email_per_15_minutes(), + password_reset_ip_per_15_minutes: default_password_reset_ip_per_15_minutes(), + password_reset_token_per_15_minutes: default_password_reset_token_per_15_minutes(), + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -252,4 +566,32 @@ mod tests { let config: AuthConfigStored = serde_json::from_value(serde_json::json!({})).unwrap(); assert!(config.email_verification_required); } + + #[test] + fn rate_limit_defaults_support_legacy_config_rows() { + let config: RateLimitsConfigStored = serde_json::from_value(serde_json::json!({ + "anonymous_per_minute": 7, + "user_per_minute": 55 + })) + .unwrap(); + assert_eq!(config.anonymous_per_minute, 7); + assert_eq!(config.user_per_minute, 55); + assert_eq!(config.login_identity_per_5_minutes, 10); + assert_eq!(config.password_reset_token_per_15_minutes, 5); + } + + #[test] + fn invalid_runtime_limits_are_rejected_before_persisting() { + let result = validate_runtime_config_value( + "rate_limits", + &serde_json::json!({ "anonymous_per_minute": 0 }), + ); + assert!(result.is_err()); + + let result = validate_runtime_config_value( + "file_limits", + &serde_json::json!({ "max_image_pixels": 999_999 }), + ); + assert!(result.is_err()); + } } diff --git a/src/services/storage.rs b/src/services/storage.rs index e4fd7b9..3584bfa 100644 --- a/src/services/storage.rs +++ b/src/services/storage.rs @@ -161,7 +161,7 @@ where if let Some(endpoint) = active_endpoint(state).await? { match store_bytes_s3(state, &endpoint, key, bytes.clone(), content_type).await { Ok(stored) => return Ok(stored), - Err(err) => log_local_fallback(&endpoint, key, &err), + Err(err) => log_local_fallback(state, &endpoint, key, &err), } } @@ -233,7 +233,7 @@ pub async fn store_file( if let Some(endpoint) = active_endpoint(state).await? { match store_file_s3(state, &endpoint, key, path, content_type, metadata.len()).await { Ok(stored) => return Ok(stored), - Err(err) => log_local_fallback(&endpoint, key, &err), + Err(err) => log_local_fallback(state, &endpoint, key, &err), } } @@ -302,7 +302,8 @@ async fn store_file_local( }) } -fn log_local_fallback(endpoint: &StorageEndpoint, key: &str, err: &AppError) { +fn log_local_fallback(state: &AppState, endpoint: &StorageEndpoint, key: &str, err: &AppError) { + crate::services::metrics::record_storage_fallback(state); tracing::warn!( storage_endpoint_id = %endpoint.id, storage_endpoint = %endpoint.name, diff --git a/src/state.rs b/src/state.rs index a8660fd..4b444d5 100644 --- a/src/state.rs +++ b/src/state.rs @@ -1,5 +1,6 @@ use crate::config::Config; use crate::services::mail::Mailer; +use crate::services::settings::RuntimePolicyCache; #[derive(Clone)] pub struct AppState { @@ -8,4 +9,5 @@ pub struct AppState { pub redis: redis::aio::ConnectionManager, pub mailer: std::sync::Arc, pub image_processing_semaphore: std::sync::Arc, + pub runtime_policy_cache: RuntimePolicyCache, } diff --git a/src/worker/mod.rs b/src/worker/mod.rs index 2f39421..ba25436 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -1,6 +1,7 @@ use crate::error::{AppError, ErrorCode}; use crate::services::billing; use crate::services::compress; +use crate::services::metrics; use crate::services::quota; use crate::services::storage; use crate::state::AppState; @@ -15,9 +16,9 @@ use tokio::sync::Semaphore; use tokio::task::JoinSet; use uuid::Uuid; -const STREAM_KEY: &str = "stream:compress_jobs"; -const GROUP_NAME: &str = "compress_workers"; -const DEAD_STREAM_KEY: &str = "stream:compress_jobs:dead"; +const STREAM_KEY: &str = metrics::QUEUE_STREAM_KEY; +const GROUP_NAME: &str = metrics::QUEUE_GROUP_NAME; +const DEAD_STREAM_KEY: &str = metrics::DEAD_STREAM_KEY; const MAX_DELIVERIES: usize = 3; const STALE_MESSAGE_IDLE_MS: usize = 5 * 60 * 1000; @@ -136,6 +137,7 @@ async fn handle_message( write_dead_letter(conn, &msg.id, task_id, deliveries, &err).await?; mark_task_dead_letter(state, task_id, &err.message).await?; ack_message(conn, &msg.id).await?; + metrics::record_dead_letter(state); tracing::error!( task_id = %task_id, deliveries,