feat(host-agent): Phase 1c — SteamCMD update + jailed file manager
steam_update func runs SteamCMD per game (rust/conan/soulmask app-ids;
dune rejected), streaming stdout to {instance}.steam_status. Jailed
file manager on {instance}.files.cmd: list/read/write/delete/rename/
mkdir/mkfile/move/copy, all confined to instance root via two-stage
lexical-normalize + canonicalize (defeats ../ traversal AND symlink
escape — incl chained symlinks). Replaces the Go agent's UNJAILED
legacy files API (retired, not ported). 5MiB read cap.
42/42 tests green: 24 filemanager incl 7 jail-escape attempts
(dotdot, deep dotdot, absolute, symlink-inside, direct symlink,
chained symlink), 5 steamcmd app-id (cfg-gated win/linux soulmask).
Jail logic reviewed line-by-line: Path::starts_with is component-wise
(no sibling-prefix bypass), non-existent suffix components can't be
symlinks, leading .. normalizes to / and fails the prefix check.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Vantz Stockwell
@@ -11,6 +11,7 @@ use std::collections::HashSet;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use crate::rcon::RconConfig;
|
||||
use crate::steamcmd::SteamcmdConfig;
|
||||
|
||||
/// Instance ids share the NATS subject namespace with host-level segments.
|
||||
const RESERVED_INSTANCE_IDS: &[&str] = &["host", "cmd", "files", "update", "agent"];
|
||||
@@ -65,6 +66,10 @@ pub struct InstanceConfig {
|
||||
/// Protocol defaults to WebRcon for rust, Source for conan/soulmask.
|
||||
#[serde(default)]
|
||||
pub rcon: Option<RconConfig>,
|
||||
/// SteamCMD update settings. Absent = defaults apply (steamcmd on PATH,
|
||||
/// validate = false).
|
||||
#[serde(default)]
|
||||
pub steamcmd: Option<SteamcmdConfig>,
|
||||
}
|
||||
|
||||
impl InstanceConfig {
|
||||
|
||||
527
corrosion-host-agent/src/filemanager.rs
Normal file
527
corrosion-host-agent/src/filemanager.rs
Normal file
@@ -0,0 +1,527 @@
|
||||
//! Jailed file manager for game-server install directories.
|
||||
//!
|
||||
//! Every path operation is confined to the instance `root` — the directory
|
||||
//! declared as `root` in `[[instance]]` config. A two-stage check (lexical
|
||||
//! Clean + `std::fs::canonicalize`) prevents both `../..` traversals and
|
||||
//! symlink-based escapes: even if an attacker plants a symlink inside the root
|
||||
//! that points outside it, `canonicalize` resolves the target and the prefix
|
||||
//! check catches the escape.
|
||||
//!
|
||||
//! The NATS request/reply contract mirrors the Go companion agent's jailed file
|
||||
//! manager (see `companion-agent/internal/filemanager/`) but uses a simpler
|
||||
//! flat JSON envelope rather than the VueFinder storage-path protocol — the
|
||||
//! Rust agent is the replacement, and the panel's backend talks to whichever
|
||||
//! agent is present.
|
||||
//!
|
||||
//! Subject: `corrosion.{license}.{instance}.files.cmd`
|
||||
//! Request: `{"op":"list"|"read"|"write"|"delete"|"rename"|"mkdir"|"mkfile"|"move"|"copy",
|
||||
//! "path":"rel/path", "dest"?:"...", "content"?:"...", "name"?:"..."}`
|
||||
//! Response: `{"status":"success","data":...}` or `{"status":"error","message":"..."}`
|
||||
|
||||
use anyhow::{bail, Context};
|
||||
use chrono::{DateTime, SecondsFormat, Utc};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::fs;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
/// Maximum size for a `read` operation (5 MiB). Larger files must be
|
||||
/// transferred through a dedicated download endpoint, not the file manager.
|
||||
const MAX_READ_SIZE: u64 = 5 * 1024 * 1024;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Wire types
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub struct FileRequest {
|
||||
pub op: String,
|
||||
/// Relative path within the instance root (the "subject" of the operation).
|
||||
#[serde(default)]
|
||||
pub path: String,
|
||||
/// Destination for `rename`, `move`, `copy` — relative to instance root.
|
||||
#[serde(default)]
|
||||
pub dest: Option<String>,
|
||||
/// Text content for `write`.
|
||||
#[serde(default)]
|
||||
pub content: Option<String>,
|
||||
/// Bare filename for `mkdir` and `mkfile`.
|
||||
#[serde(default)]
|
||||
pub name: Option<String>,
|
||||
}
|
||||
|
||||
/// A single directory entry returned by `list`.
|
||||
#[derive(Debug, Serialize)]
|
||||
pub struct FileEntry {
|
||||
pub name: String,
|
||||
/// Path relative to the instance root, using forward slashes.
|
||||
pub path: String,
|
||||
pub is_dir: bool,
|
||||
/// File size in bytes. Zero for directories.
|
||||
pub size: u64,
|
||||
/// RFC 3339 modification timestamp.
|
||||
pub modified: String,
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Jail helper — the security core of this module
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Resolve `rel` against `root`, then canonicalize to reject any form of
|
||||
/// escape including `../..` traversals and symlinks that point outside root.
|
||||
///
|
||||
/// For paths that do not yet exist (e.g. write targets), we canonicalize the
|
||||
/// nearest existing ancestor and then re-join the remaining components, which
|
||||
/// are lexically-clean because they went through `std::path::Path` building.
|
||||
///
|
||||
/// Returns the absolute, canonicalized path if it is within `root`.
|
||||
pub fn jail(root: &Path, rel: &str) -> anyhow::Result<PathBuf> {
|
||||
// Canonicalize root once to get a stable prefix for comparison.
|
||||
// We do this on every call rather than caching so the function stays
|
||||
// pure and testable without Agent state.
|
||||
let canon_root = fs::canonicalize(root)
|
||||
.with_context(|| format!("canonicalize instance root '{}'", root.display()))?;
|
||||
|
||||
// Build the candidate absolute path. We use Path joining so that an
|
||||
// absolute `rel` (e.g. "/etc/passwd") replaces the root entirely — we
|
||||
// detect and reject that case immediately.
|
||||
let candidate = if rel.is_empty() || rel == "." {
|
||||
root.to_path_buf()
|
||||
} else {
|
||||
let rel_path = Path::new(rel);
|
||||
if rel_path.is_absolute() {
|
||||
bail!(
|
||||
"absolute path '{}' is not allowed; supply a path relative to the instance root",
|
||||
rel
|
||||
);
|
||||
}
|
||||
root.join(rel_path)
|
||||
};
|
||||
|
||||
// Normalize lexically first (removes `..` / `.` without filesystem access).
|
||||
// This is a defence-in-depth step; the authoritative check is below.
|
||||
let lexical = normalize_lexical(&candidate);
|
||||
|
||||
// Canonicalize: resolve symlinks and `..` via the kernel.
|
||||
// For a not-yet-existing path we walk up to the nearest existing ancestor.
|
||||
let canon = canonicalize_lenient(&lexical)?;
|
||||
|
||||
// Authoritative prefix check: the resolved path must be equal to or a
|
||||
// child of the canonicalized root.
|
||||
if canon != canon_root && !canon.starts_with(&canon_root) {
|
||||
bail!(
|
||||
"path '{}' resolves to '{}' which is outside the instance root '{}'",
|
||||
rel,
|
||||
canon.display(),
|
||||
canon_root.display()
|
||||
);
|
||||
}
|
||||
|
||||
Ok(canon)
|
||||
}
|
||||
|
||||
/// Canonicalize a path that may not fully exist yet by walking up to the
|
||||
/// nearest existing ancestor, canonicalizing it, then re-joining the remaining
|
||||
/// (lexically-clean) suffix.
|
||||
fn canonicalize_lenient(path: &Path) -> anyhow::Result<PathBuf> {
|
||||
// Fast path: path already exists.
|
||||
if let Ok(c) = fs::canonicalize(path) {
|
||||
return Ok(c);
|
||||
}
|
||||
|
||||
// Walk up until we find an ancestor that exists.
|
||||
let mut existing = path.to_path_buf();
|
||||
let mut suffix: Vec<std::ffi::OsString> = Vec::new();
|
||||
|
||||
loop {
|
||||
match fs::canonicalize(&existing) {
|
||||
Ok(canon) => {
|
||||
// Re-attach the non-existing suffix.
|
||||
let mut result = canon;
|
||||
for component in suffix.iter().rev() {
|
||||
result = result.join(component);
|
||||
}
|
||||
return Ok(result);
|
||||
}
|
||||
Err(_) => {
|
||||
let file_name = match existing.file_name() {
|
||||
Some(n) => n.to_os_string(),
|
||||
None => bail!("cannot resolve path '{}'", path.display()),
|
||||
};
|
||||
suffix.push(file_name);
|
||||
existing = match existing.parent() {
|
||||
Some(p) => p.to_path_buf(),
|
||||
None => bail!("cannot resolve path '{}'", path.display()),
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Lexically normalize a path (remove `.` and `..` components) without
|
||||
/// touching the filesystem. This mirrors `filepath.Clean` in Go.
|
||||
fn normalize_lexical(path: &Path) -> PathBuf {
|
||||
let mut components: Vec<std::path::Component> = Vec::new();
|
||||
for component in path.components() {
|
||||
match component {
|
||||
std::path::Component::CurDir => {}
|
||||
std::path::Component::ParentDir => {
|
||||
// Only pop a normal component — we cannot pop a root prefix.
|
||||
if matches!(components.last(), Some(std::path::Component::Normal(_))) {
|
||||
components.pop();
|
||||
} else {
|
||||
components.push(component);
|
||||
}
|
||||
}
|
||||
other => components.push(other),
|
||||
}
|
||||
}
|
||||
components.iter().collect()
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Operations
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// List the contents of a directory. Returns an entry per item, sorted
|
||||
/// (directories first, then files, both alphabetical).
|
||||
pub fn list(root: &Path, rel: &str) -> anyhow::Result<Vec<FileEntry>> {
|
||||
let abs = jail(root, rel)?;
|
||||
// Use the canonicalized root as the prefix for relative path computation so
|
||||
// that symlinked root paths (e.g. macOS /var → /private/var) don't cause
|
||||
// strip_prefix to fail and fall back to leaking the absolute path.
|
||||
let canon_root = fs::canonicalize(root)
|
||||
.with_context(|| format!("canonicalize root '{}'", root.display()))?;
|
||||
|
||||
let rd = fs::read_dir(&abs)
|
||||
.with_context(|| format!("read_dir '{}'", abs.display()))?;
|
||||
|
||||
let mut entries: Vec<FileEntry> = Vec::new();
|
||||
for item in rd {
|
||||
let item = item.with_context(|| format!("reading directory entry in '{}'", abs.display()))?;
|
||||
let meta = item.metadata().with_context(|| format!("stat '{}'", item.path().display()))?;
|
||||
|
||||
let name = item.file_name().to_string_lossy().into_owned();
|
||||
let is_dir = meta.is_dir();
|
||||
let size = if is_dir { 0 } else { meta.len() };
|
||||
|
||||
// Build the relative path from the canonicalized root.
|
||||
let entry_abs = item.path();
|
||||
let entry_rel = entry_abs
|
||||
.strip_prefix(&canon_root)
|
||||
.unwrap_or(&entry_abs)
|
||||
.to_string_lossy()
|
||||
.replace('\\', "/");
|
||||
|
||||
let modified = meta
|
||||
.modified()
|
||||
.ok()
|
||||
.map(|t| {
|
||||
let dt: DateTime<Utc> = t.into();
|
||||
dt.to_rfc3339_opts(SecondsFormat::Secs, true)
|
||||
})
|
||||
.unwrap_or_default();
|
||||
|
||||
entries.push(FileEntry { name, path: entry_rel, is_dir, size, modified });
|
||||
}
|
||||
|
||||
// Stable sort: dirs first, then alphabetical within each group.
|
||||
entries.sort_by(|a, b| {
|
||||
b.is_dir.cmp(&a.is_dir).then_with(|| a.name.cmp(&b.name))
|
||||
});
|
||||
|
||||
Ok(entries)
|
||||
}
|
||||
|
||||
/// Read a text file. Capped at `MAX_READ_SIZE` bytes.
|
||||
pub fn read(root: &Path, rel: &str) -> anyhow::Result<String> {
|
||||
let abs = jail(root, rel)?;
|
||||
|
||||
let meta = fs::metadata(&abs)
|
||||
.with_context(|| format!("stat '{}'", abs.display()))?;
|
||||
|
||||
if meta.is_dir() {
|
||||
bail!("'{}' is a directory, not a file", rel);
|
||||
}
|
||||
if meta.len() > MAX_READ_SIZE {
|
||||
bail!(
|
||||
"file '{}' is {} bytes which exceeds the {} byte read limit",
|
||||
rel,
|
||||
meta.len(),
|
||||
MAX_READ_SIZE
|
||||
);
|
||||
}
|
||||
|
||||
fs::read_to_string(&abs).with_context(|| format!("read '{}'", abs.display()))
|
||||
}
|
||||
|
||||
/// Write (create or overwrite) a file. Parent directories are created as
|
||||
/// needed.
|
||||
pub fn write(root: &Path, rel: &str, content: &str) -> anyhow::Result<()> {
|
||||
let abs = jail(root, rel)?;
|
||||
|
||||
if let Some(parent) = abs.parent() {
|
||||
fs::create_dir_all(parent)
|
||||
.with_context(|| format!("create_dir_all '{}'", parent.display()))?;
|
||||
}
|
||||
|
||||
fs::write(&abs, content.as_bytes())
|
||||
.with_context(|| format!("write '{}'", abs.display()))
|
||||
}
|
||||
|
||||
/// Delete a file or directory tree.
|
||||
pub fn delete(root: &Path, rel: &str) -> anyhow::Result<()> {
|
||||
let abs = jail(root, rel)?;
|
||||
|
||||
let meta = fs::metadata(&abs)
|
||||
.with_context(|| format!("stat '{}'", abs.display()))?;
|
||||
|
||||
if meta.is_dir() {
|
||||
fs::remove_dir_all(&abs).with_context(|| format!("remove_dir_all '{}'", abs.display()))
|
||||
} else {
|
||||
fs::remove_file(&abs).with_context(|| format!("remove_file '{}'", abs.display()))
|
||||
}
|
||||
}
|
||||
|
||||
/// Rename/move `rel` to a new bare name (`new_name`) within the same parent.
|
||||
/// `new_name` must not contain path separators.
|
||||
pub fn rename(root: &Path, rel: &str, new_name: &str) -> anyhow::Result<()> {
|
||||
if new_name.is_empty() || new_name == "." || new_name == ".." {
|
||||
bail!("new_name '{}' is not a valid filename", new_name);
|
||||
}
|
||||
if new_name.contains('/') || new_name.contains('\\') {
|
||||
bail!("new_name '{}' must not contain path separators", new_name);
|
||||
}
|
||||
|
||||
let src_abs = jail(root, rel)?;
|
||||
|
||||
// Construct the destination relative path by replacing the filename part
|
||||
// of `rel` with `new_name`. This keeps everything in relative-path space
|
||||
// so we never hand an absolute path to `jail`.
|
||||
let src_rel = Path::new(rel);
|
||||
let dest_rel = match src_rel.parent() {
|
||||
Some(parent) if parent != Path::new("") => {
|
||||
parent.join(new_name).to_string_lossy().replace('\\', "/")
|
||||
}
|
||||
_ => new_name.to_string(),
|
||||
};
|
||||
|
||||
let dest_abs = jail(root, &dest_rel)?;
|
||||
|
||||
fs::rename(&src_abs, &dest_abs)
|
||||
.with_context(|| format!("rename '{}' -> '{}'", src_abs.display(), dest_abs.display()))
|
||||
}
|
||||
|
||||
/// Create a directory (and any missing parents) at `rel`.
|
||||
pub fn mkdir(root: &Path, rel: &str) -> anyhow::Result<()> {
|
||||
let abs = jail(root, rel)?;
|
||||
fs::create_dir_all(&abs).with_context(|| format!("mkdir '{}'", abs.display()))
|
||||
}
|
||||
|
||||
/// Create an empty file at `rel`. Fails if it already exists.
|
||||
pub fn mkfile(root: &Path, rel: &str) -> anyhow::Result<()> {
|
||||
let abs = jail(root, rel)?;
|
||||
|
||||
if let Some(parent) = abs.parent() {
|
||||
fs::create_dir_all(parent)
|
||||
.with_context(|| format!("create_dir_all '{}'", parent.display()))?;
|
||||
}
|
||||
|
||||
let _ = std::fs::OpenOptions::new()
|
||||
.create_new(true)
|
||||
.write(true)
|
||||
.open(&abs)
|
||||
.with_context(|| format!("mkfile '{}'", abs.display()))?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Move `src` to `dest` (both relative to root).
|
||||
pub fn move_path(root: &Path, src: &str, dest: &str) -> anyhow::Result<()> {
|
||||
let src_abs = jail(root, src)?;
|
||||
let dest_abs = jail(root, dest)?;
|
||||
|
||||
if let Some(parent) = dest_abs.parent() {
|
||||
fs::create_dir_all(parent)
|
||||
.with_context(|| format!("create_dir_all '{}'", parent.display()))?;
|
||||
}
|
||||
|
||||
fs::rename(&src_abs, &dest_abs).or_else(|_| {
|
||||
// Cross-device move: copy then delete.
|
||||
copy_recursive(&src_abs, &dest_abs)?;
|
||||
fs::remove_dir_all(&src_abs)
|
||||
.with_context(|| format!("remove source '{}' after cross-device move", src_abs.display()))
|
||||
}).with_context(|| format!("move '{}' -> '{}'", src_abs.display(), dest_abs.display()))
|
||||
}
|
||||
|
||||
/// Copy `src` to `dest` (both relative to root).
|
||||
pub fn copy(root: &Path, src: &str, dest: &str) -> anyhow::Result<()> {
|
||||
let src_abs = jail(root, src)?;
|
||||
let dest_abs = jail(root, dest)?;
|
||||
|
||||
if let Some(parent) = dest_abs.parent() {
|
||||
fs::create_dir_all(parent)
|
||||
.with_context(|| format!("create_dir_all '{}'", parent.display()))?;
|
||||
}
|
||||
|
||||
copy_recursive(&src_abs, &dest_abs)
|
||||
.with_context(|| format!("copy '{}' -> '{}'", src_abs.display(), dest_abs.display()))
|
||||
}
|
||||
|
||||
/// Recursive copy helper (mirrors Go's `copyRecursive`).
|
||||
fn copy_recursive(src: &Path, dest: &Path) -> anyhow::Result<()> {
|
||||
let meta = fs::metadata(src)
|
||||
.with_context(|| format!("stat source '{}'", src.display()))?;
|
||||
|
||||
if meta.is_dir() {
|
||||
fs::create_dir_all(dest)
|
||||
.with_context(|| format!("create_dir_all '{}'", dest.display()))?;
|
||||
|
||||
for entry in fs::read_dir(src)
|
||||
.with_context(|| format!("read_dir '{}'", src.display()))?
|
||||
{
|
||||
let entry = entry?;
|
||||
copy_recursive(&entry.path(), &dest.join(entry.file_name()))?;
|
||||
}
|
||||
} else {
|
||||
fs::copy(src, dest)
|
||||
.with_context(|| format!("copy '{}' -> '{}'", src.display(), dest.display()))?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// NATS request dispatch
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Dispatch a `FileRequest` against `root` and return a JSON `serde_json::Value`
|
||||
/// ready for the NATS reply.
|
||||
pub fn dispatch(root: &Path, req: &FileRequest) -> serde_json::Value {
|
||||
use serde_json::json;
|
||||
|
||||
let result = match req.op.as_str() {
|
||||
"list" => {
|
||||
list(root, &req.path).map(|entries| json!({ "entries": entries }))
|
||||
}
|
||||
"read" => {
|
||||
read(root, &req.path).map(|content| json!({ "content": content }))
|
||||
}
|
||||
"write" => {
|
||||
let content = req.content.as_deref().unwrap_or("");
|
||||
write(root, &req.path, content).map(|_| json!(null))
|
||||
}
|
||||
"delete" => {
|
||||
delete(root, &req.path).map(|_| json!(null))
|
||||
}
|
||||
"rename" => {
|
||||
let new_name = req.name.as_deref().unwrap_or("");
|
||||
rename(root, &req.path, new_name).map(|_| json!(null))
|
||||
}
|
||||
"mkdir" => {
|
||||
mkdir(root, &req.path).map(|_| json!(null))
|
||||
}
|
||||
"mkfile" => {
|
||||
mkfile(root, &req.path).map(|_| json!(null))
|
||||
}
|
||||
"move" => {
|
||||
let dest = req.dest.as_deref().unwrap_or("");
|
||||
move_path(root, &req.path, dest).map(|_| json!(null))
|
||||
}
|
||||
"copy" => {
|
||||
let dest = req.dest.as_deref().unwrap_or("");
|
||||
copy(root, &req.path, dest).map(|_| json!(null))
|
||||
}
|
||||
other => Err(anyhow::anyhow!(
|
||||
"unknown op '{}' (supported: list, read, write, delete, rename, mkdir, mkfile, move, copy)",
|
||||
other
|
||||
)),
|
||||
};
|
||||
|
||||
match result {
|
||||
Ok(data) => json!({ "status": "success", "data": data }),
|
||||
Err(e) => {
|
||||
tracing::warn!("filemanager op='{}' path='{}': {e:#}", req.op, req.path);
|
||||
json!({ "status": "error", "message": format!("{e:#}") })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Subscribe to `corrosion.{license}.{instance}.files.cmd` and serve file
|
||||
/// manager requests for `instance_id` jailed to `root`.
|
||||
///
|
||||
/// This function runs until the agent's cancellation token fires or the NATS
|
||||
/// subscription ends. It is spawned once per instance in `main.rs`.
|
||||
pub async fn run(
|
||||
agent: std::sync::Arc<crate::agent::Agent>,
|
||||
instance_id: String,
|
||||
root: PathBuf,
|
||||
) -> anyhow::Result<()> {
|
||||
use futures::StreamExt;
|
||||
|
||||
let subject = crate::subjects::instance_files_cmd(&agent.cfg.license_id, &instance_id);
|
||||
let mut sub = agent.nats.subscribe(subject.clone()).await?;
|
||||
tracing::info!("file manager handler listening on {subject}");
|
||||
|
||||
let cancel = agent.shutdown.clone();
|
||||
loop {
|
||||
tokio::select! {
|
||||
msg = sub.next() => {
|
||||
match msg {
|
||||
Some(msg) => {
|
||||
let agent = agent.clone();
|
||||
let root = root.clone();
|
||||
let instance_id = instance_id.clone();
|
||||
tokio::spawn(async move { handle(agent, &instance_id, &root, msg).await });
|
||||
}
|
||||
None => {
|
||||
tracing::warn!("file manager subscription ended for '{instance_id}'");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
_ = cancel.cancelled() => {
|
||||
tracing::info!("file manager handler stopping for '{instance_id}'");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn handle(
|
||||
agent: std::sync::Arc<crate::agent::Agent>,
|
||||
instance_id: &str,
|
||||
root: &Path,
|
||||
msg: async_nats::Message,
|
||||
) {
|
||||
let Some(reply) = msg.reply.clone() else {
|
||||
tracing::warn!("file manager message without reply subject ignored (instance '{instance_id}')");
|
||||
return;
|
||||
};
|
||||
|
||||
let response = match serde_json::from_slice::<FileRequest>(&msg.payload) {
|
||||
Ok(req) => {
|
||||
// Blocking fs calls — offload from the async executor.
|
||||
let root = root.to_path_buf();
|
||||
tokio::task::spawn_blocking(move || dispatch(&root, &req))
|
||||
.await
|
||||
.unwrap_or_else(|e| {
|
||||
serde_json::json!({ "status": "error", "message": format!("internal error: {e}") })
|
||||
})
|
||||
}
|
||||
Err(e) => {
|
||||
serde_json::json!({ "status": "error", "message": format!("invalid request payload: {e}") })
|
||||
}
|
||||
};
|
||||
|
||||
let bytes = match serde_json::to_vec(&response) {
|
||||
Ok(b) => b,
|
||||
Err(e) => {
|
||||
tracing::error!("file manager response serialize failed: {e}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
if let Err(e) = agent.nats.publish(reply, bytes.into()).await {
|
||||
tracing::warn!("file manager response publish failed: {e}");
|
||||
}
|
||||
}
|
||||
@@ -15,6 +15,7 @@ use std::sync::Arc;
|
||||
use crate::agent::Agent;
|
||||
use crate::process::ProcessSupervisor;
|
||||
use crate::subjects;
|
||||
use crate::steamcmd;
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct InstanceCommand {
|
||||
@@ -175,10 +176,84 @@ async fn dispatch(
|
||||
}),
|
||||
};
|
||||
}
|
||||
"steam_update" => {
|
||||
// Look up instance config for game name, root, and optional steamcmd
|
||||
// settings. The supervisor only carries process-control state, not
|
||||
// the full config, so we reach into agent.cfg.instances here as the
|
||||
// rcon dispatch does.
|
||||
let inst_cfg = agent.cfg.instances.iter().find(|i| i.id == sup.instance_id);
|
||||
|
||||
let Some(inst_cfg) = inst_cfg else {
|
||||
return json!({
|
||||
"status": "error",
|
||||
"func": "steam_update",
|
||||
"instance_id": sup.instance_id,
|
||||
"message": format!("no config found for instance '{}'", sup.instance_id),
|
||||
});
|
||||
};
|
||||
|
||||
let game = inst_cfg.game.as_str();
|
||||
let root = inst_cfg.root.clone();
|
||||
|
||||
// Resolve steamcmd path and validate flag from config or use defaults.
|
||||
let (steamcmd_path, validate) = match inst_cfg.steamcmd.as_ref() {
|
||||
Some(s) => {
|
||||
let path = s
|
||||
.steamcmd_path
|
||||
.as_ref()
|
||||
.and_then(|p| p.to_str().map(|s| s.to_string()))
|
||||
.unwrap_or_else(|| "steamcmd".to_string());
|
||||
(path, s.validate)
|
||||
}
|
||||
None => ("steamcmd".to_string(), false),
|
||||
};
|
||||
|
||||
let license = agent.cfg.license_id.clone();
|
||||
let instance_id = sup.instance_id.clone();
|
||||
let nats = agent.nats.clone();
|
||||
|
||||
// Publish each progress line to the steam_status subject.
|
||||
let on_progress = move |line: &str| {
|
||||
let subject = subjects::instance_steam_status(&license, &instance_id);
|
||||
let event = json!({
|
||||
"timestamp": Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
|
||||
"instance_id": instance_id,
|
||||
"line": line,
|
||||
});
|
||||
match serde_json::to_vec(&event) {
|
||||
Ok(bytes) => {
|
||||
// Fire-and-forget; the async publish is non-blocking on
|
||||
// the caller side. We create a mini-runtime task via
|
||||
// a oneshot since on_progress is Fn (not async).
|
||||
let nats = nats.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = nats.publish(subject, bytes.into()).await {
|
||||
tracing::warn!("steam_status publish failed: {e}");
|
||||
}
|
||||
});
|
||||
}
|
||||
Err(e) => tracing::error!("steam_status serialize failed: {e}"),
|
||||
}
|
||||
};
|
||||
|
||||
return match steamcmd::update(game, &root, &steamcmd_path, validate, on_progress).await {
|
||||
Ok(()) => json!({
|
||||
"status": "success",
|
||||
"func": "steam_update",
|
||||
"instance_id": sup.instance_id,
|
||||
}),
|
||||
Err(e) => json!({
|
||||
"status": "error",
|
||||
"func": "steam_update",
|
||||
"instance_id": sup.instance_id,
|
||||
"message": format!("{e:#}"),
|
||||
}),
|
||||
};
|
||||
}
|
||||
other => {
|
||||
return json!({
|
||||
"status": "error",
|
||||
"message": format!("unknown func '{other}' (supported: start, stop, restart, status, rcon)"),
|
||||
"message": format!("unknown func '{other}' (supported: start, stop, restart, status, rcon, steam_update)"),
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
@@ -4,11 +4,13 @@
|
||||
pub mod agent;
|
||||
pub mod bus;
|
||||
pub mod config;
|
||||
pub mod filemanager;
|
||||
pub mod hostcmd;
|
||||
pub mod instancecmd;
|
||||
pub mod prober;
|
||||
pub mod process;
|
||||
pub mod rcon;
|
||||
pub mod steamcmd;
|
||||
pub mod subjects;
|
||||
pub mod telemetry;
|
||||
pub mod version;
|
||||
|
||||
@@ -5,7 +5,8 @@
|
||||
//! game adapters arrive in Phase 1+ (see PROTOCOL.md).
|
||||
|
||||
use corrosion_host_agent::{
|
||||
agent, bus, config, hostcmd, instancecmd, prober, process, subjects, telemetry, version,
|
||||
agent, bus, config, filemanager, hostcmd, instancecmd, prober, process, subjects, telemetry,
|
||||
version,
|
||||
};
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
@@ -117,7 +118,7 @@ async fn run(settings: config::Settings) -> Result<()> {
|
||||
}
|
||||
}));
|
||||
}
|
||||
for sup in agent.supervisors.values() {
|
||||
for (instance_id, sup) in &agent.supervisors {
|
||||
{
|
||||
let agent = agent.clone();
|
||||
let sup = sup.clone();
|
||||
@@ -131,6 +132,24 @@ async fn run(settings: config::Settings) -> Result<()> {
|
||||
agent.clone(),
|
||||
sup.clone(),
|
||||
)));
|
||||
// File manager: one handler task per instance, jailed to root.
|
||||
{
|
||||
let agent = agent.clone();
|
||||
let inst_cfg = agent
|
||||
.cfg
|
||||
.instances
|
||||
.iter()
|
||||
.find(|i| &i.id == instance_id)
|
||||
.cloned();
|
||||
if let Some(cfg) = inst_cfg {
|
||||
let id = instance_id.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
if let Err(e) = filemanager::run(agent, id, cfg.root).await {
|
||||
tracing::error!("file manager handler failed: {e:#}");
|
||||
}
|
||||
}));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
wait_for_shutdown_signal().await;
|
||||
|
||||
126
corrosion-host-agent/src/steamcmd.rs
Normal file
126
corrosion-host-agent/src/steamcmd.rs
Normal file
@@ -0,0 +1,126 @@
|
||||
//! SteamCMD update integration for process-managed game instances.
|
||||
//!
|
||||
//! Wraps the `steamcmd` binary to perform an `+app_update` for a given game
|
||||
//! instance, streaming stdout lines to a caller-supplied progress callback so
|
||||
//! the panel can display live update output. The agent already runs a task per
|
||||
//! command in a separate `tokio::spawn`, so the blocking-until-done semantics
|
||||
//! here are intentional — the NATS reply is sent only when SteamCMD exits.
|
||||
//!
|
||||
//! Dune is Docker-image-based and explicitly has no SteamCMD integration — any
|
||||
//! attempt to invoke `update` on a Dune instance returns a clear error rather
|
||||
//! than a silent no-op.
|
||||
|
||||
use std::path::Path;
|
||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||
use tokio::process::Command;
|
||||
|
||||
/// Return the Steam app ID for a given game name, or `None` for Dune (Docker).
|
||||
///
|
||||
/// Soulmask returns the Windows or Linux server app ID depending on the compile
|
||||
/// target so this function is `#[cfg]`-gated at the platform level.
|
||||
pub fn app_id_for_game(game: &str) -> Option<u32> {
|
||||
match game {
|
||||
"rust" => Some(258550),
|
||||
"conan" => Some(443030),
|
||||
"soulmask" => {
|
||||
#[cfg(windows)]
|
||||
{
|
||||
Some(3017310)
|
||||
}
|
||||
#[cfg(not(windows))]
|
||||
{
|
||||
Some(3017300)
|
||||
}
|
||||
}
|
||||
// Dune uses Docker images — SteamCMD has no role here.
|
||||
"dune" => None,
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Configuration controlling SteamCMD behaviour for one instance.
|
||||
/// Serialised as `[instance.steamcmd]` in agent.toml.
|
||||
#[derive(Debug, Clone, serde::Deserialize, Default)]
|
||||
pub struct SteamcmdConfig {
|
||||
/// Absolute or relative path to the `steamcmd` binary.
|
||||
/// Defaults to `"steamcmd"` (resolved via `PATH`) when absent.
|
||||
#[serde(default)]
|
||||
pub steamcmd_path: Option<std::path::PathBuf>,
|
||||
|
||||
/// Whether to pass `validate` to `+app_update`. Adds a file-hash check
|
||||
/// pass that catches corruption at the cost of a longer update time.
|
||||
#[serde(default)]
|
||||
pub validate: bool,
|
||||
}
|
||||
|
||||
/// Run a SteamCMD update for `game` into `install_dir`.
|
||||
///
|
||||
/// - `steamcmd_path`: path to the binary (or `"steamcmd"` to use PATH).
|
||||
/// - `validate`: appends `validate` to the `+app_update` call.
|
||||
/// - `on_progress`: receives each stdout line as it arrives so callers can
|
||||
/// forward progress to the panel in real time.
|
||||
///
|
||||
/// Returns `Ok(())` on a zero exit code, otherwise an error describing the
|
||||
/// failure. Dune is rejected before any process is spawned.
|
||||
pub async fn update(
|
||||
game: &str,
|
||||
install_dir: &Path,
|
||||
steamcmd_path: &str,
|
||||
validate: bool,
|
||||
on_progress: impl Fn(&str),
|
||||
) -> anyhow::Result<()> {
|
||||
use anyhow::Context;
|
||||
|
||||
let app_id = app_id_for_game(game).ok_or_else(|| {
|
||||
anyhow::anyhow!(
|
||||
"dune uses Docker images, not SteamCMD — cannot run app_update for game '{game}'"
|
||||
)
|
||||
})?;
|
||||
|
||||
let install_dir_str = install_dir
|
||||
.to_str()
|
||||
.with_context(|| format!("install_dir '{}' is not valid UTF-8", install_dir.display()))?;
|
||||
|
||||
let mut args: Vec<String> = vec![
|
||||
"+force_install_dir".to_string(),
|
||||
install_dir_str.to_string(),
|
||||
"+login".to_string(),
|
||||
"anonymous".to_string(),
|
||||
"+app_update".to_string(),
|
||||
app_id.to_string(),
|
||||
];
|
||||
if validate {
|
||||
args.push("validate".to_string());
|
||||
}
|
||||
args.push("+quit".to_string());
|
||||
|
||||
tracing::info!(
|
||||
"steamcmd: starting update for game={game} app_id={app_id} install_dir={} validate={validate}",
|
||||
install_dir.display()
|
||||
);
|
||||
|
||||
let mut child = Command::new(steamcmd_path)
|
||||
.args(&args)
|
||||
.stdout(std::process::Stdio::piped())
|
||||
.stderr(std::process::Stdio::null())
|
||||
.spawn()
|
||||
.with_context(|| format!("spawning steamcmd binary '{steamcmd_path}'"))?;
|
||||
|
||||
let stdout = child.stdout.take().expect("stdout was piped");
|
||||
let mut lines = BufReader::new(stdout).lines();
|
||||
|
||||
while let Some(line) = lines.next_line().await.context("reading steamcmd stdout")? {
|
||||
tracing::debug!("steamcmd: {line}");
|
||||
on_progress(&line);
|
||||
}
|
||||
|
||||
let status = child.wait().await.context("waiting for steamcmd to exit")?;
|
||||
if status.success() {
|
||||
tracing::info!("steamcmd: update completed successfully for game={game}");
|
||||
Ok(())
|
||||
} else {
|
||||
let code = status.code().unwrap_or(-1);
|
||||
anyhow::bail!("steamcmd exited with non-zero status {code} for game={game}")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -26,3 +26,14 @@ pub fn instance_cmd(license: &str, instance: &str) -> String {
|
||||
pub fn instance_status(license: &str, instance: &str) -> String {
|
||||
format!("corrosion.{license}.{instance}.status")
|
||||
}
|
||||
|
||||
/// Per-instance SteamCMD progress stream. Lines from `steamcmd` stdout are
|
||||
/// published here so the panel can display live update output.
|
||||
pub fn instance_steam_status(license: &str, instance: &str) -> String {
|
||||
format!("corrosion.{license}.{instance}.steam_status")
|
||||
}
|
||||
|
||||
/// Per-instance file manager command channel (request-reply).
|
||||
pub fn instance_files_cmd(license: &str, instance: &str) -> String {
|
||||
format!("corrosion.{license}.{instance}.files.cmd")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user