forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
16.2 KiB
1//! Repo mirrors in the web process: the sync worker (see `worker`), the push debounce, the
2//! pull-side ref diff, and the SSRF checks run before every sync.
3//!
4//! A pull mirror is fetched with `forge_git::mirror::mirror_fetch`, then every branch and tag that
5//! moved is passed to `post_update::after_ref_update` with the system actor, so webhooks and pull
6//! request sync behave as they do after a push. A push mirror runs `mirror_push`.
7
8pub mod worker;
9
10use std::collections::BTreeMap;
11use std::future::Future;
12use std::net::IpAddr;
13use std::path::Path;
14use std::time::Duration;
15
16use forge_core::mirrors::{Mirror, MirrorDirection};
17use forge_core::time::{now_stamp, stamp_after};
18use forge_core::{AuditAction, Repo};
19use forge_git::command::GitCommand;
20use forge_git::mirror::{PinnedHost, RemoteAuth, mirror_fetch, mirror_push};
21use reqwest::Url;
22use serde_json::json;
23
24use crate::audit::{self, Actor, Target};
25use crate::post_update::after_ref_update;
26use crate::state::AppState;
27use crate::webhooks::client::{WebhookPolicy, is_forbidden_ip, validate_url};
28
29/// How long a push waits after the last ref update before it is sent.
30pub const PUSH_DEBOUNCE: Duration = Duration::from_secs(10);
31/// Hard limit on one sync run. The git process is killed when it passes.
32pub const RUN_TIMEOUT: Duration = Duration::from_secs(10 * 60);
33/// The all-zero object id git uses for a ref that does not exist on one side.
34pub const ZERO_OID: &str = "0000000000000000000000000000000000000000";
35/// Next run for a mirror whose interval is `0` (manual only).
36const FAR_FUTURE: &str = "9999-12-31T23:59:59.999999999Z";
37
38/// Queues every enabled push mirror of a repo for a sync `PUSH_DEBOUNCE` from now, and wakes the
39/// worker. Called after a branch or tag update.
40pub async fn enqueue_push_mirrors(state: &AppState, repo_id: &str) -> forge_core::Result<()> {
41 let due = stamp_after(PUSH_DEBOUNCE);
42 let mut queued = false;
43 for mirror in state.store.list_mirrors(repo_id).await? {
44 if mirror.direction == MirrorDirection::Push && mirror.enabled {
45 state.store.schedule_mirror(&mirror.id, &due).await?;
46 queued = true;
47 }
48 }
49 if queued {
50 state.mirror_waker.notify_one();
51 }
52 Ok(())
53}
54
55/// The policy the mirror worker runs under, from the `[webhooks]` config. An invalid config falls
56/// back to the strictest policy: no internal hosts and no plain HTTP.
57pub fn policy(state: &AppState) -> WebhookPolicy {
58 WebhookPolicy::from_config(&state.config.webhooks).unwrap_or_else(|e| {
59 tracing::error!(error = %e, "mirrors: invalid [webhooks] config, using the default policy");
60 WebhookPolicy {
61 allow_http: false,
62 allowed_hosts: Vec::new(),
63 }
64 })
65}
66
67/// Checks a mirror URL against the webhook policy before a run: the URL shape, and every address
68/// the host resolves to. Returns the address git must connect to (see [`pin_checked`]). `None`
69/// when there is nothing to pin: an address literal was checked by `validate_url`, and a host in
70/// `allowed_hosts` is trusted.
71pub async fn check_remote(url: &str, policy: &WebhookPolicy) -> Result<Option<PinnedHost>, String> {
72 let parsed = validate_url(url, policy).map_err(|e| format!("mirror URL refused: {e}"))?;
73 let host = parsed.host_str().unwrap_or_default();
74 let bare = host.trim_start_matches('[').trim_end_matches(']');
75 if bare.parse::<IpAddr>().is_ok() || policy.host_allowed(host) {
76 return Ok(None);
77 }
78 let port = parsed.port_or_known_default().unwrap_or(443);
79 // One lookup for the whole run: the addresses checked here are the ones git is pinned to.
80 let resolved: Vec<IpAddr> = tokio::net::lookup_host((host, port))
81 .await
82 .map_err(|e| format!("cannot resolve the mirror host: {e}"))?
83 .map(|addr| addr.ip())
84 .collect();
85 pin_checked(host, port, &resolved, policy).map(Some)
86}
87
88/// Checks every resolved address of `host` and pins the first one. One forbidden address refuses
89/// the whole host, so a mixed answer cannot slip a private address past the check.
90pub fn pin_checked(
91 host: &str,
92 port: u16,
93 resolved: &[IpAddr],
94 policy: &WebhookPolicy,
95) -> Result<PinnedHost, String> {
96 let Some(ip) = resolved.first().copied() else {
97 return Err("the mirror host did not resolve".into());
98 };
99 if resolved
100 .iter()
101 .any(|ip| is_forbidden_ip(*ip) && !policy.ip_allowed(*ip))
102 {
103 return Err("mirror host resolves to a forbidden address".into());
104 }
105 Ok(PinnedHost {
106 host: host.to_string(),
107 port,
108 ip,
109 })
110}
111
112/// The URL with userinfo and query removed: scheme, host (and port) and path. Used in audit
113/// entries and anywhere a URL is shown.
114pub fn redact_url(url: &str) -> String {
115 match Url::parse(url) {
116 Ok(parsed) => {
117 let host = parsed.host_str().unwrap_or_default();
118 let port = parsed.port().map(|p| format!(":{p}")).unwrap_or_default();
119 format!("{}://{host}{port}{}", parsed.scheme(), parsed.path())
120 }
121 Err(_) => "[invalid url]".to_string(),
122 }
123}
124
125/// The credentials a run presents. `None` when the mirror has neither a user nor a password.
126fn auth_for(mirror: &Mirror, secret: Option<Vec<u8>>) -> Option<RemoteAuth> {
127 match secret {
128 Some(secret) => Some(RemoteAuth {
129 username: mirror.username.clone(),
130 secret,
131 }),
132 None if !mirror.username.is_empty() => Some(RemoteAuth {
133 username: mirror.username.clone(),
134 secret: Vec::new(),
135 }),
136 None => None,
137 }
138}
139
140/// Every branch and tag of the bare repo at `repo_dir`, as full ref name to object id.
141pub async fn ref_snapshot(
142 state: &AppState,
143 repo_dir: &Path,
144) -> Result<BTreeMap<String, String>, String> {
145 let out = GitCommand::new(&state.git, "for-each-ref")
146 .git_dir(repo_dir)
147 .arg("--format=%(objectname) %(refname)")
148 .arg("refs/heads/")
149 .arg("refs/tags/")
150 .output()
151 .await
152 .map_err(|e| e.to_string())?;
153 let mut refs = BTreeMap::new();
154 for line in String::from_utf8_lossy(&out).lines() {
155 if let Some((oid, name)) = line.split_once(' ') {
156 refs.insert(name.to_string(), oid.to_string());
157 }
158 }
159 Ok(refs)
160}
161
162/// The refs that differ between two snapshots as `(name, old, new)`. A new ref has the zero old
163/// id; a removed ref has the zero new id.
164pub fn changed_refs(
165 before: &BTreeMap<String, String>,
166 after: &BTreeMap<String, String>,
167) -> Vec<(String, String, String)> {
168 let mut changes = Vec::new();
169 for (name, new) in after {
170 match before.get(name) {
171 Some(old) if old == new => {}
172 Some(old) => changes.push((name.clone(), old.clone(), new.clone())),
173 None => changes.push((name.clone(), ZERO_OID.to_string(), new.clone())),
174 }
175 }
176 for (name, old) in before {
177 if !after.contains_key(name) {
178 changes.push((name.clone(), old.clone(), ZERO_OID.to_string()));
179 }
180 }
181 changes
182}
183
184/// Snapshots the refs, runs `fetch`, snapshots again, and passes every changed ref to
185/// `after_ref_update` as the system actor. Returns how many refs changed. A failed `after_ref_update`
186/// is logged and does not stop the others.
187pub async fn pull_refs<F>(state: &AppState, repo: &Repo, fetch: F) -> Result<usize, String>
188where
189 F: Future<Output = Result<(), String>>,
190{
191 let repo_dir =
192 forge_git::repo::repo_path(&state.git.data_dir, &repo.id).map_err(|e| e.to_string())?;
193 let before = ref_snapshot(state, &repo_dir).await?;
194 fetch.await?;
195 let after = ref_snapshot(state, &repo_dir).await?;
196 let changes = changed_refs(&before, &after);
197 for (name, old, new) in &changes {
198 if let Err(e) = after_ref_update(state, repo, "", name, old, new).await {
199 tracing::error!(error = %e, repo_id = %repo.id, refname = %name, "mirrors: post-update work failed");
200 }
201 }
202 Ok(changes.len())
203}
204
205/// One sync of one mirror, without recording it. `Err` is the scrubbed error text.
206async fn sync_one(state: &AppState, mirror: &Mirror, policy: &WebhookPolicy) -> Result<(), String> {
207 let repo = state
208 .store
209 .repo_by_id(&mirror.repo_id)
210 .await
211 .map_err(|e| e.to_string())?
212 .ok_or("the repository is gone")?;
213 let secret = state
214 .store
215 .mirror_secret(&mirror.id)
216 .await
217 .map_err(|e| e.to_string())?;
218 let auth = auth_for(mirror, secret);
219 let pin = check_remote(&mirror.url, policy).await?;
220 let repo_dir =
221 forge_git::repo::repo_path(&state.git.data_dir, &repo.id).map_err(|e| e.to_string())?;
222 let allow_http = policy.allow_http;
223 match mirror.direction {
224 MirrorDirection::Pull => {
225 let fetch = async {
226 mirror_fetch(
227 &state.git,
228 &repo_dir,
229 &mirror.url,
230 auth.as_ref(),
231 allow_http,
232 pin.as_ref(),
233 RUN_TIMEOUT,
234 )
235 .await
236 .map_err(|e| e.to_string())
237 };
238 pull_refs(state, &repo, fetch).await.map(|_| ())
239 }
240 MirrorDirection::Push => mirror_push(
241 &state.git,
242 &repo_dir,
243 &mirror.url,
244 auth.as_ref(),
245 allow_http,
246 pin.as_ref(),
247 RUN_TIMEOUT,
248 )
249 .await
250 .map_err(|e| e.to_string()),
251 }
252}
253
254/// When the next run of a mirror is due, from now.
255fn next_run_at(interval_minutes: u32) -> String {
256 if interval_minutes == 0 {
257 return FAR_FUTURE.to_string();
258 }
259 stamp_after(Duration::from_secs(u64::from(interval_minutes) * 60))
260}
261
262/// Runs one sync and records the result. Audits only the transition from not-failed to failed.
263/// A debounce or "sync now" that arrives during the run keeps its earlier due time.
264pub async fn run_one(state: &AppState, mirror: &Mirror, policy: &WebhookPolicy) {
265 let started = now_stamp();
266 let outcome = sync_one(state, mirror, policy).await;
267 let ok = outcome.is_ok();
268 let error = outcome.err();
269 let mut next = next_run_at(mirror.interval_minutes);
270 if let Ok(Some(current)) = state.store.mirror(&mirror.id).await
271 && current.next_sync_at > started
272 && current.next_sync_at < next
273 {
274 next = current.next_sync_at;
275 }
276 if let Err(e) = state
277 .store
278 .record_mirror_result(&mirror.id, ok, error.as_deref(), &next)
279 .await
280 {
281 tracing::error!(error = %e, mirror_id = %mirror.id, "mirrors: could not record the sync result");
282 }
283 if let Some(error) = &error {
284 tracing::warn!(mirror_id = %mirror.id, error = %error, "mirrors: sync failed");
285 if mirror.last_ok != Some(false) {
286 let data = json!({
287 "direction": mirror.direction.as_str(),
288 "url": redact_url(&mirror.url),
289 "error": error,
290 });
291 audit::record(
292 &state.store,
293 None,
294 Actor::Anonymous,
295 AuditAction::MirrorSyncFailed,
296 Target {
297 kind: "mirror",
298 id: &mirror.id,
299 name: &redact_url(&mirror.url),
300 },
301 data,
302 )
303 .await;
304 }
305 }
306}
307
308#[cfg(test)]
309mod tests {
310 use super::*;
311
312 fn refs(pairs: &[(&str, &str)]) -> BTreeMap<String, String> {
313 pairs
314 .iter()
315 .map(|(k, v)| (k.to_string(), v.to_string()))
316 .collect()
317 }
318
319 #[test]
320 fn changed_refs_finds_new_moved_and_removed() {
321 let before = refs(&[
322 ("refs/heads/main", "a"),
323 ("refs/heads/gone", "b"),
324 ("refs/tags/v1", "c"),
325 ]);
326 let after = refs(&[
327 ("refs/heads/main", "d"),
328 ("refs/heads/new", "e"),
329 ("refs/tags/v1", "c"),
330 ]);
331 let mut changes = changed_refs(&before, &after);
332 changes.sort();
333 assert_eq!(
334 changes,
335 vec![
336 (
337 "refs/heads/gone".to_string(),
338 "b".to_string(),
339 ZERO_OID.to_string()
340 ),
341 (
342 "refs/heads/main".to_string(),
343 "a".to_string(),
344 "d".to_string()
345 ),
346 (
347 "refs/heads/new".to_string(),
348 ZERO_OID.to_string(),
349 "e".to_string()
350 ),
351 ]
352 );
353 assert!(changed_refs(&after, &after).is_empty());
354 }
355
356 #[test]
357 fn redacted_url_drops_userinfo_and_query() {
358 assert_eq!(
359 redact_url("https://user:pw@example.com:8443/a/b.git?token=x"),
360 "https://example.com:8443/a/b.git"
361 );
362 }
363
364 #[test]
365 fn auth_needs_a_user_or_a_secret() {
366 let mut mirror = sample_mirror();
367 assert!(auth_for(&mirror, None).is_none());
368 mirror.username = "bot".into();
369 let anon = auth_for(&mirror, None).expect("user only");
370 assert!(anon.secret.is_empty());
371 let full = auth_for(&mirror, Some(b"tok".to_vec())).expect("with secret");
372 assert_eq!(full.secret, b"tok".to_vec());
373 }
374
375 #[test]
376 fn manual_interval_is_far_future() {
377 assert_eq!(next_run_at(0), FAR_FUTURE);
378 assert!(next_run_at(480).as_str() < FAR_FUTURE);
379 }
380
381 #[test]
382 fn pin_uses_the_first_checked_address() {
383 let policy = WebhookPolicy {
384 allow_http: false,
385 allowed_hosts: Vec::new(),
386 };
387 let public_v4: IpAddr = "93.184.216.34".parse().expect("ip");
388 let public_v6: IpAddr = "2606:2800:220:1::1".parse().expect("ip");
389 let pinned = pin_checked("example.com", 443, &[public_v4, public_v6], &policy)
390 .expect("public addresses pass");
391 assert_eq!(pinned.ip, public_v4);
392 assert_eq!(pinned.port, 443);
393 assert_eq!(
394 pinned.config_pair().expect("pair"),
395 "http.curloptResolve=example.com:443:93.184.216.34"
396 );
397 let pinned_v6 = pin_checked("example.com", 80, &[public_v6], &policy).expect("v6");
398 assert_eq!(
399 pinned_v6.config_pair().expect("pair"),
400 "http.curloptResolve=example.com:80:[2606:2800:220:1::1]"
401 );
402 // A private address anywhere in the answer refuses the host, even when listed second.
403 let loopback: IpAddr = "127.0.0.1".parse().expect("ip");
404 assert!(pin_checked("example.com", 443, &[public_v4, loopback], &policy).is_err());
405 assert!(pin_checked("example.com", 443, &[loopback, public_v4], &policy).is_err());
406 assert!(pin_checked("example.com", 443, &[], &policy).is_err());
407 }
408
409 #[tokio::test]
410 async fn remote_checks_refuse_loopback_and_userinfo() {
411 let policy = WebhookPolicy {
412 allow_http: true,
413 allowed_hosts: Vec::new(),
414 };
415 assert!(
416 check_remote("https://127.0.0.1/x.git", &policy)
417 .await
418 .is_err()
419 );
420 assert!(check_remote("https://[::1]/x.git", &policy).await.is_err());
421 assert!(
422 check_remote("https://u:p@example.com/x.git", &policy)
423 .await
424 .is_err()
425 );
426 assert!(check_remote("file:///etc/passwd", &policy).await.is_err());
427 let allowed = WebhookPolicy {
428 allow_http: false,
429 allowed_hosts: vec![crate::webhooks::client::AllowEntry::Net(
430 "127.0.0.0/8".parse().expect("cidr"),
431 )],
432 };
433 assert!(
434 check_remote("https://127.0.0.1/x.git", &allowed)
435 .await
436 .is_ok()
437 );
438 }
439
440 pub(super) fn sample_mirror() -> Mirror {
441 Mirror {
442 id: "m1".into(),
443 repo_id: "r1".into(),
444 direction: MirrorDirection::Pull,
445 url: "https://example.com/x.git".into(),
446 username: String::new(),
447 has_secret: false,
448 interval_minutes: 480,
449 enabled: true,
450 next_sync_at: now_stamp(),
451 last_sync_at: None,
452 last_ok: None,
453 last_error: None,
454 created_at: now_stamp(),
455 }
456 }
457}