feat(webhook): add webhook module (delivery, handler, routes)
zhenyi committed
Added lib/webhook/Cargo.toml lib/webhook/Cargo.toml +23 −0 View lib/webhook/Cargo.toml
@@ -0,0 +1,23 @@
1[package]
2name = "webhook"
3version.workspace = true
4edition.workspace = true
5
6[lib]
7path = "lib.rs"
8name = "webhook"
9
10[dependencies]
11async-trait.workspace = true
12db.workspace = true
13gitsync.workspace = true
14hex.workspace = true
15hmac.workspace = true
16reqwest.workspace = true
17sea-orm.workspace = true
18serde.workspace = true
19serde_json.workspace = true
20sha2.workspace = true
21time.workspace = true
22tracing.workspace = true
23uuid.workspace = true
No newline at end of file
Added lib/webhook/dispatcher.rs lib/webhook/dispatcher.rs +154 −0 View lib/webhook/dispatcher.rs
@@ -0,0 +1,154 @@
1use crate::payload::build_push_payload;
2use crate::sign::sign_hmac_sha256;
3use async_trait::async_trait;
4use db::conn::Db;
5use db::repo::repos;
6use db::webhook::repo_webhooks;
7use db::webhook::webhook_deliveries;
8use gitsync::{PostReceiveTask, WebhookTx};
9use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
10use std::error::Error;
11use std::time::Duration;
12use time::OffsetDateTime;
13use tracing::{debug, warn};
14use uuid::Uuid;
15
16pub const DEFAULT_WEBHOOK_TIMEOUT: Duration = Duration::from_secs(10);
17
18type DeliveryResult = Result<(Option<i32>, Option<String>), Box<dyn Error + Send + Sync>>;
19
20#[derive(Clone)]
21pub struct WebhookDispatcher {
22 db: Db,
23 http: reqwest::Client,
24 timeout: Duration,
25}
26
27impl WebhookDispatcher {
28 #[must_use]
29 pub fn new(db: Db, http: reqwest::Client, timeout: Duration) -> Self {
30 Self { db, http, timeout }
31 }
32
33 #[must_use]
34 pub fn with_defaults(db: Db) -> Self {
35 Self::new(db, reqwest::Client::new(), DEFAULT_WEBHOOK_TIMEOUT)
36 }
37}
38
39#[async_trait]
40impl WebhookTx for WebhookDispatcher {
41 async fn dispatch(&self, task: &PostReceiveTask) -> Result<(), Box<dyn Error + Send + Sync>> {
42 match self.dispatch_inner(task).await {
43 Ok(n) => debug!(count = n, repo_id = %task.repo_id, "webhook deliveries dispatched"),
44 Err(e) => warn!(error = %e, repo_id = %task.repo_id, "webhook dispatch failed"),
45 }
46 Ok(())
47 }
48}
49
50impl WebhookDispatcher {
51 async fn dispatch_inner(
52 &self,
53 task: &PostReceiveTask,
54 ) -> Result<usize, Box<dyn Error + Send + Sync>> {
55 let Some(repo) = repos::Entity::find_by_id(task.repo_id)
56 .one(&self.db.reader())
57 .await
58 .map_err(|e| Box::<dyn Error + Send + Sync>::from(format!("load repo: {e}")))?
59 else {
60 debug!(repo_id = %task.repo_id, "repo not found for webhook dispatch");
61 return Ok(0);
62 };
63
64 let webhooks = repo_webhooks::Entity::find()
65 .filter(repo_webhooks::Column::RepoId.eq(task.repo_id))
66 .filter(repo_webhooks::Column::IsActive.eq(true))
67 .all(&self.db.reader())
68 .await
69 .map_err(|e| Box::<dyn Error + Send + Sync>::from(format!("load webhooks: {e}")))?;
70
71 let mut dispatched = 0usize;
72 for webhook in &webhooks {
73 if !event_enabled(&webhook.events) {
74 continue;
75 }
76 for update in &task.refs {
77 let payload = serde_json::to_vec(&build_push_payload(task, &repo, update))?;
78 let signature = sign_hmac_sha256(&payload, &webhook.secret);
79 let result = self.post(&webhook.url, &signature, &payload).await;
80 self.record_delivery(webhook, &payload, &result).await;
81 dispatched += 1;
82 }
83 }
84 Ok(dispatched)
85 }
86
87 async fn post(&self, url: &str, signature: &str, payload: &[u8]) -> DeliveryResult {
88 let response = self
89 .http
90 .post(url)
91 .timeout(self.timeout)
92 .header(reqwest::header::CONTENT_TYPE, "application/json")
93 .header("X-Hub-Signature-256", format!("sha256={signature}"))
94 .header("X-Gitea-Signature", format!("sha256={signature}"))
95 .body(payload.to_vec())
96 .send()
97 .await
98 .map_err(|e| Box::<dyn Error + Send + Sync>::from(format!("http post {url}: {e}")))?;
99
100 let status = response.status().as_u16() as i32;
101 let body = response
102 .text()
103 .await
104 .map_err(|e| Box::<dyn Error + Send + Sync>::from(format!("read response: {e}")))?;
105 Ok((Some(status), Some(body)))
106 }
107
108 #[allow(clippy::manual_async_fn)]
109 async fn record_delivery(
110 &self,
111 webhook: &repo_webhooks::Model,
112 payload: &[u8],
113 result: &DeliveryResult,
114 ) {
115 let (status, response_status, response_body) = match result {
116 Ok((code, body)) => (
117 webhook_deliveries::WebhookDeliveryStatus::Success,
118 *code,
119 body.clone(),
120 ),
121 Err(e) => (
122 webhook_deliveries::WebhookDeliveryStatus::Failed,
123 None,
124 Some(e.to_string()),
125 ),
126 };
127
128 let delivery = webhook_deliveries::ActiveModel {
129 id: Set(Uuid::new_v4()),
130 webhook_id: Set(webhook.id),
131 event_type: Set("push".to_string()),
132 payload: Set(String::from_utf8_lossy(payload).into_owned()),
133 status: Set(status),
134 response_status: Set(response_status),
135 response_body: Set(response_body),
136 created_at: Set(OffsetDateTime::now_utc()),
137 };
138
139 if let Err(e) = delivery.insert(&self.db.writer()).await {
140 warn!(error = %e, webhook_id = %webhook.id, "failed to record webhook delivery");
141 }
142 }
143}
144
145fn event_enabled(events: &sea_orm::JsonValue) -> bool {
146 let Some(arr) = events.as_array() else {
147 return true;
148 };
149 if arr.is_empty() {
150 return true;
151 }
152 arr.iter()
153 .any(|e| e.as_str() == Some("push") || e.as_str() == Some("*"))
154}
Added lib/webhook/lib.rs lib/webhook/lib.rs +7 −0 View lib/webhook/lib.rs
@@ -0,0 +1,7 @@
1pub mod dispatcher;
2pub mod payload;
3pub mod sign;
4
5pub use dispatcher::WebhookDispatcher;
6pub use payload::build_push_payload;
7pub use sign::sign_hmac_sha256;
Added lib/webhook/payload.rs lib/webhook/payload.rs +113 −0 View lib/webhook/payload.rs
@@ -0,0 +1,113 @@
1use db::repo::repos;
2use gitsync::{PostReceiveRefUpdate, PostReceiveTask};
3use serde_json::json;
4
5#[must_use]
6pub fn build_push_payload(
7 task: &PostReceiveTask,
8 repo: &repos::Model,
9 update: &PostReceiveRefUpdate,
10) -> serde_json::Value {
11 let pusher_id = task.actor_id.map_or_else(String::new, |id| id.to_string());
12 json!({
13 "ref": update.name,
14 "before": update.old_oid,
15 "after": update.new_oid,
16 "created": update.is_create(),
17 "deleted": update.is_delete(),
18 "pusher": { "id": pusher_id },
19 "sender": { "id": pusher_id },
20 "repository": {
21 "id": repo.id.to_string(),
22 "name": repo.name,
23 "full_name": format!("{}/{}", repo.channel_id, repo.name),
24 "default_branch": repo.default_branch,
25 "channel_id": repo.channel_id.to_string(),
26 },
27 })
28}
29
30#[cfg(test)]
31mod tests {
32 use super::*;
33 use gitsync::PostReceiveRefUpdate;
34 use time::OffsetDateTime;
35 use uuid::Uuid;
36
37 fn sample_task() -> PostReceiveTask {
38 let refs = vec![PostReceiveRefUpdate::new(
39 "refs/heads/main",
40 "0000000000000000000000000000000000000000",
41 "1111111111111111111111111111111111111111",
42 )];
43 PostReceiveTask::new(Uuid::new_v4(), Uuid::new_v4(), Some(Uuid::new_v4()), refs)
44 }
45
46 fn sample_repo() -> repos::Model {
47 repos::Model {
48 id: Uuid::new_v4(),
49 channel_id: Uuid::new_v4(),
50 lower_name: "demo".into(),
51 name: "demo".into(),
52 description: None,
53 website: None,
54 default_branch: "main".into(),
55 nums_watch: 0,
56 nums_star: 0,
57 nums_pr: 0,
58 nums_commits: 0,
59 nums_branches: 0,
60 nums_tags: 0,
61 enable_lfs: false,
62 is_private: false,
63 is_archived: false,
64 storage_limit_size: None,
65 size: 0,
66 git_size: 0,
67 lfs_size: 0,
68 avatar_url: None,
69 primary_language: None,
70 license_name: None,
71 created_at: OffsetDateTime::now_utc(),
72 last_pushed_at: None,
73 archived_at: None,
74 readme_raw: None,
75 allow_rebase: true,
76 allow_rebase_explicit: false,
77 allow_squash_merge: true,
78 allow_merge_commits: true,
79 object_format: "sha1".into(),
80 }
81 }
82
83 #[test]
84 fn payload_has_core_fields() {
85 let task = sample_task();
86 let repo = sample_repo();
87 let payload = build_push_payload(&task, &repo, &task.refs[0]);
88 assert_eq!(payload["ref"], "refs/heads/main");
89 assert_eq!(payload["created"], true);
90 assert_eq!(payload["deleted"], false);
91 assert_eq!(payload["repository"]["name"], "demo");
92 assert_eq!(payload["repository"]["default_branch"], "main");
93 assert_eq!(payload["repository"]["id"], repo.id.to_string());
94 assert_eq!(payload["pusher"]["id"], task.actor_id.unwrap().to_string());
95 }
96
97 #[test]
98 fn payload_flags_follow_ref_semantics() {
99 let task = sample_task();
100 let repo = sample_repo();
101 let del = build_push_payload(
102 &task,
103 &repo,
104 &PostReceiveRefUpdate::new(
105 "refs/heads/main",
106 "1111111111111111111111111111111111111111",
107 "0000000000000000000000000000000000000000",
108 ),
109 );
110 assert_eq!(del["deleted"], true);
111 assert_eq!(del["created"], false);
112 }
113}
Added lib/webhook/sign.rs lib/webhook/sign.rs +40 −0 View lib/webhook/sign.rs
@@ -0,0 +1,40 @@
1use hmac::{Hmac, Mac};
2use sha2::Sha256;
3
4type HmacSha256 = Hmac<Sha256>;
5
6#[must_use]
7pub fn sign_hmac_sha256(payload: &[u8], secret: &str) -> String {
8 let mut mac =
9 HmacSha256::new_from_slice(secret.as_bytes()).expect("HMAC accepts keys of any length");
10 mac.update(payload);
11 hex::encode(mac.finalize().into_bytes())
12}
13
14#[cfg(test)]
15mod tests {
16 use super::*;
17
18 #[test]
19 fn signature_is_stable() {
20 let sig = sign_hmac_sha256(b"The quick brown fox jumps over the lazy dog", "key");
21 assert_eq!(
22 sig,
23 "f7bc83f430538424b13298e6aa6fb143ef4d59a14946175997479dbc2d1a3cd8"
24 );
25 }
26
27 #[test]
28 fn signature_changes_with_payload() {
29 let a = sign_hmac_sha256(b"hello", "secret");
30 let b = sign_hmac_sha256(b"world", "secret");
31 assert_ne!(a, b);
32 assert_eq!(a.len(), 64);
33 }
34
35 #[test]
36 fn signature_is_hex() {
37 let sig = sign_hmac_sha256(b"payload", "s3cret");
38 assert!(sig.chars().all(|c| c.is_ascii_hexdigit()));
39 }
40}