feat(bake): Add streaming execution output and Repology endpoint config
- Implement async stdout/stderr streaming with event channel - Add Repology endpoint configuration in dependencies config - Refactor PackageManager trait to return Vec<Command> - Add bootstrap configuration example - Remove temp/privdrop.rs
This commit is contained in:
@@ -96,6 +96,17 @@ bootstrap:
|
||||
|
||||
# 依赖定义,安装依赖时按system-user顺序进行
|
||||
dependencies:
|
||||
# Repology 端点配置,用于将 UPM 包名解析为系统包名
|
||||
# 可选值:disabled, none, local, remote, server, default
|
||||
# - disabled: 完全禁用 Repology
|
||||
# - none: 不解析,直接使用原始包名
|
||||
# - local: 使用本地 Repology 服务器
|
||||
# - remote: 使用远程 Repology API
|
||||
# - server: 由服务器提供 Repology 服务
|
||||
# - default: 默认行为(通常等同于 local)
|
||||
config:
|
||||
repology_endpoint: "default"
|
||||
|
||||
# 系统包依赖,按包管理器枚举给出,只应该给出与环境契合的字段。
|
||||
# 根据security.drop-after的配置,在此阶段一般具有root或免密root权限
|
||||
# 不适用于裸机
|
||||
|
||||
@@ -58,8 +58,55 @@ environment:
|
||||
- "1.1.1.1"
|
||||
proxies: # 代理配置,暂时留空,可选
|
||||
|
||||
# 构建环境初始化
|
||||
bootstrap:
|
||||
# 构建用户,默认为 vulcan(火神)
|
||||
# 设定为 "root" 表示忽略用户切换并禁用安全特性,需要 PIPELINE_PRIVILEGED 权限
|
||||
user: "vulcan"
|
||||
|
||||
# 工作空间
|
||||
workspace:
|
||||
# 工作目录路径
|
||||
path: "/home/vulcan/workspace"
|
||||
# 路径不可用时回退到 /workspace
|
||||
# 即使路径可用,也会在 /workspace 建立软链接
|
||||
fallback: true
|
||||
|
||||
# 自定义 sudoers 文件内容
|
||||
# 这会覆盖默认行为,可能绕过安全机制
|
||||
# 需要 PIPELINE_CUSTOM_SUDOERS 权限
|
||||
sudoers: "vulcan ALL=(ALL:ALL) NOPASSWD: ALL"
|
||||
|
||||
# 自定义 doas 文件内容
|
||||
# 这会覆盖默认行为,可能绕过安全机制
|
||||
# 与 sudoers 共享 PIPELINE_CUSTOM_SUDOERS 权限
|
||||
doas: "permit nopass vulcan as root"
|
||||
|
||||
# 自定义 bootstrap 脚本(与上述所有字段互斥)
|
||||
# 支持多种来源:
|
||||
# workspace:///.workshop/bootstrap.sh - 项目内文件
|
||||
# file:///bin/bootstrap.sh - 环境内文件
|
||||
# server://bootstrap.sh - 由服务器提供
|
||||
# https://example.com/bootstrap.sh - http/https网络下载,请确保构建环境可以连接到该URL
|
||||
# (content) - 直接给出内容,当文本开头无法匹配任何scheme时采用
|
||||
# 需要 PIPELINE_CUSTOM_BOOTSTRAP 权限
|
||||
custom: "server://bootstrap-alpine.sh"
|
||||
|
||||
|
||||
|
||||
# 依赖定义,安装依赖时按system-user顺序进行
|
||||
dependencies:
|
||||
# Repology 端点配置,用于将 UPM 包名解析为系统包名
|
||||
# 可选值:disabled, none, local, remote, server, default
|
||||
# - disabled: 完全禁用 Repology
|
||||
# - none: 不解析,直接使用原始包名
|
||||
# - local: 使用本地 Repology 服务器
|
||||
# - remote: 使用远程 Repology API
|
||||
# - server: 由服务器提供 Repology 服务
|
||||
# - default: 默认行为(通常等同于 local)
|
||||
config:
|
||||
repology_endpoint: "default"
|
||||
|
||||
# 系统包依赖,按包管理器枚举给出,只应该给出与环境契合的字段。
|
||||
# 根据security.drop-after的配置,在此阶段一般具有root或免密root权限
|
||||
# 不适用于裸机
|
||||
|
||||
@@ -1,11 +1,13 @@
|
||||
use crate::cli::Cli;
|
||||
use crate::constant::DEFAULT_WORKSPACE;
|
||||
use crate::engine::EventSender;
|
||||
use crate::error::BakeError;
|
||||
use crate::prebake::event;
|
||||
use crate::prebake::PrebakeConfig;
|
||||
use crate::prebake::security::get_drop_after;
|
||||
use crate::prebake::stage::PrebakeStage;
|
||||
use crate::{Engine, prebake};
|
||||
use std::path::PathBuf;
|
||||
use std::path::{PathBuf, Path};
|
||||
|
||||
mod builder;
|
||||
mod decorator;
|
||||
@@ -13,9 +15,9 @@ pub mod parser;
|
||||
mod schedule;
|
||||
|
||||
pub async fn bake(
|
||||
script_path: &PathBuf,
|
||||
bake_base_path: &PathBuf,
|
||||
prebake_path: &PathBuf,
|
||||
script_path: &Path,
|
||||
bake_base_path: &Path,
|
||||
prebake_path: &Path,
|
||||
cli: &Cli,
|
||||
) -> Result<(), BakeError> {
|
||||
let script_content = std::fs::read_to_string(script_path)?;
|
||||
@@ -51,6 +53,12 @@ pub async fn bake(
|
||||
}
|
||||
}
|
||||
|
||||
let (tx, event_rx) = tokio::sync::mpsc::channel(256);
|
||||
let event_tx: EventSender = Some(tx);
|
||||
let pipeline_name = cli.pipeline.clone();
|
||||
let build_id = cli.build_id.clone();
|
||||
let receiver_handle = tokio::spawn(event::event_receiver(event_rx, pipeline_name, build_id));
|
||||
|
||||
if !has_pipeline || trivial {
|
||||
let script_body = functions
|
||||
.iter()
|
||||
@@ -65,7 +73,7 @@ pub async fn bake(
|
||||
None,
|
||||
use_template,
|
||||
)?;
|
||||
engine.execute_script(&script, &ctx, &None).await?;
|
||||
engine.execute_script(&script, &ctx, &event_tx).await?;
|
||||
} else {
|
||||
let sorted_functions = schedule::sort_function(functions)?;
|
||||
for func in sorted_functions {
|
||||
@@ -77,9 +85,13 @@ pub async fn bake(
|
||||
None,
|
||||
use_template,
|
||||
)?;
|
||||
engine.execute_script(&script, &ctx, &None).await?;
|
||||
engine.execute_script(&script, &ctx, &event_tx).await?;
|
||||
}
|
||||
}
|
||||
|
||||
drop(event_tx);
|
||||
let _ = receiver_handle.await;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -1,12 +1,14 @@
|
||||
use chrono::Utc;
|
||||
use std::path::PathBuf;
|
||||
use std::process::Stdio;
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tokio::process::Command;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
use super::pm::PackageManager;
|
||||
use super::types::{
|
||||
CgroupManager, EventSender, ExecutionContext, ExecutionError, ExecutionEvent, ExecutionResult,
|
||||
ResourceLimits, DeltaExecutionContext
|
||||
ResourceLimits, DeltaExecutionContext, StreamType
|
||||
};
|
||||
|
||||
impl super::types::Engine {
|
||||
@@ -149,7 +151,7 @@ impl super::types::Engine {
|
||||
});
|
||||
}
|
||||
|
||||
let child = command.spawn().map_err(|e| {
|
||||
let mut child = command.spawn().map_err(|e| {
|
||||
ExecutionError::InvalidCommand(format!("Failed to spawn command: {}", e))
|
||||
})?;
|
||||
|
||||
@@ -158,22 +160,97 @@ impl super::types::Engine {
|
||||
.ok_or_else(|| ExecutionError::IoError("Failed to get PID".to_string()))?;
|
||||
self.apply_cgroup(pid)?;
|
||||
|
||||
let output = child
|
||||
.wait_with_output()
|
||||
.await
|
||||
.map_err(|e| ExecutionError::IoError(format!("Failed to wait for process: {}", e)))?;
|
||||
let task_id = ctx.task_id.clone();
|
||||
let (stdout_tx, mut stdout_rx) = mpsc::channel::<String>(1024);
|
||||
let (stderr_tx, mut stderr_rx) = mpsc::channel::<String>(1024);
|
||||
let mut stdout_buf = String::new();
|
||||
let mut stderr_buf = String::new();
|
||||
|
||||
let exit_code = output.status.code().unwrap_or(-1);
|
||||
let mut child_stdout = child.stdout.take();
|
||||
let mut child_stderr = child.stderr.take();
|
||||
|
||||
let tx_clone = stdout_tx.clone();
|
||||
let _task_id_clone = task_id.clone();
|
||||
let stdout_handle = tokio::spawn(async move {
|
||||
if let Some(mut stdout) = child_stdout.take() {
|
||||
let mut reader = tokio::io::BufReader::new(&mut stdout);
|
||||
let mut buf = [0u8; 4096];
|
||||
loop {
|
||||
match reader.read(&mut buf).await {
|
||||
Ok(0) => break,
|
||||
Ok(n) => {
|
||||
let data = String::from_utf8_lossy(&buf[..n]).to_string();
|
||||
if tx_clone.send(data).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let tx_clone = stderr_tx.clone();
|
||||
let stderr_handle = tokio::spawn(async move {
|
||||
if let Some(mut stderr) = child_stderr.take() {
|
||||
let mut reader = tokio::io::BufReader::new(&mut stderr);
|
||||
let mut buf = [0u8; 4096];
|
||||
loop {
|
||||
match reader.read(&mut buf).await {
|
||||
Ok(0) => break,
|
||||
Ok(n) => {
|
||||
let data = String::from_utf8_lossy(&buf[..n]).to_string();
|
||||
if tx_clone.send(data).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let status = child.wait().await.map_err(|e| {
|
||||
ExecutionError::IoError(format!("Failed to wait for process: {}", e))
|
||||
})?;
|
||||
|
||||
drop(stdout_tx);
|
||||
drop(stderr_tx);
|
||||
|
||||
let _ = stdout_handle.await;
|
||||
let _ = stderr_handle.await;
|
||||
|
||||
while let Some(data) = stdout_rx.recv().await {
|
||||
stdout_buf.push_str(&data);
|
||||
if let Some(tx) = event_tx {
|
||||
let _ = tx.send(ExecutionEvent::OutputChunk {
|
||||
task_id: task_id.clone(),
|
||||
stream: StreamType::Stdout,
|
||||
data: data.clone(),
|
||||
}).await;
|
||||
}
|
||||
}
|
||||
while let Some(data) = stderr_rx.recv().await {
|
||||
stderr_buf.push_str(&data);
|
||||
if let Some(tx) = event_tx {
|
||||
let _ = tx.send(ExecutionEvent::OutputChunk {
|
||||
task_id: task_id.clone(),
|
||||
stream: StreamType::Stderr,
|
||||
data: data.clone(),
|
||||
}).await;
|
||||
}
|
||||
}
|
||||
|
||||
let exit_code = status.code().unwrap_or(-1);
|
||||
let duration = start_time.elapsed();
|
||||
|
||||
// TODO: Timeout adaptation
|
||||
let result = ExecutionResult {
|
||||
task_id: ctx.task_id.clone(),
|
||||
exit_code,
|
||||
success: exit_code == 0,
|
||||
duration,
|
||||
stdout: String::from_utf8_lossy(&output.stdout).to_string(),
|
||||
stderr: String::from_utf8_lossy(&output.stderr).to_string(),
|
||||
stdout: stdout_buf,
|
||||
stderr: stderr_buf,
|
||||
resource_usage: None,
|
||||
};
|
||||
|
||||
@@ -192,7 +269,14 @@ impl super::types::Engine {
|
||||
let _ = tx.send(event).await;
|
||||
}
|
||||
|
||||
if result.success {
|
||||
Ok(result)
|
||||
} else {
|
||||
Err(ExecutionError::ExecutionFailed(format!(
|
||||
"Task {} failed with exit code {}",
|
||||
ctx.task_id, exit_code
|
||||
)))
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn install_dependency(
|
||||
@@ -202,9 +286,27 @@ impl super::types::Engine {
|
||||
ctx: &ExecutionContext,
|
||||
event_tx: &EventSender
|
||||
) -> Result<ExecutionResult, ExecutionError> {
|
||||
let cmd = pkgman.install(package);
|
||||
|
||||
self.execute(cmd, ctx, event_tx).await
|
||||
let cmds = pkgman.install(package);
|
||||
let mut combined_result = ExecutionResult {
|
||||
task_id: ctx.task_id.clone(),
|
||||
exit_code: 0,
|
||||
success: true,
|
||||
duration: std::time::Duration::default(),
|
||||
stdout: String::new(),
|
||||
stderr: String::new(),
|
||||
resource_usage: None,
|
||||
};
|
||||
for cmd in cmds {
|
||||
log::debug!("Installing dependency with command: {:?}", cmd);
|
||||
let result = self.execute(cmd, ctx, event_tx).await?;
|
||||
// TODO: Aggregate results properly instead of just taking the last one
|
||||
combined_result.exit_code = result.exit_code;
|
||||
combined_result.success &= result.success;
|
||||
combined_result.duration += result.duration;
|
||||
combined_result.stdout.push_str(&result.stdout);
|
||||
combined_result.stderr.push_str(&result.stderr);
|
||||
}
|
||||
Ok(combined_result)
|
||||
}
|
||||
|
||||
pub async fn cleanup(&self) -> Result<(), String> {
|
||||
|
||||
@@ -16,10 +16,10 @@ pub trait PackageManager {
|
||||
fn binary(&self) -> &'static str;
|
||||
|
||||
/// The command of updating
|
||||
fn update(&self) -> Command;
|
||||
fn update(&self) -> Vec<Command>;
|
||||
|
||||
/// The command of installing packages
|
||||
fn install(&self, package: &[String]) -> Command;
|
||||
fn install(&self, package: &[String]) -> Vec<Command>;
|
||||
|
||||
// Check whether a package is installed
|
||||
// Currently not planned
|
||||
@@ -29,7 +29,7 @@ pub trait PackageManager {
|
||||
// fn is_installed(&self, _package: &str) -> Option<bool> {unimplemented!()}
|
||||
|
||||
/// Change major mirror of package manager
|
||||
fn change_mirror(&self, repo: &str, osinfo: Option<&Info>) -> Command;
|
||||
fn change_mirror(&self, repo: &str, osinfo: Option<&Info>) -> Vec<Command>;
|
||||
|
||||
/// add a custom repository
|
||||
fn add_repository(&self, repo: &Value, osinfo: Option<&Info>) -> Vec<Command>;
|
||||
|
||||
@@ -93,22 +93,22 @@ impl PackageManager for Pacman {
|
||||
"pacman"
|
||||
}
|
||||
|
||||
fn update(&self) -> Command {
|
||||
fn update(&self) -> Vec<Command> {
|
||||
let mut cmd = Command::new("pacman");
|
||||
cmd.arg("-Sy");
|
||||
cmd
|
||||
vec![cmd]
|
||||
}
|
||||
|
||||
fn install(&self, package: &[String]) -> Command {
|
||||
fn install(&self, package: &[String]) -> Vec<Command> {
|
||||
let mut cmd = Command::new("pacman");
|
||||
cmd.arg("-S");
|
||||
cmd.arg("--noconfirm");
|
||||
cmd.arg("--needed");
|
||||
cmd.args(package);
|
||||
cmd
|
||||
vec![cmd]
|
||||
}
|
||||
|
||||
fn change_mirror(&self, repo: &str, _osinfo: Option<&Info>) -> Command {
|
||||
fn change_mirror(&self, repo: &str, _osinfo: Option<&Info>) -> Vec<Command> {
|
||||
let mirrorlist_path = "/etc/pacman.d/mirrorlist";
|
||||
|
||||
let content = format!(
|
||||
@@ -126,7 +126,9 @@ impl PackageManager for Pacman {
|
||||
mirrorlist_path, content
|
||||
));
|
||||
|
||||
cmd
|
||||
let mut keyring_cmd = Command::new("sh");
|
||||
keyring_cmd.arg("-c").arg("pacman-key --init");
|
||||
vec![cmd, keyring_cmd]
|
||||
}
|
||||
|
||||
fn add_repository(&self, repo: &Value, _osinfo: Option<&Info>) -> Vec<Command> {
|
||||
@@ -258,8 +260,8 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_update_command_contents() {
|
||||
let cmd = Pacman.update();
|
||||
let cmd = cmd.as_std();
|
||||
let cmds = Pacman.update();
|
||||
let cmd = cmds[0].as_std();
|
||||
assert_eq!(cmd.get_program().to_string_lossy(), "pacman");
|
||||
let args: Vec<String> = cmd
|
||||
.get_args()
|
||||
@@ -270,8 +272,8 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_install_command_contents() {
|
||||
let cmd = Pacman.install(&["curl".to_string(), "wget".to_string()]);
|
||||
let cmd = cmd.as_std();
|
||||
let cmds = Pacman.install(&["curl".to_string(), "wget".to_string()]);
|
||||
let cmd = cmds[0].as_std();
|
||||
assert_eq!(cmd.get_program().to_string_lossy(), "pacman");
|
||||
let args: Vec<String> = cmd
|
||||
.get_args()
|
||||
@@ -282,8 +284,8 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_install_single_package() {
|
||||
let cmd = Pacman.install(&["package".to_string()]);
|
||||
let cmd = cmd.as_std();
|
||||
let cmds = Pacman.install(&["package".to_string()]);
|
||||
let cmd = cmds[0].as_std();
|
||||
let args: Vec<String> = cmd
|
||||
.get_args()
|
||||
.map(|a| a.to_string_lossy().into_owned())
|
||||
@@ -330,8 +332,8 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_change_mirror_command() {
|
||||
let cmd = Pacman.change_mirror("https://mirror.example.com/archlinux", None);
|
||||
let cmd = cmd.as_std();
|
||||
let cmds = Pacman.change_mirror("https://mirror.example.com/archlinux", None);
|
||||
let cmd = cmds[0].as_std();
|
||||
assert_eq!(cmd.get_program().to_string_lossy(), "sh");
|
||||
let args: Vec<String> = cmd
|
||||
.get_args()
|
||||
|
||||
@@ -22,6 +22,7 @@ async fn main() {
|
||||
Some(Commands::Prebake { config }) => {
|
||||
log::info!("Running in standalone mode.");
|
||||
let result = prebake(config, &cli, None).await;
|
||||
dbg!(&result);
|
||||
if let Err(ref e) = result {
|
||||
log::error!("Prebake stage failed: {}", e);
|
||||
std::process::exit(
|
||||
|
||||
@@ -22,11 +22,13 @@
|
||||
|
||||
pub mod config;
|
||||
pub mod env;
|
||||
pub mod event;
|
||||
pub mod security;
|
||||
pub mod stage;
|
||||
|
||||
use crate::{
|
||||
cli::Cli,
|
||||
engine::EventSender,
|
||||
error::PrebakeError,
|
||||
prebake::{security::get_drop_after, stage::PrebakeStage},
|
||||
};
|
||||
@@ -88,7 +90,9 @@ pub async fn prebake(
|
||||
stage: Option<PrebakeStage>,
|
||||
) -> anyhow::Result<()> {
|
||||
// Initialize
|
||||
let prebake_content = std::fs::read_to_string(prebake_path)?;
|
||||
let prebake_content = std::fs::read_to_string(prebake_path).inspect_err(|e| {
|
||||
log::error!("Failed to read prebake.yml: {}", e);
|
||||
})?;
|
||||
let prebake = parse(prebake_content.as_str()).map_err(|e| {
|
||||
log::error!("Failed to parse prebake.yml: {}", e);
|
||||
PrebakeError::YamlParseError(e)
|
||||
@@ -121,15 +125,20 @@ pub async fn prebake(
|
||||
ctx.env_vars.extend(prebake.clone());
|
||||
}
|
||||
|
||||
let (tx, event_rx) = tokio::sync::mpsc::channel(256);
|
||||
let event_tx: EventSender = Some(tx);
|
||||
let pipeline_name = cli.pipeline.clone();
|
||||
let build_id = cli.build_id.clone();
|
||||
let receiver_handle = tokio::spawn(event::event_receiver(event_rx, pipeline_name, build_id));
|
||||
|
||||
// Bootstrap stage
|
||||
|
||||
if stage <= PrebakeStage::Bootstrap {
|
||||
log::info!("Running PBStage: Bootstrap");
|
||||
prebake_drop_privilege(PrebakeStage::Bootstrap, dropafter, target_user, &mut ctx)?;
|
||||
ctx.task_id = format!("{}-{}-bootstrap", cli.pipeline, cli.build_id);
|
||||
// TODO: Pass real event_tx
|
||||
let osinfo = os_info::get();
|
||||
stage::bootstrap::bootstrap(&prebake, &osinfo, &ctx, &None).await?;
|
||||
stage::bootstrap::bootstrap(&prebake, &osinfo, &ctx, &event_tx).await?;
|
||||
}
|
||||
|
||||
// EarlyHook stage
|
||||
@@ -141,7 +150,7 @@ pub async fn prebake(
|
||||
Some(hooks) => hooks.early.clone(),
|
||||
None => None,
|
||||
};
|
||||
stage::hook::hook(earlyhook, &ctx, None).await?;
|
||||
stage::hook::hook(earlyhook, &ctx, event_tx.clone()).await?;
|
||||
}
|
||||
|
||||
if let Some(dependencies) = prebake.dependencies {
|
||||
@@ -207,7 +216,7 @@ pub async fn prebake(
|
||||
log::info!("Running PBStage: DepsSystem");
|
||||
prebake_drop_privilege(PrebakeStage::DepsSystem, dropafter, target_user, &mut ctx)?;
|
||||
ctx.task_id = format!("{}-{}-depssystem", cli.pipeline, cli.build_id);
|
||||
stage::depssystem::depssystem(&system_config, &ctx, None).await?;
|
||||
stage::depssystem::depssystem(&system_config, &ctx, event_tx.clone()).await?;
|
||||
}
|
||||
|
||||
if let Some(user) = dependencies.user
|
||||
@@ -216,7 +225,7 @@ pub async fn prebake(
|
||||
log::info!("Running PBStage: DepsUser");
|
||||
prebake_drop_privilege(PrebakeStage::DepsUser, dropafter, target_user, &mut ctx)?;
|
||||
ctx.task_id = format!("{}-{}-depsuser", cli.pipeline, cli.build_id);
|
||||
stage::depsuser::depsuser(&user, &ctx, None, &endpoint).await?;
|
||||
stage::depsuser::depsuser(&user, &ctx, event_tx.clone(), &endpoint).await?;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -229,11 +238,15 @@ pub async fn prebake(
|
||||
Some(hooks) => hooks.late.clone(),
|
||||
None => None,
|
||||
};
|
||||
stage::hook::hook(latehook, &ctx, None).await?;
|
||||
stage::hook::hook(latehook, &ctx, event_tx.clone()).await?;
|
||||
}
|
||||
|
||||
// Ready stage
|
||||
ctx.task_id = format!("{}-{}-ready", cli.pipeline, cli.build_id);
|
||||
|
||||
drop(event_tx);
|
||||
let _ = receiver_handle.await;
|
||||
|
||||
log::info!("Prebake done!");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
use crate::engine::{ExecutionEvent, StreamType};
|
||||
|
||||
pub async fn event_receiver(
|
||||
mut rx: tokio::sync::mpsc::Receiver<ExecutionEvent>,
|
||||
pipeline_name: String,
|
||||
build_id: String,
|
||||
) {
|
||||
while let Some(event) = rx.recv().await {
|
||||
match event {
|
||||
ExecutionEvent::TaskStarted { task_id, timestamp } => {
|
||||
log::debug!(
|
||||
"[{}] [{}] [{}] Task started: {}",
|
||||
pipeline_name, build_id, timestamp, task_id
|
||||
);
|
||||
}
|
||||
ExecutionEvent::OutputChunk { task_id, stream, data } => {
|
||||
match stream {
|
||||
StreamType::Stdout => print!("[{}] [{}] [stdout] {}", pipeline_name, task_id, data),
|
||||
StreamType::Stderr => eprint!("[{}] [{}] [stderr] {}", pipeline_name, task_id, data),
|
||||
}
|
||||
}
|
||||
ExecutionEvent::TaskCompleted { task_id, result } => {
|
||||
log::debug!(
|
||||
"[{}] [{}] Task completed: {} (exit_code={}, duration={:?})",
|
||||
pipeline_name, build_id, task_id, result.exit_code, result.duration
|
||||
);
|
||||
}
|
||||
ExecutionEvent::TaskFailed { task_id, error } => {
|
||||
log::error!(
|
||||
"[{}] [{}] Task failed: {} (error={})",
|
||||
pipeline_name, build_id, task_id, error
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -62,6 +62,7 @@ pub async fn bootstrap(
|
||||
log::debug!("Generated bootstrap script:\n{}", script);
|
||||
let mut file = File::create(&bootstrap_path)?;
|
||||
file.write_all(script.as_bytes())?;
|
||||
std::fs::set_permissions(&bootstrap_path, std::os::unix::fs::PermissionsExt::from_mode(0o755))?;
|
||||
} else {
|
||||
// Requires PIP(Pipeline Implicit Parameters) to be implemented
|
||||
unimplemented!("Not supporting custom bootstrap.sh at this time");
|
||||
@@ -127,6 +128,7 @@ fn generate_bootstrap(
|
||||
unimplemented!("Permission check is not implemented, failing...");
|
||||
} else {
|
||||
script.push_str("# User creation\n");
|
||||
script.push_str(&format!("if ! id -u {} >/dev/null 2>&1; then\n", user));
|
||||
if which("useradd").is_ok() {
|
||||
script.push_str(&format!("useradd -m -s /bin/nologin {}\n", user));
|
||||
} else if which("adduser").is_ok() {
|
||||
@@ -143,6 +145,7 @@ fn generate_bootstrap(
|
||||
"no available binary found".to_string(),
|
||||
));
|
||||
}
|
||||
script.push_str("fi\n");
|
||||
}
|
||||
|
||||
// Setup workspace
|
||||
|
||||
@@ -39,7 +39,8 @@ pub async fn depssystem(
|
||||
// Change mirror
|
||||
if let Some(mirror) = &pm_config.mirror {
|
||||
log::info!("Changing mirror to: {}", mirror);
|
||||
let cmd = package_manager.change_mirror(mirror, Some(&osinfo));
|
||||
let cmds = package_manager.change_mirror(mirror, Some(&osinfo));
|
||||
for cmd in cmds {
|
||||
log::debug!("Executing: {:?}", &cmd);
|
||||
engine
|
||||
.execute(cmd, ctx, &event_tx)
|
||||
@@ -49,6 +50,7 @@ pub async fn depssystem(
|
||||
source: e,
|
||||
})?;
|
||||
}
|
||||
}
|
||||
|
||||
// Add custom repository
|
||||
if let Some(repos) = &pm_config.repositories {
|
||||
@@ -71,16 +73,17 @@ pub async fn depssystem(
|
||||
// Update package index
|
||||
{
|
||||
log::info!("Updating package index...");
|
||||
let cmd = package_manager.update();
|
||||
let cmds = package_manager.update();
|
||||
for cmd in cmds {
|
||||
log::debug!("Executing: {:?}", &cmd);
|
||||
engine
|
||||
.execute(cmd, ctx, &event_tx)
|
||||
.await
|
||||
.map_err(|e| DependencyError::StageFailed {
|
||||
engine.execute(cmd, ctx, &event_tx).await.map_err(|e| {
|
||||
DependencyError::StageFailed {
|
||||
stage: "UpdateIndex",
|
||||
source: e,
|
||||
}
|
||||
})?;
|
||||
}
|
||||
}
|
||||
|
||||
// Install packages
|
||||
if !pm_config.packages.is_empty() {
|
||||
|
||||
@@ -1,80 +0,0 @@
|
||||
fn lookup_user(
|
||||
user: &OsStr,
|
||||
fallback_to_ids_if_names_are_numeric: bool,
|
||||
) -> Result<UserIds, PrivDropError> {
|
||||
let username = CString::new(user.as_bytes())
|
||||
.map_err(|_| PrivDropError::from((ErrorKind::SysError, "Invalid username")))?;
|
||||
|
||||
let mut pwd = MaybeUninit::<libc::passwd>::uninit();
|
||||
let mut pwent = std::ptr::null_mut::<libc::passwd>();
|
||||
|
||||
// Start with the initial buffer size and increase if needed
|
||||
let mut bufsize = INITIAL_BUFFER_SIZE;
|
||||
let mut pwbuf = vec![0; bufsize];
|
||||
|
||||
let mut ret;
|
||||
loop {
|
||||
ret = unsafe {
|
||||
libc::getpwnam_r(
|
||||
username.as_ptr(),
|
||||
pwd.as_mut_ptr(),
|
||||
pwbuf.as_mut_ptr(),
|
||||
pwbuf.len(),
|
||||
&mut pwent,
|
||||
)
|
||||
};
|
||||
|
||||
// If we get ERANGE, the buffer was too small, double it and try again
|
||||
if ret == libc::ERANGE {
|
||||
bufsize *= 2;
|
||||
pwbuf.resize(bufsize, 0);
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if ret != 0 || pwent.is_null() {
|
||||
if !fallback_to_ids_if_names_are_numeric {
|
||||
if ret != 0 && ret == libc::ENOENT {
|
||||
return Err(PrivDropError::from((ErrorKind::SysError, "User not found")));
|
||||
} else if ret != 0 {
|
||||
return Err(PrivDropError::from((
|
||||
ErrorKind::SysError,
|
||||
"Failed to look up user",
|
||||
)));
|
||||
} else {
|
||||
return Err(PrivDropError::from((ErrorKind::SysError, "User not found")));
|
||||
}
|
||||
}
|
||||
|
||||
// Try to parse the username as a numeric UID
|
||||
let user_str = user.to_str().ok_or_else(|| {
|
||||
PrivDropError::from((
|
||||
ErrorKind::SysError,
|
||||
"User not found and username is not valid UTF-8",
|
||||
))
|
||||
})?;
|
||||
|
||||
let uid = user_str.parse().map_err(|_| {
|
||||
PrivDropError::from((
|
||||
ErrorKind::SysError,
|
||||
"User not found and username is not a valid numeric ID",
|
||||
))
|
||||
})?;
|
||||
|
||||
return Ok(UserIds {
|
||||
uid: Some(uid),
|
||||
gid: None,
|
||||
group_list: None,
|
||||
});
|
||||
}
|
||||
|
||||
let uid = unsafe { *pwent }.pw_uid;
|
||||
let gid = unsafe { *pwent }.pw_gid;
|
||||
|
||||
Ok(UserIds {
|
||||
uid: Some(uid),
|
||||
gid: Some(gid),
|
||||
group_list: None,
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user