swarm-nats-auth: grant the wanted-state bucket, scoped by direction

The bucket the previous commit adds had no grants, so the controller
could not create or write it and no hive could read its own key.

Measured against nats-server 2.14.4 rather than extended by analogy,
because the shapes are not symmetric:

- controller: `STREAM.INFO` + `STREAM.CREATE` on `KV_hive-wanted` and
  `$KV.hive-wanted.>`. With only today's status grants, `get_key_value`
  timed out and the server named the two missing stream subjects.
- hive: `STREAM.INFO` plus **one** direct-get subject carrying its own
  key. A KV read is a publish — `store.get` is a request — and the
  direct-get subject embeds the key, so the read scopes per hive. By
  analogy with `reader_subjects` this would have been `.>`, handing
  every hive every hive's wanted set.

Both refusals were verified to fire, not assumed: as `alpha`,
`get beta` was refused naming
`$JS.API.DIRECT.GET.KV_hive-wanted.$KV.hive-wanted.beta`, and
`put alpha` was refused naming `$KV.hive-wanted.alpha`.

`a_reader_may_list_and_fetch_but_not_write` asserted a reader holds no
`$KV.` subject at all, which held only while status was the sole
bucket. Narrowed to the status bucket — the invariant it defends is
that the controller cannot forge a hive's own report, and the wanted
bucket runs the other way.

Refs #3124

swarm-nats-auth/src/policy.rs

@@ -138,6 +138,16 @@ impl Policy {
         format!("KV_{}", self.bucket)
     }
 
+    /// The stream backing the wanted-state bucket.
+    ///
+    /// Derived from the crate constant rather than from a second configurable
+    /// bucket name: `wanted::BUCKET` is a `const` precisely so writer and
+    /// reader cannot disagree about it, and a flag here would reintroduce the
+    /// disagreement one layer out.
+    fn wanted_stream() -> String {
+        format!("KV_{}", swarm_queue_client::wanted::BUCKET)
+    }
+
     /// What *any* `JetStream` client must be able to ask before it can do
     /// anything at all, bucket-specific or not.
     ///
@@ -238,6 +248,21 @@ impl Policy {
         // pointing at a subject nobody publishes to. Not `extra_hive_subjects`:
         // that is for streams this crate does not know about.
         subjects.push(swarm_queue_client::notices::subject(hive));
+        // Reading the wanted-state bucket, and **only this hive's key**.
+        //
+        // A KV read is a publish: `store.get` is a request, and the direct-get
+        // subject carries the key, so the grant scopes to one hive rather than
+        // to the bucket. `$KV.<wanted>.<hive>` is deliberately absent — the
+        // controller declares this and a hive converges to it, so a hive that
+        // could write its own key could declare its own desired state.
+        subjects.extend([
+            format!("$JS.API.STREAM.INFO.{}", Self::wanted_stream()),
+            format!(
+                "$JS.API.DIRECT.GET.{}.$KV.{}.{hive}",
+                Self::wanted_stream(),
+                swarm_queue_client::wanted::BUCKET
+            ),
+        ]);
         subjects.extend(
             self.extra_hive_subjects
                 .iter()
@@ -291,6 +316,13 @@ impl Policy {
             // admission). Same failure mode as the knowledge event above: a
             // refused publish reaches the client as a timeout.
             swarm_queue_client::DEPLOY_SUBJECT_WILDCARD.to_owned(),
+            // The wanted-state bucket, which the controller creates and writes
+            // for every hive. `.>` here against one key per hive above: this is
+            // the single writer, so scoping it per hive would be a list of
+            // hives to keep, which `hive_name`'s doc argues against.
+            format!("$JS.API.STREAM.INFO.{}", Self::wanted_stream()),
+            format!("$JS.API.STREAM.CREATE.{}", Self::wanted_stream()),
+            format!("$KV.{}.>", swarm_queue_client::wanted::BUCKET),
         ]);
         subjects
     }
@@ -513,7 +545,12 @@ mod tests {
             p.publish
                 .contains(&"$JS.API.DIRECT.GET.KV_hive-status.>".to_owned())
         );
-        assert!(!p.publish.iter().any(|s| s.starts_with("$KV.")));
+        // Scoped to the status bucket, not to `$KV.` as a whole: the hives
+        // write that one and a controller able to write it could forge a
+        // hive's own report. The wanted bucket runs the other way and the
+        // controller is its writer, so a blanket assertion here would forbid
+        // the grant that bucket exists for.
+        assert!(!p.publish.iter().any(|s| s.starts_with("$KV.hive-status")));
     }
 
     #[test]
@@ -570,6 +607,59 @@ mod tests {
         assert!(!p.publish.iter().any(|s| s == "hive-notices.>"));
     }
 
+    #[test]
+    fn a_hive_may_read_its_own_wanted_key() {
+        // A KV read is a publish, and the direct-get subject carries the key.
+        let p = policy()
+            .permissions("hive-alpha")
+            .expect("a hive is admitted");
+        assert!(
+            p.publish
+                .contains(&"$JS.API.DIRECT.GET.KV_hive-wanted.$KV.hive-wanted.alpha".to_owned()),
+            "got: {:?}",
+            p.publish
+        );
+    }
+
+    #[test]
+    fn a_hive_may_not_read_another_hives_wanted_key() {
+        // The whole reason the read is scoped by key rather than by bucket.
+        let p = policy()
+            .permissions("hive-alpha")
+            .expect("a hive is admitted");
+        assert!(!p.publish.iter().any(|s| s.contains("hive-wanted.beta")));
+        assert!(
+            !p.publish
+                .iter()
+                .any(|s| s.starts_with("$JS.API.DIRECT.GET.KV_hive-wanted") && s.ends_with('>'))
+        );
+    }
+
+    #[test]
+    fn a_hive_may_not_write_its_own_wanted_key() {
+        // The controller declares wanted state; a hive that could write its own
+        // key could declare its own.
+        let p = policy()
+            .permissions("hive-alpha")
+            .expect("a hive is admitted");
+        assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted.alpha"));
+        assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted.>"));
+    }
+
+    #[test]
+    fn the_controller_may_create_and_write_the_wanted_bucket() {
+        // Presence control for the three above: the subjects exist, on the one
+        // identity that is meant to hold them.
+        let p = policy()
+            .permissions("swarm-controller")
+            .expect("a reader is admitted");
+        assert!(p.publish.contains(&"$KV.hive-wanted.>".to_owned()));
+        assert!(
+            p.publish
+                .contains(&"$JS.API.STREAM.CREATE.KV_hive-wanted".to_owned())
+        );
+    }
+
     #[test]
     fn the_notices_stream_grant_is_identical_across_hives() {
         // Unlike `$KV.<bucket>.<hive>` or an `extra_hive_subjects`