f9743bb29a
Full-stack implementation: FastAPI backend with PAM auth, WebSocket stats/terminal, and vanilla JS frontend with tiling desktop shell. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
89 lines
3 KiB
Python
89 lines
3 KiB
Python
"""Process list and signal management."""
|
|
|
|
import os
|
|
import signal
|
|
|
|
import psutil
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from backend.auth import get_current_user
|
|
|
|
router = APIRouter(prefix="/api/processes", tags=["processes"])
|
|
|
|
|
|
class SignalRequest(BaseModel):
|
|
pid: int
|
|
signal: str = "SIGTERM" # SIGTERM, SIGKILL, SIGSTOP, SIGCONT, SIGHUP
|
|
|
|
|
|
@router.get("")
|
|
async def list_processes(_user: str = Depends(get_current_user)):
|
|
"""List all running processes."""
|
|
procs = []
|
|
for p in psutil.process_iter(["pid", "name", "username", "cpu_percent",
|
|
"memory_percent", "status", "create_time",
|
|
"cmdline"]):
|
|
try:
|
|
info = p.info
|
|
procs.append({
|
|
"pid": info["pid"],
|
|
"name": info["name"],
|
|
"user": info["username"],
|
|
"cpu": info["cpu_percent"],
|
|
"mem": round(info["memory_percent"], 1) if info["memory_percent"] else 0,
|
|
"status": info["status"],
|
|
"started": info["create_time"],
|
|
"cmdline": " ".join(info["cmdline"][:5]) if info["cmdline"] else "",
|
|
})
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
|
continue
|
|
return procs
|
|
|
|
|
|
@router.get("/{pid}")
|
|
async def process_detail(pid: int, _user: str = Depends(get_current_user)):
|
|
try:
|
|
p = psutil.Process(pid)
|
|
with p.oneshot():
|
|
return {
|
|
"pid": p.pid,
|
|
"name": p.name(),
|
|
"user": p.username(),
|
|
"cpu": p.cpu_percent(),
|
|
"mem": round(p.memory_percent(), 1),
|
|
"status": p.status(),
|
|
"started": p.create_time(),
|
|
"cmdline": p.cmdline(),
|
|
"cwd": p.cwd(),
|
|
"num_threads": p.num_threads(),
|
|
"open_files": len(p.open_files()),
|
|
"connections": len(p.net_connections()),
|
|
"memory_info": {
|
|
"rss": p.memory_info().rss,
|
|
"vms": p.memory_info().vms,
|
|
},
|
|
}
|
|
except psutil.NoSuchProcess:
|
|
raise HTTPException(404, f"Process {pid} not found")
|
|
except psutil.AccessDenied:
|
|
raise HTTPException(403, f"Access denied to process {pid}")
|
|
|
|
|
|
@router.post("/signal")
|
|
async def send_signal(req: SignalRequest, _user: str = Depends(get_current_user)):
|
|
sig_name = req.signal.upper()
|
|
if not sig_name.startswith("SIG"):
|
|
sig_name = "SIG" + sig_name
|
|
try:
|
|
sig = getattr(signal, sig_name)
|
|
except AttributeError:
|
|
raise HTTPException(400, f"Unknown signal: {req.signal}")
|
|
|
|
try:
|
|
os.kill(req.pid, sig)
|
|
except ProcessLookupError:
|
|
raise HTTPException(404, f"Process {req.pid} not found")
|
|
except PermissionError:
|
|
raise HTTPException(403, f"Permission denied for pid {req.pid}")
|
|
return {"pid": req.pid, "signal": sig_name, "sent": True}
|