kaushik/rmm-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: authenticate installer download, rotate-capable agent token, thermal alerts, vitals history

Browse Source 
- deploy_agent.sh download now requires auth in EVERY mode (agent token via
  header/?token=, or a dashboard session). It was gated on strict mode only,
  so in grace mode the served installer published the live fleet token to the
  internet
- accept AGENT_TOKEN_PREVIOUS alongside AGENT_TOKEN so a rotation can roll
  through the fleet; record agent_token_current per node to know when the
  previous token can be dropped
- /api/agent-token (dashboard-auth) so the UI can build the install command
- installer: fall back to the default server URL when an inherited one is
  unreachable (stale Tailscale address stranded a fresh install), and derive
  the version banner from the embedded agent instead of a hardcoded string
- CPU thermal alerts to Rocket.Chat at 85C with hysteresis clearing at 75C
- telemetry_history: 1-minute vitals samples, 7-day TTL (~37 MB fleet-wide),
  plus /api/history and /api/history-bulk for dashboard sparklines
- server watches its own disk (85%) after the 2026-07-23 full-disk outage that
  killed mongod; per-heartbeat telemetry logging now opt-in via VERBOSE_TELEMETRY
- /api/logs returns the newest 25 slim entries per client instead of the full
  history (2.2 MB every 3s was most of the server's egress); gzip middleware
- agent speed probe right-sized to 4/2 MB once a day
- server file listing/delete endpoints, search clearing, WAN IP capture

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
kaushik
2026-08-04 11:13:42 +05:30
parent c929176cfa
commit 0025a65f3d
4 changed files with 494 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
CLAUDE.md
View File

@@ -4,6 +4,54 @@ This VM (`vm3-mint`, LAN `192.168.1.201`, Tailscale `100.79.183.41`) is the
**production server** for the SeekRight Pulse RMM system. Field agents and the
public dashboard both talk to services running here.
## FIELD-SYSTEM SAFETY IS THE TOP PRIORITY
The agents run on ~26 production toll-plaza machines across India with no easy
physical access. A broken agent rollout can strand the whole fleet. Every
change MUST follow these rules:
1. **Backward compatible always.** Old agents must keep working against a new
backend, and new agents against an old backend. Additive changes only:
new heartbeat-response keys (old agents ignore them) and new Optional
telemetry fields (old backends… see rule 2). Never rename/remove existing
API fields, endpoints, or whitelist command keys agents depend on.
2. **A telemetry field takes three places**: agent payload (deploy_agent.sh),
`TelemetryPayload` in central_api_prototype.py, and the UI. Pydantic
silently DROPS unknown fields — a missing model field fails invisibly.
3. **Agent code is fail-soft.** Anything added to the agent must swallow its
own errors and never break the heartbeat loop. Follow the existing
try/except-print pattern. No new inbound ports, no shells for data the
Python stdlib/psutil can read, outbound connections to our server (plus
the existing ipify/Rocket.Chat calls) only.
4. **Canary before fleet.** Update NH-8 (or one expendable node) first,
verify version + token_ok + features on its card, then batch the rest.
Each update = ~5 min agent downtime + Rocket.Chat down/up alert pair.
5. **Whitelist discipline.** Agents execute ONLY commands in commands.json.
Keep new commands read-only unless explicitly required; remember the file
is served to ALL agents (no per-client commands) and read per request.
6. **Token injection is sacred.** The server replaces exactly the assignment
`AGENT_TOKEN="__AGENT_TOKEN__"` in the served installer (count=1) and the
installer's sentinel check uses a split placeholder (`"__AGENT_""TOKEN__"`)
so injection can't rewrite it. Broke once (2026-07-16, agents got empty
tokens); don't reintroduce.
7. **Never flip strict auth** (`set_strict.py`) until every system that
matters shows `agent_token_ok: true` — strict mode locks tokenless agents
out AND blocks them from downloading the installer to fix themselves.
8. **Verify before rollout**: `bash -n deploy_agent.sh`, extract the embedded
agent (between the heredoc markers) and `ast.parse` it, and exercise new
functions against a mock where feasible.
## Agent version history
- **legacy / pre-3.2** — no token, no version reporting, old two-call poll
(`/api/telemetry` + `/api/get-command`). Still on: HYDTOT pair + offline nodes.
- **3.2-auth** — X-Agent-Token auth, single `/api/heartbeat` call, sends
cpu_temp + agent_version. Deployed fleet-wide 2026-07-16.
- **3.4-net** (current template, NOT yet fleet-deployed) — dashboard-configured
SHIFT path for video fetch, SHIFT folder search, 6-hourly link-speed probe
vs our own server, local interface IPs in every heartbeat (`import socket`
is stdlib IP discovery — NOT a websocket; no packets sent, nothing listens).
## Architecture
```
@@ -73,6 +121,45 @@ Field agents (X-Agent-Token) ─────────────────
- Installed/updated via `deploy_agent.sh` (served by the backend; in strict
mode downloading it requires a valid agent token).
## Agent token rotation (and why the download is authenticated)
The served installer has the LIVE agent token injected into it. Until
2026-08-03 that download was only gated in strict mode, so in grace mode
`https://rmm-backend.seekright.com/deploy_agent.sh` handed the fleet token to
anyone on the internet. It is now authenticated in EVERY mode (agent token via
`X-Agent-Token` header or `?token=`, or a logged-in dashboard session).
To rotate the token:
1. `.env`: move the current value to `AGENT_TOKEN_PREVIOUS=`, set a new
`AGENT_TOKEN=` (`python3 -c "import secrets; print(secrets.token_urlsafe(32))"`).
Both are accepted while both are set — agents authenticate the self-update
with the old token and receive the new one.
2. Restart the backend, then run `update_agent` across the fleet.
3. When every active node shows `agent_token_ok: true` AND is on the new token,
delete `AGENT_TOKEN_PREVIOUS` from `.env` and restart. Only then flip strict
(`set_strict.py`) — that is what actually closes the agent endpoints.
Note: grace mode means the agent endpoints accept UNAUTHENTICATED requests
today. Rotating the token stops the leak, but strict mode is what enforces it.
## Installing the agent on a NEW system
```bash
curl -fsSL "https://rmm-backend.seekright.com/deploy_agent.sh" -o /tmp/deploy_agent.sh
sudo bash /tmp/deploy_agent.sh <CLIENT_ID> # e.g. KRBOT-Narwana
```
- ALWAYS pass a CLIENT_ID on a fresh install — the fallback is the machine's
hostname (that's how the stale `hamsadmin-MS-7E07` client entry happened).
- ALWAYS download from the server, never copy deploy_agent.sh out of the repo:
the server injects the live agent token at download time.
- Re-running with no args is a safe in-place upgrade (keeps the node's
identity) — the `update_agent` fleet command relies on this.
- In strict auth mode the download itself needs the token:
append `?token=<AGENT_TOKEN from .env>` to the URL.
- After install: the node appears in the dashboard in ~30s; set its SHIFT
folder path in the UI card so video fetch/search works.
## How to deploy
Backend:
@@ -117,6 +204,14 @@ curl -sk -o /dev/null -w '%{http_code}\n' https://rmm.seekright.com/ # via Syn
(localhost only), 7514 + others auditor portal, 4430–4433 MeshCentral,
Grafana also runs here. Check `ss -tlnp` before assigning a new port.
- 2026-07-23 INCIDENT: the server's own disk hit 100% → mongod fatally aborted
("Writing to log file failed") and stayed down ~22h (no Restart= in its unit)
→ dashboard showed zero systems while agents kept heartbeating into a dead DB.
Mitigations now in code: per-heartbeat telemetry stdout dumps are opt-in
(`VERBOSE_TELEMETRY=true` in .env to re-enable) and the backend Rocket.Chat
alerts when its own `/` passes 85%. Recommended systemd hardening: a mongod
override with `Restart=on-failure`, and `SystemMaxUse=2G` in journald.conf.
## Other services on this VM (not part of RMM deploys)
- `meshcentral.service` — MeshCentral remote management (node, ports 4430–4433)

339
central_api_prototype.py
View File

@@ -14,6 +14,7 @@ load_dotenv()
from fastapi import FastAPI, HTTPException, Depends, Header, Request, BackgroundTasks
from fastapi.responses import PlainTextResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from pydantic import BaseModel
from typing import Dict, Any, List, Optional
import uvicorn
@@ -24,6 +25,10 @@ import json
from datetime import datetime, timedelta
app = FastAPI(title="RMM Central API")
# Dashboard polling was ~90% of this server's 428 GB/month egress; JSON
# compresses ~15x. Agents' urllib doesn't send Accept-Encoding, so agent
# traffic (incl. speedtest blobs) is unaffected.
app.add_middleware(GZipMiddleware, minimum_size=1024)
# Enable CORS for frontend dashboard queries (React dev server runs on a separate port)
app.add_middleware(
@@ -37,6 +42,10 @@ app.add_middleware(
# JWT-like HMAC Secure Token Utilities
JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY", "seekright_rmm_secret_key_2026_default")
# Per-heartbeat telemetry dumps flooded journald+syslog and helped fill the disk
# (2026-07-23: / hit 100% and mongod fatally aborted). Now opt-in via env.
VERBOSE_TELEMETRY = os.getenv("VERBOSE_TELEMETRY", "false").lower() in ("1", "true", "yes")
def generate_token(username: str) -> str:
# Expire in 1 day (24 hours)
expiry = (datetime.utcnow() + timedelta(days=1)).isoformat()
@@ -106,6 +115,16 @@ clients_collection = db["clients"]
logs_collection = db["logs"]
config_collection = db["config"]
# Rolling vitals history: one slim sample per client per minute, expired
# automatically by MongoDB after 7 days (~1.1 MB/week/system, ~35 MB fleet).
history_collection = db["telemetry_history"]
HISTORY_RETENTION_DAYS = 7
try:
history_collection.create_index("ts", expireAfterSeconds=HISTORY_RETENTION_DAYS * 24 * 3600)
history_collection.create_index([("client_id", 1), ("ts", 1)])
except Exception as e:
print(f"[!] Could not create telemetry_history indexes: {e}")
# --- Agent authentication ---
# Shared secret agents present in X-Agent-Token. Auth mode ("grace"|"strict") is
# stored in the config collection so it can be flipped live from the dashboard:
@@ -117,8 +136,19 @@ def get_agent_auth_mode() -> str:
doc = config_collection.find_one({"_id": "agent_auth"})
return (doc or {}).get("mode", "grace")
# During a token rotation both the new and the outgoing token must validate,
# otherwise agents can't authenticate the very download that gives them the new
# one. Set AGENT_TOKEN_PREVIOUS in .env while rotating, then clear it once the
# whole fleet reports the new token.
AGENT_TOKEN_PREVIOUS = os.getenv("AGENT_TOKEN_PREVIOUS", "")
def is_valid_agent_token(token: Optional[str]) -> bool:
return bool(AGENT_TOKEN) and token is not None and hmac.compare_digest(token, AGENT_TOKEN)
if token is None:
return False
for accepted in (AGENT_TOKEN, AGENT_TOKEN_PREVIOUS):
if accepted and hmac.compare_digest(token, accepted):
return True
return False
async def require_agent_token(x_agent_token: str = Header(None)):
"""Dependency guarding every agent-facing endpoint. Enforces only in strict mode;
@@ -198,6 +228,37 @@ def append_to_logs(log_type: str, client_id: str, detail: Any):
except Exception as e:
print(f"[!] Error writing log to MongoDB: {e}")
def record_history_sample(client_id: str, payload):
"""
Store one slim vitals sample per client per minute. The _id is the
client+minute bucket so concurrent uvicorn workers can't duplicate a
sample, and $setOnInsert keeps the first write of each minute.
"""
try:
bucket = datetime.now().replace(second=0, microsecond=0)
gpu_temp = None
for g in (payload.gpus or []):
raw = str(g.get("temp", "")).strip().rstrip("C").strip()
try:
gpu_temp = float(raw)
break
except ValueError:
continue
history_collection.update_one(
{"_id": f"{client_id}|{bucket.isoformat()}"},
{"$setOnInsert": {
"client_id": client_id,
"ts": bucket,
"cpu": payload.cpu_percent,
"cpu_temp": payload.cpu_temp,
"ram": payload.memory_percent,
"gpu_temp": gpu_temp,
}},
upsert=True,
)
except Exception as e:
print(f"[!] History sample failed for {client_id}: {e}")
def mark_client_active(client_id: str):
try:
now_str = datetime.now().isoformat()
@@ -257,9 +318,41 @@ def load_clients() -> Dict[str, Any]:
return {}
import asyncio
import shutil
CPU_TEMP_ALERT_C = float(os.getenv("CPU_TEMP_ALERT_C", "85"))
CPU_TEMP_CLEAR_C = float(os.getenv("CPU_TEMP_CLEAR_C", "75"))
SERVER_DISK_ALERT_PERCENT = 85.0
_server_disk_state = {"alerted_at": None}
def check_server_disk():
# The RMM watches every site's disks but died on its own full disk
# (2026-07-23: / hit 100%, mongod aborted). Watch our own disk too.
try:
usage = shutil.disk_usage("/")
pct = round(usage.used / usage.total * 100, 1)
free_gb = round(usage.free / (1024 ** 3), 1)
now = datetime.now()
if pct >= SERVER_DISK_ALERT_PERCENT:
last = _server_disk_state["alerted_at"]
if last is None or (now - last).total_seconds() >= 3600:
_server_disk_state["alerted_at"] = now
send_rocketchat_notification(
text=f"🚨 **RMM SERVER DISK:** {pct}% used, only {free_gb} GB free on the central server. At 100% MongoDB dies — clean up now.",
color="#e74c3c",
)
elif _server_disk_state["alerted_at"] is not None:
_server_disk_state["alerted_at"] = None
send_rocketchat_notification(
text=f"✅ **RMM SERVER DISK recovered:** {pct}% used, {free_gb} GB free.",
color="#2ecc71",
)
except Exception as e:
print(f"[!] Server disk self-check failed: {e}")
async def monitor_heartbeats_loop():
while True:
check_server_disk()
try:
now = datetime.now()
timeout_time = (now - timedelta(seconds=30)).isoformat()
@@ -326,6 +419,7 @@ class TelemetryPayload(BaseModel):
disks: List[Dict[str, Any]]
gpus: List[Dict[str, Any]]
cpu_temp: Optional[float] = None
local_ips: Optional[List[str]] = None
agent_version: Optional[str] = None
speed_upload_mbps: Optional[float] = None
speed_download_mbps: Optional[float] = None
@@ -467,7 +561,8 @@ async def receive_telemetry(client_id: str, payload: TelemetryPayload, token_val
)
mark_client_active(client_id)
record_history_sample(client_id, payload)
# Get current alert states from database to prevent duplicate alerts
client_doc = clients_collection.find_one({"_id": client_id}) or {}
storage_alerts = client_doc.get("active_storage_alerts", [])
@@ -516,6 +611,27 @@ async def receive_telemetry(client_id: str, payload: TelemetryPayload, token_val
color="#2ecc71"
)
# CPU thermal alerting. Hysteresis (alert at 85C, clear at 75C) stops a node
# hovering on the threshold from spamming Rocket.Chat every 10 seconds.
temp_alert_sent = client_doc.get("cpu_temp_alert_sent", False)
temp_modified = False
cpu_temp = payload.cpu_temp
if cpu_temp is not None:
if cpu_temp >= CPU_TEMP_ALERT_C and not temp_alert_sent:
temp_alert_sent = True
temp_modified = True
send_rocketchat_notification(
text=f"🔥 **CPU OVERHEATING:** Client `{client_id}` CPU at **{cpu_temp}°C** (threshold {CPU_TEMP_ALERT_C}°C). Check cooling — thermal shutdown risk.",
color="#e74c3c",
)
elif cpu_temp <= CPU_TEMP_CLEAR_C and temp_alert_sent:
temp_alert_sent = False
temp_modified = True
send_rocketchat_notification(
text=f"✅ **CPU temperature normal:** Client `{client_id}` back down to {cpu_temp}°C.",
color="#2ecc71",
)
update_fields = {
"telemetry": payload.dict()
}
@@ -523,6 +639,8 @@ async def receive_telemetry(client_id: str, payload: TelemetryPayload, token_val
update_fields["active_storage_alerts"] = storage_alerts
if gpu_modified:
update_fields["gpu_alert_sent"] = gpu_alert_sent
if temp_modified:
update_fields["cpu_temp_alert_sent"] = temp_alert_sent
clients_collection.update_one(
{"_id": client_id},
@@ -535,19 +653,20 @@ async def receive_telemetry(client_id: str, payload: TelemetryPayload, token_val
data = payload.dict()
client_telemetry[client_id] = data
# Log received stats
print(f"\n[TELEMETRY RECEIVED] from {client_id}:")
print(f" CPU: {data['cpu_percent']}%")
print(f" RAM: {data['memory_percent']}% ({data['memory_free_gb']} GB Free / {data['memory_total_gb']} GB Total)")
for disk in data['disks']:
print(f" Drive {disk['mount']}: {disk['percent']}% Used ({disk['free_gb']} GB Free / {disk['total_gb']} GB Total)")
for gpu in data['gpus']:
power_str = f"Power: {gpu.get('power_draw', 'N/A')} / {gpu.get('power_limit', 'N/A')}"
fan_str = f"Fan: {gpu.get('fan_speed', 'N/A')}"
print(f" GPU [{gpu['name']}]: Core: {gpu['utilization']}, Temp: {gpu['temp']}, VRAM: {gpu['memory_used']}/{gpu['memory_total']}, {power_str}, {fan_str}")
# Log received stats (opt-in: this was the main journald/syslog flood source)
if VERBOSE_TELEMETRY:
print(f"\n[TELEMETRY RECEIVED] from {client_id}:")
print(f" CPU: {data['cpu_percent']}%")
print(f" RAM: {data['memory_percent']}% ({data['memory_free_gb']} GB Free / {data['memory_total_gb']} GB Total)")
for disk in data['disks']:
print(f" Drive {disk['mount']}: {disk['percent']}% Used ({disk['free_gb']} GB Free / {disk['total_gb']} GB Total)")
for gpu in data['gpus']:
power_str = f"Power: {gpu.get('power_draw', 'N/A')} / {gpu.get('power_limit', 'N/A')}"
fan_str = f"Fan: {gpu.get('fan_speed', 'N/A')}"
print(f" GPU [{gpu['name']}]: Core: {gpu['utilization']}, Temp: {gpu['temp']}, VRAM: {gpu['memory_used']}/{gpu['memory_total']}, {power_str}, {fan_str}")
return {"status": "success"}
data = payload.dict()
@@ -589,21 +708,32 @@ async def get_logs_history(current_user: str = Depends(get_current_user)):
Endpoint to retrieve logs history grouped by client_id.
"""
try:
cursor = logs_collection.find().sort("timestamp", 1)
# Newest 25 per client with slim details. The full-fat version shipped
# every stored telemetry payload (2 MB+ JSON) and the dashboard re-polls
# this every 3s — it was hanging the browser.
grouped_logs = {}
cursor = logs_collection.find().sort("timestamp", -1)
for doc in cursor:
cid = doc.get("client_id")
if not cid:
continue
if cid not in grouped_logs:
grouped_logs[cid] = []
entry = {
bucket = grouped_logs.setdefault(cid, [])
if len(bucket) >= 25:
continue
dtype = doc.get("type")
detail = doc.get("detail")
if dtype == "telemetry" and isinstance(detail, dict):
detail = {"cpu_percent": detail.get("cpu_percent"),
"memory_percent": detail.get("memory_percent")}
elif isinstance(detail, dict):
detail = {k: (v[:400] if isinstance(v, str) else v) for k, v in detail.items()}
bucket.append({
"timestamp": doc.get("timestamp"),
"type": doc.get("type"),
"detail": doc.get("detail")
}
grouped_logs[cid].append(entry)
"type": dtype,
"detail": detail
})
for cid in grouped_logs:
grouped_logs[cid].reverse()
return grouped_logs
except Exception as e:
print(f"[!] Error loading logs from MongoDB: {e}")
@@ -776,6 +906,64 @@ async def request_search(client_id: str, query: str, current_user: str = Depends
append_to_logs("search_requested", client_id, {"query": q})
return {"status": "success"}
@app.get("/api/history")
async def get_history(client_id: str, hours: int = 6, current_user: str = Depends(get_current_user)):
"""Vitals history for one node — powers the dashboard sparklines."""
hours = min(max(hours, 1), HISTORY_RETENTION_DAYS * 24)
since = datetime.now() - timedelta(hours=hours)
try:
cursor = history_collection.find(
{"client_id": client_id, "ts": {"$gte": since}},
{"_id": 0, "client_id": 0},
).sort("ts", 1)
return [{
"ts": d["ts"].isoformat(),
"cpu": d.get("cpu"),
"cpu_temp": d.get("cpu_temp"),
"ram": d.get("ram"),
"gpu_temp": d.get("gpu_temp"),
} for d in cursor]
except Exception as e:
print(f"[!] Error loading history for {client_id}: {e}")
return []
@app.get("/api/agent-token")
async def get_agent_token(current_user: str = Depends(get_current_user)):
"""Current agent token, for building the install command in the dashboard.
Logged-in operators only — the installer download itself is now authenticated."""
return {"token": AGENT_TOKEN}
@app.get("/api/history-bulk")
async def get_history_bulk(points: int = 60, current_user: str = Depends(get_current_user)):
"""
Recent vitals for every node in one request — the dashboard sparklines poll
this instead of one request per card.
"""
points = min(max(points, 5), 240)
since = datetime.now() - timedelta(minutes=points)
out = {}
try:
cursor = history_collection.find(
{"ts": {"$gte": since}},
{"_id": 0, "ts": 1, "client_id": 1, "cpu": 1, "cpu_temp": 1, "ram": 1},
).sort("ts", 1)
for d in cursor:
out.setdefault(d["client_id"], []).append({
"ts": d["ts"].isoformat(),
"cpu": d.get("cpu"),
"cpu_temp": d.get("cpu_temp"),
"ram": d.get("ram"),
})
except Exception as e:
print(f"[!] Error loading bulk history: {e}")
return out
@app.post("/api/clear-search")
async def clear_search(client_id: str, current_user: str = Depends(get_current_user)):
"""Drop a node's stored search result so the dashboard stops showing it."""
clients_collection.update_one({"_id": client_id}, {"$unset": {"file_search": ""}})
return {"status": "success"}
@app.post("/api/search-results")
async def receive_search_results(client_id: str, payload: SearchResultsPayload, token_valid: bool = Depends(require_agent_token)):
now_str = datetime.now().isoformat()
@@ -884,6 +1072,71 @@ async def upload_chunk(client_id: str, filename: str, chunk_index: int, request:
return {"status": "success", "chunk_index": chunk_index}
@app.get("/api/server-files")
async def list_server_files(current_user: str = Depends(get_current_user)):
"""
All videos currently stored on the server (per client), so the dashboard
can show and manage what is occupying the 20 GB transfer store.
"""
data = {}
try:
for cid in sorted(os.listdir(FILE_TRANSFER_DIR)):
client_dir = os.path.join(FILE_TRANSFER_DIR, cid)
if not os.path.isdir(client_dir):
continue
files = []
for name in sorted(os.listdir(client_dir)):
if ".part" in name:
continue
path = os.path.join(client_dir, name)
if os.path.isfile(path):
st = os.stat(path)
files.append({
"name": name,
"size_mb": round(st.st_size / (1024 * 1024), 2),
"modified": datetime.fromtimestamp(st.st_mtime).isoformat(timespec="seconds"),
})
if files:
data[cid] = files
except Exception as e:
print(f"[!] Error listing server files: {e}")
return data
@app.post("/api/delete-server-file")
async def delete_server_file(client_id: str, filename: str, current_user: str = Depends(get_current_user)):
safe_client = os.path.basename(client_id.strip())
safe_name = os.path.basename(filename.strip())
if not safe_client or not safe_name:
raise HTTPException(status_code=400, detail="client_id and filename are required")
path = os.path.join(FILE_TRANSFER_DIR, safe_client, safe_name)
if not os.path.isfile(path):
raise HTTPException(status_code=404, detail="File not found on server")
os.remove(path)
# Drop the stale 'ready' transfer card if it points at the deleted file
doc = clients_collection.find_one({"_id": client_id}, {"file_transfer": 1})
if doc and (doc.get("file_transfer") or {}).get("filename") == safe_name:
clients_collection.update_one({"_id": client_id}, {"$unset": {"file_transfer": ""}})
append_to_logs("server_file_deleted", client_id, {"filename": safe_name})
return {"status": "success", "message": f"Deleted '{safe_name}' from server"}
@app.get("/api/speedtest-blob")
async def speedtest_blob(size_mb: int = 8, token_valid: bool = Depends(require_agent_token)):
# Agents time this download to estimate their downlink to the server.
size_mb = max(1, min(int(size_mb), 32))
def gen():
chunk = b"\0" * (1024 * 1024)
for _ in range(size_mb):
yield chunk
return StreamingResponse(gen(), media_type="application/octet-stream")
@app.post("/api/speedtest-sink")
async def speedtest_sink(request: Request, token_valid: bool = Depends(require_agent_token)):
# Agents time this upload; the body is read and discarded.
total = 0
async for chunk in request.stream():
total += len(chunk)
return {"status": "success", "received_bytes": total}
def enforce_storage_limit(max_bytes: int = 20 * 1024**3):
"""Scan FILE_TRANSFER_DIR recursively and delete oldest completed files if total size > max_bytes."""
all_files = []
@@ -950,13 +1203,19 @@ async def get_file_transfers(current_user: str = Depends(get_current_user)):
print(f"[!] Error loading file transfers: {e}")
return {}
from fastapi.responses import FileResponse, Response
from fastapi.responses import FileResponse, Response, StreamingResponse
@app.get("/deploy_agent.sh")
async def get_agent_installer(x_agent_token: str = Header(None), token: Optional[str] = None):
# In strict mode the installer carries the live token, so downloading it must
# itself be authenticated (header for agent self-update, ?token= for humans).
if get_agent_auth_mode() == "strict" and not (is_valid_agent_token(x_agent_token) or is_valid_agent_token(token)):
async def get_agent_installer(x_agent_token: str = Header(None), token: Optional[str] = None,
authorization: str = Header(None)):
# The served installer carries the LIVE agent token, so this download is
# authenticated in EVERY mode — gating it on strict mode published the
# fleet's token to the internet. Accepted proof: an agent token (header for
# self-update, ?token= for a human running the install by hand) or a
# logged-in dashboard session.
dashboard_ok = bool(authorization and authorization.startswith("Bearer ")
and verify_token(authorization.split(" ")[1]))
if not (is_valid_agent_token(x_agent_token) or is_valid_agent_token(token) or dashboard_ok):
raise HTTPException(status_code=401, detail="Invalid or missing agent token")
path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "deploy_agent.sh")
if not os.path.isfile(path):
@@ -1000,18 +1259,28 @@ async def download_file(client_id: str, filename: str, token: Optional[str] = No
return FileResponse(path, media_type=media_type, filename=safe_name)
@app.post("/api/heartbeat")
async def receive_heartbeat(client_id: str, payload: TelemetryPayload,
async def receive_heartbeat(client_id: str, payload: TelemetryPayload, request: Request,
token_valid: bool = Depends(require_agent_token)):
await receive_telemetry(client_id, payload)
# Record migration status so the dashboard can confirm the fleet is upgraded
# and every node is authenticated before auth is flipped to strict.
clients_collection.update_one(
{"_id": client_id},
{"$set": {"agent_version": payload.agent_version or "unknown",
# During rotation both tokens validate, so also record whether the agent
# holds the CURRENT one — that is the signal for "safe to drop the previous
# token and flip strict".
presented = request.headers.get("x-agent-token")
token_is_current = bool(AGENT_TOKEN and presented
and hmac.compare_digest(presented, AGENT_TOKEN))
set_fields = {"agent_version": payload.agent_version or "unknown",
"speed_upload_mbps": payload.speed_upload_mbps,
"speed_download_mbps": payload.speed_download_mbps,
"agent_token_ok": token_valid}},
)
"agent_token_ok": token_valid,
"agent_token_current": token_is_current}
# The Synology reverse proxy forwards the site's real public IP — record it
# so every node's WAN address is visible without any agent-side lookup.
xff = (request.headers.get("x-forwarded-for") or "").split(",")[0].strip()
if xff:
set_fields["public_ip"] = xff
clients_collection.update_one({"_id": client_id}, {"$set": set_fields})
doc = clients_collection.find_one_and_update(
{"_id": client_id},
{"$set": {"pending_command": "none", "pending_file_request": "none", "pending_search": "none"}},

5
commands.json
View File

@@ -337,6 +337,11 @@
"bash",
"-c",
"echo HOST:$(hostname); echo USER:$(id -un); echo MACHINE_ID:$(cat /etc/machine-id 2>/dev/null); echo AGENT_PROCS:; pgrep -af client_agent_prototype 2>/dev/null; echo APT_PROCS:; ps aux | grep -E \"apt-get|apt |dpkg|unattended\" | grep -v grep; echo UPDATE_UNITS:; systemctl list-units \"run-r*\" --no-legend --all 2>&1 | head -5"
],
"public_ip": [
"bash",
"-c",
"curl -s --max-time 10 http://api.ipify.org || curl -sk --max-time 10 https://api.ipify.org || curl -sk --max-time 10 https://ifconfig.me"
]
}
}

93
deploy_agent.sh
View File

@@ -32,8 +32,19 @@ if [ -f "$EXISTING_PY" ]; then
fi
# Fallbacks when nothing was provided and there is no prior install
DEFAULT_CENTRAL_URL="http://rmm-backend.seekright.com"
if [ -z "$CLIENT_ID" ]; then CLIENT_ID="$(hostname)"; fi
if [ -z "$CENTRAL_URL" ]; then CENTRAL_URL="http://rmm-backend.seekright.com"; fi
if [ -z "$CENTRAL_URL" ]; then CENTRAL_URL="$DEFAULT_CENTRAL_URL"; fi
# A URL inherited from a previous install can be stale (old Tailscale/LAN
# address), which silently strands the agent in a heartbeat-timeout loop.
# Probe it, and fall back to the public URL when it is unreachable.
if [ -z "$2" ] && [ "$CENTRAL_URL" != "$DEFAULT_CENTRAL_URL" ]; then
if ! curl -fsS --max-time 8 -o /dev/null "$CENTRAL_URL/docs" 2>/dev/null; then
echo "⚠️ Inherited server URL $CENTRAL_URL is unreachable — falling back to $DEFAULT_CENTRAL_URL"
CENTRAL_URL="$DEFAULT_CENTRAL_URL"
fi
fi
# Agent token. The server injects the live token into the line below at download
# time (replacing __AGENT_TOKEN__). Priority: CLI arg > server-injected > existing install.
@@ -50,7 +61,10 @@ elif [ "$AGENT_TOKEN" = "__AGENT_""TOKEN__" ]; then
fi
fi
echo "🚀 Preparing SeekRight RMM Agent installation (agent v3.2-auth)..."
# Read the version out of the embedded agent template so the banner can never
# drift from what actually gets installed.
AGENT_VERSION_LABEL="$(sed -n 's/^AGENT_VERSION = "\(.*\)"/\1/p' "$0" | head -1)"
echo "🚀 Preparing SeekRight RMM Agent installation (agent v${AGENT_VERSION_LABEL:-unknown})..."
echo "📍 Target Client ID: $CLIENT_ID"
echo "🌐 Central Server URL: $CENTRAL_URL"
@@ -81,6 +95,7 @@ import subprocess
import psutil
import os
import sys
import socket
import string
import threading
@@ -101,7 +116,7 @@ def get_cpu_temp():
pass
return None
AGENT_VERSION = "3.3-shift"
AGENT_VERSION = "3.4-net"
CLIENT_ID = "TEMPLATE_CLIENT_ID"
CENTRAL_BASE = "TEMPLATE_CENTRAL_URL"
@@ -127,6 +142,8 @@ command_in_progress = False
file_transfer_in_progress = False
SHIFT_PATH = None # per-site SHIFT folder, configured on the dashboard, delivered via heartbeat
search_in_progress = False
speed_download_mbps = None
speed_upload_mbps = None
def run_command_async(command_key, safe_cmd_list, is_shutdown_or_restart):
global command_in_progress
@@ -343,6 +360,9 @@ def send_heartbeat():
"memory_free_gb": round(psutil.virtual_memory().available / (1024**3), 2),
"disks": disks,
"gpus": gpus,
"speed_download_mbps": speed_download_mbps,
"speed_upload_mbps": speed_upload_mbps,
"local_ips": get_local_ips(),
"agent_version": AGENT_VERSION
}
@@ -705,6 +725,70 @@ def list_takeleap_subfolder_files(subfolder_name):
def list_takeleap_subfolder(subfolder_name):
list_takeleap_subfolder_files(subfolder_name)
def get_local_ips():
# All local interface addresses, primary outbound first — reported in every
# heartbeat so the dashboard shows them without any remote command.
ips = []
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(2)
s.connect(("8.8.8.8", 80))
ips.append(s.getsockname()[0])
s.close()
except Exception:
pass
try:
for iface, addrs in psutil.net_if_addrs().items():
if iface == "lo":
continue
for a in addrs:
if int(a.family) == int(socket.AF_INET) and a.address and not a.address.startswith("127."):
entry = f"{a.address} ({iface})"
if a.address not in [i.split(" ")[0] for i in ips]:
ips.append(entry)
except Exception:
pass
return ips[:6]
def run_speed_test():
# Brief probe against our own server (4 MB down / 2 MB up) — measures the
# link that actually matters for video fetches. Never runs during a transfer.
global speed_download_mbps, speed_upload_mbps
try:
req = urllib.request.Request(f"{CENTRAL_BASE}/api/speedtest-blob?size_mb=4")
req.add_header('ngrok-skip-browser-warning', 'true')
start = time.time()
received = 0
with urllib.request.urlopen(req, timeout=180) as r:
while True:
chunk = r.read(65536)
if not chunk:
break
received += len(chunk)
elapsed = max(time.time() - start, 0.001)
speed_download_mbps = round(received * 8 / (elapsed * 1000 * 1000), 2)
except Exception as e:
print(f" -> [!] Download speed test failed: {e}")
try:
blob = b"\0" * (2 * 1024 * 1024)
req = urllib.request.Request(f"{CENTRAL_BASE}/api/speedtest-sink", data=blob, method='POST')
req.add_header('Content-Type', 'application/octet-stream')
req.add_header('ngrok-skip-browser-warning', 'true')
start = time.time()
urllib.request.urlopen(req, timeout=180)
elapsed = max(time.time() - start, 0.001)
speed_upload_mbps = round(len(blob) * 8 / (elapsed * 1000 * 1000), 2)
except Exception as e:
print(f" -> [!] Upload speed test failed: {e}")
print(f" -> Link speed: down {speed_download_mbps} Mbps / up {speed_upload_mbps} Mbps")
def speed_test_loop():
time.sleep(120) # let the agent settle before the first probe
while True:
if not file_transfer_in_progress:
run_speed_test()
time.sleep(24 * 3600)
def find_shift_roots():
# The dashboard-configured path wins; without one, fall back to any
# */TAKELEAP/SHIFT discovered by the legacy drive scan.
@@ -1023,6 +1107,9 @@ if __name__ == "__main__":
sys.exit(1)
else:
print(f"Starting Agent v{AGENT_VERSION} for {CLIENT_ID}...")
t = threading.Thread(target=speed_test_loop)
t.daemon = True
t.start()
try:
while True:
send_heartbeat()