use std::collections::BTreeMap;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use areev_core::types::capability::Declaration;
use crate::egress::{EgressDenied, EgressPolicy};
type Result<T> = std::result::Result<T, String>;
const MAX_BODY: usize = 1024 * 1024;
#[derive(Clone)]
pub enum Credential {
Bearer(String),
Header { name: String, value: String },
}
impl std::fmt::Debug for Credential {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Credential::Bearer(_) => f.write_str("Bearer([redacted])"),
Credential::Header { name, .. } => {
write!(f, "Header {{ name: {name:?}, value: [redacted] }}")
}
}
}
}
impl Credential {
pub fn bearer_from_env(var: &str) -> Result<Credential> {
let v = std::env::var(var)
.map_err(|_| format!("credential env var {var} is not set"))?;
if v.trim().is_empty() {
return Err(format!("credential env var {var} is empty"));
}
areev_core::proc::deny_env_var(var);
Ok(Credential::Bearer(v))
}
pub fn bearer_from_env_spec(spec: &str) -> Result<(Credential, Option<String>)> {
let (var, owner) = match spec.split_once('@') {
Some((v, o)) if !o.trim().is_empty() => (v.trim(), Some(o.trim().to_string())),
Some((_, _)) => {
return Err(format!(
"credential spec {spec:?} has an empty principal after '@' — write \
NAME=VAR for an unbound credential, or NAME=VAR@principal to bind one"
))
}
None => (spec.trim(), None),
};
Ok((Self::bearer_from_env(var)?, owner))
}
}
pub const DEFAULT_CREDENTIAL_TTL: std::time::Duration = std::time::Duration::from_secs(300);
const RESOLVER_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
const RESOLVER_MAX_OUTPUT: usize = 64 * 1024;
#[derive(Debug, Clone)]
pub enum CredentialSource {
Static(Credential),
Command {
command: String,
ttl: std::time::Duration,
pass_env: Vec<String>,
},
Vault {
path: String,
field: String,
ttl: std::time::Duration,
},
}
impl From<Credential> for CredentialSource {
fn from(c: Credential) -> Self {
CredentialSource::Static(c)
}
}
impl CredentialSource {
pub fn from_spec(spec: &str) -> Result<(CredentialSource, Option<String>)> {
let spec = spec.trim();
if let Some(rest) = spec.strip_prefix("cmd:") {
let command = rest.trim();
if command.is_empty() {
return Err("credential spec 'cmd:' names no command".into());
}
return Ok((
CredentialSource::Command {
command: command.to_string(),
ttl: DEFAULT_CREDENTIAL_TTL,
pass_env: Vec::new(),
},
None,
));
}
if let Some(rest) = spec.strip_prefix("vault:") {
let (path, field) = rest.trim().split_once('#').ok_or_else(|| {
format!(
"credential spec {spec:?}: a vault source is written \
vault:<path>#<field>, e.g. vault:secret/data/google#access_token"
)
})?;
let (path, field) = (path.trim(), field.trim());
if path.is_empty() || field.is_empty() {
return Err(format!(
"credential spec {spec:?}: both the path and the field must be non-empty"
));
}
for var in ["VAULT_TOKEN", "VAULT_ADDR", "VAULT_NAMESPACE"] {
areev_core::proc::deny_env_var(var);
}
return Ok((
CredentialSource::Vault {
path: path.to_string(),
field: field.to_string(),
ttl: DEFAULT_CREDENTIAL_TTL,
},
None,
));
}
let (cred, owner) = Credential::bearer_from_env_spec(spec)?;
Ok((CredentialSource::Static(cred), owner))
}
pub fn with_resolver_config(
mut self,
ttl_secs: Option<u64>,
resolver_env: &[String],
) -> CredentialSource {
match &mut self {
CredentialSource::Static(_) => {}
CredentialSource::Command { ttl, pass_env, .. } => {
if let Some(s) = ttl_secs {
*ttl = std::time::Duration::from_secs(s.min(u64::from(u32::MAX)));
}
pass_env.extend(resolver_env.iter().cloned());
}
CredentialSource::Vault { ttl, .. } => {
if let Some(s) = ttl_secs {
*ttl = std::time::Duration::from_secs(s.min(u64::from(u32::MAX)));
}
}
}
self
}
fn is_dynamic(&self) -> bool {
!matches!(self, CredentialSource::Static(_))
}
fn ttl(&self) -> std::time::Duration {
match self {
CredentialSource::Static(_) => std::time::Duration::ZERO,
CredentialSource::Command { ttl, .. } | CredentialSource::Vault { ttl, .. } => *ttl,
}
}
fn spawn_policy(pass_env: &[String]) -> areev_core::proc::SpawnPolicy {
let mut allow = areev_core::proc::EnvPolicy::minimal_allow();
allow.extend(pass_env.iter().cloned());
areev_core::proc::SpawnPolicy {
timeout: Some(RESOLVER_TIMEOUT),
max_output_bytes: RESOLVER_MAX_OUTPUT,
env: areev_core::proc::EnvPolicy::ClearExcept { allow },
stderr: areev_core::proc::StderrMode::Pipe,
current_dir: None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CredentialDenied {
NotGranted,
WrongHost,
}
#[derive(Debug, Clone, Default)]
pub struct CallerGrant {
credentials: BTreeMap<String, Vec<areev_core::types::capability::AllowedHost>>,
methods: std::collections::BTreeSet<String>,
}
impl CallerGrant {
pub fn new() -> Self {
Self::default()
}
pub fn credential(mut self, name: &str) -> Self {
self.credentials.entry(name.to_string()).or_default();
self
}
pub fn credential_for(
mut self,
name: &str,
hosts: Vec<areev_core::types::capability::AllowedHost>,
) -> Self {
self.credentials.entry(name.to_string()).or_default().extend(hosts);
self
}
pub fn method(mut self, m: &str) -> Self {
self.methods.insert(m.trim().to_ascii_uppercase());
self
}
fn permits_method(&self, method: &str) -> bool {
if self.methods.is_empty() {
return matches!(method, "GET" | "HEAD");
}
self.methods.contains(method)
}
fn permits_credential(&self, name: &str, url: &str) -> std::result::Result<(), CredentialDenied> {
let Some(hosts) = self.credentials.get(name) else {
return Err(CredentialDenied::NotGranted);
};
if hosts.is_empty() || hosts.iter().any(|h| h.permits_url(url)) {
return Ok(());
}
Err(CredentialDenied::WrongHost)
}
}
#[derive(Debug, Clone, Default)]
pub struct EgressGrants {
by_caller: BTreeMap<String, CallerGrant>,
default_grant: Option<CallerGrant>,
}
impl EgressGrants {
pub fn new() -> Self {
Self::default()
}
pub fn grant(mut self, caller: &str, g: CallerGrant) -> Self {
self.by_caller.insert(caller.to_string(), g);
self
}
pub fn default_for_all(mut self, g: CallerGrant) -> Self {
self.default_grant = Some(g);
self
}
fn for_caller(&self, caller: &str) -> Option<&CallerGrant> {
self.by_caller.get(caller).or(self.default_grant.as_ref())
}
fn callers(&self) -> Vec<String> {
self.by_caller.keys().cloned().collect()
}
}
struct CachedCredential {
value: Credential,
expires_at: std::time::Instant,
}
type CredentialCache = Arc<std::sync::Mutex<BTreeMap<String, CachedCredential>>>;
fn resolve_credential(
name: &str,
source: &CredentialSource,
cache: &CredentialCache,
) -> std::result::Result<Credential, String> {
if let CredentialSource::Static(c) = source {
return Ok(c.clone());
}
let now = std::time::Instant::now();
if let Ok(guard) = cache.lock() {
if let Some(hit) = guard.get(name) {
if hit.expires_at > now {
return Ok(hit.value.clone());
}
}
}
let value = match source {
CredentialSource::Static(_) => unreachable!("handled above"),
CredentialSource::Command { command, pass_env, .. } => {
mint_from_command(name, command, pass_env)?
}
CredentialSource::Vault { path, field, .. } => mint_from_vault(name, path, field)?,
};
let token = Credential::Bearer(value);
if let Ok(mut guard) = cache.lock() {
let expires_at = now.checked_add(source.ttl()).unwrap_or(now);
guard.insert(
name.to_string(),
CachedCredential { value: token.clone(), expires_at },
);
}
Ok(token)
}
fn invalidate_credential(name: &str, cache: &CredentialCache) {
if let Ok(mut guard) = cache.lock() {
guard.remove(name);
}
}
fn mint_from_command(
name: &str,
command: &str,
pass_env: &[String],
) -> std::result::Result<String, String> {
use std::process::Command;
#[cfg(not(windows))]
let mut shell = Command::new("/bin/sh");
#[cfg(not(windows))]
shell.arg("-c").arg(command);
#[cfg(windows)]
let mut shell = Command::new("cmd");
#[cfg(windows)]
{
use std::os::windows::process::CommandExt;
shell.raw_arg("/C").raw_arg(command);
}
let policy = CredentialSource::spawn_policy(pass_env);
let out = areev_core::proc::run(shell, None, &[], &policy)
.map_err(|e| format!("credential '{name}': its resolver could not be started ({e})"))?;
if out.timed_out {
return Err(format!(
"credential '{name}': its resolver exceeded {}s and was killed",
RESOLVER_TIMEOUT.as_secs()
));
}
if !out.status.success() {
return Err(format!(
"credential '{name}': its resolver exited with {}",
out.status
));
}
validate_minted(name, String::from_utf8_lossy(&out.stdout).trim())
}
fn validate_minted(name: &str, value: &str) -> std::result::Result<String, String> {
if value.is_empty() {
return Err(format!("credential '{name}': its resolver returned nothing"));
}
if !areev_core::types::capability::is_valid_header_value(value) {
return Err(format!(
"credential '{name}': its resolver returned a value that is not a valid HTTP header \
value — a control character here would forge a second header on every request"
));
}
Ok(value.to_string())
}
fn mint_from_vault(name: &str, path: &str, field: &str) -> std::result::Result<String, String> {
let missing = |var: &str| {
format!("credential '{name}': a vault source needs {var} in this process's environment")
};
let addr = std::env::var("VAULT_ADDR").map_err(|_| missing("VAULT_ADDR"))?;
let token = std::env::var("VAULT_TOKEN").map_err(|_| missing("VAULT_TOKEN"))?;
let url = format!("{}/v1/{}", addr.trim_end_matches('/'), path.trim_start_matches('/'));
let agent: ureq::Agent = ureq::Agent::config_builder()
.timeout_connect(Some(std::time::Duration::from_secs(5)))
.timeout_recv_response(Some(RESOLVER_TIMEOUT))
.timeout_recv_body(Some(RESOLVER_TIMEOUT))
.timeout_global(Some(RESOLVER_TIMEOUT))
.build()
.into();
let mut req = agent.get(&url).header("X-Vault-Token", &token);
if let Ok(ns) = std::env::var("VAULT_NAMESPACE") {
req = req.header("X-Vault-Namespace", &ns);
}
let mut resp = req
.call()
.map_err(|e| format!("credential '{name}': vault at {url} did not answer ({e})"))?;
let status = resp.status().as_u16();
if status != 200 {
return Err(format!(
"credential '{name}': vault answered {status} for {path} — check the path, the \
token's policy, and (for KV v2) that the path includes its 'data/' segment"
));
}
let text = resp
.body_mut()
.read_to_string()
.map_err(|_| format!("credential '{name}': vault's answer could not be read"))?;
let body: serde_json::Value = serde_json::from_str(&text)
.map_err(|_| format!("credential '{name}': vault's answer was not JSON"))?;
let found = body
.pointer("/data/data")
.and_then(|d| d.get(field))
.or_else(|| body.pointer("/data").and_then(|d| d.get(field)))
.and_then(|v| v.as_str())
.ok_or_else(|| {
format!("credential '{name}': vault's secret at {path} has no string field '{field}'")
})?;
validate_minted(name, found.trim())
}
fn mint_token() -> String {
let mut b = [0u8; 24];
getrandom::getrandom(&mut b).expect("OS randomness for the egress token");
b.iter().map(|x| format!("{x:02x}")).collect()
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EgressRefusal {
pub caller: String,
pub destination: String,
pub reason: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EgressCall {
pub caller: String,
pub method: String,
pub url: String,
pub status: u16,
pub redirects: u32,
pub request_digest: Option<String>,
pub response_digest: String,
pub response_bytes: usize,
pub credential: Option<String>,
pub headers: BTreeMap<String, String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BlobRead {
pub caller: String,
pub uri: String,
pub bytes: usize,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CapabilityLimits {
pub max_calls: u32,
pub max_response_bytes: usize,
}
impl Default for CapabilityLimits {
fn default() -> Self {
CapabilityLimits { max_calls: 64, max_response_bytes: 1024 * 1024 }
}
}
#[derive(Debug, Clone, Default)]
struct Declared {
declaration: areev_core::types::capability::Declaration,
limits: CapabilityLimits,
calls_made: u32,
}
pub struct Broker {
url: String,
stop: Arc<AtomicBool>,
handle: Option<std::thread::JoinHandle<()>>,
refusals: Arc<std::sync::Mutex<Vec<EgressRefusal>>>,
calls: Arc<std::sync::Mutex<Vec<EgressCall>>>,
declared: Arc<std::sync::Mutex<BTreeMap<String, Declared>>>,
credential_owners: Arc<std::sync::Mutex<BTreeMap<String, String>>>,
run_principal: Arc<std::sync::Mutex<Option<String>>>,
blobs: Arc<std::sync::Mutex<Option<String>>>,
blob_reads: Arc<std::sync::Mutex<Vec<BlobRead>>>,
tokens: BTreeMap<String, String>,
default_token: Option<String>,
}
impl Broker {
pub fn start(
policy: EgressPolicy,
credentials: BTreeMap<String, CredentialSource>,
grants: EgressGrants,
refusal_code: &'static str,
) -> Result<Broker> {
let listener = TcpListener::bind("127.0.0.1:0")
.map_err(|e| format!("egress broker cannot bind loopback: {e}"))?;
let port = listener.local_addr().map(|a| a.port()).unwrap_or(0);
listener.set_nonblocking(true).ok();
let stop = Arc::new(AtomicBool::new(false));
let refusals = Arc::new(std::sync::Mutex::new(Vec::new()));
let calls = Arc::new(std::sync::Mutex::new(Vec::new()));
let declared: Arc<std::sync::Mutex<BTreeMap<String, Declared>>> =
Arc::new(std::sync::Mutex::new(BTreeMap::new()));
let credential_owners: Arc<std::sync::Mutex<BTreeMap<String, String>>> =
Arc::new(std::sync::Mutex::new(BTreeMap::new()));
let run_principal: Arc<std::sync::Mutex<Option<String>>> =
Arc::new(std::sync::Mutex::new(None));
let blobs: Arc<std::sync::Mutex<Option<String>>> = Arc::new(std::sync::Mutex::new(None));
let blob_reads: Arc<std::sync::Mutex<Vec<BlobRead>>> =
Arc::new(std::sync::Mutex::new(Vec::new()));
let (stop_t, refusals_t) = (Arc::clone(&stop), Arc::clone(&refusals));
let (calls_t, declared_t) = (Arc::clone(&calls), Arc::clone(&declared));
let (owners_t, principal_t) = (Arc::clone(&credential_owners), Arc::clone(&run_principal));
let (blobs_t, blob_reads_t) = (Arc::clone(&blobs), Arc::clone(&blob_reads));
let cache: CredentialCache = Arc::new(std::sync::Mutex::new(BTreeMap::new()));
let cache_t = Arc::clone(&cache);
let mut tokens = BTreeMap::new();
for c in grants.callers() {
tokens.insert(c, mint_token());
}
let default_token = grants.default_grant.as_ref().map(|_| mint_token());
let by_token: BTreeMap<String, String> = tokens
.iter()
.map(|(c, t)| (t.clone(), c.clone()))
.chain(default_token.iter().map(|t| (t.clone(), String::new())))
.collect();
let handle = std::thread::spawn(move || {
while !stop_t.load(Ordering::Relaxed) {
match listener.accept() {
Ok((stream, _)) => {
if stream.set_nonblocking(false).is_err() {
continue;
}
let _ = serve_one(
stream,
&policy,
&credentials,
&cache_t,
&refusals_t,
&calls_t,
&declared_t,
&owners_t,
&principal_t,
&blobs_t,
&blob_reads_t,
&grants,
&by_token,
refusal_code,
);
}
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::sleep(std::time::Duration::from_millis(5));
}
Err(_) => break,
}
}
});
Ok(Broker {
url: format!("http://127.0.0.1:{port}"),
stop,
handle: Some(handle),
refusals,
calls,
declared,
credential_owners,
run_principal,
blobs,
blob_reads,
tokens,
default_token,
})
}
pub fn serve_blobs(&self, db_path: &str) {
if let Ok(mut b) = self.blobs.lock() {
*b = Some(db_path.to_string());
}
}
pub fn blob_reads(&self) -> Vec<BlobRead> {
self.blob_reads.lock().map(|b| b.clone()).unwrap_or_default()
}
pub fn bind_credential_owner(&self, name: &str, principal: &str) {
if let Ok(mut owners) = self.credential_owners.lock() {
owners.insert(name.to_string(), principal.to_string());
}
}
pub fn bind_run_principal(&self, principal: &str) {
if let Ok(mut p) = self.run_principal.lock() {
*p = Some(principal.to_string());
}
}
pub fn declare(
&self,
caller: &str,
declaration: areev_core::types::capability::Declaration,
limits: CapabilityLimits,
) {
if let Ok(mut map) = self.declared.lock() {
let entry = map.entry(caller.to_string()).or_default();
entry.declaration = declaration;
entry.limits = limits;
}
}
pub fn token_for(&self, caller: &str) -> Option<&str> {
self.tokens
.get(caller)
.or(self.default_token.as_ref())
.map(String::as_str)
}
pub fn calls(&self) -> Vec<EgressCall> {
self.calls.lock().map(|c| c.clone()).unwrap_or_default()
}
pub fn url(&self) -> &str {
&self.url
}
pub fn refusals(&self) -> Vec<EgressRefusal> {
self.refusals.lock().map(|r| r.clone()).unwrap_or_default()
}
}
impl Drop for Broker {
fn drop(&mut self) {
self.stop.store(true, Ordering::Relaxed);
drop(self.handle.take());
}
}
#[derive(serde::Deserialize, Default)]
#[serde(default)]
struct EgressRequest {
url: String,
method: String,
credential: Option<String>,
body: Option<String>,
headers: BTreeMap<String, String>,
}
#[derive(serde::Deserialize, Default)]
#[serde(default)]
struct BlobRequest {
uri: String,
}
#[allow(clippy::too_many_arguments)]
fn serve_blob(
stream: &mut TcpStream,
body: &[u8],
caller: &str,
declared: &Arc<std::sync::Mutex<BTreeMap<String, Declared>>>,
blobs: &Arc<std::sync::Mutex<Option<String>>>,
blob_reads: &Arc<std::sync::Mutex<Vec<BlobRead>>>,
refusals: &Arc<std::sync::Mutex<Vec<EgressRefusal>>>,
refusal_code: &'static str,
) -> std::io::Result<()> {
let req: BlobRequest = match serde_json::from_slice(body) {
Ok(r) => r,
Err(e) => {
return respond(
stream,
400,
&serde_json::json!({ "error": format!("bad blob request: {e}") }).to_string(),
)
}
};
let permitted = declared
.lock()
.map(|d| d.get(caller).is_some_and(|e| e.declaration.declares_blob_read()))
.unwrap_or(false);
if !permitted {
note_refusal(
refusals,
EgressRefusal {
caller: caller.to_string(),
destination: req.uri.clone(),
reason: "read a CAS blob without declaring the 'blob' capability".into(),
},
);
return respond(
stream,
403,
&serde_json::json!({
"error": format!(
"caller '{caller}' asked to read a blob without declaring \
{{\"blob\": {{\"read\": true}}}}"
),
"code": refusal_code
})
.to_string(),
);
}
let Some(db_path) = blobs.lock().ok().and_then(|b| b.clone()) else {
return respond(
stream,
503,
&serde_json::json!({
"error": "this host wired no memory for blob reads"
})
.to_string(),
);
};
if db_path.starts_with("postgres://") || db_path.starts_with("postgresql://") {
return respond(
stream,
501,
&serde_json::json!({
"error": "blob reads from a sandboxed tool are not supported on the postgres \
backend: the lock-free path is the .blobs sidecar, which only the \
embedded backend has"
})
.to_string(),
);
}
match areev_store::read_blob_offline(&db_path, &req.uri) {
Ok(Some(bytes)) => {
note_blob_read(
blob_reads,
BlobRead {
caller: caller.to_string(),
uri: req.uri.clone(),
bytes: bytes.len(),
},
);
respond_bytes(stream, &bytes)
}
Ok(None) => respond(
stream,
409,
&serde_json::json!({
"error": format!(
"blob {} is encrypted at rest; a sandboxed tool cannot open it",
req.uri
)
})
.to_string(),
),
Err(e) => respond(
stream,
404,
&serde_json::json!({ "error": e.to_string() }).to_string(),
),
}
}
const MAX_RECORDED_CALLS: usize = 4096;
fn note_blob_read(reads: &Arc<std::sync::Mutex<Vec<BlobRead>>>, r: BlobRead) {
if let Ok(mut list) = reads.lock() {
if list.len() < MAX_RECORDED_CALLS {
list.push(r);
}
}
}
#[allow(clippy::too_many_arguments)]
fn note_refusal(refusals: &Arc<std::sync::Mutex<Vec<EgressRefusal>>>, r: EgressRefusal) {
if let Ok(mut list) = refusals.lock() {
if !list.contains(&r) {
list.push(r);
}
}
}
#[allow(clippy::too_many_arguments)]
fn serve_one(
mut stream: TcpStream,
policy: &EgressPolicy,
credentials: &BTreeMap<String, CredentialSource>,
cache: &CredentialCache,
refusals: &Arc<std::sync::Mutex<Vec<EgressRefusal>>>,
calls: &Arc<std::sync::Mutex<Vec<EgressCall>>>,
declared: &Arc<std::sync::Mutex<BTreeMap<String, Declared>>>,
credential_owners: &Arc<std::sync::Mutex<BTreeMap<String, String>>>,
run_principal: &Arc<std::sync::Mutex<Option<String>>>,
blobs: &Arc<std::sync::Mutex<Option<String>>>,
blob_reads: &Arc<std::sync::Mutex<Vec<BlobRead>>>,
grants: &EgressGrants,
by_token: &BTreeMap<String, String>,
refusal_code: &'static str,
) -> std::io::Result<()> {
stream.set_read_timeout(Some(std::time::Duration::from_secs(30))).ok();
let mut reader = BufReader::new(stream.try_clone()?);
let mut line = String::new();
reader.read_line(&mut line)?;
let path = line.split_whitespace().nth(1).unwrap_or("/").to_string();
let want_blob = path.starts_with("/blob");
let mut content_length = 0usize;
let mut presented: Option<String> = None;
loop {
let mut h = String::new();
if reader.read_line(&mut h)? == 0 || h.trim().is_empty() {
break;
}
let lower = h.to_ascii_lowercase();
if let Some(v) = lower.strip_prefix("content-length:") {
content_length = v.trim().parse().unwrap_or(0);
}
if lower.starts_with("x-areev-egress-token:") {
presented = h.split_once(':').map(|(_, v)| v.trim().to_string());
}
}
let caller = match presented.and_then(|t| by_token.get(&t).cloned()) {
Some(c) => c,
None => {
drain_body(&mut reader, content_length);
return respond(
&mut stream,
401,
&serde_json::json!({
"error": "missing or unknown X-Areev-Egress-Token",
"code": refusal_code
})
.to_string(),
)
}
};
let Some(grant) = grants.for_caller(&caller) else {
drain_body(&mut reader, content_length);
return respond(
&mut stream,
403,
&serde_json::json!({
"error": format!("caller '{caller}' has no egress grant"),
"code": refusal_code
})
.to_string(),
);
};
if content_length > MAX_BODY {
return respond(&mut stream, 413, r#"{"error":"body too large"}"#);
}
let mut body = vec![0u8; content_length];
reader.read_exact(&mut body)?;
if want_blob {
return serve_blob(
&mut stream,
&body,
&caller,
declared,
blobs,
blob_reads,
refusals,
refusal_code,
);
}
let req: EgressRequest = match serde_json::from_slice(&body) {
Ok(r) => r,
Err(e) => {
return respond(
&mut stream,
400,
&serde_json::json!({ "error": format!("bad egress request: {e}") }).to_string(),
)
}
};
for (name, value) in &req.headers {
if !areev_core::types::capability::is_valid_header_name(name)
|| !areev_core::types::capability::is_valid_header_value(value)
{
return respond(
&mut stream,
400,
&serde_json::json!({
"error": format!("header '{name}' is not a valid HTTP header name/value")
})
.to_string(),
);
}
if areev_core::types::capability::is_broker_owned_header(name) {
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: req.url.clone(),
reason: format!("tried to set the broker-owned header '{name}'"),
},
);
return respond(
&mut stream,
403,
&serde_json::json!({
"error": format!(
"caller '{caller}' tried to set header '{name}', which the broker owns — \
name a credential instead; the broker attaches it and the caller never \
holds one"
),
"code": refusal_code
})
.to_string(),
);
}
}
let guest_headers: Vec<(String, String)> =
req.headers.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
let guest_header_names: Vec<String> = req.headers.keys().cloned().collect();
if let Err(e) = policy.permits(&req.url) {
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: req.url.clone(),
reason: "destination outside the declared allowlist".into(),
},
);
return respond(
&mut stream,
403,
&serde_json::json!({
"error": format!("caller '{caller}' {e}"),
"code": refusal_code
})
.to_string(),
);
}
let method = if req.method.trim().is_empty() { "GET" } else { req.method.trim() };
let method = method.to_ascii_uppercase();
let capability: Option<(CapabilityLimits, Declaration)> = {
let mut guard = declared.lock().unwrap_or_else(|e| e.into_inner());
match guard.get_mut(&caller) {
None => None,
Some(d) => {
if d.calls_made >= d.limits.max_calls {
let max = d.limits.max_calls;
drop(guard);
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: req.url.clone(),
reason: format!("exceeded its ceiling of {max} brokered calls"),
},
);
return respond(
&mut stream,
403,
&serde_json::json!({
"error": format!(
"caller '{caller}' has made its {max} permitted brokered calls"
),
"code": refusal_code
})
.to_string(),
);
}
d.calls_made += 1;
if let Err(denied) = d.declaration.permits(
&req.url,
&method,
req.credential.as_deref(),
&guest_header_names,
) {
drop(guard);
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: req.url.clone(),
reason: format!("undeclared capability: {denied}"),
},
);
return respond(
&mut stream,
403,
&serde_json::json!({
"error": format!("caller '{caller}' {denied}"),
"code": refusal_code
})
.to_string(),
);
}
Some((d.limits, d.declaration.clone()))
}
}
};
if capability.is_some() && policy.is_unrestricted() && crate::egress::is_private_destination(&req.url)
{
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: req.url.clone(),
reason: "private or loopback destination without an explicit --allow-host entry"
.into(),
},
);
return respond(
&mut stream,
403,
&serde_json::json!({
"error": format!(
"caller '{caller}' tried to reach a private or loopback destination — a \
capability declaration alone cannot authorize one; the host must name it \
in --allow-host"
),
"code": refusal_code
})
.to_string(),
);
}
if !grant.permits_method(&method) {
let denied = EgressDenied::Method { method: method.clone() };
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: req.url.clone(),
reason: format!("method {method} is not permitted for this caller"),
},
);
return respond(
&mut stream,
403,
&serde_json::json!({
"error": format!("caller '{caller}' {denied}"),
"code": refusal_code
})
.to_string(),
);
}
let mut headers: Vec<(String, String)> = Vec::new();
let mut credential_source: Option<&CredentialSource> = None;
if let Some(name) = &req.credential {
let owner = {
let owners = credential_owners.lock().unwrap_or_else(|e| e.into_inner());
owners.get(name.as_str()).cloned()
};
if let Some(owner) = owner {
let bound = run_principal.lock().unwrap_or_else(|e| e.into_inner()).clone();
if bound.as_deref() != Some(owner.as_str()) {
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: req.url.clone(),
reason: format!(
"credential '{name}' is bound to principal '{owner}', and this run \
executes as {}",
bound.as_deref().unwrap_or("no bound principal")
),
},
);
return respond(
&mut stream,
403,
&serde_json::json!({
"error": format!(
"caller '{caller}' may not use credential '{name}' — it is bound to \
a different run principal"
),
"code": refusal_code
})
.to_string(),
);
}
}
if let Err(denied) = grant.permits_credential(name, &req.url) {
let (reason, detail) = match denied {
CredentialDenied::NotGranted => (
format!("credential '{name}' is not granted to this caller"),
format!("caller '{caller}' may not use credential '{name}'"),
),
CredentialDenied::WrongHost => (
format!(
"credential '{name}' is granted to this caller only for other hosts"
),
format!(
"caller '{caller}' may not send credential '{name}' to this host — the \
grant pairs it with a different one"
),
),
};
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: req.url.clone(),
reason,
},
);
return respond(
&mut stream,
403,
&serde_json::json!({ "error": detail, "code": refusal_code }).to_string(),
);
}
let Some(source) = credentials.get(name) else {
return respond(
&mut stream,
400,
&serde_json::json!({
"error": format!("no credential named '{name}' is configured for this run")
})
.to_string(),
);
};
match resolve_credential(name, source, cache) {
Ok(Credential::Bearer(v)) => {
headers.push(("Authorization".into(), format!("Bearer {v}")))
}
Ok(Credential::Header { name, value }) => headers.push((name.clone(), value.clone())),
Err(detail) => {
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: req.url.clone(),
reason: detail,
},
);
return respond(
&mut stream,
403,
&serde_json::json!({
"error": format!(
"caller '{caller}' could not be given credential '{name}' — its \
source did not resolve; the run's audit trail records why"
),
"code": refusal_code
})
.to_string(),
);
}
}
credential_source = Some(source);
}
for (guest_name, _) in &guest_headers {
let g = guest_name.trim().to_ascii_lowercase();
if headers.iter().any(|(k, _)| k.trim().to_ascii_lowercase() == g) {
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: req.url.clone(),
reason: format!("tried to set the broker-owned header '{guest_name}'"),
},
);
return respond(
&mut stream,
403,
&serde_json::json!({
"error": format!(
"caller '{caller}' tried to set header '{guest_name}', which carries a \
configured credential on this host — name the credential instead"
),
"code": refusal_code
})
.to_string(),
);
}
}
if !matches!(method.as_str(), "GET" | "HEAD" | "DELETE" | "POST" | "PUT" | "PATCH") {
return respond(
&mut stream,
400,
&serde_json::json!({ "error": format!("unsupported method '{method}'") }).to_string(),
);
}
let mut outcome = dispatch(
&req.url,
&method,
req.body.as_deref(),
&headers,
&guest_headers,
policy,
grant,
capability.as_ref().map(|(_, d)| d),
capability.as_ref().map(|(l, _)| l.max_response_bytes),
&caller,
refusals,
);
if let (Some(name), Some(source)) = (&req.credential, credential_source) {
let unauthorized =
matches!(&outcome, Dispatched::Answered { status: 401, credential_sent: true, .. });
if unauthorized && source.is_dynamic() {
invalidate_credential(name, cache);
if matches!(method.as_str(), "GET" | "HEAD") {
if let Ok(fresh) = resolve_credential(name, source, cache) {
if let Dispatched::Answered {
status, body, final_url, redirects, credential_sent,
} = &outcome
{
note_call(
calls,
EgressCall {
caller: caller.clone(),
method: method.clone(),
url: final_url.clone(),
status: *status,
redirects: *redirects,
request_digest: req.body.as_deref().map(digest),
response_digest: digest(&scrub_reflected(body.clone(), &headers)),
response_bytes: body.len(),
credential: if *credential_sent {
req.credential.clone()
} else {
None
},
headers: if *credential_sent {
req.headers.clone()
} else {
BTreeMap::new()
},
},
);
}
let refreshed: Vec<(String, String)> = match fresh {
Credential::Bearer(v) => {
vec![("Authorization".into(), format!("Bearer {v}"))]
}
Credential::Header { name, value } => vec![(name, value)],
};
outcome = dispatch(
&req.url,
&method,
req.body.as_deref(),
&refreshed,
&guest_headers,
policy,
grant,
capability.as_ref().map(|(_, d)| d),
capability.as_ref().map(|(l, _)| l.max_response_bytes),
&caller,
refusals,
);
headers = refreshed;
}
}
}
}
match outcome {
Dispatched::Answered { status, body, final_url, redirects, credential_sent } => {
let body = scrub_reflected(body, &headers);
note_call(
calls,
EgressCall {
caller: caller.clone(),
method: method.clone(),
url: final_url,
status,
redirects,
request_digest: req.body.as_deref().map(digest),
response_digest: digest(&body),
response_bytes: body.len(),
credential: if credential_sent { req.credential.clone() } else { None },
headers: if credential_sent {
req.headers.clone()
} else {
BTreeMap::new()
},
},
);
respond(
&mut stream,
200,
&serde_json::json!({ "status": status, "body": body }).to_string(),
)
}
Dispatched::Refused { detail } => respond(
&mut stream,
403,
&serde_json::json!({ "error": detail, "code": refusal_code }).to_string(),
),
Dispatched::TooLarge { final_url } => {
let max =
capability.as_ref().map(|(l, _)| l.max_response_bytes).unwrap_or_default();
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: final_url,
reason: format!("response exceeds its {max}-byte ceiling"),
},
);
respond(
&mut stream,
403,
&serde_json::json!({
"error": format!(
"caller '{caller}' received a response larger than its {max}-byte \
ceiling — refused rather than truncated"
),
"code": refusal_code
})
.to_string(),
)
}
Dispatched::Upstream(e) => respond(
&mut stream,
502,
&serde_json::json!({ "error": format!("upstream: {e}") }).to_string(),
),
}
}
fn scrub_reflected(mut body: String, headers: &[(String, String)]) -> String {
for (_, value) in headers {
if value.len() >= 8 && body.contains(value.as_str()) {
body = body.replace(value.as_str(), "[redacted-credential]");
}
if let Some(bare) = value.strip_prefix("Bearer ") {
if bare.len() >= 8 && body.contains(bare) {
body = body.replace(bare, "[redacted-credential]");
}
}
}
body
}
fn digest(body: &str) -> String {
use sha2::{Digest, Sha256};
let mut h = Sha256::new();
h.update(body.as_bytes());
format!("sha256:{:x}", h.finalize())
}
fn note_call(calls: &Arc<std::sync::Mutex<Vec<EgressCall>>>, c: EgressCall) {
if let Ok(mut list) = calls.lock() {
if list.len() < MAX_RECORDED_CALLS {
list.push(c);
}
}
}
const MAX_REDIRECT_HOPS: usize = 10;
enum Dispatched {
Answered { status: u16, body: String, final_url: String, redirects: u32, credential_sent: bool },
Refused { detail: String },
TooLarge { final_url: String },
Upstream(String),
}
#[allow(clippy::too_many_arguments)]
fn dispatch(
start_url: &str,
start_method: &str,
body: Option<&str>,
credential_headers: &[(String, String)],
guest_headers: &[(String, String)],
policy: &EgressPolicy,
grant: &CallerGrant,
declaration: Option<&Declaration>,
max_response_bytes: Option<usize>,
caller: &str,
refusals: &Arc<std::sync::Mutex<Vec<EgressRefusal>>>,
) -> Dispatched {
let agent: ureq::Agent = ureq::Agent::config_builder()
.timeout_connect(Some(std::time::Duration::from_secs(15)))
.timeout_recv_response(Some(std::time::Duration::from_secs(60)))
.timeout_recv_body(Some(std::time::Duration::from_secs(60)))
.max_redirects(0)
.http_status_as_error(false)
.build()
.into();
let mut url = start_url.to_string();
let mut method = start_method.to_string();
let mut body = body.map(str::to_string);
let mut left_origin = false;
let read_body = |resp: &mut ureq::http::Response<ureq::Body>| -> std::result::Result<String, ()> {
match max_response_bytes {
Some(n) => match resp.body_mut().with_config().limit(n as u64 + 1).read_to_string() {
Ok(text) if text.len() > n => Err(()),
Ok(text) => Ok(text),
Err(ureq::Error::BodyExceedsLimit(_)) => Err(()),
Err(_) => Ok(String::new()),
},
None => Ok(resp.body_mut().read_to_string().unwrap_or_default()),
}
};
for hops in 0..=MAX_REDIRECT_HOPS as u32 {
let send_credential =
hops == 0 || (!left_origin && crate::egress::same_origin(start_url, &url));
let result = perform(
&agent,
&method,
&url,
body.as_deref(),
credential_headers,
guest_headers,
send_credential,
);
let mut resp = match result {
Ok(r) => r,
Err(e) => return Dispatched::Upstream(e.to_string()),
};
let status = resp.status().as_u16();
let location = resp
.headers()
.get("location")
.and_then(|v| v.to_str().ok())
.map(str::to_string);
let Some(next_method) = redirect_method(status, &method) else {
return match read_body(&mut resp) {
Ok(text) => Dispatched::Answered {
status,
body: text,
final_url: url,
redirects: hops,
credential_sent: send_credential,
},
Err(()) => Dispatched::TooLarge { final_url: url },
};
};
let Some(location) = location else {
return match read_body(&mut resp) {
Ok(text) => Dispatched::Answered {
status,
body: text,
final_url: url,
redirects: hops,
credential_sent: send_credential,
},
Err(()) => Dispatched::TooLarge { final_url: url },
};
};
let Some(next_url) = crate::egress::resolve_location(&url, &location) else {
note_refusal(
refusals,
EgressRefusal {
caller: caller.to_string(),
destination: location.clone(),
reason: "redirected to a Location that is not a resolvable http(s) URL".into(),
},
);
return Dispatched::Refused {
detail: format!(
"caller '{caller}' was redirected to {location:?}, which does not resolve to \
an http(s) URL — refused rather than guessed at"
),
};
};
if let Err(e) = policy.permits(&next_url) {
note_refusal(
refusals,
EgressRefusal {
caller: caller.to_string(),
destination: next_url.clone(),
reason: "redirect target outside the declared allowlist".into(),
},
);
return Dispatched::Refused {
detail: format!(
"caller '{caller}' was redirected from {url} and {e} — the redirect was not \
followed"
),
};
}
if !grant.permits_method(&next_method) {
let denied = EgressDenied::Method { method: next_method.clone() };
note_refusal(
refusals,
EgressRefusal {
caller: caller.to_string(),
destination: next_url.clone(),
reason: format!(
"redirect would issue {next_method}, which is not permitted for this caller"
),
},
);
return Dispatched::Refused {
detail: format!("caller '{caller}' {denied} after a redirect"),
};
}
if let Some(d) = declaration {
if policy.is_unrestricted() && crate::egress::is_private_destination(&next_url) {
note_refusal(
refusals,
EgressRefusal {
caller: caller.to_string(),
destination: next_url.clone(),
reason: "redirected to a private or loopback destination without an \
explicit --allow-host entry"
.into(),
},
);
return Dispatched::Refused {
detail: format!(
"caller '{caller}' was redirected from {url} to a private or loopback \
destination — the redirect was not followed"
),
};
}
if let Err(denied) = d.permits(&next_url, &next_method, None, &[]) {
note_refusal(
refusals,
EgressRefusal {
caller: caller.to_string(),
destination: next_url.clone(),
reason: format!("redirect outside the declared capability: {denied}"),
},
);
return Dispatched::Refused {
detail: format!(
"caller '{caller}' was redirected from {url} and {denied} — the redirect \
was not followed"
),
};
}
}
if next_method != method && next_method == "GET" {
body = None;
}
if !crate::egress::same_origin(start_url, &next_url) {
left_origin = true;
}
method = next_method;
url = next_url;
}
note_refusal(
refusals,
EgressRefusal {
caller: caller.to_string(),
destination: url.clone(),
reason: format!("redirect chain exceeded {MAX_REDIRECT_HOPS} hops"),
},
);
Dispatched::Refused {
detail: format!(
"caller '{caller}' followed {MAX_REDIRECT_HOPS} redirects without reaching a final \
response — the chain was abandoned at {url}"
),
}
}
fn redirect_method(status: u16, method: &str) -> Option<String> {
match status {
307 | 308 => match method {
"GET" | "HEAD" => Some(method.to_string()),
_ => None,
},
301..=303 => match method {
"GET" | "HEAD" => Some(method.to_string()),
_ => Some("GET".to_string()),
},
_ => None,
}
}
fn perform(
agent: &ureq::Agent,
method: &str,
url: &str,
body: Option<&str>,
credential_headers: &[(String, String)],
guest_headers: &[(String, String)],
send_credential: bool,
) -> std::result::Result<ureq::http::Response<ureq::Body>, ureq::Error> {
let credential_headers: &[(String, String)] =
if send_credential { credential_headers } else { &[] };
let guest_headers: &[(String, String)] = if send_credential { guest_headers } else { &[] };
match method {
"POST" | "PUT" | "PATCH" => {
let mut b = match method {
"POST" => agent.post(url),
"PUT" => agent.put(url),
_ => agent.patch(url),
};
for (k, v) in guest_headers.iter().chain(credential_headers) {
b = b.header(k, v);
}
b.send(body.unwrap_or(""))
}
_ => {
let mut b = match method {
"HEAD" => agent.head(url),
"DELETE" => agent.delete(url),
_ => agent.get(url),
};
for (k, v) in guest_headers.iter().chain(credential_headers) {
b = b.header(k, v);
}
b.call()
}
}
}
fn drain_body(reader: &mut BufReader<TcpStream>, content_length: usize) {
let mut discard = vec![0u8; content_length.min(MAX_BODY)];
let _ = reader.read_exact(&mut discard);
}
fn respond(stream: &mut TcpStream, status: u16, body: &str) -> std::io::Result<()> {
let head = format!(
"HTTP/1.1 {status} \r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
stream.write_all(head.as_bytes())?;
stream.write_all(body.as_bytes())?;
stream.flush()
}
fn respond_bytes(stream: &mut TcpStream, body: &[u8]) -> std::io::Result<()> {
let head = format!(
"HTTP/1.1 200 \r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
stream.write_all(head.as_bytes())?;
stream.write_all(body)?;
stream.flush()
}
#[cfg(test)]
mod tests {
use super::*;
fn call(broker: &Broker, req: serde_json::Value) -> (u16, serde_json::Value) {
call_as(broker, "connector", req)
}
fn call_as(broker: &Broker, caller: &str, req: serde_json::Value) -> (u16, serde_json::Value) {
let token = broker.token_for(caller).unwrap_or("not-a-real-token").to_string();
call_with_token(broker, &token, req)
}
fn call_with_token(
broker: &Broker,
token: &str,
req: serde_json::Value,
) -> (u16, serde_json::Value) {
let addr = broker.url().trim_start_matches("http://").to_string();
let mut s = TcpStream::connect(addr).unwrap();
let body = req.to_string();
let head = format!(
"POST / HTTP/1.1\r\nHost: localhost\r\nX-Areev-Egress-Token: {token}\r\n\
Content-Length: {}\r\n\r\n",
body.len()
);
s.write_all(head.as_bytes()).unwrap();
s.write_all(body.as_bytes()).unwrap();
s.flush().unwrap();
let mut raw = String::new();
s.read_to_string(&mut raw).unwrap();
let status: u16 = raw
.split_whitespace()
.nth(1)
.and_then(|s| s.parse().ok())
.unwrap_or(0);
let body = raw.split("\r\n\r\n").nth(1).unwrap_or("{}");
(status, serde_json::from_str(body).unwrap_or(serde_json::Value::Null))
}
fn grants(creds: &[&str]) -> EgressGrants {
let g = creds.iter().fold(
CallerGrant::new().method("GET").method("POST").method("DELETE"),
|g, c| g.credential(c),
);
EgressGrants::new().default_for_all(g)
}
fn policy(entries: &[&str]) -> EgressPolicy {
EgressPolicy::from_config(Some(&serde_json::json!({
"int:allowed_outbound_hosts": entries
})))
.unwrap()
}
#[test]
fn a_disallowed_destination_is_refused_before_any_request_is_made() {
let b = Broker::start(policy(&["https://api.github.com"]), BTreeMap::new(), grants(&[]), "TRG-E009").unwrap();
let (status, body) = call(&b, serde_json::json!({ "url": "https://evil.com/steal" }));
assert_eq!(status, 403);
assert_eq!(body["code"], "TRG-E009");
let refusals = b.refusals();
assert_eq!(refusals.len(), 1);
assert_eq!(refusals[0].destination, "https://evil.com/steal");
assert!(refusals[0].reason.contains("allowlist"), "{:?}", refusals[0]);
}
#[test]
fn a_caller_cannot_name_a_credential_its_grant_does_not_cover() {
let b = Broker::start(
policy(&["https://api.github.com"]),
BTreeMap::new(),
grants(&[]),
"TRG-E009",
)
.unwrap();
let (status, body) = call(
&b,
serde_json::json!({ "url": "https://api.github.com/x", "credential": "gmail" }),
);
assert_eq!(status, 403);
assert!(body["error"].as_str().unwrap().contains("may not use credential"), "{body}");
}
#[test]
fn a_granted_credential_that_the_host_never_configured_is_a_client_error() {
let b = Broker::start(
policy(&["https://api.github.com"]),
BTreeMap::new(),
grants(&["gmail"]),
"TRG-E009",
)
.unwrap();
let (status, body) = call(
&b,
serde_json::json!({ "url": "https://api.github.com/x", "credential": "gmail" }),
);
assert_eq!(status, 400);
assert!(body["error"].as_str().unwrap().contains("no credential named"), "{body}");
}
#[test]
fn a_malformed_request_is_a_client_error_not_a_panic() {
let b = Broker::start(policy(&[]), BTreeMap::new(), grants(&[]), "TRG-E009").unwrap();
let (status, _) = call(&b, serde_json::json!("not an egress request"));
assert_eq!(status, 400);
}
#[test]
fn a_caller_with_no_token_is_refused_before_its_body_is_parsed() {
let b = Broker::start(policy(&[]), BTreeMap::new(), grants(&[]), "TRG-E009").unwrap();
let addr = b.url().trim_start_matches("http://").to_string();
let mut s = TcpStream::connect(addr).unwrap();
let body = "not json";
s.write_all(
format!("POST / HTTP/1.1\r\nContent-Length: {}\r\n\r\n{body}", body.len())
.as_bytes(),
)
.unwrap();
s.flush().unwrap();
let mut raw = String::new();
s.read_to_string(&mut raw).unwrap();
assert!(raw.contains("401"), "{raw}");
}
#[test]
fn a_forged_token_is_refused() {
let b = Broker::start(policy(&[]), BTreeMap::new(), grants(&[]), "TRG-E009").unwrap();
let (status, _) =
call_with_token(&b, "0".repeat(48).as_str(), serde_json::json!({ "url": "https://x/" }));
assert_eq!(status, 401);
}
#[test]
fn a_refusal_drains_the_body_so_a_large_write_never_resets() {
let b = Broker::start(policy(&[]), BTreeMap::new(), grants(&[]), "TRG-E009").unwrap();
let pad = "a".repeat(900_000);
let (status, _) =
call_with_token(&b, "0".repeat(48).as_str(), serde_json::json!({ "pad": pad }));
assert_eq!(status, 401);
}
#[test]
fn a_grant_naming_no_method_may_only_read() {
let b = Broker::start(
EgressPolicy::unrestricted(),
BTreeMap::new(),
EgressGrants::new().grant("reader", CallerGrant::new()),
"TRG-E009",
)
.unwrap();
let (status, body) = call_as(
&b,
"reader",
serde_json::json!({ "url": "https://example.com/x", "method": "POST" }),
);
assert_eq!(status, 403);
assert!(body["error"].as_str().unwrap().contains("not permitted"), "{body}");
}
#[test]
fn one_callers_token_does_not_buy_anothers_scope() {
let b = Broker::start(
EgressPolicy::unrestricted(),
BTreeMap::new(),
EgressGrants::new()
.grant("writer", CallerGrant::new().method("POST").credential("zoho"))
.grant("reader", CallerGrant::new()),
"TRG-E009",
)
.unwrap();
assert_ne!(b.token_for("writer").unwrap(), b.token_for("reader").unwrap());
let (status, _) = call_as(
&b,
"reader",
serde_json::json!({ "url": "https://example.com/x", "method": "POST" }),
);
assert_eq!(status, 403);
let (status, _) = call_as(
&b,
"reader",
serde_json::json!({ "url": "https://example.com/x", "credential": "zoho" }),
);
assert_eq!(status, 403);
}
#[test]
fn a_caller_with_no_grant_at_all_gets_no_token() {
let b = Broker::start(
EgressPolicy::unrestricted(),
BTreeMap::new(),
EgressGrants::new().grant("writer", CallerGrant::new()),
"TRG-E009",
)
.unwrap();
assert!(b.token_for("nobody").is_none(), "an ungranted caller must not see the broker");
}
#[test]
fn the_broker_binds_loopback_only() {
let b = Broker::start(policy(&[]), BTreeMap::new(), grants(&[]), "TRG-E009").unwrap();
assert!(b.url().starts_with("http://127.0.0.1:"), "{}", b.url());
}
#[test]
fn credentials_come_from_a_named_variable_never_a_literal() {
std::env::set_var("AREEV_TEST_BROKER_TOKEN", "s3cret");
let c = Credential::bearer_from_env("AREEV_TEST_BROKER_TOKEN").unwrap();
std::env::remove_var("AREEV_TEST_BROKER_TOKEN");
assert!(matches!(c, Credential::Bearer(v) if v == "s3cret"));
assert!(Credential::bearer_from_env("AREEV_TEST_BROKER_ABSENT").is_err());
std::env::set_var("AREEV_TEST_BROKER_EMPTY", " ");
assert!(Credential::bearer_from_env("AREEV_TEST_BROKER_EMPTY").is_err());
std::env::remove_var("AREEV_TEST_BROKER_EMPTY");
}
#[test]
fn an_unrestricted_policy_still_brokers_rather_than_handing_over_the_token() {
let b = Broker::start(EgressPolicy::unrestricted(), BTreeMap::new(), grants(&[]), "TRG-E009").unwrap();
let (status, _) = call(&b, serde_json::json!({ "url": "https://anywhere.example/" }));
assert_ne!(status, 403, "an absent allowlist does not refuse");
}
}