1use std::collections::HashMap;
6use std::fmt;
7use std::sync::{Arc, Mutex};
8use std::time::{Duration, SystemTime};
9
10use reqwest::header::{self, HeaderMap, HeaderValue};
11use reqwest::{Method, StatusCode};
12use serde::Deserialize;
13use serde::de::DeserializeOwned;
14
15use crate::api;
16use crate::docker::JitConfig;
17use crate::sync::lock;
18use crate::time;
19
20pub const API_URL: &str = "https://api.github.com";
22
23const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
26
27#[derive(Clone)]
29pub struct Token(String);
30
31impl fmt::Debug for Token {
32 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
33 f.write_str("Token(<redacted>)")
34 }
35}
36
37pub type TokenSource = Box<dyn Fn() -> Result<Token, Error> + Send + Sync>;
39
40pub fn gh_token() -> Result<Token, Error> {
42 let output = std::process::Command::new("gh")
43 .args(["auth", "token", "--hostname", "github.com"])
44 .output()
45 .map_err(|error| Error::Token(format!("cannot run gh: {error}")))?;
46 if !output.status.success() {
47 return Err(Error::Token(format!(
48 "gh auth token failed: {}",
49 String::from_utf8_lossy(&output.stderr).trim()
50 )));
51 }
52 let token = String::from_utf8(output.stdout)
53 .map_err(|_| Error::Token("gh auth token printed a token that is not UTF-8".to_owned()))?;
54 Ok(Token(token.trim().to_owned()))
55}
56
57pub fn install_crypto_provider() {
60 let _ = rustls::crypto::ring::default_provider().install_default();
62}
63
64#[derive(Debug, thiserror::Error)]
66pub enum Error {
67 #[error("GitHub's rate limit is used up until {}s after the epoch", secs(*until))]
69 RateLimited {
70 until: SystemTime,
72 },
73 #[error("GitHub answered {status} to {method} {path}: {message}")]
75 Status {
76 method: Method,
78 path: String,
80 status: StatusCode,
82 message: String,
84 },
85 #[error("{method} {path}: {source}")]
87 Http {
88 method: Method,
90 path: String,
92 source: reqwest::Error,
94 },
95 #[error("{path}: unexpected response: {source}")]
97 Decode {
98 path: String,
100 source: serde_json::Error,
102 },
103 #[error("no GitHub token: {0}")]
105 Token(String),
106}
107
108fn secs(time: SystemTime) -> u64 {
109 time.duration_since(SystemTime::UNIX_EPOCH)
110 .unwrap_or_default()
111 .as_secs()
112}
113
114#[derive(Debug, Clone, Default, PartialEq, Eq)]
116pub struct QueuedJob {
117 pub id: u64,
119 pub run_id: u64,
121 pub name: String,
123 pub labels: Vec<String>,
125 pub created_at: u64,
127 pub run_attempt: u32,
129 pub workflow: Option<String>,
131 pub branch: Option<String>,
133 pub url: Option<String>,
135}
136
137#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
139pub struct JobDetails {
140 pub id: u64,
142 pub run_id: u64,
144 #[serde(default = "first_attempt")]
146 pub run_attempt: u32,
147 pub name: String,
149 pub workflow_name: Option<String>,
151 pub head_branch: Option<String>,
153 pub html_url: Option<String>,
155 #[serde(default)]
157 pub labels: Vec<String>,
158 pub created_at: Option<String>,
160 pub status: String,
162 pub conclusion: Option<String>,
164 pub started_at: Option<String>,
166 pub completed_at: Option<String>,
168 pub runner_name: Option<String>,
170 #[serde(default)]
172 pub steps: Vec<JobStep>,
173}
174
175#[derive(Debug, Clone, PartialEq, Eq)]
177pub struct RunTrigger {
178 pub event: String,
180 pub head_branch: Option<String>,
182 pub head_repository: Option<String>,
184 pub head_sha: String,
187 pub by_user: bool,
190 pub pull_requests: Vec<u64>,
194}
195
196#[derive(Debug, Clone, PartialEq, Eq)]
198pub struct PullRequest {
199 pub by_user: bool,
201 pub head_repository: Option<String>,
203 pub head_branch: String,
205}
206
207#[derive(Debug, Clone, PartialEq, Eq)]
209pub struct RepositoryDetails {
210 pub default_branch: String,
212 pub visibility: api::Visibility,
214}
215
216fn first_attempt() -> u32 {
217 1
218}
219
220#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
222pub struct JobStep {
223 pub number: u32,
225 pub name: String,
227 pub status: String,
229 pub conclusion: Option<String>,
231 pub started_at: Option<String>,
233 pub completed_at: Option<String>,
235}
236
237#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
239pub struct Runner {
240 pub id: u64,
242 pub name: String,
244 pub status: String,
246 pub busy: bool,
248}
249
250#[derive(Debug, Clone, Copy, PartialEq, Eq)]
252pub enum Deleted {
253 Deleted,
255 NotFound,
257 Busy,
259}
260
261pub trait GithubApi: Send + Sync {
263 fn queued_jobs(
265 &self,
266 repository: &str,
267 ) -> impl Future<Output = Result<Vec<QueuedJob>, Error>> + Send;
268 fn runners(&self, repository: &str) -> impl Future<Output = Result<Vec<Runner>, Error>> + Send;
270 fn jit_config(
272 &self,
273 repository: &str,
274 name: &str,
275 labels: &[String],
276 ) -> impl Future<Output = Result<(Runner, JitConfig), Error>> + Send;
277 fn delete_runner(
279 &self,
280 repository: &str,
281 id: u64,
282 ) -> impl Future<Output = Result<Deleted, Error>> + Send;
283 fn job(
285 &self,
286 repository: &str,
287 id: u64,
288 ) -> impl Future<Output = Result<JobDetails, Error>> + Send;
289 fn run_trigger(
291 &self,
292 repository: &str,
293 run: u64,
294 ) -> impl Future<Output = Result<RunTrigger, Error>> + Send;
295 fn pull_request(
297 &self,
298 repository: &str,
299 number: u64,
300 ) -> impl Future<Output = Result<PullRequest, Error>> + Send;
301 fn repository(
303 &self,
304 repository: &str,
305 ) -> impl Future<Output = Result<RepositoryDetails, Error>> + Send;
306 fn on_branch(
309 &self,
310 repository: &str,
311 branch: &str,
312 sha: &str,
313 ) -> impl Future<Output = Result<bool, Error>> + Send;
314 fn cancel_run(
317 &self,
318 repository: &str,
319 run: u64,
320 ) -> impl Future<Output = Result<(), Error>> + Send;
321}
322
323pub struct Github {
326 http: reqwest::Client,
327 base: String,
328 token_source: Arc<dyn Fn() -> Result<Token, Error> + Send + Sync>,
330 token: tokio::sync::Mutex<Option<Token>>,
333 etags: Mutex<HashMap<String, (HeaderValue, Vec<u8>)>>,
335 blocked_until: Mutex<Option<SystemTime>>,
336}
337
338impl fmt::Debug for Github {
339 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
340 f.debug_struct("Github")
341 .field("base", &self.base)
342 .finish_non_exhaustive()
343 }
344}
345
346#[derive(Deserialize)]
347struct Runs {
348 workflow_runs: Vec<Run>,
349}
350
351#[derive(Deserialize)]
352struct Run {
353 id: u64,
354}
355
356#[derive(Deserialize)]
357struct Jobs {
358 jobs: Vec<Job>,
359}
360
361#[derive(Deserialize)]
362struct Job {
363 id: u64,
364 run_id: u64,
365 name: String,
366 status: String,
367 labels: Vec<String>,
368 created_at: Option<String>,
369 run_attempt: Option<u32>,
370 workflow_name: Option<String>,
371 head_branch: Option<String>,
372 html_url: Option<String>,
373}
374
375#[derive(Deserialize)]
376struct RunBody {
377 event: String,
378 head_branch: Option<String>,
379 head_repository: Option<RepositoryName>,
380 head_sha: String,
381 actor: Option<Account>,
382 triggering_actor: Option<Account>,
383 #[serde(default)]
384 pull_requests: Vec<PullRequestNumber>,
385}
386
387#[derive(Deserialize)]
388struct Account {
389 #[serde(rename = "type")]
390 kind: String,
391}
392
393impl Account {
394 fn is_user(account: Option<&Account>) -> bool {
396 account.is_some_and(|account| account.kind == "User")
397 }
398}
399
400#[derive(Deserialize)]
401struct PullRequestNumber {
402 number: u64,
403 base: Option<PullRequestBase>,
404}
405
406#[derive(Deserialize)]
407struct PullRequestBase {
408 repo: Option<RepositoryUrl>,
409}
410
411#[derive(Deserialize)]
412struct RepositoryUrl {
413 url: String,
414}
415
416impl PullRequestNumber {
417 fn targets(&self, repository: &str) -> bool {
420 let Some(url) = self
421 .base
422 .as_ref()
423 .and_then(|base| base.repo.as_ref())
424 .map(|repo| repo.url.to_ascii_lowercase())
425 else {
426 return false;
427 };
428 url.ends_with(&format!("/repos/{}", repository.to_ascii_lowercase()))
429 }
430}
431
432#[derive(Deserialize)]
433struct PullRequestBody {
434 user: Option<Account>,
435 head: PullRequestHead,
436}
437
438#[derive(Deserialize)]
439struct PullRequestHead {
440 #[serde(rename = "ref")]
441 branch: String,
442 repo: Option<RepositoryName>,
443}
444
445#[derive(Deserialize)]
446struct RepositoryName {
447 full_name: String,
448}
449
450#[derive(Deserialize)]
451struct CompareBody {
452 status: String,
453}
454
455#[derive(Deserialize)]
456struct BranchBody {
457 commit: CommitBody,
458}
459
460#[derive(Deserialize)]
461struct CommitBody {
462 sha: String,
463}
464
465#[derive(Deserialize)]
466struct RepositoryBody {
467 default_branch: String,
468 visibility: api::Visibility,
469}
470
471#[derive(Deserialize)]
472struct Runners {
473 runners: Vec<Runner>,
474}
475
476#[derive(Deserialize)]
477struct JitResponse {
478 runner: Runner,
479 encoded_jit_config: String,
480}
481
482#[derive(Deserialize)]
483struct ErrorBody {
484 message: String,
485}
486
487impl Github {
488 pub fn new(base: &str, token_source: TokenSource) -> Result<Github, Error> {
490 install_crypto_provider();
491 let mut headers = HeaderMap::new();
492 headers.insert(
493 header::ACCEPT,
494 HeaderValue::from_static("application/vnd.github+json"),
495 );
496 headers.insert(
497 "x-github-api-version",
498 HeaderValue::from_static("2022-11-28"),
499 );
500 let http = reqwest::Client::builder()
501 .user_agent(concat!("runnerd/", env!("CARGO_PKG_VERSION")))
502 .default_headers(headers)
503 .timeout(REQUEST_TIMEOUT)
504 .build()
505 .map_err(|source| Error::Http {
506 method: Method::GET,
507 path: base.to_owned(),
508 source,
509 })?;
510 Ok(Github {
511 http,
512 base: base.trim_end_matches('/').to_owned(),
513 token_source: Arc::from(token_source),
514 token: tokio::sync::Mutex::new(None),
515 etags: Mutex::new(HashMap::new()),
516 blocked_until: Mutex::new(None),
517 })
518 }
519
520 pub fn blocked_until(&self) -> Option<SystemTime> {
522 let blocked = *lock(&self.blocked_until);
523 blocked.filter(|until| *until > SystemTime::now())
524 }
525
526 async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T, Error> {
527 let response = self.send(Method::GET, path, None).await?;
528 let bytes = if response.status == StatusCode::NOT_MODIFIED {
529 let cached = lock(&self.etags).get(path).map(|(_, body)| body.clone());
530 match cached {
531 Some(body) => body,
532 None => {
533 return Err(Error::Status {
534 method: Method::GET,
535 path: path.to_owned(),
536 status: response.status,
537 message: "not modified, but nothing is cached".to_owned(),
538 });
539 }
540 }
541 } else {
542 let etag = response.etag.clone();
543 let response = expect_success(Method::GET, path, response)?;
544 if let Some(etag) = etag {
545 lock(&self.etags).insert(path.to_owned(), (etag, response.bytes.clone()));
546 }
547 response.bytes
548 };
549 decode(path, &bytes)
550 }
551
552 async fn send(
554 &self,
555 method: Method,
556 path: &str,
557 body: Option<&serde_json::Value>,
558 ) -> Result<Response, Error> {
559 if let Some(until) = self.blocked_until() {
560 return Err(Error::RateLimited { until });
561 }
562 let mut refused: Option<Token> = None;
564 loop {
565 let token = self.token(refused.as_ref()).await?;
566 let mut request = self
567 .http
568 .request(method.clone(), format!("{}{path}", self.base))
569 .bearer_auth(&token.0);
570 if let Some(body) = body {
571 request = request
572 .header(header::CONTENT_TYPE, "application/json")
573 .body(body.to_string());
574 }
575 if method == Method::GET
576 && let Some((etag, _)) = lock(&self.etags).get(path)
577 {
578 request = request.header(header::IF_NONE_MATCH, etag.clone());
579 }
580 let response = request.send().await.map_err(|source| Error::Http {
581 method: method.clone(),
582 path: path.to_owned(),
583 source: source.without_url(),
584 })?;
585 let status = response.status();
586 if status == StatusCode::UNAUTHORIZED && refused.is_none() {
587 refused = Some(token);
588 continue;
589 }
590 if let Some(until) = rate_limited_until(status, response.headers(), SystemTime::now()) {
591 *lock(&self.blocked_until) = Some(until);
592 if matches!(
593 status,
594 StatusCode::FORBIDDEN | StatusCode::TOO_MANY_REQUESTS
595 ) {
596 return Err(Error::RateLimited { until });
597 }
598 }
599 let etag = response.headers().get(header::ETAG).cloned();
600 let bytes = response.bytes().await.map_err(|source| Error::Http {
601 method: method.clone(),
602 path: path.to_owned(),
603 source: source.without_url(),
604 })?;
605 return Ok(Response {
606 status,
607 etag,
608 bytes: bytes.to_vec(),
609 });
610 }
611 }
612
613 async fn token(&self, refused: Option<&Token>) -> Result<Token, Error> {
616 let mut token = self.token.lock().await;
617 let stale = match (&*token, refused) {
618 (None, _) => true,
619 (Some(current), Some(refused)) => current.0 == refused.0,
620 (Some(_), None) => false,
621 };
622 if stale {
623 let source = self.token_source.clone();
624 *token = Some(
625 tokio::task::spawn_blocking(move || source())
626 .await
627 .map_err(|error| {
628 Error::Token(format!("the token source stopped: {error}"))
629 })??,
630 );
631 }
632 token
633 .clone()
634 .ok_or_else(|| Error::Token("the token source returned nothing".to_owned()))
635 }
636}
637
638impl GithubApi for Github {
639 async fn queued_jobs(&self, repository: &str) -> Result<Vec<QueuedJob>, Error> {
640 let mut jobs = Vec::new();
641 let mut listed = Vec::new();
642 for status in ["queued", "in_progress"] {
643 let runs: Runs = self
644 .get(&format!(
645 "/repos/{repository}/actions/runs?status={status}&per_page=100"
646 ))
647 .await?;
648 for run in runs.workflow_runs {
649 let path = jobs_path(repository, run.id);
650 let run_jobs: Jobs = self.get(&path).await?;
651 listed.push(path);
652 jobs.extend(
653 run_jobs
654 .jobs
655 .into_iter()
656 .filter(|job| job.status == "queued")
657 .map(|job| QueuedJob {
658 id: job.id,
659 run_id: job.run_id,
660 name: job.name,
661 labels: job.labels,
662 created_at: job
663 .created_at
664 .as_deref()
665 .and_then(time::parse_ms)
666 .unwrap_or(0),
667 run_attempt: job.run_attempt.unwrap_or(1),
668 workflow: job.workflow_name,
669 branch: job.head_branch,
670 url: job.html_url,
671 }),
672 );
673 }
674 }
675 let runs_prefix = format!("/repos/{repository}/actions/runs/");
677 lock(&self.etags).retain(|path, _| {
678 !(path.starts_with(&runs_prefix) && path.contains("/jobs?")) || listed.contains(path)
679 });
680 jobs.sort_by_key(|job| job.id);
681 jobs.dedup_by_key(|job| job.id);
682 Ok(jobs)
683 }
684
685 async fn runners(&self, repository: &str) -> Result<Vec<Runner>, Error> {
686 let runners: Runners = self
687 .get(&format!("/repos/{repository}/actions/runners?per_page=100"))
688 .await?;
689 Ok(runners.runners)
690 }
691
692 async fn jit_config(
693 &self,
694 repository: &str,
695 name: &str,
696 labels: &[String],
697 ) -> Result<(Runner, JitConfig), Error> {
698 let body = serde_json::json!({
699 "name": name,
700 "runner_group_id": 1,
702 "labels": labels,
703 "work_folder": "_work",
704 });
705 let path = format!("/repos/{repository}/actions/runners/generate-jitconfig");
706 let response = self.send(Method::POST, &path, Some(&body)).await?;
707 let response = expect_success(Method::POST, &path, response)?;
708 let jit: JitResponse = decode(&path, &response.bytes)?;
709 Ok((jit.runner, JitConfig(jit.encoded_jit_config)))
710 }
711
712 async fn delete_runner(&self, repository: &str, id: u64) -> Result<Deleted, Error> {
713 let path = format!("/repos/{repository}/actions/runners/{id}");
714 let response = self.send(Method::DELETE, &path, None).await?;
715 match response.status {
716 StatusCode::NOT_FOUND => Ok(Deleted::NotFound),
717 StatusCode::UNPROCESSABLE_ENTITY => Ok(Deleted::Busy),
718 _ => expect_success(Method::DELETE, &path, response).map(|_| Deleted::Deleted),
719 }
720 }
721
722 async fn job(&self, repository: &str, id: u64) -> Result<JobDetails, Error> {
723 self.get(&format!("/repos/{repository}/actions/jobs/{id}"))
724 .await
725 }
726
727 async fn run_trigger(&self, repository: &str, run: u64) -> Result<RunTrigger, Error> {
728 let path = format!("/repos/{repository}/actions/runs/{run}");
729 let run: Result<RunBody, Error> = self.get(&path).await;
730 lock(&self.etags).remove(&path);
732 let run = run?;
733 Ok(RunTrigger {
734 event: run.event,
735 head_branch: run.head_branch,
736 head_repository: run.head_repository.map(|head| head.full_name),
737 head_sha: run.head_sha,
738 by_user: Account::is_user(run.actor.as_ref())
739 && Account::is_user(run.triggering_actor.as_ref()),
740 pull_requests: run
741 .pull_requests
742 .into_iter()
743 .filter(|pull| pull.targets(repository))
744 .map(|pull| pull.number)
745 .collect(),
746 })
747 }
748
749 async fn pull_request(&self, repository: &str, number: u64) -> Result<PullRequest, Error> {
750 let path = format!("/repos/{repository}/pulls/{number}");
751 let pull: Result<PullRequestBody, Error> = self.get(&path).await;
752 lock(&self.etags).remove(&path);
753 let pull = pull?;
754 Ok(PullRequest {
755 by_user: Account::is_user(pull.user.as_ref()),
756 head_repository: pull.head.repo.map(|repo| repo.full_name),
757 head_branch: pull.head.branch,
758 })
759 }
760
761 async fn repository(&self, repository: &str) -> Result<RepositoryDetails, Error> {
762 let body: RepositoryBody = self.get(&format!("/repos/{repository}")).await?;
763 Ok(RepositoryDetails {
764 default_branch: body.default_branch,
765 visibility: body.visibility,
766 })
767 }
768
769 async fn on_branch(&self, repository: &str, branch: &str, sha: &str) -> Result<bool, Error> {
770 let commit =
771 |sha: &str| !sha.is_empty() && sha.bytes().all(|byte| byte.is_ascii_hexdigit());
772 if !commit(sha) {
773 return Ok(false);
774 }
775 let base: String = branch
777 .bytes()
778 .map(|byte| {
779 if byte.is_ascii_alphanumeric() || matches!(byte, b'/' | b'-' | b'_' | b'.') {
780 char::from(byte).to_string()
781 } else {
782 format!("%{byte:02X}")
783 }
784 })
785 .collect();
786 let path = format!("/repos/{repository}/branches/{base}");
789 let head: Result<BranchBody, Error> = self.get(&path).await;
790 lock(&self.etags).remove(&path);
791 let head = head?.commit.sha;
792 if !commit(&head) {
793 return Ok(false);
794 }
795 let path = format!("/repos/{repository}/compare/{head}...{sha}?per_page=1");
797 let body: Result<CompareBody, Error> = self.get(&path).await;
798 lock(&self.etags).remove(&path);
799 Ok(matches!(body?.status.as_str(), "identical" | "behind"))
801 }
802
803 async fn cancel_run(&self, repository: &str, run: u64) -> Result<(), Error> {
804 let path = format!("/repos/{repository}/actions/runs/{run}/cancel");
805 let response = self.send(Method::POST, &path, None).await?;
806 if response.status == StatusCode::CONFLICT {
808 return Ok(());
809 }
810 expect_success(Method::POST, &path, response).map(|_| ())
811 }
812}
813
814fn jobs_path(repository: &str, run: u64) -> String {
815 format!("/repos/{repository}/actions/runs/{run}/jobs?filter=latest&per_page=100")
816}
817
818struct Response {
819 status: StatusCode,
820 etag: Option<HeaderValue>,
821 bytes: Vec<u8>,
822}
823
824fn expect_success(method: Method, path: &str, response: Response) -> Result<Response, Error> {
825 if response.status.is_success() {
826 return Ok(response);
827 }
828 let message = serde_json::from_slice::<ErrorBody>(&response.bytes).map_or_else(
829 |_| String::from_utf8_lossy(&response.bytes).into_owned(),
830 |body| body.message,
831 );
832 Err(Error::Status {
833 method,
834 path: path.to_owned(),
835 status: response.status,
836 message,
837 })
838}
839
840fn decode<T: DeserializeOwned>(path: &str, bytes: &[u8]) -> Result<T, Error> {
841 serde_json::from_slice(bytes).map_err(|source| Error::Decode {
842 path: path.to_owned(),
843 source,
844 })
845}
846
847fn rate_limited_until(
850 status: StatusCode,
851 headers: &HeaderMap,
852 now: SystemTime,
853) -> Option<SystemTime> {
854 let number = |name: &str| {
855 headers
856 .get(name)
857 .and_then(|value| value.to_str().ok())
858 .and_then(|value| value.trim().parse::<u64>().ok())
859 };
860 if number("x-ratelimit-remaining") == Some(0)
861 && let Some(reset) = number("x-ratelimit-reset")
862 {
863 return Some(SystemTime::UNIX_EPOCH + Duration::from_secs(reset));
864 }
865 if matches!(
866 status,
867 StatusCode::FORBIDDEN | StatusCode::TOO_MANY_REQUESTS
868 ) && let Some(seconds) = number("retry-after")
869 {
870 return Some(now + Duration::from_secs(seconds));
871 }
872 None
873}
874
875#[cfg(test)]
876mod tests;