feat: add optional dev mode and analyzer tool upload role

This commit is contained in:
Luftmensch
2026-09-03 16:27:05 +07:00
parent e528419f9a
commit 0249b9eda9
5 changed files with 68 additions and 13 deletions
+47 -6
View File
@@ -53,6 +53,22 @@ def _email_enabled() -> bool:
app = FastAPI(title="Audit Hardening Tool", redirect_slashes=False)
security = HTTPBasic()
# Development mode: allows unauthenticated file upload only.
# Keep disabled by default; enable explicitly with AUDIT_DEV_MODE=true.
DEV_MODE = os.getenv("AUDIT_DEV_MODE", "false").strip().lower() in {"1", "true", "yes", "on"}
DEV_USER = {
"username": "dev_guest",
"full_name": "Development Guest",
"role": "dev_guest",
"unit_id": "",
"unit_name": "Development",
"auth_method": "dev_mode",
"can_view_outputs": True,
"can_download": True,
}
if DEV_MODE:
print(Fore.YELLOW + "[DEV MODE] Unauthenticated file upload is ENABLED; admin routes remain protected" + Fore.RESET)
# Add middleware to handle X-Forwarded-* headers from reverse proxy (Caddy)
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*")
@@ -1078,10 +1094,16 @@ async def logout(request: Request):
@app.get("/", response_class=HTMLResponse, name="index")
async def index(request: Request):
# Kiểm tra đăng nhập bắt buộc
session_id = get_or_create_session(request)
user_info = sessions.get(session_id, {}).get('user_info')
if not user_info and DEV_MODE:
user_info = dict(DEV_USER)
sessions[session_id]['username'] = DEV_USER['username']
sessions[session_id]['user_info'] = user_info
sessions[session_id]['unit_id'] = ''
sessions[session_id]['unit_name'] = 'Development'
if not user_info:
# Chưa đăng nhập → redirect về trang login
response = RedirectResponse(url=str(request.url_for('login')), status_code=303)
@@ -1639,9 +1661,15 @@ async def api_upload_async(
os_type: str = Form(...),
encrypted_files: List[UploadFile] = File(...)
):
# Bắt buộc đăng nhập để xác định unit_id người upload
# Authenticated users and explicit dev mode may upload; admin routes stay protected.
session_id = get_or_create_session(request)
user_info = sessions.get(session_id, {}).get('user_info')
if not user_info and DEV_MODE:
user_info = dict(DEV_USER)
sessions[session_id]['username'] = DEV_USER['username']
sessions[session_id]['user_info'] = user_info
sessions[session_id]['unit_id'] = ''
sessions[session_id]['unit_name'] = 'Development'
if not user_info:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
@@ -1764,6 +1792,11 @@ async def upload_files(
# Lấy hoặc tạo session
session_id = get_or_create_session(request, response)
if DEV_MODE and not sessions.get(session_id, {}).get('user_info'):
sessions[session_id]['username'] = DEV_USER['username']
sessions[session_id]['user_info'] = dict(DEV_USER)
sessions[session_id]['unit_id'] = ''
sessions[session_id]['unit_name'] = 'Development'
# Bắt đầu đo thời gian xử lý
start_time = time.time()
@@ -2530,16 +2563,24 @@ async def download_doc(doc_key: str):
# ============== ADMIN ROUTES ==============
def require_admin_session(request: Request):
"""Check session auth and return admin user or None"""
"""Check session auth and return admin user or None."""
user = get_session_user(request)
if user and user.get('role') == 'admin':
return user
return None
def require_tool_manager_session(request: Request):
"""Allow admin or analyzer to access the tool upload page."""
user = get_session_user(request)
if user and user.get('role') in {'admin', 'analyzer'}:
return user
return None
@app.get("/admin/tools", response_class=HTMLResponse, name="admin_tools")
async def admin_tools_page(request: Request, msg: str = None, msg_type: str = "success"):
"""Admin page for managing tools - requires session authentication"""
user = require_admin_session(request)
user = require_tool_manager_session(request)
if not user:
return RedirectResponse(url=request.url_for('login'), status_code=303)
@@ -2789,8 +2830,8 @@ async def admin_upload_tool(
tool_name: str = Form(...),
tool_file: UploadFile = File(...)
):
"""Upload a new tool - requires admin session authentication"""
user = require_admin_session(request)
"""Upload a new tool - requires admin or analyzer session."""
user = require_tool_manager_session(request)
if not user:
return RedirectResponse(url=request.url_for('login'), status_code=303)