From 0025a65f3d13a20b48d8a46bb7c524ed88190dfe Mon Sep 17 00:00:00 2001 From: kaushik Date: Tue, 4 Aug 2026 11:13:42 +0530 Subject: [PATCH] fix: authenticate installer download, rotate-capable agent token, thermal alerts, vitals history - 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 --- CLAUDE.md | 95 +++++++++++ central_api_prototype.py | 339 +++++++++++++++++++++++++++++++++++---- commands.json | 5 + deploy_agent.sh | 93 ++++++++++- 4 files changed, 494 insertions(+), 38 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 64c74e7..b0b8662 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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 # 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=` 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) diff --git a/central_api_prototype.py b/central_api_prototype.py index 4486dea..93945f5 100644 --- a/central_api_prototype.py +++ b/central_api_prototype.py @@ -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"}}, diff --git a/commands.json b/commands.json index b4e7ab8..a7c7caa 100644 --- a/commands.json +++ b/commands.json @@ -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" ] } } diff --git a/deploy_agent.sh b/deploy_agent.sh index 7f759c9..70814dc 100644 --- a/deploy_agent.sh +++ b/deploy_agent.sh @@ -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()