c0f17b1fe8
OIDC 原始用户名绑定使用 `oidc:{sub}:{target_user_id}` 记录占位关系，
但读取时按固定下标拆分字符串，导致 `sub` 自身包含冒号时无法回到
真实用户，回调流程会把已绑定用户误判成冲突账号。
本次改动保持现有占位格式不变，只把目标用户 ID 的解析改为从右侧
拆分，并补充覆盖 `sub` 包含冒号的单元测试；同时在 roadmap 中记录
这个回归修复，方便和 OIDC 功能演进保持同一条线索。
Constraint: 保持 OIDC 现有占位 user_id 设计，不引入新表或大范围重构
Rejected: 新增专用绑定表 | 变更范围过大，不符合当前最小修复目标
Rejected: 继续按固定位置 split | 无法兼容 subject 中合法出现的冒号
Confidence: high
Scope-risk: narrow
Reversibility: clean
Directive: 后续若继续沿用字符串占位格式，解析必须始终从尾部提取 target_user_id
Tested: OPENAI_API_KEY=dummy YUXI_SKIP_APP_INIT=1 uv run --group test pytest test/unit/services/test_oidc_service.py
Tested: OPENAI_API_KEY=dummy YUXI_SKIP_APP_INIT=1 uv run --group test ruff check package/yuxi/services/oidc_service.py test/unit/services/test_oidc_service.py
Not-tested: Docker compose 环境下的端到端 OIDC 登录流程
104 lines
3.8 KiB
Python
104 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from urllib.parse import unquote
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
os.environ.setdefault("OPENAI_API_KEY", "dummy")
|
|
|
|
from yuxi.services import oidc_service
|
|
from yuxi.storage.postgres.models_business import User
|
|
|
|
|
|
pytestmark = [pytest.mark.asyncio, pytest.mark.unit]
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def oidc_session():
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(User.__table__.create)
|
|
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with session_factory() as session:
|
|
yield session
|
|
|
|
await engine.dispose()
|
|
|
|
|
|
async def _create_user(session, user_id: str = "alice") -> User:
|
|
user = User(username="alice", user_id=user_id, password_hash="x", role="user", is_deleted=0)
|
|
session.add(user)
|
|
await session.commit()
|
|
await session.refresh(user)
|
|
return user
|
|
|
|
|
|
async def test_find_user_by_oidc_sub_resolves_placeholder_when_sub_contains_colon(oidc_session):
|
|
user = await _create_user(oidc_session)
|
|
|
|
await oidc_service._create_oidc_binding_placeholder(oidc_session, "tenant:user", user)
|
|
|
|
resolved = await oidc_service.find_user_by_oidc_sub(oidc_session, "tenant:user")
|
|
|
|
assert resolved is not None
|
|
assert resolved.id == user.id
|
|
assert resolved.user_id == user.user_id
|
|
assert resolved.is_deleted == 0
|
|
|
|
|
|
async def test_find_deleted_oidc_user_by_sub_resolves_deleted_target_when_sub_contains_colon(oidc_session):
|
|
user = await _create_user(oidc_session)
|
|
user.is_deleted = 1
|
|
await oidc_session.commit()
|
|
|
|
await oidc_service._create_oidc_binding_placeholder(oidc_session, "tenant:user", user)
|
|
|
|
resolved = await oidc_service.find_deleted_oidc_user_by_sub(oidc_session, "tenant:user")
|
|
|
|
assert resolved is not None
|
|
assert resolved.id == user.id
|
|
assert resolved.user_id == user.user_id
|
|
assert resolved.is_deleted == 1
|
|
|
|
|
|
async def test_oidc_callback_allows_existing_binding_when_sub_contains_colon(oidc_session, monkeypatch):
|
|
user = await _create_user(oidc_session)
|
|
await oidc_service._create_oidc_binding_placeholder(oidc_session, "tenant:user", user)
|
|
|
|
monkeypatch.setattr(oidc_service.oidc_config, "enabled", True)
|
|
monkeypatch.setattr(oidc_service.oidc_config, "client_id", "cid")
|
|
monkeypatch.setattr(oidc_service.oidc_config, "client_secret", "secret")
|
|
monkeypatch.setattr(oidc_service.oidc_config, "token_endpoint", "https://example/token")
|
|
monkeypatch.setattr(oidc_service.oidc_config, "authorization_endpoint", "https://example/auth")
|
|
monkeypatch.setattr(oidc_service.oidc_config, "userinfo_endpoint", "https://example/userinfo")
|
|
monkeypatch.setattr(oidc_service.oidc_config, "use_raw_username", True)
|
|
monkeypatch.setattr(oidc_service.oidc_config, "auto_create_user", False)
|
|
|
|
monkeypatch.setattr(
|
|
oidc_service.OIDCUtils,
|
|
"verify_state",
|
|
classmethod(lambda cls, state: {"redirect_path": "/"}),
|
|
)
|
|
|
|
async def fake_exchange(cls, code):
|
|
return {"access_token": "token"}
|
|
|
|
async def fake_userinfo(cls, access_token):
|
|
return {"sub": "tenant:user", "preferred_username": "alice"}
|
|
|
|
async def fake_log_operation(db, user_id, operation, request=None):
|
|
return None
|
|
|
|
monkeypatch.setattr(oidc_service.OIDCUtils, "exchange_code_for_token", classmethod(fake_exchange))
|
|
monkeypatch.setattr(oidc_service.OIDCUtils, "get_userinfo", classmethod(fake_userinfo))
|
|
monkeypatch.setattr(oidc_service, "log_operation", fake_log_operation)
|
|
|
|
response = await oidc_service.oidc_callback_handler("dummy-code", "dummy-state", oidc_session)
|
|
|
|
assert response.status_code == 302
|
|
assert unquote(response.headers["location"]).startswith("/auth/oidc/callback?code=")
|