d3d11f4
use hmac::{Hmac, KeyInit, Mac};
use models::webhook::repo_webhook_delivery::{self, RepoWebhookDeliveryStatus};
use models::webhook::repo_webhooks;
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set,
};
use sha2::Sha256;
use time::OffsetDateTime;
use uuid::Uuid;
use crate::errors::CiResult;
use crate::secrets::decrypt;
use crate::service::CiService;
const MAX_RESPONSE_BODY: usize = 2048;
const CLAIM_STUCK_MINUTES: i64 = 10;
pub async fn enqueue_for_repo(
service: &CiService,
repo_id: Uuid,
event_type: &str,
payload: &serde_json::Value,
) -> CiResult<()> {
let hooks = repo_webhooks::Entity::find()
.filter(repo_webhooks::Column::RepoId.eq(repo_id))
.filter(repo_webhooks::Column::Status.eq(models::webhook::webhooks::WebhookStatus::Active))
.all(service.db_read())
.await?;
if hooks.is_empty() {
return Ok(());
}
let now = OffsetDateTime::now_utc();
for hook in hooks {
let events: Vec<String> = serde_json::from_value(hook.events.clone()).unwrap_or_default();
if !events.iter().any(|e| e == event_type || e == "*") {
continue;
}
repo_webhook_delivery::ActiveModel {
id: Set(Uuid::now_v7()),
webhook_id: Set(hook.id),
repo_id: Set(repo_id),
event_type: Set(event_type.to_string()),
payload: Set(payload.clone()),
status: Set(RepoWebhookDeliveryStatus::Pending),
attempts: Set(0),
next_retry_at: Set(now),
response_status: Set(None),
response_body: Set(None),
last_error: Set(None),
created_at: Set(now),
updated_at: Set(now),
}
.insert(service.db_write())
.await?;
}
Ok(())
}
impl CiService {
pub async fn deliver_pending_repo_webhooks(&self) -> CiResult<u64> {
let now = OffsetDateTime::now_utc();
let stuck = now - time::Duration::minutes(CLAIM_STUCK_MINUTES);
repo_webhook_delivery::Entity::update_many()
.col_expr(
repo_webhook_delivery::Column::Status,
Expr::value(RepoWebhookDeliveryStatus::Pending),
)
.col_expr(repo_webhook_delivery::Column::UpdatedAt, Expr::value(now))
.filter(repo_webhook_delivery::Column::Status.eq(RepoWebhookDeliveryStatus::Delivering))
.filter(repo_webhook_delivery::Column::UpdatedAt.lt(stuck))
.exec(self.db_write())
.await?;
let due = repo_webhook_delivery::Entity::find()
.filter(repo_webhook_delivery::Column::Status.eq(RepoWebhookDeliveryStatus::Pending))
.filter(repo_webhook_delivery::Column::NextRetryAt.lte(now))
.order_by_asc(repo_webhook_delivery::Column::NextRetryAt)
.limit(50)
.all(self.db_read())
.await?;
let count = due.len() as u64;
let max_attempts = self.config().ci.outbound_max_attempts;
let backoff_base = self.config().ci.outbound_backoff_base_secs;
for delivery in due {
let claimed = repo_webhook_delivery::Entity::update_many()
.col_expr(
repo_webhook_delivery::Column::Status,
Expr::value(RepoWebhookDeliveryStatus::Delivering),
)
.col_expr(repo_webhook_delivery::Column::UpdatedAt, Expr::value(now))
.filter(repo_webhook_delivery::Column::Id.eq(delivery.id))
.filter(
repo_webhook_delivery::Column::Status.eq(RepoWebhookDeliveryStatus::Pending),
)
.exec(self.db_write())
.await?;
if claimed.rows_affected == 0 {
continue;
}
let hook = repo_webhooks::Entity::find_by_id(delivery.webhook_id)
.one(self.db_read())
.await?;
let Some(hook) = hook else {
self.finish_repo_delivery(
delivery.id,
RepoWebhookDeliveryStatus::Failed,
delivery.attempts,
None,
None,
Some("webhook deleted".to_string()),
)
.await?;
continue;
};
let secret = decrypt(self.encryption_key(), &hook.secret)
.unwrap_or_else(|_| hook.secret.clone());
let result = self.post_repo_payload(&hook.url, &secret, &delivery).await;
match result {
Ok((status, body)) => {
self.finish_repo_delivery(
delivery.id,
RepoWebhookDeliveryStatus::Done,
delivery.attempts + 1,
Some(status),
Some(body),
None,
)
.await?;
}
Err(error) => {
let attempts = delivery.attempts + 1;
if attempts >= max_attempts {
self.finish_repo_delivery(
delivery.id,
RepoWebhookDeliveryStatus::Failed,
attempts,
None,
None,
Some(error),
)
.await?;
} else {
let next = OffsetDateTime::now_utc()
+ time::Duration::seconds(backoff_base * i64::from(attempts));
let mut am: repo_webhook_delivery::ActiveModel = delivery.clone().into();
am.status = Set(RepoWebhookDeliveryStatus::Pending);
am.attempts = Set(attempts);
am.next_retry_at = Set(next);
am.last_error = Set(Some(error));
am.updated_at = Set(OffsetDateTime::now_utc());
am.update(self.db_write()).await?;
}
}
}
}
Ok(count)
}
async fn finish_repo_delivery(
&self,
id: Uuid,
status: RepoWebhookDeliveryStatus,
attempts: i32,
response_status: Option<i32>,
response_body: Option<String>,
last_error: Option<String>,
) -> CiResult<()> {
let am = repo_webhook_delivery::ActiveModel {
id: Set(id),
status: Set(status),
attempts: Set(attempts),
response_status: Set(response_status),
response_body: Set(response_body),
last_error: Set(last_error),
updated_at: Set(OffsetDateTime::now_utc()),
..Default::default()
};
am.update(self.db_write()).await?;
Ok(())
}
async fn post_repo_payload(
&self,
url: &str,
secret: &str,
delivery: &repo_webhook_delivery::Model,
) -> Result<(i32, String), String> {
let uri: reqwest::Url = url.parse().map_err(|e| format!("invalid url: {e}"))?;
if uri.scheme() != "https" && uri.scheme() != "http" {
return Err("webhook url must be http(s)".to_string());
}
let host = uri.host_str().unwrap_or("").to_string();
if host.is_empty() {
return Err("webhook url has no host".to_string());
}
let port = uri
.port()
.unwrap_or(if uri.scheme() == "https" { 443 } else { 80 });
let addrs = tokio::net::lookup_host((host.as_str(), port))
.await
.map_err(|e| format!("dns resolution failed: {e}"))?
.collect::<Vec<_>>();
if addrs.is_empty() {
return Err("dns resolution returned no address".to_string());
}
if addrs
.iter()
.any(|addr| !crate::net::is_public_ip(&addr.ip()))
{
return Err("url resolves to a private or loopback address".to_string());
}
let body = serde_json::to_vec(&delivery.payload).map_err(|e| e.to_string())?;
let mut mac =
Hmac::<Sha256>::new_from_slice(secret.as_bytes()).map_err(|e| e.to_string())?;
mac.update(&body);
let signature = format!(
"sha256={}",
crate::secrets::sha256_hex(&mac.finalize().into_bytes())
);
let mut builder = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(
self.config().ci.outbound_timeout_secs.max(1),
))
.redirect(reqwest::redirect::Policy::none());
for addr in &addrs {
builder = builder.resolve(&host, *addr);
}
let client = builder.build().map_err(|e| e.to_string())?;
let span = export::telemetry::http_client_span("POST", url);
let response = export::telemetry::send(
span,
client
.post(url)
.header("content-type", "application/json")
.header("x-gitdata-event", &delivery.event_type)
.header("x-gitdata-delivery", delivery.id.to_string())
.header("x-gitdata-signature", signature)
.body(body),
)
.await
.map_err(|e| {
export::metrics::ci_webhook_delivery("error");
format!("request failed: {e}")
})?;
let status = response.status().as_u16() as i32;
let text = response
.text()
.await
.unwrap_or_default()
.chars()
.take(MAX_RESPONSE_BODY)
.collect::<String>();
if (200..300).contains(&status) {
export::metrics::ci_webhook_delivery("ok");
Ok((status, text))
} else {
export::metrics::ci_webhook_delivery("error");
Err(format!("endpoint returned {status}"))
}
}
}