sasa/wireguard-admin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix systemd restart support

Browse Source
This commit is contained in:
sasa
2026-10-27 22:44:06 +02:00
parent fe7e3d61c2
commit 4b53aad20c
7 changed files with 536 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/service/dbus.rs
+428 -107
View File
@@ -12,6 +12,7 @@ use std::task::{Context, Poll, Wake, Waker};
use std::thread;
use std::time::Instant;
use zbus::connection::Builder;
use zbus::export::futures_core::Stream;
use zbus::message::Type;
use zbus::{Connection, MatchRule, MessageStream, Proxy};
@@ -26,6 +27,7 @@ const SYSTEMD_PATH: &str = "/org/freedesktop/systemd1";
const SYSTEMD_MANAGER: &str = "org.freedesktop.systemd1.Manager";
const DBUS_NAME: &str = "org.freedesktop.DBus";
const DBUS_INTERFACE: &str = "org.freedesktop.DBus";
const SYSTEMD_PRIVATE_ADDRESS: &str = "unix:path=/run/systemd/private";
/// Resolve the configured unit's primary ID, request a restart of exactly
/// `configured_unit`, and wait for its matching systemd `JobRemoved` result.
@@ -175,11 +177,17 @@ fn execution(outcome: RestartOutcome, submitted: bool, phase: RestartPhase) -> R
#[derive(Default)]
struct SystemdBus {
connection: Option<Connection>,
private_peer: bool,
manager: Option<Proxy<'static>>,
job_events: Option<zbus::proxy::SignalStream<'static>>,
job_events: Option<JobSignalStream>,
owner_events: Option<zbus::MessageStream>,
}
enum JobSignalStream {
Bus(Box<zbus::proxy::SignalStream<'static>>),
Peer(Box<MessageStream>),
}
/// The small systemd unit API used by the shared resolver. Keeping method
/// selection and canonical-ID lookup here lets synthetic tests exercise the
/// same resolution flow as the production adapter.
@@ -223,12 +231,11 @@ impl UnitLoadApi for SystemdBus {
configured_unit: &str,
deadline: RestartDeadline,
) -> Result<Self::UnitPath, RestartFailure> {
let manager = self
.manager
.as_ref()
.ok_or(RestartFailure::UnitResolutionFailed)?;
drive(manager.call("LoadUnit", &(configured_unit,)), deadline)?
.map_err(map_unit_resolution_error)
self.call_manager("LoadUnit", &(configured_unit,), deadline)
.map_err(map_unit_resolution_failure)?
.body()
.deserialize()
.map_err(|_| RestartFailure::UnitResolutionFailed)
}
fn load_state(
@@ -236,9 +243,7 @@ impl UnitLoadApi for SystemdBus {
unit_path: &Self::UnitPath,
deadline: RestartDeadline,
) -> Result<String, RestartFailure> {
let unit = self.unit_proxy(unit_path, deadline)?;
drive(unit.get_property::<String>("LoadState"), deadline)?
.map_err(map_unit_resolution_error)
self.unit_property(unit_path, "LoadState", deadline)
}
fn canonical_id(
@@ -246,87 +251,184 @@ impl UnitLoadApi for SystemdBus {
unit_path: &Self::UnitPath,
deadline: RestartDeadline,
) -> Result<String, RestartFailure> {
let unit = self.unit_proxy(unit_path, deadline)?;
drive(unit.get_property::<String>("Id"), deadline)?.map_err(map_unit_resolution_error)
self.unit_property(unit_path, "Id", deadline)
}
}
impl SystemdBus {
fn unit_proxy(
fn call_manager<B: serde::ser::Serialize + zbus::zvariant::DynamicType>(
&self,
unit_path: &zbus::zvariant::OwnedObjectPath,
method: &'static str,
body: &B,
deadline: RestartDeadline,
) -> Result<Proxy<'static>, RestartFailure> {
) -> Result<zbus::message::Message, RestartFailure> {
let connection = self
.connection
.as_ref()
.ok_or(RestartFailure::BusDisconnected)?;
let destination = manager_destination(self.private_peer);
drive(
Proxy::new_owned(
connection.clone(),
SYSTEMD_NAME,
unit_path.clone(),
"org.freedesktop.systemd1.Unit",
connection.call_method(
destination,
SYSTEMD_PATH,
Some(SYSTEMD_MANAGER),
method,
body,
),
deadline,
)?
.map_err(map_unit_resolution_error)
.map_err(map_bus_error)
}
fn unit_property(
&self,
unit_path: &zbus::zvariant::OwnedObjectPath,
property: &'static str,
deadline: RestartDeadline,
) -> Result<String, RestartFailure> {
let connection = self
.connection
.as_ref()
.ok_or(RestartFailure::BusDisconnected)?;
if !self.private_peer {
let unit = drive(
Proxy::new_owned(
connection.clone(),
SYSTEMD_NAME,
unit_path.clone(),
"org.freedesktop.systemd1.Unit",
),
deadline,
)?
.map_err(map_unit_resolution_error)?;
return drive(unit.get_property::<String>(property), deadline)?
.map_err(map_unit_resolution_error);
}
let message = drive(
connection.call_method(
Option::<&str>::None,
unit_path.clone(),
Some("org.freedesktop.DBus.Properties"),
"Get",
&("org.freedesktop.systemd1.Unit", property),
),
deadline,
)?
.map_err(map_unit_resolution_error)?;
let value: zbus::zvariant::OwnedValue = message
.body()
.deserialize()
.map_err(|_| RestartFailure::UnitResolutionFailed)?;
String::try_from(value).map_err(|_| RestartFailure::UnitResolutionFailed)
}
}
fn manager_destination(private_peer: bool) -> Option<&'static str> {
(!private_peer).then_some(SYSTEMD_NAME)
}
fn owner_monitoring_enabled(private_peer: bool) -> bool {
!private_peer
}
fn select_connection<T>(
system_bus: Result<T, (RestartFailure, bool)>,
private_peer: impl FnOnce() -> Result<T, RestartFailure>,
) -> Result<(T, bool), RestartFailure> {
match system_bus {
Ok(connection) => Ok((connection, false)),
// Once private fallback is attempted, keep its typed failure: it is
// the selected transport's actionable result.
Err((_failure, true)) => private_peer().map(|connection| (connection, true)),
Err((failure, false)) => Err(failure),
}
}
impl RestartTransport for SystemdBus {
fn connect(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure> {
let connection = drive(Connection::system(), deadline)?.map_err(map_bus_error)?;
let manager = drive(
Proxy::new_owned(
connection.clone(),
SYSTEMD_NAME,
SYSTEMD_PATH,
SYSTEMD_MANAGER,
),
deadline,
)?
.map_err(map_bus_error)?;
self.manager = Some(manager);
let bus_connection = connect_system_bus(deadline);
let ((connection, manager), private_peer) = select_connection(bus_connection, || {
let connection = drive(
Builder::address(SYSTEMD_PRIVATE_ADDRESS)
.map_err(map_bus_error)?
.p2p()
.auth_mechanism(zbus::connection::AuthMechanism::External)
.build(),
deadline,
)?
.map_err(map_private_builder_error)?;
let uid = drive(connection.peer_creds(), deadline)?
.map_err(map_peer_creds_error)?
.unix_user_id();
if !is_root_peer_uid(uid) {
return Err(RestartFailure::PermissionDenied);
}
Ok((connection, None))
})?;
self.private_peer = private_peer;
self.manager = manager;
self.connection = Some(connection);
Ok(())
}
fn subscribe(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure> {
let manager = self
.manager
let connection = self
.connection
.as_ref()
.ok_or(RestartFailure::SubscriptionFailed)?;
let job_events = drive(manager.receive_signal("JobRemoved"), deadline)?
.map_err(|_| RestartFailure::SubscriptionFailed)?;
let job_events = if self.private_peer {
let rule = MatchRule::builder()
.msg_type(Type::Signal)
.path(SYSTEMD_PATH)
.map_err(|_| RestartFailure::SubscriptionFailed)?
.interface(SYSTEMD_MANAGER)
.map_err(|_| RestartFailure::SubscriptionFailed)?
.member("JobRemoved")
.map_err(|_| RestartFailure::SubscriptionFailed)?
.build();
JobSignalStream::Peer(Box::new(
drive(
MessageStream::for_match_rule(rule, connection, Some(8)),
deadline,
)?
.map_err(|_| RestartFailure::SubscriptionFailed)?,
))
} else {
let manager = self
.manager
.as_ref()
.ok_or(RestartFailure::SubscriptionFailed)?;
JobSignalStream::Bus(Box::new(
drive(manager.receive_signal("JobRemoved"), deadline)?
.map_err(|_| RestartFailure::SubscriptionFailed)?,
))
};
let owner_rule = MatchRule::builder()
.msg_type(Type::Signal)
.sender(DBUS_NAME)
.map_err(|_| RestartFailure::SubscriptionFailed)?
.interface(DBUS_INTERFACE)
.map_err(|_| RestartFailure::SubscriptionFailed)?
.member("NameOwnerChanged")
.map_err(|_| RestartFailure::SubscriptionFailed)?
.add_arg(SYSTEMD_NAME)
.map_err(|_| RestartFailure::SubscriptionFailed)?
.build();
let owner_events = drive(
MessageStream::for_match_rule(
owner_rule,
self.connection
.as_ref()
.ok_or(RestartFailure::SubscriptionFailed)?,
Some(1),
),
deadline,
)?
.map_err(|_| RestartFailure::SubscriptionFailed)?;
// Both signal matches are installed before Subscribe and RestartUnit.
// Peer mode uses zbus's local p2p match and must not call the bus daemon.
// The bus mode additionally watches owner changes before Subscribe.
self.job_events = Some(job_events);
self.owner_events = Some(owner_events);
drive(manager.call::<_, _, ()>("Subscribe", &()), deadline)?.map_err(map_bus_error)
if owner_monitoring_enabled(self.private_peer) {
let owner_rule = MatchRule::builder()
.msg_type(Type::Signal)
.sender(DBUS_NAME)
.map_err(|_| RestartFailure::SubscriptionFailed)?
.interface(DBUS_INTERFACE)
.map_err(|_| RestartFailure::SubscriptionFailed)?
.member("NameOwnerChanged")
.map_err(|_| RestartFailure::SubscriptionFailed)?
.add_arg(SYSTEMD_NAME)
.map_err(|_| RestartFailure::SubscriptionFailed)?
.build();
self.owner_events = Some(
drive(
MessageStream::for_match_rule(owner_rule, connection, Some(1)),
deadline,
)?
.map_err(|_| RestartFailure::SubscriptionFailed)?,
);
}
self.call_manager("Subscribe", &(), deadline)?;
Ok(())
}
fn resolve_primary_unit(
@@ -342,45 +444,54 @@ impl RestartTransport for SystemdBus {
unit: &str,
deadline: RestartDeadline,
) -> Result<String, RestartFailure> {
let manager = self.manager.as_ref().ok_or(RestartFailure::RequestFailed)?;
let connection = self
.connection
.as_ref()
.ok_or(RestartFailure::BusDisconnected)?;
let owner_events = self
.owner_events
.as_mut()
.ok_or(RestartFailure::BusDisconnected)?;
let request_body = (unit, "replace");
let request = manager.call("RestartUnit", &request_body);
let job_path: zbus::zvariant::OwnedObjectPath = drive(
async {
let mut request = pin!(request);
let mut closed = pin!(connection.closed());
poll_fn(|cx| {
if connection.is_closed() {
return Poll::Ready(Err(RestartFailure::BusDisconnected));
}
if let Poll::Ready(event) = Stream::poll_next(Pin::new(&mut *owner_events), cx)
{
return Poll::Ready(match event {
None => Err(RestartFailure::BusDisconnected),
Some(Err(error)) => Err(map_bus_error(error)),
Some(Ok(_)) => Err(RestartFailure::ManagerOwnerChanged),
});
}
if let Poll::Ready(result) = Future::poll(request.as_mut(), cx) {
return Poll::Ready(result.map_err(map_bus_error));
}
if Future::poll(closed.as_mut(), cx).is_ready() {
return Poll::Ready(Err(RestartFailure::BusDisconnected));
}
Poll::Pending
})
.await
},
deadline,
)??;
let job_path: zbus::zvariant::OwnedObjectPath = if self.private_peer {
self.call_manager("RestartUnit", &request_body, deadline)?
.body()
.deserialize()
.map_err(|_| RestartFailure::RequestFailed)?
} else {
let manager = self.manager.as_ref().ok_or(RestartFailure::RequestFailed)?;
let owner_events = self
.owner_events
.as_mut()
.ok_or(RestartFailure::BusDisconnected)?;
let request =
manager.call::<_, _, zbus::zvariant::OwnedObjectPath>("RestartUnit", &request_body);
drive(
async {
let mut request = pin!(request);
let mut closed = pin!(connection.closed());
poll_fn(|cx| {
if connection.is_closed() {
return Poll::Ready(Err(RestartFailure::BusDisconnected));
}
if let Poll::Ready(event) =
Stream::poll_next(Pin::new(&mut *owner_events), cx)
{
return Poll::Ready(match event {
None => Err(RestartFailure::BusDisconnected),
Some(Err(error)) => Err(map_bus_error(error)),
Some(Ok(_)) => Err(RestartFailure::ManagerOwnerChanged),
});
}
if let Poll::Ready(result) = Future::poll(request.as_mut(), cx) {
return Poll::Ready(result.map_err(map_bus_error));
}
if Future::poll(closed.as_mut(), cx).is_ready() {
return Poll::Ready(Err(RestartFailure::BusDisconnected));
}
Poll::Pending
})
.await
},
deadline,
)??
};
Ok(job_path.to_string())
}
@@ -393,10 +504,7 @@ impl RestartTransport for SystemdBus {
.job_events
.as_mut()
.ok_or(RestartFailure::BusDisconnected)?;
let owner_events = self
.owner_events
.as_mut()
.ok_or(RestartFailure::BusDisconnected)?;
let mut owner_events = self.owner_events.as_mut();
drive(
async {
let mut closed = pin!(connection.closed());
@@ -404,8 +512,8 @@ impl RestartTransport for SystemdBus {
if connection.is_closed() {
return Poll::Ready(Ok(ServiceEvent::Disconnected));
}
if let Poll::Ready(message) =
Stream::poll_next(Pin::new(&mut *owner_events), cx)
if let Some(owner_events) = owner_events.as_mut()
&& let Poll::Ready(message) = Stream::poll_next(Pin::new(owner_events), cx)
{
return Poll::Ready(match message {
None => Ok(ServiceEvent::Disconnected),
@@ -427,11 +535,17 @@ impl RestartTransport for SystemdBus {
}
});
}
if let Poll::Ready(message) = Stream::poll_next(Pin::new(&mut *job_events), cx)
{
let message = match job_events {
JobSignalStream::Bus(stream) => {
Stream::poll_next(Pin::new(stream), cx).map(|message| message.map(Ok))
}
JobSignalStream::Peer(stream) => Stream::poll_next(Pin::new(stream), cx),
};
if let Poll::Ready(message) = message {
return Poll::Ready(match message {
None => Ok(ServiceEvent::Disconnected),
Some(message) => {
Some(Err(error)) => Err(map_bus_error(error)),
Some(Ok(message)) => {
let decoded: Result<
(u32, zbus::zvariant::OwnedObjectPath, String, String),
_,
@@ -459,6 +573,34 @@ impl RestartTransport for SystemdBus {
}
}
fn connect_system_bus(
deadline: RestartDeadline,
) -> Result<(Connection, Option<Proxy<'static>>), (RestartFailure, bool)> {
let connection = match drive(Connection::system(), deadline) {
Err(failure) => return Err((failure, false)),
Ok(Err(error)) => {
return Err((map_connect_error(&error), private_fallback_allowed(&error)));
}
Ok(Ok(connection)) => connection,
};
let manager = match drive(
Proxy::new_owned(
connection.clone(),
SYSTEMD_NAME,
SYSTEMD_PATH,
SYSTEMD_MANAGER,
),
deadline,
) {
Err(failure) => return Err((failure, false)),
Ok(Err(error)) => {
return Err((map_connect_error(&error), private_fallback_allowed(&error)));
}
Ok(Ok(manager)) => manager,
};
Ok((connection, Some(manager)))
}
fn unit_is_loaded(load_state: &str) -> bool {
load_state == "loaded"
}
@@ -483,11 +625,61 @@ fn map_bus_error(error: zbus::Error) -> RestartFailure {
{
RestartFailure::PermissionDenied
}
zbus::Error::InputOutput(_) | zbus::Error::Handshake(_) => RestartFailure::BusDisconnected,
zbus::Error::InputOutput(error) if error.kind() == std::io::ErrorKind::PermissionDenied => {
RestartFailure::PermissionDenied
}
zbus::Error::InputOutput(_) => RestartFailure::BusDisconnected,
_ => RestartFailure::RequestFailed,
}
}
fn map_connect_error(error: &zbus::Error) -> RestartFailure {
match error {
zbus::Error::Connection(error, _) | zbus::Error::InputOutput(error)
if error.kind() == std::io::ErrorKind::PermissionDenied =>
{
RestartFailure::PermissionDenied
}
zbus::Error::Connection(_, _) | zbus::Error::InputOutput(_) => {
RestartFailure::BusDisconnected
}
// A handshake rejection is an authentication/protocol failure, not an
// unavailable transport. Never bypass it via the private socket.
_ => RestartFailure::RequestFailed,
}
}
fn map_private_builder_error(error: zbus::Error) -> RestartFailure {
map_connect_error(&error)
}
fn private_fallback_allowed(error: &zbus::Error) -> bool {
match error {
zbus::Error::Connection(io, _) | zbus::Error::InputOutput(io) => matches!(
io.kind(),
std::io::ErrorKind::NotFound
| std::io::ErrorKind::ConnectionRefused
| std::io::ErrorKind::ConnectionReset
| std::io::ErrorKind::ConnectionAborted
| std::io::ErrorKind::NotConnected
| std::io::ErrorKind::BrokenPipe
),
_ => false,
}
}
fn is_root_peer_uid(uid: Option<u32>) -> bool {
uid == Some(0)
}
fn map_peer_creds_error(error: std::io::Error) -> RestartFailure {
if error.kind() == std::io::ErrorKind::PermissionDenied {
RestartFailure::PermissionDenied
} else {
RestartFailure::BusDisconnected
}
}
fn map_unit_resolution_error(error: zbus::Error) -> RestartFailure {
match map_bus_error(error) {
RestartFailure::PermissionDenied => RestartFailure::PermissionDenied,
@@ -496,6 +688,13 @@ fn map_unit_resolution_error(error: zbus::Error) -> RestartFailure {
}
}
fn map_unit_resolution_failure(failure: RestartFailure) -> RestartFailure {
match failure {
RestartFailure::RequestFailed => RestartFailure::UnitResolutionFailed,
other => other,
}
}
struct ThreadWake(thread::Thread);
impl Wake for ThreadWake {
@@ -621,6 +820,109 @@ mod tests {
RestartDeadline::new(Instant::now(), Duration::from_secs(2))
}
#[test]
fn transport_selection_falls_back_only_before_restart_setup() {
let (selected, private) =
select_connection(Ok("bus"), || Ok("private")).expect("available bus transport");
assert_eq!(selected, "bus");
assert!(!private);
let (selected, private) = select_connection(
Err::<&str, _>((RestartFailure::BusDisconnected, true)),
|| Ok("private"),
)
.expect("private peer fallback");
assert_eq!(selected, "private");
assert!(private);
assert_eq!(manager_destination(true), None);
assert_eq!(manager_destination(false), Some(SYSTEMD_NAME));
assert!(!owner_monitoring_enabled(true));
assert!(owner_monitoring_enabled(false));
let private_denied =
select_connection::<&str>(Err((RestartFailure::BusDisconnected, true)), || {
Err(RestartFailure::PermissionDenied)
});
assert_eq!(private_denied, Err(RestartFailure::PermissionDenied));
let private_timeout =
select_connection::<&str>(Err((RestartFailure::BusDisconnected, true)), || {
Err(RestartFailure::WaitTimedOutMayStillComplete)
});
assert_eq!(
private_timeout,
Err(RestartFailure::WaitTimedOutMayStillComplete)
);
let mut private_attempted = false;
let denied_bus =
select_connection::<&str>(Err((RestartFailure::PermissionDenied, false)), || {
private_attempted = true;
Ok("private")
});
assert_eq!(denied_bus, Err(RestartFailure::PermissionDenied));
assert!(!private_attempted);
let timeout_bus = select_connection::<&str>(
Err((RestartFailure::WaitTimedOutMayStillComplete, false)),
|| {
private_attempted = true;
Ok("private")
},
);
assert_eq!(
timeout_bus,
Err(RestartFailure::WaitTimedOutMayStillComplete)
);
assert!(!private_attempted);
}
#[test]
fn private_fallback_is_limited_to_missing_or_refused_bus_endpoint() {
let unavailable = zbus::Error::Connection(
Arc::new(std::io::Error::from(std::io::ErrorKind::ConnectionRefused)),
zbus::Address::try_from("unix:path=/run/dbus/system_bus_socket")
.expect("valid synthetic bus address"),
);
assert!(private_fallback_allowed(&unavailable));
assert_eq!(
map_connect_error(&unavailable),
RestartFailure::BusDisconnected
);
assert!(private_fallback_allowed(&zbus::Error::InputOutput(
Arc::new(std::io::Error::from(std::io::ErrorKind::ConnectionReset))
)));
let denied = zbus::Error::Connection(
Arc::new(std::io::Error::from(std::io::ErrorKind::PermissionDenied)),
zbus::Address::try_from("unix:path=/run/dbus/system_bus_socket")
.expect("valid synthetic bus address"),
);
assert!(!private_fallback_allowed(&denied));
assert_eq!(map_connect_error(&denied), RestartFailure::PermissionDenied);
assert_eq!(
map_private_builder_error(denied),
RestartFailure::PermissionDenied
);
assert!(!private_fallback_allowed(&zbus::Error::InputOutput(
Arc::new(std::io::Error::from(std::io::ErrorKind::PermissionDenied))
)));
assert!(!private_fallback_allowed(&zbus::Error::Handshake(
"synthetic authentication rejection".into()
)));
assert!(!is_root_peer_uid(None));
assert!(!is_root_peer_uid(Some(1000)));
assert!(is_root_peer_uid(Some(0)));
assert_eq!(
map_peer_creds_error(std::io::Error::from(std::io::ErrorKind::PermissionDenied)),
RestartFailure::PermissionDenied
);
assert_eq!(
map_peer_creds_error(std::io::Error::from(std::io::ErrorKind::ConnectionReset)),
RestartFailure::BusDisconnected
);
}
fn removed(path: &str, unit: &str, result: JobResult) -> ServiceEvent {
ServiceEvent::JobRemoved {
job_path: path.to_owned(),
@@ -747,6 +1049,21 @@ mod tests {
}
}
#[test]
fn unit_resolution_keeps_permission_timeout_and_disconnect_categories() {
assert_eq!(
map_unit_resolution_failure(RestartFailure::RequestFailed),
RestartFailure::UnitResolutionFailed
);
for failure in [
RestartFailure::PermissionDenied,
RestartFailure::WaitTimedOutMayStillComplete,
RestartFailure::BusDisconnected,
] {
assert_eq!(map_unit_resolution_failure(failure), failure);
}
}
struct FakeUnitLoadApi {
calls: Vec<String>,
configured_unit_seen: Option<String>,
@@ -903,6 +1220,10 @@ mod tests {
);
assert_eq!(execution.phase, RestartPhase::RequestRestart);
assert!(execution.submitted);
assert_eq!(
fake.calls.iter().filter(|call| **call == "restart").count(),
1
);
}
#[test]