80f9e01
use axum::{
Json, Router,
body::Bytes,
extract::{DefaultBodyLimit, Path, Query, State},
http::{HeaderMap, StatusCode},
response::{IntoResponse, Response},
routing::{get, post, put},
};
use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
use futures::StreamExt;
use hmac::{Hmac, KeyInit, Mac};
use models::ci::workflow_artifact::ArtifactState;
use models::ci::{actions_cache, actions_identity, workflow_artifact, workflow_job, workflow_run};
use sea_orm::{
ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, QueryOrder, Set,
TransactionTrait,
};
use serde::Deserialize;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use time::OffsetDateTime;
use tokio::io::{AsyncSeekExt, AsyncWriteExt};
use uuid::Uuid;
use crate::{
CiService,
errors::{CiError, CiResult},
};
pub fn router() -> Router<CiService> {
Router::new()
.route(
"/twirp/github.actions.results.api.v1.ArtifactService/{method}",
post(artifact_rpc),
)
.route(
"/twirp/github.actions.results.api.v1.CacheService/{method}",
post(cache_rpc),
)
.route("/actions-runtime/uploads/{kind}/{id}", put(upload_blob))
.route("/actions-runtime/oidc", get(oidc_token))
.route("/.well-known/openid-configuration", get(oidc_configuration))
.route("/.well-known/jwks.json", get(oidc_jwks))
.route("/_apis/artifactcache/cache", get(legacy_cache_lookup))
.route("/_apis/artifactcache/caches", post(legacy_cache_reserve))
.route(
"/_apis/artifactcache/caches/{id}",
post(legacy_cache_finalize).patch(legacy_cache_upload),
)
.layer(DefaultBodyLimit::max(64 * 1024 * 1024))
}
pub struct RuntimeError(CiError);
impl From<CiError> for RuntimeError {
fn from(error: CiError) -> Self {
Self(error)
}
}
impl From<storage::StorageError> for RuntimeError {
fn from(error: storage::StorageError) -> Self {
Self(error.into())
}
}
impl From<sea_orm::DbErr> for RuntimeError {
fn from(error: sea_orm::DbErr) -> Self {
Self(error.into())
}
}
impl IntoResponse for RuntimeError {
fn into_response(self) -> Response {
let (status, code) = match &self.0 {
CiError::Unauthorized => (StatusCode::UNAUTHORIZED, "unauthenticated"),
CiError::Forbidden => (StatusCode::FORBIDDEN, "permission_denied"),
CiError::NotFound => (StatusCode::NOT_FOUND, "not_found"),
CiError::Conflict(_) => (StatusCode::CONFLICT, "already_exists"),
CiError::InvalidArgument(_) => (StatusCode::BAD_REQUEST, "invalid_argument"),
_ => (StatusCode::INTERNAL_SERVER_ERROR, "internal"),
};
(status, Json(json!({"code":code,"msg":self.0.to_string()}))).into_response()
}
}
#[derive(Clone)]
struct RuntimeContext {
job: workflow_job::Model,
run: workflow_run::Model,
}
impl CiService {
pub fn sign_actions_token(&self, payload: &Value) -> CiResult<String> {
let header = URL_SAFE_NO_PAD.encode(br#"{"alg":"HS256","typ":"JWT"}"#);
let data = URL_SAFE_NO_PAD.encode(serde_json::to_vec(payload)?);
let message = format!("{header}.{data}");
let mut mac = Hmac::<Sha256>::new_from_slice(self.encryption_key().as_bytes())
.map_err(|error| CiError::Internal(error.to_string()))?;
mac.update(message.as_bytes());
Ok(format!(
"{message}.{}",
URL_SAFE_NO_PAD.encode(mac.finalize().into_bytes())
))
}
fn verify_actions_token(&self, token: &str) -> CiResult<Value> {
let parts: Vec<_> = token.split('.').collect();
if parts.len() != 3 {
return Err(CiError::Unauthorized);
}
let mut mac = Hmac::<Sha256>::new_from_slice(self.encryption_key().as_bytes())
.map_err(|error| CiError::Internal(error.to_string()))?;
mac.update(format!("{}.{}", parts[0], parts[1]).as_bytes());
let signature = URL_SAFE_NO_PAD
.decode(parts[2])
.map_err(|_| CiError::Unauthorized)?;
mac.verify_slice(&signature)
.map_err(|_| CiError::Unauthorized)?;
let data = URL_SAFE_NO_PAD
.decode(parts[1])
.map_err(|_| CiError::Unauthorized)?;
let value: Value = serde_json::from_slice(&data).map_err(|_| CiError::Unauthorized)?;
if value["exp"].as_i64().unwrap_or(0) <= OffsetDateTime::now_utc().unix_timestamp() {
return Err(CiError::Unauthorized);
}
Ok(value)
}
async fn runtime_context(&self, headers: &HeaderMap) -> CiResult<RuntimeContext> {
self.ensure_enabled()?;
let token = headers
.get("authorization")
.and_then(|value| value.to_str().ok())
.and_then(|value| value.strip_prefix("Bearer "))
.ok_or(CiError::Unauthorized)?;
let claims = self.verify_actions_token(token)?;
if claims["aud"] != "gitdata-actions-runtime" {
return Err(CiError::Unauthorized);
}
self.runtime_context_from_claims(&claims).await
}
async fn runtime_context_from_claims(&self, claims: &Value) -> CiResult<RuntimeContext> {
let id = claims["job_id"]
.as_str()
.and_then(|value| Uuid::parse_str(value).ok())
.ok_or(CiError::Unauthorized)?;
let job = workflow_job::Entity::find_by_id(id)
.one(self.db_write())
.await?
.ok_or(CiError::Unauthorized)?;
if !job.state.is_active() || job.lease_token.as_deref() != claims["lease_token"].as_str() {
return Err(CiError::Unauthorized);
}
let run = workflow_run::Entity::find_by_id(job.run_id)
.one(self.db_write())
.await?
.ok_or(CiError::Unauthorized)?;
Ok(RuntimeContext { job, run })
}
pub fn actions_runtime_url(&self) -> CiResult<String> {
let configured = &self.config().ci.runtime_base_url;
let base = if !configured.is_empty() {
configured.as_str()
} else if !self.config().webhook.public_base_url.is_empty() {
self.config().webhook.public_base_url.as_str()
} else {
self.config().git.base_url.as_str()
};
let url = reqwest::Url::parse(base).map_err(|error| {
CiError::InvalidArgument(format!("invalid CI runtime origin: {error}"))
})?;
if url.path() != "/" && !url.path().is_empty() {
return Err(CiError::InvalidArgument(
"ci.runtime_base_url must be a URL origin without a path prefix".into(),
));
}
Ok(base.trim_end_matches('/').to_owned())
}
pub fn actions_runtime_token(
&self,
job: &workflow_job::Model,
run: &workflow_run::Model,
) -> CiResult<String> {
self.sign_actions_token(&json!({"aud":"gitdata-actions-runtime","job_id":job.id,"run_id":run.id,"repo_id":job.repo_id,"lease_token":job.lease_token,
"scp":format!("Actions.Results:{}:{}",run.actions_id.unwrap_or(0),job.actions_id.unwrap_or(0)),"iat":OffsetDateTime::now_utc().unix_timestamp(),
"exp":OffsetDateTime::now_utc().unix_timestamp()+self.config().ci.job_timeout_max_secs as i64+600}))
}
fn signed_upload_url(
&self,
context: &RuntimeContext,
kind: &str,
id: &str,
) -> CiResult<String> {
let token=self.sign_actions_token(&json!({"aud":"gitdata-actions-upload","job_id":context.job.id,"lease_token":context.job.lease_token,"kind":kind,"id":id,"exp":OffsetDateTime::now_utc().unix_timestamp()+3600}))?;
Ok(format!(
"{}/actions-runtime/uploads/{kind}/{id}?token={token}",
self.actions_runtime_url()?
))
}
async fn numeric_actions_id(&self, entity_id: Uuid) -> CiResult<i64> {
actions_identity::Entity::find()
.filter(actions_identity::Column::EntityId.eq(entity_id))
.one(self.db_write())
.await?
.map(|identity| identity.id)
.ok_or(CiError::NotFound)
}
fn check_artifact_scope(&self, context: &RuntimeContext, body: &Value) -> CiResult<()> {
let run = field(body, "workflowRunBackendId", "workflow_run_backend_id");
if !run.is_empty() && run != context.run.actions_id.unwrap_or(0).to_string() {
return Err(CiError::Forbidden);
}
Ok(())
}
async fn artifact_call(
&self,
context: &RuntimeContext,
method: &str,
body: &Value,
) -> CiResult<Value> {
self.check_artifact_scope(context, body)?;
let name = field(body, "name", "name");
match method {
"CreateArtifact" => {
if name.trim().is_empty() || name.len() > 255 {
return Err(CiError::InvalidArgument(
"artifact name must be 1-255 characters".into(),
));
}
let txn = self.db_write().begin().await?;
txn.execute_unprepared(&format!(
"SELECT pg_advisory_xact_lock({})",
context.run.id.as_u128() as i64
))
.await?;
let existing = workflow_artifact::Entity::find()
.filter(workflow_artifact::Column::RunId.eq(context.run.id))
.filter(workflow_artifact::Column::Name.eq(&name))
.one(&txn)
.await?;
if let Some(existing) = existing {
if existing.job_id == context.job.id
&& existing.state == ArtifactState::Uploading
{
let url =
self.signed_upload_url(context, "artifact", &existing.id.to_string())?;
return Ok(json!({"ok":true,"signedUploadUrl":url}));
}
return Err(CiError::Conflict(
"artifact name already exists in this run".into(),
));
}
let id = Uuid::now_v7();
let now = OffsetDateTime::now_utc();
let configured_expiry =
now + time::Duration::days(self.config().ci.artifact_retention_days.max(1));
let requested_expiry = body
.get("expiresAt")
.or_else(|| body.get("expires_at"))
.and_then(Value::as_str)
.and_then(|value| {
OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339)
.ok()
});
let expires = requested_expiry
.map(|value| value.min(configured_expiry))
.unwrap_or(configured_expiry);
workflow_artifact::ActiveModel {
id: Set(id),
run_id: Set(context.run.id),
job_id: Set(context.job.id),
name: Set(name),
object_key: Set(CiService::artifact_object_key(
context.run.id,
context.job.id,
id,
)),
size: Set(0),
sha256: Set(None),
mime_type: Set(body
.get("mimeType")
.or_else(|| body.get("mime_type"))
.and_then(|value| value.as_str().or_else(|| value["value"].as_str()))
.map(str::to_owned)),
state: Set(ArtifactState::Uploading),
expires_at: Set(Some(expires)),
created_at: Set(now),
updated_at: Set(now),
}
.insert(&txn)
.await?;
crate::actions::identity(&txn, id, "artifact").await?;
txn.commit().await?;
Ok(
json!({"ok":true,"signedUploadUrl":self.signed_upload_url(context,"artifact",&id.to_string())?}),
)
}
"FinalizeArtifact" => {
let row = workflow_artifact::Entity::find()
.filter(workflow_artifact::Column::RunId.eq(context.run.id))
.filter(workflow_artifact::Column::JobId.eq(context.job.id))
.filter(workflow_artifact::Column::Name.eq(name))
.one(self.db_write())
.await?
.ok_or(CiError::NotFound)?;
let declared = number(body.get("size"))?;
let (size, hash) = self
.uploaded_size_and_hash(&row.object_key, self.config().ci.artifact_max_bytes)
.await?;
if size != declared {
return Err(CiError::InvalidArgument(
"artifact size does not match upload".into(),
));
}
let expected = body
.get("hash")
.and_then(|v| v.as_str().or_else(|| v["value"].as_str()))
.map(|v| v.trim_start_matches("sha256:"));
if expected.is_some_and(|value| value != hash) {
return Err(CiError::InvalidArgument(
"artifact digest does not match upload".into(),
));
}
let id = self.numeric_actions_id(row.id).await?;
let mut active: workflow_artifact::ActiveModel = row.into();
active.size = Set(size as i64);
active.sha256 = Set(Some(hash));
active.state = Set(ArtifactState::Ready);
active.updated_at = Set(OffsetDateTime::now_utc());
active.update(self.db_write()).await?;
Ok(json!({"ok":true,"artifactId":id.to_string()}))
}
"ListArtifacts" => {
let rows = workflow_artifact::Entity::find()
.filter(workflow_artifact::Column::RunId.eq(context.run.id))
.filter(workflow_artifact::Column::State.eq(ArtifactState::Ready))
.all(self.db_write())
.await?;
let name_filter = field(body, "nameFilter", "name_filter");
let id_filter = body
.get("idFilter")
.or_else(|| body.get("id_filter"))
.and_then(|v| v.as_str().or_else(|| v["value"].as_str()));
let mut artifacts = Vec::new();
for row in rows {
if row
.expires_at
.is_some_and(|expiry| expiry <= OffsetDateTime::now_utc())
|| (!name_filter.is_empty() && row.name != name_filter)
{
continue;
}
let id = self.numeric_actions_id(row.id).await?;
if id_filter.is_some_and(|filter| filter != id.to_string()) {
continue;
}
let job = workflow_job::Entity::find_by_id(row.job_id)
.one(self.db_write())
.await?
.ok_or(CiError::NotFound)?;
artifacts.push(json!({"workflowRunBackendId":context.run.actions_id.unwrap_or(0).to_string(),"workflowJobRunBackendId":job.actions_id.unwrap_or(0).to_string(),"databaseId":id.to_string(),"name":row.name,"size":row.size.to_string(),"createdAt":row.created_at.format(&time::format_description::well_known::Rfc3339).unwrap_or_default(),"hash":row.sha256.map(|hash|format!("sha256:{hash}")),"mimeType":row.mime_type}));
}
Ok(json!({"artifacts":artifacts}))
}
"GetSignedArtifactURL" | "DeleteArtifact" => {
let row = workflow_artifact::Entity::find()
.filter(workflow_artifact::Column::RunId.eq(context.run.id))
.filter(workflow_artifact::Column::Name.eq(&name))
.one(self.db_write())
.await?
.ok_or(CiError::NotFound)?;
if method == "DeleteArtifact" {
let id = self.numeric_actions_id(row.id).await?;
self.storage().client().delete(&row.object_key).await?;
workflow_artifact::Entity::delete_by_id(row.id)
.exec(self.db_write())
.await?;
Ok(json!({"ok":true,"artifactId":id.to_string()}))
} else {
if row.state != ArtifactState::Ready
|| row
.expires_at
.is_some_and(|at| at <= OffsetDateTime::now_utc())
{
return Err(CiError::NotFound);
}
Ok(
json!({"signedUrl":self.storage().client().presign_get(&row.object_key,900)?}),
)
}
}
_ => Err(CiError::InvalidArgument(format!(
"unknown artifact method `{method}`"
))),
}
}
async fn uploaded_size_and_hash(&self, key: &str, max: u64) -> CiResult<(u64, String)> {
let mut stream = self.storage().client().get_stream(key).await?;
let mut hash = Sha256::new();
let mut size = 0;
while let Some(chunk) = stream.next().await {
let chunk = chunk?;
size += chunk.len() as u64;
if size > max {
return Err(CiError::InvalidArgument(
"upload exceeds configured size limit".into(),
));
}
hash.update(&chunk);
}
Ok((
size,
hash.finalize()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect(),
))
}
async fn cache_reserve(
&self,
context: &RuntimeContext,
body: &Value,
) -> CiResult<actions_cache::Model> {
if context.run.event_payload["pull_request"]["head"]["repo"]["fork"]
.as_bool()
.unwrap_or(false)
{
return Err(CiError::Forbidden);
}
let key = field(body, "key", "key");
let version = field(body, "version", "version");
if key.is_empty() || key.len() > 512 || version.is_empty() || version.len() > 512 {
return Err(CiError::InvalidArgument(
"cache key and version must be 1-512 characters".into(),
));
}
let txn = self.db_write().begin().await?;
txn.execute_unprepared(&format!(
"SELECT pg_advisory_xact_lock({})",
context.job.repo_id.as_u128() as i64
))
.await?;
if let Some(row) = actions_cache::Entity::find()
.filter(actions_cache::Column::RepoId.eq(context.job.repo_id))
.filter(actions_cache::Column::GitRef.eq(&context.run.git_ref))
.filter(actions_cache::Column::Key.eq(&key))
.filter(actions_cache::Column::Version.eq(&version))
.one(&txn)
.await?
{
if row.job_id == context.job.id && !row.ready {
return Ok(row);
}
return Err(CiError::Conflict("cache entry already exists".into()));
}
let now = OffsetDateTime::now_utc();
let object_key = format!("ci/caches/{}/{}", context.job.repo_id, Uuid::now_v7());
let row = actions_cache::ActiveModel {
repo_id: Set(context.job.repo_id),
job_id: Set(context.job.id),
git_ref: Set(context.run.git_ref.clone()),
key: Set(key),
version: Set(version),
object_key: Set(object_key),
size: Set(0),
ready: Set(false),
created_at: Set(now),
last_accessed_at: Set(now),
expires_at: Set(
now + time::Duration::days(self.config().ci.cache_retention_days.max(1))
),
..Default::default()
}
.insert(&txn)
.await?;
txn.commit().await?;
Ok(row)
}
async fn cache_lookup(
&self,
context: &RuntimeContext,
key: &str,
restore_keys: &[String],
version: &str,
) -> CiResult<Option<actions_cache::Model>> {
let repo = models::repo::repos::Entity::find_by_id(context.job.repo_id)
.one(self.db_write())
.await?
.ok_or(CiError::NotFound)?;
let mut refs = vec![context.run.git_ref.clone()];
if let Some(base) = &context.run.base_ref {
refs.push(format!("refs/heads/{base}"))
}
refs.push(format!("refs/heads/{}", repo.default_branch));
refs.dedup();
let rows = actions_cache::Entity::find()
.filter(actions_cache::Column::RepoId.eq(repo.id))
.filter(actions_cache::Column::Version.eq(version))
.filter(actions_cache::Column::Ready.eq(true))
.filter(actions_cache::Column::ExpiresAt.gt(OffsetDateTime::now_utc()))
.order_by_desc(actions_cache::Column::CreatedAt)
.all(self.db_write())
.await?;
for reference in refs {
let entries: Vec<_> = rows.iter().filter(|row| row.git_ref == reference).collect();
let matched = entries
.iter()
.find(|row| row.key == key)
.copied()
.or_else(|| entries.iter().find(|row| row.key.starts_with(key)).copied())
.or_else(|| {
restore_keys.iter().find_map(|prefix| {
entries
.iter()
.find(|row| row.key.starts_with(prefix))
.copied()
})
});
if let Some(row) = matched {
let mut active: actions_cache::ActiveModel = row.clone().into();
active.last_accessed_at = Set(OffsetDateTime::now_utc());
active.expires_at = Set(OffsetDateTime::now_utc()
+ time::Duration::days(self.config().ci.cache_retention_days.max(1)));
return Ok(Some(active.update(self.db_write()).await?));
}
}
Ok(None)
}
async fn cache_finalize(
&self,
context: &RuntimeContext,
id: Option<i64>,
body: &Value,
) -> CiResult<actions_cache::Model> {
let mut query = actions_cache::Entity::find()
.filter(actions_cache::Column::RepoId.eq(context.job.repo_id))
.filter(actions_cache::Column::JobId.eq(context.job.id));
if let Some(id) = id {
query = query.filter(actions_cache::Column::Id.eq(id))
} else {
query = query
.filter(actions_cache::Column::Key.eq(field(body, "key", "key")))
.filter(actions_cache::Column::Version.eq(field(body, "version", "version")))
}
let row = query.one(self.db_write()).await?.ok_or(CiError::NotFound)?;
let declared = number(
body.get("sizeBytes")
.or_else(|| body.get("size_bytes"))
.or_else(|| body.get("size")),
)?;
let (size, _) = self
.uploaded_size_and_hash(&row.object_key, self.config().ci.cache_max_bytes)
.await?;
if size != declared {
return Err(CiError::InvalidArgument(
"cache size does not match upload".into(),
));
}
let mut active: actions_cache::ActiveModel = row.into();
active.size = Set(size as i64);
active.ready = Set(true);
active.last_accessed_at = Set(OffsetDateTime::now_utc());
Ok(active.update(self.db_write()).await?)
}
async fn assemble_blocks(&self, key: &str, blocks: &[String], max: u64) -> CiResult<()> {
let temporary = std::env::temp_dir().join(format!("gitdata-ci-upload-{}", Uuid::now_v7()));
let result: CiResult<()> = async {
let mut file = tokio::fs::OpenOptions::new()
.create_new(true)
.write(true)
.read(true)
.open(&temporary)
.await
.map_err(|error| CiError::Internal(error.to_string()))?;
let mut size = 0u64;
for block in blocks {
let data = self
.storage()
.client()
.get(&format!(
"{key}.blocks/{}",
crate::secrets::sha256_hex(block.as_bytes())
))
.await?;
size += data.len() as u64;
if size > max {
return Err(CiError::InvalidArgument(
"upload exceeds configured size limit".into(),
));
}
file.write_all(&data)
.await
.map_err(|error| CiError::Internal(error.to_string()))?;
}
file.seek(std::io::SeekFrom::Start(0))
.await
.map_err(|error| CiError::Internal(error.to_string()))?;
self.storage()
.client()
.put_stream(key, &mut file, "application/octet-stream")
.await?;
for block in blocks {
self.storage()
.client()
.delete(&format!(
"{key}.blocks/{}",
crate::secrets::sha256_hex(block.as_bytes())
))
.await?;
}
Ok(())
}
.await;
let _ = tokio::fs::remove_file(temporary).await;
result
}
}
fn field(value: &Value, camel: &str, snake: &str) -> String {
value
.get(camel)
.or_else(|| value.get(snake))
.and_then(|v| v.as_str().or_else(|| v["value"].as_str()))
.unwrap_or_default()
.to_owned()
}
fn number(value: Option<&Value>) -> CiResult<u64> {
value
.and_then(|value| {
value
.as_u64()
.or_else(|| value.as_str().and_then(|text| text.parse().ok()))
})
.ok_or_else(|| CiError::InvalidArgument("expected a nonnegative size".into()))
}
async fn artifact_rpc(
State(service): State<CiService>,
Path(method): Path<String>,
headers: HeaderMap,
Json(body): Json<Value>,
) -> Result<Json<Value>, RuntimeError> {
let context = service.runtime_context(&headers).await?;
Ok(Json(service.artifact_call(&context, &method, &body).await?))
}
#[derive(Deserialize)]
struct OidcQuery {
audience: Option<String>,
}
async fn oidc_token(
State(service): State<CiService>,
headers: HeaderMap,
Query(query): Query<OidcQuery>,
) -> Result<Json<Value>, RuntimeError> {
let context = service.runtime_context(&headers).await?;
if context
.job
.actions_context
.as_ref()
.and_then(|context| context["permissions"]["id-token"].as_str())
!= Some("write")
{
return Err(CiError::Forbidden.into());
}
if service.config().ci.oidc_private_key.is_empty() {
return Err(CiError::InvalidArgument(
"ci.oidc_private_key is required to issue OIDC tokens".into(),
)
.into());
}
let github = service.actions_github_context(&context.run).await?;
let repository = github["repository"].as_str().unwrap_or_default();
let sub = if let Some(environment) = &context.job.environment {
format!(
"repo:{repository}:environment:{}",
environment.replace(':', "%3A")
)
} else if context.run.pr_number.is_some() {
format!("repo:{repository}:pull_request")
} else {
format!("repo:{repository}:ref:{}", context.run.git_ref)
};
let now = OffsetDateTime::now_utc().unix_timestamp();
let issuer = service.actions_runtime_url()?;
let mut claims = github;
claims["iss"] = json!(issuer);
claims["sub"] = json!(sub);
claims["aud"] = json!(query.audience.unwrap_or_else(|| format!(
"{}/{}",
issuer,
claims["repository_owner"].as_str().unwrap_or_default()
)));
claims["iat"] = json!(now);
claims["nbf"] = json!(now - 5);
claims["exp"] = json!(now + 300);
claims["jti"] = json!(Uuid::now_v7());
claims.as_object_mut().map(|value| value.remove("token"));
let value = service
.actions_engine(
json!({"op":"oidc_sign","token":service.config().ci.oidc_private_key,"value":claims}),
)
.await?;
Ok(Json(json!({"value":value})))
}
async fn oidc_configuration(State(service): State<CiService>) -> Result<Json<Value>, RuntimeError> {
let issuer = service.actions_runtime_url()?;
Ok(Json(
json!({"issuer":issuer,"jwks_uri":format!("{issuer}/.well-known/jwks.json"),"id_token_signing_alg_values_supported":["RS256"],"subject_types_supported":["public"],"response_types_supported":["id_token"]}),
))
}
async fn oidc_jwks(State(service): State<CiService>) -> Result<Json<Value>, RuntimeError> {
if service.config().ci.oidc_private_key.is_empty() {
return Ok(Json(json!({"keys":[]})));
}
Ok(Json(
service
.actions_engine(json!({"op":"oidc_jwks","token":service.config().ci.oidc_private_key}))
.await?,
))
}
async fn cache_rpc(
State(service): State<CiService>,
Path(method): Path<String>,
headers: HeaderMap,
Json(body): Json<Value>,
) -> Result<Json<Value>, RuntimeError> {
let context = service.runtime_context(&headers).await?;
let result = match method.as_str() {
"CreateCacheEntry" => match service.cache_reserve(&context, &body).await {
Ok(row) => {
json!({"ok":true,"signedUploadUrl":service.signed_upload_url(&context,"cache",&row.id.to_string())?,"message":""})
}
Err(CiError::Conflict(message)) => {
json!({"ok":false,"signedUploadUrl":"","message":message})
}
Err(error) => return Err(error.into()),
},
"FinalizeCacheEntryUpload" => {
let row = service.cache_finalize(&context, None, &body).await?;
json!({"ok":true,"entryId":row.id.to_string(),"message":""})
}
"GetCacheEntryDownloadURL" => {
let restore = body
.get("restoreKeys")
.or_else(|| body.get("restore_keys"))
.map(|value| ci_protocol::actions::strings(Some(value)))
.unwrap_or_default();
match service
.cache_lookup(
&context,
&field(&body, "key", "key"),
&restore,
&field(&body, "version", "version"),
)
.await?
{
Some(row) => {
json!({"ok":true,"signedDownloadUrl":service.storage().client().presign_get(&row.object_key,900)? ,"matchedKey":row.key})
}
None => json!({"ok":false,"signedDownloadUrl":"","matchedKey":""}),
}
}
_ => {
return Err(
CiError::InvalidArgument(format!("unknown cache method `{method}`")).into(),
);
}
};
Ok(Json(result))
}
#[derive(Deserialize)]
struct UploadQuery {
token: String,
comp: Option<String>,
blockid: Option<String>,
}
async fn upload_blob(
State(service): State<CiService>,
Path((kind, id)): Path<(String, String)>,
Query(query): Query<UploadQuery>,
body: Bytes,
) -> Result<Response, RuntimeError> {
let claims = service.verify_actions_token(&query.token)?;
if claims["aud"] != "gitdata-actions-upload" || claims["kind"] != kind || claims["id"] != id {
return Err(CiError::Forbidden.into());
}
let context = service.runtime_context_from_claims(&claims).await?;
let (key, max) = if kind == "artifact" {
let id = Uuid::parse_str(&id).map_err(|_| CiError::NotFound)?;
let row = workflow_artifact::Entity::find_by_id(id)
.one(service.db_write())
.await?
.ok_or(CiError::NotFound)?;
if row.job_id != context.job.id || row.state != ArtifactState::Uploading {
return Err(CiError::Forbidden.into());
}
(row.object_key, service.config().ci.artifact_max_bytes)
} else if kind == "cache" {
let id = id.parse::<i64>().map_err(|_| CiError::NotFound)?;
let row = actions_cache::Entity::find_by_id(id)
.one(service.db_write())
.await?
.ok_or(CiError::NotFound)?;
if row.job_id != context.job.id || row.ready {
return Err(CiError::Forbidden.into());
}
(row.object_key, service.config().ci.cache_max_bytes)
} else {
return Err(CiError::NotFound.into());
};
match query.comp.as_deref() {
Some("block") => {
let block = query
.blockid
.ok_or_else(|| CiError::InvalidArgument("blockid is required".into()))?;
if block.len() > 256 {
return Err(CiError::InvalidArgument("blockid is too long".into()).into());
}
service
.storage()
.client()
.put(
&format!(
"{key}.blocks/{}",
crate::secrets::sha256_hex(block.as_bytes())
),
&body,
)
.await
.map_err(CiError::from)?;
}
Some("blocklist") => {
let text = std::str::from_utf8(&body)
.map_err(|_| CiError::InvalidArgument("blocklist is not UTF-8".into()))?;
let mut blocks = Vec::new();
for part in text.split('<') {
if let Some(value) = part
.strip_prefix("Latest>")
.or_else(|| part.strip_prefix("Uncommitted>"))
.or_else(|| part.strip_prefix("Committed>"))
{
blocks.push(value.split('<').next().unwrap_or_default().to_owned());
}
}
if blocks.len() > 50000 {
return Err(CiError::InvalidArgument("too many upload blocks".into()).into());
}
service.assemble_blocks(&key, &blocks, max).await?;
}
None => {
if body.len() as u64 > max {
return Err(CiError::InvalidArgument(
"upload exceeds configured size limit".into(),
)
.into());
}
service
.storage()
.client()
.put(&key, &body)
.await
.map_err(CiError::from)?;
}
_ => return Err(CiError::InvalidArgument("unsupported upload operation".into()).into()),
}
Ok((
StatusCode::CREATED,
[
("x-ms-request-id", Uuid::now_v7().to_string()),
("x-ms-version", "2023-11-03".into()),
("etag", format!("\"{}\"", Uuid::now_v7())),
],
)
.into_response())
}
#[derive(Deserialize)]
struct LegacyLookup {
keys: String,
version: String,
}
async fn legacy_cache_lookup(
State(service): State<CiService>,
headers: HeaderMap,
Query(query): Query<LegacyLookup>,
) -> Result<Response, RuntimeError> {
let context = service.runtime_context(&headers).await?;
let keys: Vec<String> = query.keys.split(',').map(str::to_owned).collect();
match service.cache_lookup(&context,keys.first().map(String::as_str).unwrap_or_default(),keys.get(1..).unwrap_or_default(),&query.version).await?{
Some(row)=>Ok(Json(json!({"cacheKey":row.key,"scope":row.git_ref,"creationTime":row.created_at.format(&time::format_description::well_known::Rfc3339).unwrap_or_default(),"archiveLocation":service.storage().client().presign_get(&row.object_key,900)?})).into_response()),
None=>Ok(StatusCode::NO_CONTENT.into_response()),
}
}
async fn legacy_cache_reserve(
State(service): State<CiService>,
headers: HeaderMap,
Json(body): Json<Value>,
) -> Result<Json<Value>, RuntimeError> {
let context = service.runtime_context(&headers).await?;
let row = service.cache_reserve(&context, &body).await?;
Ok(Json(json!({"cacheId":row.id})))
}
async fn legacy_cache_upload(
State(service): State<CiService>,
Path(id): Path<i64>,
headers: HeaderMap,
body: Bytes,
) -> Result<StatusCode, RuntimeError> {
let context = service.runtime_context(&headers).await?;
let row = actions_cache::Entity::find_by_id(id)
.one(service.db_write())
.await
.map_err(CiError::from)?
.ok_or(CiError::NotFound)?;
if row.job_id != context.job.id || row.ready {
return Err(CiError::Forbidden.into());
}
let range = headers
.get("content-range")
.and_then(|value| value.to_str().ok())
.ok_or_else(|| CiError::InvalidArgument("content-range is required".into()))?;
let span = range
.strip_prefix("bytes ")
.and_then(|value| value.split('/').next())
.and_then(|value| value.split_once('-'))
.ok_or_else(|| CiError::InvalidArgument("invalid content-range".into()))?;
let start = span
.0
.parse::<u64>()
.map_err(|_| CiError::InvalidArgument("invalid content-range".into()))?;
let end = span
.1
.parse::<u64>()
.map_err(|_| CiError::InvalidArgument("invalid content-range".into()))?;
if end < start
|| end - start + 1 != body.len() as u64
|| end >= service.config().ci.cache_max_bytes
{
return Err(CiError::InvalidArgument("cache upload range is invalid".into()).into());
}
service
.storage()
.client()
.put(&format!("{}.ranges/{start:020}", row.object_key), &body)
.await
.map_err(CiError::from)?;
Ok(StatusCode::NO_CONTENT)
}
async fn legacy_cache_finalize(
State(service): State<CiService>,
Path(id): Path<i64>,
headers: HeaderMap,
Json(body): Json<Value>,
) -> Result<StatusCode, RuntimeError> {
let context = service.runtime_context(&headers).await?;
let row = actions_cache::Entity::find_by_id(id)
.one(service.db_write())
.await
.map_err(CiError::from)?
.ok_or(CiError::NotFound)?;
if row.job_id != context.job.id {
return Err(CiError::Forbidden.into());
}
let mut ranges = service
.storage()
.client()
.list(&format!("{}.ranges/", row.object_key))
.await
.map_err(CiError::from)?;
ranges.sort_by(|a, b| a.key.cmp(&b.key));
let temporary = std::env::temp_dir().join(format!("gitdata-cache-upload-{}", Uuid::now_v7()));
let assembled: CiResult<()> = async {
let mut file = tokio::fs::OpenOptions::new()
.create_new(true)
.write(true)
.read(true)
.open(&temporary)
.await
.map_err(|error| CiError::Internal(error.to_string()))?;
let mut expected = 0;
for range in &ranges {
let offset = range
.key
.rsplit('/')
.next()
.and_then(|value| value.parse::<u64>().ok())
.ok_or_else(|| CiError::InvalidArgument("invalid cache segment".into()))?;
if offset != expected {
return Err(CiError::InvalidArgument(
"cache segments overlap or have gaps".into(),
));
}
let data = service.storage().client().get(&range.key).await?;
expected += data.len() as u64;
file.write_all(&data)
.await
.map_err(|error| CiError::Internal(error.to_string()))?;
}
file.seek(std::io::SeekFrom::Start(0))
.await
.map_err(|error| CiError::Internal(error.to_string()))?;
service
.storage()
.client()
.put_stream(&row.object_key, &mut file, "application/octet-stream")
.await?;
Ok(())
}
.await;
let _ = tokio::fs::remove_file(&temporary).await;
assembled?;
service.cache_finalize(&context, Some(id), &body).await?;
for range in ranges {
service
.storage()
.client()
.delete(&range.key)
.await
.map_err(CiError::from)?;
}
Ok(StatusCode::NO_CONTENT)
}