Menu
Public repository / 8d41f1e10aceed387d425b43038f2070e399a349
AkurAI-Build / src/server.rs
Latest change 8d41f1e10aceed387d425b43038f2070e399a349 - OLI-277: schedules, reusable workflows (include + inputs), per-repository secrets (value or PassVault), native caches by Ólafur Búi Ólafsson
use std::{
io::Write,
net::SocketAddr,
path::{Path as FsPath, 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, delete, get, post, put},
};
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::hosted_git;
use crate::linking;
use crate::notifications;
use crate::oauth::{Denial, ResourceServer};
use crate::protection;
use crate::webhooks;
use crate::{
config,
db::{Database, RepositoryDeleteError},
forks, git_process, history, mcp,
repo_delete::{self, RepositoryRemoval},
runner::Runner,
statuses,
tree::RepoTree,
};
mod access_api;
mod forks_api;
mod insights_api;
mod notifications_api;
mod protection_api;
mod pulls_api;
mod releases_api;
mod search_api;
mod secrets_api;
mod ssh_api;
mod webhooks_api;
pub use search_api::{CurlSearcher, IndexConfig, Searcher};
pub use ssh_api::{AuthorizedKeysSink, sink_for};
const MAX_BODY: usize = 1024 * 1024;
const MAX_GIT_BODY: usize = 512 * 1024 * 1024;
const MAX_CONNECTIONS: usize = 1024;
const MAX_BLOCKING_GIT: usize = 4;
const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
const LANDING: &str = include_str!("../app/pages/landing.html");
const DOCS: &str = include_str!("../app/pages/docs/index.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 REPOSITORY: &str = include_str!("../app/pages/repository.html");
const HISTORY: &str = include_str!("../app/pages/repository-history.html");
const RELEASES: &str = include_str!("../app/pages/repository-releases.html");
const NOTIFICATIONS: &str = include_str!("../app/pages/notifications.html");
const INSIGHTS: &str = include_str!("../app/pages/repository-insights.html");
const SEARCH: &str = include_str!("../app/pages/search.html");
const SSH_KEYS: &str = include_str!("../app/pages/ssh-keys.html");
const CODE: &str = include_str!("../app/pages/repository-code.html");
const RUN: &str = include_str!("../app/pages/run.html");
const BASE: &str = include_str!("../app/templates/layouts/base.html");
const DOCS_LAYOUT: &str = include_str!("../app/templates/layouts/docs.html");
const DOCS_NAV: &str = include_str!("../app/templates/partials/docs-nav.html");
const PUBLIC_WORDMARK: &str = include_str!("../app/templates/partials/public-wordmark.html");
const PUBLIC_HEADER: &str = include_str!("../app/templates/partials/public-header.html");
const PUBLIC_FOOTER: &str = include_str!("../app/templates/partials/public-footer.html");
const CODE_CONTENT: &str = include_str!("../app/templates/partials/code-content.html");
const NOT_FOUND: &str = include_str!("../app/pages/_404.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,
pub oauth_issuer: String,
pub oauth_audience: String,
/// Paperclip base URL and board key for issue linking (OLI-274); None = off.
pub paperclip: Option<(String, String)>,
/// Sender address and command for notification mail (OLI-275); None = feed only.
pub mail: Option<(String, String)>,
/// Git over SSH (OLI-276): the authorized_keys file the forge owns; None = off.
pub authorized_keys: Option<Arc<AuthorizedKeysSink>>,
/// `user@host` shown as the SSH clone address when SSH is on.
pub ssh_address: Option<String>,
/// akurai-index for cross-repository code search (OLI-276); None = off.
pub index: Option<IndexConfig>,
}
#[derive(Clone)]
struct AppState {
database: Database,
runner: Runner,
tree: RepoTree,
hosted_root: PathBuf,
blocking_git: BlockingGit,
token: Arc<str>,
webhook_secret: Arc<str>,
public_origin: Arc<str>,
templates: Arc<Environment<'static>>,
mcp: Arc<mcp::Service>,
oauth: Arc<ResourceServer>,
/// Issue tracker for commit and pull request links (OLI-274).
tracker: Arc<dyn linking::Tracker>,
/// Git over SSH (OLI-276); None when the forge has no authorized_keys file.
authorized_keys: Option<Arc<AuthorizedKeysSink>>,
ssh_address: Option<Arc<str>>,
/// Code search backend (OLI-276); None when not configured.
searcher: Option<Arc<dyn Searcher>>,
}
#[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>,
before: Option<i64>,
limit: Option<usize>,
offset: Option<usize>,
}
#[derive(Deserialize, Default)]
struct DeliveryMetricsQuery {
repo: Option<String>,
environment: Option<String>,
window: Option<i64>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct DeleteRepositoryQuery {
/// Must repeat the repository name; guards against accidental deletes.
confirm: Option<String>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct VisibilityRequest {
visibility: String,
}
#[derive(Serialize)]
struct RepositoryView {
id: i64,
name: String,
url: String,
default_branch: String,
visibility: String,
created_at: i64,
fork_of: Option<String>,
mirror_of: Option<String>,
synced_at: Option<i64>,
/// `user@host:Name.git`, only when the forge serves Git over SSH.
ssh_url: Option<String>,
}
fn repository_client_url(
repository_name: &str,
repository_url: &str,
data_root: &FsPath,
public_origin: &str,
) -> String {
let repository_path = FsPath::new(repository_url);
let hosted_root = data_root.join("hosted");
if repository_path.starts_with(&hosted_root)
&& repository_path
.parent()
.and_then(FsPath::file_name)
.is_some_and(|name| name == "hosted")
&& repository_path
.extension()
.is_some_and(|extension| extension == "git")
{
return format!(
"{}/git/{repository_name}.git",
public_origin.trim_end_matches('/')
);
}
repository_url.to_owned()
}
fn repository_view(
repository: crate::db::Repository,
data_root: &FsPath,
public_origin: &str,
) -> RepositoryView {
let url = repository_client_url(&repository.name, &repository.url, data_root, public_origin);
RepositoryView {
id: repository.id,
name: repository.name,
url,
default_branch: repository.default_branch,
visibility: repository.visibility,
created_at: repository.created_at,
fork_of: repository.fork_of,
mirror_of: repository.mirror_of,
synced_at: repository.synced_at,
ssh_url: None,
}
}
#[derive(Serialize)]
struct StateResponse {
repositories: Vec<RepositoryView>,
runs: Vec<crate::db::Run>,
workers: Vec<crate::db::Worker>,
queued_runs: i64,
run_cursor: Option<i64>,
run_total: i64,
}
#[derive(Serialize)]
struct CodeEntryView {
name: String,
kind: String,
size: u64,
commit_message: Option<String>,
href: String,
}
#[derive(Serialize)]
struct CodeBreadcrumb {
label: String,
href: String,
}
#[derive(Serialize)]
struct CodePage {
repository: RepositoryView,
reference: String,
path: String,
entries: Vec<CodeEntryView>,
breadcrumbs: Vec<CodeBreadcrumb>,
source: Option<String>,
is_file: bool,
parent_href: String,
latest_commit: Option<crate::tree::LatestCommit>,
commit_href: Option<String>,
public: bool,
}
#[derive(Serialize)]
struct IdResponse {
id: i64,
}
#[derive(Serialize)]
struct LandingRepository {
name: String,
default_branch: String,
owner: String,
run_count: i64,
successful_runs: i64,
}
#[derive(Serialize)]
struct PublicBuilder {
name: String,
initial: String,
repository_count: usize,
}
#[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 forbidden(message: impl Into<String>) -> Self {
Self {
status: StatusCode::FORBIDDEN,
code: "forbidden",
message: message.into(),
}
}
fn bad(error: impl std::fmt::Display) -> Self {
Self {
status: StatusCode::BAD_REQUEST,
code: "invalid_request",
message: error.to_string(),
}
}
fn not_found(message: impl Into<String>) -> Self {
Self {
status: StatusCode::NOT_FOUND,
code: "not_found",
message: message.into(),
}
}
fn conflict(message: impl Into<String>) -> Self {
Self {
status: StatusCode::CONFLICT,
code: "conflict",
message: message.into(),
}
}
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(),
}
}
}
#[derive(Clone)]
struct BlockingGit {
permits: Arc<tokio::sync::Semaphore>,
}
impl BlockingGit {
fn new(limit: usize) -> Self {
Self {
permits: Arc::new(tokio::sync::Semaphore::new(limit)),
}
}
async fn run<T, F>(&self, operation: F) -> Result<T, ApiError>
where
T: Send + 'static,
F: FnOnce() -> Result<T, ApiError> + Send + 'static,
{
let permit = self
.permits
.clone()
.acquire_owned()
.await
.map_err(ApiError::internal)?;
tokio::task::spawn_blocking(move || {
let _permit = permit;
operation()
})
.await
.map_err(ApiError::internal)?
}
}
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()
}
}
fn template_environment() -> Result<Environment<'static>> {
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("layouts/docs.html", DOCS_LAYOUT)?;
templates.add_template("partials/docs-nav.html", DOCS_NAV)?;
templates.add_template("partials/public-wordmark.html", PUBLIC_WORDMARK)?;
templates.add_template("partials/public-header.html", PUBLIC_HEADER)?;
templates.add_template("partials/public-footer.html", PUBLIC_FOOTER)?;
templates.add_template("partials/code-content.html", CODE_CONTENT)?;
templates.add_template("landing.html", LANDING)?;
templates.add_template("docs.html", DOCS)?;
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("run.html", RUN)?;
templates.add_template("repository.html", REPOSITORY)?;
templates.add_template("repository-code.html", CODE)?;
templates.add_template("repository-history.html", HISTORY)?;
templates.add_template("repository-releases.html", RELEASES)?;
templates.add_template("notifications.html", NOTIFICATIONS)?;
templates.add_template("repository-insights.html", INSIGHTS)?;
templates.add_template("search.html", SEARCH)?;
templates.add_template("ssh-keys.html", SSH_KEYS)?;
templates.add_template("not-found.html", NOT_FOUND)?;
Ok(templates)
}
fn request_prefers_html(headers: &HeaderMap) -> bool {
let Some(accept) = headers
.get(header::ACCEPT)
.and_then(|value| value.to_str().ok())
else {
return false;
};
let media_types = accept
.split(',')
.map(|value| value.split(';').next().unwrap_or_default().trim());
let mut html = false;
for media_type in media_types {
if media_type == "application/json" || media_type.ends_with("+json") {
return false;
}
html |= media_type == "text/html" || media_type == "application/xhtml+xml";
}
html
}
fn public_error_response(
templates: &Environment<'_>,
headers: &HeaderMap,
error: ApiError,
title: &str,
heading: &str,
description: &str,
) -> Result<Response, ApiError> {
if !request_prefers_html(headers) {
return Ok(error.into_response());
}
let status = error.status;
let page = templates
.get_template("not-found.html")
.map_err(ApiError::internal)?
.render(context! {
title => title,
heading => heading,
description => description
})
.map_err(ApiError::internal)?;
Ok((status, Html(page)).into_response())
}
async fn public_not_found(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Response, ApiError> {
public_error_response(
&state.templates,
&headers,
ApiError::not_found("route not found"),
"Page not found",
"This page could not be found.",
"The address may be outdated or mistyped. Return home or open the documentation.",
)
}
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!(
options.worker_count <= 16,
"worker count must be 0..=16 (0 = serve-only, no execution)"
);
database.validate_schema()?;
if options.worker_count > 0 {
// Recovery belongs to the executor role: only the worker instance may
// mark orphaned runs interrupted and reset worker rows. A serve-only
// restart must never clobber a live worker's in-flight run.
let recovered = database.recover_interrupted()?;
if recovered > 0 {
warn!(recovered, "marked interrupted work after restart");
}
database.reset_workers()?;
let configured: Vec<String> = (0..options.worker_count)
.map(|index| format!("titan-{}", index + 1))
.collect();
let pruned = database.prune_workers_except(&configured)?;
if pruned > 0 {
warn!(
pruned,
"removed stale worker rows beyond the configured count"
);
}
}
let templates = template_environment()?;
let hosted_root = options.data_root.clone();
match protection::seed_defaults(&database) {
Ok(added) if added > 0 => info!(added, "seeded default branch protection rules"),
Ok(_) => {}
Err(error) => warn!(%error, "could not seed branch protection rules"),
}
match hosted_git::install_hooks_all(&hosted_root.join("hosted")) {
Ok(installed) if installed > 0 => info!(installed, "installed pre-receive hooks"),
Ok(_) => {}
Err(error) => warn!(%error, "could not install pre-receive hooks"),
}
let tree = RepoTree::new(&options.data_root)?;
let runner = Runner::new(
database.clone(),
options.data_root.clone(),
options.allow_native,
)?
.with_admins(IdpConfig::from_env(&options.public_origin).admin_emails)
.with_vault(Arc::new(crate::secrets::CommandVault(
std::env::var("AKURAI_BUILD_PASSVAULT_CMD").unwrap_or_else(|_| "akurai-secret".into()),
)));
let mcp = mcp::Service::new(database.clone(), options.data_root, options.allow_native)?;
// Same admin allowlist and userinfo endpoint as browser login.
let idp = IdpConfig::from_env(&options.public_origin);
if idp.admin_emails.is_empty() {
warn!("AKURAI_BUILD_ADMIN_EMAILS is empty; IdP tokens cannot reach /mcp");
}
let oauth = ResourceServer::new(
&options.oauth_issuer,
&options.oauth_audience,
&options.public_origin,
)
.with_admins(idp.admin_emails.clone())
.with_userinfo_url(idp.userinfo_url());
let state = AppState {
database: database.clone(),
runner: runner.clone(),
tree,
blocking_git: BlockingGit::new(MAX_BLOCKING_GIT),
hosted_root,
token: options.token.into(),
webhook_secret: options.webhook_secret.into(),
public_origin: options.public_origin.into(),
templates: Arc::new(templates),
mcp: Arc::new(mcp),
oauth: Arc::new(oauth),
tracker: match &options.paperclip {
Some((url, token)) => {
info!(%url, "issue linking to Paperclip is on");
Arc::new(linking::PaperclipTracker::new(url, token))
}
None => Arc::new(linking::NoTracker),
},
authorized_keys: options.authorized_keys.clone(),
ssh_address: options.ssh_address.as_deref().map(Arc::from),
searcher: options
.index
.clone()
.map(|index| Arc::new(CurlSearcher(index)) as Arc<dyn Searcher>),
};
if state.authorized_keys.is_some() {
info!(address = ?options.ssh_address, "git over SSH is on");
ssh_api::publish_authorized_keys(&state);
}
if state.searcher.is_some() {
info!("code search through akurai-index is on");
}
{
// Scheduled runs: every cron in a hosted pipeline fires once per matching minute (OLI-277).
let database = database.clone();
let runner = runner.clone();
let root = state.hosted_root.clone();
tokio::spawn(async move {
let mut ticker = tokio::time::interval(Duration::from_secs(30));
loop {
ticker.tick().await;
let database = database.clone();
let runner = runner.clone();
let root = root.clone();
let result = tokio::task::spawn_blocking(move || {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |d| d.as_secs()) as i64;
crate::schedule::tick(&database, &runner, &root, now)
})
.await;
match result {
Ok(Ok(fired)) if !fired.is_empty() => {
for fire in fired {
info!(repository = %fire.repository, branch = %fire.branch, cron = %fire.cron, run = fire.run_id, "scheduled run queued");
}
}
Ok(Ok(_)) => {}
Ok(Err(error)) => warn!(%error, "schedule tick failed"),
Err(error) => warn!(%error, "schedule task failed"),
}
}
});
}
{
// Mirrors pull from their source every ten minutes (OLI-276).
let database = database.clone();
let root = state.hosted_root.clone();
tokio::spawn(async move {
let mut ticker = tokio::time::interval(Duration::from_secs(60));
loop {
ticker.tick().await;
let database = database.clone();
let root = root.clone();
let result = tokio::task::spawn_blocking(move || -> anyhow::Result<usize> {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |d| d.as_secs()) as i64;
let mut updated = 0;
for name in forks::due_mirrors(&database, now, 600)? {
match forks::sync(&database, &root, &name, now) {
Ok(outcome) if outcome.updated => updated += 1,
Ok(_) => {}
Err(error) => warn!(%error, repository = %name, "mirror sync failed"),
}
}
Ok(updated)
})
.await;
match result {
Ok(Ok(updated)) if updated > 0 => info!(updated, "mirrors synced"),
Ok(Ok(_)) => {}
Ok(Err(error)) => warn!(%error, "mirror sync round failed"),
Err(error) => warn!(%error, "mirror sync task failed"),
}
}
});
}
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));
}
if let Some((from, command)) = &options.mail {
match notifications::CommandMailer::new(from, command) {
Ok(mailer) => {
info!(%from, "notification mail is on");
let database = database.clone();
let mailer = Arc::new(mailer);
tokio::spawn(async move {
let mut ticker = tokio::time::interval(Duration::from_secs(30));
loop {
ticker.tick().await;
let database = database.clone();
let mailer = mailer.clone();
let result = tokio::task::spawn_blocking(move || {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |d| d.as_secs())
as i64;
notifications::mail_pending(&database, mailer.as_ref(), now, 50)
})
.await;
match result {
Ok(Ok((mailed, failed))) if mailed + failed > 0 => {
info!(mailed, failed, "notification mail");
}
Ok(Ok(_)) => {}
Ok(Err(error)) => warn!(%error, "notification mail round failed"),
Err(error) => warn!(%error, "notification mail task failed"),
}
}
});
}
Err(error) => warn!(%error, "notification mail is misconfigured; feed only"),
}
}
{
// Outbound webhooks: one delivery loop per process, every few seconds.
let database = database.clone();
tokio::spawn(async move {
let mut ticker = tokio::time::interval(Duration::from_secs(5));
loop {
ticker.tick().await;
let database = database.clone();
let result = tokio::task::spawn_blocking(move || {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |d| d.as_secs()) as i64;
webhooks::deliver_due(&database, &webhooks::CurlSender, now)
})
.await;
match result {
Ok(Ok((delivered, failed))) if delivered + failed > 0 => {
info!(delivered, failed, "webhook deliveries");
}
Ok(Ok(_)) => {}
Ok(Err(error)) => warn!(%error, "webhook delivery round failed"),
Err(error) => warn!(%error, "webhook delivery task failed"),
}
}
});
}
let app = Router::new()
.route("/", get(index))
.route("/docs", get(documentation))
.route("/docs/", get(documentation))
.route("/repos/{repository}", get(public_repository))
.route(
"/repos/{repository}/tree/{reference}",
get(public_code_root),
)
.route(
"/repos/{repository}/tree/{reference}/{*path}",
get(public_code_path),
)
.route(
"/app/repositories/{repository}/tree/{reference}",
get(app_code_root),
)
.route(
"/app/repositories/{repository}/tree/{reference}/{*path}",
get(app_code_path),
)
.route(
"/repos/{repository}/commits/{reference}",
get(history_commits_page),
)
.route(
"/repos/{repository}/releases",
get(releases_api::releases_page),
)
.route(
"/app/repositories/{repository}/releases",
get(releases_api::releases_page),
)
.route("/repos/{repository}/commit/{sha}", get(history_commit_page))
.route(
"/repos/{repository}/compare/{spec}",
get(history_compare_page),
)
.route(
"/repos/{repository}/blame/{reference}/{*path}",
get(history_blame_page),
)
.route(
"/app/repositories/{repository}/commits/{reference}",
get(history_commits_page),
)
.route(
"/app/repositories/{repository}/commit/{sha}",
get(history_commit_page),
)
.route(
"/app/repositories/{repository}/compare/{spec}",
get(history_compare_page),
)
.route(
"/app/repositories/{repository}/blame/{reference}/{*path}",
get(history_blame_page),
)
.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("/app/search", get(search_api::page))
.route(
"/app/settings/ssh-keys",
get(ssh_api::page).post(ssh_api::page_add),
)
.route(
"/app/settings/ssh-keys/{id}/remove",
post(ssh_api::page_remove),
)
.route(
"/app/repositories/{repository}/fork",
post(forks_api::page_fork),
)
.route(
"/app/repositories/{repository}/sync",
post(forks_api::page_sync),
)
.route(
"/app/repositories/{repository}/insights",
get(insights_api::page),
)
.route("/repos/{repository}/insights", get(insights_api::page))
.route("/app/notifications", get(notifications_api::page))
.route(
"/app/notifications/read-all",
post(notifications_api::page_read_all),
)
.route(
"/app/notifications/{id}/read",
post(notifications_api::page_read_one),
)
.route("/api/search", get(search_api::api))
.route("/api/ssh-keys", get(ssh_api::list).post(ssh_api::create))
.route("/api/ssh-keys/{id}", delete(ssh_api::remove))
.route("/api/mirrors", post(forks_api::create_mirror))
.route("/api/repos/{repository}/fork", post(forks_api::fork))
.route("/api/repos/{repository}/forks", get(forks_api::list_forks))
.route("/api/repos/{repository}/sync", post(forks_api::sync))
.route("/api/repos/{repository}/insights", get(insights_api::api))
.route("/api/repos/{repository}/secrets", get(secrets_api::list))
.route(
"/api/repos/{repository}/secrets/{name}",
put(secrets_api::put).delete(secrets_api::remove),
)
.route(
"/api/repos/{repository}/schedules",
get(secrets_api::schedules),
)
.route("/api/notifications", get(notifications_api::feed))
.route(
"/api/notifications/read-all",
post(notifications_api::read_all),
)
.route(
"/api/notifications/{id}/read",
post(notifications_api::read_one),
)
.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/repos/{repository}", delete(delete_repository))
.route("/api/runs/{id}", get(run_detail))
.route("/api/metrics/delivery", get(delivery_metrics))
.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}/commits/{sha}/statuses",
get(commit_statuses),
)
.route("/api/repos/{repository}/statuses", get(batch_statuses))
.route("/api/repos/{repository}/commits", get(repo_commits))
.route("/api/repos/{repository}/commits/{sha}", get(repo_commit))
.route("/api/repos/{repository}/compare/{spec}", get(repo_compare))
.route("/api/repos/{repository}/blame", get(repo_blame))
.route(
"/api/repos/{repository}/pulls",
get(pulls_api::list_pulls).post(pulls_api::create_pull),
)
.route(
"/api/repos/{repository}/pulls/{number}",
get(pulls_api::pull_detail),
)
.route(
"/api/repos/{repository}/pulls/{number}/close",
post(pulls_api::close_pull),
)
.route(
"/api/repos/{repository}/pulls/{number}/reopen",
post(pulls_api::reopen_pull),
)
.route(
"/api/repos/{repository}/pulls/{number}/reviews",
post(pulls_api::create_review),
)
.route(
"/api/repos/{repository}/pulls/{number}/comments",
post(pulls_api::create_comment),
)
.route(
"/api/repos/{repository}/pulls/{number}/merge",
post(pulls_api::merge_pull),
)
.route(
"/api/repos/{repository}/protection",
get(protection_api::list_rules).put(protection_api::put_rule),
)
.route(
"/api/repos/{repository}/protection/{branch}",
delete(protection_api::delete_rule),
)
.route(
"/api/repos/{repository}/protection/check",
post(protection_api::check),
)
.route(
"/api/repos/{repository}/collaborators",
get(access_api::list_collaborators).put(access_api::put_collaborator),
)
.route(
"/api/repos/{repository}/collaborators/{subject}",
delete(access_api::delete_collaborator),
)
.route(
"/api/tokens",
get(access_api::list_pats).post(access_api::create_pat),
)
.route("/api/tokens/{id}", delete(access_api::revoke_pat))
.route(
"/api/repos/{repository}/deploy-keys",
get(access_api::list_deploy_keys).post(access_api::create_deploy_key),
)
.route(
"/api/repos/{repository}/deploy-keys/{id}",
delete(access_api::revoke_deploy_key),
)
.route(
"/api/repos/{repository}/webhooks",
get(webhooks_api::list_webhooks).post(webhooks_api::create_webhook),
)
.route(
"/api/repos/{repository}/webhooks/{id}",
delete(webhooks_api::delete_webhook).patch(webhooks_api::set_webhook_active),
)
.route(
"/api/repos/{repository}/webhooks/{id}/deliveries",
get(webhooks_api::list_deliveries),
)
.route(
"/api/repos/{repository}/webhooks/{id}/deliveries/{delivery}/redeliver",
post(webhooks_api::redeliver),
)
.route(
"/api/repos/{repository}/events/push",
post(webhooks_api::push_event),
)
.route("/api/repos/{repository}/tags", get(releases_api::list_tags))
.route(
"/api/repos/{repository}/releases",
get(releases_api::list_releases).post(releases_api::create_release),
)
.route(
"/api/repos/{repository}/releases/{tag}",
get(releases_api::get_release).delete(releases_api::delete_release),
)
.route(
"/api/repos/{repository}/releases/{tag}/assets",
post(releases_api::attach_asset),
)
.route(
"/api/repos/{repository}/releases/{tag}/assets/{name}",
get(releases_api::download_asset),
)
.route("/api/repos/{repository}/tree", get(repo_tree))
.merge(mcp_routes())
.fallback(public_not_found)
.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 repositories = state
.database
.public_repositories()
.map_err(ApiError::internal)?
.into_iter()
.map(|repository| LandingRepository {
owner: repository_owner(&repository.url),
name: repository.name,
default_branch: repository.default_branch,
run_count: repository.run_count,
successful_runs: repository.successful_runs,
})
.collect::<Vec<_>>();
let mut builder_counts = std::collections::BTreeMap::new();
for repository in &repositories {
*builder_counts.entry(repository.owner.clone()).or_insert(0) += 1;
}
let mut builders = builder_counts
.into_iter()
.map(|(name, repository_count)| PublicBuilder {
initial: name.chars().next().unwrap_or('A').to_uppercase().collect(),
name,
repository_count,
})
.collect::<Vec<_>>();
builders.sort_by(|left, right| {
right
.repository_count
.cmp(&left.repository_count)
.then_with(|| left.name.cmp(&right.name))
});
let page = state
.templates
.get_template("landing.html")
.map_err(ApiError::internal)?
.render(context! {
version => env!("CARGO_PKG_VERSION"),
repositories => repositories,
builders => builders
})
.map_err(ApiError::internal)?;
Ok(Html(page))
}
fn encode_code_segment(value: &str) -> String {
value
.bytes()
.map(|byte| {
if byte.is_ascii_alphanumeric() || b"-._~".contains(&byte) {
(byte as char).to_string()
} else {
format!("%{byte:02X}")
}
})
.collect()
}
fn code_base(public: bool, repository: &str, reference: &str) -> String {
let prefix = if public {
"/repos"
} else {
"/app/repositories"
};
format!(
"{prefix}/{}/tree/{}",
encode_code_segment(repository),
encode_code_segment(reference),
)
}
fn code_path_href(base: &str, path: &str) -> String {
if path.is_empty() {
return base.to_owned();
}
format!(
"{base}/{}",
path.split('/')
.filter(|segment| !segment.is_empty())
.map(encode_code_segment)
.collect::<Vec<_>>()
.join("/")
)
}
fn code_breadcrumbs(
public: bool,
repository: &str,
reference: &str,
path: &str,
) -> Vec<CodeBreadcrumb> {
let base = code_base(public, repository, reference);
let mut breadcrumbs = vec![CodeBreadcrumb {
label: repository.to_owned(),
href: base.clone(),
}];
let mut current = String::new();
for segment in path.split('/').filter(|segment| !segment.is_empty()) {
if !current.is_empty() {
current.push('/');
}
current.push_str(segment);
breadcrumbs.push(CodeBreadcrumb {
label: segment.to_owned(),
href: code_path_href(&base, ¤t),
});
}
breadcrumbs
}
async fn render_code(
state: &AppState,
repository_name: String,
reference: String,
path: String,
public: bool,
) -> Result<Html<String>, ApiError> {
let repository = state
.database
.repository(&repository_name)
.map_err(|_| ApiError::not_found(format!("unknown repository {repository_name}")))?;
if public && repository.visibility != "public" {
return Err(ApiError::not_found("repository not found"));
}
let view = repository_view(repository.clone(), &state.hosted_root, &state.public_origin);
let base = code_base(public, &repository_name, &reference);
enum Lookup {
Tree(crate::tree::TreeResponse),
Blob(crate::tree::BlobResponse),
}
let tree = state.tree.clone();
let name = repository.name.clone();
let url = repository.url.clone();
let lookup_reference = reference.clone();
let lookup_path = path.clone();
let lookup = state
.blocking_git
.run(
move || match tree.tree(&name, &url, Some(&lookup_reference), &lookup_path) {
Ok(response) => Ok(Lookup::Tree(response)),
Err(_) => tree
.blob(&name, &url, &lookup_reference, &lookup_path)
.map(Lookup::Blob)
.map_err(|_| ApiError::not_found("repository ref or path not found")),
},
)
.await?;
let (entries, source, is_file, latest_commit) = match lookup {
Lookup::Tree(tree) => (
tree.entries
.into_iter()
.map(|entry| CodeEntryView {
href: code_path_href(
&base,
&if path.is_empty() {
entry.name.clone()
} else {
format!("{path}/{}", entry.name)
},
),
name: entry.name,
kind: entry.kind,
size: entry.size,
commit_message: entry.commit_message,
})
.collect(),
None,
false,
tree.latest_commit,
),
Lookup::Blob(blob) => (Vec::new(), Some(blob.content), true, blob.latest_commit),
};
let commit_href = latest_commit
.as_ref()
.map(|commit| code_base(public, &repository_name, &commit.sha));
let parent_href = if path.is_empty() {
if public {
format!("/repos/{}", encode_code_segment(&repository_name))
} else {
format!(
"/app/repositories/{}",
encode_code_segment(&repository_name)
)
}
} else {
let parent = path.rsplit_once('/').map_or("", |(parent, _)| parent);
code_path_href(&base, parent)
};
let page = CodePage {
repository: view,
reference: reference.clone(),
path: path.clone(),
entries,
breadcrumbs: code_breadcrumbs(public, &repository_name, &reference, &path),
source,
is_file,
parent_href,
latest_commit,
commit_href,
public,
};
let rendered = state
.templates
.get_template("repository-code.html")
.map_err(ApiError::internal)?
.render(context! { page => page })
.map_err(ApiError::internal)?;
Ok(Html(rendered))
}
const HISTORY_PAGE_SIZE: usize = 30;
enum HistoryRequest {
Commits { reference: String, skip: usize },
Commit { sha: String },
Compare { base: String, head: String },
Blame { reference: String, path: String },
}
fn history_view(
request: &HistoryRequest,
mirror: &std::path::Path,
commit_prefix: &str,
list_href: &str,
) -> anyhow::Result<minijinja::Value> {
Ok(match request {
HistoryRequest::Commits { reference, skip } => {
let page = history::commits(mirror, reference, "", *skip, HISTORY_PAGE_SIZE)?;
let next_href = page
.has_more
.then(|| format!("{list_href}?skip={}", skip + HISTORY_PAGE_SIZE));
context! { kind => "commits", heading => format!("Commits on {reference}"),
commits => page.commits, next_href => next_href, commit_prefix => commit_prefix }
}
HistoryRequest::Commit { sha } => {
let detail = history::commit(mirror, sha)?;
context! { kind => "commit", heading => format!("Commit {}", &detail.summary.sha[..8]),
sha => detail.summary.sha, author => detail.summary.author, body => detail.body,
files => detail.diff.files, patch => detail.diff.patch,
truncated => detail.diff.truncated, commit_prefix => commit_prefix }
}
HistoryRequest::Compare { base, head } => {
let comparison = history::compare(mirror, base, head)?;
context! { kind => "compare", heading => format!("Compare {base}...{head}"),
commits => comparison.commits, merge_base => comparison.merge_base,
files => comparison.diff.files, patch => comparison.diff.patch,
truncated => comparison.diff.truncated, commit_prefix => commit_prefix }
}
HistoryRequest::Blame { reference, path } => {
let blame = history::blame(mirror, reference, path)?;
context! { kind => "blame", heading => format!("Blame {path} @ {reference}"),
blame_lines => blame.lines, commit_prefix => commit_prefix }
}
})
}
async fn render_history(
state: &AppState,
headers: &HeaderMap,
uri: &axum::http::Uri,
repository_name: String,
request: HistoryRequest,
) -> Result<Response, ApiError> {
let public = !uri.path().starts_with("/app/");
if !public && !valid_session(headers, state) {
return Ok(Redirect::to("/login").into_response());
}
let repository = state
.database
.repository(&repository_name)
.map_err(|_| ApiError::not_found(format!("unknown repository {repository_name}")))?;
if public && repository.visibility != "public" {
return Err(ApiError::not_found("repository not found"));
}
let prefix = if public {
"/repos"
} else {
"/app/repositories"
};
let encoded = encode_code_segment(&repository_name);
let repo_href = format!("{prefix}/{encoded}");
let commit_prefix = format!("{repo_href}/commit/");
let list_href = uri.path().to_owned();
let view = repository_view(repository.clone(), &state.hosted_root, &state.public_origin);
let tree = state.tree.clone();
let name = repository.name.clone();
let url = repository.url.clone();
let data = state
.blocking_git
.run(move || {
let mirror = tree.sync_mirror(&name, &url).map_err(ApiError::internal)?;
history_view(&request, &mirror, &commit_prefix, &list_href)
.map_err(|_| ApiError::not_found("repository ref or path not found"))
})
.await?;
let rendered = state
.templates
.get_template("repository-history.html")
.map_err(ApiError::internal)?
.render(
context! { page => context! { repository => view, public => public,
repo_href => repo_href, ..data } },
)
.map_err(ApiError::internal)?;
Ok(Html(rendered).into_response())
}
#[derive(Deserialize)]
struct SkipQuery {
skip: Option<usize>,
}
async fn history_commits_page(
State(state): State<AppState>,
headers: HeaderMap,
uri: axum::http::Uri,
Path((repository, reference)): Path<(String, String)>,
Query(query): Query<SkipQuery>,
) -> Result<Response, ApiError> {
let skip = query.skip.unwrap_or(0);
render_history(
&state,
&headers,
&uri,
repository,
HistoryRequest::Commits { reference, skip },
)
.await
}
async fn history_commit_page(
State(state): State<AppState>,
headers: HeaderMap,
uri: axum::http::Uri,
Path((repository, sha)): Path<(String, String)>,
) -> Result<Response, ApiError> {
render_history(
&state,
&headers,
&uri,
repository,
HistoryRequest::Commit { sha },
)
.await
}
async fn history_compare_page(
State(state): State<AppState>,
headers: HeaderMap,
uri: axum::http::Uri,
Path((repository, spec)): Path<(String, String)>,
) -> Result<Response, ApiError> {
let (base, head) = spec
.split_once("...")
.map(|(base, head)| (base.to_owned(), head.to_owned()))
.ok_or_else(|| ApiError::bad("compare spec must be base...head"))?;
render_history(
&state,
&headers,
&uri,
repository,
HistoryRequest::Compare { base, head },
)
.await
}
async fn history_blame_page(
State(state): State<AppState>,
headers: HeaderMap,
uri: axum::http::Uri,
Path((repository, reference, path)): Path<(String, String, String)>,
) -> Result<Response, ApiError> {
render_history(
&state,
&headers,
&uri,
repository,
HistoryRequest::Blame { reference, path },
)
.await
}
async fn public_code_root(
State(state): State<AppState>,
headers: HeaderMap,
Path((repository, reference)): Path<(String, String)>,
) -> Result<Response, ApiError> {
match render_code(&state, repository, reference, String::new(), true).await {
Ok(page) => Ok(page.into_response()),
Err(error) if error.status == StatusCode::NOT_FOUND => public_error_response(
&state.templates,
&headers,
error,
"Repository content not found",
"Repository content could not be found.",
"The repository, ref, or path may have moved. Return to the site or open the documentation.",
),
Err(error) => Err(error),
}
}
async fn public_code_path(
State(state): State<AppState>,
headers: HeaderMap,
Path((repository, reference, path)): Path<(String, String, String)>,
) -> Result<Response, ApiError> {
match render_code(&state, repository, reference, path, true).await {
Ok(page) => Ok(page.into_response()),
Err(error) if error.status == StatusCode::NOT_FOUND => public_error_response(
&state.templates,
&headers,
error,
"Repository content not found",
"Repository content could not be found.",
"The repository, ref, or path may have moved. Return to the site or open the documentation.",
),
Err(error) => Err(error),
}
}
async fn app_code_root(
State(state): State<AppState>,
headers: HeaderMap,
Path((repository, reference)): Path<(String, String)>,
) -> Result<Response, ApiError> {
if !valid_session(&headers, &state) {
return Ok(Redirect::to("/login").into_response());
}
match render_code(&state, repository, reference, String::new(), false).await {
Ok(page) => Ok(page.into_response()),
Err(error) if error.status == StatusCode::NOT_FOUND => public_error_response(
&state.templates,
&headers,
error,
"Repository content not found",
"Repository content could not be found.",
"The repository, ref, or path may have moved. Return to the site or open the documentation.",
),
Err(error) => Err(error),
}
}
async fn app_code_path(
State(state): State<AppState>,
headers: HeaderMap,
Path((repository, reference, path)): Path<(String, String, String)>,
) -> Result<Response, ApiError> {
if !valid_session(&headers, &state) {
return Ok(Redirect::to("/login").into_response());
}
match render_code(&state, repository, reference, path, false).await {
Ok(page) => Ok(page.into_response()),
Err(error) if error.status == StatusCode::NOT_FOUND => public_error_response(
&state.templates,
&headers,
error,
"Repository content not found",
"Repository content could not be found.",
"The repository, ref, or path may have moved. Return to the site or open the documentation.",
),
Err(error) => Err(error),
}
}
async fn documentation(State(state): State<AppState>) -> Result<Html<String>, ApiError> {
let page = state
.templates
.get_template("docs.html")
.map_err(ApiError::internal)?
.render(context! { version => env!("CARGO_PKG_VERSION") })
.map_err(ApiError::internal)?;
Ok(Html(page))
}
fn repository_owner(url: &str) -> String {
let (host, path) = if let Some((_, remote)) = url.split_once("://") {
remote
.split_once('/')
.map(|(host, path)| (host.rsplit('@').next().unwrap_or(host), path))
.unwrap_or((remote, ""))
} else if let Some((_, remote)) = url.split_once('@') {
remote.split_once(':').unwrap_or((remote, ""))
} else {
return "AkurAI Build".into();
};
let mut segments = path.split('/').filter(|segment| !segment.is_empty());
match (segments.next(), segments.next()) {
(Some("git"), Some(_)) => "AkurAI Build".into(),
(Some(owner), Some(_)) => owner.into(),
_ => host.into(),
}
}
async fn public_repository(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository_name): Path<String>,
) -> Result<Response, ApiError> {
let repository = match state.database.repository(&repository_name) {
Ok(repository) if repository.visibility == "public" => repository,
_ => {
return public_error_response(
&state.templates,
&headers,
ApiError::not_found(format!("unknown repository {repository_name}")),
"Repository not found",
"This repository could not be found.",
"The repository may be private, renamed, or unavailable. Explore public work or open the documentation.",
);
}
};
let code_href = code_base(true, &repository.name, &repository.default_branch);
let repository = repository_view(repository, &state.hosted_root, &state.public_origin);
let page = state
.templates
.get_template("repository.html")
.map_err(ApiError::internal)?
.render(context! { repository => repository, code_href => code_href })
.map_err(ApiError::internal)?;
Ok(Html(page).into_response())
}
async fn git_http(
State(state): State<AppState>,
Path(path): Path<String>,
request: Request<Body>,
) -> Result<Response, ApiError> {
validate_git_path(&path).map_err(ApiError::bad)?;
let repository = path
.split('/')
.next()
.and_then(|r| r.strip_suffix(".git"))
.unwrap_or_default()
.to_owned();
let push = access_api::is_push(&path, request.uri().query());
let remote_user = match access_api::authorize_git(request.headers(), &state, &repository, push)
{
Ok(user) => user,
// git only sends credentials after a challenge
Err(error) if error.status == StatusCode::UNAUTHORIZED => {
let mut response = error.into_response();
response.headers_mut().insert(
header::WWW_AUTHENTICATE,
HeaderValue::from_static("Basic realm=\"AkurAI Build\""),
);
return Ok(response);
}
Err(error) => return Err(error),
};
let forge_url = request_origin(request.headers());
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");
let hook = GitHookContext {
forge_url,
token: state.token.to_string(),
repository,
remote_user,
};
tokio::task::spawn_blocking(move || run_git_http(root, path, parts, body, &hook))
.await
.map_err(ApiError::internal)?
.map_err(ApiError::internal)
}
/// Where this forge answers for the client that is pushing (loopback or the
/// edge); the pre-receive hook calls back through it.
fn request_origin(headers: &HeaderMap) -> String {
let host = headers
.get(header::HOST)
.and_then(|v| v.to_str().ok())
.unwrap_or("127.0.0.1:3210");
let https = headers
.get("x-forwarded-proto")
.and_then(|v| v.to_str().ok())
.is_some_and(|p| p.eq_ignore_ascii_case("https"));
format!("{}://{host}", if https { "https" } else { "http" })
}
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(())
}
/// What git's hooks learn about the push: who, which repository, and where the
/// forge answers (the pre-receive hook asks it about branch protection).
struct GitHookContext {
forge_url: String,
token: String,
repository: String,
remote_user: String,
}
fn run_git_http(
root: PathBuf,
path: String,
parts: axum::http::request::Parts,
body: Bytes,
hook: &GitHookContext,
) -> Result<Response> {
let mut command = Command::new("git");
command
.arg("http-backend")
.env("REMOTE_USER", &hook.remote_user)
.env("AKURAI_FORGE_URL", &hook.forge_url)
.env("AKURAI_FORGE_TOKEN", &hook.token)
.env("AKURAI_FORGE_REPO", &hook.repository)
.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 mut repository = repository_view(repository, &state.hosted_root, &state.public_origin);
repository.ssh_url = state
.ssh_address
.as_deref()
.map(|address| format!("{address}:{}.git", repository.name));
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 mut response = Redirect::to(&config.authorize_url(&login_state)).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 admin = valid_admin_session(&headers, &state) || valid_bearer(&headers, &state);
let all_repositories = state.database.repositories().map_err(ApiError::internal)?;
let visible_repositories = all_repositories
.iter()
.filter(|repository| admin || repository.visibility == "public")
.collect::<Vec<_>>();
let repository_filter = if let Some(requested) = query.repo.as_deref() {
let requested = requested
.split(',')
.map(str::trim)
.filter(|name| !name.is_empty())
.collect::<Vec<_>>();
if requested.is_empty() {
return Err(ApiError::bad("repository filter is empty"));
}
if !requested.iter().all(|name| {
visible_repositories
.iter()
.any(|repository| repository.name == *name)
}) {
return Err(ApiError::not_found("repository not found"));
}
Some(requested.join(","))
} else {
// Visibility is filtered by `public_only` below, so the implicit
// "all visible repositories" view is not bound by the explicit
// repository filter cap.
None
};
let repositories = visible_repositories
.into_iter()
.cloned()
.map(|repository| repository_view(repository, &state.hosted_root, &state.public_origin))
.collect();
let mut run_query = crate::db::RunQuery {
repository: repository_filter,
statuses: list_filter(query.status),
git_ref: query.git_ref,
triggers: list_filter(query.trigger),
search: query.q,
before_id: query.before,
public_only: !admin,
limit: query.limit.unwrap_or(50),
offset: query.offset.unwrap_or(0),
};
let run_cursor = match run_query.before_id {
Some(cursor) => Some(cursor),
None => state
.database
.max_run_id(&run_query)
.map_err(ApiError::bad)?,
};
run_query.before_id = run_cursor;
let run_total = state
.database
.count_runs(&run_query)
.map_err(ApiError::bad)?;
let runs = state
.database
.query_runs(&run_query)
.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,
run_total,
run_cursor,
}))
}
async fn delivery_metrics(
State(state): State<AppState>,
headers: HeaderMap,
Query(query): Query<DeliveryMetricsQuery>,
) -> Result<Json<Envelope<crate::db::DeliveryMetrics>>, ApiError> {
authorize(&headers, &state)?;
let repository_name = query
.repo
.as_deref()
.ok_or_else(|| ApiError::bad("repo is required for repository-scoped metrics"))?;
let repository = state
.database
.repository(repository_name)
.map_err(|_| ApiError::not_found("repository not found"))?;
if repository.visibility != "public"
&& !valid_admin_session(&headers, &state)
&& !valid_bearer(&headers, &state)
{
return Err(ApiError::not_found("repository not found"));
}
let window = query.window.unwrap_or(7 * 86_400);
Ok(ok(state
.database
.delivery_metrics(Some(repository_name), query.environment.as_deref(), window)
.map_err(ApiError::bad)?))
}
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)?))
}
fn visible_run_detail(
state: &AppState,
headers: &HeaderMap,
id: i64,
) -> Result<crate::db::RunDetail, ApiError> {
let mut detail = state.database.detail(id).map_err(ApiError::bad)?;
let repository = state
.database
.repository(&detail.run.repository)
.map_err(ApiError::bad)?;
let privileged = valid_admin_session(headers, state) || valid_bearer(headers, state);
if repository.visibility != "public" && !privileged {
return Err(ApiError::not_found("run not found"));
}
if !privileged {
for deployment in &mut detail.deployments {
deployment.organization_id = None;
deployment.workspace_id = None;
deployment.installation_id = None;
deployment.app_id = None;
deployment.installation_revision = None;
deployment.commit_sha = None;
deployment.artifact_digest = None;
deployment.result = None;
deployment.rollback_of_deployment_id = None;
deployment.recovery_of_deployment_id = None;
}
}
Ok(detail)
}
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(visible_run_detail(&state, &headers, id)?))
}
async fn run_events(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<i64>,
) -> Result<Response, ApiError> {
authorize(&headers, &state)?;
visible_run_detail(&state, &headers, id)?;
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<RepositoryView>>, ApiError> {
authorize_mutation(&headers, &state)?;
config::validate_repository(&request.name, &request.url, &request.branch)
.map_err(ApiError::bad)?;
let repository = state
.database
.add_repository(&request.name, &request.url, &request.branch)
.map_err(ApiError::bad)?;
Ok(ok(repository_view(
repository,
&state.hosted_root,
&state.public_origin,
)))
}
/// `DELETE /api/repos/{repository}?confirm={repository}`: deletes the
/// repository's rows, hosted bare repository, and run files from disk.
async fn delete_repository(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
Query(query): Query<DeleteRepositoryQuery>,
) -> Result<Json<Envelope<RepositoryRemoval>>, ApiError> {
authorize_mutation(&headers, &state)?;
config::validate_repo_name(&repository).map_err(ApiError::bad)?;
if query.confirm.as_deref() != Some(repository.as_str()) {
return Err(ApiError::bad(format!(
"deleting a repository requires confirm={repository}"
)));
}
let (database, data_root) = (state.database.clone(), state.hosted_root.clone());
let removal = tokio::task::spawn_blocking(move || {
repo_delete::delete(&database, &data_root, &repository)
})
.await
.map_err(ApiError::internal)?
.map_err(
|error| match error.downcast_ref::<RepositoryDeleteError>() {
Some(RepositoryDeleteError::NotFound(_)) => ApiError::not_found(error.to_string()),
Some(RepositoryDeleteError::ActiveRuns { .. }) => ApiError::conflict(error.to_string()),
None => ApiError::internal(format!("{error:#}")),
},
)?;
info!(
repository = %removal.deleted.repository.name,
runs = removal.deleted.run_ids.len(),
paths = removal.removed_paths.len(),
"deleted repository"
);
Ok(ok(removal))
}
async fn change_repository_visibility(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
Json(request): Json<VisibilityRequest>,
) -> Result<Json<Envelope<RepositoryView>>, ApiError> {
authorize_mutation(&headers, &state)?;
let repository = state
.database
.set_repository_visibility(&repository, &request.visibility)
.map_err(ApiError::bad)?;
Ok(ok(repository_view(
repository,
&state.hosted_root,
&state.public_origin,
)))
}
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<git_process::RemoteBranch>>>, ApiError> {
authorize(&headers, &state)?;
let repo = state
.database
.repository(&repository)
.map_err(|_| ApiError::bad("repository not found"))?;
let default_branch = repo.default_branch;
let url = repo.url;
let branches = state
.blocking_git
.run(move || git_process::remote_branches(&url, &default_branch).map_err(ApiError::bad))
.await?;
Ok(ok(branches))
}
fn validate_sha(sha: &str) -> Result<(), ApiError> {
let well_formed = matches!(sha.len(), 40 | 64) && sha.bytes().all(|b| b.is_ascii_hexdigit());
well_formed
.then_some(())
.ok_or_else(|| ApiError::bad(format!("invalid commit sha {sha:?}")))
}
fn known_repository(state: &AppState, repository: &str) -> Result<(), ApiError> {
state
.database
.repository(repository)
.map(|_| ())
.map_err(|_| ApiError::not_found(format!("unknown repository {repository}")))
}
/// `GET /api/repos/{repository}/commits/{sha}/statuses`: per-job statuses of
/// the commit's runs plus their rollup.
async fn commit_statuses(
State(state): State<AppState>,
headers: HeaderMap,
Path((repository, sha)): Path<(String, String)>,
) -> Result<Json<Envelope<statuses::CombinedStatus>>, ApiError> {
authorize(&headers, &state)?;
validate_sha(&sha)?;
known_repository(&state, &repository)?;
let database = state.database.clone();
let combined =
tokio::task::spawn_blocking(move || statuses::combined(&database, &repository, &sha))
.await
.map_err(ApiError::internal)?
.map_err(ApiError::internal)?;
Ok(ok(combined))
}
#[derive(Deserialize)]
struct BatchStatusQuery {
/// Comma-separated commit SHAs.
shas: String,
}
/// `GET /api/repos/{repository}/statuses?shas=a,b`: rollup per SHA, the
/// badge data for branch lists.
async fn batch_statuses(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
Query(query): Query<BatchStatusQuery>,
) -> Result<Json<Envelope<std::collections::BTreeMap<String, statuses::Rollup>>>, ApiError> {
authorize(&headers, &state)?;
known_repository(&state, &repository)?;
let shas: Vec<String> = query.shas.split(',').map(str::to_owned).collect();
if shas.len() > statuses::MAX_BATCH_SHAS {
return Err(ApiError::bad(format!(
"at most {} shas per request",
statuses::MAX_BATCH_SHAS
)));
}
shas.iter().try_for_each(|sha| validate_sha(sha))?;
let database = state.database.clone();
let rollups = tokio::task::spawn_blocking(move || {
shas.into_iter()
.map(|sha| statuses::combined(&database, &repository, &sha).map(|c| (sha, c.state)))
.collect::<anyhow::Result<_>>()
})
.await
.map_err(ApiError::internal)?
.map_err(ApiError::internal)?;
Ok(ok(rollups))
}
#[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 tree = state.tree.clone();
let name = repo.name;
let url = repo.url;
let reference = query.reference;
let response = state
.blocking_git
.run(move || {
tree.tree(&name, &url, reference.as_deref(), &sub_path)
.map_err(ApiError::bad)
})
.await?;
Ok(ok(response))
}
#[derive(Deserialize)]
struct HistoryQuery {
#[serde(rename = "ref")]
reference: Option<String>,
path: Option<String>,
skip: Option<usize>,
limit: Option<usize>,
}
const DEFAULT_HISTORY_PAGE: usize = 30;
/// Runs a `history` query against the repository's synced browse mirror.
async fn with_mirror<T, F>(
state: &AppState,
headers: &HeaderMap,
repository: &str,
query: F,
) -> Result<Json<Envelope<T>>, ApiError>
where
T: Serialize + Send + 'static,
F: FnOnce(&std::path::Path) -> anyhow::Result<T> + Send + 'static,
{
authorize(headers, state)?;
let repo = state
.database
.repository(repository)
.map_err(|_| ApiError::not_found(format!("unknown repository {repository}")))?;
let tree = state.tree.clone();
let value = state
.blocking_git
.run(move || {
let mirror = tree
.sync_mirror(&repo.name, &repo.url)
.map_err(ApiError::bad)?;
query(&mirror).map_err(ApiError::bad)
})
.await?;
Ok(ok(value))
}
/// `GET /api/repos/{repository}/commits?ref=&path=&skip=&limit=`.
async fn repo_commits(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
Query(query): Query<HistoryQuery>,
) -> Result<Json<Envelope<history::CommitPage>>, ApiError> {
let reference = query.reference.unwrap_or_else(|| "HEAD".to_owned());
let path = query.path.unwrap_or_default();
let skip = query.skip.unwrap_or(0);
let limit = query.limit.unwrap_or(DEFAULT_HISTORY_PAGE);
with_mirror(&state, &headers, &repository, move |mirror| {
history::commits(mirror, &reference, &path, skip, limit)
})
.await
}
/// `GET /api/repos/{repository}/commits/{sha}`: metadata plus diff.
async fn repo_commit(
State(state): State<AppState>,
headers: HeaderMap,
Path((repository, sha)): Path<(String, String)>,
) -> Result<Json<Envelope<history::CommitDetail>>, ApiError> {
with_mirror(&state, &headers, &repository, move |mirror| {
history::commit(mirror, &sha)
})
.await
}
/// `GET /api/repos/{repository}/compare/{base}...{head}`.
async fn repo_compare(
State(state): State<AppState>,
headers: HeaderMap,
Path((repository, spec)): Path<(String, String)>,
) -> Result<Json<Envelope<history::Comparison>>, ApiError> {
let (base, head) = spec
.split_once("...")
.map(|(base, head)| (base.to_owned(), head.to_owned()))
.ok_or_else(|| ApiError::bad("compare spec must be base...head"))?;
with_mirror(&state, &headers, &repository, move |mirror| {
history::compare(mirror, &base, &head)
})
.await
}
/// `GET /api/repos/{repository}/blame?ref=&path=`.
async fn repo_blame(
State(state): State<AppState>,
headers: HeaderMap,
Path(repository): Path<String>,
Query(query): Query<HistoryQuery>,
) -> Result<Json<Envelope<history::Blame>>, ApiError> {
let reference = query.reference.unwrap_or_else(|| "HEAD".to_owned());
let path = query.path.unwrap_or_default();
with_mirror(&state, &headers, &repository, move |mirror| {
history::blame(mirror, &reference, &path)
})
.await
}
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 valid_bearer(headers: &HeaderMap, state: &AppState) -> bool {
headers
.get(header::AUTHORIZATION)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.strip_prefix("Bearer "))
.is_some_and(|supplied| constant_time_equal(supplied.as_bytes(), state.token.as_bytes()))
}
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()
}
/// Streamable-HTTP MCP transport (JSON responses only) plus its RFC 9728
/// protected-resource metadata.
fn mcp_routes() -> Router<AppState> {
Router::new()
.route("/mcp", post(mcp_http))
.route(
"/.well-known/oauth-protected-resource",
get(protected_resource_metadata),
)
.route(
"/.well-known/oauth-protected-resource/mcp",
get(protected_resource_metadata),
)
}
async fn protected_resource_metadata(State(state): State<AppState>) -> Json<Value> {
Json(state.oauth.metadata())
}
async fn mcp_http(State(state): State<AppState>, headers: HeaderMap, body: Bytes) -> Response {
let origin = headers
.get(header::ORIGIN)
.and_then(|value| value.to_str().ok());
if !state.oauth.origin_allowed(origin) {
return ApiError::forbidden("origin is not allowed").into_response();
}
if let Err(denial) = authorize_mcp(&headers, &state).await {
return mcp_challenge(&state, denial);
}
let service = state.mcp.clone();
let reply = tokio::task::spawn_blocking(move || match serde_json::from_slice::<Value>(&body) {
Ok(request) => service.handle(&request),
Err(error) => Some(mcp::parse_error(&error)),
})
.await;
match reply {
Ok(Some(response)) => Json(response).into_response(),
Ok(None) => StatusCode::ACCEPTED.into_response(),
Err(error) => ApiError::internal(error).into_response(),
}
}
/// Why an MCP request was refused before reaching the tool set.
enum McpDenial {
/// No bearer token: plain challenge.
Missing,
/// Token presented but invalid: 401 `invalid_token`.
Invalid,
/// Valid IdP user outside the admin allowlist: 403 `insufficient_scope`.
Forbidden,
}
/// Accepts the controller API token (existing agents) or an IdP access token
/// whose audience includes this app, held by an allowlisted admin.
async fn authorize_mcp(headers: &HeaderMap, state: &AppState) -> Result<(), McpDenial> {
let token = headers
.get(header::AUTHORIZATION)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.strip_prefix("Bearer "))
.map(str::trim)
.filter(|token| !token.is_empty())
.ok_or(McpDenial::Missing)?;
if constant_time_equal(token.as_bytes(), state.token.as_bytes()) {
return Ok(());
}
let oauth = state.oauth.clone();
let token = token.to_owned();
match tokio::task::spawn_blocking(move || oauth.authenticate(&token)).await {
Ok(Ok(identity)) => {
info!(email = %identity.email, tenant = %identity.tenant, "MCP access via IdP token");
Ok(())
}
Ok(Err(Denial::Invalid(reason))) => {
warn!(reason, "rejected MCP access token");
Err(McpDenial::Invalid)
}
Ok(Err(Denial::Forbidden(reason))) => {
warn!(reason, "refused MCP access for IdP user");
Err(McpDenial::Forbidden)
}
Err(error) => {
warn!(?error, "MCP token validation task failed");
Err(McpDenial::Invalid)
}
}
}
fn mcp_challenge(state: &AppState, denial: McpDenial) -> Response {
let (mut response, error) = match denial {
McpDenial::Missing => (ApiError::unauthorized().into_response(), None),
McpDenial::Invalid => (
ApiError::unauthorized().into_response(),
Some("invalid_token"),
),
McpDenial::Forbidden => (
ApiError::forbidden("an AkurAI Build admin is required").into_response(),
Some("insufficient_scope"),
),
};
if let Ok(value) = HeaderValue::from_str(&state.oauth.challenge(error)) {
response
.headers_mut()
.insert(header::WWW_AUTHENTICATE, value);
}
response
}
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 valid_admin_session(headers: &HeaderMap, state: &AppState) -> bool {
let Some(cookie) = request_cookie(headers, "ab_session") else {
return false;
};
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, &state.token) 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| value == "admin")
}
fn hex_to_text(value: &str) -> Option<String> {
if value.len() > 1024 || !value.len().is_multiple_of(2) {
return None;
}
value
.as_bytes()
.as_chunks::<2>()
.0
.iter()
.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()
.as_chunks::<2>()
.0
.iter()
.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 crate::oauth::{
self,
testing::{self as oauth_testing, TestIdp, user_claims},
};
use tower::Service;
#[test]
fn templates_compile_and_render_without_inline_code() -> Result<()> {
let environment = template_environment()?;
let page = environment.get_template("landing.html")?.render(context! {
version => "test",
repositories => vec![LandingRepository {
name: "AkurAI-Build".into(),
default_branch: "main".into(),
owner: "olibuijr".into(),
run_count: 4,
successful_runs: 3,
}],
builders => vec![PublicBuilder {
name: "olibuijr".into(),
initial: "O".into(),
repository_count: 1,
}]
})?;
let docs = environment
.get_template("docs.html")?
.render(context! { version => "test" })?;
assert!(docs.contains("AkurAI Build documentation"));
assert!(docs.contains("akurai_run_queue"));
assert!(!docs.contains("Bunfork"));
assert!(docs.contains("AkurAI <b>Build</b>"));
assert!(!docs.contains("AkurAI <b>/Build</b>"));
assert!(page.contains("olibuijr /</span> AkurAI-Build"));
assert!(page.contains("Trending repositories"));
assert!(page.contains("Public builders"));
assert!(!page.contains("<script>"));
Ok(())
}
#[tokio::test]
async fn public_errors_negotiate_branded_html_and_json() -> Result<()> {
let environment = template_environment()?;
let mut html_headers = HeaderMap::new();
html_headers.insert(
header::ACCEPT,
HeaderValue::from_static("text/html,application/xhtml+xml"),
);
let html = public_error_response(
&environment,
&html_headers,
ApiError::not_found("missing"),
"Page not found",
"This page could not be found.",
"Return home or open the documentation.",
)
.map_err(|error| anyhow::anyhow!(error.message))?;
assert_eq!(html.status(), StatusCode::NOT_FOUND);
assert_eq!(
html.headers().get(header::CONTENT_TYPE),
Some(&HeaderValue::from_static("text/html; charset=utf-8"))
);
let html_body = to_bytes(html.into_body(), MAX_BODY).await?;
let html_body = String::from_utf8(html_body.to_vec())?;
assert!(html_body.contains("AkurAI <b>Build</b>"));
assert!(html_body.contains("Go home"));
assert!(!html_body.contains("{\"ok\":false"));
let mut json_headers = HeaderMap::new();
json_headers.insert(header::ACCEPT, HeaderValue::from_static("application/json"));
let json = public_error_response(
&environment,
&json_headers,
ApiError::not_found("missing"),
"Page not found",
"This page could not be found.",
"Return home or open the documentation.",
)
.map_err(|error| anyhow::anyhow!(error.message))?;
assert_eq!(json.status(), StatusCode::NOT_FOUND);
assert_eq!(
json.headers().get(header::CONTENT_TYPE),
Some(&HeaderValue::from_static("application/json"))
);
let json_body = to_bytes(json.into_body(), MAX_BODY).await?;
let json_body: Value = serde_json::from_slice(&json_body)?;
assert_eq!(json_body["error"]["code"], "not_found");
assert_eq!(json_body["error"]["message"], "missing");
Ok(())
}
fn authenticated_session_cookie() -> Result<String> {
let payload = format!(
"{}:test-user:test@example.com:member",
unix_time()
.map_err(|error| anyhow::anyhow!(error.message))?
.saturating_add(3600)
);
let signature = sign_with_secret(&payload, "api-token-123")
.map_err(|error| anyhow::anyhow!(error.message))?;
Ok(format!(
"ab_session={}.{}",
encode_hex(payload.as_bytes()),
signature
))
}
#[tokio::test]
async fn not_found_handlers_negotiate_branded_html_and_json_end_to_end() -> Result<()> {
let session = authenticated_session_cookie()?;
let cases = [
("/missing", "This page could not be found.", None),
(
"/repos/missing",
"This repository could not be found.",
None,
),
(
"/repos/missing/tree/main",
"Repository content could not be found.",
None,
),
(
"/repos/missing/tree/main/src/missing.rs",
"Repository content could not be found.",
None,
),
(
"/app/repositories/missing/tree/main",
"Repository content could not be found.",
Some(session.as_str()),
),
(
"/app/repositories/missing/tree/main/src/missing.rs",
"Repository content could not be found.",
Some(session.as_str()),
),
];
for (uri, heading, cookie) in cases {
let mut request = Request::builder()
.uri(uri)
.header(header::ACCEPT, "text/html,application/xhtml+xml");
if let Some(cookie) = cookie {
request = request.header(header::COOKIE, cookie);
}
let mut app = test_router();
let response = app.call(request.body(Body::empty())?).await?;
assert_eq!(response.status(), StatusCode::NOT_FOUND, "{uri}");
assert_eq!(
response.headers().get(header::CONTENT_TYPE),
Some(&HeaderValue::from_static("text/html; charset=utf-8")),
"{uri}"
);
let body = to_bytes(response.into_body(), MAX_BODY).await?;
let body = String::from_utf8(body.to_vec())?;
assert!(body.contains("AkurAI <b>Build</b>"), "{uri}");
assert!(body.contains(heading), "{uri}");
assert!(!body.contains("{\"ok\":false"), "{uri}");
let mut request = Request::builder()
.uri(uri)
.header(header::ACCEPT, "application/json");
if let Some(cookie) = cookie {
request = request.header(header::COOKIE, cookie);
}
let mut app = test_router();
let response = app.call(request.body(Body::empty())?).await?;
assert_eq!(response.status(), StatusCode::NOT_FOUND, "{uri}");
assert_eq!(
response.headers().get(header::CONTENT_TYPE),
Some(&HeaderValue::from_static("application/json")),
"{uri}"
);
let body = json_body(response).await;
assert_eq!(body["ok"], false, "{uri}");
assert_eq!(body["error"]["code"], "not_found", "{uri}");
}
Ok(())
}
#[test]
fn directory_sizes_render_as_not_applicable() -> Result<()> {
let environment = template_environment()?;
let page = CodePage {
repository: RepositoryView {
id: 1,
name: "AkurAI-Build".into(),
url: "https://example.com/AkurAI-Build.git".into(),
default_branch: "main".into(),
visibility: "public".into(),
created_at: 0,
fork_of: None,
mirror_of: None,
synced_at: None,
ssh_url: None,
},
reference: "main".into(),
path: String::new(),
entries: vec![
CodeEntryView {
name: "src".into(),
kind: "tree".into(),
size: 0,
commit_message: None,
href: "/repos/AkurAI-Build/tree/main/src".into(),
},
CodeEntryView {
name: "Cargo.toml".into(),
kind: "blob".into(),
size: 42,
commit_message: None,
href: "/repos/AkurAI-Build/tree/main/Cargo.toml".into(),
},
],
breadcrumbs: vec![CodeBreadcrumb {
label: "AkurAI-Build".into(),
href: "/repos/AkurAI-Build/tree/main".into(),
}],
source: None,
is_file: false,
parent_href: "/repos/AkurAI-Build".into(),
latest_commit: None,
commit_href: None,
public: true,
};
let rendered = environment
.get_template("repository-code.html")?
.render(context! { page => page })?;
assert!(rendered.contains(
"<td data-label=\"Size\"><span aria-label=\"Not applicable\">—</span></td>"
));
assert!(rendered.contains("<td data-label=\"Size\">42</td>"));
assert!(!rendered.contains("<td data-label=\"Size\">0</td>"));
Ok(())
}
#[test]
fn code_path_segments_encode_special_characters_and_separators() {
assert_eq!(encode_code_segment("!*'()"), "%21%2A%27%28%29");
assert_eq!(
code_path_href(
"/app/repositories/repo/tree/main",
"bang!/star*/quote'/left(/right)"
),
"/app/repositories/repo/tree/main/bang%21/star%2A/quote%27/left%28/right%29"
);
assert_eq!(encode_code_segment("../secret\\file"), "..%2Fsecret%5Cfile");
}
#[test]
fn public_repository_owner_uses_remote_namespace() {
assert_eq!(
repository_owner("https://github.com/olibuijr/AkurAI-Build.git"),
"olibuijr"
);
assert_eq!(
repository_owner("git@gitlab.example.com:platform/service.git"),
"platform"
);
assert_eq!(
repository_owner("https://akurai-build.example.com/git/service.git"),
"AkurAI Build"
);
}
#[test]
fn hosted_repository_url_uses_public_clone_route() {
let root = FsPath::new("/home/olafurbui/.local/share/akurai-build");
assert_eq!(
repository_client_url(
"AkurAI-Build",
"/home/olafurbui/.local/share/akurai-build/hosted/AkurAI-Build.git",
root,
"https://akurai-build.olibuijr.com",
),
"https://akurai-build.olibuijr.com/git/AkurAI-Build.git"
);
assert_eq!(
repository_client_url(
"AkurAI-Build",
"https://github.com/olibuijr/AkurAI-Build.git",
root,
"https://akurai-build.olibuijr.com",
),
"https://github.com/olibuijr/AkurAI-Build.git"
);
}
#[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 ──
pub(super) 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");
AppState {
database: database.clone(),
runner,
tree,
blocking_git: BlockingGit::new(MAX_BLOCKING_GIT),
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(template_environment().expect("templates")),
mcp: Arc::new(
mcp::Service::new(database.clone(), temp.path().to_owned(), false).expect("mcp"),
),
oauth: Arc::new(ResourceServer::new(
oauth::DEFAULT_ISSUER,
oauth::DEFAULT_AUDIENCE,
"https://build.example",
)),
tracker: Arc::new(linking::testing::RecordingTracker::default()),
authorized_keys: None,
ssh_address: None,
searcher: None,
}
}
#[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("/repos/{repository}", get(public_repository))
.route(
"/repos/{repository}/tree/{reference}",
get(public_code_root),
)
.route(
"/repos/{repository}/tree/{reference}/{*path}",
get(public_code_path),
)
.route(
"/app/repositories/{repository}/tree/{reference}",
get(app_code_root),
)
.route(
"/app/repositories/{repository}/tree/{reference}/{*path}",
get(app_code_path),
)
.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))
.fallback(public_not_found)
.with_state(state)
.layer(DefaultBodyLimit::max(MAX_BODY))
.layer(middleware::from_fn(security_headers))
}
pub(super) 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);
}
// ── GET /api/state ──
/// 40 repositories with one run each; even-numbered ones are public.
fn state_router_with_many_repositories() -> Result<Router> {
let state = test_state("api-token-123", "webhook-secret");
for index in 0..40 {
let name = format!("repo-{index:02}");
let repository =
state
.database
.add_repository(&name, "https://example.com/r.git", "main")?;
if index % 2 == 0 {
state.database.set_repository_visibility(&name, "public")?;
}
state
.database
.create_run(repository.id, "main", None, "manual")?;
}
Ok(Router::new()
.route("/api/state", get(api_state))
.with_state(state))
}
async fn state_runs(app: &mut Router, credential: (HeaderName, String)) -> Result<Value> {
let request = Request::builder()
.uri("/api/state?limit=200")
.header(credential.0, credential.1)
.body(Body::empty())?;
let response = app.call(request).await?;
assert_eq!(response.status(), StatusCode::OK);
Ok(json_body(response).await["data"].clone())
}
#[tokio::test]
async fn api_state_lists_runs_beyond_the_explicit_filter_cap() -> Result<()> {
let mut app = state_router_with_many_repositories()?;
let admin = (header::AUTHORIZATION, "Bearer api-token-123".to_owned());
let data = state_runs(&mut app, admin).await?;
assert_eq!(data["run_total"], 40);
assert_eq!(data["repositories"].as_array().map(Vec::len), Some(40));
let member = (header::COOKIE, authenticated_session_cookie()?);
let data = state_runs(&mut app, member).await?;
assert_eq!(
data["run_total"], 20,
"members see public repositories only"
);
let runs = data["runs"].as_array().cloned().unwrap_or_default();
assert!(runs.iter().all(|run| {
run["repository"]
.as_str()
.and_then(|name| name.strip_prefix("repo-"))
.and_then(|index| index.parse::<u32>().ok())
.is_some_and(|index| index % 2 == 0)
}));
Ok(())
}
// ── DELETE /api/repos/{repository} ──
/// Router serving only repository deletion over a data root the caller
/// owns, plus the database behind it.
fn delete_router(data_root: &FsPath) -> (Router, Database) {
let mut state = test_state("api-token-123", "webhook-secret");
state.hosted_root = data_root.to_owned();
let database = state.database.clone();
let router = Router::new()
.route("/api/repos/{repository}", delete(delete_repository))
.with_state(state)
.layer(middleware::from_fn(security_headers));
(router, database)
}
fn delete_request(uri: &str, authorized: bool) -> Request<Body> {
let mut builder = Request::builder().method("DELETE").uri(uri);
if authorized {
builder = builder.header(header::AUTHORIZATION, "Bearer api-token-123");
}
builder.body(Body::empty()).expect("build request")
}
fn statuses_router() -> (Router, Database) {
let state = test_state("api-token-123", "webhook-secret");
let database = state.database.clone();
let router = Router::new()
.route(
"/api/repos/{repository}/commits/{sha}/statuses",
get(commit_statuses),
)
.route("/api/repos/{repository}/statuses", get(batch_statuses))
.route("/api/repos/{repository}/commits", get(repo_commits))
.route("/api/repos/{repository}/commits/{sha}", get(repo_commit))
.route("/api/repos/{repository}/compare/{spec}", get(repo_compare))
.route("/api/repos/{repository}/blame", get(repo_blame))
.with_state(state);
(router, database)
}
async fn get_json(
app: &mut Router,
uri: &str,
authorized: bool,
) -> (StatusCode, serde_json::Value) {
let mut builder = Request::builder().uri(uri);
if authorized {
builder = builder.header(header::AUTHORIZATION, "Bearer api-token-123");
}
let response = app
.call(builder.body(Body::empty()).expect("request"))
.await
.expect("response");
(response.status(), json_body(response).await)
}
#[tokio::test]
async fn commit_status_endpoints_report_run_results_per_sha() -> Result<()> {
let sha = "a".repeat(40);
let (mut app, database) = statuses_router();
let repo = database.add_repository("demo", "https://example.com/demo.git", "main")?;
let run = database.create_run(repo.id, "main", Some(&sha), "manual")?;
database.insert_jobs(run, &[crate::db::tests::dummy_job_spec("verify", "verify")])?;
let job = database.jobs(run)?[0].id;
database.set_job_status(job, "failed", None, Some(1))?;
let (status, _) = get_json(
&mut app,
&format!("/api/repos/demo/commits/{sha}/statuses"),
false,
)
.await;
assert_eq!(status, StatusCode::UNAUTHORIZED);
let (status, body) = get_json(
&mut app,
&format!("/api/repos/demo/commits/{sha}/statuses"),
true,
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["data"]["state"], "failure");
assert_eq!(body["data"]["statuses"][0]["context"], "verify");
assert_eq!(body["data"]["statuses"][0]["run_id"], run);
let other = "b".repeat(40);
let (_, body) = get_json(
&mut app,
&format!("/api/repos/demo/statuses?shas={sha},{other}"),
true,
)
.await;
assert_eq!(body["data"][&sha], "failure");
assert_eq!(body["data"][&other], "none");
for (uri, expected) in [
(
"/api/repos/demo/commits/zz/statuses".to_owned(),
StatusCode::BAD_REQUEST,
),
(
format!("/api/repos/ghost/commits/{sha}/statuses"),
StatusCode::NOT_FOUND,
),
(
"/api/repos/demo/statuses?shas=nothex".to_owned(),
StatusCode::BAD_REQUEST,
),
] {
let (status, _) = get_json(&mut app, &uri, true).await;
assert_eq!(status, expected, "{uri}");
}
Ok(())
}
#[tokio::test]
async fn history_endpoints_serve_commits_diff_compare_and_blame() -> Result<()> {
let source = tempfile::tempdir()?;
let repo = source.path();
let git = |args: &[&str]| {
let status = crate::git_process::command(repo)
.args(args)
.status()
.expect("git");
assert!(status.success(), "git {args:?}");
};
git(&["init", "-b", "main"]);
git(&["config", "user.name", "Ada"]);
git(&["config", "user.email", "ada@example.invalid"]);
std::fs::write(repo.join("a.txt"), "one\n")?;
git(&["add", "."]);
git(&["commit", "-m", "first"]);
git(&["checkout", "-b", "topic"]);
std::fs::write(repo.join("a.txt"), "one\ntwo\n")?;
git(&["commit", "-am", "second"]);
let head = crate::git_process::output(repo, ["rev-parse", "HEAD"])?;
let (mut app, database) = statuses_router();
database.add_repository("demo", &repo.to_string_lossy(), "main")?;
let (status, _) = get_json(&mut app, "/api/repos/demo/commits", false).await;
assert_eq!(status, StatusCode::UNAUTHORIZED);
let (status, body) = get_json(&mut app, "/api/repos/demo/commits?ref=topic", true).await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["data"]["commits"][0]["sha"], head);
assert_eq!(body["data"]["commits"][0]["subject"], "second");
let (_, body) = get_json(&mut app, &format!("/api/repos/demo/commits/{head}"), true).await;
assert!(
body["data"]["diff"]["patch"]
.as_str()
.expect("patch is a string")
.contains("+two")
);
let (_, body) = get_json(&mut app, "/api/repos/demo/compare/main...topic", true).await;
assert_eq!(body["data"]["commits"].as_array().map(Vec::len), Some(1));
let (_, body) =
get_json(&mut app, "/api/repos/demo/blame?ref=topic&path=a.txt", true).await;
assert_eq!(body["data"]["lines"][1]["content"], "two");
assert_eq!(body["data"]["lines"][1]["author"], "Ada");
for (uri, expected) in [
("/api/repos/demo/compare/nodots", StatusCode::BAD_REQUEST),
(
"/api/repos/demo/commits?ref=--output=x",
StatusCode::BAD_REQUEST,
),
("/api/repos/demo/blame?ref=main", StatusCode::BAD_REQUEST),
("/api/repos/ghost/commits", StatusCode::NOT_FOUND),
] {
let (status, _) = get_json(&mut app, uri, true).await;
assert_eq!(status, expected, "{uri}");
}
Ok(())
}
#[tokio::test]
async fn history_pages_render_for_public_repositories_and_escape_content() -> Result<()> {
let source = tempfile::tempdir()?;
let repo = source.path();
let git = |args: &[&str]| {
assert!(
crate::git_process::command(repo)
.args(args)
.status()
.expect("git")
.success()
);
};
git(&["init", "-b", "main"]);
git(&["config", "user.name", "Ada"]);
git(&["config", "user.email", "ada@example.invalid"]);
std::fs::write(repo.join("a.txt"), "<script>alert(1)</script>\n")?;
git(&["add", "."]);
git(&["commit", "-m", "add <b>file</b>"]);
let sha = crate::git_process::output(repo, ["rev-parse", "HEAD"])?;
let state = test_state("api-token-123", "webhook-secret");
let database = state.database.clone();
database.add_repository("demo", &repo.to_string_lossy(), "main")?;
database.set_repository_visibility("demo", "public")?;
let mut app = Router::new()
.route(
"/repos/{repository}/commits/{reference}",
get(history_commits_page),
)
.route("/repos/{repository}/commit/{sha}", get(history_commit_page))
.route(
"/repos/{repository}/blame/{reference}/{*path}",
get(history_blame_page),
)
.route(
"/app/repositories/{repository}/commits/{reference}",
get(history_commits_page),
)
.with_state(state);
let mut page = |uri: String| {
let request = Request::builder()
.uri(uri)
.body(Body::empty())
.expect("request");
let call = app.call(request);
async move {
let response = call.await.expect("response");
let status = response.status();
let bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024)
.await
.expect("body");
(status, String::from_utf8_lossy(&bytes).into_owned())
}
};
let (status, html) = page("/repos/demo/commits/main".into()).await;
assert_eq!(status, StatusCode::OK);
assert!(
html.contains(&sha[..8]) && html.contains("add <b>file</b>"),
"{html}"
);
let (_, html) = page(format!("/repos/demo/commit/{sha}")).await;
assert!(
html.contains("<script>") && !html.contains("<script>alert"),
"{html}"
);
let (_, html) = page("/repos/demo/blame/main/a.txt".into()).await;
assert!(html.contains("Ada"));
let (status, _) = page("/repos/ghost/commits/main".into()).await;
assert_eq!(status, StatusCode::NOT_FOUND);
let (status, _) = page("/app/repositories/demo/commits/main".into()).await;
assert_eq!(status, StatusCode::SEE_OTHER, "app pages need a session");
Ok(())
}
#[tokio::test]
async fn delete_repository_requires_auth_and_exact_confirmation() -> Result<()> {
let root = tempfile::tempdir()?;
let (mut app, database) = delete_router(root.path());
database.add_repository("demo", "https://example.com/demo.git", "main")?;
for (uri, authorized, status) in [
(
"/api/repos/demo?confirm=demo",
false,
StatusCode::UNAUTHORIZED,
),
("/api/repos/demo", true, StatusCode::BAD_REQUEST),
(
"/api/repos/demo?confirm=other",
true,
StatusCode::BAD_REQUEST,
),
(
"/api/repos/ghost?confirm=ghost",
true,
StatusCode::NOT_FOUND,
),
] {
let response = app.call(delete_request(uri, authorized)).await?;
assert_eq!(response.status(), status, "{uri}");
assert_eq!(json_body(response).await["ok"], false, "{uri}");
}
assert!(database.repository("demo").is_ok(), "nothing was deleted");
Ok(())
}
#[tokio::test]
async fn delete_repository_refuses_while_a_run_is_active() -> Result<()> {
let root = tempfile::tempdir()?;
let hosted = root.path().join("hosted/busy.git");
std::fs::create_dir_all(&hosted)?;
let (mut app, database) = delete_router(root.path());
let repository = database.add_repository("busy", &hosted.to_string_lossy(), "main")?;
let run = database.create_run(repository.id, "main", None, "manual")?;
assert!(database.claim_run(run)?);
let response = app
.call(delete_request("/api/repos/busy?confirm=busy", true))
.await?;
assert_eq!(response.status(), StatusCode::CONFLICT);
assert_eq!(json_body(response).await["error"]["code"], "conflict");
assert!(hosted.is_dir(), "hosted repository is kept");
assert!(database.run(run).is_ok(), "run is kept");
Ok(())
}
#[tokio::test]
async fn delete_repository_removes_rows_and_hosted_directory() -> Result<()> {
let root = tempfile::tempdir()?;
let hosted = root.path().join("hosted/demo.git");
std::fs::create_dir_all(&hosted)?;
let (mut app, database) = delete_router(root.path());
let repository = database.add_repository("demo", &hosted.to_string_lossy(), "main")?;
let run = database.create_run(repository.id, "main", None, "manual")?;
database.finish_run(run, "failed", Some("boom"))?;
let artifacts = root.path().join(format!("artifacts/{run}"));
std::fs::create_dir_all(&artifacts)?;
let response = app
.call(delete_request("/api/repos/demo?confirm=demo", true))
.await?;
assert_eq!(response.status(), StatusCode::OK);
let body = json_body(response).await;
assert_eq!(body["ok"], true);
assert_eq!(body["data"]["repository"]["name"], "demo");
assert_eq!(body["data"]["run_ids"], serde_json::json!([run]));
assert_eq!(
body["data"]["removed_paths"].as_array().map(Vec::len),
Some(2)
);
assert!(!hosted.exists(), "hosted bare repository is removed");
assert!(!artifacts.exists(), "artifact store is removed");
assert!(database.repository("demo").is_err());
assert!(database.run(run).is_err());
Ok(())
}
#[tokio::test]
async fn blocking_git_limits_concurrency_and_preserves_api_errors() {
use std::sync::atomic::{AtomicUsize, Ordering};
let gate = BlockingGit::new(1);
let active = Arc::new(AtomicUsize::new(0));
let maximum = Arc::new(AtomicUsize::new(0));
let operation = |active: Arc<AtomicUsize>, maximum: Arc<AtomicUsize>| {
move || {
let current = active.fetch_add(1, Ordering::SeqCst) + 1;
maximum.fetch_max(current, Ordering::SeqCst);
std::thread::sleep(Duration::from_millis(25));
active.fetch_sub(1, Ordering::SeqCst);
Ok(())
}
};
let first = gate.run(operation(active.clone(), maximum.clone()));
let second = gate.run(operation(active, maximum.clone()));
let (first, second) = tokio::join!(first, second);
assert!(first.is_ok());
assert!(second.is_ok());
assert_eq!(maximum.load(Ordering::SeqCst), 1);
let error = gate
.run(|| Err::<(), _>(ApiError::bad("specific git failure")))
.await
.expect_err("operation error");
assert_eq!(error.status, StatusCode::BAD_REQUEST);
assert_eq!(error.message, "specific git failure");
}
// ── HTTP MCP endpoint (OAuth protected resource) ──
const MCP_AUD: [&str; 2] = ["akurai-agent-akurai-build", "akurai-build"];
fn mcp_router(idp: &TestIdp) -> Router {
mcp_router_for(idp, "admin@example.com")
}
/// Router whose IdP userinfo reports `email` (verified) for `user-1`.
fn mcp_router_for(idp: &TestIdp, email: &'static str) -> Router {
let mut state = test_state("api-token-123", "webhook-secret");
state.oauth = Arc::new(
ResourceServer::new(
oauth_testing::ISSUER,
"akurai-build",
"https://build.example",
)
.with_jwks(idp.jwks())
.with_admins(vec!["admin@example.com".to_owned()])
.with_userinfo(move |_| {
Ok(serde_json::json!({"sub": "user-1", "email": email, "email_verified": true}))
}),
);
mcp_routes()
.with_state(state)
.layer(middleware::from_fn(security_headers))
}
fn mcp_request(authorization: Option<&str>, origin: Option<&str>) -> Request<Body> {
let mut builder = Request::builder()
.method("POST")
.uri("/mcp")
.header(header::CONTENT_TYPE, "application/json");
if let Some(value) = authorization {
builder = builder.header(header::AUTHORIZATION, value);
}
if let Some(value) = origin {
builder = builder.header(header::ORIGIN, value);
}
builder
.body(Body::from(
r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}"#,
))
.expect("build request")
}
fn challenge(response: &Response) -> &str {
response
.headers()
.get(header::WWW_AUTHENTICATE)
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
}
#[tokio::test]
async fn mcp_protected_resource_metadata_is_public() {
let idp = TestIdp::new();
for uri in [
"/.well-known/oauth-protected-resource",
"/.well-known/oauth-protected-resource/mcp",
] {
let request = Request::builder()
.uri(uri)
.body(Body::empty())
.expect("request");
let response = mcp_router(&idp).call(request).await.expect("call");
assert_eq!(response.status(), StatusCode::OK);
let body = json_body(response).await;
assert_eq!(body["resource"], "https://build.example/mcp");
assert_eq!(body["authorization_servers"][0], oauth_testing::ISSUER);
}
}
#[tokio::test]
async fn mcp_without_token_gets_resource_metadata_challenge() {
let idp = TestIdp::new();
let response = mcp_router(&idp)
.call(mcp_request(None, None))
.await
.expect("call");
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
assert_eq!(
challenge(&response),
"Bearer resource_metadata=\"https://build.example/.well-known/oauth-protected-resource/mcp\""
);
}
#[tokio::test]
async fn mcp_rejects_invalid_token_with_invalid_token_challenge() {
let idp = TestIdp::new();
let wrong_audience = idp.sign(
oauth_testing::KID,
&user_claims(serde_json::json!(["akurai-notes"])),
);
for token in ["wrong-value".to_owned(), wrong_audience] {
let response = mcp_router(&idp)
.call(mcp_request(Some(&format!("Bearer {token}")), None))
.await
.expect("call");
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
assert!(challenge(&response).starts_with("Bearer error=\"invalid_token\""));
}
}
#[tokio::test]
async fn mcp_accepts_idp_access_token() {
let idp = TestIdp::new();
let token = idp.sign(oauth_testing::KID, &user_claims(serde_json::json!(MCP_AUD)));
let response = mcp_router(&idp)
.call(mcp_request(Some(&format!("Bearer {token}")), None))
.await
.expect("call");
assert_eq!(response.status(), StatusCode::OK);
let body = json_body(response).await;
assert_eq!(body["result"]["serverInfo"]["name"], "akurai-build");
}
#[tokio::test]
async fn mcp_forbids_idp_users_outside_the_admin_allowlist() {
let idp = TestIdp::new();
let token = idp.sign(oauth_testing::KID, &user_claims(serde_json::json!(MCP_AUD)));
let response = mcp_router_for(&idp, "member@example.com")
.call(mcp_request(Some(&format!("Bearer {token}")), None))
.await
.expect("call");
assert_eq!(response.status(), StatusCode::FORBIDDEN);
assert!(challenge(&response).starts_with("Bearer error=\"insufficient_scope\""));
}
#[tokio::test]
async fn mcp_keeps_accepting_the_api_token() {
let idp = TestIdp::new();
let response = mcp_router(&idp)
.call(mcp_request(Some("Bearer api-token-123"), None))
.await
.expect("call");
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(json_body(response).await["id"], 1);
}
#[tokio::test]
async fn mcp_rejects_foreign_origin() {
let idp = TestIdp::new();
let response = mcp_router(&idp)
.call(mcp_request(
Some("Bearer api-token-123"),
Some("https://evil.example"),
))
.await
.expect("call");
assert_eq!(response.status(), StatusCode::FORBIDDEN);
}
#[tokio::test]
async fn mcp_notification_is_accepted_without_body() {
let idp = TestIdp::new();
let request = Request::builder()
.method("POST")
.uri("/mcp")
.header(header::AUTHORIZATION, "Bearer api-token-123")
.body(Body::from(
r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#,
))
.expect("request");
let response = mcp_router(&idp).call(request).await.expect("call");
assert_eq!(response.status(), StatusCode::ACCEPTED);
}
}