use std::collections::{BTreeSet, HashMap};
use std::fmt;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, PoisonError, RwLock};
use std::time::Duration;
use futures_util::future::BoxFuture;
use crate::error::{Error, WireKind};
pub const PROXY_AUTH_HEADER: &str = "X-aws-proxy-auth";
pub const PROXY_PORT_HEADER: &str = "X-aws-proxy-port";
pub const DEFAULT_AGENT_PORT: u16 = 9000;
pub const WS_SUBPROTOCOL: &str = protocol::tunnel::WS_MARKER_SUBPROTOCOL;
pub const WS_AUTH_SUBPROTOCOL_PREFIX: &str = "lambda-microvms.authentication.";
pub const WS_PORT_SUBPROTOCOL_PREFIX: &str = "lambda-microvms.port.";
pub const MAX_TOKEN_LIFETIME: Duration = Duration::from_secs(60 * 60);
pub const DEFAULT_REFRESH_AFTER: Duration = Duration::from_secs(30 * 60);
#[derive(Clone, Default)]
pub struct ProxyToken {
headers: HashMap<String, String>,
}
impl ProxyToken {
pub fn new(headers: HashMap<String, String>) -> Self {
Self { headers }
}
pub fn from_pairs<I, K, V>(pairs: I) -> Self
where
I: IntoIterator<Item = (K, V)>,
K: Into<String>,
V: Into<String>,
{
Self::new(
pairs
.into_iter()
.map(|(name, value)| (name.into(), value.into()))
.collect(),
)
}
pub fn get(&self, name: &str) -> Option<&str> {
self.headers
.iter()
.find(|(key, _)| key.eq_ignore_ascii_case(name))
.map(|(_, value)| value.as_str())
}
pub fn headers(&self) -> impl Iterator<Item = (&str, &str)> {
self.headers
.iter()
.map(|(name, value)| (name.as_str(), value.as_str()))
}
pub fn auth_value(&self) -> Result<&str, Error> {
self.get(PROXY_AUTH_HEADER).ok_or_else(|| {
Error::wire(
WireKind::AuthTokenMint,
format!(
"the minted authToken map has no {PROXY_AUTH_HEADER} key (it carries {:?}); \
the response is a header map, not a token string — see docs/PLATFORM.md, \
\"Endpoint authentication\"",
self.headers.keys().collect::<Vec<_>>()
),
)
})
}
pub fn is_empty(&self) -> bool {
self.headers.is_empty()
}
}
impl From<crate::control::ProxyToken> for ProxyToken {
fn from(token: crate::control::ProxyToken) -> Self {
Self::from_pairs(token.headers())
}
}
impl fmt::Debug for ProxyToken {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let mut keys: Vec<&str> = self.headers.keys().map(String::as_str).collect();
keys.sort_unstable();
f.debug_struct("ProxyToken")
.field("headers", &keys)
.finish()
}
}
pub trait TokenMinter: Send + Sync {
fn mint(&self) -> BoxFuture<'_, Result<ProxyToken, Error>>;
fn mint_for_ports(&self, ports: &[u16]) -> BoxFuture<'_, Result<ProxyToken, Error>> {
let _ = ports;
self.mint()
}
}
pub trait Clock: Send + Sync + fmt::Debug {
fn elapsed(&self) -> Duration;
}
#[derive(Debug)]
pub struct TokioClock {
base: tokio::time::Instant,
}
impl Default for TokioClock {
fn default() -> Self {
Self {
base: tokio::time::Instant::now(),
}
}
}
impl Clock for TokioClock {
fn elapsed(&self) -> Duration {
self.base.elapsed()
}
}
struct Cached {
token: ProxyToken,
minted_at: Duration,
ports: BTreeSet<u16>,
}
pub struct ProxyAuth {
minter: Arc<dyn TokenMinter>,
port: u16,
refresh_after: Duration,
clock: Arc<dyn Clock>,
cached: RwLock<Option<Cached>>,
mint_lock: tokio::sync::Mutex<()>,
mint_count: AtomicU64,
}
impl ProxyAuth {
pub fn new(minter: Arc<dyn TokenMinter>, port: u16) -> Self {
Self {
minter,
port,
refresh_after: DEFAULT_REFRESH_AFTER,
clock: Arc::new(TokioClock::default()),
cached: RwLock::new(None),
mint_lock: tokio::sync::Mutex::new(()),
mint_count: AtomicU64::new(0),
}
}
pub fn with_refresh_after(
minter: Arc<dyn TokenMinter>,
port: u16,
refresh_after: Duration,
clock: Arc<dyn Clock>,
) -> Result<Self, Error> {
if refresh_after >= MAX_TOKEN_LIFETIME {
return Err(Error::invalid_arg(format!(
"a proxy-token refresh interval of {}s is not below the service's \
{}s ceiling, so a request can be issued with a token that has already \
expired; see docs/PLATFORM.md, \"Endpoint authentication\"",
refresh_after.as_secs(),
MAX_TOKEN_LIFETIME.as_secs()
)));
}
Ok(Self {
minter,
port,
refresh_after,
clock,
cached: RwLock::new(None),
mint_lock: tokio::sync::Mutex::new(()),
mint_count: AtomicU64::new(0),
})
}
pub fn port(&self) -> u16 {
self.port
}
pub fn mint_count(&self) -> u64 {
self.mint_count.load(Ordering::SeqCst)
}
pub fn invalidate(&self) {
*self.cached.write().unwrap_or_else(PoisonError::into_inner) = None;
}
pub fn is_cached(&self) -> bool {
self.fresh_token_for(self.port).is_some()
}
pub async fn headers(&self) -> Result<Vec<(String, String)>, Error> {
Ok(self.headers_from(&self.token().await?))
}
pub async fn headers_for_port(&self, port: u16) -> Result<Vec<(String, String)>, Error> {
let token = self.token_for(port).await?;
let mut headers: Vec<(String, String)> = token
.headers()
.filter(|(name, _)| !name.eq_ignore_ascii_case(PROXY_PORT_HEADER))
.map(|(name, value)| (name.to_string(), value.to_string()))
.collect();
headers.push((PROXY_PORT_HEADER.to_string(), port.to_string()));
Ok(headers)
}
pub async fn subprotocols(&self, port: u16) -> Result<[String; 3], Error> {
let token = self.token_for(port).await?;
let auth = token.auth_value()?;
Ok([
WS_SUBPROTOCOL.to_string(),
format!("{WS_AUTH_SUBPROTOCOL_PREFIX}{auth}"),
format!("{WS_PORT_SUBPROTOCOL_PREFIX}{port}"),
])
}
async fn token(&self) -> Result<ProxyToken, Error> {
self.token_for(self.port).await
}
async fn token_for(&self, port: u16) -> Result<ProxyToken, Error> {
if let Some(token) = self.fresh_token_for(port) {
return Ok(token);
}
let _guard = self.mint_lock.lock().await;
if let Some(token) = self.fresh_token_for(port) {
return Ok(token);
}
let mut wanted: BTreeSet<u16> = self
.cached
.read()
.unwrap_or_else(PoisonError::into_inner)
.as_ref()
.map(|entry| entry.ports.clone())
.unwrap_or_default();
wanted.insert(self.port);
wanted.insert(port);
let ports: Vec<u16> = wanted.iter().copied().collect();
let token = self.minter.mint_for_ports(&ports).await.map_err(|err| {
let message = format!(
"could not mint a proxy auth token for port(s) {ports:?}: {err}; minting is \
inside the request path, so the identical request may succeed"
);
Error::wire(WireKind::AuthTokenMint, message).with_source(err)
})?;
token.auth_value()?;
*self.cached.write().unwrap_or_else(PoisonError::into_inner) = Some(Cached {
token: token.clone(),
minted_at: self.clock.elapsed(),
ports: wanted,
});
self.mint_count.fetch_add(1, Ordering::SeqCst);
Ok(token)
}
fn fresh_token_for(&self, port: u16) -> Option<ProxyToken> {
let cached = self.cached.read().unwrap_or_else(PoisonError::into_inner);
let entry = cached.as_ref()?;
if !entry.ports.contains(&port) {
return None;
}
let age = self.clock.elapsed().saturating_sub(entry.minted_at);
if age > self.refresh_after {
return None;
}
Some(entry.token.clone())
}
fn headers_from(&self, token: &ProxyToken) -> Vec<(String, String)> {
let mut headers: Vec<(String, String)> = token
.headers()
.map(|(name, value)| (name.to_string(), value.to_string()))
.collect();
if !headers
.iter()
.any(|(name, _)| name.eq_ignore_ascii_case(PROXY_PORT_HEADER))
{
headers.push((PROXY_PORT_HEADER.to_string(), self.port.to_string()));
}
headers
}
}
impl fmt::Debug for ProxyAuth {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ProxyAuth")
.field("port", &self.port)
.field("refresh_after", &self.refresh_after)
.field("mint_count", &self.mint_count())
.field("cached", &self.is_cached())
.finish()
}
}
#[cfg(test)]
pub(crate) mod testing {
use super::*;
#[derive(Debug, Default)]
pub(crate) struct ManualClock {
millis: AtomicU64,
}
impl ManualClock {
pub(crate) fn advance(&self, by: Duration) {
self.millis
.fetch_add(by.as_millis() as u64, Ordering::SeqCst);
}
}
impl Clock for ManualClock {
fn elapsed(&self) -> Duration {
Duration::from_millis(self.millis.load(Ordering::SeqCst))
}
}
#[derive(Debug, Default)]
pub(crate) struct CountingMinter {
issued: AtomicU64,
failures: AtomicU64,
omit_auth_header: bool,
requested: RwLock<Vec<Vec<u16>>>,
}
impl CountingMinter {
pub(crate) fn failing(times: u64) -> Self {
Self {
failures: AtomicU64::new(times),
..Self::default()
}
}
pub(crate) fn without_auth_header() -> Self {
Self {
omit_auth_header: true,
..Self::default()
}
}
pub(crate) fn value(nth: u64) -> String {
format!("jwe-{nth}")
}
pub(crate) fn requested(&self) -> Vec<Vec<u16>> {
self.requested
.read()
.unwrap_or_else(PoisonError::into_inner)
.clone()
}
fn issue(&self, ports: &[u16]) -> Result<ProxyToken, Error> {
if self
.failures
.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| {
left.checked_sub(1)
})
.is_ok()
{
return Err(Error::wire(
WireKind::AuthTokenMint,
"ThrottlingException from CreateMicrovmAuthToken",
));
}
let mut sorted = ports.to_vec();
sorted.sort_unstable();
self.requested
.write()
.unwrap_or_else(PoisonError::into_inner)
.push(sorted);
let nth = self.issued.fetch_add(1, Ordering::SeqCst);
if self.omit_auth_header {
return Ok(ProxyToken::from_pairs([("X-aws-something-else", "junk")]));
}
Ok(ProxyToken::from_pairs([(
PROXY_AUTH_HEADER.to_string(),
Self::value(nth),
)]))
}
}
impl TokenMinter for CountingMinter {
fn mint(&self) -> BoxFuture<'_, Result<ProxyToken, Error>> {
Box::pin(async move { self.issue(&[DEFAULT_AGENT_PORT]) })
}
fn mint_for_ports(&self, ports: &[u16]) -> BoxFuture<'_, Result<ProxyToken, Error>> {
let ports = ports.to_vec();
Box::pin(async move { self.issue(&ports) })
}
}
}
#[cfg(test)]
mod tests {
use super::testing::{CountingMinter, ManualClock};
use super::*;
use crate::error::ErrorKind;
fn auth(clock: Arc<ManualClock>, refresh_after: Duration) -> ProxyAuth {
auth_with(clock, refresh_after).0
}
fn auth_with(
clock: Arc<ManualClock>,
refresh_after: Duration,
) -> (ProxyAuth, Arc<CountingMinter>) {
let minter = Arc::new(CountingMinter::default());
let auth = ProxyAuth::with_refresh_after(
Arc::clone(&minter) as Arc<dyn TokenMinter>,
DEFAULT_AGENT_PORT,
refresh_after,
clock,
)
.expect("the interval is below the ceiling");
(auth, minter)
}
fn value_of(headers: &[(String, String)], name: &str) -> Option<String> {
headers
.iter()
.find(|(key, _)| key.eq_ignore_ascii_case(name))
.map(|(_, value)| value.clone())
}
#[test]
fn the_default_refresh_interval_is_strictly_below_the_service_ceiling() {
assert!(
DEFAULT_REFRESH_AFTER < MAX_TOKEN_LIFETIME,
"refreshing at or after the ceiling puts the expiry inside the window \
between building the headers and the proxy reading them"
);
assert_eq!(MAX_TOKEN_LIFETIME, Duration::from_secs(3600));
assert_eq!(DEFAULT_REFRESH_AFTER, MAX_TOKEN_LIFETIME / 2);
}
#[test]
fn a_refresh_interval_at_the_ceiling_is_refused() {
let err = ProxyAuth::with_refresh_after(
Arc::new(CountingMinter::default()),
DEFAULT_AGENT_PORT,
MAX_TOKEN_LIFETIME,
Arc::new(ManualClock::default()),
)
.expect_err("an interval at the ceiling must be refused");
assert_eq!(err.kind(), ErrorKind::InvalidArg);
assert!(
err.to_string().contains("ceiling"),
"the message must name the finding: {err}"
);
assert!(
ProxyAuth::with_refresh_after(
Arc::new(CountingMinter::default()),
DEFAULT_AGENT_PORT,
MAX_TOKEN_LIFETIME - Duration::from_secs(1),
Arc::new(ManualClock::default()),
)
.is_ok()
);
}
#[tokio::test]
async fn every_header_set_carries_both_the_auth_and_the_port_header() {
let auth = auth(Arc::new(ManualClock::default()), DEFAULT_REFRESH_AFTER);
let headers = auth.headers().await.expect("the fake minter succeeds");
assert_eq!(
value_of(&headers, PROXY_AUTH_HEADER).as_deref(),
Some(CountingMinter::value(0).as_str())
);
assert_eq!(
value_of(&headers, PROXY_PORT_HEADER).as_deref(),
Some("9000"),
"without the port header the proxy answers as if the token were bad"
);
assert_eq!(
headers.len(),
2,
"no third header was invented: {headers:?}"
);
}
#[tokio::test]
async fn a_token_map_without_the_auth_header_is_a_retryable_mint_failure() {
let auth = ProxyAuth::with_refresh_after(
Arc::new(CountingMinter::without_auth_header()),
DEFAULT_AGENT_PORT,
DEFAULT_REFRESH_AFTER,
Arc::new(ManualClock::default()),
)
.expect("interval accepted");
let err = auth.headers().await.expect_err("the map has no auth key");
assert_eq!(err.wire_kind(), Some(WireKind::AuthTokenMint));
assert!(err.retryable());
assert!(
err.to_string().contains(PROXY_AUTH_HEADER),
"the message must name the missing key: {err}"
);
assert_eq!(
auth.mint_count(),
0,
"an unusable token must not be counted as minted or cached"
);
}
#[tokio::test]
async fn a_token_inside_the_refresh_window_is_reused() {
let clock = Arc::new(ManualClock::default());
let auth = auth(Arc::clone(&clock), DEFAULT_REFRESH_AFTER);
let first = auth.headers().await.expect("mints");
clock.advance(DEFAULT_REFRESH_AFTER - Duration::from_secs(1));
let second = auth.headers().await.expect("reuses");
assert_eq!(first, second);
assert_eq!(auth.mint_count(), 1);
}
#[tokio::test]
async fn crossing_the_refresh_window_mints_a_new_token_and_sends_it() {
let clock = Arc::new(ManualClock::default());
let auth = auth(Arc::clone(&clock), DEFAULT_REFRESH_AFTER);
let first = auth.headers().await.expect("mints");
assert_eq!(
value_of(&first, PROXY_AUTH_HEADER).as_deref(),
Some(CountingMinter::value(0).as_str())
);
clock.advance(DEFAULT_REFRESH_AFTER + Duration::from_secs(1));
let second = auth.headers().await.expect("re-mints");
assert_eq!(auth.mint_count(), 2);
assert_eq!(
value_of(&second, PROXY_AUTH_HEADER).as_deref(),
Some(CountingMinter::value(1).as_str()),
"the refreshed token never reached the headers"
);
}
async fn walk_two_ceilings(refresh_after: Duration) -> (Duration, u64) {
let clock = Arc::new(ManualClock::default());
let auth = ProxyAuth {
minter: Arc::new(CountingMinter::default()),
port: DEFAULT_AGENT_PORT,
refresh_after,
clock: Arc::clone(&clock) as Arc<dyn Clock>,
cached: RwLock::new(None),
mint_lock: tokio::sync::Mutex::new(()),
mint_count: AtomicU64::new(0),
};
let mut minted_at = Duration::ZERO;
let mut oldest = Duration::ZERO;
for minute in 0..120u64 {
let now = Duration::from_secs(minute * 60);
let before = auth.mint_count();
auth.headers().await.expect("the fake minter succeeds");
if auth.mint_count() > before {
minted_at = now;
}
oldest = oldest.max(now - minted_at);
clock.advance(Duration::from_secs(60));
}
(oldest, auth.mint_count())
}
#[tokio::test]
async fn a_token_is_never_presented_after_the_service_ceiling() {
let (oldest, mints) = walk_two_ceilings(DEFAULT_REFRESH_AFTER).await;
assert!(
oldest < MAX_TOKEN_LIFETIME,
"a request carried a token {}s old, past the {}s ceiling",
oldest.as_secs(),
MAX_TOKEN_LIFETIME.as_secs()
);
assert!(
MAX_TOKEN_LIFETIME - oldest >= MAX_TOKEN_LIFETIME - DEFAULT_REFRESH_AFTER,
"the oldest token presented had only {}s of life left",
(MAX_TOKEN_LIFETIME - oldest).as_secs()
);
assert_eq!(mints, 4);
let (oldest_at_ceiling, _) = walk_two_ceilings(MAX_TOKEN_LIFETIME).await;
assert!(
oldest_at_ceiling >= MAX_TOKEN_LIFETIME,
"refreshing at the ceiling presented a token only {}s old, so this \
property cannot tell a safe schedule from an unsafe one",
oldest_at_ceiling.as_secs()
);
}
#[tokio::test]
async fn invalidate_forces_the_next_request_to_mint() {
let clock = Arc::new(ManualClock::default());
let auth = auth(Arc::clone(&clock), DEFAULT_REFRESH_AFTER);
auth.headers().await.expect("mints");
assert_eq!(auth.mint_count(), 1);
assert!(auth.is_cached());
auth.invalidate();
assert!(!auth.is_cached(), "invalidate did not drop the cache");
let refreshed = auth.headers().await.expect("re-mints");
assert_eq!(auth.mint_count(), 2);
assert_eq!(
value_of(&refreshed, PROXY_AUTH_HEADER).as_deref(),
Some(CountingMinter::value(1).as_str())
);
}
#[tokio::test]
async fn a_failed_mint_is_retryable_and_the_next_attempt_succeeds() {
let auth = ProxyAuth::with_refresh_after(
Arc::new(CountingMinter::failing(1)),
DEFAULT_AGENT_PORT,
DEFAULT_REFRESH_AFTER,
Arc::new(ManualClock::default()),
)
.expect("interval accepted");
let err = auth.headers().await.expect_err("the first mint fails");
assert_eq!(err.kind(), ErrorKind::Retryable);
assert_eq!(err.wire_kind(), Some(WireKind::AuthTokenMint));
assert_eq!(auth.mint_count(), 0);
let headers = auth.headers().await.expect("the second mint succeeds");
assert_eq!(auth.mint_count(), 1);
assert!(value_of(&headers, PROXY_AUTH_HEADER).is_some());
}
#[tokio::test]
async fn concurrent_requests_on_a_cold_cache_mint_once() {
let auth = Arc::new(auth(
Arc::new(ManualClock::default()),
DEFAULT_REFRESH_AFTER,
));
let mut sets = Vec::new();
for _ in 0..8 {
sets.push(Arc::clone(&auth).headers_owned());
}
let results = futures_util::future::join_all(sets).await;
for result in &results {
assert!(result.is_ok(), "{result:?}");
}
assert_eq!(
auth.mint_count(),
1,
"eight concurrent requests burned more than one control-plane call"
);
}
impl ProxyAuth {
#[cfg(test)]
async fn headers_owned(self: Arc<Self>) -> Result<Vec<(String, String)>, Error> {
self.headers().await
}
}
#[tokio::test]
async fn the_subprotocols_are_the_platforms_three_exact_strings_in_order() {
let auth = auth(Arc::new(ManualClock::default()), DEFAULT_REFRESH_AFTER);
let offered = auth
.subprotocols(8080)
.await
.expect("the fake minter mints");
assert_eq!(
offered,
[
"lambda-microvms".to_string(),
format!(
"lambda-microvms.authentication.{}",
CountingMinter::value(0)
),
"lambda-microvms.port.8080".to_string(),
],
"the handshake offered something the proxy does not match"
);
assert_eq!(auth.port(), DEFAULT_AGENT_PORT);
assert!(
offered[2].ends_with("8080"),
"the port was taken from the instance rather than the argument: {}",
offered[2]
);
}
#[tokio::test]
async fn a_handshake_reuses_the_cached_token_and_re_mints_for_an_uncovered_port() {
let clock = Arc::new(ManualClock::default());
let (auth, minter) = auth_with(Arc::clone(&clock), DEFAULT_REFRESH_AFTER);
let first = auth.subprotocols(9000).await.expect("mints");
clock.advance(DEFAULT_REFRESH_AFTER - Duration::from_secs(1));
let second = auth.subprotocols(9000).await.expect("reuses");
let headers = auth.headers().await.expect("reuses");
assert_eq!(first, second, "the same cached token produced two values");
assert_eq!(
auth.mint_count(),
1,
"the handshake path burned its own control-plane call for a port the cached token \
already covers, so TRAP-9 is now two schedules to get right instead of one"
);
assert_eq!(
value_of(&headers, PROXY_AUTH_HEADER).as_deref(),
Some(CountingMinter::value(0).as_str())
);
let third = auth.subprotocols(3000).await.expect("mints a wider token");
assert_eq!(
auth.mint_count(),
2,
"a port outside the cached token's scope reused it, which is a handshake the proxy \
refuses as 1006 with no reason"
);
assert_ne!(
third[1], first[1],
"the wider token carries the same auth value as the 9000-only one, so nothing was \
actually re-scoped"
);
assert_eq!(third[2], "lambda-microvms.port.3000");
assert_eq!(
minter.requested(),
vec![vec![9000], vec![3000, 9000]],
"the widened mint dropped the session's own port, so the next request on 9000 \
re-mints and two alternating ports mint forever"
);
auth.subprotocols(3000).await.expect("reuses");
auth.headers().await.expect("reuses");
assert_eq!(
auth.mint_count(),
2,
"the widened token did not serve both ports, so the scope was replaced rather than \
extended"
);
}
#[tokio::test]
async fn a_handshake_after_the_refresh_window_carries_the_freshly_minted_token() {
let clock = Arc::new(ManualClock::default());
let auth = auth(Arc::clone(&clock), DEFAULT_REFRESH_AFTER);
let first = auth.subprotocols(9000).await.expect("mints");
assert!(first[1].ends_with(&CountingMinter::value(0)));
clock.advance(DEFAULT_REFRESH_AFTER + Duration::from_secs(1));
let second = auth.subprotocols(9000).await.expect("re-mints");
assert_eq!(auth.mint_count(), 2);
assert_eq!(
second[1],
format!("{WS_AUTH_SUBPROTOCOL_PREFIX}{}", CountingMinter::value(1)),
"the handshake offered the expired token, which the proxy rejects as if it were \
the wrong token"
);
}
#[tokio::test]
async fn a_token_map_without_the_auth_key_fails_the_handshake_rather_than_offering_nothing() {
let auth = ProxyAuth::with_refresh_after(
Arc::new(CountingMinter::without_auth_header()),
DEFAULT_AGENT_PORT,
DEFAULT_REFRESH_AFTER,
Arc::new(ManualClock::default()),
)
.expect("interval accepted");
let err = auth
.subprotocols(9000)
.await
.expect_err("the map has no auth key");
assert_eq!(err.wire_kind(), Some(WireKind::AuthTokenMint));
assert!(err.retryable());
assert!(
err.to_string().contains(PROXY_AUTH_HEADER),
"the message must name the missing key: {err}"
);
}
#[tokio::test]
async fn nothing_that_survives_a_handshake_can_print_the_credential() {
let auth = ProxyAuth::with_refresh_after(
Arc::new(CountingMinter::default()),
DEFAULT_AGENT_PORT,
DEFAULT_REFRESH_AFTER,
Arc::new(ManualClock::default()),
)
.expect("interval accepted");
let offered = auth.subprotocols(9000).await.expect("mints");
let credential = CountingMinter::value(0);
assert!(
offered[1].contains(&credential),
"the test is measuring nothing unless the value really is in the list"
);
let rendered = format!("{auth:?}");
assert!(
!rendered.contains(&credential),
"the auth token reached ProxyAuth's Debug: {rendered}"
);
assert!(
rendered.contains("mint_count"),
"the Debug still has to be useful: {rendered}"
);
}
#[tokio::test]
async fn headers_for_a_port_are_backed_by_a_token_minted_for_that_port() {
let (auth, minter) = auth_with(Arc::new(ManualClock::default()), DEFAULT_REFRESH_AFTER);
auth.headers().await.expect("mints for the session's port");
let for_workload = auth.headers_for_port(8080).await.expect("mints wider");
assert_eq!(
value_of(&for_workload, PROXY_PORT_HEADER).as_deref(),
Some("8080"),
"the request would target the session's port rather than the caller's"
);
assert_eq!(
for_workload.len(),
2,
"no third header was invented and none was duplicated: {for_workload:?}"
);
assert_eq!(
minter.requested(),
vec![vec![9000], vec![8080, 9000]],
"the port header said 8080 behind a token nobody scoped to 8080, which the proxy \
refuses with 403 'Access to port denied'"
);
assert_eq!(
value_of(&for_workload, PROXY_AUTH_HEADER).as_deref(),
Some(CountingMinter::value(1).as_str()),
"the headers carried the narrow first token rather than the widened second one"
);
auth.headers().await.expect("reuses the widened token");
assert_eq!(
auth.mint_count(),
2,
"widening the scope for 8080 evicted 9000, so alternating ports mint forever"
);
}
#[tokio::test]
async fn a_requested_port_replaces_one_the_token_map_carried() {
struct PortCarryingMinter;
impl TokenMinter for PortCarryingMinter {
fn mint(&self) -> BoxFuture<'_, Result<ProxyToken, Error>> {
Box::pin(async move {
Ok(ProxyToken::from_pairs([
(PROXY_AUTH_HEADER, "jwe-from-control"),
(PROXY_PORT_HEADER, "9000"),
]))
})
}
}
let auth = ProxyAuth::with_refresh_after(
Arc::new(PortCarryingMinter),
DEFAULT_AGENT_PORT,
DEFAULT_REFRESH_AFTER,
Arc::new(ManualClock::default()),
)
.expect("interval accepted");
let session_headers = auth.headers().await.expect("mints");
assert_eq!(
value_of(&session_headers, PROXY_PORT_HEADER).as_deref(),
Some("9000")
);
let for_workload = auth.headers_for_port(8080).await.expect("reuses");
assert_eq!(
for_workload
.iter()
.filter(|(name, _)| name.eq_ignore_ascii_case(PROXY_PORT_HEADER))
.count(),
1,
"two port headers went out, so which port is targeted depends on ordering: \
{for_workload:?}"
);
assert_eq!(
value_of(&for_workload, PROXY_PORT_HEADER).as_deref(),
Some("8080"),
"the caller's port lost to the token map's"
);
}
#[test]
fn a_proxy_token_debug_does_not_print_the_credential() {
let token = ProxyToken::from_pairs([(PROXY_AUTH_HEADER, "eyJhbGciOi-secret")]);
let rendered = format!("{token:?}");
assert!(rendered.contains(PROXY_AUTH_HEADER), "{rendered}");
assert!(
!rendered.contains("secret"),
"the token value reached a Debug string: {rendered}"
);
}
#[test]
fn the_auth_header_is_found_whatever_case_the_service_used() {
let token = ProxyToken::from_pairs([("x-AWS-Proxy-AUTH", "jwe")]);
assert_eq!(token.auth_value().expect("found"), "jwe");
assert!(!token.is_empty());
}
#[test]
fn a_control_plane_token_converts_with_both_headers_intact() {
let mut map = std::collections::BTreeMap::new();
map.insert(
crate::control::microvm::PROXY_AUTH_HEADER.to_string(),
"jwe-from-control".to_string(),
);
let minted =
crate::control::ProxyToken::from_map(&map, 9000).expect("the map has the auth key");
let converted: ProxyToken = minted.into();
assert_eq!(
converted.auth_value().expect("the auth value survived"),
"jwe-from-control"
);
assert_eq!(
converted.get(PROXY_PORT_HEADER),
Some("9000"),
"the port header did not survive the conversion, so every request built \
from a control-plane token would be rejected as if the token were bad"
);
}
}