node: Add block command to control socket
Introduces the ability to explicitly block a peer via the node control socket. Previously, the node only exposed follow and unfollow commands. While the underlying policy database schema supported a Block variant, there was no mechanism to trigger this state via the client handle. The new block command: 1. Updates the node's follow policy to Block. 2. Immediately disconnects the peer if a session is active.
This commit is contained in:
parent
4286590fd9
commit
94e0a5128d
|
|
@ -170,6 +170,14 @@ where
|
||||||
return Err(CommandError::Runtime(e));
|
return Err(CommandError::Runtime(e));
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
Command::Block { nid } => match handle.block(nid) {
|
||||||
|
Ok(result) => {
|
||||||
|
CommandResult::updated(result).to_writer(writer)?;
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
return Err(CommandError::Runtime(e));
|
||||||
|
}
|
||||||
|
},
|
||||||
Command::Unfollow { nid } => match handle.unfollow(nid) {
|
Command::Unfollow { nid } => match handle.unfollow(nid) {
|
||||||
Ok(result) => {
|
Ok(result) => {
|
||||||
CommandResult::updated(result).to_writer(writer)?;
|
CommandResult::updated(result).to_writer(writer)?;
|
||||||
|
|
|
||||||
|
|
@ -247,6 +247,12 @@ impl radicle::node::Handle for Handle {
|
||||||
Ok(receiver.recv()??)
|
Ok(receiver.recv()??)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn block(&mut self, id: NodeId) -> Result<bool, Self::Error> {
|
||||||
|
let (sender, receiver) = chan::bounded(1);
|
||||||
|
self.command(service::Command::Block(id, sender))?;
|
||||||
|
receiver.recv().map_err(Error::from)
|
||||||
|
}
|
||||||
|
|
||||||
fn seed(&mut self, id: RepoId, scope: policy::Scope) -> Result<bool, Error> {
|
fn seed(&mut self, id: RepoId, scope: policy::Scope) -> Result<bool, Error> {
|
||||||
let (responder, receiver) = service::command::Responder::oneshot();
|
let (responder, receiver) = service::command::Responder::oneshot();
|
||||||
self.command(service::Command::Seed(id, scope, responder))?;
|
self.command(service::Command::Seed(id, scope, responder))?;
|
||||||
|
|
|
||||||
|
|
@ -18,6 +18,7 @@ pub struct Handle {
|
||||||
pub updates: Arc<Mutex<Vec<(RepoId, PublicKey)>>>,
|
pub updates: Arc<Mutex<Vec<(RepoId, PublicKey)>>>,
|
||||||
pub seeding: Arc<Mutex<HashSet<RepoId>>>,
|
pub seeding: Arc<Mutex<HashSet<RepoId>>>,
|
||||||
pub following: Arc<Mutex<HashSet<NodeId>>>,
|
pub following: Arc<Mutex<HashSet<NodeId>>>,
|
||||||
|
pub blocked: Arc<Mutex<HashSet<NodeId>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl radicle::node::Handle for Handle {
|
impl radicle::node::Handle for Handle {
|
||||||
|
|
@ -85,6 +86,7 @@ impl radicle::node::Handle for Handle {
|
||||||
}
|
}
|
||||||
|
|
||||||
fn follow(&mut self, id: NodeId, _alias: Option<Alias>) -> Result<bool, Self::Error> {
|
fn follow(&mut self, id: NodeId, _alias: Option<Alias>) -> Result<bool, Self::Error> {
|
||||||
|
self.blocked.lock().unwrap().remove(&id);
|
||||||
Ok(self.following.lock().unwrap().insert(id))
|
Ok(self.following.lock().unwrap().insert(id))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -93,7 +95,14 @@ impl radicle::node::Handle for Handle {
|
||||||
}
|
}
|
||||||
|
|
||||||
fn unfollow(&mut self, id: NodeId) -> Result<bool, Self::Error> {
|
fn unfollow(&mut self, id: NodeId) -> Result<bool, Self::Error> {
|
||||||
Ok(self.following.lock().unwrap().remove(&id))
|
let f = self.following.lock().unwrap().remove(&id);
|
||||||
|
let b = self.blocked.lock().unwrap().remove(&id);
|
||||||
|
Ok(f || b)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn block(&mut self, id: NodeId) -> Result<bool, Self::Error> {
|
||||||
|
self.following.lock().unwrap().remove(&id);
|
||||||
|
Ok(self.blocked.lock().unwrap().insert(id))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn announce_refs_for(
|
fn announce_refs_for(
|
||||||
|
|
|
||||||
|
|
@ -856,6 +856,16 @@ where
|
||||||
.expect("Service::command: error unfollowing node");
|
.expect("Service::command: error unfollowing node");
|
||||||
resp.ok(updated).ok();
|
resp.ok(updated).ok();
|
||||||
}
|
}
|
||||||
|
Command::Block(id, resp) => {
|
||||||
|
let updated = self
|
||||||
|
.policies
|
||||||
|
.set_follow_policy(&id, policy::Policy::Block)
|
||||||
|
.expect("Service::command: error blocking node");
|
||||||
|
if updated {
|
||||||
|
self.outbox.disconnect(id, DisconnectReason::Policy);
|
||||||
|
}
|
||||||
|
resp.send(updated).ok();
|
||||||
|
}
|
||||||
Command::AnnounceRefs(id, namespaces, resp) => {
|
Command::AnnounceRefs(id, namespaces, resp) => {
|
||||||
let doc = match self.storage.get(id) {
|
let doc = match self.storage.get(id) {
|
||||||
Ok(Some(doc)) => doc,
|
Ok(Some(doc)) => doc,
|
||||||
|
|
@ -1344,6 +1354,7 @@ where
|
||||||
DisconnectReason::Session(e) => e.severity(),
|
DisconnectReason::Session(e) => e.severity(),
|
||||||
DisconnectReason::Command
|
DisconnectReason::Command
|
||||||
| DisconnectReason::Conflict
|
| DisconnectReason::Conflict
|
||||||
|
| DisconnectReason::Policy
|
||||||
| DisconnectReason::SelfConnection => Severity::Low,
|
| DisconnectReason::SelfConnection => Severity::Low,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
@ -2684,6 +2695,8 @@ pub enum DisconnectReason {
|
||||||
Conflict,
|
Conflict,
|
||||||
/// Connection to self.
|
/// Connection to self.
|
||||||
SelfConnection,
|
SelfConnection,
|
||||||
|
/// Peer is blocked by policy
|
||||||
|
Policy,
|
||||||
/// User requested disconnect
|
/// User requested disconnect
|
||||||
Command,
|
Command,
|
||||||
}
|
}
|
||||||
|
|
@ -2712,6 +2725,7 @@ impl fmt::Display for DisconnectReason {
|
||||||
Self::Command => write!(f, "command"),
|
Self::Command => write!(f, "command"),
|
||||||
Self::SelfConnection => write!(f, "self-connection"),
|
Self::SelfConnection => write!(f, "self-connection"),
|
||||||
Self::Conflict => write!(f, "conflict"),
|
Self::Conflict => write!(f, "conflict"),
|
||||||
|
Self::Policy => write!(f, "policy"),
|
||||||
Self::Session(err) => write!(f, "{err}"),
|
Self::Session(err) => write!(f, "{err}"),
|
||||||
Self::Fetch(err) => write!(f, "fetch: {err}"),
|
Self::Fetch(err) => write!(f, "fetch: {err}"),
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -99,6 +99,8 @@ pub enum Command {
|
||||||
Follow(NodeId, Option<Alias>, Responder<bool>),
|
Follow(NodeId, Option<Alias>, Responder<bool>),
|
||||||
/// Unfollow the given node.
|
/// Unfollow the given node.
|
||||||
Unfollow(NodeId, Responder<bool>),
|
Unfollow(NodeId, Responder<bool>),
|
||||||
|
/// Block the given node.
|
||||||
|
Block(NodeId, Sender<bool>),
|
||||||
/// Query the internal service state.
|
/// Query the internal service state.
|
||||||
QueryState(Arc<QueryState>, Sender<Result<()>>),
|
QueryState(Arc<QueryState>, Sender<Result<()>>),
|
||||||
}
|
}
|
||||||
|
|
@ -194,6 +196,7 @@ impl fmt::Debug for Command {
|
||||||
Self::Unseed(id, _) => write!(f, "Unseed({id})"),
|
Self::Unseed(id, _) => write!(f, "Unseed({id})"),
|
||||||
Self::Follow(id, _, _) => write!(f, "Follow({id})"),
|
Self::Follow(id, _, _) => write!(f, "Follow({id})"),
|
||||||
Self::Unfollow(id, _) => write!(f, "Unfollow({id})"),
|
Self::Unfollow(id, _) => write!(f, "Unfollow({id})"),
|
||||||
|
Self::Block(id, _) => write!(f, "Block({id})"),
|
||||||
Self::QueryState { .. } => write!(f, "QueryState(..)"),
|
Self::QueryState { .. } => write!(f, "QueryState(..)"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||||
|
|
||||||
### Added
|
### Added
|
||||||
|
|
||||||
|
- `radicle::node::Handle` added a `block` method to allow setting the follow
|
||||||
|
policy to `Policy::Block`.
|
||||||
- `radicle::node::Handle::announce_refs_for` now allows specifying for which
|
- `radicle::node::Handle::announce_refs_for` now allows specifying for which
|
||||||
namespaces changes should be announced. A corresponding enum variant
|
namespaces changes should be announced. A corresponding enum variant
|
||||||
`radicle::node::Command::AnnounceRefsFor` is added.
|
`radicle::node::Command::AnnounceRefsFor` is added.
|
||||||
|
|
|
||||||
|
|
@ -907,6 +907,8 @@ pub trait Handle: Clone + Sync + Send {
|
||||||
fn seed(&mut self, id: RepoId, scope: policy::Scope) -> Result<bool, Self::Error>;
|
fn seed(&mut self, id: RepoId, scope: policy::Scope) -> Result<bool, Self::Error>;
|
||||||
/// Start following the given peer.
|
/// Start following the given peer.
|
||||||
fn follow(&mut self, id: NodeId, alias: Option<Alias>) -> Result<bool, Self::Error>;
|
fn follow(&mut self, id: NodeId, alias: Option<Alias>) -> Result<bool, Self::Error>;
|
||||||
|
/// Set the following policy to block for the given peer.
|
||||||
|
fn block(&mut self, id: NodeId) -> Result<bool, Self::Error>;
|
||||||
/// Un-seed the given repo and delete it from storage.
|
/// Un-seed the given repo and delete it from storage.
|
||||||
fn unseed(&mut self, id: RepoId) -> Result<bool, Self::Error>;
|
fn unseed(&mut self, id: RepoId) -> Result<bool, Self::Error>;
|
||||||
/// Unfollow the given peer.
|
/// Unfollow the given peer.
|
||||||
|
|
@ -1191,6 +1193,13 @@ impl Handle for Node {
|
||||||
Ok(response.updated)
|
Ok(response.updated)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn block(&mut self, nid: NodeId) -> Result<bool, Error> {
|
||||||
|
let mut lines = self.call::<Success>(Command::Block { nid }, DEFAULT_TIMEOUT)?;
|
||||||
|
let response = lines.next().ok_or(Error::EmptyResponse)??;
|
||||||
|
|
||||||
|
Ok(response.updated)
|
||||||
|
}
|
||||||
|
|
||||||
fn seed(&mut self, rid: RepoId, scope: policy::Scope) -> Result<bool, Error> {
|
fn seed(&mut self, rid: RepoId, scope: policy::Scope) -> Result<bool, Error> {
|
||||||
let mut lines = self.call::<Success>(Command::Seed { rid, scope }, DEFAULT_TIMEOUT)?;
|
let mut lines = self.call::<Success>(Command::Seed { rid, scope }, DEFAULT_TIMEOUT)?;
|
||||||
let response = lines.next().ok_or(Error::EmptyResponse)??;
|
let response = lines.next().ok_or(Error::EmptyResponse)??;
|
||||||
|
|
|
||||||
|
|
@ -120,6 +120,10 @@ pub enum Command {
|
||||||
#[serde(rename_all = "camelCase")]
|
#[serde(rename_all = "camelCase")]
|
||||||
Unfollow { nid: NodeId },
|
Unfollow { nid: NodeId },
|
||||||
|
|
||||||
|
/// Block the given node.
|
||||||
|
#[serde(rename_all = "camelCase")]
|
||||||
|
Block { nid: NodeId },
|
||||||
|
|
||||||
/// Get the node's status.
|
/// Get the node's status.
|
||||||
Status,
|
Status,
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -119,14 +119,15 @@ impl Store<Write> {
|
||||||
/// Follow a node.
|
/// Follow a node.
|
||||||
pub fn follow(&mut self, id: &NodeId, alias: Option<&Alias>) -> Result<bool, Error> {
|
pub fn follow(&mut self, id: &NodeId, alias: Option<&Alias>) -> Result<bool, Error> {
|
||||||
let mut stmt = self.db.prepare(
|
let mut stmt = self.db.prepare(
|
||||||
"INSERT INTO `following` (id, alias)
|
"INSERT INTO `following` (id, alias, policy)
|
||||||
VALUES (?1, ?2)
|
VALUES (?1, ?2, ?3)
|
||||||
ON CONFLICT DO UPDATE
|
ON CONFLICT (id) DO UPDATE
|
||||||
SET alias = ?2 WHERE alias != ?2",
|
SET alias = ?2, policy = ?3 WHERE alias != ?2 OR policy != ?3",
|
||||||
)?;
|
)?;
|
||||||
|
|
||||||
stmt.bind((1, id))?;
|
stmt.bind((1, id))?;
|
||||||
stmt.bind((2, alias.map_or("", |alias| alias.as_str())))?;
|
stmt.bind((2, alias.map_or("", |alias| alias.as_str())))?;
|
||||||
|
stmt.bind((3, Policy::Allow))?;
|
||||||
stmt.next()?;
|
stmt.next()?;
|
||||||
|
|
||||||
Ok(self.db.change_count() > 0)
|
Ok(self.db.change_count() > 0)
|
||||||
|
|
@ -248,6 +249,18 @@ impl<T> Store<T> {
|
||||||
))
|
))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Returns `true` if there is a follow policy for the given node, and that
|
||||||
|
/// policy is [`Policy::Block`].
|
||||||
|
pub fn is_blocked(&self, id: &NodeId) -> Result<bool, Error> {
|
||||||
|
Ok(matches!(
|
||||||
|
self.follow_policy(id)?,
|
||||||
|
Some(FollowPolicy {
|
||||||
|
policy: Policy::Block,
|
||||||
|
..
|
||||||
|
})
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
/// Get a node's follow policy.
|
/// Get a node's follow policy.
|
||||||
pub fn follow_policy(&self, id: &NodeId) -> Result<Option<FollowPolicy>, Error> {
|
pub fn follow_policy(&self, id: &NodeId) -> Result<Option<FollowPolicy>, Error> {
|
||||||
let mut stmt = self
|
let mut stmt = self
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue