Skip to content
Sign up

flagon-io/g1t

public
0
426 linesCodeBlameRaw
1//! Actions from GitHub, kept on g1t.
2//!
3//! An action `uses:` names that g1t does not host (reach.rs) comes from
4//! GitHub. Before a job fetches it, g1t resolves its ref to a commit, as
5//! GitHub's runner does, from the repository's advertisement
6//! (`info/refs`, which no API rate limit applies to), and keeps that
7//! commit's archive in R2. Every later job that uses the action at that
8//! commit downloads it from g1t: faster than GitHub, the same bytes
9//! whatever happens to the tag, and there while GitHub is not.
10//!
11//! - A ref is looked up again once `REF_FRESH_MS` has passed, so a moved
12//! tag (`v4` moving to a new release) is followed within minutes. When
13//! GitHub does not answer, the commit the ref last named is used, and
14//! the job says so.
15//! - A commit's archive is the one GitHub serves for it, which never
16//! changes. One larger than `ARCHIVE_MAX_BYTES` is not kept: the job
17//! fetches it from GitHub itself.
18//! - An archive no job has used for `ARCHIVE_UNUSED_DAYS` is deleted by
19//! the hourly sweep. Archives are g1t's own, shared by every workspace,
20//! and charged to none.
21
22use futures_util::StreamExt;
23use g1t_contracts::time::rfc3339;
24use g1t_kit::now_ms;
25use serde::Deserialize;
26use serde_json::{Value, json};
27use worker::Result;
28
29use crate::Actions;
30use crate::reach::UsesRef;
31
32/// The largest archive kept: the Worker holds it whole while saving it.
33pub(crate) const ARCHIVE_MAX_BYTES: usize = 32 * 1024 * 1024;
34/// How long a ref's commit is trusted before GitHub is asked again.
35pub(crate) const REF_FRESH_MS: u64 = 5 * 60 * 1000;
36/// An archive unused this long is deleted.
37pub(crate) const ARCHIVE_UNUSED_DAYS: u64 = 30;
38/// An advertisement larger than this is not read: a ref is never worth it.
39const ADVERTISEMENT_MAX_BYTES: usize = 8 * 1024 * 1024;
40const DAY_MS: u64 = 24 * 60 * 60 * 1000;
41
42/// Whether `text` is a whole commit id.
43pub(crate) fn is_sha(text: &str) -> bool {
44 text.len() == 40 && text.chars().all(|c| c.is_ascii_hexdigit())
45}
46
47/// The refs a smart HTTP advertisement (`info/refs?service=git-upload-pack`)
48/// lists, as (commit, name). Peeled tags keep their `^{}`.
49pub(crate) fn parse_advertisement(bytes: &[u8]) -> Vec<(String, String)> {
50 let mut refs = Vec::new();
51 let mut at = 0;
52 while at + 4 <= bytes.len() {
53 let Some(length) = std::str::from_utf8(&bytes[at..at + 4]).ok().and_then(|hex| usize::from_str_radix(hex, 16).ok()) else {
54 break;
55 };
56 // A flush or delimiter packet.
57 if length < 4 {
58 at += 4;
59 continue;
60 }
61 if at + length > bytes.len() {
62 break;
63 }
64 let line = String::from_utf8_lossy(&bytes[at + 4..at + length]);
65 at += length;
66 let line = line.trim_end_matches('\n');
67 if line.starts_with('#') {
68 continue;
69 }
70 // The first ref carries the capabilities after a NUL.
71 let line = line.split('\0').next().unwrap_or_default();
72 if let Some((sha, name)) = line.split_once(' ')
73 && is_sha(sha)
74 {
75 refs.push((sha.to_ascii_lowercase(), name.to_owned()));
76 }
77 }
78 refs
79}
80
81/// The commit `wanted` names among `refs`, as git reads a short name: a
82/// commit id as it is, else a tag, else a branch (`refs/…` names one
83/// exactly). A tag's commit is its peeled one.
84pub(crate) fn pick(refs: &[(String, String)], wanted: &str) -> Option<String> {
85 if is_sha(wanted) {
86 return Some(wanted.to_ascii_lowercase());
87 }
88 let names: Vec<String> = if wanted.starts_with("refs/") {
89 vec![wanted.to_owned()]
90 } else {
91 vec![format!("refs/tags/{wanted}"), format!("refs/heads/{wanted}")]
92 };
93 names.iter().find_map(|name| {
94 let peeled = format!("{name}^{{}}");
95 refs.iter()
96 .find(|(_, listed)| *listed == peeled)
97 .or_else(|| refs.iter().find(|(_, listed)| listed == name))
98 .map(|(sha, _)| sha.clone())
99 })
100}
101
102/// Where a commit's archive is kept in R2.
103pub(crate) fn archive_object(repository: &str, sha: &str) -> String {
104 format!("action-archives/{}/{sha}.tar.gz", repository.to_ascii_lowercase())
105}
106
107/// A GET to GitHub, whose body is read up to `max` bytes. None when it
108/// does not answer 200, or sends more.
109async fn fetch_capped(url: &str, max: usize) -> Option<Vec<u8>> {
110 let headers = worker::Headers::new();
111 // GitHub answers a smart HTTP advertisement to git, which it knows by
112 // its agent.
113 headers.set("user-agent", "git/2.47.0 (g1t)").ok()?;
114 let mut init = worker::RequestInit::new();
115 init.with_method(worker::Method::Get).with_headers(headers);
116 let request = worker::Request::new_with_init(url, &init).ok()?;
117 let mut response = worker::Fetch::Request(request).send().await.ok()?;
118 if response.status_code() != 200 {
119 return None;
120 }
121 let mut stream = response.stream().ok()?;
122 let mut body = Vec::new();
123 while let Some(chunk) = stream.next().await {
124 body.extend_from_slice(&chunk.ok()?);
125 if body.len() > max {
126 return None;
127 }
128 }
129 Some(body)
130}
131
132/// The status GitHub answers a GET with, or None when it doesn't answer.
133async fn status_of(url: &str) -> Option<u16> {
134 let headers = worker::Headers::new();
135 headers.set("user-agent", "git/2.47.0 (g1t)").ok()?;
136 let mut init = worker::RequestInit::new();
137 init.with_method(worker::Method::Get).with_headers(headers);
138 let request = worker::Request::new_with_init(url, &init).ok()?;
139 let response = worker::Fetch::Request(request).send().await.ok()?;
140 Some(response.status_code())
141}
142
143/// What GitHub's answer to a repository's advertisement says about it:
144/// Some(true) when it publishes the repository, Some(false) when it says
145/// there is none it will show (GitHub answers 401 for a repository that
146/// doesn't exist as for a private one), None for anything else, such as a
147/// rate limit or an error.
148pub(crate) fn published_by(status: Option<u16>) -> Option<bool> {
149 match status? {
150 200 => Some(true),
151 401 | 404 => Some(false),
152 _ => None,
153 }
154}
155
156#[derive(Deserialize)]
157struct RefRow {
158 sha: String,
159 resolved_at: String,
160}
161
162impl Actions {
163 /// Where a job gets `used`, an action g1t does not host:
164 /// `{"source": "github", "sha", "archive", "note"}`. `sha` is the commit
165 /// its ref names (absent when GitHub does not know it, or never
166 /// answered), `archive` whether g1t keeps that commit's archive for the
167 /// job to download, and `note` anything the job should say about it.
168 pub(crate) async fn github_action(&self, used: &UsesRef) -> Result<Value> {
169 let repository = used.full_name().to_ascii_lowercase();
170 let now = now_ms();
171 let (sha, note) = self.resolve_github(&repository, &used.git_ref, now).await?;
172 let Some(sha) = sha else {
173 return Ok(json!({ "source": "github", "note": note }));
174 };
175 let archive = self.keep_archive(&repository, &sha, now).await?;
176 Ok(json!({ "source": "github", "sha": sha, "archive": archive, "note": note }))
177 }
178
179 /// Whether GitHub publishes `repository` (`owner/repo`): Some(true) when
180 /// it does, or once named a commit of it for a job; Some(false) when it
181 /// says it has no such repository; None when it could not say.
182 pub(crate) async fn github_publishes(&self, repository: &str) -> Result<Option<bool>> {
183 let repository = repository.to_ascii_lowercase();
184 let seen = self
185 .db
186 .prepare("SELECT sha FROM action_refs WHERE repository = ? LIMIT 1")
187 .bind(&[repository.as_str().into()])?
188 .first::<Value>(None)
189 .await?;
190 if seen.is_some() {
191 return Ok(Some(true));
192 }
193 Ok(published_by(status_of(&format!("https://github.com/{repository}.git/info/refs?service=git-upload-pack")).await))
194 }
195
196 /// The commit `git_ref` names in GitHub's `repository`, and a note when
197 /// it is the one it last named because GitHub did not answer.
198 async fn resolve_github(&self, repository: &str, git_ref: &str, now: u64) -> Result<(Option<String>, Option<String>)> {
199 if is_sha(git_ref) {
200 return Ok((Some(git_ref.to_ascii_lowercase()), None));
201 }
202 let known = self
203 .db
204 .prepare("SELECT sha, resolved_at FROM action_refs WHERE repository = ? AND ref = ?")
205 .bind(&[repository.into(), git_ref.into()])?
206 .first::<RefRow>(None)
207 .await?;
208 if let Some(row) = &known
209 && row.resolved_at >= rfc3339(now.saturating_sub(REF_FRESH_MS))
210 {
211 return Ok((Some(row.sha.clone()), None));
212 }
213 let url = format!("https://github.com/{repository}.git/info/refs?service=git-upload-pack");
214 match fetch_capped(&url, ADVERTISEMENT_MAX_BYTES).await {
215 Some(advertisement) => {
216 let Some(sha) = pick(&parse_advertisement(&advertisement), git_ref) else {
217 return Ok((None, None));
218 };
219 self.db
220 .prepare(
221 "INSERT INTO action_refs (repository, ref, sha, resolved_at) VALUES (?1, ?2, ?3, ?4)
222 ON CONFLICT (repository, ref) DO UPDATE SET sha = ?3, resolved_at = ?4",
223 )
224 .bind(&[repository.into(), git_ref.into(), sha.as_str().into(), rfc3339(now).into()])?
225 .run()
226 .await?;
227 Ok((Some(sha), None))
228 }
229 None => Ok(match known {
230 Some(row) => {
231 let note = format!(
232 "GitHub did not say which commit {repository}@{git_ref} is; using {}, which it named on {}.",
233 &row.sha,
234 row.resolved_at.get(..10).unwrap_or(&row.resolved_at)
235 );
236 (Some(row.sha), Some(note))
237 }
238 None => (None, None),
239 }),
240 }
241 }
242
243 /// Whether g1t keeps the archive of `repository` at `sha`, fetching it
244 /// from GitHub the first time.
245 async fn keep_archive(&self, repository: &str, sha: &str, now: u64) -> Result<bool> {
246 let Some(bucket) = &self.cache else { return Ok(false) };
247 #[derive(Deserialize)]
248 struct Kept {
249 used_at: String,
250 }
251 let kept = self
252 .db
253 .prepare("SELECT used_at FROM action_archives WHERE repository = ? AND sha = ?")
254 .bind(&[repository.into(), sha.into()])?
255 .first::<Kept>(None)
256 .await?;
257 if let Some(kept) = kept {
258 // Marked used at most once a day: the sweep needs no more.
259 if kept.used_at < rfc3339(now.saturating_sub(DAY_MS)) {
260 self.db
261 .prepare("UPDATE action_archives SET used_at = ? WHERE repository = ? AND sha = ?")
262 .bind(&[rfc3339(now).into(), repository.into(), sha.into()])?
263 .run()
264 .await?;
265 }
266 return Ok(true);
267 }
268 let url = format!("https://codeload.github.com/{repository}/tar.gz/{sha}");
269 let Some(archive) = fetch_capped(&url, ARCHIVE_MAX_BYTES).await else {
270 return Ok(false);
271 };
272 let object = archive_object(repository, sha);
273 let size = archive.len();
274 if let Err(error) = bucket.put(&object, archive).execute().await {
275 worker::console_error!("actions: the archive of {repository}@{sha} was not kept: {error}");
276 return Ok(false);
277 }
278 let at = rfc3339(now);
279 self.db
280 .prepare(
281 "INSERT INTO action_archives (repository, sha, object, size, stored_at, used_at) VALUES (?1, ?2, ?3, ?4, ?5, ?5)
282 ON CONFLICT (repository, sha) DO UPDATE SET used_at = ?5",
283 )
284 .bind(&[repository.into(), sha.into(), object.into(), (size as f64).into(), at.into()])?
285 .run()
286 .await?;
287 Ok(true)
288 }
289
290 /// `job_action_archive`: a running job downloads an action's archive
291 /// g1t keeps, by `report.repository` and `report.sha`. Answers the R2
292 /// object to send, `{"object", "size"}`. Any job may: they are public
293 /// repositories' code.
294 pub async fn job_action_archive(&self, a: g1t_contracts::actions::JobCallArgs) -> Result<g1t_contracts::Outcome<Value>> {
295 crate::check!(self.job_for_token(&a).await?);
296 let repository = a.report["repository"].as_str().unwrap_or_default().to_ascii_lowercase();
297 let sha = a.report["sha"].as_str().unwrap_or_default().to_ascii_lowercase();
298 if !is_sha(&sha) || crate::reach::parse_uses(&format!("{repository}@{sha}")).is_none() {
299 return Ok(crate::fail(g1t_contracts::FailureCode::Invalid, "Name the action's repository as owner/repo, and its commit."));
300 }
301 #[derive(Deserialize)]
302 struct Kept {
303 object: String,
304 size: f64,
305 }
306 let kept = self
307 .db
308 .prepare("SELECT object, size FROM action_archives WHERE repository = ? AND sha = ?")
309 .bind(&[repository.as_str().into(), sha.as_str().into()])?
310 .first::<Kept>(None)
311 .await?;
312 Ok(match kept {
313 Some(kept) => g1t_contracts::Outcome::Ok(json!({ "object": kept.object, "size": kept.size })),
314 None => crate::fail(g1t_contracts::FailureCode::NotFound, format!("g1t does not keep {repository}@{sha}.")),
315 })
316 }
317
318 /// Hourly: archives no job has used for `ARCHIVE_UNUSED_DAYS`, and refs
319 /// not looked up as long.
320 pub async fn sweep_action_archives(&self, now: u64) -> Result<()> {
321 let before = rfc3339(now.saturating_sub(ARCHIVE_UNUSED_DAYS * DAY_MS));
322 #[derive(Deserialize)]
323 struct Gone {
324 repository: String,
325 sha: String,
326 object: String,
327 }
328 let gone = self
329 .db
330 .prepare("SELECT repository, sha, object FROM action_archives WHERE used_at < ? LIMIT 100")
331 .bind(&[before.as_str().into()])?
332 .all()
333 .await?
334 .results::<Gone>()?;
335 if let Some(bucket) = &self.cache
336 && !gone.is_empty()
337 {
338 bucket.delete_multiple(gone.iter().map(|g| g.object.as_str()).collect()).await?;
339 for archive in &gone {
340 self.db
341 .prepare("DELETE FROM action_archives WHERE repository = ? AND sha = ?")
342 .bind(&[archive.repository.as_str().into(), archive.sha.as_str().into()])?
343 .run()
344 .await?;
345 }
346 }
347 self.db.prepare("DELETE FROM action_refs WHERE resolved_at < ?").bind(&[before.into()])?.run().await?;
348 Ok(())
349 }
350}
351
352#[cfg(test)]
353mod tests {
354 use super::*;
355
356 fn pkt(payload: &str) -> Vec<u8> {
357 let mut out = format!("{:04x}", payload.len() + 4).into_bytes();
358 out.extend_from_slice(payload.as_bytes());
359 out
360 }
361
362 const MAIN: &str = "1111111111111111111111111111111111111111";
363 const V4_TAG: &str = "2222222222222222222222222222222222222222";
364 const V4_COMMIT: &str = "3333333333333333333333333333333333333333";
365 const V1_BRANCH: &str = "4444444444444444444444444444444444444444";
366 const V1_TAG: &str = "5555555555555555555555555555555555555555";
367
368 fn advertisement() -> Vec<u8> {
369 let mut bytes = pkt("# service=git-upload-pack\n");
370 bytes.extend_from_slice(b"0000");
371 bytes.extend(pkt(&format!("{MAIN} HEAD\0multi_ack thin-pack symref=HEAD:refs/heads/main\n")));
372 bytes.extend(pkt(&format!("{MAIN} refs/heads/main\n")));
373 bytes.extend(pkt(&format!("{V1_BRANCH} refs/heads/v1\n")));
374 bytes.extend(pkt(&format!("{V1_TAG} refs/tags/v1\n")));
375 bytes.extend(pkt(&format!("{V4_TAG} refs/tags/v4\n")));
376 bytes.extend(pkt(&format!("{V4_COMMIT} refs/tags/v4^{{}}\n")));
377 bytes.extend_from_slice(b"0000");
378 bytes
379 }
380
381 #[test]
382 fn only_an_answer_from_github_says_it_has_no_such_repository() {
383 assert_eq!(published_by(Some(200)), Some(true));
384 assert_eq!(published_by(Some(401)), Some(false));
385 assert_eq!(published_by(Some(404)), Some(false));
386 // A rate limit, an error or no answer says nothing, so the name stays GitHub's.
387 assert_eq!(published_by(Some(429)), None);
388 assert_eq!(published_by(Some(503)), None);
389 assert_eq!(published_by(None), None);
390 }
391
392 #[test]
393 fn an_advertisement_lists_its_refs_capabilities_aside() {
394 let refs = parse_advertisement(&advertisement());
395 assert_eq!(refs.len(), 6);
396 assert_eq!(refs[0], (MAIN.to_owned(), "HEAD".to_owned()));
397 assert_eq!(refs[5], (V4_COMMIT.to_owned(), "refs/tags/v4^{}".to_owned()));
398 // Cut short: what came whole is kept.
399 let whole = advertisement();
400 assert_eq!(parse_advertisement(&whole[..whole.len() - 30]).len(), 5);
401 assert!(parse_advertisement(b"not git").is_empty());
402 }
403
404 #[test]
405 fn a_ref_is_read_as_git_reads_a_short_name() {
406 let refs = parse_advertisement(&advertisement());
407 // A tag's commit, not the tag object.
408 assert_eq!(pick(&refs, "v4").as_deref(), Some(V4_COMMIT));
409 // A tag before a branch of the same name.
410 assert_eq!(pick(&refs, "v1").as_deref(), Some(V1_TAG));
411 assert_eq!(pick(&refs, "refs/heads/v1").as_deref(), Some(V1_BRANCH));
412 assert_eq!(pick(&refs, "main").as_deref(), Some(MAIN));
413 // A commit is itself, known or not.
414 let other = "ABCDEF0123456789ABCDEF0123456789ABCDEF01";
415 assert_eq!(pick(&refs, other).as_deref(), Some("abcdef0123456789abcdef0123456789abcdef01"));
416 assert_eq!(pick(&refs, "v9"), None);
417 assert_eq!(pick(&refs, "abc123"), None);
418 }
419
420 #[test]
421 fn archives_are_kept_by_repository_and_commit() {
422 assert_eq!(archive_object("Actions/Checkout", V4_COMMIT), format!("action-archives/actions/checkout/{V4_COMMIT}.tar.gz"));
423 assert!(is_sha(MAIN));
424 assert!(!is_sha("v4"));
425 }
426}