Reject conflicting SSH sync metadata

crates/geth-node/src/lib.rs

@@ -910,13 +910,17 @@ async fn ssh_cert_sync_from_peer(
                 });
             }
             let store = Store::open(&node.paths.metadata_db())?;
-            let requests_imported = requests.len();
-            let certificates_imported = certificates.len();
+            let mut requests_imported = 0;
+            let mut certificates_imported = 0;
             for request in &requests {
-                store.insert_ssh_cert_request(&stored_from_ssh_cert_request(request))?;
+                if insert_ssh_cert_request_if_not_conflicting(&store, request)? {
+                    requests_imported += 1;
+                }
             }
             for certificate in &certificates {
-                store.insert_ssh_certificate(&stored_from_ssh_certificate(certificate))?;
+                if insert_ssh_certificate_if_not_conflicting(&store, certificate)? {
+                    certificates_imported += 1;
+                }
             }
             store_live_sync_cursor(&store, peer_node, "ssh-certs", high_water_ms)?;
             Ok(ControlResponse::SshCertSynced {
@@ -988,9 +992,11 @@ async fn ssh_revocation_sync_from_peer(
                 });
             }
             let store = Store::open(&node.paths.metadata_db())?;
-            let revocations_imported = revocations.len();
+            let mut revocations_imported = 0;
             for revocation in &revocations {
-                store.insert_ssh_revocation(&stored_from_ssh_revocation(revocation))?;
+                if insert_ssh_revocation_if_not_conflicting(&store, revocation)? {
+                    revocations_imported += 1;
+                }
             }
             store_live_sync_cursor(&store, peer_node, "ssh-revocations", high_water_ms)?;
             Ok(ControlResponse::SshRevocationSynced {
@@ -4792,6 +4798,24 @@ fn stored_from_ssh_cert_request(request: &SshCertRequest) -> StoredSshCertReques
     }
 }
 
+fn insert_ssh_cert_request_if_not_conflicting(
+    store: &Store,
+    request: &SshCertRequest,
+) -> Result<bool, NodeError> {
+    let stored = stored_from_ssh_cert_request(request);
+    if let Some(existing) = store.get_ssh_cert_request(&stored.request_id)? {
+        if existing != stored {
+            tracing::warn!(
+                request_id = %stored.request_id,
+                "rejected conflicting SSH certificate request during sync"
+            );
+        }
+        return Ok(false);
+    }
+    store.insert_ssh_cert_request(&stored)?;
+    Ok(true)
+}
+
 fn ssh_cert_request_from_stored(stored: StoredSshCertRequest) -> Result<SshCertRequest, NodeError> {
     let cert_kind = stored
         .cert_kind
@@ -4826,6 +4850,28 @@ fn stored_from_ssh_certificate(certificate: &SshCertificateRecord) -> StoredSshC
     }
 }
 
+fn insert_ssh_certificate_if_not_conflicting(
+    store: &Store,
+    certificate: &SshCertificateRecord,
+) -> Result<bool, NodeError> {
+    let stored = stored_from_ssh_certificate(certificate);
+    if let Some(existing) = store
+        .list_ssh_certificates()?
+        .into_iter()
+        .find(|existing| existing.cert_id == stored.cert_id)
+    {
+        if existing != stored {
+            tracing::warn!(
+                cert_id = %stored.cert_id,
+                "rejected conflicting SSH certificate during sync"
+            );
+        }
+        return Ok(false);
+    }
+    store.insert_ssh_certificate(&stored)?;
+    Ok(true)
+}
+
 fn ssh_certificate_from_stored(stored: StoredSshCertificate) -> SshCertificateRecord {
     SshCertificateRecord {
         id: SshCertId::new(stored.cert_id),
@@ -4847,6 +4893,28 @@ fn stored_from_ssh_revocation(revocation: &SshRevocationEntry) -> StoredSshRevoc
     }
 }
 
+fn insert_ssh_revocation_if_not_conflicting(
+    store: &Store,
+    revocation: &SshRevocationEntry,
+) -> Result<bool, NodeError> {
+    let stored = stored_from_ssh_revocation(revocation);
+    if let Some(existing) = store
+        .list_ssh_revocations()?
+        .into_iter()
+        .find(|existing| existing.revocation_id == stored.revocation_id)
+    {
+        if existing != stored {
+            tracing::warn!(
+                revocation_id = %stored.revocation_id,
+                "rejected conflicting SSH revocation during sync"
+            );
+        }
+        return Ok(false);
+    }
+    store.insert_ssh_revocation(&stored)?;
+    Ok(true)
+}
+
 fn ssh_revocation_from_stored(
     stored: StoredSshRevocation,
 ) -> Result<SshRevocationEntry, NodeError> {
@@ -4951,6 +5019,70 @@ mod tests {
         .expect("store auth grant");
     }
 
+    #[test]
+    fn ssh_sync_import_rejects_conflicting_records() {
+        let store = Store::open_memory().expect("open");
+        let request = SshCertRequest {
+            id: SshCertRequestId::new("ssh-cert-request:1"),
+            requester_node: NodeId::new("node:right"),
+            public_key: "ssh-ed25519 AAAA".to_owned(),
+            public_key_fingerprint: "SHA256:one".to_owned(),
+            cert_kind: SshCertKind::User,
+            principals: vec!["eric".to_owned()],
+            requested_validity: None,
+            renewal_of: None,
+            reason: Some("original".to_owned()),
+            status: SshCertRequestStatus::Pending,
+            created_at: UnixMillis(1),
+        };
+        assert!(
+            insert_ssh_cert_request_if_not_conflicting(&store, &request).expect("insert request")
+        );
+        let mut conflicting_request = request.clone();
+        conflicting_request.reason = Some("conflict".to_owned());
+        assert!(
+            !insert_ssh_cert_request_if_not_conflicting(&store, &conflicting_request)
+                .expect("reject conflicting request")
+        );
+        assert_eq!(
+            store
+                .get_ssh_cert_request(request.id.as_str())
+                .expect("get request")
+                .expect("request")
+                .reason
+                .as_deref(),
+            Some("original")
+        );
+
+        let revocation = SshRevocationEntry {
+            id: geth_types::SshRevocationId::new("ssh-revocation:1"),
+            kind: SshRevocationKind::KeyId,
+            target: "old-key".to_owned(),
+            reason: Some("original".to_owned()),
+            created_at: UnixMillis(2),
+            published: true,
+        };
+        assert!(
+            insert_ssh_revocation_if_not_conflicting(&store, &revocation)
+                .expect("insert revocation")
+        );
+        let mut conflicting_revocation = revocation.clone();
+        conflicting_revocation.target = "other-key".to_owned();
+        assert!(
+            !insert_ssh_revocation_if_not_conflicting(&store, &conflicting_revocation)
+                .expect("reject conflicting revocation")
+        );
+        assert_eq!(
+            store
+                .list_ssh_revocations()
+                .expect("list revocations")
+                .first()
+                .expect("revocation")
+                .target,
+            "old-key"
+        );
+    }
+
     #[test]
     fn sync_watermarks_include_only_authorized_streams() {
         let store = Store::open_memory().expect("open");