v0.5.0: Add RBAC auth, LLM router, audit logging, output sanitization
- 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
This commit is contained in:
+115
@@ -0,0 +1,115 @@
|
||||
#!/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
|
||||
Reference in New Issue
Block a user