eric/geth
Watch
1
0
Fork
You've already forked geth
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fetch CAS blobs over Iroh

Browse source
This commit is contained in:
Eric Wendland 2026-05-18 17:18:25 +02:00
commit e4b788fec2
10 changed files with 425 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

281
crates/geth-node/src/lib.rs
View file

@ -1,5 +1,6 @@
pub mod service;
use base64::Engine;
use geth_auth::{AuthExplanation, AuthOp, AuthOpKind};
use geth_cas::{
BlobInfoSummary, FileConflict, FileConflictKind, FileConflictResolution, FileConflictStatus,
@ -36,8 +37,8 @@ use geth_store::{
StoredResourceSecret, StoredSshCertRequest, StoredSshCertificate, StoredSshRevocation,
};
use geth_types::{
AuthOpId, Capability, KeyId, NodeId, PrincipalId, ResourceId, ResourceKind, ResourceName,
SshCertId, SshCertRequestId, UnixMillis,
AuthOpId, BlobHash, Capability, KeyId, NodeId, PrincipalId, ResourceId, ResourceKind,
ResourceName, SshCertId, SshCertRequestId, UnixMillis,
};
use std::collections::{BTreeMap, VecDeque};
use std::path::Path;
@ -239,6 +240,10 @@ pub async fn handle_request_async(
resource,
capability,
} => peer_auth_check(node, &peer_node, resource, capability).await,
ControlRequest::CasFetch {
node: peer_node,
hash,
} => cas_fetch_from_peer(node, &peer_node, hash).await,
other => handle_request(node, other),
}
}
@ -512,6 +517,9 @@ async fn peer_ping(node: &LocalNode, peer_node: &str) -> Result<ControlResponse,
PeerControlResponse::AuthChecked { .. } => Err(NodeError::IrohPeer(
"peer returned auth-check response to ping request".to_owned(),
)),
PeerControlResponse::CasFetched { .. } => Err(NodeError::IrohPeer(
"peer returned CAS fetch response to ping request".to_owned(),
)),
}
}
@ -614,6 +622,142 @@ async fn peer_auth_check(
PeerControlResponse::Pong { .. } => Err(NodeError::IrohPeer(
"peer returned pong to auth-check request".to_owned(),
)),
PeerControlResponse::CasFetched { .. } => Err(NodeError::IrohPeer(
"peer returned CAS fetch response to auth-check request".to_owned(),
)),
}
}
async fn cas_fetch_from_peer(
node: &LocalNode,
peer_node: &str,
hash: BlobHash,
) -> Result<ControlResponse, NodeError> {
let store = Store::open(&node.paths.metadata_db())?;
let stored = store
.get_peer_card(peer_node)?
.ok_or_else(|| NodeError::PeerNotFound(peer_node.to_owned()))?;
let peer_card: PeerCard = serde_json::from_str(&stored.card_json)?;
peer_card.validate_candidate()?;
let candidate = peer_card
.endpoints
.first()
.ok_or(geth_discovery::DiscoveryError::MissingEndpoint)?;
ensure_peer_card_matches_endpoint(&peer_card, &candidate.endpoint_id)?;
let node_addr = iroh_node_addr_from_candidate(candidate)?;
let endpoint = node
.iroh_endpoint
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)?
.clone()
.ok_or(NodeError::IrohEndpointUnavailable)?;
let self_card = local_peer_card(node, DiscoverySource::PeerExchange, true).await?;
let nonce = geth_crypto::blake3_hex(
format!(
"{}\0{}\0{}\0{}",
node.node_id,
peer_node,
hash.as_str(),
geth_store::now_ms()
)
.as_bytes(),
);
let request = PeerControlRequest::CasFetch {
peer_card: self_card,
hash: hash.clone(),
nonce: nonce.clone(),
};
let conn = endpoint
.endpoint()
.connect(node_addr, geth_iroh::ALPN_CONTROL)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let (mut send, mut recv) = conn
.open_bi()
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
send.write_all(geth_control::encode_peer_request(&request)?.as_bytes())
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
send.finish()
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let bytes = recv
.read_to_end(64 * 1024 * 1024)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let text =
std::str::from_utf8(&bytes).map_err(|error| NodeError::IrohPeer(error.to_string()))?;
match geth_control::decode_peer_response(text)? {
PeerControlResponse::CasFetched {
node_id,
agent_id,
endpoint_id,
hash: response_hash,
size_bytes,
content_base64,
allowed,
reason,
nonce: response_nonce,
note,
..
} if response_nonce == nonce && response_hash == hash => {
if !allowed {
return Ok(ControlResponse::CasFetched {
peer_node_id: node_id,
peer_agent_id: agent_id,
endpoint_id,
hash: response_hash,
size_bytes: 0,
allowed,
reason,
note,
});
}
let content = content_base64
.ok_or_else(|| NodeError::IrohPeer("peer omitted CAS content".to_owned()))
.and_then(|content| {
base64::engine::general_purpose::STANDARD
.decode(content)
.map_err(|error| NodeError::IrohPeer(error.to_string()))
})?;
if content.len() as u64 != size_bytes {
return Err(NodeError::IrohPeer(format!(
"peer announced {size_bytes} CAS bytes but returned {} bytes",
content.len()
)));
}
let cas = LocalCas::new(node.paths.cas_dir());
let info = cas.add_bytes(&content)?;
if info.hash != response_hash {
return Err(NodeError::IrohPeer(format!(
"peer returned content hash {} for requested {}",
info.hash, response_hash
)));
}
store.record_cas_object(
info.hash.as_str(),
info.size_bytes,
&info.path.to_string_lossy(),
)?;
Ok(ControlResponse::CasFetched {
peer_node_id: node_id,
peer_agent_id: agent_id,
endpoint_id,
hash: info.hash,
size_bytes: info.size_bytes,
allowed,
reason,
note,
})
}
PeerControlResponse::CasFetched { .. } => Err(NodeError::IrohPeer(
"peer CAS fetch response did not match request".to_owned(),
)),
PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
PeerControlResponse::Pong { .. } | PeerControlResponse::AuthChecked { .. } => Err(
NodeError::IrohPeer("peer returned wrong response type to CAS fetch".to_owned()),
),
}
}
@ -720,6 +864,74 @@ async fn handle_iroh_control_connection(
note: "protected peer request authenticated endpoint/card binding before resource capability evaluation".to_owned(),
}
}
PeerControlRequest::CasFetch {
peer_card,
hash,
nonce,
} => {
peer_card.validate_candidate()?;
ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
let discovered = DiscoveredPeer::candidate(
peer_card.clone(),
UnixMillis(geth_store::now_ms()),
DiscoverySource::PeerExchange,
)?;
let store = Store::open(&node.paths.metadata_db())?;
store.upsert_peer_card(&StoredPeerCard {
peer_id: peer_card.node_id.to_string(),
card_json: serde_json::to_string(&peer_card)?,
updated_at_ms: discovered.discovered_at.0,
})?;
let resource = "resource:cas:local".to_owned();
let capability = "cas.fetch".to_owned();
let explanation = geth_auth::explain_auth_ops(
&load_auth_ops_for_resource(&store, &resource)?,
PrincipalId::new(peer_card.node_id.to_string()),
ResourceId::new(resource),
Capability::new(capability),
);
if explanation.allowed {
match LocalCas::new(node.paths.cas_dir()).read_bytes(&hash) {
Ok(content) => {
let size_bytes = content.len() as u64;
PeerControlResponse::CasFetched {
node_id: node.node_id.clone(),
agent_id: node.agent_id.clone(),
endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
remote_endpoint_id,
hash,
size_bytes,
content_base64: Some(
base64::engine::general_purpose::STANDARD.encode(content),
),
allowed: true,
reason: explanation.reason,
evaluated_ops: explanation.evaluated_ops,
nonce,
note: "CAS fetch authenticated endpoint/card binding and required cas.fetch on resource:cas:local".to_owned(),
}
}
Err(error) => PeerControlResponse::Error {
message: format!("CAS blob {hash} is not available: {error}"),
},
}
} else {
PeerControlResponse::CasFetched {
node_id: node.node_id.clone(),
agent_id: node.agent_id.clone(),
endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
remote_endpoint_id,
hash,
size_bytes: 0,
content_base64: None,
allowed: false,
reason: explanation.reason,
evaluated_ops: explanation.evaluated_ops,
nonce,
note: "CAS fetch authenticated endpoint/card binding and required cas.fetch on resource:cas:local".to_owned(),
}
}
}
};
send.write_all(geth_control::encode_peer_response(&response)?.as_bytes())
.await
@ -855,6 +1067,7 @@ pub fn handle_request(
}
ControlRequest::PeerPing { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::PeerAuthCheck { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::CasFetch { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::ResourceList => Ok(ControlResponse::ResourceList {
resources: store
.list_resources()?
@ -2330,6 +2543,9 @@ mod tests {
updated_at_ms: geth_store::now_ms(),
})
.expect("insert right peer");
let right_blob = LocalCas::new(right.paths.cas_dir())
.add_bytes(b"remote cas bytes")
.expect("right cas add");
let ping = handle_request_async(
&left,
@ -2380,6 +2596,34 @@ mod tests {
other => panic!("unexpected denied auth response: {other:?}"),
}
let denied_fetch = handle_request_async(
&left,
ControlRequest::CasFetch {
node: right_card.node_id.to_string(),
hash: right_blob.hash.clone(),
},
)
.await
.expect("denied cas fetch");
match denied_fetch {
ControlResponse::CasFetched {
allowed,
reason,
size_bytes,
..
} => {
assert!(!allowed);
assert_eq!(size_bytes, 0);
assert!(reason.contains("no active direct or group grant"));
assert!(
!LocalCas::new(left.paths.cas_dir())
.has(&right_blob.hash)
.expect("left cas has after denied fetch")
);
}
other => panic!("unexpected denied CAS fetch response: {other:?}"),
}
handle_request(
&right,
ControlRequest::AuthGrant {
@ -2415,6 +2659,39 @@ mod tests {
other => panic!("unexpected allowed auth response: {other:?}"),
}
let fetched = handle_request_async(
&left,
ControlRequest::CasFetch {
node: right_card.node_id.to_string(),
hash: right_blob.hash.clone(),
},
)
.await
.expect("allowed cas fetch");
match fetched {
ControlResponse::CasFetched {
allowed,
hash,
size_bytes,
reason,
note,
..
} => {
assert!(allowed);
assert_eq!(hash, right_blob.hash);
assert_eq!(size_bytes, right_blob.size_bytes);
assert!(reason.contains("direct grant"));
assert!(note.contains("required cas.fetch"));
assert_eq!(
LocalCas::new(left.paths.cas_dir())
.read_bytes(&right_blob.hash)
.expect("left cas read fetched blob"),
b"remote cas bytes"
);
}
other => panic!("unexpected allowed CAS fetch response: {other:?}"),
}
left_endpoint.shutdown().await;
right_endpoint.shutdown().await;
}