hyperhive/hive-c0re/src/priv_client.rs

//! Async client for the `hive-priv` privileged-helper socket.
//!
//! Exposes a standalone async function per operation. Each call opens a
//! fresh connection to `/run/hive/priv.sock`, sends one JSON line, reads
//! the response, and closes. Connection-per-call is intentional: priv
//! calls are infrequent (once per rebuild step), so simplicity wins over
//! a persistent connection.

use anyhow::{Context as _, Result, bail};
use hive_sh4re::priv_proto::{BindMount, JournalOutput, PRIV_SOCK, PrivRequest, PrivResponse};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;

/// Send a single request to `hive-priv` and return the response.
pub async fn call(req: &PrivRequest) -> Result<PrivResponse> {
    let mut stream = UnixStream::connect(PRIV_SOCK)
        .await
        .context("connect to hive-priv socket")?;
    let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n";
    stream
        .write_all(line.as_bytes())
        .await
        .context("send request to hive-priv")?;
    stream.shutdown().await.context("shutdown write half")?;
    let mut resp_line = String::new();
    BufReader::new(stream)
        .read_line(&mut resp_line)
        .await
        .context("read response from hive-priv")?;
    serde_json::from_str(&resp_line).context("parse PrivResponse")
}

pub async fn start_container(name: &str) -> Result<()> {
    ok(call(&PrivRequest::StartContainer {
        name: name.to_owned(),
    })
    .await?)
}

pub async fn stop_container(name: &str) -> Result<()> {
    ok(call(&PrivRequest::StopContainer {
        name: name.to_owned(),
    })
    .await?)
}

pub async fn kill_container(name: &str) -> Result<()> {
    ok(call(&PrivRequest::KillContainer {
        name: name.to_owned(),
    })
    .await?)
}

pub async fn update_container(name: &str) -> Result<(String, String)> {
    check(
        call(&PrivRequest::UpdateContainer {
            name: name.to_owned(),
        })
        .await?,
    )
}

pub async fn create_container(name: &str) -> Result<(String, String)> {
    check(
        call(&PrivRequest::CreateContainer {
            name: name.to_owned(),
        })
        .await?,
    )
}

pub async fn destroy_container(name: &str) -> Result<()> {
    ok(call(&PrivRequest::DestroyContainer {
        name: name.to_owned(),
    })
    .await?)
}

pub async fn list_containers() -> Result<String> {
    let (stdout, _) = check(call(&PrivRequest::ListContainers).await?)?;
    Ok(stdout)
}

/// Read a container's journal via the root helper (`journalctl -M`).
/// Returns `(stdout, stderr)`; a non-zero journalctl exit is reported in
/// `stderr` rather than as an `Err`, so callers can surface either.
#[allow(clippy::too_many_arguments)]
pub async fn read_container_journal(
    container: &str,
    lines: u32,
    boot: bool,
    output: JournalOutput,
    unit: Option<String>,
    priority: Option<String>,
    grep: Option<String>,
    since: Option<String>,
    until: Option<String>,
) -> Result<(String, String)> {
    check(
        call(&PrivRequest::ReadContainerJournal {
            container: container.to_owned(),
            lines,
            boot,
            output,
            unit,
            priority,
            grep,
            since,
            until,
        })
        .await?,
    )
}

pub async fn write_nspawn_flags(container: &str, binds: &[BindMount]) -> Result<()> {
    ok(call(&PrivRequest::WriteNspawnFlags {
        container: container.to_owned(),
        binds: binds.to_vec(),
    })
    .await?)
}

pub async fn write_resource_limits(
    container: &str,
    memory_max: &str,
    cpu_quota: &str,
) -> Result<()> {
    ok(call(&PrivRequest::WriteResourceLimits {
        container: container.to_owned(),
        memory_max: memory_max.to_owned(),
        cpu_quota: cpu_quota.to_owned(),
    })
    .await?)
}

pub async fn remove_service_dropin(container: &str) -> Result<()> {
    ok(call(&PrivRequest::RemoveServiceDropin {
        container: container.to_owned(),
    })
    .await?)
}

pub async fn daemon_reload() -> Result<()> {
    ok(call(&PrivRequest::DaemonReload).await?)
}

pub async fn reload_gateway_nginx() -> Result<()> {
    ok(call(&PrivRequest::ReloadGatewayNginx).await?)
}

pub async fn chown_socket_dir(agent_name: &str, uid: u32, gid: u32) -> Result<()> {
    ok(call(&PrivRequest::ChownSocketDir {
        agent_name: agent_name.to_owned(),
        uid,
        gid,
    })
    .await?)
}

pub async fn chmod_socket_dir(agent_name: &str, mode: u32) -> Result<()> {
    ok(call(&PrivRequest::ChmodSocketDir {
        agent_name: agent_name.to_owned(),
        mode,
    })
    .await?)
}

fn check(resp: PrivResponse) -> Result<(String, String)> {
    if resp.ok {
        Ok((resp.stdout, resp.stderr))
    } else {
        bail!(
            "{}",
            resp.error.as_deref().unwrap_or("hive-priv returned error")
        )
    }
}

fn ok(resp: PrivResponse) -> Result<()> {
    check(resp)?;
    Ok(())
}