438 lines
18 KiB
Python
438 lines
18 KiB
Python
"""Pheby attachment store — adapter-owned copies of Hermes deliverables.
|
|
|
|
Design (spec: "Generated attachments / Deliverable Mode"):
|
|
|
|
* The agent produces files via Hermes's normal deliverable pipeline. The
|
|
adapter intercepts ``send_document`` / ``send_image_file`` / ``send_voice``
|
|
/ ``send_video`` and *copies* the source file into adapter-owned storage,
|
|
registering an opaque 32-hex attachment ID.
|
|
* The client only ever sees attachment IDs — never server paths. Downloads
|
|
resolve ID → registered file inside the storage root; path traversal and
|
|
arbitrary filesystem reads are impossible by construction.
|
|
* Metadata (JSON, one file per attachment) persists across restarts so valid
|
|
attachments survive a Hermes/Pheby restart.
|
|
* Cleanup deletes only files this store owns (inside its own storage root,
|
|
matched by registered IDs) after the retention window (default 7 days).
|
|
Hermes-owned originals elsewhere on disk are never touched.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import mimetypes
|
|
import os
|
|
import shutil
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from . import protocol
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
IMAGE_MIME_PREFIXES = ("image/",)
|
|
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".heic"}
|
|
|
|
# Extensions Hermes's deliverable system treats as audio/video (used to pick
|
|
# a sensible kind for inline preview decisions).
|
|
AUDIO_EXTS = {".ogg", ".opus", ".mp3", ".wav", ".m4a", ".flac"}
|
|
VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm"}
|
|
|
|
|
|
def guess_mime(filename: str, fallback: str = "application/octet-stream") -> str:
|
|
"""Best-effort MIME type for a filename (stdlib mimetypes + extras)."""
|
|
ext = Path(filename).suffix.lower()
|
|
explicit = {
|
|
".md": "text/markdown", ".yml": "application/yaml",
|
|
".yaml": "application/yaml", ".toml": "application/toml",
|
|
".log": "text/plain", ".apk": "application/vnd.android.package-archive",
|
|
".ogg": "audio/ogg", ".opus": "audio/opus",
|
|
}
|
|
if ext in explicit:
|
|
return explicit[ext]
|
|
guessed, _ = mimetypes.guess_type(filename)
|
|
return guessed or fallback
|
|
|
|
|
|
class AttachmentStore:
|
|
"""Owns adapter-managed attachment copies and their metadata."""
|
|
|
|
def __init__(self, root: Path, retention_days: int = 7,
|
|
index_path: Optional[Path] = None):
|
|
self._root = Path(root).resolve()
|
|
self._retention_days = max(0, int(retention_days))
|
|
self._index_path = (
|
|
Path(index_path) if index_path else self._root / "attachments.json"
|
|
)
|
|
self._lock = asyncio.Lock()
|
|
self._meta: Dict[str, Dict[str, Any]] = {}
|
|
self._loaded = False
|
|
|
|
# ── paths ────────────────────────────────────────────────────────────
|
|
@property
|
|
def root(self) -> Path:
|
|
return self._root
|
|
|
|
def _blob_path(self, attachment_id: str, filename: str) -> Path:
|
|
"""Blob location: ``blobs/<aa>/<id>__<sanitized-filename>``."""
|
|
safe_name = self._sanitize_filename(filename)
|
|
return self._root / "blobs" / attachment_id[:2] / f"{attachment_id}__{safe_name}"
|
|
|
|
def _meta_path(self, attachment_id: str) -> Path:
|
|
return self._root / "meta" / f"{attachment_id}.json"
|
|
|
|
@staticmethod
|
|
def _sanitize_filename(filename: str) -> str:
|
|
"""Strip path separators/control chars from a stored filename."""
|
|
name = os.path.basename(str(filename or "").replace("\\", "/")).strip()
|
|
name = "".join(c for c in name if c.isprintable() and c not in '/\\')
|
|
return name[:120] or "file.bin"
|
|
|
|
# ── persistence ──────────────────────────────────────────────────────
|
|
def _load_index(self) -> None:
|
|
if self._loaded:
|
|
return
|
|
self._loaded = True
|
|
try:
|
|
if self._index_path.exists():
|
|
data = json.loads(self._index_path.read_text(encoding="utf-8"))
|
|
if isinstance(data, dict):
|
|
self._meta = {
|
|
k: v for k, v in data.items()
|
|
if isinstance(k, str) and isinstance(v, dict)
|
|
and protocol.is_valid_attachment_id(k)
|
|
}
|
|
except Exception:
|
|
logger.warning("[pheby] attachment index unreadable; starting empty",
|
|
exc_info=True)
|
|
|
|
def _save_index(self) -> None:
|
|
try:
|
|
self._index_path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = self._index_path.with_suffix(".tmp")
|
|
tmp.write_text(
|
|
json.dumps(self._meta, ensure_ascii=False, indent=1),
|
|
encoding="utf-8")
|
|
os.replace(tmp, self._index_path)
|
|
except Exception:
|
|
logger.error("[pheby] failed to persist attachment index",
|
|
exc_info=True)
|
|
|
|
# ── registration ─────────────────────────────────────────────────────
|
|
async def register_file(
|
|
self,
|
|
source_path: str,
|
|
*,
|
|
conversation_id: str,
|
|
message_id: Optional[str] = None,
|
|
filename: Optional[str] = None,
|
|
kind_hint: Optional[str] = None,
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Copy *source_path* into adapter storage and register metadata.
|
|
|
|
Returns the attachment descriptor dict, or ``None`` when the source
|
|
is missing/unsafe. The original file is never modified or deleted.
|
|
"""
|
|
try:
|
|
src = Path(source_path).expanduser().resolve(strict=True)
|
|
except (OSError, RuntimeError, ValueError):
|
|
logger.warning("[pheby] deliverable not found: %s",
|
|
protocol.safe_str(source_path, 120))
|
|
return None
|
|
if not src.is_file():
|
|
return None
|
|
|
|
fname = self._sanitize_filename(filename or src.name)
|
|
attachment_id = protocol.new_id()
|
|
mime = guess_mime(fname)
|
|
is_image = mime.startswith(IMAGE_MIME_PREFIXES) or (
|
|
Path(fname).suffix.lower() in IMAGE_EXTS)
|
|
if kind_hint == "voice":
|
|
kind = "voice"
|
|
elif kind_hint == "video" or Path(fname).suffix.lower() in VIDEO_EXTS:
|
|
kind = "video"
|
|
elif kind_hint == "audio" or Path(fname).suffix.lower() in AUDIO_EXTS:
|
|
kind = "audio"
|
|
elif is_image:
|
|
kind = "image"
|
|
else:
|
|
kind = "document"
|
|
|
|
try:
|
|
size = src.stat().st_size
|
|
async with self._lock:
|
|
self._load_index()
|
|
blob = self._blob_path(attachment_id, fname)
|
|
blob.parent.mkdir(parents=True, exist_ok=True)
|
|
# Copy under the lock so cleanup can never race a half-written
|
|
# blob (cleanup only deletes registered+expired entries).
|
|
await asyncio.to_thread(shutil.copy2, str(src), str(blob))
|
|
meta: Dict[str, Any] = {
|
|
"attachment_id": attachment_id,
|
|
"filename": fname,
|
|
"mime_type": mime,
|
|
"size": size,
|
|
"kind": kind,
|
|
"conversation_id": conversation_id,
|
|
"message_id": message_id,
|
|
"direction": "outbound",
|
|
"created_at": protocol.now_iso(),
|
|
"created_epoch": time.time(),
|
|
"retention_days": self._retention_days,
|
|
"blob": blob.name,
|
|
"blob_subdir": blob.parent.name,
|
|
}
|
|
self._meta[attachment_id] = meta
|
|
self._save_index()
|
|
except Exception:
|
|
logger.error("[pheby] attachment registration failed for %s",
|
|
protocol.safe_str(source_path, 120), exc_info=True)
|
|
return None
|
|
|
|
logger.info(
|
|
"[pheby] attachment registered: id=%s kind=%s size=%d conv=%s",
|
|
attachment_id, kind, size, conversation_id)
|
|
return self.describe(attachment_id)
|
|
|
|
async def register_bytes(
|
|
self,
|
|
data: bytes,
|
|
*,
|
|
conversation_id: str,
|
|
filename: str,
|
|
mime_type: Optional[str] = None,
|
|
message_id: Optional[str] = None,
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Register an *inbound* upload: raw client bytes → adapter storage.
|
|
|
|
Returns the attachment descriptor dict, or ``None`` on failure.
|
|
"""
|
|
fname = self._sanitize_filename(filename)
|
|
attachment_id = protocol.new_id()
|
|
mime = (mime_type or "").strip()
|
|
if mime in ("", "application/octet-stream"):
|
|
mime = guess_mime(fname)
|
|
if not isinstance(mime, str) or "/" not in mime or len(mime) > 120:
|
|
mime = "application/octet-stream"
|
|
is_image = mime.startswith(IMAGE_MIME_PREFIXES) or (
|
|
Path(fname).suffix.lower() in IMAGE_EXTS)
|
|
if Path(fname).suffix.lower() in VIDEO_EXTS:
|
|
kind = "video"
|
|
elif Path(fname).suffix.lower() in AUDIO_EXTS:
|
|
kind = "audio"
|
|
elif is_image:
|
|
kind = "image"
|
|
else:
|
|
kind = "document"
|
|
size = len(data)
|
|
|
|
try:
|
|
async with self._lock:
|
|
self._load_index()
|
|
blob = self._blob_path(attachment_id, fname)
|
|
blob.parent.mkdir(parents=True, exist_ok=True)
|
|
await asyncio.to_thread(blob.write_bytes, data)
|
|
meta: Dict[str, Any] = {
|
|
"attachment_id": attachment_id,
|
|
"filename": fname,
|
|
"mime_type": mime,
|
|
"size": size,
|
|
"kind": kind,
|
|
"conversation_id": conversation_id,
|
|
"message_id": message_id,
|
|
"direction": "inbound",
|
|
"created_at": protocol.now_iso(),
|
|
"created_epoch": time.time(),
|
|
"retention_days": self._retention_days,
|
|
"blob": blob.name,
|
|
"blob_subdir": blob.parent.name,
|
|
}
|
|
self._meta[attachment_id] = meta
|
|
self._save_index()
|
|
except Exception:
|
|
logger.error("[pheby] inbound attachment registration failed",
|
|
exc_info=True)
|
|
return None
|
|
|
|
logger.info(
|
|
"[pheby] inbound attachment registered: id=%s kind=%s size=%d "
|
|
"conv=%s", attachment_id, kind, size, conversation_id)
|
|
return self.describe(attachment_id)
|
|
|
|
def anchor_message(self, attachment_id: str, message_id: str) -> None:
|
|
"""Attach *message_id* to a registered attachment (thread anchoring)."""
|
|
self._load_index()
|
|
meta = self._meta.get(attachment_id)
|
|
if meta is not None:
|
|
meta["message_id"] = str(message_id)
|
|
self._save_index()
|
|
|
|
def get_blob_path(self, attachment_id: str) -> Optional[Path]:
|
|
"""Filesystem path for an attachment (server-side use only)."""
|
|
return self.resolve_blob(attachment_id)
|
|
|
|
# ── lookup / download ────────────────────────────────────────────────
|
|
def describe(self, attachment_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Public descriptor for an attachment (no server paths)."""
|
|
meta = self._meta.get(attachment_id)
|
|
if not meta or self._is_expired(meta):
|
|
return None
|
|
return {
|
|
"attachment_id": attachment_id,
|
|
"filename": meta.get("filename", "file.bin"),
|
|
"mime_type": meta.get("mime_type", "application/octet-stream"),
|
|
"size": int(meta.get("size", 0)),
|
|
"kind": meta.get("kind", "document"),
|
|
"inline_image": meta.get("kind") == "image",
|
|
"direction": meta.get("direction", "outbound"),
|
|
"conversation_id": meta.get("conversation_id"),
|
|
"message_id": meta.get("message_id"),
|
|
"created_at": meta.get("created_at"),
|
|
"expires_at": self._expires_at_iso(meta),
|
|
"download_path": f"/attachments/{attachment_id}",
|
|
}
|
|
|
|
def _expires_at_iso(self, meta: Dict[str, Any]) -> Optional[str]:
|
|
retention = int(meta.get("retention_days", self._retention_days))
|
|
if retention <= 0:
|
|
return None
|
|
created = float(meta.get("created_epoch", 0) or 0)
|
|
if not created:
|
|
return None
|
|
import datetime as _dt
|
|
return _dt.datetime.fromtimestamp(
|
|
created + retention * 86400, tz=_dt.timezone.utc).isoformat()
|
|
|
|
def resolve_blob(self, attachment_id: str) -> Optional[Path]:
|
|
"""Resolve an ID to its blob path — only for registered IDs.
|
|
|
|
Returns ``None`` for unknown, expired, or malformed IDs. The
|
|
returned path is always inside the storage root (the blob filename
|
|
comes from sanitized metadata, never client input).
|
|
"""
|
|
if not protocol.is_valid_attachment_id(attachment_id):
|
|
return None
|
|
meta = self._meta.get(attachment_id)
|
|
if not meta:
|
|
return None
|
|
if self._is_expired(meta):
|
|
return None
|
|
blob = (self._root / "blobs" / str(meta.get("blob_subdir", "")) /
|
|
str(meta.get("blob", "")))
|
|
try:
|
|
resolved = blob.resolve(strict=True)
|
|
except (OSError, RuntimeError, ValueError):
|
|
return None
|
|
# Defense in depth: blob must live inside our storage root.
|
|
try:
|
|
resolved.relative_to(self._root)
|
|
except ValueError:
|
|
return None
|
|
if not resolved.is_file():
|
|
return None
|
|
return resolved
|
|
|
|
def list_for_conversation(self, conversation_id: str) -> List[Dict[str, Any]]:
|
|
out = []
|
|
for aid in list(self._meta):
|
|
desc = self.describe(aid)
|
|
if desc and desc.get("conversation_id") == conversation_id and (
|
|
desc.get("direction") != "inbound" or desc.get("message_id")):
|
|
out.append(desc)
|
|
return out
|
|
|
|
# ── cleanup ──────────────────────────────────────────────────────────
|
|
def _is_expired(self, meta: Dict[str, Any]) -> bool:
|
|
retention = int(meta.get("retention_days", self._retention_days))
|
|
if retention <= 0:
|
|
return False
|
|
created = float(meta.get("created_epoch", 0) or 0)
|
|
return created > 0 and (time.time() - created) > retention * 86400
|
|
|
|
async def cleanup_expired(self) -> int:
|
|
"""Delete expired adapter-owned blobs + metadata. Returns count.
|
|
|
|
Only deletes blobs this store registered (inside its own root, keyed
|
|
by ID). Never touches anything outside the storage root.
|
|
"""
|
|
async with self._lock:
|
|
self._load_index()
|
|
expired = [aid for aid, m in self._meta.items() if self._is_expired(m)]
|
|
removed = 0
|
|
for aid in expired:
|
|
meta = self._meta.pop(aid, None)
|
|
if not meta:
|
|
continue
|
|
blob = (self._root / "blobs" / str(meta.get("blob_subdir", "")) /
|
|
str(meta.get("blob", "")))
|
|
try:
|
|
resolved = blob.resolve(strict=False)
|
|
resolved.relative_to(self._root) # containment check
|
|
if resolved.is_file():
|
|
await asyncio.to_thread(resolved.unlink)
|
|
removed += 1
|
|
except (OSError, RuntimeError, ValueError):
|
|
logger.warning("[pheby] cleanup skipped blob for %s", aid)
|
|
try:
|
|
mp = self._meta_path(aid)
|
|
if mp.exists():
|
|
await asyncio.to_thread(mp.unlink)
|
|
except OSError:
|
|
pass
|
|
if expired:
|
|
self._save_index()
|
|
logger.info("[pheby] cleaned %d expired attachment(s)", removed)
|
|
return removed
|
|
|
|
# ── legacy per-id meta files (restart durability helper) ─────────────
|
|
def hydrate_legacy_meta(self) -> None:
|
|
"""Read any per-ID ``meta/*.json`` files from older versions."""
|
|
self._load_index()
|
|
try:
|
|
meta_dir = self._root / "meta"
|
|
if not meta_dir.is_dir():
|
|
return
|
|
for mp in meta_dir.glob("*.json"):
|
|
aid = mp.stem
|
|
if aid in self._meta or not protocol.is_valid_attachment_id(aid):
|
|
continue
|
|
try:
|
|
data = json.loads(mp.read_text(encoding="utf-8"))
|
|
if isinstance(data, dict):
|
|
self._meta[aid] = data
|
|
except Exception:
|
|
continue
|
|
except Exception:
|
|
logger.debug("[pheby] legacy meta hydration skipped", exc_info=True)
|
|
|
|
def wipe_all(self) -> None:
|
|
"""Test helper: remove everything this store owns."""
|
|
self._meta = {}
|
|
self._loaded = True
|
|
if self._root.exists():
|
|
shutil.rmtree(self._root, ignore_errors=True)
|
|
|
|
|
|
def constant_time_equals(a: str, b: str) -> bool:
|
|
"""Length-safe constant-time string comparison for secrets."""
|
|
a_b = a.encode("utf-8")
|
|
b_b = b.encode("utf-8")
|
|
return len(a_b) == len(b_b) and hmac.compare_digest(a_b, b_b)
|
|
|
|
|
|
def hash_secret_for_log(secret: str) -> str:
|
|
"""Short non-reversible fingerprint for log lines (never the secret)."""
|
|
if not secret:
|
|
return "unset"
|
|
return hashlib.sha256(secret.encode("utf-8")).hexdigest()[:8]
|
|
|
|
|
|
__all__ = [
|
|
"AttachmentStore", "constant_time_equals", "hash_secret_for_log",
|
|
"guess_mime", "IMAGE_EXTS", "AUDIO_EXTS", "VIDEO_EXTS",
|
|
]
|