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

Represent SSH distribution sync as a resource log

Browse source
This commit is contained in:
Eric Wendland 2026-05-22 16:51:30 +02:00
commit 42d7301bdb
6 changed files with 247 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

7
crates/geth-control/src/lib.rs
View file

@ -10,7 +10,8 @@ use geth_pubsub::PubsubMessage;
use geth_resource::ResourceDescriptor;
use geth_secrets::{BearerAccess, BearerChallenge, BearerProof, ResourceMasterSecret};
use geth_ssh_identity::{
SshCertApproval, SshCertRequest, SshCertificateRecord, SshRevocationEntry,
SshCertApproval, SshCertRequest, SshCertificateRecord, SshDistributionLogEntry,
SshRevocationEntry,
};
use geth_ssh_proxy::SshProxyConnection;
use geth_types::BlobHash;
@ -1177,6 +1178,7 @@ pub enum PeerControlResponse {
remote_endpoint_id: String,
requests: Vec<SshCertRequest>,
certificates: Vec<SshCertificateRecord>,
log_entries: Vec<SshDistributionLogEntry>,
high_water_ms: i64,
allowed: bool,
reason: String,
@ -1190,6 +1192,7 @@ pub enum PeerControlResponse {
endpoint_id: String,
remote_endpoint_id: String,
revocations: Vec<SshRevocationEntry>,
log_entries: Vec<SshDistributionLogEntry>,
high_water_ms: i64,
allowed: bool,
reason: String,
@ -2316,6 +2319,7 @@ mod tests {
remote_endpoint_id: "endpoint:caller".to_owned(),
requests: Vec::new(),
certificates: Vec::new(),
log_entries: Vec::new(),
high_water_ms: 42,
allowed: false,
reason: "no grant".to_owned(),
@ -2335,6 +2339,7 @@ mod tests {
endpoint_id: "endpoint:peer".to_owned(),
remote_endpoint_id: "endpoint:caller".to_owned(),
revocations: Vec::new(),
log_entries: Vec::new(),
high_water_ms: 42,
allowed: false,
reason: "no grant".to_owned(),

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

@ -35,9 +35,9 @@ use geth_secrets::{BearerAccess, BearerChallenge, BearerProof, ResourceMasterSec
use geth_ssh_identity::{
SSH_CERT_ISSUANCE_NAMESPACE, SSH_CERT_REQUEST_NAMESPACE, SSH_REVOCATION_LIST_NAMESPACE,
SshCertApproval, SshCertKind, SshCertRequest, SshCertRequestStatus, SshCertificateRecord,
SshRecordProvenance, SshRevocationEntry, SshRevocationExportFormat, SshRevocationKind,
build_ssh_cert_sign_command, cert_request_id, certificate_id, openssh_krl_spec,
parse_openssh_krl_spec, revocation_id, ssh_cert_request_signing_payload,
SshDistributionLogEntry, SshRecordProvenance, SshRevocationEntry, SshRevocationExportFormat,
SshRevocationKind, build_ssh_cert_sign_command, cert_request_id, certificate_id,
openssh_krl_spec, parse_openssh_krl_spec, revocation_id, ssh_cert_request_signing_payload,
ssh_certificate_signing_payload, ssh_public_key_fingerprint, ssh_revocation_signing_payload,
write_openssh_krl,
};
@ -1731,6 +1731,7 @@ async fn ssh_cert_sync_from_peer(
endpoint_id,
requests,
certificates,
log_entries,
high_water_ms,
allowed,
reason,
@ -1750,18 +1751,13 @@ async fn ssh_cert_sync_from_peer(
});
}
let store = Store::open(&node.paths.metadata_db())?;
let mut requests_imported = 0;
let mut certificates_imported = 0;
for request in &requests {
if insert_ssh_cert_request_if_not_conflicting(&store, request)? {
requests_imported += 1;
}
}
for certificate in &certificates {
if insert_ssh_certificate_if_not_conflicting(&store, certificate)? {
certificates_imported += 1;
}
}
let entries = if log_entries.is_empty() {
ssh_cert_distribution_log_entries(requests, certificates)
} else {
log_entries
};
let (requests_imported, certificates_imported, _) =
import_ssh_distribution_log_entries(&store, &entries)?;
store_live_sync_cursor(&store, peer_node, "ssh-certs", high_water_ms)?;
Ok(ControlResponse::SshCertSynced {
peer_node_id: node_id,
@ -1814,6 +1810,7 @@ async fn ssh_revocation_sync_from_peer(
agent_id,
endpoint_id,
revocations,
log_entries,
high_water_ms,
allowed,
reason,
@ -1832,12 +1829,13 @@ async fn ssh_revocation_sync_from_peer(
});
}
let store = Store::open(&node.paths.metadata_db())?;
let mut revocations_imported = 0;
for revocation in &revocations {
if insert_ssh_revocation_if_not_conflicting(&store, revocation)? {
revocations_imported += 1;
}
}
let entries = if log_entries.is_empty() {
ssh_revocation_distribution_log_entries(revocations)
} else {
log_entries
};
let (_, _, revocations_imported) =
import_ssh_distribution_log_entries(&store, &entries)?;
store_live_sync_cursor(&store, peer_node, "ssh-revocations", high_water_ms)?;
Ok(ControlResponse::SshRevocationSynced {
peer_node_id: node_id,
@ -5259,7 +5257,7 @@ async fn handle_iroh_control_connection(
&nonce,
bearer_proof.as_ref(),
)?;
let (requests, certificates) = if explanation.allowed {
let (requests, certificates, log_entries) = if explanation.allowed {
(
store
.list_ssh_cert_requests_since(since_ms)?
@ -5271,9 +5269,15 @@ async fn handle_iroh_control_connection(
.into_iter()
.map(ssh_certificate_from_stored)
.collect::<Result<Vec<_>, _>>()?,
Vec::new(),
)
} else {
(Vec::new(), Vec::new())
(Vec::new(), Vec::new(), Vec::new())
};
let log_entries = if explanation.allowed {
ssh_cert_distribution_log_entries(requests.clone(), certificates.clone())
} else {
log_entries
};
PeerControlResponse::SshCertSynced {
node_id: node.node_id.clone(),
@ -5282,6 +5286,7 @@ async fn handle_iroh_control_connection(
remote_endpoint_id,
requests,
certificates,
log_entries,
high_water_ms,
allowed: explanation.allowed,
reason: explanation.reason,
@ -5320,14 +5325,19 @@ async fn handle_iroh_control_connection(
&nonce,
bearer_proof.as_ref(),
)?;
let revocations = if explanation.allowed {
let (revocations, log_entries) = if explanation.allowed {
store
.list_ssh_revocations_since(since_ms)?
.into_iter()
.map(ssh_revocation_from_stored)
.collect::<Result<Vec<_>, _>>()?
.collect::<Result<Vec<_>, _>>()
.map(|revocations| {
let log_entries =
ssh_revocation_distribution_log_entries(revocations.clone());
(revocations, log_entries)
})?
} else {
Vec::new()
(Vec::new(), Vec::new())
};
PeerControlResponse::SshRevocationSynced {
node_id: node.node_id.clone(),
@ -5335,6 +5345,7 @@ async fn handle_iroh_control_connection(
endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(),
remote_endpoint_id,
revocations,
log_entries,
high_water_ms,
allowed: explanation.allowed,
reason: explanation.reason,
@ -10587,6 +10598,77 @@ fn insert_ssh_revocation_if_not_conflicting(
Ok(true)
}
fn ssh_cert_distribution_log_entries(
requests: Vec<SshCertRequest>,
certificates: Vec<SshCertificateRecord>,
) -> Vec<SshDistributionLogEntry> {
let mut entries = requests
.into_iter()
.map(SshDistributionLogEntry::cert_request)
.chain(
certificates
.into_iter()
.map(SshDistributionLogEntry::certificate),
)
.collect::<Vec<_>>();
entries.sort_by(|left, right| {
left.timestamp()
.0
.cmp(&right.timestamp().0)
.then_with(|| left.entry_id().cmp(right.entry_id()))
});
entries
}
fn ssh_revocation_distribution_log_entries(
revocations: Vec<SshRevocationEntry>,
) -> Vec<SshDistributionLogEntry> {
let mut entries = revocations
.into_iter()
.map(SshDistributionLogEntry::revocation)
.collect::<Vec<_>>();
entries.sort_by(|left, right| {
left.timestamp()
.0
.cmp(&right.timestamp().0)
.then_with(|| left.entry_id().cmp(right.entry_id()))
});
entries
}
fn import_ssh_distribution_log_entries(
store: &Store,
entries: &[SshDistributionLogEntry],
) -> Result<(usize, usize, usize), NodeError> {
let mut requests_imported = 0;
let mut certificates_imported = 0;
let mut revocations_imported = 0;
for entry in entries {
match entry {
SshDistributionLogEntry::CertRequest { request, .. } => {
if insert_ssh_cert_request_if_not_conflicting(store, request)? {
requests_imported += 1;
}
}
SshDistributionLogEntry::Certificate { certificate, .. } => {
if insert_ssh_certificate_if_not_conflicting(store, certificate)? {
certificates_imported += 1;
}
}
SshDistributionLogEntry::Revocation { revocation, .. } => {
if insert_ssh_revocation_if_not_conflicting(store, revocation)? {
revocations_imported += 1;
}
}
}
}
Ok((
requests_imported,
certificates_imported,
revocations_imported,
))
}
fn ssh_revocation_from_stored(
stored: StoredSshRevocation,
) -> Result<SshRevocationEntry, NodeError> {

98
crates/geth-ssh-identity/src/lib.rs
View file

@ -220,6 +220,79 @@ pub struct SshRevocationEntry {
pub provenance: Option<SshRecordProvenance>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum SshDistributionLogEntry {
CertRequest {
entry_id: String,
resource: String,
timestamp: UnixMillis,
request: SshCertRequest,
},
Certificate {
entry_id: String,
resource: String,
timestamp: UnixMillis,
certificate: SshCertificateRecord,
},
Revocation {
entry_id: String,
resource: String,
timestamp: UnixMillis,
revocation: SshRevocationEntry,
},
}
impl SshDistributionLogEntry {
#[must_use]
pub fn cert_request(request: SshCertRequest) -> Self {
Self::CertRequest {
entry_id: format!("ssh-log:cert-request:{}", request.id),
resource: "resource:ssh:certs".to_owned(),
timestamp: request.created_at,
request,
}
}
#[must_use]
pub fn certificate(certificate: SshCertificateRecord) -> Self {
Self::Certificate {
entry_id: format!("ssh-log:certificate:{}", certificate.id),
resource: "resource:ssh:certs".to_owned(),
timestamp: certificate.imported_at,
certificate,
}
}
#[must_use]
pub fn revocation(revocation: SshRevocationEntry) -> Self {
Self::Revocation {
entry_id: format!("ssh-log:revocation:{}", revocation.id),
resource: "resource:ssh:revocations".to_owned(),
timestamp: revocation.created_at,
revocation,
}
}
#[must_use]
pub fn timestamp(&self) -> UnixMillis {
match self {
Self::CertRequest { timestamp, .. }
| Self::Certificate { timestamp, .. }
| Self::Revocation { timestamp, .. } => *timestamp,
}
}
#[must_use]
pub fn entry_id(&self) -> &str {
match self {
Self::CertRequest { entry_id, .. }
| Self::Certificate { entry_id, .. }
| Self::Revocation { entry_id, .. } => entry_id,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SshCertRequestSigningPayload {
pub id: SshCertRequestId,
@ -547,6 +620,31 @@ mod tests {
assert_eq!(decoded, request);
}
#[test]
fn ssh_distribution_log_entries_roundtrip_with_stable_ids() {
let request = SshCertRequest {
id: "ssh-cert-request:1".into(),
requester_node: "node:laptop".into(),
public_key: "ssh-ed25519 AAAA test".to_owned(),
public_key_fingerprint: ssh_public_key_fingerprint("ssh-ed25519 AAAA test"),
cert_kind: SshCertKind::User,
principals: vec!["eric".to_owned()],
requested_validity: Some("+52w".to_owned()),
renewal_of: None,
reason: Some("bootstrap".to_owned()),
status: SshCertRequestStatus::Pending,
created_at: UnixMillis(10),
provenance: None,
};
let entry = SshDistributionLogEntry::cert_request(request);
assert_eq!(entry.entry_id(), "ssh-log:cert-request:ssh-cert-request:1");
assert_eq!(entry.timestamp(), UnixMillis(10));
let json = serde_json::to_string(&entry).expect("json");
let decoded: SshDistributionLogEntry = serde_json::from_str(&json).expect("decode");
assert_eq!(decoded, entry);
}
#[test]
fn sign_command_includes_host_flag_for_host_certs() {
let request = SshCertRequest {