GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

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

Browse source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Eureka 2026-11-11 20:24:49 +00:00
commit 979b7307ca
7 changed files with 519 additions and 115 deletions
Download patch file Download diff file Expand all files Collapse all files

4
crates/huginn-daemon/src/index/fakes.rs
View file

@ -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<String>,
/// Every chat delete that went through: the room and the ids (`None` is all of the room).
pub(crate) chat_deletes: Vec<(String, Option<Vec<String>>)>,
/// 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(())

6
crates/huginn-daemon/src/index/qdrant.rs
View file

@ -467,7 +467,11 @@ impl SourceIndex for QdrantIndex {
fn chat_tail(&mut self, room_id: &str) -> Result<Option<StoredChat>> {
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<Vec<StoredChat>> {

166
crates/huginn-daemon/src/sources/chat.rs
View file

@ -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<String>,
}
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<ScrubHit>,
}
impl ChatReport {
fn empty(room_id: &str, alias: Option<String>) -> 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<String>, 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<S: SourceIndex>(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<String>,
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<S: SourceIndex>(index: &mut S, keep: &BTreeSet<String>) -> Result<Vec<String>> {
/// 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<S: SourceIndex>(index: &mut S, keep: &BTreeSet<String>) -> Result<Retired> {
let rooms: Vec<String> = 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<ChatMessage>,
ops: Vec<Op>,
/// Every event id of the page: a redaction of one of them is known.
seen: HashSet<String>,
}
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<Pending> = Vec::new();
let mut ops: Vec<Op> = 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<E: Embedder, S: SourceIndex>(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<E: Embedder, S: SourceIndex>(embedder: &mut E, index: &mut S, dimensions: u32, room: &Room, mut folded: Folded) -> Result<u32> {
fn write_page<E: Embedder, S: SourceIndex>(
embedder: &mut E,
index: &mut S,
dimensions: u32,
room: &Room,
mut folded: Folded,
may_rebuild: bool,
) -> Result<Option<u32>> {
let room_id = &room.facts.room_id;
let mut asked: Vec<String> = 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<ChatMessage>, Vec<ChatMessage>) = folded.pending.drain(..).partition(|message| held.contains(message.event_id.as_str()));
let mut ops: Vec<Op> = 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<E: Embedder, S: SourceIndex>(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<E: Embedder, S: SourceIndex, H: RoomHistory>(
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<String> = 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<E: Embedder, S: SourceIndex, H: RoomHistory>(
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<E: Embedder, S: SourceIndex, H: RoomHistory>(
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<E: Embedder, S: SourceIndex, H: RoomHistory>(
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<E: Embedder, S: SourceIndex, H: RoomHistory>(
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);
}
}
}

254
crates/huginn-daemon/src/sources/chat_tests.rs
View file

@ -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<RoomEvent>,
@ -28,11 +36,15 @@ struct FakeHistory {
joined: Vec<String>,
rooms: BTreeMap<String, Scripted>,
calls: Vec<String>,
/// Rooms whose state cannot be read.
facts_fail: BTreeSet<String>,
/// 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<RoomEvent>) -> &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<RoomFacts> {
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<RoomEvent> = 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<ChatRefreshed>, Vec<String>) {
@ -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<String> = 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::<Vec<_>>().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::<Vec<_>>());
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 <redacted>")));
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 <redacted> 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));
}

67
crates/huginn-daemon/src/sources/room_scope.rs
View file

@ -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<String>,
@ -31,13 +32,25 @@ pub struct RoomFacts {
pub bridge: Option<String>,
}
/// 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<String>,
ids: BTreeSet<String>,
}
impl Estate {
pub fn new(ids: BTreeSet<String>) -> anyhow::Result<Self> {
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<Refusal> {
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");

108
crates/huginn-mind/src/redact.rs
View file

@ -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 = "<redacted>";
@ -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<ScrubKind>,
}
/// 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 `<redacted>`. 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<usize>); 5] = [
let mut stripped: String = text.chars().filter(|c| !zero_width(*c)).collect();
let passes: [(ScrubKind, fn(&str, usize) -> Option<usize>); 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<usize>, hits: &mut Vec<ScrubKind>) -> String {
@ -212,12 +223,14 @@ fn pem_block(text: &str, at: usize) -> Option<usize> {
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<usize> {
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<usize> {
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<usize> {
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<usize> {
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<usize> {
(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<redacted>");
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 <redacted> end");
let jwt = "eyJhbGciOiJIUzI1NiJ9.eyJz\u{feff}dWIiOiJmYWtlIn0.c2lnbmF0dXJl\u{200d}ZmFrZQ";
assert_eq!(scrub(jwt).text, "<redacted>");
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}");
}
}
}

29
docs/chat-retrieval-map.md
View file

@ -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.