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_cal::AreevFacade;
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;
const DEFAULT_TRANSFER: usize = areev_core::types::capability::DEFAULT_TRANSFER_BYTES as usize;
const MAX_TRANSFER: usize = areev_core::types::capability::MAX_TRANSFER_BYTES as usize;
#[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"));
}
if !areev_core::types::capability::is_valid_header_value(&v) {
return Err(format!(
"credential env var {var} holds a control character (such as CR/LF) and cannot \
be sent in an HTTP header"
));
}
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>,
header: Option<String>,
},
Vault {
path: String,
field: String,
ttl: std::time::Duration,
header: Option<String>,
},
}
const CREDENTIAL_HEADER_REFUSED: [&str; 8] = [
"connection",
"content-length",
"expect",
"keep-alive",
"te",
"trailer",
"transfer-encoding",
"upgrade",
];
fn check_credential_header(name: &str) -> Result<()> {
if !areev_core::types::capability::is_valid_header_name(name) {
return Err(format!(
"credential header {name:?} is not a valid HTTP header name — write \
header:<Name>=<source>, e.g. header:X-Api-Key=cmd:…"
));
}
let lower = name.to_ascii_lowercase();
if areev_core::types::capability::is_broker_owned_header(name) {
return Err(format!(
"credential header {name:?} is owned by the broker — a bearer credential already \
rides in Authorization; drop the header: prefix to use it"
));
}
if CREDENTIAL_HEADER_REFUSED.contains(&lower.as_str()) {
return Err(format!(
"credential header {name:?} frames the HTTP message and cannot carry a credential"
));
}
Ok(())
}
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("header:") {
let (header, inner) = rest.split_once('=').ok_or_else(|| {
format!(
"credential spec {spec:?}: a header credential is written \
header:<Name>=<source>, e.g. header:X-Api-Key=cmd:cat /run/secrets/key"
)
})?;
let header = header.trim();
check_credential_header(header)?;
if inner.trim_start().starts_with("header:") {
return Err(format!("credential spec {spec:?} selects a header twice"));
}
let (source, owner) = Self::from_spec(inner)?;
return Ok((source.in_header(header), owner));
}
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(),
header: None,
},
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,
header: None,
},
None,
));
}
let (cred, owner) = Credential::bearer_from_env_spec(spec)?;
Ok((CredentialSource::Static(cred), owner))
}
pub fn source_spec(spec: &str) -> &str {
let spec = spec.trim();
match spec.strip_prefix("header:").and_then(|r| r.split_once('=')) {
Some((_, inner)) => inner.trim(),
None => spec,
}
}
fn in_header(self, name: &str) -> CredentialSource {
let name = name.to_string();
match self {
CredentialSource::Static(Credential::Bearer(value))
| CredentialSource::Static(Credential::Header { value, .. }) => {
CredentialSource::Static(Credential::Header { name, value })
}
CredentialSource::Command { command, ttl, pass_env, .. } => {
CredentialSource::Command { command, ttl, pass_env, header: Some(name) }
}
CredentialSource::Vault { path, field, ttl, .. } => {
CredentialSource::Vault { path, field, ttl, header: Some(name) }
}
}
}
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, header) = match source {
CredentialSource::Static(_) => unreachable!("handled above"),
CredentialSource::Command { command, pass_env, header, .. } => {
(mint_from_command(name, command, pass_env)?, header)
}
CredentialSource::Vault { path, field, header, .. } => {
(mint_from_vault(name, path, field)?, header)
}
};
let token = match header {
Some(h) => Credential::Header { name: h.clone(), value },
None => 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::fill(&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 response_mime: Option<String>,
pub response_ref: Option<String>,
pub credential: Option<String>,
pub headers: BTreeMap<String, String>,
pub response_headers: BTreeMap<String, String>,
}
pub const EXPOSABLE_RESPONSE_HEADERS: &[&str] = &["content-type", "mcp-session-id"];
const MAX_EXPOSED_HEADER_VALUE: usize = 1024;
#[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,
pub max_request_bytes: usize,
}
impl Default for CapabilityLimits {
fn default() -> Self {
CapabilityLimits {
max_calls: 64,
max_response_bytes: DEFAULT_TRANSFER,
max_request_bytes: DEFAULT_TRANSFER,
}
}
}
#[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>>>,
artifact_store: Arc<std::sync::Mutex<Option<Arc<AreevFacade>>>>,
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 artifact_store = 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 artifact_store_t = Arc::clone(&artifact_store);
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,
&artifact_store_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,
artifact_store,
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 bind_artifact_store(&self, store: Arc<AreevFacade>) {
if let Ok(mut slot) = self.artifact_store.lock() {
*slot = Some(store);
}
}
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);
if let Ok(mut slot) = self.artifact_store.lock() {
slot.take();
}
drop(self.handle.take());
}
}
#[derive(serde::Deserialize, Default)]
#[serde(default)]
struct EgressRequest {
url: String,
method: String,
credential: Option<String>,
body: Option<String>,
response_mode: Option<String>,
body_ref: Option<String>,
content_type: Option<String>,
headers: BTreeMap<String, String>,
response_headers: Vec<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(),
);
};
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>>>,
artifact_store: &Arc<std::sync::Mutex<Option<Arc<AreevFacade>>>>,
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 mut expose: Vec<String> = Vec::new();
for name in &req.response_headers {
let lower = name.trim().to_ascii_lowercase();
if !EXPOSABLE_RESPONSE_HEADERS.contains(&lower.as_str()) {
return respond(
&mut stream,
400,
&serde_json::json!({
"error": format!(
"response header '{name}' cannot be handed back — the broker exposes \
only {}",
EXPOSABLE_RESPONSE_HEADERS.join(", ")
)
})
.to_string(),
);
}
if !expose.contains(&lower) {
expose.push(lower);
}
}
if req.response_mode.as_deref().is_some_and(|m| m != "artifact") {
return respond(&mut stream, 400, r#"{"error":"unsupported response_mode (expected artifact)"}"#);
}
if req.body.is_some() && req.body_ref.is_some() {
return respond(&mut stream, 400, r#"{"error":"body and body_ref are mutually exclusive"}"#);
}
if req.content_type.is_some() != req.body_ref.is_some() {
return respond(&mut stream, 400, r#"{"error":"content_type is required exactly when body_ref is set"}"#);
}
let mut guest_headers: Vec<(String, String)> =
req.headers.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
if let Some(mime) = &req.content_type {
if !areev_core::types::capability::is_valid_header_value(mime) || mime.is_empty()
|| req.headers.keys().any(|k| k.eq_ignore_ascii_case("content-type"))
{
return respond(&mut stream, 400, r#"{"error":"invalid or duplicate content_type"}"#);
}
guest_headers.push(("Content-Type".into(), mime.clone()));
}
let guest_header_names: Vec<String> = guest_headers.iter().map(|(k, _)| k.clone()).collect();
let sent_header_names: BTreeMap<String, String> = guest_headers.iter().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 requested_method = if req.method.trim().is_empty() { "GET" } else { req.method.trim() };
let requested_method = requested_method.to_ascii_uppercase();
let method = match (req.body_ref.is_some(), requested_method.as_str()) {
(true, "POST_ARTIFACT") => "POST",
(true, "PUT_ARTIFACT") => "PUT",
(true, "PATCH_ARTIFACT") => "PATCH",
(false, "POST_ARTIFACT" | "PUT_ARTIFACT" | "PATCH_ARTIFACT") => {
return respond(&mut stream, 400, r#"{"error":"artifact method requires body_ref"}"#)
}
(true, _) => return respond(&mut stream, 400, r#"{"error":"body_ref requires POST_ARTIFACT, PUT_ARTIFACT, or PATCH_ARTIFACT"}"#),
(false, _) => requested_method.as_str(),
}.to_string();
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 (req.body_ref.is_some() || req.response_mode.as_deref() == Some("artifact"))
&& artifact_store.lock().ok().and_then(|s| s.clone()).is_none()
{
return respond(&mut stream, 503, r#"{"error":"artifact mode requires a run-bound writable memory"}"#);
}
if req.body_ref.is_some() && !capability.as_ref().is_some_and(|(_, d)| d.declares_blob_read()) {
return respond(&mut stream, 403, &serde_json::json!({
"error": "body_ref requires a declared blob read capability", "code": refusal_code
}).to_string());
}
let artifact_mode = req.response_mode.as_deref() == Some("artifact");
let response_cap: Option<usize> = match (&capability, artifact_mode) {
(Some((l, _)), _) => Some(l.max_response_bytes),
(None, true) => Some(DEFAULT_TRANSFER),
(None, false) => None,
};
let request_cap = capability.as_ref().map_or(DEFAULT_TRANSFER, |(l, _)| l.max_request_bytes);
let out_of_range = |key: &str, n: usize| -> Option<String> {
(n == 0 || n > MAX_TRANSFER).then(|| {
areev_run_core::RunError::TransferLimitInvalid {
node: caller.clone(),
detail: format!(
"{key} is {n}, outside 1..={MAX_TRANSFER} — refused rather than clamped"
),
}
.to_string()
})
};
let invalid = if artifact_mode {
out_of_range("max_response_bytes", response_cap.unwrap_or(DEFAULT_TRANSFER))
} else {
None
}
.or_else(|| req.body_ref.as_ref().and_then(|_| out_of_range("max_request_bytes", request_cap)));
if let Some(detail) = invalid {
return respond(
&mut stream,
400,
&serde_json::json!({ "error": detail, "code": "RUN-E028" }).to_string(),
);
}
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 request_bytes = if let Some(uri) = &req.body_ref {
let store = artifact_store.lock().unwrap().clone().unwrap();
let sized = match store.with_store(|m| m.blob_len(uri)) {
Ok(n) if n > request_cap as u64 => Err(n),
Ok(_) => match store.with_store(|m| m.get_blob(uri)) {
Ok(bytes) if bytes.len() <= request_cap => Ok(bytes),
Ok(bytes) => Err(bytes.len() as u64),
Err(e) => return respond(&mut stream, 400, &serde_json::json!({"error": e.to_string()}).to_string()),
},
Err(e) => return respond(&mut stream, 400, &serde_json::json!({"error": e.to_string()}).to_string()),
};
match sized {
Ok(bytes) => {
note_blob_read(blob_reads, BlobRead { caller: caller.clone(), uri: uri.clone(), bytes: bytes.len() });
Some(bytes)
}
Err(size) => {
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: req.url.clone(),
reason: format!("artifact upload exceeds its {request_cap}-byte ceiling"),
},
);
return respond(
&mut stream,
413,
&serde_json::json!({
"error": format!(
"caller '{caller}' asked to upload a {size}-byte artifact, larger than \
its {request_cap}-byte max_request_bytes ceiling — refused before \
any byte was sent upstream"
),
"code": refusal_code
})
.to_string(),
);
}
}
} else { req.body.as_ref().map(|s| s.as_bytes().to_vec()) };
let mut outcome = dispatch(
&req.url,
&method,
request_bytes.as_deref(),
&headers,
&guest_headers,
policy,
grant,
capability.as_ref().map(|(_, d)| d),
response_cap,
artifact_mode,
&expose,
&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: request_bytes.as_deref().map(digest_bytes),
response_digest: if req.response_mode.as_deref() == Some("artifact") {
digest_bytes(body)
} else {
digest_bytes(scrub_reflected(
String::from_utf8(body.clone()).expect("text decoded at dispatch"),
&headers).as_bytes())
},
response_bytes: body.len(),
response_mime: None,
response_ref: None,
credential: if *credential_sent {
req.credential.clone()
} else {
None
},
headers: if *credential_sent {
sent_header_names.clone()
} else {
BTreeMap::new()
},
response_headers: 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,
request_bytes.as_deref(),
&refreshed,
&guest_headers,
policy,
grant,
capability.as_ref().map(|(_, d)| d),
response_cap,
artifact_mode,
&expose,
&caller,
refusals,
);
headers = refreshed;
}
}
}
}
match outcome {
Dispatched::Answered { status, body, mime, final_url, redirects, credential_sent, exposed } => {
if headers.iter().any(|(_, value)| value.len() >= 8 && mime.contains(value)) {
return respond(&mut stream, 502, r#"{"error":"upstream reflected a credential in media type"}"#);
}
let exposed: BTreeMap<String, String> = exposed
.into_iter()
.map(|(k, v)| (k, scrub_reflected(v, &headers)))
.collect();
let body = if req.response_mode.as_deref() == Some("artifact") {
if headers.iter().any(|(_, value)| value.len() >= 8 &&
(body.windows(value.len()).any(|w| w == value.as_bytes()) ||
value.strip_prefix("Bearer ").is_some_and(|bare| bare.len() >= 8 &&
body.windows(bare.len()).any(|w| w == bare.as_bytes())))) {
return respond(&mut stream, 502, r#"{"error":"upstream reflected a credential in binary response"}"#);
}
body
} else {
scrub_reflected(String::from_utf8(body).expect("text decoded at dispatch"), &headers).into_bytes()
};
let artifact_ref = if req.response_mode.as_deref() == Some("artifact") {
let store = artifact_store.lock().unwrap().clone().unwrap();
match store.with_store(|m| m.put_blob(&body)) {
Ok(uri) => Some(uri),
Err(e) => return respond(&mut stream, 502, &serde_json::json!({"error": format!("artifact storage failed: {e}")}).to_string()),
}
} else { None };
note_call(
calls,
EgressCall {
caller: caller.clone(),
method: method.clone(),
url: final_url,
status,
redirects,
request_digest: request_bytes.as_deref().map(digest_bytes),
response_digest: digest_bytes(&body),
response_bytes: body.len(),
response_mime: req.response_mode.as_deref().map(|_| mime.clone()),
response_ref: artifact_ref.clone(),
credential: if credential_sent { req.credential.clone() } else { None },
headers: if credential_sent {
sent_header_names.clone()
} else {
BTreeMap::new()
},
response_headers: exposed.clone(),
},
);
let mut answer = if req.response_mode.as_deref() == Some("artifact") {
serde_json::json!({ "status": status,
"ref": artifact_ref,
"sha256": digest_bytes(&body), "bytes": body.len(), "mime": mime })
} else {
serde_json::json!({ "status": status,
"body": String::from_utf8(body).expect("text decoded at dispatch") })
};
if !expose.is_empty() {
answer["headers"] = serde_json::json!(exposed);
}
respond(&mut stream, 200, &answer.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 = response_cap.unwrap_or(0);
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::NotText { final_url, detail } => {
note_refusal(
refusals,
EgressRefusal {
caller: caller.clone(),
destination: final_url,
reason: format!("response body is not UTF-8 text: {detail}"),
},
);
respond(
&mut stream,
502,
&serde_json::json!({
"error": format!(
"upstream: response body is not UTF-8 text ({detail}) — the \
broker carries text bodies only, and refuses rather than \
handing back an empty one with the upstream's status"
),
"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_bytes(body: &[u8]) -> String {
use sha2::{Digest, Sha256};
let mut h = Sha256::new();
h.update(body);
format!("sha256:{}", hex::encode(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: Vec<u8>,
mime: String,
final_url: String,
redirects: u32,
credential_sent: bool,
exposed: BTreeMap<String, String>,
},
Refused { detail: String },
TooLarge { final_url: String },
NotText { final_url: String, detail: String },
Upstream(String),
}
enum BodyErr {
TooLarge,
NotText(String),
Transport(String),
}
#[allow(clippy::too_many_arguments)]
fn dispatch(
start_url: &str,
start_method: &str,
body: Option<&[u8]>,
credential_headers: &[(String, String)],
guest_headers: &[(String, String)],
policy: &EgressPolicy,
grant: &CallerGrant,
declaration: Option<&Declaration>,
max_response_bytes: Option<usize>,
binary: bool,
expose: &[String],
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(<[u8]>::to_vec);
let mut left_origin = false;
let read_body = |resp: &mut ureq::http::Response<ureq::Body>| -> std::result::Result<Vec<u8>, BodyErr> {
let cap = max_response_bytes.unwrap_or(usize::MAX - 1);
let bytes = resp.body_mut().with_config().limit(cap as u64 + 1).read_to_vec()
.map_err(|e| match e {
ureq::Error::BodyExceedsLimit(_) => BodyErr::TooLarge,
other if binary => BodyErr::Transport(other.to_string()),
other => BodyErr::NotText(other.to_string()),
})?;
if bytes.len() > cap { return Err(BodyErr::TooLarge); }
if !binary {
std::str::from_utf8(&bytes).map_err(|e| BodyErr::NotText(e.to_string()))?;
}
Ok(bytes)
};
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 media = resp.headers().get("content-type")
.and_then(|v| v.to_str().ok()).unwrap_or("application/octet-stream")
.split(';').next().unwrap_or("").trim();
let mime = if media.len() <= 128 && media.contains('/') &&
media.bytes().all(|b| b.is_ascii_alphanumeric() || matches!(b, b'/' | b'-' | b'+' | b'.')) {
media.to_string()
} else {
"application/octet-stream".to_string()
};
let exposed: BTreeMap<String, String> = expose
.iter()
.filter_map(|name| {
let v = resp.headers().get(name.as_str())?.to_str().ok()?.trim();
(!v.is_empty()
&& v.len() <= MAX_EXPOSED_HEADER_VALUE
&& v.bytes().all(|b| (0x20..0x7f).contains(&b)))
.then(|| (name.clone(), v.to_string()))
})
.collect();
let Some(next_method) = redirect_method(status, &method) else {
return match read_body(&mut resp) {
Ok(text) => Dispatched::Answered {
status,
body: text,
mime,
final_url: url,
redirects: hops,
credential_sent: send_credential,
exposed: exposed.clone(),
},
Err(BodyErr::TooLarge) => Dispatched::TooLarge { final_url: url },
Err(BodyErr::NotText(detail)) => {
Dispatched::NotText { final_url: url, detail }
}
Err(BodyErr::Transport(detail)) => Dispatched::Upstream(format!(
"the response body was interrupted before it completed ({detail}) — \
refused rather than stored short"
)),
};
};
let Some(location) = location else {
return match read_body(&mut resp) {
Ok(text) => Dispatched::Answered {
status,
body: text,
mime,
final_url: url,
redirects: hops,
credential_sent: send_credential,
exposed: exposed.clone(),
},
Err(BodyErr::TooLarge) => Dispatched::TooLarge { final_url: url },
Err(BodyErr::NotText(detail)) => {
Dispatched::NotText { final_url: url, detail }
}
Err(BodyErr::Transport(detail)) => Dispatched::Upstream(format!(
"the response body was interrupted before it completed ({detail}) — \
refused rather than stored short"
)),
};
};
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<&[u8]>,
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(b""))
}
_ => {
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 fixture(expected: Vec<u8>, response: Vec<u8>, mime: &'static str) -> (String, std::thread::JoinHandle<()>) {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let url = format!("http://{}/artifact", listener.local_addr().unwrap());
let thread = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
stream.set_read_timeout(Some(std::time::Duration::from_secs(3))).unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
let mut length = 0;
loop {
line.clear();
reader.read_line(&mut line).unwrap();
if line == "\r\n" { break; }
if let Some(n) = line.to_ascii_lowercase().strip_prefix("content-length:") {
length = n.trim().parse().unwrap();
}
}
let mut sent = vec![0; length];
reader.read_exact(&mut sent).unwrap();
assert_eq!(sent, expected);
write!(stream, "HTTP/1.1 200 OK\r\nContent-Type: {mime}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", response.len()).unwrap();
let _ = stream.write_all(&response); });
(url, thread)
}
#[test]
fn binary_request_and_response_are_byte_exact_and_audited_without_bytes() {
for (bytes, mime) in [
(vec![0, 0x80, 0xff, 10], "application/octet-stream"),
(b"%PDF-1.7\n\0\x80\xff".to_vec(), "application/pdf"),
(b"PK\x03\x04\0\xff\x80".to_vec(), "application/zip"),
] {
let (url, server) = fixture(bytes.clone(), bytes.clone(), mime);
let origin = url.trim_end_matches("/artifact");
let broker = Broker::start(policy(&[origin]), BTreeMap::new(), grants(&[]), "RUN-E022").unwrap();
let dir = tempfile::tempdir().unwrap();
let store = Arc::new(AreevFacade::new(areev_store::Areev::open(
dir.path().join("m.db").to_str().unwrap()).unwrap()));
let request_ref = store.with_store(|m| m.put_blob(&bytes)).unwrap();
broker.bind_artifact_store(Arc::clone(&store));
broker.declare("", Declaration::parse(&serde_json::json!([
{"blob":{"read":true}},
{"http":{"hosts":[origin],"methods":["POST"],"headers":["Content-Type"]}}
])).unwrap(), CapabilityLimits::default());
let (status, answer) = call(&broker, serde_json::json!({
"url": url, "method": "POST_ARTIFACT", "response_mode": "artifact",
"body_ref": request_ref,
"content_type": mime
}));
assert_eq!(status, 200, "{answer}");
server.join().unwrap();
assert_eq!(answer["status"], 200);
assert_eq!(answer["mime"], mime);
assert_eq!(answer["bytes"], bytes.len());
assert_eq!(answer["sha256"], digest_bytes(&bytes));
assert_eq!(answer["ref"], request_ref);
assert_eq!(store.with_store(|m| m.get_blob(answer["ref"].as_str().unwrap())).unwrap(), bytes);
assert!(answer.get("body").is_none());
let calls = broker.calls();
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].request_digest.as_deref(), Some(digest_bytes(&bytes).as_str()));
assert_eq!(calls[0].response_digest, digest_bytes(&bytes));
assert_eq!(calls[0].response_bytes, bytes.len());
assert_eq!(calls[0].response_ref.as_deref(), Some(request_ref.as_str()));
store.with_store(|m| {
crate::journal::write_egress_call(m, "binary", &calls[0], 1_788_134_400_000,
"user:test", None).unwrap();
assert_eq!(m.gc_blobs().unwrap(), 0);
assert_eq!(m.get_blob(&request_ref).unwrap(), bytes);
});
}
}
#[test]
fn unsupported_modes_and_malformed_binary_requests_fail_before_dispatch() {
let broker = Broker::start(policy(&["http://127.0.0.1:1"]), BTreeMap::new(), grants(&[]), "RUN-E022").unwrap();
for req in [
serde_json::json!({"response_mode":"future"}),
serde_json::json!({"body_ref":"cas://sha256:abc"}),
serde_json::json!({"body":"x", "body_ref":"cas://sha256:abc", "content_type":"application/pdf"}),
serde_json::json!({"url":"http://127.0.0.1:1/x", "method":"POST_ARTIFACT"}),
serde_json::json!({"url":"http://127.0.0.1:1/x", "method":"POST", "body_ref":"cas://sha256:abc", "content_type":"application/pdf"}),
] {
let (status, answer) = call(&broker, req);
assert_eq!(status, 400, "{answer}");
}
assert!(broker.calls().is_empty());
for marker in ["POST_ARTIFACT", "PUT_ARTIFACT", "PATCH_ARTIFACT"] {
assert!(!matches!(marker, "GET" | "HEAD" | "DELETE" | "POST" | "PUT" | "PATCH"));
}
let (status, answer) = call(&broker, serde_json::json!({
"url":"http://127.0.0.1:1/artifact", "response_mode":"artifact"
}));
assert_eq!(status, 503, "{answer}");
}
#[test]
fn downloaded_artifact_uses_the_memorys_encryption_key() {
let bytes = b"SECRET-PDF-BYTES-\x00\x80\xff".to_vec();
let (url, server) = fixture(Vec::new(), bytes.clone(), "application/pdf");
let origin = url.trim_end_matches("/artifact");
let dir = tempfile::tempdir().unwrap();
let db = dir.path().join("encrypted.db");
let opts = areev_store::AreevOptions {
encryption_key: Some([42u8; 32]), ..Default::default()
};
let store = Arc::new(AreevFacade::new(areev_store::Areev::open_with(
db.to_str().unwrap(), opts).unwrap()));
let broker = Broker::start(policy(&[origin]), BTreeMap::new(), grants(&[]), "RUN-E022").unwrap();
broker.bind_artifact_store(Arc::clone(&store));
let (status, answer) = call(&broker, serde_json::json!({"url":url,"response_mode":"artifact"}));
assert_eq!(status, 200, "{answer}");
server.join().unwrap();
let uri = answer["ref"].as_str().unwrap();
assert_eq!(store.with_store(|m| m.get_blob(uri)).unwrap(), bytes);
let hex = uri.strip_prefix("cas://sha256:").unwrap();
let sidecar = db.with_file_name("encrypted.db.blobs")
.join(&hex[..2]).join(&hex[2..]);
let stored = std::fs::read(sidecar).unwrap();
assert!(!stored.windows(bytes.len()).any(|w| w == bytes));
}
#[test]
fn binary_overrun_is_a_typed_error_not_a_empty_success() {
let bytes = vec![0x80; MAX_BODY + 1];
let (url, server) = fixture(Vec::new(), bytes, "application/pdf");
let origin = url.trim_end_matches("/artifact");
let broker = Broker::start(policy(&[origin]), BTreeMap::new(), grants(&[]), "RUN-E022").unwrap();
bind_test_store(&broker);
let (status, answer) = call(&broker, serde_json::json!({"url":url,"response_mode":"artifact"}));
server.join().unwrap();
assert_eq!(status, 403, "{answer}");
assert_eq!(answer["code"], "RUN-E022");
assert!(broker.calls().is_empty());
}
#[test]
fn interrupted_binary_response_is_not_a_success() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let origin = format!("http://{}", listener.local_addr().unwrap());
let url = format!("{origin}/artifact");
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
loop {
line.clear();
reader.read_line(&mut line).unwrap();
if line == "\r\n" { break; }
}
stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 16\r\n\r\n\x00\x80").unwrap();
});
let broker = Broker::start(policy(&[&origin]), BTreeMap::new(), grants(&[]), "RUN-E022").unwrap();
bind_test_store(&broker);
let (status, answer) = call(&broker, serde_json::json!({"url":url,"response_mode":"artifact"}));
server.join().unwrap();
assert_eq!(status, 502, "{answer}");
assert!(answer["error"].as_str().unwrap().starts_with("upstream:"));
assert!(broker.calls().is_empty());
}
fn kept_store() -> (tempfile::TempDir, Arc<AreevFacade>) {
let dir = tempfile::tempdir().unwrap();
let store = Arc::new(AreevFacade::new(areev_store::Areev::open(
dir.path().join("m.db").to_str().unwrap()).unwrap()));
(dir, store)
}
#[test]
fn dropping_the_broker_releases_the_bound_store_synchronously() {
let (_dir, store) = kept_store();
let broker = Broker::start(policy(&["http://127.0.0.1:1"]), BTreeMap::new(), grants(&[]), "RUN-E022").unwrap();
broker.bind_artifact_store(Arc::clone(&store));
assert_eq!(Arc::strong_count(&store), 2, "the broker holds the store while it serves");
drop(broker);
assert_eq!(
Arc::strong_count(&store),
1,
"the store must be released by drop itself, not by the detached thread's exit"
);
}
fn upstream<F>(answer: F) -> (String, std::thread::JoinHandle<Vec<u8>>)
where
F: FnOnce(&mut TcpStream) + Send + 'static,
{
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let origin = format!("http://{}", listener.local_addr().unwrap());
let thread = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
stream.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap();
let mut reader = BufReader::new(stream.try_clone().unwrap());
let mut line = String::new();
let mut length = 0;
loop {
line.clear();
reader.read_line(&mut line).unwrap();
if line == "\r\n" || line.is_empty() { break; }
if let Some(n) = line.to_ascii_lowercase().strip_prefix("content-length:") {
length = n.trim().parse().unwrap();
}
}
let mut sent = vec![0; length];
reader.read_exact(&mut sent).unwrap();
answer(&mut stream);
sent
});
(origin, thread)
}
#[derive(Clone, Copy, Debug)]
enum Framing { Length, Chunked, CloseDelimited }
fn framed(body: Vec<u8>, framing: Framing) -> impl FnOnce(&mut TcpStream) + Send + 'static {
move |s: &mut TcpStream| {
let _ = match framing {
Framing::Length => write!(s,
"HTTP/1.1 200 OK\r\nContent-Type: application/pdf\r\nContent-Length: {}\r\n\
Connection: close\r\n\r\n", body.len())
.and_then(|_| s.write_all(&body)),
Framing::Chunked => write!(s,
"HTTP/1.1 200 OK\r\nContent-Type: application/pdf\r\n\
Transfer-Encoding: chunked\r\nConnection: close\r\n\r\n")
.and_then(|_| {
for chunk in body.chunks(1000) {
write!(s, "{:x}\r\n", chunk.len())?;
s.write_all(chunk)?;
s.write_all(b"\r\n")?;
}
s.write_all(b"0\r\n\r\n")
}),
Framing::CloseDelimited => write!(s,
"HTTP/1.1 200 OK\r\nContent-Type: application/pdf\r\nConnection: close\r\n\r\n")
.and_then(|_| s.write_all(&body)),
};
let _ = s.flush();
let _ = s.shutdown(std::net::Shutdown::Write);
}
}
fn artifact_broker(
origin: &str,
limits: CapabilityLimits,
store: &Arc<AreevFacade>,
credentials: BTreeMap<String, CredentialSource>,
) -> Broker {
let creds: Vec<String> = credentials.keys().cloned().collect();
let broker = Broker::start(
policy(&[origin]),
credentials,
grants(&creds.iter().map(String::as_str).collect::<Vec<_>>()),
"RUN-E022",
)
.unwrap();
broker.bind_artifact_store(Arc::clone(store));
broker.declare("", Declaration::parse(&serde_json::json!([
{"blob": {"read": true}},
{"http": {"hosts": [origin], "methods": ["GET", "POST"],
"headers": ["Content-Type"], "credentials": creds}}
])).unwrap(), limits);
broker
}
fn limits(response: usize, request: usize) -> CapabilityLimits {
CapabilityLimits { max_response_bytes: response, max_request_bytes: request, ..Default::default() }
}
fn document_fixture() -> Vec<u8> {
let mut bytes: Vec<u8> = (0..2 * 1024 * 1024 + 13u32)
.map(|i| 0x80 | (i.wrapping_mul(31) % 127) as u8)
.collect();
bytes[0] = 0xff;
let marker = b"RAW-DOCUMENT-MARKER-339";
bytes[64..64 + marker.len()].copy_from_slice(marker);
assert!(std::str::from_utf8(&bytes).is_err());
bytes
}
const SECRET: &str = "sk-live-339-never-journaled-value";
#[test]
fn a_25_mib_declaration_downloads_a_2_mib_document_exactly_and_journals_no_secret_or_byte() {
let bytes = document_fixture();
let (origin, server) = upstream(framed(bytes.clone(), Framing::Length));
let (_dir, store) = kept_store();
let creds: BTreeMap<String, CredentialSource> =
[("vendor".to_string(), Credential::Bearer(SECRET.into()).into())].into();
let broker = artifact_broker(&origin, limits(26_214_400, DEFAULT_TRANSFER), &store, creds);
let (status, answer) = call(&broker, serde_json::json!({
"url": format!("{origin}/statement.pdf"), "response_mode": "artifact",
"credential": "vendor"
}));
server.join().unwrap();
assert_eq!(status, 200, "{answer}");
assert_eq!(answer["bytes"], 2 * 1024 * 1024 + 13);
assert_eq!(answer["sha256"], digest_bytes(&bytes));
let uri = answer["ref"].as_str().unwrap().to_string();
let stored = store.with_store(|m| m.get_blob(&uri)).unwrap();
assert_eq!(stored.len(), bytes.len());
assert_eq!(digest_bytes(&stored), digest_bytes(&bytes));
assert_eq!(stored, bytes);
let calls = broker.calls();
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].credential.as_deref(), Some("vendor"));
let h = store.with_store(|m| {
crate::journal::write_egress_call(m, "r339", &calls[0], 1_788_134_400_000,
"user:test", None)
}).unwrap();
let grain = store.with_store(|m| m.get(&h)).unwrap();
let recorded = format!("{:?}", grain.fields);
assert!(!recorded.contains(SECRET), "credential value journaled: {recorded}");
assert!(!recorded.contains("RAW-DOCUMENT-MARKER-339"), "raw bytes journaled");
assert!(recorded.contains(&digest_bytes(&bytes)), "{recorded}");
assert!(recorded.len() < 4096, "the record is metadata, not the document");
assert!(format!("{calls:?}").len() < 4096);
}
#[test]
fn a_1_mib_declaration_refuses_the_same_document_naming_1048576() {
let bytes = document_fixture();
let (origin, server) = upstream(framed(bytes, Framing::Length));
let (_dir, store) = kept_store();
let broker = artifact_broker(&origin, limits(1_048_576, DEFAULT_TRANSFER), &store, BTreeMap::new());
let (status, answer) = call(&broker, serde_json::json!({
"url": format!("{origin}/statement.pdf"), "response_mode": "artifact"
}));
server.join().unwrap();
assert_eq!(status, 403, "{answer}");
assert_eq!(answer["code"], "RUN-E022");
let error = answer["error"].as_str().unwrap();
assert!(error.contains("1048576-byte"), "names the effective limit: {error}");
assert!(answer.get("ref").is_none() && answer.get("bytes").is_none(), "no partial success");
assert!(broker.calls().is_empty(), "an overrun is not a call that succeeded");
let refusals = broker.refusals();
assert_eq!(refusals.len(), 1);
assert!(refusals[0].reason.contains("1048576-byte"), "{:?}", refusals[0]);
let h = store.with_store(|m| {
crate::journal::write_egress_refusal(m, "r339", &refusals[0], 1_788_134_400_000,
"user:test", None)
}).unwrap();
let recorded = format!("{:?}", store.with_store(|m| m.get(&h)).unwrap().fields);
assert!(!recorded.contains("RAW-DOCUMENT-MARKER-339"));
}
#[test]
fn an_undeclared_artifact_keeps_the_1_mib_default_and_says_so() {
let (origin, server) = upstream(framed(vec![0x80; DEFAULT_TRANSFER + 1], Framing::Length));
let broker = Broker::start(policy(&[&origin]), BTreeMap::new(), grants(&[]), "RUN-E022").unwrap();
let (_dir, store) = kept_store();
broker.bind_artifact_store(store);
let (status, answer) = call(&broker, serde_json::json!({
"url": format!("{origin}/a"), "response_mode": "artifact"
}));
server.join().unwrap();
assert_eq!(status, 403, "{answer}");
assert!(answer["error"].as_str().unwrap().contains("1048576-byte"), "{answer}");
}
#[test]
fn the_ceiling_is_exact_at_the_limit_and_refuses_limit_plus_one_in_every_framing() {
const LIMIT: usize = 4096;
for framing in [Framing::Length, Framing::Chunked, Framing::CloseDelimited] {
for (size, admitted) in [(LIMIT, true), (LIMIT + 1, false)] {
let body: Vec<u8> = (0..size).map(|i| 0x80 | (i % 97) as u8).collect();
let (origin, server) = upstream(framed(body.clone(), framing));
let (_dir, store) = kept_store();
let broker = artifact_broker(&origin, limits(LIMIT, DEFAULT_TRANSFER), &store, BTreeMap::new());
let (status, answer) = call(&broker, serde_json::json!({
"url": format!("{origin}/x"), "response_mode": "artifact"
}));
server.join().unwrap();
if admitted {
assert_eq!(status, 200, "{framing:?} at the limit: {answer}");
assert_eq!(answer["bytes"], LIMIT);
let got = store.with_store(|m| m.get_blob(answer["ref"].as_str().unwrap())).unwrap();
assert_eq!(got, body, "{framing:?}");
} else {
assert_eq!(status, 403, "{framing:?} at limit + 1: {answer}");
assert_eq!(answer["code"], "RUN-E022");
assert!(answer["error"].as_str().unwrap().contains("4096-byte"), "{answer}");
assert!(broker.calls().is_empty(), "{framing:?}");
}
}
}
}
#[test]
fn an_interrupted_artifact_fails_rather_than_succeeding_short() {
let cases: [(&str, &[u8]); 2] = [
("length", b"HTTP/1.1 200 OK\r\nContent-Length: 5000\r\n\r\n"),
("chunked", b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n1000\r\n"),
];
for (name, head) in cases {
let head = head.to_vec();
let (origin, server) = upstream(move |s| {
let _ = s.write_all(&head);
let _ = s.write_all(&[0x80; 100]);
let _ = s.flush();
let _ = s.shutdown(std::net::Shutdown::Both);
});
let (_dir, store) = kept_store();
let broker = artifact_broker(&origin, limits(26_214_400, DEFAULT_TRANSFER), &store, BTreeMap::new());
let (status, answer) = call(&broker, serde_json::json!({
"url": format!("{origin}/x"), "response_mode": "artifact"
}));
server.join().unwrap();
assert_eq!(status, 502, "{name}: {answer}");
let error = answer["error"].as_str().unwrap();
assert!(error.contains("interrupted"), "{name}: {error}");
assert!(answer.get("ref").is_none(), "{name}");
assert!(broker.calls().is_empty(), "{name}");
}
}
#[test]
fn a_declared_16_mib_upload_is_sent_exactly() {
const SIXTEEN: usize = 16 * 1024 * 1024;
let bytes: Vec<u8> = (0..SIXTEEN).map(|i| 0x80 | (i.wrapping_mul(7) % 127) as u8).collect();
let (origin, server) = upstream(|s| {
let _ = s.write_all(b"HTTP/1.1 201 Created\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok");
});
let (_dir, store) = kept_store();
let request_ref = store.with_store(|m| m.put_blob(&bytes)).unwrap();
let broker = artifact_broker(&origin, limits(DEFAULT_TRANSFER, SIXTEEN), &store, BTreeMap::new());
let (status, answer) = call(&broker, serde_json::json!({
"url": format!("{origin}/upload"), "method": "POST_ARTIFACT",
"body_ref": request_ref, "content_type": "application/pdf"
}));
let received = server.join().unwrap();
assert_eq!(status, 200, "{answer}");
assert_eq!(answer["status"], 201);
assert_eq!(received.len(), SIXTEEN);
assert_eq!(digest_bytes(&received), digest_bytes(&bytes));
let calls = broker.calls();
assert_eq!(calls[0].request_digest.as_deref(), Some(digest_bytes(&bytes).as_str()));
assert_eq!(broker.blob_reads()[0].bytes, SIXTEEN);
}
#[test]
fn an_over_ceiling_upload_is_refused_before_the_upstream_sees_a_byte() {
for (declared, size) in [
(16 * 1024 * 1024, 16 * 1024 * 1024 + 1),
(DEFAULT_TRANSFER, DEFAULT_TRANSFER + 1),
] {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let origin = format!("http://{}", listener.local_addr().unwrap());
let (_dir, store) = kept_store();
let request_ref = store.with_store(|m| m.put_blob(&vec![0xfe; size])).unwrap();
let broker = artifact_broker(&origin, limits(DEFAULT_TRANSFER, declared), &store, BTreeMap::new());
let (status, answer) = call(&broker, serde_json::json!({
"url": format!("{origin}/upload"), "method": "POST_ARTIFACT",
"body_ref": request_ref, "content_type": "application/pdf"
}));
assert_eq!(status, 413, "{answer}");
assert_eq!(answer["code"], "RUN-E022");
let error = answer["error"].as_str().unwrap();
assert!(error.contains(&format!("{declared}-byte")), "names the effective limit: {error}");
assert!(
matches!(listener.accept(), Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock),
"the upstream was never connected to"
);
assert!(broker.calls().is_empty());
assert!(broker.blob_reads().is_empty(), "nothing was handed on");
assert_eq!(broker.refusals().len(), 1);
}
}
#[test]
fn an_oversized_upload_is_refused_by_size_before_its_bytes_are_loaded() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let origin = format!("http://{}", listener.local_addr().unwrap());
let dir = tempfile::tempdir().unwrap();
let db = dir.path().join("m.db");
let store = Arc::new(AreevFacade::new(areev_store::Areev::open(db.to_str().unwrap()).unwrap()));
let uri = store.with_store(|m| m.put_blob(b"%PDF-small")).unwrap();
let hex = uri.strip_prefix("cas://sha256:").unwrap();
let sidecar = db.with_file_name("m.db.blobs").join(&hex[..2]).join(&hex[2..]);
let gib: u64 = 1 << 30;
std::fs::OpenOptions::new().write(true).open(&sidecar).unwrap().set_len(gib).unwrap();
assert_eq!(store.with_store(|m| m.blob_len(&uri)).unwrap(), gib);
let broker = artifact_broker(&origin, limits(DEFAULT_TRANSFER, DEFAULT_TRANSFER), &store, BTreeMap::new());
let (status, answer) = call(&broker, serde_json::json!({
"url": format!("{origin}/upload"), "method": "POST_ARTIFACT",
"body_ref": uri, "content_type": "application/pdf"
}));
assert_eq!(status, 413, "refused by size, not by a failed read: {answer}");
let error = answer["error"].as_str().unwrap();
assert!(error.contains(&format!("{gib}-byte artifact")), "{error}");
assert!(error.contains("1048576-byte"), "{error}");
assert!(matches!(listener.accept(), Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock));
assert!(broker.blob_reads().is_empty());
}
#[test]
fn out_of_range_limits_are_refused_before_any_upstream_io_never_clamped() {
for (l, req) in [
(limits(0, DEFAULT_TRANSFER), serde_json::json!({"response_mode": "artifact"})),
(limits(MAX_TRANSFER + 1, DEFAULT_TRANSFER), serde_json::json!({"response_mode": "artifact"})),
(limits(DEFAULT_TRANSFER, MAX_TRANSFER + 1), serde_json::json!({
"method": "POST_ARTIFACT", "content_type": "application/pdf"})),
] {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let origin = format!("http://{}", listener.local_addr().unwrap());
let (_dir, store) = kept_store();
let r = store.with_store(|m| m.put_blob(b"%PDF")).unwrap();
let broker = artifact_broker(&origin, l, &store, BTreeMap::new());
let mut req = req;
req["url"] = serde_json::json!(format!("{origin}/x"));
if req.get("method").is_some() { req["body_ref"] = serde_json::json!(r); }
let (status, answer) = call(&broker, req);
assert_eq!(status, 400, "{answer}");
assert_eq!(answer["code"], "RUN-E028");
assert!(answer["error"].as_str().unwrap().starts_with("RUN-E028: "), "{answer}");
assert!(matches!(listener.accept(), Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock));
}
}
#[test]
fn text_mode_keeps_its_contract_under_a_large_declaration() {
let (origin, server) = upstream(framed(b"plain text".to_vec(), Framing::Length));
let (_dir, store) = kept_store();
let broker = artifact_broker(&origin, limits(26_214_400, DEFAULT_TRANSFER), &store, BTreeMap::new());
let (status, answer) = call(&broker, serde_json::json!({"url": format!("{origin}/t")}));
server.join().unwrap();
assert_eq!(status, 200, "{answer}");
assert_eq!(answer["body"], "plain text");
assert!(answer.get("ref").is_none());
}
fn bind_test_store(broker: &Broker) {
let dir = tempfile::tempdir().unwrap();
let store = Arc::new(AreevFacade::new(areev_store::Areev::open(
dir.path().join("m.db").to_str().unwrap()).unwrap()));
broker.bind_artifact_store(store);
}
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");
}
fn mcp_answer(session: &'static str) -> impl FnOnce(&mut TcpStream) + Send + 'static {
move |s: &mut TcpStream| {
let body = "data: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}\n\n";
let _ = write!(
s,
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nMcp-Session-Id: {session}\r\n\
Set-Cookie: sid=cookie-secret\r\nWWW-Authenticate: Bearer realm=x\r\n\
Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = s.flush();
}
}
#[test]
fn response_headers_are_absent_unless_named_so_old_answers_are_byte_identical() {
let (origin, server) = upstream(mcp_answer("sess-1"));
let b = Broker::start(policy(&[&origin]), BTreeMap::new(), grants(&[]), "RUN-E022").unwrap();
let (status, answer) =
call(&b, serde_json::json!({ "url": format!("{origin}/mcp"), "method": "POST", "body": "{}" }));
server.join().unwrap();
assert_eq!(status, 200, "{answer}");
let keys: Vec<&String> = answer.as_object().unwrap().keys().collect();
assert_eq!(keys, ["body", "status"], "the pre-#395 answer shape: {answer}");
assert!(b.calls()[0].response_headers.is_empty());
}
#[test]
fn named_response_headers_cross_back_and_are_journaled() {
let (origin, server) = upstream(mcp_answer("sess-1"));
let b = Broker::start(policy(&[&origin]), BTreeMap::new(), grants(&[]), "RUN-E022").unwrap();
let (status, answer) = call(
&b,
serde_json::json!({ "url": format!("{origin}/mcp"), "method": "POST", "body": "{}",
"response_headers": ["Mcp-Session-Id", "content-type", "MCP-SESSION-ID"] }),
);
server.join().unwrap();
assert_eq!(status, 200, "{answer}");
assert_eq!(
answer["headers"],
serde_json::json!({ "mcp-session-id": "sess-1", "content-type": "text/event-stream" }),
"exactly the names asked for, lowercased, deduplicated — never Set-Cookie"
);
let calls = b.calls();
assert_eq!(calls[0].response_headers.get("mcp-session-id").map(String::as_str), Some("sess-1"));
let (_dir, store) = kept_store();
store.with_store(|m| {
let h = crate::journal::write_egress_call(m, "r395", &calls[0], 1_788_134_400_000,
"user:test", None).unwrap();
let g = m.get(&h).unwrap();
assert!(format!("{g:?}").contains("sess-1"), "{g:?}");
});
}
#[test]
fn a_response_header_outside_the_closed_list_is_refused_before_any_io() {
let b = Broker::start(policy(&["http://127.0.0.1:1"]), BTreeMap::new(), grants(&[]), "RUN-E022").unwrap();
for name in ["Set-Cookie", "WWW-Authenticate", "Authorization", "Location", "X-Anything"] {
let (status, answer) = call(
&b,
serde_json::json!({ "url": "http://127.0.0.1:1/mcp", "response_headers": [name] }),
);
assert_eq!(status, 400, "{name}: {answer}");
assert!(answer["error"].as_str().unwrap().contains("mcp-session-id"), "{answer}");
}
assert!(b.calls().is_empty() && b.refusals().is_empty());
}
#[test]
fn an_exposed_header_reflecting_the_credential_is_scrubbed_like_the_body() {
let (origin, server) = upstream(mcp_answer("sk-live-395-reflected-token"));
let creds: BTreeMap<String, CredentialSource> = [(
"mcp".to_string(),
Credential::Bearer("sk-live-395-reflected-token".into()).into(),
)]
.into();
let b = Broker::start(policy(&[&origin]), creds, grants(&["mcp"]), "RUN-E022").unwrap();
let (status, answer) = call(
&b,
serde_json::json!({ "url": format!("{origin}/mcp"), "method": "POST", "body": "{}",
"credential": "mcp", "response_headers": ["Mcp-Session-Id"] }),
);
server.join().unwrap();
assert_eq!(status, 200, "{answer}");
assert_eq!(answer["headers"]["mcp-session-id"], "[redacted-credential]");
assert!(!format!("{:?}", b.calls()).contains("sk-live-395"), "nor in the audit record");
}
}