Compare commits
19 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 660474a6ce | |||
| 93dd14967d | |||
| 923a3872fe | |||
| 5f10e34341 | |||
| b80d322650 | |||
| fff386f9d5 | |||
| 93f4e1feea | |||
| 97255b42fb | |||
| 554a2c349f | |||
| 10501168c5 | |||
| d346c61c28 | |||
| 7652095109 | |||
| 6b58ac67f6 | |||
| 3ad1d52fe2 | |||
| 2e8cc24e81 | |||
| 36e73fd9ed | |||
| 3d14e4a370 | |||
| c6b8c9c905 | |||
| f980b6ace9 |
+81
-13
@@ -5,16 +5,16 @@ on:
|
|||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
build:
|
build:
|
||||||
|
runs-on: ${{ matrix.os }}
|
||||||
strategy:
|
strategy:
|
||||||
matrix:
|
matrix:
|
||||||
os: [ubuntu-latest, windows-latest]
|
os: [ubuntu-latest, windows-latest]
|
||||||
runs-on: ${{ matrix.os }}
|
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v4
|
- uses: actions/checkout@v4
|
||||||
- uses: dtolnay/rust-toolchain@stable
|
- uses: dtolnay/rust-toolchain@stable
|
||||||
with:
|
with:
|
||||||
components: clippy, rustfmt
|
components: clippy, rustfmt
|
||||||
- name: Install system dependencies
|
- name: Install system dependencies (Linux)
|
||||||
if: matrix.os == 'ubuntu-latest'
|
if: matrix.os == 'ubuntu-latest'
|
||||||
run: sudo apt-get update && sudo apt-get install -y libdrm-dev libdrm-amdgpu1
|
run: sudo apt-get update && sudo apt-get install -y libdrm-dev libdrm-amdgpu1
|
||||||
- name: Cargo fmt
|
- name: Cargo fmt
|
||||||
@@ -23,22 +23,89 @@ jobs:
|
|||||||
run: cargo clippy --all-targets --all-features -- -D warnings
|
run: cargo clippy --all-targets --all-features -- -D warnings
|
||||||
- name: Build (release)
|
- name: Build (release)
|
||||||
run: cargo build --release --workspace
|
run: cargo build --release --workspace
|
||||||
|
|
||||||
|
- name: "Linux: start agent and run WS probe"
|
||||||
|
if: matrix.os == 'ubuntu-latest'
|
||||||
|
shell: bash
|
||||||
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
RUST_LOG=info SOCKTOP_ENABLE_SSL=0 SOCKTOP_AGENT_GPU=0 SOCKTOP_AGENT_TEMP=0 ./target/release/socktop_agent -p 3000 > agent.log 2>&1 &
|
||||||
|
AGENT_PID=$!
|
||||||
|
for i in {1..60}; do
|
||||||
|
if curl -fsS http://127.0.0.1:3000/healthz >/dev/null; then break; fi
|
||||||
|
sleep 1
|
||||||
|
done
|
||||||
|
if ! curl -fsS http://127.0.0.1:3000/healthz >/dev/null; then
|
||||||
|
echo "--- agent.log (tail) ---"
|
||||||
|
tail -n 200 agent.log || true
|
||||||
|
(command -v ss >/dev/null && ss -ltnp || netstat -ltnp) || true
|
||||||
|
kill $AGENT_PID || true
|
||||||
|
exit 1
|
||||||
|
fi
|
||||||
|
SOCKTOP_WS=ws://127.0.0.1:3000/ws cargo test -p socktop --test ws_probe -- --nocapture
|
||||||
|
kill $AGENT_PID || true
|
||||||
|
|
||||||
|
- name: "Windows: start agent and run WS probe"
|
||||||
|
if: matrix.os == 'windows-latest'
|
||||||
|
shell: pwsh
|
||||||
|
run: |
|
||||||
|
$env:SOCKTOP_ENABLE_SSL = "0"
|
||||||
|
$env:SOCKTOP_AGENT_GPU = "0"
|
||||||
|
$env:SOCKTOP_AGENT_TEMP = "0"
|
||||||
|
$out = Join-Path $PWD "agent.out.txt"
|
||||||
|
$err = Join-Path $PWD "agent.err.txt"
|
||||||
|
$p = Start-Process -FilePath "${PWD}\target\release\socktop_agent.exe" -ArgumentList "-p 3000" -RedirectStandardOutput $out -RedirectStandardError $err -PassThru -NoNewWindow
|
||||||
|
$ready = $false
|
||||||
|
for ($i = 0; $i -lt 60; $i++) {
|
||||||
|
$pinfo = New-Object System.Diagnostics.ProcessStartInfo
|
||||||
|
$pinfo.FileName = "curl.exe"
|
||||||
|
$pinfo.Arguments = "-fsS http://127.0.0.1:3000/healthz"
|
||||||
|
$pinfo.RedirectStandardOutput = $true
|
||||||
|
$pinfo.RedirectStandardError = $true
|
||||||
|
$pinfo.UseShellExecute = $false
|
||||||
|
$proc = [System.Diagnostics.Process]::Start($pinfo)
|
||||||
|
$proc.WaitForExit()
|
||||||
|
if ($proc.ExitCode -eq 0) { $ready = $true; break }
|
||||||
|
Start-Sleep -Seconds 1
|
||||||
|
}
|
||||||
|
if (-not $ready) {
|
||||||
|
Write-Warning "TCP connect to (127.0.0.1 : 3000) failed"
|
||||||
|
if (Test-Path $out) { Write-Host "--- agent.out (full) ---"; Get-Content $out }
|
||||||
|
if (Test-Path $err) { Write-Host "--- agent.err (full) ---"; Get-Content $err }
|
||||||
|
Write-Host "--- netstat ---"
|
||||||
|
netstat -ano | Select-String ":3000" | ForEach-Object { $_.Line }
|
||||||
|
if ($p -and !$p.HasExited) { Stop-Process -Id $p.Id -Force -ErrorAction SilentlyContinue }
|
||||||
|
throw "agent did not become ready"
|
||||||
|
}
|
||||||
|
$env:SOCKTOP_WS = "ws://127.0.0.1:3000/ws"
|
||||||
|
try {
|
||||||
|
cargo test -p socktop --test ws_probe -- --nocapture
|
||||||
|
} finally {
|
||||||
|
if ($p -and !$p.HasExited) { Stop-Process -Id $p.Id -Force -ErrorAction SilentlyContinue }
|
||||||
|
}
|
||||||
|
|
||||||
- name: Smoke test (client --help)
|
- name: Smoke test (client --help)
|
||||||
run: cargo run -p socktop -- --help
|
run: cargo run -p socktop -- --help
|
||||||
- name: Package artifacts
|
|
||||||
|
- name: Package artifacts (Linux)
|
||||||
|
if: matrix.os == 'ubuntu-latest'
|
||||||
shell: bash
|
shell: bash
|
||||||
run: |
|
run: |
|
||||||
set -e
|
set -e
|
||||||
mkdir dist
|
mkdir -p dist
|
||||||
if [[ "${{ matrix.os }}" == "windows-latest" ]]; then
|
cp target/release/socktop dist/
|
||||||
cp target/release/socktop.exe dist/
|
cp target/release/socktop_agent dist/
|
||||||
cp target/release/socktop_agent.exe dist/
|
tar czf socktop-${{ matrix.os }}.tar.gz -C dist .
|
||||||
7z a socktop-${{ matrix.os }}.zip dist/*
|
|
||||||
else
|
- name: Package artifacts (Windows)
|
||||||
cp target/release/socktop dist/
|
if: matrix.os == 'windows-latest'
|
||||||
cp target/release/socktop_agent dist/
|
shell: pwsh
|
||||||
tar czf socktop-${{ matrix.os }}.tar.gz -C dist .
|
run: |
|
||||||
fi
|
New-Item -ItemType Directory -Force -Path dist | Out-Null
|
||||||
|
Copy-Item target\release\socktop.exe dist\
|
||||||
|
Copy-Item target\release\socktop_agent.exe dist\
|
||||||
|
Compress-Archive -Path dist\* -DestinationPath socktop-${{ matrix.os }}.zip -Force
|
||||||
|
|
||||||
- name: Upload build artifacts (ephemeral)
|
- name: Upload build artifacts (ephemeral)
|
||||||
uses: actions/upload-artifact@v4
|
uses: actions/upload-artifact@v4
|
||||||
with:
|
with:
|
||||||
@@ -46,6 +113,7 @@ jobs:
|
|||||||
path: |
|
path: |
|
||||||
*.tar.gz
|
*.tar.gz
|
||||||
*.zip
|
*.zip
|
||||||
|
|
||||||
- name: Upload to rolling GitHub Release (main only)
|
- name: Upload to rolling GitHub Release (main only)
|
||||||
if: github.ref == 'refs/heads/main' && github.event_name == 'push'
|
if: github.ref == 'refs/heads/main' && github.event_name == 'push'
|
||||||
uses: softprops/action-gh-release@v2
|
uses: softprops/action-gh-release@v2
|
||||||
|
|||||||
Generated
+824
-8
File diff suppressed because it is too large
Load Diff
+14
-1
@@ -13,7 +13,7 @@ futures-util = "0.3"
|
|||||||
anyhow = "1.0"
|
anyhow = "1.0"
|
||||||
|
|
||||||
# websocket
|
# websocket
|
||||||
tokio-tungstenite = "0.24"
|
tokio-tungstenite = { version = "0.24", features = ["__rustls-tls", "connect"] }
|
||||||
tungstenite = "0.24"
|
tungstenite = "0.24"
|
||||||
url = "2.5"
|
url = "2.5"
|
||||||
|
|
||||||
@@ -34,3 +34,16 @@ chrono = { version = "0.4", features = ["serde"] }
|
|||||||
|
|
||||||
# web server (remote-agent)
|
# web server (remote-agent)
|
||||||
axum = { version = "0.7", features = ["ws"] }
|
axum = { version = "0.7", features = ["ws"] }
|
||||||
|
|
||||||
|
# protobuf
|
||||||
|
prost = "0.13"
|
||||||
|
prost-types = "0.13"
|
||||||
|
bytes = "1"
|
||||||
|
|
||||||
|
[profile.release]
|
||||||
|
# Favor smaller, simpler binaries with good runtime perf
|
||||||
|
lto = "thin"
|
||||||
|
codegen-units = 1
|
||||||
|
panic = "abort"
|
||||||
|
opt-level = 3
|
||||||
|
strip = "symbols"
|
||||||
@@ -12,6 +12,7 @@ socktop is a remote system monitor with a rich TUI, inspired by top/btop, talkin
|
|||||||
## Features
|
## Features
|
||||||
|
|
||||||
- Remote monitoring via WebSocket (JSON over WS)
|
- Remote monitoring via WebSocket (JSON over WS)
|
||||||
|
- Optional WSS (TLS): agent auto‑generates a self‑signed cert on first run; client pins the cert via --tls-ca/-t
|
||||||
- TUI built with ratatui
|
- TUI built with ratatui
|
||||||
- CPU
|
- CPU
|
||||||
- Overall sparkline + per-core mini bars
|
- Overall sparkline + per-core mini bars
|
||||||
@@ -50,7 +51,7 @@ exec bash # or: exec zsh / exec fish
|
|||||||
|
|
||||||
Windows (for the brave): install from https://rustup.rs with the MSVC toolchain. Yes, you’ll need Visual Studio Build Tools. You chose Windows — enjoy the ride.
|
Windows (for the brave): install from https://rustup.rs with the MSVC toolchain. Yes, you’ll need Visual Studio Build Tools. You chose Windows — enjoy the ride.
|
||||||
|
|
||||||
### Raspberry Pi (required)
|
### Raspberry Pi / Ubuntu / PopOS (required)
|
||||||
|
|
||||||
Install GPU support with apt command below
|
Install GPU support with apt command below
|
||||||
|
|
||||||
@@ -67,7 +68,7 @@ Two components:
|
|||||||
|
|
||||||
1) Agent (remote): small Rust WS server using sysinfo + /proc. It collects on demand when the client asks (fast metrics ~500 ms, processes ~2 s, disks ~5 s). No background loop when nobody is connected.
|
1) Agent (remote): small Rust WS server using sysinfo + /proc. It collects on demand when the client asks (fast metrics ~500 ms, processes ~2 s, disks ~5 s). No background loop when nobody is connected.
|
||||||
|
|
||||||
2) Client (local): TUI that connects to ws://HOST:PORT/ws and renders updates.
|
2) Client (local): TUI that connects to ws://HOST:PORT/ws (or wss://HOST:PORT/ws when TLS is enabled) and renders updates.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
@@ -95,6 +96,30 @@ cargo build --release
|
|||||||
|
|
||||||
Tip: Add ?token=... if you enable auth (see Security).
|
Tip: Add ?token=... if you enable auth (see Security).
|
||||||
|
|
||||||
|
TLS quick start (optional, recommended on untrusted networks):
|
||||||
|
|
||||||
|
- Start the agent with TLS enabled (default TLS port 8443). On first run it will generate a self‑signed certificate and key under your config directory.
|
||||||
|
|
||||||
|
```bash
|
||||||
|
./target/release/socktop_agent --enableSSL --port 8443 # or: -p 8443
|
||||||
|
# First run prints the cert and key paths, e.g.:
|
||||||
|
# socktop_agent: generated self-signed TLS certificate at /home/you/.config/socktop_agent/tls/cert.pem
|
||||||
|
# socktop_agent: private key at /home/you/.config/socktop_agent/tls/key.pem
|
||||||
|
```
|
||||||
|
|
||||||
|
- Copy the certificate file to the client machine (keep the key private on the server):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
scp /home/you/.config/socktop_agent/tls/cert.pem you@client:/tmp/socktop-agent-ca.pem
|
||||||
|
```
|
||||||
|
|
||||||
|
- Connect with the TUI, pinning the server cert:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
./target/release/socktop --tls-ca /tmp/socktop-agent-ca.pem wss://REMOTE_HOST:8443/ws
|
||||||
|
# Note: if you pass --tls-ca but use ws://, the client auto-upgrades to wss://
|
||||||
|
```
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## Install (from crates.io)
|
## Install (from crates.io)
|
||||||
@@ -135,6 +160,8 @@ Agent (server):
|
|||||||
socktop_agent --port 3000
|
socktop_agent --port 3000
|
||||||
# or env: SOCKTOP_PORT=3000 socktop_agent
|
# or env: SOCKTOP_PORT=3000 socktop_agent
|
||||||
# optional auth: SOCKTOP_TOKEN=changeme socktop_agent
|
# optional auth: SOCKTOP_TOKEN=changeme socktop_agent
|
||||||
|
# enable TLS (self‑signed cert, default port 8443; you can also use -p):
|
||||||
|
socktop_agent --enableSSL --port 8443
|
||||||
```
|
```
|
||||||
|
|
||||||
Client (TUI):
|
Client (TUI):
|
||||||
@@ -143,6 +170,11 @@ Client (TUI):
|
|||||||
socktop ws://HOST:3000/ws
|
socktop ws://HOST:3000/ws
|
||||||
# with token:
|
# with token:
|
||||||
socktop "ws://HOST:3000/ws?token=changeme"
|
socktop "ws://HOST:3000/ws?token=changeme"
|
||||||
|
# TLS with pinned server certificate (recommended over the internet):
|
||||||
|
socktop --tls-ca /path/to/cert.pem wss://HOST:8443/ws
|
||||||
|
# shorthand:
|
||||||
|
socktop -t /path/to/cert.pem wss://HOST:8443/ws
|
||||||
|
# Note: providing --tls-ca/-t automatically upgrades ws:// to wss:// if you forget
|
||||||
```
|
```
|
||||||
|
|
||||||
Intervals (client-driven):
|
Intervals (client-driven):
|
||||||
@@ -188,6 +220,13 @@ Tip: If only the binary changed, restart is enough. If the unit file changed, ru
|
|||||||
- Flag: --port 8080 or -p 8080
|
- Flag: --port 8080 or -p 8080
|
||||||
- Positional: socktop_agent 8080
|
- Positional: socktop_agent 8080
|
||||||
- Env: SOCKTOP_PORT=8080
|
- Env: SOCKTOP_PORT=8080
|
||||||
|
- TLS (self‑signed):
|
||||||
|
- Enable: --enableSSL
|
||||||
|
- Default TLS port: 8443 (override with --port/-p)
|
||||||
|
- Certificate/Key location (created on first TLS run):
|
||||||
|
- Linux (XDG): $XDG_CONFIG_HOME/socktop_agent/tls/{cert.pem,key.pem} (defaults to ~/.config)
|
||||||
|
- The agent prints these paths on creation.
|
||||||
|
- You can set XDG_CONFIG_HOME before first run to control where certs are written.
|
||||||
- Auth token (optional): SOCKTOP_TOKEN=changeme
|
- Auth token (optional): SOCKTOP_TOKEN=changeme
|
||||||
- Disable GPU metrics: SOCKTOP_AGENT_GPU=0
|
- Disable GPU metrics: SOCKTOP_AGENT_GPU=0
|
||||||
- Disable CPU temperature: SOCKTOP_AGENT_TEMP=0
|
- Disable CPU temperature: SOCKTOP_AGENT_TEMP=0
|
||||||
@@ -250,6 +289,27 @@ Client:
|
|||||||
socktop "ws://HOST:3000/ws?token=changeme"
|
socktop "ws://HOST:3000/ws?token=changeme"
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### TLS / WSS
|
||||||
|
|
||||||
|
For encrypted connections, enable TLS on the agent and pin the server certificate on the client.
|
||||||
|
|
||||||
|
Server (generates self‑signed cert and key on first run):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
socktop_agent --enableSSL --port 8443
|
||||||
|
```
|
||||||
|
|
||||||
|
Client (trust/pin the server cert; copy cert.pem from the agent):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
socktop --tls-ca /path/to/agent/cert.pem wss://HOST:8443/ws
|
||||||
|
```
|
||||||
|
|
||||||
|
Notes:
|
||||||
|
- Do not copy the private key off the server; only the cert.pem is needed by clients.
|
||||||
|
- When --tls-ca/-t is supplied, the client auto‑upgrades ws:// to wss:// to avoid protocol mismatch.
|
||||||
|
- You can run multiple clients with different cert paths by passing --tls-ca per invocation.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## Using tmux to monitor multiple hosts
|
## Using tmux to monitor multiple hosts
|
||||||
@@ -319,7 +379,8 @@ Tips:
|
|||||||
cargo fmt
|
cargo fmt
|
||||||
cargo clippy --all-targets --all-features
|
cargo clippy --all-targets --all-features
|
||||||
cargo run -p socktop -- ws://127.0.0.1:3000/ws
|
cargo run -p socktop -- ws://127.0.0.1:3000/ws
|
||||||
cargo run -p socktop_agent -- --port 3000
|
# TLS (dev): first run will create certs under ~/.config/socktop_agent/tls/
|
||||||
|
cargo run -p socktop_agent -- --enableSSL --port 8443
|
||||||
```
|
```
|
||||||
|
|
||||||
---
|
---
|
||||||
@@ -331,7 +392,7 @@ cargo run -p socktop_agent -- --port 3000
|
|||||||
- [x] Sort top processes in the TUI
|
- [x] Sort top processes in the TUI
|
||||||
- [ ] Configurable refresh intervals (client)
|
- [ ] Configurable refresh intervals (client)
|
||||||
- [ ] Export metrics to file
|
- [ ] Export metrics to file
|
||||||
- [ ] TLS / WSS support
|
- [x] TLS / WSS support (self‑signed server cert + client pinning)
|
||||||
- [x] Split processes/disks to separate WS calls with independent cadences (already logical on client; formalize API)
|
- [x] Split processes/disks to separate WS calls with independent cadences (already logical on client; formalize API)
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|||||||
@@ -0,0 +1,8 @@
|
|||||||
|
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8433/ws
|
||||||
|
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8433/ws
|
||||||
|
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8433/ws
|
||||||
|
Error: Address already in use (os error 98)
|
||||||
|
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8433/ws
|
||||||
|
Error: Address already in use (os error 98)
|
||||||
|
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8443/ws
|
||||||
|
socktop_agent: TLS enabled. Listening on wss://0.0.0.0:8443/ws
|
||||||
@@ -0,0 +1,15 @@
|
|||||||
|
syntax = "proto3";
|
||||||
|
package socktop;
|
||||||
|
|
||||||
|
// All running processes. Sorting is done client-side.
|
||||||
|
message Processes {
|
||||||
|
uint64 process_count = 1; // total processes in the system
|
||||||
|
repeated Process rows = 2; // all processes
|
||||||
|
}
|
||||||
|
|
||||||
|
message Process {
|
||||||
|
uint32 pid = 1;
|
||||||
|
string name = 2;
|
||||||
|
float cpu_usage = 3; // 0..100
|
||||||
|
uint64 mem_bytes = 4; // RSS bytes
|
||||||
|
}
|
||||||
@@ -0,0 +1,3 @@
|
|||||||
|
[toolchain]
|
||||||
|
channel = "stable"
|
||||||
|
components = ["clippy", "rustfmt"]
|
||||||
+12
-2
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "socktop"
|
name = "socktop"
|
||||||
version = "0.1.1"
|
version = "0.1.11"
|
||||||
authors = ["Jason Witty <jasonpwitty+socktop@proton.me>"]
|
authors = ["Jason Witty <jasonpwitty+socktop@proton.me>"]
|
||||||
description = "Remote system monitor over WebSocket, TUI like top"
|
description = "Remote system monitor over WebSocket, TUI like top"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
@@ -19,4 +19,14 @@ crossterm = { workspace = true }
|
|||||||
chrono = { workspace = true }
|
chrono = { workspace = true }
|
||||||
anyhow = { workspace = true }
|
anyhow = { workspace = true }
|
||||||
flate2 = { version = "1", default-features = false, features = ["rust_backend"] }
|
flate2 = { version = "1", default-features = false, features = ["rust_backend"] }
|
||||||
tungstenite = "0.27.0"
|
rustls = "0.23"
|
||||||
|
rustls-pemfile = "2.1"
|
||||||
|
prost = { workspace = true }
|
||||||
|
bytes = { workspace = true }
|
||||||
|
|
||||||
|
[dev-dependencies]
|
||||||
|
assert_cmd = "2.0"
|
||||||
|
|
||||||
|
[build-dependencies]
|
||||||
|
prost-build = "0.13"
|
||||||
|
protoc-bin-vendored = "3"
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
fn main() {
|
||||||
|
let protoc = protoc_bin_vendored::protoc_bin_path().expect("protoc");
|
||||||
|
std::env::set_var("PROTOC", protoc);
|
||||||
|
let mut cfg = prost_build::Config::new();
|
||||||
|
cfg.out_dir(std::env::var("OUT_DIR").unwrap());
|
||||||
|
cfg.compile_protos(&["../proto/processes.proto"], &["../proto"])
|
||||||
|
.expect("compile protos");
|
||||||
|
}
|
||||||
+13
-2
@@ -63,6 +63,9 @@ pub struct App {
|
|||||||
last_disks_poll: Instant,
|
last_disks_poll: Instant,
|
||||||
procs_interval: Duration,
|
procs_interval: Duration,
|
||||||
disks_interval: Duration,
|
disks_interval: Duration,
|
||||||
|
|
||||||
|
// For reconnects
|
||||||
|
ws_url: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl App {
|
impl App {
|
||||||
@@ -91,12 +94,19 @@ impl App {
|
|||||||
.unwrap_or_else(Instant::now),
|
.unwrap_or_else(Instant::now),
|
||||||
procs_interval: Duration::from_secs(2),
|
procs_interval: Duration::from_secs(2),
|
||||||
disks_interval: Duration::from_secs(5),
|
disks_interval: Duration::from_secs(5),
|
||||||
|
ws_url: String::new(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn run(&mut self, url: &str) -> Result<(), Box<dyn std::error::Error>> {
|
pub async fn run(
|
||||||
|
&mut self,
|
||||||
|
url: &str,
|
||||||
|
tls_ca: Option<&str>,
|
||||||
|
) -> Result<(), Box<dyn std::error::Error>> {
|
||||||
// Connect to agent
|
// Connect to agent
|
||||||
let mut ws = connect(url).await?;
|
//let mut ws = connect(url, tls_ca).await?;
|
||||||
|
self.ws_url = url.to_string();
|
||||||
|
let mut ws = connect(url, tls_ca).await?;
|
||||||
|
|
||||||
// Terminal setup
|
// Terminal setup
|
||||||
enable_raw_mode()?;
|
enable_raw_mode()?;
|
||||||
@@ -461,6 +471,7 @@ impl Default for App {
|
|||||||
.unwrap_or_else(Instant::now),
|
.unwrap_or_else(Instant::now),
|
||||||
procs_interval: Duration::from_secs(2),
|
procs_interval: Duration::from_secs(2),
|
||||||
disks_interval: Duration::from_secs(5),
|
disks_interval: Duration::from_secs(5),
|
||||||
|
ws_url: String::new(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,4 @@
|
|||||||
|
//! Library surface for integration tests and reuse.
|
||||||
|
|
||||||
|
pub mod types;
|
||||||
|
pub mod ws;
|
||||||
+49
-11
@@ -9,22 +9,60 @@ mod ws;
|
|||||||
use app::App;
|
use app::App;
|
||||||
use std::env;
|
use std::env;
|
||||||
|
|
||||||
|
fn parse_args<I: IntoIterator<Item = String>>(args: I) -> Result<(String, Option<String>), String> {
|
||||||
|
let mut it = args.into_iter();
|
||||||
|
let prog = it.next().unwrap_or_else(|| "socktop".into());
|
||||||
|
let mut url: Option<String> = None;
|
||||||
|
let mut tls_ca: Option<String> = None;
|
||||||
|
|
||||||
|
while let Some(arg) = it.next() {
|
||||||
|
match arg.as_str() {
|
||||||
|
"-h" | "--help" => {
|
||||||
|
return Err(format!(
|
||||||
|
"Usage: {prog} [--tls-ca CERT_PEM|-t CERT_PEM] ws://HOST:PORT/ws"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
"--tls-ca" | "-t" => {
|
||||||
|
tls_ca = it.next();
|
||||||
|
}
|
||||||
|
_ if arg.starts_with("--tls-ca=") => {
|
||||||
|
if let Some((_, v)) = arg.split_once('=') {
|
||||||
|
if !v.is_empty() {
|
||||||
|
tls_ca = Some(v.to_string());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_ => {
|
||||||
|
if url.is_none() {
|
||||||
|
url = Some(arg);
|
||||||
|
} else {
|
||||||
|
return Err(format!(
|
||||||
|
"Unexpected argument. Usage: {prog} [--tls-ca CERT_PEM|-t CERT_PEM] ws://HOST:PORT/ws"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
match url {
|
||||||
|
Some(u) => Ok((u, tls_ca)),
|
||||||
|
None => Err(format!(
|
||||||
|
"Usage: {prog} [--tls-ca CERT_PEM|-t CERT_PEM] ws://HOST:PORT/ws"
|
||||||
|
)),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::main]
|
#[tokio::main]
|
||||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||||
let mut args = env::args();
|
// Reuse the same parsing logic for testability
|
||||||
let prog = args.next().unwrap_or_else(|| "socktop".into());
|
let (url, tls_ca) = match parse_args(env::args()) {
|
||||||
let url = match args.next() {
|
Ok(v) => v,
|
||||||
Some(flag) if flag == "-h" || flag == "--help" => {
|
Err(msg) => {
|
||||||
println!("Usage: {prog} ws://HOST:PORT/ws");
|
eprintln!("{msg}");
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
Some(url) => url,
|
|
||||||
None => {
|
|
||||||
eprintln!("Usage: {prog} ws://HOST:PORT/ws");
|
|
||||||
std::process::exit(1);
|
|
||||||
}
|
|
||||||
};
|
};
|
||||||
|
|
||||||
let mut app = App::new();
|
let mut app = App::new();
|
||||||
app.run(&url).await
|
app.run(&url, tls_ca.as_deref()).await
|
||||||
}
|
}
|
||||||
|
|||||||
+90
-83
@@ -2,18 +2,62 @@
|
|||||||
|
|
||||||
use flate2::bufread::GzDecoder;
|
use flate2::bufread::GzDecoder;
|
||||||
use futures_util::{SinkExt, StreamExt};
|
use futures_util::{SinkExt, StreamExt};
|
||||||
|
use prost::Message as _;
|
||||||
|
use rustls::{ClientConfig, RootCertStore};
|
||||||
|
use rustls_pemfile::Item;
|
||||||
use std::io::Read;
|
use std::io::Read;
|
||||||
|
use std::{fs::File, io::BufReader, sync::Arc};
|
||||||
use tokio::net::TcpStream;
|
use tokio::net::TcpStream;
|
||||||
use tokio::time::{interval, Duration};
|
use tokio_tungstenite::{
|
||||||
use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
|
connect_async, connect_async_tls_with_config, tungstenite::client::IntoClientRequest,
|
||||||
|
tungstenite::Message, Connector, MaybeTlsStream, WebSocketStream,
|
||||||
|
};
|
||||||
|
use url::Url;
|
||||||
|
|
||||||
use crate::types::{DiskInfo, Metrics, ProcessesPayload};
|
use crate::types::{DiskInfo, Metrics, ProcessInfo, ProcessesPayload};
|
||||||
|
|
||||||
|
mod pb {
|
||||||
|
// generated by build.rs
|
||||||
|
include!(concat!(env!("OUT_DIR"), "/socktop.rs"));
|
||||||
|
}
|
||||||
|
|
||||||
pub type WsStream = WebSocketStream<MaybeTlsStream<TcpStream>>;
|
pub type WsStream = WebSocketStream<MaybeTlsStream<TcpStream>>;
|
||||||
|
|
||||||
// Connect to the agent and return the WS stream
|
// Connect to the agent and return the WS stream
|
||||||
pub async fn connect(url: &str) -> Result<WsStream, Box<dyn std::error::Error>> {
|
pub async fn connect(
|
||||||
let (ws, _) = connect_async(url).await?;
|
url: &str,
|
||||||
|
tls_ca: Option<&str>,
|
||||||
|
) -> Result<WsStream, Box<dyn std::error::Error>> {
|
||||||
|
let mut u = Url::parse(url)?;
|
||||||
|
if let Some(ca_path) = tls_ca {
|
||||||
|
if u.scheme() == "ws" {
|
||||||
|
let _ = u.set_scheme("wss");
|
||||||
|
}
|
||||||
|
return connect_with_ca(u.as_str(), ca_path).await;
|
||||||
|
}
|
||||||
|
let (ws, _) = connect_async(u.as_str()).await?;
|
||||||
|
Ok(ws)
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn connect_with_ca(url: &str, ca_path: &str) -> Result<WsStream, Box<dyn std::error::Error>> {
|
||||||
|
let mut root = RootCertStore::empty();
|
||||||
|
let mut reader = BufReader::new(File::open(ca_path)?);
|
||||||
|
let mut der_certs = Vec::new();
|
||||||
|
while let Ok(Some(item)) = rustls_pemfile::read_one(&mut reader) {
|
||||||
|
if let Item::X509Certificate(der) = item {
|
||||||
|
der_certs.push(der);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
root.add_parsable_certificates(der_certs);
|
||||||
|
|
||||||
|
let cfg = ClientConfig::builder()
|
||||||
|
.with_root_certificates(root)
|
||||||
|
.with_no_client_auth();
|
||||||
|
let cfg = Arc::new(cfg);
|
||||||
|
|
||||||
|
let req = url.into_client_request()?;
|
||||||
|
let (ws, _) =
|
||||||
|
connect_async_tls_with_config(req, None, true, Some(Connector::Rustls(cfg))).await?;
|
||||||
Ok(ws)
|
Ok(ws)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -39,6 +83,16 @@ fn gunzip_to_string(bytes: &[u8]) -> Option<String> {
|
|||||||
Some(out)
|
Some(out)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn gunzip_to_vec(bytes: &[u8]) -> Option<Vec<u8>> {
|
||||||
|
let mut dec = GzDecoder::new(bytes);
|
||||||
|
let mut out = Vec::new();
|
||||||
|
dec.read_to_end(&mut out).ok()?;
|
||||||
|
Some(out)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn is_gzip(bytes: &[u8]) -> bool {
|
||||||
|
bytes.len() >= 2 && bytes[0] == 0x1f && bytes[1] == 0x8b
|
||||||
|
}
|
||||||
// Suppress dead_code until these are wired into the app
|
// Suppress dead_code until these are wired into the app
|
||||||
#[allow(dead_code)]
|
#[allow(dead_code)]
|
||||||
pub enum Payload {
|
pub enum Payload {
|
||||||
@@ -47,23 +101,6 @@ pub enum Payload {
|
|||||||
Processes(ProcessesPayload),
|
Processes(ProcessesPayload),
|
||||||
}
|
}
|
||||||
|
|
||||||
#[allow(dead_code)]
|
|
||||||
fn parse_any_payload(json: &str) -> Result<Payload, serde_json::Error> {
|
|
||||||
if let Ok(m) = serde_json::from_str::<Metrics>(json) {
|
|
||||||
return Ok(Payload::Metrics(m));
|
|
||||||
}
|
|
||||||
if let Ok(d) = serde_json::from_str::<Vec<DiskInfo>>(json) {
|
|
||||||
return Ok(Payload::Disks(d));
|
|
||||||
}
|
|
||||||
if let Ok(p) = serde_json::from_str::<ProcessesPayload>(json) {
|
|
||||||
return Ok(Payload::Processes(p));
|
|
||||||
}
|
|
||||||
Err(serde_json::Error::io(std::io::Error::new(
|
|
||||||
std::io::ErrorKind::InvalidData,
|
|
||||||
"unknown payload",
|
|
||||||
)))
|
|
||||||
}
|
|
||||||
|
|
||||||
// Send a "get_disks" request and await a JSON Vec<DiskInfo>
|
// Send a "get_disks" request and await a JSON Vec<DiskInfo>
|
||||||
pub async fn request_disks(ws: &mut WsStream) -> Option<Vec<DiskInfo>> {
|
pub async fn request_disks(ws: &mut WsStream) -> Option<Vec<DiskInfo>> {
|
||||||
if ws.send(Message::Text("get_disks".into())).await.is_err() {
|
if ws.send(Message::Text("get_disks".into())).await.is_err() {
|
||||||
@@ -78,7 +115,7 @@ pub async fn request_disks(ws: &mut WsStream) -> Option<Vec<DiskInfo>> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Send a "get_processes" request and await a JSON ProcessesPayload
|
// Send a "get_processes" request and await a ProcessesPayload decoded from protobuf (binary, may be gzipped)
|
||||||
pub async fn request_processes(ws: &mut WsStream) -> Option<ProcessesPayload> {
|
pub async fn request_processes(ws: &mut WsStream) -> Option<ProcessesPayload> {
|
||||||
if ws
|
if ws
|
||||||
.send(Message::Text("get_processes".into()))
|
.send(Message::Text("get_processes".into()))
|
||||||
@@ -89,68 +126,38 @@ pub async fn request_processes(ws: &mut WsStream) -> Option<ProcessesPayload> {
|
|||||||
}
|
}
|
||||||
match ws.next().await {
|
match ws.next().await {
|
||||||
Some(Ok(Message::Binary(b))) => {
|
Some(Ok(Message::Binary(b))) => {
|
||||||
gunzip_to_string(&b).and_then(|s| serde_json::from_str::<ProcessesPayload>(&s).ok())
|
let gz = is_gzip(&b);
|
||||||
|
let data = if gz { gunzip_to_vec(&b)? } else { b };
|
||||||
|
match pb::Processes::decode(data.as_slice()) {
|
||||||
|
Ok(pb) => {
|
||||||
|
let rows: Vec<ProcessInfo> = pb
|
||||||
|
.rows
|
||||||
|
.into_iter()
|
||||||
|
.map(|p: pb::Process| ProcessInfo {
|
||||||
|
pid: p.pid,
|
||||||
|
name: p.name,
|
||||||
|
cpu_usage: p.cpu_usage,
|
||||||
|
mem_bytes: p.mem_bytes,
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
Some(ProcessesPayload {
|
||||||
|
process_count: pb.process_count as usize,
|
||||||
|
top_processes: rows,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
if std::env::var("SOCKTOP_DEBUG").ok().as_deref() == Some("1") {
|
||||||
|
eprintln!("protobuf decode failed: {e}");
|
||||||
|
}
|
||||||
|
// Fallback: maybe it's JSON (bytes already decompressed if gz)
|
||||||
|
match String::from_utf8(data) {
|
||||||
|
Ok(s) => serde_json::from_str::<ProcessesPayload>(&s).ok(),
|
||||||
|
Err(_) => None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
Some(Ok(Message::Text(json))) => serde_json::from_str::<ProcessesPayload>(&json).ok(),
|
Some(Ok(Message::Text(json))) => serde_json::from_str::<ProcessesPayload>(&json).ok(),
|
||||||
_ => None,
|
_ => None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[allow(dead_code)]
|
|
||||||
pub async fn start_ws_polling(mut ws: WsStream) {
|
|
||||||
let mut t_fast = interval(Duration::from_millis(500));
|
|
||||||
let mut t_procs = interval(Duration::from_secs(2));
|
|
||||||
let mut t_disks = interval(Duration::from_secs(5));
|
|
||||||
|
|
||||||
let _ = ws.send(Message::Text("get_metrics".into())).await;
|
|
||||||
let _ = ws.send(Message::Text("get_processes".into())).await;
|
|
||||||
let _ = ws.send(Message::Text("get_disks".into())).await;
|
|
||||||
|
|
||||||
loop {
|
|
||||||
tokio::select! {
|
|
||||||
_ = t_fast.tick() => {
|
|
||||||
let _ = ws.send(Message::Text("get_metrics".into())).await;
|
|
||||||
}
|
|
||||||
_ = t_procs.tick() => {
|
|
||||||
let _ = ws.send(Message::Text("get_processes".into())).await;
|
|
||||||
}
|
|
||||||
_ = t_disks.tick() => {
|
|
||||||
let _ = ws.send(Message::Text("get_disks".into())).await;
|
|
||||||
}
|
|
||||||
maybe = ws.next() => {
|
|
||||||
let Some(result) = maybe else { break; };
|
|
||||||
let Ok(msg) = result else { break; };
|
|
||||||
match msg {
|
|
||||||
Message::Binary(b) => {
|
|
||||||
if let Some(json) = gunzip_to_string(&b) {
|
|
||||||
if let Ok(payload) = parse_any_payload(&json) {
|
|
||||||
match payload {
|
|
||||||
Payload::Metrics(_m) => {
|
|
||||||
// update your app state with fast metrics
|
|
||||||
}
|
|
||||||
Payload::Disks(_d) => {
|
|
||||||
// update your app state with disks
|
|
||||||
}
|
|
||||||
Payload::Processes(_p) => {
|
|
||||||
// update your app state with processes
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Message::Text(s) => {
|
|
||||||
if let Ok(payload) = parse_any_payload(&s) {
|
|
||||||
match payload {
|
|
||||||
Payload::Metrics(_m) => {}
|
|
||||||
Payload::Disks(_d) => {}
|
|
||||||
Payload::Processes(_p) => {}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Message::Close(_) => break,
|
|
||||||
_ => {}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -0,0 +1,56 @@
|
|||||||
|
//! CLI arg parsing tests for socktop (client)
|
||||||
|
use std::process::Command;
|
||||||
|
|
||||||
|
// We test the parsing by invoking the binary with --help and ensuring the help mentions short and long flags.
|
||||||
|
// Also directly test the parse_args function via a tiny helper in a doctest-like fashion using a small
|
||||||
|
// reimplementation here kept in sync with main (compile-time test).
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_help_mentions_short_and_long_flags() {
|
||||||
|
let output = Command::new(env!("CARGO_BIN_EXE_socktop"))
|
||||||
|
.arg("--help")
|
||||||
|
.output()
|
||||||
|
.expect("run socktop --help");
|
||||||
|
let text = format!(
|
||||||
|
"{}{}",
|
||||||
|
String::from_utf8_lossy(&output.stdout),
|
||||||
|
String::from_utf8_lossy(&output.stderr)
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
text.contains("--tls-ca") && text.contains("-t"),
|
||||||
|
"help text missing --tls-ca/-t\n{text}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_tlc_ca_arg_long_and_short_parsed() {
|
||||||
|
// Use --help combined with flags to avoid network and still exercise arg acceptance
|
||||||
|
let exe = env!("CARGO_BIN_EXE_socktop");
|
||||||
|
// Long form with help
|
||||||
|
let out = Command::new(exe)
|
||||||
|
.args(["--tls-ca", "/tmp/cert.pem", "--help"])
|
||||||
|
.output()
|
||||||
|
.expect("run socktop");
|
||||||
|
assert!(
|
||||||
|
out.status.success(),
|
||||||
|
"socktop --tls-ca … --help did not succeed"
|
||||||
|
);
|
||||||
|
let text = format!(
|
||||||
|
"{}{}",
|
||||||
|
String::from_utf8_lossy(&out.stdout),
|
||||||
|
String::from_utf8_lossy(&out.stderr)
|
||||||
|
);
|
||||||
|
assert!(text.contains("Usage:"));
|
||||||
|
// Short form with help
|
||||||
|
let out2 = Command::new(exe)
|
||||||
|
.args(["-t", "/tmp/cert.pem", "--help"])
|
||||||
|
.output()
|
||||||
|
.expect("run socktop");
|
||||||
|
assert!(out2.status.success(), "socktop -t … --help did not succeed");
|
||||||
|
let text2 = format!(
|
||||||
|
"{}{}",
|
||||||
|
String::from_utf8_lossy(&out2.stdout),
|
||||||
|
String::from_utf8_lossy(&out2.stderr)
|
||||||
|
);
|
||||||
|
assert!(text2.contains("Usage:"));
|
||||||
|
}
|
||||||
@@ -0,0 +1,29 @@
|
|||||||
|
use socktop::ws::{connect, request_metrics, request_processes};
|
||||||
|
|
||||||
|
// Integration probe: only runs when SOCKTOP_WS is set to an agent WebSocket URL.
|
||||||
|
// Example: SOCKTOP_WS=ws://127.0.0.1:3000/ws cargo test -p socktop --test ws_probe -- --nocapture
|
||||||
|
#[tokio::test]
|
||||||
|
async fn probe_ws_endpoints() {
|
||||||
|
// Gate the test to avoid CI failures when no agent is running.
|
||||||
|
let url = match std::env::var("SOCKTOP_WS") {
|
||||||
|
Ok(v) if !v.is_empty() => v,
|
||||||
|
_ => {
|
||||||
|
eprintln!(
|
||||||
|
"skipping ws_probe: set SOCKTOP_WS=ws://host:port/ws to run this integration test"
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// Optional pinned CA for WSS/self-signed setups
|
||||||
|
let tls_ca = std::env::var("SOCKTOP_TLS_CA").ok();
|
||||||
|
let mut ws = connect(&url, tls_ca.as_deref()).await.expect("connect ws");
|
||||||
|
|
||||||
|
// Should get fast metrics quickly
|
||||||
|
let m = request_metrics(&mut ws).await;
|
||||||
|
assert!(m.is_some(), "expected Metrics payload within timeout");
|
||||||
|
|
||||||
|
// Processes may be gzipped and a bit slower, but should arrive
|
||||||
|
let p = request_processes(&mut ws).await;
|
||||||
|
assert!(p.is_some(), "expected Processes payload within timeout");
|
||||||
|
}
|
||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "socktop_agent"
|
name = "socktop_agent"
|
||||||
version = "0.1.1"
|
version = "0.1.11"
|
||||||
authors = ["Jason Witty <jasonpwitty+socktop@proton.me>"]
|
authors = ["Jason Witty <jasonpwitty+socktop@proton.me>"]
|
||||||
description = "Remote system monitor over WebSocket, TUI like top"
|
description = "Remote system monitor over WebSocket, TUI like top"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
@@ -20,4 +20,21 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
|
|||||||
nvml-wrapper = "0.10"
|
nvml-wrapper = "0.10"
|
||||||
gfxinfo = "0.1.2"
|
gfxinfo = "0.1.2"
|
||||||
tungstenite = "0.27.0"
|
tungstenite = "0.27.0"
|
||||||
once_cell = "1.19"
|
once_cell = "1.19"
|
||||||
|
axum-server = { version = "0.6", features = ["tls-rustls"] }
|
||||||
|
rustls = "0.23"
|
||||||
|
rustls-pemfile = "2.1"
|
||||||
|
openssl = { version = "0.10", features = ["vendored"] } # for cross‑platform self‑signed generation
|
||||||
|
anyhow = "1"
|
||||||
|
hostname = "0.3"
|
||||||
|
bytes = { workspace = true }
|
||||||
|
prost = { workspace = true }
|
||||||
|
|
||||||
|
[build-dependencies]
|
||||||
|
prost-build = "0.13"
|
||||||
|
prost-types = { workspace = true }
|
||||||
|
tonic-build = { version = "0.12", default-features = false, optional = true }
|
||||||
|
protoc-bin-vendored = "3"
|
||||||
|
[dev-dependencies]
|
||||||
|
assert_cmd = "2.0"
|
||||||
|
tempfile = "3.10"
|
||||||
@@ -0,0 +1,11 @@
|
|||||||
|
fn main() {
|
||||||
|
// Ensure protoc exists (vendored for reproducible builds)
|
||||||
|
let protoc = protoc_bin_vendored::protoc_bin_path().expect("protoc");
|
||||||
|
std::env::set_var("PROTOC", protoc);
|
||||||
|
|
||||||
|
// Compile protobuf definitions for processes
|
||||||
|
let mut cfg = prost_build::Config::new();
|
||||||
|
cfg.out_dir(std::env::var("OUT_DIR").unwrap());
|
||||||
|
cfg.compile_protos(&["../proto/processes.proto"], &["../proto"])
|
||||||
|
.expect("compile protos");
|
||||||
|
}
|
||||||
+96
-59
@@ -3,20 +3,38 @@
|
|||||||
|
|
||||||
mod gpu;
|
mod gpu;
|
||||||
mod metrics;
|
mod metrics;
|
||||||
|
mod proto;
|
||||||
mod sampler;
|
mod sampler;
|
||||||
mod state;
|
mod state;
|
||||||
mod types;
|
mod types;
|
||||||
mod ws;
|
mod ws;
|
||||||
|
|
||||||
use axum::{routing::get, Router};
|
use axum::{http::StatusCode, routing::get, Router};
|
||||||
use std::net::SocketAddr;
|
use std::net::SocketAddr;
|
||||||
|
use std::str::FromStr;
|
||||||
|
|
||||||
|
mod tls;
|
||||||
|
|
||||||
use crate::sampler::{spawn_disks_sampler, spawn_process_sampler, spawn_sampler};
|
use crate::sampler::{spawn_disks_sampler, spawn_process_sampler, spawn_sampler};
|
||||||
use state::AppState;
|
use state::AppState;
|
||||||
use ws::ws_handler;
|
|
||||||
|
fn arg_flag(name: &str) -> bool {
|
||||||
|
std::env::args().any(|a| a == name)
|
||||||
|
}
|
||||||
|
fn arg_value(name: &str) -> Option<String> {
|
||||||
|
let mut it = std::env::args();
|
||||||
|
while let Some(a) = it.next() {
|
||||||
|
if a == name {
|
||||||
|
return it.next();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None
|
||||||
|
}
|
||||||
|
|
||||||
|
// (tests moved to end of file to satisfy clippy::items_after_test_module)
|
||||||
|
|
||||||
#[tokio::main]
|
#[tokio::main]
|
||||||
async fn main() {
|
async fn main() -> anyhow::Result<()> {
|
||||||
tracing_subscriber::fmt::init();
|
tracing_subscriber::fmt::init();
|
||||||
|
|
||||||
let state = AppState::new();
|
let state = AppState::new();
|
||||||
@@ -29,71 +47,90 @@ async fn main() {
|
|||||||
// 5s disks
|
// 5s disks
|
||||||
let _h_disks = spawn_disks_sampler(state.clone(), std::time::Duration::from_secs(5));
|
let _h_disks = spawn_disks_sampler(state.clone(), std::time::Duration::from_secs(5));
|
||||||
|
|
||||||
// Web app
|
// Web app: route /ws to the websocket handler
|
||||||
let port = resolve_port();
|
async fn healthz() -> StatusCode {
|
||||||
|
println!("/healthz request");
|
||||||
|
StatusCode::OK
|
||||||
|
}
|
||||||
let app = Router::new()
|
let app = Router::new()
|
||||||
.route("/ws", get(ws_handler))
|
.route("/ws", get(ws::ws_handler))
|
||||||
.with_state(state);
|
.route("/healthz", get(healthz))
|
||||||
|
.with_state(state.clone());
|
||||||
|
|
||||||
|
let enable_ssl =
|
||||||
|
arg_flag("--enableSSL") || std::env::var("SOCKTOP_ENABLE_SSL").ok().as_deref() == Some("1");
|
||||||
|
if enable_ssl {
|
||||||
|
// Port can be overridden by --port or SOCKTOP_PORT; default to 8443 when SSL
|
||||||
|
let port = arg_value("--port")
|
||||||
|
.or_else(|| arg_value("-p"))
|
||||||
|
.or_else(|| std::env::var("SOCKTOP_PORT").ok())
|
||||||
|
.and_then(|s| s.parse::<u16>().ok())
|
||||||
|
.unwrap_or(8443);
|
||||||
|
|
||||||
|
let (cert_path, key_path) = tls::ensure_self_signed_cert()?;
|
||||||
|
let cfg = axum_server::tls_rustls::RustlsConfig::from_pem_file(cert_path, key_path).await?;
|
||||||
|
|
||||||
|
let addr = SocketAddr::from_str(&format!("0.0.0.0:{port}"))?;
|
||||||
|
println!("socktop_agent: TLS enabled. Listening on wss://{addr}/ws");
|
||||||
|
axum_server::bind_rustls(addr, cfg)
|
||||||
|
.serve(app.into_make_service())
|
||||||
|
.await?;
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
|
||||||
|
// Non-TLS HTTP/WS path
|
||||||
|
let port = arg_value("--port")
|
||||||
|
.or_else(|| arg_value("-p"))
|
||||||
|
.or_else(|| std::env::var("SOCKTOP_PORT").ok())
|
||||||
|
.and_then(|s| s.parse::<u16>().ok())
|
||||||
|
.unwrap_or(3000);
|
||||||
let addr = SocketAddr::from(([0, 0, 0, 0], port));
|
let addr = SocketAddr::from(([0, 0, 0, 0], port));
|
||||||
|
println!("socktop_agent: Listening on ws://{addr}/ws");
|
||||||
//output to console
|
axum_server::bind(addr)
|
||||||
println!("Remote agent running at http://{addr}");
|
.serve(app.into_make_service())
|
||||||
println!("WebSocket endpoint: ws://{addr}/ws");
|
.await?;
|
||||||
|
Ok(())
|
||||||
//trace logging
|
|
||||||
tracing::info!("Remote agent running at http://{} (ws at /ws)", addr);
|
|
||||||
tracing::info!("WebSocket endpoint: ws://{}/ws", addr);
|
|
||||||
|
|
||||||
let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
|
|
||||||
axum::serve(listener, app).await.unwrap();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Resolve the listening port from CLI args/env with a 3000 default.
|
#[cfg(test)]
|
||||||
// Supports: --port <PORT>, -p <PORT>, a bare numeric positional arg, or SOCKTOP_PORT.
|
mod tests_cli_agent {
|
||||||
fn resolve_port() -> u16 {
|
// Local helper for testing port parsing
|
||||||
const DEFAULT: u16 = 3000;
|
fn parse_port<I: IntoIterator<Item = String>>(args: I, default_port: u16) -> u16 {
|
||||||
|
let mut it = args.into_iter();
|
||||||
// Env takes precedence over positional, but is overridden by explicit flags if present.
|
let _ = it.next(); // prog
|
||||||
if let Ok(s) = std::env::var("SOCKTOP_PORT") {
|
let mut long: Option<String> = None;
|
||||||
if let Ok(p) = s.parse::<u16>() {
|
let mut short: Option<String> = None;
|
||||||
if p != 0 {
|
while let Some(a) = it.next() {
|
||||||
return p;
|
match a.as_str() {
|
||||||
}
|
"--port" => long = it.next(),
|
||||||
}
|
"-p" => short = it.next(),
|
||||||
eprintln!("Warning: invalid SOCKTOP_PORT='{s}'; using default {DEFAULT}");
|
_ if a.starts_with("--port=") => {
|
||||||
}
|
if let Some((_, v)) = a.split_once('=') {
|
||||||
|
long = Some(v.to_string());
|
||||||
let mut args = std::env::args().skip(1);
|
|
||||||
while let Some(arg) = args.next() {
|
|
||||||
match arg.as_str() {
|
|
||||||
"--port" | "-p" => {
|
|
||||||
if let Some(v) = args.next() {
|
|
||||||
match v.parse::<u16>() {
|
|
||||||
Ok(p) if p != 0 => return p,
|
|
||||||
_ => {
|
|
||||||
eprintln!("Invalid port '{v}'; using default {DEFAULT}");
|
|
||||||
return DEFAULT;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
eprintln!("Missing value for {arg} ; using default {DEFAULT}");
|
|
||||||
return DEFAULT;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
"--help" | "-h" => {
|
|
||||||
println!("Usage: socktop_agent [--port <PORT>] [PORT]\n SOCKTOP_PORT=<PORT> socktop_agent");
|
|
||||||
std::process::exit(0);
|
|
||||||
}
|
|
||||||
s => {
|
|
||||||
if let Ok(p) = s.parse::<u16>() {
|
|
||||||
if p != 0 {
|
|
||||||
return p;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
_ => {}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
long.or(short)
|
||||||
|
.and_then(|s| s.parse::<u16>().ok())
|
||||||
|
.unwrap_or(default_port)
|
||||||
}
|
}
|
||||||
|
|
||||||
DEFAULT
|
#[test]
|
||||||
|
fn port_long_short_and_assign() {
|
||||||
|
assert_eq!(
|
||||||
|
parse_port(vec!["agent".into(), "--port".into(), "9001".into()], 8443),
|
||||||
|
9001
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
parse_port(vec!["agent".into(), "-p".into(), "9002".into()], 8443),
|
||||||
|
9002
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
parse_port(vec!["agent".into(), "--port=9003".into()], 8443),
|
||||||
|
9003
|
||||||
|
);
|
||||||
|
assert_eq!(parse_port(vec!["agent".into()], 8443), 8443);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,8 +4,11 @@ use crate::gpu::collect_all_gpus;
|
|||||||
use crate::state::AppState;
|
use crate::state::AppState;
|
||||||
use crate::types::{DiskInfo, Metrics, NetworkInfo, ProcessInfo, ProcessesPayload};
|
use crate::types::{DiskInfo, Metrics, NetworkInfo, ProcessInfo, ProcessesPayload};
|
||||||
use once_cell::sync::OnceCell;
|
use once_cell::sync::OnceCell;
|
||||||
|
#[cfg(target_os = "linux")]
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
|
#[cfg(target_os = "linux")]
|
||||||
use std::fs;
|
use std::fs;
|
||||||
|
#[cfg(target_os = "linux")]
|
||||||
use std::io;
|
use std::io;
|
||||||
use std::sync::Mutex;
|
use std::sync::Mutex;
|
||||||
use std::time::{Duration, Instant};
|
use std::time::{Duration, Instant};
|
||||||
@@ -198,6 +201,8 @@ pub async fn collect_disks(state: &AppState) -> Vec<DiskInfo> {
|
|||||||
.collect()
|
.collect()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Linux-only helpers and implementation using /proc deltas for accurate CPU%.
|
||||||
|
#[cfg(target_os = "linux")]
|
||||||
#[inline]
|
#[inline]
|
||||||
fn read_total_jiffies() -> io::Result<u64> {
|
fn read_total_jiffies() -> io::Result<u64> {
|
||||||
// /proc/stat first line: "cpu user nice system idle iowait irq softirq steal ..."
|
// /proc/stat first line: "cpu user nice system idle iowait irq softirq steal ..."
|
||||||
@@ -216,6 +221,7 @@ fn read_total_jiffies() -> io::Result<u64> {
|
|||||||
Err(io::Error::other("no cpu line"))
|
Err(io::Error::other("no cpu line"))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(target_os = "linux")]
|
||||||
#[inline]
|
#[inline]
|
||||||
fn read_proc_jiffies(pid: u32) -> Option<u64> {
|
fn read_proc_jiffies(pid: u32) -> Option<u64> {
|
||||||
let path = format!("/proc/{pid}/stat");
|
let path = format!("/proc/{pid}/stat");
|
||||||
@@ -230,11 +236,10 @@ fn read_proc_jiffies(pid: u32) -> Option<u64> {
|
|||||||
Some(utime.saturating_add(stime))
|
Some(utime.saturating_add(stime))
|
||||||
}
|
}
|
||||||
|
|
||||||
// Replace the body of collect_processes_top_k to use /proc deltas.
|
/// Collect all processes (Linux): compute CPU% via /proc jiffies delta; sorting moved to client.
|
||||||
// This makes CPU% = (delta_proc / delta_total) * 100 over the 2s interval.
|
#[cfg(target_os = "linux")]
|
||||||
pub async fn collect_processes_top_k(state: &AppState, k: usize) -> ProcessesPayload {
|
pub async fn collect_processes_all(state: &AppState) -> ProcessesPayload {
|
||||||
// Fresh view to avoid lingering entries and select "no tasks" (no per-thread rows).
|
// Fresh view to avoid lingering entries and select "no tasks" (no per-thread rows).
|
||||||
// Only processes, no per-thread entries.
|
|
||||||
let mut sys = System::new();
|
let mut sys = System::new();
|
||||||
sys.refresh_processes_specifics(
|
sys.refresh_processes_specifics(
|
||||||
ProcessesToUpdate::All,
|
ProcessesToUpdate::All,
|
||||||
@@ -256,12 +261,20 @@ pub async fn collect_processes_top_k(state: &AppState, k: usize) -> ProcessesPay
|
|||||||
|
|
||||||
// Compute deltas vs last sample
|
// Compute deltas vs last sample
|
||||||
let (last_total, mut last_map) = {
|
let (last_total, mut last_map) = {
|
||||||
let mut t = state.proc_cpu.lock().await;
|
#[cfg(target_os = "linux")]
|
||||||
let lt = t.last_total;
|
{
|
||||||
let lm = std::mem::take(&mut t.last_per_pid);
|
let mut t = state.proc_cpu.lock().await;
|
||||||
t.last_total = total_now;
|
let lt = t.last_total;
|
||||||
t.last_per_pid = current.clone();
|
let lm = std::mem::take(&mut t.last_per_pid);
|
||||||
(lt, lm)
|
t.last_total = total_now;
|
||||||
|
t.last_per_pid = current.clone();
|
||||||
|
(lt, lm)
|
||||||
|
}
|
||||||
|
#[cfg(not(target_os = "linux"))]
|
||||||
|
{
|
||||||
|
let _: u64 = total_now; // silence unused warning
|
||||||
|
(0u64, HashMap::new())
|
||||||
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
// On first run or if total delta is tiny, report zeros
|
// On first run or if total delta is tiny, report zeros
|
||||||
@@ -278,7 +291,7 @@ pub async fn collect_processes_top_k(state: &AppState, k: usize) -> ProcessesPay
|
|||||||
.collect();
|
.collect();
|
||||||
return ProcessesPayload {
|
return ProcessesPayload {
|
||||||
process_count: total_count,
|
process_count: total_count,
|
||||||
top_processes: top_k_sorted(procs, k),
|
top_processes: procs,
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -304,24 +317,48 @@ pub async fn collect_processes_top_k(state: &AppState, k: usize) -> ProcessesPay
|
|||||||
|
|
||||||
ProcessesPayload {
|
ProcessesPayload {
|
||||||
process_count: total_count,
|
process_count: total_count,
|
||||||
top_processes: top_k_sorted(procs, k),
|
top_processes: procs,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Collect all processes (non-Linux): use sysinfo's internal CPU% by doing a double refresh.
|
||||||
|
#[cfg(not(target_os = "linux"))]
|
||||||
|
pub async fn collect_processes_all(state: &AppState) -> ProcessesPayload {
|
||||||
|
use tokio::time::sleep;
|
||||||
|
|
||||||
|
let mut sys = state.sys.lock().await;
|
||||||
|
|
||||||
|
// First refresh to set baseline
|
||||||
|
sys.refresh_processes_specifics(
|
||||||
|
ProcessesToUpdate::All,
|
||||||
|
false,
|
||||||
|
ProcessRefreshKind::everything().without_tasks(),
|
||||||
|
);
|
||||||
|
// Small delay so sysinfo can compute CPU deltas on next refresh
|
||||||
|
sleep(Duration::from_millis(250)).await;
|
||||||
|
sys.refresh_processes_specifics(
|
||||||
|
ProcessesToUpdate::All,
|
||||||
|
false,
|
||||||
|
ProcessRefreshKind::everything().without_tasks(),
|
||||||
|
);
|
||||||
|
|
||||||
|
let total_count = sys.processes().len();
|
||||||
|
|
||||||
|
let procs: Vec<ProcessInfo> = sys
|
||||||
|
.processes()
|
||||||
|
.values()
|
||||||
|
.map(|p| ProcessInfo {
|
||||||
|
pid: p.pid().as_u32(),
|
||||||
|
name: p.name().to_string_lossy().into_owned(),
|
||||||
|
cpu_usage: p.cpu_usage(),
|
||||||
|
mem_bytes: p.memory(),
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
ProcessesPayload {
|
||||||
|
process_count: total_count,
|
||||||
|
top_processes: procs,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Small helper to select and sort top-k by cpu
|
// Small helper to select and sort top-k by cpu
|
||||||
fn top_k_sorted(mut v: Vec<ProcessInfo>, k: usize) -> Vec<ProcessInfo> {
|
// Client now handles sorting/pagination.
|
||||||
if v.len() > k {
|
|
||||||
v.select_nth_unstable_by(k, |a, b| {
|
|
||||||
b.cpu_usage
|
|
||||||
.partial_cmp(&a.cpu_usage)
|
|
||||||
.unwrap_or(std::cmp::Ordering::Equal)
|
|
||||||
});
|
|
||||||
v.truncate(k);
|
|
||||||
}
|
|
||||||
v.sort_by(|a, b| {
|
|
||||||
b.cpu_usage
|
|
||||||
.partial_cmp(&a.cpu_usage)
|
|
||||||
.unwrap_or(std::cmp::Ordering::Equal)
|
|
||||||
});
|
|
||||||
v
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -1,32 +1,5 @@
|
|||||||
use serde::{Deserialize, Serialize};
|
// Generated protobuf modules live under OUT_DIR; include them here.
|
||||||
|
// This module will expose socktop::Processes and socktop::Process types.
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
pub mod pb {
|
||||||
pub struct Metrics {
|
include!(concat!(env!("OUT_DIR"), "/socktop.rs"));
|
||||||
pub ts_unix_ms: i64,
|
|
||||||
pub host: String,
|
|
||||||
pub uptime_secs: u64,
|
|
||||||
pub cpu_overall: f32,
|
|
||||||
pub cpu_per_core: Vec<f32>,
|
|
||||||
pub load_avg: (f64, f64, f64),
|
|
||||||
pub mem_total_mb: u64,
|
|
||||||
pub mem_used_mb: u64,
|
|
||||||
pub swap_total_mb: u64,
|
|
||||||
pub swap_used_mb: u64,
|
|
||||||
pub net_aggregate: NetTotals,
|
|
||||||
pub top_processes: Vec<Proc>,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct NetTotals {
|
|
||||||
pub rx_bytes: u64,
|
|
||||||
pub tx_bytes: u64,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct Proc {
|
|
||||||
pub pid: i32,
|
|
||||||
pub name: String,
|
|
||||||
pub cpu: f32,
|
|
||||||
pub mem_mb: u64,
|
|
||||||
pub status: String,
|
|
||||||
}
|
|
||||||
@@ -1,5 +1,6 @@
|
|||||||
//! Shared agent state: sysinfo handles and hot JSON cache.
|
//! Shared agent state: sysinfo handles and hot JSON cache.
|
||||||
|
|
||||||
|
#[cfg(target_os = "linux")]
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::sync::atomic::AtomicUsize;
|
use std::sync::atomic::AtomicUsize;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
@@ -11,6 +12,7 @@ pub type SharedComponents = Arc<Mutex<Components>>;
|
|||||||
pub type SharedDisks = Arc<Mutex<Disks>>;
|
pub type SharedDisks = Arc<Mutex<Disks>>;
|
||||||
pub type SharedNetworks = Arc<Mutex<Networks>>;
|
pub type SharedNetworks = Arc<Mutex<Networks>>;
|
||||||
|
|
||||||
|
#[cfg(target_os = "linux")]
|
||||||
#[derive(Default)]
|
#[derive(Default)]
|
||||||
pub struct ProcCpuTracker {
|
pub struct ProcCpuTracker {
|
||||||
pub last_total: u64,
|
pub last_total: u64,
|
||||||
@@ -24,7 +26,8 @@ pub struct AppState {
|
|||||||
pub disks: SharedDisks,
|
pub disks: SharedDisks,
|
||||||
pub networks: SharedNetworks,
|
pub networks: SharedNetworks,
|
||||||
|
|
||||||
// For correct per-process CPU% using /proc deltas
|
// For correct per-process CPU% using /proc deltas (Linux only path uses this tracker)
|
||||||
|
#[cfg(target_os = "linux")]
|
||||||
pub proc_cpu: Arc<Mutex<ProcCpuTracker>>,
|
pub proc_cpu: Arc<Mutex<ProcCpuTracker>>,
|
||||||
|
|
||||||
// Connection tracking (to allow future idle sleeps if desired)
|
// Connection tracking (to allow future idle sleeps if desired)
|
||||||
@@ -45,6 +48,7 @@ impl AppState {
|
|||||||
components: Arc::new(Mutex::new(components)),
|
components: Arc::new(Mutex::new(components)),
|
||||||
disks: Arc::new(Mutex::new(disks)),
|
disks: Arc::new(Mutex::new(disks)),
|
||||||
networks: Arc::new(Mutex::new(networks)),
|
networks: Arc::new(Mutex::new(networks)),
|
||||||
|
#[cfg(target_os = "linux")]
|
||||||
proc_cpu: Arc::new(Mutex::new(ProcCpuTracker::default())),
|
proc_cpu: Arc::new(Mutex::new(ProcCpuTracker::default())),
|
||||||
client_count: Arc::new(AtomicUsize::new(0)),
|
client_count: Arc::new(AtomicUsize::new(0)),
|
||||||
auth_token: std::env::var("SOCKTOP_TOKEN")
|
auth_token: std::env::var("SOCKTOP_TOKEN")
|
||||||
|
|||||||
@@ -0,0 +1,96 @@
|
|||||||
|
use openssl::asn1::Asn1Time;
|
||||||
|
use openssl::hash::MessageDigest;
|
||||||
|
use openssl::nid::Nid;
|
||||||
|
use openssl::pkey::PKey;
|
||||||
|
use openssl::rsa::Rsa;
|
||||||
|
use openssl::x509::extension::{
|
||||||
|
BasicConstraints, ExtendedKeyUsage, KeyUsage, SubjectAlternativeName,
|
||||||
|
};
|
||||||
|
use openssl::x509::{X509NameBuilder, X509};
|
||||||
|
use std::{
|
||||||
|
fs,
|
||||||
|
io::Write,
|
||||||
|
net::{IpAddr, Ipv4Addr},
|
||||||
|
path::{Path, PathBuf},
|
||||||
|
};
|
||||||
|
|
||||||
|
fn config_dir() -> PathBuf {
|
||||||
|
std::env::var_os("XDG_CONFIG_HOME")
|
||||||
|
.map(PathBuf::from)
|
||||||
|
.or_else(|| std::env::var_os("HOME").map(|h| Path::new(&h).join(".config")))
|
||||||
|
.unwrap_or_else(|| PathBuf::from("."))
|
||||||
|
.join("socktop_agent")
|
||||||
|
.join("tls")
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn cert_paths() -> (PathBuf, PathBuf) {
|
||||||
|
let dir = config_dir();
|
||||||
|
(dir.join("cert.pem"), dir.join("key.pem"))
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn ensure_self_signed_cert() -> anyhow::Result<(PathBuf, PathBuf)> {
|
||||||
|
let (cert_path, key_path) = cert_paths();
|
||||||
|
if cert_path.exists() && key_path.exists() {
|
||||||
|
return Ok((cert_path, key_path));
|
||||||
|
}
|
||||||
|
fs::create_dir_all(cert_path.parent().unwrap())?;
|
||||||
|
|
||||||
|
// Key
|
||||||
|
let rsa = Rsa::generate(4096)?;
|
||||||
|
let pkey = PKey::from_rsa(rsa)?;
|
||||||
|
|
||||||
|
// Subject/issuer
|
||||||
|
let hostname = hostname::get()
|
||||||
|
.ok()
|
||||||
|
.and_then(|s| s.into_string().ok())
|
||||||
|
.unwrap_or_else(|| "localhost".to_string());
|
||||||
|
let mut name = X509NameBuilder::new()?;
|
||||||
|
name.append_entry_by_nid(Nid::COMMONNAME, &hostname)?;
|
||||||
|
let name = name.build();
|
||||||
|
|
||||||
|
// Cert builder
|
||||||
|
let mut builder = X509::builder()?;
|
||||||
|
builder.set_version(2)?;
|
||||||
|
builder.set_subject_name(&name)?;
|
||||||
|
builder.set_issuer_name(&name)?;
|
||||||
|
builder.set_pubkey(&pkey)?;
|
||||||
|
|
||||||
|
builder.set_not_before(Asn1Time::days_from_now(0)?.as_ref())?;
|
||||||
|
builder.set_not_after(Asn1Time::days_from_now(397)?.as_ref())?;
|
||||||
|
|
||||||
|
// SANs: hostname + localhost loopbacks
|
||||||
|
let mut san = SubjectAlternativeName::new();
|
||||||
|
san.dns(&hostname)
|
||||||
|
.dns("localhost")
|
||||||
|
.ip("127.0.0.1")
|
||||||
|
.ip("::1");
|
||||||
|
// Add a generic 0.0.0.0 for convenience; some TLS libs ignore this, but harmless.
|
||||||
|
let _ = san.ip(&IpAddr::V4(Ipv4Addr::UNSPECIFIED).to_string());
|
||||||
|
let san = san.build(&builder.x509v3_context(None, None))?;
|
||||||
|
// End-entity cert: not a CA
|
||||||
|
builder.append_extension(BasicConstraints::new().critical().build()?)?;
|
||||||
|
builder.append_extension(
|
||||||
|
KeyUsage::new()
|
||||||
|
.digital_signature()
|
||||||
|
.key_encipherment()
|
||||||
|
.build()?,
|
||||||
|
)?;
|
||||||
|
// TLS server usage
|
||||||
|
builder.append_extension(ExtendedKeyUsage::new().server_auth().build()?)?;
|
||||||
|
builder.append_extension(san)?;
|
||||||
|
|
||||||
|
builder.sign(&pkey, MessageDigest::sha256())?;
|
||||||
|
let cert: X509 = builder.build();
|
||||||
|
|
||||||
|
let mut f = fs::File::create(&cert_path)?;
|
||||||
|
f.write_all(&cert.to_pem()?)?;
|
||||||
|
let mut k = fs::File::create(&key_path)?;
|
||||||
|
k.write_all(&pkey.private_key_to_pem_pkcs8()?)?;
|
||||||
|
|
||||||
|
println!(
|
||||||
|
"socktop_agent: generated self-signed TLS certificate at {}",
|
||||||
|
cert_path.display()
|
||||||
|
);
|
||||||
|
println!("socktop_agent: private key at {}", key_path.display());
|
||||||
|
Ok((cert_path, key_path))
|
||||||
|
}
|
||||||
+35
-3
@@ -10,7 +10,8 @@ use futures_util::StreamExt;
|
|||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::io::Write;
|
use std::io::Write;
|
||||||
|
|
||||||
use crate::metrics::{collect_disks, collect_fast_metrics, collect_processes_top_k};
|
use crate::metrics::{collect_disks, collect_fast_metrics, collect_processes_all};
|
||||||
|
use crate::proto::pb;
|
||||||
use crate::state::AppState;
|
use crate::state::AppState;
|
||||||
|
|
||||||
pub async fn ws_handler(
|
pub async fn ws_handler(
|
||||||
@@ -44,8 +45,39 @@ async fn handle_socket(mut socket: WebSocket, state: AppState) {
|
|||||||
let _ = send_json(&mut socket, &d).await;
|
let _ = send_json(&mut socket, &d).await;
|
||||||
}
|
}
|
||||||
Message::Text(ref text) if text == "get_processes" => {
|
Message::Text(ref text) if text == "get_processes" => {
|
||||||
let p = collect_processes_top_k(&state, 50).await;
|
let payload = collect_processes_all(&state).await;
|
||||||
let _ = send_json(&mut socket, &p).await;
|
// Map to protobuf message
|
||||||
|
let rows: Vec<pb::Process> = payload
|
||||||
|
.top_processes
|
||||||
|
.into_iter()
|
||||||
|
.map(|p| pb::Process {
|
||||||
|
pid: p.pid,
|
||||||
|
name: p.name,
|
||||||
|
cpu_usage: p.cpu_usage,
|
||||||
|
mem_bytes: p.mem_bytes,
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
let pb = pb::Processes {
|
||||||
|
process_count: payload.process_count as u64,
|
||||||
|
rows,
|
||||||
|
};
|
||||||
|
let mut buf = Vec::with_capacity(8 * 1024);
|
||||||
|
if prost::Message::encode(&pb, &mut buf).is_err() {
|
||||||
|
let _ = socket.send(Message::Close(None)).await;
|
||||||
|
} else {
|
||||||
|
// compress if large
|
||||||
|
if buf.len() <= 768 {
|
||||||
|
let _ = socket.send(Message::Binary(buf)).await;
|
||||||
|
} else {
|
||||||
|
let mut enc = GzEncoder::new(Vec::new(), Compression::fast());
|
||||||
|
if enc.write_all(&buf).is_ok() {
|
||||||
|
let bin = enc.finish().unwrap_or(buf);
|
||||||
|
let _ = socket.send(Message::Binary(bin)).await;
|
||||||
|
} else {
|
||||||
|
let _ = socket.send(Message::Binary(buf)).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
Message::Close(_) => break,
|
Message::Close(_) => break,
|
||||||
_ => {}
|
_ => {}
|
||||||
|
|||||||
@@ -0,0 +1,28 @@
|
|||||||
|
//! CLI arg parsing tests for socktop_agent (server)
|
||||||
|
use std::process::Command;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_help_and_port_short_long() {
|
||||||
|
// We verify port flags are accepted by ensuring the process starts (then we kill quickly).
|
||||||
|
// Use an unlikely port to avoid conflicts.
|
||||||
|
let exe = env!("CARGO_BIN_EXE_socktop_agent");
|
||||||
|
|
||||||
|
// TLS enabled with long --port
|
||||||
|
let mut child = Command::new(exe)
|
||||||
|
.args(["--enableSSL", "--port", "9555"])
|
||||||
|
.spawn()
|
||||||
|
.expect("spawn agent");
|
||||||
|
// Give it a moment to bind
|
||||||
|
std::thread::sleep(std::time::Duration::from_millis(150));
|
||||||
|
let _ = child.kill();
|
||||||
|
let _ = child.wait();
|
||||||
|
|
||||||
|
// TLS enabled with short -p
|
||||||
|
let mut child2 = Command::new(exe)
|
||||||
|
.args(["--enableSSL", "-p", "9556"])
|
||||||
|
.spawn()
|
||||||
|
.expect("spawn agent");
|
||||||
|
std::thread::sleep(std::time::Duration::from_millis(150));
|
||||||
|
let _ = child2.kill();
|
||||||
|
let _ = child2.wait();
|
||||||
|
}
|
||||||
@@ -0,0 +1,59 @@
|
|||||||
|
use assert_cmd::prelude::*;
|
||||||
|
use std::fs;
|
||||||
|
use std::path::PathBuf;
|
||||||
|
use std::process::Command;
|
||||||
|
use std::time::Duration;
|
||||||
|
use std::time::Instant;
|
||||||
|
|
||||||
|
fn expected_paths(config_home: &std::path::Path) -> (PathBuf, PathBuf) {
|
||||||
|
let base = config_home.join("socktop_agent").join("tls");
|
||||||
|
(base.join("cert.pem"), base.join("key.pem"))
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn generates_self_signed_cert_and_key_in_xdg_path() {
|
||||||
|
// Create an isolated fake XDG_CONFIG_HOME
|
||||||
|
let tmpdir = tempfile::tempdir().expect("tempdir");
|
||||||
|
let xdg = tmpdir.path().to_path_buf();
|
||||||
|
|
||||||
|
// Run the agent once with --enableSSL, short timeout so it exits quickly when killed
|
||||||
|
let mut cmd = Command::cargo_bin("socktop_agent").expect("binary exists");
|
||||||
|
// Bind to an ephemeral port (-p 0) to avoid conflicts/flakes
|
||||||
|
cmd.env("XDG_CONFIG_HOME", &xdg)
|
||||||
|
.arg("--enableSSL")
|
||||||
|
.arg("-p")
|
||||||
|
.arg("0");
|
||||||
|
|
||||||
|
// Spawn the process and poll for cert generation
|
||||||
|
let mut child = cmd.spawn().expect("spawn agent");
|
||||||
|
|
||||||
|
// Poll up to ~3s for files to appear to avoid timing flakes
|
||||||
|
let (cert_path, key_path) = expected_paths(&xdg);
|
||||||
|
let start = Instant::now();
|
||||||
|
let timeout = Duration::from_millis(3000);
|
||||||
|
let interval = Duration::from_millis(50);
|
||||||
|
while start.elapsed() < timeout {
|
||||||
|
if cert_path.exists() && key_path.exists() {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
std::thread::sleep(interval);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Terminate the process regardless
|
||||||
|
let _ = child.kill();
|
||||||
|
let _ = child.wait();
|
||||||
|
|
||||||
|
// Verify files exist at expected paths
|
||||||
|
assert!(
|
||||||
|
cert_path.exists(),
|
||||||
|
"cert not found at {}",
|
||||||
|
cert_path.display()
|
||||||
|
);
|
||||||
|
assert!(key_path.exists(), "key not found at {}", key_path.display());
|
||||||
|
|
||||||
|
// Also ensure they are non-empty
|
||||||
|
let cert_md = fs::metadata(&cert_path).expect("cert metadata");
|
||||||
|
let key_md = fs::metadata(&key_path).expect("key metadata");
|
||||||
|
assert!(cert_md.len() > 0, "cert is empty");
|
||||||
|
assert!(key_md.len() > 0, "key is empty");
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user