Matrix mouth: fetch and store chat images (cut A), ChatDelivery v3 #12
11 changed files with 847 additions and 28 deletions
|
|
@ -196,7 +196,7 @@ The command is a dry-run shape probe for public surfaces that are real enough to
|
|||
|
||||
Bifrost's native Discord gateway is the conversation observer. A direct mention of the Bifrost bot, or a reply to one of its messages, becomes an immutable typed `bifrost.discord.persona_conversation_event.v0` only when the guild and channel have one exact Persona target binding. Bot-authored, ambient, empty, over-1200-character, unbound, and ambiguously bound messages do not enter the conversation path. The 1200-character cap is the shared signed alpha boundary with Epiphany. The observation store is Bifrost-private working state, never an actuator queue, and insert collisions must contain exactly equal content. VoidBot does not observe, schedule, or answer for the target Persona.
|
||||
|
||||
Persona speech goes out through Matrix, not Discord. `tools/persona-matrix-mouth.mjs` is the one process for every Persona (`PERSONAS`, a list of slugs): it holds each Persona's Matrix access token as its own systemd credential (`matrix-token-<persona>`) and nobody else holds any. Per Persona it keeps one `/sync` loop over the configured rooms and the account's direct rooms (a DM invitation from an account on `MATRIX_SERVER_NAME` is accepted while that Persona's brake is released, and one from any other server is declined), one cursor and one delivery sequence. A delivery carries the sender's Matrix id, the event's `m.mentions` user ids and whether the room is a DM. A media event (`m.image`, `m.file`, `m.audio`, `m.video`) is delivered as a typed text line, `[image filename="cat.png" mime=image/png size=2048 url=mxc://server/id]`, followed by its caption when it has one, in `text` and in context; the wire stays v1. A Persona turn is a tool-less text call and cannot open a file or see an image, so the mouth fetches nothing from the media API. Reading wakes nothing and acks nothing: the heartbeat polls with the non-consuming `gamecult.persona.chat_pending_request.v1`, and only `chat_ack` advances a position. The mouth speaks on a `gamecult.persona.chat_send_request.v1` and joins rooms only while that Persona's brake (`/etc/gamecult/minds/<persona>/control.cc`) is released. One loopback CultNet RUDP endpoint serves all of them; a request naming an unlisted Persona gets a typed `chat_refusal` (`UnknownPersona`). Document shapes: `tools/persona-chat-documents.mjs`; golden bytes: `tests/fixtures/persona-chat-v1/`. The Discord posting actuator, its permit and its RUDP listener are removed.
|
||||
Persona speech goes out through Matrix, not Discord. `tools/persona-matrix-mouth.mjs` is the one process for every Persona (`PERSONAS`, a list of slugs): it holds each Persona's Matrix access token as its own systemd credential (`matrix-token-<persona>`) and nobody else holds any. Per Persona it keeps one `/sync` loop over the configured rooms and the account's direct rooms (a DM invitation from an account on `MATRIX_SERVER_NAME` is accepted while that Persona's brake is released, and one from any other server is declined), one cursor and one delivery sequence. A delivery carries the sender's Matrix id, the event's `m.mentions` user ids and whether the room is a DM. A media event (`m.image`, `m.file`, `m.audio`, `m.video`) is delivered as a typed text line, `[image filename="cat.png" mime=image/png size=2048 url=mxc://server/id]`, followed by its caption when it has one, in `text` and in context. Only `m.image` is also fetched, and only when `IMAGE_STORE_ROOT` is set (unset, the line above is all there is and nothing is downloaded). The fetch is made as the receiving Persona, with its own token, from `/_matrix/client/v1/media/download/<server>/<mediaId>` on its homeserver, for a message the Persona is delivered (not its own words, a room it is not joined to, or history before the ear), and only from the event's own `content.url`; a URL in the message text is never followed. An encrypted attachment (`m.room.encrypted`, or `content.file`) is never fetched. The type is decided by the magic bytes alone (png, jpeg, gif, webp; SVG and anything else is refused, the declared mime is ignored), the body is capped at 5 MiB by `Content-Length` and by the bytes actually read, and the pixel count read from the image header (never decoded) is capped at 40 MP. A refused or failed fetch leaves the typed line with a fixed-word reason (`refused=encrypted|no-url|fetch-failed|too-big|wrong-type|bad-header|too-many-pixels|store-failed`) and no image. A stored image is written to `IMAGE_STORE_ROOT/<persona>/<hex sha256 of the room id>/<hex sha256 of the bytes>[.png|.jpg|.gif|.webp]`, the name generated by the mouth, never the sender's: files `0640`, room directories `2750` (the Persona directory is setgid to the Persona's group, so the Persona's reader opens them by group; the mouth is the only writer), each written under a temporary name and renamed. It is deleted once it is at least 24 hours old and its delivery has been acked, whichever is later; a room keeps at most 50 MiB, the oldest evicted first. The sweep runs when an image is stored and when an ack arrives. A delivery is the 12-field tuple of `gamecult.persona.chat_pending.v2`: field 11, after `is_direct`, is nil or `[path, mime, size, sha256]`, a reference and never bytes; `path` is the store-relative `<room hash>/<file>` above, `mime` the sniffed type, `size` the file length (at most 5 MiB) and `sha256` its lowercase hex digest. The reader re-checks all four against the file. A Persona turn that has no reader for it still sees the typed line. Reading wakes nothing and acks nothing: the heartbeat polls with the non-consuming `gamecult.persona.chat_pending_request.v1`, and only `chat_ack` advances a position. The mouth speaks on a `gamecult.persona.chat_send_request.v1` and joins rooms only while that Persona's brake (`/etc/gamecult/minds/<persona>/control.cc`) is released. One loopback CultNet RUDP endpoint serves all of them; a request naming an unlisted Persona gets a typed `chat_refusal` (`UnknownPersona`). Document shapes: `tools/persona-chat-documents.mjs`; golden bytes: `tests/fixtures/persona-chat-v1/`, `-v2/` and `-v3/` (the pending answer; Huginn pins the same bytes). The Discord posting actuator, its permit and its RUDP listener are removed.
|
||||
|
||||
Bifrost admits an observation only when guild, channel, target runtime, canonical repository, Persona, producer, and producer runtime match a preconfigured `bifrost.persona_feedback.target_binding.v0` document in Bifrost's private provider store. That binding also owns source visibility, data classification, and the delivery-store target. Bifrost never infers public visibility from Discord identifiers. It classifies the author using its own `bifrost.discord.actor_link.v0` state: a match yields `linked_governance_feedback`; absence yields `unlinked_social_feedback`. Neither class grants adoption, work, release, or deployment authority.
|
||||
|
||||
|
|
|
|||
Binary file not shown.
BIN
tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.image.msgpack
vendored
Normal file
BIN
tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.image.msgpack
vendored
Normal file
Binary file not shown.
BIN
tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.no-image.msgpack
vendored
Normal file
BIN
tests/fixtures/persona-chat-v3/gamecult.persona.chat_pending.v2.no-image.msgpack
vendored
Normal file
Binary file not shown.
28
tests/image-bytes.mjs
Normal file
28
tests/image-bytes.mjs
Normal file
|
|
@ -0,0 +1,28 @@
|
|||
// Header-only image bytes for tests: just enough of each format for the mouth to sniff its type and size.
|
||||
export const png = (w = 4, h = 4, pad = 0) => {
|
||||
const b = Buffer.alloc(33 + pad);
|
||||
Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0, 0, 0, 13]).copy(b);
|
||||
b.write("IHDR", 12, "latin1");
|
||||
b.writeUInt32BE(w, 16);
|
||||
b.writeUInt32BE(h, 20);
|
||||
return b;
|
||||
};
|
||||
export const jpeg = (w = 4, h = 4) => Buffer.from([0xff, 0xd8, 0xff, 0xe0, 0, 4, 0, 0, 0xff, 0xc0, 0, 11, 8, h >> 8, h & 255, w >> 8, w & 255, 1, 1, 0x11, 0, 0xff, 0xda, 0, 2]);
|
||||
export const gif = (w = 4, h = 4) => Buffer.from([...Buffer.from("GIF89a", "latin1"), w & 255, w >> 8, h & 255, h >> 8, 0, 0, 0]);
|
||||
const riff = (chunk, body) => { const b = Buffer.alloc(12 + 8 + body.length); b.write("RIFF", 0, "latin1"); b.writeUInt32LE(b.length - 8, 4); b.write("WEBP", 8, "latin1"); b.write(chunk, 12, "latin1"); b.writeUInt32LE(body.length, 16); body.copy(b, 20); return b; };
|
||||
export const webpX = (w = 4, h = 4) => { const x = Buffer.alloc(10); x.writeUIntLE(w - 1, 4, 3); x.writeUIntLE(h - 1, 7, 3); return riff("VP8X", x); };
|
||||
export const webpL = (w = 4, h = 4) => { const x = Buffer.alloc(5); x[0] = 0x2f; const bits = (w - 1) | ((h - 1) << 14); x.writeUInt32LE(bits >>> 0, 1); return riff("VP8L", x); };
|
||||
export const webpLossy = (w = 4, h = 4) => { const x = Buffer.alloc(10); Buffer.from([0x9d, 0x01, 0x2a]).copy(x, 3); x.writeUInt16LE(w, 6); x.writeUInt16LE(h, 8); return riff("VP8 ", x); };
|
||||
export const svg = Buffer.from('<svg xmlns="http://www.w3.org/2000/svg" width="1" height="1"/>');
|
||||
|
||||
|
||||
// A Persona's image directory as the installer leaves it: group-inherited and setgid (2750), empty. Not the mouth's to chmod.
|
||||
import { chmodSync, mkdirSync, mkdtempSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
export function personaStoreRoot() {
|
||||
const root = join(mkdtempSync(join(tmpdir(), "images-")), "eureka");
|
||||
mkdirSync(root);
|
||||
chmodSync(root, 0o2750);
|
||||
return root;
|
||||
}
|
||||
308
tests/persona-chat-images.test.mjs
Normal file
308
tests/persona-chat-images.test.mjs
Normal file
|
|
@ -0,0 +1,308 @@
|
|||
// The image fetch and store behind the Matrix mouth's chat images: what is accepted by its bytes, what is bounded,
|
||||
// what the store writes and when it lets go. Fake fetch, temp directories, no network and no CultLib.
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { createHash } from "node:crypto";
|
||||
import { chmodSync, chownSync, existsSync, mkdirSync, readFileSync, readdirSync, statSync, writeFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { FETCH_TIMEOUT_MS, MAX_BYTES, MAX_PIXELS, RETAIN_MS, ROOM_CAP_BYTES, classify, createImageStore, fetchImage, sniff } from "../tools/persona-chat-images.mjs";
|
||||
import * as docs from "../tools/persona-chat-documents.mjs";
|
||||
import { gif, jpeg, personaStoreRoot, png, svg, webpL, webpLossy, webpX } from "./image-bytes.mjs";
|
||||
|
||||
const sha = data => createHash("sha256").update(data).digest("hex");
|
||||
const HS = "http://hs.test";
|
||||
const mxc = "mxc://gamecult.test/AbC_123-x";
|
||||
|
||||
// A fake fetch answering one request with `body` (bytes, or a function giving a ReadableStream) and `headers`.
|
||||
function fakeFetch({ status = 200, headers = {}, body }) {
|
||||
const calls = [];
|
||||
const fn = async (url, init) => {
|
||||
calls.push({ url, init });
|
||||
return new Response(typeof body === "function" ? body() : body, { status, headers });
|
||||
};
|
||||
fn.calls = calls;
|
||||
return fn;
|
||||
}
|
||||
const get = fetchLike => fetchImage({ fetch: fetchLike, homeserver: HS, token: "tok-own", mxc });
|
||||
|
||||
test("fetch uses this Persona's token on the authenticated media path of its own homeserver, without following redirects", async () => {
|
||||
const f = fakeFetch({ body: png() });
|
||||
const got = await get(f);
|
||||
assert.equal(got.mime, "image/png");
|
||||
assert.equal(f.calls.length, 1);
|
||||
assert.equal(f.calls[0].url, `${HS}/_matrix/client/v1/media/download/gamecult.test/AbC_123-x`);
|
||||
assert.equal(f.calls[0].init.headers.authorization, "Bearer tok-own");
|
||||
assert.equal(f.calls[0].init.redirect, "error");
|
||||
});
|
||||
|
||||
test("a redirect is refused, not followed: a fetch that honours redirect:error fails the request", async () => {
|
||||
// The stub behaves as fetch does: with redirect "error" a redirect response throws; with "follow" it would serve the image from elsewhere.
|
||||
const redirecting = async (url, init) => {
|
||||
if (init.redirect === "error") throw new TypeError("fetch failed: redirect");
|
||||
return new Response(png(), { status: 200 });
|
||||
};
|
||||
assert.deepEqual(await get(redirecting), { refusal: "fetch-failed" });
|
||||
});
|
||||
|
||||
test("a fetch that never answers is abandoned at the timeout, and the default is 30 s", async () => {
|
||||
assert.equal(FETCH_TIMEOUT_MS, 30000);
|
||||
let seen;
|
||||
const hanging = (url, init) => new Promise((_, reject) => { seen = init.signal; init.signal.addEventListener("abort", () => reject(init.signal.reason)); });
|
||||
const started = Date.now();
|
||||
assert.deepEqual(await fetchImage({ fetch: hanging, homeserver: HS, token: "tok-own", mxc, timeoutMs: 50 }), { refusal: "fetch-failed" });
|
||||
assert.ok(seen instanceof AbortSignal && seen.aborted);
|
||||
assert.ok(Date.now() - started < 5000);
|
||||
});
|
||||
|
||||
test("only a plain mxc url is fetched: no other scheme, no path climbing, nothing missing", async () => {
|
||||
for (const url of [undefined, null, 5, "https://evil.example/x.png", "mxc://gamecult.test/../etc", "mxc://gamecult.test/a/b", "mxc://gamecult.test/a?x=1", "mxc://gamecult.test/", "http://gamecult.test/_matrix/media/x"]) {
|
||||
const f = fakeFetch({ body: png() });
|
||||
assert.deepEqual(await fetchImage({ fetch: f, homeserver: HS, token: "t", mxc: url }), { refusal: "no-url" }, String(url));
|
||||
assert.equal(f.calls.length, 0);
|
||||
}
|
||||
});
|
||||
|
||||
test("the type is decided by the magic bytes: png, jpeg, gif and webp pass; svg and everything else are refused", async () => {
|
||||
for (const [bytes, mime] of [[png(), "image/png"], [jpeg(), "image/jpeg"], [gif(), "image/gif"], [webpX(), "image/webp"], [webpL(), "image/webp"], [webpLossy(), "image/webp"]]) {
|
||||
const got = await get(fakeFetch({ body: bytes }));
|
||||
assert.equal(got.mime, mime);
|
||||
assert.deepEqual(got.bytes, bytes);
|
||||
}
|
||||
for (const bytes of [svg, Buffer.from("<html></html>"), Buffer.from("%PDF-1.7"), Buffer.alloc(0), Buffer.from([0x42, 0x4d, 0, 0, 0, 0, 0, 0]), Buffer.from("RIFF....WAVEfmt ", "latin1")]) {
|
||||
assert.deepEqual(await get(fakeFetch({ body: bytes, headers: { "content-type": "image/png" } })), { refusal: "wrong-type" });
|
||||
}
|
||||
});
|
||||
|
||||
test("the declared type is never read: an svg labelled image/png is refused, a png labelled svg is accepted as png", async () => {
|
||||
assert.deepEqual(await get(fakeFetch({ body: svg, headers: { "content-type": "image/png" } })), { refusal: "wrong-type" });
|
||||
const got = await get(fakeFetch({ body: png(), headers: { "content-type": "image/svg+xml" } }));
|
||||
assert.equal(got.mime, "image/png");
|
||||
});
|
||||
|
||||
test("a Content-Length over 5 MiB is refused whatever the body turns out to be", async () => {
|
||||
const body = () => new ReadableStream({ pull(c) { c.enqueue(png()); c.close(); } });
|
||||
assert.deepEqual(await get(fakeFetch({ headers: { "content-length": String(MAX_BYTES + 1) }, body })), { refusal: "too-big" });
|
||||
assert.equal((await get(fakeFetch({ headers: { "content-length": String(MAX_BYTES) }, body }))).mime, "image/png", "at the cap it is read");
|
||||
});
|
||||
|
||||
test("the bytes actually read are capped at 5 MiB with a lying or absent Content-Length, and the stream is abandoned", async () => {
|
||||
for (const headers of [{}, { "content-length": "100" }]) {
|
||||
let pulled = 0;
|
||||
const chunk = Buffer.alloc(1024 * 1024);
|
||||
const body = () => new ReadableStream({ pull(c) { pulled++; c.enqueue(chunk); if (pulled === 8) c.close(); } });
|
||||
assert.deepEqual(await get(fakeFetch({ headers, body })), { refusal: "too-big" });
|
||||
assert.ok(pulled <= 7, `the stream was read ${pulled} chunks past the cap`);
|
||||
}
|
||||
});
|
||||
|
||||
test("exactly 5 MiB is allowed", async () => {
|
||||
const got = await get(fakeFetch({ body: png(4, 4, MAX_BYTES - 33) }));
|
||||
assert.equal(got.bytes.length, MAX_BYTES);
|
||||
});
|
||||
|
||||
test("the pixel cap is read from each header without decoding: 40 MP passes, one pixel row more is refused", () => {
|
||||
assert.equal(MAX_PIXELS, 8000 * 5000);
|
||||
for (const make of [png, jpeg, gif, webpX, webpL, webpLossy]) {
|
||||
assert.ok(classify(make(8000, 5000)).mime, `${make.name} at the cap`);
|
||||
assert.equal(classify(make(8000, 5001)).refusal, "too-many-pixels", `${make.name} over the cap`);
|
||||
}
|
||||
});
|
||||
|
||||
test("a header that will not parse is refused: truncated, no dimensions, zero size, or no frame before the scan", () => {
|
||||
assert.equal(classify(png().subarray(0, 20)).refusal, "bad-header");
|
||||
assert.equal(classify(png(0, 5)).refusal, "bad-header");
|
||||
assert.equal(classify(Buffer.concat([png().subarray(0, 12), Buffer.from("IDAT"), png().subarray(16)])).refusal, "bad-header");
|
||||
assert.equal(classify(jpeg().subarray(0, 12)).refusal, "bad-header");
|
||||
assert.equal(classify(Buffer.from([0xff, 0xd8, 0xff, 0xda, 0, 2, 0, 0])).refusal, "bad-header");
|
||||
assert.equal(classify(gif().subarray(0, 8)).refusal, "bad-header");
|
||||
assert.equal(classify(riffOnly()).refusal, "bad-header");
|
||||
assert.equal(sniff(Buffer.from([1, 2, 3])), null);
|
||||
});
|
||||
function riffOnly() { const b = Buffer.alloc(24); b.write("RIFF", 0, "latin1"); b.write("WEBP", 8, "latin1"); b.write("VP8X", 12, "latin1"); return b; }
|
||||
|
||||
test("an error status, a missing body or a throwing fetch is a fetch failure", async () => {
|
||||
assert.deepEqual(await get(fakeFetch({ status: 404, body: png() })), { refusal: "fetch-failed" });
|
||||
assert.deepEqual(await get(fakeFetch({ status: 200, body: null })), { refusal: "fetch-failed" });
|
||||
assert.deepEqual(await fetchImage({ fetch: async () => { throw new Error("boom tok-own"); }, homeserver: HS, token: "tok-own", mxc }), { refusal: "fetch-failed" });
|
||||
});
|
||||
|
||||
// ---- the store ----
|
||||
|
||||
function storeIn(clock = { t: Date.now() }) {
|
||||
const root = personaStoreRoot();
|
||||
return { root, clock, store: createImageStore({ root, now: () => clock.t }) };
|
||||
}
|
||||
const files = root => readdirSync(root).flatMap(d => readdirSync(join(root, d)).map(f => `${d}/${f}`));
|
||||
|
||||
test("the layout is <sha256 of the room id>/<sha256 of the bytes>.<ext>, generated here, relative and without climbing", () => {
|
||||
const { root, store } = storeIn();
|
||||
const room = "!Room:gamecult.test";
|
||||
const bytes = png(7, 9);
|
||||
const ref = store.put(room, bytes, "image/png");
|
||||
assert.deepEqual(ref, { path: `${sha(room)}/${sha(bytes)}.png`, mime: "image/png", size: bytes.length, sha256: sha(bytes) });
|
||||
assert.deepEqual(readFileSync(join(root, ref.path)), bytes);
|
||||
assert.ok(!ref.path.startsWith("/") && !ref.path.split("/").includes(".."));
|
||||
assert.match(ref.path, /^[0-9a-f]{64}\/[0-9a-f]{64}\.[a-z0-9]{1,5}$/);
|
||||
for (const [make, mime, ext] of [[jpeg, "image/jpeg", "jpg"], [gif, "image/gif", "gif"], [webpX, "image/webp", "webp"]]) assert.ok(store.put(room, make(), mime).path.endsWith(`.${ext}`));
|
||||
assert.equal(files(root).length, 4);
|
||||
});
|
||||
|
||||
test("a sender-chosen name never reaches the disk: the store takes no name at all, and a hostile room id is only hashed", () => {
|
||||
const { root, store } = storeIn();
|
||||
assert.equal(store.put.length, 3);
|
||||
store.put("!../../../etc:x", png(), "image/png");
|
||||
assert.deepEqual(readdirSync(root).filter(d => !/^[0-9a-f]{64}$/.test(d)), []);
|
||||
});
|
||||
|
||||
test("files are 0640 and room directories 0750 plus the setgid they inherit, whatever the umask, and no temp file is left", () => {
|
||||
const old = process.umask(0o077);
|
||||
try {
|
||||
const { root, store } = storeIn();
|
||||
const ref = store.put("!r:x", png(), "image/png");
|
||||
assert.equal(statSync(join(root, ref.path)).mode & 0o7777, 0o640);
|
||||
assert.equal(statSync(join(root, ref.path, "..")).mode & 0o7777, 0o2750);
|
||||
assert.equal(statSync(root).mode & 0o7777, 0o2750, "the Persona's directory is the installer's: untouched");
|
||||
assert.deepEqual(readdirSync(join(root, ref.path, "..")).filter(n => n.startsWith(".tmp-")), []);
|
||||
} finally { process.umask(old); }
|
||||
});
|
||||
|
||||
test("a room directory takes the Persona directory's group by inheritance: it is never chmodded, so a non-member's chmod cannot strip setgid", () => {
|
||||
// bifrost is not in the Persona's group. Stand in for that with a group the test user belongs to that is not its own.
|
||||
const other = process.getgroups().find(g => g !== process.getgid());
|
||||
if (other === undefined) return; // no second group to inherit from: the next test covers the missing-setgid refusal
|
||||
const { root, store } = storeIn();
|
||||
try { chownSync(root, process.getuid(), other); } catch { return; } // a sandbox that will not hand the directory to that group cannot show inheritance
|
||||
chmodSync(root, 0o2750);
|
||||
const ref = store.put("!r:x", png(), "image/png");
|
||||
assert.equal(statSync(join(root, ref.path, "..")).gid, other);
|
||||
assert.equal(statSync(join(root, ref.path)).gid, other);
|
||||
assert.equal(statSync(join(root, ref.path, "..")).mode & 0o2000, 0o2000);
|
||||
});
|
||||
|
||||
test("a Persona directory without setgid refuses the image: nothing is written, and the directories are not chmodded to hide it", () => {
|
||||
const { root, store } = storeIn();
|
||||
chmodSync(root, 0o750);
|
||||
assert.throws(() => store.put("!r:x", png(), "image/png"), { code: "ENOSETGID" });
|
||||
assert.deepEqual(files(root), []);
|
||||
assert.equal(statSync(root).mode & 0o7777, 0o750, "the installer's directory keeps its mode");
|
||||
assert.equal(statSync(join(root, sha("!r:x"))).mode & 0o7777, 0o750, "the room directory it made is left as mkdir made it");
|
||||
});
|
||||
|
||||
|
||||
test("a room directory whose group is not the Persona directory's is refused: nothing is written", () => {
|
||||
const root = personaStoreRoot();
|
||||
const wrong = createImageStore({ root, stat: path => { const s = statSync(path); return path === root ? s : Object.assign(Object.create(Object.getPrototypeOf(s)), s, { gid: s.gid + 1 }); } });
|
||||
assert.throws(() => wrong.put("!r:x", png(), "image/png"), { code: "ENOGROUP" });
|
||||
assert.deepEqual(files(root), []);
|
||||
const right = createImageStore({ root });
|
||||
assert.ok(right.put("!r:x", png(), "image/png").path, "the same directory is accepted when the group matches");
|
||||
});
|
||||
|
||||
test("a missing Persona directory is refused, not created, and nothing above or below it appears", () => {
|
||||
const parent = join(personaStoreRoot(), "..");
|
||||
const root = join(parent, "absent", "eureka");
|
||||
const store = createImageStore({ root });
|
||||
assert.throws(() => store.put("!r:x", png(), "image/png"), { code: "ENOPERSONADIR" });
|
||||
assert.equal(existsSync(join(parent, "absent")), false);
|
||||
const file = join(parent, "not-a-dir");
|
||||
writeFileSync(file, "x");
|
||||
assert.throws(() => createImageStore({ root: file }).put("!r:x", png(), "image/png"), { code: "ENOPERSONADIR" });
|
||||
});
|
||||
|
||||
test("a write that cannot finish leaves no temp file and no partial image under the final name", () => {
|
||||
const { root, store } = storeIn();
|
||||
const bytes = png(6, 6);
|
||||
mkdirSync(join(root, sha("!r:x"), `${sha(bytes)}.png`), { recursive: true }); // the final name is taken by a directory: the rename fails
|
||||
assert.throws(() => store.put("!r:x", bytes, "image/png"));
|
||||
assert.deepEqual(readdirSync(join(root, sha("!r:x"))).filter(n => n.startsWith(".tmp-")), []);
|
||||
});
|
||||
|
||||
test("a stored image stays until it is at least 24 h old and acked, whichever is later", () => {
|
||||
const { root, clock, store } = storeIn();
|
||||
const ref = store.put("!r:x", png(), "image/png");
|
||||
store.hold(ref.path, 5);
|
||||
clock.t += RETAIN_MS + 1000;
|
||||
store.sweep();
|
||||
assert.ok(existsSync(join(root, ref.path)), "old but not acked: kept");
|
||||
store.acked(4);
|
||||
assert.ok(existsSync(join(root, ref.path)), "an ack short of its delivery: kept");
|
||||
store.acked(5);
|
||||
assert.ok(!existsSync(join(root, ref.path)), "old and acked: gone");
|
||||
|
||||
const young = store.put("!r:x", png(9, 9), "image/png");
|
||||
store.hold(young.path, 6);
|
||||
store.acked(6);
|
||||
clock.t += RETAIN_MS - 1000;
|
||||
store.sweep();
|
||||
assert.ok(existsSync(join(root, young.path)), "acked but under 24 h: kept");
|
||||
clock.t += 2000;
|
||||
store.sweep();
|
||||
assert.ok(!existsSync(join(root, young.path)), "acked and 24 h old: gone");
|
||||
});
|
||||
|
||||
test("after a restart what is on disk counts as acked and a stray temp file is removed", () => {
|
||||
const { root, clock, store } = storeIn();
|
||||
const ref = store.put("!r:x", png(), "image/png");
|
||||
const dir = join(root, ref.path, "..");
|
||||
const stray = join(dir, ".tmp-dead");
|
||||
writeFileSync(stray, "x");
|
||||
const again = createImageStore({ root, now: () => clock.t + RETAIN_MS * 2 });
|
||||
again.sweep();
|
||||
assert.ok(!existsSync(join(root, ref.path)));
|
||||
assert.ok(!existsSync(stray));
|
||||
});
|
||||
|
||||
test("a room holds at most 50 MiB: the oldest go first, other rooms are untouched, the newest is kept", () => {
|
||||
const { root, clock, store } = storeIn();
|
||||
const big = n => png(4, 4, 5 * 1024 * 1024 - 33 - n); // distinct bytes, just under 5 MiB each, so ten fit and eleven do not
|
||||
const other = store.put("!other:x", big(900), "image/png");
|
||||
clock.t += 1;
|
||||
const refs = [];
|
||||
for (let i = 0; i < 10; i++) { clock.t += 1; const b = big(i); b[40] = i + 1; refs.push(store.put("!r:x", b, "image/png")); }
|
||||
assert.ok(refs.every(r => existsSync(join(root, r.path))), "ten images under the cap are all kept");
|
||||
clock.t += 1;
|
||||
const b = big(50); b[40] = 99;
|
||||
const newest = store.put("!r:x", b, "image/png");
|
||||
assert.ok(!existsSync(join(root, refs[0].path)), "the oldest is evicted");
|
||||
assert.ok(refs.slice(1).every(r => existsSync(join(root, r.path))));
|
||||
assert.ok(existsSync(join(root, newest.path)));
|
||||
assert.ok(existsSync(join(root, other.path)), "another room's image is not counted");
|
||||
const total = refs.slice(1).concat(newest).reduce((n, r) => n + r.size, 0);
|
||||
assert.ok(total <= ROOM_CAP_BYTES);
|
||||
});
|
||||
|
||||
// ---- the delivery reference ----
|
||||
|
||||
const ref = { path: `${"cd".repeat(32)}/${"ab".repeat(32)}.png`, mime: "image/png", size: 2048, sha256: "ab".repeat(32) };
|
||||
const base = { seq: 1, roomAlias: "#r:x", roomId: "!r:x", eventId: "$e", sender: "@a:x", senderDisplay: "A", text: "t", atUnixMs: 1, context: [], mentions: [], isDirect: false };
|
||||
|
||||
test("a delivery is a 12-tuple whose index 11 is nil or [path, mime, size, sha256]", () => {
|
||||
const none = docs.deliveryTuple({ ...base, image: null });
|
||||
assert.equal(none.length, 12);
|
||||
assert.equal(none[11], null);
|
||||
assert.equal(docs.deliveryTuple(base)[11], null, "no image given reads as none");
|
||||
const some = docs.deliveryTuple({ ...base, image: ref });
|
||||
assert.equal(some.length, 12);
|
||||
assert.deepEqual(some[11], [ref.path, ref.mime, ref.size, ref.sha256]);
|
||||
assert.deepEqual(docs.parseDelivery(some).image, ref);
|
||||
assert.equal(docs.parseDelivery(none).image, null);
|
||||
assert.throws(() => docs.parseDelivery(some.slice(0, 11)), /12-field/, "the 11-field delivery is gone: one shape");
|
||||
assert.equal(docs.kinds.pending, "gamecult.persona.chat_pending.v2");
|
||||
});
|
||||
|
||||
test("an image path is never absolute, never climbs, never another shape; mime, size and hash are bounded", () => {
|
||||
const bad = patch => () => docs.deliveryTuple({ ...base, image: { ...ref, ...patch } });
|
||||
for (const path of [`/${ref.path}`, `../${ref.path}`, `${"cd".repeat(32)}/../${"ab".repeat(32)}.png`, `${"cd".repeat(32)}/${"ab".repeat(32)}.png/`, "etc/passwd", `${"CD".repeat(32)}/${"ab".repeat(32)}.png`,
|
||||
`${"cd".repeat(32)}/${"ab".repeat(32)}.toolong`, `${"cd".repeat(32)}/${"ab".repeat(32)}.jpg`, `${"cd".repeat(32)}/${"ef".repeat(32)}.png`, `${"cd".repeat(32)}/${"ab".repeat(32)}.`, `${"cd".repeat(31)}/${"ab".repeat(32)}.png`, "", 5]) {
|
||||
assert.throws(bad({ path }), Error, String(path));
|
||||
}
|
||||
assert.throws(bad({ mime: "image/svg+xml" }));
|
||||
assert.throws(bad({ mime: "image/png ; x" }));
|
||||
assert.throws(bad({ size: MAX_BYTES + 1 }));
|
||||
assert.throws(bad({ size: 0 }));
|
||||
assert.throws(bad({ size: 1.5 }));
|
||||
assert.throws(bad({ sha256: "AB".repeat(32) }));
|
||||
assert.throws(() => docs.parseDelivery([...docs.deliveryTuple({ ...base, image: null }).slice(0, 11), ["../x", "image/png", 1, "a".repeat(64)]]));
|
||||
const bare = { ...ref, path: `${"cd".repeat(32)}/${"ab".repeat(32)}` };
|
||||
assert.deepEqual(docs.deliveryTuple({ ...base, image: bare })[11][0], bare.path, "the extension is optional");
|
||||
});
|
||||
|
|
@ -21,11 +21,11 @@ const delivery = {
|
|||
seq: 7, roomAlias: "#operator:gamecult.org", roomId: "!abc:gamecult.org", eventId: "$e1", sender: "@meta:gamecult.org", senderDisplay: "Meta",
|
||||
text: "hello eureka", atUnixMs: 1761700000000,
|
||||
context: [{ sender: "@eureka:gamecult.org", text: "earlier", atUnixMs: 1761699999000 }],
|
||||
mentions: ["@eureka:gamecult.org", "@nova:gamecult.org"], isDirect: true,
|
||||
mentions: ["@eureka:gamecult.org", "@nova:gamecult.org"], isDirect: true, image: null,
|
||||
};
|
||||
const image = { path: `${"cd".repeat(32)}/${"ab".repeat(32)}.png`, mime: "image/png", size: 2048, sha256: "ab".repeat(32) };
|
||||
const samples = {
|
||||
[docs.kinds.pendingRequest]: { tuple: docs.pendingRequestTuple({ persona: "eureka", afterSeq: 3 }), parse: docs.parsePendingRequest },
|
||||
[docs.kinds.pending]: { tuple: docs.pendingTuple({ persona: "eureka", deliveries: [delivery] }), parse: docs.parsePending },
|
||||
[docs.kinds.ack]: { tuple: docs.ackTuple({ persona: "eureka", throughSeq: 7 }), parse: docs.parseAck },
|
||||
[docs.kinds.sendRequest]: { tuple: docs.sendRequestTuple({ requestId: "req-1", persona: "eureka", roomAlias: "#operator:gamecult.org", text: "hi" }), parse: docs.parseSendRequest },
|
||||
[docs.kinds.sendReceipt]: { tuple: docs.sendReceiptTuple({ requestId: "req-1", eventId: "$sent" }), parse: docs.parseSendReceipt },
|
||||
|
|
@ -49,11 +49,11 @@ test("the chat documents refuse malformed tuples", () => {
|
|||
assert.throws(() => docs.parseSendRequest(["r", "eureka", "#a:b", "x".repeat(4001)]));
|
||||
assert.throws(() => docs.parseSendRequest(["r", "eureka", "#a:b", 5]));
|
||||
const wire = docs.deliveryTuple(delivery);
|
||||
assert.equal(wire.length, 11);
|
||||
assert.equal(wire.length, 12);
|
||||
assert.deepEqual(docs.parseDelivery(wire), delivery);
|
||||
assert.throws(() => docs.parseDelivery(wire.slice(0, 9)), /11-field/, "the nine-field delivery is gone: one shape");
|
||||
assert.throws(() => docs.parseDelivery([...wire.slice(0, 10), "yes"]), /boolean/);
|
||||
assert.throws(() => docs.parseDelivery([...wire.slice(0, 9), [""], true]));
|
||||
assert.throws(() => docs.parseDelivery(wire.slice(0, 11)), /12-field/, "the eleven-field delivery is gone: one shape");
|
||||
assert.throws(() => docs.parseDelivery([...wire.slice(0, 10), "yes", null]), /boolean/);
|
||||
assert.throws(() => docs.parseDelivery([...wire.slice(0, 9), [""], true, null]));
|
||||
assert.throws(() => docs.refusalTuple({ persona: "p", refusal: "Invented" }));
|
||||
assert.throws(() => docs.parseSendReceipt(["r", "$e", "Brake"]), /exactly one/);
|
||||
assert.throws(() => docs.parseSendReceipt(["r", null, null]), /exactly one/);
|
||||
|
|
@ -280,6 +280,23 @@ test("a reply is paged to the byte budget, oldest first, and one delivery that w
|
|||
assert.equal(pageOf("eureka", [mk(1, "x".repeat(4000))], msgpack)[1].length, 1, "a single delivery is always served");
|
||||
});
|
||||
|
||||
// The pending answer v2 (ChatDelivery v3): the bytes Huginn pins in its crates/mind-body/tests/fixtures/persona-chat-v3/.
|
||||
const v3Dir = resolve(root, "tests", "fixtures", "persona-chat-v3");
|
||||
const v3Samples = {
|
||||
"gamecult.persona.chat_pending.v2.image": { tuple: docs.pendingTuple({ persona: "eureka", deliveries: [{ ...delivery, image }] }), parse: docs.parsePending },
|
||||
"gamecult.persona.chat_pending.v2.no-image": { tuple: docs.pendingTuple({ persona: "eureka", deliveries: [delivery] }), parse: docs.parsePending },
|
||||
};
|
||||
for (const [name, { tuple, parse }] of Object.entries(v3Samples)) {
|
||||
test(`golden fixture: ${name} is byte-stable and parses back`, async () => {
|
||||
const bytes = Buffer.from(msgpack.encode(tuple));
|
||||
const file = resolve(v3Dir, `${name}.msgpack`);
|
||||
if (process.env.UPDATE_FIXTURES === "1") { await mkdir(v3Dir, { recursive: true }); await writeFile(file, bytes); }
|
||||
assert.deepEqual(bytes, await readFile(file), "wire bytes changed: Huginn decodes these fixtures");
|
||||
assert.deepEqual(parse(msgpack.decode(bytes)), parse(tuple));
|
||||
assert.equal(msgpack.decode(bytes)[1][0].length, 12);
|
||||
});
|
||||
}
|
||||
|
||||
// The send pair v2: the bytes the Huginn organ pins from its side.
|
||||
const v2Dir = resolve(root, "tests", "fixtures", "persona-chat-v2");
|
||||
const v2Request = basis => docs.sendRequestV2Tuple({ requestId: "req-1", persona: "eureka", roomAlias: "#swarm:gamecult.org", text: "hi", basis });
|
||||
|
|
|
|||
|
|
@ -1,8 +1,15 @@
|
|||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { createServer } from "node:http";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { readFileSync, readdirSync, statSync } from "node:fs";
|
||||
import { resolve, join } from "node:path";
|
||||
import { existsSync, mkdtempSync, statSync as stat } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { createHash } from "node:crypto";
|
||||
import { createImageStore, RETAIN_MS } from "../tools/persona-chat-images.mjs";
|
||||
import { personaStoreRoot } from "./image-bytes.mjs";
|
||||
import * as docs from "../tools/persona-chat-documents.mjs";
|
||||
import { gif, png, svg } from "./image-bytes.mjs";
|
||||
import { createMouth, bodyText, createBrakeReader, guarded, redact, roomsFor, projectRoomsVariable } from "../tools/persona-matrix-mouth.mjs";
|
||||
|
||||
const TOKEN = "syt_SECRET_TOKEN_9f3";
|
||||
|
|
@ -16,12 +23,13 @@ const tick = () => new Promise(r => setImmediate(r));
|
|||
// A fake Matrix client-server API: just the endpoints the mouth uses.
|
||||
// It is also the record of every room: events the sync script emits and messages sent are kept, and
|
||||
// `preload` (room id -> events, oldest first) stands for history from before the mouth's process began.
|
||||
async function fakeHomeserver({ sync, failWith, users, member, direct, preload = {}, failContext = false } = {}) {
|
||||
async function fakeHomeserver({ sync, failWith, users, member, direct, preload = {}, failContext = false, media } = {}) {
|
||||
const seen = [];
|
||||
const record_ = new Map(Object.entries(preload).map(([room, events]) => [room, [...events]]));
|
||||
const remember = (room, event) => { const list = record_.get(room) ?? []; if (!list.some(e => e.event_id === event.event_id)) list.push(event); record_.set(room, list); };
|
||||
const sent = new Map(); // txnId -> event_id
|
||||
const server = createServer((req, res) => {
|
||||
// The homeserver is answered in-process (no socket): the suite runs under a capped address space, where undici's HTTP client cannot start.
|
||||
const handler = (req, res) => {
|
||||
let raw = "";
|
||||
req.on("data", c => (raw += c));
|
||||
req.on("end", () => {
|
||||
|
|
@ -32,6 +40,11 @@ async function fakeHomeserver({ sync, failWith, users, member, direct, preload =
|
|||
if (failWith) return reply(failWith, { errcode: "M_UNKNOWN_TOKEN", error: `bad token ${record.auth}` });
|
||||
const p = record.path.replace("/_matrix/client/v3", "").replace("/_matrix/client/v1", "");
|
||||
const bearer = (record.auth ?? "").replace("Bearer ", "");
|
||||
if (p.startsWith("/media/download/")) {
|
||||
const served = media?.(bearer, ...p.slice("/media/download/".length).split("/")) ?? { status: 404, body: "" };
|
||||
res.writeHead(served.status ?? 200, served.headers ?? {});
|
||||
return res.end(served.body);
|
||||
}
|
||||
if (p === "/account/whoami") return reply(200, { user_id: users?.[bearer] ?? SELF });
|
||||
if (p === "/joined_rooms") return reply(200, { joined_rooms: member ?? Object.values(ROOMS) });
|
||||
if (req.method === "PUT" && p.includes("/account_data/m.direct")) return reply(200, {});
|
||||
|
|
@ -63,13 +76,24 @@ async function fakeHomeserver({ sync, failWith, users, member, direct, preload =
|
|||
}
|
||||
reply(404, { errcode: "M_NOT_FOUND" });
|
||||
});
|
||||
};
|
||||
const fetch = (url, init = {}) => new Promise(resolve => {
|
||||
const u = new URL(url);
|
||||
const req = Object.assign(new EventEmitter(), { method: init.method ?? "GET", url: u.pathname + u.search, headers: init.headers ?? {} });
|
||||
const res = {
|
||||
writeHead(status, headers) { this.status = status; this.headers = headers; },
|
||||
end(body) { resolve(new Response(body === "" ? null : body, { status: this.status, headers: this.headers })); },
|
||||
};
|
||||
handler(req, res);
|
||||
process.nextTick(() => { if (init.body !== undefined) req.emit("data", init.body); req.emit("end"); });
|
||||
});
|
||||
await new Promise(ok => server.listen(0, "127.0.0.1", ok));
|
||||
return {
|
||||
url: `http://127.0.0.1:${server.address().port}`,
|
||||
url: "http://hs.test",
|
||||
fetch,
|
||||
seen,
|
||||
puts: () => seen.filter(r => r.method === "PUT"),
|
||||
close: () => new Promise(ok => server.close(ok)),
|
||||
downloads: () => seen.filter(r => r.path.includes("/media/download/")),
|
||||
close: async () => {},
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -81,7 +105,7 @@ function memoryState(initial = null) {
|
|||
const joined = events => ({ rooms: { join: events } });
|
||||
function mouthFor(hs, overrides = {}) {
|
||||
return createMouth({
|
||||
persona: "eureka", homeserver: hs.url, rooms: Object.keys(ROOMS), token: TOKEN, state: memoryState(),
|
||||
persona: "eureka", homeserver: hs.url, rooms: Object.keys(ROOMS), token: TOKEN, state: memoryState(), fetch: hs.fetch,
|
||||
serverName: HS_NAME, brake: async () => true, log: () => {}, syncTimeoutMs: 1, ...overrides,
|
||||
});
|
||||
}
|
||||
|
|
@ -429,7 +453,7 @@ test("join_while_braked: no join and no send while the brake is held; joining ha
|
|||
|
||||
const failing = await fakeHomeserver({ sync: script, member: [] });
|
||||
try {
|
||||
const mouth = mouthFor(failing, { fetch: (url, init) => String(url).includes("/join/") ? Promise.reject(new Error(`join refused for ${init.headers.authorization}`)) : fetch(url, init), log: l => logs.push(l) });
|
||||
const mouth = mouthFor(failing, { fetch: (url, init) => String(url).includes("/join/") ? Promise.reject(new Error(`join refused for ${init.headers.authorization}`)) : failing.fetch(url, init), log: l => logs.push(l) });
|
||||
await mouth.syncOnce(); // no throw: the unit does not crash-loop on a failed join
|
||||
assert.ok(logs.some(l => l.includes("not joined")) && logs.every(l => !l.includes(TOKEN)));
|
||||
} finally { await failing.close(); }
|
||||
|
|
@ -753,3 +777,167 @@ test("media_delivery: a Matrix image reaches the delivery and a later delivery's
|
|||
assert.equal(JSON.stringify(mouth.pendingAfter(0)).includes(TOKEN), false);
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
// ---- chat images ----
|
||||
|
||||
const IMG_ROOM = "!swarm:gamecult.test";
|
||||
const sha = data => createHash("sha256").update(data).digest("hex");
|
||||
const imageEvent = (id, content, extra = {}) => ({ type: "m.room.message", event_id: id, sender: `@alice:${HS_NAME}`, origin_server_ts: 1000, content: media("m.image", { body: "cat.png", url: MXC_URL, info: { mimetype: "image/png", size: 2048 }, ...content }), ...extra });
|
||||
const imageTimeline = (events, room = IMG_ROOM) => ({
|
||||
null: { next_batch: "b0", ...joined({ [IMG_ROOM]: { timeline: { events: [] } } }) },
|
||||
b0: { next_batch: "b1", ...joined({ [room]: { timeline: { events } } }) },
|
||||
});
|
||||
const served = bytes => (bearer, server, id) => (bearer === TOKEN && server === "gamecult.test" && id === "AbC_123-x" ? { status: 200, body: bytes } : { status: 401, body: "" });
|
||||
function imageSetup(clock = { t: Date.now() }) {
|
||||
const root = personaStoreRoot();
|
||||
return { root, clock, images: createImageStore({ root, now: () => clock.t }) };
|
||||
}
|
||||
async function receive(events, { media: mediaFn, room, images, mouthOptions = {} } = {}) {
|
||||
const tl = imageTimeline(events, room);
|
||||
const hs = await fakeHomeserver({ sync: since => tl[since] ?? { next_batch: since }, media: mediaFn });
|
||||
const mouth = mouthFor(hs, { images, ...mouthOptions });
|
||||
await mouth.syncOnce();
|
||||
const made = await mouth.syncOnce();
|
||||
return { hs, mouth, made };
|
||||
}
|
||||
|
||||
test("image_delivery: an m.image event is fetched as this Persona, stored by hash, and the delivery carries the reference in the v3 tuple", async () => {
|
||||
const bytes = png(40, 30);
|
||||
const { root, images } = imageSetup();
|
||||
const { hs, made } = await receive([imageEvent("$img", {})], { media: served(bytes), images });
|
||||
try {
|
||||
assert.equal(made.length, 1);
|
||||
const [d] = made;
|
||||
assert.deepEqual(d.image, { path: `${sha(IMG_ROOM)}/${sha(bytes)}.png`, mime: "image/png", size: bytes.length, sha256: sha(bytes) });
|
||||
assert.ok(existsSync(join(root, d.image.path)));
|
||||
assert.deepEqual(hs.downloads().map(r => [r.method, r.path, r.auth]), [["GET", "/_matrix/client/v1/media/download/gamecult.test/AbC_123-x", `Bearer ${TOKEN}`]]);
|
||||
assert.equal(d.text, `[image filename="cat.png" mime=image/png size=2048 url=${MXC_URL}]`, "the text line is unchanged when the image is stored");
|
||||
const tuple = docs.deliveryTuple(d);
|
||||
assert.equal(tuple.length, 12);
|
||||
assert.deepEqual(tuple[11], [d.image.path, "image/png", bytes.length, sha(bytes)]);
|
||||
assert.ok(!d.image.path.startsWith("/") && !d.image.path.includes(".."));
|
||||
assert.equal(stat(join(root, d.image.path)).mode & 0o777, 0o640);
|
||||
assert.equal(stat(join(root, d.image.path, "..")).mode & 0o777, 0o750);
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
test("image_delivery: the file name is ours, never the sender's filename, body or info", async () => {
|
||||
const bytes = png(5, 5);
|
||||
const { root, images } = imageSetup();
|
||||
const { hs, made } = await receive([imageEvent("$img", { body: "../../etc/passwd", filename: "/abs/evil.sh", info: { mimetype: "text/x-shellscript", size: 1 } })], { media: served(bytes), images });
|
||||
try {
|
||||
assert.equal(made[0].image.path, `${sha(IMG_ROOM)}/${sha(bytes)}.png`);
|
||||
assert.equal(made[0].image.mime, "image/png", "the sniffed type, not the declared one");
|
||||
assert.ok(!JSON.stringify(made[0].image).includes("evil"));
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
test("image_delivery: only an event the Persona is delivered is fetched: not her own, not another room's, not history, not a plain message naming an mxc url", async () => {
|
||||
const { images } = imageSetup();
|
||||
const events = [
|
||||
imageEvent("$own", {}, { sender: SELF }),
|
||||
imageEvent("$x", {}),
|
||||
{ ...message("$text", `@alice:${HS_NAME}`, `see ${MXC_URL} and https://evil.example/a.png`, 1100), content: { msgtype: "m.text", body: `see ${MXC_URL}` } },
|
||||
];
|
||||
const tl = {
|
||||
null: { next_batch: "b0", ...joined({ [IMG_ROOM]: { timeline: { events: [imageEvent("$hist", {})] } } }) },
|
||||
b0: { next_batch: "b1", ...joined({ [IMG_ROOM]: { timeline: { events: [events[0], events[2]] } }, "!elsewhere:gamecult.test": { timeline: { events: [events[1]] } } }) },
|
||||
};
|
||||
const hs = await fakeHomeserver({ sync: since => tl[since] ?? { next_batch: since }, media: served(png()) });
|
||||
try {
|
||||
const mouth = mouthFor(hs, { images });
|
||||
await mouth.syncOnce();
|
||||
const made = await mouth.syncOnce();
|
||||
assert.deepEqual(made.map(d => d.eventId), ["$text"]);
|
||||
assert.equal(made[0].image, null);
|
||||
assert.deepEqual(hs.downloads(), []);
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
test("image_delivery: the url is the event's own content.url: a body or caption that names an mxc is not followed", async () => {
|
||||
const { images } = imageSetup();
|
||||
const { hs, made } = await receive([imageEvent("$img", { url: undefined, body: MXC_URL })], { media: served(png()), images });
|
||||
try {
|
||||
assert.equal(made[0].image, null);
|
||||
assert.match(made[0].text, /refused=no-url\]/);
|
||||
assert.deepEqual(hs.downloads(), []);
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
test("image_delivery: an encrypted attachment is never fetched, whether or not it also names a url", async () => {
|
||||
const { images } = imageSetup();
|
||||
const file = { url: MXC_URL, key: { k: "SECRET_KEY_MATERIAL" }, iv: "IV", hashes: { sha256: "H" } };
|
||||
const { hs, made } = await receive([imageEvent("$sealed", { url: undefined, file }), imageEvent("$both", { file }, { origin_server_ts: 1001 })], { media: served(png()), images });
|
||||
try {
|
||||
assert.deepEqual(made.map(d => d.image), [null, null]);
|
||||
assert.ok(made.every(d => d.text.includes("refused=encrypted") && !d.text.includes("SECRET_KEY_MATERIAL")));
|
||||
assert.deepEqual(hs.downloads(), []);
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
test("image_delivery: a refused or failed fetch gives a nil image and today's line with a fixed-word reason", async () => {
|
||||
const cases = [
|
||||
["wrong-type", served(svg)],
|
||||
["fetch-failed", () => ({ status: 404, body: "" })],
|
||||
["too-big", (b, s, i) => ({ ...served(png())(b, s, i), headers: { "content-length": String(6 * 1024 * 1024) } })],
|
||||
["too-many-pixels", served(png(9000, 9000))],
|
||||
["bad-header", served(png(0, 0))],
|
||||
];
|
||||
for (const [reason, mediaFn] of cases) {
|
||||
const { images } = imageSetup();
|
||||
const { hs, made } = await receive([imageEvent("$img", { filename: "evil\nrefused=fake" })], { media: mediaFn, images });
|
||||
try {
|
||||
assert.equal(made[0].image, null, reason);
|
||||
assert.equal(made[0].text.split("\n")[0].endsWith(`refused=${reason}]`), true, `${reason}: ${made[0].text}`);
|
||||
assert.ok(made[0].text.startsWith("[image filename="));
|
||||
assert.equal(docs.deliveryTuple(made[0])[11], null);
|
||||
} finally { await hs.close(); }
|
||||
}
|
||||
});
|
||||
|
||||
test("image_delivery: with no image store the line is today's and nothing is fetched", async () => {
|
||||
const { hs, made } = await receive([imageEvent("$img", {})], { media: served(png()), images: null });
|
||||
try {
|
||||
assert.equal(made[0].image, null);
|
||||
assert.equal(made[0].text, `[image filename="cat.png" mime=image/png size=2048 url=${MXC_URL}]`);
|
||||
assert.deepEqual(hs.downloads(), []);
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
test("image_delivery: a stored image stays until its delivery is acked and it is 24 h old, and nothing but a reason word is logged", async () => {
|
||||
const bytes = gif(3, 3);
|
||||
const { root, clock, images } = imageSetup();
|
||||
const lines = [];
|
||||
const { hs, mouth, made } = await receive([imageEvent("$img", {})], { media: served(bytes), images, mouthOptions: { log: l => lines.push(l) } });
|
||||
try {
|
||||
const file = join(root, made[0].image.path);
|
||||
clock.t += RETAIN_MS * 2;
|
||||
images.sweep();
|
||||
assert.ok(existsSync(file), "a day old but unacked: kept");
|
||||
await mouth.ack(made[0].seq);
|
||||
assert.ok(!existsSync(file), "acked and a day old: gone");
|
||||
assert.deepEqual(lines.filter(l => l.includes(MXC_URL) || l.includes(root) || l.includes(sha(bytes))), []);
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
test("image_delivery: an acked image under 24 h stays", async () => {
|
||||
const { root, clock, images } = imageSetup();
|
||||
const { hs, mouth, made } = await receive([imageEvent("$img", {})], { media: served(png()), images });
|
||||
try {
|
||||
await mouth.ack(made[0].seq);
|
||||
clock.t += RETAIN_MS - 5000;
|
||||
images.sweep();
|
||||
assert.ok(existsSync(join(root, made[0].image.path)));
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
||||
test("image_delivery: a store that cannot write leaves the line and a nil image, and the sync goes on", async () => {
|
||||
const dead = { put() { throw Object.assign(new Error("no space"), { code: "ENOSPC" }); }, hold() {}, acked() {} };
|
||||
const lines = [];
|
||||
const { hs, made } = await receive([imageEvent("$img", {})], { media: served(png()), images: dead, mouthOptions: { log: l => lines.push(l) } });
|
||||
try {
|
||||
assert.equal(made[0].image, null);
|
||||
assert.match(made[0].text, /refused=store-failed\]/);
|
||||
assert.ok(lines.some(l => l.includes("ENOSPC")));
|
||||
} finally { await hs.close(); }
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,11 +1,14 @@
|
|||
// gamecult.persona.chat_*.v1 and the v2 send pair: the documents Huginn and the Matrix mouth exchange.
|
||||
// gamecult.persona.chat_*.v1, the v2 send pair and the v2 pending answer: the documents Huginn and the Matrix mouth exchange.
|
||||
// Each kind is a positional tuple (the CultCache payload shape); this module owns
|
||||
// the field order, the parsers and the envelope check. It does no I/O and loads no
|
||||
// runtime, so the wire encoder (MessagePack) stays with the caller.
|
||||
// runtime, so the wire encoder (MessagePack) stays with the caller. The image path, mime and size bounds it checks
|
||||
// come from persona-chat-images.mjs, which owns them.
|
||||
|
||||
import { IMAGE_PATH, MAX_BYTES, MIMES, extensionOf } from "./persona-chat-images.mjs";
|
||||
|
||||
export const kinds = Object.freeze({
|
||||
pendingRequest: "gamecult.persona.chat_pending_request.v1",
|
||||
pending: "gamecult.persona.chat_pending.v1",
|
||||
pending: "gamecult.persona.chat_pending.v2",
|
||||
ack: "gamecult.persona.chat_ack.v1",
|
||||
sendRequest: "gamecult.persona.chat_send_request.v1",
|
||||
sendReceipt: "gamecult.persona.chat_send_receipt.v1",
|
||||
|
|
@ -119,11 +122,31 @@ export function deliveryTuple(d) {
|
|||
seq(d.seq, "seq"), text(d.roomAlias, "room_alias"), text(d.roomId, "room_id"), text(d.eventId, "event_id"),
|
||||
text(d.sender, "sender"), text(d.senderDisplay, "sender_display"), String(d.text), seq(d.atUnixMs, "at_unix_ms"),
|
||||
d.context.slice(-maxContext).map(c => [text(c.sender, "sender"), String(c.text), seq(c.atUnixMs, "at_unix_ms")]),
|
||||
d.mentions.map(id => text(id, "mention")), flag(d.isDirect, "is_direct"),
|
||||
d.mentions.map(id => text(id, "mention")), flag(d.isDirect, "is_direct"), imageTuple(d.image ?? null),
|
||||
];
|
||||
}
|
||||
|
||||
// The image the mouth stored for a delivery, by reference: [path, mime, size, sha256] or null. The path is relative to the
|
||||
// Persona's image store and is exactly `<hex sha256 of the room id>/<hex sha256 of the bytes>[.<ext of the mime>]`, so it can
|
||||
// be neither absolute nor climb out of the store. Whether the file matches the rest is the reader's to re-check.
|
||||
function imageTuple(image) {
|
||||
if (image === null) return null;
|
||||
const { path, mime, size, sha256 } = image;
|
||||
if (typeof path !== "string" || !IMAGE_PATH.test(path)) throw new Error("image path must be <room hash>/<bytes hash>[.ext]");
|
||||
if (!MIMES.includes(mime)) throw new Error("image mime must be image/png, image/jpeg, image/gif or image/webp");
|
||||
if (path.includes(".") && !path.endsWith(`.${extensionOf(mime)}`)) throw new Error("image path extension must match its mime");
|
||||
if (!Number.isSafeInteger(size) || size < 1 || size > MAX_BYTES) throw new Error("image size must be 1 to 5 MiB");
|
||||
if (typeof sha256 !== "string" || !/^[0-9a-f]{64}$/.test(sha256)) throw new Error("image sha256 must be lowercase hex");
|
||||
if (!path.split("/")[1].startsWith(sha256)) throw new Error("image path must name its sha256");
|
||||
return [path, mime, size, sha256];
|
||||
}
|
||||
function parseImage(t) {
|
||||
const [path, mime, size, sha256] = tuple(t, 4, "image");
|
||||
imageTuple({ path, mime, size, sha256 });
|
||||
return { path, mime, size, sha256 };
|
||||
}
|
||||
export function parseDelivery(t) {
|
||||
const [s, roomAlias, roomId, eventId, sender, senderDisplay, body, atUnixMs, context, mentions, isDirect] = tuple(t, 11, "delivery");
|
||||
const [s, roomAlias, roomId, eventId, sender, senderDisplay, body, atUnixMs, context, mentions, isDirect, image] = tuple(t, 12, "delivery");
|
||||
if (!Array.isArray(context)) throw new Error("delivery context must be a list");
|
||||
if (!Array.isArray(mentions)) throw new Error("delivery mentions must be a list");
|
||||
return {
|
||||
|
|
@ -131,6 +154,7 @@ export function parseDelivery(t) {
|
|||
sender: text(sender, "sender"), senderDisplay: text(senderDisplay, "sender_display"), text: String(body), atUnixMs: seq(atUnixMs, "at_unix_ms"),
|
||||
context: context.map(c => { const [cs, ct, ca] = tuple(c, 3, "context"); return { sender: text(cs, "sender"), text: String(ct), atUnixMs: seq(ca, "at_unix_ms") }; }),
|
||||
mentions: mentions.map(id => text(id, "mention")), isDirect: flag(isDirect, "is_direct"),
|
||||
image: image === null ? null : parseImage(image),
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
228
tools/persona-chat-images.mjs
Normal file
228
tools/persona-chat-images.mjs
Normal file
|
|
@ -0,0 +1,228 @@
|
|||
// Chat images for the Matrix mouth: fetch one m.image attachment from the Persona's own homeserver, decide what it
|
||||
// is from its bytes alone, and keep it in a per-Persona store the Persona's reader opens by reference.
|
||||
//
|
||||
// Nothing in the bytes or in the event is trusted. The type is the magic bytes (the declared mime and the filename
|
||||
// are never read), the size and pixel counts are bounded before anything decodes the image, and every name on disk is
|
||||
// generated here from hashes. Refusals are a fixed word list so a refusal can be printed without printing sender text.
|
||||
import { createHash, randomBytes } from "node:crypto";
|
||||
import { closeSync, fsyncSync, mkdirSync, openSync, readdirSync, renameSync, statSync, unlinkSync, writeSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
|
||||
export const MAX_BYTES = 5 * 1024 * 1024;
|
||||
export const MAX_PIXELS = 40_000_000;
|
||||
export const ROOM_CAP_BYTES = 50 * 1024 * 1024;
|
||||
export const RETAIN_MS = 24 * 60 * 60 * 1000;
|
||||
export const FILE_MODE = 0o640;
|
||||
export const DIR_MODE = 0o750; // a room directory's mode at mkdir; its setgid bit comes from the Persona's directory, never from a chmod
|
||||
export const FETCH_TIMEOUT_MS = 30000;
|
||||
|
||||
// Why an image was not stored. Printed in the delivery's text line, so only these words ever are.
|
||||
export const refusals = Object.freeze(["encrypted", "no-url", "fetch-failed", "too-big", "wrong-type", "bad-header", "too-many-pixels", "store-failed"]);
|
||||
|
||||
const MXC = /^mxc:\/\/([A-Za-z0-9.:_-]+)\/([A-Za-z0-9_-]+)$/;
|
||||
const HEX64 = /^[0-9a-f]{64}$/;
|
||||
export const IMAGE_PATH = /^[0-9a-f]{64}\/[0-9a-f]{64}(\.[a-z0-9]{1,5})?$/;
|
||||
const EXT = { "image/png": "png", "image/jpeg": "jpg", "image/gif": "gif", "image/webp": "webp" };
|
||||
export const MIMES = Object.freeze(Object.keys(EXT));
|
||||
export const extensionOf = mime => EXT[mime];
|
||||
|
||||
const u16le = (b, o) => b[o] | (b[o + 1] << 8);
|
||||
const u16be = (b, o) => (b[o] << 8) | b[o + 1];
|
||||
const u32be = (b, o) => b.readUInt32BE(o);
|
||||
const ascii = (b, from, to) => b.toString("latin1", from, to);
|
||||
|
||||
// The type and the pixel dimensions from the header alone, never decoding. null when the bytes are none of the four
|
||||
// types; { mime, width: 0, height: 0 } when the type is known and the header will not parse.
|
||||
export function sniff(b) {
|
||||
if (b.length >= 8 && b.subarray(0, 8).equals(Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]))) {
|
||||
const ok = b.length >= 24 && ascii(b, 12, 16) === "IHDR";
|
||||
return { mime: "image/png", width: ok ? u32be(b, 16) : 0, height: ok ? u32be(b, 20) : 0 };
|
||||
}
|
||||
if (b.length >= 3 && b[0] === 0xff && b[1] === 0xd8 && b[2] === 0xff) return { mime: "image/jpeg", ...jpegSize(b) };
|
||||
if (b.length >= 6 && (ascii(b, 0, 6) === "GIF87a" || ascii(b, 0, 6) === "GIF89a")) {
|
||||
return { mime: "image/gif", width: b.length >= 10 ? u16le(b, 6) : 0, height: b.length >= 10 ? u16le(b, 8) : 0 };
|
||||
}
|
||||
if (b.length >= 12 && ascii(b, 0, 4) === "RIFF" && ascii(b, 8, 12) === "WEBP") return { mime: "image/webp", ...webpSize(b) };
|
||||
return null;
|
||||
}
|
||||
|
||||
function jpegSize(b) {
|
||||
let i = 2;
|
||||
while (i + 4 <= b.length) {
|
||||
if (b[i] !== 0xff) return { width: 0, height: 0 };
|
||||
while (b[i] === 0xff && i + 1 < b.length) i++; // fill bytes
|
||||
const marker = b[i++];
|
||||
if (marker === 0x01 || (marker >= 0xd0 && marker <= 0xd9)) continue; // no length
|
||||
if (i + 2 > b.length) break;
|
||||
const length = u16be(b, i);
|
||||
const frame = marker >= 0xc0 && marker <= 0xcf && marker !== 0xc4 && marker !== 0xc8 && marker !== 0xcc;
|
||||
if (frame) return i + 7 <= b.length ? { height: u16be(b, i + 3), width: u16be(b, i + 5) } : { width: 0, height: 0 };
|
||||
if (marker === 0xda || length < 2) break; // scan data before any frame header: unparsable
|
||||
i += length;
|
||||
}
|
||||
return { width: 0, height: 0 };
|
||||
}
|
||||
|
||||
function webpSize(b) {
|
||||
const bad = { width: 0, height: 0 };
|
||||
const chunk = ascii(b, 12, 16);
|
||||
if (chunk === "VP8 ") return b.length >= 30 && b[23] === 0x9d && b[24] === 0x01 && b[25] === 0x2a ? { width: u16le(b, 26) & 0x3fff, height: u16le(b, 28) & 0x3fff } : bad;
|
||||
if (chunk === "VP8L") {
|
||||
if (b.length < 25 || b[20] !== 0x2f) return bad;
|
||||
return { width: 1 + (b[21] | ((b[22] & 0x3f) << 8)), height: 1 + ((b[22] >> 6) | (b[23] << 2) | ((b[24] & 0x0f) << 10)) };
|
||||
}
|
||||
if (chunk === "VP8X") return b.length >= 30 ? { width: 1 + (b[24] | (b[25] << 8) | (b[26] << 16)), height: 1 + (b[27] | (b[28] << 8) | (b[29] << 16)) } : bad;
|
||||
return bad;
|
||||
}
|
||||
|
||||
// The checked type of `bytes`: { mime, ext } or { refusal }.
|
||||
export function classify(bytes) {
|
||||
const found = sniff(bytes);
|
||||
if (!found) return { refusal: "wrong-type" };
|
||||
if (!(found.width > 0 && found.height > 0)) return { refusal: "bad-header" };
|
||||
if (found.width * found.height > MAX_PIXELS) return { refusal: "too-many-pixels" };
|
||||
return { mime: found.mime, ext: EXT[found.mime] };
|
||||
}
|
||||
|
||||
// GET the attachment named by `mxc` as this Persona: its token, its homeserver, the authenticated media path. The
|
||||
// body is read stream-wise and abandoned the moment it passes the cap, whatever Content-Length said.
|
||||
// Returns { bytes, mime, ext } or { refusal }.
|
||||
export async function fetchImage({ fetch: httpFetch, homeserver, token, mxc, timeoutMs = FETCH_TIMEOUT_MS }) {
|
||||
const named = typeof mxc === "string" ? MXC.exec(mxc) : null;
|
||||
if (!named) return { refusal: "no-url" };
|
||||
const signal = AbortSignal.timeout(timeoutMs);
|
||||
try {
|
||||
const response = await httpFetch(`${homeserver}/_matrix/client/v1/media/download/${named[1]}/${named[2]}`, {
|
||||
method: "GET", headers: { authorization: `Bearer ${token}` }, redirect: "error", signal,
|
||||
});
|
||||
if (!response.ok || !response.body) return { refusal: "fetch-failed" };
|
||||
const declared = Number(response.headers.get("content-length"));
|
||||
if (response.headers.get("content-length") !== null && Number.isFinite(declared) && declared > MAX_BYTES) {
|
||||
await response.body.cancel().catch(() => {});
|
||||
return { refusal: "too-big" };
|
||||
}
|
||||
const chunks = [];
|
||||
let total = 0;
|
||||
const reader = response.body.getReader();
|
||||
for (;;) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
total += value.length;
|
||||
if (total > MAX_BYTES) {
|
||||
await reader.cancel().catch(() => {});
|
||||
return { refusal: "too-big" };
|
||||
}
|
||||
chunks.push(value);
|
||||
}
|
||||
const bytes = Buffer.concat(chunks);
|
||||
const kind = classify(bytes);
|
||||
return kind.refusal ? kind : { bytes, ...kind };
|
||||
} catch {
|
||||
return { refusal: "fetch-failed" };
|
||||
}
|
||||
}
|
||||
|
||||
const sha256 = data => createHash("sha256").update(data).digest("hex");
|
||||
|
||||
// One Persona's store: `<root>/<hex sha256 of room id>/<hex sha256 of bytes>[.ext]`. The mouth is its only writer.
|
||||
// A stored image leaves when it is at least RETAIN_MS old and the delivery that carried it is acked, whichever is
|
||||
// later; a room holds at most ROOM_CAP_BYTES, oldest evicted first. What is on disk at start is old and acked: a
|
||||
// delivery still pending after a restart is made again by the next sync and stores its image again.
|
||||
export function createImageStore({ root, now = Date.now, log = () => {}, stat = statSync }) {
|
||||
const entries = new Map(); // relative path -> { room, size, storedAt, seq } (seq: the newest delivery carrying it, null once acked)
|
||||
let loaded = false;
|
||||
|
||||
function load() {
|
||||
if (loaded) return;
|
||||
loaded = true;
|
||||
for (const room of safeList(root)) {
|
||||
if (!HEX64.test(room)) continue;
|
||||
for (const name of safeList(join(root, room))) {
|
||||
const path = `${room}/${name}`;
|
||||
if (name.startsWith(".tmp-")) { try { unlinkSync(join(root, path)); } catch { /* gone */ } continue; }
|
||||
if (!IMAGE_PATH.test(path)) continue;
|
||||
try { const s = statSync(join(root, path)); if (s.isFile()) entries.set(path, { room, size: s.size, storedAt: s.mtimeMs, seq: null }); } catch { /* gone */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function drop(path) {
|
||||
try { unlinkSync(join(root, path)); } catch { /* already gone */ }
|
||||
entries.delete(path);
|
||||
}
|
||||
|
||||
function sweep() {
|
||||
load();
|
||||
const at = now();
|
||||
for (const [path, e] of [...entries]) if (e.seq === null && at - e.storedAt >= RETAIN_MS) drop(path);
|
||||
}
|
||||
|
||||
// Store `bytes` (already classified as `mime`) for `roomId`. Returns the delivery reference.
|
||||
function put(roomId, bytes, mime) {
|
||||
load();
|
||||
const room = sha256(roomId);
|
||||
const digest = sha256(bytes);
|
||||
const ext = EXT[mime];
|
||||
const path = `${room}/${digest}.${ext}`;
|
||||
const dir = join(root, room);
|
||||
// The Persona's directory (`root`) belongs to the installer: bifrost:mind-<slug>, 2750. It is never chmodded here, and
|
||||
// neither is a room directory: bifrost is not in the Persona's group, and a chmod by a non-member clears setgid. A room
|
||||
// directory inherits group and setgid from `root` at mkdir; if it did not, a file in it would take bifrost's group and the
|
||||
// Persona's reader could not open it, so the image is refused instead.
|
||||
// No chmod can set the group bits without risking setgid, so the creation modes must be exact: a fixed umask (it only
|
||||
// withholds other-access) for the mkdir and the open, restored at once. Synchronous, so nothing else runs under it.
|
||||
const umask = process.umask(0o027);
|
||||
let temp, fd;
|
||||
try {
|
||||
// Only the room directory is created: a missing Persona directory is the installer's to make, and is not made up here.
|
||||
let persona;
|
||||
try { persona = stat(root); } catch { persona = null; }
|
||||
if (!persona?.isDirectory()) throw Object.assign(new Error("the Persona's image directory does not exist"), { code: "ENOPERSONADIR" });
|
||||
try { mkdirSync(dir, { mode: DIR_MODE }); } catch (error) { if (error.code !== "EEXIST") throw error; }
|
||||
const made = stat(dir);
|
||||
if ((made.mode & 0o2000) === 0) throw Object.assign(new Error("image directory lacks the setgid bit"), { code: "ENOSETGID" });
|
||||
if (made.gid !== persona.gid) throw Object.assign(new Error("image directory is not in the Persona directory's group"), { code: "ENOGROUP" });
|
||||
temp = join(dir, `.tmp-${randomBytes(8).toString("hex")}`);
|
||||
fd = openSync(temp, "wx", FILE_MODE);
|
||||
} finally { process.umask(umask); }
|
||||
try {
|
||||
writeSync(fd, bytes);
|
||||
fsyncSync(fd);
|
||||
} finally { closeSync(fd); }
|
||||
try { renameSync(temp, join(dir, `${digest}.${ext}`)); } catch (error) { try { unlinkSync(temp); } catch { /* gone */ } throw error; }
|
||||
entries.set(path, { room, size: bytes.length, storedAt: now(), seq: entries.get(path)?.seq ?? null });
|
||||
evict(room, path);
|
||||
sweep();
|
||||
return { path, mime, size: bytes.length, sha256: digest };
|
||||
}
|
||||
|
||||
function evict(room, keep) {
|
||||
const here = [...entries].filter(([, e]) => e.room === room);
|
||||
let total = here.reduce((n, [, e]) => n + e.size, 0);
|
||||
for (const [path, e] of here.sort((a, b) => a[1].storedAt - b[1].storedAt)) {
|
||||
if (total <= ROOM_CAP_BYTES) break;
|
||||
if (path === keep) continue;
|
||||
drop(path);
|
||||
total -= e.size;
|
||||
}
|
||||
}
|
||||
|
||||
// The delivery numbered `seq` carries the image at `path`: it is kept at least until that delivery is acked.
|
||||
function hold(path, seq) {
|
||||
const e = entries.get(path);
|
||||
if (e) e.seq = Math.max(e.seq ?? 0, seq);
|
||||
}
|
||||
|
||||
// Deliveries through `throughSeq` are acked.
|
||||
function acked(throughSeq) {
|
||||
load();
|
||||
for (const e of entries.values()) if (e.seq !== null && e.seq <= throughSeq) e.seq = null;
|
||||
sweep();
|
||||
}
|
||||
|
||||
return { put, hold, acked, sweep, get size() { load(); return entries.size; } };
|
||||
}
|
||||
|
||||
function safeList(dir) {
|
||||
try { return readdirSync(dir); } catch { return []; }
|
||||
}
|
||||
|
|
@ -13,7 +13,9 @@
|
|||
// reads the room's head from Matrix and answers RoomMoved, naming the messages that moved it, instead
|
||||
// of the PUT. A v1 send, and a v2 send with no basis, are not checked; both still pass through the gate.
|
||||
//
|
||||
// Media: an m.image/m.file/m.audio/m.video event is delivered as one typed line (see bodyText), not its bare filename. Nothing is downloaded.
|
||||
// Media: an m.image/m.file/m.audio/m.video event is delivered as one typed line (see bodyText), not its bare filename.
|
||||
// An m.image event in a delivered message is also fetched, as this Persona and only when an image store is configured
|
||||
// (persona-chat-images.mjs): the delivery then carries a reference to the stored file, or the line says why not.
|
||||
//
|
||||
// Context: Matrix is the record. A delivery's `context` is read from the homeserver when the delivery
|
||||
// is built (GET /messages, or the thread's relations for a threaded message), as this Persona's
|
||||
|
|
@ -34,6 +36,7 @@ import { createSocket } from "node:dgram";
|
|||
import { resolve, join } from "node:path";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import * as docs from "./persona-chat-documents.mjs";
|
||||
import { createImageStore, fetchImage } from "./persona-chat-images.mjs";
|
||||
|
||||
const API = "/_matrix/client/v3";
|
||||
const BRAKE_TYPE = "eureka.control.brake.v1";
|
||||
|
|
@ -57,12 +60,13 @@ export const movesRoom = (event, selfId) => isLine(event) && event.sender !== se
|
|||
// [image filename="cat.png" mime=image/png size=2048 url=mxc://gamecult.test/abc]
|
||||
// A Persona turn is a tool-less text call (it has no way to open a file or see an image), so the mouth names the
|
||||
// attachment and fetches nothing. Every field is checked before it is printed; a field that fails is left out.
|
||||
// An encrypted attachment names its `file.url` and carries `encrypted`; its key is never read.
|
||||
// An encrypted attachment names its `file.url` and carries `encrypted`; its key is never read. `refused` is the
|
||||
// fixed-word reason an image was not stored (persona-chat-images `refusals`); it is never sender text.
|
||||
const MEDIA_KIND = { "m.image": "image", "m.file": "file", "m.audio": "audio", "m.video": "video" };
|
||||
const MXC = /^mxc:\/\/[A-Za-z0-9.:_-]+\/[A-Za-z0-9_-]+$/;
|
||||
const MIME = /^[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]{0,126}\/[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]{0,126}$/;
|
||||
const FILENAME_BYTES = 255;
|
||||
export function bodyText(content, maxBytes) {
|
||||
export function bodyText(content, maxBytes, refused = null) {
|
||||
const kind = MEDIA_KIND[content?.msgtype];
|
||||
if (!kind) return docs.clip(content.body, maxBytes);
|
||||
const encrypted = typeof content.file?.url === "string" && typeof content.url !== "string";
|
||||
|
|
@ -74,6 +78,7 @@ export function bodyText(content, maxBytes) {
|
|||
if (Number.isSafeInteger(content.info?.size) && content.info.size >= 0) parts.push(`size=${content.info.size}`);
|
||||
if (typeof url === "string" && MXC.test(url)) parts.push(`url=${url}`);
|
||||
if (encrypted) parts.push("encrypted");
|
||||
if (refused !== null) parts.push(`refused=${refused}`);
|
||||
const line = docs.clip(`[${parts.join(" ")}]`, maxBytes);
|
||||
const caption = named && content.body !== filename ? content.body : "";
|
||||
return caption === "" ? line : docs.clip(`${line}\n${caption}`, maxBytes);
|
||||
|
|
@ -97,6 +102,7 @@ export function createRoomGate() {
|
|||
export function createMouth({
|
||||
persona, homeserver, serverName, rooms, token, fetch: httpFetch = globalThis.fetch, state, brake,
|
||||
now = Date.now, log = () => {}, syncTimeoutMs = 30000, maxPending = MAX_PENDING, gate = createRoomGate(), sendTimeoutMs = SEND_TIMEOUT_MS,
|
||||
images = null, // this Persona's image store (createImageStore), or null: images stay the typed text line
|
||||
}) {
|
||||
if (!persona || !homeserver || !token || !rooms?.length) throw new Error("persona, homeserver, rooms and token are required");
|
||||
const say = message => log(redact(String(message), token));
|
||||
|
|
@ -257,6 +263,20 @@ export function createMouth({
|
|||
return lines;
|
||||
}
|
||||
|
||||
// The image of one delivered m.image event: { image } (a stored reference) or { refused } (a reason word). With no
|
||||
// store nothing is attempted and nothing is said. The url is the event's own content.url; encrypted events and
|
||||
// anything the checks refuse leave only the text line.
|
||||
async function imageOf(roomId, content) {
|
||||
if (!images || content?.msgtype !== "m.image") return {};
|
||||
if (content.file !== undefined) return { refused: "encrypted" };
|
||||
const got = await fetchImage({ fetch: httpFetch, homeserver, token, mxc: content.url });
|
||||
if (got.refusal) return { refused: got.refusal };
|
||||
try { return { image: images.put(roomId, got.bytes, got.mime) }; } catch (error) {
|
||||
say(`image not stored: ${error?.code ?? "write failed"}`);
|
||||
return { refused: "store-failed" };
|
||||
}
|
||||
}
|
||||
|
||||
// One /sync round. Returns the deliveries it made.
|
||||
async function syncOnce() {
|
||||
if (!started || unjoined()) await start();
|
||||
|
|
@ -283,20 +303,23 @@ export function createMouth({
|
|||
if (typeof event.sender !== "string" || event.sender === "" || typeof event.event_id !== "string" || event.event_id === "") continue;
|
||||
if (processed.has(event.event_id)) continue; // read already in a round the cap kept from advancing
|
||||
const ts = event.origin_server_ts;
|
||||
const entry = { sender: event.sender, text: bodyText(event.content, docs.limits.textBytes), atUnixMs: Number.isSafeInteger(ts) && ts >= 0 ? ts : now() };
|
||||
const atUnixMs = Number.isSafeInteger(ts) && ts >= 0 ? ts : now();
|
||||
const deliverable = !baseline && !freshRooms.has(roomId) && event.sender !== selfId;
|
||||
if (deliverable && pending.length + made.length >= maxPending) { truncated = true; break rounds; }
|
||||
processed.add(event.event_id);
|
||||
if (deliverable) {
|
||||
const context = await contextFor(roomId, event, round);
|
||||
const { image = null, refused = null } = await imageOf(roomId, event.content);
|
||||
if (refused !== null) say(`image refused: ${refused}`);
|
||||
const mentioned = event.content["m.mentions"]?.user_ids;
|
||||
made.push({
|
||||
seq: ++seqCounter, roomAlias: alias, roomId, eventId: event.event_id, sender: event.sender,
|
||||
senderDisplay: displayNames.get(event.sender) ?? localpart(event.sender), text: entry.text, atUnixMs: entry.atUnixMs,
|
||||
context,
|
||||
senderDisplay: displayNames.get(event.sender) ?? localpart(event.sender), text: bodyText(event.content, docs.limits.textBytes, refused), atUnixMs,
|
||||
context, image,
|
||||
mentions: Array.isArray(mentioned) ? mentioned.filter(id => typeof id === "string" && id !== "" && id.length <= 255).slice(0, docs.limits.mentions) : [],
|
||||
isDirect: directRooms.has(roomId),
|
||||
});
|
||||
if (image) images.hold(image.path, seqCounter);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -325,11 +348,12 @@ export function createMouth({
|
|||
|
||||
async function ack(throughSeq) {
|
||||
pending = pending.filter(d => d.seq > throughSeq);
|
||||
images?.acked(throughSeq);
|
||||
await commit(pending.length ? pending[0].seq - 1 : seqCounter);
|
||||
}
|
||||
|
||||
function pendingAfter(afterSeq) {
|
||||
return pending.filter(d => d.seq > afterSeq).map(d => ({ ...d, context: d.context.map(c => ({ ...c })), mentions: [...d.mentions] }));
|
||||
return pending.filter(d => d.seq > afterSeq).map(d => ({ ...d, context: d.context.map(c => ({ ...c })), mentions: [...d.mentions], image: d.image && { ...d.image } }));
|
||||
}
|
||||
|
||||
// The ids of the messages that moved the room since `basis`, oldest first. Walks the head newest to oldest
|
||||
|
|
@ -547,7 +571,7 @@ export function roomsFor(env, persona) {
|
|||
|
||||
// PERSONAS is the comma-separated list of Persona slugs this process speaks for.
|
||||
// Each slug names its credential (matrix-token-<slug>), its cursor directory under
|
||||
// STATE_DIRECTORY and its brake at CONTROL_ROOT/<slug>/control.cc.
|
||||
// STATE_DIRECTORY, its image store at IMAGE_STORE_ROOT/<slug> (optional) and its brake at CONTROL_ROOT/<slug>/control.cc.
|
||||
export async function main(env = process.env) {
|
||||
const personas = required(env, "PERSONAS").split(",").map(s => s.trim()).filter(Boolean);
|
||||
if (personas.length === 0 || new Set(personas).size !== personas.length || !personas.every(p => SLUG.test(p))) throw new Error("PERSONAS must be distinct slugs");
|
||||
|
|
@ -563,11 +587,13 @@ export async function main(env = process.env) {
|
|||
const roomsByPersona = new Map(personas.map(persona => [persona, roomsFor(env, persona)]));
|
||||
const stateRoot = required(env, "STATE_DIRECTORY");
|
||||
const controlRoot = env.CONTROL_ROOT ?? "/etc/gamecult/minds";
|
||||
const imageRoot = env.IMAGE_STORE_ROOT ? resolve(env.IMAGE_STORE_ROOT) : null; // unset: chat images stay the typed text line
|
||||
const gate = createRoomGate(); // one for every Persona: two Minds in one room take turns
|
||||
const mouths = new Map(personas.map(persona => {
|
||||
mkdirSync(join(stateRoot, persona), { recursive: true });
|
||||
return [persona, createMouth({
|
||||
persona, homeserver, serverName, rooms: roomsByPersona.get(persona), token: tokens.get(persona), gate, log: line => log(`${persona}: ${line}`),
|
||||
images: imageRoot ? createImageStore({ root: join(imageRoot, persona), log: line => log(`${persona}: ${line}`) }) : null,
|
||||
state: cultcacheState({ runtime, path: join(stateRoot, persona, "cursor.cc") }),
|
||||
brake: createBrakeReader({ controlPath: join(controlRoot, persona, "control.cc"), readStore: runtime.readSingleFileStore, decode: runtime.msgpack.decode }),
|
||||
})];
|
||||
|
|
|
|||
Loading…
Reference in a new issue