Menu
AkurAI-Build
publicLatest change 30e6b8c347048d64c2c048dfd05a8ded219d37f6 - Host authenticated Git repositories over Smart HTTP with repo host/sync CLI by Ólafur Búi Ólafsson
use std::{
collections::{BTreeMap, BTreeSet},
env,
fs::{self, File, OpenOptions},
io::{Read, Write},
path::{Path, PathBuf},
process::{Command, ExitStatus, Stdio},
thread,
time::{Duration, Instant},
};
use anyhow::{Context, Result, bail, ensure};
use serde::Serialize;
use sha2::{Digest, Sha256};
use crate::{
config::{self, JobSpec},
db::{Artifact, Database, RunDetail},
};
const MAX_LOG_BYTES: usize = 2 * 1024 * 1024;
const MAX_ARTIFACT_FILES: usize = 10_000;
const MAX_ARTIFACT_BYTES: u64 = 512 * 1024 * 1024;
#[derive(Clone)]
pub struct Runner {
database: Database,
root: PathBuf,
allow_native: bool,
}
#[derive(Serialize)]
struct DeploymentArtifact<'a> {
id: i64,
name: &'a str,
sha256: &'a str,
bytes: i64,
}
struct ProcessResult {
status: ExitStatus,
logs: String,
timed_out: bool,
}
impl Runner {
pub fn new(database: Database, root: PathBuf, allow_native: bool) -> Result<Self> {
for path in [
root.join("workspaces"),
root.join("artifacts"),
root.join("cache"),
] {
fs::create_dir_all(&path)?;
private_directory(&path)?;
}
Ok(Self {
database,
root,
allow_native,
})
}
pub fn queue(
&self,
repository: &str,
git_ref: Option<&str>,
commit: Option<&str>,
trigger: &str,
) -> Result<i64> {
let repository = self.database.repository(repository)?;
let git_ref = git_ref.unwrap_or(&repository.default_branch);
config::validate_ref(git_ref)?;
if let Some(commit) = commit {
config::validate_commit(commit)?;
}
self.database
.create_run(repository.id, git_ref, commit, trigger)
}
pub fn process(&self, run_id: i64) -> Result<RunDetail> {
if !self.database.claim_run(run_id)? {
return self.database.detail(run_id);
}
self.process_claimed_run(run_id)
}
pub fn process_claimed_run(&self, run_id: i64) -> Result<RunDetail> {
ensure!(
self.database.run(run_id)?.status == "running",
"run {run_id} was not claimed"
);
if let Err(error) = self.execute_claimed(run_id) {
let message = bounded(&format!("{error:#}"), 16 * 1024);
self.database.finish_run(run_id, "failed", Some(&message))?;
}
self.database.detail(run_id)
}
fn execute_claimed(&self, run_id: i64) -> Result<()> {
let run = self.database.run(run_id)?;
let repository = self.database.repository(&run.repository)?;
let workspace = self.root.join("workspaces").join(run_id.to_string());
if self.database.jobs(run_id)?.is_empty() {
if workspace.exists() {
fs::remove_dir_all(&workspace)?;
}
fs::create_dir_all(&workspace)?;
let commit = checkout(
&repository.url,
&run.git_ref,
run.commit_sha.as_deref(),
&workspace,
)?;
self.database.set_run_commit(run_id, &commit)?;
let pipeline_path = [workspace.join(".akurai.yml"), workspace.join("akurai.yml")]
.into_iter()
.find(|path| path.is_file())
.context("repository has no .akurai.yml")?;
ensure!(
pipeline_path.metadata()?.len() <= 256 * 1024,
"pipeline is too large"
);
let source = fs::read_to_string(pipeline_path)?;
let jobs = config::parse(&source)?;
self.database.insert_jobs(run_id, &jobs)?;
}
loop {
let jobs = self.database.jobs(run_id)?;
if jobs.iter().all(|job| {
matches!(
job.status.as_str(),
"succeeded" | "failed" | "skipped" | "canceled" | "interrupted"
)
}) {
let failed = jobs.iter().any(|job| {
matches!(job.status.as_str(), "failed" | "canceled" | "interrupted")
});
self.database.finish_run(
run_id,
if failed { "failed" } else { "succeeded" },
None,
)?;
if !failed {
let _ = fs::remove_dir_all(&workspace);
}
return Ok(());
}
let mut progressed = false;
for job in &jobs {
if job.status != "queued" {
continue;
}
let dependencies = jobs
.iter()
.filter(|candidate| job.needs.contains(&candidate.base_name))
.collect::<Vec<_>>();
if dependencies.iter().any(|dependency| {
matches!(
dependency.status.as_str(),
"failed" | "skipped" | "canceled" | "interrupted"
)
}) {
self.database.set_job_status(
job.id,
"skipped",
Some("dependency did not succeed"),
None,
)?;
progressed = true;
continue;
}
if !dependencies
.iter()
.all(|dependency| dependency.status == "succeeded")
{
continue;
}
if job.approval_required && job.approved_at.is_none() {
self.database
.set_job_status(job.id, "waiting", None, None)?;
progressed = true;
continue;
}
self.execute_job(run_id, job.id, &workspace)?;
progressed = true;
break;
}
if !progressed {
let waiting = self
.database
.jobs(run_id)?
.iter()
.any(|job| job.status == "waiting");
if waiting {
self.database.finish_run(run_id, "waiting", None)?;
return Ok(());
}
bail!("pipeline made no progress");
}
}
}
fn execute_job(&self, run_id: i64, job_id: i64, workspace: &Path) -> Result<()> {
let run = self.database.run(run_id)?;
let jobs = self.database.jobs(run_id)?;
let job = jobs
.iter()
.find(|job| job.id == job_id)
.context("job disappeared")?;
let spec: JobSpec = serde_json::from_str(&job.spec_json)?;
let branch = run
.git_ref
.strip_prefix("refs/heads/")
.unwrap_or(&run.git_ref);
if !spec.branches.is_empty() && !spec.branches.iter().any(|allowed| allowed == branch) {
self.database.set_job_status(
job_id,
"skipped",
Some("Git ref is not allowed for this job"),
None,
)?;
return Ok(());
}
if spec.executor == "native" && !self.allow_native {
self.database.set_job_status(job_id, "failed", Some("native execution is disabled; use a trusted repository and AKURAI_ALLOW_NATIVE=1"), None)?;
return Ok(());
}
let artifacts = self.database.artifacts(run_id)?;
let dependency_ids = jobs
.iter()
.filter(|candidate| spec.needs.contains(&candidate.base_name))
.map(|candidate| candidate.id)
.collect::<BTreeSet<_>>();
let promoted = artifacts
.iter()
.filter(|artifact| dependency_ids.contains(&artifact.job_id))
.collect::<Vec<_>>();
let deployment = if let Some(environment) = &spec.environment {
let manifest = promoted
.iter()
.map(|artifact| DeploymentArtifact {
id: artifact.id,
name: &artifact.name,
sha256: &artifact.sha256,
bytes: artifact.bytes,
})
.collect::<Vec<_>>();
Some(self.database.begin_deployment(
run_id,
job_id,
environment,
&serde_json::to_string(&manifest)?,
)?)
} else {
None
};
self.database
.set_job_status(job_id, "running", None, None)?;
let execution = self.execute(&spec, workspace, run_id, job_id);
match execution {
Ok(result) if result.status.success() && !result.timed_out => {
if let Err(error) =
self.collect_artifacts(run_id, job_id, workspace, &spec.artifacts)
{
let logs = format!("{}\nartifact error: {error:#}", result.logs);
self.database.set_job_status(
job_id,
"failed",
Some(&logs),
result.status.code(),
)?;
if let Some(id) = deployment {
self.database.finish_deployment(id, "failed")?;
}
} else {
self.database.set_job_status(
job_id,
"succeeded",
Some(&result.logs),
result.status.code(),
)?;
if let Some(id) = deployment {
self.database.finish_deployment(id, "succeeded")?;
}
}
}
Ok(result) => {
let logs = if result.timed_out {
format!("{}\njob timed out", result.logs)
} else {
result.logs
};
self.database.set_job_status(
job_id,
"failed",
Some(&logs),
result.status.code(),
)?;
if let Some(id) = deployment {
self.database.finish_deployment(id, "failed")?;
}
}
Err(error) => {
self.database.set_job_status(
job_id,
"failed",
Some(&bounded(&format!("{error:#}"), MAX_LOG_BYTES)),
None,
)?;
if let Some(id) = deployment {
self.database.finish_deployment(id, "failed")?;
}
}
}
Ok(())
}
fn execute(
&self,
spec: &JobSpec,
workspace: &Path,
run_id: i64,
job_id: i64,
) -> Result<ProcessResult> {
let secrets = declared_secrets(&spec.secrets)?;
let matrix = matrix_environment(&spec.matrix);
if spec.executor == "native" {
let mut command = native_shell(spec.shell.as_deref(), &spec.command);
command.current_dir(workspace).env_clear();
copy_host_environment(&mut command);
command.env("HOME", workspace.join(".home"));
for (name, value) in matrix.iter().chain(&secrets) {
command.env(name, value);
}
run_process(command, spec.timeout, &secrets)
} else {
self.execute_docker(spec, workspace, run_id, job_id, &matrix, &secrets)
}
}
fn execute_docker(
&self,
spec: &JobSpec,
workspace: &Path,
run_id: i64,
job_id: i64,
matrix: &BTreeMap<String, String>,
secrets: &BTreeMap<String, String>,
) -> Result<ProcessResult> {
let name = format!("akurai-{run_id}-{job_id}");
let mut command = Command::new("docker");
command.args([
"run",
"--rm",
"--name",
&name,
"--pull=missing",
"--cap-drop=ALL",
"--security-opt=no-new-privileges",
"--pids-limit=512",
"--memory=4g",
"--cpus=2",
"--read-only",
"--tmpfs=/tmp:rw,nosuid,nodev,size=512m",
"--workdir=/workspace",
]);
if !spec.network {
command.arg("--network=none");
}
if let Some(platform) = spec.matrix.get("platform") {
command.args(["--platform", platform]);
}
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
let metadata = workspace.metadata()?;
command.args(["--user", &format!("{}:{}", metadata.uid(), metadata.gid())]);
}
command.args(["--volume", &format!("{}:/workspace", workspace.display())]);
let repository = self.database.run(run_id)?.repository;
for path in &spec.cache {
let host = self
.root
.join("cache")
.join(safe_component(&repository))
.join(safe_component(path));
fs::create_dir_all(&host)?;
fs::create_dir_all(workspace.join(path))?;
command.args(["--volume", &format!("{}:/workspace/{path}", host.display())]);
}
command.args(["--env", "HOME=/workspace/.home"]);
for name in matrix.keys().chain(secrets.keys()) {
command.args(["--env", name]);
}
let image = spec.image.as_deref().context("docker job has no image")?;
command.arg(image);
match spec.shell.as_deref().unwrap_or("sh") {
"cmd" => command.args(["cmd.exe", "/d", "/s", "/c", &spec.command]),
"powershell" => command.args([
"powershell",
"-NoProfile",
"-NonInteractive",
"-Command",
&spec.command,
]),
"sh" => command.args(["sh", "-eu", "-c", &spec.command]),
other => bail!("unsupported shell {other}"),
};
command.env_clear();
copy_host_environment(&mut command);
for (key, value) in matrix.iter().chain(secrets) {
command.env(key, value);
}
let result = run_process(command, spec.timeout, secrets)?;
if result.timed_out {
let _ = Command::new("docker")
.args(["rm", "--force", &name])
.output();
}
Ok(result)
}
fn collect_artifacts(
&self,
run_id: i64,
job_id: i64,
workspace: &Path,
patterns: &[String],
) -> Result<()> {
if patterns.is_empty() {
return Ok(());
}
let workspace = workspace.canonicalize()?;
let artifact_root = self.root.join("artifacts");
fs::create_dir_all(&artifact_root)?;
let existing = self.database.artifacts(run_id)?;
let mut bytes = existing.iter().try_fold(0_u64, |total, artifact| {
Ok::<_, anyhow::Error>(total + u64::try_from(artifact.bytes)?)
})?;
let mut count = existing.len();
let mut files = BTreeSet::new();
for pattern in patterns {
let full = workspace.join(pattern).to_string_lossy().into_owned();
for matched in
glob::glob(&full).with_context(|| format!("invalid artifact glob {pattern}"))?
{
collect_files(&matched?, &mut files)?;
}
}
for source in files {
let canonical = source.canonicalize()?;
ensure!(
canonical.starts_with(&workspace),
"artifact escapes workspace"
);
ensure!(
!source.symlink_metadata()?.file_type().is_symlink(),
"artifact symlinks are not allowed"
);
let relative = canonical.strip_prefix(&workspace)?;
let size = canonical.metadata()?.len();
bytes = bytes.checked_add(size).context("artifact size overflow")?;
count += 1;
ensure!(
bytes <= MAX_ARTIFACT_BYTES && count <= MAX_ARTIFACT_FILES,
"artifact limits exceeded"
);
let relative_store = PathBuf::from("artifacts")
.join(run_id.to_string())
.join(job_id.to_string())
.join(relative);
let destination = self.root.join(&relative_store);
let parent = destination.parent().context("artifact has no parent")?;
fs::create_dir_all(parent)?;
let temporary =
destination.with_extension(format!("akurai-tmp-{}", std::process::id()));
let (digest, copied) = copy_hashed(&canonical, &temporary)?;
fs::rename(&temporary, &destination)?;
self.database.add_artifact(
run_id,
job_id,
&relative.to_string_lossy(),
&relative_store.to_string_lossy(),
&digest,
copied,
)?;
}
Ok(())
}
pub fn artifact_path(&self, artifact: &Artifact) -> Result<PathBuf> {
let root = self.root.canonicalize()?;
let path = self.root.join(&artifact.path).canonicalize()?;
ensure!(
path.starts_with(&root) && path.is_file(),
"artifact path is invalid"
);
let (digest, bytes) = hash_file(&path)?;
ensure!(
digest == artifact.sha256 && i64::try_from(bytes)? == artifact.bytes,
"artifact integrity check failed"
);
Ok(path)
}
}
fn checkout(url: &str, git_ref: &str, requested: Option<&str>, workspace: &Path) -> Result<String> {
run_git(workspace, ["init"])?;
run_git(workspace, ["remote", "add", "origin", url])?;
run_git(workspace, ["fetch", "--depth=64", "origin", git_ref])?;
let branch_tip = git_output(workspace, ["rev-parse", "FETCH_HEAD"])?;
let commit = if let Some(commit) = requested {
run_git(workspace, ["fetch", "--depth=64", "origin", commit])?;
let status = git_status(
workspace,
["merge-base", "--is-ancestor", commit, &branch_tip],
)?;
ensure!(
status.success(),
"requested commit is not reachable from fetched ref"
);
commit.to_owned()
} else {
branch_tip
};
run_git(workspace, ["checkout", "--detach", &commit])?;
git_output(workspace, ["rev-parse", "HEAD"])
}
fn run_git<const N: usize>(cwd: &Path, arguments: [&str; N]) -> Result<()> {
let output = crate::git_process::command(cwd).args(arguments).output()?;
ensure!(
output.status.success(),
"git failed: {}",
bounded(&String::from_utf8_lossy(&output.stderr), 16 * 1024)
);
Ok(())
}
fn git_output<const N: usize>(cwd: &Path, arguments: [&str; N]) -> Result<String> {
crate::git_process::output(cwd, arguments)
}
fn git_status<const N: usize>(cwd: &Path, arguments: [&str; N]) -> Result<ExitStatus> {
Ok(crate::git_process::command(cwd).args(arguments).status()?)
}
fn native_shell(shell: Option<&str>, script: &str) -> Command {
#[cfg(windows)]
{
let mut command = Command::new(match shell.unwrap_or("cmd") {
"powershell" => "powershell",
_ => "cmd.exe",
});
if shell == Some("powershell") {
command.args(["-NoProfile", "-NonInteractive", "-Command", script]);
} else {
command.args(["/d", "/s", "/c", script]);
}
command
}
#[cfg(not(windows))]
{
let _ = shell;
let mut command = Command::new("/bin/sh");
command.args(["-eu", "-c", script]);
command
}
}
fn run_process(
mut command: Command,
timeout_seconds: u64,
secrets: &BTreeMap<String, String>,
) -> Result<ProcessResult> {
command.stdout(Stdio::piped()).stderr(Stdio::piped());
let mut child = command.spawn().context("start build process")?;
let stdout = child.stdout.take().context("capture stdout")?;
let stderr = child.stderr.take().context("capture stderr")?;
let out = thread::spawn(move || read_bounded(stdout));
let err = thread::spawn(move || read_bounded(stderr));
let deadline = Instant::now() + Duration::from_secs(timeout_seconds);
let (status, timed_out) = loop {
if let Some(status) = child.try_wait()? {
break (status, false);
}
if Instant::now() >= deadline {
child.kill()?;
break (child.wait()?, true);
}
thread::sleep(Duration::from_millis(100));
};
let stdout = out
.join()
.map_err(|_| anyhow::anyhow!("stdout reader panicked"))??;
let stderr = err
.join()
.map_err(|_| anyhow::anyhow!("stderr reader panicked"))??;
let mut logs = String::from_utf8_lossy(&stdout).into_owned();
if !stderr.is_empty() {
if !logs.is_empty() {
logs.push('\n');
}
logs.push_str(&String::from_utf8_lossy(&stderr));
}
for value in secrets.values().filter(|value| !value.is_empty()) {
logs = logs.replace(value, "[redacted]");
}
Ok(ProcessResult {
status,
logs: bounded(&logs, MAX_LOG_BYTES),
timed_out,
})
}
fn read_bounded(mut reader: impl Read) -> Result<Vec<u8>> {
let mut retained = Vec::new();
let mut chunk = [0_u8; 8192];
loop {
let read = reader.read(&mut chunk)?;
if read == 0 {
break;
}
let remaining = MAX_LOG_BYTES.saturating_sub(retained.len());
retained.extend_from_slice(&chunk[..read.min(remaining)]);
}
Ok(retained)
}
fn declared_secrets(names: &[String]) -> Result<BTreeMap<String, String>> {
names
.iter()
.map(|name| {
let value = env::var(format!("AKURAI_SECRET_{name}"))
.with_context(|| format!("missing AKURAI_SECRET_{name}"))?;
Ok((name.clone(), value))
})
.collect()
}
fn matrix_environment(matrix: &BTreeMap<String, String>) -> BTreeMap<String, String> {
matrix
.iter()
.map(|(key, value)| {
let key = key
.chars()
.map(|character| {
if character.is_ascii_alphanumeric() {
character.to_ascii_uppercase()
} else {
'_'
}
})
.collect::<String>();
(format!("AKURAI_MATRIX_{key}"), value.clone())
})
.collect()
}
fn copy_host_environment(command: &mut Command) {
for name in [
"PATH",
"HOME",
"USER",
"LANG",
"LC_ALL",
"DOCKER_CONFIG",
"SSH_AUTH_SOCK",
] {
if let Some(value) = env::var_os(name) {
command.env(name, value);
}
}
}
fn collect_files(path: &Path, files: &mut BTreeSet<PathBuf>) -> Result<()> {
let metadata = path.symlink_metadata()?;
ensure!(
!metadata.file_type().is_symlink(),
"artifact symlinks are not allowed"
);
if metadata.is_file() {
files.insert(path.to_owned());
} else if metadata.is_dir() {
for entry in fs::read_dir(path)? {
collect_files(&entry?.path(), files)?;
}
} else {
bail!("artifact is not a regular file or directory");
}
Ok(())
}
fn copy_hashed(source: &Path, destination: &Path) -> Result<(String, u64)> {
let mut input = File::open(source)?;
let mut output = OpenOptions::new()
.write(true)
.create_new(true)
.open(destination)?;
let mut hasher = Sha256::new();
let mut total = 0_u64;
let mut buffer = [0_u8; 64 * 1024];
loop {
let read = input.read(&mut buffer)?;
if read == 0 {
break;
}
total += u64::try_from(read)?;
ensure!(total <= MAX_ARTIFACT_BYTES, "artifact file is too large");
hasher.update(&buffer[..read]);
output.write_all(&buffer[..read])?;
}
output.sync_all()?;
Ok((format!("{:x}", hasher.finalize()), total))
}
fn hash_file(path: &Path) -> Result<(String, u64)> {
let mut input = File::open(path)?;
let mut hasher = Sha256::new();
let mut total = 0_u64;
let mut buffer = [0_u8; 64 * 1024];
loop {
let read = input.read(&mut buffer)?;
if read == 0 {
break;
}
total += u64::try_from(read)?;
ensure!(total <= MAX_ARTIFACT_BYTES, "artifact file is too large");
hasher.update(&buffer[..read]);
}
Ok((format!("{:x}", hasher.finalize()), total))
}
fn safe_component(value: &str) -> String {
value
.chars()
.map(|character| {
if character.is_ascii_alphanumeric() || matches!(character, '.' | '-' | '_') {
character
} else {
'_'
}
})
.collect()
}
fn private_directory(path: &Path) -> Result<()> {
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
fs::set_permissions(path, fs::Permissions::from_mode(0o700))?;
}
Ok(())
}
fn bounded(value: &str, max: usize) -> String {
if value.len() <= max {
return value.to_owned();
}
let mut end = max;
while !value.is_char_boundary(end) {
end -= 1;
}
format!("{}\n[truncated]", &value[..end])
}
#[cfg(test)]
mod tests {
use super::*;
use crate::db::Database;
const KEY: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
#[test]
fn queue_validates_refs() -> Result<()> {
let root = tempfile::tempdir()?;
let database = Database::memory(KEY)?;
database.add_repository("app", "https://example.com/app.git", "main")?;
let runner = Runner::new(database, root.path().to_owned(), false)?;
assert!(
runner
.queue("app", Some("--upload-pack=x"), None, "manual")
.is_err()
);
Ok(())
}
}