Menu
AkurAI-Build
publicLatest change 0e969f8ad7050ab59b61ec03b8caef0fd2d19bed - Stream canonical run progress events by Ólafur Búi Ólafsson
use std::{
io::Write,
net::SocketAddr,
path::PathBuf,
process::{Command, Stdio},
sync::Arc,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use anyhow::{Context, Result, ensure};
use axum::{
Json, Router,
body::{Body, Bytes, to_bytes},
extract::{DefaultBodyLimit, Path, Query, State},
http::{HeaderMap, HeaderName, HeaderValue, Request, StatusCode, header},
middleware::{self, Next},
response::{Html, IntoResponse, Redirect, Response},
routing::{any, get, post},
};
use hmac::{Hmac, Mac};
use minijinja::{AutoEscape, Environment, context};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::Sha256;
use subtle::ConstantTimeEq;
use tokio::io::AsyncWriteExt;
use tokio_util::io::ReaderStream;
use tower::limit::GlobalConcurrencyLimitLayer;
use tower_http::{compression::CompressionLayer, timeout::TimeoutLayer, trace::TraceLayer};
use tracing::{info, warn};
use crate::auth::{IdpConfig, UserInfo};
use crate::{config, db::Database, runner::Runner, tree::RepoTree};
const MAX_BODY: usize = 1024 * 1024;
const MAX_GIT_BODY: usize = 512 * 1024 * 1024;
const MAX_CONNECTIONS: usize = 1024;
const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
const LANDING: &str = include_str!("../app/pages/landing.html");
const DASHBOARD: &str = include_str!("../app/pages/index.html");
const REPOSITORIES: &str = include_str!("../app/pages/repositories.html");
const REPOSITORY_APP: &str = include_str!("../app/pages/repository-app.html");
const RUNS: &str = include_str!("../app/pages/runs.html");
const WORKERS: &str = include_str!("../app/pages/workers.html");
const LOGIN: &str = include_str!("../app/pages/login.html");
const PROFILE: &str = include_str!("../app/pages/profile.html");
const ADMIN: &str = include_str!("../app/pages/admin.html");
const REPOSITORY: &str = include_str!("../app/pages/repository.html");
const RUN: &str = include_str!("../app/pages/run.html");
const BASE: &str = include_str!("../app/templates/layouts/base.html");
const CSS: &str = include_str!("../public/app.css");
const JS: &str = include_str!("../public/app.js");
type HmacSha256 = Hmac<Sha256>;
#[derive(Clone)]
pub struct ServerOptions {
pub address: SocketAddr,
pub token: String,
pub worker_count: usize,
pub webhook_secret: String,
pub public_origin: String,
pub data_root: PathBuf,
pub allow_native: bool,
}
#[derive(Clone)]
struct AppState {
database: Database,
runner: Runner,
tree: RepoTree,
hosted_root: PathBuf,
token: Arc<str>,
webhook_secret: Arc<str>,
public_origin: Arc<str>,
templates: Arc<Environment<'static>>,
}
#[derive(Serialize)]
struct Envelope<T: Serialize> {
ok: bool,
data: T,
}
#[derive(Serialize)]
struct ErrorEnvelope {
ok: bool,
error: ErrorMessage,
}
#[derive(Serialize)]
struct ErrorMessage {
code: &'static str,
message: String,
}
struct ApiError {
status: StatusCode,
code: &'static str,
message: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct TriggerRequest {
git_ref: Option<String>,
commit: Option<String>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RepositoryRequest {
name: String,
url: String,
branch: String,
}
#[derive(Serialize)]
struct RunProgress {
run_id: i64,
repository: String,
git_ref: String,
commit_sha: Option<String>,
status: String,
stage: String,
progress: u8,
jobs: Vec<ProgressJob>,
deployments: Vec<ProgressDeployment>,
error: Option<String>,
}
#[derive(Serialize)]
struct ProgressJob {
id: i64,
name: String,
status: String,
environment: Option<String>,
approval_required: bool,
}
#[derive(Serialize)]
struct ProgressDeployment {
id: i64,
environment: String,
status: String,
}
#[derive(Deserialize, Default)]
struct StateQuery {
repo: Option<String>,
status: Option<String>,
git_ref: Option<String>,
trigger: Option<String>,
q: Option<String>,
limit: Option<usize>,
offset: Option<usize>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct VisibilityRequest {
visibility: String,
}
#[derive(Serialize)]
struct StateResponse {
repositories: Vec<crate::db::Repository>,
runs: Vec<crate::db::Run>,
workers: Vec<crate::db::Worker>,
queued_runs: i64,
}
#[derive(Serialize)]
struct IdResponse {
id: i64,
}
#[derive(Serialize)]
struct BranchInfo {
name: String,
is_default: bool,
}
#[derive(Deserialize)]
struct CallbackQuery {
code: Option<String>,
state: Option<String>,
error: Option<String>,
}
#[derive(Deserialize)]
struct TokenResponse {
access_token: String,
}
impl ApiError {
fn unauthorized() -> Self {
Self {
status: StatusCode::UNAUTHORIZED,
code: "unauthorized",
message: "valid bearer token required".into(),
}
}
fn bad(error: impl std::fmt::Display) -> Self {
Self {
status: StatusCode::BAD_REQUEST,
code: "invalid_request",
message: error.to_string(),
}
}
fn internal(error: impl std::fmt::Display) -> Self {
tracing::error!(error = %error, "request failed");
Self {
status: StatusCode::INTERNAL_SERVER_ERROR,
code: "internal",
message: "request failed".into(),
}
}
}
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
(
self.status,
Json(ErrorEnvelope {
ok: false,
error: ErrorMessage {
code: self.code,
message: self.message,
},
}),
)
.into_response()
}
}
pub async fn serve(database: Database, options: ServerOptions) -> Result<()> {
ensure!(!options.token.is_empty(), "API token must not be empty");
ensure!(
!options.webhook_secret.is_empty(),
"webhook secret must not be empty"
);
ensure!(
(1..=16).contains(&options.worker_count),
"worker count must be 1..=16"
);
database.validate_schema()?;
let recovered = database.recover_interrupted()?;
if recovered > 0 {
warn!(recovered, "marked interrupted work after restart");
}
database.reset_workers()?;
let mut templates = Environment::new();
templates.set_auto_escape_callback(|name| {
if name.ends_with(".html") {
AutoEscape::Html
} else {
AutoEscape::None
}
});
templates.add_template("layouts/base.html", BASE)?;
templates.add_template("landing.html", LANDING)?;
templates.add_template("dashboard.html", DASHBOARD)?;
templates.add_template("repositories.html", REPOSITORIES)?;
templates.add_template("repository-app.html", REPOSITORY_APP)?;
templates.add_template("runs.html", RUNS)?;
templates.add_template("workers.html", WORKERS)?;
templates.add_template("login.html", LOGIN)?;
templates.add_template("profile.html", PROFILE)?;
templates.add_template("admin.html", ADMIN)?;
templates.add_template("run.html", RUN)?;
templates.add_template("repository.html", REPOSITORY)?;
let hosted_root = options.data_root.clone();
let tree = RepoTree::new(&options.data_root)?;
let runner = Runner::new(database.clone(), options.data_root, options.allow_native)?;
let state = AppState {
database: database.clone(),
runner: runner.clone(),
tree,
hosted_root,
token: options.token.into(),
webhook_secret: options.webhook_secret.into(),
public_origin: options.public_origin.into(),
templates: Arc::new(templates),
};
for index in 0..options.worker_count {
let worker_id = format!("titan-{}", index + 1);
database.register_worker(&worker_id, "titan", "docker,native", 1)?;
tokio::spawn(worker(database.clone(), runner.clone(), worker_id));
}
let app = Router::new()
.route("/", get(index))
.route("/repos/{repository}", get(public_repository))
.route("/app", get(dashboard))
.route("/app/repositories", get(repositories_page))
.route("/app/repositories/{repository}", get(repository_dashboard))
.route("/app/run/{id}", get(run_page))
.route("/app/runs", get(runs_page))
.route("/app/workers", get(workers_page))
.route("/login", get(login))
.route("/auth/callback", get(auth_callback))
.route("/assets/app.css", get(css))
.route("/assets/app.js", get(js))
.route("/git/{*path}", any(git_http))
.route("/api/health", get(health))
.route("/api/state", get(api_state))
.route("/api/repos", post(create_repository))
.route("/api/runs/{id}", get(run_detail))
.route("/api/runs/{id}/events", get(run_events))
.route("/api/runs/{id}/retry", post(retry))
.route("/api/workers", get(api_workers))
.route("/api/runs/{id}/promote/{environment}", post(promote))
.route("/api/repos/{repository}/runs", post(trigger))
.route(
"/api/repos/{repository}/visibility",
post(change_repository_visibility),
)
.route("/api/hooks/{repository}", post(webhook))
.route("/api/artifacts/{id}", get(artifact))
.route("/api/repos/{repository}/branches", get(branches))
.route("/api/repos/{repository}/tree", get(repo_tree))
.with_state(state)
.layer(DefaultBodyLimit::max(MAX_BODY))
.layer(CompressionLayer::new())
.layer(TimeoutLayer::with_status_code(
StatusCode::REQUEST_TIMEOUT,
REQUEST_TIMEOUT,
))
.layer(TraceLayer::new_for_http())
.layer(middleware::from_fn(security_headers))
.layer(GlobalConcurrencyLimitLayer::new(MAX_CONNECTIONS));
let listener = tokio::net::TcpListener::bind(options.address).await?;
serve_bounded(listener, app).await
}
async fn worker(database: Database, runner: Runner, worker_id: String) {
if let Err(error) = database.heartbeat_worker(&worker_id, "idle", None, false) {
warn!(%worker_id, ?error, "worker registration heartbeat failed");
}
loop {
match database.claim_next_queued_run() {
Ok(Some(id)) => {
if let Err(error) =
database.heartbeat_worker(&worker_id, "running", Some(id), false)
{
warn!(%worker_id, ?error, "worker heartbeat failed");
}
let heartbeat_database = database.clone();
let heartbeat_worker_id = worker_id.clone();
let heartbeat = tokio::spawn(async move {
let mut ticker = tokio::time::interval(Duration::from_secs(5));
loop {
ticker.tick().await;
if let Err(error) = heartbeat_database.heartbeat_worker(
&heartbeat_worker_id,
"running",
Some(id),
false,
) {
warn!(worker_id = %heartbeat_worker_id, ?error, "worker heartbeat failed");
}
}
});
let process_runner = runner.clone();
let run =
tokio::task::spawn_blocking(move || process_runner.process_claimed_run(id))
.await;
heartbeat.abort();
let completed = match run {
Ok(Ok(detail)) => matches!(
detail.run.status.as_str(),
"waiting" | "succeeded" | "failed" | "canceled" | "interrupted"
),
Ok(Err(error)) => {
tracing::error!(run_id = id, ?error, "run failed");
false
}
Err(error) => {
tracing::error!(run_id = id, ?error, "run worker panicked");
false
}
};
if let Err(error) = database.heartbeat_worker(&worker_id, "idle", None, completed) {
warn!(%worker_id, ?error, "worker idle heartbeat failed");
}
}
Ok(None) => {
if let Err(error) = database.heartbeat_worker(&worker_id, "idle", None, false) {
warn!(%worker_id, ?error, "worker heartbeat failed");
}
tokio::time::sleep(Duration::from_millis(500)).await;
}
Err(error) => {
tracing::error!(?error, "queue poll failed");
tokio::time::sleep(Duration::from_secs(2)).await;
}
}
}
}
async fn index(State(state): State<AppState>) -> Result<Html<String>, ApiError> {
let page = state
.templates
.get_template("landing.html")
.map_err(ApiError::internal)?
.render(context! { version => env!("CARGO_PKG_VERSION"), repositories => state.database.public_repositories().map_err(ApiError::internal)? })
.map_err(ApiError::internal)?;
Ok(Html(page))
}
async fn public_repository(
State(state): State<AppState>,
Path(repository): Path<String>,
) -> Result<Html<String>, ApiError> {
let repository = state
.database
.repository(&repository)
.map_err(ApiError::bad)?;
if repository.visibility != "public" {
return Err(ApiError::bad("unknown public repository"));
}
let page = state
.templates
.get_template("repository.html")
.map_err(ApiError::internal)?
.render(context! { repository => repository })
.map_err(ApiError::internal)?;
Ok(Html(page))
}
async fn git_http(
State(state): State<AppState>,
Path(path): Path<String>,
request: Request<Body>,
) -> Result<Response, ApiError> {
authorize(request.headers(), &state)?;
validate_git_path(&path).map_err(ApiError::bad)?;
let (parts, body) = request.into_parts();
let body = to_bytes(body, MAX_GIT_BODY).await.map_err(ApiError::bad)?;
let root = state.hosted_root.join("hosted");
tokio::task::spawn_blocking(move || run_git_http(root, path, parts, body))
.await
.map_err(ApiError::internal)?
.map_err(ApiError::internal)
}
fn validate_git_path(path: &str) -> Result<()> {
ensure!(!path.is_empty() && path.len() <= 512, "invalid Git path");
let repository = path.split('/').next().unwrap_or_default();
ensure!(repository.ends_with(".git"), "invalid Git repository path");
ensure!(
path.split('/').all(|component| {
!component.is_empty()
&& component
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || b"._-".contains(&byte))
}),
"invalid Git path"
);
Ok(())
}
fn run_git_http(
root: PathBuf,
path: String,
parts: axum::http::request::Parts,
body: Bytes,
) -> Result<Response> {
let mut command = Command::new("git");
command
.arg("http-backend")
.env("GIT_PROJECT_ROOT", root)
.env("GIT_HTTP_EXPORT_ALL", "1")
.env("PATH_INFO", format!("/{path}"))
.env("REQUEST_METHOD", parts.method.as_str())
.env("QUERY_STRING", parts.uri.query().unwrap_or_default())
.env("CONTENT_LENGTH", body.len().to_string())
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
if let Some(value) = parts.headers.get(header::CONTENT_TYPE) {
command.env("CONTENT_TYPE", value.to_str()?);
}
if let Some(value) = parts.headers.get("git-protocol") {
command.env("HTTP_GIT_PROTOCOL", value.to_str()?);
}
let mut child = command
.spawn()
.context("failed to start git http-backend")?;
child
.stdin
.take()
.context("git http-backend stdin is unavailable")?
.write_all(&body)?;
let output = child.wait_with_output()?;
ensure!(
output.status.success(),
"git http-backend failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
);
git_cgi_response(&output.stdout)
}
fn git_cgi_response(output: &[u8]) -> Result<Response> {
let (header_bytes, body) = output
.windows(4)
.position(|window| window == b"\r\n\r\n")
.map(|index| (&output[..index], &output[index + 4..]))
.or_else(|| {
output
.windows(2)
.position(|window| window == b"\n\n")
.map(|index| (&output[..index], &output[index + 2..]))
})
.context("git http-backend returned malformed CGI output")?;
let mut status = StatusCode::OK;
let mut response = Response::builder();
for line in String::from_utf8_lossy(header_bytes).lines() {
let (name, value) = line.split_once(':').context("invalid Git CGI header")?;
if name.eq_ignore_ascii_case("status") {
let code = value
.split_whitespace()
.next()
.context("missing Git CGI status")?
.parse::<u16>()?;
status = StatusCode::from_u16(code)?;
} else {
response = response.header(
HeaderName::from_bytes(name.trim().as_bytes())?,
HeaderValue::from_str(value.trim())?,
);
}
}
Ok(response.status(status).body(Body::from(body.to_vec()))?)
}
async fn dashboard(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Response, ApiError> {
if !valid_session(&headers, &state) {
return Ok(Redirect::to("/login").into_response());
}
let page = state
.templates
.get_template("dashboard.html")
.map_err(ApiError::internal)?
.render(context! { version => env!("CARGO_PKG_VERSION") })
.map_err(ApiError::internal)?;
Ok(Html(page).into_response())
}
async fn repositories_page(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Response, ApiError> {
if !valid_session(&headers, &state) {
return Ok(Redirect::to("/login").into_response());
}
let page = state
.templates
.get_template("repositories.html")
.map_err(ApiError::internal)?
.render(context! { version => env!("CARGO_PKG_VERSION") })
.map_err(ApiError::internal)?;
Ok(Html(page).into_response())
}
async fn repository_dashboard(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
) -> Result<Response, ApiError> {
if !valid_session(&headers, &state) {
return Ok(Redirect::to("/login").into_response());
}
let repository = state
.database
.repository(&repository)
.map_err(ApiError::bad)?;
let page = state
.templates
.get_template("repository-app.html")
.map_err(ApiError::internal)?
.render(context! {
version => env!("CARGO_PKG_VERSION"),
repository => repository
})
.map_err(ApiError::internal)?;
Ok(Html(page).into_response())
}
async fn runs_page(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Response, ApiError> {
if !valid_session(&headers, &state) {
return Ok(Redirect::to("/login").into_response());
}
let page = state
.templates
.get_template("runs.html")
.map_err(ApiError::internal)?
.render(context! { version => env!("CARGO_PKG_VERSION") })
.map_err(ApiError::internal)?;
Ok(Html(page).into_response())
}
async fn run_page(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<i64>,
) -> Result<Response, ApiError> {
if !valid_session(&headers, &state) {
return Ok(Redirect::to("/login").into_response());
}
let page = state
.templates
.get_template("run.html")
.map_err(ApiError::internal)?
.render(context! { version => env!("CARGO_PKG_VERSION"), run_id => id })
.map_err(ApiError::internal)?;
Ok(Html(page).into_response())
}
async fn workers_page(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Response, ApiError> {
if !valid_session(&headers, &state) {
return Ok(Redirect::to("/login").into_response());
}
let page = state
.templates
.get_template("workers.html")
.map_err(ApiError::internal)?
.render(context! { version => env!("CARGO_PKG_VERSION") })
.map_err(ApiError::internal)?;
Ok(Html(page).into_response())
}
async fn login(State(state): State<AppState>) -> Result<Response, ApiError> {
let config = IdpConfig::from_env(&state.public_origin);
if config.client_id.is_empty() || config.client_secret.is_empty() {
return Err(ApiError::internal("AkurAI Auth client is not configured"));
}
let login_state = login_state(&state)?;
let page = state
.templates
.get_template("login.html")
.map_err(ApiError::internal)?
.render(context! { authorize_url => config.authorize_url(&login_state) })
.map_err(ApiError::internal)?;
let mut response = Html(page).into_response();
response.headers_mut().insert(
header::SET_COOKIE,
HeaderValue::from_str(&format!(
"ab_oidc_state={login_state}; Path=/auth/callback; HttpOnly; Secure; SameSite=Lax; Max-Age=300"
))
.map_err(ApiError::internal)?,
);
Ok(response)
}
async fn auth_callback(
State(state): State<AppState>,
headers: HeaderMap,
Query(query): Query<CallbackQuery>,
) -> Result<Response, ApiError> {
let code = query.code.filter(|code| !code.is_empty()).ok_or_else(|| {
ApiError::bad(
query
.error
.unwrap_or_else(|| "authentication failed".into()),
)
})?;
let returned_state = query
.state
.ok_or_else(|| ApiError::bad("missing authentication state"))?;
let expected_state = request_cookie(&headers, "ab_oidc_state")
.ok_or_else(|| ApiError::bad("missing authentication state"))?;
if !constant_time_equal(returned_state.as_bytes(), expected_state.as_bytes())
|| !valid_login_state(expected_state, &state)
{
return Err(ApiError::bad("invalid authentication state"));
}
let config = IdpConfig::from_env(&state.public_origin);
if config.client_id.is_empty() || config.client_secret.is_empty() {
return Err(ApiError::internal("AkurAI Auth client is not configured"));
}
let user = idp_user(config.clone(), code).await?;
let role = if config.admin_emails.iter().any(|email| email == &user.email) {
"admin"
} else {
"member"
};
let mut response = Redirect::to("/app").into_response();
response.headers_mut().append(
header::SET_COOKIE,
HeaderValue::from_str(&session_cookie(&user, role, &state)?).map_err(ApiError::internal)?,
);
response.headers_mut().append(
header::SET_COOKIE,
HeaderValue::from_static(
"ab_oidc_state=; Path=/auth/callback; HttpOnly; Secure; SameSite=Lax; Max-Age=0",
),
);
Ok(response)
}
async fn idp_user(config: IdpConfig, code: String) -> Result<UserInfo, ApiError> {
tokio::task::spawn_blocking(move || idp_user_blocking(config, code))
.await
.map_err(ApiError::internal)?
}
fn idp_user_blocking(config: IdpConfig, code: String) -> Result<UserInfo, ApiError> {
let body = serde_json::to_vec(&serde_json::json!({
"grant_type": "authorization_code",
"code": code,
"redirect_uri": config.redirect_uri,
"client_id": config.client_id,
"client_secret": config.client_secret,
}))
.map_err(ApiError::internal)?;
let token: TokenResponse = serde_json::from_slice(&curl_with_stdin(
&[
"--fail",
"--silent",
"--show-error",
"--max-time",
"15",
"--header",
"Content-Type: application/json",
"--data-binary",
"@-",
&config.token_url(),
],
&body,
)?)
.map_err(ApiError::internal)?;
let userinfo_config = format!(
"url = \"{}\"\nheader = \"Authorization: Bearer {}\"\n",
config.userinfo_url(),
token.access_token
);
serde_json::from_slice(&curl_with_stdin(
&[
"--fail",
"--silent",
"--show-error",
"--max-time",
"15",
"--config",
"-",
],
userinfo_config.as_bytes(),
)?)
.map_err(ApiError::internal)
}
fn curl_with_stdin(arguments: &[&str], stdin: &[u8]) -> Result<Vec<u8>, ApiError> {
let mut child = Command::new("curl")
.args(arguments)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.map_err(ApiError::internal)?;
child
.stdin
.take()
.ok_or_else(|| ApiError::internal("curl stdin unavailable"))?
.write_all(stdin)
.map_err(ApiError::internal)?;
let output = child.wait_with_output().map_err(ApiError::internal)?;
if output.status.success() {
Ok(output.stdout)
} else {
Err(ApiError::bad("identity provider request failed"))
}
}
async fn css() -> Response {
asset(CSS, "text/css; charset=utf-8")
}
async fn js() -> Response {
asset(JS, "text/javascript; charset=utf-8")
}
fn asset(body: &'static str, content_type: &'static str) -> Response {
let mut response = Response::new(Body::from(body));
response
.headers_mut()
.insert(header::CONTENT_TYPE, HeaderValue::from_static(content_type));
response.headers_mut().insert(
header::CACHE_CONTROL,
HeaderValue::from_static("public, max-age=300"),
);
response
}
async fn health(State(state): State<AppState>) -> Result<Json<Envelope<Value>>, ApiError> {
state.database.check_ready().map_err(ApiError::internal)?;
let workers = state.database.workers().map_err(ApiError::internal)?;
let active_workers = workers
.iter()
.filter(|worker| worker.status == "running")
.count();
let queued_runs = state.database.queued_runs().map_err(ApiError::internal)?;
Ok(ok(serde_json::json!({
"service": "akurai-build",
"status": "ok",
"version": env!("CARGO_PKG_VERSION"),
"workers": workers.len(),
"active_workers": active_workers,
"queued_runs": queued_runs,
})))
}
async fn api_state(
State(state): State<AppState>,
headers: HeaderMap,
Query(query): Query<StateQuery>,
) -> Result<Json<Envelope<StateResponse>>, ApiError> {
authorize(&headers, &state)?;
let repositories = state.database.repositories().map_err(ApiError::internal)?;
let runs = state
.database
.query_runs(&crate::db::RunQuery {
repository: query.repo,
statuses: list_filter(query.status),
git_ref: query.git_ref,
triggers: list_filter(query.trigger),
search: query.q,
limit: query.limit.unwrap_or(50),
offset: query.offset.unwrap_or(0),
})
.map_err(ApiError::bad)?;
let workers = state.database.workers().map_err(ApiError::internal)?;
let queued_runs = state.database.queued_runs().map_err(ApiError::internal)?;
Ok(ok(StateResponse {
repositories,
runs,
workers,
queued_runs,
}))
}
async fn api_workers(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Json<Envelope<Vec<crate::db::Worker>>>, ApiError> {
authorize(&headers, &state)?;
Ok(ok(state.database.workers().map_err(ApiError::internal)?))
}
async fn run_detail(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<i64>,
) -> Result<Json<Envelope<crate::db::RunDetail>>, ApiError> {
authorize(&headers, &state)?;
Ok(ok(state.database.detail(id).map_err(ApiError::bad)?))
}
async fn run_events(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<i64>,
) -> Result<Response, ApiError> {
authorize(&headers, &state)?;
state.database.detail(id).map_err(ApiError::bad)?;
let database = state.database.clone();
let (reader, mut writer) = tokio::io::duplex(64 * 1024);
tokio::spawn(async move {
let mut previous = String::new();
let mut ticker = tokio::time::interval(Duration::from_millis(500));
loop {
ticker.tick().await;
let detail = match database.detail(id) {
Ok(detail) => detail,
Err(error) => {
let payload = serde_json::json!({"run_id": id, "status": "failed", "error": error.to_string()});
let _ = writer
.write_all(format!("event: error\ndata: {payload}\n\n").as_bytes())
.await;
break;
}
};
let terminal = is_terminal_run(&detail.run.status);
let payload = match serde_json::to_string(&run_progress(detail)) {
Ok(payload) => payload,
Err(_) => break,
};
if payload != previous {
let frame = format!("event: progress\ndata: {payload}\n\n");
if writer.write_all(frame.as_bytes()).await.is_err() {
break;
}
previous = payload;
} else if writer.write_all(b": keepalive\n\n").await.is_err() {
break;
}
if terminal {
break;
}
}
});
let mut response = Response::new(Body::from_stream(ReaderStream::new(reader)));
response.headers_mut().insert(
header::CONTENT_TYPE,
HeaderValue::from_static("text/event-stream"),
);
response.headers_mut().insert(
header::CACHE_CONTROL,
HeaderValue::from_static("no-cache, no-transform"),
);
response.headers_mut().insert(
header::CONTENT_ENCODING,
HeaderValue::from_static("identity"),
);
response.headers_mut().insert(
HeaderName::from_static("x-accel-buffering"),
HeaderValue::from_static("no"),
);
Ok(response)
}
fn run_progress(detail: crate::db::RunDetail) -> RunProgress {
let job_count = detail.jobs.len() as u32;
let job_units = detail
.jobs
.iter()
.map(|job| match job.status.as_str() {
"succeeded" | "failed" | "canceled" | "skipped" => 4,
"waiting" => 3,
"running" => 2,
_ => 0,
})
.sum::<u32>();
let progress = match detail.run.status.as_str() {
"succeeded" => 100,
"queued" => 2,
"waiting" => 85,
_ if job_count > 0 => (5 + (80 * job_units / (job_count * 4))).min(95) as u8,
_ => 5,
};
let stage = if detail.run.status == "succeeded" {
"succeeded".to_string()
} else if is_terminal_run(&detail.run.status) {
detail.run.status.clone()
} else if detail.run.status == "waiting"
|| detail
.jobs
.iter()
.any(|job| job.status == "waiting" && job.approval_required)
{
"waiting_for_approval".to_string()
} else if let Some(deployment) = detail
.deployments
.iter()
.rev()
.find(|deployment| deployment.status != "succeeded")
{
format!("deploying:{}", deployment.environment)
} else if let Some(job) = detail.jobs.iter().find(|job| job.status == "running") {
format!("running:{}", job.name)
} else {
detail.run.status.clone()
};
RunProgress {
run_id: detail.run.id,
repository: detail.run.repository,
git_ref: detail.run.git_ref,
commit_sha: detail.run.commit_sha,
status: detail.run.status,
stage,
progress,
jobs: detail
.jobs
.into_iter()
.map(|job| ProgressJob {
id: job.id,
name: job.name,
status: job.status,
environment: job.environment,
approval_required: job.approval_required,
})
.collect(),
deployments: detail
.deployments
.into_iter()
.map(|deployment| ProgressDeployment {
id: deployment.id,
environment: deployment.environment,
status: deployment.status,
})
.collect(),
error: detail.run.error,
}
}
fn is_terminal_run(status: &str) -> bool {
matches!(status, "succeeded" | "failed" | "canceled" | "interrupted")
}
async fn create_repository(
State(state): State<AppState>,
headers: HeaderMap,
Json(request): Json<RepositoryRequest>,
) -> Result<Json<Envelope<crate::db::Repository>>, ApiError> {
authorize_mutation(&headers, &state)?;
config::validate_repository(&request.name, &request.url, &request.branch)
.map_err(ApiError::bad)?;
Ok(ok(state
.database
.add_repository(&request.name, &request.url, &request.branch)
.map_err(ApiError::bad)?))
}
async fn change_repository_visibility(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
Json(request): Json<VisibilityRequest>,
) -> Result<Json<Envelope<crate::db::Repository>>, ApiError> {
authorize_mutation(&headers, &state)?;
Ok(ok(state
.database
.set_repository_visibility(&repository, &request.visibility)
.map_err(ApiError::bad)?))
}
async fn trigger(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
Json(request): Json<TriggerRequest>,
) -> Result<Json<Envelope<IdResponse>>, ApiError> {
authorize_mutation(&headers, &state)?;
let id = state
.runner
.queue(
&repository,
request.git_ref.as_deref(),
request.commit.as_deref(),
"manual",
)
.map_err(ApiError::bad)?;
Ok(ok(IdResponse { id }))
}
async fn branches(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
) -> Result<Json<Envelope<Vec<BranchInfo>>>, ApiError> {
authorize(&headers, &state)?;
let repo = state
.database
.repository(&repository)
.map_err(|_| ApiError::bad("repository not found"))?;
let output = std::process::Command::new("git")
.args(["ls-remote", "--heads", &repo.url])
.env("GIT_TERMINAL_PROMPT", "0")
.output()
.map_err(|e| ApiError::bad(format!("failed to list branches: {e}")))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
return Err(ApiError::bad(format!(
"git ls-remote failed: {}",
stderr.lines().next().unwrap_or("unknown error")
)));
}
let mut names: Vec<String> = String::from_utf8(output.stdout)
.map_err(|e| ApiError::bad(format!("invalid git output: {e}")))?
.lines()
.filter_map(|line| line.split('\t').nth(1))
.filter_map(|refname| refname.strip_prefix("refs/heads/"))
.map(str::to_owned)
.collect();
names.sort();
let branches: Vec<BranchInfo> = names
.into_iter()
.map(|name| {
let is_default = name == repo.default_branch;
BranchInfo { name, is_default }
})
.collect();
Ok(ok(branches))
}
#[derive(Deserialize)]
struct TreeQuery {
#[serde(rename = "ref")]
reference: Option<String>,
path: Option<String>,
}
async fn repo_tree(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
Query(query): Query<TreeQuery>,
) -> Result<Json<Envelope<crate::tree::TreeResponse>>, ApiError> {
authorize(&headers, &state)?;
let repo = state
.database
.repository(&repository)
.map_err(|_| ApiError::bad("repository not found"))?;
let sub_path = query.path.unwrap_or_default();
let response = state
.tree
.tree(&repo.name, &repo.url, query.reference.as_deref(), &sub_path)
.map_err(ApiError::bad)?;
Ok(ok(response))
}
async fn retry(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<i64>,
) -> Result<Json<Envelope<IdResponse>>, ApiError> {
authorize_mutation(&headers, &state)?;
let id = state.database.retry(id).map_err(ApiError::bad)?;
Ok(ok(IdResponse { id }))
}
async fn promote(
State(state): State<AppState>,
headers: HeaderMap,
Path((id, environment)): Path<(i64, String)>,
) -> Result<Json<Envelope<Value>>, ApiError> {
authorize_mutation(&headers, &state)?;
let jobs = state
.database
.approve_environment(id, &environment)
.map_err(ApiError::bad)?;
Ok(ok(
serde_json::json!({"run_id":id,"environment":environment,"jobs":jobs}),
))
}
async fn artifact(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<i64>,
) -> Result<Response, ApiError> {
authorize(&headers, &state)?;
let artifact = state.database.artifact(id).map_err(ApiError::bad)?;
let path = state
.runner
.artifact_path(&artifact)
.map_err(ApiError::internal)?;
let file = tokio::fs::File::open(path)
.await
.map_err(ApiError::internal)?;
let mut response = Response::new(Body::from_stream(ReaderStream::new(file)));
response.headers_mut().insert(
header::CONTENT_TYPE,
HeaderValue::from_static("application/octet-stream"),
);
response.headers_mut().insert(
header::CONTENT_LENGTH,
HeaderValue::from_str(&artifact.bytes.to_string()).map_err(ApiError::internal)?,
);
let filename = artifact
.name
.rsplit('/')
.next()
.unwrap_or("artifact")
.replace(['"', '\r', '\n'], "_");
response.headers_mut().insert(
header::CONTENT_DISPOSITION,
HeaderValue::from_str(&format!("attachment; filename=\"{filename}\""))
.map_err(ApiError::internal)?,
);
Ok(response)
}
async fn webhook(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
body: Bytes,
) -> Result<Json<Envelope<IdResponse>>, ApiError> {
authorize_webhook(&headers, &state, &body)?;
let payload: Value = serde_json::from_slice(&body).map_err(ApiError::bad)?;
let git_ref = payload
.get("ref")
.and_then(Value::as_str)
.context("webhook has no ref")
.map_err(ApiError::bad)?;
let commit = payload
.get("after")
.or_else(|| payload.get("checkout_sha"))
.and_then(Value::as_str)
.or_else(|| {
payload
.get("head_commit")
.and_then(|value| value.get("id"))
.and_then(Value::as_str)
});
let commit = commit.filter(|value| value.bytes().any(|byte| byte != b'0'));
let id = state
.runner
.queue(
&repository,
Some(git_ref.strip_prefix("refs/heads/").unwrap_or(git_ref)),
commit,
"webhook",
)
.map_err(ApiError::bad)?;
Ok(ok(IdResponse { id }))
}
fn login_state(state: &AppState) -> Result<String, ApiError> {
let mut nonce = [0_u8; 16];
getrandom::fill(&mut nonce).map_err(ApiError::internal)?;
let payload = format!("{}:{}", encode_hex(&nonce), unix_time()?);
Ok(format!("{payload}.{}", sign(&payload, state)?))
}
fn list_filter(value: Option<String>) -> Vec<String> {
value
.into_iter()
.flat_map(|value| {
value
.split(',')
.map(str::trim)
.filter(|item| !item.is_empty())
.map(str::to_owned)
.collect::<Vec<_>>()
})
.collect()
}
fn valid_login_state(value: &str, state: &AppState) -> bool {
let Some((payload, supplied)) = value.rsplit_once('.') else {
return false;
};
let Some((_, created)) = payload.rsplit_once(':') else {
return false;
};
let Ok(created) = created.parse::<u64>() else {
return false;
};
unix_time().is_ok_and(|now| now.saturating_sub(created) <= 300)
&& sign(payload, state)
.is_ok_and(|expected| constant_time_equal(supplied.as_bytes(), expected.as_bytes()))
}
fn session_cookie(user: &UserInfo, role: &str, state: &AppState) -> Result<String, ApiError> {
let payload = format!(
"{}:{}:{}:{}",
unix_time()? + 3600,
user.sub,
user.email,
role
);
let value = format!(
"{}.{}",
encode_hex(payload.as_bytes()),
sign(&payload, state)?
);
Ok(format!(
"ab_session={value}; Path=/; HttpOnly; Secure; SameSite=Lax; Max-Age=3600"
))
}
fn request_cookie<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> {
headers
.get(header::COOKIE)?
.to_str()
.ok()?
.split(';')
.map(str::trim)
.find_map(|pair| pair.strip_prefix(name)?.strip_prefix('='))
}
fn unix_time() -> Result<u64, ApiError> {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_secs())
.map_err(ApiError::internal)
}
fn sign(value: &str, state: &AppState) -> Result<String, ApiError> {
sign_with_secret(value, &state.token)
}
fn sign_with_secret(value: &str, secret: &str) -> Result<String, ApiError> {
let mut mac = HmacSha256::new_from_slice(secret.as_bytes()).map_err(ApiError::internal)?;
mac.update(value.as_bytes());
Ok(encode_hex(&mac.finalize().into_bytes()))
}
fn encode_hex(value: &[u8]) -> String {
value.iter().map(|byte| format!("{byte:02x}")).collect()
}
fn authorize(headers: &HeaderMap, state: &AppState) -> Result<(), ApiError> {
if let Some(supplied) = headers
.get(header::AUTHORIZATION)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.strip_prefix("Bearer "))
{
return constant_time_equal(supplied.as_bytes(), state.token.as_bytes())
.then_some(())
.ok_or_else(ApiError::unauthorized);
}
valid_session(headers, state)
.then_some(())
.ok_or_else(ApiError::unauthorized)
}
fn valid_session(headers: &HeaderMap, state: &AppState) -> bool {
request_cookie(headers, "ab_session")
.is_some_and(|cookie| valid_session_cookie(cookie, &state.token))
}
fn valid_session_cookie(cookie: &str, secret: &str) -> bool {
let Some((encoded_payload, supplied_signature)) = cookie.rsplit_once('.') else {
return false;
};
let Some(payload) = hex_to_text(encoded_payload) else {
return false;
};
let Ok(expected_signature) = sign_with_secret(&payload, secret) else {
return false;
};
if !constant_time_equal(supplied_signature.as_bytes(), expected_signature.as_bytes()) {
return false;
}
let mut fields = payload.splitn(4, ':');
let Some(expires_at) = fields.next().and_then(|value| value.parse::<u64>().ok()) else {
return false;
};
unix_time().is_ok_and(|now| expires_at >= now)
&& fields.next().is_some_and(|value| !value.is_empty())
&& fields.next().is_some_and(|value| !value.is_empty())
&& fields
.next()
.is_some_and(|value| matches!(value, "admin" | "member"))
}
fn hex_to_text(value: &str) -> Option<String> {
if value.len() > 1024 || !value.len().is_multiple_of(2) {
return None;
}
value
.as_bytes()
.chunks_exact(2)
.map(|pair| {
std::str::from_utf8(pair)
.ok()
.and_then(|pair| u8::from_str_radix(pair, 16).ok())
})
.collect::<Option<Vec<_>>>()
.and_then(|bytes| String::from_utf8(bytes).ok())
}
fn authorize_mutation(headers: &HeaderMap, state: &AppState) -> Result<(), ApiError> {
authorize(headers, state)?;
if let Some(origin) = headers
.get(header::ORIGIN)
.and_then(|value| value.to_str().ok())
&& origin != state.public_origin.as_ref()
{
return Err(ApiError::bad("origin is not allowed"));
}
Ok(())
}
fn authorize_webhook(headers: &HeaderMap, state: &AppState, body: &[u8]) -> Result<(), ApiError> {
if let Some(token) = headers
.get("x-gitlab-token")
.and_then(|value| value.to_str().ok())
&& constant_time_equal(token.as_bytes(), state.webhook_secret.as_bytes())
{
return Ok(());
}
for name in ["x-hub-signature-256", "x-gitea-signature"] {
if let Some(signature) = headers.get(name).and_then(|value| value.to_str().ok()) {
let signature = signature.strip_prefix("sha256=").unwrap_or(signature);
let supplied = decode_hex(signature).map_err(ApiError::bad)?;
let mut mac = HmacSha256::new_from_slice(state.webhook_secret.as_bytes())
.map_err(ApiError::internal)?;
mac.update(body);
if constant_time_equal(&supplied, &mac.finalize().into_bytes()) {
return Ok(());
}
}
}
Err(ApiError::unauthorized())
}
fn decode_hex(value: &str) -> Result<Vec<u8>> {
ensure!(
value.len().is_multiple_of(2) && value.len() <= 128,
"invalid signature"
);
value
.as_bytes()
.chunks_exact(2)
.map(|pair| {
let text = std::str::from_utf8(pair)?;
Ok(u8::from_str_radix(text, 16)?)
})
.collect()
}
fn constant_time_equal(left: &[u8], right: &[u8]) -> bool {
left.len() == right.len() && left.ct_eq(right).into()
}
fn ok<T: Serialize>(data: T) -> Json<Envelope<T>> {
Json(Envelope { ok: true, data })
}
async fn security_headers(request: Request<Body>, next: Next) -> Response {
let mut response = next.run(request).await;
let is_json = response
.headers()
.get(header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.is_some_and(|value| value.contains("json"));
let headers = response.headers_mut();
headers.insert(
header::X_CONTENT_TYPE_OPTIONS,
HeaderValue::from_static("nosniff"),
);
headers.insert(header::X_FRAME_OPTIONS, HeaderValue::from_static("DENY"));
headers.insert(
header::REFERRER_POLICY,
HeaderValue::from_static("no-referrer"),
);
headers.insert(header::CONTENT_SECURITY_POLICY, HeaderValue::from_static("default-src 'self'; base-uri 'none'; object-src 'none'; frame-ancestors 'none'; form-action 'self'; script-src 'self'; style-src 'self'; connect-src 'self'"));
headers.insert(
HeaderName::from_static("permissions-policy"),
HeaderValue::from_static("camera=(), geolocation=(), microphone=()"),
);
headers.insert(
HeaderName::from_static("cross-origin-opener-policy"),
HeaderValue::from_static("same-origin"),
);
headers.insert(
HeaderName::from_static("cross-origin-resource-policy"),
HeaderValue::from_static("same-origin"),
);
if is_json {
headers.insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
}
response
}
async fn serve_bounded(listener: tokio::net::TcpListener, app: Router) -> Result<()> {
use hyper_util::{
rt::{TokioExecutor, TokioIo, TokioTimer},
server::{conn::auto::Builder, graceful::GracefulShutdown},
service::TowerToHyperService,
};
info!(address = %listener.local_addr()?, "AkurAI Build listening");
let connections = Arc::new(tokio::sync::Semaphore::new(MAX_CONNECTIONS));
let graceful = GracefulShutdown::new();
let mut shutdown = std::pin::pin!(shutdown_signal());
loop {
let permit = tokio::select! { () = &mut shutdown => break, permit = connections.clone().acquire_owned() => permit? };
let (socket, _) = tokio::select! { () = &mut shutdown => break, accepted = listener.accept() => accepted? };
let mut builder = Builder::new(TokioExecutor::new());
builder
.http1()
.timer(TokioTimer::new())
.header_read_timeout(REQUEST_TIMEOUT);
let connection = graceful.watch(
builder
.serve_connection_with_upgrades(
TokioIo::new(socket),
TowerToHyperService::new(app.clone()),
)
.into_owned(),
);
tokio::spawn(async move {
if let Err(error) = connection.await {
tracing::debug!(?error, "connection closed");
}
drop(permit);
});
}
drop(listener);
tokio::select! { () = graceful.shutdown() => {}, () = tokio::time::sleep(REQUEST_TIMEOUT) => warn!("graceful shutdown timed out") }
Ok(())
}
async fn shutdown_signal() {
let ctrl_c = async {
if let Err(error) = tokio::signal::ctrl_c().await {
warn!(?error, "failed to install Ctrl-C handler");
}
};
#[cfg(unix)]
let terminate = async {
match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
Ok(mut signal) => {
signal.recv().await;
}
Err(error) => warn!(?error, "failed to install terminate handler"),
}
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! { () = ctrl_c => {}, () = terminate => {} }
}
#[cfg(test)]
mod tests {
use super::*;
use tower::Service;
#[test]
fn templates_compile_and_render_without_inline_code() -> Result<()> {
let mut environment = Environment::new();
environment.add_template("layouts/base.html", BASE)?;
environment.add_template("landing.html", LANDING)?;
environment.add_template("dashboard.html", DASHBOARD)?;
environment.add_template("repositories.html", REPOSITORIES)?;
environment.add_template("repository-app.html", REPOSITORY_APP)?;
environment.add_template("runs.html", RUNS)?;
environment.add_template("workers.html", WORKERS)?;
environment.add_template("login.html", LOGIN)?;
environment.add_template("run.html", RUN)?;
environment.add_template("profile.html", PROFILE)?;
environment.add_template("admin.html", ADMIN)?;
environment.add_template("repository.html", REPOSITORY)?;
let page = environment.get_template("landing.html")?.render(
context! { version => "test", repositories => Vec::<crate::db::Repository>::new() },
)?;
assert!(page.contains("AkurAI Build"));
assert!(!page.contains("<script>"));
Ok(())
}
#[test]
fn authentication_helpers_fail_closed() {
assert!(constant_time_equal(b"same", b"same"));
assert!(!constant_time_equal(b"same", b"different"));
assert!(decode_hex("not-hex").is_err());
}
#[test]
fn signed_browser_session_authorizes_without_bearer_token() {
let secret = "test-secret";
let payload = "4102444800:user-1:olafurbui@example.com:member";
let signature = match sign_with_secret(payload, secret) {
Ok(signature) => signature,
Err(_) => panic!("valid test signing key"),
};
let cookie = format!("{}.{}", encode_hex(payload.as_bytes()), signature);
assert!(valid_session_cookie(&cookie, secret));
assert!(!valid_session_cookie(&cookie, "other-secret"));
assert!(!valid_session_cookie("00.invalid", secret));
}
#[test]
fn git_smart_http_paths_and_cgi_responses_are_bounded() -> Result<()> {
validate_git_path("tasks.git/info/refs")?;
assert!(validate_git_path("../tasks.git/info/refs").is_err());
assert!(validate_git_path("tasks/info/refs").is_err());
let response = git_cgi_response(
b"Status: 200 OK\r\nContent-Type: application/x-git-upload-pack-result\r\n\r\npack",
)?;
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response.headers().get(header::CONTENT_TYPE),
Some(&HeaderValue::from_static(
"application/x-git-upload-pack-result"
))
);
Ok(())
}
// ── constant_time_equal boundaries ──
#[test]
fn constant_time_equal_rejects_empty_input() {
assert!(!constant_time_equal(b"", b"token"));
assert!(!constant_time_equal(b"token", b""));
}
#[test]
fn constant_time_equal_rejects_wrong_token() {
assert!(!constant_time_equal(b"correct", b"wrong-token"));
assert!(!constant_time_equal(b"xyz", b"abc"));
}
#[test]
fn constant_time_equal_rejects_too_short_token() {
assert!(!constant_time_equal(b"secret", b"sec"));
}
#[test]
fn constant_time_equal_rejects_correct_prefix_but_longer_token() {
// "secret" is a prefix of "secret-extended" — must be rejected
assert!(!constant_time_equal(b"secret", b"secret-extended"));
assert!(!constant_time_equal(b"secret-extended", b"secret"));
}
#[test]
fn constant_time_equal_accepts_exact_match_only() {
assert!(constant_time_equal(
b"exact-token-value",
b"exact-token-value"
));
assert!(constant_time_equal(b"a", b"a"));
}
// ── decode_hex boundaries ──
#[test]
fn decode_hex_rejects_odd_length() {
assert!(decode_hex("abc").is_err());
}
#[test]
fn decode_hex_rejects_non_hex_characters() {
assert!(decode_hex("zz").is_err());
assert!(decode_hex("gg12").is_err());
}
#[test]
fn decode_hex_accepts_valid_hex() -> Result<()> {
assert_eq!(decode_hex("48656c6c6f")?, b"Hello");
assert_eq!(decode_hex("00ff")?, vec![0x00, 0xff]);
assert_eq!(decode_hex("")?, Vec::<u8>::new());
Ok(())
}
// ── authorize header parsing ──
fn test_state(token: &str, webhook_secret: &str) -> AppState {
const KEY: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
let temp = tempfile::tempdir().expect("tempdir");
let database = Database::memory(KEY).expect("database");
let runner = Runner::new(database.clone(), temp.path().to_owned(), false).expect("runner");
let tree = RepoTree::new(temp.path()).expect("tree");
let mut env = Environment::new();
env.add_template("layouts/base.html", BASE).ok();
AppState {
database,
runner,
tree,
hosted_root: temp.path().to_owned(),
token: Arc::from(token),
webhook_secret: Arc::from(webhook_secret),
public_origin: Arc::from("https://build.example"),
templates: Arc::new(env),
}
}
#[test]
fn authorize_rejects_missing_authorization_header() {
let state = test_state("api-token-123", "wh-secret");
let headers = HeaderMap::new();
let result = authorize(&headers, &state);
assert!(result.is_err());
assert_eq!(
result.expect_err("should be unauthorized").status,
StatusCode::UNAUTHORIZED
);
}
#[test]
fn authorize_rejects_malformed_scheme_not_bearer() {
let state = test_state("api-token-123", "wh-secret");
let mut headers = HeaderMap::new();
headers.insert(
header::AUTHORIZATION,
HeaderValue::from_static("Basic some-value"),
);
let result = authorize(&headers, &state);
assert!(result.is_err());
}
#[test]
fn authorize_rejects_bearer_with_wrong_token() {
let state = test_state("api-token-123", "wh-secret");
let mut headers = HeaderMap::new();
headers.insert(
header::AUTHORIZATION,
HeaderValue::from_static("Bearer wrong-value"),
);
let result = authorize(&headers, &state);
assert!(result.is_err());
}
#[test]
fn authorize_accepts_correct_bearer_token() {
let state = test_state("api-token-123", "wh-secret");
let mut headers = HeaderMap::new();
headers.insert(
header::AUTHORIZATION,
HeaderValue::from_static("Bearer api-token-123"),
);
assert!(authorize(&headers, &state).is_ok());
}
#[test]
fn authorize_webhook_rejects_missing_header() {
let state = test_state("api-token", "webhook-secret");
let headers = HeaderMap::new();
let result = authorize_webhook(&headers, &state, b"{}");
assert!(result.is_err());
assert_eq!(
result.expect_err("should be unauthorized").status,
StatusCode::UNAUTHORIZED
);
}
#[test]
fn authorize_webhook_rejects_wrong_token() {
let state = test_state("api-token", "webhook-secret");
let mut headers = HeaderMap::new();
headers.insert("x-gitlab-token", HeaderValue::from_static("wrong-token"));
let result = authorize_webhook(&headers, &state, b"{}");
assert!(result.is_err());
}
#[test]
fn authorize_webhook_accepts_correct_x_gitlab_token() {
let state = test_state("api-token", "webhook-secret");
let mut headers = HeaderMap::new();
headers.insert("x-gitlab-token", HeaderValue::from_static("webhook-secret"));
assert!(authorize_webhook(&headers, &state, b"{}").is_ok());
}
// ── ApiError envelope shape ──
#[tokio::test]
async fn unauthorized_error_response_has_correct_envelope_shape() {
let error = ApiError::unauthorized();
let response = error.into_response();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
let bytes = axum::body::to_bytes(response.into_body(), 4096)
.await
.expect("read body");
let body = String::from_utf8(bytes.to_vec()).expect("utf-8");
let parsed: serde_json::Value = serde_json::from_str(&body).expect("json");
assert_eq!(parsed["ok"], false);
assert_eq!(parsed["error"]["code"], "unauthorized");
assert!(
parsed["error"]["message"]
.as_str()
.is_some_and(|m| !m.is_empty())
);
}
// ── In-process app / router tests ──
fn test_router() -> Router {
let state = test_state("api-token-123", "webhook-secret");
Router::new()
.route("/api/health", get(health))
.route("/api/state", get(api_state))
.route("/api/runs/{id}/promote/{environment}", post(promote))
.route("/api/hooks/{repository}", post(webhook))
.route("/api/repos", post(create_repository))
.with_state(state)
.layer(DefaultBodyLimit::max(MAX_BODY))
.layer(middleware::from_fn(security_headers))
}
async fn json_body(response: Response) -> serde_json::Value {
let bytes = axum::body::to_bytes(response.into_body(), 64 * 1024)
.await
.expect("read body")
.to_vec();
serde_json::from_slice(&bytes).expect("json")
}
#[tokio::test]
async fn app_unauthenticated_api_returns_401() {
let mut app = test_router();
let request = Request::builder()
.uri("/api/state")
.body(Body::empty())
.expect("build request");
let response = app.call(request).await.expect("call");
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
let body = json_body(response).await;
assert_eq!(body["ok"], false);
assert_eq!(body["error"]["code"], "unauthorized");
}
#[tokio::test]
async fn app_health_is_public_200() {
let mut app = test_router();
let request = Request::builder()
.uri("/api/health")
.body(Body::empty())
.expect("build request");
let response = app.call(request).await.expect("call");
assert_eq!(response.status(), StatusCode::OK);
let body = json_body(response).await;
assert_eq!(body["ok"], true);
}
#[tokio::test]
async fn app_security_headers_are_present() {
let mut app = test_router();
let request = Request::builder()
.uri("/api/health")
.body(Body::empty())
.expect("build request");
let response = app.call(request).await.expect("call");
let headers = response.headers();
assert_eq!(
headers
.get(header::X_CONTENT_TYPE_OPTIONS)
.and_then(|v| v.to_str().ok()),
Some("nosniff")
);
assert_eq!(
headers
.get(header::X_FRAME_OPTIONS)
.and_then(|v| v.to_str().ok()),
Some("DENY")
);
let csp = headers
.get(header::CONTENT_SECURITY_POLICY)
.and_then(|v| v.to_str().ok())
.expect("CSP header");
assert!(
csp.contains("script-src 'self'"),
"CSP must include script-src 'self': {csp}"
);
}
#[tokio::test]
async fn app_webhook_rejects_bad_token() {
let mut app = test_router();
let request = Request::builder()
.uri("/api/hooks/my-repo")
.method("POST")
.header("x-gitlab-token", "wrong-value")
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(r#"{"ref":"refs/heads/main","after":"abc123"}"#))
.expect("build request");
let response = app.call(request).await.expect("call");
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}
#[tokio::test]
async fn app_promote_requires_authorization() {
let mut app = test_router();
let request = Request::builder()
.uri("/api/runs/1/promote/staging")
.method("POST")
.header(header::CONTENT_TYPE, "application/json")
.body(Body::empty())
.expect("build request");
let response = app.call(request).await.expect("call");
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}
#[tokio::test]
async fn app_body_limit_rejects_oversized_request() {
let state = test_state("api-token-123", "webhook-secret");
let mut app = Router::new()
.route("/api/repos", post(create_repository))
.with_state(state)
.layer(DefaultBodyLimit::max(8))
.layer(middleware::from_fn(security_headers));
let oversized = vec![b'a'; 64];
let request = Request::builder()
.uri("/api/repos")
.method("POST")
.header(header::CONTENT_TYPE, "application/json")
.header(
header::AUTHORIZATION,
HeaderValue::from_static("Bearer api-token-123"),
)
.body(Body::from(oversized))
.expect("build request");
let response = app.call(request).await.expect("call");
assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
}
#[tokio::test]
async fn app_unknown_route_returns_404() {
let mut app = test_router();
let request = Request::builder()
.uri("/nonexistent")
.body(Body::empty())
.expect("build request");
let response = app.call(request).await.expect("call");
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn app_wrong_method_returns_405() {
let mut app = test_router();
// GET /api/repos is not defined, only POST
let request = Request::builder()
.uri("/api/repos")
.body(Body::empty())
.expect("build request");
let response = app.call(request).await.expect("call");
assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED);
}
#[tokio::test]
async fn app_authorized_api_state_accepts_valid_token() {
let mut app = test_router();
let request = Request::builder()
.uri("/api/state")
.header(
header::AUTHORIZATION,
HeaderValue::from_static("Bearer api-token-123"),
)
.body(Body::empty())
.expect("build request");
let response = app.call(request).await.expect("call");
// Should be OK even though DB queries might fail on empty tables;
// the authorize gate passes so we either get 200 or a DB error,
// but never 401.
assert_ne!(response.status(), StatusCode::UNAUTHORIZED);
}
}