80f9e01
use models::chat::outbound_webhooks;
use models::chat::webhook_deliveries::{self, WebhookDeliveryStatus};
use sea_orm::sea_query::Expr;
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;
use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::authz::can_manage_space;
use crate::share::lookup::require_room_readable;
const MAX_DELIVERY_ATTEMPTS: i32 = 5;
/// 多数站点（知乎、微博、CSDN）会对默认爬虫 UA 直接 403，浏览器 UA 才有正常 HTML。
const USER_AGENT: &str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) \
Chrome/131.0.0.0 Safari/537.36";
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct OutboundWebhookDto {
pub id: Uuid,
pub room_id: Uuid,
pub name: String,
pub url: String,
pub disabled: bool,
pub last_status: Option<String>,
pub last_used_at: Option<OffsetDateTime>,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct OutboundWebhookSecretDto {
pub id: Uuid,
pub room_id: Uuid,
pub name: String,
pub url: String,
pub secret: String,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateOutboundWebhookCmd {
pub name: String,
pub url: String,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdateOutboundWebhookCmd {
#[serde(default)]
pub name: Option<String>,
#[serde(default)]
pub url: Option<String>,
#[serde(default)]
pub disabled: Option<bool>,
}
impl From<outbound_webhooks::Model> for OutboundWebhookDto {
fn from(w: outbound_webhooks::Model) -> Self {
Self {
id: w.id,
room_id: w.room_id,
name: w.name,
url: w.url,
disabled: w.disabled,
last_status: w.last_status,
last_used_at: w.last_used_at,
created_at: w.created_at,
}
}
}
pub(crate) fn validate_outbound_url(url: &str) -> Result<http::Uri, ChatError> {
let uri: http::Uri = url
.parse()
.map_err(|_| ChatError::BadRequest("invalid url".to_string()))?;
let scheme = uri.scheme_str().unwrap_or("");
if scheme != "http" && scheme != "https" {
return Err(ChatError::BadRequest(
"url scheme must be http or https".to_string(),
));
}
let host = uri
.host()
.ok_or_else(|| ChatError::BadRequest("url host is required".to_string()))?;
if host.is_empty() {
return Err(ChatError::BadRequest("url host is required".to_string()));
}
if let Ok(ip) = host.parse::<std::net::IpAddr>()
&& !is_public_ip(&ip)
{
return Err(ChatError::BadRequest(
"url must not point at a private or loopback address".to_string(),
));
}
Ok(uri)
}
pub(crate) fn is_public_ip(ip: &std::net::IpAddr) -> bool {
match ip {
std::net::IpAddr::V4(v4) => {
!v4.is_loopback()
&& !v4.is_private()
&& !v4.is_link_local()
&& !v4.is_broadcast()
&& !v4.is_documentation()
&& !v4.is_unspecified()
&& !(v4.octets()[0] == 100 && (v4.octets()[1] & 0b1100_0000) == 0b0100_0000)
}
std::net::IpAddr::V6(v6) => {
!v6.is_loopback()
&& !v6.is_unspecified()
&& !v6.is_multicast()
&& {
let segments = v6.segments();
// 64:ff9b::/96 (NAT64), fc00::/7 (ULA), fe80::/10 (link-local),
// ::ffff:0:0/96 (IPv4-mapped) handled via to_ipv4 below.
!(segments[0] == 0x0064 && segments[1] == 0xff9b)
&& (segments[0] & 0xfe00) != 0xfc00
&& (segments[0] & 0xffc0) != 0xfe80
}
&& match v6.to_ipv4_mapped() {
Some(v4) => is_public_ip(&std::net::IpAddr::V4(v4)),
None => true,
}
}
}
}
/// 抓取结果：正文 + 最终落地的 URL。
///
/// 短链（`b23.tv` / `t.co` / `xhslink.com`）只有跟随重定向才能拿到真实内容，
/// 而落地地址又决定这条链接该按哪个平台渲染，所以两者一起返回。
pub(crate) struct Fetched {
pub body: String,
pub final_url: String,
}
/// 从 `content-type` 里摘 charset；缺失时按 `None`（交给调用方嗅探）。
fn charset_of(content_type: &str) -> Option<String> {
let lower = content_type.to_ascii_lowercase();
let start = lower.find("charset=")? + "charset=".len();
let rest = &content_type[start..];
let value = rest
.split(|c: char| c == ';' || c == ',' || c.is_whitespace())
.next()
.unwrap_or("")
.trim_matches(|c: char| c == '"' || c == '\'');
(!value.is_empty()).then(|| value.to_string())
}
/// 按声明编码解码；没声明时先看 UTF-8 能否直通，否则回退 GB18030（中文站常见）。
pub(crate) fn decode_body(body: &[u8], content_type: &str) -> String {
if let Some(label) = charset_of(content_type)
&& let Some(encoding) = encoding_rs::Encoding::for_label(label.as_bytes())
{
let (text, _, _) = encoding.decode(body);
return text.into_owned();
}
match std::str::from_utf8(body) {
Ok(text) => text.to_string(),
Err(_) => {
let (text, _, _) = encoding_rs::GB18030.decode(body);
text.into_owned()
}
}
}
/// SSRF 防护后的 GET：只允许公网地址、限时限量，跟随重定向并对每一跳重做校验。
///
/// `expect_content_type` 非空时，响应 `content-type` 必须包含该片段（读 body 之前检查）。
/// 重定向是手工跟的：`reqwest` 自带的跳转策略不会拦截落到内网地址的 `Location`，
/// 所以每一跳都要重新过一遍 [`validate_outbound_url`] 与 DNS 归属检查。
pub(crate) async fn fetch_following(
url: &str,
accept: &str,
expect_content_type: Option<&str>,
max_bytes: usize,
) -> Result<Fetched, String> {
fetch(url, accept, expect_content_type, max_bytes).await
}
const MAX_REDIRECTS: usize = 6;
async fn fetch(
url: &str,
accept: &str,
expect_content_type: Option<&str>,
max_bytes: usize,
) -> Result<Fetched, String> {
let mut current = url.to_string();
for _ in 0..=MAX_REDIRECTS {
let uri = validate_outbound_url(¤t).map_err(|e| e.to_string())?;
let host = uri.host().unwrap_or("").to_string();
let port = uri.port_u16();
let addrs = tokio::net::lookup_host((host.as_str(), port.unwrap_or(443)))
.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| !is_public_ip(&addr.ip())) {
return Err("url resolves to a private or loopback address".to_string());
}
let mut client_builder = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(10))
.redirect(reqwest::redirect::Policy::none())
.user_agent(USER_AGENT);
for addr in &addrs {
client_builder = client_builder.resolve(&host, *addr);
}
let client = client_builder.build().map_err(|e| e.to_string())?;
let response = client
.get(¤t)
.header("accept", accept)
.send()
.await
.map_err(|e| format!("request failed: {e}"))?;
if response.status().is_redirection() {
let location = response
.headers()
.get("location")
.and_then(|value| value.to_str().ok())
.ok_or_else(|| "redirect without a location header".to_string())?;
let base = url::Url::parse(¤t).map_err(|e| e.to_string())?;
current = base
.join(location)
.map_err(|e| format!("invalid redirect target: {e}"))?
.to_string();
continue;
}
if !response.status().is_success() {
return Err(format!("endpoint returned {}", response.status().as_u16()));
}
let content_type = response
.headers()
.get("content-type")
.and_then(|value| value.to_str().ok())
.unwrap_or("")
.to_string();
let lowered = content_type.to_ascii_lowercase();
if let Some(expected) = expect_content_type
&& !lowered.is_empty()
&& !lowered.contains(expected)
{
return Err(format!("unexpected content-type {lowered}"));
}
let mut body: Vec<u8> = Vec::new();
let mut stream = response;
while let Some(chunk) = stream.chunk().await.map_err(|e| e.to_string())? {
body.extend_from_slice(&chunk);
if body.len() >= max_bytes {
break;
}
}
return Ok(Fetched {
body: decode_body(&body, &content_type),
final_url: current,
});
}
Err("too many redirects".to_string())
}
async fn require_room_manager(
service: &ChatService,
room_id: Uuid,
caller: Uuid,
) -> Result<(), ChatError> {
let ctx = require_room_readable(service, room_id, caller).await?;
match &ctx.space {
Some(space) => {
if !can_manage_space(service, space, caller).await? {
return Err(ChatError::Forbidden);
}
}
None => return Err(ChatError::Forbidden),
}
Ok(())
}
fn generate_secret() -> String {
use rand::Rng;
const CHARSET: &[u8] = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
let mut rng = rand::thread_rng();
(0..48)
.map(|_| CHARSET[rng.gen_range(0..CHARSET.len())] as char)
.collect()
}
impl ChatService {
pub async fn create_outbound_webhook(
&self,
caller: Uuid,
room_id: Uuid,
cmd: CreateOutboundWebhookCmd,
) -> Result<OutboundWebhookSecretDto, ChatError> {
require_room_manager(self, room_id, caller).await?;
let name = cmd.name.trim();
if name.is_empty() || name.chars().count() > self.config().chat.max_webhook_name_len {
return Err(ChatError::BadRequest("invalid webhook name".to_string()));
}
validate_outbound_url(cmd.url.trim())?;
let now = OffsetDateTime::now_utc();
let row = outbound_webhooks::ActiveModel {
id: Set(Uuid::now_v7()),
room_id: Set(room_id),
user_id: Set(caller),
name: Set(name.to_string()),
url: Set(cmd.url.trim().to_string()),
secret: Set(generate_secret()),
disabled: Set(false),
last_status: Set(None),
last_used_at: Set(None),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db().write())
.await?;
Ok(OutboundWebhookSecretDto {
id: row.id,
room_id: row.room_id,
name: row.name,
url: row.url,
secret: row.secret,
created_at: row.created_at,
})
}
pub async fn list_outbound_webhooks(
&self,
caller: Uuid,
room_id: Uuid,
) -> Result<Vec<OutboundWebhookDto>, ChatError> {
require_room_manager(self, room_id, caller).await?;
let rows = outbound_webhooks::Entity::find()
.filter(outbound_webhooks::Column::RoomId.eq(room_id))
.all(self.db().read())
.await?;
Ok(rows.into_iter().map(OutboundWebhookDto::from).collect())
}
pub async fn update_outbound_webhook(
&self,
caller: Uuid,
room_id: Uuid,
webhook_id: Uuid,
cmd: UpdateOutboundWebhookCmd,
) -> Result<OutboundWebhookDto, ChatError> {
require_room_manager(self, room_id, caller).await?;
let webhook = outbound_webhooks::Entity::find_by_id(webhook_id)
.filter(outbound_webhooks::Column::RoomId.eq(room_id))
.one(self.db().write())
.await?
.ok_or(ChatError::NotFound)?;
let mut am: outbound_webhooks::ActiveModel = webhook.into();
if let Some(name) = cmd.name {
let name = name.trim();
if name.is_empty() || name.chars().count() > self.config().chat.max_webhook_name_len {
return Err(ChatError::BadRequest("invalid webhook name".to_string()));
}
am.name = Set(name.to_string());
}
if let Some(url) = cmd.url {
validate_outbound_url(url.trim())?;
am.url = Set(url.trim().to_string());
}
if let Some(disabled) = cmd.disabled {
am.disabled = Set(disabled);
}
am.updated_at = Set(OffsetDateTime::now_utc());
let updated = am.update(self.db().write()).await?;
Ok(OutboundWebhookDto::from(updated))
}
pub async fn delete_outbound_webhook(
&self,
caller: Uuid,
room_id: Uuid,
webhook_id: Uuid,
) -> Result<(), ChatError> {
require_room_manager(self, room_id, caller).await?;
let deleted = outbound_webhooks::Entity::delete_many()
.filter(outbound_webhooks::Column::Id.eq(webhook_id))
.filter(outbound_webhooks::Column::RoomId.eq(room_id))
.exec(self.db().write())
.await?;
if deleted.rows_affected == 0 {
return Err(ChatError::NotFound);
}
webhook_deliveries::Entity::delete_many()
.filter(webhook_deliveries::Column::WebhookId.eq(webhook_id))
.filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending))
.exec(self.db().write())
.await?;
Ok(())
}
#[tracing::instrument(name = "chat.deliver_webhooks", skip_all)]
pub async fn deliver_pending_webhook_events(&self) -> Result<u64, ChatError> {
let now = OffsetDateTime::now_utc();
let _ = webhook_deliveries::Entity::update_many()
.col_expr(
webhook_deliveries::Column::Status,
Expr::value(WebhookDeliveryStatus::Pending),
)
.filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Delivering))
.filter(
webhook_deliveries::Column::NextRetryAt
.lt(now - time::Duration::minutes(crate::worker::STUCK_CLAIM_MINUTES)),
)
.exec(self.db().write())
.await?;
let due = webhook_deliveries::Entity::find()
.filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending))
.filter(webhook_deliveries::Column::NextRetryAt.lte(now))
.limit(50)
.all(self.db().write())
.await?;
let count = due.len() as u64;
for delivery in due {
let claimed = webhook_deliveries::Entity::update_many()
.col_expr(
webhook_deliveries::Column::Status,
Expr::value(WebhookDeliveryStatus::Delivering),
)
.filter(webhook_deliveries::Column::Id.eq(delivery.id))
.filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending))
.exec(self.db().write())
.await?;
if claimed.rows_affected == 0 {
continue;
}
let Some(webhook) = outbound_webhooks::Entity::find_by_id(delivery.webhook_id)
.one(self.db().read())
.await?
else {
self.finish_delivery(
delivery.id,
WebhookDeliveryStatus::Failed,
Some("webhook deleted".to_string()),
delivery.attempts,
)
.await;
continue;
};
let result = if webhook.disabled {
Err("webhook disabled".to_string())
} else {
self.post_webhook_payload(&webhook, &delivery).await
};
let (status, error) = match result {
Ok(status_line) => {
let mut am: outbound_webhooks::ActiveModel = webhook.into();
am.last_status = Set(Some(status_line.clone()));
am.last_used_at = Set(Some(OffsetDateTime::now_utc()));
am.updated_at = Set(OffsetDateTime::now_utc());
let _ = am.update(self.db().write()).await;
(WebhookDeliveryStatus::Done, None)
}
Err(error) => {
let attempts = delivery.attempts + 1;
if attempts >= MAX_DELIVERY_ATTEMPTS {
(WebhookDeliveryStatus::Failed, Some(error))
} else {
let mut am: webhook_deliveries::ActiveModel = delivery.clone().into();
am.status = Set(WebhookDeliveryStatus::Pending);
am.attempts = Set(attempts);
am.next_retry_at = Set(OffsetDateTime::now_utc()
+ time::Duration::seconds(60 * attempts as i64));
am.last_error = Set(Some(error.clone()));
am.updated_at = Set(OffsetDateTime::now_utc());
let _ = am.update(self.db().write()).await;
continue;
}
}
};
self.finish_delivery(delivery.id, status, error, delivery.attempts)
.await;
}
Ok(count)
}
async fn finish_delivery(
&self,
id: Uuid,
status: WebhookDeliveryStatus,
error: Option<String>,
attempts: i32,
) {
let am = webhook_deliveries::ActiveModel {
id: Set(id),
status: Set(status),
attempts: Set(attempts + 1),
last_error: Set(error),
updated_at: Set(OffsetDateTime::now_utc()),
..Default::default()
};
let _ = am.update(self.db().write()).await;
}
async fn post_webhook_payload(
&self,
webhook: &outbound_webhooks::Model,
delivery: &webhook_deliveries::Model,
) -> Result<String, String> {
use hmac::{Hmac, KeyInit, Mac};
use sha2::Sha256;
let uri = validate_outbound_url(&webhook.url).map_err(|e| e.to_string())?;
let host = uri.host().unwrap_or("").to_string();
let port = uri.port_u16();
let addrs = tokio::net::lookup_host((host.as_str(), port.unwrap_or(443)))
.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| !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(webhook.secret.as_bytes()).map_err(|e| e.to_string())?;
mac.update(&body);
let signature = format!("sha256={}", hex_lower(&mac.finalize().into_bytes()));
let mut client_builder = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(10))
.redirect(reqwest::redirect::Policy::none());
for addr in &addrs {
client_builder = client_builder.resolve(&host, *addr);
}
let client = client_builder.build().map_err(|e| e.to_string())?;
let span = export::telemetry::http_client_span("POST", &webhook.url);
let response = export::telemetry::send(
span,
client
.post(&webhook.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::chat_webhook_delivery("error");
format!("request failed: {e}")
})?;
let status = response.status().as_u16();
if (200..300).contains(&status) {
export::metrics::chat_webhook_delivery("ok");
Ok(format!("delivered {status}"))
} else {
export::metrics::chat_webhook_delivery("error");
Err(format!("endpoint returned {status}"))
}
}
}
fn hex_lower(bytes: &[u8]) -> String {
bytes.iter().map(|b| format!("{b:02x}")).collect()
}
pub async fn enqueue_outbound_event(
service: &ChatService,
room_id: Uuid,
event_type: &str,
payload: &serde_json::Value,
) {
let hooks = outbound_webhooks::Entity::find()
.filter(outbound_webhooks::Column::RoomId.eq(room_id))
.filter(outbound_webhooks::Column::Disabled.eq(false))
.all(service.db().read())
.await
.unwrap_or_default();
if hooks.is_empty() {
return;
}
let now = OffsetDateTime::now_utc();
for hook in hooks {
let row = webhook_deliveries::ActiveModel {
id: Set(Uuid::now_v7()),
webhook_id: Set(hook.id),
room_id: Set(room_id),
event_type: Set(event_type.to_string()),
payload: Set(payload.clone()),
status: Set(WebhookDeliveryStatus::Pending),
attempts: Set(0),
next_retry_at: Set(now),
last_error: Set(None),
created_at: Set(now),
updated_at: Set(now),
};
if let Err(error) = row.insert(service.db().write()).await {
tracing::warn!(%error, "failed to enqueue outbound webhook delivery");
}
}
}