root
f2fdf95897
- New: auth.py – Role-based access (operator/approver/viewer) - New: llm_router.py – Configurable primary/fallback provider chain - New: audit_log.py + audit_query.py – DSGVO-compliant audit trail - New: sanitize_output() – masks secrets in bot responses - Updated: bot.py – Integrated auth guards, router, audit logging - Updated: ollama_client.py – Router-compatible interface - Updated: CLOWNFISCHSERVER.md – Documentation updated - Updated: README.md – Documentation updated
116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
🐠 Clownfischserver – Role-Based Access Control
|
||
Author: Mehlzoerwer-Claude (https://github.com/mehlzoerwer-claude)
|
||
License: GPL-3.0
|
||
|
||
Implements DSGVO Art. 32 access control:
|
||
- operator: full execution rights (/shell, /code, /run, /ja, /ssh, /snapshot)
|
||
- approver: can approve pending operations (/ja)
|
||
- viewer: read-only access (/status, /snapshots, /help)
|
||
|
||
Backwards compatible: if no role env vars are set but TELEGRAM_CHAT_ID
|
||
is configured, that user is auto-granted the operator role.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
ROLE_OPERATOR = "operator"
|
||
ROLE_APPROVER = "approver"
|
||
ROLE_VIEWER = "viewer"
|
||
|
||
VALID_ROLES = (ROLE_OPERATOR, ROLE_APPROVER, ROLE_VIEWER)
|
||
|
||
# Operators implicitly hold approver + viewer rights.
|
||
# Approvers implicitly hold viewer rights.
|
||
_ROLE_INHERITANCE: dict[str, tuple[str, ...]] = {
|
||
ROLE_OPERATOR: (ROLE_OPERATOR, ROLE_APPROVER, ROLE_VIEWER),
|
||
ROLE_APPROVER: (ROLE_APPROVER, ROLE_VIEWER),
|
||
ROLE_VIEWER: (ROLE_VIEWER,),
|
||
}
|
||
|
||
|
||
def _parse_id_list(raw: str | None) -> set[str]:
|
||
if not raw:
|
||
return set()
|
||
return {part.strip() for part in raw.split(",") if part.strip()}
|
||
|
||
|
||
def _load_role_map() -> dict[str, str]:
|
||
"""Read role config from env vars.
|
||
|
||
Returns {user_id_str: highest_role}. If a user appears in multiple
|
||
role lists, the strongest role wins (operator > approver > viewer).
|
||
"""
|
||
operators = _parse_id_list(os.getenv("CLOWNFISCH_OPERATORS"))
|
||
approvers = _parse_id_list(os.getenv("CLOWNFISCH_APPROVERS"))
|
||
viewers = _parse_id_list(os.getenv("CLOWNFISCH_VIEWERS"))
|
||
|
||
role_map: dict[str, str] = {}
|
||
|
||
for uid in viewers:
|
||
role_map[uid] = ROLE_VIEWER
|
||
for uid in approvers:
|
||
role_map[uid] = ROLE_APPROVER
|
||
for uid in operators:
|
||
role_map[uid] = ROLE_OPERATOR
|
||
|
||
# Backwards compatibility: legacy TELEGRAM_CHAT_ID becomes operator
|
||
# when no explicit role lists are defined.
|
||
if not role_map:
|
||
legacy = os.getenv("TELEGRAM_CHAT_ID", "").strip()
|
||
if legacy:
|
||
role_map[legacy] = ROLE_OPERATOR
|
||
logger.info(
|
||
"Auth: no role lists configured – granting operator to legacy "
|
||
"TELEGRAM_CHAT_ID for backwards compatibility."
|
||
)
|
||
|
||
return role_map
|
||
|
||
|
||
def get_role(user_id: str | int | None) -> str | None:
|
||
"""Return the role of a user, or None if unauthorized."""
|
||
if user_id is None:
|
||
return None
|
||
return _load_role_map().get(str(user_id))
|
||
|
||
|
||
def has_role(user_id: str | int | None, *required: str) -> bool:
|
||
"""Check if the user satisfies ANY of the required roles.
|
||
|
||
Inheritance applies: an operator passes a check for approver/viewer.
|
||
"""
|
||
role = get_role(user_id)
|
||
if role is None:
|
||
return False
|
||
granted = _ROLE_INHERITANCE.get(role, (role,))
|
||
return any(r in granted for r in required)
|
||
|
||
|
||
def is_authorized(user_id: str | int | None) -> bool:
|
||
"""True if the user has any role at all."""
|
||
return get_role(user_id) is not None
|
||
|
||
|
||
def list_users_with_role(role: str) -> list[str]:
|
||
"""Return all user_ids that hold (or inherit) the given role."""
|
||
if role not in VALID_ROLES:
|
||
return []
|
||
return [uid for uid, r in _load_role_map().items()
|
||
if role in _ROLE_INHERITANCE.get(r, (r,))]
|
||
|
||
|
||
def describe_roles() -> dict[str, list[str]]:
|
||
"""Snapshot of current role assignments – useful for /status or audits."""
|
||
role_map = _load_role_map()
|
||
result: dict[str, list[str]] = {r: [] for r in VALID_ROLES}
|
||
for uid, role in role_map.items():
|
||
result[role].append(uid)
|
||
return result
|