use std::collections::HashMap;
use std::fmt;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime};
use reqwest::header::{self, HeaderMap, HeaderValue};
use reqwest::{Method, StatusCode};
use serde::Deserialize;
use serde::de::DeserializeOwned;
use crate::api;
use crate::docker::JitConfig;
use crate::sync::lock;
use crate::time;
pub const API_URL: &str = "https://api.github.com";
const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Clone)]
pub struct Token(String);
impl fmt::Debug for Token {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("Token(<redacted>)")
}
}
pub type TokenSource = Box<dyn Fn() -> Result<Token, Error> + Send + Sync>;
pub fn gh_token() -> Result<Token, Error> {
let output = std::process::Command::new("gh")
.args(["auth", "token", "--hostname", "github.com"])
.output()
.map_err(|error| Error::Token(format!("cannot run gh: {error}")))?;
if !output.status.success() {
return Err(Error::Token(format!(
"gh auth token failed: {}",
String::from_utf8_lossy(&output.stderr).trim()
)));
}
let token = String::from_utf8(output.stdout)
.map_err(|_| Error::Token("gh auth token printed a token that is not UTF-8".to_owned()))?;
Ok(Token(token.trim().to_owned()))
}
pub fn install_crypto_provider() {
let _ = rustls::crypto::ring::default_provider().install_default();
}
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error("GitHub's rate limit is used up until {}s after the epoch", secs(*until))]
RateLimited {
until: SystemTime,
},
#[error("GitHub answered {status} to {method} {path}: {message}")]
Status {
method: Method,
path: String,
status: StatusCode,
message: String,
},
#[error("{method} {path}: {source}")]
Http {
method: Method,
path: String,
source: reqwest::Error,
},
#[error("{path}: unexpected response: {source}")]
Decode {
path: String,
source: serde_json::Error,
},
#[error("no GitHub token: {0}")]
Token(String),
}
fn secs(time: SystemTime) -> u64 {
time.duration_since(SystemTime::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct QueuedJob {
pub id: u64,
pub run_id: u64,
pub name: String,
pub labels: Vec<String>,
pub created_at: u64,
pub run_attempt: u32,
pub workflow: Option<String>,
pub branch: Option<String>,
pub url: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
pub struct JobDetails {
pub id: u64,
pub run_id: u64,
#[serde(default = "first_attempt")]
pub run_attempt: u32,
pub name: String,
pub workflow_name: Option<String>,
pub head_branch: Option<String>,
pub html_url: Option<String>,
#[serde(default)]
pub labels: Vec<String>,
pub created_at: Option<String>,
pub status: String,
pub conclusion: Option<String>,
pub started_at: Option<String>,
pub completed_at: Option<String>,
pub runner_name: Option<String>,
#[serde(default)]
pub steps: Vec<JobStep>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunTrigger {
pub event: String,
pub head_branch: Option<String>,
pub head_repository: Option<String>,
pub head_sha: String,
pub by_user: bool,
pub pull_requests: Vec<u64>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PullRequest {
pub by_user: bool,
pub head_repository: Option<String>,
pub head_branch: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RepositoryDetails {
pub default_branch: String,
pub visibility: api::Visibility,
}
fn first_attempt() -> u32 {
1
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
pub struct JobStep {
pub number: u32,
pub name: String,
pub status: String,
pub conclusion: Option<String>,
pub started_at: Option<String>,
pub completed_at: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
pub struct Runner {
pub id: u64,
pub name: String,
pub status: String,
pub busy: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Deleted {
Deleted,
NotFound,
Busy,
}
pub trait GithubApi: Send + Sync {
fn queued_jobs(
&self,
repository: &str,
) -> impl Future<Output = Result<Vec<QueuedJob>, Error>> + Send;
fn runners(&self, repository: &str) -> impl Future<Output = Result<Vec<Runner>, Error>> + Send;
fn jit_config(
&self,
repository: &str,
name: &str,
labels: &[String],
) -> impl Future<Output = Result<(Runner, JitConfig), Error>> + Send;
fn delete_runner(
&self,
repository: &str,
id: u64,
) -> impl Future<Output = Result<Deleted, Error>> + Send;
fn job(
&self,
repository: &str,
id: u64,
) -> impl Future<Output = Result<JobDetails, Error>> + Send;
fn run_trigger(
&self,
repository: &str,
run: u64,
) -> impl Future<Output = Result<RunTrigger, Error>> + Send;
fn pull_request(
&self,
repository: &str,
number: u64,
) -> impl Future<Output = Result<PullRequest, Error>> + Send;
fn repository(
&self,
repository: &str,
) -> impl Future<Output = Result<RepositoryDetails, Error>> + Send;
fn on_branch(
&self,
repository: &str,
branch: &str,
sha: &str,
) -> impl Future<Output = Result<bool, Error>> + Send;
fn cancel_run(
&self,
repository: &str,
run: u64,
) -> impl Future<Output = Result<(), Error>> + Send;
}
pub struct Github {
http: reqwest::Client,
base: String,
token_source: Arc<dyn Fn() -> Result<Token, Error> + Send + Sync>,
token: tokio::sync::Mutex<Option<Token>>,
etags: Mutex<HashMap<String, (HeaderValue, Vec<u8>)>>,
blocked_until: Mutex<Option<SystemTime>>,
}
impl fmt::Debug for Github {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Github")
.field("base", &self.base)
.finish_non_exhaustive()
}
}
#[derive(Deserialize)]
struct Runs {
workflow_runs: Vec<Run>,
}
#[derive(Deserialize)]
struct Run {
id: u64,
}
#[derive(Deserialize)]
struct Jobs {
jobs: Vec<Job>,
}
#[derive(Deserialize)]
struct Job {
id: u64,
run_id: u64,
name: String,
status: String,
labels: Vec<String>,
created_at: Option<String>,
run_attempt: Option<u32>,
workflow_name: Option<String>,
head_branch: Option<String>,
html_url: Option<String>,
}
#[derive(Deserialize)]
struct RunBody {
event: String,
head_branch: Option<String>,
head_repository: Option<RepositoryName>,
head_sha: String,
actor: Option<Account>,
triggering_actor: Option<Account>,
#[serde(default)]
pull_requests: Vec<PullRequestNumber>,
}
#[derive(Deserialize)]
struct Account {
#[serde(rename = "type")]
kind: String,
}
impl Account {
fn is_user(account: Option<&Account>) -> bool {
account.is_some_and(|account| account.kind == "User")
}
}
#[derive(Deserialize)]
struct PullRequestNumber {
number: u64,
base: Option<PullRequestBase>,
}
#[derive(Deserialize)]
struct PullRequestBase {
repo: Option<RepositoryUrl>,
}
#[derive(Deserialize)]
struct RepositoryUrl {
url: String,
}
impl PullRequestNumber {
fn targets(&self, repository: &str) -> bool {
let Some(url) = self
.base
.as_ref()
.and_then(|base| base.repo.as_ref())
.map(|repo| repo.url.to_ascii_lowercase())
else {
return false;
};
url.ends_with(&format!("/repos/{}", repository.to_ascii_lowercase()))
}
}
#[derive(Deserialize)]
struct PullRequestBody {
user: Option<Account>,
head: PullRequestHead,
}
#[derive(Deserialize)]
struct PullRequestHead {
#[serde(rename = "ref")]
branch: String,
repo: Option<RepositoryName>,
}
#[derive(Deserialize)]
struct RepositoryName {
full_name: String,
}
#[derive(Deserialize)]
struct CompareBody {
status: String,
}
#[derive(Deserialize)]
struct BranchBody {
commit: CommitBody,
}
#[derive(Deserialize)]
struct CommitBody {
sha: String,
}
#[derive(Deserialize)]
struct RepositoryBody {
default_branch: String,
visibility: api::Visibility,
}
#[derive(Deserialize)]
struct Runners {
runners: Vec<Runner>,
}
#[derive(Deserialize)]
struct JitResponse {
runner: Runner,
encoded_jit_config: String,
}
#[derive(Deserialize)]
struct ErrorBody {
message: String,
}
impl Github {
pub fn new(base: &str, token_source: TokenSource) -> Result<Github, Error> {
install_crypto_provider();
let mut headers = HeaderMap::new();
headers.insert(
header::ACCEPT,
HeaderValue::from_static("application/vnd.github+json"),
);
headers.insert(
"x-github-api-version",
HeaderValue::from_static("2022-11-28"),
);
let http = reqwest::Client::builder()
.user_agent(concat!("runnerd/", env!("CARGO_PKG_VERSION")))
.default_headers(headers)
.timeout(REQUEST_TIMEOUT)
.build()
.map_err(|source| Error::Http {
method: Method::GET,
path: base.to_owned(),
source,
})?;
Ok(Github {
http,
base: base.trim_end_matches('/').to_owned(),
token_source: Arc::from(token_source),
token: tokio::sync::Mutex::new(None),
etags: Mutex::new(HashMap::new()),
blocked_until: Mutex::new(None),
})
}
pub fn blocked_until(&self) -> Option<SystemTime> {
let blocked = *lock(&self.blocked_until);
blocked.filter(|until| *until > SystemTime::now())
}
async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T, Error> {
let response = self.send(Method::GET, path, None).await?;
let bytes = if response.status == StatusCode::NOT_MODIFIED {
let cached = lock(&self.etags).get(path).map(|(_, body)| body.clone());
match cached {
Some(body) => body,
None => {
return Err(Error::Status {
method: Method::GET,
path: path.to_owned(),
status: response.status,
message: "not modified, but nothing is cached".to_owned(),
});
}
}
} else {
let etag = response.etag.clone();
let response = expect_success(Method::GET, path, response)?;
if let Some(etag) = etag {
lock(&self.etags).insert(path.to_owned(), (etag, response.bytes.clone()));
}
response.bytes
};
decode(path, &bytes)
}
async fn send(
&self,
method: Method,
path: &str,
body: Option<&serde_json::Value>,
) -> Result<Response, Error> {
if let Some(until) = self.blocked_until() {
return Err(Error::RateLimited { until });
}
let mut refused: Option<Token> = None;
loop {
let token = self.token(refused.as_ref()).await?;
let mut request = self
.http
.request(method.clone(), format!("{}{path}", self.base))
.bearer_auth(&token.0);
if let Some(body) = body {
request = request
.header(header::CONTENT_TYPE, "application/json")
.body(body.to_string());
}
if method == Method::GET
&& let Some((etag, _)) = lock(&self.etags).get(path)
{
request = request.header(header::IF_NONE_MATCH, etag.clone());
}
let response = request.send().await.map_err(|source| Error::Http {
method: method.clone(),
path: path.to_owned(),
source: source.without_url(),
})?;
let status = response.status();
if status == StatusCode::UNAUTHORIZED && refused.is_none() {
refused = Some(token);
continue;
}
if let Some(until) = rate_limited_until(status, response.headers(), SystemTime::now()) {
*lock(&self.blocked_until) = Some(until);
if matches!(
status,
StatusCode::FORBIDDEN | StatusCode::TOO_MANY_REQUESTS
) {
return Err(Error::RateLimited { until });
}
}
let etag = response.headers().get(header::ETAG).cloned();
let bytes = response.bytes().await.map_err(|source| Error::Http {
method: method.clone(),
path: path.to_owned(),
source: source.without_url(),
})?;
return Ok(Response {
status,
etag,
bytes: bytes.to_vec(),
});
}
}
async fn token(&self, refused: Option<&Token>) -> Result<Token, Error> {
let mut token = self.token.lock().await;
let stale = match (&*token, refused) {
(None, _) => true,
(Some(current), Some(refused)) => current.0 == refused.0,
(Some(_), None) => false,
};
if stale {
let source = self.token_source.clone();
*token = Some(
tokio::task::spawn_blocking(move || source())
.await
.map_err(|error| {
Error::Token(format!("the token source stopped: {error}"))
})??,
);
}
token
.clone()
.ok_or_else(|| Error::Token("the token source returned nothing".to_owned()))
}
}
impl GithubApi for Github {
async fn queued_jobs(&self, repository: &str) -> Result<Vec<QueuedJob>, Error> {
let mut jobs = Vec::new();
let mut listed = Vec::new();
for status in ["queued", "in_progress"] {
let runs: Runs = self
.get(&format!(
"/repos/{repository}/actions/runs?status={status}&per_page=100"
))
.await?;
for run in runs.workflow_runs {
let path = jobs_path(repository, run.id);
let run_jobs: Jobs = self.get(&path).await?;
listed.push(path);
jobs.extend(
run_jobs
.jobs
.into_iter()
.filter(|job| job.status == "queued")
.map(|job| QueuedJob {
id: job.id,
run_id: job.run_id,
name: job.name,
labels: job.labels,
created_at: job
.created_at
.as_deref()
.and_then(time::parse_ms)
.unwrap_or(0),
run_attempt: job.run_attempt.unwrap_or(1),
workflow: job.workflow_name,
branch: job.head_branch,
url: job.html_url,
}),
);
}
}
let runs_prefix = format!("/repos/{repository}/actions/runs/");
lock(&self.etags).retain(|path, _| {
!(path.starts_with(&runs_prefix) && path.contains("/jobs?")) || listed.contains(path)
});
jobs.sort_by_key(|job| job.id);
jobs.dedup_by_key(|job| job.id);
Ok(jobs)
}
async fn runners(&self, repository: &str) -> Result<Vec<Runner>, Error> {
let runners: Runners = self
.get(&format!("/repos/{repository}/actions/runners?per_page=100"))
.await?;
Ok(runners.runners)
}
async fn jit_config(
&self,
repository: &str,
name: &str,
labels: &[String],
) -> Result<(Runner, JitConfig), Error> {
let body = serde_json::json!({
"name": name,
"runner_group_id": 1,
"labels": labels,
"work_folder": "_work",
});
let path = format!("/repos/{repository}/actions/runners/generate-jitconfig");
let response = self.send(Method::POST, &path, Some(&body)).await?;
let response = expect_success(Method::POST, &path, response)?;
let jit: JitResponse = decode(&path, &response.bytes)?;
Ok((jit.runner, JitConfig(jit.encoded_jit_config)))
}
async fn delete_runner(&self, repository: &str, id: u64) -> Result<Deleted, Error> {
let path = format!("/repos/{repository}/actions/runners/{id}");
let response = self.send(Method::DELETE, &path, None).await?;
match response.status {
StatusCode::NOT_FOUND => Ok(Deleted::NotFound),
StatusCode::UNPROCESSABLE_ENTITY => Ok(Deleted::Busy),
_ => expect_success(Method::DELETE, &path, response).map(|_| Deleted::Deleted),
}
}
async fn job(&self, repository: &str, id: u64) -> Result<JobDetails, Error> {
self.get(&format!("/repos/{repository}/actions/jobs/{id}"))
.await
}
async fn run_trigger(&self, repository: &str, run: u64) -> Result<RunTrigger, Error> {
let path = format!("/repos/{repository}/actions/runs/{run}");
let run: Result<RunBody, Error> = self.get(&path).await;
lock(&self.etags).remove(&path);
let run = run?;
Ok(RunTrigger {
event: run.event,
head_branch: run.head_branch,
head_repository: run.head_repository.map(|head| head.full_name),
head_sha: run.head_sha,
by_user: Account::is_user(run.actor.as_ref())
&& Account::is_user(run.triggering_actor.as_ref()),
pull_requests: run
.pull_requests
.into_iter()
.filter(|pull| pull.targets(repository))
.map(|pull| pull.number)
.collect(),
})
}
async fn pull_request(&self, repository: &str, number: u64) -> Result<PullRequest, Error> {
let path = format!("/repos/{repository}/pulls/{number}");
let pull: Result<PullRequestBody, Error> = self.get(&path).await;
lock(&self.etags).remove(&path);
let pull = pull?;
Ok(PullRequest {
by_user: Account::is_user(pull.user.as_ref()),
head_repository: pull.head.repo.map(|repo| repo.full_name),
head_branch: pull.head.branch,
})
}
async fn repository(&self, repository: &str) -> Result<RepositoryDetails, Error> {
let body: RepositoryBody = self.get(&format!("/repos/{repository}")).await?;
Ok(RepositoryDetails {
default_branch: body.default_branch,
visibility: body.visibility,
})
}
async fn on_branch(&self, repository: &str, branch: &str, sha: &str) -> Result<bool, Error> {
let commit =
|sha: &str| !sha.is_empty() && sha.bytes().all(|byte| byte.is_ascii_hexdigit());
if !commit(sha) {
return Ok(false);
}
let base: String = branch
.bytes()
.map(|byte| {
if byte.is_ascii_alphanumeric() || matches!(byte, b'/' | b'-' | b'_' | b'.') {
char::from(byte).to_string()
} else {
format!("%{byte:02X}")
}
})
.collect();
let path = format!("/repos/{repository}/branches/{base}");
let head: Result<BranchBody, Error> = self.get(&path).await;
lock(&self.etags).remove(&path);
let head = head?.commit.sha;
if !commit(&head) {
return Ok(false);
}
let path = format!("/repos/{repository}/compare/{head}...{sha}?per_page=1");
let body: Result<CompareBody, Error> = self.get(&path).await;
lock(&self.etags).remove(&path);
Ok(matches!(body?.status.as_str(), "identical" | "behind"))
}
async fn cancel_run(&self, repository: &str, run: u64) -> Result<(), Error> {
let path = format!("/repos/{repository}/actions/runs/{run}/cancel");
let response = self.send(Method::POST, &path, None).await?;
if response.status == StatusCode::CONFLICT {
return Ok(());
}
expect_success(Method::POST, &path, response).map(|_| ())
}
}
fn jobs_path(repository: &str, run: u64) -> String {
format!("/repos/{repository}/actions/runs/{run}/jobs?filter=latest&per_page=100")
}
struct Response {
status: StatusCode,
etag: Option<HeaderValue>,
bytes: Vec<u8>,
}
fn expect_success(method: Method, path: &str, response: Response) -> Result<Response, Error> {
if response.status.is_success() {
return Ok(response);
}
let message = serde_json::from_slice::<ErrorBody>(&response.bytes).map_or_else(
|_| String::from_utf8_lossy(&response.bytes).into_owned(),
|body| body.message,
);
Err(Error::Status {
method,
path: path.to_owned(),
status: response.status,
message,
})
}
fn decode<T: DeserializeOwned>(path: &str, bytes: &[u8]) -> Result<T, Error> {
serde_json::from_slice(bytes).map_err(|source| Error::Decode {
path: path.to_owned(),
source,
})
}
fn rate_limited_until(
status: StatusCode,
headers: &HeaderMap,
now: SystemTime,
) -> Option<SystemTime> {
let number = |name: &str| {
headers
.get(name)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.trim().parse::<u64>().ok())
};
if number("x-ratelimit-remaining") == Some(0)
&& let Some(reset) = number("x-ratelimit-reset")
{
return Some(SystemTime::UNIX_EPOCH + Duration::from_secs(reset));
}
if matches!(
status,
StatusCode::FORBIDDEN | StatusCode::TOO_MANY_REQUESTS
) && let Some(seconds) = number("retry-after")
{
return Some(now + Duration::from_secs(seconds));
}
None
}
#[cfg(test)]
mod tests;