okooooooo/NodeFlare
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

完善初始化、远程执行与安全检查

Browse Source
This commit is contained in:
imengying committed 2026-09-04 15:11:41 +08:00
1 parent b7f54cddac
commit 6e3819dedf
19 files changed
+1108 -189
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.github/dependabot.yml
+42
View File
@@ -0,0 +1,42 @@
version: 2
updates:
- package-ecosystem: cargo
directory: /backend
schedule:
interval: weekly
groups:
backend-rust:
patterns:
- "*"
open-pull-requests-limit: 5
- package-ecosystem: cargo
directory: /agent
schedule:
interval: weekly
groups:
agent-rust:
patterns:
- "*"
open-pull-requests-limit: 5
- package-ecosystem: npm
directory: /
schedule:
interval: weekly
groups:
web:
patterns:
- "*"
open-pull-requests-limit: 5
- package-ecosystem: github-actions
directory: /
schedule:
interval: weekly
groups:
actions:
patterns:
- "*"
open-pull-requests-limit: 5
.github/workflows/release.yml
+24 -1
View File
@@ -45,6 +45,29 @@ jobs:
echo "version=$version" >> "$GITHUB_OUTPUT"
echo "release=$RELEASE_BUILD" >> "$GITHUB_OUTPUT"
security:
name: Dependency security
runs-on: ubuntu-latest
timeout-minutes: 20
steps:
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
with:
persist-credentials: false
- uses: dtolnay/rust-toolchain@4360b52568e2003a75bf9bc1d59f33a8e3fc893c # stable
- uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2.2.0
with:
bun-version: 1.4.0
- name: Audit web dependencies
timeout-minutes: 5
run: bun audit
- name: Install Rust dependency auditor
timeout-minutes: 10
run: cargo install cargo-audit --version 0.22.2 --locked --no-default-features
- name: Audit backend dependencies
run: cargo audit --file backend/Cargo.lock
- name: Audit Agent dependencies
run: cargo audit --file agent/Cargo.lock
quality:
name: Quality checks
needs: metadata
@@ -420,7 +443,7 @@ jobs:
release:
name: Publish GitHub Release
if: startsWith(github.ref, 'refs/tags/v')
needs: [metadata, quality, server, agent, frontend]
needs: [metadata, security, quality, server, agent, frontend]
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
README.md
+6 -6
View File
@@ -12,7 +12,7 @@ NodeFlare 是一个自部署的服务器监控面板。服务端使用 Rust/Axum
sudo ./install.sh
```
首次安装会在终端询问管理员用户名，并隐藏输入、二次确认管理员密码；随后可直接填写 SQLite 或 PostgreSQL URL，回车则使用 `/etc/nodeflare/nodeflare.db`。脚本会构建 NodeFlare，将后端安装为 `/opt/nodeflare/nodeflare`、静态资源安装到 `/opt/nodeflare/share`、systemd 服务安装为 `nodeflare.service`，并将配置和服务端持久数据统一放在 `/etc/nodeflare`。重复运行安装脚本会保留已有配置和数据库。
首次安装会在终端询问管理员用户名，并隐藏输入、二次确认管理员密码；随后可直接填写 SQLite 或 PostgreSQL URL，回车则使用 `/etc/nodeflare/nodeflare.db`。脚本会构建 NodeFlare，将后端安装为 `/opt/nodeflare/nodeflare`、静态资源安装到 `/opt/nodeflare/share`、systemd 服务安装为 `nodeflare.service`，并将配置和服务端持久数据统一放在 `/etc/nodeflare`。数据库初始化成功后，配置中的初始管理员密码会自动清空；重复运行安装脚本会保留已有配置和数据库。
面板不会创建额外的 Linux 系统用户，systemd 服务的运行方式与 Agent 一致。
Linux 默认目录结构：
@@ -21,13 +21,13 @@ Linux 默认目录结构：
/opt/nodeflare/nodeflare # 后端
/opt/nodeflare/agent # Agent（安装时才存在）
/opt/nodeflare/share/ # 前端与安装脚本
/opt/nodeflare/data/agent/ # Agent 暂存数据
/etc/nodeflare/config.toml # 后端配置
/etc/nodeflare/nodeflare.db # 默认 SQLite 数据库
/etc/nodeflare/themes/ # 已安装主题
/etc/nodeflare/agent/ # Agent 暂存数据与任务日志
```
默认可从 `http://127.0.0.1:8080` 访问公开面板，管理页位于 `/admin`。“登录与安全”可启用 TOTP、查看登录设备并将指定设备踢下线；远程执行使用一次性命令，可同时选择多台服务器，每次下发都必须提交当前 TOTP 验证码。等待超过 24 小时的命令会自动取消，已完成结果随历史保留周期清理。
默认可从 `http://127.0.0.1:8080` 访问公开面板，管理页位于 `/admin`。“登录与安全”可启用 TOTP、查看登录设备并将指定设备踢下线；远程执行使用一次性命令，可同时选择多台服务器，每次下发都必须提交当前 TOTP 验证码。Agent 会在执行前持久化任务 ID，重启时不会重复运行中断的命令；结果会重传到后端确认落库为止。等待超过 24 小时的命令会自动取消，已完成结果随历史保留周期清理。
后台每个侧边栏入口都有独立地址，例如远程执行为 `/admin/remote`、主题商店为 `/admin/themes`、关于为 `/admin/about`。刷新页面或使用浏览器前进、后退时会保留当前栏目。
@@ -53,7 +53,7 @@ sudo ./install.sh --uninstall --purge
## 配置
正式安装的配置文件是 `/etc/nodeflare/config.toml`。安装时会交互式询问用户名、密码和可跳过的数据库 URL；监听地址、数据库、Turnstile、静态资源与会话时长等均可在该 TOML 文件中修改。相对路径以配置文件所在目录为基准解析。
正式安装的配置文件是 `/etc/nodeflare/config.toml`。安装时会交互式询问用户名、仅供首次初始化使用的密码和可跳过的数据库 URL；监听地址、数据库、Turnstile、静态资源与会话时长等均可在该 TOML 文件中修改。相对路径以配置文件所在目录为基准解析。
SQLite 示例：
@@ -93,7 +93,7 @@ curl -fsSL https://raw.githubusercontent.com/imengying/NodeFlare/main/agent/agen
Agent 通过 `/api/agent/ws` 建立 WebSocket 连接。除 localhost 调试外，安装脚本要求 HTTPS。
Agent Token 由面板生成，安装脚本将可执行文件安装为 `agent`（Windows 为 `agent.exe`），服务名为 `nodeflare-agent`。Linux 使用 `/opt/nodeflare`；macOS 与 FreeBSD 使用 `/usr/local/libexec/nodeflare`；Windows 使用 `%ProgramFiles%\NodeFlare`。程序、静态资源和持久数据均收拢在各平台的 NodeFlare 目录中，Agent 暂存数据位于其中的 `data/agent`。服务端只保存 Token 哈希，Token 仅在创建节点或主动重置时显示；重置会使旧 Agent 立即离线。Unix 服务通过受保护的服务环境传入 Token，避免出现在进程命令行；Windows 计划任务仍保存为 SYSTEM 任务参数。安装过程不会额外生成 Agent 配置文件。
Agent Token 由面板生成，安装脚本将可执行文件安装为 `agent`（Windows 为 `agent.exe`），服务名为 `nodeflare-agent`。Linux 可执行文件位于 `/opt/nodeflare`，状态目录为 `/etc/nodeflare/agent`；macOS 使用 `/usr/local/libexec/nodeflare` 与 `/Library/Application Support/NodeFlare/Agent`；FreeBSD 使用 `/usr/local/libexec/nodeflare` 与 `/var/db/nodeflare/agent`；Windows 使用 `%ProgramFiles%\NodeFlare` 与 `%ProgramData%\NodeFlare\Agent`。状态目录只保存待上报指标和远程任务日志。服务端只保存 Token 哈希，Token 仅在创建节点或主动重置时显示；重置会使旧 Agent 立即离线。Unix 服务通过受保护的服务环境传入 Token，避免出现在进程命令行；Windows 计划任务仍保存为 SYSTEM 任务参数。安装过程不会额外生成 Agent 配置文件或 Token 文件。
远程执行由系统级 Agent 服务直接运行：Linux、macOS、FreeBSD 默认是 `root`，Windows 是 `SYSTEM`，因此等同于服务器控制权。请只通过 HTTPS 暴露 NodeFlare、启用 TOTP、保护管理员会话，并仅安装可信主题；Linux Agent 不启用文件系统隔离，否则运维命令无法管理宿主机。
@@ -183,7 +183,7 @@ pg_dump --format=custom --file=/var/backups/nodeflare.dump \
'postgres://nodeflare:password@127.0.0.1:5432/nodeflare?sslmode=prefer'
```
恢复前请停止 NodeFlare，并先在独立环境验证备份。配置文件包含管理员初始凭据和可选的 Turnstile 密钥，也应通过权限受控的方式单独备份。
恢复前请停止 NodeFlare，并先在独立环境验证备份。配置文件可能包含数据库凭据和可选的 Turnstile 密钥，也应通过权限受控的方式单独备份。
## 架构
agent/agent.sh
+5 -5
View File
@@ -4,8 +4,8 @@ set -eu
SERVICE_NAME="nodeflare-agent"
INSTALL_DIR="/opt/nodeflare"
AGENT_FILE="$INSTALL_DIR/agent"
DATA_DIR="$INSTALL_DIR/data"
STATE_DIR="$DATA_DIR/agent"
STATE_ROOT="/etc/nodeflare"
STATE_DIR="$STATE_ROOT/agent"
SERVICE_FILE="/etc/systemd/system/$SERVICE_NAME.service"
OPENRC_FILE="/etc/init.d/$SERVICE_NAME"
@@ -136,9 +136,9 @@ install_agent() {
init_system=$(detect_init_system)
[ "$init_system" != "unknown" ] || fail "未检测到正在运行的 systemd 或 OpenRC"
mkdir -p "$INSTALL_DIR" "$DATA_DIR" "$STATE_DIR"
mkdir -p "$INSTALL_DIR" "$STATE_ROOT" "$STATE_DIR"
chmod 755 "$INSTALL_DIR"
chmod 750 "$DATA_DIR"
chmod 700 "$STATE_ROOT"
chmod 750 "$STATE_DIR"
temporary="$INSTALL_DIR/.agent.$$.download"
trap 'rm -f "$temporary"' EXIT HUP INT TERM
@@ -289,7 +289,7 @@ uninstall_agent() {
[ "$init_system" != "systemd" ] || systemctl daemon-reload 2>/dev/null || true
rm -f "$AGENT_FILE"
rm -rf "$STATE_DIR"
rmdir "$DATA_DIR" 2>/dev/null || true
rmdir "$STATE_ROOT" 2>/dev/null || true
rmdir "$INSTALL_DIR" 2>/dev/null || true
echo "NodeFlare Agent 已卸载"
}
agent/install-freebsd.sh
+5 -5
View File
@@ -5,8 +5,8 @@ SERVICE_NAME="nodeflare-agent"
RC_NAME="nodeflare_agent"
INSTALL_DIR="/usr/local/libexec/nodeflare"
AGENT_FILE="$INSTALL_DIR/agent"
DATA_DIR="$INSTALL_DIR/data"
STATE_DIR="$DATA_DIR/agent"
STATE_ROOT="/var/db/nodeflare"
STATE_DIR="$STATE_ROOT/agent"
SERVICE_FILE="/usr/local/etc/rc.d/$SERVICE_NAME"
log() {
@@ -68,7 +68,7 @@ if [ "${1:-}" = "--uninstall" ]; then
sysrc -x "${RC_NAME}_enable" >/dev/null 2>&1 || true
rm -f "$SERVICE_FILE" "$AGENT_FILE"
rm -rf "$STATE_DIR"
rmdir "$DATA_DIR" 2>/dev/null || true
rmdir "$STATE_ROOT" 2>/dev/null || true
rmdir "$INSTALL_DIR" 2>/dev/null || true
echo "NodeFlare Agent 已卸载"
exit 0
@@ -141,9 +141,9 @@ if [ -n "$mirror" ]; then
case "$mirror" in *@*) fail "下载加速前缀不能包含用户信息" ;; esac
fi
mkdir -p "$INSTALL_DIR" "$DATA_DIR" "$STATE_DIR"
mkdir -p "$INSTALL_DIR" "$STATE_ROOT" "$STATE_DIR"
chmod 755 "$INSTALL_DIR"
chmod 750 "$DATA_DIR"
chmod 750 "$STATE_ROOT"
chmod 750 "$STATE_DIR"
temporary="$INSTALL_DIR/.agent.$$.download"
trap 'rm -f "$temporary"' EXIT HUP INT TERM
agent/install-macos.sh
+5 -5
View File
@@ -4,8 +4,8 @@ set -eu
LABEL="nodeflare-agent"
INSTALL_DIR="/usr/local/libexec/nodeflare"
AGENT_FILE="$INSTALL_DIR/agent"
DATA_DIR="$INSTALL_DIR/data"
STATE_DIR="$DATA_DIR/agent"
STATE_ROOT="/Library/Application Support/NodeFlare"
STATE_DIR="$STATE_ROOT/Agent"
PLIST_FILE="/Library/LaunchDaemons/$LABEL.plist"
log() {
@@ -43,7 +43,7 @@ if [ "${1:-}" = "--uninstall" ]; then
launchctl bootout system "$PLIST_FILE" 2>/dev/null || true
rm -f "$PLIST_FILE" "$AGENT_FILE"
rm -rf "$STATE_DIR"
rmdir "$DATA_DIR" 2>/dev/null || true
rmdir "$STATE_ROOT" 2>/dev/null || true
rmdir "$INSTALL_DIR" 2>/dev/null || true
echo "NodeFlare Agent 已卸载"
exit 0
@@ -109,9 +109,9 @@ if [ -n "$mirror" ]; then
case "$mirror" in *@*) fail "下载加速前缀不能包含用户信息" ;; esac
fi
mkdir -p "$INSTALL_DIR" "$DATA_DIR" "$STATE_DIR"
mkdir -p "$INSTALL_DIR" "$STATE_ROOT" "$STATE_DIR"
chmod 755 "$INSTALL_DIR"
chmod 750 "$DATA_DIR"
chmod 750 "$STATE_ROOT"
chmod 750 "$STATE_DIR"
temporary="$INSTALL_DIR/.agent.$$.download"
trap 'rm -f "$temporary"' EXIT HUP INT TERM
agent/install.ps1
+2 -2
View File
@@ -11,8 +11,8 @@ param(
$ErrorActionPreference = "Stop"
$TaskName = "nodeflare-agent"
$InstallDir = Join-Path $env:ProgramFiles "NodeFlare"
$DataDir = Join-Path $InstallDir "data"
$StateDir = Join-Path $DataDir "agent"
$DataDir = Join-Path $env:ProgramData "NodeFlare"
$StateDir = Join-Path $DataDir "Agent"
$AgentFile = Join-Path $InstallDir "agent.exe"
function Write-Step([string]$Message) {
agent/src/main.rs
+499 -77
View File
@@ -45,6 +45,10 @@ const REMOTE_TASK_TIMEOUT: Duration = Duration::from_secs(10 * 60);
const REMOTE_STREAM_OUTPUT_BYTES: u64 = 512 * 1024;
const REMOTE_RESULT_OUTPUT_BYTES: usize = 900 * 1024;
const REMOTE_OUTPUT_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
const REMOTE_RESULT_RETRY_INTERVAL: Duration = Duration::from_secs(5);
const REMOTE_TASK_JOURNAL_VERSION: u32 = 1;
const MAX_REMOTE_TASK_JOURNAL_ENTRIES: usize = 32;
const MAX_REMOTE_TASK_JOURNAL_BYTES: u64 = 32 * 1024 * 1024;
const LATENCY_WORKERS: usize = 4;
const MAX_PENDING_LATENCY_RESULTS: usize = 4096;
const MAX_REPORT_AGE_SECONDS: i64 = 7_000;
@@ -72,6 +76,8 @@ const PUBLIC_IP_V6_URL: &str = "https://ipv6.icanhazip.com/";
const RUNTIME_STATS_INTERVAL: Duration = Duration::from_secs(5 * 60);
const CLOCK_CALIBRATION_MAX_AGE: Duration = Duration::from_secs(24 * 60 * 60);
const CLOCK_CALIBRATION_MIN_CHANGE_MS: i64 = 20_000;
const AGENT_PROTOCOL_VERSION: &str = "1";
const AGENT_CAPABILITIES: &str = "metrics-v1,config-v1,remote-exec-v1,task-ack-v1";
type Error = Box<dyn std::error::Error + Send + Sync>;
type Result<T> = std::result::Result<T, Error>;
@@ -1755,6 +1761,14 @@ fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option<i64>)
request
.headers_mut()
.insert("User-Agent", format!("nodeflare-agent/{VERSION}").parse()?);
request.headers_mut().insert(
"X-NodeFlare-Agent-Protocol",
AGENT_PROTOCOL_VERSION.parse()?,
);
request.headers_mut().insert(
"X-NodeFlare-Agent-Capabilities",
AGENT_CAPABILITIES.parse()?,
);
let started_ms = unix_timestamp_millis();
let (socket, response) = connect(request)?;
let ended_ms = unix_timestamp_millis();
@@ -1813,7 +1827,7 @@ struct RemoteTaskMessage {
command: String,
}
#[derive(Debug, Serialize)]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
struct TaskResultMessage {
#[serde(rename = "type")]
message_type: String,
@@ -1823,6 +1837,28 @@ struct TaskResultMessage {
exit_code: Option<i64>,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct TaskResultAckMessage {
#[serde(rename = "type")]
message_type: String,
task_id: String,
}
#[derive(Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct RemoteTaskJournal {
version: u32,
entries: Vec<RemoteTaskJournalEntry>,
}
#[derive(Debug, Deserialize, Serialize)]
#[serde(tag = "state", rename_all = "snake_case")]
enum RemoteTaskJournalEntry {
Active { task_id: String },
Completed { result: TaskResultMessage },
}
#[derive(Default)]
struct CapturedOutput {
bytes: Vec<u8>,
@@ -2039,12 +2075,37 @@ fn execute_remote_task(task: &RemoteTaskMessage) -> TaskResultMessage {
struct RemoteExecutor {
task_tx: mpsc::SyncSender<RemoteTaskMessage>,
result_rx: mpsc::Receiver<TaskResultMessage>,
seen: HashSet<String>,
seen_order: VecDeque<String>,
active: HashSet<String>,
completed: HashMap<String, TaskResultMessage>,
task_order: VecDeque<String>,
last_sent: HashMap<String, Instant>,
journal_path: PathBuf,
disabled: bool,
}
impl RemoteExecutor {
fn new() -> Result<Self> {
Self::new_at(remote_task_journal_path()?)
}
fn new_at(journal_path: PathBuf) -> Result<Self> {
let journal = load_remote_task_journal(&journal_path)?;
let mut active = HashSet::new();
let mut completed = HashMap::new();
let mut task_order = VecDeque::new();
for entry in journal.entries {
match entry {
RemoteTaskJournalEntry::Active { task_id } => {
active.insert(task_id.clone());
task_order.push_back(task_id);
}
RemoteTaskJournalEntry::Completed { result } => {
task_order.push_back(result.task_id.clone());
completed.insert(result.task_id.clone(), result);
}
}
}
let (task_tx, task_rx) = mpsc::sync_channel::<RemoteTaskMessage>(16);
let (result_tx, result_rx) = mpsc::channel::<TaskResultMessage>();
thread::Builder::new()
@@ -2056,59 +2117,289 @@ impl RemoteExecutor {
}
}
})?;
Ok(Self {
let interrupted = active.drain().collect::<Vec<_>>();
for task_id in interrupted {
completed.insert(
task_id.clone(),
TaskResultMessage {
message_type: "task_result".to_string(),
task_id,
status: "failed".to_string(),
result: "Agent 在命令执行期间重启，无法确认原命令状态；为避免重复操作，任务不会再次执行"
.to_string(),
exit_code: Some(-1),
},
);
}
let executor = Self {
task_tx,
result_rx,
seen: HashSet::new(),
seen_order: VecDeque::new(),
})
active,
completed,
task_order,
last_sent: HashMap::new(),
journal_path,
disabled: false,
};
if !executor.completed.is_empty() {
executor.persist_snapshot(None)?;
}
Ok(executor)
}
fn disabled() -> Self {
let (task_tx, task_rx) = mpsc::sync_channel(1);
let (result_tx, result_rx) = mpsc::channel();
drop(task_rx);
drop(result_tx);
Self {
task_tx,
result_rx,
active: HashSet::new(),
completed: HashMap::new(),
task_order: VecDeque::new(),
last_sent: HashMap::new(),
journal_path: PathBuf::new(),
disabled: true,
}
}
fn enqueue(&mut self, task: RemoteTaskMessage) -> Option<TaskResultMessage> {
if self.seen.contains(&task.task_id) {
if self.disabled {
return Some(remote_task_rejected(
task.task_id,
"Agent 任务日志异常，远程执行已禁用",
));
}
if self.completed.contains_key(&task.task_id) {
self.last_sent.remove(&task.task_id);
return None;
}
if self.active.contains(&task.task_id) {
return None;
}
let task_id = task.task_id.clone();
self.seen.insert(task_id.clone());
self.seen_order.push_back(task_id.clone());
while self.seen_order.len() > 1024 {
if let Some(expired) = self.seen_order.pop_front() {
self.seen.remove(&expired);
}
if self.task_order.len() >= MAX_REMOTE_TASK_JOURNAL_ENTRIES {
return Some(remote_task_rejected(
task_id,
"待确认的远程任务过多，已拒绝执行",
));
}
self.active.insert(task_id.clone());
self.task_order.push_back(task_id.clone());
if let Err(error) = self.persist_snapshot(None) {
self.active.remove(&task_id);
self.task_order.retain(|id| id != &task_id);
eprintln!("remote task journal write failed: {error}");
return Some(remote_task_rejected(
task_id,
"Agent 无法持久化任务状态，已拒绝执行",
));
}
match self.task_tx.try_send(task) {
Ok(()) => None,
Err(mpsc::TrySendError::Full(_)) => Some(TaskResultMessage {
message_type: "task_result".to_string(),
task_id,
status: "failed".to_string(),
result: "远程命令队列已满，请稍后重试".to_string(),
exit_code: Some(-1),
}),
Err(mpsc::TrySendError::Disconnected(_)) => Some(TaskResultMessage {
message_type: "task_result".to_string(),
task_id,
status: "failed".to_string(),
result: "远程命令执行线程不可用".to_string(),
exit_code: Some(-1),
}),
Err(error) => {
let message = match error {
mpsc::TrySendError::Full(_) => "远程命令队列已满，请稍后重试",
mpsc::TrySendError::Disconnected(_) => "远程命令执行线程不可用",
};
let result = remote_task_rejected(task_id, message);
self.remember_completed(result.clone());
None
}
}
}
fn drain(&self) -> impl Iterator<Item = TaskResultMessage> + '_ {
self.result_rx.try_iter()
fn drain_completed(&mut self) {
let results = self.result_rx.try_iter().collect::<Vec<_>>();
for result in results {
self.remember_completed(result);
}
}
fn acknowledge(&mut self, task_id: &str) {
if !self.completed.contains_key(task_id) {
return;
}
match self.persist_snapshot(Some(task_id)) {
Ok(()) => {
self.completed.remove(task_id);
self.task_order.retain(|id| id != task_id);
self.last_sent.remove(task_id);
}
Err(error) => {
eprintln!("remote task ACK persistence failed: {error}");
}
}
}
fn remember_completed(&mut self, result: TaskResultMessage) {
let task_id = result.task_id.clone();
let was_active = self.active.remove(&task_id);
if !was_active && !self.completed.contains_key(&task_id) {
self.task_order.push_back(task_id.clone());
}
self.completed.insert(task_id.clone(), result);
self.last_sent.remove(&task_id);
if let Err(error) = self.persist_snapshot(None) {
eprintln!("remote task result persistence failed: {error}");
}
}
fn due_results(&mut self) -> Vec<TaskResultMessage> {
self.drain_completed();
let now = Instant::now();
self.task_order
.iter()
.filter_map(|task_id| {
let result = self.completed.get(task_id)?;
let due = self.last_sent.get(task_id).is_none_or(|sent_at| {
now.duration_since(*sent_at) >= REMOTE_RESULT_RETRY_INTERVAL
});
due.then(|| result.clone())
})
.collect()
}
fn mark_sent(&mut self, task_id: &str) {
self.last_sent.insert(task_id.to_string(), Instant::now());
}
fn reset_delivery(&mut self) {
self.last_sent.clear();
}
fn persist_snapshot(&self, excluded_task_id: Option<&str>) -> Result<()> {
let entries = self
.task_order
.iter()
.filter(|task_id| excluded_task_id != Some(task_id.as_str()))
.filter_map(|task_id| {
if self.active.contains(task_id) {
Some(RemoteTaskJournalEntry::Active {
task_id: task_id.clone(),
})
} else {
self.completed
.get(task_id)
.cloned()
.map(|result| RemoteTaskJournalEntry::Completed { result })
}
})
.collect::<Vec<_>>();
write_remote_task_journal(&self.journal_path, entries)
}
}
fn flush_remote_results(
socket: &mut LiveSocket,
executor: &RemoteExecutor,
pending: &mut VecDeque<TaskResultMessage>,
) -> Result<()> {
pending.extend(executor.drain());
while let Some(result) = pending.front() {
socket.send(Message::Text(serde_json::to_string(result)?.into()))?;
pending.pop_front();
fn remote_task_rejected(task_id: String, result: &str) -> TaskResultMessage {
TaskResultMessage {
message_type: "task_result".to_string(),
task_id,
status: "failed".to_string(),
result: result.to_string(),
exit_code: Some(-1),
}
}
fn valid_remote_task_id(value: &str) -> bool {
value.len() == 36
&& value.bytes().enumerate().all(|(index, byte)| {
if matches!(index, 8 | 13 | 18 | 23) {
byte == b'-'
} else {
byte.is_ascii_hexdigit()
}
})
}
fn load_remote_task_journal(path: &Path) -> Result<RemoteTaskJournal> {
let bytes = match fs::read(path) {
Ok(bytes) => bytes,
Err(error) if error.kind() == ErrorKind::NotFound => {
return Ok(RemoteTaskJournal {
version: REMOTE_TASK_JOURNAL_VERSION,
entries: Vec::new(),
});
}
Err(error) => return Err(error.into()),
};
if bytes.len() as u64 > MAX_REMOTE_TASK_JOURNAL_BYTES {
return Err("remote task journal is too large".into());
}
let journal = serde_json::from_slice::<RemoteTaskJournal>(&bytes)?;
if journal.version != REMOTE_TASK_JOURNAL_VERSION {
return Err("unsupported remote task journal version".into());
}
if journal.entries.len() > MAX_REMOTE_TASK_JOURNAL_ENTRIES {
return Err("remote task journal has too many entries".into());
}
let mut task_ids = HashSet::new();
for entry in &journal.entries {
let (task_id, valid_result) = match entry {
RemoteTaskJournalEntry::Active { task_id } => (task_id, true),
RemoteTaskJournalEntry::Completed { result } => (
&result.task_id,
result.message_type == "task_result"
&& matches!(result.status.as_str(), "success" | "failed")
&& result.result.len() <= REMOTE_RESULT_OUTPUT_BYTES + 64,
),
};
if !valid_remote_task_id(task_id) || !valid_result || !task_ids.insert(task_id) {
return Err("remote task journal contains an invalid entry".into());
}
}
Ok(journal)
}
fn write_remote_task_journal(path: &Path, entries: Vec<RemoteTaskJournalEntry>) -> Result<()> {
if entries.is_empty() {
return match fs::remove_file(path) {
Ok(()) => Ok(()),
Err(error) if error.kind() == ErrorKind::NotFound => Ok(()),
Err(error) => Err(error.into()),
};
}
let journal = RemoteTaskJournal {
version: REMOTE_TASK_JOURNAL_VERSION,
entries,
};
let mut encoded = serde_json::to_vec(&journal)?;
encoded.push(b'\n');
if encoded.len() as u64 > MAX_REMOTE_TASK_JOURNAL_BYTES {
return Err("remote task journal is too large".into());
}
let temporary = path.with_file_name(format!(
".{}.{}.tmp",
path.file_name()
.and_then(|value| value.to_str())
.unwrap_or("remote-tasks"),
std::process::id()
));
let mut file = OpenOptions::new()
.create(true)
.truncate(true)
.write(true)
.open(&temporary)?;
#[cfg(unix)]
fs::set_permissions(&temporary, fs::Permissions::from_mode(0o600))?;
file.write_all(&encoded)?;
file.sync_all()?;
drop(file);
#[cfg(target_os = "windows")]
if path.exists() {
fs::remove_file(path)?;
}
if let Err(error) = fs::rename(&temporary, path) {
let _ = fs::remove_file(&temporary);
return Err(error.into());
}
Ok(())
}
fn flush_remote_results(socket: &mut LiveSocket, executor: &mut RemoteExecutor) -> Result<()> {
for result in executor.due_results() {
socket.send(Message::Text(serde_json::to_string(&result)?.into()))?;
executor.mark_sent(&result.task_id);
}
Ok(())
}
@@ -2127,6 +2418,7 @@ enum LiveRead {
Ack(LiveAck),
Config(RemoteConfig),
RemoteTask(RemoteTaskMessage),
TaskResultAck(String),
Closed,
Pending,
}
@@ -2161,8 +2453,7 @@ fn read_live_ack(socket: &mut LiveSocket) -> Result<LiveRead> {
return Ok(LiveRead::Pending);
};
if task.message_type != "remote_task"
|| task.task_id.is_empty()
|| task.task_id.len() > 80
|| !valid_remote_task_id(&task.task_id)
|| task.command.trim().is_empty()
|| task.command.len() > 16_384
{
@@ -2170,6 +2461,16 @@ fn read_live_ack(socket: &mut LiveSocket) -> Result<LiveRead> {
}
Ok(LiveRead::RemoteTask(task))
}
Some("task_result_ack") => {
let Ok(ack) = serde_json::from_value::<TaskResultAckMessage>(value) else {
return Ok(LiveRead::Pending);
};
if ack.message_type != "task_result_ack" || !valid_remote_task_id(&ack.task_id)
{
return Ok(LiveRead::Pending);
}
Ok(LiveRead::TaskResultAck(ack.task_id))
}
_ => Ok(LiveRead::Pending),
}
}
@@ -2188,11 +2489,32 @@ fn read_live_ack(socket: &mut LiveSocket) -> Result<LiveRead> {
}
}
fn accept_remote_task(
socket: &mut LiveSocket,
executor: &mut RemoteExecutor,
task: RemoteTaskMessage,
) -> Result<()> {
let task_id = task.task_id.clone();
let rejected = executor.enqueue(task);
socket.send(Message::Text(
serde_json::json!({
"type": "task_received",
"task_id": task_id,
})
.to_string()
.into(),
))?;
if let Some(result) = rejected {
socket.send(Message::Text(serde_json::to_string(&result)?.into()))?;
}
flush_remote_results(socket, executor)?;
Ok(())
}
fn wait_for_live_ack(
socket: &mut LiveSocket,
remote_config: &Arc<Mutex<Option<RemoteConfig>>>,
remote_executor: &mut RemoteExecutor,
pending_remote_results: &mut VecDeque<TaskResultMessage>,
) -> Result<LiveRead> {
let deadline = Instant::now() + LIVE_ACK_READ_TIMEOUT;
loop {
@@ -2211,9 +2533,10 @@ fn wait_for_live_ack(
}
}
LiveRead::RemoteTask(task) => {
if let Some(result) = remote_executor.enqueue(task) {
pending_remote_results.push_back(result);
}
accept_remote_task(socket, remote_executor, task)?;
}
LiveRead::TaskResultAck(task_id) => {
remote_executor.acknowledge(&task_id);
}
LiveRead::Closed => return Ok(LiveRead::Closed),
LiveRead::Ack(_) | LiveRead::Pending => {}
@@ -2282,11 +2605,10 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
let mut remote_executor = match RemoteExecutor::new() {
Ok(executor) => executor,
Err(error) => {
eprintln!("remote executor failed to start: {error}");
return;
eprintln!("remote executor disabled: {error}");
RemoteExecutor::disabled()
}
};
let mut pending_remote_results = VecDeque::new();
'sender: loop {
if socket.is_none() {
match connect_live(endpoint, token) {
@@ -2298,6 +2620,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
continue;
}
socket = Some(connected);
remote_executor.reset_delivery();
accepted_through = persisted_through.load(Ordering::Acquire);
next_send_at = Instant::now();
next_probe_at = Instant::now();
@@ -2312,8 +2635,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
}
if let Some(connected) = socket.as_mut()
&& flush_remote_results(connected, &remote_executor, &mut pending_remote_results)
.is_err()
&& flush_remote_results(connected, &mut remote_executor).is_err()
{
socket = None;
thread::sleep(LIVE_RECONNECT_DELAY);
@@ -2356,8 +2678,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
let mut drop_socket = false;
if let Some(connected) = socket.as_mut() {
if flush_remote_results(connected, &remote_executor, &mut pending_remote_results)
.is_err()
if flush_remote_results(connected, &mut remote_executor).is_err()
|| set_live_read_timeout(connected, Some(LIVE_HINT_READ_TIMEOUT)).is_err()
{
drop_socket = true;
@@ -2383,10 +2704,13 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
}
}
Ok(LiveRead::RemoteTask(task)) => {
if let Some(result) = remote_executor.enqueue(task) {
pending_remote_results.push_back(result);
if accept_remote_task(connected, &mut remote_executor, task).is_err() {
drop_socket = true;
}
}
Ok(LiveRead::TaskResultAck(task_id)) => {
remote_executor.acknowledge(&task_id);
}
Ok(LiveRead::Pending) => {}
Err(error) => {
eprintln!("live hint read failed: {error}");
@@ -2437,12 +2761,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
let mut drop_socket = false;
if let Some(socket) = socket.as_mut() {
match wait_for_live_ack(
socket,
&remote_config,
&mut remote_executor,
&mut pending_remote_results,
) {
match wait_for_live_ack(socket, &remote_config, &mut remote_executor) {
Ok(LiveRead::Closed) => drop_socket = true,
Ok(LiveRead::Ack(ack)) => {
observe_persisted_through(&persisted_through, &ack);
@@ -2469,6 +2788,9 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) {
Ok(LiveRead::RemoteTask(_task)) => {
// 在 wait_for_live_ack 中已处理，这里不应该出现
}
Ok(LiveRead::TaskResultAck(_)) => {
// 在 wait_for_live_ack 中已处理，这里不应该出现
}
Ok(LiveRead::Pending) => {
eprintln!("live ACK timed out");
drop_socket = true;
@@ -2581,25 +2903,31 @@ fn prune_report_samples(samples: &mut Vec<Report>, now: i64) {
}
}
fn pending_spool_path() -> Result<PathBuf> {
fn agent_state_directory() -> Result<PathBuf> {
if let Some(value) = env::var_os("NODEFLARE_STATE_DIR") {
let directory = PathBuf::from(value);
if !directory.is_absolute() {
return Err("NODEFLARE_STATE_DIR must be an absolute path".into());
}
fs::create_dir_all(&directory)?;
return Ok(directory.join("pending.jsonl"));
return Ok(directory);
}
#[cfg(target_os = "windows")]
{
let executable = env::current_exe()?;
let directory = executable
.parent()
.ok_or("agent executable path has no parent directory")?
.join("data")
.join("agent");
let directory = env::var_os("ProgramData")
.map(PathBuf::from)
.filter(|path| path.is_absolute())
.map(|path| path.join("NodeFlare").join("Agent"))
.unwrap_or_else(|| {
env::current_exe()
.ok()
.and_then(|path| path.parent().map(Path::to_path_buf))
.unwrap_or_else(|| PathBuf::from("."))
.join("data")
.join("agent")
});
fs::create_dir_all(&directory)?;
return Ok(directory.join("pending.jsonl"));
return Ok(directory);
}
#[cfg(not(target_os = "windows"))]
{
@@ -2607,10 +2935,18 @@ fn pending_spool_path() -> Result<PathBuf> {
let directory = executable
.parent()
.ok_or("agent executable path has no parent directory")?;
Ok(directory.join("pending.jsonl"))
Ok(directory.to_path_buf())
}
}
fn pending_spool_path() -> Result<PathBuf> {
Ok(agent_state_directory()?.join("pending.jsonl"))
}
fn remote_task_journal_path() -> Result<PathBuf> {
Ok(agent_state_directory()?.join("remote-tasks.json"))
}
fn load_pending_spool(path: &Path) -> Result<Vec<Report>> {
let file = match fs::File::open(path) {
Ok(file) => file,
@@ -3124,20 +3460,23 @@ fn main() {
#[cfg(test)]
mod tests {
use clap::Parser;
use std::time::{Duration, Instant};
use std::fs;
use std::path::PathBuf;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use super::{
CLOCK_CALIBRATION_MAX_AGE, CapturedOutput, CliOptions, ClockCalibration,
GithubReleaseAsset, LatencyResult, LatencyTask, LiveAck, MAX_PENDING_LATENCY_RESULTS,
PUBLIC_IP_STALE_AFTER, PublicIpValue, REMOTE_RESULT_OUTPUT_BYTES, RemoteTaskMessage,
Report, UPDATE_CHECK_JITTER_MAX_SECONDS, ack_persist_interval, ack_wss_interval,
advance_deadline, clock_offset_from_http_date, corrected_timestamp,
execute_remote_task_with_timeout, gpu_name_from_uevent, is_public_probe_ip, live_endpoint,
live_update_payload, monotonic_report_timestamp, normalized_version, parse_lspci_gpu_names,
PUBLIC_IP_STALE_AFTER, PublicIpValue, REMOTE_RESULT_OUTPUT_BYTES, RemoteExecutor,
RemoteTaskJournalEntry, RemoteTaskMessage, Report, TaskResultMessage,
UPDATE_CHECK_JITTER_MAX_SECONDS, ack_persist_interval, ack_wss_interval, advance_deadline,
clock_offset_from_http_date, corrected_timestamp, execute_remote_task_with_timeout,
gpu_name_from_uevent, is_public_probe_ip, live_endpoint, live_update_payload,
monotonic_report_timestamp, normalized_version, parse_lspci_gpu_names,
parse_pciconf_gpu_names, parse_probe_target, parse_public_ip,
parse_system_profiler_gpu_names, ping_latencies, ping_latency, prune_report_samples,
release_asset_sha256, remote_result_text, sanitize_latency_tasks, update_check_jitter,
valid_endpoint, version_triplet,
valid_endpoint, version_triplet, write_remote_task_journal,
};
#[cfg(any(target_os = "windows", target_os = "macos", target_os = "freebsd"))]
@@ -3348,6 +3687,89 @@ mod tests {
assert!(result.ends_with("[输出已截断]"));
}
fn temporary_remote_task_journal(label: &str) -> PathBuf {
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos();
let directory = std::env::temp_dir().join(format!(
"nodeflare-agent-{label}-{}-{nonce}",
std::process::id()
));
fs::create_dir_all(&directory).unwrap();
directory.join("remote-tasks.json")
}
#[test]
fn interrupted_remote_tasks_are_failed_without_reexecution() {
let path = temporary_remote_task_journal("interrupted-task");
let task_id = "8dd70536-f721-4d47-af63-e2f17c83de25";
write_remote_task_journal(
&path,
vec![RemoteTaskJournalEntry::Active {
task_id: task_id.to_string(),
}],
)
.unwrap();
let mut executor = RemoteExecutor::new_at(path.clone()).unwrap();
assert!(executor.active.is_empty());
let due = executor.due_results();
assert_eq!(due.len(), 1);
assert_eq!(due[0].task_id, task_id);
assert_eq!(due[0].status, "failed");
assert!(due[0].result.contains("不会再次执行"));
executor.mark_sent(task_id);
assert!(executor.due_results().is_empty());
executor.reset_delivery();
assert_eq!(executor.due_results().len(), 1);
executor.acknowledge(task_id);
assert!(!path.exists());
let directory = path.parent().unwrap().to_path_buf();
drop(executor);
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn completed_remote_tasks_return_the_cached_result() {
let path = temporary_remote_task_journal("completed-task");
let task_id = "6cbf4d36-f33a-40ef-b952-6af1b8491820";
let result = TaskResultMessage {
message_type: "task_result".to_string(),
task_id: task_id.to_string(),
status: "success".to_string(),
result: "already finished".to_string(),
exit_code: Some(0),
};
write_remote_task_journal(
&path,
vec![RemoteTaskJournalEntry::Completed {
result: result.clone(),
}],
)
.unwrap();
let mut executor = RemoteExecutor::new_at(path.clone()).unwrap();
assert!(
executor
.enqueue(RemoteTaskMessage {
message_type: "remote_task".to_string(),
task_id: task_id.to_string(),
command: "this command must not run".to_string(),
})
.is_none()
);
assert!(executor.active.is_empty());
assert_eq!(executor.due_results(), [result]);
executor.acknowledge(task_id);
let directory = path.parent().unwrap().to_path_buf();
drop(executor);
fs::remove_dir_all(directory).unwrap();
}
#[test]
fn encodes_realtime_samples_as_a_batch() {
let payload = live_update_payload(&[
backend/README.md
+1 -1
View File
@@ -47,7 +47,7 @@ theme_dir = "/etc/nodeflare/themes"
session_ttl_hours = 168
```
PostgreSQL 可将 `database_url` 改为 `postgres://user:password@host:5432/database?sslmode=prefer`。相对 SQLite 路径和静态资源路径均相对于配置文件目录解析；URL 中的特殊字符需要进行百分号编码。安装脚本询问管理员用户名、密码和数据库 URL，其他项直接编辑 `/etc/nodeflare/config.toml`。面板服务不创建额外的系统用户。
PostgreSQL 可将 `database_url` 改为 `postgres://user:password@host:5432/database?sslmode=prefer`。相对 SQLite 路径和静态资源路径均相对于配置文件目录解析；URL 中的特殊字符需要进行百分号编码。安装脚本询问管理员用户名、首次初始化密码和数据库 URL；初始化成功后密码会从配置中自动清空，其他项直接编辑 `/etc/nodeflare/config.toml`。面板服务不创建额外的系统用户。
## 主要端点
backend/config.example.toml
+1 -1
View File
@@ -12,7 +12,7 @@ bind_addr = "127.0.0.1:8080"
# 管理员用户名
admin_username = "admin"
# 管理员密码（8-128 个字符；启动前必须替换此占位值）
# 管理员密码（8-128 个字符；仅首次初始化数据库需要，成功后自动清空）
admin_password = "CHANGE_ME_WITH_A_STRONG_PASSWORD"
# Cloudflare Turnstile 站点密钥（公开）
backend/migrations/postgres/01_initial.sql
+33 -27
View File
@@ -5,37 +5,41 @@ CREATE TABLE settings (
CREATE TABLE servers (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
name TEXT NOT NULL CHECK(char_length(btrim(name)) BETWEEN 1 AND 80),
region TEXT NOT NULL DEFAULT '',
group_name TEXT NOT NULL DEFAULT '默认',
tags TEXT NOT NULL DEFAULT '',
hidden BIGINT NOT NULL DEFAULT 0,
hidden BIGINT NOT NULL DEFAULT 0 CHECK(hidden IN (0, 1)),
sort_order BIGINT NOT NULL DEFAULT 0,
expires_at BIGINT,
traffic_limit BIGINT NOT NULL DEFAULT 0,
traffic_limit_type TEXT NOT NULL DEFAULT 'sum',
price DOUBLE PRECISION NOT NULL DEFAULT 0,
billing_cycle BIGINT NOT NULL DEFAULT 30,
currency TEXT NOT NULL DEFAULT 'CNY',
auto_renewal BIGINT NOT NULL DEFAULT 0,
traffic_limit BIGINT NOT NULL DEFAULT 0 CHECK(traffic_limit >= 0),
traffic_limit_type TEXT NOT NULL DEFAULT 'sum'
CHECK(traffic_limit_type IN ('sum', 'max', 'min', 'up', 'down')),
price DOUBLE PRECISION NOT NULL DEFAULT 0 CHECK(price BETWEEN -1 AND 1000000000),
billing_cycle BIGINT NOT NULL DEFAULT 30 CHECK(billing_cycle BETWEEN 0 AND 3650),
currency TEXT NOT NULL DEFAULT 'CNY' CHECK(currency ~ '^[A-Z]{3}$'),
auto_renewal BIGINT NOT NULL DEFAULT 0 CHECK(auto_renewal IN (0, 1)),
last_ip TEXT NOT NULL DEFAULT '',
ip_v4 TEXT NOT NULL DEFAULT '',
ip_v6 TEXT NOT NULL DEFAULT '',
network_interface TEXT NOT NULL DEFAULT '',
reset_day BIGINT NOT NULL DEFAULT 1,
report_interval BIGINT NOT NULL DEFAULT 60,
collect_interval BIGINT NOT NULL DEFAULT 1,
reset_day BIGINT NOT NULL DEFAULT 1 CHECK(reset_day BETWEEN 1 AND 31),
report_interval BIGINT NOT NULL DEFAULT 60 CHECK(report_interval BETWEEN 15 AND 3600),
collect_interval BIGINT NOT NULL DEFAULT 1 CHECK(collect_interval BETWEEN 1 AND 60),
rx_correction BIGINT NOT NULL DEFAULT 0,
tx_correction BIGINT NOT NULL DEFAULT 0,
agent_mirror TEXT NOT NULL DEFAULT '',
offline_notify_disabled BIGINT NOT NULL DEFAULT 0,
auto_update BIGINT NOT NULL DEFAULT 1,
offline_notify_disabled BIGINT NOT NULL DEFAULT 0 CHECK(offline_notify_disabled IN (0, 1)),
auto_update BIGINT NOT NULL DEFAULT 1 CHECK(auto_update IN (0, 1)),
token_hash TEXT NOT NULL UNIQUE,
created_at BIGINT NOT NULL,
updated_at BIGINT NOT NULL
updated_at BIGINT NOT NULL,
CHECK(collect_interval <= report_interval),
CHECK((report_interval + collect_interval - 1) / collect_interval <= 720)
);
CREATE INDEX servers_sort ON servers(sort_order, created_at);
CREATE INDEX servers_public_sort ON servers(hidden, sort_order, created_at);
CREATE TABLE metric_history (
server_id TEXT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
@@ -90,7 +94,7 @@ CREATE INDEX server_latest_state_time ON server_latest_state(latest_timestamp);
CREATE TABLE admin_2fa (
username TEXT PRIMARY KEY,
totp_secret TEXT NOT NULL,
enabled BIGINT NOT NULL DEFAULT 1,
enabled BIGINT NOT NULL DEFAULT 1 CHECK(enabled IN (0, 1)),
created_at BIGINT NOT NULL
);
@@ -133,7 +137,9 @@ CREATE TABLE sessions (
user_agent TEXT NOT NULL DEFAULT '',
created_at BIGINT NOT NULL,
last_seen_at BIGINT NOT NULL,
expires_at BIGINT NOT NULL
expires_at BIGINT NOT NULL,
CHECK(last_seen_at >= created_at),
CHECK(expires_at > created_at)
);
CREATE INDEX sessions_expires_at ON sessions(expires_at);
@@ -142,7 +148,7 @@ CREATE INDEX sessions_user_activity ON sessions(username, last_seen_at DESC);
CREATE TABLE dashboard_proofs (
token_hash TEXT PRIMARY KEY,
created_at BIGINT NOT NULL,
expires_at BIGINT NOT NULL
expires_at BIGINT NOT NULL CHECK(expires_at > created_at)
);
CREATE INDEX dashboard_proofs_expires_at ON dashboard_proofs(expires_at);
@@ -150,12 +156,12 @@ CREATE INDEX dashboard_proofs_expires_at ON dashboard_proofs(expires_at);
CREATE TABLE server_traffic_state (
server_id TEXT PRIMARY KEY REFERENCES servers(id) ON DELETE CASCADE,
cycle_key BIGINT NOT NULL DEFAULT 0,
reset_day BIGINT NOT NULL DEFAULT 1,
reset_day BIGINT NOT NULL DEFAULT 1 CHECK(reset_day BETWEEN 1 AND 31),
timestamp BIGINT NOT NULL DEFAULT 0,
raw_rx BIGINT NOT NULL DEFAULT 0,
raw_tx BIGINT NOT NULL DEFAULT 0,
used_rx BIGINT NOT NULL DEFAULT 0,
used_tx BIGINT NOT NULL DEFAULT 0
raw_rx BIGINT NOT NULL DEFAULT 0 CHECK(raw_rx >= 0),
raw_tx BIGINT NOT NULL DEFAULT 0 CHECK(raw_tx >= 0),
used_rx BIGINT NOT NULL DEFAULT 0 CHECK(used_rx >= 0),
used_tx BIGINT NOT NULL DEFAULT 0 CHECK(used_tx >= 0)
);
CREATE TABLE latency_tasks (
@@ -165,7 +171,7 @@ CREATE TABLE latency_tasks (
target TEXT NOT NULL,
port BIGINT CHECK(port IS NULL OR port BETWEEN 1 AND 65535),
interval_seconds BIGINT NOT NULL CHECK(interval_seconds BETWEEN 30 AND 3600),
default_enabled BIGINT NOT NULL DEFAULT 0,
default_enabled BIGINT NOT NULL DEFAULT 0 CHECK(default_enabled IN (0, 1)),
sort_order BIGINT NOT NULL DEFAULT 0,
created_at BIGINT NOT NULL,
updated_at BIGINT NOT NULL,
@@ -187,8 +193,8 @@ CREATE TABLE latency_results (
task_id TEXT NOT NULL REFERENCES latency_tasks(id) ON DELETE CASCADE,
server_id TEXT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
timestamp BIGINT NOT NULL,
latency_ms DOUBLE PRECISION NOT NULL,
packet_loss DOUBLE PRECISION NOT NULL,
latency_ms DOUBLE PRECISION NOT NULL CHECK(latency_ms >= 0),
packet_loss DOUBLE PRECISION NOT NULL CHECK(packet_loss BETWEEN 0 AND 100),
PRIMARY KEY(task_id, server_id, timestamp)
);
@@ -202,7 +208,7 @@ CREATE TABLE alert_rules (
threshold DOUBLE PRECISION NOT NULL CHECK(threshold > 0),
duration_minutes BIGINT NOT NULL CHECK(duration_minutes BETWEEN 1 AND 1440),
aggregation TEXT NOT NULL CHECK(aggregation IN ('average', 'continuous')),
enabled BIGINT NOT NULL DEFAULT 1,
enabled BIGINT NOT NULL DEFAULT 1 CHECK(enabled IN (0, 1)),
created_at BIGINT NOT NULL,
updated_at BIGINT NOT NULL
);
@@ -219,7 +225,7 @@ CREATE INDEX alert_rule_servers_server ON alert_rule_servers(server_id, rule_id)
CREATE TABLE alert_states (
state_key TEXT PRIMARY KEY,
active BIGINT NOT NULL DEFAULT 0,
active BIGINT NOT NULL DEFAULT 0 CHECK(active IN (0, 1)),
updated_at BIGINT NOT NULL,
details_json TEXT NOT NULL DEFAULT '{}'
);
backend/migrations/sqlite/01_initial.sql
+34 -27
View File
@@ -7,37 +7,42 @@ CREATE TABLE settings (
CREATE TABLE servers (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
name TEXT NOT NULL CHECK(length(trim(name)) BETWEEN 1 AND 80),
region TEXT NOT NULL DEFAULT '',
group_name TEXT NOT NULL DEFAULT '默认',
tags TEXT NOT NULL DEFAULT '',
hidden INTEGER NOT NULL DEFAULT 0,
hidden INTEGER NOT NULL DEFAULT 0 CHECK(hidden IN (0, 1)),
sort_order INTEGER NOT NULL DEFAULT 0,
expires_at INTEGER,
traffic_limit INTEGER NOT NULL DEFAULT 0,
traffic_limit_type TEXT NOT NULL DEFAULT 'sum',
price REAL NOT NULL DEFAULT 0,
billing_cycle INTEGER NOT NULL DEFAULT 30,
currency TEXT NOT NULL DEFAULT 'CNY',
auto_renewal INTEGER NOT NULL DEFAULT 0,
traffic_limit INTEGER NOT NULL DEFAULT 0 CHECK(traffic_limit >= 0),
traffic_limit_type TEXT NOT NULL DEFAULT 'sum'
CHECK(traffic_limit_type IN ('sum', 'max', 'min', 'up', 'down')),
price REAL NOT NULL DEFAULT 0 CHECK(price BETWEEN -1 AND 1000000000),
billing_cycle INTEGER NOT NULL DEFAULT 30 CHECK(billing_cycle BETWEEN 0 AND 3650),
currency TEXT NOT NULL DEFAULT 'CNY'
CHECK(length(currency) = 3 AND currency NOT GLOB '*[^A-Z]*'),
auto_renewal INTEGER NOT NULL DEFAULT 0 CHECK(auto_renewal IN (0, 1)),
last_ip TEXT NOT NULL DEFAULT '',
ip_v4 TEXT NOT NULL DEFAULT '',
ip_v6 TEXT NOT NULL DEFAULT '',
network_interface TEXT NOT NULL DEFAULT '',
reset_day INTEGER NOT NULL DEFAULT 1,
report_interval INTEGER NOT NULL DEFAULT 60,
collect_interval INTEGER NOT NULL DEFAULT 1,
reset_day INTEGER NOT NULL DEFAULT 1 CHECK(reset_day BETWEEN 1 AND 31),
report_interval INTEGER NOT NULL DEFAULT 60 CHECK(report_interval BETWEEN 15 AND 3600),
collect_interval INTEGER NOT NULL DEFAULT 1 CHECK(collect_interval BETWEEN 1 AND 60),
rx_correction INTEGER NOT NULL DEFAULT 0,
tx_correction INTEGER NOT NULL DEFAULT 0,
agent_mirror TEXT NOT NULL DEFAULT '',
offline_notify_disabled INTEGER NOT NULL DEFAULT 0,
auto_update INTEGER NOT NULL DEFAULT 1,
offline_notify_disabled INTEGER NOT NULL DEFAULT 0 CHECK(offline_notify_disabled IN (0, 1)),
auto_update INTEGER NOT NULL DEFAULT 1 CHECK(auto_update IN (0, 1)),
token_hash TEXT NOT NULL UNIQUE,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
updated_at INTEGER NOT NULL,
CHECK(collect_interval <= report_interval),
CHECK((report_interval + collect_interval - 1) / collect_interval <= 720)
) WITHOUT ROWID;
CREATE INDEX servers_sort ON servers(sort_order, created_at);
CREATE INDEX servers_public_sort ON servers(hidden, sort_order, created_at);
CREATE TABLE metric_history (
server_id TEXT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
@@ -92,7 +97,7 @@ CREATE INDEX server_latest_state_time ON server_latest_state(latest_timestamp);
CREATE TABLE admin_2fa (
username TEXT PRIMARY KEY,
totp_secret TEXT NOT NULL,
enabled INTEGER NOT NULL DEFAULT 1,
enabled INTEGER NOT NULL DEFAULT 1 CHECK(enabled IN (0, 1)),
created_at INTEGER NOT NULL
) WITHOUT ROWID;
@@ -135,7 +140,9 @@ CREATE TABLE sessions (
user_agent TEXT NOT NULL DEFAULT '',
created_at INTEGER NOT NULL,
last_seen_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL
expires_at INTEGER NOT NULL,
CHECK(last_seen_at >= created_at),
CHECK(expires_at > created_at)
) WITHOUT ROWID;
CREATE INDEX sessions_expires_at ON sessions(expires_at);
@@ -144,7 +151,7 @@ CREATE INDEX sessions_user_activity ON sessions(username, last_seen_at DESC);
CREATE TABLE dashboard_proofs (
token_hash TEXT PRIMARY KEY,
created_at INTEGER NOT NULL,
expires_at INTEGER NOT NULL
expires_at INTEGER NOT NULL CHECK(expires_at > created_at)
) WITHOUT ROWID;
CREATE INDEX dashboard_proofs_expires_at ON dashboard_proofs(expires_at);
@@ -152,12 +159,12 @@ CREATE INDEX dashboard_proofs_expires_at ON dashboard_proofs(expires_at);
CREATE TABLE server_traffic_state (
server_id TEXT PRIMARY KEY REFERENCES servers(id) ON DELETE CASCADE,
cycle_key INTEGER NOT NULL DEFAULT 0,
reset_day INTEGER NOT NULL DEFAULT 1,
reset_day INTEGER NOT NULL DEFAULT 1 CHECK(reset_day BETWEEN 1 AND 31),
timestamp INTEGER NOT NULL DEFAULT 0,
raw_rx INTEGER NOT NULL DEFAULT 0,
raw_tx INTEGER NOT NULL DEFAULT 0,
used_rx INTEGER NOT NULL DEFAULT 0,
used_tx INTEGER NOT NULL DEFAULT 0
raw_rx INTEGER NOT NULL DEFAULT 0 CHECK(raw_rx >= 0),
raw_tx INTEGER NOT NULL DEFAULT 0 CHECK(raw_tx >= 0),
used_rx INTEGER NOT NULL DEFAULT 0 CHECK(used_rx >= 0),
used_tx INTEGER NOT NULL DEFAULT 0 CHECK(used_tx >= 0)
) WITHOUT ROWID;
CREATE TABLE latency_tasks (
@@ -167,7 +174,7 @@ CREATE TABLE latency_tasks (
target TEXT NOT NULL,
port INTEGER CHECK(port IS NULL OR port BETWEEN 1 AND 65535),
interval_seconds INTEGER NOT NULL CHECK(interval_seconds BETWEEN 30 AND 3600),
default_enabled INTEGER NOT NULL DEFAULT 0,
default_enabled INTEGER NOT NULL DEFAULT 0 CHECK(default_enabled IN (0, 1)),
sort_order INTEGER NOT NULL DEFAULT 0,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
@@ -189,8 +196,8 @@ CREATE TABLE latency_results (
task_id TEXT NOT NULL REFERENCES latency_tasks(id) ON DELETE CASCADE,
server_id TEXT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
timestamp INTEGER NOT NULL,
latency_ms REAL NOT NULL,
packet_loss REAL NOT NULL,
latency_ms REAL NOT NULL CHECK(latency_ms >= 0),
packet_loss REAL NOT NULL CHECK(packet_loss BETWEEN 0 AND 100),
PRIMARY KEY(task_id, server_id, timestamp)
) WITHOUT ROWID;
@@ -206,7 +213,7 @@ CREATE TABLE alert_rules (
threshold REAL NOT NULL CHECK(threshold > 0),
duration_minutes INTEGER NOT NULL CHECK(duration_minutes BETWEEN 1 AND 1440),
aggregation TEXT NOT NULL CHECK(aggregation IN ('average', 'continuous')),
enabled INTEGER NOT NULL DEFAULT 1,
enabled INTEGER NOT NULL DEFAULT 1 CHECK(enabled IN (0, 1)),
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
) WITHOUT ROWID;
@@ -223,7 +230,7 @@ CREATE INDEX alert_rule_servers_server ON alert_rule_servers(server_id, rule_id)
CREATE TABLE alert_states (
state_key TEXT PRIMARY KEY,
active INTEGER NOT NULL DEFAULT 0,
active INTEGER NOT NULL DEFAULT 0 CHECK(active IN (0, 1)),
updated_at INTEGER NOT NULL,
details_json TEXT NOT NULL DEFAULT '{}'
) WITHOUT ROWID;
backend/src/config.rs
+66 -5
View File
@@ -1,6 +1,7 @@
use anyhow::{Context, Result};
use clap::Parser;
use serde::Deserialize;
use std::io::Write;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
@@ -32,6 +33,7 @@ pub struct Config {
#[serde(default = "default_bind_addr")]
pub bind_addr: String,
pub admin_username: String,
#[serde(default)]
pub admin_password: String,
#[serde(default)]
pub turnstile_site_key: String,
@@ -99,10 +101,12 @@ impl Config {
{
anyhow::bail!("admin_username must be 1-64 characters without whitespace");
}
if !(8..=128).contains(&config.admin_password.chars().count()) {
if !config.admin_password.is_empty()
&& !(8..=128).contains(&config.admin_password.chars().count())
{
anyhow::bail!("admin_password must be 8-128 characters");
}
if is_example_password(&config.admin_password) {
if !config.admin_password.is_empty() && is_example_password(&config.admin_password) {
anyhow::bail!(
"admin_password still contains the example placeholder; replace it before starting NodeFlare"
);
@@ -111,6 +115,46 @@ impl Config {
}
}
pub fn clear_bootstrap_password(path: &Path) -> Result<bool> {
let content = std::fs::read_to_string(path)
.with_context(|| format!("failed to read configuration {}", path.display()))?;
let mut output = String::with_capacity(content.len());
let mut replaced = false;
for segment in content.split_inclusive('\n') {
let (line, newline) = segment
.strip_suffix('\n')
.map_or((segment, ""), |line| (line, "\n"));
let trimmed = line.trim_start();
let is_password = !trimmed.starts_with('#')
&& trimmed
.strip_prefix("admin_password")
.is_some_and(|rest| rest.trim_start().starts_with('='));
if is_password {
let indentation = &line[..line.len() - trimmed.len()];
output.push_str(indentation);
output.push_str("admin_password = \"\"");
output.push_str(newline);
replaced = true;
} else {
output.push_str(segment);
}
}
if !replaced || output == content {
return Ok(false);
}
let parent = path.parent().unwrap_or_else(|| Path::new("."));
let permissions = std::fs::metadata(path)?.permissions();
let mut temporary = tempfile::NamedTempFile::new_in(parent)?;
temporary.write_all(output.as_bytes())?;
temporary.as_file().sync_all()?;
std::fs::set_permissions(temporary.path(), permissions)?;
temporary
.persist(path)
.map_err(|error| anyhow::anyhow!(error.error))?;
Ok(true)
}
fn is_example_password(value: &str) -> bool {
matches!(
value.trim().to_ascii_lowercase().as_str(),
@@ -158,9 +202,9 @@ fn resolve_database_url(base: &Path, value: &str) -> Result<String> {
#[cfg(test)]
mod tests {
use super::{
Args, default_admin_frontend_dir, default_agent_dir, default_bind_addr,
default_database_url, default_public_frontend_dir, default_theme_dir, is_example_password,
resolve_database_url,
Args, clear_bootstrap_password, default_admin_frontend_dir, default_agent_dir,
default_bind_addr, default_database_url, default_public_frontend_dir, default_theme_dir,
is_example_password, resolve_database_url,
};
use clap::Parser;
use std::path::Path;
@@ -216,4 +260,21 @@ mod tests {
assert!(is_example_password("CHANGE_ME_WITH_A_STRONG_PASSWORD"));
assert!(!is_example_password("a-real-password-123"));
}
#[test]
fn clears_only_the_bootstrap_password() {
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("config.toml");
std::fs::write(
&path,
"admin_username = \"admin\"\nadmin_password = \"secret-value\"\n# admin_password = \"comment\"\ndatabase_url = \"sqlite::memory:\"\n",
)
.unwrap();
assert!(clear_bootstrap_password(&path).unwrap());
assert_eq!(
std::fs::read_to_string(&path).unwrap(),
"admin_username = \"admin\"\nadmin_password = \"\"\n# admin_password = \"comment\"\ndatabase_url = \"sqlite::memory:\"\n"
);
assert!(!clear_bootstrap_password(&path).unwrap());
}
}
backend/src/db/mod.rs
+11 -1
View File
@@ -330,12 +330,20 @@ pub async fn initialize(pool: &Database, config: &Config) -> Result<()> {
.await?;
}
let password_hash =
sqlx::query_scalar::<_, String>(pool.sql("SELECT value FROM settings WHERE key = ?"))
.bind("admin_password_hash")
.fetch_optional(&mut *transaction)
.await?;
let scheme =
sqlx::query_scalar::<_, String>(pool.sql("SELECT value FROM settings WHERE key = ?"))
.bind("password_scheme")
.fetch_optional(&mut *transaction)
.await?;
if scheme.as_deref() != Some(PASSWORD_SCHEME) {
if password_hash.is_none() {
if !(8..=128).contains(&config.admin_password.chars().count()) {
anyhow::bail!("admin_password is required when initializing a new database");
}
let derived = auth::derive_client_password(&config.admin_password, &salt);
let password_hash = auth::hash_password(&derived)?;
for (key, value) in [
@@ -360,6 +368,8 @@ pub async fn initialize(pool: &Database, config: &Config) -> Result<()> {
.execute(&mut *transaction)
.await?;
tracing::info!(timestamp = now, "initialized password scheme");
} else if scheme.as_deref() != Some(PASSWORD_SCHEME) {
anyhow::bail!("database uses an unsupported administrator password scheme");
}
transaction.commit().await?;
Ok(())
backend/src/db/queries.rs
+69 -5
View File
@@ -1543,7 +1543,7 @@ pub async fn pending_remote_tasks(db: &Database, server_id: &str) -> Result<Vec<
sqlx::query(db.sql(
"UPDATE remote_tasks SET status='failed', completed_at=?, \
result=CASE WHEN result='' THEN '任务等待超过 24 小时，已自动取消' ELSE result END \
WHERE server_id=? AND status='pending' AND requested_at<?",
WHERE server_id=? AND status IN ('pending','sent') AND requested_at<?",
))
.bind(current)
.bind(server_id)
@@ -1552,7 +1552,8 @@ pub async fn pending_remote_tasks(db: &Database, server_id: &str) -> Result<Vec<
.await?;
let rows = sqlx::query(db.sql(
"SELECT id, server_id, command, status, requested_by, requested_at, started_at, \
completed_at, result, exit_code FROM remote_tasks WHERE server_id=? AND status='pending' \
completed_at, result, exit_code FROM remote_tasks \
WHERE server_id=? AND status IN ('pending','sent') \
ORDER BY requested_at LIMIT 50",
))
.bind(server_id)
@@ -1615,7 +1616,18 @@ pub async fn update_remote_task_result(
.bind(server_id)
.execute(db.pool())
.await?;
Ok(result.rows_affected() > 0)
if result.rows_affected() > 0 {
return Ok(true);
}
let completed = sqlx::query_scalar::<_, i64>(db.sql(
"SELECT COUNT(*) FROM remote_tasks \
WHERE id=? AND server_id=? AND status IN ('success','failed')",
))
.bind(id)
.bind(server_id)
.fetch_one(db.pool())
.await?;
Ok(completed > 0)
}
pub async fn telegram_settings(db: &Database) -> Result<Option<TelegramSettingsView>> {
@@ -2065,6 +2077,42 @@ mod tests {
assert_eq!(active_count, 1);
}
#[tokio::test]
async fn remote_task_delivery_and_results_are_idempotent() {
let db = super::super::connect("sqlite::memory:").await.unwrap();
db.migrate().await.unwrap();
let (server_id, _) = create_server(&db, &server_input(0)).await.unwrap();
let task = create_remote_task(&db, &server_id, "uptime", "admin")
.await
.unwrap();
mark_remote_task_sent(&db, &task.id, &server_id)
.await
.unwrap();
mark_remote_task_sent(&db, &task.id, &server_id)
.await
.unwrap();
assert_eq!(
remote_task(&db, &task.id).await.unwrap().unwrap().status,
"sent"
);
assert!(
update_remote_task_result(&db, &server_id, &task.id, "success", "done", Some(0))
.await
.unwrap()
);
assert!(
update_remote_task_result(&db, &server_id, &task.id, "success", "done", Some(0))
.await
.unwrap()
);
let completed = remote_task(&db, &task.id).await.unwrap().unwrap();
assert_eq!(completed.status, "success");
assert_eq!(completed.result, "done");
assert_eq!(completed.exit_code, Some(0));
}
#[tokio::test]
async fn sqlite_migration_contains_optimized_indexes() {
let db = super::super::connect("sqlite::memory:").await.unwrap();
@@ -2073,14 +2121,30 @@ mod tests {
"SELECT name FROM sqlite_master WHERE type='index' AND name IN (\
'metric_history_time',\
'latency_results_time',\
'remote_tasks_server_status_time')",
'remote_tasks_server_status_time',\
'servers_public_sort')",
)
.fetch_all(db.pool())
.await
.unwrap()
.into_iter()
.collect::<HashSet<_>>();
assert_eq!(indexes.len(), 3);
assert_eq!(indexes.len(), 4);
}
#[tokio::test]
async fn sqlite_migration_rejects_invalid_server_configuration() {
let db = super::super::connect("sqlite::memory:").await.unwrap();
db.migrate().await.unwrap();
let result = sqlx::query(
"INSERT INTO servers(\
id, name, hidden, traffic_limit_type, currency, reset_day, report_interval, \
collect_interval, token_hash, created_at, updated_at\
) VALUES ('invalid', 'Invalid', 2, 'sum', 'CNY', 1, 60, 5, 'token', 1, 1)",
)
.execute(db.pool())
.await;
assert!(result.is_err());
}
#[test]
backend/src/main.rs
+12
View File
@@ -125,6 +125,18 @@ async fn main() -> Result<()> {
tracing::info!(database = ?database.kind(), "connected to database");
database.migrate().await?;
db::initialize(&database, &config).await?;
if !config.admin_password.is_empty() {
match config::clear_bootstrap_password(&args.config) {
Ok(true) => {
tracing::info!(path = %args.config.display(), "cleared bootstrap password from configuration")
}
Ok(false) => {}
Err(error) => {
tracing::warn!(%error, path = %args.config.display(), "failed to clear bootstrap password from configuration")
}
}
config.admin_password.clear();
}
database.optimize().await?;
for (label, path) in [
backend/src/routes/remote.rs
+1 -5
View File
@@ -101,11 +101,7 @@ pub async fn create_task(
crate::db::queries::create_remote_task(&state.db, &server_id, command, &user.username)
.await
.map_err(ApiResponse::internal)?;
if state.send_remote_task(&task).await {
crate::db::queries::mark_remote_task_sent(&state.db, &task.id, &task.server_id)
.await
.map_err(ApiResponse::internal)?;
}
state.send_remote_task(&task).await;
tasks.push(CreatedRemoteTask {
server_id,
task_id: task.id,
backend/src/websocket/agent.rs
+100 -12
View File
@@ -16,6 +16,11 @@ use tokio::sync::mpsc;
const MAX_AGENT_MESSAGE_BYTES: usize = 2 * 1024 * 1024;
const MAX_TASK_RESULT_BYTES: usize = 1024 * 1024;
const AGENT_PROTOCOL_VERSION: &str = "1";
const AGENT_PROTOCOL_HEADER: &str = "x-nodeflare-agent-protocol";
const AGENT_CAPABILITIES_HEADER: &str = "x-nodeflare-agent-capabilities";
const REQUIRED_AGENT_CAPABILITIES: [&str; 4] =
["metrics-v1", "config-v1", "remote-exec-v1", "task-ack-v1"];
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
@@ -38,6 +43,14 @@ struct TaskResultMessage {
exit_code: Option<i64>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct TaskReceivedMessage {
#[serde(rename = "type")]
_message_type: String,
task_id: String,
}
pub async fn handle(
ws: WebSocketUpgrade,
State(state): State<Arc<AppState>>,
@@ -51,6 +64,13 @@ pub async fn handle(
)
.into_response();
}
if !agent_protocol_supported(&headers) {
return ApiResponse::error(
axum::http::StatusCode::UPGRADE_REQUIRED,
format!("Agent 协议不兼容，需要协议版本 {AGENT_PROTOCOL_VERSION}"),
)
.into_response();
}
let token = headers
.get(axum::http::header::AUTHORIZATION)
.and_then(|value| value.to_str().ok())
@@ -131,14 +151,7 @@ async fn run(
crate::db::queries::pending_remote_tasks(&state.db, &identity.server_id).await
{
for task in tasks {
if send_task(&outbound_tx, &task) {
let _ = crate::db::queries::mark_remote_task_sent(
&state.db,
&task.id,
&identity.server_id,
)
.await;
}
send_task(&outbound_tx, &task);
}
}
@@ -272,14 +285,13 @@ async fn handle_text(
let Ok(result) = serde_json::from_value::<TaskResultMessage>(value) else {
return;
};
if result.task_id.is_empty()
|| result.task_id.len() > 80
if !valid_task_id(&result.task_id)
|| !matches!(result.status.as_str(), "success" | "failed")
|| result.result.len() > MAX_TASK_RESULT_BYTES
{
return;
}
if let Err(error) = crate::db::queries::update_remote_task_result(
match crate::db::queries::update_remote_task_result(
&state.db,
&identity.server_id,
&result.task_id,
@@ -289,7 +301,36 @@ async fn handle_text(
)
.await
{
tracing::error!(%error, server_id = %identity.server_id, "task result persistence failed");
Ok(true) => {
let _ = outbound.try_send(AgentCommand::Text(
serde_json::json!({
"type": "task_result_ack",
"task_id": result.task_id,
})
.to_string(),
));
}
Ok(false) => {}
Err(error) => {
tracing::error!(%error, server_id = %identity.server_id, "task result persistence failed");
}
}
}
Some("task_received") => {
let Ok(receipt) = serde_json::from_value::<TaskReceivedMessage>(value) else {
return;
};
if !valid_task_id(&receipt.task_id) {
return;
}
if let Err(error) = crate::db::queries::mark_remote_task_sent(
&state.db,
&receipt.task_id,
&identity.server_id,
)
.await
{
tracing::error!(%error, server_id = %identity.server_id, "task receipt persistence failed");
}
}
_ => {}
@@ -394,6 +435,53 @@ fn valid_batch_id(value: &str) -> bool {
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':'))
}
fn valid_task_id(value: &str) -> bool {
!value.is_empty() && value.len() <= 80 && uuid::Uuid::parse_str(value).is_ok()
}
fn agent_protocol_supported(headers: &HeaderMap) -> bool {
let protocol = headers
.get(AGENT_PROTOCOL_HEADER)
.and_then(|value| value.to_str().ok())
.map(str::trim);
if protocol != Some(AGENT_PROTOCOL_VERSION) {
return false;
}
let capabilities = headers
.get(AGENT_CAPABILITIES_HEADER)
.and_then(|value| value.to_str().ok())
.unwrap_or("")
.split(',')
.map(str::trim)
.collect::<std::collections::HashSet<_>>();
REQUIRED_AGENT_CAPABILITIES
.iter()
.all(|capability| capabilities.contains(capability))
}
pub fn queue_remote_task(connection: &AgentConnection, task: &RemoteTaskInfo) -> bool {
send_task(&connection.sender, task)
}
#[cfg(test)]
mod tests {
use super::*;
use axum::http::HeaderValue;
#[test]
fn requires_the_current_agent_protocol_and_capabilities() {
let mut headers = HeaderMap::new();
assert!(!agent_protocol_supported(&headers));
headers.insert(
AGENT_PROTOCOL_HEADER,
HeaderValue::from_static(AGENT_PROTOCOL_VERSION),
);
headers.insert(
AGENT_CAPABILITIES_HEADER,
HeaderValue::from_static("metrics-v1,config-v1,remote-exec-v1,task-ack-v1"),
);
assert!(agent_protocol_supported(&headers));
headers.insert(AGENT_PROTOCOL_HEADER, HeaderValue::from_static("999"));
assert!(!agent_protocol_supported(&headers));
}
}
scripts/websocket-smoke.mjs
+192 -4
View File
@@ -9,6 +9,11 @@ const expectTaskAssigned = process.env.MONITOR_EXPECT_TASK_ASSIGNED !== "0";
const configOnly = process.env.MONITOR_CONFIG_ONLY === "1";
const expectAgentRejected = process.env.MONITOR_EXPECT_AGENT_REJECTED === "1";
const rotateAgentToken = process.env.MONITOR_ROTATE_AGENT_TOKEN === "1";
const agentProtocolHeaders = {
"X-NodeFlare-Agent-Protocol": "1",
"X-NodeFlare-Agent-Capabilities":
"metrics-v1,config-v1,remote-exec-v1,task-ack-v1",
};
if (!baseUrl || !agentToken || (!expectAgentRejected && (!adminToken || !serverId || !latencyTaskId))) {
throw new Error(
@@ -24,10 +29,95 @@ function websocketUrl(path) {
return url.toString();
}
function openSocket(path, token, headers = {}) {
function decodeBase32(value) {
const alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567";
let buffer = 0;
let bits = 0;
const bytes = [];
for (const character of value.toUpperCase()) {
const index = alphabet.indexOf(character);
if (index < 0) throw new Error("Invalid TOTP secret");
buffer = (buffer << 5) | index;
bits += 5;
if (bits >= 8) {
bits -= 8;
bytes.push((buffer >> bits) & 0xff);
}
}
return new Uint8Array(bytes);
}
async function currentTotp(secret) {
const key = await crypto.subtle.importKey(
"raw",
decodeBase32(secret),
{ name: "HMAC", hash: "SHA-1" },
false,
["sign"],
);
const counter = new Uint8Array(8);
let value = BigInt(Math.floor(Date.now() / 30_000));
for (let index = counter.length - 1; index >= 0; index -= 1) {
counter[index] = Number(value & 0xffn);
value >>= 8n;
}
const digest = new Uint8Array(await crypto.subtle.sign("HMAC", key, counter));
const offset = digest[digest.length - 1] & 0x0f;
const binary =
((digest[offset] & 0x7f) << 24) |
(digest[offset + 1] << 16) |
(digest[offset + 2] << 8) |
digest[offset + 3];
return String(binary % 1_000_000).padStart(6, "0");
}
async function setTotpEnabled(secret, enabled) {
const response = await fetch(
new URL(`/api/admin/2fa/${enabled ? "enable" : "disable"}`, baseUrl),
{
method: "POST",
headers: {
Authorization: `Bearer ${adminToken}`,
"Content-Type": "application/json",
},
body: JSON.stringify({ totp_code: await currentTotp(secret) }),
},
);
if (!response.ok) {
throw new Error(
`${enabled ? "Enable" : "Disable"} TOTP returned HTTP ${response.status}: ${await response.text()}`,
);
}
}
async function waitForRemoteTask(taskId, expectedStatus) {
for (let attempt = 0; attempt < 30; attempt += 1) {
const response = await fetch(
new URL(`/api/admin/remote/task/${encodeURIComponent(taskId)}`, baseUrl),
{ headers: { Authorization: `Bearer ${adminToken}` } },
);
if (!response.ok) {
throw new Error(`Remote task returned HTTP ${response.status}: ${await response.text()}`);
}
const task = await response.json();
if (task.status === expectedStatus) return task;
await new Promise((resolve) => setTimeout(resolve, 100));
}
throw new Error(`Remote task ${taskId} did not reach ${expectedStatus}`);
}
function socketHeaders(path, token, headers = {}, includeAgentProtocol = true) {
return {
Authorization: `Bearer ${token}`,
...(includeAgentProtocol && path === "/api/agent/ws" ? agentProtocolHeaders : {}),
...headers,
};
}
function openSocket(path, token, headers = {}, includeAgentProtocol = true) {
return new Promise((resolve, reject) => {
const socket = new WebSocket(websocketUrl(path), {
headers: { Authorization: `Bearer ${token}`, ...headers },
headers: socketHeaders(path, token, headers, includeAgentProtocol),
});
const timer = setTimeout(() => {
socket.terminate();
@@ -48,10 +138,10 @@ function openSocket(path, token, headers = {}) {
});
}
function expectSocketStatus(path, token, statusCode) {
function expectSocketStatus(path, token, statusCode, headers = {}, includeAgentProtocol = true) {
return new Promise((resolve, reject) => {
const socket = new WebSocket(websocketUrl(path), {
headers: { Authorization: `Bearer ${token}` },
headers: socketHeaders(path, token, headers, includeAgentProtocol),
});
const timer = setTimeout(() => {
socket.terminate();
@@ -187,6 +277,22 @@ function waitForSocketClose(socket, expected) {
});
}
await expectSocketStatus("/api/agent/ws", "invalid-agent-token", 426, {}, false);
await expectSocketStatus(
"/api/agent/ws",
"invalid-agent-token",
426,
{ "X-NodeFlare-Agent-Protocol": "999" },
);
await expectSocketStatus(
"/api/agent/ws",
"invalid-agent-token",
426,
{
"X-NodeFlare-Agent-Capabilities":
"metrics-v1,config-v1,remote-exec-v1",
},
);
await expectSocketStatus("/api/agent/ws", "invalid-agent-token", 401);
if (expectAgentRejected) {
await expectSocketStatus("/api/agent/ws", agentToken, 401);
@@ -195,6 +301,7 @@ if (expectAgentRejected) {
const dashboard = configOnly ? null : await openSocket("/api/ws", adminToken);
let agent;
let enabledTotpSecret = "";
try {
if (dashboard) {
const pong = waitForMessage(dashboard, "pong");
@@ -273,6 +380,80 @@ try {
await closeSocket(rotatedSocket);
process.stdout.write(`${rotatedToken}\n`);
} else if (!configOnly) {
const setupResponse = await fetch(new URL("/api/admin/2fa/setup", baseUrl), {
method: "POST",
headers: { Authorization: `Bearer ${adminToken}` },
});
if (!setupResponse.ok) {
throw new Error(`TOTP setup returned HTTP ${setupResponse.status}: ${await setupResponse.text()}`);
}
const setup = await setupResponse.json();
if (!setup.secret) throw new Error("TOTP setup did not return a secret");
await setTotpEnabled(setup.secret, true);
enabledTotpSecret = setup.secret;
const command = "printf nodeflare-smoke";
const assignedTaskPromise = waitForJsonMessage(
agent,
"remote task",
(message) => message.type === "remote_task" && message.command === command,
);
const createTaskResponse = await fetch(new URL("/api/admin/remote/task", baseUrl), {
method: "POST",
headers: {
Authorization: `Bearer ${adminToken}`,
"Content-Type": "application/json",
},
body: JSON.stringify({
command,
server_ids: [serverId],
totp_code: await currentTotp(setup.secret),
}),
});
if (!createTaskResponse.ok) {
throw new Error(
`Remote task creation returned HTTP ${createTaskResponse.status}: ${await createTaskResponse.text()}`,
);
}
const createdTask = await createTaskResponse.json();
const taskId = createdTask.tasks?.[0]?.task_id;
const assignedTask = await assignedTaskPromise;
if (!taskId || assignedTask.task_id !== taskId) {
throw new Error(`Invalid remote task assignment: ${JSON.stringify(assignedTask)}`);
}
agent.send(JSON.stringify({ type: "task_received", task_id: taskId }));
await waitForRemoteTask(taskId, "sent");
const result = {
type: "task_result",
task_id: taskId,
status: "success",
result: "nodeflare-smoke",
exit_code: 0,
};
const resultAckPromise = waitForJsonMessage(
agent,
"remote task result ACK",
(message) => message.type === "task_result_ack" && message.task_id === taskId,
);
agent.send(JSON.stringify(result));
await resultAckPromise;
const completedTask = await waitForRemoteTask(taskId, "success");
if (completedTask.result !== "nodeflare-smoke" || completedTask.exit_code !== 0) {
throw new Error(`Invalid remote task result: ${JSON.stringify(completedTask)}`);
}
const duplicateAckPromise = waitForJsonMessage(
agent,
"duplicate remote task result ACK",
(message) => message.type === "task_result_ack" && message.task_id === taskId,
);
agent.send(JSON.stringify(result));
await duplicateAckPromise;
await setTotpEnabled(setup.secret, false);
enabledTotpSecret = "";
const wakeHintPromise = waitForJsonMessage(
agent,
"batched overview wake hint",
@@ -460,6 +641,13 @@ try {
}
}
} finally {
if (enabledTotpSecret) {
try {
await setTotpEnabled(enabledTotpSecret, false);
} catch (error) {
console.error(error);
}
}
await Promise.all([
agent ? closeSocket(agent) : Promise.resolve(),
dashboard ? closeSocket(dashboard) : Promise.resolve(),