87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
import os
|
|
import uuid
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, Form, Response
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from app.auth import create_jwt_token, verify_jwt_token
|
|
from app.models import get_all_rules, set_rule, get_audit_logs, get_all_keys, add_api_key
|
|
|
|
router = APIRouter()
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
ADMIN_USER = os.environ.get("ADMIN_USER", "nabil")
|
|
ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "password")
|
|
|
|
def get_current_admin(request: Request):
|
|
try:
|
|
payload = verify_jwt_token(request)
|
|
if payload.get("user") != ADMIN_USER:
|
|
raise HTTPException(status_code=401)
|
|
return payload
|
|
except HTTPException:
|
|
raise HTTPException(status_code=401, detail="Unauthorized")
|
|
|
|
def get_current_admin_optional(request: Request):
|
|
try:
|
|
return get_current_admin(request)
|
|
except HTTPException:
|
|
return None
|
|
|
|
@router.get("/login", response_class=HTMLResponse)
|
|
async def login_page(request: Request):
|
|
if get_current_admin_optional(request):
|
|
return RedirectResponse(url="/dashboard")
|
|
return templates.TemplateResponse(request=request, name="login.html")
|
|
|
|
@router.post("/auth/login")
|
|
async def login(response: Response, username: str = Form(...), password: str = Form(...)):
|
|
if username == ADMIN_USER and password == ADMIN_PASSWORD:
|
|
token = create_jwt_token({"user": username})
|
|
resp = RedirectResponse(url="/dashboard", status_code=302)
|
|
resp.set_cookie(key="access_token", value=token, httponly=True, samesite="strict", max_age=12 * 3600)
|
|
return resp
|
|
else:
|
|
# Simplistic rate limiting could be handled by a middleware, but for now we just return 401
|
|
raise HTTPException(status_code=401, detail="Invalid credentials")
|
|
|
|
@router.get("/auth/logout")
|
|
async def logout(response: Response):
|
|
resp = RedirectResponse(url="/login", status_code=302)
|
|
resp.delete_cookie("access_token")
|
|
return resp
|
|
|
|
@router.get("/dashboard", response_class=HTMLResponse)
|
|
async def dashboard(request: Request, admin=Depends(get_current_admin)):
|
|
logs = await get_audit_logs(limit=20)
|
|
rules = await get_all_rules()
|
|
return templates.TemplateResponse(request=request, name="dashboard.html", context={"logs": logs, "rules": rules})
|
|
|
|
@router.get("/editor/{scope}", response_class=HTMLResponse)
|
|
async def editor(request: Request, scope: str, admin=Depends(get_current_admin)):
|
|
rules = await get_all_rules()
|
|
content = rules.get(scope, {})
|
|
return templates.TemplateResponse(request=request, name="editor.html", context={"scope": scope, "content": content})
|
|
|
|
@router.post("/api/admin/rules/{scope}")
|
|
async def update_rule(scope: str, request: Request, admin=Depends(get_current_admin)):
|
|
data = await request.json()
|
|
await set_rule(scope, data)
|
|
return {"status": "success"}
|
|
|
|
@router.get("/audit", response_class=HTMLResponse)
|
|
async def audit_page(request: Request, admin=Depends(get_current_admin)):
|
|
logs = await get_audit_logs(limit=100)
|
|
return templates.TemplateResponse(request=request, name="audit.html", context={"logs": logs})
|
|
|
|
@router.get("/keys", response_class=HTMLResponse)
|
|
async def keys_page(request: Request, admin=Depends(get_current_admin)):
|
|
keys = await get_all_keys()
|
|
return templates.TemplateResponse(request=request, name="keys.html", context={"keys": keys})
|
|
|
|
@router.post("/api/keys/rotate/{agent}")
|
|
async def rotate_key(agent: str, admin=Depends(get_current_admin)):
|
|
# Simple key generation strategy
|
|
new_key = f"ctx-{agent.lower()}-{uuid.uuid4().hex[:16]}"
|
|
await add_api_key(agent.upper(), new_key)
|
|
return {"status": "success", "new_key": new_key}
|