Fix Telegram fallback binding group gate

src/bot/handlers/basic.py

@@ -134,8 +134,8 @@ class BasicCommandHandler:
             payload = str(parts[1] if len(parts) > 1 else "").strip()
             if payload.startswith("bind_"):
                 token = payload[len("bind_") :].strip()
-                result = self._prompt_bind_from_web_token(message, token)
-                trace.set_status("ok" if result == "confirm_prompted" else "error", result)
+                result = self._bind_from_web_token(message, message.from_user, token)
+                trace.set_status("ok" if result == "bound" else "error", result)
                 return
             self.bot.reply_to(message, self.io_layer.build_welcome_text(), parse_mode="HTML")
             trace.set_status("ok")

tests/test_bot_basic_handler.py

@@ -102,13 +102,21 @@ def test_basic_handler_diag_returns_html():
 
 def test_start_bind_token_binds_telegram_to_web_account():
     bot = DummyBot()
-    db = SimpleNamespace(
-        peek_web_bind_token=lambda token: {
+    consumed = []
+    bound = []
+
+    def _consume(token):
+        consumed.append(token)
+        return {
             "supabase_user_id": "user-1",
             "supabase_email": "u@example.com",
         }
-        if token == "abc123"
-        else None,
+
+    db = SimpleNamespace(
+        consume_web_bind_token=_consume,
+        upsert_user=lambda *_args, **_kwargs: None,
+        bind_supabase_identity=lambda **kwargs: bound.append(kwargs)
+        or {"ok": True, "reason": "bound"},
     )
     io_layer = SimpleNamespace(
         build_welcome_text=lambda: "WELCOME",
@@ -130,9 +138,11 @@ def test_start_bind_token_binds_telegram_to_web_account():
 
     handler.handle_start_help(_message("/start bind_abc123"))
 
+    assert consumed == ["abc123"]
+    assert bound[0]["telegram_id"] == 1
+    assert bound[0]["supabase_user_id"] == "user-1"
     assert len(bot.replies) == 1
-    assert "确认绑定" in bot.replies[0]["text"]
-    assert "u***@example.com" in bot.replies[0]["text"]
+    assert "账号绑定完成" in bot.replies[0]["text"]
 
 
 def test_confirm_bind_callback_consumes_token_and_binds_account():

tests/test_web_observability.py

@@ -1811,6 +1811,62 @@ def test_telegram_identity_endpoints_skip_subscription_gate(monkeypatch):
     assert link_payload["bot_url"].endswith("?start=bind_bind-token")
 
 
+def test_telegram_bind_by_token_does_not_require_existing_group_member(monkeypatch):
+    monkeypatch.setattr(web_core.SUPABASE_ENTITLEMENT, "enabled", True)
+    monkeypatch.setattr(web_core.SUPABASE_ENTITLEMENT, "require_subscription", True)
+    monkeypatch.setattr(web_core.SUPABASE_ENTITLEMENT, "supabase_url", "https://example.supabase.co")
+    monkeypatch.setattr(web_core.SUPABASE_ENTITLEMENT, "anon_key", "anon-key")
+    monkeypatch.setattr(web_core, "_SUPABASE_AUTH_REQUIRED", True)
+
+    class _Identity:
+        user_id = "user-1"
+        email = "user@example.com"
+        points = 0
+        created_at = "2026-05-01T00:00:00+00:00"
+
+    class _FakePricing:
+        configured = True
+
+        @staticmethod
+        def get_member_status(telegram_id):
+            raise AssertionError("bind fallback must not require existing group membership")
+
+        @staticmethod
+        def resolve_price_for_telegram_id(telegram_id):
+            return {"telegram_id": telegram_id, "pricing_source": "telegram_public"}
+
+    class _FakeDB:
+        @staticmethod
+        def consume_bind_token(token):
+            return 12345
+
+        @staticmethod
+        def upsert_user(telegram_id, username):
+            return None
+
+        @staticmethod
+        def bind_supabase_identity(*, telegram_id, supabase_user_id, supabase_email):
+            return {
+                "ok": True,
+                "telegram_id": telegram_id,
+                "supabase_user_id": supabase_user_id,
+                "supabase_email": supabase_email,
+            }
+
+    monkeypatch.setattr(web_core.SUPABASE_ENTITLEMENT, "get_identity", lambda token: _Identity())
+    monkeypatch.setattr(auth_api, "TelegramGroupPricing", lambda: _FakePricing())
+    monkeypatch.setattr(auth_api, "DBManager", lambda: _FakeDB())
+
+    response = client.post(
+        "/api/auth/telegram/bind-by-token",
+        headers={"Authorization": "Bearer access-token"},
+        json={"token": "token-12345"},
+    )
+
+    assert response.status_code == 200
+    assert response.json()["telegram_id"] == 12345
+
+
 def test_auth_me_does_not_reconcile_on_status_probe(monkeypatch):
     monkeypatch.setattr(routes, "_assert_entitlement", lambda request: None)
 

web/services/auth_api.py

@@ -416,11 +416,6 @@ def bind_telegram_by_token(request: Request, body) -> Dict[str, Any]:
     if telegram_id is None:
         raise HTTPException(status_code=400, detail="invalid or expired bind token")
 
-    pricing = TelegramGroupPricing()
-    member_status = pricing.get_member_status(telegram_id) if pricing.configured else None
-    if not member_status or member_status not in ("creator", "administrator", "member"):
-        raise HTTPException(status_code=403, detail="not a group member")
-
     db.upsert_user(telegram_id, "")
     bind_result = db.bind_supabase_identity(
         telegram_id=telegram_id,
@@ -433,7 +428,7 @@ def bind_telegram_by_token(request: Request, body) -> Dict[str, Any]:
             detail=str(bind_result.get("reason") or "telegram bind failed"),
         )
 
-    price = pricing.resolve_price_for_telegram_id(telegram_id)
+    price = TelegramGroupPricing().resolve_price_for_telegram_id(telegram_id)
     return {
         "ok": True,
         "telegram_id": telegram_id,