264 lines
12 KiB
Python
264 lines
12 KiB
Python
"""Workspaces mit serverseitiger Autorisierung (→ Konzept 6.2, 6.3).
|
|
|
|
Jede Funktion verlangt einen Principal und prüft selbst. Fremde Workspaces gibt
|
|
es nicht: Wer kein Mitglied ist, bekommt NotFound, auch globale Administratoren.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import audit
|
|
from app.core.errors import Forbidden, Invalid, NotFound
|
|
from app.identity import Principal
|
|
from app.identity.models import Team, TeamMembership, User
|
|
from app.workspaces.models import APPROVAL_MODES, APPROVER_ROLES, ROLES, Workspace, WorkspaceMembership
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Membership:
|
|
workspace: Workspace
|
|
role: str
|
|
|
|
@property
|
|
def is_admin(self) -> bool:
|
|
return self.role == "admin"
|
|
|
|
@property
|
|
def may_approve(self) -> bool:
|
|
return self.role in APPROVER_ROLES
|
|
|
|
|
|
def list_for(session: Session, principal: Principal) -> list[Membership]:
|
|
rows = session.execute(
|
|
select(Workspace, WorkspaceMembership.role)
|
|
.join(WorkspaceMembership, WorkspaceMembership.workspace_id == Workspace.id)
|
|
.where(WorkspaceMembership.user_id == principal.user_id, Workspace.team_id.in_(principal.team_ids),
|
|
Workspace.deleted.is_(None))
|
|
.order_by(Workspace.name))
|
|
return [Membership(w, role) for w, role in rows]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Stats:
|
|
chats: int = 0
|
|
sources: int = 0
|
|
last: datetime | None = None
|
|
|
|
|
|
def stats(session: Session, workspace_ids: list[int]) -> dict[int, Stats]:
|
|
"""Zahl der Chats und Quellen und letzte Aktivität je Workspace (für die Übersicht)."""
|
|
from app.chat.models import ChatSession, Message
|
|
from app.data_sources.models import DataSource
|
|
if not workspace_ids:
|
|
return {}
|
|
chats = dict(session.execute(select(ChatSession.workspace_id, func.count()).where(
|
|
ChatSession.workspace_id.in_(workspace_ids)).group_by(ChatSession.workspace_id)).all())
|
|
sources = dict(session.execute(select(DataSource.workspace_id, func.count()).where(
|
|
DataSource.workspace_id.in_(workspace_ids)).group_by(DataSource.workspace_id)).all())
|
|
last = dict(session.execute(select(ChatSession.workspace_id, func.max(Message.created))
|
|
.join(Message, Message.chat_session_id == ChatSession.id)
|
|
.where(ChatSession.workspace_id.in_(workspace_ids))
|
|
.group_by(ChatSession.workspace_id)).all())
|
|
return {i: Stats(chats.get(i, 0), sources.get(i, 0), last.get(i)) for i in workspace_ids}
|
|
|
|
|
|
def get(session: Session, principal: Principal, workspace_id: int) -> Membership:
|
|
row = session.execute(
|
|
select(Workspace, WorkspaceMembership.role)
|
|
.join(WorkspaceMembership, WorkspaceMembership.workspace_id == Workspace.id)
|
|
.where(Workspace.id == workspace_id, WorkspaceMembership.user_id == principal.user_id,
|
|
# Zweite Sicherung: wer das Team verlassen hat, kommt nicht mehr hinein
|
|
Workspace.team_id.in_(principal.team_ids),
|
|
# Workspaces im Papierkorb gibt es für alle Funktionen nicht mehr
|
|
Workspace.deleted.is_(None))).first()
|
|
if row is None:
|
|
raise NotFound
|
|
return Membership(*row)
|
|
|
|
|
|
def require_admin(session: Session, principal: Principal, workspace_id: int) -> Membership:
|
|
membership = get(session, principal, workspace_id)
|
|
if not membership.is_admin:
|
|
raise Forbidden
|
|
return membership
|
|
|
|
|
|
def teams_for(session: Session, principal: Principal) -> list[Team]:
|
|
return list(session.scalars(select(Team).where(Team.id.in_(principal.team_ids)).order_by(Team.name)))
|
|
|
|
|
|
def create(session: Session, principal: Principal, team_id: int, name: str, description: str = "") -> Workspace:
|
|
name = name.strip()
|
|
if not name:
|
|
raise Invalid("Name fehlt.")
|
|
if team_id not in principal.team_ids:
|
|
raise Invalid("Workspaces können nur für eigene Teams angelegt werden.")
|
|
workspace = Workspace(team_id=team_id, name=name[:200], description=description.strip(),
|
|
created_by=principal.user_id)
|
|
session.add(workspace)
|
|
session.flush()
|
|
session.add(WorkspaceMembership(workspace_id=workspace.id, user_id=principal.user_id, role="admin"))
|
|
audit.record(session, "workspace.created", user_id=principal.user_id, workspace_id=workspace.id,
|
|
target_kind="workspace", target_id=workspace.id, team_id=team_id)
|
|
return workspace
|
|
|
|
|
|
def update(session: Session, principal: Principal, workspace_id: int, name: str, description: str) -> Workspace:
|
|
workspace = require_admin(session, principal, workspace_id).workspace
|
|
name = " ".join(name.split())
|
|
if not name:
|
|
raise Invalid("Name fehlt.")
|
|
changed = {k: v for k, v in (("name", name[:200]), ("description", description.strip()[:2000]))
|
|
if getattr(workspace, k) != v}
|
|
for key, value in changed.items():
|
|
setattr(workspace, key, value)
|
|
if changed:
|
|
audit.record(session, "workspace.updated", user_id=principal.user_id, workspace_id=workspace_id,
|
|
target_kind="workspace", target_id=workspace_id, **changed)
|
|
return workspace
|
|
|
|
|
|
MAX_EXAMPLES, MAX_EXAMPLE_LENGTH = 8, 300
|
|
|
|
|
|
def examples(workspace: Workspace) -> list[str]:
|
|
"""Beispielfragen für Übersicht und leeren Chat; ein Klick startet einen Chat damit."""
|
|
value = (workspace.settings or {}).get("examples")
|
|
return [e for e in value if isinstance(e, str)] if isinstance(value, list) else []
|
|
|
|
|
|
def set_examples(session: Session, principal: Principal, workspace_id: int, text: str) -> Workspace:
|
|
"""Eine Frage je Zeile; leer = keine."""
|
|
workspace = require_admin(session, principal, workspace_id).workspace
|
|
found = [" ".join(line.split()) for line in text.splitlines() if line.strip()]
|
|
if len(found) > MAX_EXAMPLES:
|
|
raise Invalid(f"Höchstens {MAX_EXAMPLES} Beispielfragen.")
|
|
if any(len(e) > MAX_EXAMPLE_LENGTH for e in found):
|
|
raise Invalid(f"Eine Beispielfrage darf höchstens {MAX_EXAMPLE_LENGTH} Zeichen haben.")
|
|
current = dict(workspace.settings or {})
|
|
if found:
|
|
current["examples"] = found
|
|
else:
|
|
current.pop("examples", None)
|
|
if current != (workspace.settings or {}):
|
|
workspace.settings = current
|
|
audit.record(session, "workspace.examples_set", user_id=principal.user_id, workspace_id=workspace_id,
|
|
target_kind="workspace", target_id=workspace_id, count=len(found))
|
|
return workspace
|
|
|
|
|
|
def tool_rounds(workspace: Workspace, settings) -> int:
|
|
"""Tool-Runden je Antwort: Einstellung des Workspace, sonst AIW_MAX_TOOL_ROUNDS; nie über dem Limit."""
|
|
value = (workspace.settings or {}).get("max_tool_rounds")
|
|
rounds = value if isinstance(value, int) and value > 0 else settings.max_tool_rounds
|
|
return min(rounds, settings.max_tool_rounds_limit)
|
|
|
|
|
|
def chat_only(workspace: Workspace) -> bool:
|
|
"""Chat ohne Datenquellen erlaubt (Ausnahme vom Grundsatz „Context first“)."""
|
|
return bool((workspace.settings or {}).get("chat_only"))
|
|
|
|
|
|
def set_chat_only(session: Session, principal: Principal, workspace_id: int, enabled: bool) -> Workspace:
|
|
workspace = require_admin(session, principal, workspace_id).workspace
|
|
current = dict(workspace.settings or {})
|
|
if enabled:
|
|
current["chat_only"] = True
|
|
else:
|
|
current.pop("chat_only", None)
|
|
if current != (workspace.settings or {}):
|
|
workspace.settings = current
|
|
audit.record(session, "workspace.chat_only_set", user_id=principal.user_id, workspace_id=workspace_id,
|
|
target_kind="workspace", target_id=workspace_id, enabled=enabled)
|
|
return workspace
|
|
|
|
|
|
def set_tool_rounds(session: Session, principal: Principal, settings, workspace_id: int, value: str) -> Workspace:
|
|
"""Leer = Standard; sonst 4 bis AIW_MAX_TOOL_ROUNDS_LIMIT."""
|
|
workspace = require_admin(session, principal, workspace_id).workspace
|
|
value = value.strip()
|
|
if value and (not value.isdigit() or not 4 <= int(value) <= settings.max_tool_rounds_limit):
|
|
raise Invalid(f"Tool-Schritte je Antwort: eine Zahl von 4 bis {settings.max_tool_rounds_limit}, "
|
|
"oder leer für den Standard.")
|
|
current = dict(workspace.settings or {})
|
|
if value:
|
|
current["max_tool_rounds"] = int(value)
|
|
else:
|
|
current.pop("max_tool_rounds", None)
|
|
if current != (workspace.settings or {}):
|
|
workspace.settings = current
|
|
audit.record(session, "workspace.tool_rounds_set", user_id=principal.user_id, workspace_id=workspace_id,
|
|
target_kind="workspace", target_id=workspace_id, value=current.get("max_tool_rounds"))
|
|
return workspace
|
|
|
|
|
|
def set_approval_mode(session: Session, principal: Principal, workspace_id: int, mode: str) -> Workspace:
|
|
"""Modus A (risk): Einstufung entscheidet. Modus B (all_writes): jede Schreibaktion in ein
|
|
Fremdsystem braucht eine Freigabe, auch wenn das Tool erlaubt ist (→ Konzept 10.3)."""
|
|
workspace = require_admin(session, principal, workspace_id).workspace
|
|
if mode not in APPROVAL_MODES:
|
|
raise Invalid("Unbekannter Freigabemodus.")
|
|
if workspace.approval_mode != mode:
|
|
workspace.approval_mode = mode
|
|
audit.record(session, "workspace.approval_mode_set", user_id=principal.user_id, workspace_id=workspace_id,
|
|
target_kind="workspace", target_id=workspace_id, mode=mode)
|
|
return workspace
|
|
|
|
|
|
def members(session: Session, principal: Principal, workspace_id: int) -> list[tuple[User, str]]:
|
|
get(session, principal, workspace_id)
|
|
rows = session.execute(
|
|
select(User, WorkspaceMembership.role)
|
|
.join(WorkspaceMembership, WorkspaceMembership.user_id == User.id)
|
|
.where(WorkspaceMembership.workspace_id == workspace_id)
|
|
.order_by(User.display_name))
|
|
return [(u, role) for u, role in rows]
|
|
|
|
|
|
def set_member(session: Session, principal: Principal, workspace_id: int, username: str, role: str) -> None:
|
|
"""Mitglied hinzufügen oder Rolle ändern. Nur Mitglieder des Teams kommen in Frage."""
|
|
workspace = require_admin(session, principal, workspace_id).workspace
|
|
if role not in ROLES:
|
|
raise Invalid("Unbekannte Rolle.")
|
|
user = session.scalar(select(User).where(User.username == username.strip().lower(), User.active))
|
|
in_team = user and session.get(TeamMembership, (workspace.team_id, user.id))
|
|
if not in_team:
|
|
raise Invalid("Diesen Nutzer gibt es im Team des Workspace nicht.")
|
|
membership = session.get(WorkspaceMembership, (workspace_id, user.id))
|
|
if membership is None:
|
|
session.add(WorkspaceMembership(workspace_id=workspace_id, user_id=user.id, role=role))
|
|
elif membership.role != role:
|
|
if membership.role == "admin" and _admin_count(session, workspace_id) == 1:
|
|
raise Invalid("Der letzte Admin kann nicht herabgestuft werden.")
|
|
membership.role = role
|
|
audit.record(session, "workspace.member_set", user_id=principal.user_id, workspace_id=workspace_id,
|
|
target_kind="user", target_id=user.id, role=role)
|
|
|
|
|
|
def set_role(session: Session, principal: Principal, workspace_id: int, user_id: int, role: str) -> None:
|
|
"""Rolle eines bestehenden Mitglieds ändern."""
|
|
require_admin(session, principal, workspace_id)
|
|
user = session.get(User, user_id)
|
|
if user is None or session.get(WorkspaceMembership, (workspace_id, user_id)) is None:
|
|
raise NotFound
|
|
set_member(session, principal, workspace_id, user.username, role)
|
|
|
|
|
|
def remove_member(session: Session, principal: Principal, workspace_id: int, user_id: int) -> None:
|
|
require_admin(session, principal, workspace_id)
|
|
membership = session.get(WorkspaceMembership, (workspace_id, user_id))
|
|
if membership is None:
|
|
raise NotFound
|
|
if membership.role == "admin" and _admin_count(session, workspace_id) == 1:
|
|
raise Invalid("Der letzte Admin kann nicht entfernt werden.")
|
|
session.delete(membership)
|
|
audit.record(session, "workspace.member_removed", user_id=principal.user_id, workspace_id=workspace_id,
|
|
target_kind="user", target_id=user_id)
|
|
|
|
|
|
def _admin_count(session: Session, workspace_id: int) -> int:
|
|
return len(session.scalars(select(WorkspaceMembership.user_id).where(
|
|
WorkspaceMembership.workspace_id == workspace_id, WorkspaceMembership.role == "admin")).all())
|