diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..f92cee4 --- /dev/null +++ b/.github/dependabot.yml @@ -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 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 5478e95..d85e80d 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -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: diff --git a/README.md b/README.md index 9161546..8112279 100644 --- a/README.md +++ b/README.md @@ -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 密钥,也应通过权限受控的方式单独备份。 ## 架构 diff --git a/agent/agent.sh b/agent/agent.sh index e47a3a0..0b0bafa 100644 --- a/agent/agent.sh +++ b/agent/agent.sh @@ -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 已卸载" } diff --git a/agent/install-freebsd.sh b/agent/install-freebsd.sh index 5eca97d..28de1f1 100755 --- a/agent/install-freebsd.sh +++ b/agent/install-freebsd.sh @@ -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 diff --git a/agent/install-macos.sh b/agent/install-macos.sh index c7e0a0c..271e022 100644 --- a/agent/install-macos.sh +++ b/agent/install-macos.sh @@ -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 diff --git a/agent/install.ps1 b/agent/install.ps1 index c0c5199..39a4c5b 100644 --- a/agent/install.ps1 +++ b/agent/install.ps1 @@ -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) { diff --git a/agent/src/main.rs b/agent/src/main.rs index 7cdad6e..03f6e28 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -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; type Result = std::result::Result; @@ -1755,6 +1761,14 @@ fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option) 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, } +#[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, +} + +#[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, @@ -2039,12 +2075,37 @@ fn execute_remote_task(task: &RemoteTaskMessage) -> TaskResultMessage { struct RemoteExecutor { task_tx: mpsc::SyncSender, result_rx: mpsc::Receiver, - seen: HashSet, - seen_order: VecDeque, + active: HashSet, + completed: HashMap, + task_order: VecDeque, + last_sent: HashMap, + journal_path: PathBuf, + disabled: bool, } impl RemoteExecutor { fn new() -> Result { + Self::new_at(remote_task_journal_path()?) + } + + fn new_at(journal_path: PathBuf) -> Result { + 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::(16); let (result_tx, result_rx) = mpsc::channel::(); thread::Builder::new() @@ -2056,59 +2117,289 @@ impl RemoteExecutor { } } })?; - Ok(Self { + let interrupted = active.drain().collect::>(); + 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 { - 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 + '_ { - self.result_rx.try_iter() + fn drain_completed(&mut self) { + let results = self.result_rx.try_iter().collect::>(); + 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 { + 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::>(); + write_remote_task_journal(&self.journal_path, entries) } } -fn flush_remote_results( - socket: &mut LiveSocket, - executor: &RemoteExecutor, - pending: &mut VecDeque, -) -> 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 { + 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::(&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) -> 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 { 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 { } Ok(LiveRead::RemoteTask(task)) } + Some("task_result_ack") => { + let Ok(ack) = serde_json::from_value::(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 { } } +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>>, remote_executor: &mut RemoteExecutor, - pending_remote_results: &mut VecDeque, ) -> Result { 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, now: i64) { } } -fn pending_spool_path() -> Result { +fn agent_state_directory() -> Result { 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 { 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 { + Ok(agent_state_directory()?.join("pending.jsonl")) +} + +fn remote_task_journal_path() -> Result { + Ok(agent_state_directory()?.join("remote-tasks.json")) +} + fn load_pending_spool(path: &Path) -> Result> { 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(&[ diff --git a/backend/README.md b/backend/README.md index 81df908..37c3e06 100644 --- a/backend/README.md +++ b/backend/README.md @@ -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`。面板服务不创建额外的系统用户。 ## 主要端点 diff --git a/backend/config.example.toml b/backend/config.example.toml index aff4d08..8067082 100644 --- a/backend/config.example.toml +++ b/backend/config.example.toml @@ -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 站点密钥(公开) diff --git a/backend/migrations/postgres/01_initial.sql b/backend/migrations/postgres/01_initial.sql index 3e1bbcf..d1387c8 100644 --- a/backend/migrations/postgres/01_initial.sql +++ b/backend/migrations/postgres/01_initial.sql @@ -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 '{}' ); diff --git a/backend/migrations/sqlite/01_initial.sql b/backend/migrations/sqlite/01_initial.sql index 9611443..1b4adac 100644 --- a/backend/migrations/sqlite/01_initial.sql +++ b/backend/migrations/sqlite/01_initial.sql @@ -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; diff --git a/backend/src/config.rs b/backend/src/config.rs index bd5d9a3..be6e950 100644 --- a/backend/src/config.rs +++ b/backend/src/config.rs @@ -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 { + 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 { #[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()); + } } diff --git a/backend/src/db/mod.rs b/backend/src/db/mod.rs index 5f68056..6303a22 100644 --- a/backend/src/db/mod.rs +++ b/backend/src/db/mod.rs @@ -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(()) diff --git a/backend/src/db/queries.rs b/backend/src/db/queries.rs index 37fe687..7bc4abd 100644 --- a/backend/src/db/queries.rs +++ b/backend/src/db/queries.rs @@ -1543,7 +1543,7 @@ pub async fn pending_remote_tasks(db: &Database, server_id: &str) -> Result Result 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> { @@ -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::>(); - 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] diff --git a/backend/src/main.rs b/backend/src/main.rs index 125d790..fd9f832 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -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 [ diff --git a/backend/src/routes/remote.rs b/backend/src/routes/remote.rs index 7e8a7e8..e9b7182 100644 --- a/backend/src/routes/remote.rs +++ b/backend/src/routes/remote.rs @@ -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, diff --git a/backend/src/websocket/agent.rs b/backend/src/websocket/agent.rs index 859e326..9a5061e 100644 --- a/backend/src/websocket/agent.rs +++ b/backend/src/websocket/agent.rs @@ -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, } +#[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>, @@ -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::(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::(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::>(); + 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)); + } +} diff --git a/scripts/websocket-smoke.mjs b/scripts/websocket-smoke.mjs index 5935502..638ae67 100644 --- a/scripts/websocket-smoke.mjs +++ b/scripts/websocket-smoke.mjs @@ -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(),