bvandeusen
f3e919892d
Implements #253's framework: a small core capability registry (steward/core/capabilities.py) where a module/plugin publishes a named, role-gated action and a consumer discovers it via has_capability() and runs it via invoke_capability() — no hard import, graceful degradation, permission propagation (actor role checked against the capability's required_role). Core publishes "ansible.run_playbook" (operator) wrapping ansible.runner. trigger_run (extended to accept a caller-built inventory). First consumer: the host_agent plugin gains "Deploy via Ansible" on its settings page — pick an inventory target/group and it installs/updates the agent via the bundled host_agent/install.yml, minting a fresh token per host and injecting it as an inventory hostvar (turning per-host curl|sh into one run). Exposed role ordering as middleware.role_meets. Unit tests for the registry + role checks. Task #253 (milestone #37). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
from __future__ import annotations
|
|
import functools
|
|
from quart import session, redirect, url_for, abort, g
|
|
from steward.models.users import UserRole
|
|
|
|
_ROLE_ORDER = [UserRole.viewer, UserRole.operator, UserRole.admin]
|
|
|
|
|
|
def role_meets(user_role: UserRole, minimum_role: UserRole) -> bool:
|
|
"""True when user_role is at least minimum_role in the viewer<operator<admin order."""
|
|
return _ROLE_ORDER.index(user_role) >= _ROLE_ORDER.index(minimum_role)
|
|
|
|
|
|
def require_role(minimum_role: UserRole):
|
|
"""Decorator: requires authenticated user with at least minimum_role.
|
|
Also allows access for validated share-token requests (viewer level only).
|
|
"""
|
|
def decorator(f):
|
|
@functools.wraps(f)
|
|
async def wrapper(*args, **kwargs):
|
|
# Share-token requests bypass session auth for viewer-level endpoints
|
|
if getattr(g, "share_viewer", False) and minimum_role == UserRole.viewer:
|
|
return await f(*args, **kwargs)
|
|
user_id = session.get("user_id")
|
|
if not user_id:
|
|
return redirect(url_for("auth.login"))
|
|
user_role = UserRole(session.get("user_role", "viewer"))
|
|
if _ROLE_ORDER.index(user_role) < _ROLE_ORDER.index(minimum_role):
|
|
abort(403)
|
|
return await f(*args, **kwargs)
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
def login_user(user) -> None:
|
|
"""Write user identity into the session."""
|
|
session["user_id"] = user.id
|
|
session["user_role"] = user.role.value
|
|
session["username"] = user.username
|
|
|
|
|
|
def logout_user() -> None:
|
|
session.clear()
|
|
|
|
|
|
def current_user_id() -> str | None:
|
|
return session.get("user_id")
|