github-featured/dataelement--bisheng
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(knowledge): optimize permission-filtered pagination

Browse Source
This commit is contained in:
GuoQing Zhang
2026-07-30 17:38:08 +08:00
parent 0bb2d42ecc
commit cd2e78a7e1
15 changed files with 936 additions and 757 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/bisheng/knowledge/domain/models/knowledge.py
+6
View File
@@ -364,6 +364,10 @@ class KnowledgeDao(KnowledgeBase):
# Cursor mode: append keyset WHERE and LIMIT only.
statement = cls._apply_keyset_where(statement, sort_by, cursor)
statement = statement.limit(limit)
elif not page and limit:
# The first cursor page uses page=0 and cursor=None. Keep it
# bounded instead of accidentally returning every candidate row.
statement = statement.limit(limit)
order_clauses = []
# Pinning is a first-page-only UX; suppress it when continuing a cursor.
@@ -579,6 +583,8 @@ class KnowledgeDao(KnowledgeBase):
statement = statement.limit(limit)
elif page and limit:
statement = statement.offset((page - 1) * limit).limit(limit)
elif limit:
statement = statement.limit(limit)
order_clauses = []
if preferred_ids and cursor is None:
src/backend/bisheng/knowledge/domain/services/knowledge_service.py
+101 -21
View File
@@ -92,6 +92,7 @@ _KNOWLEDGE_LIST_PERMISSION_IDS = [
"manage_kb_manager",
"manage_kb_viewer",
]
_KNOWLEDGE_PERMISSION_SCAN_BATCH_SIZE = 50
class KnowledgeService(KnowledgeUtils):
@@ -226,6 +227,90 @@ class KnowledgeService(KnowledgeUtils):
and current_tid != DEFAULT_TENANT_ID
)
@classmethod
async def _scan_visible_knowledge(
cls,
*,
login_user: UserPayload,
candidate_ids: list[int],
knowledge_type: KnowledgeTypeEnum,
name: str | None,
sort_by: str,
page_num: int | None,
cursor: list | None,
page_size: int,
permission_id: str,
preferred_ids: list[int] | None,
) -> tuple[list[Knowledge], dict[int, set[str]], int, int]:
"""Scan bounded DB batches and refill rows removed by exact permission checks."""
visible: list[Knowledge] = []
permission_map: dict[int, set[str]] = {}
scanned_rows = 0
batch_count = 0
# Name sorting exposes an offset-style pseudo cursor. To preserve pages
# after permission filtering, scan from the start and skip preceding
# visible rows. Pinned first pages also require offset batches because
# their rank is not represented by the timestamp cursor.
offset_scan = sort_by == "name" or (cursor is None and bool(preferred_ids))
visible_start = ((page_num or 1) - 1) * page_size if sort_by == "name" else 0
target_visible_count = visible_start + page_size + 1
batch_page = 1
batch_cursor = list(cursor) if cursor else None
requested_permission_ids = list(dict.fromkeys([permission_id, *_KNOWLEDGE_LIST_PERMISSION_IDS]))
while len(visible) < target_visible_count:
batch = await KnowledgeDao.aget_user_knowledge(
login_user.user_id,
candidate_ids,
knowledge_type,
name,
sort_by,
page=batch_page if offset_scan else 0,
limit=_KNOWLEDGE_PERMISSION_SCAN_BATCH_SIZE,
preferred_ids=preferred_ids,
cursor=None if offset_scan else batch_cursor,
)
batch_count += 1
scanned_rows += len(batch)
if not batch:
break
batch_permission_map = await cls.permission_service.get_knowledge_permission_map_async(
login_user,
[int(one.id) for one in batch],
requested_permission_ids,
)
permission_map.update(batch_permission_map)
visible.extend(one for one in batch if permission_id in batch_permission_map.get(int(one.id), set()))
if len(batch) < _KNOWLEDGE_PERMISSION_SCAN_BATCH_SIZE:
break
if offset_scan:
batch_page += 1
continue
last_db_row = batch[-1]
next_batch_cursor = [
last_db_row.create_time if sort_by == "create_time" else last_db_row.update_time,
last_db_row.id,
]
if next_batch_cursor == batch_cursor:
logger.warning(
"knowledge permission scan cursor did not advance: sort_by={} cursor={}",
sort_by,
batch_cursor,
)
break
batch_cursor = next_batch_cursor
return (
visible[visible_start : visible_start + page_size + 1],
permission_map,
scanned_rows,
batch_count,
)
@classmethod
async def get_knowledge(
cls,
@@ -289,35 +374,30 @@ class KnowledgeService(KnowledgeUtils):
login_user.user_id,
knowledge_type,
)
merged = set(int(k) for k in accessible_ids) | set(creator_ids)
permission_map = await cls.permission_service.get_knowledge_permission_map_async(
login_user,
list(merged),
_KNOWLEDGE_LIST_PERMISSION_IDS,
)
knowledge_id_extra = [
knowledge_id for knowledge_id in merged if permission_id in permission_map.get(int(knowledge_id), set())
]
res = await KnowledgeDao.aget_user_knowledge(
login_user.user_id,
knowledge_id_extra,
knowledge_type,
name,
sort_by,
page=(page_num if is_name_sort else 0),
limit=fetch_limit,
preferred_ids=preferred_ids,
merged = {int(k) for k in accessible_ids} | set(creator_ids)
res, permission_map, scanned_rows, batch_count = await cls._scan_visible_knowledge(
login_user=login_user,
candidate_ids=list(merged),
knowledge_type=knowledge_type,
name=name,
sort_by=sort_by,
page_num=page_num,
cursor=keyset_cursor,
page_size=page_size,
permission_id=permission_id,
preferred_ids=preferred_ids,
)
logger.info(
"[perf][knowledge.list.filter] user_id={} permission_id={} type={} accessible_ids={} creator_ids={} "
"filtered_ids={} sort_by={} page_size={} rows={} took_ms={:.2f}",
"candidate_ids={} scanned_rows={} batches={} sort_by={} page_size={} rows={} took_ms={:.2f}",
login_user.user_id,
permission_id,
knowledge_type.value,
len(accessible_ids),
len(creator_ids),
len(knowledge_id_extra),
len(merged),
scanned_rows,
batch_count,
sort_by,
page_size,
len(res),
@@ -370,7 +450,7 @@ class KnowledgeService(KnowledgeUtils):
)
# ---- 5. Compute next_cursor (None if has_more is False) ----
next_cursor: Optional[str] = None
next_cursor: str | None = None
if has_more and result_data:
last = result_data[-1]
if is_name_sort:
src/backend/bisheng/permission/domain/services/fine_grained_permission_service.py
+36 -51
View File
@@ -416,15 +416,6 @@ class FineGrainedPermissionService:
return [(object_type, str(object_id))]
@staticmethod
async def _tuple_resource_types(resource_type: str, resource_id: str) -> list[str]:
resource_types = [resource_type]
if resource_type == "knowledge_library":
resource_types.extend(
await PermissionService._legacy_alias_object_types(resource_type, resource_id),
)
return list(dict.fromkeys(resource_types))
@classmethod
async def get_effective_permission_ids_async(
cls,
@@ -473,49 +464,43 @@ class FineGrainedPermissionService:
for resource_type, resource_id in lineage:
level_permissions: set[str] = set()
level_saw_tuple = False
for tuple_resource_type in await cls._tuple_resource_types(resource_type, str(resource_id)):
tuple_object = f"{tuple_resource_type}:{resource_id}"
if tuple_cache is not None and tuple_object in tuple_cache:
tuples = tuple_cache[tuple_object]
else:
tuples = await fga.read_tuples(object=tuple_object)
if tuple_cache is not None:
tuple_cache[tuple_object] = tuples
binding_resource_type = (
resource_type
if tuple_resource_type != "knowledge_space" or resource_type != "knowledge_library"
else "knowledge_library"
tuple_object = f"{resource_type}:{resource_id}"
if tuple_cache is not None and tuple_object in tuple_cache:
tuples = tuple_cache[tuple_object]
else:
tuples = await fga.read_tuples(object=tuple_object)
if tuple_cache is not None:
tuple_cache[tuple_object] = tuples
for tuple_data in tuples:
tuple_user = tuple_data.get("user")
relation = tuple_data.get("relation")
if tuple_user not in user_subject_strings:
continue
binding = await cls._resolve_binding_for_tuple(
resource_type,
resource_id,
tuple_user,
relation,
bindings,
binding_department_paths,
tuple_department_paths,
binding_index=binding_index,
)
if cls._is_legacy_subscription_viewer_tuple(
resource_type,
tuple_user,
relation,
binding,
):
saw_legacy_subscription_viewer_tuple = True
continue
model = models.get(binding.get("model_id")) if binding and binding.get("model_id") else None
if binding and binding.get("model_id"):
saw_bound_model_tuple = True
level_saw_tuple = True
level_permissions.update(
cls._permission_ids_for_relation(resource_type, relation, model),
)
for tuple_data in tuples:
tuple_user = tuple_data.get("user")
relation = tuple_data.get("relation")
if tuple_user not in user_subject_strings:
continue
binding = await cls._resolve_binding_for_tuple(
binding_resource_type,
resource_id,
tuple_user,
relation,
bindings,
binding_department_paths,
tuple_department_paths,
binding_index=binding_index,
)
if cls._is_legacy_subscription_viewer_tuple(
tuple_resource_type,
tuple_user,
relation,
binding,
):
saw_legacy_subscription_viewer_tuple = True
continue
model = models.get(binding.get("model_id")) if binding and binding.get("model_id") else None
if binding and binding.get("model_id"):
saw_bound_model_tuple = True
level_saw_tuple = True
level_permissions.update(
cls._permission_ids_for_relation(resource_type, relation, model),
)
if nearest_binding_wins and level_saw_tuple:
matched_lineage_binding = True
effective_permissions.update(level_permissions)
src/backend/bisheng/permission/domain/services/legacy_rbac_sync_service.py
+114 -123
View File
@@ -9,8 +9,8 @@ from __future__ import annotations
import json
import logging
from collections.abc import Iterable
from dataclasses import dataclass
from typing import Iterable, Optional
from sqlalchemy import text as sa_text
from sqlalchemy.sql import bindparam
@@ -24,34 +24,26 @@ logger = logging.getLogger(__name__)
ACCESS_TYPE_TO_FGA: dict[int, tuple[str, str]] = {
1: ('knowledge_library', 'viewer'),
3: ('knowledge_library', 'editor'),
5: ('assistant', 'viewer'),
6: ('assistant', 'editor'),
7: ('tool', 'viewer'),
8: ('tool', 'editor'),
9: ('workflow', 'viewer'),
10: ('workflow', 'editor'),
11: ('dashboard', 'viewer'),
12: ('dashboard', 'editor'),
1: ("knowledge_library", "viewer"),
3: ("knowledge_library", "editor"),
5: ("assistant", "viewer"),
6: ("assistant", "editor"),
7: ("tool", "viewer"),
8: ("tool", "editor"),
9: ("workflow", "viewer"),
10: ("workflow", "editor"),
11: ("dashboard", "viewer"),
12: ("dashboard", "editor"),
}
RELATION_PRIORITY = {
'viewer': 1,
'editor': 2,
"viewer": 1,
"editor": 2,
}
ROLE_ACCESS_RELATIONS = frozenset({'viewer', 'editor'})
ROLE_ACCESS_OBJECT_TYPES = frozenset(
candidate_type
for object_type, _relation in ACCESS_TYPE_TO_FGA.values()
for candidate_type in (
(object_type, 'knowledge_space')
if object_type == 'knowledge_library'
else (object_type,)
)
)
_RELATION_BINDINGS_CONFIG_KEY = 'permission_relation_model_bindings_v1'
ROLE_ACCESS_RELATIONS = frozenset({"viewer", "editor"})
ROLE_ACCESS_OBJECT_TYPES = frozenset(object_type for object_type, _relation in ACCESS_TYPE_TO_FGA.values())
_RELATION_BINDINGS_CONFIG_KEY = "permission_relation_model_bindings_v1"
@dataclass(frozen=True)
@@ -62,11 +54,11 @@ class RoleAccessSignature:
resource_id: str
@property
def user_id(self) -> Optional[int]:
if not self.user.startswith('user:'):
def user_id(self) -> int | None:
if not self.user.startswith("user:"):
return None
try:
return int(self.user.split(':', 1)[1])
return int(self.user.split(":", 1)[1])
except Exception:
return None
@@ -75,7 +67,7 @@ class RoleAccessSignature:
action=action,
user=self.user,
relation=self.relation,
object=f'{self.object_type}:{self.resource_id}',
object=f"{self.object_type}:{self.resource_id}",
)
@@ -94,15 +86,18 @@ class LegacyRBACSyncService:
raw_new_role_ids = {int(role_id) for role_id in (new_role_ids or []) if role_id is not None}
if (AdminRole in raw_old_role_ids) != (AdminRole in raw_new_role_ids):
action = 'write' if AdminRole in raw_new_role_ids else 'delete'
await cls._write_operations([
TupleOperation(
action=action,
user=f'user:{user_id}',
relation='super_admin',
object='system:global',
),
], [user_id])
action = "write" if AdminRole in raw_new_role_ids else "delete"
await cls._write_operations(
[
TupleOperation(
action=action,
user=f"user:{user_id}",
relation="super_admin",
object="system:global",
),
],
[user_id],
)
old_role_ids = cls._clean_role_ids(raw_old_role_ids)
new_role_ids = cls._clean_role_ids(raw_new_role_ids)
@@ -126,19 +121,23 @@ class LegacyRBACSyncService:
operations: list[TupleOperation] = []
for group_id in member_group_ids or []:
operations.append(TupleOperation(
action='write',
user=f'user:{user_id}',
relation='member',
object=f'user_group:{group_id}',
))
operations.append(
TupleOperation(
action="write",
user=f"user:{user_id}",
relation="member",
object=f"user_group:{group_id}",
)
)
for group_id in admin_group_ids or []:
operations.append(TupleOperation(
action='write',
user=f'user:{user_id}',
relation='admin',
object=f'user_group:{group_id}',
))
operations.append(
TupleOperation(
action="write",
user=f"user:{user_id}",
relation="admin",
object=f"user_group:{group_id}",
)
)
await cls._write_operations(operations, [user_id])
@classmethod
@@ -202,7 +201,7 @@ class LegacyRBACSyncService:
actual = await cls._actual_user_role_access_signatures(user_id)
if actual is None:
await cls._write_operations([sig.to_operation('write') for sig in desired], [user_id])
await cls._write_operations([sig.to_operation("write") for sig in desired], [user_id])
return
await cls._apply_signature_diff(actual, desired)
@@ -214,24 +213,26 @@ class LegacyRBACSyncService:
fga = PermissionService._get_fga()
if fga is None:
logger.warning('FGAClient not available for user_group subject cleanup: %s', group_id)
logger.warning("FGAClient not available for user_group subject cleanup: %s", group_id)
return
operations: list[TupleOperation] = []
for relation in ('member', 'admin'):
user = f'user_group:{group_id}#{relation}'
for relation in ("member", "admin"):
user = f"user_group:{group_id}#{relation}"
try:
tuples = await fga.read_tuples(user=user)
except Exception as exc:
logger.warning('Failed to read tuples for %s: %s', user, exc)
logger.warning("Failed to read tuples for %s: %s", user, exc)
continue
for t in tuples or []:
operations.append(TupleOperation(
action='delete',
user=t.get('user', ''),
relation=t.get('relation', ''),
object=t.get('object', ''),
))
operations.append(
TupleOperation(
action="delete",
user=t.get("user", ""),
relation=t.get("relation", ""),
object=t.get("object", ""),
)
)
await cls._write_operations(operations, [])
@classmethod
@@ -245,18 +246,8 @@ class LegacyRBACSyncService:
protected = await cls._resource_permission_user_binding_set(stale) if stale else set()
to_delete = stale - protected
operations = [
sig.to_operation('delete')
for sig in to_delete
] + [
sig.to_operation('write')
for sig in to_write
]
affected = {
uid
for sig in (to_delete | to_write)
if (uid := sig.user_id) is not None
}
operations = [sig.to_operation("delete") for sig in to_delete] + [sig.to_operation("write") for sig in to_write]
affected = {uid for sig in (to_delete | to_write) if (uid := sig.user_id) is not None}
await cls._write_operations(operations, affected)
@classmethod
@@ -280,8 +271,9 @@ class LegacyRBACSyncService:
await PermissionCache.invalidate_user(uid)
try:
from bisheng.core.cache.redis_manager import get_redis_client
redis = await get_redis_client()
await redis.adelete(f'user:{uid}:is_super')
await redis.adelete(f"user:{uid}:is_super")
except Exception:
pass
@@ -290,7 +282,7 @@ class LegacyRBACSyncService:
cls,
user_id: int,
role_ids: Iterable[int],
override: Optional[tuple[int, int, set[str]]] = None,
override: tuple[int, int, set[str]] | None = None,
) -> set[RoleAccessSignature]:
role_ids = cls._clean_role_ids(role_ids)
if not role_ids:
@@ -335,7 +327,7 @@ class LegacyRBACSyncService:
object_type, relation = mapping
return [
RoleAccessSignature(
user=f'user:{user_id}',
user=f"user:{user_id}",
relation=relation,
object_type=fga_type,
resource_id=str(resource_id),
@@ -361,19 +353,11 @@ class LegacyRBACSyncService:
@staticmethod
def _fga_object_types(object_type: str) -> tuple[str, ...]:
if object_type == 'knowledge_library':
return 'knowledge_library', 'knowledge_space'
if object_type == 'knowledge_space':
return 'knowledge_space', 'knowledge_library'
return (object_type,)
@staticmethod
def _clean_role_ids(role_ids: Iterable[int]) -> set[int]:
return {
int(role_id)
for role_id in (role_ids or [])
if role_id is not None and int(role_id) != AdminRole
}
return {int(role_id) for role_id in (role_ids or []) if role_id is not None and int(role_id) != AdminRole}
@classmethod
async def _role_access_rows(
@@ -383,76 +367,83 @@ class LegacyRBACSyncService:
if not role_ids:
return []
statement = (
sa_text('SELECT role_id, third_id, type FROM roleaccess '
'WHERE role_id IN :role_ids AND type IN :access_types')
.bindparams(bindparam('role_ids', expanding=True))
.bindparams(bindparam('access_types', expanding=True))
sa_text(
"SELECT role_id, third_id, type FROM roleaccess WHERE role_id IN :role_ids AND type IN :access_types"
)
.bindparams(bindparam("role_ids", expanding=True))
.bindparams(bindparam("access_types", expanding=True))
)
async with get_async_db_session() as session:
with bypass_tenant_filter():
rows = (await session.execute(
statement,
{
'role_ids': sorted(role_ids),
'access_types': sorted(ACCESS_TYPE_TO_FGA.keys()),
},
)).all()
rows = (
await session.execute(
statement,
{
"role_ids": sorted(role_ids),
"access_types": sorted(ACCESS_TYPE_TO_FGA.keys()),
},
)
).all()
return [(int(row[0]), str(row[1]), int(row[2])) for row in rows]
@classmethod
async def _role_ids_for_user(cls, user_id: int) -> set[int]:
statement = sa_text('SELECT role_id FROM userrole WHERE user_id = :user_id AND role_id != :admin_rid')
statement = sa_text("SELECT role_id FROM userrole WHERE user_id = :user_id AND role_id != :admin_rid")
async with get_async_db_session() as session:
with bypass_tenant_filter():
rows = (await session.execute(
statement,
{'user_id': int(user_id), 'admin_rid': AdminRole},
)).all()
rows = (
await session.execute(
statement,
{"user_id": int(user_id), "admin_rid": AdminRole},
)
).all()
return {int(row[0]) for row in rows}
@classmethod
async def _user_ids_for_role(cls, role_id: int) -> list[int]:
statement = sa_text('SELECT DISTINCT user_id FROM userrole WHERE role_id = :role_id')
statement = sa_text("SELECT DISTINCT user_id FROM userrole WHERE role_id = :role_id")
async with get_async_db_session() as session:
with bypass_tenant_filter():
rows = (await session.execute(statement, {'role_id': int(role_id)})).all()
rows = (await session.execute(statement, {"role_id": int(role_id)})).all()
return [int(row[0]) for row in rows]
@classmethod
async def _actual_user_role_access_signatures(
cls,
user_id: int,
) -> Optional[set[RoleAccessSignature]]:
) -> set[RoleAccessSignature] | None:
from bisheng.permission.domain.services.permission_service import PermissionService
fga = PermissionService._get_fga()
if fga is None:
return None
try:
tuples = await fga.read_tuples(user=f'user:{user_id}')
tuples = await fga.read_tuples(user=f"user:{user_id}")
except Exception as exc:
logger.warning('Failed to read FGA tuples for user %s: %s', user_id, exc)
logger.warning("Failed to read FGA tuples for user %s: %s", user_id, exc)
return None
actual: set[RoleAccessSignature] = set()
for t in tuples or []:
user = t.get('user', '')
relation = t.get('relation', '')
obj = t.get('object', '')
user = t.get("user", "")
relation = t.get("relation", "")
obj = t.get("object", "")
if relation not in ROLE_ACCESS_RELATIONS:
continue
parts = obj.split(':', 1)
parts = obj.split(":", 1)
if len(parts) != 2:
continue
object_type, resource_id = parts
if object_type not in ROLE_ACCESS_OBJECT_TYPES:
continue
actual.add(RoleAccessSignature(
user=user,
relation=relation,
object_type=object_type,
resource_id=resource_id,
))
actual.add(
RoleAccessSignature(
user=user,
relation=relation,
object_type=object_type,
resource_id=resource_id,
)
)
return actual
@classmethod
@@ -481,12 +472,12 @@ class LegacyRBACSyncService:
from bisheng.common.models.config import ConfigDao
row = await ConfigDao.aget_config_by_key(_RELATION_BINDINGS_CONFIG_KEY)
if not row or not (row.value or '').strip():
if not row or not (row.value or "").strip():
return []
try:
bindings = json.loads(row.value or '[]')
bindings = json.loads(row.value or "[]")
except Exception:
logger.warning('Failed to parse resource permission bindings config')
logger.warning("Failed to parse resource permission bindings config")
return []
if not isinstance(bindings, list):
return []
@@ -501,10 +492,10 @@ class LegacyRBACSyncService:
) -> bool:
check_types = set(cls._fga_object_types(sig.object_type))
return any(
binding.get('resource_type') in check_types
and str(binding.get('resource_id')) == sig.resource_id
and binding.get('subject_type') == 'user'
and str(binding.get('subject_id')) == str(user_id)
and binding.get('relation') == sig.relation
binding.get("resource_type") in check_types
and str(binding.get("resource_id")) == sig.resource_id
and binding.get("subject_type") == "user"
and str(binding.get("subject_id")) == str(user_id)
and binding.get("relation") == sig.relation
for binding in bindings
)
src/backend/bisheng/permission/domain/services/permission_service.py
+15 -97
View File
@@ -153,17 +153,6 @@ class PermissionService:
consistency=consistency,
)
if not allowed:
for legacy_type in await cls._legacy_alias_object_types(object_type, object_id):
allowed = await fga.check(
user=f"user:{user_id}",
relation=relation,
object=f"{legacy_type}:{object_id}",
consistency=consistency,
)
if allowed:
break
# L4: Owner fallback — if FGA says no, check DB creator field. Owner
# and creator are decoupled, so an explicitly revoked creator-owner
# must not be resurrected: only fall back to creator ownership when no
@@ -272,19 +261,6 @@ class PermissionService:
if len(parts) == 2:
ids.append(parts[1])
for legacy_type in await cls._legacy_alias_object_types(object_type):
legacy_objects = await fga.list_objects(
user=f"user:{user_id}",
relation=relation,
type=legacy_type,
)
legacy_ids = []
for obj in legacy_objects:
parts = obj.split(":", 1)
if len(parts) == 2:
legacy_ids.append(parts[1])
ids.extend(await cls._filter_legacy_alias_ids(object_type, legacy_ids))
ids = await cls._finalize_accessible_ids(
ids,
user_id,
@@ -327,9 +303,7 @@ class PermissionService:
"""
operations: list[TupleOperation] = []
affected_user_ids: set[int] = set()
fga_objects = [f"{object_type}:{object_id}"]
for legacy_type in await cls._legacy_alias_object_types(object_type, object_id):
fga_objects.append(f"{legacy_type}:{object_id}")
fga_object = f"{object_type}:{object_id}"
for grant in grants or []:
fga_users = await cls._expand_subject(
@@ -338,15 +312,14 @@ class PermissionService:
grant.include_children,
)
for fga_user in fga_users:
for fga_object in fga_objects:
operations.append(
TupleOperation(
action="write",
user=fga_user,
relation=grant.relation,
object=fga_object,
)
operations.append(
TupleOperation(
action="write",
user=fga_user,
relation=grant.relation,
object=fga_object,
)
)
affected_user_ids.update(
await cls._affected_user_ids_for_subject(
grant.subject_type,
@@ -362,15 +335,14 @@ class PermissionService:
revoke.include_children,
)
for fga_user in fga_users:
for fga_object in fga_objects:
operations.append(
TupleOperation(
action="delete",
user=fga_user,
relation=revoke.relation,
object=fga_object,
)
operations.append(
TupleOperation(
action="delete",
user=fga_user,
relation=revoke.relation,
object=fga_object,
)
)
affected_user_ids.update(
await cls._affected_user_ids_for_subject(
revoke.subject_type,
@@ -624,8 +596,6 @@ class PermissionService:
return []
tuples = await fga.read_tuples(object=f"{object_type}:{object_id}")
for legacy_type in await cls._legacy_alias_object_types(object_type, object_id):
tuples.extend(await fga.read_tuples(object=f"{legacy_type}:{object_id}"))
if not tuples:
return []
@@ -927,16 +897,6 @@ class PermissionService:
if allowed:
return level.value
for legacy_type in await cls._legacy_alias_object_types(object_type, object_id):
legacy_checks = [
{"user": f"user:{user_id}", "relation": level.value, "object": f"{legacy_type}:{object_id}"}
for level in PermissionLevel
]
legacy_results = await fga.batch_check(legacy_checks)
for level, allowed in zip(PermissionLevel, legacy_results):
if allowed:
return level.value
# Owner/creator decoupled: creator counts as owner only when no other
# owner tuple remains (FGA reachable here). See check() L4 fallback.
return await cls._get_implicit_permission_level_after_gate(
@@ -1109,8 +1069,6 @@ class PermissionService:
if fga is None:
return False
tuples = await fga.read_tuples(object=f"{object_type}:{object_id}", relation="owner")
for legacy_type in await cls._legacy_alias_object_types(object_type, object_id):
tuples.extend(await fga.read_tuples(object=f"{legacy_type}:{object_id}", relation="owner"))
exclude_user = f"user:{exclude_user_id}" if exclude_user_id is not None else None
return any(t.get("relation") == "owner" and t.get("user") != exclude_user for t in (tuples or []))
except Exception as e:
@@ -1446,46 +1404,6 @@ class PermissionService:
login_user,
)
@classmethod
async def _legacy_alias_object_types(
cls,
object_type: str,
object_id: str | None = None,
) -> list[str]:
if object_type != "knowledge_library":
return []
if object_id is None:
return ["knowledge_space"]
try:
from bisheng.knowledge.domain.models.knowledge import KnowledgeDao, KnowledgeTypeEnum
obj = await KnowledgeDao.aquery_by_id(int(object_id))
if obj and obj.type != KnowledgeTypeEnum.SPACE.value:
return ["knowledge_space"]
except Exception as e:
logger.debug("Could not resolve legacy alias object type for %s:%s: %s", object_type, object_id, e)
return []
@classmethod
async def _filter_legacy_alias_ids(
cls,
object_type: str,
ids: list[str],
) -> list[str]:
if object_type != "knowledge_library" or not ids:
return ids
try:
from bisheng.knowledge.domain.models.knowledge import KnowledgeDao, KnowledgeTypeEnum
numeric_ids = [int(one) for one in ids if str(one).isdigit()]
if not numeric_ids:
return []
objects = await KnowledgeDao.aget_list_by_ids(numeric_ids)
return [str(obj.id) for obj in objects if obj.type != KnowledgeTypeEnum.SPACE.value]
except Exception as e:
logger.debug("Could not filter legacy alias ids for %s: %s", object_type, e)
return []
@classmethod
async def _expand_subject(
cls,
src/backend/bisheng/user/domain/services/auth.py
+3 -11
View File
@@ -131,18 +131,10 @@ def _effective_web_menu_strip_orphans(web_menu: list[str]) -> list[str]:
# ── AccessType → ReBAC mapping (F008, AD-02) ────────────────
# Maps old RBAC AccessType to one-or-more (relation, object_type) pairs for
# ReBAC delegation. Knowledge libraries are migrating from the historical
# knowledge_space object type to the dedicated knowledge_library type, so the
# adapter must temporarily accept both.
# ReBAC delegation.
_ACCESS_TYPE_TO_REBAC: dict[int, tuple[tuple[str, str], ...]] = {
AccessType.KNOWLEDGE: (
("can_read", "knowledge_library"),
("can_read", "knowledge_space"),
),
AccessType.KNOWLEDGE_WRITE: (
("can_edit", "knowledge_library"),
("can_edit", "knowledge_space"),
),
AccessType.KNOWLEDGE: (("can_read", "knowledge_library"),),
AccessType.KNOWLEDGE_WRITE: (("can_edit", "knowledge_library"),),
AccessType.WORKFLOW: (("can_read", "workflow"),),
AccessType.WORKFLOW_WRITE: (("can_edit", "workflow"),),
AccessType.ASSISTANT_READ: (("can_read", "assistant"),),
src/backend/test/knowledge/test_knowledge_list_cursor.py
+185 -25
View File
@@ -10,6 +10,7 @@ Covers:
- AD-15 sort_by=name fallback: pseudo-cursor (page_num offset), same response shape
- type=0 (document KB) and type=1 (QA KB) both follow cursor protocol
"""
from __future__ import annotations
from types import SimpleNamespace
@@ -62,10 +63,19 @@ async def test_first_page_returns_cursor_envelope_no_total():
# 21 rows ≥ page_size+1 → has_more=True
rows = [_make_knowledge_row(i) for i in range(1, 22)]
with patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]), \
patch.object(KnowledgeService.permission_service, "get_knowledge_permission_map_async", new_callable=AsyncMock, return_value={1: {"view_kb"}, 2: {"view_kb"}, 3: {"view_kb"}}), \
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows), \
patch.object(KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r):
with (
patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]),
patch.object(
KnowledgeService.permission_service,
"get_knowledge_permission_map_async",
new_callable=AsyncMock,
return_value={i: {"view_kb"} for i in range(1, 22)},
),
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows),
patch.object(
KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r
),
):
result = await KnowledgeService.get_knowledge(
request=MagicMock(),
login_user=login_user,
@@ -92,10 +102,19 @@ async def test_last_page_returns_has_more_false_and_null_cursor():
# Only 5 rows < page_size+1 → has_more=False, next_cursor=None
rows = [_make_knowledge_row(i) for i in range(1, 6)]
with patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]), \
patch.object(KnowledgeService.permission_service, "get_knowledge_permission_map_async", new_callable=AsyncMock, return_value={i: {"view_kb"} for i in [1, 2, 3]}), \
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows), \
patch.object(KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r):
with (
patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]),
patch.object(
KnowledgeService.permission_service,
"get_knowledge_permission_map_async",
new_callable=AsyncMock,
return_value={i: {"view_kb"} for i in range(1, 6)},
),
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows),
patch.object(
KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r
),
):
result = await KnowledgeService.get_knowledge(
request=MagicMock(),
login_user=login_user,
@@ -111,6 +130,62 @@ async def test_last_page_returns_has_more_false_and_null_cursor():
assert len(result.data) == 5
@pytest.mark.asyncio
async def test_permission_scan_refills_from_next_database_batch():
mod = _load_service()
KnowledgeService = mod.KnowledgeService
login_user = _make_login_user(accessible=tuple(str(i) for i in range(1, 101)))
first_batch = [_make_knowledge_row(i) for i in range(1, 51)]
second_batch = [_make_knowledge_row(i) for i in range(51, 101)]
async def permission_map(_user, knowledge_ids, _permission_ids):
allowed_ids = set(range(1, 11)) if knowledge_ids[0] == 1 else set(range(51, 62))
return {knowledge_id: {"view_kb"} for knowledge_id in knowledge_ids if knowledge_id in allowed_ids}
with (
patch.object(
mod.KnowledgeDao,
"aget_knowledge_ids_created_by",
new_callable=AsyncMock,
return_value=[],
),
patch.object(
KnowledgeService.permission_service,
"get_knowledge_permission_map_async",
new_callable=AsyncMock,
side_effect=permission_map,
) as permission_mock,
patch.object(
mod.KnowledgeDao,
"aget_user_knowledge",
new_callable=AsyncMock,
side_effect=[first_batch, second_batch],
) as dao_mock,
patch.object(
KnowledgeService,
"aconvert_knowledge_read",
new_callable=AsyncMock,
side_effect=lambda _user, rows, **_kwargs: rows,
),
):
result = await KnowledgeService.get_knowledge(
request=MagicMock(),
login_user=login_user,
knowledge_type=KnowledgeTypeEnum.NORMAL,
cursor=None,
page_size=20,
sort_by="update_time",
permission_id="view_kb",
)
assert [one.id for one in result.data] == [*range(1, 11), *range(51, 61)]
assert result.has_more is True
assert dao_mock.await_count == 2
assert permission_mock.await_count == 2
assert all(len(call.args[1]) == 50 for call in permission_mock.await_args_list)
assert dao_mock.await_args_list[1].kwargs["cursor"] == [first_batch[-1].update_time, 50]
# ---------------------------------------------------------------------------
# AC-05: cursor continuation decodes correctly and the next_cursor matches the
# last visible item
@@ -125,10 +200,19 @@ async def test_next_cursor_encodes_last_visible_sort_key_and_id():
rows = [_make_knowledge_row(i) for i in range(10, 32)] # 22 rows
with patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]), \
patch.object(KnowledgeService.permission_service, "get_knowledge_permission_map_async", new_callable=AsyncMock, return_value={i: {"view_kb"} for i in [1, 2, 3]}), \
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows), \
patch.object(KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r):
with (
patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]),
patch.object(
KnowledgeService.permission_service,
"get_knowledge_permission_map_async",
new_callable=AsyncMock,
return_value={i: {"view_kb"} for i in range(10, 32)},
),
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows),
patch.object(
KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r
),
):
result = await KnowledgeService.get_knowledge(
request=MagicMock(),
login_user=login_user,
@@ -210,11 +294,20 @@ async def test_acount_user_knowledge_not_called():
login_user = _make_login_user()
rows = [_make_knowledge_row(i) for i in range(1, 6)]
with patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]), \
patch.object(KnowledgeService.permission_service, "get_knowledge_permission_map_async", new_callable=AsyncMock, return_value={i: {"view_kb"} for i in [1, 2, 3]}), \
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows), \
patch.object(mod.KnowledgeDao, "acount_user_knowledge", new_callable=AsyncMock, return_value=0) as count_mock, \
patch.object(KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r):
with (
patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]),
patch.object(
KnowledgeService.permission_service,
"get_knowledge_permission_map_async",
new_callable=AsyncMock,
return_value={i: {"view_kb"} for i in range(1, 6)},
),
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows),
patch.object(mod.KnowledgeDao, "acount_user_knowledge", new_callable=AsyncMock, return_value=0) as count_mock,
patch.object(
KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r
),
):
await KnowledgeService.get_knowledge(
request=MagicMock(),
login_user=login_user,
@@ -240,10 +333,19 @@ async def test_sort_by_name_uses_pseudo_cursor_encoding_page_num():
login_user = _make_login_user()
rows = [_make_knowledge_row(i) for i in range(1, 22)] # 21 rows → has_more
with patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]), \
patch.object(KnowledgeService.permission_service, "get_knowledge_permission_map_async", new_callable=AsyncMock, return_value={i: {"view_kb"} for i in [1, 2, 3]}), \
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows), \
patch.object(KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r):
with (
patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]),
patch.object(
KnowledgeService.permission_service,
"get_knowledge_permission_map_async",
new_callable=AsyncMock,
return_value={i: {"view_kb"} for i in range(1, 22)},
),
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows),
patch.object(
KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r
),
):
result = await KnowledgeService.get_knowledge(
request=MagicMock(),
login_user=login_user,
@@ -265,6 +367,55 @@ async def test_sort_by_name_uses_pseudo_cursor_encoding_page_num():
assert decoded == [2] # first page → next page is 2
@pytest.mark.asyncio
async def test_sort_by_name_second_page_slices_after_visible_rows():
mod = _load_service()
KnowledgeService = mod.KnowledgeService
login_user = _make_login_user(accessible=tuple(str(i) for i in range(1, 11)))
rows = [_make_knowledge_row(i) for i in range(1, 11)]
cursor = encode_cursor((2,), context="knowledge|sort_by=name")
with (
patch.object(
mod.KnowledgeDao,
"aget_knowledge_ids_created_by",
new_callable=AsyncMock,
return_value=[],
),
patch.object(
KnowledgeService.permission_service,
"get_knowledge_permission_map_async",
new_callable=AsyncMock,
return_value={i: {"view_kb"} for i in (2, 4, 6, 8, 10)},
),
patch.object(
mod.KnowledgeDao,
"aget_user_knowledge",
new_callable=AsyncMock,
return_value=rows,
) as dao_mock,
patch.object(
KnowledgeService,
"aconvert_knowledge_read",
new_callable=AsyncMock,
side_effect=lambda _user, visible_rows, **_kwargs: visible_rows,
),
):
result = await KnowledgeService.get_knowledge(
request=MagicMock(),
login_user=login_user,
knowledge_type=KnowledgeTypeEnum.NORMAL,
cursor=cursor,
page_size=2,
sort_by="name",
permission_id="view_kb",
)
assert [one.id for one in result.data] == [6, 8]
assert result.has_more is True
assert dao_mock.await_args.kwargs["page"] == 1
# ---------------------------------------------------------------------------
# AC-01: type=0 and type=1 both follow cursor protocol
# ---------------------------------------------------------------------------
@@ -278,10 +429,19 @@ async def test_both_kb_types_follow_cursor_protocol(ktype):
login_user = _make_login_user()
rows = [_make_knowledge_row(i) for i in range(1, 6)]
with patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]), \
patch.object(KnowledgeService.permission_service, "get_knowledge_permission_map_async", new_callable=AsyncMock, return_value={i: {"view_kb"} for i in [1, 2, 3]}), \
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows), \
patch.object(KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r):
with (
patch.object(mod.KnowledgeDao, "aget_knowledge_ids_created_by", new_callable=AsyncMock, return_value=[]),
patch.object(
KnowledgeService.permission_service,
"get_knowledge_permission_map_async",
new_callable=AsyncMock,
return_value={i: {"view_kb"} for i in range(1, 6)},
),
patch.object(mod.KnowledgeDao, "aget_user_knowledge", new_callable=AsyncMock, return_value=rows),
patch.object(
KnowledgeService, "aconvert_knowledge_read", new_callable=AsyncMock, side_effect=lambda u, r, **kw: r
),
):
result = await KnowledgeService.get_knowledge(
request=MagicMock(),
login_user=login_user,
src/backend/test/knowledge/test_knowledge_list_dao_limit.py
+60
View File
@@ -0,0 +1,60 @@
from contextlib import asynccontextmanager
import pytest
from bisheng.knowledge.domain.models import knowledge as knowledge_module
from bisheng.knowledge.domain.models.knowledge import KnowledgeDao, KnowledgeTypeEnum
class _EmptyResult:
def all(self):
return []
class _CapturingSession:
def __init__(self):
self.statement = None
async def exec(self, statement):
self.statement = statement
return _EmptyResult()
def _session_factory(session):
@asynccontextmanager
async def _context():
yield session
return _context
@pytest.mark.asyncio
async def test_first_user_cursor_page_applies_limit(monkeypatch):
session = _CapturingSession()
monkeypatch.setattr(knowledge_module, "get_async_db_session", _session_factory(session))
await KnowledgeDao.aget_user_knowledge(
user_id=7,
knowledge_id_extra=[1, 2, 3],
knowledge_type=KnowledgeTypeEnum.NORMAL,
page=0,
limit=21,
cursor=None,
)
assert session.statement._limit_clause.value == 21
@pytest.mark.asyncio
async def test_first_admin_cursor_page_applies_limit(monkeypatch):
session = _CapturingSession()
monkeypatch.setattr(knowledge_module, "get_async_db_session", _session_factory(session))
await KnowledgeDao.aget_all_knowledge(
knowledge_type=KnowledgeTypeEnum.NORMAL,
page=0,
limit=21,
cursor=None,
)
assert session.statement._limit_clause.value == 21
src/backend/test/knowledge/test_knowledge_permission_service.py
+208 -157
View File
@@ -3,7 +3,6 @@ from unittest.mock import AsyncMock, patch
import pytest
from bisheng.common.errcode.http_error import UnAuthorizedError
from bisheng.database.models.role_access import AccessType
from bisheng.knowledge.domain.services.knowledge_permission_service import (
_PERMISSION_SYNC_TIMEOUT_SECONDS,
@@ -17,7 +16,7 @@ async def test_ensure_knowledge_read_async_uses_view_permission_id():
login_user = SimpleNamespace(user_id=7)
with patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.check_permission_id_async',
"bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.check_permission_id_async",
new_callable=AsyncMock,
return_value=True,
) as mock_check_permission_id:
@@ -27,7 +26,7 @@ async def test_ensure_knowledge_read_async_uses_view_permission_id():
knowledge_id=12,
)
mock_check_permission_id.assert_awaited_once_with(login_user, 12, 'view_kb')
mock_check_permission_id.assert_awaited_once_with(login_user, 12, "view_kb")
@pytest.mark.asyncio
@@ -36,7 +35,7 @@ async def test_check_access_async_uses_view_permission_id():
login_user = SimpleNamespace(user_id=7)
with patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.check_permission_id_async',
"bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.check_permission_id_async",
new_callable=AsyncMock,
return_value=True,
) as mock_check_permission_id:
@@ -48,7 +47,7 @@ async def test_check_access_async_uses_view_permission_id():
)
assert allowed is True
mock_check_permission_id.assert_awaited_once_with(login_user, 16, 'view_kb')
mock_check_permission_id.assert_awaited_once_with(login_user, 16, "view_kb")
@pytest.mark.asyncio
@@ -57,7 +56,7 @@ async def test_ensure_knowledge_write_async_uses_edit_permission_id():
login_user = SimpleNamespace(user_id=7)
with patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.check_permission_id_async',
"bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.check_permission_id_async",
new_callable=AsyncMock,
return_value=True,
) as mock_check_permission_id:
@@ -67,7 +66,7 @@ async def test_ensure_knowledge_write_async_uses_edit_permission_id():
knowledge_id=18,
)
mock_check_permission_id.assert_awaited_once_with(login_user, 18, 'edit_kb')
mock_check_permission_id.assert_awaited_once_with(login_user, 18, "edit_kb")
@pytest.mark.asyncio
@@ -76,7 +75,7 @@ async def test_ensure_knowledge_delete_async_uses_delete_permission_id():
login_user = SimpleNamespace(user_id=7)
with patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.check_permission_id_async',
"bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.check_permission_id_async",
new_callable=AsyncMock,
return_value=True,
) as mock_check_permission_id:
@@ -86,7 +85,7 @@ async def test_ensure_knowledge_delete_async_uses_delete_permission_id():
knowledge_id=19,
)
mock_check_permission_id.assert_awaited_once_with(login_user, 19, 'delete_kb')
mock_check_permission_id.assert_awaited_once_with(login_user, 19, "delete_kb")
@pytest.mark.asyncio
@@ -97,13 +96,16 @@ async def test_ensure_access_async_raises_when_rebac_denies():
class _Denied(Exception):
pass
with patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.check_permission_id_async',
new_callable=AsyncMock,
return_value=False,
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.UnAuthorizedError',
_Denied,
with (
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.check_permission_id_async",
new_callable=AsyncMock,
return_value=False,
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.UnAuthorizedError",
_Denied,
),
):
with pytest.raises(_Denied):
await service.ensure_access_async(
@@ -123,7 +125,7 @@ def test_check_access_sync_uses_view_permission_id():
return True
with patch(
'bisheng.knowledge.domain.services.knowledge_permission_service._run_async_safe',
"bisheng.knowledge.domain.services.knowledge_permission_service._run_async_safe",
side_effect=_close_and_allow,
) as mock_run_async:
allowed = service.check_access_sync(
@@ -134,8 +136,8 @@ def test_check_access_sync_uses_view_permission_id():
)
assert allowed is True
assert mock_run_async.call_args.args[0].cr_code.co_name == 'check_permission_id_async'
assert mock_run_async.call_args.kwargs['timeout'] == _PERMISSION_SYNC_TIMEOUT_SECONDS
assert mock_run_async.call_args.args[0].cr_code.co_name == "check_permission_id_async"
assert mock_run_async.call_args.kwargs["timeout"] == _PERMISSION_SYNC_TIMEOUT_SECONDS
@pytest.mark.asyncio
@@ -143,14 +145,14 @@ async def test_check_permission_id_async_reads_effective_permission_ids():
login_user = SimpleNamespace(user_id=7)
with patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.get_effective_permission_ids_async',
"bisheng.knowledge.domain.services.knowledge_permission_service.KnowledgePermissionService.get_effective_permission_ids_async",
new_callable=AsyncMock,
return_value={'view_kb'},
return_value={"view_kb"},
) as mock_get_effective_permission_ids:
allowed = await KnowledgePermissionService.check_permission_id_async(
login_user=login_user,
knowledge_id=31,
permission_id='view_kb',
permission_id="view_kb",
)
assert allowed is True
@@ -171,12 +173,15 @@ def test_ensure_access_sync_raises_when_rebac_denies():
coro.close()
return False
with patch(
'bisheng.knowledge.domain.services.knowledge_permission_service._run_async_safe',
side_effect=_close_and_deny,
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.UnAuthorizedError',
_Denied,
with (
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service._run_async_safe",
side_effect=_close_and_deny,
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.UnAuthorizedError",
_Denied,
),
):
with pytest.raises(_Denied):
service.ensure_access_sync(
@@ -195,107 +200,140 @@ async def test_filter_knowledge_ids_by_permission_async_honors_custom_model_perm
get_user_group_ids=AsyncMock(return_value=[]),
)
fake_fga = SimpleNamespace(
read_tuples=AsyncMock(return_value=[
{'user': 'user:7', 'relation': 'viewer', 'object': 'knowledge_library:12'},
]),
read_tuples=AsyncMock(
return_value=[
{"user": "user:7", "relation": "viewer", "object": "knowledge_library:12"},
]
),
)
with patch(
'bisheng.knowledge.domain.services.knowledge_permission_service._get_relation_models',
new_callable=AsyncMock,
return_value=[{
'id': 'custom_view_only',
'name': '只看不用',
'relation': 'viewer',
'grant_tier': 'usage',
'permissions': ['view_kb'],
'permissions_explicit': True,
'is_system': False,
}],
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service._get_bindings',
new_callable=AsyncMock,
return_value=[{
'resource_type': 'knowledge_library',
'resource_id': '12',
'subject_type': 'user',
'subject_id': 7,
'relation': 'viewer',
'include_children': None,
'model_id': 'custom_view_only',
}],
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.PermissionService._get_fga',
return_value=fake_fga,
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.UserDepartmentDao.aget_user_departments',
new_callable=AsyncMock,
return_value=[],
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.DepartmentDao.aget_by_ids',
new_callable=AsyncMock,
return_value=[],
with (
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service._get_relation_models",
new_callable=AsyncMock,
return_value=[
{
"id": "custom_view_only",
"name": "只看不用",
"relation": "viewer",
"grant_tier": "usage",
"permissions": ["view_kb"],
"permissions_explicit": True,
"is_system": False,
}
],
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service._get_bindings",
new_callable=AsyncMock,
return_value=[
{
"resource_type": "knowledge_library",
"resource_id": "12",
"subject_type": "user",
"subject_id": 7,
"relation": "viewer",
"include_children": None,
"model_id": "custom_view_only",
}
],
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.PermissionService._get_fga",
return_value=fake_fga,
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.UserDepartmentDao.aget_user_departments",
new_callable=AsyncMock,
return_value=[],
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.DepartmentDao.aget_by_ids",
new_callable=AsyncMock,
return_value=[],
),
):
assert await service.filter_knowledge_ids_by_permission_async(login_user, [12], 'use_kb') == []
assert await service.filter_knowledge_ids_by_permission_async(login_user, [12], 'view_kb') == [12]
assert await service.filter_knowledge_ids_by_permission_async(login_user, [12], "use_kb") == []
assert await service.filter_knowledge_ids_by_permission_async(login_user, [12], "view_kb") == [12]
@pytest.mark.asyncio
async def test_filter_knowledge_ids_by_permission_async_reads_legacy_knowledge_space_tuples():
async def test_filter_knowledge_ids_by_permission_async_ignores_knowledge_space_tuples():
service = KnowledgePermissionService()
login_user = SimpleNamespace(
user_id=7,
is_admin=lambda: False,
get_user_group_ids=AsyncMock(return_value=[]),
)
fake_fga = SimpleNamespace(
read_tuples=AsyncMock(side_effect=lambda object: (
[] if object == 'knowledge_library:13'
else [{'user': 'user:7', 'relation': 'viewer', 'object': 'knowledge_space:13'}]
)),
read_tuples=AsyncMock(
side_effect=lambda object: (
[]
if object == "knowledge_library:13"
else [{"user": "user:7", "relation": "viewer", "object": "knowledge_space:13"}]
)
),
)
with patch(
'bisheng.knowledge.domain.services.knowledge_permission_service._get_relation_models',
new_callable=AsyncMock,
return_value=[{
'id': 'custom_view_only',
'name': '只看不用',
'relation': 'viewer',
'grant_tier': 'usage',
'permissions': ['view_kb'],
'permissions_explicit': True,
'is_system': False,
}],
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service._get_bindings',
new_callable=AsyncMock,
return_value=[{
'resource_type': 'knowledge_library',
'resource_id': '13',
'subject_type': 'user',
'subject_id': 7,
'relation': 'viewer',
'include_children': None,
'model_id': 'custom_view_only',
}],
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.PermissionService._get_fga',
return_value=fake_fga,
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.PermissionService._legacy_alias_object_types',
new_callable=AsyncMock,
return_value=['knowledge_space'],
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.UserDepartmentDao.aget_user_departments',
new_callable=AsyncMock,
return_value=[],
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.DepartmentDao.aget_by_ids',
new_callable=AsyncMock,
return_value=[],
with (
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service._get_relation_models",
new_callable=AsyncMock,
return_value=[
{
"id": "custom_view_only",
"name": "只看不用",
"relation": "viewer",
"grant_tier": "usage",
"permissions": ["view_kb"],
"permissions_explicit": True,
"is_system": False,
}
],
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service._get_bindings",
new_callable=AsyncMock,
return_value=[
{
"resource_type": "knowledge_library",
"resource_id": "13",
"subject_type": "user",
"subject_id": 7,
"relation": "viewer",
"include_children": None,
"model_id": "custom_view_only",
}
],
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.PermissionService._get_fga",
return_value=fake_fga,
),
patch(
"bisheng.permission.domain.services.fine_grained_permission_service.PermissionService.get_implicit_permission_level",
new_callable=AsyncMock,
return_value=None,
),
patch(
"bisheng.permission.domain.services.fine_grained_permission_service.PermissionService.get_permission_level",
new_callable=AsyncMock,
return_value=None,
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.UserDepartmentDao.aget_user_departments",
new_callable=AsyncMock,
return_value=[],
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.DepartmentDao.aget_by_ids",
new_callable=AsyncMock,
return_value=[],
),
):
assert await service.filter_knowledge_ids_by_permission_async(login_user, [13], 'use_kb') == []
assert await service.filter_knowledge_ids_by_permission_async(login_user, [13], 'view_kb') == [13]
assert await service.filter_knowledge_ids_by_permission_async(login_user, [13], "use_kb") == []
assert await service.filter_knowledge_ids_by_permission_async(login_user, [13], "view_kb") == []
assert {call.kwargs["object"] for call in fake_fga.read_tuples.await_args_list} == {"knowledge_library:13"}
@pytest.mark.asyncio
@@ -306,49 +344,62 @@ async def test_check_permission_id_async_unions_implicit_scope_permissions():
get_user_group_ids=AsyncMock(return_value=[]),
)
fake_fga = SimpleNamespace(
read_tuples=AsyncMock(return_value=[
{'user': 'user:7', 'relation': 'viewer', 'object': 'knowledge_library:12'},
]),
read_tuples=AsyncMock(
return_value=[
{"user": "user:7", "relation": "viewer", "object": "knowledge_library:12"},
]
),
)
with patch(
'bisheng.knowledge.domain.services.knowledge_permission_service._get_relation_models',
new_callable=AsyncMock,
return_value=[{
'id': 'custom_view_only',
'name': '只看不用',
'relation': 'viewer',
'grant_tier': 'usage',
'permissions': ['view_kb'],
'permissions_explicit': True,
'is_system': False,
}],
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service._get_bindings',
new_callable=AsyncMock,
return_value=[{
'resource_type': 'knowledge_library',
'resource_id': '12',
'subject_type': 'user',
'subject_id': 7,
'relation': 'viewer',
'include_children': None,
'model_id': 'custom_view_only',
}],
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.PermissionService._get_fga',
return_value=fake_fga,
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.PermissionService.get_implicit_permission_level',
new_callable=AsyncMock,
return_value='can_manage',
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.UserDepartmentDao.aget_user_departments',
new_callable=AsyncMock,
return_value=[],
), patch(
'bisheng.knowledge.domain.services.knowledge_permission_service.DepartmentDao.aget_by_ids',
new_callable=AsyncMock,
return_value=[],
with (
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service._get_relation_models",
new_callable=AsyncMock,
return_value=[
{
"id": "custom_view_only",
"name": "只看不用",
"relation": "viewer",
"grant_tier": "usage",
"permissions": ["view_kb"],
"permissions_explicit": True,
"is_system": False,
}
],
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service._get_bindings",
new_callable=AsyncMock,
return_value=[
{
"resource_type": "knowledge_library",
"resource_id": "12",
"subject_type": "user",
"subject_id": 7,
"relation": "viewer",
"include_children": None,
"model_id": "custom_view_only",
}
],
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.PermissionService._get_fga",
return_value=fake_fga,
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.PermissionService.get_implicit_permission_level",
new_callable=AsyncMock,
return_value="can_manage",
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.UserDepartmentDao.aget_user_departments",
new_callable=AsyncMock,
return_value=[],
),
patch(
"bisheng.knowledge.domain.services.knowledge_permission_service.DepartmentDao.aget_by_ids",
new_callable=AsyncMock,
return_value=[],
),
):
assert await service.check_permission_id_async(login_user, 12, 'manage_kb_owner') is True
assert await service.check_permission_id_async(login_user, 12, "manage_kb_owner") is True
src/backend/test/knowledge/test_knowledge_service_rebac_bridge.py
+21 -63
View File
@@ -317,22 +317,10 @@ async def test_get_knowledge_lists_from_knowledge_library_object_type():
return_value=[],
),
patch.object(
KnowledgeService.permission_service,
"filter_knowledge_ids_by_permission_async",
KnowledgeService,
"_scan_visible_knowledge",
new_callable=AsyncMock,
return_value=[1],
) as mock_filter_ids,
patch.object(
service_module.KnowledgeDao,
"aget_user_knowledge",
new_callable=AsyncMock,
return_value=[],
),
patch.object(
service_module.KnowledgeDao,
"acount_user_knowledge",
new_callable=AsyncMock,
return_value=0,
return_value=([], {}, 0, 0),
),
patch.object(
service_module.KnowledgeService,
@@ -341,20 +329,14 @@ async def test_get_knowledge_lists_from_knowledge_library_object_type():
return_value=[],
),
):
result, total = await KnowledgeService.get_knowledge(
result = await KnowledgeService.get_knowledge(
request=MagicMock(),
login_user=login_user,
knowledge_type=KnowledgeTypeEnum.NORMAL,
)
assert result == []
assert total == 0
assert result.data == []
login_user.rebac_list_accessible.assert_awaited_once_with("can_read", "knowledge_library")
mock_filter_ids.assert_awaited_once_with(
login_user,
[1],
"use_kb",
)
@pytest.mark.asyncio
@@ -375,23 +357,11 @@ async def test_get_knowledge_merges_creator_owned_ids_into_use_kb_filter_candida
return_value=[9],
),
patch.object(
KnowledgeService.permission_service,
"filter_knowledge_ids_by_permission_async",
KnowledgeService,
"_scan_visible_knowledge",
new_callable=AsyncMock,
return_value=[1, 9],
) as mock_filter_ids,
patch.object(
service_module.KnowledgeDao,
"aget_user_knowledge",
new_callable=AsyncMock,
return_value=[],
),
patch.object(
service_module.KnowledgeDao,
"acount_user_knowledge",
new_callable=AsyncMock,
return_value=0,
),
return_value=([], {}, 0, 0),
) as scan_mock,
patch.object(
service_module.KnowledgeService,
"aconvert_knowledge_read",
@@ -405,10 +375,10 @@ async def test_get_knowledge_merges_creator_owned_ids_into_use_kb_filter_candida
knowledge_type=KnowledgeTypeEnum.NORMAL,
)
filter_args = mock_filter_ids.await_args.args
assert filter_args[0] is login_user
assert set(filter_args[1]) == {1, 9}
assert filter_args[2] == "use_kb"
scan_kwargs = scan_mock.await_args.kwargs
assert scan_kwargs["login_user"] is login_user
assert set(scan_kwargs["candidate_ids"]) == {1, 9}
assert scan_kwargs["permission_id"] == "use_kb"
@pytest.mark.asyncio
@@ -429,23 +399,11 @@ async def test_get_knowledge_supports_view_permission_filter_override():
return_value=[],
),
patch.object(
KnowledgeService.permission_service,
"filter_knowledge_ids_by_permission_async",
KnowledgeService,
"_scan_visible_knowledge",
new_callable=AsyncMock,
return_value=[1],
) as mock_filter_ids,
patch.object(
service_module.KnowledgeDao,
"aget_user_knowledge",
new_callable=AsyncMock,
return_value=[],
),
patch.object(
service_module.KnowledgeDao,
"acount_user_knowledge",
new_callable=AsyncMock,
return_value=0,
),
return_value=([], {}, 0, 0),
) as scan_mock,
patch.object(
service_module.KnowledgeService,
"aconvert_knowledge_read",
@@ -460,10 +418,10 @@ async def test_get_knowledge_supports_view_permission_filter_override():
permission_id="view_kb",
)
filter_args = mock_filter_ids.await_args.args
assert filter_args[0] is login_user
assert filter_args[1] == [1]
assert filter_args[2] == "view_kb"
scan_kwargs = scan_mock.await_args.kwargs
assert scan_kwargs["login_user"] is login_user
assert scan_kwargs["candidate_ids"] == [1]
assert scan_kwargs["permission_id"] == "view_kb"
@pytest.mark.asyncio
src/backend/test/permission/test_auth_knowledge_library_adapter.py
+25 -42
View File
@@ -3,80 +3,63 @@ from unittest.mock import AsyncMock
import pytest
from bisheng.database.models.role_access import AccessType
from bisheng.user.domain.services import auth as auth_mod
@pytest.fixture
def login_user(monkeypatch):
monkeypatch.setattr(auth_mod.UserRoleDao, 'get_user_roles', lambda user_id: [])
return auth_mod.LoginUser(user_id=7, user_name='tester', user_role=[2])
monkeypatch.setattr(auth_mod.UserRoleDao, "get_user_roles", lambda user_id: [])
return auth_mod.LoginUser(user_id=7, user_name="tester", user_role=[2])
def test_sync_access_check_accepts_legacy_and_new_knowledge_targets(login_user, monkeypatch):
monkeypatch.setitem(
auth_mod._ACCESS_TYPE_TO_REBAC,
'legacy-read',
(
('can_read', 'knowledge_library'),
('can_read', 'knowledge_space'),
),
)
def test_sync_access_check_uses_canonical_knowledge_library_target(login_user, monkeypatch):
rebac_check = AsyncMock(side_effect=lambda relation, object_type, object_id: object_type == "knowledge_library")
monkeypatch.setattr(
auth_mod.LoginUser,
'rebac_check',
AsyncMock(side_effect=lambda relation, object_type, object_id: object_type == 'knowledge_space'),
"rebac_check",
rebac_check,
)
monkeypatch.setattr(
'bisheng.permission.domain.services.owner_service._run_async_safe',
"bisheng.permission.domain.services.owner_service._run_async_safe",
lambda coro: asyncio.run(coro),
)
assert login_user.access_check(owner_user_id=99, target_id='12', access_type='legacy-read') is True
assert login_user.access_check(owner_user_id=99, target_id="12", access_type=AccessType.KNOWLEDGE) is True
rebac_check.assert_awaited_once_with("can_read", "knowledge_library", "12")
@pytest.mark.asyncio
async def test_async_access_check_unions_new_and_legacy_knowledge_targets(login_user, monkeypatch):
monkeypatch.setitem(
auth_mod._ACCESS_TYPE_TO_REBAC,
'legacy-read',
(
('can_read', 'knowledge_library'),
('can_read', 'knowledge_space'),
),
)
async def test_async_access_check_uses_canonical_knowledge_library_target(login_user, monkeypatch):
rebac_check = AsyncMock(side_effect=lambda relation, object_type, object_id: object_type == "knowledge_library")
monkeypatch.setattr(
auth_mod.LoginUser,
'rebac_check',
AsyncMock(side_effect=lambda relation, object_type, object_id: object_type == 'knowledge_library'),
"rebac_check",
rebac_check,
)
allowed = await login_user.async_access_check(
owner_user_id=99,
target_id='15',
access_type='legacy-read',
target_id="15",
access_type=AccessType.KNOWLEDGE,
)
assert allowed is True
rebac_check.assert_awaited_once_with("can_read", "knowledge_library", "15")
@pytest.mark.asyncio
async def test_async_accessible_ids_merge_new_and_legacy_knowledge_targets(login_user, monkeypatch):
monkeypatch.setitem(
auth_mod._ACCESS_TYPE_TO_REBAC,
'legacy-read',
(
('can_read', 'knowledge_library'),
('can_read', 'knowledge_space'),
),
async def test_async_accessible_ids_use_canonical_knowledge_library_target(login_user, monkeypatch):
rebac_list_accessible = AsyncMock(
side_effect=lambda relation, object_type: ["1", "2"] if object_type == "knowledge_library" else ["3"]
)
monkeypatch.setattr(
auth_mod.LoginUser,
'rebac_list_accessible',
AsyncMock(side_effect=lambda relation, object_type: (
['1', '2'] if object_type == 'knowledge_library' else ['2', '3']
)),
"rebac_list_accessible",
rebac_list_accessible,
)
ids = await login_user.aget_user_access_resource_ids(['legacy-read'])
ids = await login_user.aget_user_access_resource_ids([AccessType.KNOWLEDGE])
assert set(ids) == {'1', '2', '3'}
assert set(ids) == {"1", "2"}
rebac_list_accessible.assert_awaited_once_with("can_read", "knowledge_library")
src/backend/test/permission/test_legacy_rbac_sync_service.py
+104 -86
View File
@@ -12,20 +12,19 @@ async def test_role_access_signature_expansion_dedupes_to_highest_relation():
with patch.object(
LegacyRBACSyncService,
'_role_access_rows',
"_role_access_rows",
new_callable=AsyncMock,
return_value=[
(2, 'kb-1', 1),
(2, 'kb-1', 3),
(2, 'asst-1', 5),
(2, "kb-1", 1),
(2, "kb-1", 3),
(2, "asst-1", 5),
],
):
result = await LegacyRBACSyncService._role_access_signatures_for_roles(7, {2})
assert result == {
RoleAccessSignature('user:7', 'editor', 'knowledge_library', 'kb-1'),
RoleAccessSignature('user:7', 'editor', 'knowledge_space', 'kb-1'),
RoleAccessSignature('user:7', 'viewer', 'assistant', 'asst-1'),
RoleAccessSignature("user:7", "editor", "knowledge_library", "kb-1"),
RoleAccessSignature("user:7", "viewer", "assistant", "asst-1"),
}
@@ -36,31 +35,35 @@ async def test_sync_user_role_change_deletes_stale_except_explicit_bindings():
RoleAccessSignature,
)
protected = RoleAccessSignature('user:5', 'viewer', 'workflow', 'wf-protected')
stale = RoleAccessSignature('user:5', 'editor', 'workflow', 'wf-old')
desired = RoleAccessSignature('user:5', 'editor', 'workflow', 'wf-new')
protected = RoleAccessSignature("user:5", "viewer", "workflow", "wf-protected")
stale = RoleAccessSignature("user:5", "editor", "workflow", "wf-old")
desired = RoleAccessSignature("user:5", "editor", "workflow", "wf-new")
with patch.object(
LegacyRBACSyncService,
'_role_access_signatures_for_roles',
new_callable=AsyncMock,
side_effect=[{protected, stale}, {desired}],
), patch.object(
LegacyRBACSyncService,
'_resource_permission_user_binding_set',
new_callable=AsyncMock,
return_value={protected},
), patch.object(
LegacyRBACSyncService,
'_write_operations',
new_callable=AsyncMock,
) as write_ops:
with (
patch.object(
LegacyRBACSyncService,
"_role_access_signatures_for_roles",
new_callable=AsyncMock,
side_effect=[{protected, stale}, {desired}],
),
patch.object(
LegacyRBACSyncService,
"_resource_permission_user_binding_set",
new_callable=AsyncMock,
return_value={protected},
),
patch.object(
LegacyRBACSyncService,
"_write_operations",
new_callable=AsyncMock,
) as write_ops,
):
await LegacyRBACSyncService.sync_user_role_change(5, [2], [3])
operations, affected = write_ops.await_args.args
assert {(op.action, op.user, op.relation, op.object) for op in operations} == {
('delete', 'user:5', 'editor', 'workflow:wf-old'),
('write', 'user:5', 'editor', 'workflow:wf-new'),
("delete", "user:5", "editor", "workflow:wf-old"),
("write", "user:5", "editor", "workflow:wf-new"),
}
assert affected == {5}
@@ -72,39 +75,45 @@ async def test_sync_role_deleted_preserves_grants_from_remaining_roles():
RoleAccessSignature,
)
shared = RoleAccessSignature('user:5', 'viewer', 'workflow', 'wf-shared')
removed = RoleAccessSignature('user:5', 'editor', 'workflow', 'wf-old')
shared = RoleAccessSignature("user:5", "viewer", "workflow", "wf-shared")
removed = RoleAccessSignature("user:5", "editor", "workflow", "wf-old")
with patch.object(
LegacyRBACSyncService,
'_user_ids_for_role',
new_callable=AsyncMock,
return_value=[5],
), patch.object(
LegacyRBACSyncService,
'_role_ids_for_user',
new_callable=AsyncMock,
return_value={2, 3},
), patch.object(
LegacyRBACSyncService,
'_role_access_signatures_for_roles',
new_callable=AsyncMock,
side_effect=[{shared, removed}, {shared}],
), patch.object(
LegacyRBACSyncService,
'_resource_permission_user_binding_set',
new_callable=AsyncMock,
return_value=set(),
), patch.object(
LegacyRBACSyncService,
'_write_operations',
new_callable=AsyncMock,
) as write_ops:
with (
patch.object(
LegacyRBACSyncService,
"_user_ids_for_role",
new_callable=AsyncMock,
return_value=[5],
),
patch.object(
LegacyRBACSyncService,
"_role_ids_for_user",
new_callable=AsyncMock,
return_value={2, 3},
),
patch.object(
LegacyRBACSyncService,
"_role_access_signatures_for_roles",
new_callable=AsyncMock,
side_effect=[{shared, removed}, {shared}],
),
patch.object(
LegacyRBACSyncService,
"_resource_permission_user_binding_set",
new_callable=AsyncMock,
return_value=set(),
),
patch.object(
LegacyRBACSyncService,
"_write_operations",
new_callable=AsyncMock,
) as write_ops,
):
await LegacyRBACSyncService.sync_role_deleted(2)
operations, affected = write_ops.await_args.args
assert [(op.action, op.user, op.relation, op.object) for op in operations] == [
('delete', 'user:5', 'editor', 'workflow:wf-old'),
("delete", "user:5", "editor", "workflow:wf-old"),
]
assert affected == {5}
@@ -116,21 +125,24 @@ async def test_sync_user_role_change_writes_super_admin_tuple_for_admin_role():
LegacyRBACSyncService,
)
with patch.object(
LegacyRBACSyncService,
'_role_access_signatures_for_roles',
new_callable=AsyncMock,
return_value=set(),
), patch.object(
LegacyRBACSyncService,
'_write_operations',
new_callable=AsyncMock,
) as write_ops:
with (
patch.object(
LegacyRBACSyncService,
"_role_access_signatures_for_roles",
new_callable=AsyncMock,
return_value=set(),
),
patch.object(
LegacyRBACSyncService,
"_write_operations",
new_callable=AsyncMock,
) as write_ops,
):
await LegacyRBACSyncService.sync_user_role_change(1, [], [AdminRole])
operations, affected = write_ops.await_args_list[0].args
assert [(op.action, op.user, op.relation, op.object) for op in operations] == [
('write', 'user:1', 'super_admin', 'system:global'),
("write", "user:1", "super_admin", "system:global"),
]
assert affected == [1]
@@ -141,15 +153,18 @@ async def test_sync_user_auth_created_writes_group_memberships():
LegacyRBACSyncService,
)
with patch.object(
LegacyRBACSyncService,
'sync_user_role_change',
new_callable=AsyncMock,
) as role_sync, patch.object(
LegacyRBACSyncService,
'_write_operations',
new_callable=AsyncMock,
) as write_ops:
with (
patch.object(
LegacyRBACSyncService,
"sync_user_role_change",
new_callable=AsyncMock,
) as role_sync,
patch.object(
LegacyRBACSyncService,
"_write_operations",
new_callable=AsyncMock,
) as write_ops,
):
await LegacyRBACSyncService.sync_user_auth_created(
5,
[2],
@@ -160,8 +175,8 @@ async def test_sync_user_auth_created_writes_group_memberships():
role_sync.assert_awaited_once_with(5, [], [2])
operations, affected = write_ops.await_args.args
assert {(op.action, op.user, op.relation, op.object) for op in operations} == {
('write', 'user:5', 'member', 'user_group:7'),
('write', 'user:5', 'admin', 'user_group:8'),
("write", "user:5", "member", "user_group:7"),
("write", "user:5", "admin", "user_group:8"),
}
assert affected == [5]
@@ -172,17 +187,20 @@ async def test_group_change_handler_invalidates_direct_user_cache():
from bisheng.user_group.domain.services.group_change_handler import GroupChangeHandler
ops = [
TupleOperation(action='write', user='user:9', relation='member', object='user_group:3'),
TupleOperation(action='write', user='user_group:3#member', relation='viewer', object='workflow:wf-1'),
TupleOperation(action="write", user="user:9", relation="member", object="user_group:3"),
TupleOperation(action="write", user="user_group:3#member", relation="viewer", object="workflow:wf-1"),
]
with patch(
'bisheng.permission.domain.services.permission_service.PermissionService.batch_write_tuples',
new_callable=AsyncMock,
) as batch_write, patch(
'bisheng.permission.domain.services.permission_cache.PermissionCache.invalidate_user',
new_callable=AsyncMock,
) as invalidate:
with (
patch(
"bisheng.permission.domain.services.permission_service.PermissionService.batch_write_tuples",
new_callable=AsyncMock,
) as batch_write,
patch(
"bisheng.permission.domain.services.permission_cache.PermissionCache.invalidate_user",
new_callable=AsyncMock,
) as invalidate,
):
await GroupChangeHandler.execute_async(ops)
batch_write.assert_awaited_once_with(ops, crash_safe=True)
src/backend/test/permission/test_permission_enrichment.py
+2 -16
View File
@@ -400,8 +400,7 @@ class TestGetResourcePermissionsIntegration:
assert result == []
@pytest.mark.asyncio
async def test_knowledge_library_permissions_merge_legacy_knowledge_space_tuples(self, mock_fga):
"""knowledge_library should still surface historical knowledge_space tuples."""
async def test_knowledge_library_permissions_ignore_knowledge_space_tuples(self, mock_fga):
from bisheng.permission.domain.services.permission_service import PermissionService
await mock_fga.write_tuples(
@@ -410,25 +409,12 @@ class TestGetResourcePermissionsIntegration:
]
)
mock_user1 = AsyncMock()
mock_user1.user_id = 1
mock_user1.user_name = "Admin"
with (
patch.object(PermissionService, "_get_fga", return_value=mock_fga),
patch.object(
PermissionService,
"_legacy_alias_object_types",
new_callable=AsyncMock,
return_value=["knowledge_space"],
),
patch.object(UserDao, "aget_user_by_ids", new_callable=AsyncMock, return_value=[mock_user1]),
):
result = await PermissionService.get_resource_permissions(
object_type="knowledge_library",
object_id="42",
)
assert len(result) == 1
assert result[0].relation == "owner"
assert result[0].subject_name == "Admin"
assert result == []
src/backend/test/permission/test_permission_service.py
+8 -33
View File
@@ -333,7 +333,7 @@ class TestPermissionServiceCheck:
assert result is True
@pytest.mark.asyncio
async def test_check_knowledge_library_accepts_legacy_knowledge_space_tuple(
async def test_check_knowledge_library_ignores_knowledge_space_tuple(
self,
mock_fga,
mock_login_user_normal,
@@ -346,12 +346,6 @@ class TestPermissionServiceCheck:
with (
patch.object(PermissionService, "_get_fga", return_value=mock_fga),
patch.object(
PermissionService,
"_legacy_alias_object_types",
new_callable=AsyncMock,
return_value=["knowledge_space"],
),
patch(
"bisheng.permission.domain.services.permission_cache.PermissionCache.get_check",
new_callable=AsyncMock,
@@ -370,7 +364,7 @@ class TestPermissionServiceCheck:
login_user=mock_login_user_normal,
)
assert result is True
assert result is False
class TestPermissionServiceListAccessible:
@@ -426,7 +420,7 @@ class TestPermissionServiceListAccessible:
assert sorted(result) == ["abc", "def"]
@pytest.mark.asyncio
async def test_knowledge_library_list_unions_legacy_ids(self, mock_fga, mock_login_user_normal):
async def test_knowledge_library_list_uses_canonical_object_type(self, mock_fga, mock_login_user_normal):
from bisheng.permission.domain.services.permission_service import PermissionService
await mock_fga.write_tuples(
@@ -444,13 +438,6 @@ class TestPermissionServiceListAccessible:
new_callable=AsyncMock,
side_effect=lambda ids, *_args, **_kwargs: ids,
),
patch.object(
PermissionService,
"_legacy_alias_object_types",
new_callable=AsyncMock,
return_value=["knowledge_space"],
),
patch.object(PermissionService, "_filter_legacy_alias_ids", new_callable=AsyncMock, return_value=["def"]),
patch(
"bisheng.permission.domain.services.permission_cache.PermissionCache.get_list_objects",
new_callable=AsyncMock,
@@ -468,7 +455,7 @@ class TestPermissionServiceListAccessible:
login_user=mock_login_user_normal,
)
assert sorted(result) == ["abc", "def"]
assert result == ["abc"]
@pytest.mark.asyncio
async def test_fga_unavailable_still_returns_creator_owned_ids(self, mock_login_user_normal):
@@ -563,17 +550,11 @@ class TestPermissionServiceAuthorize:
mock_fga.assert_tuple_count(0)
@pytest.mark.asyncio
async def test_authorize_knowledge_library_dual_writes_legacy_knowledge_space(self, mock_fga):
async def test_authorize_knowledge_library_writes_only_canonical_object(self, mock_fga):
from bisheng.permission.domain.services.permission_service import PermissionService
with (
patch.object(PermissionService, "_get_fga", return_value=mock_fga),
patch.object(
PermissionService,
"_legacy_alias_object_types",
new_callable=AsyncMock,
return_value=["knowledge_space"],
),
patch(
"bisheng.permission.domain.services.permission_cache.PermissionCache.invalidate_user",
new_callable=AsyncMock,
@@ -592,7 +573,7 @@ class TestPermissionServiceAuthorize:
)
mock_fga.assert_tuple_exists("user:5", "viewer", "knowledge_library:abc")
mock_fga.assert_tuple_exists("user:5", "viewer", "knowledge_space:abc")
mock_fga.assert_tuple_count(1)
@pytest.mark.asyncio
async def test_authorize_department_invalidates_expanded_users(self, mock_fga):
@@ -780,7 +761,7 @@ class TestPermissionServiceBatchWrite:
class TestPermissionServiceGetPermissionLevel:
@pytest.mark.asyncio
async def test_knowledge_library_permission_level_uses_legacy_knowledge_space_tuples(
async def test_knowledge_library_permission_level_ignores_knowledge_space_tuples(
self,
mock_fga,
mock_login_user_normal,
@@ -793,12 +774,6 @@ class TestPermissionServiceGetPermissionLevel:
with (
patch.object(PermissionService, "_get_fga", return_value=mock_fga),
patch.object(
PermissionService,
"_legacy_alias_object_types",
new_callable=AsyncMock,
return_value=["knowledge_space"],
),
patch.object(PermissionService, "_get_resource_creator", new_callable=AsyncMock, return_value=None),
):
result = await PermissionService.get_permission_level(
@@ -808,7 +783,7 @@ class TestPermissionServiceGetPermissionLevel:
login_user=mock_login_user_normal,
)
assert result == "can_edit"
assert result is None
class TestExpandSubject:
src/backend/test/permission/test_permission_service_authorize_collision.py
+48 -32
View File
@@ -17,28 +17,36 @@ async def test_authorize_write_wins_over_same_tuple_delete():
captured = {}
async def _capture(operations, **kwargs):
captured['operations'] = operations
captured["operations"] = operations
with patch.object(
PermissionService, '_legacy_alias_object_types', new=AsyncMock(return_value=[]),
), patch.object(
PermissionService, '_expand_subject', new=AsyncMock(return_value=['user:2']),
), patch.object(
PermissionService, '_affected_user_ids_for_subject', new=AsyncMock(return_value=set()),
), patch.object(
PermissionService, 'batch_write_tuples', new=AsyncMock(side_effect=_capture),
with (
patch.object(
PermissionService,
"_expand_subject",
new=AsyncMock(return_value=["user:2"]),
),
patch.object(
PermissionService,
"_affected_user_ids_for_subject",
new=AsyncMock(return_value=set()),
),
patch.object(
PermissionService,
"batch_write_tuples",
new=AsyncMock(side_effect=_capture),
),
):
await PermissionService.authorize(
object_type='channel',
object_id='c1',
grants=[AuthorizeGrantItem(subject_type='user', subject_id=2, relation='manager')],
revokes=[AuthorizeRevokeItem(subject_type='user', subject_id=2, relation='manager')],
object_type="channel",
object_id="c1",
grants=[AuthorizeGrantItem(subject_type="user", subject_id=2, relation="manager")],
revokes=[AuthorizeRevokeItem(subject_type="user", subject_id=2, relation="manager")],
)
ops = captured['operations']
ops = captured["operations"]
actions = {(op.action, op.user, op.relation, op.object) for op in ops}
assert ('write', 'user:2', 'manager', 'channel:c1') in actions
assert ('delete', 'user:2', 'manager', 'channel:c1') not in actions
assert ("write", "user:2", "manager", "channel:c1") in actions
assert ("delete", "user:2", "manager", "channel:c1") not in actions
@pytest.mark.asyncio
@@ -48,24 +56,32 @@ async def test_authorize_keeps_delete_of_a_different_relation():
captured = {}
async def _capture(operations, **kwargs):
captured['operations'] = operations
captured["operations"] = operations
with patch.object(
PermissionService, '_legacy_alias_object_types', new=AsyncMock(return_value=[]),
), patch.object(
PermissionService, '_expand_subject', new=AsyncMock(return_value=['user:2']),
), patch.object(
PermissionService, '_affected_user_ids_for_subject', new=AsyncMock(return_value=set()),
), patch.object(
PermissionService, 'batch_write_tuples', new=AsyncMock(side_effect=_capture),
with (
patch.object(
PermissionService,
"_expand_subject",
new=AsyncMock(return_value=["user:2"]),
),
patch.object(
PermissionService,
"_affected_user_ids_for_subject",
new=AsyncMock(return_value=set()),
),
patch.object(
PermissionService,
"batch_write_tuples",
new=AsyncMock(side_effect=_capture),
),
):
await PermissionService.authorize(
object_type='channel',
object_id='c1',
grants=[AuthorizeGrantItem(subject_type='user', subject_id=2, relation='editor')],
revokes=[AuthorizeRevokeItem(subject_type='user', subject_id=2, relation='viewer')],
object_type="channel",
object_id="c1",
grants=[AuthorizeGrantItem(subject_type="user", subject_id=2, relation="editor")],
revokes=[AuthorizeRevokeItem(subject_type="user", subject_id=2, relation="viewer")],
)
actions = {(op.action, op.user, op.relation, op.object) for op in captured['operations']}
assert ('write', 'user:2', 'editor', 'channel:c1') in actions
assert ('delete', 'user:2', 'viewer', 'channel:c1') in actions
actions = {(op.action, op.user, op.relation, op.object) for op in captured["operations"]}
assert ("write", "user:2", "editor", "channel:c1") in actions
assert ("delete", "user:2", "viewer", "channel:c1") in actions