feat: add wildcard token support for allowed extensions and access token CLI commands
Introduce WILDCARD_TOKENS set in attachments.py to treat "*", ".*", "*.*" as wildcards that fall back to ALLOWED_UPLOAD_TYPES. Add cmd_token_issue and cmd_token_list CLI commands for issuing and listing DevPlace access tokens. Register access_tokens table in SOFT_DELETE_TABLES and initialize its columns and indexes in init_db. Replace Form() with Depends(json_or_form(...)) in admin backup, container, notification, settings, and user routers to support both JSON and form data.
This commit is contained in:
@@ -2,7 +2,7 @@
|
||||
|
||||
import logging
|
||||
from typing import Annotated
|
||||
from fastapi import APIRouter, Request, Form
|
||||
from fastapi import Depends, APIRouter, Request
|
||||
from fastapi.responses import HTMLResponse, JSONResponse
|
||||
from devplacepy.models import AdminRoleForm, AdminPasswordForm
|
||||
from devplacepy.database import (
|
||||
@@ -24,15 +24,14 @@ from devplacepy.services.audit import record as audit
|
||||
from devplacepy.services.manager import service_manager
|
||||
from devplacepy.services.openai_gateway.analytics import build_user_usage
|
||||
from devplacepy.services.openai_gateway.usage import pricing_from_cfg
|
||||
from devplacepy.dependencies import json_or_form
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
def _seniority_key(u: dict) -> tuple[str, int]:
|
||||
return (u.get("created_at") or "", u.get("id") or 0)
|
||||
|
||||
|
||||
def _is_senior_admin(actor: dict, target: dict | None) -> bool:
|
||||
if not target or target.get("role") != "Admin":
|
||||
return False
|
||||
@@ -40,7 +39,6 @@ def _is_senior_admin(actor: dict, target: dict | None) -> bool:
|
||||
return False
|
||||
return _seniority_key(target) < _seniority_key(actor)
|
||||
|
||||
|
||||
def _deny_senior(request: Request, admin: dict, uid: str, target: dict, event_key: str):
|
||||
audit.record(
|
||||
request,
|
||||
@@ -55,7 +53,6 @@ def _deny_senior(request: Request, admin: dict, uid: str, target: dict, event_ke
|
||||
)
|
||||
return action_result(request, "/admin/users")
|
||||
|
||||
|
||||
@router.get("/users/{uid}/ai-usage")
|
||||
async def admin_user_ai_usage(request: Request, uid: str, hours: int = 24):
|
||||
user = get_current_user(request)
|
||||
@@ -67,7 +64,6 @@ async def admin_user_ai_usage(request: Request, uid: str, hours: int = 24):
|
||||
pricing = pricing_from_cfg(service.get_config()) if service is not None else None
|
||||
return JSONResponse(build_user_usage(uid, hours=hours, pricing=pricing))
|
||||
|
||||
|
||||
@router.get("/users", response_class=HTMLResponse)
|
||||
async def admin_users(request: Request, page: int = 1):
|
||||
admin = require_admin(request)
|
||||
@@ -110,10 +106,9 @@ async def admin_users(request: Request, page: int = 1):
|
||||
model=AdminUsersOut,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/users/{uid}/role")
|
||||
async def admin_user_role(
|
||||
request: Request, uid: str, data: Annotated[AdminRoleForm, Form()]
|
||||
request: Request, uid: str, data: Annotated[AdminRoleForm, Depends(json_or_form(AdminRoleForm))]
|
||||
):
|
||||
admin = require_admin(request)
|
||||
role = data.role.capitalize()
|
||||
@@ -156,10 +151,9 @@ async def admin_user_role(
|
||||
)
|
||||
return action_result(request, "/admin/users")
|
||||
|
||||
|
||||
@router.post("/users/{uid}/password")
|
||||
async def admin_user_password(
|
||||
request: Request, uid: str, data: Annotated[AdminPasswordForm, Form()]
|
||||
request: Request, uid: str, data: Annotated[AdminPasswordForm, Depends(json_or_form(AdminPasswordForm))]
|
||||
):
|
||||
admin = require_admin(request)
|
||||
users = get_table("users")
|
||||
@@ -184,7 +178,6 @@ async def admin_user_password(
|
||||
)
|
||||
return action_result(request, "/admin/users")
|
||||
|
||||
|
||||
@router.post("/users/{uid}/toggle")
|
||||
async def admin_user_toggle(request: Request, uid: str):
|
||||
admin = require_admin(request)
|
||||
@@ -225,7 +218,6 @@ async def admin_user_toggle(request: Request, uid: str):
|
||||
)
|
||||
return action_result(request, "/admin/users")
|
||||
|
||||
|
||||
@router.post("/users/{uid}/reset-ai-quota")
|
||||
async def admin_user_reset_ai_quota(request: Request, uid: str):
|
||||
admin = require_admin(request)
|
||||
|
||||
Reference in New Issue
Block a user