use std::fs;
use std::os::unix::fs::{MetadataExt, PermissionsExt};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result, anyhow, bail};
use serde::{Deserialize, Serialize};
use super::{
JobId, JobPhase, ManagedProduct, PeerCredentials, RedeployJob, RedeployOutcome,
ResolvedRelease, SystemdManager, UserInstallContext,
};
const STATE_ROOT: &str = "/var/lib/capulus/jobs";
const RUNTIME_ROOT: &str = "/run/capulus/jobs";
const LOCK_ROOT: &str = "/run/capulus/locks";
#[derive(Clone, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RedeployRequest {
pub job: JobId,
pub product: String,
pub package: String,
pub release: ResolvedRelease,
pub required_user: Option<UserInstallContext>,
}
impl std::fmt::Debug for RedeployRequest {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("RedeployRequest")
.field("job", &self.job)
.field("product", &self.product)
.field("package", &self.package)
.field("version", &self.release.version)
.field(
"required_user",
&self.required_user.as_ref().map(|user| user.uid),
)
.finish_non_exhaustive()
}
}
impl RedeployRequest {
fn validate(&self, product: &ManagedProduct) -> Result<()> {
if self.product != product.name() || self.package != product.package() {
bail!("redeploy request does not match the installed managed product");
}
self.release.validate()?;
if let Some(user) = &self.required_user {
user.revalidate(&product.user_binary().cargo_name)?;
}
Ok(())
}
}
#[derive(Clone, Debug)]
pub struct RedeployCoordinator {
product: Arc<ManagedProduct>,
store: JobStore,
systemd: SystemdManager,
}
impl RedeployCoordinator {
pub fn new(product: Arc<ManagedProduct>) -> Result<Self> {
Ok(Self {
store: JobStore::new(product.name())?,
product,
systemd: SystemdManager,
})
}
pub async fn schedule(
&self,
peer: PeerCredentials,
release: ResolvedRelease,
reinstall_requesting_user: bool,
) -> Result<RedeployOutcome> {
release.validate()?;
let required_user = (reinstall_requesting_user && peer.uid != 0)
.then(|| UserInstallContext::capture(peer, &self.product.user_binary().cargo_name))
.transpose()?
.flatten();
let _lock = capulus_lock(self.product.name())?;
if let Some(active) = self.store.active()? {
let status = self.store.status(active.job)?;
if !status.phase.is_terminal() {
if active.version == release.version
&& active.required_uid == required_user.as_ref().map(|user| user.uid)
{
return Ok(RedeployOutcome {
job: active.job,
unit: status.unit,
version: active.version.to_string(),
started: false,
});
}
bail!(
"redeploy {} is already active for version {}",
active.job,
active.version
);
}
self.store.clear_active(active.job)?;
}
let job = JobId::random();
let unit = SystemdManager::redeploy_unit_name(&self.product, job);
let request = RedeployRequest {
job,
product: self.product.name().to_string(),
package: self.product.package().to_string(),
release,
required_user,
};
request.validate(&self.product)?;
self.store.write_request(&request)?;
self.store.initialize(&request, unit.clone())?;
if let Err(error) = self.systemd.start_redeploy(&self.product, job).await {
self.store.fail(
job,
format!("failed to start transient unit: {error}"),
false,
None,
)?;
self.store.remove_request(job)?;
return Err(error.into());
}
Ok(RedeployOutcome {
job,
unit,
version: request.release.version.to_string(),
started: true,
})
}
pub fn status(&self, job: JobId) -> Result<RedeployJob> {
self.store.status(job)
}
pub fn active(&self) -> Result<Option<RedeployJob>> {
self.store
.active()?
.map(|active| self.store.status(active.job))
.transpose()
}
pub fn take_worker_request(&self, job: JobId) -> Result<RedeployRequest> {
let request = self.store.read_request(job)?;
request.validate(&self.product)?;
self.store.remove_request(job)?;
Ok(request)
}
pub fn update(
&self,
job: JobId,
phase: JobPhase,
detail: impl Into<String>,
system_committed: bool,
) -> Result<()> {
self.store
.update(job, phase, detail.into(), system_committed)
}
pub fn fail(
&self,
job: JobId,
detail: impl Into<String>,
system_committed: bool,
rollback_succeeded: Option<bool>,
) -> Result<()> {
self.store
.fail(job, detail.into(), system_committed, rollback_succeeded)
}
pub fn complete(&self, job: JobId, required_user_reinstalled: Option<bool>) -> Result<()> {
self.store.complete(job, required_user_reinstalled)
}
}
#[derive(Clone, Debug)]
struct JobStore {
product: String,
state_directory: PathBuf,
runtime_directory: PathBuf,
}
impl JobStore {
fn new(product: &str) -> Result<Self> {
if !rustix::process::geteuid().is_root() {
bail!("Capulus job store requires root");
}
let store = Self {
product: product.to_string(),
state_directory: Path::new(STATE_ROOT).join(product),
runtime_directory: Path::new(RUNTIME_ROOT).join(product),
};
ensure_root_directory(Path::new(STATE_ROOT), 0o700)?;
ensure_root_directory(Path::new(RUNTIME_ROOT), 0o700)?;
ensure_root_directory(Path::new(LOCK_ROOT), 0o700)?;
ensure_root_directory(&store.state_directory, 0o700)?;
ensure_root_directory(&store.runtime_directory, 0o700)?;
store.reconcile_boot()?;
Ok(store)
}
fn initialize(&self, request: &RedeployRequest, unit: String) -> Result<()> {
let status = StoredJob {
boot_id: boot_id()?,
status: RedeployJob {
job: request.job,
product: request.product.clone(),
version: request.release.version.to_string(),
unit,
phase: JobPhase::Queued,
detail: "queued for transient systemd execution".to_string(),
system_committed: false,
rollback_succeeded: None,
required_user_reinstalled: None,
},
};
write_json(&self.status_path(request.job), &status, 0o600)?;
write_json(
&self.active_path(),
&ActiveJob {
job: request.job,
version: request.release.version.clone(),
required_uid: request.required_user.as_ref().map(|user| user.uid),
},
0o600,
)
}
fn status(&self, job: JobId) -> Result<RedeployJob> {
let stored: StoredJob = read_json(&self.status_path(job))?;
if stored.status.job != job || stored.status.product != self.product {
bail!("stored redeploy status identity does not match its path");
}
Ok(stored.status)
}
fn active(&self) -> Result<Option<ActiveJob>> {
match read_json(&self.active_path()) {
Ok(active) => Ok(Some(active)),
Err(error)
if error
.downcast_ref::<std::io::Error>()
.is_some_and(|error| error.kind() == std::io::ErrorKind::NotFound) =>
{
Ok(None)
}
Err(error) => Err(error),
}
}
fn clear_active(&self, job: JobId) -> Result<()> {
if self.active()?.is_some_and(|active| active.job == job) {
remove_file(&self.active_path())?;
}
Ok(())
}
fn write_request(&self, request: &RedeployRequest) -> Result<()> {
let mut bytes = Vec::new();
ciborium::into_writer(request, &mut bytes).context("failed to encode redeploy request")?;
crate::store::atomic_write(&self.request_path(request.job), &bytes, Some(0o600), None)
.context("failed to persist redeploy request")
}
fn read_request(&self, job: JobId) -> Result<RedeployRequest> {
let path = self.request_path(job);
validate_private_root_file(&path)?;
let bytes = fs::read(path).context("failed to read redeploy request")?;
ciborium::from_reader(bytes.as_slice()).context("failed to decode redeploy request")
}
fn remove_request(&self, job: JobId) -> Result<()> {
remove_file(&self.request_path(job))
}
fn update(
&self,
job: JobId,
phase: JobPhase,
detail: String,
system_committed: bool,
) -> Result<()> {
let mut stored: StoredJob = read_json(&self.status_path(job))?;
validate_transition(&stored.status.phase, &phase)?;
stored.status.phase = phase;
stored.status.detail = bounded_detail(detail);
stored.status.system_committed = system_committed;
write_json(&self.status_path(job), &stored, 0o600)
}
fn fail(
&self,
job: JobId,
detail: String,
system_committed: bool,
rollback_succeeded: Option<bool>,
) -> Result<()> {
let mut stored: StoredJob = read_json(&self.status_path(job))?;
if stored.status.phase.is_terminal() {
bail!("cannot fail terminal redeploy job {job}");
}
stored.status.phase = JobPhase::Failed;
stored.status.detail = bounded_detail(detail);
stored.status.system_committed = system_committed;
stored.status.rollback_succeeded = rollback_succeeded;
write_json(&self.status_path(job), &stored, 0o600)?;
self.clear_active(job)
}
fn complete(&self, job: JobId, required_user_reinstalled: Option<bool>) -> Result<()> {
let mut stored: StoredJob = read_json(&self.status_path(job))?;
validate_transition(&stored.status.phase, &JobPhase::Complete)?;
stored.status.phase = JobPhase::Complete;
stored.status.detail = "managed redeploy completed".to_string();
stored.status.system_committed = true;
stored.status.required_user_reinstalled = required_user_reinstalled;
write_json(&self.status_path(job), &stored, 0o600)?;
self.clear_active(job)
}
fn reconcile_boot(&self) -> Result<()> {
let Some(active) = self.active()? else {
return Ok(());
};
let stored: StoredJob = read_json(&self.status_path(active.job))?;
if !stored.status.phase.is_terminal() && stored.boot_id != boot_id()? {
self.fail(
active.job,
"redeploy was interrupted by a system reboot".to_string(),
stored.status.system_committed,
None,
)?;
}
Ok(())
}
fn active_path(&self) -> PathBuf {
self.state_directory.join("active.json")
}
fn status_path(&self, job: JobId) -> PathBuf {
self.state_directory.join(format!("{job}.json"))
}
fn request_path(&self, job: JobId) -> PathBuf {
self.runtime_directory.join(format!("{job}.cbor"))
}
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct StoredJob {
boot_id: String,
status: RedeployJob,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct ActiveJob {
job: JobId,
version: semver::Version,
required_uid: Option<u32>,
}
fn validate_transition(current: &JobPhase, next: &JobPhase) -> Result<()> {
if current.is_terminal() {
bail!("cannot transition a terminal redeploy job");
}
if next != &JobPhase::Failed && phase_rank(next) <= phase_rank(current) {
bail!("invalid redeploy phase transition from {current:?} to {next:?}");
}
Ok(())
}
fn phase_rank(phase: &JobPhase) -> u8 {
match phase {
JobPhase::Queued => 0,
JobPhase::Preparing => 1,
JobPhase::Toolchain => 2,
JobPhase::Resolving => 3,
JobPhase::Building => 4,
JobPhase::Validating => 5,
JobPhase::Staging => 6,
JobPhase::CommittingSystem => 7,
JobPhase::RestartingAgent => 8,
JobPhase::ReinstallingUser => 9,
JobPhase::Complete => 10,
JobPhase::Failed => u8::MAX,
}
}
fn capulus_lock(product: &str) -> Result<crate::InvocationLock> {
crate::acquire_named_in(LOCK_ROOT, &format!("{product}-schedule"), true)
.context("failed to acquire Capulus scheduling lock")
}
fn ensure_root_directory(path: &Path, mode: u32) -> Result<()> {
match fs::symlink_metadata(path) {
Ok(metadata)
if !metadata.file_type().is_dir() || metadata.uid() != 0 || metadata.gid() != 0 =>
{
bail!(
"Capulus directory is not a root-owned real directory: {}",
path.display()
)
}
Ok(_) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
fs::create_dir_all(path)
.with_context(|| format!("failed to create {}", path.display()))?;
}
Err(error) => {
return Err(error).with_context(|| format!("failed to inspect {}", path.display()));
}
}
fs::set_permissions(path, fs::Permissions::from_mode(mode))
.with_context(|| format!("failed to chmod {}", path.display()))
}
fn write_json<T: Serialize>(path: &Path, value: &T, mode: u32) -> Result<()> {
let bytes = serde_json::to_vec(value).context("failed to encode Capulus job state")?;
crate::store::atomic_write(path, &bytes, Some(mode), None)
.with_context(|| format!("failed to persist {}", path.display()))
}
fn read_json<T: for<'de> Deserialize<'de>>(path: &Path) -> Result<T> {
validate_private_root_file(path)?;
serde_json::from_slice(&fs::read(path)?)
.with_context(|| format!("failed to decode {}", path.display()))
}
fn validate_private_root_file(path: &Path) -> Result<()> {
let metadata = fs::symlink_metadata(path)?;
if !metadata.file_type().is_file()
|| metadata.uid() != 0
|| metadata.gid() != 0
|| metadata.mode() & 0o077 != 0
{
bail!(
"Capulus state is not a private root-owned regular file: {}",
path.display()
);
}
Ok(())
}
fn remove_file(path: &Path) -> Result<()> {
match fs::remove_file(path) {
Ok(()) => Ok(()),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(error).with_context(|| format!("failed to remove {}", path.display())),
}
}
fn boot_id() -> Result<String> {
let value = fs::read_to_string("/proc/sys/kernel/random/boot_id")
.context("failed to read kernel boot ID")?;
let value = value.trim();
if value.len() != 36
|| !value
.bytes()
.all(|byte| byte.is_ascii_hexdigit() || byte == b'-')
{
return Err(anyhow!("kernel boot ID is malformed"));
}
Ok(value.to_string())
}
fn bounded_detail(mut detail: String) -> String {
const MAX_DETAIL_CHARS: usize = 2048;
if detail.chars().count() <= MAX_DETAIL_CHARS {
return detail;
}
detail = detail.chars().take(MAX_DETAIL_CHARS - 1).collect();
detail.push('…');
detail
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn phase_transitions_are_forward_only() {
assert!(validate_transition(&JobPhase::Queued, &JobPhase::Building).is_ok());
assert!(validate_transition(&JobPhase::Building, &JobPhase::Failed).is_ok());
assert!(validate_transition(&JobPhase::Building, &JobPhase::Preparing).is_err());
assert!(validate_transition(&JobPhase::Complete, &JobPhase::Failed).is_err());
}
#[test]
fn public_job_detail_is_bounded() {
let detail = bounded_detail("x".repeat(3000));
assert_eq!(detail.chars().count(), 2048);
assert!(detail.ends_with('…'));
}
}