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

Sync SSH metadata over Iroh

Browse source
This commit is contained in:
Eric Wendland 2026-05-18 17:24:10 +02:00
commit 9887b47a40
7 changed files with 686 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -244,6 +244,12 @@ pub async fn handle_request_async(
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
}
other => handle_request(node, other),
}
}
@ -517,8 +523,10 @@ 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(),
PeerControlResponse::CasFetched { .. }
| PeerControlResponse::SshCertSynced { .. }
| PeerControlResponse::SshRevocationSynced { .. } => Err(NodeError::IrohPeer(
"peer returned wrong response type to ping request".to_owned(),
)),
}
}
@ -622,8 +630,10 @@ 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(),
PeerControlResponse::CasFetched { .. }
| PeerControlResponse::SshCertSynced { .. }
| PeerControlResponse::SshRevocationSynced { .. } => Err(NodeError::IrohPeer(
"peer returned wrong response type to auth-check request".to_owned(),
)),
}
}
@ -755,9 +765,199 @@ async fn cas_fetch_from_peer(
"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()),
),
PeerControlResponse::Pong { .. }
| PeerControlResponse::AuthChecked { .. }
| PeerControlResponse::SshCertSynced { .. }
| PeerControlResponse::SshRevocationSynced { .. } => Err(NodeError::IrohPeer(
"peer returned wrong response type to CAS fetch".to_owned(),
)),
}
}
async fn ssh_cert_sync_from_peer(
node: &LocalNode,
peer_node: &str,
) -> Result<ControlResponse, NodeError> {
let response = request_peer_control(node, peer_node, "ssh-cert-sync", |peer_card, nonce| {
PeerControlRequest::SshCertSync { peer_card, nonce }
})
.await?;
match response {
PeerControlResponse::SshCertSynced {
node_id,
agent_id,
endpoint_id,
requests,
certificates,
allowed,
reason,
note,
..
} => {
if !allowed {
return Ok(ControlResponse::SshCertSynced {
peer_node_id: node_id,
peer_agent_id: agent_id,
endpoint_id,
requests_imported: 0,
certificates_imported: 0,
allowed,
reason,
note,
});
}
let store = Store::open(&node.paths.metadata_db())?;
let requests_imported = requests.len();
let certificates_imported = certificates.len();
for request in &requests {
store.insert_ssh_cert_request(&stored_from_ssh_cert_request(request))?;
}
for certificate in &certificates {
store.insert_ssh_certificate(&stored_from_ssh_certificate(certificate))?;
}
Ok(ControlResponse::SshCertSynced {
peer_node_id: node_id,
peer_agent_id: agent_id,
endpoint_id,
requests_imported,
certificates_imported,
allowed,
reason,
note,
})
}
PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
_ => Err(NodeError::IrohPeer(
"peer returned wrong response type to SSH cert sync".to_owned(),
)),
}
}
async fn ssh_revocation_sync_from_peer(
node: &LocalNode,
peer_node: &str,
) -> Result<ControlResponse, NodeError> {
let response = request_peer_control(
node,
peer_node,
"ssh-revocation-sync",
|peer_card, nonce| PeerControlRequest::SshRevocationSync { peer_card, nonce },
)
.await?;
match response {
PeerControlResponse::SshRevocationSynced {
node_id,
agent_id,
endpoint_id,
revocations,
allowed,
reason,
note,
..
} => {
if !allowed {
return Ok(ControlResponse::SshRevocationSynced {
peer_node_id: node_id,
peer_agent_id: agent_id,
endpoint_id,
revocations_imported: 0,
allowed,
reason,
note,
});
}
let store = Store::open(&node.paths.metadata_db())?;
let revocations_imported = revocations.len();
for revocation in &revocations {
store.insert_ssh_revocation(&stored_from_ssh_revocation(revocation))?;
}
Ok(ControlResponse::SshRevocationSynced {
peer_node_id: node_id,
peer_agent_id: agent_id,
endpoint_id,
revocations_imported,
allowed,
reason,
note,
})
}
PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)),
_ => Err(NodeError::IrohPeer(
"peer returned wrong response type to SSH revocation sync".to_owned(),
)),
}
}
async fn request_peer_control(
node: &LocalNode,
peer_node: &str,
operation: &str,
build_request: impl FnOnce(PeerCard, String) -> PeerControlRequest,
) -> Result<PeerControlResponse, 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,
operation,
geth_store::now_ms()
)
.as_bytes(),
);
let request = build_request(self_card, 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(16 * 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()))?;
let response = geth_control::decode_peer_response(text)?;
match &response {
PeerControlResponse::SshCertSynced {
nonce: response_nonce,
..
}
| PeerControlResponse::SshRevocationSynced {
nonce: response_nonce,
..
} if response_nonce == &nonce => Ok(response),
PeerControlResponse::Error { .. } => Ok(response),
_ => Err(NodeError::IrohPeer(format!(
"peer {operation} response did not match request"
))),
}
}
@ -932,6 +1132,102 @@ async fn handle_iroh_control_connection(
}
}
}
PeerControlRequest::SshCertSync { peer_card, 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:ssh:certs".to_owned();
let capability = "ssh_cert.sync".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 (requests, certificates) = if explanation.allowed {
(
store
.list_ssh_cert_requests()?
.into_iter()
.map(ssh_cert_request_from_stored)
.collect::<Result<Vec<_>, _>>()?,
store
.list_ssh_certificates()?
.into_iter()
.map(ssh_certificate_from_stored)
.collect(),
)
} else {
(Vec::new(), Vec::new())
};
PeerControlResponse::SshCertSynced {
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,
requests,
certificates,
allowed: explanation.allowed,
reason: explanation.reason,
evaluated_ops: explanation.evaluated_ops,
nonce,
note: "SSH certificate metadata sync authenticated endpoint/card binding and required ssh_cert.sync on resource:ssh:certs".to_owned(),
}
}
PeerControlRequest::SshRevocationSync { peer_card, 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:ssh:revocations".to_owned();
let capability = "ssh_revocation.sync".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 revocations = if explanation.allowed {
store
.list_ssh_revocations()?
.into_iter()
.map(ssh_revocation_from_stored)
.collect::<Result<Vec<_>, _>>()?
} else {
Vec::new()
};
PeerControlResponse::SshRevocationSynced {
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,
revocations,
allowed: explanation.allowed,
reason: explanation.reason,
evaluated_ops: explanation.evaluated_ops,
nonce,
note: "SSH revocation sync authenticated endpoint/card binding and required ssh_revocation.sync on resource:ssh:revocations".to_owned(),
}
}
};
send.write_all(geth_control::encode_peer_response(&response)?.as_bytes())
.await
@ -1068,6 +1364,8 @@ pub fn handle_request(
ControlRequest::PeerPing { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::PeerAuthCheck { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::CasFetch { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::SshCertSync { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::SshRevocationSync { .. } => Err(NodeError::IrohEndpointUnavailable),
ControlRequest::ResourceList => Ok(ControlResponse::ResourceList {
resources: store
.list_resources()?
@ -2546,6 +2844,41 @@ mod tests {
let right_blob = LocalCas::new(right.paths.cas_dir())
.add_bytes(b"remote cas bytes")
.expect("right cas add");
let right_pubkey = right_home.path().join("request.pub");
std::fs::write(
&right_pubkey,
"ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIGV0aA== node\n",
)
.expect("write right public key");
let requested = handle_request(
&right,
ControlRequest::SshCertRequest {
public_key_path: right_pubkey,
cert_kind: "user".to_owned(),
principals: vec!["eric".to_owned()],
requested_validity: Some("+52w".to_owned()),
renewal_of: None,
reason: Some("test sync".to_owned()),
},
)
.expect("right ssh cert request");
let right_request_id = match requested {
ControlResponse::SshCertRequested { request } => request.id,
other => panic!("unexpected SSH cert request response: {other:?}"),
};
let added_revocation = handle_request(
&right,
ControlRequest::SshRevocationAdd {
kind: "key-id".to_owned(),
target: "old-node-key".to_owned(),
reason: Some("test sync".to_owned()),
},
)
.expect("right ssh revocation");
let right_revocation_id = match added_revocation {
ControlResponse::SshRevocationAdded { revocation } => revocation.id,
other => panic!("unexpected SSH revocation response: {other:?}"),
};
let ping = handle_request_async(
&left,
@ -2692,6 +3025,119 @@ mod tests {
other => panic!("unexpected allowed CAS fetch response: {other:?}"),
}
let denied_cert_sync = handle_request_async(
&left,
ControlRequest::SshCertSync {
node: right_card.node_id.to_string(),
},
)
.await
.expect("denied ssh cert sync");
match denied_cert_sync {
ControlResponse::SshCertSynced {
allowed,
requests_imported,
certificates_imported,
reason,
..
} => {
assert!(!allowed);
assert_eq!(requests_imported, 0);
assert_eq!(certificates_imported, 0);
assert!(reason.contains("no active direct or group grant"));
}
other => panic!("unexpected denied SSH cert sync response: {other:?}"),
}
handle_request(
&right,
ControlRequest::AuthGrant {
subject: left.node_id.clone(),
resource: "resource:ssh:certs".to_owned(),
capability: "ssh_cert.sync".to_owned(),
grant_id: Some("grant:left-ssh-cert-sync".to_owned()),
},
)
.expect("grant left ssh cert sync");
handle_request(
&right,
ControlRequest::AuthGrant {
subject: left.node_id.clone(),
resource: "resource:ssh:revocations".to_owned(),
capability: "ssh_revocation.sync".to_owned(),
grant_id: Some("grant:left-ssh-revocation-sync".to_owned()),
},
)
.expect("grant left ssh revocation sync");
let cert_sync = handle_request_async(
&left,
ControlRequest::SshCertSync {
node: right_card.node_id.to_string(),
},
)
.await
.expect("allowed ssh cert sync");
match cert_sync {
ControlResponse::SshCertSynced {
allowed,
requests_imported,
certificates_imported,
reason,
note,
..
} => {
assert!(allowed);
assert_eq!(requests_imported, 1);
assert_eq!(certificates_imported, 0);
assert!(reason.contains("direct grant"));
assert!(note.contains("ssh_cert.sync"));
}
other => panic!("unexpected allowed SSH cert sync response: {other:?}"),
}
let left_requests = Store::open(&left_paths.metadata_db())
.expect("open left after cert sync")
.list_ssh_cert_requests()
.expect("list synced requests");
assert!(
left_requests
.iter()
.any(|request| request.request_id == right_request_id.as_str())
);
let revocation_sync = handle_request_async(
&left,
ControlRequest::SshRevocationSync {
node: right_card.node_id.to_string(),
},
)
.await
.expect("allowed ssh revocation sync");
match revocation_sync {
ControlResponse::SshRevocationSynced {
allowed,
revocations_imported,
reason,
note,
..
} => {
assert!(allowed);
assert_eq!(revocations_imported, 1);
assert!(reason.contains("direct grant"));
assert!(note.contains("ssh_revocation.sync"));
}
other => panic!("unexpected allowed SSH revocation sync response: {other:?}"),
}
let left_revocations = Store::open(&left_paths.metadata_db())
.expect("open left after revocation sync")
.list_ssh_revocations()
.expect("list synced revocations");
assert!(
left_revocations
.iter()
.any(|revocation| revocation.revocation_id == right_revocation_id.as_str())
);
left_endpoint.shutdown().await;
right_endpoint.shutdown().await;
}