feat: 支持按需触发短信图形验证

ai_knowledge_base_v2/apps/backend/app/schemas/auth.py

@@ -9,8 +9,8 @@ from app.schemas.user import UserProfile
 
 class SendSmsRequest(BaseModel):
     phone: str = Field(min_length=11, max_length=20)
-    captchaId: str = Field(min_length=1, max_length=100)
+    captchaId: str | None = Field(default=None, min_length=1, max_length=100)
-    captchaCode: str = Field(min_length=4, max_length=8)
+    captchaCode: str | None = Field(default=None, min_length=4, max_length=8)
 
 
 class LoginRequest(BaseModel):

ai_knowledge_base_v2/apps/backend/app/services/auth_service.py

@@ -14,11 +14,29 @@ from app.services.security_state_service import SecurityStateService
 
 
 class AuthService:
+    CAPTCHA_REQUIRED_SECONDS = 60 * 60
+
     @classmethod
-    def send_sms_code(cls, db: Session, phone: str, captcha_id: str, captcha_code: str, ip: str) -> None:
+    def send_sms_code(
-        CaptchaService.verify(captcha_id, captcha_code)
+        cls,
+        db: Session,
+        phone: str,
+        captcha_id: str | None,
+        captcha_code: str | None,
+        ip: str,
+    ) -> None:
         user = cls._get_existing_user(db, phone)
         cls._ensure_user_can_login(user)
+        captcha_key = cls._captcha_required_key(phone)
+        has_captcha = bool(captcha_id or captcha_code)
+        if has_captcha:
+            if not captcha_id or not captcha_code:
+                raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请完整填写图形验证码")
+            CaptchaService.verify(captcha_id, captcha_code)
+            SecurityStateService.clear(captcha_key)
+        elif SecurityStateService.has_flag(captcha_key):
+            raise HTTPException(status_code=428, detail="需要完成图形验证码后再发送短信")
+
         SecurityStateService.enforce_limit(
             f"rate:sms:ip:{ip}", limit=20, window_seconds=3600, message="当前网络发送验证码过于频繁，请稍后再试"
         )
@@ -26,6 +44,7 @@ class AuthService:
             f"rate:sms:phone:{phone}", limit=1, window_seconds=60, message="验证码发送过于频繁，请一分钟后再试"
         )
         SmsCodeService.send_code(db, phone)
+        SecurityStateService.mark_flag(captcha_key, cls.CAPTCHA_REQUIRED_SECONDS)
 
     @classmethod
     def login_with_sms(cls, db: Session, phone: str, code: str) -> dict:
@@ -36,6 +55,7 @@ class AuthService:
         try:
             SmsCodeService.verify_code(db, phone, code)
         except HTTPException:
+            SecurityStateService.mark_flag(cls._captcha_required_key(phone), cls.CAPTCHA_REQUIRED_SECONDS)
             SecurityStateService.record_failure(
                 failure_key,
                 limit=5,
@@ -84,3 +104,7 @@ class AuthService:
             raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="账号尚未生效")
         if user.expired_at and user.expired_at < now:
             raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="账号已过期")
+
+    @staticmethod
+    def _captcha_required_key(phone: str) -> str:
+        return f"auth:sms:captcha_required:{phone}"

ai_knowledge_base_v2/apps/backend/app/services/security_state_service.py

@@ -48,6 +48,22 @@ class SecurityStateService:
         with cls._lock:
             cls._counters.pop(key, None)
 
+    @classmethod
+    def mark_flag(cls, key: str, ttl_seconds: int) -> None:
+        client = get_sync_redis_client()
+        if client is not None:
+            try:
+                client.set(key, "1", ex=ttl_seconds)
+                return
+            except Exception:
+                pass
+        with cls._lock:
+            cls._counters[key] = _Counter(1, time.monotonic() + ttl_seconds)
+
+    @classmethod
+    def has_flag(cls, key: str) -> bool:
+        return cls._get(key) > 0
+
     @classmethod
     def revoke_token(cls, jti: str, expires_at: int | float | None) -> None:
         now = datetime.now(UTC)

ai_knowledge_base_v2/apps/backend/tests/test_production_readiness.py

@@ -21,6 +21,7 @@ from app.services.chat_queue_service import ChatQueueConfig, QueueRequest
 from app.services.chat_service import ChatService
 from app.services.secret_service import ENCRYPTED_PREFIX, MASKED_SECRET, SecretService
 from app.services.security_state_service import SecurityStateService
+from app.services.auth_service import AuthService
 from app.core.observability import RequestObservabilityMiddleware
 
 
@@ -145,3 +146,22 @@ def test_readiness_reports_dependency_state(monkeypatch):
     result = health.ready()
     assert result["data"]["status"] == "ready"
     assert result["data"]["checks"] == {"database": True, "redis": True}
+
+
+def test_sms_captcha_is_required_only_after_first_send(monkeypatch):
+    engine = create_engine(
+        "sqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(engine)
+    monkeypatch.setattr("app.services.auth_service.SmsCodeService.send_code", lambda *_args: None)
+    monkeypatch.setattr("app.services.auth_service.CaptchaService.verify", lambda *_args: None)
+
+    with Session(engine) as db:
+        db.add(User(id=2, phone="13800000002", name="按需验证用户", daily_chat_limit=10, daily_chat_used=0))
+        db.commit()
+        AuthService.send_sms_code(db, "13800000002", None, None, "127.0.0.1")
+        with pytest.raises(HTTPException) as exc:
+            AuthService.send_sms_code(db, "13800000002", None, None, "127.0.0.1")
+        assert exc.value.status_code == 428