pub mod demand;
pub mod device_flow;
pub mod jit;
pub mod rest;
use std::{
fmt,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
time::Duration,
};
use chrono::{DateTime, Utc};
use reqwest::{Method, StatusCode};
use runner_manager_domain::model::{Clock, Org, OwnerRepo, Timestamp};
pub use reqwest::header::HeaderMap;
use secrecy::{ExposeSecret, SecretString};
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use url::Url;
pub const GITHUB_API_VERSION: &str = "2022-11-28";
pub const GITHUB_ACCEPT: &str = "application/vnd.github+json";
pub const GITHUB_API_BASE: &str = "https://api.github.com/";
pub const GITHUB_WEB_BASE: &str = "https://github.com/";
pub const DEVICE_VERIFICATION_PATH: &str = "login/device";
pub const REVALIDATION_PATH: &str = "/user/installations";
pub const DEFAULT_LOCKOUT_BACKOFF: Duration = Duration::from_secs(60);
pub const MAX_LOCKOUT_BACKOFF: Duration = Duration::from_secs(15 * 60);
pub const MAX_PAGES: usize = 100;
pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
pub const USER_AGENT: &str = concat!("runner-manager/", env!("CARGO_PKG_VERSION"));
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AppRegistration {
client_id: String,
slug: String,
}
impl AppRegistration {
pub fn new(client_id: impl Into<String>, slug: impl Into<String>) -> Result<Self, ConfigError> {
let client_id = client_id.into();
let slug = slug.into();
if client_id.trim().is_empty() {
return Err(ConfigError::Empty { what: "client_id" });
}
if slug.trim().is_empty() {
return Err(ConfigError::Empty { what: "app slug" });
}
Ok(Self { client_id, slug })
}
#[must_use]
pub fn client_id(&self) -> &str {
&self.client_id
}
#[must_use]
pub fn slug(&self) -> &str {
&self.slug
}
#[must_use]
pub fn install_url(&self, endpoints: &Endpoints) -> Url {
endpoints
.web_base
.join("apps/")
.and_then(|u| u.join(&format!("{}/", encode_path_segment(&self.slug))))
.and_then(|u| u.join("installations/new"))
.expect("a non-empty encoded slug always joins onto the web base")
}
}
fn encode_path_segment(raw: &str) -> String {
raw.chars()
.map(|c| {
if c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.' | '~') {
c.to_string()
} else {
let mut buf = [0_u8; 4];
c.encode_utf8(&mut buf)
.as_bytes()
.iter()
.map(|b| format!("%{b:02X}"))
.collect()
}
})
.collect()
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Endpoints {
api_base: Url,
web_base: Url,
}
impl Endpoints {
#[must_use]
pub fn production() -> Self {
Self {
api_base: Url::parse(GITHUB_API_BASE).expect("GITHUB_API_BASE is a valid URL"),
web_base: Url::parse(GITHUB_WEB_BASE).expect("GITHUB_WEB_BASE is a valid URL"),
}
}
#[must_use]
pub fn new(api_base: Url, web_base: Url) -> Self {
Self {
api_base: with_trailing_slash(api_base),
web_base: with_trailing_slash(web_base),
}
}
pub fn for_test_server(root: &str) -> Result<Self, ConfigError> {
let root = Url::parse(root).map_err(|_| ConfigError::Empty {
what: "test server URL",
})?;
Ok(Self::new(root.clone(), root))
}
#[must_use]
pub fn api_base(&self) -> &Url {
&self.api_base
}
#[must_use]
pub fn web_base(&self) -> &Url {
&self.web_base
}
#[must_use]
pub fn device_code_url(&self) -> Url {
self.web_base
.join("login/device/code")
.expect("a constant path joins onto a normalised base")
}
#[must_use]
pub fn access_token_url(&self) -> Url {
self.web_base
.join("login/oauth/access_token")
.expect("a constant path joins onto a normalised base")
}
#[must_use]
pub fn verification_url(&self) -> Url {
self.web_base
.join(DEVICE_VERIFICATION_PATH)
.expect("a constant path joins onto a normalised base")
}
}
impl Default for Endpoints {
fn default() -> Self {
Self::production()
}
}
fn with_trailing_slash(mut url: Url) -> Url {
if !url.path().ends_with('/') {
let path = format!("{}/", url.path());
url.set_path(&path);
}
url
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ConfigError {
#[error("{what} must not be empty")]
Empty { what: &'static str },
}
#[derive(Clone)]
pub struct UserAccessToken {
token: SecretString,
token_type: String,
scope: Option<String>,
renewal: Option<Renewal>,
}
#[derive(Clone)]
pub struct Renewal {
refresh_token: SecretString,
pub access_expires_at: Option<DateTime<Utc>>,
pub refresh_expires_at: Option<DateTime<Utc>>,
}
impl Renewal {
#[must_use]
pub fn refresh_token(&self) -> &SecretString {
&self.refresh_token
}
}
impl fmt::Debug for Renewal {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Renewal")
.field("refresh_token", &"[redacted]")
.field("access_expires_at", &self.access_expires_at)
.field("refresh_expires_at", &self.refresh_expires_at)
.finish()
}
}
impl UserAccessToken {
#[must_use]
pub fn new(token: SecretString) -> Self {
Self {
token,
token_type: "bearer".to_string(),
scope: None,
renewal: None,
}
}
pub(crate) fn from_parts(
token: SecretString,
token_type: String,
scope: Option<String>,
) -> Self {
Self {
token,
token_type,
scope,
renewal: None,
}
}
#[must_use]
pub(crate) fn with_renewal(
mut self,
refresh_token: Option<SecretString>,
access_expires_in: Option<u64>,
refresh_expires_in: Option<u64>,
) -> Self {
self.renewal = refresh_token.map(|refresh_token| {
let at = |secs: Option<u64>| {
secs.and_then(|s| i64::try_from(s).ok())
.and_then(|s| Utc::now().checked_add_signed(chrono::TimeDelta::seconds(s)))
};
Renewal {
refresh_token,
access_expires_at: at(access_expires_in),
refresh_expires_at: at(refresh_expires_in),
}
});
self
}
#[must_use]
pub fn renewal(&self) -> Option<&Renewal> {
self.renewal.as_ref()
}
#[must_use]
pub fn from_stored(token: SecretString) -> Self {
Self::from_stored_document(&token)
}
#[must_use]
pub fn from_stored_document(stored: &SecretString) -> Self {
#[derive(Deserialize)]
struct Document {
access_token: String,
#[serde(default)]
refresh_token: Option<String>,
#[serde(default)]
access_expires_at: Option<DateTime<Utc>>,
#[serde(default)]
refresh_expires_at: Option<DateTime<Utc>>,
}
match serde_json::from_str::<Document>(stored.expose_secret()) {
Ok(document) => Self {
token: SecretString::from(document.access_token),
token_type: "bearer".to_string(),
scope: None,
renewal: document.refresh_token.map(|refresh_token| Renewal {
refresh_token: SecretString::from(refresh_token),
access_expires_at: document.access_expires_at,
refresh_expires_at: document.refresh_expires_at,
}),
},
Err(_) => Self::new(stored.clone()),
}
}
#[must_use]
pub fn to_stored_document(&self) -> SecretString {
#[derive(Serialize)]
struct Document<'a> {
access_token: &'a str,
#[serde(skip_serializing_if = "Option::is_none")]
refresh_token: Option<&'a str>,
#[serde(skip_serializing_if = "Option::is_none")]
access_expires_at: Option<DateTime<Utc>>,
#[serde(skip_serializing_if = "Option::is_none")]
refresh_expires_at: Option<DateTime<Utc>>,
}
let document = Document {
access_token: self.token.expose_secret(),
refresh_token: self
.renewal
.as_ref()
.map(|r| r.refresh_token.expose_secret()),
access_expires_at: self.renewal.as_ref().and_then(|r| r.access_expires_at),
refresh_expires_at: self.renewal.as_ref().and_then(|r| r.refresh_expires_at),
};
SecretString::from(
serde_json::to_string(&document)
.unwrap_or_else(|_| self.token.expose_secret().to_string()),
)
}
#[must_use]
pub fn secret(&self) -> &SecretString {
&self.token
}
#[must_use]
pub fn token_type(&self) -> &str {
&self.token_type
}
#[must_use]
pub fn scope(&self) -> Option<&str> {
self.scope.as_deref()
}
#[must_use]
pub fn family(&self) -> &str {
let raw = self.token.expose_secret();
match raw.find('_') {
Some(idx) if idx < 8 => &raw[..=idx],
_ => "",
}
}
#[must_use]
pub fn is_user_to_server(&self) -> bool {
self.family() == "ghu_"
}
}
impl fmt::Debug for UserAccessToken {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("UserAccessToken")
.field("token", &"[REDACTED]")
.field("family", &self.family())
.finish_non_exhaustive()
}
}
impl PartialEq for UserAccessToken {
fn eq(&self, other: &Self) -> bool {
self.token.expose_secret() == other.token.expose_secret()
&& self.token_type == other.token_type
&& self.scope == other.scope
}
}
impl Eq for UserAccessToken {}
#[derive(Debug, thiserror::Error)]
pub enum GithubError {
#[error(
"GitHub rejected the stored credential; run `runner-manager auth login` to sign in again"
)]
AuthenticationFailed,
#[error(
"GitHub has temporarily locked out authentication for this credential; \
back off for {}s and do not retry — the credential itself is not the problem",
retry_after.as_secs()
)]
AuthenticationLockout { retry_after: Duration },
#[error(
"GitHub denied {method} {path}: the App installation does not grant it{}",
message.as_deref().map(|m| format!(" ({m})")).unwrap_or_default()
)]
Forbidden {
method: String,
path: String,
message: Option<String>,
headers: Box<HeaderMap>,
},
#[error(
"GitHub returned {status} for {method} {path}{}",
message.as_deref().map(|m| format!(": {m}")).unwrap_or_default()
)]
Status {
status: u16,
method: String,
path: String,
message: Option<String>,
headers: Box<HeaderMap>,
},
#[error("GitHub was unreachable")]
Transport(#[source] reqwest::Error),
#[error("a {what} response from GitHub could not be decoded as {expected}")]
Decode {
what: &'static str,
expected: &'static str,
#[source]
source: serde_json::Error,
},
#[error("GitHub returned {value:?} for {what}, which this client cannot use")]
Malformed { what: &'static str, value: String },
#[error(transparent)]
Config(#[from] ConfigError),
}
impl GithubError {
#[must_use]
pub fn is_authentication(&self) -> bool {
matches!(
self,
Self::AuthenticationFailed | Self::AuthenticationLockout { .. }
)
}
#[must_use]
pub fn is_lockout(&self) -> bool {
matches!(self, Self::AuthenticationLockout { .. })
}
#[must_use]
pub fn headers(&self) -> Option<&HeaderMap> {
match self {
Self::Status { headers, .. } | Self::Forbidden { headers, .. } => Some(headers),
_ => None,
}
}
#[must_use]
pub fn retry_after(&self) -> Option<Duration> {
self.headers().and_then(retry_after)
}
#[must_use]
pub fn rate_limit(&self) -> Option<RateLimitEvidence> {
let headers = self.headers()?;
let read = |name: &str| {
headers
.get(name)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.trim().parse::<u64>().ok())
};
let remaining = read("x-ratelimit-remaining");
let reset = read("x-ratelimit-reset");
if remaining.is_none() && reset.is_none() {
return None;
}
Some(RateLimitEvidence {
remaining,
reset_unix_secs: reset,
retry_after: self.retry_after(),
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RateLimitEvidence {
pub remaining: Option<u64>,
pub reset_unix_secs: Option<u64>,
pub retry_after: Option<Duration>,
}
fn transport(err: reqwest::Error) -> GithubError {
GithubError::Transport(err.without_url())
}
#[derive(Clone)]
pub struct ApiRequest {
method: Method,
path: String,
query: Vec<(String, String)>,
body: Option<serde_json::Value>,
}
impl ApiRequest {
#[must_use]
pub fn get(path: impl Into<String>) -> Self {
Self::new(Method::GET, path)
}
#[must_use]
pub fn delete(path: impl Into<String>) -> Self {
Self::new(Method::DELETE, path)
}
#[must_use]
pub fn new(method: Method, path: impl Into<String>) -> Self {
Self {
method,
path: path.into(),
query: Vec::new(),
body: None,
}
}
pub fn post_json<T: Serialize>(path: impl Into<String>, body: &T) -> Result<Self, GithubError> {
let value = serde_json::to_value(body).map_err(|source| GithubError::Decode {
what: "request",
expected: "JSON",
source,
})?;
Ok(Self {
method: Method::POST,
path: path.into(),
query: Vec::new(),
body: Some(value),
})
}
#[must_use]
pub fn query(mut self, key: impl Into<String>, value: impl fmt::Display) -> Self {
self.query.push((key.into(), value.to_string()));
self
}
#[must_use]
pub fn method(&self) -> &Method {
&self.method
}
#[must_use]
pub fn path(&self) -> &str {
&self.path
}
}
impl fmt::Debug for ApiRequest {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ApiRequest")
.field("method", &self.method.as_str())
.field("path", &self.path)
.field(
"query_keys",
&self.query.iter().map(|(k, _)| k).collect::<Vec<_>>(),
)
.field(
"body",
&self.body.as_ref().map_or("none", |_| "[REDACTED JSON]"),
)
.finish()
}
}
#[derive(Clone)]
pub struct ApiResponse {
status: StatusCode,
headers: HeaderMap,
body: Vec<u8>,
}
impl ApiResponse {
#[must_use]
pub fn status(&self) -> StatusCode {
self.status
}
#[must_use]
pub fn headers(&self) -> &HeaderMap {
&self.headers
}
#[must_use]
pub fn header(&self, name: &str) -> Option<&str> {
self.headers.get(name).and_then(|v| v.to_str().ok())
}
pub fn json<T: DeserializeOwned>(&self) -> Result<T, GithubError> {
serde_json::from_slice(&self.body).map_err(|source| GithubError::Decode {
what: "response",
expected: std::any::type_name::<T>(),
source,
})
}
#[must_use]
pub fn next_page(&self) -> Option<Url> {
self.header("link").and_then(parse_link_next)
}
}
impl fmt::Debug for ApiResponse {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ApiResponse")
.field("status", &self.status.as_u16())
.field("body_bytes", &self.body.len())
.finish_non_exhaustive()
}
}
fn parse_link_next(link: &str) -> Option<Url> {
let mut rest = link;
while let Some(open) = rest.find('<') {
let after_open = &rest[open + 1..];
let Some(close) = after_open.find('>') else {
return None;
};
let target = after_open[..close].trim();
let tail = &after_open[close + 1..];
let cut = tail
.match_indices(',')
.find(|(i, _)| tail[i + 1..].trim_start().starts_with('<'))
.map_or(tail.len(), |(i, _)| i);
let (params, next) = tail.split_at(cut);
let is_next = params.split(';').any(|param| {
let param = param.trim().replace(['"', '\''], "");
param.eq_ignore_ascii_case("rel=next")
});
if is_next {
return Url::parse(target).ok();
}
rest = next.strip_prefix(',').unwrap_or(next);
}
None
}
#[derive(Debug, Deserialize)]
struct ErrorEnvelope {
message: Option<String>,
}
fn error_message(body: &[u8]) -> Option<String> {
serde_json::from_slice::<ErrorEnvelope>(body)
.ok()
.and_then(|e| e.message)
.filter(|m| !m.is_empty())
}
fn retry_after(headers: &HeaderMap) -> Option<Duration> {
headers
.get("retry-after")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.trim().parse::<u64>().ok())
.map(Duration::from_secs)
}
#[async_trait::async_trait]
pub trait Sleeper: Send + Sync + fmt::Debug {
async fn sleep(&self, duration: Duration);
}
#[derive(Debug, Clone, Copy, Default)]
pub struct TokioSleeper;
#[async_trait::async_trait]
impl Sleeper for TokioSleeper {
async fn sleep(&self, duration: Duration) {
tokio::time::sleep(duration).await;
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Revalidation {
Valid,
Rejected,
Unavailable,
}
#[derive(Debug)]
struct LockoutState {
until: Option<Timestamp>,
backoff: Duration,
}
pub struct AuthenticatedClient {
http: reqwest::Client,
endpoints: Endpoints,
credential: std::sync::Mutex<UserAccessToken>,
source: Option<Arc<dyn CredentialSource>>,
renewal: Option<Arc<dyn CredentialRenewal>>,
clock: Arc<dyn Clock>,
revalidation_generation: AtomicU64,
revalidation_gate: tokio::sync::Mutex<()>,
last_revalidation: std::sync::Mutex<Revalidation>,
revalidations_performed: AtomicU64,
consecutive_unauthorized: AtomicU64,
lockout: std::sync::Mutex<LockoutState>,
}
impl fmt::Debug for AuthenticatedClient {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let locked_out = match self.lockout.try_lock() {
Ok(state) => {
if state.until.is_some_and(|until| self.clock.now() < until) {
"yes"
} else {
"no"
}
}
Err(_) => "unknown (the lockout state is being updated)",
};
f.debug_struct("AuthenticatedClient")
.field("api_base", &self.endpoints.api_base.as_str())
.field(
"credential",
&self
.credential
.try_lock()
.map_or("[in use]", |_| "[redacted]"),
)
.field(
"revalidations_performed",
&self.revalidations_performed.load(Ordering::Relaxed),
)
.field("locked_out", &locked_out)
.finish_non_exhaustive()
}
}
#[async_trait::async_trait]
pub trait CredentialRenewal: fmt::Debug + Send + Sync {
async fn renew(&self, refresh_token: &SecretString) -> Result<UserAccessToken, String>;
}
pub trait CredentialSource: fmt::Debug + Send + Sync {
fn reload(&self) -> Option<UserAccessToken>;
}
impl AuthenticatedClient {
pub fn new(
endpoints: Endpoints,
credential: UserAccessToken,
clock: Arc<dyn Clock>,
) -> Result<Self, GithubError> {
let http = reqwest::Client::builder()
.timeout(DEFAULT_REQUEST_TIMEOUT)
.build()
.map_err(transport)?;
Ok(Self::with_http_client(http, endpoints, credential, clock))
}
#[must_use]
pub fn with_http_client(
http: reqwest::Client,
endpoints: Endpoints,
credential: UserAccessToken,
clock: Arc<dyn Clock>,
) -> Self {
Self {
http,
endpoints,
credential: std::sync::Mutex::new(credential),
source: None,
renewal: None,
clock,
revalidation_generation: AtomicU64::new(0),
revalidation_gate: tokio::sync::Mutex::new(()),
last_revalidation: std::sync::Mutex::new(Revalidation::Valid),
revalidations_performed: AtomicU64::new(0),
consecutive_unauthorized: AtomicU64::new(0),
lockout: std::sync::Mutex::new(LockoutState {
until: None,
backoff: DEFAULT_LOCKOUT_BACKOFF,
}),
}
}
#[must_use]
pub fn endpoints(&self) -> &Endpoints {
&self.endpoints
}
#[must_use]
pub fn revalidations_performed(&self) -> u64 {
self.revalidations_performed.load(Ordering::SeqCst)
}
#[must_use]
pub fn is_locked_out(&self) -> bool {
self.lockout_remaining().is_some()
}
#[must_use]
pub fn lockout_remaining(&self) -> Option<Duration> {
let state = self.lockout.lock().expect("lockout lock poisoned");
let until = state.until?;
let now = self.clock.now();
if now >= until {
return None;
}
(until - now).to_std().ok()
}
pub fn clear_lockout(&self) {
self.lockout.lock().expect("lockout lock poisoned").until = None;
self.consecutive_unauthorized.store(0, Ordering::SeqCst);
}
pub async fn send(&self, request: &ApiRequest) -> Result<ApiResponse, GithubError> {
if let Some(remaining) = self.lockout_remaining() {
tracing::debug!(
method = request.method.as_str(),
path = %request.path,
remaining_secs = remaining.as_secs(),
"suppressed a request: GitHub authentication lockout is still backing off"
);
return Err(GithubError::AuthenticationLockout {
retry_after: remaining,
});
}
let first = self.send_raw(request).await?;
match self.classify(request, &first, Attempt::First) {
Classified::Ok => Ok(first),
Classified::Unauthorized => self.revalidate_and_retry_once(request).await,
Classified::Error(err) => Err(err),
}
}
pub async fn get_json<T: DeserializeOwned>(&self, path: &str) -> Result<T, GithubError> {
self.send(&ApiRequest::get(path)).await?.json()
}
fn bearer(&self) -> String {
self.credential
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.secret()
.expose_secret()
.to_string()
}
#[must_use]
pub fn with_renewal(mut self, renewal: Arc<dyn CredentialRenewal>) -> Self {
self.renewal = Some(renewal);
self
}
#[must_use]
pub fn with_credential_source(mut self, source: Arc<dyn CredentialSource>) -> Self {
self.source = Some(source);
self
}
async fn swap_credential_once<F>(&self, produce: F, note: &'static str) -> bool
where
F: AsyncFnOnce(&Self) -> Option<UserAccessToken>,
{
let before = self.revalidation_generation.load(Ordering::SeqCst);
let _gate = self.revalidation_gate.lock().await;
if self.revalidation_generation.load(Ordering::SeqCst) != before {
return true;
}
let Some(fresh) = produce(self).await else {
return false;
};
{
let mut guard = self
.credential
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if guard.secret().expose_secret() == fresh.secret().expose_secret() {
return false;
}
*guard = fresh;
}
self.revalidation_generation.fetch_add(1, Ordering::SeqCst);
tracing::info!("{note}");
true
}
async fn reload_once(&self) -> bool {
let Some(source) = self.source.clone() else {
return false;
};
self.swap_credential_once(
async |_| source.reload(),
"the stored credential changed and was picked up without a restart",
)
.await
}
async fn renew_once(&self) -> bool {
let Some(renewal) = self.renewal.clone() else {
return false;
};
self.swap_credential_once(
async |client: &Self| {
let refresh = {
let guard = client
.credential
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
guard.renewal().map(|r| r.refresh_token().clone())
}?;
match renewal.renew(&refresh).await {
Ok(fresh) => Some(fresh),
Err(error) => {
tracing::warn!(
%error,
"the user access token could not be renewed; an interactive \
sign-in may be required"
);
None
}
}
},
"the user access token was renewed",
)
.await
}
pub async fn post_json<B: Serialize, T: DeserializeOwned>(
&self,
path: &str,
body: &B,
) -> Result<T, GithubError> {
self.send(&ApiRequest::post_json(path, body)?).await?.json()
}
pub async fn revalidate(&self) -> Result<Revalidation, GithubError> {
self.revalidate_from(Attempt::First).await
}
async fn revalidate_after_unauthorized(&self) -> Result<Revalidation, GithubError> {
self.revalidate_from(Attempt::Retry).await
}
async fn revalidate_from(&self, attempt: Attempt) -> Result<Revalidation, GithubError> {
if let Some(retry_after) = self.lockout_remaining() {
return Err(GithubError::AuthenticationLockout { retry_after });
}
let sampled = self.revalidation_generation.load(Ordering::SeqCst);
let _guard = self.revalidation_gate.lock().await;
let outcome = if self.revalidation_generation.load(Ordering::SeqCst) != sampled {
let shared = *self
.last_revalidation
.lock()
.expect("re-validation lock poisoned");
tracing::debug!(
outcome = ?shared,
"reused an in-flight credential re-validation instead of starting another"
);
shared
} else {
self.revalidations_performed.fetch_add(1, Ordering::SeqCst);
let fresh = self.probe_credential(attempt).await;
*self
.last_revalidation
.lock()
.expect("re-validation lock poisoned") = fresh;
self.revalidation_generation.fetch_add(1, Ordering::SeqCst);
tracing::info!(
outcome = ?fresh,
"re-validated the stored credential; it carries no refresh half to renew"
);
fresh
};
if let Some(retry_after) = self.lockout_remaining() {
return Err(GithubError::AuthenticationLockout { retry_after });
}
Ok(outcome)
}
async fn probe_credential(&self, attempt: Attempt) -> Revalidation {
let probe = ApiRequest::get(REVALIDATION_PATH).query("per_page", 1);
match self.send_raw(&probe).await {
Ok(response) if response.status.is_success() => Revalidation::Valid,
Ok(response) if response.status == StatusCode::UNAUTHORIZED => {
self.consecutive_unauthorized.fetch_add(1, Ordering::SeqCst);
Revalidation::Rejected
}
Ok(response) if response.status == StatusCode::FORBIDDEN => {
if self.is_lockout_403(&response, attempt) {
self.latch_lockout(&response.headers);
}
Revalidation::Unavailable
}
Ok(_) | Err(_) => Revalidation::Unavailable,
}
}
async fn revalidate_and_retry_once(
&self,
request: &ApiRequest,
) -> Result<ApiResponse, GithubError> {
if self.renew_once().await || self.reload_once().await {
let second = self.send_raw(request).await?;
return match self.classify(request, &second, Attempt::Retry) {
Classified::Ok => Ok(second),
Classified::Unauthorized => Err(GithubError::AuthenticationFailed),
Classified::Error(err) => Err(err),
};
}
match self.revalidate_after_unauthorized().await? {
Revalidation::Rejected => {
tracing::warn!(
method = request.method.as_str(),
path = %request.path,
"GitHub rejected the stored credential; re-authentication is required"
);
Err(GithubError::AuthenticationFailed)
}
Revalidation::Valid | Revalidation::Unavailable => {
if let Some(remaining) = self.lockout_remaining() {
return Err(GithubError::AuthenticationLockout {
retry_after: remaining,
});
}
let second = self.send_raw(request).await?;
match self.classify(request, &second, Attempt::Retry) {
Classified::Ok => Ok(second),
Classified::Unauthorized => Err(GithubError::AuthenticationFailed),
Classified::Error(err) => Err(err),
}
}
}
}
fn classify(
&self,
request: &ApiRequest,
response: &ApiResponse,
attempt: Attempt,
) -> Classified {
let status = response.status;
if status.is_success() {
self.consecutive_unauthorized.store(0, Ordering::SeqCst);
return Classified::Ok;
}
if status == StatusCode::UNAUTHORIZED {
self.consecutive_unauthorized.fetch_add(1, Ordering::SeqCst);
return Classified::Unauthorized;
}
if status == StatusCode::FORBIDDEN && self.is_lockout_403(response, attempt) {
let backoff = self.latch_lockout(&response.headers);
tracing::warn!(
method = request.method.as_str(),
path = %request.path,
backoff_secs = backoff.as_secs(),
"GitHub answered 403 after 401s: temporary authentication lockout, backing off"
);
return Classified::Error(GithubError::AuthenticationLockout {
retry_after: backoff,
});
}
let headers = Box::new(response.headers.clone());
let message = error_message(&response.body);
if status == StatusCode::FORBIDDEN {
return Classified::Error(GithubError::Forbidden {
method: request.method.as_str().to_string(),
path: request.path.clone(),
message,
headers,
});
}
Classified::Error(GithubError::Status {
status: status.as_u16(),
method: request.method.as_str().to_string(),
path: request.path.clone(),
message,
headers,
})
}
fn is_lockout_403(&self, response: &ApiResponse, attempt: Attempt) -> bool {
if is_rate_limited(response) {
return false;
}
match attempt {
Attempt::Retry => true,
Attempt::First => is_lockout_continuation(response),
}
}
fn latch_lockout(&self, headers: &HeaderMap) -> Duration {
let requested = retry_after(headers).unwrap_or(DEFAULT_LOCKOUT_BACKOFF);
let clamped = requested.min(MAX_LOCKOUT_BACKOFF);
let delta = chrono::TimeDelta::from_std(clamped).unwrap_or_else(|_| {
chrono::TimeDelta::from_std(DEFAULT_LOCKOUT_BACKOFF)
.expect("sixty seconds is a representable span")
});
let backoff = delta.to_std().unwrap_or(DEFAULT_LOCKOUT_BACKOFF);
let mut state = self.lockout.lock().expect("lockout lock poisoned");
state.backoff = backoff;
state.until = Some(self.clock.now() + delta);
backoff
}
async fn send_raw(&self, request: &ApiRequest) -> Result<ApiResponse, GithubError> {
let url = self.resolve(&request.path)?;
let mut builder = self
.http
.request(request.method.clone(), url)
.header(reqwest::header::ACCEPT, GITHUB_ACCEPT)
.header(reqwest::header::USER_AGENT, USER_AGENT)
.header("X-GitHub-Api-Version", GITHUB_API_VERSION)
.header(
reqwest::header::AUTHORIZATION,
format!("Bearer {}", self.bearer()),
);
if !request.query.is_empty() {
builder = builder.query(&request.query);
}
if let Some(body) = &request.body {
builder = builder.json(body);
}
let response = builder.send().await.map_err(transport)?;
let status = response.status();
let headers = response.headers().clone();
let body = response.bytes().await.map_err(transport)?.to_vec();
tracing::debug!(
method = request.method.as_str(),
path = %request.path,
status = status.as_u16(),
body_bytes = body.len(),
"github api request"
);
Ok(ApiResponse {
status,
headers,
body,
})
}
fn resolve(&self, path: &str) -> Result<Url, GithubError> {
if path.starts_with("http://") || path.starts_with("https://") {
return Url::parse(path).map_err(|_| GithubError::Malformed {
what: "an absolute request URL",
value: path.to_string(),
});
}
self.endpoints
.api_base
.join(path.trim_start_matches('/'))
.map_err(|_| GithubError::Malformed {
what: "a request path",
value: path.to_string(),
})
}
}
enum Classified {
Ok,
Unauthorized,
Error(GithubError),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Attempt {
First,
Retry,
}
fn is_rate_limited(response: &ApiResponse) -> bool {
if response.status == StatusCode::TOO_MANY_REQUESTS {
return true;
}
if response
.header("x-ratelimit-remaining")
.is_some_and(|v| v.trim() == "0")
{
return true;
}
error_message(&response.body).is_some_and(|m| m.to_ascii_lowercase().contains("rate limit"))
}
fn is_lockout_continuation(response: &ApiResponse) -> bool {
response.headers.contains_key("retry-after") && error_message(&response.body).is_none()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RepositorySelection {
All,
Selected,
}
impl RepositorySelection {
#[must_use]
pub fn is_over_broad(self) -> bool {
matches!(self, Self::All)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum InstallationAccount {
User(String),
Organization(Org),
Enterprise(String),
}
impl InstallationAccount {
#[must_use]
pub fn login(&self) -> &str {
match self {
Self::User(login) | Self::Enterprise(login) => login,
Self::Organization(org) => org.as_str(),
}
}
#[must_use]
pub fn organization(&self) -> Option<&Org> {
match self {
Self::Organization(org) => Some(org),
Self::User(_) | Self::Enterprise(_) => None,
}
}
#[must_use]
pub fn kind(&self) -> &'static str {
match self {
Self::User(_) => "user",
Self::Organization(_) => "organization",
Self::Enterprise(_) => "enterprise",
}
}
}
impl fmt::Display for InstallationAccount {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.login())
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Installation {
pub id: u64,
pub account: InstallationAccount,
pub repository_selection: RepositorySelection,
pub repositories: Vec<OwnerRepo>,
pub permissions: Vec<(String, String)>,
}
impl Installation {
#[must_use]
pub fn is_over_broad(&self) -> bool {
self.repository_selection.is_over_broad()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReachableTargets {
installations: Vec<Installation>,
skipped: usize,
}
impl ReachableTargets {
#[must_use]
pub fn installations(&self) -> &[Installation] {
&self.installations
}
#[must_use]
pub fn skipped(&self) -> usize {
self.skipped
}
#[must_use]
pub fn repositories(&self) -> Vec<OwnerRepo> {
let mut all: Vec<OwnerRepo> = self
.installations
.iter()
.flat_map(|i| i.repositories.iter().cloned())
.collect();
all.sort();
all.dedup();
all
}
#[must_use]
pub fn organizations(&self) -> Vec<Org> {
let mut all: Vec<Org> = self
.installations
.iter()
.filter_map(|i| i.account.organization().cloned())
.collect();
all.sort();
all.dedup();
all
}
#[must_use]
pub fn over_broad(&self) -> Vec<&Installation> {
self.installations
.iter()
.filter(|i| i.is_over_broad())
.collect()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.repositories().is_empty() && self.organizations().is_empty()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum InstallationDiscovery {
NotInstalled { install_url: Url },
Indeterminate { skipped: usize },
Installed(ReachableTargets),
}
impl InstallationDiscovery {
#[must_use]
pub fn targets(&self) -> Option<&ReachableTargets> {
match self {
Self::Installed(t) => Some(t),
Self::NotInstalled { .. } | Self::Indeterminate { .. } => None,
}
}
#[must_use]
pub fn install_url(&self) -> Option<&Url> {
match self {
Self::NotInstalled { install_url } => Some(install_url),
Self::Installed(_) | Self::Indeterminate { .. } => None,
}
}
#[must_use]
pub fn skipped(&self) -> usize {
match self {
Self::NotInstalled { .. } => 0,
Self::Indeterminate { skipped } => *skipped,
Self::Installed(targets) => targets.skipped(),
}
}
}
#[derive(Debug, Deserialize)]
struct InstallationsPage {
#[serde(default)]
total_count: Option<u64>,
#[serde(default)]
installations: Vec<RawInstallation>,
}
#[derive(Debug, Deserialize)]
struct RawInstallation {
id: u64,
#[serde(default)]
account: Option<RawAccount>,
#[serde(default)]
repository_selection: Option<String>,
#[serde(default)]
permissions: std::collections::BTreeMap<String, String>,
}
#[derive(Debug, Deserialize)]
struct RawAccount {
#[serde(default)]
login: Option<String>,
#[serde(default)]
slug: Option<String>,
#[serde(default)]
name: Option<String>,
#[serde(rename = "type", default)]
account_type: Option<String>,
}
impl RawAccount {
fn display_login(&self) -> Option<&str> {
[
self.login.as_deref(),
self.slug.as_deref(),
self.name.as_deref(),
]
.into_iter()
.flatten()
.find(|value| !value.is_empty())
}
fn is_enterprise_shaped(&self) -> bool {
self.login.as_deref().is_none_or(str::is_empty)
&& (self.slug.as_deref().is_some_and(|s| !s.is_empty())
|| self.name.as_deref().is_some_and(|s| !s.is_empty()))
}
}
#[derive(Debug, Deserialize)]
struct RepositoriesPage {
#[serde(default)]
total_count: Option<u64>,
#[serde(default)]
repositories: Vec<RawRepository>,
}
#[derive(Debug, Deserialize)]
struct RawRepository {
full_name: String,
}
fn under_collected(collected: usize, total_count: Option<u64>) -> Option<u64> {
let total = total_count?;
(total > collected as u64).then_some(total)
}
impl AuthenticatedClient {
pub async fn discover_installations(
&self,
app: &AppRegistration,
) -> Result<InstallationDiscovery, GithubError> {
let mut installations = Vec::new();
let mut skipped = 0_usize;
for raw in self.all_installations().await? {
let Some(login) = raw.account.as_ref().and_then(RawAccount::display_login) else {
skipped += 1;
tracing::warn!(
installation_id = raw.id,
"skipping an installation GitHub reported with no nameable account; \
anything it reaches is missing from this report"
);
continue;
};
let account_type = raw.account.as_ref().and_then(|a| a.account_type.as_deref());
let account = match account_type {
Some("Organization") => {
InstallationAccount::Organization(Org::new(login).map_err(|_| {
GithubError::Malformed {
what: "an installation account login",
value: login.to_string(),
}
})?)
}
Some("Enterprise") => InstallationAccount::Enterprise(login.to_string()),
_ if raw
.account
.as_ref()
.is_some_and(RawAccount::is_enterprise_shaped) =>
{
InstallationAccount::Enterprise(login.to_string())
}
_ => InstallationAccount::User(login.to_string()),
};
let repository_selection = match raw.repository_selection.as_deref() {
Some("all") => RepositorySelection::All,
_ => RepositorySelection::Selected,
};
installations.push(Installation {
id: raw.id,
account,
repository_selection,
repositories: self.installation_repositories(raw.id).await?,
permissions: raw.permissions.into_iter().collect(),
});
}
let targets = ReachableTargets {
installations,
skipped,
};
if targets.is_empty() {
if skipped > 0 {
tracing::warn!(
skipped,
"every installation GitHub reported was skipped; whether the App is \
installed cannot be determined from this credential"
);
return Ok(InstallationDiscovery::Indeterminate { skipped });
}
let install_url = app.install_url(&self.endpoints);
tracing::info!(
install_url = %install_url,
"the published App is not installed on anything this credential can reach"
);
return Ok(InstallationDiscovery::NotInstalled { install_url });
}
tracing::info!(
repositories = targets.repositories().len(),
organizations = targets.organizations().len(),
over_broad = targets.over_broad().len(),
skipped,
"discovered the targets this credential can reach"
);
Ok(InstallationDiscovery::Installed(targets))
}
async fn all_installations(&self) -> Result<Vec<RawInstallation>, GithubError> {
let mut out = Vec::new();
let mut total_count = None;
let mut next = Some(ApiRequest::get("/user/installations").query("per_page", 100));
let mut pages = 0_usize;
while let Some(request) = next.take() {
let response = self.send(&request).await?;
let page: InstallationsPage = response.json()?;
total_count = page.total_count.or(total_count);
out.extend(page.installations);
pages += 1;
if pages >= MAX_PAGES {
tracing::warn!(
pages,
collected = out.len(),
"stopped following installation pages at the ceiling; a `Link: rel=next` \
that never ends would otherwise loop forever"
);
break;
}
next = response
.next_page()
.map(|url| ApiRequest::get(url.as_str()));
}
if let Some(expected) = under_collected(out.len(), total_count) {
tracing::warn!(
expected,
collected = out.len(),
"GitHub reported more installations than pagination collected; the reported \
reach is incomplete"
);
}
Ok(out)
}
async fn installation_repositories(&self, id: u64) -> Result<Vec<OwnerRepo>, GithubError> {
let mut out = Vec::new();
let mut total_count = None;
let mut next = Some(
ApiRequest::get(format!("/user/installations/{id}/repositories"))
.query("per_page", 100),
);
let mut pages = 0_usize;
while let Some(request) = next.take() {
let response = self.send(&request).await?;
let page: RepositoriesPage = response.json()?;
total_count = page.total_count.or(total_count);
for repo in page.repositories {
out.push(OwnerRepo::parse(&repo.full_name).map_err(|_| {
GithubError::Malformed {
what: "a repository full_name",
value: repo.full_name.clone(),
}
})?);
}
pages += 1;
if pages >= MAX_PAGES {
tracing::warn!(
installation_id = id,
pages,
collected = out.len(),
"stopped following repository pages at the ceiling; a `Link: rel=next` \
that never ends would otherwise loop forever"
);
break;
}
next = response
.next_page()
.map(|url| ApiRequest::get(url.as_str()));
}
if let Some(expected) = under_collected(out.len(), total_count) {
tracing::warn!(
installation_id = id,
expected,
collected = out.len(),
"GitHub reported more repositories than pagination collected; this \
installation's reach is under-reported"
);
}
Ok(out)
}
}
#[cfg(test)]
pub(crate) mod testing {
use super::*;
use serde_json::{Value, json};
use std::sync::{Mutex, atomic::AtomicUsize};
use wiremock::{Request, Respond, ResponseTemplate};
pub const FIXTURE_TOKEN: &str = "ghu_fixtureTOKENnotARealCredential00";
pub const FIXTURE_DEVICE_CODE: &str = "fixture-device-code-0e37a9c1b4d84f2a";
pub const FIXTURE_USER_CODE: &str = "WDJB-MJHT";
#[derive(Debug)]
pub struct TestClock {
now: Mutex<Timestamp>,
}
impl TestClock {
pub fn advance_secs(&self, secs: i64) {
let mut now = self.now.lock().expect("TestClock lock poisoned");
*now += chrono::TimeDelta::seconds(secs);
}
}
impl Default for TestClock {
fn default() -> Self {
Self {
now: Mutex::new(
chrono::DateTime::from_timestamp(1_787_270_400, 0).expect("a valid instant"),
),
}
}
}
impl Clock for TestClock {
fn now(&self) -> Timestamp {
*self.now.lock().expect("TestClock lock poisoned")
}
}
#[derive(Debug, Default)]
pub struct RecordingSleeper {
recorded: Mutex<Vec<Duration>>,
}
impl RecordingSleeper {
pub fn recorded(&self) -> Vec<Duration> {
self.recorded.lock().expect("sleeper lock poisoned").clone()
}
}
#[async_trait::async_trait]
impl Sleeper for RecordingSleeper {
async fn sleep(&self, duration: Duration) {
self.recorded
.lock()
.expect("sleeper lock poisoned")
.push(duration);
}
}
pub struct Script {
responses: Vec<ResponseTemplate>,
calls: AtomicUsize,
}
impl Script {
#[must_use]
pub fn new(responses: Vec<ResponseTemplate>) -> Self {
assert!(
!responses.is_empty(),
"a script needs at least one response"
);
Self {
responses,
calls: AtomicUsize::new(0),
}
}
}
impl Respond for Script {
fn respond(&self, _: &Request) -> ResponseTemplate {
let i = self.calls.fetch_add(1, Ordering::SeqCst);
self.responses[i.min(self.responses.len() - 1)].clone()
}
}
#[must_use]
pub fn device_code_body(server_uri: &str, interval: u64, expires_in: u64) -> Value {
json!({
"device_code": FIXTURE_DEVICE_CODE,
"user_code": FIXTURE_USER_CODE,
"verification_uri": format!("{server_uri}/login/device"),
"expires_in": expires_in,
"interval": interval
})
}
#[must_use]
pub fn error_body(code: &str, interval: Option<u64>) -> Value {
let mut body = json!({
"error": code,
"error_description": "see the OAuth 2.0 Device Authorization Grant",
"error_uri": "https://docs.github.com/developers/apps/authorizing-oauth-apps"
});
if let Some(interval) = interval {
body["interval"] = json!(interval);
}
body
}
#[must_use]
pub fn token_body() -> Value {
json!({ "access_token": FIXTURE_TOKEN, "token_type": "bearer", "scope": "" })
}
#[must_use]
pub fn installations_body(entries: &[(u64, &str, &str, &str)]) -> Value {
let installations: Vec<Value> = entries
.iter()
.map(|(id, login, account_type, selection)| {
json!({
"id": id,
"account": { "login": login, "type": account_type },
"repository_selection": selection,
"permissions": {
"actions": "read",
"administration": "write",
"metadata": "read",
"organization_self_hosted_runners": "write"
}
})
})
.collect();
json!({ "total_count": installations.len(), "installations": installations })
}
#[must_use]
pub fn repositories_body(full_names: &[&str]) -> Value {
let repositories: Vec<Value> = full_names
.iter()
.map(|full_name| json!({ "full_name": full_name }))
.collect();
json!({ "total_count": repositories.len(), "repositories": repositories })
}
}
#[cfg(test)]
mod tests {
#[test]
fn both_stored_shapes_load_and_a_pair_survives_a_round_trip() {
let legacy = UserAccessToken::from_stored_document(&SecretString::from("ghu_legacy123"));
assert_eq!(legacy.secret().expose_secret(), "ghu_legacy123");
assert!(
legacy.renewal().is_none(),
"a bare token has no renewal half, and inventing one would make the client try to refresh a credential the App never issued a refresh token for"
);
let pair = UserAccessToken::new(SecretString::from("ghu_new")).with_renewal(
Some(SecretString::from("ghr_new")),
Some(28_800),
Some(15_897_600),
);
let stored = pair.to_stored_document();
let read = UserAccessToken::from_stored_document(&stored);
assert_eq!(read.secret().expose_secret(), "ghu_new");
let renewal = read.renewal().expect("the pair survives the round trip");
assert_eq!(renewal.refresh_token().expose_secret(), "ghr_new");
assert!(renewal.access_expires_at.is_some());
assert!(renewal.refresh_expires_at.is_some());
let bare_round_trip = UserAccessToken::from_stored_document(&legacy.to_stored_document());
assert_eq!(bare_round_trip.secret().expose_secret(), "ghu_legacy123");
assert!(bare_round_trip.renewal().is_none());
}
#[test]
fn a_refresh_token_never_appears_in_debug_output() {
let pair = UserAccessToken::new(SecretString::from("ghu_x")).with_renewal(
Some(SecretString::from("ghr_SUPERSECRET")),
Some(1),
Some(2),
);
let rendered = format!("{:?}", pair.renewal().expect("a renewal"));
assert!(!rendered.contains("ghr_SUPERSECRET"), "{rendered}");
assert!(rendered.contains("redacted"), "{rendered}");
}
use super::*;
use crate::testing::{FIXTURE_TOKEN, Script, TestClock, installations_body, repositories_body};
use serde_json::json;
use wiremock::{
Mock, MockServer, ResponseTemplate,
matchers::{header, method, path},
};
fn client(server: &MockServer, clock: Arc<TestClock>) -> AuthenticatedClient {
AuthenticatedClient::new(
Endpoints::for_test_server(&server.uri()).unwrap(),
UserAccessToken::new(SecretString::from(FIXTURE_TOKEN)),
clock,
)
.unwrap()
}
fn app() -> AppRegistration {
AppRegistration::new("Iv23liTESTCLIENTID", "runner-manager").unwrap()
}
#[derive(Debug)]
struct StoreHolding(Option<&'static str>);
impl CredentialSource for StoreHolding {
fn reload(&self) -> Option<UserAccessToken> {
self.0
.map(|token| UserAccessToken::new(SecretString::from(token)))
}
}
#[tokio::test]
async fn a_daemon_picks_up_a_sign_in_that_happened_after_it_started() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/repos/acme/app"))
.and(header("authorization", "Bearer ghu_dead"))
.respond_with(ResponseTemplate::new(401))
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/repos/acme/app"))
.and(header("authorization", "Bearer ghu_freshly_signed_in"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({"id": 1})))
.expect(1)
.mount(&server)
.await;
let client = AuthenticatedClient::new(
Endpoints::for_test_server(&server.uri()).unwrap(),
UserAccessToken::new(SecretString::from("ghu_dead")),
Arc::new(TestClock::default()),
)
.unwrap()
.with_credential_source(Arc::new(StoreHolding(Some("ghu_freshly_signed_in"))));
client
.send(&ApiRequest::get("/repos/acme/app"))
.await
.expect(
"the 401 is retried with what the store holds now, without anybody \n restarting the daemon",
);
}
#[tokio::test]
async fn a_store_holding_the_same_dead_token_is_not_worth_a_retry() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/repos/acme/app"))
.respond_with(ResponseTemplate::new(401))
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(401))
.expect(1)
.mount(&server)
.await;
let client = AuthenticatedClient::new(
Endpoints::for_test_server(&server.uri()).unwrap(),
UserAccessToken::new(SecretString::from("ghu_revoked")),
Arc::new(TestClock::default()),
)
.unwrap()
.with_credential_source(Arc::new(StoreHolding(Some("ghu_revoked"))));
let failure = client
.send(&ApiRequest::get("/repos/acme/app"))
.await
.expect_err("a revoked credential is still revoked when the store agrees");
assert!(
matches!(failure, GithubError::AuthenticationFailed),
"{failure:?}"
);
}
#[tokio::test]
async fn an_unreadable_store_changes_nothing_about_the_rejection() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/repos/acme/app"))
.respond_with(ResponseTemplate::new(401))
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(401))
.expect(1)
.mount(&server)
.await;
let client = AuthenticatedClient::new(
Endpoints::for_test_server(&server.uri()).unwrap(),
UserAccessToken::new(SecretString::from("ghu_revoked")),
Arc::new(TestClock::default()),
)
.unwrap()
.with_credential_source(Arc::new(StoreHolding(None)));
let failure = client
.send(&ApiRequest::get("/repos/acme/app"))
.await
.expect_err("nothing to pick up means the rejection stands");
assert!(
matches!(failure, GithubError::AuthenticationFailed),
"{failure:?}"
);
}
#[tokio::test]
async fn every_request_states_its_api_version_and_accept_header() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.and(header("x-github-api-version", GITHUB_API_VERSION))
.and(header("accept", GITHUB_ACCEPT))
.and(header("authorization", format!("Bearer {FIXTURE_TOKEN}")))
.and(header("user-agent", USER_AGENT))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.expect(1)
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
client
.send(&ApiRequest::get("/user/installations"))
.await
.expect("the mock only matches when all four headers are present");
}
#[tokio::test]
async fn a_401_revalidates_once_and_retries_once_then_succeeds() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(Script::new(vec![
ResponseTemplate::new(401).set_body_json(json!({"message": "Bad credentials"})),
ResponseTemplate::new(200).set_body_json(json!({"total_count": 0})),
]))
.expect(2)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.expect(1)
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let response = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect("the retry succeeds");
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
client.revalidations_performed(),
1,
"one 401 must produce exactly one re-validation"
);
}
#[tokio::test]
async fn a_second_401_after_the_retry_is_terminal_authentication_failure() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(ResponseTemplate::new(401))
.expect(2)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("two 401s is terminal");
assert!(matches!(err, GithubError::AuthenticationFailed), "{err:?}");
assert!(err.is_authentication());
assert!(!err.is_lockout());
}
#[tokio::test]
async fn a_rejected_revalidation_fails_without_spending_the_retry() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(ResponseTemplate::new(401))
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(401))
.expect(1)
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("the credential is dead");
assert!(matches!(err, GithubError::AuthenticationFailed), "{err:?}");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn eight_concurrent_401s_produce_one_revalidation_not_eight() {
const CALLERS: usize = 8;
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(ResponseTemplate::new(401))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(installations_body(&[]))
.set_delay(Duration::from_millis(250)),
)
.expect(1)
.mount(&server)
.await;
let client = Arc::new(client(&server, Arc::new(TestClock::default())));
let barrier = Arc::new(tokio::sync::Barrier::new(CALLERS));
let mut tasks = Vec::new();
for _ in 0..CALLERS {
let client = Arc::clone(&client);
let barrier = Arc::clone(&barrier);
tasks.push(tokio::spawn(async move {
barrier.wait().await;
client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("every caller sees a dead endpoint")
}));
}
let mut outcomes = Vec::new();
for task in tasks {
outcomes.push(task.await.expect("no caller panicked"));
}
assert_eq!(outcomes.len(), CALLERS);
for err in &outcomes {
assert!(matches!(err, GithubError::AuthenticationFailed), "{err:?}");
}
assert_eq!(
client.revalidations_performed(),
1,
"{CALLERS} concurrent 401s must produce ONE attempt, not {CALLERS}"
);
let seen = server.received_requests().await.expect("recording is on");
let probes = seen
.iter()
.filter(|r| r.url.path() == "/user/installations")
.count();
assert_eq!(probes, 1, "GitHub itself saw exactly one re-validation");
let attempts = seen
.iter()
.filter(|r| r.url.path() == "/orgs/acme/actions/runners")
.count();
assert_eq!(
attempts,
CALLERS * 2,
"each caller still gets its own single retry"
);
}
#[tokio::test]
async fn a_403_after_401s_is_a_lockout_and_not_an_authentication_failure() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(Script::new(vec![
ResponseTemplate::new(401),
ResponseTemplate::new(403).insert_header("retry-after", "42"),
]))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("403 after a 401");
match err {
GithubError::AuthenticationLockout { retry_after } => {
assert_eq!(retry_after, Duration::from_secs(42), "honours retry-after");
}
other => panic!("expected a lockout, got {other:?}"),
}
assert!(client.is_locked_out());
}
#[tokio::test]
async fn a_403_with_no_preceding_401_is_a_permissions_answer_not_a_lockout() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(
ResponseTemplate::new(403)
.set_body_json(json!({"message": "Resource not accessible by integration"})),
)
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("403");
assert!(matches!(err, GithubError::Forbidden { .. }), "{err:?}");
assert!(!err.is_lockout());
assert!(!err.is_authentication());
assert!(!client.is_locked_out(), "a permissions 403 must not latch");
}
#[tokio::test]
async fn a_locked_out_client_issues_no_further_http_until_the_backoff_elapses() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(Script::new(vec![
ResponseTemplate::new(401),
ResponseTemplate::new(403).insert_header("retry-after", "60"),
ResponseTemplate::new(200).set_body_json(json!({"total_count": 0})),
]))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.mount(&server)
.await;
let clock = Arc::new(TestClock::default());
let client = client(&server, Arc::clone(&clock));
let request = ApiRequest::get("/orgs/acme/actions/runners");
let err = client.send(&request).await.expect_err("locks out");
assert!(err.is_lockout(), "{err:?}");
let after_lockout = server.received_requests().await.unwrap().len();
for _ in 0..3 {
let err = client.send(&request).await.expect_err("still locked out");
assert!(err.is_lockout(), "{err:?}");
}
assert_eq!(
server.received_requests().await.unwrap().len(),
after_lockout,
"a backed-off client must open no sockets at all"
);
clock.advance_secs(61);
assert!(!client.is_locked_out(), "the back-off expires on the clock");
let response = client.send(&request).await.expect("traffic resumes");
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
server.received_requests().await.unwrap().len(),
after_lockout + 1
);
}
#[tokio::test]
async fn a_stale_401_does_not_turn_a_later_permissions_403_into_a_lockout() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(Script::new(vec![
ResponseTemplate::new(401),
ResponseTemplate::new(404).set_body_json(json!({"message": "Not Found"})),
]))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/orgs/acme/actions/runners/generate-jitconfig"))
.respond_with(
ResponseTemplate::new(403)
.set_body_json(json!({"message": "Resource not accessible by integration"})),
)
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("404");
assert!(
matches!(err, GithubError::Status { status: 404, .. }),
"{err:?}"
);
let err = client
.send(&ApiRequest::new(
Method::POST,
"/orgs/acme/actions/runners/generate-jitconfig",
))
.await
.expect_err("403");
assert!(
matches!(err, GithubError::Forbidden { .. }),
"a fresh first-attempt 403 is a permissions answer, not a lockout: {err:?}"
);
assert!(!err.is_lockout());
assert!(
!client.is_locked_out(),
"a stale 401 must not be able to silence the client for a minute"
);
}
#[tokio::test]
async fn a_rate_limited_403_is_not_reported_as_an_authentication_lockout() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(Script::new(vec![
ResponseTemplate::new(401),
ResponseTemplate::new(403)
.insert_header("x-ratelimit-remaining", "0")
.insert_header("x-ratelimit-reset", "1787270460")
.insert_header("retry-after", "30")
.set_body_json(json!({"message": "API rate limit exceeded"})),
]))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("rate limited");
assert!(
matches!(err, GithubError::Forbidden { .. }),
"a rate limit is not an authentication outcome: {err:?}"
);
assert!(!err.is_lockout());
assert!(!err.is_authentication());
assert!(
!client.is_locked_out(),
"a rate limit must not latch this crate's authentication back-off"
);
let evidence = err
.rate_limit()
.expect("the headers survived classification");
assert_eq!(evidence.remaining, Some(0));
assert_eq!(evidence.reset_unix_secs, Some(1_787_270_460));
assert_eq!(evidence.retry_after, Some(Duration::from_secs(30)));
}
#[tokio::test]
async fn a_429_carries_its_retry_after_across_the_c2_c3_seam() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(
ResponseTemplate::new(429)
.insert_header("retry-after", "17")
.insert_header("x-ratelimit-remaining", "0")
.set_body_json(json!({"message": "You have exceeded a secondary rate limit"})),
)
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("429");
assert!(
matches!(err, GithubError::Status { status: 429, .. }),
"{err:?}"
);
assert_eq!(
err.retry_after(),
Some(Duration::from_secs(17)),
"destroying this header is what made `c3`'s Definition of Done unmeetable"
);
assert_eq!(
err.headers().and_then(|h| h.get("x-ratelimit-remaining")),
Some(&reqwest::header::HeaderValue::from_static("0"))
);
}
#[tokio::test]
async fn an_extreme_retry_after_is_clamped_and_never_fails_open() {
async fn lockout_for(header: &str) -> (GithubError, bool, Option<Duration>) {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(Script::new(vec![
ResponseTemplate::new(401),
ResponseTemplate::new(403).insert_header("retry-after", header),
]))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.mount(&server)
.await;
let client = AuthenticatedClient::new(
Endpoints::for_test_server(&server.uri()).unwrap(),
UserAccessToken::new(SecretString::from(FIXTURE_TOKEN)),
Arc::new(TestClock::default()),
)
.unwrap();
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("403 after a 401");
let locked = client.is_locked_out();
let remaining = client.lockout_remaining();
(err, locked, remaining)
}
let (err, locked, remaining) = lockout_for("86400").await;
let GithubError::AuthenticationLockout { retry_after } = &err else {
panic!("expected a lockout, got {err:?}");
};
assert_eq!(
*retry_after, MAX_LOCKOUT_BACKOFF,
"an unclamped Retry-After lets a remote party decide how long this product \
stays down"
);
assert!(locked);
assert!(remaining.is_some_and(|r| r <= MAX_LOCKOUT_BACKOFF));
let (err, locked, remaining) = lockout_for(&u64::MAX.to_string()).await;
assert!(err.is_lockout(), "{err:?}");
assert!(
locked,
"an absurd Retry-After must not mean `not locked out at all` — that fails open"
);
assert!(remaining.is_some_and(|r| r <= MAX_LOCKOUT_BACKOFF));
}
#[tokio::test]
async fn a_direct_revalidation_is_refused_while_the_lockout_is_backing_off() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(Script::new(vec![
ResponseTemplate::new(401),
ResponseTemplate::new(403).insert_header("retry-after", "60"),
]))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("locks out");
assert!(client.is_locked_out());
let before = server.received_requests().await.unwrap().len();
let err = client
.revalidate()
.await
.expect_err("the documented lockout error is now reachable");
assert!(err.is_lockout(), "{err:?}");
assert_eq!(
server.received_requests().await.unwrap().len(),
before,
"a locked-out client opens no socket, and the probe is not an exception"
);
}
#[tokio::test]
async fn a_directly_requested_probe_does_not_latch_a_lockout_from_a_stale_401() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(Script::new(vec![
ResponseTemplate::new(401),
ResponseTemplate::new(404).set_body_json(json!({"message": "Not Found"})),
]))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(Script::new(vec![
ResponseTemplate::new(200).set_body_json(installations_body(&[])),
ResponseTemplate::new(403)
.set_body_json(json!({"message": "Resource not accessible by integration"})),
]))
.mount(&server)
.await;
let clock = Arc::new(TestClock::default());
let client = client(&server, clock.clone());
client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("the retry 404s");
assert!(
!client.is_locked_out(),
"a 404 on the retry is not a lockout"
);
clock.advance_secs(300);
let outcome = client
.revalidate()
.await
.expect("a direct probe is not a lockout error");
assert_eq!(
outcome,
Revalidation::Unavailable,
"a 403 on the probe teaches this client nothing about the credential"
);
assert!(
!client.is_locked_out(),
"a caller-initiated probe is a *first* attempt, not the retry that follows a 401: \
latching here converts a stale 401 into a 60-second client-wide outage and \
reports a missing permission as `the credential is fine, please wait`"
);
assert_eq!(client.lockout_remaining(), None);
}
#[tokio::test]
async fn a_directly_requested_probe_latches_a_continuation_shaped_lockout() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(403).insert_header("retry-after", "60"))
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let err = client
.revalidate()
.await
.expect_err("a probe that latches a lockout reports it rather than `Unavailable`");
assert!(
err.is_lockout(),
"`revalidate` latched a client-wide lockout and must say so; answering \
`Ok(Unavailable)` leaves `f1` to discover a 15-minute outage through a separate \
`is_locked_out()` call it has no reason to make: {err:?}"
);
assert!(
client.is_locked_out(),
"a 403 carrying `retry-after` with no message is GitHub continuing a lockout, \
whoever asked for the request that met it"
);
assert_eq!(client.lockout_remaining(), Some(Duration::from_secs(60)));
let before = server.received_requests().await.unwrap().len();
let err = client.revalidate().await.expect_err("still locked out");
assert!(err.is_lockout(), "{err:?}");
assert_eq!(
server.received_requests().await.unwrap().len(),
before,
"latching must actually stop traffic"
);
}
#[tokio::test]
async fn a_date_form_retry_after_is_still_recognised_as_a_continuation() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(
ResponseTemplate::new(403)
.insert_header("retry-after", "Wed, 21 Oct 2026 07:28:00 GMT"),
)
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let err = client
.revalidate()
.await
.expect_err("a date-form `retry-after` is still GitHub asking to be left alone");
assert!(
err.is_lockout(),
"gating detection on an integer parse hands the continuation bug back for every \
lockout GitHub chose to date-stamp: {err:?}"
);
assert_eq!(
client.lockout_remaining(),
Some(DEFAULT_LOCKOUT_BACKOFF),
"the date form is recognised for detection; the duration falls back to the \
default, which is what `latch_lockout` already did with a header it could not \
parse as seconds"
);
}
#[tokio::test]
async fn a_lockout_outliving_its_backoff_re_latches_instead_of_reporting_a_permissions_answer()
{
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(Script::new(vec![
ResponseTemplate::new(401),
ResponseTemplate::new(403).insert_header("retry-after", "60"),
ResponseTemplate::new(403).insert_header("retry-after", "60"),
]))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.mount(&server)
.await;
let clock = Arc::new(TestClock::default());
let client = client(&server, clock.clone());
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("403 on the retry");
assert!(err.is_lockout(), "{err:?}");
assert!(client.is_locked_out());
clock.advance_secs(61);
assert!(!client.is_locked_out(), "the back-off has run out");
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("GitHub is still locking the credential out");
assert!(
err.is_lockout(),
"a 403 carrying `retry-after` with no message is GitHub continuing the lockout, \
not the App installation refusing a permission; reporting `Forbidden` here \
tells the operator to fix a grant that is not missing: {err:?}"
);
assert!(
client.is_locked_out(),
"`backs off without retrying` fails for any lockout that outlives one back-off \
if the continuation does not re-latch"
);
let GithubError::AuthenticationLockout { retry_after } = err else {
unreachable!("asserted above")
};
assert_eq!(
retry_after,
Duration::from_secs(60),
"the continuation's own `retry-after` sets the new back-off"
);
let before = server.received_requests().await.unwrap().len();
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("still locked out");
assert!(err.is_lockout(), "{err:?}");
assert_eq!(
server.received_requests().await.unwrap().len(),
before,
"re-latching must actually stop traffic, not merely rename the error"
);
}
#[tokio::test]
async fn a_first_attempt_permissions_403_is_still_reported_as_forbidden() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/orgs/acme/actions/runners"))
.respond_with(
ResponseTemplate::new(403)
.set_body_json(json!({"message": "Resource not accessible by integration"})),
)
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let err = client
.send(&ApiRequest::get("/orgs/acme/actions/runners"))
.await
.expect_err("403");
assert!(
matches!(err, GithubError::Forbidden { .. }),
"a message and no `retry-after` is GitHub naming a missing grant: {err:?}"
);
assert!(!client.is_locked_out());
}
#[tokio::test]
async fn a_concurrent_success_cannot_downgrade_a_lockout_to_a_permissions_answer() {
let server = MockServer::start().await;
let client = client(&server, Arc::new(TestClock::default()));
client
.consecutive_unauthorized
.fetch_add(1, Ordering::SeqCst);
client.consecutive_unauthorized.store(0, Ordering::SeqCst);
let mut headers = HeaderMap::new();
headers.insert("retry-after", "60".parse().unwrap());
let lockout = ApiResponse {
status: StatusCode::FORBIDDEN,
headers,
body: Vec::new(),
};
assert!(
client.is_lockout_403(&lockout, Attempt::Retry),
"`Attempt::Retry` already means this request's own 401 incremented the counter, so \
reading the counter again adds no signal and only lets an unrelated success \
downgrade a real lockout to `Forbidden`"
);
client.consecutive_unauthorized.store(7, Ordering::SeqCst);
let permissions = ApiResponse {
status: StatusCode::FORBIDDEN,
headers: HeaderMap::new(),
body: br#"{"message":"Resource not accessible by integration"}"#.to_vec(),
};
assert!(!client.is_lockout_403(&permissions, Attempt::First));
}
#[tokio::test]
async fn discovery_returns_the_reachable_repository_and_organization_set() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(
ResponseTemplate::new(200).set_body_json(installations_body(&[
(11, "IvanMurzak", "User", "selected"),
(22, "Tap-Top-Fun", "Organization", "all"),
])),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations/11/repositories"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(repositories_body(&["IvanMurzak/GitHub-Runner-Scaler-UI"])),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations/22/repositories"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(repositories_body(&["Tap-Top-Fun/game", "Tap-Top-Fun/site"])),
)
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let discovery = client.discover_installations(&app()).await.unwrap();
let targets = discovery.targets().expect("installed");
assert_eq!(
targets
.repositories()
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>(),
[
"IvanMurzak/GitHub-Runner-Scaler-UI",
"Tap-Top-Fun/game",
"Tap-Top-Fun/site"
]
);
assert_eq!(
targets
.organizations()
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>(),
["Tap-Top-Fun"],
"a User account is not an organization target"
);
assert!(discovery.install_url().is_none());
}
#[tokio::test]
async fn an_over_broad_installation_is_visible_rather_than_assumed() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(
ResponseTemplate::new(200).set_body_json(installations_body(&[
(11, "IvanMurzak", "User", "selected"),
(22, "Tap-Top-Fun", "Organization", "all"),
])),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations/11/repositories"))
.respond_with(ResponseTemplate::new(200).set_body_json(repositories_body(&["a/b"])))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations/22/repositories"))
.respond_with(ResponseTemplate::new(200).set_body_json(repositories_body(&["c/d"])))
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let targets = client
.discover_installations(&app())
.await
.unwrap()
.targets()
.cloned()
.expect("installed");
let over_broad = targets.over_broad();
assert_eq!(over_broad.len(), 1);
assert_eq!(over_broad[0].account.login(), "Tap-Top-Fun");
assert!(over_broad[0].is_over_broad());
assert_eq!(
over_broad[0].repository_selection,
RepositorySelection::All,
"`repository_selection: all` reaches repositories created later too"
);
assert!(
targets.installations().iter().any(|i| i
.permissions
.iter()
.any(|(k, v)| k == "administration" && v == "write")),
"the grant GitHub reports is surfaced verbatim, not assumed from the design"
);
}
#[tokio::test]
async fn discovery_returns_the_installation_url_when_the_set_is_empty() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let discovery = client.discover_installations(&app()).await.unwrap();
let url = discovery
.install_url()
.expect("an empty set must yield the installation URL");
assert_eq!(url.path(), "/apps/runner-manager/installations/new");
assert!(discovery.targets().is_none());
}
#[tokio::test]
async fn an_installation_that_reaches_no_repository_is_still_not_installed() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(
ResponseTemplate::new(200).set_body_json(installations_body(&[(
11,
"IvanMurzak",
"User",
"selected",
)])),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations/11/repositories"))
.respond_with(ResponseTemplate::new(200).set_body_json(repositories_body(&[])))
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let discovery = client.discover_installations(&app()).await.unwrap();
assert!(
discovery.install_url().is_some(),
"a user installation that selected no repository reaches nothing"
);
}
#[tokio::test]
async fn discovery_follows_every_page_rather_than_trusting_the_first() {
let server = MockServer::start().await;
let next = format!("<{}/user/installations?page=2>; rel=\"next\"", server.uri());
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(Script::new(vec![
ResponseTemplate::new(200)
.set_body_json(installations_body(&[(
11,
"one",
"Organization",
"selected",
)]))
.insert_header("link", next.as_str()),
ResponseTemplate::new(200).set_body_json(installations_body(&[(
22,
"two",
"Organization",
"selected",
)])),
]))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations/11/repositories"))
.respond_with(ResponseTemplate::new(200).set_body_json(repositories_body(&["one/a"])))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/user/installations/22/repositories"))
.respond_with(ResponseTemplate::new(200).set_body_json(repositories_body(&["two/b"])))
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let targets = client
.discover_installations(&app())
.await
.unwrap()
.targets()
.cloned()
.expect("installed");
assert_eq!(
targets
.organizations()
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>(),
["one", "two"],
"the second page must not be dropped"
);
}
#[tokio::test]
async fn an_installation_with_a_null_or_enterprise_account_does_not_fail_the_whole_decode() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"total_count": 3,
"installations": [
{ "id": 10, "account": null, "repository_selection": "selected" },
{
"id": 20,
"account": { "slug": "acme-enterprise", "name": "Acme Inc" },
"repository_selection": "selected"
},
{
"id": 30,
"account": { "login": "IvanMurzak", "type": "User" },
"repository_selection": "selected"
}
]
})))
.mount(&server)
.await;
for (id, repo) in [(20_u64, "acme-enterprise/tools"), (30, "IvanMurzak/app")] {
Mock::given(method("GET"))
.and(path(format!("/user/installations/{id}/repositories")))
.respond_with(ResponseTemplate::new(200).set_body_json(repositories_body(&[repo])))
.mount(&server)
.await;
}
let client = client(&server, Arc::new(TestClock::default()));
let targets = client
.discover_installations(&app())
.await
.expect("one odd account must not fail the whole discovery")
.targets()
.cloned()
.expect("installed");
let reached = targets
.repositories()
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>();
assert!(
reached.contains(&"acme-enterprise/tools".to_string()),
"the enterprise installation is named from `slug` rather than dropped: {reached:?}"
);
assert!(
reached.contains(&"IvanMurzak/app".to_string()),
"the ordinary installation alongside it survives too: {reached:?}"
);
assert_eq!(reached.len(), 2);
assert_eq!(
targets.installations().len(),
2,
"the null account is skipped, and only it"
);
assert_eq!(
targets.skipped(),
1,
"the skip is the right trade, but it must travel with the answer: everything the \
skipped installation reaches is missing from the lists above, and a short list \
reads exactly like a complete one"
);
let enterprise = targets
.installations()
.iter()
.find(|i| i.id == 20)
.expect("the enterprise installation survived");
assert_eq!(
enterprise.account,
InstallationAccount::Enterprise("acme-enterprise".to_string()),
"an account with no `login` that names itself through `slug` is an enterprise, \
and calling it a user is a wrong statement about the operator's own account"
);
assert_eq!(enterprise.account.kind(), "enterprise");
assert!(
enterprise.account.organization().is_none(),
"an enterprise is not an organization target: `GET /orgs/{{org}}/actions/runners` \
does not accept one, so contributing nothing to `organizations()` is correct"
);
assert!(
!targets
.organizations()
.iter()
.any(|o| o.as_str() == "acme-enterprise"),
"and it must not be smuggled in as one either"
);
}
#[tokio::test]
async fn a_credential_whose_only_installation_was_skipped_is_not_reported_as_not_installed() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"total_count": 1,
"installations": [
{ "id": 10, "account": null, "repository_selection": "selected" }
]
})))
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let discovery = client.discover_installations(&app()).await.unwrap();
assert_eq!(
discovery,
InstallationDiscovery::Indeterminate { skipped: 1 },
"GitHub reported an installation; this client could not describe it. That is not \
the same answer as `not installed`, and only one of the two is fixed by \
installing the App"
);
assert_eq!(
discovery.install_url(),
None,
"offering the install URL here is the wrong remediation, and putting it one field \
over from the right verdict would just relocate the defect"
);
assert_eq!(discovery.skipped(), 1);
assert!(discovery.targets().is_none());
}
#[tokio::test]
async fn an_empty_reach_with_nothing_skipped_is_still_not_installed() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(ResponseTemplate::new(200).set_body_json(installations_body(&[])))
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let discovery = client.discover_installations(&app()).await.unwrap();
assert!(discovery.install_url().is_some(), "{discovery:?}");
assert_eq!(discovery.skipped(), 0);
}
#[tokio::test]
async fn a_self_referential_link_header_stops_at_the_page_ceiling() {
let server = MockServer::start().await;
let self_link = format!("<{}/user/installations?page=2>; rel=\"next\"", server.uri());
Mock::given(method("GET"))
.and(path("/user/installations"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(installations_body(&[]))
.insert_header("link", self_link.as_str()),
)
.mount(&server)
.await;
let client = client(&server, Arc::new(TestClock::default()));
let discovery = client
.discover_installations(&app())
.await
.expect("the ceiling is what makes this return at all");
assert!(
discovery.install_url().is_some(),
"no installation was found"
);
assert_eq!(
server.received_requests().await.unwrap().len(),
MAX_PAGES,
"pagination must stop at the ceiling rather than follow the loop forever"
);
}
#[test]
fn a_link_header_yields_only_the_next_relation() {
let header = "<https://api.github.com/user/installations?page=3>; rel=\"next\", \
<https://api.github.com/user/installations?page=9>; rel=\"last\"";
assert_eq!(
parse_link_next(header).map(|u| u.to_string()),
Some("https://api.github.com/user/installations?page=3".to_string())
);
assert!(parse_link_next("<https://x/>; rel=\"last\"").is_none());
assert!(parse_link_next("nonsense").is_none());
}
#[test]
fn a_next_url_containing_a_comma_still_paginates() {
let header = "<https://api.github.com/repos/o/r/actions/runners\
?labels=self-hosted,windows&page=2>; rel=\"next\"";
assert_eq!(
parse_link_next(header).map(|u| u.to_string()),
Some(
"https://api.github.com/repos/o/r/actions/runners\
?labels=self-hosted,windows&page=2"
.to_string()
),
"a comma inside the URL must not end the link-value"
);
let header = "<https://api.github.com/x?a=1,2&page=1>; rel=\"prev\", \
<https://api.github.com/x?a=1,2&page=3>; rel=\"next\"";
assert_eq!(
parse_link_next(header).map(|u| u.to_string()),
Some("https://api.github.com/x?a=1,2&page=3".to_string())
);
}
#[test]
fn the_next_relation_is_found_wherever_it_sits_in_the_header() {
let not_first = "<https://api.github.com/u?page=1>; rel=\"first\", \
<https://api.github.com/u?page=9>; rel=\"last\", \
<https://api.github.com/u?page=4>; rel=\"next\"";
assert_eq!(
parse_link_next(not_first).map(|u| u.to_string()),
Some("https://api.github.com/u?page=4".to_string())
);
let unquoted = "<https://api.github.com/u?page=1>; rel=prev, \
<https://api.github.com/u?page=3>; rel=next";
assert_eq!(
parse_link_next(unquoted).map(|u| u.to_string()),
Some("https://api.github.com/u?page=3".to_string())
);
assert!(
parse_link_next("<https://api.github.com/u?page=2; rel=\"next\"").is_none(),
"an unterminated target is not a link-value"
);
}
#[test]
fn a_short_collection_is_measured_against_the_count_github_reported() {
assert_eq!(under_collected(1, Some(2)), Some(2), "page 2 was dropped");
assert_eq!(under_collected(2, Some(2)), None, "complete");
assert_eq!(
under_collected(3, Some(2)),
None,
"a collection that grew between pages is not an under-collection"
);
assert_eq!(under_collected(0, None), None, "no count, no claim");
}
#[test]
fn no_type_in_this_crate_renders_a_secret_through_debug() {
let token = UserAccessToken::new(SecretString::from(FIXTURE_TOKEN));
let rendered = format!("{token:?}");
assert!(!rendered.contains(FIXTURE_TOKEN), "{rendered}");
assert!(rendered.contains("[REDACTED]"));
assert!(
rendered.contains("ghu_"),
"the family prefix is diagnostic and is not the secret"
);
let request = ApiRequest::post_json("/x", &json!({"encoded_jit_config": "SECRETBLOB"}))
.expect("serializes");
let rendered = format!("{request:?}");
assert!(!rendered.contains("SECRETBLOB"), "{rendered}");
let response = ApiResponse {
status: StatusCode::OK,
headers: HeaderMap::new(),
body: b"{\"encoded_jit_config\":\"SECRETBLOB\"}".to_vec(),
};
let rendered = format!("{response:?}");
assert!(!rendered.contains("SECRETBLOB"), "{rendered}");
}
const CONFIDENTIAL: &[&str] = &[concat!("client", "secret"), concat!("app", "secret")];
const MANIFEST: (&str, &str) = ("Cargo.toml", include_str!("../Cargo.toml"));
const CRATE_SOURCES: &[(&str, &str)] = &[
("demand.rs", include_str!("demand.rs")),
("device_flow.rs", include_str!("device_flow.rs")),
("jit.rs", include_str!("jit.rs")),
("lib.rs", include_str!("lib.rs")),
("rest.rs", include_str!("rest.rs")),
];
const SOURCES_OWNED_BY_C2: &[(&str, &str)] = &[
("device_flow.rs", include_str!("device_flow.rs")),
("lib.rs", include_str!("lib.rs")),
MANIFEST,
];
fn normalise_source(source: &str) -> String {
source.to_ascii_lowercase().replace('_', "")
}
fn normalise_manifest(manifest: &str) -> String {
manifest.to_ascii_lowercase().replace(['_', '-'], "")
}
fn normalise(name: &str, contents: &str) -> String {
if name == MANIFEST.0 {
normalise_manifest(contents)
} else {
normalise_source(contents)
}
}
fn non_test_prefix(source: &str) -> &str {
let mut offset = 0;
for line in source.split_inclusive('\n') {
if line.trim() == "#[cfg(test)]" {
return &source[..offset];
}
offset += line.len();
}
source
}
#[test]
fn no_confidential_credential_in_this_crate() {
for &(name, source) in CRATE_SOURCES.iter().chain(std::iter::once(&MANIFEST)) {
let haystack = normalise(name, source);
for forbidden in CONFIDENTIAL {
assert!(
!haystack.contains(forbidden),
"{name} names {forbidden:?} in some spelling: a public client cannot \
secure a confidential credential, and this design never tries to (D3)"
);
}
}
}
#[test]
fn the_confidential_credential_scan_covers_every_source_file() {
fn collect(dir: &std::path::Path, prefix: &str, found: &mut Vec<String>) {
for entry in std::fs::read_dir(dir).expect("the source directory is readable") {
let entry = entry.expect("a readable directory entry");
let name = entry.file_name().to_string_lossy().into_owned();
let relative = if prefix.is_empty() {
name.clone()
} else {
format!("{prefix}/{name}")
};
if entry.file_type().expect("a readable entry type").is_dir() {
collect(&entry.path(), &relative, found);
} else if name.ends_with(".rs") {
found.push(relative);
}
}
}
let mut on_disk = Vec::new();
collect(
std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/src")),
"",
&mut on_disk,
);
on_disk.sort();
let mut scanned: Vec<String> = CRATE_SOURCES
.iter()
.map(|(name, _)| (*name).to_string())
.collect();
scanned.sort();
assert_eq!(
scanned, on_disk,
"`src/` and the scanned list have diverged. Add the new file to `CRATE_SOURCES` \
with an `include_str!`, naming it by its `/`-joined path relative to `src/`; \
leaving it out means the confidential-credential scan silently stops covering \
`every source file in the crate`, which is the claim it makes."
);
}
#[test]
fn the_confidential_credential_scan_is_not_evaded_by_naming() {
let source_evasions = [
format!("let {}Token = fetch()", "refresh"),
format!("struct {}Token;", "Refresh"),
format!("{}_TOKEN", "REFRESH"),
format!("{}Secret", "client"),
format!("{}_SECRET", "CLIENT"),
format!("{}_secret", "app"),
];
for evasion in &source_evasions {
let normalised = normalise("lib.rs", evasion);
assert!(
normalised.contains(concat!("refresh", "token"))
|| normalised.contains(concat!("client", "secret"))
|| normalised.contains(concat!("app", "secret")),
"{evasion:?} would walk straight through the scan"
);
}
let manifest_evasion = format!("{}-secret = \"...\"", "client");
assert!(
normalise(MANIFEST.0, &manifest_evasion).contains(concat!("client", "secret")),
"a kebab-case TOML key is an identifier, and the manifest normaliser must \
still collapse it"
);
for allowed in [
"a public client cannot hold a client secret",
"the published App issues no renewal token",
"//! This gateway is deliberately client-secret-free, as D3 requires.",
"a refresh-free credential model",
] {
let normalised = normalise("lib.rs", allowed);
assert!(
!normalised.contains(concat!("client", "secret"))
&& !normalised.contains(concat!("refresh", "token")),
"{allowed:?} is English, not an identifier, and must not trip the scan"
);
}
}
#[test]
fn this_crate_persists_nothing_and_does_not_depend_on_the_platform_crate() {
assert!(
!MANIFEST.1.contains("runner-manager-platform"),
"the gateway must be testable with no platform dependency at all"
);
for &(name, source) in SOURCES_OWNED_BY_C2 {
if name == MANIFEST.0 {
continue;
}
let non_test = non_test_prefix(source);
for forbidden in [
"std::fs",
"fs::write",
"File::create",
"File::options",
"OpenOptions",
"std::io::Write",
"tokio::fs",
] {
assert!(
!non_test.contains(forbidden),
"{name} performs a filesystem operation ({forbidden:?}) outside its tests"
);
}
}
}
#[test]
fn the_non_test_boundary_is_a_line_and_not_a_mention() {
let file = "//! Test helpers live in an inline #[cfg(test)] module near the bottom.\n\
\n\
fn persist() { std::fs::write(\"x\", b\"y\").unwrap(); }\n\
\n\
#[cfg(test)]\n\
mod tests {\n\
fn helper() { std::fs::write(\"ok-in-tests\", b\"\").unwrap(); }\n\
}\n";
let non_test = non_test_prefix(file);
assert!(
non_test.contains("fn persist"),
"a prose mention of the attribute truncated the scanned region, and every \
filesystem call below it stopped being scanned — silently:\n{non_test}"
);
assert!(
!non_test.contains("ok-in-tests"),
"the boundary must still exclude the real test module:\n{non_test}"
);
for &(name, source) in SOURCES_OWNED_BY_C2 {
if name == MANIFEST.0 {
continue;
}
let expected = source
.lines()
.position(|line| line.trim() == "#[cfg(test)]")
.expect("each source file has an inline test module");
let scanned = non_test_prefix(source).lines().count();
assert_eq!(
scanned, expected,
"{name}: the scanned region ends at line {scanned} but the test module starts \
at line {expected}. The gap is code that claims to be scanned and is not."
);
}
}
}