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

Authorize remote modules with bearer proofs

Browse source
This commit is contained in:
Eric Wendland 2026-05-19 19:16:30 +02:00
commit 7e39a19c18
8 changed files with 572 additions and 120 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -255,42 +255,52 @@ pub async fn handle_request_async(
ControlRequest::CasFetch {
node: peer_node,
hash,
} => cas_fetch_from_peer(node, &peer_node, hash).await,
ControlRequest::SshCertSync { node: peer_node } => {
ssh_cert_sync_from_peer(node, &peer_node).await
}
ControlRequest::SshRevocationSync { node: peer_node } => {
ssh_revocation_sync_from_peer(node, &peer_node).await
}
bearer_secret,
} => cas_fetch_from_peer(node, &peer_node, hash, bearer_secret).await,
ControlRequest::SshCertSync {
node: peer_node,
bearer_secret,
} => ssh_cert_sync_from_peer(node, &peer_node, bearer_secret).await,
ControlRequest::SshRevocationSync {
node: peer_node,
bearer_secret,
} => ssh_revocation_sync_from_peer(node, &peer_node, bearer_secret).await,
ControlRequest::KvSync {
node: peer_node,
name,
} => kv_sync_from_peer(node, &peer_node, &name).await,
bearer_secret,
} => kv_sync_from_peer(node, &peer_node, &name, bearer_secret).await,
ControlRequest::DbSync {
node: peer_node,
name,
limit,
} => db_sync_from_peer(node, &peer_node, &name, limit).await,
bearer_secret,
} => db_sync_from_peer(node, &peer_node, &name, limit, bearer_secret).await,
ControlRequest::PubsubPub {
topic,
message,
node: Some(peer_node),
} => pubsub_publish_to_peer(node, &peer_node, topic, message).await,
bearer_secret,
} => pubsub_publish_to_peer(node, &peer_node, topic, message, bearer_secret).await,
ControlRequest::PubsubSub {
topic,
node: Some(peer_node),
} => pubsub_subscribe_from_peer(node, &peer_node, topic).await,
bearer_secret,
} => pubsub_subscribe_from_peer(node, &peer_node, topic, bearer_secret).await,
ControlRequest::PipeConnect {
target,
node: Some(peer_node),
} => pipe_connect_to_peer(node, &peer_node, target).await,
ControlRequest::SshProxyConnect { node: peer_node } => {
ssh_proxy_connect_to_peer(node, &peer_node).await
}
bearer_secret,
} => pipe_connect_to_peer(node, &peer_node, target, bearer_secret).await,
ControlRequest::SshProxyConnect {
node: peer_node,
bearer_secret,
} => ssh_proxy_connect_to_peer(node, &peer_node, bearer_secret).await,
ControlRequest::DocumentSync {
node: peer_node,
name,
} => document_sync_from_peer(node, &peer_node, &name).await,
bearer_secret,
} => document_sync_from_peer(node, &peer_node, &name, bearer_secret).await,
other => handle_request(node, other),
}
}
@ -699,6 +709,7 @@ async fn cas_fetch_from_peer(
node: &LocalNode,
peer_node: &str,
hash: BlobHash,
bearer_secret: Option<String>,
) -> Result<ControlResponse, NodeError> {
let store = Store::open(&node.paths.metadata_db())?;
let stored = store
@ -733,6 +744,7 @@ async fn cas_fetch_from_peer(
peer_card: self_card,
hash: hash.clone(),
nonce: nonce.clone(),
bearer_proof: bearer_proof(bearer_secret, "resource:cas:local", "cas.fetch", &nonce),
};
let conn = endpoint
@ -843,6 +855,7 @@ async fn cas_fetch_from_peer(
async fn ssh_cert_sync_from_peer(
node: &LocalNode,
peer_node: &str,
bearer_secret: Option<String>,
) -> Result<ControlResponse, NodeError> {
let since_ms = load_live_sync_cursor(
&Store::open(&node.paths.metadata_db())?,
@ -853,7 +866,13 @@ async fn ssh_cert_sync_from_peer(
PeerControlRequest::SshCertSync {
peer_card,
since_ms,
nonce,
nonce: nonce.clone(),
bearer_proof: bearer_proof(
bearer_secret,
"resource:ssh:certs",
"ssh_cert.sync",
&nonce,
),
}
})
.await?;
@ -913,6 +932,7 @@ async fn ssh_cert_sync_from_peer(
async fn ssh_revocation_sync_from_peer(
node: &LocalNode,
peer_node: &str,
bearer_secret: Option<String>,
) -> Result<ControlResponse, NodeError> {
let since_ms = load_live_sync_cursor(
&Store::open(&node.paths.metadata_db())?,
@ -926,7 +946,13 @@ async fn ssh_revocation_sync_from_peer(
|peer_card, nonce| PeerControlRequest::SshRevocationSync {
peer_card,
since_ms,
nonce,
nonce: nonce.clone(),
bearer_proof: bearer_proof(
bearer_secret,
"resource:ssh:revocations",
"ssh_revocation.sync",
&nonce,
),
},
)
.await?;
@ -980,6 +1006,7 @@ async fn kv_sync_from_peer(
node: &LocalNode,
peer_node: &str,
name: &str,
bearer_secret: Option<String>,
) -> Result<ControlResponse, NodeError> {
geth_kv::validate_kv_name(name).map_err(|_| NodeError::InvalidKvName(name.to_owned()))?;
let stream = format!("kv:{name}");
@ -990,7 +1017,13 @@ async fn kv_sync_from_peer(
peer_card,
name: name.to_owned(),
since_ms,
nonce,
nonce: nonce.clone(),
bearer_proof: bearer_proof(
bearer_secret,
&format!("resource:kv:{name}"),
"kv.read",
&nonce,
),
}
})
.await?;
@ -1065,6 +1098,7 @@ async fn pubsub_publish_to_peer(
peer_node: &str,
topic: String,
message: String,
bearer_secret: Option<String>,
) -> Result<ControlResponse, NodeError> {
geth_pubsub::validate_topic(&topic)?;
geth_pubsub::validate_message(&message)?;
@ -1073,7 +1107,13 @@ async fn pubsub_publish_to_peer(
peer_card,
topic: topic.clone(),
message: message.clone(),
nonce,
nonce: nonce.clone(),
bearer_proof: bearer_proof(
bearer_secret,
&format!("resource:pubsub:{topic}"),
"pubsub.publish",
&nonce,
),
}
})
.await?;
@ -1114,13 +1154,20 @@ async fn pubsub_subscribe_from_peer(
node: &LocalNode,
peer_node: &str,
topic: String,
bearer_secret: Option<String>,
) -> Result<ControlResponse, NodeError> {
geth_pubsub::validate_topic(&topic)?;
let response = request_peer_control(node, peer_node, "pubsub-subscribe", |peer_card, nonce| {
PeerControlRequest::PubsubSubscribe {
peer_card,
topic: topic.clone(),
nonce,
nonce: nonce.clone(),
bearer_proof: bearer_proof(
bearer_secret,
&format!("resource:pubsub:{topic}"),
"pubsub.subscribe",
&nonce,
),
}
})
.await?;
@ -1159,13 +1206,20 @@ async fn pipe_connect_to_peer(
node: &LocalNode,
peer_node: &str,
target: String,
bearer_secret: Option<String>,
) -> Result<ControlResponse, NodeError> {
geth_pipe::validate_pipe_name(&target)?;
let response = request_peer_control(node, peer_node, "pipe-connect", |peer_card, nonce| {
PeerControlRequest::PipeConnect {
peer_card,
target: target.clone(),
nonce,
nonce: nonce.clone(),
bearer_proof: bearer_proof(
bearer_secret,
&format!("resource:pipe:{target}"),
"pipe.connect",
&nonce,
),
}
})
.await?;
@ -1206,10 +1260,20 @@ async fn pipe_connect_to_peer(
async fn ssh_proxy_connect_to_peer(
node: &LocalNode,
peer_node: &str,
bearer_secret: Option<String>,
) -> Result<ControlResponse, NodeError> {
let response =
request_peer_control(node, peer_node, "ssh-proxy-connect", |peer_card, nonce| {
PeerControlRequest::SshProxyConnect { peer_card, nonce }
PeerControlRequest::SshProxyConnect {
peer_card,
nonce: nonce.clone(),
bearer_proof: bearer_proof(
bearer_secret,
"resource:ssh-proxy:local",
"ssh_proxy.connect",
&nonce,
),
}
})
.await?;
match response {
@ -1242,6 +1306,7 @@ async fn document_sync_from_peer(
node: &LocalNode,
peer_node: &str,
name: &str,
bearer_secret: Option<String>,
) -> Result<ControlResponse, NodeError> {
geth_document::validate_document_name(name)
.map_err(|_| NodeError::InvalidDocumentName(name.to_owned()))?;
@ -1253,7 +1318,13 @@ async fn document_sync_from_peer(
peer_card,
name: name.to_owned(),
since_ms,
nonce,
nonce: nonce.clone(),
bearer_proof: bearer_proof(
bearer_secret,
&format!("resource:document:{name}"),
"document.read",
&nonce,
),
}
})
.await?;
@ -1347,6 +1418,7 @@ async fn db_sync_from_peer(
peer_node: &str,
name: &str,
limit: u32,
bearer_secret: Option<String>,
) -> Result<ControlResponse, NodeError> {
geth_db::validate_db_name(name).map_err(|_| NodeError::InvalidDbName(name.to_owned()))?;
let store = Store::open(&node.paths.metadata_db())?;
@ -1362,7 +1434,13 @@ async fn db_sync_from_peer(
name: name.to_owned(),
after_db_version,
limit,
nonce,
nonce: nonce.clone(),
bearer_proof: bearer_proof(
bearer_secret,
&format!("resource:db:{name}"),
"db.sync",
&nonce,
),
}
})
.await?;
@ -1585,6 +1663,123 @@ fn can_sync_resource(
.allowed)
}
fn bearer_proof(
bearer_secret: Option<String>,
resource: &str,
capability: &str,
nonce: &str,
) -> Option<BearerProof> {
bearer_secret.map(|secret| {
let secret = geth_types::SecretId::new(secret);
let resource = ResourceId::new(resource.to_owned());
let capabilities = vec![Capability::new(capability.to_owned())];
let response = geth_secrets::bearer_response(&secret, &resource, &capabilities, nonce);
BearerProof {
secret,
resource,
capabilities,
nonce: nonce.to_owned(),
response,
}
})
}
fn explain_peer_or_bearer(
store: &Store,
peer_node: &str,
resource: &str,
capability: &str,
nonce: &str,
bearer_proof: Option<&BearerProof>,
) -> Result<AuthExplanation, NodeError> {
let ops = load_auth_ops_for_resource(store, resource)?;
let resource_id = ResourceId::new(resource.to_owned());
let capability_id = Capability::new(capability.to_owned());
let peer_explanation = geth_auth::explain_auth_ops(
&ops,
PrincipalId::new(peer_node.to_owned()),
resource_id.clone(),
capability_id.clone(),
);
if peer_explanation.allowed {
return Ok(peer_explanation);
}
let Some(proof) = bearer_proof else {
return Ok(peer_explanation);
};
let bearer_subject = format!("bearer:{}", proof.secret);
let denied = |reason: String| AuthExplanation {
subject: bearer_subject.clone(),
resource: resource.to_owned(),
capability: capability.to_owned(),
allowed: false,
reason,
evaluated_ops: peer_explanation.evaluated_ops,
};
if proof.resource != resource_id {
return Ok(denied(format!(
"bearer proof resource {} does not match requested resource {resource}",
proof.resource
)));
}
if proof.nonce != nonce {
return Ok(denied(
"bearer proof nonce does not match request".to_owned(),
));
}
if !proof
.capabilities
.iter()
.any(|granted| geth_auth::capability_allows(granted, &capability_id))
{
return Ok(denied(
"bearer proof does not include the requested capability".to_owned(),
));
}
if !geth_secrets::verify_bearer_response(
&proof.secret,
&proof.resource,
&proof.capabilities,
nonce,
&proof.response,
) {
return Ok(denied("bearer proof response is invalid".to_owned()));
}
let now = UnixMillis(geth_store::now_ms());
let Some(access) = load_bearer_access(store)?.into_iter().find(|access| {
access.secret == proof.secret
&& access.resource == proof.resource
&& access.expires_at.is_none_or(|expires| expires.0 > now.0)
}) else {
return Ok(denied(
"bearer access is not active for requested resource".to_owned(),
));
};
if !access
.capabilities
.iter()
.any(|granted| geth_auth::capability_allows(granted, &capability_id))
{
return Ok(denied(
"bearer access lacks requested capability".to_owned(),
));
}
Ok(AuthExplanation {
subject: bearer_subject,
resource: resource.to_owned(),
capability: capability.to_owned(),
allowed: true,
reason:
"bearer proof allows this resource-scoped capability without granting node identity"
.to_owned(),
evaluated_ops: peer_explanation.evaluated_ops,
})
}
async fn request_peer_control(
node: &LocalNode,
peer_node: &str,
@ -1754,7 +1949,7 @@ async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError> {
"ssh-certs",
remote_watermarks.as_ref(),
)? {
if let Err(error) = ssh_cert_sync_from_peer(node, &peer.peer_id).await {
if let Err(error) = ssh_cert_sync_from_peer(node, &peer.peer_id, None).await {
tracing::debug!(peer = %peer.peer_id, %error, "SSH cert live sync failed");
}
}
@ -1764,7 +1959,7 @@ async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError> {
"ssh-revocations",
remote_watermarks.as_ref(),
)? {
if let Err(error) = ssh_revocation_sync_from_peer(node, &peer.peer_id).await {
if let Err(error) = ssh_revocation_sync_from_peer(node, &peer.peer_id, None).await {
tracing::debug!(peer = %peer.peer_id, %error, "SSH revocation live sync failed");
}
}
@ -1772,7 +1967,7 @@ async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError> {
let stream = format!("kv:{}", kv.name);
if should_live_sync_stream(&store, &peer.peer_id, &stream, remote_watermarks.as_ref())?
{
if let Err(error) = kv_sync_from_peer(node, &peer.peer_id, &kv.name).await {
if let Err(error) = kv_sync_from_peer(node, &peer.peer_id, &kv.name, None).await {
tracing::debug!(peer = %peer.peer_id, kv = %kv.name, %error, "KV live sync failed");
}
}
@ -1782,7 +1977,7 @@ async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError> {
if should_live_sync_stream(&store, &peer.peer_id, &stream, remote_watermarks.as_ref())?
{
if let Err(error) =
document_sync_from_peer(node, &peer.peer_id, &document.name).await
document_sync_from_peer(node, &peer.peer_id, &document.name, None).await
{
tracing::debug!(peer = %peer.peer_id, document = %document.name, %error, "document live sync failed");
}
@ -1792,7 +1987,9 @@ async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError> {
let stream = format!("db:{}", db.name);
if should_live_sync_stream(&store, &peer.peer_id, &stream, remote_watermarks.as_ref())?
{
if let Err(error) = db_sync_from_peer(node, &peer.peer_id, &db.name, 100).await {
if let Err(error) =
db_sync_from_peer(node, &peer.peer_id, &db.name, 100, None).await
{
tracing::debug!(peer = %peer.peer_id, db = %db.name, %error, "DB live sync failed");
}
}
@ -1919,6 +2116,7 @@ async fn handle_iroh_control_connection(
peer_card,
hash,
nonce,
bearer_proof,
} => {
peer_card.validate_candidate()?;
ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
@ -1935,12 +2133,14 @@ async fn handle_iroh_control_connection(
})?;
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),
);
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&resource,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
if explanation.allowed {
match LocalCas::new(node.paths.cas_dir()).read_bytes(&hash) {
Ok(content) => {
@ -1987,6 +2187,7 @@ async fn handle_iroh_control_connection(
peer_card,
since_ms,
nonce,
bearer_proof,
} => {
peer_card.validate_candidate()?;
ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
@ -2004,12 +2205,14 @@ async fn handle_iroh_control_connection(
let resource = "resource:ssh:certs".to_owned();
let capability = "ssh_cert.sync".to_owned();
let high_water_ms = geth_store::now_ms();
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),
);
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&resource,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
let (requests, certificates) = if explanation.allowed {
(
store
@ -2045,6 +2248,7 @@ async fn handle_iroh_control_connection(
peer_card,
since_ms,
nonce,
bearer_proof,
} => {
peer_card.validate_candidate()?;
ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
@ -2062,12 +2266,14 @@ async fn handle_iroh_control_connection(
let resource = "resource:ssh:revocations".to_owned();
let capability = "ssh_revocation.sync".to_owned();
let high_water_ms = geth_store::now_ms();
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),
);
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&resource,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
let revocations = if explanation.allowed {
store
.list_ssh_revocations_since(since_ms)?
@ -2096,6 +2302,7 @@ async fn handle_iroh_control_connection(
name,
since_ms,
nonce,
bearer_proof,
} => {
geth_kv::validate_kv_name(&name).map_err(|_| NodeError::InvalidKvName(name.clone()))?;
peer_card.validate_candidate()?;
@ -2114,12 +2321,14 @@ async fn handle_iroh_control_connection(
if let Some(kv) = store.get_kv_store_by_name(&name)? {
let capability = "kv.read".to_owned();
let high_water_ms = geth_store::now_ms();
let explanation = geth_auth::explain_auth_ops(
&load_auth_ops_for_resource(&store, &kv.resource_id)?,
PrincipalId::new(peer_card.node_id.to_string()),
ResourceId::new(kv.resource_id.clone()),
Capability::new(capability),
);
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&kv.resource_id,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
let entries = if explanation.allowed {
store
.list_kv_entries_since(&kv.kv_id, since_ms)?
@ -2158,6 +2367,7 @@ async fn handle_iroh_control_connection(
topic,
message,
nonce,
bearer_proof,
} => {
geth_pubsub::validate_topic(&topic)?;
geth_pubsub::validate_message(&message)?;
@ -2176,12 +2386,14 @@ async fn handle_iroh_control_connection(
})?;
let resource = format!("resource:pubsub:{topic}");
let capability = "pubsub.publish".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),
);
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&resource,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
let published = if explanation.allowed {
Some(record_pubsub_message(&node, topic, message)?)
} else {
@ -2204,6 +2416,7 @@ async fn handle_iroh_control_connection(
peer_card,
topic,
nonce,
bearer_proof,
} => {
geth_pubsub::validate_topic(&topic)?;
peer_card.validate_candidate()?;
@ -2221,12 +2434,14 @@ async fn handle_iroh_control_connection(
})?;
let resource = format!("resource:pubsub:{topic}");
let capability = "pubsub.subscribe".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),
);
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&resource,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
let messages = if explanation.allowed {
pubsub_messages_for_topic(&node, &topic)?
} else {
@ -2250,6 +2465,7 @@ async fn handle_iroh_control_connection(
peer_card,
target,
nonce,
bearer_proof,
} => {
geth_pipe::validate_pipe_name(&target)?;
peer_card.validate_candidate()?;
@ -2267,12 +2483,14 @@ async fn handle_iroh_control_connection(
})?;
let resource = format!("resource:pipe:{target}");
let capability = "pipe.connect".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),
);
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&resource,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
let connection = if explanation.allowed {
Some(record_pipe_connection(
&node,
@ -2295,7 +2513,11 @@ async fn handle_iroh_control_connection(
note: "pipe connect authenticated endpoint/card binding and required pipe.connect on the remote pipe resource; byte streams are not implemented yet".to_owned(),
}
}
PeerControlRequest::SshProxyConnect { peer_card, nonce } => {
PeerControlRequest::SshProxyConnect {
peer_card,
nonce,
bearer_proof,
} => {
peer_card.validate_candidate()?;
ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?;
let discovered = DiscoveredPeer::candidate(
@ -2311,12 +2533,14 @@ async fn handle_iroh_control_connection(
})?;
let resource = "resource:ssh-proxy:local".to_owned();
let capability = "ssh_proxy.connect".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),
);
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&resource,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
let connection = if explanation.allowed {
Some(SshProxyConnection {
target_node: NodeId::new(node.node_id.clone()),
@ -2346,6 +2570,7 @@ async fn handle_iroh_control_connection(
name,
since_ms,
nonce,
bearer_proof,
} => {
geth_document::validate_document_name(&name)
.map_err(|_| NodeError::InvalidDocumentName(name.clone()))?;
@ -2365,12 +2590,14 @@ async fn handle_iroh_control_connection(
if let Some(document) = store.get_document_resource_by_name(&name)? {
let capability = "document.read".to_owned();
let high_water_ms = geth_store::now_ms();
let explanation = geth_auth::explain_auth_ops(
&load_auth_ops_for_resource(&store, &document.resource_id)?,
PrincipalId::new(peer_card.node_id.to_string()),
ResourceId::new(document.resource_id.clone()),
Capability::new(capability),
);
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&document.resource_id,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
let state = if explanation.allowed && document.updated_at_ms >= since_ms {
Some(document_state_from_stored(&document))
} else {
@ -2402,6 +2629,7 @@ async fn handle_iroh_control_connection(
after_db_version,
limit,
nonce,
bearer_proof,
} => {
geth_db::validate_db_name(&name).map_err(|_| NodeError::InvalidDbName(name.clone()))?;
peer_card.validate_candidate()?;
@ -2419,12 +2647,14 @@ async fn handle_iroh_control_connection(
})?;
if let Some(db) = store.get_db_resource_by_name(&name)? {
let capability = "db.sync".to_owned();
let explanation = geth_auth::explain_auth_ops(
&load_auth_ops_for_resource(&store, &db.resource_id)?,
PrincipalId::new(peer_card.node_id.to_string()),
ResourceId::new(db.resource_id.clone()),
Capability::new(capability),
);
let explanation = explain_peer_or_bearer(
&store,
peer_card.node_id.as_str(),
&db.resource_id,
&capability,
&nonce,
bearer_proof.as_ref(),
)?;
let sync_result: Result<_, String> = if explanation.allowed {
let path = Path::new(&db.path);
match geth_db::extract_crsqlite_changes(path, after_db_version, limit) {
@ -3731,6 +3961,7 @@ pub fn handle_request(
topic,
message,
node: None,
..
} => {
geth_pubsub::validate_topic(&topic)?;
geth_pubsub::validate_message(&message)?;
@ -3738,7 +3969,9 @@ pub fn handle_request(
Ok(ControlResponse::PubsubPublished { message })
}
ControlRequest::PubsubPub { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::PubsubSub { topic, node: None } => {
ControlRequest::PubsubSub {
topic, node: None, ..
} => {
geth_pubsub::validate_topic(&topic)?;
let messages = pubsub_messages_for_topic(node, &topic)?;
Ok(ControlResponse::PubsubMessages {
@ -3764,7 +3997,9 @@ pub fn handle_request(
runtime.listeners.insert(name, listener.clone());
Ok(ControlResponse::PipeListening { listener })
}
ControlRequest::PipeConnect { target, node: None } => {
ControlRequest::PipeConnect {
target, node: None, ..
} => {
geth_pipe::validate_pipe_name(&target)?;
let connection = record_pipe_connection(
node,
@ -4899,6 +5134,7 @@ mod tests {
ControlRequest::CasFetch {
node: right_card.node_id.to_string(),
hash: right_blob.hash.clone(),
bearer_secret: None,
},
)
.await
@ -4927,6 +5163,7 @@ mod tests {
ControlRequest::KvSync {
node: right_card.node_id.to_string(),
name: "prefs".to_owned(),
bearer_secret: None,
},
)
.await
@ -4951,6 +5188,7 @@ mod tests {
topic: "presence/test".to_owned(),
message: "hello".to_owned(),
node: Some(right_card.node_id.to_string()),
bearer_secret: None,
},
)
.await
@ -4974,6 +5212,7 @@ mod tests {
ControlRequest::PubsubSub {
topic: "presence/test".to_owned(),
node: Some(right_card.node_id.to_string()),
bearer_secret: None,
},
)
.await
@ -4997,6 +5236,7 @@ mod tests {
ControlRequest::PipeConnect {
target: "inbox".to_owned(),
node: Some(right_card.node_id.to_string()),
bearer_secret: None,
},
)
.await
@ -5019,6 +5259,7 @@ mod tests {
&left,
ControlRequest::SshProxyConnect {
node: right_card.node_id.to_string(),
bearer_secret: None,
},
)
.await
@ -5042,6 +5283,7 @@ mod tests {
ControlRequest::DocumentSync {
node: right_card.node_id.to_string(),
name: "notes".to_owned(),
bearer_secret: None,
},
)
.await
@ -5066,6 +5308,7 @@ mod tests {
node: right_card.node_id.to_string(),
name: "notes".to_owned(),
limit: 10,
bearer_secret: None,
},
)
.await
@ -5088,6 +5331,45 @@ mod tests {
other => panic!("unexpected denied DB sync response: {other:?}"),
}
let bearer_secret = match handle_request(
&right,
ControlRequest::SecretBearerCreate {
resource: "resource:cas:local".to_owned(),
capabilities: vec!["cas.fetch".to_owned()],
expires_at_ms: None,
},
)
.expect("create remote CAS bearer access")
{
ControlResponse::SecretBearerCreated { access } => access.secret.to_string(),
other => panic!("unexpected bearer create response: {other:?}"),
};
let bearer_fetch = handle_request_async(
&left,
ControlRequest::CasFetch {
node: right_card.node_id.to_string(),
hash: right_blob.hash.clone(),
bearer_secret: Some(bearer_secret),
},
)
.await
.expect("bearer cas fetch");
match bearer_fetch {
ControlResponse::CasFetched {
allowed,
hash,
size_bytes,
reason,
..
} => {
assert!(allowed);
assert_eq!(hash, right_blob.hash);
assert_eq!(size_bytes, right_blob.size_bytes);
assert!(reason.contains("bearer proof"));
}
other => panic!("unexpected bearer CAS fetch response: {other:?}"),
}
handle_request(
&right,
ControlRequest::AuthGrant {
@ -5198,6 +5480,7 @@ mod tests {
ControlRequest::CasFetch {
node: right_card.node_id.to_string(),
hash: right_blob.hash.clone(),
bearer_secret: None,
},
)
.await
@ -5237,6 +5520,7 @@ mod tests {
ControlRequest::KvSync {
node: right_card.node_id.to_string(),
name: "prefs".to_owned(),
bearer_secret: None,
},
)
.await
@ -5277,6 +5561,7 @@ mod tests {
topic: "presence/test".to_owned(),
message: "hello".to_owned(),
node: Some(right_card.node_id.to_string()),
bearer_secret: None,
},
)
.await
@ -5302,6 +5587,7 @@ mod tests {
ControlRequest::PubsubSub {
topic: "presence/test".to_owned(),
node: None,
bearer_secret: None,
},
)
.expect("right pubsub sub after remote publish");
@ -5318,6 +5604,7 @@ mod tests {
ControlRequest::PubsubSub {
topic: "presence/test".to_owned(),
node: Some(right_card.node_id.to_string()),
bearer_secret: None,
},
)
.await
@ -5344,6 +5631,7 @@ mod tests {
ControlRequest::PipeConnect {
target: "inbox".to_owned(),
node: Some(right_card.node_id.to_string()),
bearer_secret: None,
},
)
.await
@ -5368,6 +5656,7 @@ mod tests {
&left,
ControlRequest::SshProxyConnect {
node: right_card.node_id.to_string(),
bearer_secret: None,
},
)
.await
@ -5398,6 +5687,7 @@ mod tests {
ControlRequest::DocumentSync {
node: right_card.node_id.to_string(),
name: "notes".to_owned(),
bearer_secret: None,
},
)
.await
@ -5437,6 +5727,7 @@ mod tests {
node: right_card.node_id.to_string(),
name: "notes".to_owned(),
limit: 10,
bearer_secret: None,
},
)
.await
@ -5472,6 +5763,7 @@ mod tests {
&left,
ControlRequest::SshCertSync {
node: right_card.node_id.to_string(),
bearer_secret: None,
},
)
.await
@ -5517,6 +5809,7 @@ mod tests {
&left,
ControlRequest::SshCertSync {
node: right_card.node_id.to_string(),
bearer_secret: None,
},
)
.await
@ -5552,6 +5845,7 @@ mod tests {
&left,
ControlRequest::SshRevocationSync {
node: right_card.node_id.to_string(),
bearer_secret: None,
},
)
.await