brian/bebot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rearrange the code a bit

Browse Source 
I'm prepping to make bebot do more things than just be a gitlab webhook
handler, so I've moved the gitlab stuff into its own module (and some of
the matrix helper functions too, for good measure).

The config file also now puts all the gitlab-specific configuration
under a 'gitlab' key.
This commit is contained in:
Brian Tarricone 2023-09-20 00:54:26 -07:00
parent 61febd2745
commit 092b29637f
6 changed files with 408 additions and 345 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
sample-config.yaml
View File

@ -2,16 +2,18 @@
bind_address: 127.0.0.1
# Port the webhook listener should bind to (default is 3000).
bind_port: 3000
# Optional prefix to serve the webhook path under (default is empty string).
url_prefix: "/bebot"
# Matrix user to sign in as.
user_id: "@mybebot:example.com"
# Password for Matrix user.
password: "secret-matrix-account-password"
# Default Matrix room to publish events to.
default_room: "#my-project-commits:example.com"
# Default set of events to publish. If left out, all events will be published.
default_publish_events:
# All Gitlab-specific settings are under here.
gitlab:
# Optional prefix to serve the webhook path under (default is empty string).
url_prefix: "/bebot"
# Default Matrix room to publish Gitlab events to.
default_room: "#my-project-commits:example.com"
# Default set of events to publish. If left out, all events will be published.
default_publish_events:
- name: push
# Regexes of branch names. Leave out entirely for "all branches".
branches:
@ -35,8 +37,8 @@ default_publish_events:
# actions will be published.
statuses:
- failed
# Key-value configuration for repositories.
repo_configs:
# Key-value configuration for repositories.
repo_configs:
# Keys are the instance name / namespace / repository name
"gitlab.example.com/myorg/my-cool-app":
# Each repository should use a unique, randomly-generated token. Enter

110
src/config.rs
View File

@ -14,14 +14,13 @@
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::{collections::HashMap, fmt, fs::File, io::BufReader};
use std::{collections::HashMap, fs::File, io::BufReader};
use anyhow::Context;
use matrix_sdk::ruma::{OwnedRoomOrAliasId, OwnedUserId, RoomOrAliasId, UserId};
use matrix_sdk::ruma::{OwnedRoomOrAliasId, OwnedUserId};
use regex::Regex;
use serde::de;
use crate::event::{IssueAction, MergeRequestAction, PipelineStatus};
use crate::gitlab_event::{IssueAction, MergeRequestAction, PipelineStatus};
#[derive(Deserialize)]
#[serde(tag = "name", rename_all = "snake_case")]
@ -47,111 +46,30 @@ pub enum PublishEvent {
pub struct RepoConfig {
pub token: String,
#[serde(default)]
#[serde(deserialize_with = "deser_optional_room_or_alias_id")]
#[serde(deserialize_with = "crate::matrix::deser_optional_room_or_alias_id")]
pub room: Option<OwnedRoomOrAliasId>,
pub publish_events: Option<Vec<PublishEvent>>,
}
#[derive(Deserialize)]
pub struct Config {
pub bind_address: Option<String>,
pub bind_port: Option<u16>,
pub struct GitlabConfig {
pub url_prefix: Option<String>,
#[serde(deserialize_with = "deser_user_id")]
pub user_id: OwnedUserId,
pub password: String,
#[serde(default)]
#[serde(deserialize_with = "deser_optional_room_or_alias_id")]
#[serde(deserialize_with = "crate::matrix::deser_optional_room_or_alias_id")]
pub default_room: Option<OwnedRoomOrAliasId>,
pub default_publish_events: Option<Vec<PublishEvent>>,
pub repo_configs: HashMap<String, RepoConfig>, // key is repo url without scheme; e.g.
// gitlab.xfce.org/xfce/xfdesktop
}
fn deser_user_id<'de, D>(deserializer: D) -> Result<OwnedUserId, D::Error>
where
D: de::Deserializer<'de>,
{
struct UserIdVisitor;
impl<'de> de::Visitor<'de> for UserIdVisitor {
type Value = OwnedUserId;
fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("a matrix user ID")
}
fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
where
E: de::Error,
{
UserId::parse(v).map_err(E::custom)
}
}
deserializer.deserialize_any(UserIdVisitor)
}
fn deser_room_or_alias_id<'de, D>(deserializer: D) -> Result<OwnedRoomOrAliasId, D::Error>
where
D: de::Deserializer<'de>,
{
struct RoomOrAliasIdVisitor;
impl<'de> de::Visitor<'de> for RoomOrAliasIdVisitor {
type Value = OwnedRoomOrAliasId;
fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("a matrix room ID")
}
fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
where
E: de::Error,
{
RoomOrAliasId::parse(v).map_err(E::custom)
}
}
deserializer.deserialize_any(RoomOrAliasIdVisitor)
}
fn deser_optional_room_or_alias_id<'de, D>(deserializer: D) -> Result<Option<OwnedRoomOrAliasId>, D::Error>
where
D: de::Deserializer<'de>,
{
struct OptionalRoomOrAliasIdVisitor;
impl<'de> de::Visitor<'de> for OptionalRoomOrAliasIdVisitor {
type Value = Option<OwnedRoomOrAliasId>;
fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("null or matrix room ID")
}
fn visit_none<E>(self) -> Result<Self::Value, E>
where
E: de::Error,
{
Ok(None)
}
fn visit_some<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
where
D: serde::Deserializer<'de>,
{
Ok(Some(deser_room_or_alias_id(deserializer)?))
}
fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
where
E: de::Error,
{
RoomOrAliasId::parse(v).map(Some).map_err(E::custom)
}
}
deserializer.deserialize_any(OptionalRoomOrAliasIdVisitor)
#[derive(Deserialize)]
pub struct Config {
pub bind_address: Option<String>,
pub bind_port: Option<u16>,
#[serde(deserialize_with = "crate::matrix::deser_user_id")]
pub user_id: OwnedUserId,
pub password: String,
pub gitlab: GitlabConfig,
}
fn load_blocking(path: &String) -> anyhow::Result<Config> {

144
src/gitlab.rs Normal file
View File

@ -0,0 +1,144 @@
// bebot
// Copyright (C) 2023 Brian Tarricone <brian@tarricone.org>
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::sync::Arc;
use constant_time_eq::constant_time_eq;
use http::StatusCode;
use matrix_sdk::{
ruma::{events::room::message::RoomMessageEventContent, OwnedRoomOrAliasId},
Client,
};
use tokio::sync::mpsc;
use warp::{filters::BoxedFilter, reply::Reply, Filter};
use crate::{
config::Config,
gitlab_event::{parse_ref, GitlabEvent, GitlabEventExt},
matrix,
};
pub fn build_gitlab_messages(event: &GitlabEvent) -> Vec<String> {
let project = event.project();
let refname = event.r#ref().map(parse_ref);
event
.titles()
.iter()
.map(|title| {
format!(
"\\[{}\\] {}*{}* {}",
project.path_with_namespace,
refname
.as_ref()
.map(|rn| format!("`{}` ", rn))
.unwrap_or_else(|| "".to_string()),
event.user(),
title,
)
})
.collect()
}
pub async fn handle_gitlab_event(
event: GitlabEvent,
room_id: &OwnedRoomOrAliasId,
matrix_client: &Client,
) -> anyhow::Result<()> {
let room = matrix::ensure_room_joined(matrix_client, room_id).await?;
for msg in build_gitlab_messages(&event) {
debug!("Sending message to {}: {}", room_id, msg);
let msg_content = RoomMessageEventContent::text_markdown(&msg);
room.send(msg_content, None).await?;
}
Ok(())
}
pub fn build_webhook_route(config: Arc<Config>, matrix_client: Client) -> anyhow::Result<BoxedFilter<(impl Reply,)>> {
let (event_tx, mut event_rx) = mpsc::channel::<(GitlabEvent, OwnedRoomOrAliasId)>(100);
tokio::spawn(async move {
while let Some((event, room)) = event_rx.recv().await {
if let Err(err) = handle_gitlab_event(event, &room, &matrix_client).await {
warn!("Failed to handle payload: {}", err);
}
}
});
let gitlab_root_path = if let Some(url_prefix) = config.gitlab.url_prefix.as_ref() {
url_prefix.split('/').fold(warp::any().boxed(), |last, segment| {
if segment.is_empty() {
last
} else {
last.and(warp::path(segment.to_string())).boxed()
}
})
} else {
warp::any().boxed()
};
let gitlab = gitlab_root_path
.and(warp::path!("hooks" / "gitlab"))
.and(warp::post())
.and(warp::header::<String>("x-gitlab-token"))
.and(warp::body::json())
.then(move |token: String, event: GitlabEvent| {
let config = Arc::clone(&config);
let event_tx = event_tx.clone();
async move {
match event {
GitlabEvent::Other => {
warp::reply::with_status("Unsupported Gitlab event type", StatusCode::BAD_REQUEST)
}
_ => {
let project = event.project();
let config_key = project.web_url.replace("http://", "").replace("https://", "");
if let Some(repo_config) = config.gitlab.repo_configs.get(&config_key) {
if !constant_time_eq(token.as_bytes(), repo_config.token.as_bytes()) {
warn!("Invalid token for repo '{}'", config_key);
warp::reply::with_status("Invalid token", StatusCode::FORBIDDEN)
} else {
debug!("payload: {:?}", event);
if let Some(room) = repo_config.room.as_ref().or(config.gitlab.default_room.as_ref()) {
let publish_events = repo_config
.publish_events
.as_ref()
.or(config.gitlab.default_publish_events.as_ref());
if publish_events.map(|ecs| event.should_publish(ecs)).unwrap_or(true) {
if let Err(err) = event_tx.send((event, room.clone())).await {
warn!("Failed to enqueue payload: {}", err);
}
}
warp::reply::with_status("OK", StatusCode::OK)
} else {
info!("Channel not configured for repo '{}'", config_key);
warp::reply::with_status(
"Matrix room not configured for repo",
StatusCode::NOT_FOUND,
)
}
}
} else {
info!("Repo '{}' unconfigured", config_key);
warp::reply::with_status("Repo not configured", StatusCode::NOT_FOUND)
}
}
}
}
})
.boxed();
Ok(gitlab)
}

0
src/event.rs → src/gitlab_event.rs
View File

209
src/main.rs
View File

@ -22,138 +22,26 @@ extern crate log;
extern crate serde;
mod config;
mod event;
mod gitlab;
mod gitlab_event;
mod matrix;
use std::{env, net::IpAddr, process::exit, sync::Arc, time::Duration};
use std::{env, net::IpAddr, process::exit, sync::Arc};
use anyhow::Context;
use constant_time_eq::constant_time_eq;
use event::{GitlabEvent, GitlabEventExt};
use http::StatusCode;
use matrix_sdk::{
config::SyncSettings,
room::Joined,
ruma::{events::room::message::RoomMessageEventContent, OwnedRoomOrAliasId},
BaseRoom, Client,
};
use tokio::sync::mpsc;
use warp::Filter;
async fn build_sync_settings(matrix_client: &Client) -> SyncSettings {
let mut settings = SyncSettings::default().timeout(Duration::from_secs(30));
if let Some(token) = matrix_client.sync_token().await {
settings = settings.token(token);
}
settings
}
async fn matrix_connect(config: &config::Config) -> anyhow::Result<Client> {
let client = Client::builder()
.server_name(config.user_id.server_name())
.user_agent(format!("{}/{}", env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION")))
.build()
.await?;
client
.login_username(&config.user_id, &config.password)
.initial_device_display_name("Bebot")
.send()
.await?;
info!("Connected to matrix as {}; waiting for first sync", config.user_id);
let settings = build_sync_settings(&client).await;
client.sync_once(settings).await?;
info!("First matrix sync complete");
let sync_client = client.clone();
tokio::spawn(async move {
let settings = build_sync_settings(&sync_client).await;
if let Err(err) = sync_client.sync(settings).await {
error!("Matrix sync failed: {}", err);
exit(1);
}
});
Ok(client)
}
async fn ensure_matrix_room_joined(matrix_client: &Client, room_id: &OwnedRoomOrAliasId) -> anyhow::Result<Joined> {
fn room_matches(a_room: &BaseRoom, our_room: &OwnedRoomOrAliasId) -> bool {
let our_room_str = our_room.as_str();
a_room.room_id().as_str() == our_room_str
|| a_room
.canonical_alias()
.iter()
.any(|alias| alias.as_str() == our_room_str)
|| a_room.alt_aliases().iter().any(|alias| alias.as_str() == our_room_str)
}
let mut room = matrix_client
.joined_rooms()
.iter()
.find(|a_room| room_matches(a_room, room_id))
.cloned();
if room.is_none() {
if let Some(invited) = matrix_client
.invited_rooms()
.iter()
.find(|a_room| room_matches(a_room, room_id))
{
invited.accept_invitation().await?;
} else {
matrix_client.join_room_by_id_or_alias(room_id, &[]).await?;
}
let settings = build_sync_settings(matrix_client).await;
matrix_client.sync_once(settings).await?;
room = matrix_client
.joined_rooms()
.iter()
.find(|a_room| room_matches(a_room, room_id))
.cloned();
}
room.ok_or_else(|| anyhow!("Unable to join room {}", room_id))
}
fn build_gitlab_messages(event: &GitlabEvent) -> Vec<String> {
let project = event.project();
let refname = event.r#ref().map(event::parse_ref);
event
.titles()
.iter()
.map(|title| {
format!(
"\\[{}\\] {}*{}* {}",
project.path_with_namespace,
refname
.as_ref()
.map(|rn| format!("`{}` ", rn))
.unwrap_or_else(|| "".to_string()),
event.user(),
title,
)
})
.collect()
}
async fn handle_gitlab_event(
event: GitlabEvent,
room_id: &OwnedRoomOrAliasId,
matrix_client: &Client,
) -> anyhow::Result<()> {
let room = ensure_matrix_room_joined(matrix_client, room_id).await?;
for msg in build_gitlab_messages(&event) {
debug!("Sending message to {}: {}", room_id, msg);
let msg_content = RoomMessageEventContent::text_markdown(&msg);
room.send(msg_content, None).await?;
}
Ok(())
}
async fn run() -> anyhow::Result<()> {
let config_path = env::args()
.nth(1)
.ok_or_else(|| anyhow!("Config file should be passed as only parameter"))?;
let config = Arc::new(config::load(config_path).await?);
let matrix_client = matrix::connect(&config).await.context("Failed to connect to Matrix")?;
let gitlab = gitlab::build_webhook_route(Arc::clone(&config), matrix_client)?;
let routes = gitlab.with(warp::log("bebot"));
let addr = config
.bind_address
.as_ref()
@ -162,83 +50,6 @@ async fn run() -> anyhow::Result<()> {
.parse::<IpAddr>()
.context("Failed to parse bind_address")?;
let port = config.bind_port.unwrap_or(3000);
let matrix_client = matrix_connect(&config).await.context("Failed to connect to Matrix")?;
let (event_tx, mut event_rx) = mpsc::channel::<(GitlabEvent, OwnedRoomOrAliasId)>(100);
tokio::spawn(async move {
while let Some((event, room)) = event_rx.recv().await {
if let Err(err) = handle_gitlab_event(event, &room, &matrix_client).await {
warn!("Failed to handle payload: {}", err);
}
}
});
let gitlab_root_path = if let Some(url_prefix) = config.url_prefix.as_ref() {
url_prefix.split('/').fold(warp::any().boxed(), |last, segment| {
if segment.is_empty() {
last
} else {
last.and(warp::path(segment.to_string())).boxed()
}
})
} else {
warp::any().boxed()
};
let gitlab = gitlab_root_path
.and(warp::path!("hooks" / "gitlab"))
.and(warp::post())
.and(warp::header::<String>("x-gitlab-token"))
.and(warp::body::json())
.then(move |token: String, event: event::GitlabEvent| {
let config = Arc::clone(&config);
let event_tx = event_tx.clone();
async move {
match event {
GitlabEvent::Other => {
warp::reply::with_status("Unsupported Gitlab event type", StatusCode::BAD_REQUEST)
}
_ => {
let project = event.project();
let config_key = project.web_url.replace("http://", "").replace("https://", "");
if let Some(repo_config) = config.repo_configs.get(&config_key) {
if !constant_time_eq(token.as_bytes(), repo_config.token.as_bytes()) {
warn!("Invalid token for repo '{}'", config_key);
warp::reply::with_status("Invalid token", StatusCode::FORBIDDEN)
} else {
debug!("payload: {:?}", event);
if let Some(room) = repo_config.room.as_ref().or(config.default_room.as_ref()) {
let publish_events = repo_config
.publish_events
.as_ref()
.or(config.default_publish_events.as_ref());
if publish_events.map(|ecs| event.should_publish(ecs)).unwrap_or(true) {
if let Err(err) = event_tx.send((event, room.clone())).await {
warn!("Failed to enqueue payload: {}", err);
}
}
warp::reply::with_status("OK", StatusCode::OK)
} else {
info!("Channel not configured for repo '{}'", config_key);
warp::reply::with_status(
"Matrix room not configured for repo",
StatusCode::NOT_FOUND,
)
}
}
} else {
info!("Repo '{}' unconfigured", config_key);
warp::reply::with_status("Repo not configured", StatusCode::NOT_FOUND)
}
}
}
}
});
let routes = gitlab.with(warp::log("bebot"));
warp::serve(routes).run((addr, port)).await;
Ok(())

188
src/matrix.rs Normal file
View File

@ -0,0 +1,188 @@
// bebot
// Copyright (C) 2023 Brian Tarricone <brian@tarricone.org>
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::{fmt, process::exit, time::Duration};
use matrix_sdk::{
config::SyncSettings,
room::Joined,
ruma::{OwnedRoomOrAliasId, OwnedUserId, RoomOrAliasId, UserId},
BaseRoom, Client,
};
use serde::de;
use crate::config::Config;
async fn build_sync_settings(matrix_client: &Client) -> SyncSettings {
let mut settings = SyncSettings::default().timeout(Duration::from_secs(30));
if let Some(token) = matrix_client.sync_token().await {
settings = settings.token(token);
}
settings
}
pub async fn connect(config: &Config) -> anyhow::Result<Client> {
let client = Client::builder()
.server_name(config.user_id.server_name())
.user_agent(format!("{}/{}", env!("CARGO_PKG_NAME"), env!("CARGO_PKG_VERSION")))
.build()
.await?;
client
.login_username(&config.user_id, &config.password)
.initial_device_display_name("Bebot")
.send()
.await?;
info!("Connected to matrix as {}; waiting for first sync", config.user_id);
let settings = build_sync_settings(&client).await;
client.sync_once(settings).await?;
info!("First matrix sync complete");
let sync_client = client.clone();
tokio::spawn(async move {
let settings = build_sync_settings(&sync_client).await;
if let Err(err) = sync_client.sync(settings).await {
error!("Matrix sync failed: {}", err);
exit(1);
}
});
Ok(client)
}
pub async fn ensure_room_joined(matrix_client: &Client, room_id: &OwnedRoomOrAliasId) -> anyhow::Result<Joined> {
fn room_matches(a_room: &BaseRoom, our_room: &OwnedRoomOrAliasId) -> bool {
let our_room_str = our_room.as_str();
a_room.room_id().as_str() == our_room_str
|| a_room
.canonical_alias()
.iter()
.any(|alias| alias.as_str() == our_room_str)
|| a_room.alt_aliases().iter().any(|alias| alias.as_str() == our_room_str)
}
let mut room = matrix_client
.joined_rooms()
.iter()
.find(|a_room| room_matches(a_room, room_id))
.cloned();
if room.is_none() {
if let Some(invited) = matrix_client
.invited_rooms()
.iter()
.find(|a_room| room_matches(a_room, room_id))
{
invited.accept_invitation().await?;
} else {
matrix_client.join_room_by_id_or_alias(room_id, &[]).await?;
}
let settings = build_sync_settings(matrix_client).await;
matrix_client.sync_once(settings).await?;
room = matrix_client
.joined_rooms()
.iter()
.find(|a_room| room_matches(a_room, room_id))
.cloned();
}
room.ok_or_else(|| anyhow!("Unable to join room {}", room_id))
}
pub fn deser_user_id<'de, D>(deserializer: D) -> Result<OwnedUserId, D::Error>
where
D: de::Deserializer<'de>,
{
struct UserIdVisitor;
impl<'de> de::Visitor<'de> for UserIdVisitor {
type Value = OwnedUserId;
fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("a matrix user ID")
}
fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
where
E: de::Error,
{
UserId::parse(v).map_err(E::custom)
}
}
deserializer.deserialize_any(UserIdVisitor)
}
fn deser_room_or_alias_id<'de, D>(deserializer: D) -> Result<OwnedRoomOrAliasId, D::Error>
where
D: de::Deserializer<'de>,
{
struct RoomOrAliasIdVisitor;
impl<'de> de::Visitor<'de> for RoomOrAliasIdVisitor {
type Value = OwnedRoomOrAliasId;
fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("a matrix room ID")
}
fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
where
E: de::Error,
{
RoomOrAliasId::parse(v).map_err(E::custom)
}
}
deserializer.deserialize_any(RoomOrAliasIdVisitor)
}
pub fn deser_optional_room_or_alias_id<'de, D>(deserializer: D) -> Result<Option<OwnedRoomOrAliasId>, D::Error>
where
D: de::Deserializer<'de>,
{
struct OptionalRoomOrAliasIdVisitor;
impl<'de> de::Visitor<'de> for OptionalRoomOrAliasIdVisitor {
type Value = Option<OwnedRoomOrAliasId>;
fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("null or matrix room ID")
}
fn visit_none<E>(self) -> Result<Self::Value, E>
where
E: de::Error,
{
Ok(None)
}
fn visit_some<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
where
D: serde::Deserializer<'de>,
{
Ok(Some(deser_room_or_alias_id(deserializer)?))
}
fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
where
E: de::Error,
{
RoomOrAliasId::parse(v).map(Some).map_err(E::custom)
}
}
deserializer.deserialize_any(OptionalRoomOrAliasIdVisitor)
}