244 lines
7.8 KiB
Python
244 lines
7.8 KiB
Python
from fastapi import APIRouter, Request, Depends, HTTPException, Query
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import desc
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Optional
|
|
|
|
from database import get_db
|
|
from auth import verify_token, require_admin
|
|
from models import Computer, Heartbeat, User, Notification, AuditLog, Setting
|
|
from notifications import check_offline_computers, get_unread_count
|
|
from settings_store import get_all_settings
|
|
from timezone_utils import localtime, localtime_iso, get_common_timezones, get_current_timezone
|
|
|
|
router = APIRouter()
|
|
templates = Jinja2Templates(directory="templates")
|
|
templates.env.filters["localtime"] = localtime
|
|
templates.env.filters["localtime_iso"] = localtime_iso
|
|
templates.env.globals["current_timezone"] = get_current_timezone
|
|
|
|
|
|
def user_or_redirect(request: Request, db: Session):
|
|
token = request.cookies.get("access_token")
|
|
if not token:
|
|
return None
|
|
payload = verify_token(token, "access")
|
|
if not payload or "sub" not in payload:
|
|
return None
|
|
return db.query(User).filter(User.username == payload["sub"]).first()
|
|
|
|
|
|
def has_admin(db: Session) -> bool:
|
|
return db.query(User).filter(User.role == "admin").count() > 0
|
|
|
|
|
|
@router.get("/", response_class=HTMLResponse)
|
|
def index(
|
|
request: Request,
|
|
status: Optional[str] = Query(None),
|
|
user: Optional[str] = Query(None),
|
|
search: Optional[str] = Query(None),
|
|
sort: str = Query("hostname"),
|
|
order: str = Query("asc"),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
current_user = user_or_redirect(request, db)
|
|
if not current_user:
|
|
return RedirectResponse(url="/login")
|
|
|
|
check_offline_computers(db)
|
|
|
|
query = db.query(Computer)
|
|
if status and status in ("online", "offline"):
|
|
query = query.filter(Computer.status == status)
|
|
if user:
|
|
query = query.filter(Computer.current_user.ilike(f"%{user}%"))
|
|
if search:
|
|
query = query.filter(
|
|
Computer.hostname.ilike(f"%{search}%")
|
|
| Computer.os_info.ilike(f"%{search}%")
|
|
| Computer.current_ip.ilike(f"%{search}%")
|
|
)
|
|
|
|
allowed_sort = {"hostname", "last_seen", "current_user", "status", "current_cpu_percent", "current_ram_percent"}
|
|
if sort in allowed_sort:
|
|
column = getattr(Computer, sort)
|
|
if order == "desc":
|
|
query = query.order_by(desc(column))
|
|
else:
|
|
query = query.order_by(column)
|
|
else:
|
|
query = query.order_by(Computer.hostname)
|
|
|
|
computers = query.all()
|
|
|
|
return templates.TemplateResponse(
|
|
"dashboard.html",
|
|
{
|
|
"request": request,
|
|
"user": current_user,
|
|
"computers": computers,
|
|
"filters": {
|
|
"status": status or "",
|
|
"user": user or "",
|
|
"search": search or "",
|
|
"sort": sort,
|
|
"order": order,
|
|
},
|
|
"unread_count": get_unread_count(db),
|
|
"now": datetime.now(timezone.utc),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/login", response_class=HTMLResponse)
|
|
def login_page(request: Request, db: Session = Depends(get_db)):
|
|
# Allow public registration only while no admin exists
|
|
registration_open = not has_admin(db)
|
|
return templates.TemplateResponse("login.html", {"request": request, "error": None, "registration_open": registration_open})
|
|
|
|
|
|
@router.get("/computers/{computer_id}", response_class=HTMLResponse)
|
|
def computer_page(computer_id: int, request: Request, db: Session = Depends(get_db)):
|
|
current_user = user_or_redirect(request, db)
|
|
if not current_user:
|
|
return RedirectResponse(url="/login")
|
|
|
|
computer = db.query(Computer).filter(Computer.id == computer_id).first()
|
|
if not computer:
|
|
raise HTTPException(status_code=404, detail="Computer not found")
|
|
|
|
since = datetime.now(timezone.utc) - timedelta(hours=24)
|
|
heartbeats = (
|
|
db.query(Heartbeat)
|
|
.filter(Heartbeat.computer_id == computer_id, Heartbeat.timestamp >= since)
|
|
.order_by(Heartbeat.timestamp)
|
|
.all()
|
|
)
|
|
|
|
hb_data = []
|
|
latest_sessions = []
|
|
for h in heartbeats:
|
|
hb_data.append({
|
|
"timestamp": localtime_iso(h.timestamp),
|
|
"username": h.username,
|
|
"local_ip": h.local_ip,
|
|
"cpu_percent": h.cpu_percent,
|
|
"ram_percent": h.ram_percent,
|
|
"ram_total_gb": h.ram_total_gb,
|
|
"ram_used_gb": h.ram_used_gb,
|
|
"ram_available_gb": h.ram_available_gb,
|
|
"swap_total_gb": h.swap_total_gb,
|
|
"swap_used_gb": h.swap_used_gb,
|
|
"swap_free_gb": h.swap_free_gb,
|
|
"swap_percent": h.swap_percent,
|
|
"cpu_temp": h.cpu_temp,
|
|
"processes": h.processes or [],
|
|
"sessions": h.sessions or [],
|
|
})
|
|
# Take the most recent heartbeat that actually has session data
|
|
for h in reversed(heartbeats):
|
|
if h.sessions:
|
|
latest_sessions = h.sessions
|
|
break
|
|
|
|
return templates.TemplateResponse(
|
|
"computer.html",
|
|
{
|
|
"request": request,
|
|
"user": current_user,
|
|
"computer": computer,
|
|
"heartbeats": hb_data,
|
|
"latest_sessions": latest_sessions,
|
|
"unread_count": get_unread_count(db),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/notifications", response_class=HTMLResponse)
|
|
def notifications_page(request: Request, db: Session = Depends(get_db)):
|
|
current_user = user_or_redirect(request, db)
|
|
if not current_user:
|
|
return RedirectResponse(url="/login")
|
|
|
|
notifications = (
|
|
db.query(Notification)
|
|
.order_by(desc(Notification.sent_at))
|
|
.limit(100)
|
|
.all()
|
|
)
|
|
|
|
return templates.TemplateResponse(
|
|
"notifications.html",
|
|
{
|
|
"request": request,
|
|
"user": current_user,
|
|
"notifications": notifications,
|
|
"unread_count": get_unread_count(db),
|
|
},
|
|
)
|
|
|
|
|
|
# Admin pages
|
|
|
|
@router.get("/admin/users", response_class=HTMLResponse)
|
|
def admin_users_page(request: Request, db: Session = Depends(get_db)):
|
|
current_user = user_or_redirect(request, db)
|
|
if not current_user or current_user.role != "admin":
|
|
return RedirectResponse(url="/login")
|
|
|
|
users = db.query(User).order_by(User.username).all()
|
|
return templates.TemplateResponse(
|
|
"admin_users.html",
|
|
{
|
|
"request": request,
|
|
"user": current_user,
|
|
"users": users,
|
|
"unread_count": get_unread_count(db),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/admin/audit", response_class=HTMLResponse)
|
|
def admin_audit_page(request: Request, db: Session = Depends(get_db)):
|
|
current_user = user_or_redirect(request, db)
|
|
if not current_user or current_user.role != "admin":
|
|
return RedirectResponse(url="/login")
|
|
|
|
audit = (
|
|
db.query(AuditLog)
|
|
.order_by(desc(AuditLog.timestamp))
|
|
.limit(200)
|
|
.all()
|
|
)
|
|
return templates.TemplateResponse(
|
|
"admin_audit.html",
|
|
{
|
|
"request": request,
|
|
"user": current_user,
|
|
"audit": audit,
|
|
"unread_count": get_unread_count(db),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/admin/settings", response_class=HTMLResponse)
|
|
def admin_settings_page(request: Request, db: Session = Depends(get_db)):
|
|
current_user = user_or_redirect(request, db)
|
|
if not current_user or current_user.role != "admin":
|
|
return RedirectResponse(url="/login")
|
|
|
|
settings = get_all_settings(db)
|
|
return templates.TemplateResponse(
|
|
"admin_settings.html",
|
|
{
|
|
"request": request,
|
|
"user": current_user,
|
|
"settings": settings,
|
|
"timezones": get_common_timezones(),
|
|
"unread_count": get_unread_count(db),
|
|
},
|
|
)
|