diff --git a/AGENTS.md b/AGENTS.md index 1b69cd5..bf7ef07 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -155,8 +155,10 @@ Roadmap items should be actionable and checkable: and true CRDT sync are still roadmap work. - Pubsub supports local daemon-lifetime publish/subscribe snapshots through a 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. + ALPN requires `pubsub.publish` on `resource:pubsub:`. Remote subscribe + over the same path requires `pubsub.subscribe` and returns the peer's current + daemon-lifetime snapshot. Iroh-gossip replication and private topics are still + roadmap work. - Pipe listen/connect supports a daemon-lifetime registry. Remote `geth pipe connect --node ` uses the protected Iroh control ALPN and requires `pipe.connect` on `resource:pipe:`. Iroh byte streams, diff --git a/README.md b/README.md index e46097b..8e9ba9d 100644 --- a/README.md +++ b/README.md @@ -109,7 +109,8 @@ The bootstrap implementation provides: - local JSON document commands: `geth document create/status/set/get`; `geth document sync ` pulls authorized remote JSON state - local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub`; `geth pubsub - pub --node ` publishes to an authorized peer + pub --node ` publishes to an authorized peer; + `geth pubsub sub --node ` reads an authorized peer snapshot - SSH certificate flow metadata: - `geth ssh cert request --public-key --principal ` - `geth ssh cert requests` @@ -157,7 +158,9 @@ 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. +not durable storage. Remote pubsub subscribe uses the same protected path and +requires `pubsub.subscribe` on `resource:pubsub:` before returning the +peer's current daemon-lifetime snapshot for that topic. Remote pipe connect uses the same protected Iroh control path and requires `pipe.connect` on `resource:pipe:`. The current prototype records a remote connection attempt and whether a listener exists; byte streaming and forwarding diff --git a/crates/geth-cli/src/lib.rs b/crates/geth-cli/src/lib.rs index 5a15601..fe7798f 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -320,6 +320,8 @@ pub enum PubsubCommand { }, Sub { topic: String, + #[arg(long)] + node: Option, }, } @@ -653,7 +655,7 @@ fn request_for_command(command: Command) -> Result { message, node, }, - PubsubCommand::Sub { topic } => ControlRequest::PubsubSub { topic }, + PubsubCommand::Sub { topic, node } => ControlRequest::PubsubSub { topic, node }, }, Command::Pipe { command } => match command { PipeCommand::Listen { name } => ControlRequest::PipeListen { name }, @@ -1477,6 +1479,32 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { } println!("note: {note}"); } + ControlResponse::PubsubRemoteMessages { + peer_node_id, + peer_agent_id, + endpoint_id, + topic, + messages, + allowed, + reason, + note, + } => { + if allowed { + println!("topic: {topic}"); + println!("peer: {peer_node_id}"); + println!("messages: {}", messages.len()); + for message in messages { + println!("{}\t{}", message.published_at.0, message.message); + } + } else { + println!("pubsub subscribe denied by {peer_node_id}"); + } + println!("agent: {peer_agent_id}"); + println!("endpoint: {endpoint_id}"); + println!("allowed: {allowed}"); + println!("reason: {reason}"); + println!("note: {note}"); + } ControlResponse::PipeListening { listener } => { println!("listening pipe: {}", listener.name); println!("id: {}", listener.id); diff --git a/crates/geth-control/src/lib.rs b/crates/geth-control/src/lib.rs index b7c9e28..07824a5 100644 --- a/crates/geth-control/src/lib.rs +++ b/crates/geth-control/src/lib.rs @@ -229,6 +229,7 @@ pub enum ControlRequest { }, PubsubSub { topic: String, + node: Option, }, PipeListen { name: String, @@ -498,6 +499,16 @@ pub enum ControlResponse { messages: Vec, note: String, }, + PubsubRemoteMessages { + peer_node_id: String, + peer_agent_id: String, + endpoint_id: String, + topic: String, + messages: Vec, + allowed: bool, + reason: String, + note: String, + }, PipeListening { listener: PipeListener, }, @@ -602,6 +613,11 @@ pub enum PeerControlRequest { message: String, nonce: String, }, + PubsubSubscribe { + peer_card: PeerCard, + topic: String, + nonce: String, + }, PipeConnect { peer_card: PeerCard, target: String, @@ -723,6 +739,19 @@ pub enum PeerControlResponse { nonce: String, note: String, }, + PubsubSubscribed { + node_id: String, + agent_id: String, + endpoint_id: String, + remote_endpoint_id: String, + topic: String, + messages: Vec, + allowed: bool, + reason: String, + evaluated_ops: usize, + nonce: String, + note: String, + }, PipeConnected { node_id: String, agent_id: String, @@ -987,6 +1016,15 @@ mod tests { request ); + let request = ControlRequest::PubsubSub { + topic: "presence/test".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(), @@ -1005,6 +1043,25 @@ mod tests { response ); + let response = ControlResponse::PubsubRemoteMessages { + peer_node_id: "node:peer".to_owned(), + peer_agent_id: "agent:peer".to_owned(), + endpoint_id: "endpoint:peer".to_owned(), + topic: "presence/test".to_owned(), + messages: vec![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 response = ControlResponse::PipeRemoteConnected { peer_node_id: "node:peer".to_owned(), peer_agent_id: "agent:peer".to_owned(), @@ -1303,6 +1360,50 @@ mod tests { response ); + let request = PeerControlRequest::PubsubSubscribe { + peer_card: PeerCard { + node_id: "node:caller".into(), + agent_id: "agent:caller".into(), + endpoints: Vec::new(), + issued_at: geth_types::UnixMillis(1), + signature: geth_discovery::SignatureMetadata { + namespace: "geth.peer-card.v1@geth.local".to_owned(), + signer: "agent:caller".to_owned(), + public_key: "key".to_owned(), + signature: "sig".to_owned(), + }, + }, + topic: "presence/test".to_owned(), + nonce: "nonce".to_owned(), + }; + assert_eq!( + decode_peer_request(&encode_peer_request(&request).expect("encode")).expect("decode"), + request + ); + + let response = PeerControlResponse::PubsubSubscribed { + 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(), + topic: "presence/test".to_owned(), + messages: vec![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 + ); + let response = PeerControlResponse::PipeConnected { node_id: "node:peer".to_owned(), agent_id: "agent:peer".to_owned(), diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 109020f..fdaa362 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -273,6 +273,10 @@ pub async fn handle_request_async( message, node: Some(peer_node), } => pubsub_publish_to_peer(node, &peer_node, topic, message).await, + ControlRequest::PubsubSub { + topic, + node: Some(peer_node), + } => pubsub_subscribe_from_peer(node, &peer_node, topic).await, ControlRequest::PipeConnect { target, node: Some(peer_node), @@ -560,6 +564,7 @@ async fn peer_ping(node: &LocalNode, peer_node: &str) -> Result Err(NodeError::IrohPeer( @@ -673,6 +678,7 @@ async fn peer_auth_check( | PeerControlResponse::SshRevocationSynced { .. } | PeerControlResponse::KvSynced { .. } | PeerControlResponse::PubsubPublished { .. } + | PeerControlResponse::PubsubSubscribed { .. } | PeerControlResponse::PipeConnected { .. } | PeerControlResponse::DocumentSynced { .. } | PeerControlResponse::DbSynced { .. } => Err(NodeError::IrohPeer( @@ -815,6 +821,7 @@ async fn cas_fetch_from_peer( | PeerControlResponse::SshRevocationSynced { .. } | PeerControlResponse::KvSynced { .. } | PeerControlResponse::PubsubPublished { .. } + | PeerControlResponse::PubsubSubscribed { .. } | PeerControlResponse::PipeConnected { .. } | PeerControlResponse::DocumentSynced { .. } | PeerControlResponse::DbSynced { .. } => Err(NodeError::IrohPeer( @@ -1093,6 +1100,51 @@ async fn pubsub_publish_to_peer( } } +async fn pubsub_subscribe_from_peer( + node: &LocalNode, + peer_node: &str, + topic: String, +) -> Result { + 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, + } + }) + .await?; + match response { + PeerControlResponse::PubsubSubscribed { + node_id, + agent_id, + endpoint_id, + topic: response_topic, + messages, + allowed, + reason, + note, + .. + } if response_topic == topic => Ok(ControlResponse::PubsubRemoteMessages { + peer_node_id: node_id, + peer_agent_id: agent_id, + endpoint_id, + topic: response_topic, + messages, + allowed, + reason, + note, + }), + PeerControlResponse::PubsubSubscribed { .. } => Err(NodeError::IrohPeer( + "peer pubsub subscribe response did not match request".to_owned(), + )), + PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)), + _ => Err(NodeError::IrohPeer( + "peer returned wrong response type to pubsub subscribe".to_owned(), + )), + } +} + async fn pipe_connect_to_peer( node: &LocalNode, peer_node: &str, @@ -1549,6 +1601,10 @@ async fn request_peer_control( nonce: response_nonce, .. } + | PeerControlResponse::PubsubSubscribed { + nonce: response_nonce, + .. + } | PeerControlResponse::PipeConnected { nonce: response_nonce, .. @@ -2082,6 +2138,52 @@ async fn handle_iroh_control_connection( note: "pubsub publish authenticated endpoint/card binding and required pubsub.publish on the remote topic resource; pubsub is lossy".to_owned(), } } + PeerControlRequest::PubsubSubscribe { + peer_card, + topic, + nonce, + } => { + geth_pubsub::validate_topic(&topic)?; + 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.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 messages = if explanation.allowed { + pubsub_messages_for_topic(&node, &topic)? + } else { + Vec::new() + }; + PeerControlResponse::PubsubSubscribed { + 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, + topic, + messages, + allowed: explanation.allowed, + reason: explanation.reason, + evaluated_ops: explanation.evaluated_ops, + nonce, + note: "pubsub subscribe authenticated endpoint/card binding and required pubsub.subscribe on the remote topic resource; pubsub is lossy daemon-lifetime state".to_owned(), + } + } PeerControlRequest::PipeConnect { peer_card, target, @@ -3266,25 +3368,16 @@ pub fn handle_request( Ok(ControlResponse::PubsubPublished { message }) } ControlRequest::PubsubPub { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable), - ControlRequest::PubsubSub { topic } => { + ControlRequest::PubsubSub { topic, node: None } => { geth_pubsub::validate_topic(&topic)?; - let runtime = node - .runtime - .pubsub - .lock() - .map_err(|_| NodeError::RuntimeLockPoisoned)?; - let messages = runtime - .messages - .iter() - .filter(|message| message.topic.as_str() == topic) - .cloned() - .collect(); + let messages = pubsub_messages_for_topic(node, &topic)?; Ok(ControlResponse::PubsubMessages { topic, messages, note: geth_pubsub::pubsub_storage_warning().to_owned(), }) } + ControlRequest::PubsubSub { node: Some(_), .. } => Err(NodeError::IrohEndpointUnavailable), ControlRequest::PipeListen { name } => { geth_pipe::validate_pipe_name(&name)?; let listener = PipeListener { @@ -3428,6 +3521,23 @@ fn record_pubsub_message( Ok(message) } +fn pubsub_messages_for_topic( + node: &LocalNode, + topic: &str, +) -> Result, NodeError> { + let runtime = node + .runtime + .pubsub + .lock() + .map_err(|_| NodeError::RuntimeLockPoisoned)?; + Ok(runtime + .messages + .iter() + .filter(|message| message.topic.as_str() == topic) + .cloned() + .collect()) +} + fn record_pipe_connection( node: &LocalNode, target: String, @@ -4319,6 +4429,29 @@ mod tests { other => panic!("unexpected denied pubsub publish response: {other:?}"), } + let denied_pubsub_subscribe = handle_request_async( + &left, + ControlRequest::PubsubSub { + topic: "presence/test".to_owned(), + node: Some(right_card.node_id.to_string()), + }, + ) + .await + .expect("denied remote pubsub subscribe"); + match denied_pubsub_subscribe { + ControlResponse::PubsubRemoteMessages { + allowed, + messages, + reason, + .. + } => { + assert!(!allowed); + assert!(messages.is_empty()); + assert!(reason.contains("no active direct or group grant")); + } + other => panic!("unexpected denied pubsub subscribe response: {other:?}"), + } + let denied_pipe = handle_request_async( &left, ControlRequest::PipeConnect { @@ -4421,6 +4554,16 @@ mod tests { }, ) .expect("grant left pubsub publish"); + handle_request( + &right, + ControlRequest::AuthGrant { + subject: left.node_id.clone(), + resource: "resource:pubsub:presence/test".to_owned(), + capability: "pubsub.subscribe".to_owned(), + grant_id: Some("grant:left-pubsub-subscribe".to_owned()), + }, + ) + .expect("grant left pubsub subscribe"); handle_request( &right, ControlRequest::AuthGrant { @@ -4578,6 +4721,7 @@ mod tests { &right, ControlRequest::PubsubSub { topic: "presence/test".to_owned(), + node: None, }, ) .expect("right pubsub sub after remote publish"); @@ -4589,6 +4733,32 @@ mod tests { other => panic!("unexpected remote pubsub messages response: {other:?}"), } + let subscribed = handle_request_async( + &left, + ControlRequest::PubsubSub { + topic: "presence/test".to_owned(), + node: Some(right_card.node_id.to_string()), + }, + ) + .await + .expect("allowed remote pubsub subscribe"); + match subscribed { + ControlResponse::PubsubRemoteMessages { + allowed, + messages, + reason, + note, + .. + } => { + assert!(allowed); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].message, "hello"); + assert!(reason.contains("direct grant")); + assert!(note.contains("pubsub.subscribe")); + } + other => panic!("unexpected allowed pubsub subscribe response: {other:?}"), + } + let remote_pipe = handle_request_async( &left, ControlRequest::PipeConnect { diff --git a/crates/geth/tests/bootstrap.rs b/crates/geth/tests/bootstrap.rs index f093707..30c5758 100644 --- a/crates/geth/tests/bootstrap.rs +++ b/crates/geth/tests/bootstrap.rs @@ -1241,6 +1241,7 @@ fn pubsub_pub_sub_uses_lossy_in_memory_runtime() { &node, geth_control::ControlRequest::PubsubSub { topic: "presence/laptop".to_owned(), + node: None, }, ) .expect("subscribe snapshot"); @@ -1263,6 +1264,7 @@ fn pubsub_pub_sub_uses_lossy_in_memory_runtime() { &reopened, geth_control::ControlRequest::PubsubSub { topic: "presence/laptop".to_owned(), + node: None, }, ) .expect("subscribe reopened snapshot"); diff --git a/docs/architecture.md b/docs/architecture.md index ab4338f..d9e8623 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -177,8 +177,10 @@ 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. +its local ring buffer. `geth pubsub sub --node ` can read an +authorized peer's current snapshot for that topic over the same protected path +when the caller has `pubsub.subscribe` on `resource:pubsub:`. Iroh-gossip +replication and private topics are future work. `geth-pipe` currently supports `pipe listen/connect` against a daemon-lifetime registry. `geth pipe connect --node ` sends an authorized remote diff --git a/docs/roadmap.md b/docs/roadmap.md index 93a35c8..0dd3731 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -294,8 +294,12 @@ authorization and durable-state boundaries clear. imported peer over Iroh. - `[x]` Remote pubsub publish requires `pubsub.publish` on `resource:pubsub:`. + - `[x]` `geth pubsub sub --node ` reads an authorized peer + snapshot over Iroh. + - `[x]` Remote pubsub subscribe requires `pubsub.subscribe` on + `resource:pubsub:`. + - `[x]` Tests cover denied and allowed remote pubsub subscribe. - `[ ]` 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