eba4fd1
use std::collections::BTreeMap;
use axum::extract::FromRequestParts;
use axum::http::header;
use ci_protocol::runner::{LeaseDto, RegisterResponse, StepPayload};
use models::ci::runner::{self, RunnerState};
use models::ci::runner_token;
use models::ci::workflow_job;
use models::ci::workflow_run;
use models::ci::workflow_step;
use sea_orm::{
ActiveModelTrait, ColumnTrait, Condition, EntityTrait, QueryFilter, QueryOrder, Set,
};
use sha2::{Digest, Sha256};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::errors::{CiError, CiResult};
use crate::secrets::{constant_time_eq, random_secret};
use crate::service::CiService;
#[derive(Debug, Clone)]
pub struct RunnerAuth {
pub runner_id: Uuid,
}
pub fn hash_token(secret: &str, salt: &str) -> String {
let digest = Sha256::digest(format!("{secret}{salt}").as_bytes());
let mut out = String::with_capacity(digest.len() * 2);
for byte in digest {
out.push_str(&format!("{byte:02x}"));
}
out
}
pub fn generate_credential(runner_id: Uuid) -> (String, String, String, String) {
let secret = random_secret(32);
let salt = random_secret(16);
let hash = hash_token(&secret, &salt);
let last_eight: String = secret
.chars()
.rev()
.take(8)
.collect::<String>()
.chars()
.rev()
.collect();
let token = format!("{runner_id}.{secret}");
(token, hash, salt, last_eight)
}
pub async fn verify_runner_token(service: &CiService, token: &str) -> CiResult<Uuid> {
let Some((id, secret)) = token.split_once('.') else {
return Err(CiError::Unauthorized);
};
let Ok(runner_id) = Uuid::parse_str(id) else {
return Err(CiError::Unauthorized);
};
let row = runner_token::Entity::find()
.filter(runner_token::Column::RunnerId.eq(runner_id))
.one(service.db_read())
.await?;
let Some(row) = row else {
return Err(CiError::Unauthorized);
};
if row.revoked_at.is_some() {
return Err(CiError::Unauthorized);
}
if let Some(expires_at) = row.expires_at
&& expires_at <= OffsetDateTime::now_utc()
{
return Err(CiError::Unauthorized);
}
let computed = hash_token(secret, &row.token_salt);
if !constant_time_eq(computed.as_bytes(), row.token_hash.as_bytes()) {
return Err(CiError::Unauthorized);
}
let runner = runner::Entity::find_by_id(runner_id)
.one(service.db_read())
.await?
.ok_or(CiError::Unauthorized)?;
if runner.state != RunnerState::Active {
return Err(CiError::Forbidden);
}
Ok(runner_id)
}
impl FromRequestParts<CiService> for RunnerAuth {
type Rejection = crate::errors::ApiError;
async fn from_request_parts(
parts: &mut axum::http::request::Parts,
state: &CiService,
) -> Result<Self, Self::Rejection> {
let Some(value) = parts.headers.get(header::AUTHORIZATION) else {
return Err(crate::errors::ApiError::unauthorized());
};
let raw = value
.to_str()
.map_err(|_| crate::errors::ApiError::bad_request("malformed authorization header"))?;
let Some(token) = raw.strip_prefix("Bearer ") else {
return Err(crate::errors::ApiError::unauthorized());
};
let runner_id = verify_runner_token(state, token)
.await
.map_err(crate::errors::ApiError::from)?;
Ok(Self { runner_id })
}
}
impl CiService {
pub async fn register_runner(
&self,
token: &str,
name: &str,
labels: Vec<String>,
capacity: i32,
version: Option<String>,
) -> CiResult<RegisterResponse> {
self.register_runner_inner(token, name, labels, capacity, version)
.await
}
pub async fn configure_runner_protocol(
&self,
runner_id: Uuid,
versions: &[u32],
capabilities: &[String],
) -> CiResult<u32> {
let version = if versions.contains(&2) {
2
} else if versions.is_empty() || versions.contains(&1) {
1
} else {
return Err(CiError::InvalidArgument(
"no common runner protocol version".into(),
));
};
let row = runner::Entity::find_by_id(runner_id)
.one(self.db_write())
.await?
.ok_or(CiError::NotFound)?;
let mut active: runner::ActiveModel = row.into();
active.protocol_version = Set(Some(version as i32));
active.capabilities = Set(Some(serde_json::to_value(capabilities)?));
active.update(self.db_write()).await?;
Ok(version)
}
async fn register_runner_inner(
&self,
token: &str,
name: &str,
labels: Vec<String>,
capacity: i32,
version: Option<String>,
) -> CiResult<RegisterResponse> {
self.ensure_enabled()?;
if name.trim().is_empty() || name.len() > 200 {
return Err(CiError::InvalidArgument(
"runner name must be 1-200 characters".to_string(),
));
}
let mut labels: Vec<String> = labels
.into_iter()
.map(|l| l.trim().to_string())
.filter(|l| !l.is_empty())
.collect();
labels.sort();
labels.dedup();
if labels.is_empty() {
return Err(CiError::InvalidArgument(
"runner must declare at least one label".to_string(),
));
}
if capacity <= 0 || capacity > 256 {
return Err(CiError::InvalidArgument(
"runner capacity must be between 1 and 256".to_string(),
));
}
let (registration_token, space_scope) = self.verify_registration_token(token).await?;
let now = OffsetDateTime::now_utc();
let runner_id = Uuid::now_v7();
runner::ActiveModel {
id: Set(runner_id),
name: Set(name.trim().to_string()),
labels: Set(serde_json::to_value(&labels)?),
capacity: Set(capacity),
version: Set(version),
state: Set(RunnerState::Active),
space_id: Set(space_scope),
last_seen_at: Set(Some(now)),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
}
.insert(self.db_write())
.await?;
let (credential, hash, salt, last_eight) = generate_credential(runner_id);
runner_token::ActiveModel {
id: Set(Uuid::now_v7()),
runner_id: Set(Some(runner_id)),
name: Set(format!("runner:{name}")),
token_hash: Set(hash),
token_salt: Set(salt),
token_last_eight: Set(last_eight),
space_id: Set(space_scope),
single_use: Set(false),
used_at: Set(None),
revoked_at: Set(None),
expires_at: Set(None),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db_write())
.await?;
if let Some(registration_token_id) = registration_token {
let mut am: runner_token::ActiveModel = registration_token_id.into();
am.used_at = Set(Some(now));
am.updated_at = Set(now);
let _ = am.update(self.db_write()).await;
}
tracing::info!(runner_id = %runner_id, name, "runner registered");
Ok(RegisterResponse {
runner_id,
credential,
heartbeat_interval_secs: self.config().ci.heartbeat_interval_secs,
poll_timeout_secs: self.config().ci.long_poll_timeout_secs,
protocol_version: 1,
})
}
async fn verify_registration_token(
&self,
token: &str,
) -> CiResult<(Option<runner_token::Model>, Option<Uuid>)> {
let Some((id, secret)) = token.split_once('.') else {
return Err(CiError::Unauthorized);
};
let Ok(token_id) = Uuid::parse_str(id) else {
return Err(CiError::Unauthorized);
};
let row = runner_token::Entity::find_by_id(token_id)
.one(self.db_read())
.await?
.ok_or(CiError::Unauthorized)?;
if row.runner_id.is_some() || row.revoked_at.is_some() {
return Err(CiError::Unauthorized);
}
if row.single_use && row.used_at.is_some() {
return Err(CiError::Unauthorized);
}
if let Some(expires_at) = row.expires_at
&& expires_at <= OffsetDateTime::now_utc()
{
return Err(CiError::Unauthorized);
}
let computed = hash_token(secret, &row.token_salt);
if !constant_time_eq(computed.as_bytes(), row.token_hash.as_bytes()) {
return Err(CiError::Unauthorized);
}
Ok((Some(row.clone()), row.space_id))
}
pub async fn runner_heartbeat(&self, runner_id: Uuid) -> CiResult<Vec<Uuid>> {
let now = OffsetDateTime::now_utc();
runner::Entity::update_many()
.col_expr(
runner::Column::LastSeenAt,
sea_orm::sea_query::Expr::value(Some(now)),
)
.col_expr(
runner::Column::UpdatedAt,
sea_orm::sea_query::Expr::value(now),
)
.filter(runner::Column::Id.eq(runner_id))
.exec(self.db_write())
.await?;
self.cancel_list_for_runner(runner_id).await
}
/// Disables runners that stopped heartbeating and revokes their credentials.
pub async fn reap_stale_runners(&self) -> CiResult<u64> {
let now = OffsetDateTime::now_utc();
let stale_secs = self.config().ci.runner_stale_secs.max(1) as i64;
let cutoff = now - time::Duration::seconds(stale_secs);
let stale = runner::Entity::find()
.filter(runner::Column::State.eq(RunnerState::Active))
.filter(
Condition::any()
.add(runner::Column::LastSeenAt.lt(cutoff))
.add(runner::Column::LastSeenAt.is_null()),
)
.all(self.db_write())
.await?;
let mut ids = Vec::with_capacity(stale.len());
for row in stale {
let id = row.id;
let mut am: runner::ActiveModel = row.into();
am.state = Set(RunnerState::Disabled);
am.updated_at = Set(now);
am.update(self.db_write()).await?;
ids.push(id);
}
if ids.is_empty() {
return Ok(0);
}
runner_token::Entity::update_many()
.col_expr(
runner_token::Column::RevokedAt,
sea_orm::sea_query::Expr::value(Some(now)),
)
.col_expr(
runner_token::Column::UpdatedAt,
sea_orm::sea_query::Expr::value(now),
)
.filter(runner_token::Column::RunnerId.is_in(ids.clone()))
.filter(runner_token::Column::RevokedAt.is_null())
.exec(self.db_write())
.await?;
tracing::info!(count = ids.len(), "stale runners disabled");
Ok(ids.len() as u64)
}
pub async fn runner_labels(&self, runner_id: Uuid) -> CiResult<Vec<String>> {
let runner = runner::Entity::find_by_id(runner_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
Ok(serde_json::from_value(runner.labels).unwrap_or_default())
}
/// Builds the full claim payload for a job that this runner has leased.
pub async fn build_job_payload(
&self,
job_id: Uuid,
_runner_id: Uuid,
lease_token: String,
) -> CiResult<ci_protocol::runner::JobPayload> {
let job = workflow_job::Entity::find_by_id(job_id)
.one(self.db_write())
.await?
.ok_or(CiError::NotFound)?;
let run = workflow_run::Entity::find_by_id(job.run_id)
.one(self.db_write())
.await?
.ok_or(CiError::NotFound)?;
if job.actions_context.is_some() {
return self
.build_actions_job_payload(&job, &run, lease_token)
.await;
}
let repo = self.repo_by_id(job.repo_id).await?;
let space = self.space_of_repo(&repo).await?;
let definition: serde_json::Value = job.definition.clone();
let steps = workflow_step::Entity::find()
.filter(workflow_step::Column::JobId.eq(job_id))
.order_by_asc(workflow_step::Column::StepIndex)
.all(self.db_read())
.await?;
let step_env: BTreeMap<String, String> = definition
.get("env")
.and_then(|v| serde_json::from_value(v.clone()).ok())
.unwrap_or_default();
let mut step_payloads = Vec::with_capacity(steps.len());
for step in steps {
step_payloads.push(StepPayload {
index: step.step_index,
name: step.name.clone(),
run: step.script.clone(),
shell: step.shell.clone(),
working_directory: step.working_directory.clone(),
env: step_env.clone(),
timeout_secs: definition
.get("timeout-minutes")
.and_then(|v| v.as_u64())
.map(|m| m * 60),
});
}
let workflow_env: BTreeMap<String, String> = run
.definition
.get("env")
.and_then(|v| serde_json::from_value(v.clone()).ok())
.unwrap_or_default();
let mut env = BTreeMap::new();
env.insert("GITDATA_RUN_ID".to_string(), run.id.to_string());
env.insert("GITDATA_JOB_ID".to_string(), job.id.to_string());
env.insert("GITDATA_REPO_ID".to_string(), repo.id.to_string());
env.insert("GITDATA_SPACE".to_string(), space.lower_name.clone());
env.insert("GITDATA_REPO".to_string(), repo.name.clone());
env.insert("GITDATA_SHA".to_string(), run.commit_sha.clone());
env.insert("GITDATA_REF".to_string(), run.git_ref.clone());
env.insert("GITDATA_EVENT".to_string(), run.event.as_str().to_string());
env.insert("GITDATA_WORKFLOW".to_string(), run.workflow_name.clone());
env.insert("CI".to_string(), "true".to_string());
env.insert("GITDATA_CI".to_string(), "true".to_string());
if let Some(branch) = &run.branch {
env.insert("GITDATA_BRANCH".to_string(), branch.clone());
}
if let Some(tag) = &run.tag {
env.insert("GITDATA_TAG".to_string(), tag.clone());
}
if let Some(base_sha) = &run.base_sha {
env.insert("GITDATA_BASE_SHA".to_string(), base_sha.clone());
}
if let Some(base_ref) = &run.base_ref {
env.insert("GITDATA_BASE_REF".to_string(), base_ref.clone());
}
if let Some(number) = run.pr_number {
env.insert("GITDATA_PR_NUMBER".to_string(), number.to_string());
}
if let Some(environment) = &job.environment {
env.insert("GITDATA_ENVIRONMENT".to_string(), environment.clone());
}
for (key, value) in workflow_env {
env.insert(key, value);
}
let secrets: BTreeMap<String, String> = self
.resolved_secrets(repo.id, space.id, job.environment.as_deref())
.await?
.into_iter()
.collect();
let contents_write = definition
.get("permissions")
.and_then(|p| p.get("contents"))
.and_then(|c| c.as_str())
.is_some_and(|c| c == "write");
let artifacts: Vec<ci_protocol::runner::ArtifactSpecDto> = definition
.get("artifacts")
.and_then(|v| serde_json::from_value(v.clone()).ok())
.unwrap_or_default();
let needs_artifacts = self.needs_artifacts(run.id, job.id).await?;
let artifacts: Vec<ci_protocol::runner::ArtifactSpecDto> = artifacts
.into_iter()
.map(|spec| ci_protocol::runner::ArtifactSpecDto {
name: spec.name,
path: spec.path,
always: spec.always,
})
.collect();
let clone_url = format!(
"{}/{}/{}.git",
self.config().git.base_url.trim_end_matches('/'),
space.lower_name,
repo.name
);
let payload = ci_protocol::runner::JobPayload {
job_id: job.id,
run_id: run.id,
repo_id: repo.id,
space_id: space.id,
run_attempt: job.attempt,
job_name: job.name.clone(),
clone_url,
commit_sha: run.commit_sha.clone(),
git_ref: run.git_ref.clone(),
branch: run.branch.clone(),
tag: run.tag.clone(),
base_sha: run.base_sha.clone(),
base_ref: run.base_ref.clone(),
pr_number: run.pr_number,
event: run.event.as_str().to_string(),
workflow_name: run.workflow_name.clone(),
workflow_file: run.workflow_file.clone(),
env,
secrets,
allow_git_token: contents_write,
container: definition
.get("container")
.and_then(|v| v.as_str())
.map(|s| s.to_string()),
timeout_secs: definition
.get("timeout-minutes")
.and_then(|v| v.as_u64())
.map(|m| m * 60)
.unwrap_or(self.config().ci.job_timeout_default_secs)
.min(self.config().ci.job_timeout_max_secs),
steps: step_payloads,
artifacts,
needs_artifacts,
lease: LeaseDto {
token: lease_token,
expires_at: OffsetDateTime::now_utc()
+ time::Duration::seconds(self.config().ci.lease_secs as i64),
heartbeat_interval_secs: self.config().ci.heartbeat_interval_secs,
attempt: job.attempt,
},
actions: None,
};
Ok(payload)
}
}
impl CiService {
/// Returns the claim payload (with a fresh lease token) for a leased job.
pub async fn claim_payload(
&self,
job_id: Uuid,
runner_id: Uuid,
) -> CiResult<Option<ci_protocol::runner::JobPayload>> {
let job = workflow_job::Entity::find_by_id(job_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
if job.runner_id != Some(runner_id) {
return Ok(None);
}
let lease_token = job
.lease_token
.clone()
.unwrap_or_else(|| crate::secrets::random_secret(24));
self.build_job_payload(job_id, runner_id, lease_token)
.await
.map(Some)
}
}