diff --git a/AGENTS.md b/AGENTS.md index f6c89a7..b0287da 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -147,8 +147,9 @@ Roadmap items should be actionable and checkable: - Document resources can be registered locally and updated with validated local JSON state. Automerge editing/state and sync are still roadmap work. - Pubsub supports local daemon-lifetime publish/subscribe snapshots through a - bounded in-memory ring buffer. Iroh-gossip replication, private topics, and - pubsub capability enforcement are still roadmap work. + bounded in-memory ring buffer. Remote publish over the protected Iroh control + ALPN requires `pubsub.publish` on `resource:pubsub:`. Iroh-gossip + replication, private topics, and remote subscribe are still roadmap work. - Pipe listen/connect supports a local daemon-lifetime registry only. Iroh byte streams, TCP/Unix forwarding, and pipe capability enforcement are still roadmap work. diff --git a/README.md b/README.md index 6bdd982..7fca023 100644 --- a/README.md +++ b/README.md @@ -106,7 +106,8 @@ The bootstrap implementation provides: `--subject ` to exercise local capability checks for non-local callers; `geth kv sync ` pulls authorized remote updates - local JSON document commands: `geth document create/status/set/get` -- local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub` +- local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub`; `geth pubsub + pub --node ` publishes to an authorized peer - SSH certificate flow metadata: - `geth ssh cert request --public-key --principal ` - `geth ssh cert requests` @@ -146,6 +147,10 @@ Named KV stores participate in the same live-sync loop once they exist locally: manual `geth kv sync ` and background ticks require `kv.read` on the remote `resource:kv:` and import only remote entries that are not older than the local value. +Remote pubsub publish uses the protected Iroh control path too. The remote peer +requires `pubsub.publish` on `resource:pubsub:` before recording the +message in its local daemon-lifetime ring buffer. Pubsub remains lossy and is +not durable storage. Importing or pinging a peer card never grants capabilities by itself. When `[iroh].local_discovery = true`, the daemon also advertises and discovers signed peer cards on LAN using a geth-specific mDNS TXT payload. That payload is diff --git a/crates/geth-cli/src/lib.rs b/crates/geth-cli/src/lib.rs index 1389a19..398882c 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -312,8 +312,15 @@ pub enum KvCommand { #[derive(Debug, Subcommand)] pub enum PubsubCommand { - Pub { topic: String, message: String }, - Sub { topic: String }, + Pub { + topic: String, + message: String, + #[arg(long)] + node: Option, + }, + Sub { + topic: String, + }, } #[derive(Debug, Subcommand)] @@ -624,7 +631,15 @@ fn request_for_command(command: Command) -> Result { KvCommand::Sync { node, name } => ControlRequest::KvSync { node, name }, }, Command::Pubsub { command } => match command { - PubsubCommand::Pub { topic, message } => ControlRequest::PubsubPub { topic, message }, + PubsubCommand::Pub { + topic, + message, + node, + } => ControlRequest::PubsubPub { + topic, + message, + node, + }, PubsubCommand::Sub { topic } => ControlRequest::PubsubSub { topic }, }, Command::Pipe { command } => match command { @@ -1361,6 +1376,28 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { println!("published: {}", message.topic); println!("published_at_ms: {}", message.published_at.0); } + ControlResponse::PubsubRemotePublished { + peer_node_id, + peer_agent_id, + endpoint_id, + message, + allowed, + reason, + note, + } => { + if allowed { + println!("published: {}", message.topic); + println!("peer: {peer_node_id}"); + println!("published_at_ms: {}", message.published_at.0); + } else { + println!("pubsub publish denied by {peer_node_id}"); + } + println!("agent: {peer_agent_id}"); + println!("endpoint: {endpoint_id}"); + println!("allowed: {allowed}"); + println!("reason: {reason}"); + println!("note: {note}"); + } ControlResponse::PubsubMessages { topic, messages, diff --git a/crates/geth-control/src/lib.rs b/crates/geth-control/src/lib.rs index ba1f9e0..75b87ce 100644 --- a/crates/geth-control/src/lib.rs +++ b/crates/geth-control/src/lib.rs @@ -216,6 +216,7 @@ pub enum ControlRequest { PubsubPub { topic: String, message: String, + node: Option, }, PubsubSub { topic: String, @@ -451,6 +452,15 @@ pub enum ControlResponse { PubsubPublished { message: PubsubMessage, }, + PubsubRemotePublished { + peer_node_id: String, + peer_agent_id: String, + endpoint_id: String, + message: PubsubMessage, + allowed: bool, + reason: String, + note: String, + }, PubsubMessages { topic: String, messages: Vec, @@ -541,6 +551,12 @@ pub enum PeerControlRequest { since_ms: i64, nonce: String, }, + PubsubPublish { + peer_card: PeerCard, + topic: String, + message: String, + nonce: String, + }, } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] @@ -623,6 +639,18 @@ pub enum PeerControlResponse { nonce: String, note: String, }, + PubsubPublished { + node_id: String, + agent_id: String, + endpoint_id: String, + remote_endpoint_id: String, + message: Option, + allowed: bool, + reason: String, + evaluated_ops: usize, + nonce: String, + note: String, + }, Error { message: String, }, @@ -822,6 +850,34 @@ mod tests { request ); + let request = ControlRequest::PubsubPub { + topic: "presence/test".to_owned(), + message: "online".to_owned(), + node: Some("node:peer".to_owned()), + }; + assert_eq!( + decode_request(&encode_request(&request).expect("encode")).expect("decode"), + request + ); + + let response = ControlResponse::PubsubRemotePublished { + peer_node_id: "node:peer".to_owned(), + peer_agent_id: "agent:peer".to_owned(), + endpoint_id: "endpoint:peer".to_owned(), + message: PubsubMessage { + topic: "presence/test".into(), + message: "online".to_owned(), + published_at: geth_types::UnixMillis(1), + }, + allowed: true, + reason: "direct grant".to_owned(), + note: "lossy".to_owned(), + }; + assert_eq!( + decode_response(&encode_response(&response).expect("encode")).expect("decode"), + response + ); + let request = ControlRequest::DbChanges { name: "notes".to_owned(), after_db_version: Some(7), @@ -990,5 +1046,27 @@ mod tests { .expect("decode"), response ); + + let response = PeerControlResponse::PubsubPublished { + node_id: "node:peer".to_owned(), + agent_id: "agent:peer".to_owned(), + endpoint_id: "endpoint:peer".to_owned(), + remote_endpoint_id: "endpoint:caller".to_owned(), + message: Some(PubsubMessage { + topic: "presence/test".into(), + message: "online".to_owned(), + published_at: geth_types::UnixMillis(1), + }), + allowed: true, + reason: "direct grant".to_owned(), + evaluated_ops: 1, + nonce: "nonce".to_owned(), + note: "lossy".to_owned(), + }; + assert_eq!( + decode_peer_response(&encode_peer_response(&response).expect("encode")) + .expect("decode"), + response + ); } } diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 38db66f..e2bfea7 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -263,6 +263,11 @@ pub async fn handle_request_async( node: peer_node, name, } => kv_sync_from_peer(node, &peer_node, &name).await, + ControlRequest::PubsubPub { + topic, + message, + node: Some(peer_node), + } => pubsub_publish_to_peer(node, &peer_node, topic, message).await, other => handle_request(node, other), } } @@ -539,7 +544,8 @@ async fn peer_ping(node: &LocalNode, peer_node: &str) -> Result Err(NodeError::IrohPeer( + | PeerControlResponse::KvSynced { .. } + | PeerControlResponse::PubsubPublished { .. } => Err(NodeError::IrohPeer( "peer returned wrong response type to ping request".to_owned(), )), } @@ -647,7 +653,8 @@ async fn peer_auth_check( PeerControlResponse::CasFetched { .. } | PeerControlResponse::SshCertSynced { .. } | PeerControlResponse::SshRevocationSynced { .. } - | PeerControlResponse::KvSynced { .. } => Err(NodeError::IrohPeer( + | PeerControlResponse::KvSynced { .. } + | PeerControlResponse::PubsubPublished { .. } => Err(NodeError::IrohPeer( "peer returned wrong response type to auth-check request".to_owned(), )), } @@ -784,7 +791,8 @@ async fn cas_fetch_from_peer( | PeerControlResponse::AuthChecked { .. } | PeerControlResponse::SshCertSynced { .. } | PeerControlResponse::SshRevocationSynced { .. } - | PeerControlResponse::KvSynced { .. } => Err(NodeError::IrohPeer( + | PeerControlResponse::KvSynced { .. } + | PeerControlResponse::PubsubPublished { .. } => Err(NodeError::IrohPeer( "peer returned wrong response type to CAS fetch".to_owned(), )), } @@ -1010,6 +1018,56 @@ async fn kv_sync_from_peer( } } +async fn pubsub_publish_to_peer( + node: &LocalNode, + peer_node: &str, + topic: String, + message: String, +) -> Result { + geth_pubsub::validate_topic(&topic)?; + geth_pubsub::validate_message(&message)?; + let response = request_peer_control(node, peer_node, "pubsub-publish", |peer_card, nonce| { + PeerControlRequest::PubsubPublish { + peer_card, + topic: topic.clone(), + message: message.clone(), + nonce, + } + }) + .await?; + match response { + PeerControlResponse::PubsubPublished { + node_id, + agent_id, + endpoint_id, + message, + allowed, + reason, + note, + .. + } => { + let message = message.unwrap_or_else(|| PubsubMessage { + topic: topic.into(), + message: String::new(), + published_at: UnixMillis(0), + }); + Ok(ControlResponse::PubsubRemotePublished { + peer_node_id: node_id, + peer_agent_id: agent_id, + endpoint_id, + message, + allowed, + reason, + note, + }) + } + PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)), + _ => Err(NodeError::IrohPeer( + "peer returned wrong response type to pubsub publish".to_owned(), + )), + } +} + fn live_sync_cursor_key(peer_node: &str, stream: &str) -> String { format!("live-sync:{peer_node}:{stream}") } @@ -1106,6 +1164,10 @@ async fn request_peer_control( | PeerControlResponse::KvSynced { nonce: response_nonce, .. + } + | PeerControlResponse::PubsubPublished { + nonce: response_nonce, + .. } if response_nonce == &nonce => Ok(response), PeerControlResponse::Error { .. } => Ok(response), _ => Err(NodeError::IrohPeer(format!( @@ -1497,6 +1559,53 @@ async fn handle_iroh_control_connection( } } } + PeerControlRequest::PubsubPublish { + peer_card, + topic, + message, + nonce, + } => { + geth_pubsub::validate_topic(&topic)?; + geth_pubsub::validate_message(&message)?; + 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 = 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 published = if explanation.allowed { + Some(record_pubsub_message(&node, topic, message)?) + } else { + None + }; + PeerControlResponse::PubsubPublished { + 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, + message: published, + allowed: explanation.allowed, + reason: explanation.reason, + evaluated_ops: explanation.evaluated_ops, + nonce, + note: "pubsub publish authenticated endpoint/card binding and required pubsub.publish on the remote topic resource; pubsub is lossy".to_owned(), + } + } }; send.write_all(geth_control::encode_peer_response(&response)?.as_bytes()) .await @@ -2492,25 +2601,17 @@ pub fn handle_request( state: document_state_from_stored(&stored), }) } - ControlRequest::PubsubPub { topic, message } => { + ControlRequest::PubsubPub { + topic, + message, + node: None, + } => { geth_pubsub::validate_topic(&topic)?; geth_pubsub::validate_message(&message)?; - let message = PubsubMessage { - topic: topic.clone().into(), - message, - published_at: UnixMillis(geth_store::now_ms()), - }; - let mut runtime = node - .runtime - .pubsub - .lock() - .map_err(|_| NodeError::RuntimeLockPoisoned)?; - runtime.messages.push_back(message.clone()); - while runtime.messages.len() > PUBSUB_RING_LIMIT { - runtime.messages.pop_front(); - } + let message = record_pubsub_message(node, topic, message)?; Ok(ControlResponse::PubsubPublished { message }) } + ControlRequest::PubsubPub { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable), ControlRequest::PubsubSub { topic } => { geth_pubsub::validate_topic(&topic)?; let runtime = node @@ -2659,6 +2760,28 @@ fn ensure_local_kv_store(store: &Store, name: &str) -> Result Result { + let message = PubsubMessage { + topic: topic.into(), + message, + published_at: UnixMillis(geth_store::now_ms()), + }; + let mut runtime = node + .runtime + .pubsub + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)?; + runtime.messages.push_back(message.clone()); + while runtime.messages.len() > PUBSUB_RING_LIMIT { + runtime.messages.pop_front(); + } + Ok(message) +} + fn document_resource_from_stored(stored: &StoredDocumentResource) -> DocumentResource { DocumentResource { id: stored.document_id.clone().into(), @@ -3280,6 +3403,30 @@ mod tests { other => panic!("unexpected denied KV sync response: {other:?}"), } + let denied_pubsub = handle_request_async( + &left, + ControlRequest::PubsubPub { + topic: "presence/test".to_owned(), + message: "hello".to_owned(), + node: Some(right_card.node_id.to_string()), + }, + ) + .await + .expect("denied remote pubsub publish"); + match denied_pubsub { + ControlResponse::PubsubRemotePublished { + allowed, + reason, + message, + .. + } => { + assert!(!allowed); + assert!(message.message.is_empty()); + assert!(reason.contains("no active direct or group grant")); + } + other => panic!("unexpected denied pubsub publish response: {other:?}"), + } + handle_request( &right, ControlRequest::AuthGrant { @@ -3300,6 +3447,16 @@ mod tests { }, ) .expect("grant left kv read"); + handle_request( + &right, + ControlRequest::AuthGrant { + subject: left.node_id.clone(), + resource: "resource:pubsub:presence/test".to_owned(), + capability: "pubsub.publish".to_owned(), + grant_id: Some("grant:left-pubsub-publish".to_owned()), + }, + ) + .expect("grant left pubsub publish"); let allowed = handle_request_async( &left, @@ -3397,6 +3554,47 @@ mod tests { other => panic!("unexpected synced KV get response: {other:?}"), } + let published = handle_request_async( + &left, + ControlRequest::PubsubPub { + topic: "presence/test".to_owned(), + message: "hello".to_owned(), + node: Some(right_card.node_id.to_string()), + }, + ) + .await + .expect("allowed remote pubsub publish"); + match published { + ControlResponse::PubsubRemotePublished { + allowed, + message, + reason, + note, + .. + } => { + assert!(allowed); + assert_eq!(message.topic.as_str(), "presence/test"); + assert_eq!(message.message, "hello"); + assert!(reason.contains("direct grant")); + assert!(note.contains("lossy")); + } + other => panic!("unexpected allowed pubsub publish response: {other:?}"), + } + let remote_messages = handle_request( + &right, + ControlRequest::PubsubSub { + topic: "presence/test".to_owned(), + }, + ) + .expect("right pubsub sub after remote publish"); + match remote_messages { + ControlResponse::PubsubMessages { messages, .. } => { + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].message, "hello"); + } + other => panic!("unexpected remote pubsub messages response: {other:?}"), + } + let denied_cert_sync = handle_request_async( &left, ControlRequest::SshCertSync { diff --git a/crates/geth/tests/bootstrap.rs b/crates/geth/tests/bootstrap.rs index 9eb020c..65896bd 100644 --- a/crates/geth/tests/bootstrap.rs +++ b/crates/geth/tests/bootstrap.rs @@ -1225,6 +1225,7 @@ fn pubsub_pub_sub_uses_lossy_in_memory_runtime() { geth_control::ControlRequest::PubsubPub { topic: "presence/laptop".to_owned(), message: "online".to_owned(), + node: None, }, ) .expect("publish"); @@ -1278,6 +1279,7 @@ fn pubsub_pub_sub_uses_lossy_in_memory_runtime() { geth_control::ControlRequest::PubsubPub { topic: "presence/laptop".to_owned(), message: String::new(), + node: None, }, ) .is_err() diff --git a/docs/architecture.md b/docs/architecture.md index 7b9b713..10008cd 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -158,8 +158,12 @@ Automerge CRDT state. Automerge state encoding and sync are future work. `geth-pubsub` currently supports local publish/subscribe snapshots through the daemon control protocol. Messages live in a bounded in-memory ring buffer and are lost when the daemon stops. This is deliberate: pubsub is a lossy wakeup and -presence channel, not authoritative storage. Iroh-gossip replication, private -topics, and capability enforcement are future work. +presence channel, not authoritative storage. `geth pubsub pub +--node ` can publish to an imported peer over the protected Iroh +control ALPN. The remote daemon validates endpoint/card binding and requires +`pubsub.publish` on `resource:pubsub:` before recording the message in +its local ring buffer. Iroh-gossip replication and private topics are future +work. `geth-pipe` currently supports `pipe listen/connect` against a local daemon-lifetime registry. This is a control-plane scaffold for names and diff --git a/docs/roadmap.md b/docs/roadmap.md index 62214ad..0d8196f 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -278,8 +278,12 @@ authorization and durable-state boundaries clear. buffer. - `[x]` Pubsub messages are documented and tested as lossy notifications, not durable facts. - - `[ ]` `geth pubsub pub/sub` works across local test nodes over Iroh. - - `[ ]` Pubsub publish/subscribe access is capability checked. + - `[x]` `geth pubsub pub --node ` publishes to an + imported peer over Iroh. + - `[x]` Remote pubsub publish requires `pubsub.publish` on + `resource:pubsub:`. + - `[ ]` Replace bootstrap remote publish with iroh-gossip topics. + - `[ ]` Pubsub subscribe works across local test nodes over Iroh or gossip. - `[ ]` Docs and tests keep durable state in CAS/KV/document/db instead. ## Phase 4: Pipes, SSH Proxy, And SSH Distribution