Skip to content

feat(gateways): store inbound media as HTTP reference, not inline RDF #127

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
retog wants to merge 1 commit into
base: main
Choose a base branch
from
+448 −90
Open

feat(gateways): store inbound media as HTTP reference, not inline RDF #127

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
96 changes: 94 additions & 2 deletions scripts/inbound_store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,12 +57,32 @@
P_TEXT = KB + "text"
P_MESSAGE_ID = KB + "messageId"
P_DELIVERED = KB + "delivered"
# A message's media (voice note, image) is NOT embedded in the graph: consistency
# over data-in-graph means every attachment — regardless of size — is a *reference*
# resolved over HTTP, never an inline data-URI literal. This predicate carries that
# reference as an IRI object (the gateway's own token-gated GET /media/<id> URL);
# it is multi-valued, so a message with several images gets one triple each. The
# attachment's media type is not stored here on purpose — it is returned by the
# HTTP response's Content-Type header when the reference is resolved, which is
# where a media type belongs once the payload lives behind a URL.
P_ATTACHMENT = KB + "attachment"

# Subdirectory (under the gateway's store dir) that holds the per-message files.
# The gateway owns this folder read-write; the life store mounts it read-only.
MESSAGES_SUBDIR = "messages"

# Subdirectory holding the durable media blobs referenced by P_ATTACHMENT. Blobs
# are named by a server-generated hex id (never an untrusted filename) with no
# RDF extension, so qlever-dir — which indexes only .nt/.ttl/.n3 plus declared
# converters — ignores them: the binaries sit on the same volume as the message
# .nt files without ever entering the triple store.
MEDIA_SUBDIR = "media"

_SLUG_RE = re.compile(r"[^a-z0-9]+")
# A stored media id is exactly token_hex(16) — 32 lowercase hex chars. Validating
# against this on read makes the GET /media/<id> path traversal-safe: an id that
# is not pure hex can never resolve to a file outside the media dir.
_MEDIA_ID_RE = re.compile(r"^[0-9a-f]{32}$")


def _slug(value: str) -> str:
Expand All @@ -74,6 +94,10 @@ def messages_dir(store_dir: str | Path) -> Path:
return Path(store_dir) / MESSAGES_SUBDIR


def media_dir(store_dir: str | Path) -> Path:
return Path(store_dir) / MEDIA_SUBDIR


# -- N-Triples serialization --------------------------------------------------
# A tiny, self-contained N-Triples reader/writer. It supports exactly the three
# object shapes this store uses: an IRI object (rdf:type), a plain string
Expand Down Expand Up @@ -138,12 +162,18 @@ def _render(fields: dict) -> str:
lines.append(_lit(subj, P_GROUP, fields["group"]))
if fields.get("message_id"):
lines.append(_lit(subj, P_MESSAGE_ID, fields["message_id"]))
# Multi-valued: one IRI triple per attachment reference (deduped, order-free
# since the whole record is sorted before serialization).
for url in dict.fromkeys(fields.get("attachments") or []):
if url:
lines.append(_iri(subj, P_ATTACHMENT, url))
return "".join(l + "\n" for l in sorted(lines))


def _parse(text: str) -> dict | None:
"""Read a message file back into a ``fields`` dict, or None if unparseable."""
fields: dict = {"delivered": False, "group": None, "message_id": None}
fields: dict = {"delivered": False, "group": None, "message_id": None,
"attachments": []}
subject = None
for line in text.splitlines():
line = line.strip()
Expand All @@ -169,6 +199,10 @@ def _parse(text: str) -> dict | None:
fields["text"] = value
elif pred == P_MESSAGE_ID:
fields["message_id"] = value
elif pred == P_ATTACHMENT:
# IRI object → obj_iri is set; append the reference URL.
if value:
fields["attachments"].append(value)
elif pred == P_DELIVERED:
fields["delivered"] = value.strip().lower() == "true"
if subject is None or "channel" not in fields:
Expand Down Expand Up @@ -238,13 +272,18 @@ def write_message(
message_id: str | None = None,
timestamp: float | None = None,
delivered: bool = False,
attachment_urls: list[str] | None = None,
) -> tuple[str, Path]:
"""Persist one inbound message as a deterministic N-Triples file.

Returns ``(subject_uri, path)``. ``delivered=False`` (the default) marks the
message as still owed to triage; pass ``delivered=True`` for a message the
gateway is deliberately *not* forwarding (blacklisted, group-blocked or
no-action-class) so the daily drain never re-surfaces it.

``attachment_urls`` are HTTP-resolvable references to this message's media
(voice note, image), each emitted as a ``kb:attachment`` IRI. The bytes are
never inlined into the graph — see :func:`store_media`.
"""
ts = time.time() if timestamp is None else float(timestamp)
token = secrets.token_hex(8)
Expand All @@ -258,6 +297,7 @@ def write_message(
"message_id": message_id or None,
"received_at": _iso(ts),
"delivered": bool(delivered),
"attachments": [u for u in (attachment_urls or []) if u],
}
# Filename: zero-padded epoch millis (sortable) + token (unique, IRI-safe).
fname = f"{int(ts * 1000):016d}-{token}.nt"
Expand All @@ -266,6 +306,56 @@ def write_message(
return subject, path


def store_media(store_dir: str | Path, data: bytes, content_type: str | None) -> str:
"""Persist one inbound media blob durably and return its server-generated id.

The blob is keyed by ``token_hex(16)`` — never by an untrusted filename — so
the id is path-safe by construction and reveals nothing about the sender. The
``content_type`` is written to a ``<id>.type`` sidecar so the serving endpoint
can set the right ``Content-Type`` without trusting anything client-supplied.
Neither file carries an RDF extension, so the life store never indexes them.

The caller builds the HTTP reference (``…/media/<id>``) and passes it to
:func:`write_message` as an ``attachment_urls`` entry; the bytes stay on disk
and out of the graph.
"""
media_id = secrets.token_hex(16)
d = media_dir(store_dir)
d.mkdir(parents=True, exist_ok=True)
blob = d / media_id
tmp = blob.with_suffix(".tmp")
tmp.write_bytes(data or b"")
os.replace(tmp, blob)
ct = (content_type or "application/octet-stream").strip() or "application/octet-stream"
_atomic_write(ct + "\n", d / (media_id + ".type"))
return media_id


def load_media(store_dir: str | Path, media_id: str) -> tuple[bytes, str] | None:
"""Return ``(bytes, content_type)`` for a stored media id, or None.

Validates the id against :data:`_MEDIA_ID_RE` before touching the filesystem,
so a crafted ``media_id`` can never escape the media dir (path traversal). The
content type falls back to ``application/octet-stream`` if the sidecar is
missing or unreadable.
"""
if not _MEDIA_ID_RE.match(media_id or ""):
return None
d = media_dir(store_dir)
try:
data = (d / media_id).read_bytes()
except OSError:
return None
ct = "application/octet-stream"
try:
sidecar = (d / (media_id + ".type")).read_text(encoding="utf-8").strip()
if sidecar:
ct = sidecar
except OSError:
pass
return data, ct


def undelivered(
store_dir: str | Path,
since: str | float | None = None,
Expand All @@ -280,7 +370,8 @@ def undelivered(
anything; only the daily triage drain does.

Each returned dict has: ``subject``, ``channel``, ``sender``, ``group``,
``message_id``, ``received_at`` (ISO-8601), ``text``.
``message_id``, ``received_at`` (ISO-8601), ``text``, ``attachments`` (a
possibly-empty list of HTTP media reference URLs).
"""
mdir = messages_dir(store_dir)
if not mdir.is_dir():
Expand Down Expand Up @@ -311,5 +402,6 @@ def undelivered(
"message_id": fields.get("message_id"),
"received_at": fields["received_at"],
"text": fields["text"],
"attachments": fields.get("attachments") or [],
})
return out
120 changes: 99 additions & 21 deletions scripts/signal-gateway.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,13 +201,23 @@ def _inbound_gate_decision(sender: str, group_id: str | None) -> dict:
return {"forward": True, "flagged_unknown": False, "delivered_if_held": True, "reason": "policy-error"}


# This gateway's own base URL on the internal Docker network, used to build the
# HTTP references stored for inbound media (GET /media/<id> below). Defaults to
# the compose service name + HTTP port; a deployment running more than one Signal
# identity (e.g. the user's personal account) overrides it per container.
GATEWAY_SELF_URL = os.environ.get(
"SIGNAL_GATEWAY_SELF_URL", f"http://signal-gateway:{HTTP_PORT}"
).rstrip("/")


def _persist_inbound(question: str, sender: str, group_id: str | None,
delivered: bool) -> None:
delivered: bool, attachment_urls: list[str] | None = None) -> None:
"""Best-effort persist of one inbound message to the store; never raises."""
try:
_ibstore.write_message(
INBOUND_STORE_DIR, channel=INBOUND_CHANNEL, sender=sender or "unknown",
text=question, group=group_id or None, delivered=delivered,
attachment_urls=attachment_urls or None,
)
except Exception as exc:
print(f"[signal-gateway] could not persist inbound message: {exc}", flush=True)
Expand All @@ -223,6 +233,23 @@ def _forward_news(question: str, source: str, group_id: str | None, lang: str) -
print(f"[signal-gateway] forwarded news-flagged message from {source}", flush=True)


def _store_media_ref(data: bytes, content_type: str | None) -> str | None:
"""Persist inbound media durably and return its HTTP-resolvable reference URL.

Best-effort: any failure returns None so the message still forwards and
persists with its transcript — only the media link is skipped. The bytes go
to disk (out of the graph); the returned URL is what lands in the message's
``kb:attachment`` triple."""
if not data:
return None
try:
media_id = _ibstore.store_media(INBOUND_STORE_DIR, data, content_type)
except Exception as exc:
print(f"[signal-gateway] could not store inbound media: {exc}", flush=True)
return None
return f"{GATEWAY_SELF_URL}/media/{media_id}"


# Outbound send-control policy — the messenger analogue of EMAIL_SEND_POLICY.
# Keyed by the *sending* account number (this gateway's own SIGNAL_ACCOUNT), NOT
# the recipient: the category is resolved for the identity a message goes out as,
Expand Down Expand Up @@ -385,19 +412,27 @@ def _attachment_path(att: dict) -> Path | None:
return None


def _split_attachments(event: dict) -> tuple[Path | None, list[dict]]:
"""Partition inbound attachments into (voice_note_path, image_files).
def _split_attachments(event: dict) -> tuple[Path | None, list[dict], list[str]]:
"""Partition inbound attachments into (voice_note_path, files, attachment_urls).

signal-cli labels each attachment with its contentType: ``audio/*`` is a
voice note to transcribe, ``image/*`` is forwarded to the agent as a file
payload (``{"filename", "content_type", "data"(base64)}`` — the shape the
retinue gateway's POST /message accepts as ``files``). An attachment with
no contentType keeps the legacy voice-note treatment, since before this
split every attachment was handed to the transcriber."""
split every attachment was handed to the transcriber.

Every attachment (image or voice note) is ALSO persisted durably and its
HTTP reference collected in ``attachment_urls`` — the ``kb:attachment`` triple
on the stored message. The voice note is additionally added to ``files`` so
the original audio rides into the conversation alongside its transcript. The
durable reference is stored regardless of size (consistency: a reference,
never inline); only the transient ``files`` payload honours the size cap."""
msg = event.get("envelope", {}).get("dataMessage") or {}
attachments = msg.get("attachments") or []
voice: Path | None = None
images: list[dict] = []
files: list[dict] = []
attachment_urls: list[str] = []
for att in attachments:
if not isinstance(att, dict):
continue
Expand All @@ -406,26 +441,46 @@ def _split_attachments(event: dict) -> tuple[Path | None, list[dict]]:
print(f"[signal-gateway] attachment metadata present but file not found: {att}", flush=True)
continue
content_type = str(att.get("contentType") or "").lower()
try:
data = path.read_bytes()
except OSError as exc:
print(f"[signal-gateway] could not read inbound attachment {path}: {exc}", flush=True)
data = b""
if content_type.startswith("image/"):
try:
data = path.read_bytes()
except OSError as exc:
print(f"[signal-gateway] could not read inbound image {path}: {exc}", flush=True)
continue
if not data:
continue
ref = _store_media_ref(data, content_type)
if ref:
attachment_urls.append(ref)
if len(data) > MAX_INBOUND_FILE_BYTES:
print(f"[signal-gateway] inbound image too large to forward ({len(data)} bytes)", flush=True)
continue
suffix = path.suffix or mimetypes.guess_extension(content_type) or ".jpg"
images.append({
files.append({
"filename": f"signal-image{suffix}",
"content_type": content_type,
"data": base64.b64encode(data).decode("ascii"),
})
elif voice is None:
voice = path
return voice, images
else:
# Voice note (audio/*) or an unlabeled attachment: transcribe the
# first one. Persist it durably as a reference, and — when it fits —
# attach the audio itself so the conversation carries it.
mime = content_type or "audio/ogg"
if data:
ref = _store_media_ref(data, mime)
if ref:
attachment_urls.append(ref)
if len(data) <= MAX_INBOUND_FILE_BYTES:
suffix = path.suffix or mimetypes.guess_extension(mime) or ".ogg"
label = "voice" if mime.startswith("audio/") else "attachment"
files.append({
"filename": f"signal-{label}{suffix}",
"content_type": mime,
"data": base64.b64encode(data).decode("ascii"),
})
if voice is None:
voice = path
return voice, files, attachment_urls


def _extract_sender(event: dict) -> str | None:
Expand Down Expand Up @@ -1053,7 +1108,7 @@ def _handle_event(event: dict) -> None:
except Exception as exc:
print(f"[signal-gateway] could not record recent sender: {exc}", flush=True)

voice, files = _split_attachments(event)
voice, files, attachment_urls = _split_attachments(event)
if voice is not None:
print(f"[signal-gateway] processing voice message from {sender}", flush=True)
question, lang = _transcribe(voice)
Expand All @@ -1079,7 +1134,7 @@ def _handle_event(event: dict) -> None:
# account hands it to the user's triage and stays silent towards the sender.
if SIGNAL_GATEWAY_MODE == "inbox":
_forward_to_inbox(question, lang, sender, group_id=_extract_group_id(event),
files=files)
files=files, attachment_urls=attachment_urls)
else:
_handle_control_message(question, lang, sender, files=files)

Expand Down Expand Up @@ -1122,7 +1177,8 @@ def _handle_control_message(question: str, lang: str, sender: str,

def _forward_to_inbox(question: str, lang: str, sender: str,
group_id: str | None = None,
files: list[dict] | None = None) -> None:
files: list[dict] | None = None,
attachment_urls: list[str] | None = None) -> None:
"""Hand an inbox-account message to the user's triage, notifying the user.

The account is one of the user's own message sources, so the message is the
Expand Down Expand Up @@ -1150,7 +1206,8 @@ def _forward_to_inbox(question: str, lang: str, sender: str,
if gate.get("news"):
_forward_news(question, group_id if is_group else sender, group_id, lang)
if not gate["forward"]:
_persist_inbound(question, sender, group_id, delivered=gate["delivered_if_held"])
_persist_inbound(question, sender, group_id, delivered=gate["delivered_if_held"],
attachment_urls=attachment_urls)
print(
f"[signal-gateway] gate held inbox message from {sender_label} "
f"({gate['reason']}); no model turn",
Expand Down Expand Up @@ -1188,8 +1245,11 @@ def _forward_to_inbox(question: str, lang: str, sender: str,
if gate["flagged_unknown"] else ""
)
attachment_line = (
(f"\nThe message includes {len(files)} attached image(s), forwarded "
f"with this prompt; their saved on-disk paths are listed at the end.\n")
(f"\nThe message includes {len(files)} attached file(s) (image(s) and/or "
f"the original voice note), forwarded with this prompt; their saved "
f"on-disk paths are listed at the end. When a voice note is attached, "
f"include the audio itself in the dashboard conversation (not only its "
f"transcript).\n")
if files else ""
)
prompt = (
Expand Down Expand Up @@ -1234,7 +1294,8 @@ def _forward_to_inbox(question: str, lang: str, sender: str,
# Persist AFTER forwarding so the delivered flag reflects reality: a message
# handed to triage is delivered; one whose forward failed stays undelivered
# and the daily drain retries it.
_persist_inbound(question, sender, group_id, delivered=forwarded)
_persist_inbound(question, sender, group_id, delivered=forwarded,
attachment_urls=attachment_urls)



Expand Down Expand Up @@ -1673,6 +1734,23 @@ def do_GET(self):
else:
self._reply(status, body)
return
if self.path.split("?", 1)[0].rstrip("/").startswith("/media/"):
# Resolve a durable inbound-media reference (kb:attachment). The bytes
# live on the store volume, out of the graph; this serves them back
# over HTTP. Token-gated like /qr — it is the user's private inbound
# content. load_media validates the id, so a crafted path cannot
# escape the media dir.
if not self._authorized():
self._reply(401, {"error": "unauthorized"})
return
media_id = self.path.split("?", 1)[0].rstrip("/")[len("/media/"):]
loaded = _ibstore.load_media(INBOUND_STORE_DIR, media_id)
if loaded is None:
self._reply(404, {"error": "not found"})
return
data, content_type = loaded
self._reply_raw(200, data, content_type)
return
if self.path.rstrip("/") == "/pending-sends":
if not self._authorized():
self._reply(401, {"error": "unauthorized"})
Expand Down
Loading
Loading