feat: isolate external application conversations

ai_knowledge_base_v2/apps/backend/tests/test_chat_source_isolation.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+from fastapi import HTTPException
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+from sqlalchemy.pool import StaticPool
+
+from app.core.auth_context import ChatAccessScope
+from app.core.dependencies import get_current_user_context
+from app.models import Base
+from app.models.sso import SsoClient
+from app.models.user import User
+from app.services.chat_service import ChatService
+
+
+def _db() -> Session:
+    engine = create_engine(
+        "sqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(engine)
+    return Session(engine)
+
+
+def test_chat_sessions_are_isolated_by_login_source_and_application():
+    with _db() as db:
+        user = User(
+            phone="13800138000",
+            name="测试学员",
+            status=1,
+            daily_chat_limit=100,
+            daily_chat_used=0,
+            is_deleted=0,
+        )
+        db.add(user)
+        db.commit()
+        db.refresh(user)
+
+        direct = ChatService.create_session(db, user)
+        app_a_scope = ChatAccessScope(source_type="sso", source_client_id=101)
+        app_b_scope = ChatAccessScope(source_type="sso", source_client_id=202)
+        app_a = ChatService.create_session(db, user, app_a_scope)
+        app_b = ChatService.create_session(db, user, app_b_scope)
+
+        assert [item.id for item in ChatService.list_sessions(db, user)] == [direct.id]
+        assert [item.id for item in ChatService.list_sessions(db, user, app_a_scope)] == [app_a.id]
+        assert [item.id for item in ChatService.list_sessions(db, user, app_b_scope)] == [app_b.id]
+
+        with pytest.raises(HTTPException) as cross_source:
+            ChatService.get_history(db, user, app_a.id)
+        assert cross_source.value.status_code == 404
+
+        with pytest.raises(HTTPException) as cross_application:
+            ChatService.get_history(db, user, app_b.id, app_a_scope)
+        assert cross_application.value.status_code == 404
+
+
+def test_auth_context_defaults_old_tokens_to_direct_and_rejects_disabled_sso_client():
+    with _db() as db:
+        user = User(
+            phone="13700137000",
+            name="登录来源验收",
+            status=1,
+            daily_chat_limit=100,
+            daily_chat_used=0,
+            is_deleted=0,
+        )
+        client = SsoClient(
+            app_id="context-test-app",
+            name="鉴权验收应用",
+            client_secret="encrypted-placeholder",
+            redirect_uris="[]",
+            status=1,
+        )
+        db.add_all([user, client])
+        db.commit()
+
+        direct = get_current_user_context({"sub": str(user.id), "type": "user"}, db)
+        assert direct.chat_scope == ChatAccessScope.direct()
+
+        sso = get_current_user_context(
+            {
+                "sub": str(user.id),
+                "type": "user",
+                "auth_source": "sso",
+                "sso_client_id": client.id,
+            },
+            db,
+        )
+        assert sso.chat_scope == ChatAccessScope(source_type="sso", source_client_id=client.id)
+
+        client.status = 0
+        db.commit()
+        with pytest.raises(HTTPException) as disabled:
+            get_current_user_context(
+                {
+                    "sub": str(user.id),
+                    "type": "user",
+                    "auth_source": "sso",
+                    "sso_client_id": client.id,
+                },
+                db,
+            )
+        assert disabled.value.status_code == 401

ai_knowledge_base_v2/apps/backend/tests/test_sso_service.py

@@ -15,6 +15,8 @@ from sqlalchemy.pool import StaticPool
 from app.models import Base
 from app.core.config import get_settings
 from app.models.ai_config import SystemConfig
+from app.models.entitlement import EntitlementPlan, UserEntitlement
+from app.core.security import decode_access_token
 from app.models.sso import SsoClient, SsoLoginAudit, UserExternalIdentity
 from app.models.user import User
 from app.schemas.sso import SsoTicketRequest
@@ -211,3 +213,59 @@ def test_sso_first_login_requires_verified_existing_phone(monkeypatch: pytest.Mo
         audit = db.scalar(select(SsoLoginAudit).order_by(SsoLoginAudit.id.desc()))
         assert audit is not None
         assert audit.status == "FAILED"
+
+
+def test_sso_can_auto_register_and_assign_configured_entitlement(monkeypatch: pytest.MonkeyPatch):
+    redis = FakeRedis()
+    monkeypatch.setattr("app.services.sso_service.get_sync_redis_client", lambda: redis)
+    monkeypatch.setattr(get_settings(), "jwt_secret_key", "test-sso-jwt-secret-key-32-bytes-long")
+    with _db() as db:
+        _user, client, secret = _seed(db)
+        plan = EntitlementPlan(
+            name="第三方默认权益",
+            plan_type="basic",
+            validity_days=90,
+            status=1,
+            sort_order=1,
+        )
+        db.add(plan)
+        db.flush()
+        client.allow_auto_register = 1
+        client.default_entitlement_plan_id = plan.id
+        db.commit()
+
+        raw_body = json.dumps(
+            {
+                "externalUserId": "external-new-user",
+                "verifiedPhone": "13900139000",
+                "displayName": "新接入学员",
+            },
+            ensure_ascii=False,
+            separators=(",", ":"),
+        ).encode()
+        timestamp = str(int(time.time()))
+        nonce = "auto-register-nonce-123"
+        ticket = SsoService.issue_ticket(
+            db,
+            app_id=client.app_id,
+            timestamp=timestamp,
+            nonce=nonce,
+            signature=_signature(secret, timestamp, nonce, raw_body),
+            raw_body=raw_body,
+            payload=SsoTicketRequest.model_validate_json(raw_body),
+            ip="127.0.0.1",
+        )
+
+        user = db.scalar(select(User).where(User.phone == "13900139000"))
+        assert user is not None
+        assert user.name == "新接入学员"
+        assert user.registration_source == "sso"
+        assert user.registration_client_id == client.id
+        entitlement = db.scalar(select(UserEntitlement).where(UserEntitlement.user_id == user.id))
+        assert entitlement is not None
+        assert entitlement.plan_id == plan.id
+
+        login = SsoService.exchange(db, code=ticket["code"], ip="127.0.0.1")
+        claims = decode_access_token(login["token"])
+        assert claims["auth_source"] == "sso"
+        assert claims["sso_client_id"] == client.id