Watch-mapper status-amplification mitigation (#48)

Signed-off-by: Erick Bourgeois <erick@jeb.ca>

Author: ebourgeois

.claude/CHANGELOG.md

@@ -9,6 +9,71 @@ The format is based on the regulated environment requirements:
 
 ---
 
+## [2026-04-25 22:00] - Phase 3 of security audit: route Node→SM via canonical CAPI Machine
+
+**Author:** Erick Bourgeois
+
+### Changed
+- `src/reconcilers/helpers.rs`: new
+  `node_to_scheduled_machines_via_machine(node, machines)` mapper. Routes
+  Node events to owning ScheduledMachines via the canonical CAPI Machine
+  ownership chain — `Machine.status.nodeRef.name` (written only by the
+  CAPI controller) plus the controller-stamped
+  `LABEL_SCHEDULED_MACHINE` label on the Machine. Both legs are
+  write-controlled by trusted controllers, not by the tenant.
+- `src/reconcilers/helpers.rs`: `node_to_scheduled_machines` (the
+  pre-existing mapper that trusted `SM.status.nodeRef.name`) marked
+  `#[deprecated(since = "0.1.2")]` with a note pointing at the
+  replacement. Kept exported and tested for one release so any external
+  caller (none expected) gets a compile-time pointer.
+- `src/reconcilers/mod.rs`: re-export the new symbol; gate the legacy
+  re-export with `#[allow(deprecated)]`.
+- `src/main.rs`: wire a standalone `kube::runtime::reflector` for CAPI
+  Machines (filtered by `LABEL_SCHEDULED_MACHINE`). Spawn its watcher in
+  the background; pass the reader handle into the Node watch closure.
+  Switch the Node mapper from `node_to_scheduled_machines` to
+  `node_to_scheduled_machines_via_machine`. The
+  `Controller::watches_with` Machine watch keeps its own internal
+  reflector for the existing Machine→SM route — kube-rs does not expose
+  that internal store, so the secondary watch is intentional. Doubled
+  watch traffic is minor (same kind, same label selector).
+- `src/reconcilers/helpers_tests.rs`: 9 new tests covering the
+  canonical-Machine mapper:
+  - happy-path single-Machine match returns owning SM
+  - Node with no owning Machine → NOT enqueued (closes spoof window)
+  - Machine for a different node → NOT enqueued (canonical wins over
+    spoofed SM status)
+  - unlabelled / namespace-less / no-nodeRef Machine → ignored
+  - Node without metadata.name → ignored
+  - whitespace-only label value → ignored (mirrors
+    `machine_to_scheduled_machine`'s defensive trim)
+  - multiple Machines for same Node → both owning SMs enqueued
+- Existing `test_node_to_sms_*` tests for the deprecated mapper kept
+  with per-test `#[allow(deprecated)]` annotations so the legacy
+  surface stays covered until removal.
+
+### Why
+Phase 3 of the 2026-04-25 security audit roadmap. The pre-existing
+`node_to_scheduled_machines` mapper filtered SMs by
+`status.nodeRef.name`, which is writable by anyone with `patch
+scheduledmachines/status` on the resource. A tenant could write a
+fast-changing node name (e.g. a busy worker that's frequently
+updating) into status to amplify how many SM reconciliations they
+caused per Node update — a CPU DoS amplifier. The drain target itself
+was never affected (the reconciler always reads canonical
+`Machine.status.nodeRef` via `get_node_from_machine`), so this was
+verified as Low severity, but the routing change closes the
+amplification regardless of how the reconciler evolves.
+
+### Impact
+- [ ] Breaking change (deprecated symbol still exported)
+- [ ] Requires cluster rollout (no schema or RBAC changes)
+- [x] Config change only (in-binary; one extra Machine watch on the
+      same selector that already exists)
+- [ ] Documentation only
+
+---
+
 ## [2026-04-25 18:00] - Phase 2 of security audit: finalizer cleanup force-remove on PDB stall
 
 **Author:** Erick Bourgeois

src/main.rs

@@ -28,15 +28,19 @@ use five_spot::health::{start_health_server, HealthState};
 use five_spot::labels::LABEL_SCHEDULED_MACHINE;
 use five_spot::metrics::init_controller_info;
 use five_spot::reconcilers::{
-    error_policy, machine_to_scheduled_machine, node_to_scheduled_machines,
+    error_policy, machine_to_scheduled_machine, node_to_scheduled_machines_via_machine,
     reconcile_scheduled_machine, Context,
 };
-use futures::StreamExt;
+use futures::{StreamExt, TryStreamExt};
 use k8s_openapi::api::core::v1::Node;
 use kube::{
     api::{ApiResource, GroupVersionKind, ListParams},
     core::DynamicObject,
-    runtime::{watcher::Config, Controller},
+    runtime::{
+        reflector,
+        watcher::{self, Config},
+        Controller, WatchStreamExt,
+    },
     Api, Client,
 };
 use tracing::{error, info, warn};
@@ -276,19 +280,51 @@ async fn main() -> Result<()> {
     info!("Starting controller for ScheduledMachine resources");
 
     // Secondary watches — event-driven reactivity without polling.
-    // 1. CAPI Machine (dynamic GVK) — filtered by the scheduled-machine label we
-    //    already stamp on every Machine we create. Reverse-mapped via that label.
-    // 2. Kubernetes Node — name-matched against every SM's status.nodeRef.name
-    //    using a snapshot of the controller's own primary-resource Store.
+    //
+    // 1. CAPI Machine (dynamic GVK), filtered by the scheduled-machine label
+    //    we stamp on every Machine we create. Reverse-mapped via that label
+    //    (`machine_to_scheduled_machine`). The `Controller::watches_with`
+    //    call below maintains its own internal reflector for this watch.
+    //
+    // 2. Kubernetes Node, mapped to owning ScheduledMachines via the
+    //    canonical CAPI Machine ownership chain
+    //    (`node_to_scheduled_machines_via_machine`). This requires its own
+    //    Machine reflector store, separate from #1's internal one — kube-rs
+    //    `watches_with` does not expose its internal store. The doubled
+    //    watch traffic is minor: same kind, same label selector. Routing
+    //    via the canonical Machine instead of the tenant-writable
+    //    `SM.status.nodeRef.name` closes the watch-amplification vector
+    //    documented in Phase 3 of the 2026-04-25 security audit roadmap.
     let machine_ar = ApiResource::from_gvk_with_plural(
         &GroupVersionKind::gvk(CAPI_GROUP, CAPI_MACHINE_API_VERSION, "Machine"),
         CAPI_RESOURCE_MACHINES,
     );
     let machines_api: Api<DynamicObject> = Api::all_with(client.clone(), &machine_ar);
     let nodes_api: Api<Node> = Api::all(client.clone());
 
+    // Standalone Machine reflector for the Node→SM mapper. We build its
+    // store here, spawn the watcher in the background, and clone the
+    // reader handle into the Node closure below. The store starts empty
+    // and fills as Machine watch events arrive — Node events that fire
+    // during the warmup window will see an empty store and enqueue
+    // nothing, which is correct (no Machine yet ⇒ no SM owns this Node).
+    let machine_writer: reflector::store::Writer<DynamicObject> =
+        reflector::store::Writer::new(machine_ar.clone());
+    let machine_store = machine_writer.as_reader();
+    let machine_reflector_machines_api = machines_api.clone();
+    tokio::spawn(async move {
+        let stream = watcher::watcher(
+            machine_reflector_machines_api,
+            watcher::Config::default().labels(LABEL_SCHEDULED_MACHINE),
+        )
+        .reflect(machine_writer)
+        .applied_objects();
+        if let Err(e) = stream.try_for_each(|_| async { Ok(()) }).await {
+            error!(error = %e, "Machine reflector for Node→SM mapping terminated");
+        }
+    });
+
     let controller = Controller::new(scheduled_machines, Config::default());
-    let sm_store = controller.store();
 
     controller
         .watches_with(
@@ -298,8 +334,14 @@ async fn main() -> Result<()> {
             |machine: DynamicObject| machine_to_scheduled_machine(&machine),
         )
         .watches(nodes_api, Config::default(), move |node: Node| {
-            let snapshot = sm_store.state();
-            node_to_scheduled_machines(&node, snapshot.iter().map(std::convert::AsRef::as_ref))
+            // Snapshot the standalone Machine reflector's store. Each
+            // entry is an Arc<DynamicObject>; deref through AsRef to feed
+            // the pure mapper.
+            let machines_snapshot = machine_store.state();
+            node_to_scheduled_machines_via_machine(
+                &node,
+                machines_snapshot.iter().map(std::convert::AsRef::as_ref),
+            )
         })
         .shutdown_on_signal()
         .run(reconcile_scheduled_machine, error_policy, context)

src/reconcilers/helpers.rs

@@ -1600,10 +1600,32 @@ pub fn machine_to_scheduled_machine(
 /// Map a `Node` event to all `ScheduledMachine`s whose
 /// `status.nodeRef.name == node.metadata.name`.
 ///
+/// **DEPRECATED — replaced by [`node_to_scheduled_machines_via_machine`].**
+/// This function trusts the SM's own `status.nodeRef.name`, which is
+/// writable by anyone with `patch scheduledmachines/status` on the
+/// resource. A tenant could write a fast-changing node name into
+/// status to amplify how many SM reconciliations they trigger from
+/// unrelated Node events (CPU DoS amplifier — see Phase 3 of the
+/// 2026-04-25 security audit roadmap).
+///
+/// The replacement routes Node → Machine (canonical
+/// `Machine.status.nodeRef.name` written by CAPI) → SM (via the
+/// controller-stamped `LABEL_SCHEDULED_MACHINE` label on the Machine),
+/// closing the spoof window.
+///
+/// Kept exported with `#[deprecated]` for one release so external
+/// callers (none expected) get a compile-time pointer at the
+/// replacement.
+///
 /// Runs `O(N)` over the supplied SM iterator. Fine at small scale (tens to
 /// hundreds of SMs); if the cluster ever hosts thousands, swap in a reverse
 /// index keyed by the last-observed `nodeRef.name`.
 #[must_use]
+#[deprecated(
+    since = "0.1.2",
+    note = "Trusts tenant-writable SM.status.nodeRef.name; use \
+            node_to_scheduled_machines_via_machine for canonical-Machine routing."
+)]
 pub fn node_to_scheduled_machines<'a, I>(
     node: &k8s_openapi::api::core::v1::Node,
     scheduled_machines: I,
@@ -1634,6 +1656,95 @@ where
         .collect()
 }
 
+/// Map a `Node` event to all `ScheduledMachine`s whose **owning CAPI Machine**
+/// has `status.nodeRef.name == node.metadata.name`.
+///
+/// This is the canonical-Machine replacement for
+/// [`node_to_scheduled_machines`]. Routing via the CAPI Machine instead of the
+/// SM's own status closes the watch-amplification vector where a tenant with
+/// `patch scheduledmachines/status` could spoof `SM.status.nodeRef.name` to
+/// pin controller CPU on unrelated Node updates: only the CAPI controller
+/// writes `Machine.status.nodeRef`, and only the 5-Spot controller stamps
+/// the [`crate::labels::LABEL_SCHEDULED_MACHINE`] label that walks back to
+/// the SM. Both legs are write-controlled by trusted controllers, not by
+/// the tenant.
+///
+/// Algorithm (per Machine in the supplied iterator):
+/// 1. Read `metadata.namespace` — required to build a namespaced
+///    `ObjectRef`. Skip Machines without one.
+/// 2. Read `status.nodeRef.name` from the Machine's dynamic `data` field
+///    (`DynamicObject` because CAPI types are not statically modelled in
+///    this codebase). Skip Machines whose `nodeRef` is absent or doesn't
+///    match the node's name.
+/// 3. Read `metadata.labels[LABEL_SCHEDULED_MACHINE]` to identify the
+///    owning SM. Skip if the label is missing, empty, or whitespace-only
+///    (mirrors the same defensive trim+empty check in
+///    [`machine_to_scheduled_machine`]).
+/// 4. Emit one [`kube::runtime::reflector::ObjectRef`] per match.
+///
+/// Returns an empty `Vec` for the degenerate case where the Node itself
+/// has no `metadata.name` — there is nothing meaningful to match against.
+///
+/// Runs `O(N)` over the supplied Machine iterator. With one Machine per
+/// SM and tens-to-hundreds of SMs per cluster this is comfortably cheap;
+/// at thousands of SMs, build a reverse index keyed by node name.
+///
+/// # Threat model context
+/// Filed as Phase 3 of the 2026-04-25 security audit roadmap (severity:
+/// Low — CPU amplification, not a drain pivot). The reconciler always
+/// reads canonical `Machine.status.nodeRef` via `get_node_from_machine`
+/// before draining, so the spoof never reached the drain target — but
+/// it still amplified per-SM CPU cost. This routing change makes the
+/// amplification impossible regardless of how the reconciler evolves.
+#[must_use]
+pub fn node_to_scheduled_machines_via_machine<'a, M>(
+    node: &k8s_openapi::api::core::v1::Node,
+    machines: M,
+) -> Vec<kube::runtime::reflector::ObjectRef<ScheduledMachine>>
+where
+    M: IntoIterator<Item = &'a kube::core::DynamicObject>,
+{
+    let Some(node_name) = node.metadata.name.as_deref() else {
+        return Vec::new();
+    };
+    if node_name.is_empty() {
+        return Vec::new();
+    }
+
+    machines
+        .into_iter()
+        .filter_map(|machine| {
+            // Canonical nodeRef from Machine.status — written only by the
+            // CAPI controller, not by the tenant.
+            let machine_node_name = machine
+                .data
+                .get("status")?
+                .get("nodeRef")?
+                .get("name")?
+                .as_str()?;
+            if machine_node_name.is_empty() || machine_node_name != node_name {
+                return None;
+            }
+
+            // Walk to the owning SM via the controller-stamped label —
+            // same primitive used by machine_to_scheduled_machine.
+            let labels = machine.metadata.labels.as_ref()?;
+            let raw_sm_name = labels.get(crate::labels::LABEL_SCHEDULED_MACHINE)?;
+            let sm_name = raw_sm_name.trim();
+            if sm_name.is_empty() {
+                return None;
+            }
+
+            let sm_namespace = machine.metadata.namespace.as_deref()?;
+
+            Some(
+                kube::runtime::reflector::ObjectRef::<ScheduledMachine>::new(sm_name)
+                    .within(sm_namespace),
+            )
+        })
+        .collect()
+}
+
 /// Cordon a Kubernetes Node (mark as unschedulable)
 ///
 /// # Errors