Merge pull request #674 from lbedner/v0.6.12-rc1

v0.6.12-rc1

Author: lbedner

aegis/templates/copier-aegis-project/{{ project_slug }}/.env.example.jinja

@@ -118,8 +118,16 @@ LOGFIRE_TOKEN=
 # LOGFIRE_PROJECT_URL=https://logfire.pydantic.dev/myorg/myproject
 {%- endif %}
 
-# Authentication (set to true in production to require login)
+# Authentication
+{%- if include_auth %}
+# Auth service is included. Set to false to bypass login locally
+# (you'll be signed in as a synthetic admin dev user).
+AUTH_ENABLED=true
+{%- else %}
+# Auth service not included. A synthetic admin dev user is used so
+# handlers reading current_user don't crash. Leave as false.
 AUTH_ENABLED=false
+{%- endif %}
 
 # API Keys and Secrets (add as needed)
 # SECRET_KEY=your-super-secret-key-here

aegis/templates/copier-aegis-project/{{ project_slug }}/app/cli/auth.py.jinja

@@ -17,6 +17,9 @@ from rich.table import Table
 import typer
 
 from app.core.db import get_async_session
+{%- if include_auth_rbac %}
+from app.core.security import VALID_ROLES
+{%- endif %}
 from app.i18n import lazy_t, t
 from app.models.user import UserCreate
 from app.services.auth.user_service import UserService
@@ -255,6 +258,61 @@ async def _create_test_users(
         raise typer.Exit(1)
 
 
+{% if include_auth_rbac %}
+@app.command(help=lazy_t("auth.help_promote_user"))
+def promote_user(
+    email: str = typer.Option(..., help=lazy_t("auth.opt_promote_email")),
+    role: str = typer.Option("admin", help=lazy_t("auth.opt_promote_role")),
+) -> None:
+    asyncio.run(_promote_user(email, role))
+
+
+async def _promote_user(email: str, role: str) -> None:
+    """Async implementation of promote_user."""
+    if role not in VALID_ROLES:
+        typer.secho(
+            t("auth.invalid_role", role=role, valid=", ".join(sorted(VALID_ROLES))),
+            fg=typer.colors.RED,
+        )
+        raise typer.Exit(1)
+
+    try:
+        async with get_async_session() as session:
+            user_service = UserService(session)
+            user = await user_service.get_user_by_email(email)
+            if not user:
+                typer.secho(
+                    t("auth.user_not_found", email=email),
+                    fg=typer.colors.RED,
+                )
+                raise typer.Exit(1)
+
+            previous_role = user.role
+            updated = await user_service.update_user(user.id, role=role)
+            assert updated is not None  # we just fetched it
+
+            typer.secho(
+                t(
+                    "auth.user_promoted",
+                    email=updated.email,
+                    previous=previous_role,
+                    role=updated.role,
+                ),
+                fg=typer.colors.GREEN,
+                bold=True,
+            )
+
+    except typer.Exit:
+        raise
+    except Exception as e:
+        typer.secho(
+            t("auth.promote_user_failed", error=str(e)),
+            fg=typer.colors.RED,
+        )
+        raise typer.Exit(1)
+
+
+{% endif %}
 @app.command(help=lazy_t("auth.help_list_users"))
 def list_users() -> None:
     asyncio.run(_list_users())

aegis/templates/copier-aegis-project/{{ project_slug }}/app/components/frontend/dashboard/modals/auth_orgs_tab.py.jinja

@@ -164,13 +164,11 @@ class AuthOrgsTab(ft.Container):
 
         self._render_orgs(self._orgs_data, org_members)
 
-    def _render_orgs(
-        self,
-        orgs: list[dict[str, Any]],
-        org_members: dict[int, list[dict[str, Any]]],
-    ) -> None:
-        """Render the organizations table with expandable member rows."""
-        refresh_row = ft.Row(
+    def _refresh_row(self) -> ft.Row:
+        """Trailing-aligned refresh icon, included on render paths where a
+        retry is meaningful (loaded and error states). Skipped on the
+        unavailable state since refresh can't restore a missing database."""
+        return ft.Row(
             [
                 ft.Container(expand=True),
                 ft.IconButton(
@@ -183,6 +181,13 @@ class AuthOrgsTab(ft.Container):
             alignment=ft.MainAxisAlignment.END,
         )
 
+    def _render_orgs(
+        self,
+        orgs: list[dict[str, Any]],
+        org_members: dict[int, list[dict[str, Any]]],
+    ) -> None:
+        """Render the organizations table with expandable member rows."""
+
         columns = [
             DataTableColumn("Name", style="primary"),
             DataTableColumn("Slug", width=120, style="secondary"),
@@ -239,14 +244,17 @@ class AuthOrgsTab(ft.Container):
             empty_message="No organizations found",
         )
 
-        self._content_column.controls = [refresh_row, table]
+        self._content_column.controls = [self._refresh_row(), table]
         self._content_column.scroll = ft.ScrollMode.AUTO
         self._content_column.spacing = 0
         self.update()
 
     def _render_error(self, message: str) -> None:
-        """Render an error state."""
+        """Render an error state. The refresh button is included so the user
+        can retry after fixing the underlying problem (e.g. role bump)
+        without closing and reopening the dialog."""
         self._content_column.controls = [
+            self._refresh_row(),
             ft.Container(
                 content=ft.Icon(
                     ft.Icons.ERROR_OUTLINE,

aegis/templates/copier-aegis-project/{{ project_slug }}/app/components/frontend/dashboard/modals/auth_users_tab.py.jinja

@@ -228,15 +228,11 @@ class AuthUsersTab(ft.Container):
         return user_orgs
 
 {% endif %}
-    def _render_users(
-        self,
-        users: list[dict[str, Any]],
-{% if include_auth_org %}
-        user_orgs: dict[int, list[str]] | None = None,
-{% endif %}
-    ) -> None:
-        """Render the users table with loaded data."""
-        refresh_row = ft.Row(
+    def _refresh_row(self) -> ft.Row:
+        """Trailing-aligned refresh icon, included on render paths where a
+        retry is meaningful (loaded and error states). Skipped on the
+        unavailable state since refresh can't restore a missing database."""
+        return ft.Row(
             [
                 ft.Container(expand=True),
                 ft.IconButton(
@@ -249,8 +245,16 @@ class AuthUsersTab(ft.Container):
             alignment=ft.MainAxisAlignment.END,
         )
 
+    def _render_users(
+        self,
+        users: list[dict[str, Any]],
+{% if include_auth_org %}
+        user_orgs: dict[int, list[str]] | None = None,
+{% endif %}
+    ) -> None:
+        """Render the users table with loaded data."""
         self._content_column.controls = [
-            refresh_row,
+            self._refresh_row(),
             UsersTableSection(
                 users,
                 on_toggle=self._on_toggle_user,
@@ -265,8 +269,11 @@ class AuthUsersTab(ft.Container):
         self.update()
 
     def _render_error(self, message: str) -> None:
-        """Render an error state."""
+        """Render an error state. The refresh button is included so the user
+        can retry after fixing the underlying problem (e.g. role bump)
+        without closing and reopening the dialog."""
         self._content_column.controls = [
+            self._refresh_row(),
             ft.Container(
                 content=ft.Icon(
                     ft.Icons.ERROR_OUTLINE,

aegis/templates/copier-aegis-project/{{ project_slug }}/app/core/config.py.jinja

@@ -80,7 +80,11 @@ class Settings(
     FLET_ASSETS_DIR: str = "assets"  # Directory for Flet static assets (images, etc.)
 
     # Authentication settings
-    AUTH_ENABLED: bool = False  # Set to true in production to enable auth
+{%- if include_auth %}
+    AUTH_ENABLED: bool = True  # Auth service included; set false in .env to bypass for local dev
+{%- else %}
+    AUTH_ENABLED: bool = False  # Auth service not included; dev user is synthesized
+{%- endif %}
     DEV_USER_ROLE: str = "admin"  # Role for synthetic dev user when AUTH_ENABLED=false
     INVITE_ACCEPTANCE_MODE: str = "email"  # "email" requires email match, "token" allows anyone with token
     SECRET_KEY: str = "change-this-secret-key-in-production-use-env-variable"

aegis/templates/copier-aegis-project/{{ project_slug }}/app/i18n/locales/en.py

@@ -433,18 +433,25 @@
     "auth.users_title": "Users",
     "auth.created_column": "Created",
     "auth.list_users_failed": "Error: Failed to list users: {error}",
+    "auth.user_not_found": "Error: No user found with email '{email}'",
+    "auth.invalid_role": "Error: '{role}' is not a valid role. Valid roles: {valid}",
+    "auth.user_promoted": "Updated {email}: role {previous} -> {role}",
+    "auth.promote_user_failed": "Error: Failed to promote user: {error}",
     # Help text
     "auth.help": "Authentication management commands",
     "auth.help_create_test_user": "Create a test user for development and testing.",
     "auth.help_create_test_users": "Create multiple test users for development and testing.",
     "auth.help_list_users": "List all users in the system.",
+    "auth.help_promote_user": "Change a user's role (e.g. promote to admin).",
     "auth.opt_email": "User email address (auto-increment: test@example.com, test1@example.com, etc.)",
     "auth.opt_password": "User password (generated if not provided)",
     "auth.opt_full_name": "User full name",
     "auth.opt_prefix": "Email prefix for auto-generated emails",
     "auth.opt_domain": "Email domain for auto-generated emails",
     "auth.opt_count": "Number of test users to create",
     "auth.opt_shared_password": "Shared password (generated if not provided)",
+    "auth.opt_promote_email": "Email of the user to update",
+    "auth.opt_promote_role": "New role to assign (user, moderator, admin)",
     # ── Tasks ────────────────────────────────────────────────────────
     "tasks.listing_jobs": "Listing Scheduled Jobs",
     "tasks.no_jobs_found": "No scheduled jobs found",

aegis/templates/copier-aegis-project/{{ project_slug }}/app/i18n/locales/zh.py

@@ -433,18 +433,25 @@
     "auth.users_title": "用户列表",
     "auth.created_column": "创建时间",
     "auth.list_users_failed": "错误：获取用户列表失败：{error}",
+    "auth.user_not_found": "错误：未找到邮箱为「{email}」的用户",
+    "auth.invalid_role": "错误：「{role}」不是有效角色。可选角色：{valid}",
+    "auth.user_promoted": "已更新 {email}：角色由 {previous} 变更为 {role}",
+    "auth.promote_user_failed": "错误：更新用户角色失败：{error}",
     # 帮助文本
     "auth.help": "认证管理命令",
     "auth.help_create_test_user": "创建测试用户，用于开发和测试",
     "auth.help_create_test_users": "批量创建测试用户，用于开发和测试",
     "auth.help_list_users": "列出所有用户",
+    "auth.help_promote_user": "修改用户角色（例如提升为管理员）",
     "auth.opt_email": "用户邮箱（自动递增，如 test@、test1@ 等）",
     "auth.opt_password": "用户密码（不指定则自动生成）",
     "auth.opt_full_name": "用户全名",
     "auth.opt_prefix": "自动生成邮箱的前缀",
     "auth.opt_domain": "自动生成邮箱的域名",
     "auth.opt_count": "创建数量",
     "auth.opt_shared_password": "共享密码（不指定则自动生成）",
+    "auth.opt_promote_email": "要更新的用户邮箱",
+    "auth.opt_promote_role": "要分配的新角色（user、moderator、admin）",
     # ── 定时任务 ──────────────────────────────────────────────────────
     "tasks.listing_jobs": "已调度任务列表",
     "tasks.no_jobs_found": "没有已调度的任务",

aegis/templates/copier-aegis-project/{{ project_slug }}/app/services/auth/user_service.py.jinja

@@ -3,6 +3,9 @@
 import secrets
 from datetime import UTC, datetime, timedelta
 
+{% if include_auth_rbac %}
+from sqlalchemy import func
+{% endif %}
 {% if include_auth_org %}
 from sqlmodel import delete, select
 {% else %}
@@ -39,16 +42,32 @@ class UserService:
         self.db = db
 
     async def create_user(self, user_data: UserCreate) -> User:
-        """Create a new user asynchronously."""
+        """Create a new user asynchronously.
+{%- if include_auth_rbac %}
+
+        The very first user in the system is auto-promoted to ``admin`` so
+        the operator who runs ``aegis init`` and registers can reach the
+        admin-gated tabs (Users, Orgs) without a manual SQL update.
+{%- endif %}
+        """
         # Hash the password
         hashed_password = get_password_hash(user_data.password)
 
+{%- if include_auth_rbac %}
+        # First user becomes admin
+        count_result = await self.db.exec(select(func.count()).select_from(User))
+        is_first_user = count_result.one() == 0
+
+{%- endif %}
         # Create user object
         user = User(
             email=user_data.email.lower(),
             full_name=user_data.full_name,
             hashed_password=hashed_password,
             is_active=user_data.is_active,
+{%- if include_auth_rbac %}
+            role="admin" if is_first_user else "user",
+{%- endif %}
             created_at=datetime.now(UTC).replace(tzinfo=None),
         )
 
@@ -322,6 +341,11 @@ class UserService:
         # Path 3: brand-new user.
         if not settings.REGISTRATION_ENABLED:
             raise RegistrationClosed()
+{%- if include_auth_rbac %}
+        # First user becomes admin (matches password-registration behavior).
+        count_result = await self.db.exec(select(func.count()).select_from(User))
+        is_first_user = count_result.one() == 0
+{%- endif %}
         user = User(
             email=email,
             full_name=provider_username or email.split("@")[0],
@@ -330,6 +354,9 @@ class UserService:
             hashed_password="",
             is_active=True,
             is_verified=True,  # provider guarantees email ownership
+{%- if include_auth_rbac %}
+            role="admin" if is_first_user else "user",
+{%- endif %}
         )
         self.db.add(user)
         await self.db.flush()  # need user.id for the identity row