MaigoLabs/refray
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[+] Webhook mode

Browse Source
This commit is contained in:
Azalea Gui
2026-05-07 04:55:49 +00:00
parent 84865134b6
commit c4473d11e3
9 changed files with 2011 additions and 461 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config.rs
+48 -55
View File
@@ -13,6 +13,8 @@ pub struct Config {
pub sites: Vec<SiteConfig>,
#[serde(default)]
pub mirrors: Vec<MirrorConfig>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub webhook: Option<WebhookConfig>,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
@@ -55,6 +57,18 @@ pub struct MirrorConfig {
pub allow_force: bool,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct WebhookConfig {
#[serde(default = "default_true")]
pub install: bool,
pub url: String,
pub secret: TokenConfig,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub full_sync_interval_minutes: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reachability_check_interval_minutes: Option<u64>,
}
#[derive(Clone, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
pub struct EndpointConfig {
pub site: String,
@@ -127,23 +141,6 @@ impl Config {
}
}
pub fn remove_site(&mut self, name: &str) -> Result<()> {
if !self.sites.iter().any(|site| site.name == name) {
bail!("site '{name}' does not exist");
}
for mirror in &self.mirrors {
if mirror
.endpoints
.iter()
.any(|endpoint| endpoint.site == name)
{
bail!("site '{name}' is still used by mirror '{}'", mirror.name);
}
}
self.sites.retain(|site| site.name != name);
Ok(())
}
pub fn upsert_mirror(&mut self, mirror: MirrorConfig) {
if let Some(existing) = self
.mirrors
@@ -168,12 +165,7 @@ impl Config {
impl SiteConfig {
pub fn token(&self) -> Result<String> {
match &self.token {
TokenConfig::Value(value) => Ok(value.clone()),
TokenConfig::Env(name) => {
env::var(name).with_context(|| format!("environment variable {name} is not set"))
}
}
self.token.value("site token")
}
pub fn api_base(&self) -> String {
@@ -196,6 +188,22 @@ impl SiteConfig {
}
}
impl WebhookConfig {
pub fn secret(&self) -> Result<String> {
self.secret.value("webhook secret")
}
}
impl TokenConfig {
pub fn value(&self, label: &str) -> Result<String> {
match self {
TokenConfig::Value(value) => Ok(value.clone()),
TokenConfig::Env(name) => env::var(name)
.with_context(|| format!("environment variable {name} for {label} is not set")),
}
}
}
impl EndpointConfig {
pub fn label(&self) -> String {
format!("{}:{}:{:?}", self.site, self.namespace, self.kind)
@@ -267,6 +275,13 @@ mod tests {
fn parses_token_forms() {
let config: Config = toml::from_str(
r#"
[webhook]
install = true
url = "https://mirror.example.test/webhook"
secret = { env = "WEBHOOK_SECRET" }
full_sync_interval_minutes = 60
reachability_check_interval_minutes = 15
[[sites]]
name = "github"
provider = "github"
@@ -294,6 +309,14 @@ mod tests {
assert_eq!(config.sites.len(), 1);
assert_eq!(config.mirrors[0].endpoints.len(), 2);
let webhook = config.webhook.unwrap();
assert!(webhook.install);
assert_eq!(webhook.url, "https://mirror.example.test/webhook");
assert_eq!(
webhook.secret,
TokenConfig::Env("WEBHOOK_SECRET".to_string())
);
assert_eq!(webhook.full_sync_interval_minutes, Some(60));
}
#[test]
@@ -311,6 +334,7 @@ mod tests {
visibility: Visibility::Private,
allow_force: false,
}],
webhook: None,
};
let err = validate_config(&config).unwrap_err().to_string();
assert!(err.contains("at least two endpoints"));
@@ -335,43 +359,12 @@ mod tests {
visibility: Visibility::Private,
allow_force: false,
}],
webhook: None,
};
let err = validate_config(&config).unwrap_err().to_string();
assert!(err.contains("unknown site 'missing'"));
}
#[test]
fn removing_referenced_site_is_rejected() {
let mut config = Config {
sites: vec![
site("github", ProviderKind::Github),
site("gitea", ProviderKind::Gitea),
],
mirrors: vec![MirrorConfig {
name: "personal".to_string(),
endpoints: vec![
EndpointConfig {
site: "github".to_string(),
kind: NamespaceKind::User,
namespace: "alice".to_string(),
},
EndpointConfig {
site: "gitea".to_string(),
kind: NamespaceKind::User,
namespace: "alice".to_string(),
},
],
create_missing: true,
visibility: Visibility::Private,
allow_force: false,
}],
};
let err = config.remove_site("github").unwrap_err().to_string();
assert!(err.contains("still used by mirror 'personal'"));
assert!(config.site("github").is_some());
}
#[test]
fn api_base_defaults_match_providers() {
assert_eq!(
src/interactive.rs
+206 -2
View File
@@ -1,6 +1,8 @@
use std::fmt::Display;
use std::fs::File;
use std::io::Read;
use std::path::Path;
use std::time::Duration;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use anyhow::{Context, Result};
use console::{Term, style};
@@ -15,9 +17,10 @@ use std::io::{BufRead, Write};
use crate::config::{
Config, EndpointConfig, MirrorConfig, NamespaceKind, ProviderKind, SiteConfig, TokenConfig,
Visibility,
Visibility, WebhookConfig,
};
use crate::provider::ProviderClient;
use crate::webhook::check_webhook_url_reachable;
#[derive(Clone, Debug)]
struct ProfileTarget {
@@ -120,10 +123,89 @@ fn add_sync_group_styled(config: &mut Config, theme: &ColorfulTheme) -> Result<(
visibility: Visibility::Private,
allow_force: false,
});
prompt_webhook_setup_styled(config, theme)?;
Ok(())
}
fn prompt_webhook_setup_styled(config: &mut Config, theme: &ColorfulTheme) -> Result<()> {
if config
.webhook
.as_ref()
.is_some_and(|webhook| webhook.install)
{
println!(
"{} {}",
style("Webhooks").green().bold(),
style("already enabled").dim()
);
return Ok(());
}
println!();
println!(
"{} {}",
style("Webhooks").cyan().bold(),
style(
"strongly recommended; they sync immediately after pushes and greatly reduce conflicts"
)
.dim()
);
if !Confirm::with_theme(theme)
.with_prompt("Install webhooks for configured repositories?")
.default(true)
.interact()?
{
return Ok(());
}
let url = Input::<String>::with_theme(theme)
.with_prompt("Webhook URL reachable by GitHub/GitLab/Gitea")
.validate_with(|value: &String| validate_url(value))
.interact_text()?;
match check_webhook_url_reachable(&url) {
Ok(()) => println!(
"{} {}",
style("reachable").green().bold(),
style(&url).cyan()
),
Err(error) => {
println!(
"{} {}: {error:#}",
style("not reachable from here").yellow().bold(),
style(&url).cyan()
);
if !Confirm::with_theme(theme)
.with_prompt("Save this webhook URL anyway?")
.default(false)
.interact()?
{
return Ok(());
}
}
}
let full_sync_interval_minutes = if Confirm::with_theme(theme)
.with_prompt("Run periodic full sync while the webhook server is running?")
.default(true)
.interact()?
{
Some(
Input::<u64>::with_theme(theme)
.with_prompt("Full sync interval in minutes")
.default(60)
.interact_text()?,
)
} else {
None
};
config.webhook = Some(WebhookConfig {
install: true,
url,
secret: TokenConfig::Value(generate_webhook_secret()),
full_sync_interval_minutes,
reachability_check_interval_minutes: Some(15),
});
Ok(())
}
fn prompt_wizard_action_styled(theme: &ColorfulTheme) -> Result<WizardAction> {
let options = ["Add another sync group", "Delete an existing group", "Done"];
let index = Select::with_theme(theme)
@@ -550,6 +632,56 @@ where
visibility: Visibility::Private,
allow_force: false,
});
prompt_webhook_setup(reader, writer, config)?;
Ok(())
}
#[cfg(test)]
fn prompt_webhook_setup<R, W>(reader: &mut R, writer: &mut W, config: &mut Config) -> Result<()>
where
R: BufRead,
W: Write,
{
if config
.webhook
.as_ref()
.is_some_and(|webhook| webhook.install)
{
writeln!(writer, "Webhooks already enabled.")?;
return Ok(());
}
writeln!(
writer,
"Install webhooks? Strongly recommended because immediate sync greatly reduces conflicts."
)?;
if !prompt_bool(reader, writer, "Install webhook?", true)? {
return Ok(());
}
let url = prompt_required(reader, writer, "Webhook URL reachable by providers")?;
if let Err(error) = validate_url(&url) {
bail!(error);
}
let full_sync_interval_minutes = if prompt_bool(
reader,
writer,
"Run periodic full sync while serve is running?",
true,
)? {
Some(
prompt_with_default(reader, writer, "Full sync interval in minutes", "60")?
.parse::<u64>()
.context("full sync interval must be a number")?,
)
} else {
None
};
config.webhook = Some(WebhookConfig {
install: true,
url,
secret: TokenConfig::Value("test-webhook-secret".to_string()),
full_sync_interval_minutes,
reachability_check_interval_minutes: Some(15),
});
Ok(())
}
@@ -1128,6 +1260,36 @@ fn validate_required(value: &str) -> std::result::Result<(), String> {
}
}
fn validate_url(value: &str) -> std::result::Result<(), String> {
validate_required(value)?;
let url = Url::parse(value).map_err(|error| format!("Invalid URL: {error}"))?;
match url.scheme() {
"http" | "https" => Ok(()),
_ => Err("URL must start with http:// or https://".to_string()),
}
}
fn generate_webhook_secret() -> String {
let mut bytes = [0_u8; 32];
if File::open("/dev/urandom")
.and_then(|mut file| file.read_exact(&mut bytes))
.is_err()
{
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_nanos())
.unwrap_or_default();
for (index, byte) in bytes.iter_mut().enumerate() {
*byte = ((nanos >> ((index % 16) * 8)) & 0xff) as u8;
}
}
let mut output = String::with_capacity(bytes.len() * 2);
for byte in bytes {
output.push_str(&format!("{byte:02x}"));
}
output
}
#[cfg(test)]
mod tests {
use super::*;
@@ -1143,6 +1305,7 @@ mod tests {
"gt-token",
"",
"n",
"n",
"3",
]
.join("\n")
@@ -1198,6 +1361,7 @@ mod tests {
"gt-token",
"",
"n",
"n",
"3",
]
.join("\n")
@@ -1213,6 +1377,41 @@ mod tests {
assert_eq!(config.sites.len(), 3);
}
#[test]
fn wizard_can_enable_webhooks() {
let input = [
"https://github.com/alice",
"gh-token",
"",
"https://gitea.example.test/alice",
"gt-token",
"",
"n",
"y",
"https://mirror.example.test/webhook",
"y",
"30",
"3",
]
.join("\n")
+ "\n";
let mut reader = Cursor::new(input.as_bytes());
let mut output = Vec::new();
let config =
run_config_wizard_with_io(Config::default(), &mut reader, &mut output).unwrap();
let webhook = config.webhook.unwrap();
assert!(webhook.install);
assert_eq!(webhook.url, "https://mirror.example.test/webhook");
assert_eq!(webhook.full_sync_interval_minutes, Some(30));
assert_eq!(webhook.reachability_check_interval_minutes, Some(15));
assert_eq!(
webhook.secret,
TokenConfig::Value("test-webhook-secret".to_string())
);
}
#[test]
fn wizard_reuses_existing_credentials_for_same_instance() {
let config = Config {
@@ -1225,6 +1424,7 @@ mod tests {
git_username: None,
}],
mirrors: Vec::new(),
webhook: None,
};
let input = [
"https://github.com/alice",
@@ -1232,6 +1432,7 @@ mod tests {
"https://github.com/bob",
"",
"n",
"n",
"3",
]
.join("\n")
@@ -1285,6 +1486,7 @@ mod tests {
visibility: Visibility::Private,
allow_force: false,
}],
webhook: None,
};
let mut reader = Cursor::new(b"3\n".as_slice());
let mut output = Vec::new();
@@ -1337,6 +1539,7 @@ mod tests {
visibility: Visibility::Private,
allow_force: false,
}],
webhook: None,
};
let input = ["2", "1", "3"].join("\n") + "\n";
let mut reader = Cursor::new(input.as_bytes());
@@ -1391,6 +1594,7 @@ mod tests {
visibility: Visibility::Private,
allow_force: false,
}],
webhook: None,
};
let input = ["2", "2", "3"].join("\n") + "\n";
let mut reader = Cursor::new(input.as_bytes());
src/main.rs
+208 -338
View File
@@ -4,17 +4,20 @@ mod interactive;
mod logging;
mod provider;
mod sync;
mod webhook;
use std::env;
use std::path::PathBuf;
use anyhow::{Context, Result, bail};
use clap::{Args, Parser, Subcommand, ValueEnum};
use anyhow::{Context, Result};
use clap::{Args, Parser, Subcommand};
use crate::config::{
Config, EndpointConfig, NamespaceKind, ProviderKind, SiteConfig, TokenConfig, Visibility,
default_config_path,
};
use crate::config::{Config, default_config_path};
use crate::sync::{DEFAULT_JOBS, SyncOptions, sync_all};
use crate::webhook::{
ServeOptions, WebhookInstallOptions, WebhookUninstallOptions, install_webhooks, serve,
uninstall_webhooks,
};
#[derive(Parser, Debug)]
#[command(name = "git-sync")]
@@ -29,71 +32,15 @@ struct Cli {
#[derive(Subcommand, Debug)]
enum Command {
#[command(subcommand)]
Config(ConfigCommand),
/// Run the interactive configuration wizard
Config,
/// Sync configured mirror groups once
Sync(SyncCommand),
}
#[derive(Subcommand, Debug)]
enum ConfigCommand {
Init,
Wizard,
/// Run the webhook receiver
Serve(ServeCommand),
/// Install or uninstall repository webhooks
#[command(subcommand)]
Site(SiteCommand),
#[command(subcommand)]
Mirror(MirrorCommand),
Show,
}
#[derive(Subcommand, Debug)]
enum SiteCommand {
Add(SiteAddCommand),
Remove(NameCommand),
List,
}
#[derive(Subcommand, Debug)]
enum MirrorCommand {
Add(MirrorAddCommand),
Remove(NameCommand),
List,
}
#[derive(Args, Debug)]
struct NameCommand {
name: String,
}
#[derive(Args, Debug)]
struct SiteAddCommand {
#[arg(long)]
name: String,
#[arg(long)]
provider: ProviderArg,
#[arg(long, value_name = "URL")]
base_url: String,
#[arg(long, value_name = "URL")]
api_url: Option<String>,
#[arg(long, conflicts_with = "token_env")]
token: Option<String>,
#[arg(long, value_name = "ENV", conflicts_with = "token")]
token_env: Option<String>,
#[arg(long)]
git_username: Option<String>,
}
#[derive(Args, Debug)]
struct MirrorAddCommand {
#[arg(long)]
name: String,
#[arg(long = "endpoint", required = true, action = clap::ArgAction::Append, value_name = "SITE:KIND:NAMESPACE")]
endpoints: Vec<String>,
#[arg(long, default_value_t = true, action = clap::ArgAction::Set)]
create_missing: bool,
#[arg(long, default_value_t = VisibilityArg::Private)]
visibility: VisibilityArg,
#[arg(long, default_value_t = false)]
allow_force: bool,
Webhook(WebhookCommand),
}
#[derive(Args, Debug)]
@@ -116,27 +63,54 @@ struct SyncCommand {
jobs: usize,
}
#[derive(Clone, Debug, ValueEnum)]
enum ProviderArg {
Github,
Gitlab,
Gitea,
Forgejo,
#[derive(Args, Debug)]
struct ServeCommand {
#[arg(long, default_value = "127.0.0.1:8787", value_name = "HOST:PORT")]
listen: String,
#[arg(long, conflicts_with = "secret_env")]
secret: Option<String>,
#[arg(long, value_name = "ENV", conflicts_with = "secret")]
secret_env: Option<String>,
#[arg(long, default_value_t = DEFAULT_JOBS, value_name = "N")]
jobs: usize,
#[arg(long, value_name = "PATH")]
work_dir: Option<PathBuf>,
#[arg(long, value_name = "MINUTES")]
full_sync_interval_minutes: Option<u64>,
}
#[derive(Clone, Debug, ValueEnum)]
enum VisibilityArg {
Private,
Public,
#[derive(Subcommand, Debug)]
enum WebhookCommand {
Install(WebhookInstallCommand),
Uninstall(WebhookUninstallCommand),
}
impl std::fmt::Display for VisibilityArg {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Private => write!(f, "private"),
Self::Public => write!(f, "public"),
}
}
#[derive(Args, Debug)]
struct WebhookInstallCommand {
#[arg(long, value_name = "URL")]
url: Option<String>,
#[arg(long, conflicts_with = "secret_env")]
secret: Option<String>,
#[arg(long, value_name = "ENV", conflicts_with = "secret")]
secret_env: Option<String>,
#[arg(long, value_name = "NAME")]
group: Option<String>,
#[arg(long, value_name = "REGEX")]
repo_pattern: Option<String>,
#[arg(long)]
dry_run: bool,
#[arg(long, value_name = "PATH")]
work_dir: Option<PathBuf>,
}
#[derive(Args, Debug)]
struct WebhookUninstallCommand {
#[arg(long, value_name = "NAME")]
group: Option<String>,
#[arg(long)]
dry_run: bool,
#[arg(long, value_name = "PATH")]
work_dir: Option<PathBuf>,
}
fn main() -> Result<()> {
@@ -144,7 +118,7 @@ fn main() -> Result<()> {
let config_path = cli.config.unwrap_or_else(default_config_path);
match cli.command {
Command::Config(command) => handle_config(command, config_path),
Command::Config => interactive::run_config_wizard(&config_path),
Command::Sync(command) => {
let config = Config::load(&config_path)
.with_context(|| format!("failed to load config at {}", config_path.display()))?;
@@ -162,160 +136,89 @@ fn main() -> Result<()> {
},
)
}
}
}
fn handle_config(command: ConfigCommand, path: PathBuf) -> Result<()> {
match command {
ConfigCommand::Init => {
if path.exists() {
bail!("config already exists at {}", path.display());
}
let config = Config::default();
config.save(&path)?;
println!("created {}", path.display());
Ok(())
}
ConfigCommand::Wizard => interactive::run_config_wizard(&path),
ConfigCommand::Site(command) => handle_site(command, path),
ConfigCommand::Mirror(command) => handle_mirror(command, path),
ConfigCommand::Show => {
let config = Config::load(&path)?;
println!("{}", toml::to_string_pretty(&config)?);
Ok(())
}
}
}
fn handle_site(command: SiteCommand, path: PathBuf) -> Result<()> {
let mut config = Config::load_or_default(&path)?;
match command {
SiteCommand::Add(args) => {
let token = match (args.token, args.token_env) {
(Some(value), None) => TokenConfig::Value(value),
(None, Some(env)) => TokenConfig::Env(env),
(None, None) => bail!("pass either --token or --token-env"),
(Some(_), Some(_)) => unreachable!("clap enforces token conflicts"),
};
config.upsert_site(SiteConfig {
name: args.name,
provider: args.provider.into(),
base_url: args.base_url,
api_url: args.api_url,
token,
git_username: args.git_username,
});
config.save(&path)?;
println!("updated {}", path.display());
Ok(())
}
SiteCommand::Remove(args) => {
config.remove_site(&args.name)?;
config.save(&path)?;
println!("removed site {}", args.name);
Ok(())
}
SiteCommand::List => {
for site in &config.sites {
println!("{}\t{:?}\t{}", site.name, site.provider, site.base_url);
}
Ok(())
}
}
}
fn handle_mirror(command: MirrorCommand, path: PathBuf) -> Result<()> {
let mut config = Config::load_or_default(&path)?;
match command {
MirrorCommand::Add(args) => {
if args.endpoints.len() < 2 {
bail!("mirror groups need at least two --endpoint values");
}
let endpoints = args
.endpoints
.iter()
.map(|value| parse_endpoint(value))
.collect::<Result<Vec<_>>>()?;
for endpoint in &endpoints {
Command::Serve(command) => {
let config = Config::load(&config_path)
.with_context(|| format!("failed to load config at {}", config_path.display()))?;
let full_sync_interval_minutes = command.full_sync_interval_minutes.or_else(|| {
config
.site(&endpoint.site)
.with_context(|| format!("unknown site '{}'", endpoint.site))?;
}
config.upsert_mirror(config::MirrorConfig {
name: args.name,
endpoints,
create_missing: args.create_missing,
visibility: args.visibility.into(),
allow_force: args.allow_force,
.webhook
.as_ref()
.and_then(|webhook| webhook.full_sync_interval_minutes)
});
config.save(&path)?;
println!("updated {}", path.display());
Ok(())
let reachability_url = config.webhook.as_ref().map(|webhook| webhook.url.clone());
let reachability_check_interval_minutes = config
.webhook
.as_ref()
.and_then(|webhook| webhook.reachability_check_interval_minutes);
let secret = resolve_webhook_secret(&config, command.secret, command.secret_env)?;
serve(
config,
ServeOptions {
listen: command.listen,
secret,
workers: command.jobs,
work_dir: command.work_dir,
full_sync_interval_minutes,
reachability_url,
reachability_check_interval_minutes,
},
)
}
MirrorCommand::Remove(args) => {
config.remove_mirror(&args.name)?;
config.save(&path)?;
println!("removed mirror {}", args.name);
Ok(())
Command::Webhook(WebhookCommand::Install(command)) => {
let config = Config::load(&config_path)
.with_context(|| format!("failed to load config at {}", config_path.display()))?;
let secret = resolve_webhook_secret(&config, command.secret, command.secret_env)?;
let url = resolve_webhook_url(&config, command.url)?;
install_webhooks(
&config,
WebhookInstallOptions {
url,
secret,
group: command.group,
repo_pattern: command.repo_pattern,
dry_run: command.dry_run,
work_dir: command.work_dir,
},
)
}
MirrorCommand::List => {
for mirror in &config.mirrors {
let endpoints = mirror
.endpoints
.iter()
.map(|endpoint| {
format!(
"{}:{:?}:{}",
endpoint.site, endpoint.kind, endpoint.namespace
)
})
.collect::<Vec<_>>()
.join(", ");
println!("{}\t{}", mirror.name, endpoints);
}
Ok(())
Command::Webhook(WebhookCommand::Uninstall(command)) => {
let config = Config::load(&config_path)
.with_context(|| format!("failed to load config at {}", config_path.display()))?;
uninstall_webhooks(
&config,
WebhookUninstallOptions {
group: command.group,
dry_run: command.dry_run,
work_dir: command.work_dir,
},
)
}
}
}
fn parse_endpoint(value: &str) -> Result<EndpointConfig> {
let parts = value.splitn(3, ':').collect::<Vec<_>>();
if parts.len() != 3 {
bail!("endpoint must be SITE:KIND:NAMESPACE, got '{value}'");
}
let kind = match parts[1].to_ascii_lowercase().as_str() {
"user" => NamespaceKind::User,
"org" | "organization" => NamespaceKind::Org,
"group" => NamespaceKind::Group,
other => bail!("unsupported namespace kind '{other}'"),
};
Ok(EndpointConfig {
site: parts[0].to_string(),
kind,
namespace: parts[2].to_string(),
})
}
impl From<ProviderArg> for ProviderKind {
fn from(value: ProviderArg) -> Self {
match value {
ProviderArg::Github => Self::Github,
ProviderArg::Gitlab => Self::Gitlab,
ProviderArg::Gitea => Self::Gitea,
ProviderArg::Forgejo => Self::Forgejo,
}
fn resolve_webhook_secret(
config: &Config,
value: Option<String>,
env_name: Option<String>,
) -> Result<String> {
match (value, env_name) {
(Some(value), None) => Ok(value),
(None, Some(env_name)) => env::var(&env_name)
.with_context(|| format!("environment variable {env_name} is not set")),
(None, None) => config
.webhook
.as_ref()
.map(|webhook| webhook.secret())
.transpose()?
.ok_or_else(|| anyhow::anyhow!("pass either --secret or --secret-env")),
(Some(_), Some(_)) => unreachable!("clap enforces secret conflicts"),
}
}
impl From<VisibilityArg> for Visibility {
fn from(value: VisibilityArg) -> Self {
match value {
VisibilityArg::Private => Self::Private,
VisibilityArg::Public => Self::Public,
}
}
fn resolve_webhook_url(config: &Config, value: Option<String>) -> Result<String> {
value
.or_else(|| config.webhook.as_ref().map(|webhook| webhook.url.clone()))
.ok_or_else(|| anyhow::anyhow!("pass --url or configure [webhook].url"))
}
#[cfg(test)]
@@ -323,42 +226,23 @@ mod tests {
use super::*;
#[test]
fn cli_accepts_repeated_mirror_endpoints() {
let cli = Cli::try_parse_from([
"git-sync",
"config",
"mirror",
"add",
"--name",
"personal",
"--endpoint",
"github:user:hykilpikonna",
"--endpoint",
"gitea:user:azalea",
])
.unwrap();
fn cli_config_opens_wizard() {
let cli = Cli::try_parse_from(["git-sync", "config"]).unwrap();
let Command::Config(ConfigCommand::Mirror(MirrorCommand::Add(args))) = cli.command else {
panic!("parsed unexpected command");
};
assert_eq!(args.name, "personal");
assert_eq!(
args.endpoints,
vec![
"github:user:hykilpikonna".to_string(),
"gitea:user:azalea".to_string()
]
);
assert!(matches!(cli.command, Command::Config));
}
#[test]
fn cli_accepts_config_wizard() {
let cli = Cli::try_parse_from(["git-sync", "config", "wizard"]).unwrap();
assert!(matches!(
cli.command,
Command::Config(ConfigCommand::Wizard)
));
fn cli_rejects_removed_config_subcommands() {
for args in [
["git-sync", "config", "wizard"].as_slice(),
["git-sync", "config", "init"].as_slice(),
["git-sync", "config", "show"].as_slice(),
["git-sync", "config", "site", "list"].as_slice(),
["git-sync", "config", "mirror", "list"].as_slice(),
] {
assert!(Cli::try_parse_from(args).is_err());
}
}
#[test]
@@ -400,89 +284,75 @@ mod tests {
}
#[test]
fn cli_accepts_new_provider_kinds() {
for (name, expected) in [("forgejo", ProviderKind::Forgejo)] {
let cli = Cli::try_parse_from([
"git-sync",
"config",
"site",
"add",
"--name",
name,
"--provider",
name,
"--base-url",
"https://example.test",
"--token",
"token",
])
.unwrap();
let Command::Config(ConfigCommand::Site(SiteCommand::Add(args))) = cli.command else {
panic!("parsed unexpected command");
};
assert_eq!(ProviderKind::from(args.provider), expected);
}
}
#[test]
fn endpoint_parser_supports_aliases_and_rejects_bad_kinds() {
let endpoint = parse_endpoint("github:organization:MewoLab").unwrap();
assert_eq!(endpoint.site, "github");
assert_eq!(endpoint.kind, NamespaceKind::Org);
assert_eq!(endpoint.namespace, "MewoLab");
let endpoint = parse_endpoint("gitlab:group:parent/child").unwrap();
assert_eq!(endpoint.kind, NamespaceKind::Group);
let err = parse_endpoint("github:team:alice").unwrap_err().to_string();
assert!(err.contains("unsupported namespace kind 'team'"));
let err = parse_endpoint("github:user").unwrap_err().to_string();
assert!(err.contains("SITE:KIND:NAMESPACE"));
}
#[test]
fn site_add_requires_one_token_source() {
let missing = Cli::try_parse_from([
fn cli_accepts_webhook_serve() {
let cli = Cli::try_parse_from([
"git-sync",
"config",
"site",
"add",
"--name",
"github",
"--provider",
"github",
"--base-url",
"https://github.com",
"serve",
"--listen",
"127.0.0.1:9000",
"--secret-env",
"WEBHOOK_SECRET",
"--jobs",
"2",
"--full-sync-interval-minutes",
"30",
])
.unwrap();
let Command::Config(ConfigCommand::Site(SiteCommand::Add(args))) = missing.command else {
let Command::Serve(args) = cli.command else {
panic!("parsed unexpected command");
};
let temp = tempfile::TempDir::new().unwrap();
let err = handle_site(SiteCommand::Add(args), temp.path().join("config.toml"))
.unwrap_err()
.to_string();
assert!(err.contains("pass either --token or --token-env"));
assert_eq!(args.listen, "127.0.0.1:9000");
assert_eq!(args.secret_env, Some("WEBHOOK_SECRET".to_string()));
assert_eq!(args.jobs, 2);
assert_eq!(args.full_sync_interval_minutes, Some(30));
}
let conflict = Cli::try_parse_from([
#[test]
fn cli_accepts_webhook_install() {
let cli = Cli::try_parse_from([
"git-sync",
"config",
"site",
"add",
"--name",
"github",
"--provider",
"github",
"--base-url",
"https://github.com",
"--token",
"a",
"--token-env",
"GITHUB_TOKEN",
]);
assert!(conflict.is_err());
"webhook",
"install",
"--url",
"https://mirror.example.test/webhook",
"--secret",
"secret",
"--group",
"sync-1",
"--repo-pattern",
"^repo$",
])
.unwrap();
let Command::Webhook(WebhookCommand::Install(args)) = cli.command else {
panic!("parsed unexpected command");
};
assert_eq!(
args.url,
Some("https://mirror.example.test/webhook".to_string())
);
assert_eq!(args.secret, Some("secret".to_string()));
assert_eq!(args.group, Some("sync-1".to_string()));
assert_eq!(args.repo_pattern, Some("^repo$".to_string()));
}
#[test]
fn cli_accepts_webhook_uninstall() {
let cli = Cli::try_parse_from([
"git-sync",
"webhook",
"uninstall",
"--group",
"sync-1",
"--dry-run",
])
.unwrap();
let Command::Webhook(WebhookCommand::Uninstall(args)) = cli.command else {
panic!("parsed unexpected command");
};
assert_eq!(args.group, Some("sync-1".to_string()));
assert!(args.dry_run);
}
}
src/provider.rs
+367
View File
@@ -69,6 +69,37 @@ impl<'a> ProviderClient<'a> {
}
}
pub fn install_webhook(
&self,
endpoint: &EndpointConfig,
repo: &RemoteRepo,
url: &str,
secret: &str,
) -> Result<()> {
match self.site.provider {
ProviderKind::Github => self.github_install_webhook(endpoint, repo, url, secret),
ProviderKind::Gitlab => self.gitlab_install_webhook(endpoint, repo, url, secret),
ProviderKind::Gitea | ProviderKind::Forgejo => {
self.gitea_install_webhook(endpoint, repo, url, secret)
}
}
}
pub fn uninstall_webhook(
&self,
endpoint: &EndpointConfig,
repo_name: &str,
url: &str,
) -> Result<bool> {
match self.site.provider {
ProviderKind::Github => self.github_uninstall_webhook(endpoint, repo_name, url),
ProviderKind::Gitlab => self.gitlab_uninstall_webhook(endpoint, repo_name, url),
ProviderKind::Gitea | ProviderKind::Forgejo => {
self.gitea_uninstall_webhook(endpoint, repo_name, url)
}
}
}
pub fn validate_token(&self) -> Result<()> {
let url = format!("{}/user", self.site.api_base());
self.get(&url).map(|_| ())
@@ -169,6 +200,60 @@ impl<'a> ProviderClient<'a> {
})
}
fn github_install_webhook(
&self,
endpoint: &EndpointConfig,
repo: &RemoteRepo,
url: &str,
secret: &str,
) -> Result<()> {
if matches!(endpoint.kind, NamespaceKind::Group) {
bail!("GitHub endpoints use kind 'user' or 'org'");
}
let hooks_url = format!(
"{}/repos/{}/{}/hooks",
self.site.api_base(),
endpoint.namespace,
repo.name
);
let body = json!({
"name": "web",
"active": true,
"events": ["push"],
"config": {
"url": url,
"content_type": "json",
"secret": secret,
"insecure_ssl": "0",
},
});
if let Some(hook) = self.find_existing_hook(&hooks_url, url)? {
let update_url = format!("{hooks_url}/{}", hook.id);
self.patch_json::<serde_json::Value>(&update_url, &body)?;
} else {
self.post_json::<serde_json::Value>(&hooks_url, &body)?;
}
Ok(())
}
fn github_uninstall_webhook(
&self,
endpoint: &EndpointConfig,
repo_name: &str,
url: &str,
) -> Result<bool> {
if matches!(endpoint.kind, NamespaceKind::Group) {
bail!("GitHub endpoints use kind 'user' or 'org'");
}
let hooks_url = format!(
"{}/repos/{}/{}/hooks",
self.site.api_base(),
endpoint.namespace,
repo_name
);
self.delete_matching_hook(&hooks_url, url)
}
fn gitlab_list_repos(&self, endpoint: &EndpointConfig) -> Result<Vec<RemoteRepo>> {
match endpoint.kind {
NamespaceKind::User => {
@@ -247,6 +332,50 @@ impl<'a> ProviderClient<'a> {
.then_some(NamespaceKind::User))
}
fn gitlab_install_webhook(
&self,
endpoint: &EndpointConfig,
repo: &RemoteRepo,
url: &str,
secret: &str,
) -> Result<()> {
let project = format!("{}/{}", endpoint.namespace, repo.name);
let hooks_url = format!(
"{}/projects/{}/hooks",
self.site.api_base(),
urlencoding(&project)
);
let body = json!({
"url": url,
"push_events": true,
"tag_push_events": true,
"token": secret,
"enable_ssl_verification": true,
});
if let Some(hook) = self.find_existing_hook(&hooks_url, url)? {
let update_url = format!("{hooks_url}/{}", hook.id);
self.put_json::<serde_json::Value>(&update_url, &body)?;
} else {
self.post_json::<serde_json::Value>(&hooks_url, &body)?;
}
Ok(())
}
fn gitlab_uninstall_webhook(
&self,
endpoint: &EndpointConfig,
repo_name: &str,
url: &str,
) -> Result<bool> {
let project = format!("{}/{}", endpoint.namespace, repo_name);
let hooks_url = format!(
"{}/projects/{}/hooks",
self.site.api_base(),
urlencoding(&project)
);
self.delete_matching_hook(&hooks_url, url)
}
fn gitea_list_repos(&self, endpoint: &EndpointConfig) -> Result<Vec<RemoteRepo>> {
match endpoint.kind {
NamespaceKind::User => {
@@ -310,6 +439,75 @@ impl<'a> ProviderClient<'a> {
Ok(None)
}
fn gitea_install_webhook(
&self,
endpoint: &EndpointConfig,
repo: &RemoteRepo,
url: &str,
secret: &str,
) -> Result<()> {
if matches!(endpoint.kind, NamespaceKind::Group) {
bail!("Gitea endpoints use kind 'user' or 'org'");
}
let hooks_url = format!(
"{}/repos/{}/{}/hooks",
self.site.api_base(),
endpoint.namespace,
repo.name
);
let body = json!({
"type": "gitea",
"active": true,
"events": ["push"],
"config": {
"url": url,
"content_type": "json",
"secret": secret,
},
});
if let Some(hook) = self.find_existing_hook(&hooks_url, url)? {
let update_url = format!("{hooks_url}/{}", hook.id);
self.patch_json::<serde_json::Value>(&update_url, &body)?;
} else {
self.post_json::<serde_json::Value>(&hooks_url, &body)?;
}
Ok(())
}
fn gitea_uninstall_webhook(
&self,
endpoint: &EndpointConfig,
repo_name: &str,
url: &str,
) -> Result<bool> {
if matches!(endpoint.kind, NamespaceKind::Group) {
bail!("Gitea endpoints use kind 'user' or 'org'");
}
let hooks_url = format!(
"{}/repos/{}/{}/hooks",
self.site.api_base(),
endpoint.namespace,
repo_name
);
self.delete_matching_hook(&hooks_url, url)
}
fn find_existing_hook(&self, hooks_url: &str, target_url: &str) -> Result<Option<RepoHook>> {
let hooks: Vec<RepoHook> = self.paged_get(hooks_url)?;
Ok(hooks
.into_iter()
.find(|hook| hook.url() == Some(target_url)))
}
fn delete_matching_hook(&self, hooks_url: &str, target_url: &str) -> Result<bool> {
let Some(hook) = self.find_existing_hook(hooks_url, target_url)? else {
return Ok(false);
};
let delete_url = format!("{hooks_url}/{}", hook.id);
self.delete(&delete_url)?;
Ok(true)
}
fn paged_get<T>(&self, first_url: &str) -> Result<Vec<T>>
where
T: for<'de> Deserialize<'de>,
@@ -351,6 +549,32 @@ impl<'a> ProviderClient<'a> {
.with_context(|| format!("invalid JSON from {url}"))
}
fn put_json<T>(&self, url: &str, body: &serde_json::Value) -> Result<T>
where
T: for<'de> Deserialize<'de>,
{
self.request_headers(self.http.put(url))?
.json(body)
.send()
.with_context(|| format!("PUT {url} failed"))
.and_then(|response| check_response("PUT", url, response))?
.json()
.with_context(|| format!("invalid JSON from {url}"))
}
fn patch_json<T>(&self, url: &str, body: &serde_json::Value) -> Result<T>
where
T: for<'de> Deserialize<'de>,
{
self.request_headers(self.http.patch(url))?
.json(body)
.send()
.with_context(|| format!("PATCH {url} failed"))
.and_then(|response| check_response("PATCH", url, response))?
.json()
.with_context(|| format!("invalid JSON from {url}"))
}
fn get(&self, url: &str) -> Result<Response> {
self.request_headers(self.http.get(url))?
.send()
@@ -358,6 +582,13 @@ impl<'a> ProviderClient<'a> {
.and_then(|response| check_response("GET", url, response))
}
fn delete(&self, url: &str) -> Result<Response> {
self.request_headers(self.http.delete(url))?
.send()
.with_context(|| format!("DELETE {url} failed"))
.and_then(|response| check_response("DELETE", url, response))
}
fn request_headers(
&self,
request: reqwest::blocking::RequestBuilder,
@@ -500,6 +731,23 @@ impl From<GiteaRepo> for RemoteRepo {
}
}
#[derive(Deserialize)]
struct RepoHook {
id: u64,
#[serde(default)]
url: Option<String>,
#[serde(default)]
config: HashMap<String, String>,
}
impl RepoHook {
fn url(&self) -> Option<&str> {
self.url
.as_deref()
.or_else(|| self.config.get("url").map(String::as_str))
}
}
pub fn repos_by_name(repos: Vec<EndpointRepo>) -> HashMap<String, Vec<EndpointRepo>> {
let mut output: HashMap<String, Vec<EndpointRepo>> = HashMap::new();
for repo in repos {
@@ -677,6 +925,97 @@ mod tests {
handle.join().unwrap();
}
#[test]
fn install_webhook_posts_github_hook_when_missing() {
let (api_url, handle) = request_server(
vec![("200 OK", "[]"), ("201 Created", r#"{"id":1}"#)],
|index, request| match index {
0 => assert!(
request.starts_with("GET /repos/alice/repo/hooks "),
"request was {request}"
),
1 => {
assert!(
request.starts_with("POST /repos/alice/repo/hooks "),
"request was {request}"
);
assert!(request.contains("https://mirror.example.test/webhook"));
assert!(request.contains("secret"));
assert!(request.contains("push"));
}
_ => unreachable!(),
},
);
let site = SiteConfig {
api_url: Some(api_url),
..site(ProviderKind::Github, None)
};
let client = ProviderClient::new(&site).unwrap();
client
.install_webhook(
&EndpointConfig {
site: "github".to_string(),
kind: NamespaceKind::User,
namespace: "alice".to_string(),
},
&RemoteRepo {
name: "repo".to_string(),
clone_url: "https://github.com/alice/repo.git".to_string(),
private: true,
description: None,
},
"https://mirror.example.test/webhook",
"secret",
)
.unwrap();
handle.join().unwrap();
}
#[test]
fn uninstall_webhook_deletes_matching_github_hook() {
let (api_url, handle) = request_server(
vec![
(
"200 OK",
r#"[{"id":42,"config":{"url":"https://mirror.example.test/webhook"}}]"#,
),
("204 No Content", ""),
],
|index, request| match index {
0 => assert!(
request.starts_with("GET /repos/alice/repo/hooks "),
"request was {request}"
),
1 => assert!(
request.starts_with("DELETE /repos/alice/repo/hooks/42 "),
"request was {request}"
),
_ => unreachable!(),
},
);
let site = SiteConfig {
api_url: Some(api_url),
..site(ProviderKind::Github, None)
};
let client = ProviderClient::new(&site).unwrap();
let removed = client
.uninstall_webhook(
&EndpointConfig {
site: "github".to_string(),
kind: NamespaceKind::User,
namespace: "alice".to_string(),
},
"repo",
"https://mirror.example.test/webhook",
)
.unwrap();
assert!(removed);
handle.join().unwrap();
}
fn site(provider: ProviderKind, git_username: Option<String>) -> SiteConfig {
SiteConfig {
name: "site".to_string(),
@@ -714,4 +1053,32 @@ mod tests {
});
(format!("http://{address}"), handle)
}
fn request_server<F>(
responses: Vec<(&'static str, &'static str)>,
mut assert_request: F,
) -> (String, thread::JoinHandle<()>)
where
F: FnMut(usize, &str) + Send + 'static,
{
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let handle = thread::spawn(move || {
for (index, (status, body)) in responses.into_iter().enumerate() {
let (mut stream, _) = listener.accept().unwrap();
let mut buffer = [0_u8; 4096];
let bytes = stream.read(&mut buffer).unwrap();
let request = String::from_utf8_lossy(&buffer[..bytes]).to_string();
assert_request(index, &request);
write!(
stream,
"HTTP/1.1 {status}\r\ncontent-type: application/json\r\nconnection: close\r\ncontent-length: {}\r\n\r\n{body}",
body.len()
)
.unwrap();
}
});
(format!("http://{address}"), handle)
}
}
src/sync.rs
+37 -18
View File
@@ -16,6 +16,7 @@ use crate::git::{
};
use crate::logging;
use crate::provider::{EndpointRepo, ProviderClient, repos_by_name};
use crate::webhook;
const FAILURE_STATE_FILE: &str = "failed-repos.toml";
const REF_STATE_FILE: &str = "ref-state.toml";
@@ -361,24 +362,14 @@ fn sync_group(
.unwrap_or(mirror.create_missing);
let allow_force = context.options.force_override.unwrap_or(mirror.allow_force);
let mut all_endpoint_repos = Vec::new();
for endpoint in &mirror.endpoints {
let site = context.config.site(&endpoint.site).unwrap();
let client = ProviderClient::new(site)?;
crate::logln!(
" {} {}",
style("list").cyan().bold(),
style(endpoint.label()).dim()
);
let repos = client
.list_repos(endpoint)
.with_context(|| format!("failed to list repos for {}", endpoint.label()))?;
for repo in repos {
all_endpoint_repos.push(EndpointRepo {
endpoint: endpoint.clone(),
repo,
});
}
let all_endpoint_repos = list_group_repos(context.config, mirror)?;
if !context.options.dry_run {
webhook::ensure_configured_webhooks(
context.config,
mirror,
&all_endpoint_repos,
context.work_dir,
)?;
}
let mut repos = repos_by_name(all_endpoint_repos);
@@ -538,9 +529,37 @@ fn sync_group(
failures
});
if create_missing && !context.options.dry_run {
let repos = list_group_repos(context.config, mirror)?;
webhook::ensure_configured_webhooks(context.config, mirror, &repos, context.work_dir)?;
}
Ok(failures)
}
fn list_group_repos(config: &Config, mirror: &MirrorConfig) -> Result<Vec<EndpointRepo>> {
let mut all_endpoint_repos = Vec::new();
for endpoint in &mirror.endpoints {
let site = config.site(&endpoint.site).unwrap();
let client = ProviderClient::new(site)?;
crate::logln!(
" {} {}",
style("list").cyan().bold(),
style(endpoint.label()).dim()
);
let repos = client
.list_repos(endpoint)
.with_context(|| format!("failed to list repos for {}", endpoint.label()))?;
for repo in repos {
all_endpoint_repos.push(EndpointRepo {
endpoint: endpoint.clone(),
repo,
});
}
}
Ok(all_endpoint_repos)
}
fn pop_repo_job(queue: &Arc<Mutex<VecDeque<RepoSyncJob>>>) -> Option<RepoSyncJob> {
queue
.lock()
src/webhook.rs
+955
View File
@@ -0,0 +1,955 @@
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::fs;
use std::path::Path;
use std::path::PathBuf;
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
use std::time::Duration;
use anyhow::{Context, Result, bail};
use console::style;
use hmac::{Hmac, Mac};
use regex::escape;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::Sha256;
use tiny_http::{Header, Method, Request, Response, Server, StatusCode};
use crate::config::{
Config, EndpointConfig, MirrorConfig, ProviderKind, default_work_dir, validate_config,
};
use crate::provider::{EndpointRepo, ProviderClient, RemoteRepo};
use crate::sync::{SyncOptions, sync_all};
type HmacSha256 = Hmac<Sha256>;
const WEBHOOK_STATE_FILE: &str = "webhook-state.toml";
#[derive(Clone, Debug)]
pub struct ServeOptions {
pub listen: String,
pub secret: String,
pub workers: usize,
pub work_dir: Option<PathBuf>,
pub full_sync_interval_minutes: Option<u64>,
pub reachability_url: Option<String>,
pub reachability_check_interval_minutes: Option<u64>,
}
#[derive(Clone, Debug)]
pub struct WebhookInstallOptions {
pub url: String,
pub secret: String,
pub group: Option<String>,
pub repo_pattern: Option<String>,
pub dry_run: bool,
pub work_dir: Option<PathBuf>,
}
#[derive(Clone, Debug)]
pub struct WebhookUninstallOptions {
pub group: Option<String>,
pub dry_run: bool,
pub work_dir: Option<PathBuf>,
}
#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
struct WebhookJob {
group: String,
repo: String,
}
#[derive(Clone)]
struct JobQueue {
sender: mpsc::Sender<WebhookJob>,
pending: Arc<Mutex<BTreeSet<WebhookJob>>>,
}
pub fn serve(config: Config, options: ServeOptions) -> Result<()> {
validate_config(&config)?;
if options.workers == 0 {
bail!("--jobs must be at least 1");
}
let server = Server::http(&options.listen)
.map_err(|error| anyhow::anyhow!("failed to listen on {}: {error}", options.listen))?;
crate::logln!(
"{} {}",
style("Webhook server").cyan().bold(),
style(&options.listen).bold()
);
let config = Arc::new(config);
let (sender, receiver) = mpsc::channel::<WebhookJob>();
let pending = Arc::new(Mutex::new(BTreeSet::<WebhookJob>::new()));
let receiver = Arc::new(Mutex::new(receiver));
let sync_lock = Arc::new(Mutex::new(()));
for worker_id in 0..options.workers {
let receiver = Arc::clone(&receiver);
let pending = Arc::clone(&pending);
let config = Arc::clone(&config);
let sync_lock = Arc::clone(&sync_lock);
let work_dir = options.work_dir.clone();
thread::spawn(move || {
worker_loop(worker_id, receiver, pending, sync_lock, config, work_dir)
});
}
if let Some(minutes) = options
.full_sync_interval_minutes
.filter(|minutes| *minutes > 0)
{
let config = Arc::clone(&config);
let sync_lock = Arc::clone(&sync_lock);
let work_dir = options.work_dir.clone();
thread::spawn(move || full_sync_timer_loop(config, sync_lock, work_dir, minutes));
}
if let Some(url) = options.reachability_url.clone() {
let minutes = options
.reachability_check_interval_minutes
.filter(|minutes| *minutes > 0)
.unwrap_or(15);
thread::spawn(move || reachability_timer_loop(url, minutes));
}
let queue = JobQueue { sender, pending };
for request in server.incoming_requests() {
let response = handle_request(request, &config, &options.secret, &queue);
if let Err(error) = response {
crate::logln!("{} {error:#}", style("webhook error").red().bold());
}
}
Ok(())
}
fn full_sync_timer_loop(
config: Arc<Config>,
sync_lock: Arc<Mutex<()>>,
work_dir: Option<PathBuf>,
minutes: u64,
) {
loop {
thread::sleep(Duration::from_secs(minutes * 60));
crate::logln!(
"{} {}",
style("full sync timer").cyan().bold(),
style(format!("every {minutes} minute(s)")).dim()
);
let _sync_guard = sync_lock
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if let Err(error) = sync_all(
&config,
SyncOptions {
work_dir: work_dir.clone(),
..SyncOptions::default()
},
) {
crate::logln!("{} {error:#}", style("full sync failed").red().bold());
}
}
}
fn reachability_timer_loop(url: String, minutes: u64) {
loop {
thread::sleep(Duration::from_secs(minutes * 60));
if let Err(error) = check_webhook_url_reachable(&url) {
crate::logln!(
"{} {}: {error:#}",
style("webhook URL unreachable").yellow().bold(),
style(&url).cyan()
);
}
}
}
pub fn install_webhooks(config: &Config, options: WebhookInstallOptions) -> Result<()> {
validate_config(config)?;
let work_dir = options.work_dir.clone().unwrap_or_else(default_work_dir);
let mut state = load_webhook_state(&work_dir)?;
let repo_pattern = options
.repo_pattern
.as_deref()
.map(regex::Regex::new)
.transpose()
.with_context(|| "invalid --repo-pattern regex")?;
for mirror in &config.mirrors {
if options
.group
.as_ref()
.is_some_and(|group| group != &mirror.name)
{
continue;
}
crate::logln!();
crate::logln!(
"{} {}",
style("Webhook group").cyan().bold(),
style(&mirror.name).bold()
);
for endpoint in &mirror.endpoints {
let site = config.site(&endpoint.site).unwrap();
let client = ProviderClient::new(site)?;
crate::logln!(
" {} {}",
style("list").cyan().bold(),
style(endpoint.label()).dim()
);
let repos = client
.list_repos(endpoint)
.with_context(|| format!("failed to list repos for {}", endpoint.label()))?;
for repo in repos {
if repo_pattern
.as_ref()
.is_some_and(|pattern| !pattern.is_match(&repo.name))
{
continue;
}
install_repo_webhook(
&WebhookInstallRequest {
client: &client,
group: &mirror.name,
endpoint,
repo: &repo,
url: &options.url,
secret: &options.secret,
dry_run: options.dry_run,
},
&mut state,
)?;
}
}
}
if !options.dry_run {
save_webhook_state(&work_dir, &state)?;
}
Ok(())
}
pub fn uninstall_webhooks(config: &Config, options: WebhookUninstallOptions) -> Result<()> {
validate_config(config)?;
let work_dir = options.work_dir.clone().unwrap_or_else(default_work_dir);
let mut state = load_webhook_state(&work_dir)?;
if state.installations.is_empty() {
crate::logln!(
"{} no webhook installations recorded",
style("skip").yellow().bold()
);
return Ok(());
}
let mut removed_keys = Vec::new();
for (key, installation) in &state.installations {
if options
.group
.as_ref()
.is_some_and(|group| group != &installation.group)
{
continue;
}
crate::logln!(
" {} {} {}",
style(if options.dry_run {
"would uninstall"
} else {
"uninstall"
})
.red()
.bold(),
style(&installation.repo).cyan(),
style(format!("from {}", installation.endpoint.label())).dim()
);
if options.dry_run {
continue;
}
let Some(site) = config.site(&installation.endpoint.site) else {
crate::logln!(
" {} {} {}",
style("skip").yellow().bold(),
style(&installation.repo).cyan(),
style(format!("unknown site {}", installation.endpoint.site)).dim()
);
continue;
};
let client = ProviderClient::new(site)?;
client
.uninstall_webhook(
&installation.endpoint,
&installation.repo,
&installation.url,
)
.with_context(|| {
format!(
"failed to uninstall webhook for {} from {}",
installation.repo,
installation.endpoint.label()
)
})?;
removed_keys.push(key.clone());
}
if !options.dry_run {
for key in removed_keys {
state.installations.remove(&key);
}
save_webhook_state(&work_dir, &state)?;
}
Ok(())
}
pub fn ensure_configured_webhooks(
config: &Config,
mirror: &MirrorConfig,
repos: &[EndpointRepo],
work_dir: &Path,
) -> Result<()> {
let Some(webhook) = &config.webhook else {
return Ok(());
};
if !webhook.install {
return Ok(());
}
let secret = webhook.secret()?;
let mut state = load_webhook_state(work_dir)?;
for endpoint_repo in repos {
let Some(site) = config.site(&endpoint_repo.endpoint.site) else {
continue;
};
let client = ProviderClient::new(site)?;
install_repo_webhook(
&WebhookInstallRequest {
client: &client,
group: &mirror.name,
endpoint: &endpoint_repo.endpoint,
repo: &endpoint_repo.repo,
url: &webhook.url,
secret: &secret,
dry_run: false,
},
&mut state,
)?;
}
save_webhook_state(work_dir, &state)
}
pub fn check_webhook_url_reachable(url: &str) -> Result<()> {
let client = reqwest::blocking::Client::builder()
.timeout(Duration::from_secs(10))
.build()?;
client
.get(url)
.send()
.with_context(|| format!("failed to reach {url}"))?;
Ok(())
}
fn worker_loop(
worker_id: usize,
receiver: Arc<Mutex<mpsc::Receiver<WebhookJob>>>,
pending: Arc<Mutex<BTreeSet<WebhookJob>>>,
sync_lock: Arc<Mutex<()>>,
config: Arc<Config>,
work_dir: Option<PathBuf>,
) {
loop {
let job = {
let receiver = receiver
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
receiver.recv()
};
let Ok(job) = job else {
return;
};
crate::logln!(
"{} {} {}",
style(format!("worker {worker_id}")).cyan().bold(),
style(&job.group).bold(),
style(&job.repo).cyan()
);
let _sync_guard = sync_lock
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let result = sync_all(
&config,
SyncOptions {
group: Some(job.group.clone()),
repo_pattern: Some(format!("^{}$", escape(&job.repo))),
work_dir: work_dir.clone(),
jobs: 1,
..SyncOptions::default()
},
);
match result {
Ok(()) => crate::logln!(
"{} {}/{}",
style("webhook sync done").green().bold(),
job.group,
job.repo
),
Err(error) => crate::logln!(
"{} {}/{}: {error:#}",
style("webhook sync failed").red().bold(),
job.group,
job.repo
),
}
pending
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.remove(&job);
}
}
#[derive(Clone, Debug, Default, Deserialize, Serialize)]
struct WebhookState {
#[serde(default)]
installations: BTreeMap<String, WebhookInstallation>,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
struct WebhookInstallation {
group: String,
endpoint: EndpointConfig,
repo: String,
url: String,
}
struct WebhookInstallRequest<'a> {
client: &'a ProviderClient<'a>,
group: &'a str,
endpoint: &'a EndpointConfig,
repo: &'a RemoteRepo,
url: &'a str,
secret: &'a str,
dry_run: bool,
}
fn install_repo_webhook(
request: &WebhookInstallRequest<'_>,
state: &mut WebhookState,
) -> Result<()> {
let key = webhook_installation_key(request.group, request.endpoint, &request.repo.name);
if state
.installations
.get(&key)
.is_some_and(|installation| installation.url == request.url)
{
return Ok(());
}
crate::logln!(
" {} {} {}",
style(if request.dry_run {
"would install"
} else {
"install"
})
.green()
.bold(),
style(&request.repo.name).cyan(),
style(format!("webhook on {}", request.endpoint.label())).dim()
);
if request.dry_run {
return Ok(());
}
request
.client
.install_webhook(request.endpoint, request.repo, request.url, request.secret)
.with_context(|| {
format!(
"failed to install webhook for {} on {}",
request.repo.name,
request.endpoint.label()
)
})?;
state.installations.insert(
key,
WebhookInstallation {
group: request.group.to_string(),
endpoint: request.endpoint.clone(),
repo: request.repo.name.clone(),
url: request.url.to_string(),
},
);
Ok(())
}
fn webhook_installation_key(group: &str, endpoint: &EndpointConfig, repo: &str) -> String {
format!(
"{}\t{}\t{:?}\t{}\t{}",
group, endpoint.site, endpoint.kind, endpoint.namespace, repo
)
}
fn load_webhook_state(work_dir: &Path) -> Result<WebhookState> {
let path = webhook_state_path(work_dir);
if !path.exists() {
return Ok(WebhookState::default());
}
let contents =
fs::read_to_string(&path).with_context(|| format!("failed to read {}", path.display()))?;
toml::from_str(&contents).with_context(|| format!("failed to parse {}", path.display()))
}
fn save_webhook_state(work_dir: &Path, state: &WebhookState) -> Result<()> {
let path = webhook_state_path(work_dir);
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
}
let contents = toml::to_string_pretty(state)?;
fs::write(&path, contents).with_context(|| format!("failed to write {}", path.display()))
}
fn webhook_state_path(work_dir: &Path) -> PathBuf {
work_dir.join(WEBHOOK_STATE_FILE)
}
fn handle_request(
mut request: Request,
config: &Config,
secret: &str,
queue: &JobQueue,
) -> Result<()> {
if request.method() != &Method::Post {
respond(request, StatusCode(405), "method not allowed")?;
return Ok(());
}
let path = request.url().split('?').next().unwrap_or(request.url());
if path != "/" && path != "/webhook" {
respond(request, StatusCode(404), "not found")?;
return Ok(());
}
let headers = headers_map(request.headers());
let mut body = Vec::new();
request
.as_reader()
.read_to_end(&mut body)
.context("failed to read webhook request body")?;
let provider = detect_provider(&headers);
if !verify_signature(provider.as_ref(), &headers, &body, secret) {
respond(request, StatusCode(401), "invalid signature")?;
return Ok(());
}
let value: Value = match serde_json::from_slice(&body) {
Ok(value) => value,
Err(_) => {
respond(request, StatusCode(400), "invalid JSON")?;
return Ok(());
}
};
let Some(event) = parse_event(provider, &headers, &value) else {
respond(request, StatusCode(202), "ignored")?;
return Ok(());
};
let jobs = matching_jobs(config, &event);
if jobs.is_empty() {
respond(request, StatusCode(202), "no matching mirror group")?;
return Ok(());
}
let mut enqueued = 0;
for job in jobs {
if enqueue(queue, job)? {
enqueued += 1;
}
}
respond(request, StatusCode(202), &format!("queued {enqueued}"))?;
Ok(())
}
fn enqueue(queue: &JobQueue, job: WebhookJob) -> Result<bool> {
let mut pending = queue
.pending
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if !pending.insert(job.clone()) {
return Ok(false);
}
if queue.sender.send(job.clone()).is_err() {
pending.remove(&job);
bail!("webhook worker queue is closed");
}
Ok(true)
}
fn respond(request: Request, status: StatusCode, body: &str) -> Result<()> {
request
.respond(Response::from_string(body.to_string()).with_status_code(status))
.map_err(|error| anyhow::anyhow!("failed to send webhook response: {error}"))
}
fn headers_map(headers: &[Header]) -> HashMap<String, String> {
headers
.iter()
.map(|header| {
(
header.field.to_string().to_ascii_lowercase(),
header.value.as_str().to_string(),
)
})
.collect()
}
#[derive(Clone, Debug, Eq, PartialEq)]
struct WebhookEvent {
provider: Option<ProviderKind>,
repo: String,
namespace: Option<String>,
}
fn detect_provider(headers: &HashMap<String, String>) -> Option<ProviderKind> {
if headers.contains_key("x-forgejo-event") {
Some(ProviderKind::Forgejo)
} else if headers.contains_key("x-gitea-event") {
Some(ProviderKind::Gitea)
} else if headers.contains_key("x-gitlab-event") {
Some(ProviderKind::Gitlab)
} else if headers.contains_key("x-github-event") {
Some(ProviderKind::Github)
} else {
None
}
}
fn parse_event(
provider: Option<ProviderKind>,
headers: &HashMap<String, String>,
value: &Value,
) -> Option<WebhookEvent> {
if !is_push_event(headers) {
return None;
}
match provider {
Some(ProviderKind::Gitlab) => parse_gitlab_event(provider, value),
Some(ProviderKind::Github)
| Some(ProviderKind::Gitea)
| Some(ProviderKind::Forgejo)
| None => parse_github_like_event(provider, value),
}
}
fn is_push_event(headers: &HashMap<String, String>) -> bool {
let github = headers
.get("x-github-event")
.is_some_and(|event| event == "push");
let gitea = headers
.get("x-gitea-event")
.is_some_and(|event| event == "push");
let forgejo = headers
.get("x-forgejo-event")
.is_some_and(|event| event == "push");
let gitlab = headers
.get("x-gitlab-event")
.is_some_and(|event| event == "Push Hook" || event == "Tag Push Hook");
github || gitea || forgejo || gitlab
}
fn parse_github_like_event(provider: Option<ProviderKind>, value: &Value) -> Option<WebhookEvent> {
let repo = value.pointer("/repository/name")?.as_str()?.to_string();
let namespace = value
.pointer("/repository/owner/login")
.or_else(|| value.pointer("/repository/owner/username"))
.or_else(|| value.pointer("/repository/owner/name"))
.and_then(Value::as_str)
.map(ToOwned::to_owned)
.or_else(|| {
value
.pointer("/repository/full_name")
.and_then(Value::as_str)
.and_then(|full_name| {
full_name
.rsplit_once('/')
.map(|(owner, _)| owner.to_string())
})
});
Some(WebhookEvent {
provider,
repo,
namespace,
})
}
fn parse_gitlab_event(provider: Option<ProviderKind>, value: &Value) -> Option<WebhookEvent> {
let path = value.pointer("/project/path")?.as_str()?.to_string();
let namespace = value
.pointer("/project/path_with_namespace")
.and_then(Value::as_str)
.and_then(|path| {
path.rsplit_once('/')
.map(|(namespace, _)| namespace.to_string())
})
.or_else(|| {
value
.pointer("/project/namespace")
.and_then(Value::as_str)
.map(ToOwned::to_owned)
});
Some(WebhookEvent {
provider,
repo: path,
namespace,
})
}
fn matching_jobs(config: &Config, event: &WebhookEvent) -> Vec<WebhookJob> {
config
.mirrors
.iter()
.filter(|mirror| {
mirror.endpoints.iter().any(|endpoint| {
let Some(site) = config.site(&endpoint.site) else {
return false;
};
event
.provider
.as_ref()
.is_none_or(|provider| &site.provider == provider)
&& event
.namespace
.as_ref()
.is_none_or(|namespace| namespace == &endpoint.namespace)
})
})
.map(|mirror| WebhookJob {
group: mirror.name.clone(),
repo: event.repo.clone(),
})
.collect()
}
fn verify_signature(
provider: Option<&ProviderKind>,
headers: &HashMap<String, String>,
body: &[u8],
secret: &str,
) -> bool {
match provider {
Some(ProviderKind::Gitlab) => headers
.get("x-gitlab-token")
.is_some_and(|token| fixed_time_eq(token.as_bytes(), secret.as_bytes())),
Some(ProviderKind::Github) => {
verify_hmac_header(headers, "x-hub-signature-256", body, secret)
}
Some(ProviderKind::Gitea) | Some(ProviderKind::Forgejo) => {
verify_hmac_header(headers, "x-gitea-signature", body, secret)
|| verify_hmac_header(headers, "x-forgejo-signature", body, secret)
|| verify_hmac_header(headers, "x-hub-signature-256", body, secret)
}
None => false,
}
}
fn verify_hmac_header(
headers: &HashMap<String, String>,
header: &str,
body: &[u8],
secret: &str,
) -> bool {
let Some(signature) = headers.get(header) else {
return false;
};
let expected = hmac_sha256_hex(secret.as_bytes(), body);
let signature = signature
.trim()
.strip_prefix("sha256=")
.unwrap_or_else(|| signature.trim());
fixed_time_eq(signature.as_bytes(), expected.as_bytes())
}
fn hmac_sha256_hex(secret: &[u8], body: &[u8]) -> String {
let mut mac = HmacSha256::new_from_slice(secret).expect("HMAC accepts any key length");
mac.update(body);
let bytes = mac.finalize().into_bytes();
let mut output = String::with_capacity(bytes.len() * 2);
for byte in bytes {
output.push_str(&format!("{byte:02x}"));
}
output
}
fn fixed_time_eq(left: &[u8], right: &[u8]) -> bool {
if left.len() != right.len() {
return false;
}
let mut diff = 0_u8;
for (left, right) in left.iter().zip(right) {
diff |= left ^ right;
}
diff == 0
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::{
EndpointConfig, MirrorConfig, NamespaceKind, SiteConfig, TokenConfig, Visibility,
};
#[test]
fn verifies_github_hmac_signature() {
let body = br#"{"repository":{"name":"repo"}}"#;
let mut headers = HashMap::new();
headers.insert("x-github-event".to_string(), "push".to_string());
headers.insert(
"x-hub-signature-256".to_string(),
format!("sha256={}", hmac_sha256_hex(b"secret", body)),
);
assert!(verify_signature(
Some(&ProviderKind::Github),
&headers,
body,
"secret"
));
assert!(!verify_signature(
Some(&ProviderKind::Github),
&headers,
body,
"wrong"
));
}
#[test]
fn parses_github_push_payload() {
let mut headers = HashMap::new();
headers.insert("x-github-event".to_string(), "push".to_string());
let value: Value = serde_json::from_str(
r#"{"repository":{"name":"repo","full_name":"alice/repo","owner":{"login":"alice"}}}"#,
)
.unwrap();
let event = parse_event(Some(ProviderKind::Github), &headers, &value).unwrap();
assert_eq!(event.repo, "repo");
assert_eq!(event.namespace.as_deref(), Some("alice"));
}
#[test]
fn parses_forgejo_push_payload() {
let mut headers = HashMap::new();
headers.insert("x-forgejo-event".to_string(), "push".to_string());
let value: Value = serde_json::from_str(
r#"{"repository":{"name":"repo","full_name":"azalea/repo","owner":{"username":"azalea"}}}"#,
)
.unwrap();
let provider = detect_provider(&headers);
let event = parse_event(provider.clone(), &headers, &value).unwrap();
assert_eq!(provider, Some(ProviderKind::Forgejo));
assert_eq!(event.repo, "repo");
assert_eq!(event.namespace.as_deref(), Some("azalea"));
}
#[test]
fn verifies_forgejo_hmac_signature() {
let body = br#"{"repository":{"name":"repo"}}"#;
let mut headers = HashMap::new();
headers.insert("x-forgejo-event".to_string(), "push".to_string());
headers.insert(
"x-forgejo-signature".to_string(),
format!("sha256={}", hmac_sha256_hex(b"secret", body)),
);
assert!(verify_signature(
Some(&ProviderKind::Forgejo),
&headers,
body,
"secret"
));
}
#[test]
fn parses_gitlab_push_payload() {
let mut headers = HashMap::new();
headers.insert("x-gitlab-event".to_string(), "Push Hook".to_string());
let value: Value = serde_json::from_str(
r#"{"project":{"path":"repo","path_with_namespace":"parent/alice/repo"}}"#,
)
.unwrap();
let event = parse_event(Some(ProviderKind::Gitlab), &headers, &value).unwrap();
assert_eq!(event.repo, "repo");
assert_eq!(event.namespace.as_deref(), Some("parent/alice"));
}
#[test]
fn matches_jobs_by_provider_and_namespace() {
let config = Config {
sites: vec![
site("github", ProviderKind::Github),
site("gitea", ProviderKind::Gitea),
],
mirrors: vec![MirrorConfig {
name: "sync-1".to_string(),
endpoints: vec![
endpoint("github", NamespaceKind::User, "alice"),
endpoint("gitea", NamespaceKind::User, "azalea"),
],
create_missing: true,
visibility: Visibility::Private,
allow_force: false,
}],
webhook: None,
};
let event = WebhookEvent {
provider: Some(ProviderKind::Github),
repo: "repo".to_string(),
namespace: Some("alice".to_string()),
};
let jobs = matching_jobs(&config, &event);
assert_eq!(jobs.len(), 1);
assert_eq!(jobs[0].group, "sync-1");
assert_eq!(jobs[0].repo, "repo");
}
#[test]
fn webhook_state_persists_installations() {
let temp = tempfile::TempDir::new().unwrap();
let endpoint = endpoint("github", NamespaceKind::User, "alice");
let key = webhook_installation_key("sync-1", &endpoint, "repo");
let mut state = WebhookState::default();
state.installations.insert(
key.clone(),
WebhookInstallation {
group: "sync-1".to_string(),
endpoint,
repo: "repo".to_string(),
url: "https://mirror.example.test/webhook".to_string(),
},
);
save_webhook_state(temp.path(), &state).unwrap();
let loaded = load_webhook_state(temp.path()).unwrap();
assert_eq!(loaded.installations[&key].repo, "repo");
assert_eq!(
loaded.installations[&key].url,
"https://mirror.example.test/webhook"
);
}
fn site(name: &str, provider: ProviderKind) -> SiteConfig {
SiteConfig {
name: name.to_string(),
provider,
base_url: "https://example.test".to_string(),
api_url: None,
token: TokenConfig::Value("secret".to_string()),
git_username: None,
}
}
fn endpoint(site: &str, kind: NamespaceKind, namespace: &str) -> EndpointConfig {
EndpointConfig {
site: site.to_string(),
kind,
namespace: namespace.to_string(),
}
}
}