133 lines
4.0 KiB
Python
133 lines
4.0 KiB
Python
"""Platform access policy.
|
|
|
|
Platform roles are PMFlow-internal identities and must not be inferred from a
|
|
customer workspace role such as OWNER or REVIEWER.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
from sqlalchemy import and_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ..models import PlatformMembership, User
|
|
|
|
ACTIVE_PLATFORM_STATUS = "active"
|
|
|
|
PLATFORM_OWNER = "PLATFORM_OWNER"
|
|
PLATFORM_ADMIN = "PLATFORM_ADMIN"
|
|
AI_OPS = "AI_OPS"
|
|
SUPPORT = "SUPPORT"
|
|
BILLING_OPS = "BILLING_OPS"
|
|
SECURITY_AUDITOR = "SECURITY_AUDITOR"
|
|
|
|
PLATFORM_ROLE_CAPABILITIES: dict[str, set[str]] = {
|
|
PLATFORM_OWNER: {
|
|
"platform.admin.access",
|
|
"platform.members.manage",
|
|
"platform.ai_provider.manage",
|
|
"platform.ai_routing.manage",
|
|
"platform.route_governance.manage",
|
|
"platform.semantic_facts.manage",
|
|
"platform.tenant_metrics.view",
|
|
"platform.support.read",
|
|
"platform.billing.manage",
|
|
"platform.audit.view",
|
|
},
|
|
PLATFORM_ADMIN: {
|
|
"platform.admin.access",
|
|
"platform.ai_provider.manage",
|
|
"platform.ai_routing.manage",
|
|
"platform.route_governance.manage",
|
|
"platform.semantic_facts.manage",
|
|
"platform.tenant_metrics.view",
|
|
"platform.support.read",
|
|
"platform.audit.view",
|
|
},
|
|
AI_OPS: {
|
|
"platform.admin.access",
|
|
"platform.ai_provider.manage",
|
|
"platform.ai_routing.manage",
|
|
"platform.route_governance.manage",
|
|
"platform.semantic_facts.manage",
|
|
"platform.audit.view",
|
|
},
|
|
SUPPORT: {
|
|
"platform.admin.access",
|
|
"platform.tenant_metrics.view",
|
|
"platform.support.read",
|
|
},
|
|
BILLING_OPS: {
|
|
"platform.admin.access",
|
|
"platform.tenant_metrics.view",
|
|
"platform.billing.manage",
|
|
},
|
|
SECURITY_AUDITOR: {
|
|
"platform.admin.access",
|
|
"platform.audit.view",
|
|
},
|
|
}
|
|
|
|
|
|
def _normalize_role(value: object) -> str:
|
|
return str(value or "").strip().upper()
|
|
|
|
|
|
def _env_platform_roles(user: User) -> list[str]:
|
|
"""Bootstrap-only platform access from env, useful before DB grants exist."""
|
|
username = str(getattr(user, "username", "") or "").strip().lower()
|
|
email = str(getattr(user, "email", "") or "").strip().lower()
|
|
if not username and not email:
|
|
return []
|
|
|
|
admin_tokens = {
|
|
item.strip().lower()
|
|
for item in os.getenv("PMFLOW_PLATFORM_ADMIN_USERS", "").split(",")
|
|
if item.strip()
|
|
}
|
|
ai_ops_tokens = {
|
|
item.strip().lower()
|
|
for item in os.getenv("PMFLOW_AI_OPS_USERS", "").split(",")
|
|
if item.strip()
|
|
}
|
|
identities = {username, email}
|
|
roles: list[str] = []
|
|
if identities.intersection(admin_tokens):
|
|
roles.append(PLATFORM_ADMIN)
|
|
if identities.intersection(ai_ops_tokens):
|
|
roles.append(AI_OPS)
|
|
return roles
|
|
|
|
|
|
def _capabilities_for_roles(roles: list[str]) -> list[str]:
|
|
capabilities: set[str] = set()
|
|
for role in roles:
|
|
capabilities.update(PLATFORM_ROLE_CAPABILITIES.get(_normalize_role(role), set()))
|
|
return sorted(capabilities)
|
|
|
|
|
|
async def get_platform_context(db: AsyncSession, *, user: User) -> dict[str, Any]:
|
|
result = await db.execute(
|
|
select(PlatformMembership.role).where(
|
|
and_(
|
|
PlatformMembership.user_id == int(user.id),
|
|
PlatformMembership.status == ACTIVE_PLATFORM_STATUS,
|
|
)
|
|
)
|
|
)
|
|
roles = [_normalize_role(role) for role in result.scalars().all()]
|
|
roles.extend(role for role in _env_platform_roles(user) if role not in roles)
|
|
capabilities = _capabilities_for_roles(roles)
|
|
return {
|
|
"has_platform_access": bool(capabilities),
|
|
"roles": roles,
|
|
"capabilities": capabilities,
|
|
}
|
|
|
|
|
|
async def has_platform_capability(db: AsyncSession, *, user: User, capability: str) -> bool:
|
|
context = await get_platform_context(db, user=user)
|
|
return str(capability) in set(context.get("capabilities") or [])
|