feat(auth): 后台登录、角色权限与会话访问策略
管理员与客服分角色登录，权限由后端判定；前端补齐登录、人员与组织管理页面。
This commit is contained in:
WangShen007 committed
1 parent
9eb8613c05
commit
556b650c7a
123 files changed
+25293
No files matched your search
@@ -0,0 +1,204 @@
|
||||
"""
|
||||
Copyright 2024-2026 ReplyPier
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
"""
|
||||
|
||||
"""Turn the chat-scope toggles on the user form into a role.
|
||||
|
||||
Permissions hang off roles, not off people, so "let this agent read every
|
||||
conversation while their colleague is on leave" has to resolve to *some* role.
|
||||
Rather than inventing per-user grants — which would mean every permission check
|
||||
in the codebase learning about a second source of truth — the two toggles
|
||||
describe the chat scope a person should have, and this module finds the role in
|
||||
their organization that matches. A new role is created only when no existing
|
||||
one does, so the common case adds nothing to the Roles screen.
|
||||
"""
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, Set
|
||||
from uuid import UUID
|
||||
|
||||
from sqlalchemy.orm import Session, joinedload
|
||||
|
||||
from app.core.auth import require_grantable
|
||||
from app.core.logger import get_logger
|
||||
from app.models.permission import Permission
|
||||
from app.models.role import Role
|
||||
from app.models.user import User
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
# The unclaimed queue: sessions the AI is still handling that nobody has taken.
|
||||
AI_QUEUE_PERMISSION = "view_unassigned_chats"
|
||||
# Every conversation in the organization, including other agents' assignments.
|
||||
ALL_CHATS_PERMISSION = "view_all_chats"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ChatScope:
|
||||
"""What the two toggles on the user form are asking for."""
|
||||
|
||||
see_all_ai_chats: bool
|
||||
see_all_org_chats: bool
|
||||
|
||||
@classmethod
|
||||
def of(cls, role: Role) -> "ChatScope":
|
||||
"""The scope a role already grants — what the form shows on open."""
|
||||
names = {p.name for p in role.permissions}
|
||||
return cls(
|
||||
see_all_ai_chats=AI_QUEUE_PERMISSION in names,
|
||||
see_all_org_chats=ALL_CHATS_PERMISSION in names,
|
||||
)
|
||||
|
||||
def label(self) -> str:
|
||||
if self.see_all_org_chats and self.see_all_ai_chats:
|
||||
return "all chats"
|
||||
if self.see_all_org_chats:
|
||||
return "all chats, no AI queue"
|
||||
if self.see_all_ai_chats:
|
||||
return "assigned + AI queue"
|
||||
return "assigned only"
|
||||
|
||||
|
||||
def resolve_role(
|
||||
db: Session,
|
||||
current_user: User,
|
||||
base_role: Role,
|
||||
see_all_ai_chats: Optional[bool],
|
||||
see_all_org_chats: Optional[bool],
|
||||
) -> Role:
|
||||
"""The role that grants `base_role` plus/minus the requested chat scope.
|
||||
|
||||
Either toggle may be None, meaning "whatever the role already does" — an
|
||||
API client that has never heard of these fields gets exactly the old
|
||||
behaviour.
|
||||
"""
|
||||
current = ChatScope.of(base_role)
|
||||
wanted = ChatScope(
|
||||
see_all_ai_chats=(
|
||||
current.see_all_ai_chats if see_all_ai_chats is None else see_all_ai_chats
|
||||
),
|
||||
see_all_org_chats=(
|
||||
current.see_all_org_chats if see_all_org_chats is None else see_all_org_chats
|
||||
),
|
||||
)
|
||||
if wanted == current:
|
||||
return base_role
|
||||
|
||||
held = {p.name for p in base_role.permissions}
|
||||
# super_admin already passes every check, so widening it would create a
|
||||
# near-duplicate role that grants nothing new.
|
||||
if "super_admin" in held:
|
||||
return base_role
|
||||
|
||||
desired = set(held)
|
||||
for granted, permission in (
|
||||
(wanted.see_all_ai_chats, AI_QUEUE_PERMISSION),
|
||||
(wanted.see_all_org_chats, ALL_CHATS_PERMISSION),
|
||||
):
|
||||
if granted:
|
||||
desired.add(permission)
|
||||
else:
|
||||
desired.discard(permission)
|
||||
|
||||
# Same rule as the Roles editor: an admin cannot hand out what they do not
|
||||
# hold themselves. Only the added permissions are checked, not the base
|
||||
# role's own — someone with manage_users can already assign any role in the
|
||||
# org as it stands, so carrying its permissions across grants nothing new.
|
||||
require_grantable(current_user, desired - held)
|
||||
|
||||
existing = _role_granting(db, base_role.organization_id, desired)
|
||||
if existing is not None:
|
||||
return existing
|
||||
return _create_scoped_role(db, base_role, desired, wanted)
|
||||
|
||||
|
||||
#: Every label label() can produce, so a name can be stripped back down again.
|
||||
SCOPE_LABELS = tuple(
|
||||
ChatScope(ai, org).label() for ai in (True, False) for org in (True, False)
|
||||
)
|
||||
_SCOPE_SUFFIX = re.compile(
|
||||
r" \((?:%s)\)(?: \d+)?$" % "|".join(re.escape(label) for label in SCOPE_LABELS)
|
||||
)
|
||||
|
||||
|
||||
def _unscoped(name: str) -> str:
|
||||
"""`Agent (all chats)` -> `Agent`.
|
||||
|
||||
Re-scoping someone who is already on a derived role would otherwise stack
|
||||
suffixes: "Agent (all chats) (assigned only)".
|
||||
"""
|
||||
return _SCOPE_SUFFIX.sub("", name)
|
||||
|
||||
|
||||
def _role_granting(db: Session, organization_id: UUID, desired: Set[str]) -> Optional[Role]:
|
||||
"""An existing role in the org whose permissions are exactly `desired`.
|
||||
|
||||
Exact, not superset: a role that also grants manage_users would hand the
|
||||
person far more than the toggle asked for.
|
||||
"""
|
||||
roles = (
|
||||
db.query(Role)
|
||||
.options(joinedload(Role.permissions))
|
||||
.filter(Role.organization_id == organization_id)
|
||||
.all()
|
||||
)
|
||||
for role in roles:
|
||||
if {p.name for p in role.permissions} == desired:
|
||||
return role
|
||||
return None
|
||||
|
||||
|
||||
def _create_scoped_role(
|
||||
db: Session, base_role: Role, desired: Set[str], scope: ChatScope
|
||||
) -> Role:
|
||||
role = Role(
|
||||
name=_available_name(db, base_role, scope),
|
||||
description=f"{_unscoped(base_role.name)}, scoped to {scope.label()}",
|
||||
organization_id=base_role.organization_id,
|
||||
# Never the org default. get_default_role() picks the default for
|
||||
# invited users, and a scope chosen for one person is not that.
|
||||
is_default=False,
|
||||
)
|
||||
role.permissions = (
|
||||
db.query(Permission).filter(Permission.name.in_(desired)).all()
|
||||
)
|
||||
db.add(role)
|
||||
db.flush()
|
||||
logger.info(
|
||||
"Created chat-scope role %r (%s) for organization %s",
|
||||
role.name, scope.label(), base_role.organization_id,
|
||||
)
|
||||
return role
|
||||
|
||||
|
||||
def _available_name(db: Session, base_role: Role, scope: ChatScope) -> str:
|
||||
"""`Agent (all chats)`, or the first free numbered variant of it.
|
||||
|
||||
Role names are not unique in the schema, but two roles sharing a name in
|
||||
one org is confusing in every screen that lists them.
|
||||
"""
|
||||
taken = {
|
||||
name for (name,) in db.query(Role.name).filter(
|
||||
Role.organization_id == base_role.organization_id
|
||||
)
|
||||
}
|
||||
preferred = f"{_unscoped(base_role.name)} ({scope.label()})"
|
||||
if preferred not in taken:
|
||||
return preferred
|
||||
suffix = 2
|
||||
while f"{preferred} {suffix}" in taken:
|
||||
suffix += 1
|
||||
return f"{preferred} {suffix}"
|
||||
Reference in new issue
Block a user