127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
"""
|
|
decorators.py — Shared route decorators.
|
|
@require_role(*roles) — Enforce system role on a route.
|
|
@tenant_feature_required(flag) — Gate a route behind a plan feature flag.
|
|
@demo_readonly — Block write operations on the demo tenant.
|
|
"""
|
|
|
|
import logging
|
|
import functools
|
|
from flask import g, abort, jsonify, request
|
|
from flask_login import current_user
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def require_role(*roles):
|
|
"""
|
|
Decorator that enforces the current user holds one of the specified roles.
|
|
Works for both system_users (superadmin) and tenant users / staff.
|
|
|
|
Usage:
|
|
@require_role("tenant_admin")
|
|
@require_role("tenant_admin", "tenant_manager")
|
|
@require_role("superadmin")
|
|
"""
|
|
def decorator(fn):
|
|
@functools.wraps(fn)
|
|
def wrapper(*args, **kwargs):
|
|
if not current_user.is_authenticated:
|
|
abort(401)
|
|
|
|
user_role = getattr(current_user, "role", None)
|
|
if user_role not in roles:
|
|
logger.warning(
|
|
"Role check failed: user %s has role '%s', required one of %s",
|
|
getattr(current_user, "id", "?"), user_role, roles,
|
|
)
|
|
abort(403)
|
|
return fn(*args, **kwargs)
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
def tenant_feature_required(flag: str):
|
|
"""
|
|
Decorator that gates a route behind a plan feature flag.
|
|
Checks g.tenant.plan.features_json for the given flag.
|
|
Also respects tenant_setting_overrides for force-enable/disable.
|
|
|
|
Usage:
|
|
@tenant_feature_required("marketing")
|
|
@tenant_feature_required("inventory")
|
|
"""
|
|
def decorator(fn):
|
|
@functools.wraps(fn)
|
|
def wrapper(*args, **kwargs):
|
|
tenant = getattr(g, "tenant", None)
|
|
if tenant is None:
|
|
abort(403)
|
|
|
|
# Check for superadmin override (force-enable / force-disable)
|
|
from app.models.platform import TenantSettingOverride
|
|
override = TenantSettingOverride.query.filter_by(
|
|
tenant_id=tenant.id,
|
|
setting_key=f"feature.{flag}",
|
|
lifted_at=None,
|
|
).first()
|
|
|
|
if override is not None:
|
|
enabled = override.setting_value in ("1", "true", "True")
|
|
if not enabled:
|
|
logger.info(
|
|
"Feature '%s' force-disabled by admin override for tenant %s",
|
|
flag, tenant.slug,
|
|
)
|
|
abort(403)
|
|
# force-enabled: proceed regardless of plan
|
|
return fn(*args, **kwargs)
|
|
|
|
# Check plan feature flags
|
|
if tenant.plan is None or not tenant.plan.has_feature(flag):
|
|
logger.info(
|
|
"Feature '%s' not available on plan '%s' for tenant %s",
|
|
flag,
|
|
tenant.plan.name if tenant.plan else "unknown",
|
|
tenant.slug,
|
|
)
|
|
abort(403)
|
|
|
|
return fn(*args, **kwargs)
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
def demo_readonly(fn):
|
|
"""
|
|
Decorator that blocks all write operations (POST, PUT, PATCH, DELETE)
|
|
when the active tenant is the demo tenant.
|
|
Returns 403 with a JSON or HTML response depending on the request type.
|
|
|
|
Usage:
|
|
@demo_readonly
|
|
def create_customer():
|
|
...
|
|
"""
|
|
@functools.wraps(fn)
|
|
def wrapper(*args, **kwargs):
|
|
if request.method in ("POST", "PUT", "PATCH", "DELETE"):
|
|
tenant = getattr(g, "tenant", None)
|
|
if tenant and tenant.is_demo:
|
|
logger.info(
|
|
"Demo write blocked: %s %s", request.method, request.path
|
|
)
|
|
if request.is_json or request.path.startswith("/api/"):
|
|
return jsonify(
|
|
error="Demo account is read-only. "
|
|
"Sign up for a full account to make changes."
|
|
), 403
|
|
from flask import flash, redirect, request as req
|
|
flash(
|
|
"This is a demo account. Sign up for a full account to make changes.",
|
|
"warning",
|
|
)
|
|
return redirect(req.referrer or "/")
|
|
return fn(*args, **kwargs)
|
|
return wrapper
|