forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
21.3 KiB
1//! Smart-HTTP routes: info/refs, git-upload-pack and git-receive-pack.
2//! Streams bodies to and from `git <svc> --stateless-rpc`; gzip request bodies are decoded.
3//!
4//! Compression must be off for these responses. Topcoat's compression is configured only on the
5//! router builder (`Router::builder().compression(Compression::off())`), so boot.rs has to set it.
6
7use std::net::IpAddr;
8use std::time::Instant;
9
10use async_compression::tokio::bufread::GzipDecoder;
11use forge_core::User;
12use forge_git::{GitService, pktline};
13use futures_util::TryStreamExt;
14use http::header::{
15 AUTHORIZATION, CACHE_CONTROL, CONTENT_ENCODING, CONTENT_TYPE, HeaderValue, RETRY_AFTER,
16 WWW_AUTHENTICATE,
17};
18use http_body::Frame;
19use http_body_util::StreamBody;
20use tokio::io::AsyncReadExt;
21use tokio::process::{Child, ChildStdin};
22use tokio_util::io::{ReaderStream, StreamReader};
23use topcoat::context::{Cx, app_context};
24use topcoat::router::request::headers;
25use topcoat::router::response::Response;
26use topcoat::router::{Body, HeaderMap, StatusCode, path_param, query_params, route};
27
28use crate::access::{AccessError, Transport, find_repo, git_process, resolve_repo};
29use crate::client_ip::client_ip;
30use crate::git_auth::{parse_basic_auth, user_from_basic};
31use crate::state::AppState;
32use crate::throttle::{self, GIT_IP_LIMIT};
33
34path_param!(owner);
35path_param!(repo);
36
37#[query_params(error = bad_request)]
38struct InfoRefsQuery {
39 service: Option<String>,
40}
41
42#[route(GET "/{owner}/{repo}/info/refs")]
43async fn info_refs(cx: &Cx) -> topcoat::Result<Response> {
44 let query = query_params::<InfoRefsQuery>(cx)?;
45 let Some(service) = query.service.as_deref().and_then(GitService::from_name) else {
46 return Ok(text(StatusCode::FORBIDDEN, "dumb protocol not supported"));
47 };
48
49 let state = app_context::<AppState>(cx);
50 let headers = headers(cx);
51 let owner = path_param::<Owner>(cx);
52 let repo_name = path_param::<Repo>(cx);
53
54 let (actor, repo, path) =
55 match caller_repo(state, client_ip(cx), headers, owner, repo_name, service).await {
56 Ok(found) => found,
57 Err(response) => return Ok(*response),
58 };
59
60 let protocol = git_protocol(headers);
61 let advertisement = git_process(
62 state,
63 &repo,
64 &path,
65 actor.as_ref(),
66 service,
67 Transport::Http,
68 true,
69 protocol,
70 )
71 .output()
72 .await;
73 let refs = match advertisement {
74 Ok(refs) => refs,
75 Err(e) => {
76 tracing::error!(error = %e, service = service.name(), "info/refs failed");
77 return Ok(internal_error());
78 }
79 };
80
81 let mut payload = Vec::with_capacity(refs.len() + 64);
82 if needs_service_header(service, wants_v2(protocol)) {
83 payload.extend(pktline::service_advertisement_header(service));
84 }
85 payload.extend(refs);
86
87 Ok(respond(
88 StatusCode::OK,
89 advertisement_content_type(service),
90 Body::from(payload),
91 ))
92}
93
94#[route(POST "/{owner}/{repo}/git-upload-pack")]
95async fn upload_pack(cx: &Cx, body: Body) -> topcoat::Result<Response> {
96 Ok(serve_rpc(cx, GitService::UploadPack, body).await)
97}
98
99#[route(POST "/{owner}/{repo}/git-receive-pack")]
100async fn receive_pack(cx: &Cx, body: Body) -> topcoat::Result<Response> {
101 Ok(serve_rpc(cx, GitService::ReceivePack, body).await)
102}
103
104/// Shared body of the two POST routes: authorize, spawn `git <svc> --stateless-rpc`, pipe the
105/// request body into its stdin and stream its stdout back.
106async fn serve_rpc(cx: &Cx, service: GitService, body: Body) -> Response {
107 let state = app_context::<AppState>(cx);
108 let headers = headers(cx);
109 let owner = path_param::<Owner>(cx);
110 let repo_name = path_param::<Repo>(cx);
111
112 let (actor, repo, path) =
113 match caller_repo(state, client_ip(cx), headers, owner, repo_name, service).await {
114 Ok(found) => found,
115 Err(response) => return *response,
116 };
117
118 let gzip = match body_encoding(
119 headers
120 .get(CONTENT_ENCODING)
121 .map(|v| v.to_str().unwrap_or("")),
122 ) {
123 Some(gzip) => gzip,
124 None => {
125 return text(
126 StatusCode::UNSUPPORTED_MEDIA_TYPE,
127 "unsupported content encoding",
128 );
129 }
130 };
131
132 let mut command = git_process(
133 state,
134 &repo,
135 &path,
136 actor.as_ref(),
137 service,
138 Transport::Http,
139 false,
140 git_protocol(headers),
141 )
142 .into_tokio();
143 command
144 .stdin(std::process::Stdio::piped())
145 .stdout(std::process::Stdio::piped())
146 .stderr(std::process::Stdio::piped());
147
148 let mut child = match command.spawn() {
149 Ok(child) => child,
150 Err(e) => {
151 tracing::error!(error = %e, service = service.name(), "failed to spawn git");
152 return internal_error();
153 }
154 };
155 let (Some(stdin), Some(stdout)) = (child.stdin.take(), child.stdout.take()) else {
156 tracing::error!(service = service.name(), "git child is missing piped stdio");
157 return internal_error();
158 };
159
160 tokio::spawn(pump_request(body, gzip, stdin));
161 tokio::spawn(reap(child, service));
162
163 let chunks = ReaderStream::new(stdout).map_ok(Frame::data);
164 respond(
165 StatusCode::OK,
166 result_content_type(service),
167 Body::new(StreamBody::new(chunks)),
168 )
169}
170
171/// Copies the request body (decoding gzip when `gzip` is set) into git's stdin, then closes it.
172async fn pump_request(body: Body, gzip: bool, mut stdin: ChildStdin) {
173 let reader = StreamReader::new(body.into_data_stream().map_err(std::io::Error::other));
174 let copied = if gzip {
175 tokio::io::copy(&mut GzipDecoder::new(reader), &mut stdin).await
176 } else {
177 let mut reader = reader;
178 tokio::io::copy(&mut reader, &mut stdin).await
179 };
180 if let Err(e) = copied {
181 tracing::debug!(error = %e, "request body ended before git consumed it");
182 }
183 // Closing stdin tells git the request is complete.
184 let _ = tokio::io::AsyncWriteExt::shutdown(&mut stdin).await;
185}
186
187/// Drains stderr into the log and waits for the child, so no zombie is left behind.
188async fn reap(mut child: Child, service: GitService) {
189 let mut stderr_buf = Vec::new();
190 if let Some(mut stderr) = child.stderr.take() {
191 let _ = stderr.read_to_end(&mut stderr_buf).await;
192 }
193 let stderr_text = String::from_utf8_lossy(&stderr_buf);
194 match child.wait().await {
195 Ok(status) if status.success() => {
196 tracing::debug!(service = service.name(), "git finished");
197 }
198 Ok(status) => {
199 tracing::warn!(
200 service = service.name(),
201 %status,
202 stderr = %stderr_text.trim(),
203 "git exited with failure"
204 );
205 }
206 Err(e) => {
207 tracing::warn!(service = service.name(), error = %e, "failed to wait for git");
208 }
209 }
210}
211
212/// Extracts a `Git-Protocol` header value only if it matches `^[a-z0-9=:,._-]{1,64}$`.
213fn git_protocol(headers: &HeaderMap) -> Option<&str> {
214 let value = headers.get("git-protocol")?.to_str().ok()?;
215 valid_git_protocol(value).then_some(value)
216}
217
218fn valid_git_protocol(value: &str) -> bool {
219 (1..=64).contains(&value.len())
220 && value.bytes().all(|b| {
221 b.is_ascii_lowercase()
222 || b.is_ascii_digit()
223 || matches!(b, b'=' | b':' | b',' | b'.' | b'_' | b'-')
224 })
225}
226
227/// True when the protocol string asks for version 2.
228fn wants_v2(protocol: Option<&str>) -> bool {
229 protocol.is_some_and(|p| p.split(':').any(|kv| kv == "version=2"))
230}
231
232/// The `# service=` preamble is required for v0 advertisements. With v2 the upload-pack server
233/// writes its own version line instead. Receive-pack has no v2, so it always gets the preamble.
234fn needs_service_header(service: GitService, v2: bool) -> bool {
235 !(service == GitService::UploadPack && v2)
236}
237
238/// Decodes the request body encoding. `Some(true)` for gzip, `Some(false)` for none or identity,
239/// `None` for anything else.
240fn body_encoding(value: Option<&str>) -> Option<bool> {
241 match value.map(|v| v.trim().to_ascii_lowercase()) {
242 None => Some(false),
243 Some(v) if v == "gzip" => Some(true),
244 Some(v) if v == "identity" => Some(false),
245 Some(_) => None,
246 }
247}
248
249fn advertisement_content_type(service: GitService) -> &'static str {
250 match service {
251 GitService::UploadPack => "application/x-git-upload-pack-advertisement",
252 GitService::ReceivePack => "application/x-git-receive-pack-advertisement",
253 }
254}
255
256fn result_content_type(service: GitService) -> &'static str {
257 match service {
258 GitService::UploadPack => "application/x-git-upload-pack-result",
259 GitService::ReceivePack => "application/x-git-receive-pack-result",
260 }
261}
262
263fn respond(status: StatusCode, content_type: &'static str, body: Body) -> Response {
264 let mut res = Response::new(body);
265 *res.status_mut() = status;
266 res.headers_mut()
267 .insert(CONTENT_TYPE, HeaderValue::from_static(content_type));
268 res.headers_mut()
269 .insert(CACHE_CONTROL, HeaderValue::from_static("no-cache"));
270 res
271}
272
273fn text(status: StatusCode, message: &'static str) -> Response {
274 let mut res = Response::new(Body::from(message));
275 *res.status_mut() = status;
276 res.headers_mut().insert(
277 CONTENT_TYPE,
278 HeaderValue::from_static("text/plain; charset=utf-8"),
279 );
280 res
281}
282
283fn internal_error() -> Response {
284 text(StatusCode::INTERNAL_SERVER_ERROR, "internal server error")
285}
286
287/// Authenticates the request's credentials under the per-IP git failure limit. Only rejected
288/// credentials count, so anonymous requests and successful logins never do. Returns the response
289/// to send when the request must stop: 429 while the IP is throttled, or the access error.
290pub(crate) async fn authenticate(
291 state: &AppState,
292 ip: IpAddr,
293 headers: &HeaderMap,
294) -> Result<Option<User>, Box<Response>> {
295 let key = throttle::ip_key("git-ip", ip);
296 let now = Instant::now();
297 if let Some(retry) = state.throttle.check(&key, GIT_IP_LIMIT, now) {
298 tracing::warn!(
299 ip = %ip,
300 retry_after_secs = retry.as_secs(),
301 "git authentication throttled"
302 );
303 let mut res = text(
304 StatusCode::TOO_MANY_REQUESTS,
305 "too many failed attempts, try again later",
306 );
307 res.headers_mut()
308 .insert(RETRY_AFTER, HeaderValue::from(retry.as_secs().max(1)));
309 return Err(Box::new(res));
310 }
311 match user_from_basic(state, headers).await {
312 Ok(actor) => Ok(actor),
313 Err(AccessError::Unauthorized) if headers.contains_key(AUTHORIZATION) => {
314 state.throttle.record_failure(&key, now);
315 Err(Box::new(access_error(AccessError::Unauthorized)))
316 }
317 Err(error) => Err(Box::new(access_error(error))),
318 }
319}
320
321/// Authorizes a smart-HTTP request. A CI job token (see [`job_token_read`]) is tried first;
322/// otherwise the caller is a user or anonymous, and the normal rules apply. Returns the actor
323/// (`None` for a job), the repo and its path.
324async fn caller_repo(
325 state: &AppState,
326 ip: IpAddr,
327 headers: &HeaderMap,
328 owner: &str,
329 name: &str,
330 service: GitService,
331) -> Result<(Option<User>, forge_core::Repo, std::path::PathBuf), Box<Response>> {
332 if let Some(found) = job_token_read(state, ip, headers, owner, name, service).await {
333 return found.map(|(repo, path)| (None, repo, path));
334 }
335 let actor = authenticate(state, ip, headers).await?;
336 let (repo, path) = resolve_repo(state, actor.as_ref(), owner, name, service)
337 .await
338 .map_err(|e| Box::new(access_error(e)))?;
339 Ok((actor, repo, path))
340}
341
342/// Username of a CI job's git credential. The password is `<job id>:<job token>`.
343pub(crate) const JOB_GIT_USERNAME: &str = "ci-job";
344
345/// Read-only git access for a running CI job. Returns `None` unless the request uses the
346/// `ci-job` username, so other callers take the normal path. The token must belong to a job that
347/// is Running (see `job_by_token`), the repo must be that job's repo, and receive-pack is refused.
348/// Failed attempts count against the git failure limit.
349pub(crate) async fn job_token_read(
350 state: &AppState,
351 ip: IpAddr,
352 headers: &HeaderMap,
353 owner: &str,
354 name: &str,
355 service: GitService,
356) -> Option<Result<(forge_core::Repo, std::path::PathBuf), Box<Response>>> {
357 let (username, password) = parse_basic_auth(headers)?;
358 if !username.eq_ignore_ascii_case(JOB_GIT_USERNAME) {
359 return None;
360 }
361 let key = throttle::ip_key("git-ip", ip);
362 let now = Instant::now();
363 if let Some(retry) = state.throttle.check(&key, GIT_IP_LIMIT, now) {
364 let mut res = text(
365 StatusCode::TOO_MANY_REQUESTS,
366 "too many failed attempts, try again later",
367 );
368 res.headers_mut()
369 .insert(RETRY_AFTER, HeaderValue::from(retry.as_secs().max(1)));
370 return Some(Err(Box::new(res)));
371 }
372 if service == GitService::ReceivePack {
373 return Some(Err(Box::new(text(
374 StatusCode::FORBIDDEN,
375 "CI job tokens are read-only",
376 ))));
377 }
378 let job_token = match password.split_once(':') {
379 Some((job_id, token)) => state
380 .store
381 .job_by_token(job_id, token)
382 .await
383 .map_err(|e| internal_error_for(&e)),
384 None => Ok(None),
385 };
386 let run_repo = match job_token {
387 Ok(Some(job)) => state
388 .store
389 .run_by_id(&job.run_id)
390 .await
391 .map_err(|e| internal_error_for(&e))
392 .map(|run| run.map(|run| run.repo_id)),
393 Ok(None) => Ok(None),
394 Err(response) => Err(response),
395 };
396 let run_repo = match run_repo {
397 Ok(run_repo) => run_repo,
398 Err(response) => return Some(Err(Box::new(response))),
399 };
400 let Some(job_repo_id) = run_repo else {
401 state.throttle.record_failure(&key, now);
402 return Some(Err(Box::new(access_error(AccessError::Unauthorized))));
403 };
404 // `<repo>.wiki` names the job's repo's wiki. It is hidden while the wiki is off, as for users.
405 let (repo_name, wiki) = match forge_core::names::wiki_base(name) {
406 Some(base) => (base, true),
407 None => (name, false),
408 };
409 let repo = match find_repo(state, owner, repo_name).await {
410 Ok(Some(repo)) if repo.id == job_repo_id => repo,
411 Ok(_) => return Some(Err(Box::new(access_error(AccessError::Unauthorized)))),
412 Err(e) => return Some(Err(Box::new(access_error(e)))),
413 };
414 if wiki {
415 match state.store.wiki_enabled(&repo.id).await {
416 Ok(true) => {}
417 Ok(false) => return Some(Err(Box::new(access_error(AccessError::Unauthorized)))),
418 Err(e) => return Some(Err(Box::new(internal_error_for(&e)))),
419 }
420 return match forge_git::repo::wiki_path(&state.config.data_dir, &repo.id) {
421 Ok(path) => Some(Ok((repo, path))),
422 Err(_) => Some(Err(Box::new(internal_error()))),
423 };
424 }
425 match forge_git::repo::repo_path(&state.config.data_dir, &repo.id) {
426 Ok(path) => Some(Ok((repo, path))),
427 Err(_) => Some(Err(Box::new(internal_error()))),
428 }
429}
430
431fn internal_error_for(error: &forge_core::Error) -> Response {
432 tracing::error!(error = %error, "ci job token check failed");
433 internal_error()
434}
435
436pub(crate) fn access_error(error: AccessError) -> Response {
437 match error {
438 AccessError::Unauthorized => {
439 let mut res = text(StatusCode::UNAUTHORIZED, "unauthorized");
440 res.headers_mut().insert(
441 WWW_AUTHENTICATE,
442 HeaderValue::from_static("Basic realm=\"forge\""),
443 );
444 res
445 }
446 AccessError::NotFound => text(StatusCode::NOT_FOUND, "not found"),
447 AccessError::Forbidden => text(StatusCode::FORBIDDEN, "forbidden"),
448 AccessError::Internal(message) => {
449 tracing::error!(error = %message, "smart-http request failed");
450 internal_error()
451 }
452 }
453}
454
455#[cfg(test)]
456mod tests {
457 use super::*;
458
459 #[test]
460 fn git_protocol_accepts_documented_charset() {
461 assert!(valid_git_protocol("version=2"));
462 assert!(valid_git_protocol("version=2:object-format=sha1"));
463 assert!(valid_git_protocol("a.b,c_d-e:f"));
464 }
465
466 #[test]
467 fn git_protocol_rejects_bad_values() {
468 assert!(!valid_git_protocol(""));
469 assert!(!valid_git_protocol("Version=2"));
470 assert!(!valid_git_protocol("version 2"));
471 assert!(!valid_git_protocol("version=2;x"));
472 assert!(!valid_git_protocol("version=2\n"));
473 assert!(!valid_git_protocol(&"a".repeat(65)));
474 assert!(valid_git_protocol(&"a".repeat(64)));
475 }
476
477 #[test]
478 fn git_protocol_header_is_dropped_when_invalid() {
479 let mut headers = HeaderMap::new();
480 headers.insert("git-protocol", HeaderValue::from_static("version=2"));
481 assert_eq!(git_protocol(&headers), Some("version=2"));
482 headers.insert("git-protocol", HeaderValue::from_static("BAD VALUE"));
483 assert_eq!(git_protocol(&headers), None);
484 assert_eq!(git_protocol(&HeaderMap::new()), None);
485 }
486
487 #[test]
488 fn wants_v2_detects_version_pair() {
489 assert!(wants_v2(Some("version=2")));
490 assert!(wants_v2(Some("object-format=sha1:version=2")));
491 assert!(!wants_v2(Some("version=1")));
492 assert!(!wants_v2(None));
493 }
494
495 #[test]
496 fn service_header_rules() {
497 assert!(needs_service_header(GitService::UploadPack, false));
498 assert!(!needs_service_header(GitService::UploadPack, true));
499 assert!(needs_service_header(GitService::ReceivePack, false));
500 assert!(needs_service_header(GitService::ReceivePack, true));
501 }
502
503 #[test]
504 fn body_encoding_rules() {
505 assert_eq!(body_encoding(None), Some(false));
506 assert_eq!(body_encoding(Some("gzip")), Some(true));
507 assert_eq!(body_encoding(Some("GZIP")), Some(true));
508 assert_eq!(body_encoding(Some(" identity ")), Some(false));
509 assert_eq!(body_encoding(Some("br")), None);
510 assert_eq!(body_encoding(Some("gzip, gzip")), None);
511 assert_eq!(body_encoding(Some("")), None);
512 }
513
514 #[tokio::test]
515 async fn gzip_request_body_decodes() {
516 use async_compression::tokio::write::GzipEncoder;
517 use tokio::io::{AsyncWriteExt, BufReader};
518
519 let payload = b"0032want 0123456789abcdef0123456789abcdef01234567\n0000";
520 let mut encoder = GzipEncoder::new(Vec::new());
521 encoder.write_all(payload).await.unwrap();
522 encoder.shutdown().await.unwrap();
523 let compressed = encoder.into_inner();
524
525 let mut decoded = Vec::new();
526 let mut decoder = GzipDecoder::new(BufReader::new(&compressed[..]));
527 tokio::io::copy(&mut decoder, &mut decoded).await.unwrap();
528 assert_eq!(decoded, payload);
529 }
530
531 #[test]
532 fn content_types_match_service() {
533 assert_eq!(
534 advertisement_content_type(GitService::UploadPack),
535 "application/x-git-upload-pack-advertisement"
536 );
537 assert_eq!(
538 result_content_type(GitService::ReceivePack),
539 "application/x-git-receive-pack-result"
540 );
541 }
542
543 #[test]
544 fn access_errors_map_to_status() {
545 assert_eq!(
546 access_error(AccessError::Unauthorized).status(),
547 StatusCode::UNAUTHORIZED
548 );
549 assert_eq!(
550 access_error(AccessError::NotFound).status(),
551 StatusCode::NOT_FOUND
552 );
553 assert_eq!(
554 access_error(AccessError::Forbidden).status(),
555 StatusCode::FORBIDDEN
556 );
557 assert_eq!(
558 access_error(AccessError::Internal("x".into())).status(),
559 StatusCode::INTERNAL_SERVER_ERROR
560 );
561 let res = access_error(AccessError::Unauthorized);
562 assert_eq!(
563 res.headers()
564 .get(WWW_AUTHENTICATE)
565 .and_then(|v| v.to_str().ok()),
566 Some("Basic realm=\"forge\"")
567 );
568 }
569
570 fn basic_headers(plain: &str) -> HeaderMap {
571 let mut headers = HeaderMap::new();
572 use base64::Engine;
573
574 let value = format!("Basic {}", base64::prelude::BASE64_STANDARD.encode(plain));
575 headers.insert(AUTHORIZATION, HeaderValue::from_str(&value).unwrap());
576 headers
577 }
578
579 #[tokio::test]
580 async fn thirty_first_bad_token_is_throttled() {
581 let tmp = tempfile::tempdir().expect("tempdir");
582 let dir = tmp.path();
583 let state = crate::test_support::test_state(dir).await;
584 let ip: IpAddr = "192.0.2.50".parse().unwrap();
585 let bad = basic_headers("alice:forge_not_a_real_token");
586
587 for _ in 0..30 {
588 let res = authenticate(&state, ip, &bad).await.unwrap_err();
589 assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
590 }
591 let res = authenticate(&state, ip, &bad).await.unwrap_err();
592 assert_eq!(res.status(), StatusCode::TOO_MANY_REQUESTS);
593 assert!(res.headers().get(RETRY_AFTER).is_some());
594
595 // Anonymous requests from the same IP are stopped too, and other IPs are unaffected.
596 let other: IpAddr = "192.0.2.51".parse().unwrap();
597 assert!(
598 authenticate(&state, other, &HeaderMap::new())
599 .await
600 .unwrap()
601 .is_none()
602 );
603 let res = authenticate(&state, ip, &HeaderMap::new())
604 .await
605 .unwrap_err();
606 assert_eq!(res.status(), StatusCode::TOO_MANY_REQUESTS);
607 }
608}