feat: add admin session rbac foundation

2026-08-25 22:47:59 +08:00
1 parent 8f000811a8
commit 2cdf7f9d7a
@@ -0,0 +1,222 @@
+from datetime import datetime, timedelta, timezone
+import importlib.util
+from pathlib import Path
+
+from fastapi.testclient import TestClient
+
+from app.auth import decode_admin_access_token, hash_password
+from app.database import get_db
+from app.main import create_app
+from app.models import AdminDepartment, AdminMenu, AdminRole, AdminRoleDepartment, AdminRoleMenu, AdminUser, AdminUserDepartment, AdminUserRole, Lead, MediaAsset
+from app.rbac import DATA_SCOPE_VALUES, build_menu_tree
+from app.redis_session import (
+    InMemoryAdminSessionStore,
+    get_admin_session_store,
+    hash_refresh_token,
+)
+
+
+class LoginDb:
+    def __init__(self, user: AdminUser):
+        self.user = user
+
+    def scalar(self, _statement):
+        return self.user
+
+    def get(self, model, identifier):
+        return self.user if model is AdminUser and identifier == self.user.id else None
+
+    class _EmptyResult:
+        def all(self):
+            return []
+
+    def scalars(self, _statement):
+        return self._EmptyResult()
+
+
+def test_refresh_session_rotation_rejects_reuse():
+    store = InMemoryAdminSessionStore()
+    now = datetime.now(timezone.utc)
+    store.create(
+        session_id="session-1",
+        user_id="admin-1",
+        access_jti="access-1",
+        refresh_hash=hash_refresh_token("refresh-1"),
+        access_expires_at=now + timedelta(minutes=15),
+        refresh_expires_at=now + timedelta(days=7),
+    )
+
+    assert store.is_access_active("session-1", "access-1")
+    assert store.rotate(
+        session_id="session-1",
+        refresh_hash=hash_refresh_token("refresh-1"),
+        access_jti="access-2",
+        refresh_hash_next=hash_refresh_token("refresh-2"),
+        access_expires_at=now + timedelta(minutes=15),
+        refresh_expires_at=now + timedelta(days=7),
+    )
+    assert not store.is_access_active("session-1", "access-1")
+    assert store.is_access_active("session-1", "access-2")
+    assert not store.rotate(
+        session_id="session-1",
+        refresh_hash=hash_refresh_token("refresh-1"),
+        access_jti="access-3",
+        refresh_hash_next=hash_refresh_token("refresh-3"),
+        access_expires_at=now + timedelta(minutes=15),
+        refresh_expires_at=now + timedelta(days=7),
+    )
+
+
+def test_in_memory_store_supports_login_limit_and_permission_cache():
+    store = InMemoryAdminSessionStore()
+    assert store.allow_login_attempt("ip:admin@example.com", 2, 60)
+    assert store.allow_login_attempt("ip:admin@example.com", 2, 60)
+    assert not store.allow_login_attempt("ip:admin@example.com", 2, 60)
+    store.set_permission_context("admin-1", {"permissions": ["admin:read"]}, 60)
+    assert store.get_permission_context("admin-1") == {"permissions": ["admin:read"]}
+
+
+def test_admin_access_token_contains_scoped_session_claims():
+    user = AdminUser(
+        id="admin-1",
+        email="admin@example.com",
+        name="Admin",
+        role="admin",
+        passwordHash=hash_password("ChangeMe123!", rounds=4),
+        isActive=True,
+    )
+
+    from app.auth import create_access_token
+
+    token = create_access_token(user, session_id="session-1", access_jti="access-1")
+    payload = decode_admin_access_token(token)
+
+    assert payload["typ"] == "admin_access"
+    assert payload["aud"] == "admin"
+    assert payload["sid"] == "session-1"
+    assert payload["jti"] == "access-1"
+    assert payload["role"] == "admin"
+
+
+def test_admin_login_keeps_legacy_fields_and_sets_http_only_refresh_cookie():
+    user = AdminUser(
+        id="admin-1",
+        email="admin@example.com",
+        name="Admin",
+        role="admin",
+        passwordHash=hash_password("ChangeMe123!", rounds=4),
+        isActive=True,
+    )
+    app = create_app()
+    app.dependency_overrides[get_db] = lambda: LoginDb(user)
+    app.dependency_overrides[get_admin_session_store] = lambda: InMemoryAdminSessionStore()
+
+    try:
+        response = TestClient(app).post(
+            "/api/admin/auth/login",
+            json={"email": "admin@example.com", "password": "ChangeMe123!"},
+        )
+    finally:
+        app.dependency_overrides.clear()
+
+    assert response.status_code == 200
+    body = response.json()["data"]
+    assert body["token"] == body["accessToken"]
+    assert body["expiresIn"] > 0
+    assert body["user"]["id"] == "admin-1"
+    cookie = response.headers["set-cookie"]
+    assert "HttpOnly" in cookie
+    assert "Path=/api/admin/auth" in cookie
+
+
+def test_refresh_rotates_access_token_and_logout_revokes_it():
+    user = AdminUser(
+        id="admin-1",
+        email="admin@example.com",
+        name="Admin",
+        role="admin",
+        passwordHash=hash_password("ChangeMe123!", rounds=4),
+        isActive=True,
+    )
+    store = InMemoryAdminSessionStore()
+    app = create_app()
+    app.dependency_overrides[get_db] = lambda: LoginDb(user)
+    app.dependency_overrides[get_admin_session_store] = lambda: store
+
+    try:
+        client = TestClient(app)
+        login_response = client.post(
+            "/api/admin/auth/login",
+            json={"email": "admin@example.com", "password": "ChangeMe123!"},
+        )
+        old_token = login_response.json()["data"]["accessToken"]
+
+        refresh_response = client.post("/api/admin/auth/refresh")
+        new_token = refresh_response.json()["data"]["accessToken"]
+
+        assert refresh_response.status_code == 200
+        assert new_token != old_token
+        assert client.get("/api/admin/me", headers={"Authorization": f"Bearer {old_token}"}).status_code == 401
+        assert client.get("/api/admin/me", headers={"Authorization": f"Bearer {new_token}"}).status_code == 200
+
+        logout_response = client.post("/api/admin/auth/logout")
+        assert logout_response.status_code == 200
+        assert client.get("/api/admin/me", headers={"Authorization": f"Bearer {new_token}"}).status_code == 401
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_rbac_models_define_normalized_association_tables():
+    assert AdminRole.__tablename__ == "AdminRole"
+    assert AdminMenu.__tablename__ == "AdminMenu"
+    assert AdminDepartment.__tablename__ == "AdminDepartment"
+    assert {AdminUserRole.__table__.c.userId.name, AdminUserRole.__table__.c.roleId.name} == {"userId", "roleId"}
+    assert {AdminRoleMenu.__table__.c.roleId.name, AdminRoleMenu.__table__.c.menuId.name} == {"roleId", "menuId"}
+    assert {AdminRoleDepartment.__table__.c.roleId.name, AdminRoleDepartment.__table__.c.deptId.name} == {"roleId", "deptId"}
+    assert {AdminUserDepartment.__table__.c.userId.name, AdminUserDepartment.__table__.c.deptId.name} == {"userId", "deptId"}
+
+
+def test_rbac_migration_is_appended_to_current_head():
+    migration_path = Path(__file__).parents[1] / "alembic" / "versions" / "0025_admin_rbac.py"
+    spec = importlib.util.spec_from_file_location("admin_rbac_migration", migration_path)
+    assert spec and spec.loader
+    migration = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(migration)
+
+    assert migration.down_revision == "0024_vehicle_demand"
+    assert set(migration.RBAC_TABLES) == {
+        "AdminRole",
+        "AdminMenu",
+        "AdminDepartment",
+        "AdminUserRole",
+        "AdminRoleMenu",
+        "AdminUserDepartment",
+        "AdminRoleDepartment",
+    }
+
+
+def test_ownership_migration_is_appended_after_rbac_and_models_expose_scope_fields():
+    migration_path = Path(__file__).parents[1] / "alembic" / "versions" / "0026_admin_ownership.py"
+    spec = importlib.util.spec_from_file_location("admin_ownership_migration", migration_path)
+    assert spec and spec.loader
+    migration = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(migration)
+
+    assert migration.down_revision == "0025_admin_rbac"
+    assert {"deptId", "createdById"}.issubset(Lead.__table__.c.keys())
+    assert {"deptId", "createdById"}.issubset(MediaAsset.__table__.c.keys())
+    assert "Lead" in migration.OWNED_TABLES
+
+
+def test_dynamic_menu_tree_preserves_parent_order_and_button_permissions():
+    menus = [
+        AdminMenu(id="child", parentId="root", name="按钮", type="button", permissionCode="home:edit", sortOrder=20, isVisible=True, isActive=True),
+        AdminMenu(id="root", parentId=None, name="首页", type="page", path="/home", componentKey="HomePage", permissionCode="home:read", sortOrder=10, isVisible=True, isActive=True),
+        AdminMenu(id="hidden", parentId=None, name="隐藏", type="page", sortOrder=30, isVisible=False, isActive=True),
+    ]
+
+    tree = build_menu_tree(menus)
+
+    assert DATA_SCOPE_VALUES == {"all", "dept", "dept_and_children", "custom_dept", "self"}
+    assert [item["id"] for item in tree] == ["root"]
+    assert tree[0]["children"][0]["permission"] == "home:edit"