Skip to main content

max / makenotwork

mt+server: close internal-API replay chain (ultra-fuzz C2+E1) Delete the v1 HMAC dual-accept fallback in mt: a nonce is now mandatory, so a request without X-Internal-Nonce is a 401 rather than a downgrade to the replayable timestamp+body format. The MNW signer already emits v2 on every request, so there is no live v1 caller. Resolves the 3-run CHRONIC. Make the internal create_post path idempotent: add posts.external_ref (migration 030) + partial unique index and an ON CONFLICT-based create_post_external_ref mutation that returns the existing reply on replay without re-bumping last_activity_at. The server now sends mnw:post:<message-id> from both email-bridge callers, so a retried or replayed reply can no longer duplicate. 247 mt integration + 20 internal_auth unit + 2 server mt_client tests green; clippy clean both crates. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-19 02:29 UTC
Commit: 21d53712c5a5e1195db31f70d7a2039f17af71b1
Parent: a0a22c5
9 files changed, +317 insertions, -235 deletions
M multithreaded/crates/mt-db/src/mutations.rs +58
@@ -241,6 +241,64 @@ pub async fn create_post(
241 241 Ok(row.0)
242 242 }
243 243
244 + /// Idempotent reply insert keyed on `external_ref`, for the internal API.
245 + ///
246 + /// Mirrors [`create_thread_with_op_external_ref`]: a retried or replayed
247 + /// server→server reply carrying the same `external_ref` inserts at most once.
248 + /// `ON CONFLICT` on the partial unique index (m030) makes the repeat a no-op,
249 + /// so the m029 post_count trigger fires exactly once and `last_activity_at` is
250 + /// bumped only on the fresh insert. Returns `(post_id, created)` — `created` is
251 + /// false when an existing post was returned. User-facing replies use the plain
252 + /// [`create_post`] (no ref); this path is internal-only.
253 + #[tracing::instrument(skip_all)]
254 + pub async fn create_post_external_ref(
255 + pool: &PgPool,
256 + thread_id: Uuid,
257 + author_id: Uuid,
258 + external_ref: &str,
259 + body_markdown: &str,
260 + body_html: &str,
261 + ) -> Result<(Uuid, bool), sqlx::Error> {
262 + let mut tx = pool.begin().await?;
263 +
264 + let inserted: Option<(Uuid,)> = sqlx::query_as(
265 + "INSERT INTO posts (thread_id, author_id, body_markdown, body_html, external_ref)
266 + VALUES ($1, $2, $3, $4, $5)
267 + ON CONFLICT (external_ref) WHERE external_ref IS NOT NULL DO NOTHING
268 + RETURNING id",
269 + )
270 + .bind(thread_id)
271 + .bind(author_id)
272 + .bind(body_markdown)
273 + .bind(body_html)
274 + .bind(external_ref)
275 + .fetch_optional(&mut *tx)
276 + .await?;
277 +
278 + let result = match inserted {
279 + Some((id,)) => {
280 + // Fresh reply: bump thread activity (the trigger already counted it).
281 + sqlx::query("UPDATE threads SET last_activity_at = now() WHERE id = $1")
282 + .bind(thread_id)
283 + .execute(&mut *tx)
284 + .await?;
285 + (id, true)
286 + }
287 + None => {
288 + // Replay: the reply already exists. Return its id, don't re-bump.
289 + let existing: (Uuid,) =
290 + sqlx::query_as("SELECT id FROM posts WHERE external_ref = $1")
291 + .bind(external_ref)
292 + .fetch_one(&mut *tx)
293 + .await?;
294 + (existing.0, false)
295 + }
296 + };
297 +
298 + tx.commit().await?;
299 + Ok(result)
300 + }
301 +
244 302 /// Insert a footnote on a post. Returns the footnote ID.
245 303 #[tracing::instrument(skip_all)]
246 304 pub async fn insert_footnote(
A multithreaded/migrations/030_post_external_ref.sql +9
@@ -0,0 +1,9 @@
1 + -- Idempotency key for internally-created replies, mirroring threads.external_ref
2 + -- (m021). The internal `POST /internal/threads/{id}/posts` path is server→server
3 + -- and retry-prone (network blip, MNW-side retry, or a replayed request); without
4 + -- a dedup key a retry inserts a duplicate reply. The unique index makes a repeat
5 + -- insert with the same ref a no-op (ON CONFLICT DO NOTHING in the mutation),
6 + -- matching the thread path. Partial so user-facing replies (external_ref NULL)
7 + -- are unconstrained.
8 + ALTER TABLE posts ADD COLUMN external_ref TEXT;
9 + CREATE UNIQUE INDEX idx_posts_external_ref ON posts (external_ref) WHERE external_ref IS NOT NULL;
M multithreaded/src/internal_auth.rs +131 -229
@@ -1,19 +1,17 @@
1 1 //! HMAC-SHA256 authentication for internal API requests from MNW.
2 2 //!
3 - //! The v2 signed message binds method + path + nonce as well as timestamp +
4 - //! body — `HMAC-SHA256(timestamp \n METHOD \n PATH \n NONCE \n body)` — sent in
3 + //! The signed message binds method + path + nonce as well as timestamp + body —
4 + //! `HMAC-SHA256(timestamp \n METHOD \n PATH \n NONCE \n body)` — sent in
5 5 //! `X-Internal-{Timestamp,Signature,Nonce}`. Binding method+path stops a
6 6 //! captured signature being replayed to a different endpoint; the nonce, checked
7 7 //! against a single-use cache, stops it being re-sent at all within the 60s
8 8 //! freshness window.
9 9 //!
10 - //! **Lockstep rollout:** this verifier is in the dual-accept transition state —
11 - //! a request with no `X-Internal-Nonce` is verified against the legacy v1
12 - //! message (timestamp+body) for compatibility with a not-yet-upgraded MNW
13 - //! signer. Once the server signer is fully deployed, TIGHTEN: require a nonce
14 - //! and delete the v1 fallback (`verify_internal_signature` + the `None` branch
15 - //! of `verify_signed_request`). Until then there is a brief window where a v1
16 - //! signature is replayable — keep the dual-accept period short.
10 + //! A nonce is **mandatory**: there is exactly one verification path. A request
11 + //! with no `X-Internal-Nonce` is rejected outright (401), not downgraded — the
12 + //! legacy v1 (timestamp+body) format and its dual-accept fallback were deleted
13 + //! once the MNW signer moved fully to v2, closing the replay window an attacker
14 + //! could otherwise select by simply omitting the nonce header.
17 15
18 16 use std::collections::HashMap;
19 17 use std::sync::{LazyLock, Mutex};
@@ -116,8 +114,8 @@ impl FromRequest<AppState> for InternalAuth {
116 114 )
117 115 .map_err(|(status, msg)| (status, msg).into_response())?;
118 116
119 - // Single-use: reject a replayed nonce (only meaningful once the v2
120 - // signer is live; v1 requests carry no nonce).
117 + // Single-use: reject a replayed nonce. `verify_signed_request` has
118 + // already guaranteed the nonce is present, so this always runs.
121 119 if let Some(nonce) = nonce_header.as_deref()
122 120 && !record_nonce(nonce, now)
123 121 {
@@ -128,28 +126,15 @@ impl FromRequest<AppState> for InternalAuth {
128 126 }
129 127 }
130 128
131 - /// Compute the hex-encoded HMAC-SHA256 signature for an internal request.
132 - /// `secret` may be any length (HMAC-SHA256 accepts any key length).
133 - pub(crate) fn compute_internal_signature(secret: &str, timestamp_str: &str, body: &[u8]) -> String {
134 - // MAC over raw bytes (`timestamp\n` ++ body), not a lossy UTF-8 string. For
135 - // valid-UTF-8 bodies (all our JSON) this is byte-identical to the old
136 - // `format!`-based message, so it stays wire-compatible with MNW's signer;
137 - // it also closes the latent hole where two distinct non-UTF-8 bodies both
138 - // collapsed to the empty string and signed identically.
139 - let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes())
140 - .expect("HMAC-SHA256 accepts any key length");
141 - mac.update(timestamp_str.as_bytes());
142 - mac.update(b"\n");
143 - mac.update(body);
144 - hex::encode(mac.finalize().into_bytes())
145 - }
146 -
147 - /// Compute the v2 signature, which binds method + path + nonce in addition to
129 + /// Compute the signature, which binds method + path + nonce in addition to
148 130 /// timestamp + body. The canonical message is newline-delimited with a fixed
149 131 /// field order, body last so an embedded newline in the body can never be
150 132 /// confused with a field separator:
151 133 /// `timestamp \n METHOD \n PATH \n NONCE \n <raw body bytes>`
152 134 /// METHOD is uppercase ASCII, PATH is the request path only (no query string).
135 + /// Body is MAC'd as raw bytes, not a lossy UTF-8 string, closing the latent
136 + /// hole where two distinct non-UTF-8 bodies both collapsed to "" and signed
137 + /// identically.
153 138 pub(crate) fn compute_internal_signature_v2(
154 139 secret: &str,
155 140 timestamp_str: &str,
@@ -186,41 +171,13 @@ fn check_freshness(timestamp_str: &str, now_unix: i64) -> Result<i64, (StatusCod
186 171 Ok(timestamp)
187 172 }
188 173
189 - /// Pure verification of the legacy (v1) message: timestamp + body only.
190 - /// Retained as the back-compat path during the lockstep rollout (a request with
191 - /// no `X-Internal-Nonce` is assumed to come from a pre-upgrade signer).
174 + /// Verify a signed internal request, binding method + path + nonce. A nonce is
175 + /// mandatory — a request without one is rejected (401), never downgraded. This
176 + /// is the single verification path; the legacy v1 (timestamp+body) fallback was
177 + /// deleted once the MNW signer moved fully to v2. Freshness is checked first.
192 178 ///
193 179 /// Headers are passed as `Option<&str>` so callers can extract them with any
194 180 /// strategy (axum `HeaderMap`, manual `Bytes`, tests).
195 - pub(crate) fn verify_internal_signature(
196 - secret: &str,
197 - timestamp_header: Option<&str>,
198 - signature_header: Option<&str>,
199 - body: &[u8],
200 - now_unix: i64,
201 - ) -> Result<(), (StatusCode, &'static str)> {
202 - let timestamp_str = timestamp_header
203 - .ok_or((StatusCode::UNAUTHORIZED, "Missing X-Internal-Timestamp"))?;
204 - let signature = signature_header
205 - .ok_or((StatusCode::UNAUTHORIZED, "Missing X-Internal-Signature"))?;
206 -
207 - check_freshness(timestamp_str, now_unix)?;
208 -
209 - let expected = compute_internal_signature(secret, timestamp_str, body);
210 - if !constant_time_eq(expected.as_bytes(), signature.as_bytes()) {
211 - return Err((StatusCode::UNAUTHORIZED, "Invalid signature"));
212 - }
213 - Ok(())
214 - }
215 -
216 - /// Verify a signed internal request, binding method + path + nonce when the
217 - /// signer supplied a nonce (the v2 format), and falling back to the legacy v1
218 - /// message otherwise.
219 - ///
220 - /// This is the dual-accept transition state: MT accepts both an upgraded signer
221 - /// (nonce present → v2 + replay protection) and a pre-upgrade signer (no nonce →
222 - /// v1). Once the server signer is fully rolled out, tighten this to require a
223 - /// nonce and drop the v1 branch. Freshness is always checked first.
224 181 ///
225 182 /// Nonce replay is NOT checked here (that is stateful); the caller records the
226 183 /// nonce via [`record_nonce`] after this returns Ok.
@@ -235,10 +192,7 @@ pub(crate) fn verify_signed_request(
235 192 body: &[u8],
236 193 now_unix: i64,
237 194 ) -> Result<(), (StatusCode, &'static str)> {
238 - let Some(nonce) = nonce_header else {
239 - // No nonce → legacy signer. Verify the v1 message for back-compat.
240 - return verify_internal_signature(secret, timestamp_header, signature_header, body, now_unix);
241 - };
195 + let nonce = nonce_header.ok_or((StatusCode::UNAUTHORIZED, "Missing X-Internal-Nonce"))?;
242 196
243 197 let timestamp_str = timestamp_header
244 198 .ok_or((StatusCode::UNAUTHORIZED, "Missing X-Internal-Timestamp"))?;
@@ -323,30 +277,15 @@ mod tests {
323 277 assert!(!constant_time_eq(b"hello", b"hell"));
324 278 }
325 279
326 - #[test]
327 - fn hmac_signature_roundtrip() {
328 - let secret = "test-secret";
329 - let timestamp = "1234567890";
330 - let body = r#"{"name":"test"}"#;
331 - let message = format!("{}\n{}", timestamp, body);
332 -
333 - let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes()).unwrap();
334 - mac.update(message.as_bytes());
335 - let sig = hex::encode(mac.finalize().into_bytes());
336 -
337 - // Verify the same computation matches
338 - let mut mac2 = Hmac::<Sha256>::new_from_slice(secret.as_bytes()).unwrap();
339 - mac2.update(message.as_bytes());
340 - let expected = hex::encode(mac2.finalize().into_bytes());
341 -
342 - assert!(constant_time_eq(sig.as_bytes(), expected.as_bytes()));
343 - }
280 + // ── compute_internal_signature_v2 pins HMAC message construction ──
344 281
345 - // ── compute_internal_signature pins HMAC message construction ──
282 + fn v2(secret: &str, ts: &str, method: &str, path: &str, nonce: &str, body: &[u8]) -> String {
283 + compute_internal_signature_v2(secret, ts, method, path, nonce, body)
284 + }
346 285
347 286 #[test]
348 287 fn signature_is_64_hex_chars() {
349 - let sig = compute_internal_signature("secret", "100", b"body");
288 + let sig = v2("secret", "100", "POST", "/x", "n", b"body");
350 289 assert_eq!(sig.len(), 64, "SHA-256 hex is 64 chars");
351 290 assert!(sig.chars().all(|c| c.is_ascii_hexdigit()));
352 291 }
@@ -354,172 +293,78 @@ mod tests {
354 293 #[test]
355 294 fn signature_changes_with_secret() {
356 295 // Pins that the secret feeds into the MAC key.
357 - let s1 = compute_internal_signature("alpha", "100", b"body");
358 - let s2 = compute_internal_signature("beta", "100", b"body");
359 - assert_ne!(s1, s2);
360 - }
361 -
362 - #[test]
363 - fn signature_changes_with_timestamp() {
364 - // Pins the `format!("{}\n{}", timestamp_str, body)` ordering — a
365 - // mutation that drops the timestamp or swaps the order would make
366 - // these two signatures match.
367 - let s1 = compute_internal_signature("secret", "100", b"body");
368 - let s2 = compute_internal_signature("secret", "101", b"body");
369 - assert_ne!(s1, s2);
296 + assert_ne!(
297 + v2("alpha", "100", "POST", "/x", "n", b"body"),
298 + v2("beta", "100", "POST", "/x", "n", b"body"),
299 + );
370 300 }
371 301
372 302 #[test]
373 - fn signature_changes_with_body() {
374 - let s1 = compute_internal_signature("secret", "100", b"hello");
375 - let s2 = compute_internal_signature("secret", "100", b"hello!");
376 - assert_ne!(s1, s2);
303 + fn signature_changes_with_each_bound_field() {
304 + // Pins that timestamp, method, path, nonce, and body each feed the MAC —
305 + // a mutation dropping any field would collide one of these pairs.
306 + let base = v2("s", "100", "POST", "/x", "n", b"body");
307 + assert_ne!(base, v2("s", "101", "POST", "/x", "n", b"body"), "timestamp bound");
308 + assert_ne!(base, v2("s", "100", "GET", "/x", "n", b"body"), "method bound");
309 + assert_ne!(base, v2("s", "100", "POST", "/y", "n", b"body"), "path bound");
310 + assert_ne!(base, v2("s", "100", "POST", "/x", "m", b"body"), "nonce bound");
311 + assert_ne!(base, v2("s", "100", "POST", "/x", "n", b"body!"), "body bound");
377 312 }
378 313
379 314 #[test]
380 - fn signature_separator_is_newline_not_concat() {
381 - // Pins `format!("{}\n{}", ...)` — without the `\n`, "1" + "00body"
382 - // would collide with "10" + "0body".
383 - let collision_a = compute_internal_signature("secret", "1", b"00body");
384 - let collision_b = compute_internal_signature("secret", "10", b"0body");
315 + fn signature_separators_are_newlines_not_concat() {
316 + // Without the `\n` delimiters, field-boundary ambiguity would let two
317 + // distinct messages collide (e.g. ts "1"+"00body" vs "10"+"0body").
385 318 assert_ne!(
386 - collision_a, collision_b,
387 - "missing newline separator allows length-ambiguity collision"
319 + v2("s", "1", "POST", "/x", "n", b"00body"),
320 + v2("s", "10", "POST", "/x", "n", b"0body"),
321 + "missing separator allows length-ambiguity collision"
388 322 );
389 323 }
390 324
391 - // ── verify_internal_signature freshness + signature check ──
392 -
393 - fn valid(secret: &str, ts: &str, body: &[u8]) -> String {
394 - compute_internal_signature(secret, ts, body)
395 - }
325 + // ── verify_signed_request: signature + freshness + mandatory nonce ──
396 326
397 327 #[test]
398 328 fn verify_accepts_valid_signature_at_now() {
399 - let secret = "s";
400 - let body = b"abc";
401 - let ts = "1000";
402 - let sig = valid(secret, ts, body);
403 - assert!(verify_internal_signature(secret, Some(ts), Some(&sig), body, 1000).is_ok());
329 + let sig = v2("s", "1000", "POST", "/internal/x", "abc", b"body");
330 + assert!(verify_signed_request(
331 + "s", Some("1000"), Some(&sig), "POST", "/internal/x", Some("abc"), b"body", 1000
332 + )
333 + .is_ok());
404 334 }
405 335
406 336 #[test]
407 337 fn verify_rejects_wrong_signature() {
408 - let secret = "s";
409 - let body = b"abc";
410 - let ts = "1000";
411 - // Tamper with one hex char.
412 - let mut sig = valid(secret, ts, body);
338 + let mut sig = v2("s", "1000", "POST", "/internal/x", "abc", b"body");
413 339 let first = sig.remove(0);
414 340 sig.insert(0, if first == '0' { '1' } else { '0' });
415 - let (status, _) =
416 - verify_internal_signature(secret, Some(ts), Some(&sig), body, 1000).unwrap_err();
341 + let (status, _) = verify_signed_request(
342 + "s", Some("1000"), Some(&sig), "POST", "/internal/x", Some("abc"), b"body", 1000,
343 + )
344 + .unwrap_err();
417 345 assert_eq!(status, StatusCode::UNAUTHORIZED);
418 346 }
419 347
420 348 #[test]
421 349 fn verify_rejects_wrong_secret() {
422 - let body = b"abc";
423 - let ts = "1000";
424 - let sig = valid("real-secret", ts, body);
425 - assert!(
426 - verify_internal_signature("wrong-secret", Some(ts), Some(&sig), body, 1000).is_err()
427 - );
350 + let sig = v2("real-secret", "1000", "POST", "/internal/x", "abc", b"body");
351 + assert!(verify_signed_request(
352 + "wrong-secret", Some("1000"), Some(&sig), "POST", "/internal/x", Some("abc"), b"body", 1000
353 + )
354 + .is_err());
428 355 }
429 356
430 357 #[test]
431 358 fn verify_rejects_tampered_body() {
432 - let secret = "s";
433 - let ts = "1000";
434 - let sig = valid(secret, ts, b"original");
435 - assert!(verify_internal_signature(secret, Some(ts), Some(&sig), b"tampered", 1000).is_err());
436 - }
437 -
438 - #[test]
439 - fn verify_at_window_boundary_accepts_inside_rejects_outside() {
440 - // Asymmetric window: up to MAX_TIMESTAMP_AGE_SECS (60s) old, but only
441 - // MAX_FUTURE_SKEW_SECS (5s) into the future. `>` is strict, so exactly
442 - // at each boundary is accepted.
443 - let secret = "s";
444 - let body = b"abc";
445 - let ts = "1000";
446 - let sig = valid(secret, ts, body);
447 -
448 - // now - ts = 60 → accepted (at the age boundary)
449 - assert!(verify_internal_signature(secret, Some(ts), Some(&sig), body, 1060).is_ok());
450 - // now - ts = 61 → rejected (too old)
451 - assert!(verify_internal_signature(secret, Some(ts), Some(&sig), body, 1061).is_err());
452 - // ts - now = 5 → accepted (at the future-skew boundary)
453 - assert!(verify_internal_signature(secret, Some(ts), Some(&sig), body, 995).is_ok());
454 - // ts - now = 6 → rejected (too far in the future)
455 - assert!(verify_internal_signature(secret, Some(ts), Some(&sig), body, 994).is_err());
456 - }
457 -
458 - #[test]
459 - fn verify_rejects_missing_timestamp_header() {
460 - let secret = "s";
461 - let body = b"abc";
462 - let sig = valid(secret, "1000", body);
463 - let (status, msg) =
464 - verify_internal_signature(secret, None, Some(&sig), body, 1000).unwrap_err();
465 - assert_eq!(status, StatusCode::UNAUTHORIZED);
466 - assert!(msg.contains("Timestamp"));
467 - }
468 -
469 - #[test]
470 - fn verify_rejects_missing_signature_header() {
471 - let (status, msg) =
472 - verify_internal_signature("s", Some("1000"), None, b"abc", 1000).unwrap_err();
473 - assert_eq!(status, StatusCode::UNAUTHORIZED);
474 - assert!(msg.contains("Signature"));
475 - }
476 -
477 - #[test]
478 - fn verify_rejects_unparseable_timestamp() {
479 - let (status, msg) =
480 - verify_internal_signature("s", Some("not-an-int"), Some("zz"), b"", 1000).unwrap_err();
481 - assert_eq!(status, StatusCode::UNAUTHORIZED);
482 - assert!(msg.contains("Invalid timestamp"));
483 - }
484 -
485 - #[test]
486 - fn verify_check_order_missing_timestamp_first() {
487 - // Both headers missing: timestamp check fires first.
488 - let (_, msg) = verify_internal_signature("s", None, None, b"", 1000).unwrap_err();
489 - assert!(msg.contains("Timestamp"), "expected timestamp msg first, got: {msg}");
490 - }
491 -
492 - #[test]
493 - fn verify_check_order_freshness_before_signature() {
494 - // A stale timestamp must reject even when the (otherwise valid) sig
495 - // matches. Catches a mutation that runs the freshness check after
496 - // signature verification.
497 - let secret = "s";
498 - let body = b"abc";
499 - let ts = "1000";
500 - let sig = valid(secret, ts, body);
501 - let (_, msg) =
502 - verify_internal_signature(secret, Some(ts), Some(&sig), body, 9999).unwrap_err();
503 - assert!(msg.contains("Timestamp"), "expected freshness msg, got: {msg}");
504 - }
505 -
506 - // ── v2 (method+path+nonce) verification + nonce replay ──
507 -
508 - fn v2(secret: &str, ts: &str, method: &str, path: &str, nonce: &str, body: &[u8]) -> String {
509 - compute_internal_signature_v2(secret, ts, method, path, nonce, body)
510 - }
511 -
512 - #[test]
513 - fn verify_v2_accepts_matching_method_path_nonce() {
514 - let sig = v2("s", "1000", "POST", "/internal/x", "abc", b"body");
359 + let sig = v2("s", "1000", "POST", "/internal/x", "abc", b"original");
515 360 assert!(verify_signed_request(
516 - "s", Some("1000"), Some(&sig), "POST", "/internal/x", Some("abc"), b"body", 1000
361 + "s", Some("1000"), Some(&sig), "POST", "/internal/x", Some("abc"), b"tampered", 1000
517 362 )
518 - .is_ok());
363 + .is_err());
519 364 }
520 365
521 366 #[test]
522 - fn verify_v2_rejects_wrong_method() {
367 + fn verify_rejects_wrong_method() {
523 368 let sig = v2("s", "1000", "GET", "/internal/x", "abc", b"body");
524 369 assert!(verify_signed_request(
525 370 "s", Some("1000"), Some(&sig), "POST", "/internal/x", Some("abc"), b"body", 1000
@@ -528,7 +373,7 @@ mod tests {
528 373 }
529 374
530 375 #[test]
531 - fn verify_v2_rejects_wrong_path() {
376 + fn verify_rejects_wrong_path() {
532 377 let sig = v2("s", "1000", "POST", "/internal/a", "abc", b"body");
533 378 assert!(verify_signed_request(
534 379 "s", Some("1000"), Some(&sig), "POST", "/internal/b", Some("abc"), b"body", 1000
@@ -537,7 +382,7 @@ mod tests {
537 382 }
538 383
539 384 #[test]
540 - fn verify_v2_rejects_wrong_nonce() {
385 + fn verify_rejects_wrong_nonce() {
541 386 let sig = v2("s", "1000", "POST", "/internal/x", "abc", b"body");
542 387 assert!(verify_signed_request(
543 388 "s", Some("1000"), Some(&sig), "POST", "/internal/x", Some("zzz"), b"body", 1000
@@ -546,24 +391,81 @@ mod tests {
546 391 }
547 392
548 393 #[test]
549 - fn verify_v2_freshness_still_enforced() {
394 + fn verify_at_window_boundary_accepts_inside_rejects_outside() {
395 + // Asymmetric window: up to MAX_TIMESTAMP_AGE_SECS (60s) old, but only
396 + // MAX_FUTURE_SKEW_SECS (5s) into the future. `>` is strict, so exactly
397 + // at each boundary is accepted.
398 + let sig = v2("s", "1000", "POST", "/x", "abc", b"abc");
399 + let check = |now| verify_signed_request(
400 + "s", Some("1000"), Some(&sig), "POST", "/x", Some("abc"), b"abc", now,
401 + );
402 + assert!(check(1060).is_ok(), "now-ts=60 accepted (age boundary)");
403 + assert!(check(1061).is_err(), "now-ts=61 rejected (too old)");
404 + assert!(check(995).is_ok(), "ts-now=5 accepted (future-skew boundary)");
405 + assert!(check(994).is_err(), "ts-now=6 rejected (too far future)");
406 + }
407 +
408 + #[test]
409 + fn verify_rejects_missing_nonce() {
410 + // A request with no nonce is rejected outright — no v1 downgrade exists.
550 411 let sig = v2("s", "1000", "POST", "/internal/x", "abc", b"body");
551 - let (_, msg) = verify_signed_request(
552 - "s", Some("1000"), Some(&sig), "POST", "/internal/x", Some("abc"), b"body", 9999,
412 + let (status, msg) = verify_signed_request(
413 + "s", Some("1000"), Some(&sig), "POST", "/internal/x", None, b"body", 1000,
553 414 )
554 415 .unwrap_err();
416 + assert_eq!(status, StatusCode::UNAUTHORIZED);
417 + assert!(msg.contains("Nonce"), "expected nonce msg, got: {msg}");
418 + }
419 +
420 + #[test]
421 + fn verify_rejects_missing_timestamp_header() {
422 + let sig = v2("s", "1000", "POST", "/x", "abc", b"abc");
423 + let (status, msg) = verify_signed_request(
424 + "s", None, Some(&sig), "POST", "/x", Some("abc"), b"abc", 1000,
425 + )
426 + .unwrap_err();
427 + assert_eq!(status, StatusCode::UNAUTHORIZED);
555 428 assert!(msg.contains("Timestamp"));
556 429 }
557 430
558 431 #[test]
559 - fn verify_without_nonce_falls_back_to_v1() {
560 - // A request with no nonce is verified against the legacy v1 message
561 - // (timestamp+body), ignoring method/path — the dual-accept path.
562 - let v1_sig = compute_internal_signature("s", "1000", b"body");
563 - assert!(verify_signed_request(
564 - "s", Some("1000"), Some(&v1_sig), "POST", "/internal/x", None, b"body", 1000
432 + fn verify_rejects_missing_signature_header() {
433 + let (status, msg) = verify_signed_request(
434 + "s", Some("1000"), None, "POST", "/x", Some("abc"), b"abc", 1000,
565 435 )
566 - .is_ok());
436 + .unwrap_err();
437 + assert_eq!(status, StatusCode::UNAUTHORIZED);
438 + assert!(msg.contains("Signature"));
439 + }
440 +
441 + #[test]
442 + fn verify_rejects_unparseable_timestamp() {
443 + let (status, msg) = verify_signed_request(
444 + "s", Some("not-an-int"), Some("zz"), "POST", "/x", Some("abc"), b"", 1000,
445 + )
446 + .unwrap_err();
447 + assert_eq!(status, StatusCode::UNAUTHORIZED);
448 + assert!(msg.contains("Invalid timestamp"));
449 + }
450 +
451 + #[test]
452 + fn verify_check_order_nonce_before_timestamp() {
453 + // Nonce mandatory: a missing nonce rejects even with all else missing.
454 + let (_, msg) = verify_signed_request("s", None, None, "POST", "/x", None, b"", 1000)
455 + .unwrap_err();
456 + assert!(msg.contains("Nonce"), "expected nonce msg first, got: {msg}");
457 + }
458 +
459 + #[test]
460 + fn verify_check_order_freshness_before_signature() {
461 + // A stale timestamp must reject even when the sig is otherwise valid —
462 + // catches a mutation running the freshness check after signature verify.
463 + let sig = v2("s", "1000", "POST", "/x", "abc", b"abc");
464 + let (_, msg) = verify_signed_request(
465 + "s", Some("1000"), Some(&sig), "POST", "/x", Some("abc"), b"abc", 9999,
466 + )
467 + .unwrap_err();
468 + assert!(msg.contains("Timestamp"), "expected freshness msg, got: {msg}");
567 469 }
568 470
569 471 #[test]
M multithreaded/src/routes/internal.rs +9 -3
@@ -60,11 +60,13 @@ pub struct CreatePostRequest {
60 60 pub author_mnw_id: Uuid,
61 61 pub author_username: String,
62 62 pub author_display_name: Option<String>,
63 + pub external_ref: String,
63 64 }
64 65
65 66 #[derive(Serialize)]
66 67 pub struct CreatePostResponse {
67 68 pub post_id: Uuid,
69 + pub created: bool,
68 70 }
69 71
70 72 #[derive(Serialize)]
@@ -339,12 +341,14 @@ async fn create_post(
339 341 .await
340 342 .map_err(db_error)?;
341 343
342 - // Render markdown and create post
344 + // Render markdown and create post (idempotent on external_ref: a retried or
345 + // replayed call returns the existing reply instead of duplicating it).
343 346 let body_html = super::render_markdown(&req.body_markdown);
344 - let post_id = mt_db::mutations::create_post(
347 + let (post_id, created) = mt_db::mutations::create_post_external_ref(
345 348 &state.db,
346 349 thread_id,
347 350 req.author_mnw_id,
351 + &req.external_ref,
348 352 &req.body_markdown,
349 353 &body_html,
350 354 )
@@ -354,10 +358,12 @@ async fn create_post(
354 358 tracing::info!(
355 359 thread_id = %thread_id,
356 360 post_id = %post_id,
361 + external_ref = %req.external_ref,
362 + created,
357 363 "internal: post created"
358 364 );
359 365
360 - Ok(Json(CreatePostResponse { post_id }))
366 + Ok(Json(CreatePostResponse { post_id, created }))
361 367 }
362 368
363 369 /// Build the internal API router. Registered outside CSRF/session middleware.
M multithreaded/tests/workflows/internal_api.rs +21 -2
@@ -419,7 +419,8 @@ async fn create_post_happy_path() {
419 419 "body_markdown": "This is a reply via internal API",
420 420 "author_mnw_id": owner_id,
421 421 "author_username": "postuser",
422 - "author_display_name": "Post User"
422 + "author_display_name": "Post User",
423 + "external_ref": "mnw:post:reply-1"
423 424 });
424 425 let (status, text) = h
425 426 .signed_post(
@@ -430,12 +431,29 @@ async fn create_post_happy_path() {
430 431 assert_eq!(status, StatusCode::OK, "body: {}", text);
431 432
432 433 let post_resp: serde_json::Value = serde_json::from_str(&text).unwrap();
433 - assert!(post_resp["post_id"].as_str().is_some());
434 + let first_post_id = post_resp["post_id"].as_str().unwrap().to_string();
435 + assert!(post_resp["created"].as_bool().unwrap(), "first call creates");
434 436
435 437 // Verify thread now has 2 posts
436 438 let (_, stats_text) = h.get(&format!("/internal/threads/{}/stats", thread_id)).await;
437 439 let stats: serde_json::Value = serde_json::from_str(&stats_text).unwrap();
438 440 assert_eq!(stats["post_count"].as_i64().unwrap(), 2);
441 +
442 + // Replay the same external_ref (a retried/replayed call): same post id,
443 + // created=false, and the post count does not grow (audit E1).
444 + let (status, text2) = h
445 + .signed_post(
446 + &format!("/internal/threads/{}/posts", thread_id),
447 + &reply_body.to_string(),
448 + )
449 + .await;
450 + assert_eq!(status, StatusCode::OK, "body: {}", text2);
451 + let replay: serde_json::Value = serde_json::from_str(&text2).unwrap();
452 + assert_eq!(replay["post_id"].as_str().unwrap(), first_post_id, "replay returns original");
453 + assert!(!replay["created"].as_bool().unwrap(), "replay does not create");
454 + let (_, stats_text2) = h.get(&format!("/internal/threads/{}/stats", thread_id)).await;
455 + let stats2: serde_json::Value = serde_json::from_str(&stats_text2).unwrap();
456 + assert_eq!(stats2["post_count"].as_i64().unwrap(), 2, "no duplicate reply");
439 457 }
440 458
441 459 #[tokio::test]
@@ -447,6 +465,7 @@ async fn create_post_nonexistent_thread() {
447 465 "body_markdown": "Reply to nothing",
448 466 "author_mnw_id": Uuid::new_v4(),
449 467 "author_username": "nobody",
468 + "external_ref": "mnw:post:orphan-reply",
450 469 });
451 470 let (status, _) = h
452 471 .signed_post(
M multithreaded/tests/workflows/mutations.rs +79
@@ -469,6 +469,85 @@ async fn toggle_endorsement_db_roundtrip() {
469 469 // ============================================================================
470 470
471 471 #[tokio::test]
472 + async fn create_post_external_ref_is_idempotent() {
473 + // The internal reply path dedups on external_ref: a retried/replayed call
474 + // returns the existing post and never inserts a duplicate (audit E1).
475 + let h = TestHarness::new().await;
476 + let comm_id = h.create_community("Test", "test").await;
477 + let cat_id = h.create_category(comm_id, "General", "general").await;
478 +
479 + let author = Uuid::new_v4();
480 + sqlx::query("INSERT INTO users (mnw_account_id, username) VALUES ($1, $2)")
481 + .bind(author)
482 + .bind("author")
483 + .execute(&h.db)
484 + .await
485 + .unwrap();
486 +
487 + let (thread_id, _op) = mt_db::mutations::create_thread_with_op(
488 + &h.db, cat_id, author, "T", "op", "<p>op</p>",
489 + )
490 + .await
491 + .unwrap();
492 +
493 + let count = || async {
494 + let n: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM posts WHERE thread_id = $1")
495 + .bind(thread_id)
496 + .fetch_one(&h.db)
497 + .await
498 + .unwrap();
499 + n.0
500 + };
501 + let activity = || async {
502 + let a: (chrono::DateTime<chrono::Utc>,) =
503 + sqlx::query_as("SELECT last_activity_at FROM threads WHERE id = $1")
504 + .bind(thread_id)
505 + .fetch_one(&h.db)
506 + .await
507 + .unwrap();
508 + a.0
509 + };
510 +
511 + // First call: fresh reply, created = true.
512 + let (id1, created1) = mt_db::mutations::create_post_external_ref(
513 + &h.db, thread_id, author, "mnw:post:msg-1", "reply", "<p>reply</p>",
514 + )
515 + .await
516 + .unwrap();
517 + assert!(created1, "first call creates the reply");
518 + assert_eq!(count().await, 2, "OP + one reply");
519 + let count_for_trigger: (i32,) = sqlx::query_as("SELECT post_count FROM threads WHERE id = $1")
520 + .bind(thread_id)
521 + .fetch_one(&h.db)
522 + .await
523 + .unwrap();
524 + assert_eq!(count_for_trigger.0, 2, "m029 trigger counted the reply once");
525 + let activity_after_first = activity().await;
526 +
527 + // Replay with the same external_ref: same post id, created = false, no dup,
528 + // and last_activity_at is NOT re-bumped.
529 + let (id2, created2) = mt_db::mutations::create_post_external_ref(
530 + &h.db, thread_id, author, "mnw:post:msg-1", "reply", "<p>reply</p>",
531 + )
532 + .await
533 + .unwrap();
534 + assert_eq!(id1, id2, "replay returns the original post id");
535 + assert!(!created2, "replay does not create");
536 + assert_eq!(count().await, 2, "no duplicate reply inserted");
537 + assert_eq!(activity().await, activity_after_first, "replay does not re-bump activity");
538 +
539 + // A distinct external_ref does insert a new reply.
540 + let (id3, created3) = mt_db::mutations::create_post_external_ref(
541 + &h.db, thread_id, author, "mnw:post:msg-2", "reply2", "<p>reply2</p>",
542 + )
543 + .await
544 + .unwrap();
545 + assert_ne!(id3, id1);
546 + assert!(created3);
547 + assert_eq!(count().await, 3, "distinct ref adds a reply");
548 + }
549 +
550 + #[tokio::test]
472 551 async fn insert_flag_idempotent_per_user() {
473 552 let h = TestHarness::new().await;
474 553 let comm_id = h.create_community("Test", "test").await;
M server/src/mt_client.rs +6
@@ -73,11 +73,17 @@ pub struct CreatePostRequest {
73 73 pub author_mnw_id: Uuid,
74 74 pub author_username: String,
75 75 pub author_display_name: Option<String>,
76 + /// Idempotency key — MT dedups a retried/replayed reply on this. Use a
77 + /// stable per-message value (e.g. `mnw:post:<email-message-id>`).
78 + pub external_ref: String,
76 79 }
77 80
78 81 #[derive(Deserialize)]
79 82 pub struct CreatePostResponse {
80 83 pub post_id: Uuid,
84 + /// False when MT returned an existing reply (the ref was already seen).
85 + #[serde(default)]
86 + pub created: bool,
81 87 }
82 88
83 89 #[derive(Deserialize)]
M server/src/routes/postmark/issues.rs +3 -1
@@ -296,7 +296,7 @@ async fn handle_issue_reply(
296 296 }
297 297
298 298 // Bridge reply into the issue's MT thread (if one exists).
299 - bridge_issue_reply_to_mt(state, &issue, sender.id, &sender.username, sender.display_name.as_deref(), body_md).await;
299 + bridge_issue_reply_to_mt(state, &issue, sender.id, &sender.username, sender.display_name.as_deref(), body_md, &payload.message_id).await;
300 300
301 301 tracing::info!(
302 302 issue_id = %issue.id,
@@ -471,6 +471,7 @@ async fn bridge_issue_reply_to_mt(
471 471 sender_username: &str,
472 472 sender_display_name: Option<&str>,
473 473 body_markdown: &str,
474 + message_id: &str,
474 475 ) {
475 476 let mt = match &state.mt_client {
476 477 Some(c) => c,
@@ -492,6 +493,7 @@ async fn bridge_issue_reply_to_mt(
492 493 author_mnw_id: *sender_id,
493 494 author_username: sender_username.to_string(),
494 495 author_display_name: sender_display_name.map(String::from),
496 + external_ref: format!("mnw:post:{}", message_id),
495 497 };
496 498 if let Err(e) = mt.create_post(crate::db::MtThreadId::from(thread_id), &req).await {
497 499 tracing::warn!(error = ?e, issue_id = %issue.id, "inbound-issues: MT reply post failed");
M server/src/routes/postmark/patches.rs +1
@@ -141,6 +141,7 @@ pub(super) async fn postmark_inbound(
141 141 author_mnw_id: *sender.id,
142 142 author_username: sender.username.to_string(),
143 143 author_display_name: sender.display_name.clone(),
144 + external_ref: format!("mnw:post:{}", payload.message_id),
144 145 };
145 146 match mt.create_post(tid, &req).await {
146 147 Ok(_resp) => {