vmware-api-simulator/app/vsphere/security/authz.py

"""Role / privilege authorization for vSphere REST (and SOAP gates)."""

from __future__ import annotations

from collections.abc import Awaitable, Callable
from functools import wraps
from typing import Any

from fastapi import Depends

from app.db.pool import Database
from app.dependencies import get_database
from app.vsphere.errors import unauthorized
from app.vsphere.security.session import SessionInfo, require_session

# Privilege catalog (subset of vSphere privilege ids).
PRIVILEGES: dict[str, str] = {
    "System.Anonymous": "Anonymous access",
    "System.Read": "Read inventory",
    "System.View": "View inventory",
    "Global.ManageCustomFields": "Manage custom fields",
    "Authorization.ModifyPermissions": "Modify permissions",
    "VirtualMachine.Inventory.Create": "Create VM",
    "VirtualMachine.Inventory.Delete": "Delete VM",
    "VirtualMachine.Inventory.Move": "Move VM",
    "VirtualMachine.Interact.PowerOn": "Power on VM",
    "VirtualMachine.Interact.PowerOff": "Power off VM",
    "VirtualMachine.Interact.Suspend": "Suspend VM",
    "VirtualMachine.Interact.Reset": "Reset VM",
    "VirtualMachine.Interact.DeviceConnection": "Connect devices",
    "VirtualMachine.Interact.ConsoleInteract": "Console",
    "VirtualMachine.Config.AddNewDisk": "Add disk",
    "VirtualMachine.Config.AddExistingDisk": "Add existing disk",
    "VirtualMachine.Config.RemoveDisk": "Remove disk",
    "VirtualMachine.Config.CPUCount": "Change CPU",
    "VirtualMachine.Config.Memory": "Change memory",
    "VirtualMachine.Config.AddRemoveDevice": "Add/remove device",
    "VirtualMachine.Config.Rename": "Rename VM",
    "VirtualMachine.Provisioning.Clone": "Clone VM",
    "VirtualMachine.Provisioning.DeployTemplate": "Deploy template",
    "VirtualMachine.Provisioning.MarkAsTemplate": "Mark as template",
    "VirtualMachine.State.CreateSnapshot": "Create snapshot",
    "VirtualMachine.State.RemoveSnapshot": "Remove snapshot",
    "VirtualMachine.State.RevertToSnapshot": "Revert snapshot",
    "Datastore.Browse": "Browse datastore",
    "Datastore.FileManagement": "Manage datastore files",
    "Host.Config.Maintenance": "Host maintenance",
    "Folder.Create": "Create folder",
    "Folder.Delete": "Delete folder",
    "Folder.Rename": "Rename folder",
    "Folder.Move": "Move folder",
    "Datacenter.Create": "Create datacenter",
    "Datacenter.Delete": "Delete datacenter",
    "Cluster.Create": "Create cluster",
    "Cluster.Delete": "Delete cluster",
    "Resource.CreatePool": "Create resource pool",
    "Resource.DeletePool": "Delete resource pool",
    "Network.Assign": "Assign network",
    "ContentLibrary.CreateLocalLibrary": "Create content library",
    "ContentLibrary.AddLibraryItem": "Add library item",
    "InventoryService.Tagging.CreateCategory": "Create tag category",
    "InventoryService.Tagging.CreateTag": "Create tag",
    "InventoryService.Tagging.AttachTag": "Attach tag",
}

_ALL = frozenset(PRIVILEGES)
_READ = frozenset({"System.Anonymous", "System.Read", "System.View", "Datastore.Browse"})
_POWER = frozenset(
    {
        *_READ,
        "VirtualMachine.Interact.PowerOn",
        "VirtualMachine.Interact.PowerOff",
        "VirtualMachine.Interact.Suspend",
        "VirtualMachine.Interact.Reset",
        "VirtualMachine.Interact.ConsoleInteract",
        "VirtualMachine.State.CreateSnapshot",
        "VirtualMachine.State.RemoveSnapshot",
        "VirtualMachine.State.RevertToSnapshot",
        "VirtualMachine.Provisioning.Clone",
    }
)
_VM_ADMIN = frozenset(
    {
        *_POWER,
        "VirtualMachine.Inventory.Create",
        "VirtualMachine.Inventory.Delete",
        "VirtualMachine.Inventory.Move",
        "VirtualMachine.Config.AddNewDisk",
        "VirtualMachine.Config.AddExistingDisk",
        "VirtualMachine.Config.RemoveDisk",
        "VirtualMachine.Config.CPUCount",
        "VirtualMachine.Config.Memory",
        "VirtualMachine.Config.AddRemoveDevice",
        "VirtualMachine.Config.Rename",
        "VirtualMachine.Provisioning.DeployTemplate",
        "VirtualMachine.Provisioning.MarkAsTemplate",
        "VirtualMachine.Interact.DeviceConnection",
        "Datastore.FileManagement",
        "Network.Assign",
        "InventoryService.Tagging.CreateCategory",
        "InventoryService.Tagging.CreateTag",
        "InventoryService.Tagging.AttachTag",
        "ContentLibrary.CreateLocalLibrary",
        "ContentLibrary.AddLibraryItem",
    }
)

ROLE_PRIVILEGES: dict[str, frozenset[str]] = {
    "Administrator": _ALL,
    "ReadOnly": _READ,
    "VirtualMachinePowerUser": _POWER,
    "VirtualMachineAdministrator": _VM_ADMIN,
}


def privileges_for_roles(roles: list[str] | tuple[str, ...]) -> frozenset[str]:
    granted: set[str] = set()
    for role in roles:
        granted.update(ROLE_PRIVILEGES.get(role, ()))
    return frozenset(granted)


def has_privilege(roles: list[str] | tuple[str, ...], privilege: str) -> bool:
    granted = privileges_for_roles(roles)
    if privilege in granted:
        return True
    # Wildcard Administrator already has exact set; keep prefix convenience.
    return any(p.endswith(".*") and privilege.startswith(p[:-1]) for p in granted)


async def load_roles(database: Database, username: str) -> list[str]:
    pool = database.pool  # type: ignore[attr-defined]
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT roles FROM vsphere_credentials WHERE username = $1",
            username,
        )
    if row is None:
        return ["Administrator"] if username.endswith("@vsphere.local") else ["ReadOnly"]
    roles = list(row["roles"] or [])
    return roles or ["ReadOnly"]


def require_privilege(*needed: str) -> Callable[..., Any]:
    """FastAPI dependency factory: session must hold every listed privilege."""

    async def _dependency(
        session: SessionInfo = Depends(require_session),
        database: Database = Depends(get_database),
    ) -> SessionInfo:
        roles = list(session.roles)
        if not roles:
            roles = await load_roles(database, session.username)
        for privilege in needed:
            if not has_privilege(roles, privilege):
                raise unauthorized(f"Missing privilege: {privilege}")
        return session

    return _dependency


require_read = require_privilege("System.Read")
require_power = require_privilege("VirtualMachine.Interact.PowerOn")
require_vm_mutate = require_privilege("VirtualMachine.Inventory.Create")
require_admin = require_privilege("Authorization.ModifyPermissions")


def guard(*needed: str) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]:
    """Decorator-style helper for non-FastAPI call sites (SOAP)."""

    def decorator(fn: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]:
        @wraps(fn)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            return await fn(*args, **kwargs)

        wrapper.__vsphere_privileges__ = needed  # type: ignore[attr-defined]
        return wrapper

    return decorator