From 979b7307ca60df7fda6fa62016fcb685ee4e7e70 Mon Sep 17 00:00:00 2001 From: Eureka Date: Wed, 11 Nov 2026 20:24:49 +0000 Subject: [PATCH] chat ingest: rebuild on unknown redaction, edit revert, scrub Matrix/PGP/glued/zero-width shapes, label scrub, estate and facts fail closed, retire continues, deterministic tail Co-Authored-By: Claude Sonnet 5.5 --- crates/huginn-daemon/src/index/fakes.rs | 4 +- crates/huginn-daemon/src/index/qdrant.rs | 6 +- crates/huginn-daemon/src/sources/chat.rs | 166 +++++++++--- .../huginn-daemon/src/sources/chat_tests.rs | 254 +++++++++++++++--- .../huginn-daemon/src/sources/room_scope.rs | 67 +++-- crates/huginn-mind/src/redact.rs | 108 ++++++-- docs/chat-retrieval-map.md | 29 ++ 7 files changed, 519 insertions(+), 115 deletions(-) diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs index 30e864d..e162bca 100644 --- a/crates/huginn-daemon/src/index/fakes.rs +++ b/crates/huginn-daemon/src/index/fakes.rs @@ -270,6 +270,8 @@ pub(crate) struct SourcesState { pub(crate) fail_chat_upserts: bool, /// Every chat delete refuses while this is set. pub(crate) fail_chat_deletes: bool, + /// A chat delete naming one of these rooms refuses. + pub(crate) fail_chat_deletes_for: BTreeSet, /// Every chat delete that went through: the room and the ids (`None` is all of the room). pub(crate) chat_deletes: Vec<(String, Option>)>, /// The repo filter and limit of every search, in order. @@ -439,7 +441,7 @@ impl SourceIndex for FakeSources { fn delete_chat(&mut self, room_id: &str, ids: Option<&[String]>) -> Result<()> { let mut state = self.state.lock().unwrap(); - ensure!(!state.fail_chat_deletes, "the vector store refused the chat delete"); + ensure!(!state.fail_chat_deletes && !state.fail_chat_deletes_for.contains(room_id), "the vector store refused the chat delete"); state.chat_deletes.push((room_id.to_owned(), ids.map(<[String]>::to_vec))); state.chat.retain(|id, point| !(point.payload.room_id == room_id && ids.is_none_or(|ids| ids.contains(id)))); Ok(()) diff --git a/crates/huginn-daemon/src/index/qdrant.rs b/crates/huginn-daemon/src/index/qdrant.rs index 473789d..3a3f54c 100644 --- a/crates/huginn-daemon/src/index/qdrant.rs +++ b/crates/huginn-daemon/src/index/qdrant.rs @@ -467,7 +467,11 @@ impl SourceIndex for QdrantIndex { fn chat_tail(&mut self, room_id: &str) -> Result> { let order = json!({ "key": "ts_end_ms", "direction": "desc" }); - Ok(self.scroll_chat(chat_room_filter(room_id, vec![]), Some((order, 1)))?.into_iter().next()) + let Some(newest) = self.scroll_chat(chat_room_filter(room_id, vec![]), Some((order, 1)))?.into_iter().next() else { return Ok(None) }; + // Chunks tied on the greatest end are all read and the greatest id wins, so the tail is the same chunk on every run. + let end = newest.payload.messages.iter().map(|message| message.ts_ms).max().unwrap_or(0); + let tied = self.scroll_chat(chat_room_filter(room_id, vec![json!({ "key": "ts_end_ms", "range": { "gte": end } })]), None)?; + Ok(tied.into_iter().max_by(|a, b| a.id.cmp(&b.id))) } fn chat_chunks_holding(&mut self, room_id: &str, event_ids: &[String]) -> Result> { diff --git a/crates/huginn-daemon/src/sources/chat.rs b/crates/huginn-daemon/src/sources/chat.rs index 3c895c0..fcee66f 100644 --- a/crates/huginn-daemon/src/sources/chat.rs +++ b/crates/huginn-daemon/src/sources/chat.rs @@ -108,6 +108,8 @@ pub struct ChatReport { pub scrubbed: u32, pub purged: u32, pub seconds: f64, + /// Why the room could not be read or written this run; what the index held for it stays as it was. + pub failure: Option, } impl std::fmt::Display for ChatReport { @@ -124,7 +126,11 @@ impl std::fmt::Display for ChatReport { self.scrubbed, self.purged, self.seconds - ) + )?; + match &self.failure { + Some(why) => write!(formatter, " failed: {why}"), + None => Ok(()), + } } } @@ -154,6 +160,17 @@ pub struct ChatRefreshed { pub scrubbed: Vec, } +impl ChatReport { + fn empty(room_id: &str, alias: Option) -> Self { + Self { room_id: room_id.to_owned(), alias, first: None, last: None, events: 0, chunks: 0, scrubbed: 0, purged: 0, seconds: 0.0, failure: None } + } + + /// A room that could not be handled this run. Its points and cursor are untouched: an error is not a reclassification. + fn failed(room_id: &str, alias: Option, error: &anyhow::Error) -> Self { + Self { failure: Some(format!("{error:#}")), ..Self::empty(room_id, alias) } + } +} + impl ChatRefreshed { /// Nothing failed and no joined room had to be refused: a joined estate, direct, /// encrypted or bridged room means the reader's roster is wrong. A scrub hit does not make a run unclean. @@ -191,21 +208,32 @@ fn purge_room(index: &mut S, room_id: &str) -> Result<()> { drop_cursor(index, &cursor_key(room_id)) } +/// What `retire_chat` did: the rooms it removed, and those it could not (each with why), which it went on past. +#[derive(Debug, Default, PartialEq, Eq)] +pub struct Retired { + pub retired: Vec, + pub failed: Vec<(String, String)>, +} + /// Deletes the points and `chat:` cursor of every room not in `keep` (the reader left it, -/// or it is refused or excluded), and returns those rooms. Purge tombstones and forge -/// repos are never touched. With the empty set it removes the whole chat corpus. -pub fn retire_chat(index: &mut S, keep: &BTreeSet) -> Result> { +/// or it is refused or excluded). A room whose delete fails is reported and the others go on. Purge +/// tombstones and forge repos are never touched. With the empty set it removes the whole chat corpus. +pub fn retire_chat(index: &mut S, keep: &BTreeSet) -> Result { let rooms: Vec = read_cursors(index)?.keys().filter_map(|key| cursor_room(key)).filter(|room| !keep.contains(*room)).map(str::to_owned).collect(); - for room in &rooms { - purge_room(index, room)?; + let mut done = Retired::default(); + for room in rooms { + match purge_room(index, &room) { + Ok(()) => done.retired.push(room), + Err(error) => done.failed.push((room, format!("{error:#}"))), + } } - Ok(rooms) + Ok(done) } /// A change to chunks already stored, in event order. enum Op { /// An edit by `sender` of `target`'s body; applies only to the original's sender. - Edit { target: String, sender: String, body: String }, + Edit { id: String, target: String, sender: String, body: String }, /// A held message's body is what the page says it is (a replay). Set { target: String, body: String }, Redact { target: String }, @@ -233,9 +261,19 @@ impl Op { } } +/// A message of the page with the edits of the page that apply to it: the body is the last edit's, and a +/// redacted edit leaves the list, so the body falls back to the one before it. +struct Pending { + message: ChatMessage, + original: String, + edits: Vec<(String, String)>, +} + struct Folded { pending: Vec, ops: Vec, + /// Every event id of the page: a redaction of one of them is known. + seen: HashSet, } struct Room<'a> { @@ -247,8 +285,8 @@ struct Room<'a> { /// The page folded in order: the purge filter, then the credential scrub, then the /// event's effect. After this no raw body exists. fn fold(room: &Room, events: &[RoomEvent], report: &mut ChatReport, out: &mut ChatRefreshed) -> Folded { - let mut folded = Folded { pending: Vec::new(), ops: Vec::new() }; - let mut redacted: HashSet<&str> = HashSet::new(); + let mut pending: Vec = Vec::new(); + let mut ops: Vec = Vec::new(); let mut scrubbed = |event: &RoomEvent, body: &str| { let clean = scrub(body); for kind in clean.hits { @@ -262,10 +300,11 @@ fn fold(room: &Room, events: &[RoomEvent], report: &mut ChatReport, out: &mut Ch EventKind::Message { body } => { if covers(&room.windows, event.ts_ms) { report.purged += 1; - } else if !redacted.contains(event.event_id.as_str()) { + } else { let body = scrubbed(event, body); if !body.trim().is_empty() { - folded.pending.push(ChatMessage { event_id: event.event_id.clone(), sender: event.sender.clone(), ts_ms: event.ts_ms, body }); + let message = ChatMessage { event_id: event.event_id.clone(), sender: event.sender.clone(), ts_ms: event.ts_ms, body: body.clone() }; + pending.push(Pending { message, original: body, edits: Vec::new() }); } } } @@ -275,21 +314,29 @@ fn fold(room: &Room, events: &[RoomEvent], report: &mut ChatReport, out: &mut Ch continue; } let body = scrubbed(event, body); - match folded.pending.iter_mut().find(|message| message.event_id == *target) { - Some(message) if message.sender == event.sender && !body.trim().is_empty() => message.body = body, + match pending.iter_mut().find(|held| held.message.event_id == *target) { + Some(held) if held.message.sender == event.sender && !body.trim().is_empty() => { + held.edits.push((event.event_id.clone(), body.clone())); + held.message.body = body; + } Some(_) => {} - None => folded.ops.push(Op::Edit { target: target.clone(), sender: event.sender.clone(), body }), + None => ops.push(Op::Edit { id: event.event_id.clone(), target: target.clone(), sender: event.sender.clone(), body }), } } EventKind::Redaction { target } => { - folded.pending.retain(|message| message.event_id != *target); - redacted.insert(target); - folded.ops.push(Op::Redact { target: target.clone() }); + // The target may be a message, or an edit whose own text must stop applying. + pending.retain(|held| held.message.event_id != *target); + for held in &mut pending { + held.edits.retain(|(id, _)| id != target); + held.message.body = held.edits.last().map_or_else(|| held.original.clone(), |(_, body)| body.clone()); + } + ops.retain(|op| !matches!(op, Op::Edit { id, .. } if id == target)); + ops.push(Op::Redact { target: target.clone() }); } EventKind::Other => {} } } - folded + Folded { pending: pending.into_iter().map(|held| held.message).collect(), ops, seen: events.iter().map(|event| event.event_id.clone()).collect() } } /// Chunks to rewrite: the stored chunk ids they replace and the messages that now make them. @@ -364,16 +411,29 @@ fn apply_purge(embedder: &mut E, index: &mut S, dim /// Folds one page into the stored chunks: edits and redactions rewrite the chunk that held /// the target alone, new messages extend the open tail. -fn write_page(embedder: &mut E, index: &mut S, dimensions: u32, room: &Room, mut folded: Folded) -> Result { +fn write_page( + embedder: &mut E, + index: &mut S, + dimensions: u32, + room: &Room, + mut folded: Folded, + may_rebuild: bool, +) -> Result> { let room_id = &room.facts.room_id; let mut asked: Vec = folded.pending.iter().map(|message| message.event_id.clone()).collect(); asked.extend(folded.ops.iter().map(|op| op.target().to_owned())); if asked.is_empty() { - return Ok(0); + return Ok(Some(0)); } let holding = index.chat_chunks_holding(room_id, &asked)?; // A message the index already holds is a replay: its page body is asserted, never appended again. let held: HashSet<&str> = holding.iter().flat_map(|chunk| chunk.payload.messages.iter().map(|message| message.event_id.as_str())).collect(); + // A redaction of an event this room neither holds nor shows on the page may be of an edit whose own id is + // never stored, so a body it changed cannot be taken back here: the room is rebuilt from the history, which + // serves a redacted edit without its text. + if may_rebuild && folded.ops.iter().any(|op| matches!(op, Op::Redact { target } if !folded.seen.contains(target) && !held.contains(target.as_str()))) { + return Ok(None); + } let (replayed, fresh): (Vec, Vec) = folded.pending.drain(..).partition(|message| held.contains(message.event_id.as_str())); let mut ops: Vec = replayed.into_iter().map(|message| Op::Set { target: message.event_id, body: message.body }).collect(); ops.append(&mut folded.ops); @@ -400,7 +460,7 @@ fn write_page(embedder: &mut E, index: &mut S, dime tail_messages.extend(fresh); rewrites.push((tail_id.into_iter().collect(), tail_messages)); } - write_rewrites(embedder, index, dimensions, room, rewrites, &standing) + write_rewrites(embedder, index, dimensions, room, rewrites, &standing).map(Some) } enum Outcome { @@ -421,9 +481,13 @@ fn ingest_room( let room_id = facts.room_id.as_str(); let (key, version) = (cursor_key(room_id), chat_version()); let cursors = read_cursors(index)?; + let mut may_rebuild = false; let mut after: Option = match cursors.get(&key).map(|value| parse_cursor(value)) { None => None, - Some(Some((held, event))) if held == version => Some(event.to_owned()).filter(|event| !event.is_empty()), + Some(Some((held, event))) if held == version => { + may_rebuild = true; + Some(event.to_owned()).filter(|event| !event.is_empty()) + } // Another chunker or scrub version, or a value nobody here wrote: this room alone starts over. Some(_) => { index.delete_chat(room_id, None).with_context(|| format!("deleting the chunks of {room_id} for a re-ingest"))?; @@ -434,9 +498,20 @@ fn ingest_room( put_cursor(index, &key, &format!("{version}:"))?; } let windows = cursors.get(&purge_key(room_id)).map(|value| parse_windows(value)).transpose()?.unwrap_or_default(); - let label = facts.name.clone().or_else(|| facts.alias.clone()).unwrap_or_else(|| room_id.to_owned()); + let mut report = ChatReport::empty(room_id, facts.alias.clone()); + // The room's name is embedded and stored in every chunk, so it passes the scrub like a body. + let label = match &facts.name { + Some(name) => { + let clean = scrub(name); + for kind in clean.hits { + report.scrubbed += 1; + out.scrubbed.push(ScrubHit { room_id: room_id.to_owned(), event_id: "(name)".to_owned(), kind }); + } + clean.text + } + None => facts.alias.clone().unwrap_or_else(|| room_id.to_owned()), + }; let room = Room { facts, label, windows }; - let mut report = ChatReport { room_id: room_id.to_owned(), alias: facts.alias.clone(), first: None, last: None, events: 0, chunks: 0, scrubbed: 0, purged: 0, seconds: 0.0 }; apply_purge(embedder, index, dimensions, &room, &mut report)?; let mut restarted = false; loop { @@ -457,7 +532,15 @@ fn ingest_room( report.first.get_or_insert_with(|| page.events[0].event_id.clone()); report.events += u32::try_from(page.events.len()).unwrap_or(u32::MAX); let folded = fold(&room, &page.events, &mut report, out); - report.chunks += write_page(embedder, index, dimensions, &room, folded)?; + let Some(written) = write_page(embedder, index, dimensions, &room, folded, may_rebuild)? else { + // Start this room over from the history; nothing it read is trusted to be complete. + purge_room(index, room_id)?; + put_cursor(index, &key, &format!("{version}:"))?; + (after, may_rebuild) = (None, false); + report = ChatReport { scrubbed: report.scrubbed, ..ChatReport::empty(room_id, facts.alias.clone()) }; + continue; + }; + report.chunks += written; put_cursor(index, &key, &format!("{version}:{last}"))?; report.last = Some(last.clone()); after = Some(last); @@ -504,26 +587,31 @@ pub fn refresh_chat( match verdict { Ok(Ok(facts)) => admitted.push(facts), Ok(Err(reason)) => { + // Not kept, so the retirement below purges whatever it held. keep.remove(room_id); - match purge_room(index, room_id) { - Ok(()) => {} - Err(error) => { - journal(format!("huginn-sources: chat {room_id} could not be purged, the others go on: {error:#}")); - out.failed.push(room_id.clone()); - } - } journal(format!("refused: {room_id} {reason}")); out.refused.push(RefusedRoom { room_id: room_id.clone(), reason, event_id: None }); } + // An error is not a reclassification: the room keeps its points and cursor, and the report says why. Err(error) => { - journal(format!("huginn-sources: chat {room_id} facts failed, the others go on: {error:#}")); + let report = ChatReport::failed(room_id, None, &error.context("reading the room's facts")); + journal(report.to_string()); out.failed.push(room_id.clone()); + out.rooms.push(report); } } } - if let Err(error) = retire_chat(index, &keep) { - journal(format!("huginn-sources: retiring rooms the reader left failed, the others go on: {error:#}")); - out.failed.push("retire".to_owned()); + match retire_chat(index, &keep) { + Ok(done) => { + for (room, why) in done.failed { + journal(format!("huginn-sources: retiring {room} failed, the others go on: {why}")); + out.failed.push(room); + } + } + Err(error) => { + journal(format!("huginn-sources: retiring rooms failed: {error:#}")); + out.failed.push("retire".to_owned()); + } } for facts in admitted { match ingest_room(embedder, index, history, identity.dimensions, &facts, &mut out) { @@ -536,8 +624,10 @@ pub fn refresh_chat( out.refused.push(refused); } Err(error) => { - journal(format!("huginn-sources: chat {} failed, the others go on: {error:#}", facts.room_id)); + let report = ChatReport::failed(&facts.room_id, facts.alias.clone(), &error); + journal(report.to_string()); out.failed.push(facts.room_id.clone()); + out.rooms.push(report); } } } diff --git a/crates/huginn-daemon/src/sources/chat_tests.rs b/crates/huginn-daemon/src/sources/chat_tests.rs index 7eb6e60..838d3fe 100644 --- a/crates/huginn-daemon/src/sources/chat_tests.rs +++ b/crates/huginn-daemon/src/sources/chat_tests.rs @@ -18,6 +18,14 @@ use crate::index::fakes::{FakeEmbedder, FakeSources}; // ---- the rig ---- +fn plain(id: &str) -> RoomFacts { + RoomFacts { room_id: id.into(), alias: None, alt_aliases: vec![], name: None, encrypted: false, direct: false, bridge: None } +} + +fn estate(ids: &[&str]) -> Estate { + Estate::new(ids.iter().chain(["!pad1:s", "!pad2:s", "!pad3:s"].iter()).take(3).map(|id| (*id).to_owned()).collect()).unwrap() +} + struct Scripted { facts: RoomFacts, events: Vec, @@ -28,11 +36,15 @@ struct FakeHistory { joined: Vec, rooms: BTreeMap, calls: Vec, + /// Rooms whose state cannot be read. + facts_fail: BTreeSet, + /// Serve an event a redaction targets as the server does, without its content. + prune: bool, } impl FakeHistory { fn room(&mut self, id: &str, events: Vec) -> &mut Self { - self.rooms.insert(id.into(), Scripted { facts: RoomFacts { room_id: id.into(), ..RoomFacts::default() }, events }); + self.rooms.insert(id.into(), Scripted { facts: plain(id), events }); self.joined.push(id.into()); self } @@ -59,6 +71,9 @@ impl RoomHistory for FakeHistory { fn facts(&mut self, room_id: &str) -> Result { self.calls.push(format!("facts:{room_id}")); + if self.facts_fail.contains(room_id) { + bail!("the server did not answer for the state of {room_id}"); + } match self.rooms.get(room_id) { Some(room) => Ok(room.facts.clone()), None => bail!("no such room"), @@ -75,9 +90,13 @@ impl RoomHistory for FakeHistory { None => return Ok(None), }, }; - let rest = &room.events[from..]; - let take = rest.len().min(limit as usize); - Ok(Some(EventPage { events: rest[..take].to_vec(), more: take < rest.len() })) + let redacted: BTreeSet<&str> = room.events.iter().filter_map(|event| if let EventKind::Redaction { target } = &event.kind { Some(target.as_str()) } else { None }).collect(); + let served: Vec = room.events[from..] + .iter() + .map(|event| if self.prune && redacted.contains(event.event_id.as_str()) { RoomEvent { kind: EventKind::Other, content_keys: vec![], ..event.clone() } } else { event.clone() }) + .collect(); + let take = served.len().min(limit as usize); + Ok(Some(EventPage { events: served[..take].to_vec(), more: take < served.len() })) } } @@ -117,7 +136,7 @@ impl Rig { chunker_version: chunk::CHUNKER_VERSION, }; index.plant(DescribedSources::Labelled { meta, cursors: BTreeMap::new() }); - Self { embedder, index, estate: Estate::default() } + Self { embedder, index, estate: estate(&[]) } } fn run(&mut self, history: &mut FakeHistory) -> (Result, Vec) { @@ -188,7 +207,7 @@ fn only_joined_rooms_are_looked_at_and_a_join_rule_is_not_the_test() { let mut history = FakeHistory::default(); history.room("!a:s", vec![said("$a1", 1, "in a")]).room("!b:s", vec![said("$b1", 2, "in b")]); // A room the reader has not joined: scripted, but never listed. - history.rooms.insert("!c:s".into(), Scripted { facts: RoomFacts { room_id: "!c:s".into(), ..RoomFacts::default() }, events: vec![said("$c1", 3, "in c")] }); + history.rooms.insert("!c:s".into(), Scripted { facts: plain("!c:s"), events: vec![said("$c1", 3, "in c")] }); let mut rig = Rig::new(); let done = rig.ok(&mut history); assert!(done.is_clean()); @@ -217,7 +236,7 @@ fn a_dm_room_is_refused_unread_unindexed_and_purged() { #[test] fn each_estate_room_is_refused_by_id_and_by_alias_and_an_excluded_room_in_the_index_is_purged() { let mut rig = Rig::new(); - rig.estate = Estate { ids: ["!byid:s".to_owned()].into() }; + rig.estate = estate(&["!byid:s"]); let mut history = FakeHistory::default(); history.room("!byid:s", vec![said("$i1", 1, "x")]); for (n, alias) in ESTATE_ALIASES.iter().map(|alias| alias.to_string()).chain(["#USERS:GAMECULT.ORG".to_owned()]).enumerate() { @@ -282,6 +301,8 @@ fn a_bridged_platforms_sender_or_content_refuses_the_room_before_anything_is_emb event("$bad", "@discord_123:s", 2, EventKind::Message { body: "SECRET-BODY-TEXT".into() }), event("$bad", "@_discord_9:s", 2, EventKind::Message { body: "SECRET-BODY-TEXT".into() }), event("$bad", "@discordbot:s", 2, EventKind::Message { body: "SECRET-BODY-TEXT".into() }), + event("$bad", "@mautrix_discord:s", 2, EventKind::Message { body: "SECRET-BODY-TEXT".into() }), + event("$bad", "@_bridge_discordbot:s", 2, EventKind::Message { body: "SECRET-BODY-TEXT".into() }), RoomEvent { content_keys: vec!["body".into(), "fi.mau.discord".into()], ..said("$bad", 2, "SECRET-BODY-TEXT") }, RoomEvent { content_keys: vec!["uk.half-shot.discord".into()], ..said("$bad", 2, "SECRET-BODY-TEXT") }, ]; @@ -522,10 +543,10 @@ fn retire_chat_removes_rooms_no_longer_kept_and_given_none_all_chat_and_touches_ cursors.insert(format!("{CHAT_PURGE_PREFIX}!a:s"), "1..2".into()); rig.index.set_cursors(&cursors).unwrap(); let retired = retire_chat(&mut rig.index, &["!a:s".to_owned()].into()).unwrap(); - assert_eq!(retired, ["!b:s"]); + assert_eq!(retired.retired, ["!b:s"]); assert_eq!(rig.held_ids("!a:s"), ["$e"]); assert!(rig.held_ids("!b:s").is_empty() && rig.cursor("!b:s").is_none()); - assert_eq!(retire_chat(&mut rig.index, &BTreeSet::new()).unwrap(), ["!a:s"]); + assert_eq!(retire_chat(&mut rig.index, &BTreeSet::new()).unwrap().retired, ["!a:s"]); assert!(rig.points().is_empty()); assert_eq!(rig.index.state.lock().unwrap().points.len(), 1, "a forge point stays"); let kept: Vec = rig.index.cursors().into_keys().collect(); @@ -540,38 +561,42 @@ fn retire_chat_removes_rooms_no_longer_kept_and_given_none_all_chat_and_touches_ // ---- the credential scrub through the job ---- -fn planted() -> Vec<(ScrubKind, String)> { +fn planted() -> Vec<(ScrubKind, String, String)> { + let t = |kind, secret: String, needle: &str| (kind, secret, needle.to_owned()); vec![ - (ScrubKind::AuthorizationHeader, format!("Authorization: Bearer {}", "a1B2".repeat(8))), - (ScrubKind::Bearer, format!("Bearer {}", "z9Y8".repeat(6))), - (ScrubKind::KeyShape, format!("sk-{}", "q7W3".repeat(8))), - (ScrubKind::KeyShape, format!("sk_live_{}", "m4N5".repeat(6))), - (ScrubKind::KeyShape, format!("ghp_{}", "k2L9".repeat(9))), - (ScrubKind::KeyShape, format!("github_pat_{}_{}", "A1b2C3".repeat(4) + "ab", "d4E5f6".repeat(9) + "abcde")), - (ScrubKind::KeyShape, format!("glpat-{}", "r6T7".repeat(5))), - (ScrubKind::KeyShape, "xoxb-123456789012-987654321098-aBcDeFgHiJkLmNoP".into()), - (ScrubKind::KeyShape, "AKIA0123456789ABCDEF".into()), - (ScrubKind::KeyShape, format!("AIza{}", "Sy1_".repeat(9))), - (ScrubKind::Jwt, "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJmYWtlIn0.c2lnbmF0dXJlZmFrZQ".into()), - (ScrubKind::PemBlock, "-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKCAQEAfake\n-----END RSA PRIVATE KEY-----".into()), - (ScrubKind::PemBlock, "-----BEGIN PRIVATE KEY-----\nMIIBVQIBADANfakeNOEND".into()), - (ScrubKind::UrlUserinfo, "https://user:hunter2hunter2@host.example/r.git".into()), - (ScrubKind::UrlUserinfo, "https://tokenvalue12345@host.example/".into()), + t(ScrubKind::AuthorizationHeader, format!("Authorization: Bearer {}", "a1B2".repeat(8)), "a1B2a1B2a1B2"), + t(ScrubKind::Bearer, format!("Bearer {}", "z9Y8".repeat(6)), "z9Y8z9Y8z9Y8"), + t(ScrubKind::Bearer, format!("bearer:{}", "z9Y8".repeat(6)), "z9Y8z9Y8z9Y8"), + t(ScrubKind::KeyShape, format!("sk-{}", "q7W3".repeat(8)), "q7W3q7W3q7W3"), + t(ScrubKind::KeyShape, format!("sk_live_{}", "m4N5".repeat(6)), "m4N5m4N5m4N5"), + t(ScrubKind::KeyShape, format!("ghp_{}", "k2L9".repeat(9)), "k2L9k2L9k2L9"), + t(ScrubKind::KeyShape, format!("xxx-ghp_{}", "k2L9".repeat(9)), "k2L9k2L9k2L9"), + t(ScrubKind::KeyShape, format!("github_pat_{}_{}", "A1b2C3".repeat(4) + "ab", "d4E5f6".repeat(9) + "abcde"), "A1b2C3A1b2C3"), + t(ScrubKind::KeyShape, format!("glpat-{}", "r6T7".repeat(5)), "r6T7r6T7r6T7"), + t(ScrubKind::KeyShape, "xoxb-123456789012-987654321098-aBcDeFgHiJkLmNoP".into(), "aBcDeFgHiJkL"), + t(ScrubKind::KeyShape, "xoxa-2-123456789012-aBcDeFgHiJkLmNoP".into(), "aBcDeFgHiJkL"), + t(ScrubKind::KeyShape, "xoxp-123456789012-987654321098-aBcDeFgHiJkLmNoP".into(), "aBcDeFgHiJkL"), + t(ScrubKind::KeyShape, "xoxr-123456789012-987654321098-aBcDeFgHiJkLmNoP".into(), "aBcDeFgHiJkL"), + t(ScrubKind::KeyShape, "xoxs-123456789012-987654321098-aBcDeFgHiJkLmNoP".into(), "aBcDeFgHiJkL"), + t(ScrubKind::KeyShape, "AKIA0123456789ABCDEF".into(), "0123456789ABCDEF"), + t(ScrubKind::KeyShape, "xAKIA0123456789ABCDEF".into(), "0123456789ABCDEF"), + t(ScrubKind::KeyShape, format!("AIza{}", "Sy1_".repeat(9)), "Sy1_Sy1_Sy1_"), + t(ScrubKind::KeyShape, format!("syt_{}_{}_{}", "YWxpY2U", "aBcDeFgHiJkLmNoPqR", "0a1B2c"), "aBcDeFgHiJkLmNoPqR"), + t(ScrubKind::KeyShape, format!("mct_{}", "w4X5".repeat(8)), "w4X5w4X5w4X5"), + t(ScrubKind::AccessToken, "https://h/_matrix/client/v3/sync?access_token=abcDEF123456xyz&since=s1".into(), "abcDEF123456xyz"), + t(ScrubKind::AccessToken, "{\"access_token\": \"tokenValue987654\", \"user\": 1}".into(), "tokenValue987654"), + t(ScrubKind::Jwt, "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJmYWtlIn0.c2lnbmF0dXJlZmFrZQ".into(), "c2lnbmF0dXJlZmFrZQ"), + t(ScrubKind::PemBlock, "-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKCAQEAfake\n-----END RSA PRIVATE KEY-----".into(), "MIIEowIBAAKCAQEAfake"), + t(ScrubKind::PemBlock, "-----BEGIN PRIVATE KEY-----\nMIIBVQIBADANfakeNOEND".into(), "MIIBVQIBADANfake"), + t(ScrubKind::PemBlock, "-----BEGIN PGP PRIVATE KEY BLOCK-----\nlQOYBFakeFakeFake\n-----END PGP PRIVATE KEY BLOCK-----".into(), "lQOYBFakeFake"), + t(ScrubKind::UrlUserinfo, "https://user:hunter2hunter2@host.example/r.git".into(), "hunter2hunter2"), + t(ScrubKind::UrlUserinfo, "https://tokenvalue12345@host.example/".into(), "tokenvalue12345"), ] } -/// The part of a planted secret no replacement can contain. -fn distinct(secret: &str) -> String { - let bare = secret.rsplit(['@', ' ', '\n']).next().filter(|_| !secret.contains("://")).unwrap_or(secret); - let source = if secret.contains("://") { secret.split("://").nth(1).unwrap().split('@').next().unwrap().rsplit(':').next().unwrap() } else { bare }; - source.chars().rev().take(12).collect::>().into_iter().rev().collect() -} - #[test] fn planted_credentials_never_reach_the_chunk_the_embedder_or_the_journal_in_a_message_or_an_edit() { - for (kind, secret) in planted() { - let needle = distinct(&secret); - assert!(needle.len() >= 8, "{secret}"); + for (kind, secret, needle) in planted() { for shape in ["see {} now", "see {}", "{}\nnext line"] { let (dirty, clean) = (shape.replace("{}", &secret), "nothing here".to_owned()); let mut history = FakeHistory::default(); @@ -778,5 +803,162 @@ fn the_ingest_code_is_pure_over_its_ports_and_names_no_credential_shape_no_secon assert_eq!((code.matches("\"chat:").count(), code.matches("\"chat-purge:").count()), (defined, defined), "{name}: each prefix is defined once"); } let chat = include_str!("chat.rs"); - assert_eq!(chat.matches("scrub(").count(), 1, "scrub is called in one place"); + assert_eq!(chat.matches("scrub(").count(), 2, "scrub is called for a body and for the room name, nowhere else"); +} + +// ---- redacted edits, redacted originals, late edits ---- + +const EDIT_SECRET: &str = "EDIT-SECRET"; + +fn nothing_secret_stored(rig: &Rig) { + let stored = format!("{:?}", rig.points().iter().map(|point| (&point.payload.messages, &point.payload.text)).collect::>()); + assert!(!stored.contains(EDIT_SECRET), "{stored}"); +} + +#[test] +fn a_redacted_edit_in_the_same_page_reverts_the_message_to_its_earlier_body() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$m", 10, "original words"), edit("$e1", "@a:s", 20, "$m", "first edit"), edit("$e2", "@a:s", 30, "$m", EDIT_SECRET), redaction("$r", 40, "$e2")]); + let mut rig = Rig::new(); + rig.ok(&mut history); + assert_eq!(rig.bodies(), ["first edit"]); + assert!(rig.embedded().iter().all(|text| !text.contains(EDIT_SECRET))); + history.room("!q:s", vec![said("$m", 10, "orig"), edit("$e", "@a:s", 20, "$m", EDIT_SECRET), redaction("$r", 30, "$e")]); + rig.ok(&mut history); + assert!(rig.bodies().contains(&"orig".to_owned())); + nothing_secret_stored(&rig); +} + +#[test] +fn a_redaction_of_an_edit_a_later_run_stored_rebuilds_the_room_and_the_secret_does_not_return() { + let mut history = FakeHistory { prune: true, ..FakeHistory::default() }; + history.room("!r:s", vec![said("$m", 10, "original"), edit("$e", "@a:s", 20, "$m", EDIT_SECRET)]); + history.room("!q:s", vec![said("$q", 10, "quiet")]); + let mut rig = Rig::new(); + rig.ok(&mut history); + assert!(rig.bodies().iter().any(|body| body == EDIT_SECRET), "the edit applied while it stood"); + let recreated = rig.index.state.lock().unwrap().recreated; + history.push("!r:s", vec![redaction("$r", 30, "$e")]); + let done = rig.ok(&mut history); + assert!(done.is_clean()); + nothing_secret_stored(&rig); + assert!(rig.bodies().contains(&"original".to_owned())); + assert_eq!(rig.held_ids("!q:s"), ["$q"]); + assert_eq!(rig.index.state.lock().unwrap().recreated, recreated); + assert_eq!(rig.cursor("!r:s").unwrap(), format!("{}:$r", chat_version())); + // The rebuild does not loop: the redaction it replays is not a trigger the second time, and nothing changes. + let ids = rig.ids(); + rig.ok(&mut history); + assert_eq!(rig.ids(), ids); +} + +#[test] +fn a_redacted_original_takes_its_live_and_late_edits_with_it_in_a_page_across_runs_and_across_a_rebuild() { + let script = |history: &mut FakeHistory| { + history.room("!r:s", vec![said("$m", 10, "original"), edit("$e1", "@a:s", 20, "$m", "live edit"), redaction("$re", 25, "$e1"), redaction("$rm", 30, "$m"), edit("$late", "@a:s", 40, "$m", EDIT_SECRET)]); + history.prune = true; + }; + // One page. + let (mut history, mut rig) = (FakeHistory::default(), Rig::new()); + script(&mut history); + rig.ok(&mut history); + nothing_secret_stored(&rig); + assert!(rig.points().is_empty()); + // Across runs: the original and a live edit stand, then it is redacted, then a late edit arrives. + let (mut history, mut rig) = (FakeHistory { prune: true, ..FakeHistory::default() }, Rig::new()); + history.room("!r:s", vec![said("$m", 10, "original"), edit("$e1", "@a:s", 20, "$m", EDIT_SECRET)]); + rig.ok(&mut history); + history.push("!r:s", vec![redaction("$rm", 30, "$m")]); + rig.ok(&mut history); + nothing_secret_stored(&rig); + history.push("!r:s", vec![edit("$late", "@a:s", 40, "$m", EDIT_SECRET)]); + rig.ok(&mut history); + nothing_secret_stored(&rig); + assert!(rig.points().is_empty()); + // Across a rebuild: a lost cursor re-reads the whole history. + let mut cursors = rig.index.cursors(); + cursors.remove(&cursor_key("!r:s")); + rig.index.set_cursors(&cursors).unwrap(); + rig.ok(&mut history); + nothing_secret_stored(&rig); + assert!(rig.points().is_empty()); +} + +#[test] +fn a_stranger_cannot_edit_a_message_of_the_same_page() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$m", 10, "mine"), edit("$e", "@stranger:s", 20, "$m", "hijacked")]); + let mut rig = Rig::new(); + rig.ok(&mut history); + assert_eq!(rig.bodies(), ["mine"]); +} + +#[test] +fn the_room_name_is_scrubbed_before_it_is_embedded_or_stored() { + let mut history = FakeHistory::default(); + let token = format!("ghp_{}", "k2L9".repeat(9)); + history.room("!r:s", vec![said("$m", 10, "hello")]).facts("!r:s", |facts| facts.name = Some(format!("Swarm {token}"))); + let mut rig = Rig::new(); + let (done, journal) = rig.run(&mut history); + let done = done.unwrap(); + assert!(rig.embedded().iter().all(|text| !text.contains("k2L9k2L9"))); + assert!(rig.points().iter().all(|point| !point.payload.text.contains("k2L9k2L9") && point.payload.text.starts_with("Swarm "))); + assert!(done.scrubbed.iter().any(|hit| hit.event_id == "(name)") && journal.iter().all(|line| !line.contains("k2L9k2L9"))); +} + +#[test] +fn a_token_split_by_zero_width_characters_leaves_no_tail_in_a_stored_body() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$m", 10, &format!("key sk-{}\u{200b}{} done", "q7W3".repeat(4), "q7W3".repeat(4)))]); + let mut rig = Rig::new(); + rig.ok(&mut history); + assert_eq!(rig.bodies(), ["key done"]); +} + +#[test] +fn retire_goes_on_past_a_failed_delete_and_names_each_failure() { + let mut rig = Rig::new(); + for room in ["!a:s", "!b:s", "!c:s"] { + rig.seed(room, "$e", 1); + } + rig.index.state.lock().unwrap().fail_chat_deletes_for = ["!a:s".to_owned()].into(); + let done = retire_chat(&mut rig.index, &BTreeSet::new()).unwrap(); + assert_eq!(done.retired, ["!b:s", "!c:s"]); + assert_eq!(done.failed.len(), 1); + assert_eq!(done.failed[0].0, "!a:s"); + assert_eq!(rig.held_ids("!a:s"), ["$e"]); + assert!(rig.cursor("!a:s").is_some(), "the cursor stays so the next run finds it again"); + // The refresh reports it too. + let mut history = FakeHistory::default(); + history.room("!z:s", vec![said("$z", 1, "x")]); + let done = rig.ok(&mut history); + assert!(done.failed.contains(&"!a:s".to_owned()) && !done.is_clean()); +} + +#[test] +fn a_facts_error_on_a_joined_room_keeps_its_points_and_cursor_and_is_reported() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$a", 1, "x")]); + let mut rig = Rig::new(); + rig.ok(&mut history); + let (cursor, points) = (rig.cursor("!r:s"), rig.points()); + history.facts_fail.insert("!r:s".into()); + let (done, journal) = rig.run(&mut history); + let done = done.unwrap(); + assert_eq!((rig.cursor("!r:s"), rig.points()), (cursor, points)); + assert_eq!(done.failed, ["!r:s"]); + assert!(done.rooms.iter().any(|report| report.room_id == "!r:s" && report.failure.is_some())); + assert!(journal.iter().any(|line| line.contains("!r:s") && line.contains("failed"))); +} + +#[test] +fn the_tail_on_equal_ends_is_the_greatest_chunk_id() { + let mut rig = Rig::new(); + let (a, b) = (chunk_id("!r:s", "$a"), chunk_id("!r:s", "$b")); + rig.index.upsert_chat(&[ + ChatPoint { id: a.clone(), vector: vec![1.0; 4], payload: chunk_payload_of("!r:s", &[("$a", 5)]) }, + ChatPoint { id: b.clone(), vector: vec![1.0; 4], payload: chunk_payload_of("!r:s", &[("$b", 5)]) }, + ]) + .unwrap(); + assert_eq!(rig.index.chat_tail("!r:s").unwrap().unwrap().id, a.max(b)); } diff --git a/crates/huginn-daemon/src/sources/room_scope.rs b/crates/huginn-daemon/src/sources/room_scope.rs index 1c607be..8d7cf2d 100644 --- a/crates/huginn-daemon/src/sources/room_scope.rs +++ b/crates/huginn-daemon/src/sources/room_scope.rs @@ -16,8 +16,9 @@ use super::chat::RoomEvent; pub const ESTATE_ALIASES: [&str; 3] = ["#users:gamecult.org", "#members:gamecult.org", "#patrons:gamecult.org"]; /// What the server says about a room, as far as admission needs it. The Matrix -/// client fills it from the room's state and decides nothing. -#[derive(Clone, Debug, Default, PartialEq, Eq)] +/// client fills it from the room's state and decides nothing. It has no `Default`: a +/// room whose state could not be read is a port error, never a plain room. +#[derive(Clone, Debug, PartialEq, Eq)] pub struct RoomFacts { pub room_id: String, pub alias: Option, @@ -31,13 +32,25 @@ pub struct RoomFacts { pub bridge: Option, } -/// The estate room ids the caller knows (they exist only on the live server). -#[derive(Clone, Debug, Default, PartialEq, Eq)] +/// The estate room ids the caller knows (they exist only on the live server): exactly one per +/// estate alias, so a job that was handed none, or too few, cannot be built. +#[derive(Clone, Debug, PartialEq, Eq)] pub struct Estate { - pub ids: BTreeSet, + ids: BTreeSet, } impl Estate { + pub fn new(ids: BTreeSet) -> anyhow::Result { + anyhow::ensure!( + ids.len() == ESTATE_ALIASES.len() && ids.iter().all(|id| id.starts_with('!')), + "the estate is exactly {} room ids ({}), got {}", + ESTATE_ALIASES.len(), + ESTATE_ALIASES.join(", "), + ids.len() + ); + Ok(Self { ids }) + } + /// Whether `room_id` is a named estate room. Decided before the room's state is read. pub fn holds_id(&self, room_id: &str) -> bool { self.ids.contains(room_id) @@ -86,12 +99,14 @@ pub fn admit_room(facts: &RoomFacts, estate: &Estate) -> Result<(), Refusal> { Ok(()) } -/// Whether an event refuses its whole room: a sender whose localpart, lowercased -/// with leading underscores stripped, starts with the platform's name, or content -/// with a top-level key naming it. Deterministic identifier rules, not language reading. +/// Whether an event refuses its whole room: a sender whose localpart, lowercased, starts with the +/// platform's name after any leading underscores, has it after an underscore (`mautrix_discord`, +/// `_bridge_discordbot`), or has it followed by `bot` or an underscore; or content with a top-level key +/// naming it. Deterministic identifier rules, not language reading; `discourse` is not it. pub fn refuse_event(event: &RoomEvent) -> Option { - let localpart = event.sender.trim_start_matches('@').split(':').next().unwrap_or(""); - if localpart.to_ascii_lowercase().trim_start_matches('_').starts_with("discord") { + let localpart = event.sender.trim_start_matches('@').split(':').next().unwrap_or("").to_ascii_lowercase(); + let name = ["disc", "ord"].concat(); + if localpart.trim_start_matches('_').starts_with(&name) || localpart.contains(&format!("_{name}")) || localpart.contains(&format!("{name}bot")) || localpart.contains(&format!("{name}_")) { return Some(Refusal::DiscordSender); } event.content_keys.iter().any(|key| key.to_ascii_lowercase().contains("discord")).then_some(Refusal::DiscordContent) @@ -103,7 +118,11 @@ mod tests { use super::*; fn facts(id: &str) -> RoomFacts { - RoomFacts { room_id: id.into(), ..RoomFacts::default() } + RoomFacts { room_id: id.into(), alias: None, alt_aliases: vec![], name: None, encrypted: false, direct: false, bridge: None } + } + + fn estate(ids: &[&str]) -> Estate { + Estate::new(ids.iter().chain(["!pad1:s", "!pad2:s", "!pad3:s"].iter()).take(3).map(|id| (*id).to_owned()).collect()).unwrap() } fn event(sender: &str, keys: &[&str]) -> RoomEvent { @@ -112,7 +131,7 @@ mod tests { #[test] fn a_direct_encrypted_or_bridged_room_is_refused_and_a_plain_room_is_admitted() { - let estate = Estate::default(); + let estate = estate(&[]); assert_eq!(admit_room(&RoomFacts { direct: true, ..facts("!a:s") }, &estate), Err(Refusal::Direct)); assert_eq!(admit_room(&RoomFacts { encrypted: true, ..facts("!a:s") }, &estate), Err(Refusal::Encrypted)); for state_type in ["m.bridge", "uk.half-shot.bridge"] { @@ -123,31 +142,39 @@ mod tests { #[test] fn the_estate_is_refused_by_id_and_by_alias_in_any_case_and_a_lookalike_is_not() { - let estate = Estate { ids: ["!ids:s".to_owned()].into() }; - assert!(estate.holds_id("!ids:s") && !estate.holds_id("!other:s")); - assert_eq!(admit_room(&facts("!ids:s"), &estate), Err(Refusal::Estate)); + let named = estate(&["!ids:s"]); + assert!(named.holds_id("!ids:s") && !named.holds_id("!other:s")); + assert_eq!(admit_room(&facts("!ids:s"), &named), Err(Refusal::Estate)); for alias in ESTATE_ALIASES.iter().map(|alias| alias.to_string()).chain(["#USERS:Gamecult.Org".to_owned()]) { - assert_eq!(admit_room(&RoomFacts { alias: Some(alias.clone()), ..facts("!x:s") }, &Estate::default()), Err(Refusal::Estate), "{alias}"); - assert_eq!(admit_room(&RoomFacts { alt_aliases: vec!["#a:s".into(), alias.clone()], ..facts("!x:s") }, &Estate::default()), Err(Refusal::Estate), "alt {alias}"); + assert_eq!(admit_room(&RoomFacts { alias: Some(alias.clone()), ..facts("!x:s") }, &estate(&[])), Err(Refusal::Estate), "{alias}"); + assert_eq!(admit_room(&RoomFacts { alt_aliases: vec!["#a:s".into(), alias.clone()], ..facts("!x:s") }, &estate(&[])), Err(Refusal::Estate), "alt {alias}"); } for lookalike in ["#users-guide:gamecult.org", "#patrons-club:gamecult.org", "#users:gamecult.org.evil", "#users:other.org"] { - assert_eq!(admit_room(&RoomFacts { alias: Some(lookalike.into()), ..facts("!x:s") }, &Estate::default()), Ok(()), "{lookalike}"); + assert_eq!(admit_room(&RoomFacts { alias: Some(lookalike.into()), ..facts("!x:s") }, &estate(&[])), Ok(()), "{lookalike}"); } } #[test] fn a_bridged_platform_sender_or_content_key_refuses_and_a_lookalike_does_not() { - for sender in ["@discord_123:s", "@_discord_9:s", "@__DiscordBot:s", "@discordbot:s"] { + for sender in ["@discord_123:s", "@_discord_9:s", "@__DiscordBot:s", "@discordbot:s", "@mautrix_discord:s", "@_bridge_discordbot:s", "@a_discord:s"] { assert_eq!(refuse_event(&event(sender, &[])), Some(Refusal::DiscordSender), "{sender}"); } for key in ["fi.mau.discord", "uk.half-shot.discord", "X.DISCORD.y"] { assert_eq!(refuse_event(&event("@a:s", &["body", key])), Some(Refusal::DiscordContent), "{key}"); } - for sender in ["@discourse:s", "@disco:s", "@eureka:s", "@a_discord:s", "@x:discord.com"] { + for sender in ["@discourse:s", "@disco:s", "@eureka:s", "@x:discord.com"] { assert_eq!(refuse_event(&event(sender, &["discourse", "body"])), None, "{sender}"); } } + #[test] + fn the_estate_needs_exactly_its_room_ids() { + for ids in [vec![], vec!["!a:s"], vec!["!a:s", "!b:s", "!c:s", "!d:s"], vec!["!a:s", "!b:s", "notaroom"]] { + assert!(Estate::new(ids.iter().map(|id| (*id).to_owned()).collect()).is_err(), "{ids:?}"); + } + assert!(Estate::new(["!a:s", "!b:s", "!c:s"].map(str::to_owned).into()).is_ok()); + } + #[test] fn a_room_facts_carries_no_join_rule() { let source = include_str!("room_scope.rs"); diff --git a/crates/huginn-mind/src/redact.rs b/crates/huginn-mind/src/redact.rs index 9a17be3..23ccf57 100644 --- a/crates/huginn-mind/src/redact.rs +++ b/crates/huginn-mind/src/redact.rs @@ -107,7 +107,7 @@ fn redact_word(word: &str) -> String { /// The version of the scrub's rules. Part of the chat version: a bump makes every /// room re-ingest from the room's history and scrub it again. -pub const SCRUB_VERSION: u32 = 1; +pub const SCRUB_VERSION: u32 = 2; /// What replaces a secret. pub const REDACTED: &str = ""; @@ -121,6 +121,7 @@ pub enum ScrubKind { KeyShape, Jwt, PemBlock, + AccessToken, } impl std::fmt::Display for ScrubKind { @@ -132,6 +133,7 @@ impl std::fmt::Display for ScrubKind { Self::KeyShape => "key-shape", Self::Jwt => "jwt", Self::PemBlock => "pem-block", + Self::AccessToken => "access-token", }) } } @@ -143,26 +145,35 @@ pub struct Scrubbed { pub hits: Vec, } -/// Replaces credential shapes in `text`: a PEM private key block, an `authorization:` header line, -/// `Bearer`/`Basic`/`token` followed by a long token, a vendor key shape, a JWT, and URL userinfo. +/// Zero-width characters a secret could be split with. +fn zero_width(c: char) -> bool { + matches!(c, '\u{200b}' | '\u{200c}' | '\u{200d}' | '\u{2060}' | '\u{feff}' | '\u{00ad}' | '\u{180e}') +} + +/// Replaces credential shapes in `text`: a PEM or PGP private key block, an `authorization:` header +/// line, `Bearer`/`Basic`/`token` followed by a long token (blank or colon between), a vendor key +/// shape (Matrix tokens included), an `access_token` query or JSON value, a JWT, and URL userinfo. /// Each hit becomes ``. Token shapes only; ordinary prose, SHAs, UUIDs and ids pass byte for byte. +/// Zero-width characters are removed before matching; when anything matched, the returned text is the +/// stripped one, so a token split by them leaves no tail, and when nothing matched the text is returned as it came (emoji joiners stay). pub fn scrub(text: &str) -> Scrubbed { let mut hits = Vec::new(); - let mut text = text.to_owned(); - let passes: [(ScrubKind, fn(&str, usize) -> Option); 5] = [ + let mut stripped: String = text.chars().filter(|c| !zero_width(*c)).collect(); + let passes: [(ScrubKind, fn(&str, usize) -> Option); 6] = [ (ScrubKind::PemBlock, pem_block), (ScrubKind::AuthorizationHeader, authorization_line), + (ScrubKind::AccessToken, access_token), (ScrubKind::Bearer, bearer_token), (ScrubKind::KeyShape, key_shape), (ScrubKind::Jwt, jwt), ]; for (kind, find) in passes { - text = replace_matches(&text, kind, find, &mut hits); + stripped = replace_matches(&stripped, kind, find, &mut hits); } - let urls = redact_url_userinfo(&text); - let (before, after) = (text.matches(REDACTED).count(), urls.matches(REDACTED).count()); + let urls = redact_url_userinfo(&stripped); + let (before, after) = (stripped.matches(REDACTED).count(), urls.matches(REDACTED).count()); hits.extend(std::iter::repeat_n(ScrubKind::UrlUserinfo, after.saturating_sub(before))); - Scrubbed { text: urls, hits } + if hits.is_empty() { Scrubbed { text: text.to_owned(), hits } } else { Scrubbed { text: urls, hits } } } fn replace_matches(text: &str, kind: ScrubKind, find: fn(&str, usize) -> Option, hits: &mut Vec) -> String { @@ -212,12 +223,14 @@ fn pem_block(text: &str, at: usize) -> Option { let rest = &text[at..]; let label = rest.strip_prefix("-----BEGIN ")?; let words = run_end(label, 0, |byte| byte.is_ascii_uppercase() || byte == b' '); - label[..words].ends_with("PRIVATE KEY").then_some(())?; + (label[..words].ends_with("PRIVATE KEY") || label[..words].ends_with("PRIVATE KEY BLOCK")).then_some(())?; label[words..].starts_with("-----").then_some(())?; let body_from = at + "-----BEGIN ".len() + words + 5; - let end = text[body_from..] - .find("-----END ") - .and_then(|end| text[body_from + end..].find("PRIVATE KEY-----").map(|tail| body_from + end + tail + "PRIVATE KEY-----".len())); + let end = text[body_from..].find("-----END ").and_then(|end| { + let name = body_from + end + "-----END ".len(); + let words = run_end(text, name, |byte| byte.is_ascii_uppercase() || byte == b' '); + text[words..].starts_with("-----").then_some(words + 5) + }); Some(end.unwrap_or(text.len())) } @@ -234,20 +247,21 @@ fn bearer_token(text: &str, at: usize) -> Option { word_start(text, at).then_some(())?; let word = ["bearer", "basic", "token"].into_iter().find(|word| starts_with_ignore_case(text, at, word))?; let blanks = at + word.len(); - let value = run_end(text, blanks, |byte| byte == b' ' || byte == b'\t'); + let colon = usize::from(text.as_bytes().get(blanks) == Some(&b':')); + let value = run_end(text, blanks + colon, |byte| byte == b' ' || byte == b'\t'); (value > blanks).then_some(())?; let end = run_end(text, value, token_char); (end - value >= 16).then_some(end) } -/// A vendor key at a word start: the prefix, then a long enough run of the key alphabet. +/// A vendor key: the prefix, then a long enough run of the key alphabet. fn key_shape(text: &str, at: usize) -> Option { - word_start(text, at).then_some(())?; let rest = &text[at..]; - let by_prefix = [("sk-", 16), ("sk_live_", 16), ("pk_live_", 16), ("github_pat_", 16), ("glpat-", 16)]; + let by_prefix = [("sk-", 16), ("sk_live_", 16), ("pk_live_", 16), ("github_pat_", 16), ("glpat-", 16), ("syt_", 16), ("mct_", 16)]; let tokens = ["ghp_", "gho_", "ghu_", "ghs_", "ghr_"].into_iter().map(|prefix| (prefix, 16)); for (prefix, least) in by_prefix.into_iter().chain(tokens) { - if rest.starts_with(prefix) { + // Only the generic prefix needs a word start (`risk-averse`, `task-force`); the distinctive ones are found glued to anything. + if rest.starts_with(prefix) && (prefix != "sk-" || word_start(text, at)) { let end = run_end(text, at + prefix.len(), key_char); return (end - at - prefix.len() >= least).then_some(end); } @@ -271,9 +285,23 @@ fn key_shape(text: &str, at: usize) -> Option { None } +/// `access_token` as a query parameter or JSON member, through its value (8 or more characters). +fn access_token(text: &str, at: usize) -> Option { + starts_with_ignore_case(text, at, "access_token").then_some(())?; + let mut at = at + "access_token".len(); + let skip = |at: usize, set: &[u8]| run_end(text, at, |byte| set.contains(&byte)); + at = skip(at, b"\"'"); + at = skip(at, b" \t"); + matches!(text.as_bytes().get(at), Some(b'=' | b':')).then_some(())?; + at = skip(at + 1, b" \t"); + at = skip(at, b"\"'"); + let end = run_end(text, at, |byte| !byte.is_ascii_whitespace() && !matches!(byte, b'&' | b'"' | b'\'' | b',' | b'}' | b';' | b'#')); + (end - at >= 8).then_some(end) +} + /// Three base64url segments, the first starting `eyJ`. fn jwt(text: &str, at: usize) -> Option { - (word_start(text, at) && text[at..].starts_with("eyJ")).then_some(())?; + text[at..].starts_with("eyJ").then_some(())?; let segment = |from: usize| run_end(text, from, key_char); let first = segment(at); let mut end = first; @@ -383,6 +411,20 @@ mod tests { (ScrubKind::Jwt, "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJmYWtlIn0.c2lnbmF0dXJlZmFrZQ".into(), "c2lnbmF0dXJlZmFrZQ".into()), (ScrubKind::PemBlock, pem, "MIIEowIBAAKCAQEAfake".into()), (ScrubKind::PemBlock, "-----BEGIN PRIVATE KEY-----\nMIIBVQIBADANfakeNOEND".into(), "MIIBVQIBADANfake".into()), + (ScrubKind::KeyShape, format!("syt_{}_{}_{}", "YWxpY2U", "aBcDeFgHiJkLmNoPqR", "0a1B2c"), "aBcDeFgHiJkLmNoPqR".into()), + (ScrubKind::KeyShape, format!("mct_{}", "w4X5".repeat(8)), "w4X5w4X5w4X5".into()), + (ScrubKind::KeyShape, "xoxa-2-123456789012-aBcDeFgHiJkLmNoP".into(), "aBcDeFgHiJkL".into()), + (ScrubKind::KeyShape, "xoxp-123456789012-987654321098-aBcDeFgHiJkLmNoP".into(), "aBcDeFgHiJkL".into()), + (ScrubKind::KeyShape, "xoxr-123456789012-987654321098-aBcDeFgHiJkLmNoP".into(), "aBcDeFgHiJkL".into()), + (ScrubKind::KeyShape, "xoxs-123456789012-987654321098-aBcDeFgHiJkLmNoP".into(), "aBcDeFgHiJkL".into()), + (ScrubKind::KeyShape, "ASIA0123456789ABCDEF".into(), "0123456789ABCDEF".into()), + (ScrubKind::KeyShape, "gho_".to_owned() + &"p1Q2".repeat(9), "p1Q2p1Q2p1Q2".into()), + (ScrubKind::KeyShape, "pk_live_".to_owned() + &"e3R4".repeat(6), "e3R4e3R4e3R4".into()), + (ScrubKind::AccessToken, "https://h/_matrix/client/v3/sync?access_token=abcDEF123456xyz&since=s1".into(), "abcDEF123456xyz".into()), + (ScrubKind::AccessToken, "{\"access_token\": \"tokenValue987654\", \"user\": 1}".into(), "tokenValue987654".into()), + (ScrubKind::Bearer, format!("bearer:{}", "z9Y8x7".repeat(4)), "z9Y8x7z9Y8".into()), + (ScrubKind::Bearer, format!("token: {}", "m1N2o3".repeat(4)), "m1N2o3m1N2".into()), + (ScrubKind::PemBlock, "-----BEGIN PGP PRIVATE KEY BLOCK-----\nlQOYBFakeFakeFake\n-----END PGP PRIVATE KEY BLOCK-----".into(), "lQOYBFakeFake".into()), (ScrubKind::UrlUserinfo, "https://user:hunter2hunter2@host.example/r.git".into(), "hunter2hunter2".into()), (ScrubKind::UrlUserinfo, "https://tokenvalue12345@host.example/".into(), "tokenvalue12345".into()), ] @@ -441,4 +483,32 @@ mod tests { } } } + + #[test] + fn a_distinctive_key_is_found_glued_to_other_text_but_the_generic_prefix_still_needs_a_word_start() { + let ghp = format!("xxx{}", "ghp_".to_owned() + &"k2L9".repeat(9)); + assert_eq!(scrub(&ghp).text, "xxx"); + assert_eq!(scrub("idAKIA0123456789ABCDEFend").hits, [ScrubKind::KeyShape]); + assert_eq!(scrub(&format!("zzz{}", "xoxb-123456789012-aBcDeFgHiJkLmNoP")).hits, [ScrubKind::KeyShape]); + assert_eq!(scrub(&format!("q{}", "AIza".to_owned() + &"Sy1_".repeat(9))).hits, [ScrubKind::KeyShape]); + assert!(scrub("risk-averse-thinking-process task-force-alpha-bravo").hits.is_empty()); + } + + #[test] + fn zero_width_characters_inside_a_token_leave_no_tail_and_emoji_joiners_survive_when_nothing_matched() { + let split = format!("sk-{}\u{200b}{}", "q7W3".repeat(4), "q7W3".repeat(4)); + let out = scrub(&format!("key {split} end")); + assert_eq!(out.text, "key end"); + let jwt = "eyJhbGciOiJIUzI1NiJ9.eyJz\u{feff}dWIiOiJmYWtlIn0.c2lnbmF0dXJl\u{200d}ZmFrZQ"; + assert_eq!(scrub(jwt).text, ""); + let family = "family \u{1f468}\u{200d}\u{1f469}\u{200d}\u{1f467} here"; + assert_eq!(scrub(family).text, family); + } + + #[test] + fn matrix_tokens_and_query_values_are_found_and_prose_about_them_is_not() { + for same in ["the access_token is a thing", "access_token: short", "syt_ is the prefix", "mct_ tokens", "bearer:short"] { + assert_eq!(scrub(same).text, same, "{same}"); + } + } } diff --git a/docs/chat-retrieval-map.md b/docs/chat-retrieval-map.md index f839cd4..bee6079 100644 --- a/docs/chat-retrieval-map.md +++ b/docs/chat-retrieval-map.md @@ -169,3 +169,32 @@ the cited message and its neighbours in the same room from this index, no embedder and no Synapse; an id the index lacks is refused naming `event_id`. The listing limit above is not changed by it: the cut spec does not assign that fix, so it stays open. + +## Status: ingest core (cut chat-ingest-core) + +`sources::chat::refresh_chat` writes chat points; scope is `sources/room_scope.rs` +and the credential scrub is `huginn_mind::redact`. Rules the code holds, which a +reader of the index must know: + +- A text edit changes the stored body in place, only when its sender is the + original's sender. Edits are never stored as events, so no payload field + records which edit changed a body. +- **Rebuild on an unknown redaction.** A redaction whose target is neither a + message in the room's stored chunks nor an event of the same page may be of an + edit whose own id Huginn never stored, so the body that edit set cannot be + taken back from the chunk. The room is then rebuilt: its points and cursor are + purged and the next read re-ingests its history from the port. Synapse serves a + redacted edit without its content (verified against Synapse 1.164.1), so the + original body returns and the edited text does not. Within one page, a redacted + edit is reverted in the fold without a rebuild. Redactions the rebuild itself + replays do not trigger another rebuild. A redaction of a media or state event + also lands here (the cost is one rebuild of that room). +- **Edits of a redacted original are dropped.** Synapse keeps serving the + edits of a redacted original, with their replacement text, so Huginn drops any + edit whose target is not a message it holds or has just read (a redacted, + purged, never-stored or unknown original), live or late, in the fold and in a + rebuild alike. +- How the Matrix port maps a pruned event (`content: {}`) to `EventKind::Other` + is the ingest-job cut's to pin; this core only assumes it does. +- An error reading a joined room's facts is not a reclassification: the room + keeps its points and cursor, and the run's report for the room says why.