Fix: GroupKeyDistribute admin forgery + cap concurrent port scanners

Two pre-release fixes found during audit.

1) GroupKeyDistribute admin forgery (critical)

   `group_key_distribution::try_apply_distribution_post` trusted the
   `admin` field inside the decrypted payload without verifying it
   matched the post's author. Exploit: any peer who learns a victim's
   posting NodeId (public — appears as a recipient on any DM/group
   post) and observes a target group_id in the wild could craft an
   encrypted distribution post claiming to be from the legitimate
   admin. The victim's storage uses INSERT OR REPLACE on group_keys,
   so a successful forgery would overwrite the victim's legitimate
   group key record and stored seed, breaking future rotations / key
   distributions from the real admin.

   Fix: reject the distribution post when `content.admin != post.author`.
   Added test `forged_admin_is_rejected` that seeds a legitimate
   record, attempts a forgery, and asserts the legitimate record is
   untouched.

2) Cap concurrent port-scan hole punches at 1 (bandwidth)

   `hole_punch_with_scanning` fires ~100 QUIC ClientHellos/sec for up
   to SCAN_MAX_DURATION_SECS (300s), ~1 Mbps per active scanner. With
   no cap, the growth loop / anchor referrals / replication paths
   could spawn several scanners at once and drive sustained multi-Mbps
   upload — particularly pathological on obfuscated VPNs where every
   probe stalls at a proxy timeout, explaining the reported 10 Mbps
   sustained upload after anchor connect.

   Fix: module-level `tokio::sync::Semaphore(1)` guarding entry to the
   scanning loop. Second-and-beyond callers fall back to the cheaper
   `hole_punch_parallel` (standard punching, no 100/sec port walk)
   instead of spawning another scanner. Permit is held for the scanner
   lifetime and released on return. Added unit test
   `scanner_semaphore_caps_concurrent_scans_at_one`.

Both changes leave the successful-call path untouched (single scanner
still runs; legitimate key distributions still apply). 120 / 120 core
tests pass.

crates/core/src/connection.rs

@@ -155,6 +155,20 @@ const SCAN_PUNCH_INTERVAL_SECS: u64 = 2;
 /// Maximum scan duration (seconds) — accept the cost for otherwise-impossible connections
 const SCAN_MAX_DURATION_SECS: u64 = 300; // 5 minutes
 
+/// Global cap on concurrent port-scan hole punches. Each scanner fires
+/// ~100 QUIC ClientHellos/sec for up to `SCAN_MAX_DURATION_SECS`, which
+/// is ~1 Mbps per active scanner. Without a cap, multiple parallel
+/// referrals (growth loop, anchor referrals, replication) can spawn
+/// several scanners at once and drive sustained multi-Mbps upload —
+/// especially pathological on obfuscated VPNs where every probe stalls
+/// at proxy timeouts. A permit is acquired before the scanning loop
+/// starts and held until the scanner returns; extra callers fall back
+/// to the cheaper `hole_punch_parallel`.
+fn scanner_semaphore() -> &'static tokio::sync::Semaphore {
+    static SEM: std::sync::OnceLock<tokio::sync::Semaphore> = std::sync::OnceLock::new();
+    SEM.get_or_init(|| tokio::sync::Semaphore::new(1))
+}
+
 /// Advanced hole punch with port scanning fallback for EDM/port-restricted NAT.
 ///
 /// **Role-based behavior** (each side calls this independently):
@@ -188,6 +202,21 @@ pub(crate) async fn hole_punch_with_scanning(
         return hole_punch_parallel(endpoint, target, addresses).await;
     }
 
+    // v0.6.2: cap to one concurrent port scanner per node. Additional
+    // callers fall back to the cheaper `hole_punch_parallel` instead of
+    // spawning another 100-probes-per-second scanner. The permit is held
+    // for the lifetime of the scanner loop below (dropped on return).
+    let _scan_permit = match scanner_semaphore().try_acquire() {
+        Ok(p) => p,
+        Err(_) => {
+            tracing::debug!(
+                peer = hex::encode(target),
+                "another port scan already in progress — falling back to parallel punch"
+            );
+            return hole_punch_parallel(endpoint, target, addresses).await;
+        }
+    };
+
     // Filter to reachable families, then use observed address (first in list, injected by relay)
     let reachable = filter_reachable_families(endpoint, addresses);
     let observed_addr = reachable.first()
@@ -8379,3 +8408,21 @@ fn now_ms() -> u64 {
         .unwrap_or_default()
         .as_millis() as u64
 }
+
+#[cfg(test)]
+mod tests {
+    use super::scanner_semaphore;
+
+    #[test]
+    fn scanner_semaphore_caps_concurrent_scans_at_one() {
+        let sem = scanner_semaphore();
+        // Fresh — one permit should be available.
+        let p1 = sem.try_acquire().expect("first scan should acquire");
+        // Second concurrent caller must be rejected.
+        assert!(sem.try_acquire().is_err(), "second scan must not acquire while first holds permit");
+        // Dropping the first permit returns it to the pool.
+        drop(p1);
+        let p2 = sem.try_acquire().expect("after release, next scan should acquire");
+        drop(p2);
+    }
+}

crates/core/src/group_key_distribution.rs

@@ -96,6 +96,21 @@ pub fn try_apply_distribution_post(
                     Ok(c) => c,
                     Err(_) => continue, // Bad payload — try next persona.
                 };
+                // Critical: the `admin` claimed inside the decrypted
+                // payload must match the post author. Without this, any
+                // peer who knows a member's posting id and the group's
+                // group_id could craft an encrypted post claiming to be
+                // from the admin and overwrite the member's stored group
+                // key (create_group_key uses INSERT OR REPLACE).
+                if content.admin != post.author {
+                    tracing::warn!(
+                        post_author = hex::encode(post.author),
+                        claimed_admin = hex::encode(content.admin),
+                        group_id = hex::encode(content.group_id),
+                        "rejecting group-key-distribution post: claimed admin != post author"
+                    );
+                    continue;
+                }
                 apply_content(s, &content)?;
                 return Ok(true);
             }
@@ -173,6 +188,71 @@ mod tests {
         }
     }
 
+    #[test]
+    fn forged_admin_is_rejected() {
+        // Scenario: an attacker knows the victim's posting pubkey and the
+        // target group_id. They craft an encrypted distribution post
+        // addressed to the victim, claiming themselves as the group admin.
+        // Without the author-vs-admin check the victim would overwrite
+        // their legitimate group key record.
+        let s = temp_storage();
+        let (real_admin_sec, real_admin_id) = make_keypair(1);
+        let (attacker_sec, attacker_id) = make_keypair(9);
+        let (victim_sec, victim_id) = make_keypair(2);
+
+        // Seed the victim with a legitimate group record so we can
+        // verify it isn't overwritten by the forgery.
+        let group_id = [77u8; 32];
+        let real_pubkey = [1u8; 32];
+        let real_seed = [42u8; 32];
+        let real_record = GroupKeyRecord {
+            group_id,
+            circle_name: "real".to_string(),
+            epoch: 1,
+            group_public_key: real_pubkey,
+            admin: real_admin_id,
+            created_at: 100,
+            canonical_root_post_id: None,
+        };
+        let (_, real_post, real_vis) = build_distribution_post(
+            &real_admin_id, &real_admin_sec, &real_record, &real_seed, &[victim_id],
+        ).unwrap();
+        let victim_personas = vec![mk_persona(victim_sec, victim_id)];
+        assert!(try_apply_distribution_post(&s, &real_post, &real_vis, &victim_personas).unwrap());
+        assert_eq!(s.get_group_key(&group_id).unwrap().unwrap().admin, real_admin_id);
+
+        // Attacker authors a forgery: post.author is attacker, but the
+        // inner `admin` field claims to be the real admin.
+        let forged_content = GroupKeyDistributionContent {
+            group_id,
+            circle_name: "real".to_string(),
+            epoch: 2,
+            group_public_key: [255u8; 32],
+            admin: real_admin_id, // lies inside the encrypted payload
+            canonical_root_post_id: None,
+            group_seed: [0xFFu8; 32],
+        };
+        let plaintext = serde_json::to_string(&forged_content).unwrap();
+        let (ciphertext, wrapped) = crate::crypto::encrypt_post(
+            &plaintext, &attacker_sec, &attacker_id, &[victim_id],
+        ).unwrap();
+        let forged_post = Post {
+            author: attacker_id, // real author — attacker, not admin
+            content: ciphertext,
+            attachments: vec![],
+            timestamp_ms: 200,
+        };
+        let forged_vis = PostVisibility::Encrypted { recipients: wrapped };
+
+        let applied = try_apply_distribution_post(&s, &forged_post, &forged_vis, &victim_personas).unwrap();
+        assert!(!applied, "forged distribution post must not be applied");
+
+        // Legitimate group key must be untouched.
+        let stored = s.get_group_key(&group_id).unwrap().unwrap();
+        assert_eq!(stored.admin, real_admin_id);
+        assert_eq!(stored.group_public_key, real_pubkey);
+    }
+
     #[test]
     fn member_decrypts_and_applies() {
         let s = temp_storage();