feat: permission elevation

apps/agentstack-cli/src/agentstack_cli/__init__.py

@@ -15,6 +15,7 @@
 import agentstack_cli.commands.platform
 import agentstack_cli.commands.self
 import agentstack_cli.commands.server
+import agentstack_cli.commands.user
 from agentstack_cli.async_typer import AliasGroup, AsyncTyper
 from agentstack_cli.configuration import Configuration
 
@@ -48,6 +49,7 @@ def get_help(self, ctx):
 │ model           Configure 15+ LLM providers                                │
 │ platform        Start, stop, or delete local platform                      │
 │ server          Connect to remote Agent Stack servers                      │
+│ user            Manage users and roles                                     │
 │ self version    Show Agent Stack CLI and Platform version                  │
 │ self upgrade    Upgrade Agent Stack CLI and Platform                       │
 │ self uninstall  Uninstall Agent Stack CLI and Platform                     │
@@ -84,6 +86,12 @@ def get_help(self, ctx):
     help="Manage Agent Stack installation.",
     hidden=True,
 )
+app.add_typer(
+    agentstack_cli.commands.user.app,
+    name="user",
+    no_args_is_help=True,
+    help="Manage users.",
+)
 
 
 agent_alias = deepcopy(agentstack_cli.commands.agent.app)

apps/agentstack-cli/src/agentstack_cli/commands/user.py

@@ -0,0 +1,92 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+import typing
+from datetime import datetime
+
+import typer
+from agentstack_sdk.platform import User
+from agentstack_sdk.platform.user import UserRole
+from rich.table import Column
+
+from agentstack_cli.async_typer import AsyncTyper, console, create_table
+from agentstack_cli.configuration import Configuration
+from agentstack_cli.utils import announce_server_action, confirm_server_action
+
+app = AsyncTyper()
+configuration = Configuration()
+
+ROLE_DISPLAY = {
+    "admin": "[red]admin[/red]",
+    "developer": "[cyan]developer[/cyan]",
+    "user": "user",
+}
+
+
+@app.command("list")
+async def list_users(
+    email: typing.Annotated[str | None, typer.Option(help="Filter by email (case-insensitive partial match)")] = None,
+    limit: typing.Annotated[int, typer.Option(help="Results per page (1-100)")] = 40,
+    after: typing.Annotated[str | None, typer.Option(help="Pagination cursor (page_token)")] = None,
+):
+    """List platform users (admin only)."""
+    announce_server_action("Listing users on")
+
+    async with configuration.use_platform_client():
+        result = await User.list(email=email, limit=limit, page_token=after)
+
+        items = result.items
+        has_more = result.has_more
+        next_page_token = result.next_page_token
+
+    with create_table(
+        Column("ID", style="yellow"),
+        Column("Email"),
+        Column("Role"),
+        Column("Created"),
+        Column("Role Updated"),
+        no_wrap=True,
+    ) as table:
+        for user in items:
+            role_display = ROLE_DISPLAY.get(user.role, user.role)
+
+            created_at = _format_date(user.created_at)
+            role_updated_at = _format_date(user.role_updated_at) if user.role_updated_at else "-"
+
+            table.add_row(
+                user.id,
+                user.email,
+                role_display,
+                created_at,
+                role_updated_at,
+            )
+
+    console.print()
+    console.print(table)
+
+    if has_more and next_page_token:
+        console.print(f"\n[dim]Use --after {next_page_token} to see more[/dim]")
+
+
+@app.command("set-role")
+async def set_role(
+    user_id: typing.Annotated[str, typer.Argument(help="User UUID")],
+    role: typing.Annotated[UserRole, typer.Argument(help="Target role (admin, developer, user)")],
+    yes: typing.Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation prompts.")] = False,
+):
+    """Change user role (admin only)."""
+    url = announce_server_action(f"Changing user {user_id} to role '{role}' on")
+    await confirm_server_action("Proceed with role change on", url=url, yes=yes)
+
+    async with configuration.use_platform_client():
+        result = await User.set_role(user_id, UserRole(role))
+
+        role_display = ROLE_DISPLAY.get(result.new_role, result.new_role)
+
+        console.success(f"User role updated to [cyan]{role_display}[/cyan]")
+
+
+def _format_date(dt: datetime | None) -> str:
+    if not dt:
+        return "-"
+    return dt.strftime("%Y-%m-%d %H:%M")

apps/agentstack-sdk-py/src/agentstack_sdk/platform/__init__.py

@@ -7,5 +7,6 @@
 from .model_provider import *
 from .provider import *
 from .provider_build import *
+from .user import *
 from .user_feedback import *
 from .vector_store import *

apps/agentstack-sdk-py/src/agentstack_sdk/platform/user.py

@@ -3,23 +3,68 @@
 
 from __future__ import annotations
 
-from typing import Literal
+from enum import StrEnum
 
 import pydantic
 
 from agentstack_sdk.platform.client import PlatformClient, get_platform_client
+from agentstack_sdk.platform.common import PaginatedResult
+
+
+class UserRole(StrEnum):
+    ADMIN = "admin"
+    DEVELOPER = "developer"
+    USER = "user"
+
+
+class ChangeRoleResponse(pydantic.BaseModel):
+    user_id: str
+    new_role: UserRole
 
 
 class User(pydantic.BaseModel):
     id: str
-    role: Literal["admin", "developer", "user"]
+    role: UserRole
     email: str
     created_at: pydantic.AwareDatetime
+    role_updated_at: pydantic.AwareDatetime | None = None
 
     @staticmethod
     async def get(*, client: PlatformClient | None = None) -> User:
-        """Get the current user information."""
         async with client or get_platform_client() as client:
             return pydantic.TypeAdapter(User).validate_python(
                 (await client.get(url="/api/v1/user")).raise_for_status().json()
             )
+
+    @staticmethod
+    async def list(
+        *,
+        email: str | None = None,
+        limit: int = 40,
+        page_token: str | None = None,
+        client: PlatformClient | None = None,
+    ) -> PaginatedResult[User]:
+        async with client or get_platform_client() as client:
+            params: dict[str, int | str] = {"limit": limit}
+            if email:
+                params["email"] = email
+            if page_token:
+                params["page_token"] = page_token
+
+            return pydantic.TypeAdapter(PaginatedResult[User]).validate_python(
+                (await client.get(url="/api/v1/users", params=params)).raise_for_status().json()
+            )
+
+    @staticmethod
+    async def set_role(
+        user_id: str,
+        new_role: UserRole,
+        *,
+        client: PlatformClient | None = None,
+    ) -> ChangeRoleResponse:
+        async with client or get_platform_client() as client:
+            return pydantic.TypeAdapter(ChangeRoleResponse).validate_python(
+                (await client.put(url=f"/api/v1/users/{user_id}/role", json={"new_role": new_role}))
+                .raise_for_status()
+                .json()
+            )

apps/agentstack-server/src/agentstack_server/api/auth/auth.py

@@ -92,6 +92,7 @@ class ParsedToken(BaseModel):
     context_permissions: Permissions
     context_id: UUID
     user_id: UUID
+    iat: int
     raw: dict[str, Any]
 
 
@@ -142,6 +143,7 @@ def verify_internal_jwt(token: str, configuration: Configuration) -> ParsedToken
         context_permissions=Permissions.model_validate(payload["scope"]["context"]),
         context_id=context_id,
         user_id=UUID(payload["sub"]),
+        iat=payload["iat"],
         raw=payload,
     )
 

apps/agentstack-server/src/agentstack_server/api/dependencies.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import logging
+from datetime import UTC, datetime
 from typing import Annotated, Final
 from uuid import UUID
 
@@ -123,11 +124,19 @@ async def authorized_user(
     request: Request,
 ) -> AuthorizedUser:
     if bearer_auth:
-        # Check Bearer token first - locally this allows for "checking permissions" for development purposes
+        # Check Context token first - locally this allows for "checking permissions" for development purposes
         # even if auth is disabled (requests that would pass with no header may not pass with context token header)
         try:
             parsed_token = verify_internal_jwt(bearer_auth.credentials, configuration=configuration)
             user = await user_service.get_user(parsed_token.user_id)
+
+            iat_dt = datetime.fromtimestamp(parsed_token.iat, tz=UTC)
+            if user.role_updated_at and iat_dt < user.role_updated_at:
+                raise HTTPException(
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                    detail="Token invalidated due to role change",
+                )
+
             token = AuthorizedUser(
                 user=user,
                 global_permissions=parsed_token.global_permissions,

apps/agentstack-server/src/agentstack_server/api/routes/users.py

@@ -0,0 +1,61 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+import logging
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+
+from agentstack_server.api.dependencies import UserServiceDependency, authorized_user
+from agentstack_server.api.schema.user import ChangeRoleRequest, ChangeRoleResponse, UserListQuery, UserResponse
+from agentstack_server.domain.models.common import PaginatedResult
+from agentstack_server.domain.models.permissions import AuthorizedUser
+from agentstack_server.domain.models.user import UserRole
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(tags=["users"])
+
+
+@router.get("", response_model=PaginatedResult[UserResponse])
+async def list_users(
+    query: Annotated[UserListQuery, Query()],
+    user: Annotated[AuthorizedUser, Depends(authorized_user)],
+    user_service: UserServiceDependency,
+) -> PaginatedResult[UserResponse]:
+    if not user.user.role == UserRole.ADMIN:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin permission required")
+
+    result = await user_service.list_users(
+        limit=query.limit,
+        page_token=query.page_token,
+        email=query.email,
+    )
+
+    return PaginatedResult(
+        items=[UserResponse(**u.model_dump()) for u in result.items],
+        total_count=result.total_count,
+        has_more=result.has_more,
+    )
+
+
+@router.put("/{user_id}/role", response_model=ChangeRoleResponse)
+async def change_user_role(
+    user_id: UUID,
+    request: ChangeRoleRequest,
+    user: Annotated[AuthorizedUser, Depends(authorized_user)],
+    user_service: UserServiceDependency,
+) -> ChangeRoleResponse:
+    if not user.user.role == UserRole.ADMIN:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin permission required")
+
+    if user_id == user.user.id:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot change own role")
+
+    updated_user = await user_service.change_role(user_id=user_id, new_role=request.new_role)
+
+    return ChangeRoleResponse(
+        user_id=updated_user.id,
+        new_role=updated_user.role,
+    )

apps/agentstack-server/src/agentstack_server/api/schema/user.py

@@ -0,0 +1,31 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from uuid import UUID
+
+from pydantic import AwareDatetime, BaseModel, EmailStr, Field
+
+from agentstack_server.domain.models.user import UserRole
+
+
+class UserListQuery(BaseModel):
+    limit: int = Field(default=40, ge=1, le=100)
+    page_token: UUID | None = None
+    email: str | None = Field(default=None, description="Filter by email (case-insensitive partial match)")
+
+
+class UserResponse(BaseModel):
+    id: UUID
+    email: EmailStr
+    role: UserRole
+    created_at: AwareDatetime
+    role_updated_at: AwareDatetime | None
+
+
+class ChangeRoleRequest(BaseModel):
+    new_role: UserRole
+
+
+class ChangeRoleResponse(BaseModel):
+    user_id: UUID
+    new_role: UserRole

apps/agentstack-server/src/agentstack_server/application.py

@@ -34,6 +34,7 @@
 from agentstack_server.api.routes.providers import router as provider_router
 from agentstack_server.api.routes.user import router as user_router
 from agentstack_server.api.routes.user_feedback import router as user_feedback_router
+from agentstack_server.api.routes.users import router as users_router
 from agentstack_server.api.routes.variables import router as variables_router
 from agentstack_server.api.routes.vector_stores import router as vector_stores_router
 from agentstack_server.api.utils import format_openai_error
@@ -118,6 +119,7 @@ async def custom_http_exception_handler(request: Request, exc: Exception):
 def mount_routes(app: FastAPI):
     server_router = APIRouter()
     server_router.include_router(user_router, prefix="/user")
+    server_router.include_router(users_router, prefix="/users")
     server_router.include_router(a2a_router, prefix="/a2a")
     server_router.include_router(mcp_router, prefix="/mcp")
     server_router.include_router(provider_router, prefix="/providers", tags=["providers"])

apps/agentstack-server/src/agentstack_server/domain/models/user.py

@@ -19,4 +19,5 @@ class User(BaseModel):
     id: UUID = Field(default_factory=uuid4)
     role: UserRole = UserRole.USER
     email: EmailStr
+    role_updated_at: AwareDatetime | None = None
     created_at: AwareDatetime = Field(default_factory=utc_now)

apps/agentstack-server/src/agentstack_server/domain/repositories/user.py

@@ -1,18 +1,24 @@
 # Copyright 2025 © BeeAI a Series of LF Projects, LLC
 # SPDX-License-Identifier: Apache-2.0
 
-from collections.abc import AsyncIterator
 from typing import Protocol
 from uuid import UUID
 
+from agentstack_server.domain.models.common import PaginatedResult
 from agentstack_server.domain.models.user import User
 
 
 class IUserRepository(Protocol):
-    async def list(self) -> AsyncIterator[User]:
-        yield ...
+    async def list(
+        self,
+        *,
+        limit: int,
+        page_token: UUID | None = None,
+        email: str | None = None,
+    ) -> PaginatedResult[User]: ...
 
     async def create(self, *, user: User) -> None: ...
     async def get(self, *, user_id: UUID) -> User: ...
     async def get_by_email(self, *, email: str) -> User: ...
     async def delete(self, *, user_id: UUID) -> int: ...
+    async def update(self, *, user: User) -> None: ...