Add runtime observation and systemd restart adapter

src/domain.rs

@@ -256,7 +256,7 @@ impl ClientStates {
     }
 
     /// Record an applied state only when the verified observation matches saved clients.
-    pub fn mark_applied_after_verification(
+    pub(crate) fn mark_applied_after_verification(
         &mut self,
         observed: Snapshot,
     ) -> Result<(), DomainError> {

src/lib.rs

@@ -3,6 +3,8 @@ pub mod backup;
 pub mod config;
 pub mod domain;
 pub mod persistence;
+pub mod runtime;
+pub mod service;
 pub mod startup;
 pub mod vault;
 pub mod wireguard;

src/runtime.rs

@@ -0,0 +1,674 @@
+//! Public, read-only runtime observations and pure saved-peer verification.
+//!
+//! The transport performs bounded read-only Linux access; comparison remains
+//! pure and accepts caller-supplied observations.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use ipnet::IpNet;
+use time::OffsetDateTime;
+
+use crate::domain::PublicKey;
+use crate::wireguard::{ConfigDocument, DocumentIssue, PeerIssue};
+
+#[allow(dead_code)] // Integrated by the M5 orchestration step.
+pub(crate) mod netlink;
+#[allow(dead_code)] // Consumed by the M5 application integration step.
+pub(crate) mod transport;
+
+/// A value that could not be observed is distinct from a valid zero or empty value.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum Observed<T> {
+    Available(T),
+    Unavailable,
+}
+
+/// Current statistics for a peer, from the server interface's perspective.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct RuntimePeer {
+    pub allowed_ips: Observed<Vec<IpNet>>,
+    /// `Available(None)` means WireGuard has never recorded a handshake.
+    pub last_handshake: Observed<Option<OffsetDateTime>>,
+    /// These are current counters and may reset when the interface restarts.
+    pub received_bytes: Observed<u64>,
+    pub transmitted_bytes: Observed<u64>,
+}
+
+/// Whether the peer map is exhaustive enough to establish absence and equality.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum PeerSetCompleteness {
+    Unknown,
+    Unavailable,
+    Partial,
+    Complete,
+}
+
+/// Runtime peers keyed by their public key, with duplicate identities retained
+/// as an explicit ambiguity signal rather than silently overwritten.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct RuntimePeerSet {
+    pub peers: BTreeMap<PublicKey, RuntimePeer>,
+    pub duplicate_public_keys: BTreeSet<PublicKey>,
+    pub completeness: PeerSetCompleteness,
+}
+
+impl RuntimePeerSet {
+    /// Collect records while preserving evidence that the adapter returned a
+    /// duplicate public key. The first record remains available for diagnostics.
+    pub fn from_records(
+        completeness: PeerSetCompleteness,
+        records: impl IntoIterator<Item = (PublicKey, RuntimePeer)>,
+    ) -> Self {
+        let mut peers = BTreeMap::new();
+        let mut duplicate_public_keys = BTreeSet::new();
+        for (identity, peer) in records {
+            if let std::collections::btree_map::Entry::Vacant(entry) = peers.entry(identity) {
+                entry.insert(peer);
+            } else {
+                duplicate_public_keys.insert(identity);
+            }
+        }
+        Self {
+            peers,
+            duplicate_public_keys,
+            completeness,
+        }
+    }
+}
+
+/// Interface presence and peer observation are independent facts.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum InterfaceObservation {
+    Unknown,
+    Unavailable,
+    Missing,
+    Present {
+        /// Interface returned by the runtime adapter.
+        name: String,
+        peers: RuntimePeerSet,
+    },
+}
+
+/// One read-only sample. A missing timestamp means the sample time is unknown.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct RuntimeObservation {
+    pub observed_at: Option<OffsetDateTime>,
+    pub interface: InterfaceObservation,
+}
+
+impl RuntimeObservation {
+    pub const fn unknown() -> Self {
+        Self {
+            observed_at: None,
+            interface: InterfaceObservation::Unknown,
+        }
+    }
+}
+
+/// One saved peer's complete technical routes, independent of alias metadata
+/// and the narrower `domain::Client` address representation.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct SavedPeer {
+    pub identity: PublicKey,
+    pub allowed_ips: BTreeSet<IpNet>,
+}
+
+/// Technical peer projection from the authoritative byte-preserving document.
+/// Invalid or ambiguous records remain represented as unverifiable entries.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct SavedPeerProjection {
+    pub peers: BTreeMap<PublicKey, SavedPeer>,
+    pub duplicate_public_keys: BTreeSet<PublicKey>,
+    pub unverifiable_peer_indices: BTreeSet<usize>,
+    pub document_unverifiable: bool,
+}
+
+impl SavedPeerProjection {
+    /// Project every `[Peer]` section, retaining valid but unsupported route
+    /// shapes that cannot fit the UI's ordinary client address model.
+    pub fn from_document(document: &ConfigDocument) -> Self {
+        let mut peers = BTreeMap::new();
+        let mut duplicate_public_keys = BTreeSet::new();
+        let mut unverifiable_peer_indices = BTreeSet::new();
+
+        for peer in document.peers() {
+            let Some(identity) = peer.identity else {
+                unverifiable_peer_indices.insert(peer.index);
+                continue;
+            };
+            let invalid_route = peer
+                .issues
+                .iter()
+                .any(|issue| matches!(issue, PeerIssue::MalformedAllowedIp { .. }));
+            let duplicate_key_field = peer.issues.iter().any(|issue| {
+                matches!(
+                    issue,
+                    PeerIssue::MalformedPublicKey { .. }
+                        | PeerIssue::DuplicatePublicKeyField { .. }
+                        | PeerIssue::DuplicatePeerIdentity { .. }
+                )
+            });
+            if invalid_route || duplicate_key_field {
+                unverifiable_peer_indices.insert(peer.index);
+            }
+
+            let routes = peer
+                .allowed_ips
+                .iter()
+                .map(|network| network.trunc())
+                .collect();
+            if peers
+                .insert(
+                    identity,
+                    SavedPeer {
+                        identity,
+                        allowed_ips: routes,
+                    },
+                )
+                .is_some()
+            {
+                duplicate_public_keys.insert(identity);
+                unverifiable_peer_indices.insert(peer.index);
+            }
+        }
+
+        let document_unverifiable = document
+            .issues()
+            .iter()
+            .any(|issue| !matches!(issue, DocumentIssue::ManagedMetadata(_)));
+        Self {
+            peers,
+            duplicate_public_keys,
+            unverifiable_peer_indices,
+            document_unverifiable,
+        }
+    }
+}
+
+/// A technical discrepancy or a reason verification could not safely decide.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum ComparisonFinding {
+    InterfaceUnknown,
+    InterfaceUnavailable,
+    InterfaceMissing,
+    InterfaceMismatch { expected: String, observed: String },
+    PeerSetUnknown,
+    PeerSetUnavailable,
+    PeerSetPartial,
+    SavedDocumentUnverifiable,
+    SavedPeerUnverifiable { peer_index: usize },
+    DuplicateSavedPublicKey { identity: PublicKey },
+    DuplicateRuntimePublicKey { identity: PublicKey },
+    MissingRuntimePeer { identity: PublicKey },
+    ExtraRuntimePeer { identity: PublicKey },
+    RuntimeRoutesUnavailable { identity: PublicKey },
+    AllowedIpsMismatch { identity: PublicKey },
+}
+
+/// Outcome does not depend on handshake, counters, aliases, or record ordering.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum TechnicalMatch {
+    Match,
+    Mismatch(Vec<ComparisonFinding>),
+    Unverifiable(Vec<ComparisonFinding>),
+}
+
+/// Compare the complete saved peer set with a runtime sample, requiring the
+/// observed interface to be the interface selected by current configuration.
+pub fn compare_peers(
+    saved: &SavedPeerProjection,
+    expected_interface: &str,
+    runtime: &RuntimeObservation,
+) -> TechnicalMatch {
+    let mut findings = Vec::new();
+    let runtime_peers = match &runtime.interface {
+        InterfaceObservation::Unknown => {
+            return TechnicalMatch::Unverifiable(vec![ComparisonFinding::InterfaceUnknown]);
+        }
+        InterfaceObservation::Unavailable => {
+            return TechnicalMatch::Unverifiable(vec![ComparisonFinding::InterfaceUnavailable]);
+        }
+        InterfaceObservation::Missing => {
+            return TechnicalMatch::Mismatch(vec![ComparisonFinding::InterfaceMissing]);
+        }
+        InterfaceObservation::Present { name, peers } => {
+            if name != expected_interface {
+                return TechnicalMatch::Mismatch(vec![ComparisonFinding::InterfaceMismatch {
+                    expected: expected_interface.to_owned(),
+                    observed: name.clone(),
+                }]);
+            }
+            peers
+        }
+    };
+
+    if saved.document_unverifiable {
+        findings.push(ComparisonFinding::SavedDocumentUnverifiable);
+    }
+    findings.extend(saved.unverifiable_peer_indices.iter().map(|peer_index| {
+        ComparisonFinding::SavedPeerUnverifiable {
+            peer_index: *peer_index,
+        }
+    }));
+    findings.extend(saved.duplicate_public_keys.iter().map(|identity| {
+        ComparisonFinding::DuplicateSavedPublicKey {
+            identity: *identity,
+        }
+    }));
+    findings.extend(runtime_peers.duplicate_public_keys.iter().map(|identity| {
+        ComparisonFinding::DuplicateRuntimePublicKey {
+            identity: *identity,
+        }
+    }));
+
+    match runtime_peers.completeness {
+        PeerSetCompleteness::Unknown => findings.push(ComparisonFinding::PeerSetUnknown),
+        PeerSetCompleteness::Unavailable => findings.push(ComparisonFinding::PeerSetUnavailable),
+        PeerSetCompleteness::Partial => findings.push(ComparisonFinding::PeerSetPartial),
+        PeerSetCompleteness::Complete => {}
+    }
+
+    for (identity, expected) in &saved.peers {
+        match runtime_peers.peers.get(identity) {
+            None if runtime_peers.completeness == PeerSetCompleteness::Complete => {
+                findings.push(ComparisonFinding::MissingRuntimePeer {
+                    identity: *identity,
+                });
+            }
+            None => {}
+            Some(actual) => match &actual.allowed_ips {
+                Observed::Unavailable => {
+                    findings.push(ComparisonFinding::RuntimeRoutesUnavailable {
+                        identity: *identity,
+                    })
+                }
+                Observed::Available(routes) => {
+                    let normalized: BTreeSet<IpNet> =
+                        routes.iter().map(|network| network.trunc()).collect();
+                    if normalized != expected.allowed_ips {
+                        findings.push(ComparisonFinding::AllowedIpsMismatch {
+                            identity: *identity,
+                        });
+                    }
+                }
+            },
+        }
+    }
+    if runtime_peers.completeness == PeerSetCompleteness::Complete {
+        for identity in runtime_peers.peers.keys() {
+            if !saved.peers.contains_key(identity) {
+                findings.push(ComparisonFinding::ExtraRuntimePeer {
+                    identity: *identity,
+                });
+            }
+        }
+    }
+
+    if findings.is_empty() {
+        TechnicalMatch::Match
+    } else if findings.iter().any(is_unverifiable_finding) {
+        TechnicalMatch::Unverifiable(findings)
+    } else {
+        TechnicalMatch::Mismatch(findings)
+    }
+}
+
+fn is_unverifiable_finding(finding: &ComparisonFinding) -> bool {
+    matches!(
+        finding,
+        ComparisonFinding::InterfaceUnknown
+            | ComparisonFinding::InterfaceUnavailable
+            | ComparisonFinding::PeerSetUnknown
+            | ComparisonFinding::PeerSetUnavailable
+            | ComparisonFinding::PeerSetPartial
+            | ComparisonFinding::SavedDocumentUnverifiable
+            | ComparisonFinding::SavedPeerUnverifiable { .. }
+            | ComparisonFinding::DuplicateSavedPublicKey { .. }
+            | ComparisonFinding::DuplicateRuntimePublicKey { .. }
+            | ComparisonFinding::RuntimeRoutesUnavailable { .. }
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use std::str::FromStr;
+
+    use super::*;
+
+    const KEY_A: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
+    const KEY_B: &str = "AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=";
+
+    fn key(value: &str) -> PublicKey {
+        PublicKey::from_str(value).unwrap()
+    }
+
+    fn runtime_peer(routes: Observed<Vec<IpNet>>) -> RuntimePeer {
+        RuntimePeer {
+            allowed_ips: routes,
+            last_handshake: Observed::Available(None),
+            received_bytes: Observed::Available(0),
+            transmitted_bytes: Observed::Available(0),
+        }
+    }
+
+    fn observation(records: Vec<(PublicKey, RuntimePeer)>) -> RuntimeObservation {
+        RuntimeObservation {
+            observed_at: None,
+            interface: InterfaceObservation::Present {
+                name: "wg0".to_owned(),
+                peers: RuntimePeerSet::from_records(PeerSetCompleteness::Complete, records),
+            },
+        }
+    }
+
+    fn compare(saved: &SavedPeerProjection, runtime: &RuntimeObservation) -> TechnicalMatch {
+        compare_peers(saved, "wg0", runtime)
+    }
+
+    fn projection(source: &str) -> SavedPeerProjection {
+        SavedPeerProjection::from_document(&ConfigDocument::parse(source).unwrap())
+    }
+
+    fn route(value: &str) -> IpNet {
+        value.parse().unwrap()
+    }
+
+    #[test]
+    fn observation_distinguishes_unknown_missing_unavailable_zero_and_no_handshake() {
+        assert_eq!(
+            RuntimeObservation::unknown().interface,
+            InterfaceObservation::Unknown
+        );
+        assert_eq!(
+            RuntimeObservation {
+                observed_at: None,
+                interface: InterfaceObservation::Unavailable,
+            }
+            .interface,
+            InterfaceObservation::Unavailable
+        );
+        assert!(matches!(
+            runtime_peer(Observed::Available(Vec::new())).last_handshake,
+            Observed::Available(None)
+        ));
+        assert_eq!(
+            runtime_peer(Observed::Available(Vec::new())).received_bytes,
+            Observed::Available(0)
+        );
+        assert_eq!(
+            runtime_peer(Observed::Unavailable).allowed_ips,
+            Observed::Unavailable
+        );
+    }
+
+    #[test]
+    fn comparison_accepts_empty_set_and_normalized_reordered_dual_stack_routes() {
+        let empty = projection("");
+        assert_eq!(
+            compare(&empty, &observation(Vec::new())),
+            TechnicalMatch::Match
+        );
+
+        let source =
+            format!("[Peer]\nPublicKey = {KEY_A}\nAllowedIPs = fd00::2/128, 10.0.0.2/32\n");
+        let expected = projection(&source);
+        let actual = observation(vec![(
+            key(KEY_A),
+            runtime_peer(Observed::Available(vec![
+                route("10.0.0.2/32"),
+                route("fd00::2/128"),
+                route("10.0.0.2/32"),
+            ])),
+        )]);
+        assert_eq!(compare(&expected, &actual), TechnicalMatch::Match);
+    }
+
+    #[test]
+    fn comparison_requires_the_configured_interface_identity() {
+        let expected = projection(&format!(
+            "[Peer]\nPublicKey = {KEY_A}\nAllowedIPs = 10.0.0.2/32\n"
+        ));
+        let wrong_interface = RuntimeObservation {
+            observed_at: None,
+            interface: InterfaceObservation::Present {
+                name: "wg1".to_owned(),
+                peers: RuntimePeerSet::from_records(
+                    PeerSetCompleteness::Complete,
+                    [(
+                        key(KEY_A),
+                        runtime_peer(Observed::Available(vec![route("10.0.0.2/32")])),
+                    )],
+                ),
+            },
+        };
+        assert_eq!(
+            compare_peers(&expected, "wg0", &wrong_interface),
+            TechnicalMatch::Mismatch(vec![ComparisonFinding::InterfaceMismatch {
+                expected: "wg0".to_owned(),
+                observed: "wg1".to_owned(),
+            }])
+        );
+    }
+
+    #[test]
+    fn counters_and_handshake_changes_do_not_change_technical_match() {
+        let expected = projection(&format!(
+            "[Peer]\nPublicKey = {KEY_A}\nAllowedIPs = 10.0.0.2/32\n"
+        ));
+        let mut before_restart_peer = runtime_peer(Observed::Available(vec![route("10.0.0.2/32")]));
+        before_restart_peer.last_handshake = Observed::Available(Some(
+            OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap(),
+        ));
+        before_restart_peer.received_bytes = Observed::Available(1_000);
+        before_restart_peer.transmitted_bytes = Observed::Available(300);
+        let before_restart = observation(vec![(key(KEY_A), before_restart_peer)]);
+        let mut after_restart_peer = runtime_peer(Observed::Available(vec![route("10.0.0.2/32")]));
+        after_restart_peer.last_handshake = Observed::Available(None);
+        after_restart_peer.received_bytes = Observed::Available(0);
+        after_restart_peer.transmitted_bytes = Observed::Available(7);
+        let after_restart = observation(vec![(key(KEY_A), after_restart_peer)]);
+
+        assert_eq!(compare(&expected, &before_restart), TechnicalMatch::Match);
+        assert_eq!(compare(&expected, &after_restart), TechnicalMatch::Match);
+    }
+
+    #[test]
+    fn projection_keeps_valid_unsupported_routes_and_ipv4_only_peers() {
+        let source = format!(
+            "[Peer]\nPublicKey = {KEY_A}\nAllowedIPs = 10.0.0.2/32, 10.20.0.0/16\n[Peer]\nPublicKey = {KEY_B}\nAllowedIPs = 10.0.0.3/32\n"
+        );
+        let expected = projection(&source);
+        assert_eq!(expected.peers.len(), 2);
+        assert!(expected.unverifiable_peer_indices.is_empty());
+        let actual = observation(vec![
+            (
+                key(KEY_A),
+                runtime_peer(Observed::Available(vec![
+                    route("10.20.0.0/16"),
+                    route("10.0.0.2/32"),
+                ])),
+            ),
+            (
+                key(KEY_B),
+                runtime_peer(Observed::Available(vec![route("10.0.0.3/32")])),
+            ),
+        ]);
+        assert_eq!(compare(&expected, &actual), TechnicalMatch::Match);
+    }
+
+    #[test]
+    fn comparison_reports_missing_extra_and_route_mismatch() {
+        let expected = projection(&format!(
+            "[Peer]\nPublicKey = {KEY_A}\nAllowedIPs = 10.0.0.2/32\n"
+        ));
+        let actual = observation(vec![(
+            key(KEY_B),
+            runtime_peer(Observed::Available(vec![route("10.0.0.3/32")])),
+        )]);
+        let TechnicalMatch::Mismatch(findings) = compare(&expected, &actual) else {
+            panic!("expected concrete set differences");
+        };
+        assert!(findings.contains(&ComparisonFinding::MissingRuntimePeer {
+            identity: key(KEY_A)
+        }));
+        assert!(findings.contains(&ComparisonFinding::ExtraRuntimePeer {
+            identity: key(KEY_B)
+        }));
+
+        let actual = observation(vec![(
+            key(KEY_A),
+            runtime_peer(Observed::Available(vec![route("10.0.0.9/32")])),
+        )]);
+        assert!(matches!(
+            compare(&expected, &actual),
+            TechnicalMatch::Mismatch(findings)
+                if findings.contains(&ComparisonFinding::AllowedIpsMismatch { identity: key(KEY_A) })
+        ));
+    }
+
+    #[test]
+    fn partial_duplicate_and_unavailable_records_never_pass() {
+        let expected = projection(&format!(
+            "[Peer]\nPublicKey = {KEY_A}\nAllowedIPs = 10.0.0.2/32\n"
+        ));
+        let duplicate = RuntimePeerSet::from_records(
+            PeerSetCompleteness::Complete,
+            [
+                (
+                    key(KEY_A),
+                    runtime_peer(Observed::Available(vec![route("10.0.0.2/32")])),
+                ),
+                (
+                    key(KEY_A),
+                    runtime_peer(Observed::Available(vec![route("10.0.0.2/32")])),
+                ),
+            ],
+        );
+        assert!(matches!(
+            compare(
+                &expected,
+                &RuntimeObservation {
+                    observed_at: None,
+                    interface: InterfaceObservation::Present {
+                        name: "wg0".to_owned(),
+                        peers: duplicate,
+                    },
+                }
+            ),
+            TechnicalMatch::Unverifiable(findings)
+                if findings.contains(&ComparisonFinding::DuplicateRuntimePublicKey { identity: key(KEY_A) })
+        ));
+
+        for completeness in [
+            PeerSetCompleteness::Unknown,
+            PeerSetCompleteness::Unavailable,
+            PeerSetCompleteness::Partial,
+        ] {
+            let partial = RuntimeObservation {
+                observed_at: None,
+                interface: InterfaceObservation::Present {
+                    name: "wg0".to_owned(),
+                    peers: RuntimePeerSet {
+                        peers: BTreeMap::from([(
+                            key(KEY_A),
+                            runtime_peer(Observed::Available(vec![route("10.0.0.2/32")])),
+                        )]),
+                        duplicate_public_keys: BTreeSet::new(),
+                        completeness,
+                    },
+                },
+            };
+            let TechnicalMatch::Unverifiable(findings) = compare(&expected, &partial) else {
+                panic!("an incomplete peer set cannot verify");
+            };
+            if completeness != PeerSetCompleteness::Complete {
+                assert!(!findings.iter().any(|finding| matches!(
+                    finding,
+                    ComparisonFinding::MissingRuntimePeer { .. }
+                )));
+            }
+        }
+
+        let no_routes = observation(vec![(key(KEY_A), runtime_peer(Observed::Unavailable))]);
+        assert!(matches!(
+            compare(&expected, &no_routes),
+            TechnicalMatch::Unverifiable(_)
+        ));
+    }
+
+    #[test]
+    fn duplicate_saved_identities_are_unverifiable_even_if_the_map_would_match() {
+        let source = format!(
+            "[Peer]\nPublicKey = {KEY_A}\nAllowedIPs = 10.0.0.2/32\n[Peer]\nPublicKey = {KEY_A}\nAllowedIPs = 10.0.0.3/32\n"
+        );
+        let expected = projection(&source);
+        let actual = observation(vec![(
+            key(KEY_A),
+            runtime_peer(Observed::Available(vec![route("10.0.0.2/32")])),
+        )]);
+        assert!(matches!(
+            compare(&expected, &actual),
+            TechnicalMatch::Unverifiable(findings)
+                if findings.contains(&ComparisonFinding::DuplicateSavedPublicKey { identity: key(KEY_A) })
+        ));
+    }
+
+    #[test]
+    fn invalid_saved_peer_and_interface_states_fail_closed_but_metadata_does_not() {
+        let invalid = projection("[Peer]\nAllowedIPs = 10.0.0.2/32\n");
+        assert!(matches!(
+            compare(&invalid, &observation(Vec::new())),
+            TechnicalMatch::Unverifiable(_)
+        ));
+        let valid_with_bad_alias = projection(&format!(
+            "# wireguard-admin:v1 alias=bad%0A date=unknown\n[Peer]\nPublicKey = {KEY_A}\nAllowedIPs = 10.0.0.2/32\n"
+        ));
+        assert!(!valid_with_bad_alias.document_unverifiable);
+
+        for interface in [
+            InterfaceObservation::Unknown,
+            InterfaceObservation::Unavailable,
+            InterfaceObservation::Missing,
+        ] {
+            let result = compare(
+                &projection(""),
+                &RuntimeObservation {
+                    observed_at: None,
+                    interface,
+                },
+            );
+            assert!(!matches!(result, TechnicalMatch::Match));
+        }
+    }
+
+    #[test]
+    fn malformed_routes_and_structural_saved_issues_cannot_be_omitted_for_a_match() {
+        let malformed_route = projection(&format!(
+            "[Peer]\nPublicKey = {KEY_A}\nAllowedIPs = 10.0.0.2/32, not-a-network\n"
+        ));
+        assert!(!malformed_route.unverifiable_peer_indices.is_empty());
+        let apparent_subset_match = observation(vec![(
+            key(KEY_A),
+            runtime_peer(Observed::Available(vec![route("10.0.0.2/32")])),
+        )]);
+        assert!(matches!(
+            compare(&malformed_route, &apparent_subset_match),
+            TechnicalMatch::Unverifiable(findings)
+                if findings.iter().any(|finding| matches!(finding, ComparisonFinding::SavedPeerUnverifiable { .. }))
+        ));
+
+        let structural = projection(&format!(
+            "[Interface]\nMalformed Directive\n[Peer]\nPublicKey = {KEY_A}\nAllowedIPs = 10.0.0.2/32\n"
+        ));
+        assert!(structural.document_unverifiable);
+        let exact_runtime = observation(vec![(
+            key(KEY_A),
+            runtime_peer(Observed::Available(vec![route("10.0.0.2/32")])),
+        )]);
+        assert!(matches!(
+            compare(&structural, &exact_runtime),
+            TechnicalMatch::Unverifiable(findings)
+                if findings.contains(&ComparisonFinding::SavedDocumentUnverifiable)
+        ));
+    }
+}

src/runtime/transport.rs

@@ -0,0 +1,562 @@
+//! Bounded, read-only Linux Generic Netlink transport for WireGuard.
+//!
+//! Replies are received into zeroizing buffers. The datagram sender address
+//! must be kernel port 0; `nlmsg_pid` in request replies is the requester's
+//! port ID, as Linux Generic Netlink constructs replies using that destination.
+
+use std::time::{Duration, Instant};
+
+use rustix::event::{PollFd, PollFlags, Timespec, poll};
+use rustix::fd::OwnedFd;
+use rustix::io::Errno;
+use rustix::net::{
+    AddressFamily, RecvFlags, SendFlags, SocketFlags, SocketType, bind, getsockname,
+    netlink::SocketAddrNetlink, recvfrom, sendto, socket_with,
+};
+use zeroize::Zeroizing;
+
+use super::netlink::{DecodeContext, DecodeError, DecodeLimits, GetDeviceDecoder};
+use super::{InterfaceObservation, PeerSetCompleteness, RuntimeObservation, RuntimePeerSet};
+
+const NETLINK_GENERIC: u32 = 16;
+const CTRL_FAMILY_ID: u16 = 0x10;
+const CTRL_CMD_GETFAMILY: u8 = 3;
+const CTRL_CMD_NEWFAMILY: u8 = 1;
+const CTRL_ATTR_FAMILY_ID: u16 = 1;
+const CTRL_ATTR_FAMILY_NAME: u16 = 2;
+const WG_CMD_GET_DEVICE: u8 = 0;
+const WG_GENL_VERSION: u8 = 1;
+const WG_ATTR_IFNAME: u16 = 2;
+const NLMSG_ERROR: u16 = 2;
+const NLMSG_DONE: u16 = 3;
+const NLMSG_REQUEST: u16 = 1;
+const NLMSG_DUMP: u16 = 0x300;
+const NLMSG_MULTI: u16 = 2;
+const MAX_DATAGRAM: usize = 1024 * 1024;
+const MAX_INTERFACE_NAME: usize = 15;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)]
+pub(crate) enum TransportError {
+    #[error("configured WireGuard interface was not found")]
+    NoDevice,
+    #[error("permission denied while reading WireGuard state")]
+    PermissionDenied,
+    #[error("WireGuard netlink response was malformed or incomplete")]
+    Malformed,
+    #[error("WireGuard netlink transport failed")]
+    Transport,
+    #[error("WireGuard netlink request timed out")]
+    Timeout,
+    #[error("WireGuard netlink response exceeded configured limits")]
+    LimitExceeded,
+}
+
+#[derive(Debug)]
+struct Datagram {
+    bytes: Zeroizing<Vec<u8>>,
+    sender: u32,
+}
+
+trait Io {
+    fn now(&self) -> Duration;
+    fn send(&mut self, destination: u32, bytes: &[u8]) -> Result<(), TransportError>;
+    fn receive(&mut self, wait: Duration) -> Result<Datagram, TransportError>;
+    fn port_id(&self) -> u32;
+}
+
+struct LinuxIo {
+    socket: OwnedFd,
+    port_id: u32,
+    origin: Instant,
+}
+
+impl LinuxIo {
+    fn new() -> Result<Self, TransportError> {
+        let socket = socket_with(
+            AddressFamily::NETLINK,
+            SocketType::DGRAM,
+            SocketFlags::CLOEXEC.union(SocketFlags::NONBLOCK),
+            Some(rustix::net::netlink::GENERIC),
+        )
+        .map_err(map_errno)?;
+        bind(&socket, &SocketAddrNetlink::new(0, 0)).map_err(map_errno)?;
+        let addr = SocketAddrNetlink::try_from(getsockname(&socket).map_err(map_errno)?)
+            .map_err(|_| TransportError::Transport)?;
+        Ok(Self {
+            socket,
+            port_id: addr.pid(),
+            origin: Instant::now(),
+        })
+    }
+}
+
+impl Io for LinuxIo {
+    fn now(&self) -> Duration {
+        self.origin.elapsed()
+    }
+
+    fn port_id(&self) -> u32 {
+        self.port_id
+    }
+
+    fn send(&mut self, destination: u32, bytes: &[u8]) -> Result<(), TransportError> {
+        let sent = sendto(
+            &self.socket,
+            bytes,
+            SendFlags::empty(),
+            &SocketAddrNetlink::new(destination, 0),
+        )
+        .map_err(map_errno)?;
+        if sent == bytes.len() {
+            Ok(())
+        } else {
+            Err(TransportError::Transport)
+        }
+    }
+
+    fn receive(&mut self, wait: Duration) -> Result<Datagram, TransportError> {
+        let timeout = Timespec {
+            tv_sec: wait.as_secs().min(i64::MAX as u64) as i64,
+            tv_nsec: wait.subsec_nanos() as i64,
+        };
+        let mut fds = [PollFd::new(&self.socket, PollFlags::IN)];
+        if poll(&mut fds, Some(&timeout)).map_err(map_errno)? == 0 {
+            return Err(TransportError::Timeout);
+        }
+        let mut bytes = Zeroizing::new(vec![0; MAX_DATAGRAM]);
+        let (_, count, address) = recvfrom(
+            &self.socket,
+            &mut bytes[..],
+            RecvFlags::TRUNC.union(RecvFlags::DONTWAIT),
+        )
+        .map_err(map_errno)?;
+        if count > MAX_DATAGRAM {
+            return Err(TransportError::LimitExceeded);
+        }
+        let sender = address
+            .and_then(|a| SocketAddrNetlink::try_from(a).ok())
+            .ok_or(TransportError::Malformed)?
+            .pid();
+        bytes.truncate(count);
+        Ok(Datagram { bytes, sender })
+    }
+}
+
+fn map_errno(error: Errno) -> TransportError {
+    if error == Errno::ACCESS || error == Errno::PERM {
+        TransportError::PermissionDenied
+    } else if error == Errno::NOENT || error == Errno::NODEV {
+        TransportError::NoDevice
+    } else if error == Errno::TIMEDOUT {
+        TransportError::Timeout
+    } else {
+        TransportError::Transport
+    }
+}
+
+fn map_kernel_error(error: i32) -> TransportError {
+    if error == Errno::ACCESS.raw_os_error() || error == Errno::PERM.raw_os_error() {
+        TransportError::PermissionDenied
+    } else if error == Errno::NOENT.raw_os_error() || error == Errno::NODEV.raw_os_error() {
+        TransportError::NoDevice
+    } else {
+        TransportError::Transport
+    }
+}
+
+pub(crate) fn observe_interface(
+    interface: &str,
+    timeout: Duration,
+    limits: DecodeLimits,
+) -> Result<RuntimeObservation, TransportError> {
+    let mut io = LinuxIo::new()?;
+    observe_with_io(&mut io, interface, timeout, limits)
+}
+
+fn observe_with_io(
+    io: &mut impl Io,
+    interface: &str,
+    timeout: Duration,
+    limits: DecodeLimits,
+) -> Result<RuntimeObservation, TransportError> {
+    if interface.is_empty()
+        || interface.len() > MAX_INTERFACE_NAME
+        || interface.as_bytes().contains(&0)
+        || timeout.is_zero()
+    {
+        return Err(TransportError::Malformed);
+    }
+    let start = io.now();
+    let deadline = start.checked_add(timeout).ok_or(TransportError::Timeout)?;
+    let family = resolve_family(io, deadline)?;
+    let seq = 2;
+    let request = get_device_request(family, seq, io.port_id(), interface);
+    io.send(0, &request)?;
+    let context = DecodeContext {
+        sequence: seq,
+        port_id: io.port_id(),
+        family_id: family,
+    };
+    let mut decoder = GetDeviceDecoder::new(context, interface, limits);
+    loop {
+        let datagram = receive_before(io, deadline)?;
+        if datagram.sender != 0 {
+            return Err(TransportError::Malformed);
+        }
+        decoder.push_datagram(&datagram.bytes).map_err(map_decode)?;
+        // Decoder only returns a device after its explicit terminal DONE.
+        if decoder.is_done() {
+            let device = decoder.finish().map_err(map_decode)?;
+            return Ok(RuntimeObservation {
+                observed_at: None,
+                interface: InterfaceObservation::Present {
+                    name: device.interface_name,
+                    peers: RuntimePeerSet::from_records(
+                        PeerSetCompleteness::Complete,
+                        device.peers,
+                    ),
+                },
+            });
+        }
+    }
+}
+
+fn receive_before(io: &mut impl Io, deadline: Duration) -> Result<Datagram, TransportError> {
+    let remaining = deadline
+        .checked_sub(io.now())
+        .ok_or(TransportError::Timeout)?;
+    if remaining.is_zero() {
+        return Err(TransportError::Timeout);
+    }
+    io.receive(remaining)
+}
+
+fn resolve_family(io: &mut impl Io, deadline: Duration) -> Result<u16, TransportError> {
+    let seq = 1;
+    let mut request = message(CTRL_FAMILY_ID, NLMSG_REQUEST, seq, io.port_id());
+    request.extend_from_slice(&[CTRL_CMD_GETFAMILY, 1, 0, 0]);
+    push_attr(&mut request, CTRL_ATTR_FAMILY_NAME, b"wireguard\0");
+    set_len(&mut request);
+    io.send(0, &request)?;
+    let d = receive_before(io, deadline)?;
+    if d.sender != 0 {
+        return Err(TransportError::Malformed);
+    }
+    parse_family(&d.bytes, seq, io.port_id())
+}
+
+fn parse_family(bytes: &[u8], seq: u32, port: u32) -> Result<u16, TransportError> {
+    if bytes.len() < 20 {
+        return Err(TransportError::Malformed);
+    }
+    if read_u32(bytes, 0)? as usize != bytes.len()
+        || read_u32(bytes, 8)? != seq
+        || read_u32(bytes, 12)? != port
+    {
+        return Err(TransportError::Malformed);
+    }
+    let message_type = read_u16(bytes, 4)?;
+    if message_type == NLMSG_ERROR {
+        let status = read_u32(bytes, 16)? as i32;
+        if status == 0 {
+            return Err(TransportError::Malformed);
+        }
+        return Err(status
+            .checked_neg()
+            .map(map_kernel_error)
+            .ok_or(TransportError::Malformed)?);
+    }
+    if message_type != CTRL_FAMILY_ID {
+        return Err(TransportError::Malformed);
+    }
+    if bytes[16] != CTRL_CMD_NEWFAMILY || read_u16(bytes, 18)? != 0 {
+        return Err(TransportError::Malformed);
+    }
+    let mut off = 20;
+    while off < bytes.len() {
+        if bytes.len() - off < 4 {
+            return Err(TransportError::Malformed);
+        }
+        let len = read_u16(bytes, off)? as usize;
+        let kind = read_u16(bytes, off + 2)? & 0x3fff;
+        if len < 4 || off + len > bytes.len() {
+            return Err(TransportError::Malformed);
+        }
+        if kind == CTRL_ATTR_FAMILY_ID && len == 6 {
+            return read_u16(bytes, off + 4);
+        }
+        let aligned_len = len.checked_add(3).ok_or(TransportError::Malformed)? & !3;
+        off = off
+            .checked_add(aligned_len)
+            .filter(|&end| end <= bytes.len())
+            .ok_or(TransportError::Malformed)?;
+    }
+    Err(TransportError::NoDevice)
+}
+
+fn get_device_request(family: u16, seq: u32, port: u32, interface: &str) -> Vec<u8> {
+    let mut out = message(family, NLMSG_REQUEST | NLMSG_DUMP, seq, port);
+    out.extend_from_slice(&[WG_CMD_GET_DEVICE, WG_GENL_VERSION, 0, 0]);
+    let mut interface_name = interface.as_bytes().to_vec();
+    interface_name.push(0);
+    push_attr(&mut out, WG_ATTR_IFNAME, &interface_name);
+    set_len(&mut out);
+    out
+}
+fn message(kind: u16, flags: u16, seq: u32, port: u32) -> Vec<u8> {
+    let mut out = vec![0; 16];
+    out[4..6].copy_from_slice(&kind.to_ne_bytes());
+    out[6..8].copy_from_slice(&flags.to_ne_bytes());
+    out[8..12].copy_from_slice(&seq.to_ne_bytes());
+    out[12..16].copy_from_slice(&port.to_ne_bytes());
+    out
+}
+fn push_attr(out: &mut Vec<u8>, kind: u16, value: &[u8]) {
+    let len = (4 + value.len()) as u16;
+    out.extend_from_slice(&len.to_ne_bytes());
+    out.extend_from_slice(&kind.to_ne_bytes());
+    out.extend_from_slice(value);
+    while !out.len().is_multiple_of(4) {
+        out.push(0);
+    }
+}
+fn set_len(out: &mut [u8]) {
+    let len = out.len() as u32;
+    out[..4].copy_from_slice(&len.to_ne_bytes());
+}
+fn read_u16(bytes: &[u8], at: usize) -> Result<u16, TransportError> {
+    bytes
+        .get(at..at + 2)
+        .and_then(|b| b.try_into().ok())
+        .map(u16::from_ne_bytes)
+        .ok_or(TransportError::Malformed)
+}
+fn read_u32(bytes: &[u8], at: usize) -> Result<u32, TransportError> {
+    bytes
+        .get(at..at + 4)
+        .and_then(|b| b.try_into().ok())
+        .map(u32::from_ne_bytes)
+        .ok_or(TransportError::Malformed)
+}
+fn map_decode(error: DecodeError) -> TransportError {
+    match error {
+        DecodeError::KernelError(n) if n == Errno::NODEV.raw_os_error() => TransportError::NoDevice,
+        DecodeError::KernelError(n)
+            if n == Errno::ACCESS.raw_os_error() || n == Errno::PERM.raw_os_error() =>
+        {
+            TransportError::PermissionDenied
+        }
+        DecodeError::LimitExceeded => TransportError::LimitExceeded,
+        DecodeError::KernelError(_) | DecodeError::DumpFailed => TransportError::Transport,
+        _ => TransportError::Malformed,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn ctrl_family_message(sequence: u32, port: u32, family: u16) -> Vec<u8> {
+        let mut out = message(CTRL_FAMILY_ID, 0, sequence, port);
+        out.extend_from_slice(&[CTRL_CMD_NEWFAMILY, 2, 0, 0]);
+        push_attr(&mut out, CTRL_ATTR_FAMILY_ID, &family.to_ne_bytes());
+        set_len(&mut out);
+        out
+    }
+
+    #[test]
+    fn resolves_family_only_from_matching_kernel_reply() {
+        assert_eq!(parse_family(&ctrl_family_message(1, 55, 42), 1, 55), Ok(42));
+        assert_eq!(
+            parse_family(&ctrl_family_message(2, 55, 42), 1, 55),
+            Err(TransportError::Malformed)
+        );
+        assert_eq!(
+            parse_family(&ctrl_family_message(1, 56, 42), 1, 55),
+            Err(TransportError::Malformed)
+        );
+    }
+
+    fn netlink_error(sequence: u32, port: u32, status: i32) -> Vec<u8> {
+        let mut out = message(NLMSG_ERROR, 0, sequence, port);
+        out.extend_from_slice(&status.to_ne_bytes());
+        set_len(&mut out);
+        out
+    }
+
+    #[test]
+    fn family_errors_map_kernel_errno_after_validating_reply_identity() {
+        for errno in [Errno::NOENT, Errno::NODEV] {
+            assert_eq!(
+                parse_family(&netlink_error(1, 55, -errno.raw_os_error()), 1, 55),
+                Err(TransportError::NoDevice)
+            );
+        }
+        for errno in [Errno::ACCESS, Errno::PERM] {
+            assert_eq!(
+                parse_family(&netlink_error(1, 55, -errno.raw_os_error()), 1, 55),
+                Err(TransportError::PermissionDenied)
+            );
+        }
+        assert_eq!(
+            parse_family(&netlink_error(1, 55, -Errno::IO.raw_os_error()), 1, 55),
+            Err(TransportError::Transport)
+        );
+        assert_eq!(
+            parse_family(&netlink_error(1, 55, i32::MIN), 1, 55),
+            Err(TransportError::Malformed)
+        );
+        assert_eq!(
+            parse_family(&netlink_error(1, 55, 0), 1, 55),
+            Err(TransportError::Malformed)
+        );
+
+        assert_eq!(
+            parse_family(&netlink_error(2, 55, -Errno::NOENT.raw_os_error()), 1, 55),
+            Err(TransportError::Malformed)
+        );
+        assert_eq!(
+            parse_family(&netlink_error(1, 56, -Errno::NOENT.raw_os_error()), 1, 55),
+            Err(TransportError::Malformed)
+        );
+        assert_eq!(
+            parse_family(
+                &netlink_error(1, 55, -Errno::NOENT.raw_os_error())[..19],
+                1,
+                55
+            ),
+            Err(TransportError::Malformed)
+        );
+    }
+
+    #[test]
+    fn family_resolution_propagates_kernel_errors_from_injected_transport() {
+        for (errno, expected) in [
+            (Errno::NODEV, TransportError::NoDevice),
+            (Errno::ACCESS, TransportError::PermissionDenied),
+            (Errno::IO, TransportError::Transport),
+        ] {
+            let mut fake = FakeIo {
+                port: 55,
+                ..FakeIo::default()
+            };
+            fake.incoming
+                .push_back(datagram(netlink_error(1, 55, -errno.raw_os_error()), 0));
+            assert_eq!(
+                resolve_family(&mut fake, Duration::from_secs(1)),
+                Err(expected)
+            );
+            assert_eq!(fake.sent.len(), 1);
+        }
+    }
+
+    #[test]
+    fn get_request_targets_only_named_interface_and_dump_flags() {
+        let request = get_device_request(42, 7, 55, "wg-test");
+        assert_eq!(read_u16(&request, 4), Ok(42));
+        assert_eq!(read_u16(&request, 6), Ok(NLMSG_REQUEST | NLMSG_DUMP));
+        assert_eq!(
+            &request[16..20],
+            &[WG_CMD_GET_DEVICE, WG_GENL_VERSION, 0, 0]
+        );
+        assert_eq!(read_u16(&request, 20), Ok(12));
+        assert_eq!(&request[22..24], &2u16.to_ne_bytes());
+        assert_eq!(&request[24..32], b"wg-test\0");
+        assert_eq!(request.len(), 32);
+        assert_eq!(read_u32(&request, 0), Ok(request.len() as u32));
+    }
+
+    #[derive(Default)]
+    struct FakeIo {
+        clock: Duration,
+        port: u32,
+        sent: Vec<(u32, Vec<u8>)>,
+        incoming: std::collections::VecDeque<Datagram>,
+    }
+    impl Io for FakeIo {
+        fn now(&self) -> Duration {
+            self.clock
+        }
+        fn port_id(&self) -> u32 {
+            self.port
+        }
+        fn send(&mut self, destination: u32, bytes: &[u8]) -> Result<(), TransportError> {
+            self.sent.push((destination, bytes.to_vec()));
+            Ok(())
+        }
+        fn receive(&mut self, wait: Duration) -> Result<Datagram, TransportError> {
+            if let Some(datagram) = self.incoming.pop_front() {
+                Ok(datagram)
+            } else {
+                self.clock += wait;
+                Err(TransportError::Timeout)
+            }
+        }
+    }
+
+    fn datagram(bytes: Vec<u8>, sender: u32) -> Datagram {
+        Datagram {
+            bytes: Zeroizing::new(bytes),
+            sender,
+        }
+    }
+
+    #[test]
+    fn injected_transport_rejects_non_kernel_sender_and_bounds_resolution_wait() {
+        let mut fake = FakeIo {
+            port: 55,
+            ..FakeIo::default()
+        };
+        fake.incoming
+            .push_back(datagram(ctrl_family_message(1, 55, 42), 22));
+        assert_eq!(
+            resolve_family(&mut fake, Duration::from_secs(1)),
+            Err(TransportError::Malformed)
+        );
+        assert_eq!(fake.sent.len(), 1);
+        assert_eq!(fake.sent[0].0, 0);
+
+        let mut silent = FakeIo {
+            port: 55,
+            ..FakeIo::default()
+        };
+        assert_eq!(
+            resolve_family(&mut silent, Duration::from_millis(5)),
+            Err(TransportError::Timeout)
+        );
+        assert_eq!(silent.clock, Duration::from_millis(5));
+    }
+
+    #[test]
+    fn injected_transport_returns_only_a_complete_public_empty_observation() {
+        let mut fake = FakeIo {
+            port: 55,
+            ..FakeIo::default()
+        };
+        fake.incoming
+            .push_back(datagram(ctrl_family_message(1, 55, 42), 0));
+        let mut response = message(42, NLMSG_MULTI, 2, 55);
+        response.extend_from_slice(&[WG_CMD_GET_DEVICE, WG_GENL_VERSION, 0, 0]);
+        push_attr(&mut response, 2, b"wg0\0");
+        push_attr(&mut response, 8 | 0x8000, &[]);
+        set_len(&mut response);
+        let mut complete = response;
+        let done_start = complete.len();
+        complete.extend_from_slice(&message(NLMSG_DONE, 0, 2, 55));
+        complete.extend_from_slice(&0i32.to_ne_bytes());
+        set_len(&mut complete[done_start..]);
+        fake.incoming.push_back(datagram(complete, 0));
+        let observation = observe_with_io(
+            &mut fake,
+            "wg0",
+            Duration::from_secs(2),
+            DecodeLimits::default(),
+        )
+        .unwrap();
+        assert!(
+            matches!(observation.interface, InterfaceObservation::Present { ref name, ref peers } if name == "wg0" && peers.completeness == PeerSetCompleteness::Complete && peers.peers.is_empty())
+        );
+        assert_eq!(fake.sent.len(), 2);
+        assert!(fake.sent.iter().all(|(destination, _)| *destination == 0));
+    }
+}

src/service.rs

@@ -0,0 +1,469 @@
+//! Systemd restart job tracking and bounded D-Bus adapter.
+//!
+//! The event tracker is pure. The real D-Bus restart adapter is crate-private;
+//! callers outside the crate must use the application restart boundary.
+//!
+//! ```compile_fail
+//! use wireguard_admin::service::dbus::restart_systemd_unit_report;
+//! ```
+
+mod dbus;
+
+pub(crate) use dbus::restart_systemd_unit_report;
+
+use std::time::{Duration, Instant};
+
+/// One operation-wide monotonic deadline, created before bus setup begins.
+///
+/// The same value should be used for connection, subscription, request,
+/// waiting, and subsequent runtime verification. A timed-out wait does not
+/// imply that systemd canceled its job.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct RestartDeadline {
+    started_at: Instant,
+    timeout: Duration,
+}
+
+impl RestartDeadline {
+    pub const fn new(started_at: Instant, timeout: Duration) -> Self {
+        Self {
+            started_at,
+            timeout,
+        }
+    }
+
+    pub const fn started_at(self) -> Instant {
+        self.started_at
+    }
+
+    pub const fn timeout(self) -> Duration {
+        self.timeout
+    }
+
+    /// Remaining time at a monotonic sample. Values sampled before the start
+    /// are treated as the start, which also keeps deterministic callers safe.
+    pub fn remaining_at(self, now: Instant) -> Duration {
+        self.timeout
+            .saturating_sub(now.saturating_duration_since(self.started_at))
+    }
+
+    pub fn is_expired_at(self, now: Instant) -> bool {
+        self.remaining_at(now).is_zero()
+    }
+}
+
+/// The only event the tracker needs from systemd's `JobRemoved` signal.
+/// Unknown systemd result strings should map to [`JobResult::Other`], without
+/// retaining or surfacing the raw bus value.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum JobResult {
+    Done,
+    Failed,
+    Dependency,
+    Timeout,
+    Canceled,
+    Skipped,
+    Other,
+}
+
+/// An event delivered by the future D-Bus adapter.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum ServiceEvent {
+    JobRemoved {
+        job_path: String,
+        unit: String,
+        result: JobResult,
+    },
+    Disconnected,
+    ManagerOwnerChanged,
+    SubscriptionFailed,
+    PermissionDenied,
+    RequestFailed,
+}
+
+/// Redacted terminal reason. No D-Bus error string or signal payload is kept.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum RestartFailure {
+    JobFailed,
+    JobDependency,
+    JobTimedOut,
+    JobCanceled,
+    JobSkipped,
+    JobResultUnknown,
+    BusDisconnected,
+    ManagerOwnerChanged,
+    SubscriptionFailed,
+    UnitResolutionFailed,
+    PermissionDenied,
+    RequestFailed,
+    WaitTimedOutMayStillComplete,
+    EarlyEventBufferFull,
+    InvalidJobPath,
+}
+
+/// Current state of a single restart attempt.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum RestartOutcome {
+    Waiting,
+    Succeeded,
+    Failed(RestartFailure),
+}
+
+/// Terminal job outcome plus the request-submission boundary. A timeout after
+/// submission explicitly leaves open the possibility that the job may finish.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct RestartExecution {
+    pub outcome: RestartOutcome,
+    pub submitted: bool,
+}
+
+/// Pure event tracker for the primary systemd unit resolved from the configured
+/// target.
+///
+/// Signals can arrive before the method reply containing the job path. To
+/// close that race, up to [`MAX_EARLY_JOB_EVENTS`] events for the configured
+/// unit are retained until `job_submitted` provides the actual path. Overflow
+/// fails closed instead of silently dropping a potentially relevant event.
+pub struct RestartJobTracker {
+    configured_unit: String,
+    deadline: RestartDeadline,
+    submitted_job_path: Option<String>,
+    early_events: Vec<(String, JobResult)>,
+    outcome: RestartOutcome,
+}
+
+/// Bound for `JobRemoved` events received before the systemd method reply.
+pub const MAX_EARLY_JOB_EVENTS: usize = 64;
+
+impl RestartJobTracker {
+    pub fn new(configured_unit: impl Into<String>, deadline: RestartDeadline) -> Self {
+        Self {
+            configured_unit: configured_unit.into(),
+            deadline,
+            submitted_job_path: None,
+            early_events: Vec::new(),
+            outcome: RestartOutcome::Waiting,
+        }
+    }
+
+    pub fn configured_unit(&self) -> &str {
+        &self.configured_unit
+    }
+
+    pub const fn deadline(&self) -> RestartDeadline {
+        self.deadline
+    }
+
+    pub fn submitted_job_path(&self) -> Option<&str> {
+        self.submitted_job_path.as_deref()
+    }
+
+    pub const fn outcome(&self) -> RestartOutcome {
+        self.outcome
+    }
+
+    /// Record the actual object path returned by systemd's restart request.
+    /// Any earlier matching-unit event is reconciled against that path.
+    pub fn job_submitted(&mut self, job_path: impl Into<String>, now: Instant) -> RestartOutcome {
+        if self.outcome != RestartOutcome::Waiting {
+            return self.outcome;
+        }
+        if self.expire_if_needed(now) {
+            return self.outcome;
+        }
+
+        let job_path = job_path.into();
+        if job_path.is_empty() {
+            self.fail(RestartFailure::InvalidJobPath);
+            return self.outcome;
+        }
+        if self.submitted_job_path.is_some() {
+            self.fail(RestartFailure::RequestFailed);
+            return self.outcome;
+        }
+        self.submitted_job_path = Some(job_path.clone());
+
+        for (event_path, result) in std::mem::take(&mut self.early_events) {
+            if event_path == job_path {
+                self.finish_job(result);
+                break;
+            }
+        }
+        self.outcome
+    }
+
+    /// Apply one event using a caller-supplied monotonic time sample.
+    pub fn observe(&mut self, event: ServiceEvent, now: Instant) -> RestartOutcome {
+        if self.outcome != RestartOutcome::Waiting {
+            return self.outcome;
+        }
+        if self.expire_if_needed(now) {
+            return self.outcome;
+        }
+
+        match event {
+            ServiceEvent::JobRemoved {
+                job_path,
+                unit,
+                result,
+            } => {
+                if unit != self.configured_unit || job_path.is_empty() {
+                    return self.outcome;
+                }
+                if let Some(submitted_path) = &self.submitted_job_path {
+                    if &job_path == submitted_path {
+                        self.finish_job(result);
+                    }
+                } else if self.early_events.len() == MAX_EARLY_JOB_EVENTS {
+                    self.fail(RestartFailure::EarlyEventBufferFull);
+                } else {
+                    self.early_events.push((job_path, result));
+                }
+            }
+            ServiceEvent::Disconnected => self.fail(RestartFailure::BusDisconnected),
+            ServiceEvent::ManagerOwnerChanged => self.fail(RestartFailure::ManagerOwnerChanged),
+            ServiceEvent::SubscriptionFailed => self.fail(RestartFailure::SubscriptionFailed),
+            ServiceEvent::PermissionDenied => self.fail(RestartFailure::PermissionDenied),
+            ServiceEvent::RequestFailed => self.fail(RestartFailure::RequestFailed),
+        }
+        self.outcome
+    }
+
+    /// Check the operation deadline even when no event arrives.
+    pub fn poll(&mut self, now: Instant) -> RestartOutcome {
+        self.expire_if_needed(now);
+        self.outcome
+    }
+
+    fn expire_if_needed(&mut self, now: Instant) -> bool {
+        if self.deadline.is_expired_at(now) {
+            self.fail(RestartFailure::WaitTimedOutMayStillComplete);
+            true
+        } else {
+            false
+        }
+    }
+
+    fn finish_job(&mut self, result: JobResult) {
+        self.outcome = match result {
+            JobResult::Done => RestartOutcome::Succeeded,
+            JobResult::Failed => RestartOutcome::Failed(RestartFailure::JobFailed),
+            JobResult::Dependency => RestartOutcome::Failed(RestartFailure::JobDependency),
+            JobResult::Timeout => RestartOutcome::Failed(RestartFailure::JobTimedOut),
+            JobResult::Canceled => RestartOutcome::Failed(RestartFailure::JobCanceled),
+            JobResult::Skipped => RestartOutcome::Failed(RestartFailure::JobSkipped),
+            JobResult::Other => RestartOutcome::Failed(RestartFailure::JobResultUnknown),
+        };
+    }
+
+    fn fail(&mut self, failure: RestartFailure) {
+        self.outcome = RestartOutcome::Failed(failure);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn setup() -> (Instant, RestartJobTracker) {
+        let start = Instant::now();
+        let deadline = RestartDeadline::new(start, Duration::from_secs(5));
+        (
+            start,
+            RestartJobTracker::new("wg-quick@wg0.service", deadline),
+        )
+    }
+
+    fn removed(path: &str, unit: &str, result: JobResult) -> ServiceEvent {
+        ServiceEvent::JobRemoved {
+            job_path: path.to_owned(),
+            unit: unit.to_owned(),
+            result,
+        }
+    }
+
+    #[test]
+    fn buffers_fast_completion_until_actual_job_path_is_returned() {
+        let (start, mut tracker) = setup();
+        assert_eq!(
+            tracker.observe(
+                removed(
+                    "/org/freedesktop/systemd1/job/41",
+                    "wg-quick@wg0.service",
+                    JobResult::Done
+                ),
+                start + Duration::from_millis(1),
+            ),
+            RestartOutcome::Waiting
+        );
+        assert_eq!(
+            tracker.job_submitted(
+                "/org/freedesktop/systemd1/job/41",
+                start + Duration::from_millis(2),
+            ),
+            RestartOutcome::Succeeded
+        );
+    }
+
+    #[test]
+    fn unrelated_jobs_and_units_do_not_finish_the_attempt() {
+        let (start, mut tracker) = setup();
+        tracker.job_submitted("/job/actual", start + Duration::from_millis(1));
+        for event in [
+            removed("/job/other", "wg-quick@wg0.service", JobResult::Done),
+            removed("/job/actual", "wg-quick@wg1.service", JobResult::Done),
+        ] {
+            assert_eq!(
+                tracker.observe(event, start + Duration::from_millis(2)),
+                RestartOutcome::Waiting
+            );
+        }
+    }
+
+    #[test]
+    fn configured_alias_matches_job_removed_primary_unit_name() {
+        let start = Instant::now();
+        let deadline = RestartDeadline::new(start, Duration::from_secs(5));
+        // The adapter resolves this primary ID from configured alias
+        // "wireguard-vpn.service" before constructing the tracker.
+        let mut tracker = RestartJobTracker::new("wg-quick@wg0.service", deadline);
+        tracker.job_submitted("/job/alias-target", start + Duration::from_millis(1));
+        assert_eq!(
+            tracker.observe(
+                removed("/job/alias-target", "wg-quick@wg0.service", JobResult::Done,),
+                start + Duration::from_millis(2),
+            ),
+            RestartOutcome::Succeeded
+        );
+    }
+
+    #[test]
+    fn only_the_done_job_result_is_success() {
+        for (job_result, failure) in [
+            (JobResult::Failed, RestartFailure::JobFailed),
+            (JobResult::Dependency, RestartFailure::JobDependency),
+            (JobResult::Timeout, RestartFailure::JobTimedOut),
+            (JobResult::Canceled, RestartFailure::JobCanceled),
+            (JobResult::Skipped, RestartFailure::JobSkipped),
+            (JobResult::Other, RestartFailure::JobResultUnknown),
+        ] {
+            let (start, mut tracker) = setup();
+            tracker.job_submitted("/job/actual", start + Duration::from_millis(1));
+            assert_eq!(
+                tracker.observe(
+                    removed("/job/actual", "wg-quick@wg0.service", job_result),
+                    start + Duration::from_millis(2),
+                ),
+                RestartOutcome::Failed(failure)
+            );
+        }
+    }
+
+    #[test]
+    fn events_for_same_unit_are_still_correlated_by_returned_path() {
+        let (start, mut tracker) = setup();
+        tracker.observe(
+            removed("/job/unrelated", "wg-quick@wg0.service", JobResult::Failed),
+            start + Duration::from_millis(1),
+        );
+        assert_eq!(
+            tracker.job_submitted("/job/actual", start + Duration::from_millis(2)),
+            RestartOutcome::Waiting
+        );
+    }
+
+    #[test]
+    fn systemd_oneshot_success_needs_no_pid_or_active_state() {
+        let (start, mut tracker) = setup();
+        tracker.job_submitted("/job/actual", start + Duration::from_millis(1));
+        assert_eq!(
+            tracker.observe(
+                removed("/job/actual", "wg-quick@wg0.service", JobResult::Done),
+                start + Duration::from_millis(2),
+            ),
+            RestartOutcome::Succeeded
+        );
+    }
+
+    #[test]
+    fn monotonic_deadline_is_shared_and_timeout_does_not_claim_cancellation() {
+        let (start, mut tracker) = setup();
+        let deadline = tracker.deadline();
+        assert_eq!(
+            deadline.remaining_at(start + Duration::from_secs(2)),
+            Duration::from_secs(3)
+        );
+        assert_eq!(
+            tracker.poll(start + Duration::from_secs(5)),
+            RestartOutcome::Failed(RestartFailure::WaitTimedOutMayStillComplete)
+        );
+    }
+
+    #[test]
+    fn bus_and_request_failures_are_redacted_terminal_outcomes() {
+        for (event, failure) in [
+            (ServiceEvent::Disconnected, RestartFailure::BusDisconnected),
+            (
+                ServiceEvent::ManagerOwnerChanged,
+                RestartFailure::ManagerOwnerChanged,
+            ),
+            (
+                ServiceEvent::SubscriptionFailed,
+                RestartFailure::SubscriptionFailed,
+            ),
+            (
+                ServiceEvent::PermissionDenied,
+                RestartFailure::PermissionDenied,
+            ),
+            (ServiceEvent::RequestFailed, RestartFailure::RequestFailed),
+        ] {
+            let (start, mut tracker) = setup();
+            assert_eq!(
+                tracker.observe(event, start + Duration::from_millis(1)),
+                RestartOutcome::Failed(failure)
+            );
+        }
+    }
+
+    #[test]
+    fn early_event_buffer_overflow_fails_closed() {
+        let (start, mut tracker) = setup();
+        for index in 0..MAX_EARLY_JOB_EVENTS {
+            assert_eq!(
+                tracker.observe(
+                    removed(
+                        &format!("/job/{index}"),
+                        "wg-quick@wg0.service",
+                        JobResult::Done,
+                    ),
+                    start + Duration::from_millis(1),
+                ),
+                RestartOutcome::Waiting
+            );
+        }
+        assert_eq!(
+            tracker.observe(
+                removed("/job/overflow", "wg-quick@wg0.service", JobResult::Done),
+                start + Duration::from_millis(2),
+            ),
+            RestartOutcome::Failed(RestartFailure::EarlyEventBufferFull)
+        );
+    }
+
+    #[test]
+    fn terminal_outcome_cannot_be_replaced_by_later_events() {
+        let (start, mut tracker) = setup();
+        tracker.job_submitted("/job/actual", start + Duration::from_millis(1));
+        tracker.observe(
+            removed("/job/actual", "wg-quick@wg0.service", JobResult::Failed),
+            start + Duration::from_millis(2),
+        );
+        assert_eq!(
+            tracker.observe(
+                removed("/job/actual", "wg-quick@wg0.service", JobResult::Done),
+                start + Duration::from_millis(3),
+            ),
+            RestartOutcome::Failed(RestartFailure::JobFailed)
+        );
+    }
+}

src/service/dbus.rs

@@ -0,0 +1,680 @@
+//! Bounded asynchronous systemd D-Bus restart adapter.
+//!
+//! zbus's `async-io` backend owns its socket-driving thread. This module uses
+//! a small standard-library executor to poll each caller-side future only
+//! until the shared monotonic deadline; it does not use zbus's blocking signal
+//! iterator, whose `next` call has no application deadline.
+
+use std::future::{Future, poll_fn};
+use std::pin::{Pin, pin};
+use std::sync::Arc;
+use std::task::{Context, Poll, Wake, Waker};
+use std::thread;
+use std::time::Instant;
+
+use zbus::export::futures_core::Stream;
+use zbus::message::Type;
+use zbus::{Connection, MatchRule, MessageStream, Proxy};
+
+use super::{
+    JobResult, RestartDeadline, RestartExecution, RestartFailure, RestartJobTracker,
+    RestartOutcome, ServiceEvent,
+};
+
+const SYSTEMD_NAME: &str = "org.freedesktop.systemd1";
+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";
+
+/// Resolve the configured unit's primary ID, request a restart of exactly
+/// `configured_unit`, and wait for its matching systemd `JobRemoved` result.
+/// The caller owns backup ordering and subsequent runtime verification, and
+/// must pass the same deadline to those stages.
+/// Restart a configured unit and retain whether a timed-out request was submitted.
+pub(crate) fn restart_systemd_unit_report(
+    configured_unit: &str,
+    deadline: RestartDeadline,
+) -> RestartExecution {
+    restart_with_transport_report(&mut SystemdBus::default(), configured_unit, deadline)
+}
+
+trait RestartTransport {
+    fn connect(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure>;
+    fn subscribe(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure>;
+    fn resolve_primary_unit(
+        &mut self,
+        configured_unit: &str,
+        deadline: RestartDeadline,
+    ) -> Result<String, RestartFailure>;
+    fn restart_unit(
+        &mut self,
+        unit: &str,
+        deadline: RestartDeadline,
+    ) -> Result<String, RestartFailure>;
+    fn next_event(&mut self, deadline: RestartDeadline) -> Result<ServiceEvent, RestartFailure>;
+}
+
+#[cfg(test)]
+fn restart_with_transport(
+    transport: &mut impl RestartTransport,
+    configured_unit: &str,
+    deadline: RestartDeadline,
+) -> RestartOutcome {
+    restart_with_transport_report(transport, configured_unit, deadline).outcome
+}
+
+fn restart_with_transport_report(
+    transport: &mut impl RestartTransport,
+    configured_unit: &str,
+    deadline: RestartDeadline,
+) -> RestartExecution {
+    if configured_unit.is_empty() || configured_unit.as_bytes().contains(&0) {
+        return execution(RestartOutcome::Failed(RestartFailure::RequestFailed), false);
+    }
+
+    if let Err(failure) = transport.connect(deadline) {
+        return execution(RestartOutcome::Failed(failure), false);
+    }
+    if deadline.is_expired_at(Instant::now()) {
+        return execution(
+            RestartOutcome::Failed(RestartFailure::WaitTimedOutMayStillComplete),
+            false,
+        );
+    }
+    if let Err(failure) = transport.subscribe(deadline) {
+        return execution(RestartOutcome::Failed(failure), false);
+    }
+    if deadline.is_expired_at(Instant::now()) {
+        return execution(
+            RestartOutcome::Failed(RestartFailure::WaitTimedOutMayStillComplete),
+            false,
+        );
+    }
+    let primary_unit = match transport.resolve_primary_unit(configured_unit, deadline) {
+        Ok(unit) if !unit.is_empty() => unit,
+        Ok(_) => {
+            return execution(
+                RestartOutcome::Failed(RestartFailure::UnitResolutionFailed),
+                false,
+            );
+        }
+        Err(failure) => return execution(RestartOutcome::Failed(failure), false),
+    };
+    if deadline.is_expired_at(Instant::now()) {
+        return execution(
+            RestartOutcome::Failed(RestartFailure::WaitTimedOutMayStillComplete),
+            false,
+        );
+    }
+
+    let job_path = match transport.restart_unit(configured_unit, deadline) {
+        Ok(path) => path,
+        Err(failure) => {
+            let submitted = matches!(
+                failure,
+                RestartFailure::WaitTimedOutMayStillComplete | RestartFailure::BusDisconnected
+            );
+            return execution(RestartOutcome::Failed(failure), submitted);
+        }
+    };
+    let mut tracker = RestartJobTracker::new(primary_unit, deadline);
+    let mut outcome = tracker.job_submitted(job_path, Instant::now());
+    while outcome == RestartOutcome::Waiting {
+        let event = match transport.next_event(deadline) {
+            Ok(event) => event,
+            Err(failure) => return execution(RestartOutcome::Failed(failure), true),
+        };
+        outcome = tracker.observe(event, Instant::now());
+    }
+    execution(outcome, true)
+}
+
+fn execution(outcome: RestartOutcome, submitted: bool) -> RestartExecution {
+    RestartExecution { outcome, submitted }
+}
+
+#[derive(Default)]
+struct SystemdBus {
+    connection: Option<Connection>,
+    manager: Option<Proxy<'static>>,
+    job_events: Option<zbus::proxy::SignalStream<'static>>,
+    owner_events: Option<zbus::MessageStream>,
+}
+
+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);
+        self.connection = Some(connection);
+        Ok(())
+    }
+
+    fn subscribe(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure> {
+        let manager = self
+            .manager
+            .as_ref()
+            .ok_or(RestartFailure::SubscriptionFailed)?;
+        let job_events = 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.
+        self.job_events = Some(job_events);
+        self.owner_events = Some(owner_events);
+        drive(manager.call::<_, _, ()>("Subscribe", &()), deadline)?.map_err(map_bus_error)
+    }
+
+    fn resolve_primary_unit(
+        &mut self,
+        configured_unit: &str,
+        deadline: RestartDeadline,
+    ) -> Result<String, RestartFailure> {
+        let manager = self
+            .manager
+            .as_ref()
+            .ok_or(RestartFailure::UnitResolutionFailed)?;
+        let connection = self
+            .connection
+            .as_ref()
+            .ok_or(RestartFailure::BusDisconnected)?;
+        let unit_path: zbus::zvariant::OwnedObjectPath =
+            drive(manager.call("GetUnit", &(configured_unit,)), deadline)?
+                .map_err(map_unit_resolution_error)?;
+        let unit = drive(
+            Proxy::new_owned(
+                connection.clone(),
+                SYSTEMD_NAME,
+                unit_path,
+                "org.freedesktop.systemd1.Unit",
+            ),
+            deadline,
+        )?
+        .map_err(map_unit_resolution_error)?;
+        drive(unit.get_property::<String>("Id"), deadline)?.map_err(map_unit_resolution_error)
+    }
+
+    fn restart_unit(
+        &mut self,
+        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,
+        )??;
+        Ok(job_path.to_string())
+    }
+
+    fn next_event(&mut self, deadline: RestartDeadline) -> Result<ServiceEvent, RestartFailure> {
+        let connection = self
+            .connection
+            .as_ref()
+            .ok_or(RestartFailure::BusDisconnected)?;
+        let job_events = self
+            .job_events
+            .as_mut()
+            .ok_or(RestartFailure::BusDisconnected)?;
+        let owner_events = self
+            .owner_events
+            .as_mut()
+            .ok_or(RestartFailure::BusDisconnected)?;
+        drive(
+            async {
+                let mut closed = pin!(connection.closed());
+                poll_fn(|cx| {
+                    if connection.is_closed() {
+                        return Poll::Ready(Ok(ServiceEvent::Disconnected));
+                    }
+                    if let Poll::Ready(message) =
+                        Stream::poll_next(Pin::new(&mut *owner_events), cx)
+                    {
+                        return Poll::Ready(match message {
+                            None => Ok(ServiceEvent::Disconnected),
+                            Some(Err(error)) => Err(map_bus_error(error)),
+                            Some(Ok(message)) => {
+                                let decoded: Result<(String, String, String), _> =
+                                    message.body().deserialize();
+                                match decoded {
+                                    Ok((name, _, _)) if name == SYSTEMD_NAME => {
+                                        Ok(ServiceEvent::ManagerOwnerChanged)
+                                    }
+                                    Ok(_) => Ok(ServiceEvent::JobRemoved {
+                                        job_path: String::new(),
+                                        unit: String::new(),
+                                        result: JobResult::Other,
+                                    }),
+                                    Err(_) => Err(RestartFailure::RequestFailed),
+                                }
+                            }
+                        });
+                    }
+                    if let Poll::Ready(message) = Stream::poll_next(Pin::new(&mut *job_events), cx)
+                    {
+                        return Poll::Ready(match message {
+                            None => Ok(ServiceEvent::Disconnected),
+                            Some(message) => {
+                                let decoded: Result<
+                                    (u32, zbus::zvariant::OwnedObjectPath, String, String),
+                                    _,
+                                > = message.body().deserialize();
+                                match decoded {
+                                    Ok((_, path, unit, result)) => Ok(ServiceEvent::JobRemoved {
+                                        job_path: path.to_string(),
+                                        unit,
+                                        result: map_job_result(&result),
+                                    }),
+                                    Err(_) => Err(RestartFailure::RequestFailed),
+                                }
+                            }
+                        });
+                    }
+                    if Future::poll(closed.as_mut(), cx).is_ready() {
+                        return Poll::Ready(Ok(ServiceEvent::Disconnected));
+                    }
+                    Poll::Pending
+                })
+                .await
+            },
+            deadline,
+        )?
+    }
+}
+
+fn map_job_result(result: &str) -> JobResult {
+    match result {
+        "done" => JobResult::Done,
+        "failed" => JobResult::Failed,
+        "dependency" => JobResult::Dependency,
+        "timeout" => JobResult::Timeout,
+        "canceled" => JobResult::Canceled,
+        "skipped" => JobResult::Skipped,
+        _ => JobResult::Other,
+    }
+}
+
+fn map_bus_error(error: zbus::Error) -> RestartFailure {
+    match error {
+        zbus::Error::MethodError(name, _, _)
+            if name.as_str() == "org.freedesktop.DBus.Error.AccessDenied"
+                || name.as_str() == "org.freedesktop.systemd1.AccessDenied" =>
+        {
+            RestartFailure::PermissionDenied
+        }
+        zbus::Error::InputOutput(_) | zbus::Error::Handshake(_) => RestartFailure::BusDisconnected,
+        _ => RestartFailure::RequestFailed,
+    }
+}
+
+fn map_unit_resolution_error(error: zbus::Error) -> RestartFailure {
+    match map_bus_error(error) {
+        RestartFailure::PermissionDenied => RestartFailure::PermissionDenied,
+        RestartFailure::BusDisconnected => RestartFailure::BusDisconnected,
+        _ => RestartFailure::UnitResolutionFailed,
+    }
+}
+
+struct ThreadWake(thread::Thread);
+
+impl Wake for ThreadWake {
+    fn wake(self: Arc<Self>) {
+        self.0.unpark();
+    }
+
+    fn wake_by_ref(self: &Arc<Self>) {
+        self.0.unpark();
+    }
+}
+
+/// Poll a future on this thread, parking only for the remaining operation
+/// budget. zbus's async-io connection executor continues driving socket I/O
+/// independently and wakes this thread when the future can make progress.
+fn drive<F: Future>(future: F, deadline: RestartDeadline) -> Result<F::Output, RestartFailure> {
+    let waker = Waker::from(Arc::new(ThreadWake(thread::current())));
+    let mut context = Context::from_waker(&waker);
+    let mut future = pin!(future);
+    loop {
+        if deadline.is_expired_at(Instant::now()) {
+            return Err(RestartFailure::WaitTimedOutMayStillComplete);
+        }
+        match Future::poll(future.as_mut(), &mut context) {
+            Poll::Ready(output) => return Ok(output),
+            Poll::Pending => {
+                let remaining = deadline.remaining_at(Instant::now());
+                if remaining.is_zero() {
+                    return Err(RestartFailure::WaitTimedOutMayStillComplete);
+                }
+                thread::park_timeout(remaining);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::VecDeque;
+    use std::time::Duration;
+
+    struct FakeBus {
+        calls: Vec<&'static str>,
+        deadlines: Vec<(&'static str, RestartDeadline)>,
+        events: VecDeque<ServiceEvent>,
+        failure: Option<(&'static str, RestartFailure)>,
+        configured_unit: &'static str,
+        job_path: &'static str,
+        primary_unit: &'static str,
+    }
+
+    impl Default for FakeBus {
+        fn default() -> Self {
+            Self {
+                calls: Vec::new(),
+                deadlines: Vec::new(),
+                events: VecDeque::new(),
+                failure: None,
+                configured_unit: "wg-quick@wg0.service",
+                job_path: "",
+                primary_unit: "wg-quick@wg0.service",
+            }
+        }
+    }
+
+    impl FakeBus {
+        fn stage(&mut self, stage: &'static str) -> Result<(), RestartFailure> {
+            self.calls.push(stage);
+            match self.failure {
+                Some((failed_stage, failure)) if failed_stage == stage => Err(failure),
+                _ => Ok(()),
+            }
+        }
+    }
+
+    impl RestartTransport for FakeBus {
+        fn connect(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure> {
+            self.deadlines.push(("connect", deadline));
+            self.stage("connect")
+        }
+
+        fn subscribe(&mut self, deadline: RestartDeadline) -> Result<(), RestartFailure> {
+            self.deadlines.push(("subscribe", deadline));
+            self.stage("subscribe")
+        }
+
+        fn resolve_primary_unit(
+            &mut self,
+            configured_unit: &str,
+            deadline: RestartDeadline,
+        ) -> Result<String, RestartFailure> {
+            assert_eq!(configured_unit, self.configured_unit);
+            self.deadlines.push(("resolve", deadline));
+            self.stage("resolve")?;
+            Ok(self.primary_unit.to_owned())
+        }
+
+        fn restart_unit(
+            &mut self,
+            unit: &str,
+            deadline: RestartDeadline,
+        ) -> Result<String, RestartFailure> {
+            assert_eq!(unit, self.configured_unit);
+            self.deadlines.push(("restart", deadline));
+            self.stage("restart")?;
+            Ok(self.job_path.to_owned())
+        }
+
+        fn next_event(
+            &mut self,
+            deadline: RestartDeadline,
+        ) -> Result<ServiceEvent, RestartFailure> {
+            self.deadlines.push(("event", deadline));
+            self.stage("event")?;
+            self.events
+                .pop_front()
+                .ok_or(RestartFailure::BusDisconnected)
+        }
+    }
+
+    fn deadline() -> RestartDeadline {
+        RestartDeadline::new(Instant::now(), Duration::from_secs(2))
+    }
+
+    fn removed(path: &str, unit: &str, result: JobResult) -> ServiceEvent {
+        ServiceEvent::JobRemoved {
+            job_path: path.to_owned(),
+            unit: unit.to_owned(),
+            result,
+        }
+    }
+
+    #[test]
+    fn subscribes_before_restart_and_uses_returned_job_path() {
+        let mut fake = FakeBus {
+            job_path: "/job/actual",
+            primary_unit: "wg-quick@wg0.service",
+            events: VecDeque::from([
+                removed("/job/other", "wg-quick@wg0.service", JobResult::Done),
+                removed("/job/actual", "wg-quick@wg0.service", JobResult::Done),
+            ]),
+            ..FakeBus::default()
+        };
+        assert_eq!(
+            restart_with_transport(&mut fake, "wg-quick@wg0.service", deadline()),
+            RestartOutcome::Succeeded
+        );
+        assert_eq!(
+            fake.calls,
+            [
+                "connect",
+                "subscribe",
+                "resolve",
+                "restart",
+                "event",
+                "event"
+            ]
+        );
+    }
+
+    #[test]
+    fn fast_completion_and_failure_results_are_processed_without_a_bus() {
+        let mut fast = FakeBus {
+            job_path: "/job/fast",
+            primary_unit: "wg-quick@wg0.service",
+            events: VecDeque::from([removed(
+                "/job/fast",
+                "wg-quick@wg0.service",
+                JobResult::Done,
+            )]),
+            ..FakeBus::default()
+        };
+        assert_eq!(
+            restart_with_transport(&mut fast, "wg-quick@wg0.service", deadline()),
+            RestartOutcome::Succeeded
+        );
+
+        let mut failed = FakeBus {
+            job_path: "/job/failed",
+            primary_unit: "wg-quick@wg0.service",
+            events: VecDeque::from([removed(
+                "/job/failed",
+                "wg-quick@wg0.service",
+                JobResult::Canceled,
+            )]),
+            ..FakeBus::default()
+        };
+        assert_eq!(
+            restart_with_transport(&mut failed, "wg-quick@wg0.service", deadline()),
+            RestartOutcome::Failed(RestartFailure::JobCanceled)
+        );
+    }
+
+    #[test]
+    fn setup_and_permission_failures_stop_before_restart() {
+        for (stage, failure) in [
+            ("connect", RestartFailure::BusDisconnected),
+            ("subscribe", RestartFailure::SubscriptionFailed),
+            ("resolve", RestartFailure::UnitResolutionFailed),
+            ("restart", RestartFailure::PermissionDenied),
+        ] {
+            let mut fake = FakeBus {
+                job_path: "/job/never",
+                primary_unit: "wg-quick@wg0.service",
+                failure: Some((stage, failure)),
+                ..FakeBus::default()
+            };
+            assert_eq!(
+                restart_with_transport(&mut fake, "wg-quick@wg0.service", deadline()),
+                RestartOutcome::Failed(failure)
+            );
+            if stage != "restart" {
+                assert!(!fake.calls.contains(&"restart"));
+            }
+        }
+    }
+
+    #[test]
+    fn resolves_alias_before_restart_and_uses_one_shared_deadline() {
+        let deadline = deadline();
+        let mut fake = FakeBus {
+            job_path: "/job/alias",
+            configured_unit: "wireguard-vpn.service",
+            primary_unit: "wg-quick@wg0.service",
+            events: VecDeque::from([removed(
+                "/job/alias",
+                "wg-quick@wg0.service",
+                JobResult::Done,
+            )]),
+            ..FakeBus::default()
+        };
+        assert_eq!(
+            restart_with_transport(&mut fake, "wireguard-vpn.service", deadline),
+            RestartOutcome::Succeeded
+        );
+        assert_eq!(
+            fake.calls,
+            ["connect", "subscribe", "resolve", "restart", "event"]
+        );
+        assert!(fake.deadlines.iter().all(|(_, passed)| *passed == deadline));
+    }
+
+    #[test]
+    fn expired_shared_deadline_stops_between_setup_and_submission() {
+        let mut fake = FakeBus {
+            job_path: "/job/never",
+            ..FakeBus::default()
+        };
+        let expired = RestartDeadline::new(Instant::now(), Duration::ZERO);
+        assert_eq!(
+            restart_with_transport(&mut fake, "wg-quick@wg0.service", expired),
+            RestartOutcome::Failed(RestartFailure::WaitTimedOutMayStillComplete)
+        );
+        assert_eq!(fake.calls, ["connect"]);
+    }
+
+    #[test]
+    fn result_mapping_accepts_only_the_documented_done_value() {
+        assert_eq!(map_job_result("done"), JobResult::Done);
+        for value in ["failed", "dependency", "timeout", "canceled", "skipped"] {
+            assert_ne!(map_job_result(value), JobResult::Done);
+        }
+        assert_eq!(map_job_result("future-result"), JobResult::Other);
+    }
+
+    #[test]
+    fn std_wake_park_driver_rejects_an_already_expired_future() {
+        let expired = RestartDeadline::new(Instant::now(), Duration::ZERO);
+        assert_eq!(
+            drive(std::future::pending::<()>(), expired),
+            Err(RestartFailure::WaitTimedOutMayStillComplete)
+        );
+    }
+
+    #[test]
+    fn owner_change_and_disconnect_fail_the_attempt() {
+        for (event, failure) in [
+            (
+                ServiceEvent::ManagerOwnerChanged,
+                RestartFailure::ManagerOwnerChanged,
+            ),
+            (ServiceEvent::Disconnected, RestartFailure::BusDisconnected),
+        ] {
+            let mut fake = FakeBus {
+                job_path: "/job/actual",
+                primary_unit: "wg-quick@wg0.service",
+                events: VecDeque::from([event]),
+                ..FakeBus::default()
+            };
+            assert_eq!(
+                restart_with_transport(&mut fake, "wg-quick@wg0.service", deadline()),
+                RestartOutcome::Failed(failure)
+            );
+        }
+    }
+}

src/wireguard/document.rs

@@ -457,6 +457,45 @@ impl ConfigDocument {
         &self.issues
     }
 
+    /// Whether any interface section enables or ambiguously configures
+    /// wg-quick SaveConfig. Restarting such a service can write live peer state
+    /// back over the saved configuration during shutdown.
+    pub fn has_unsafe_save_config(&self) -> bool {
+        let mut found = false;
+        for section in self
+            .sections
+            .iter()
+            .filter(|section| section.kind == SectionKind::Interface)
+        {
+            let Some(source) = self.source_slice(section.span) else {
+                return true;
+            };
+            let mut in_interface = false;
+            for line in source.lines() {
+                let trimmed = line.trim_matches(|ch: char| ch.is_ascii_whitespace());
+                if trimmed.starts_with('[') && trimmed.ends_with(']') {
+                    in_interface = parse_section_name(trimmed)
+                        .is_some_and(|name| name.eq_ignore_ascii_case("Interface"));
+                    continue;
+                }
+                if !in_interface || trimmed.is_empty() || trimmed.starts_with(['#', ';']) {
+                    continue;
+                }
+                let Some((key, value)) = trimmed.split_once('=') else {
+                    continue;
+                };
+                if !key.trim().eq_ignore_ascii_case("SaveConfig") {
+                    continue;
+                }
+                if found || !value.trim().eq_ignore_ascii_case("false") {
+                    return true;
+                }
+                found = true;
+            }
+        }
+        false
+    }
+
     /// A raw source slice for an explicitly selected span. Callers must treat it as secret-bearing.
     /// This narrow primitive supports later edit code that owns and replaces exact spans.
     pub fn source_slice(&self, span: SourceSpan) -> Option<&str> {
@@ -491,6 +530,26 @@ mod tests {
         }
     }
 
+    #[test]
+    fn save_config_detection_is_interface_scoped_and_fails_closed_on_ambiguity() {
+        let disabled =
+            ConfigDocument::parse(b"[Interface]\nSaveConfig = false\n[Peer]\nSaveConfig = true\n")
+                .unwrap();
+        assert!(!disabled.has_unsafe_save_config());
+
+        for input in [
+            b"[Interface]\nSaveConfig = true\n".as_slice(),
+            b"[Interface]\nSaveConfig = maybe\n".as_slice(),
+            b"[Interface]\nSaveConfig = false\nSaveConfig = false\n".as_slice(),
+        ] {
+            assert!(
+                ConfigDocument::parse(input)
+                    .unwrap()
+                    .has_unsafe_save_config()
+            );
+        }
+    }
+
     #[test]
     fn wireguard_peer_inspection_preserves_multiple_routes_and_unknown_directives() {
         let input = format!(