pheby-hermes-platform-adapter/plugin/pheby/attachments.py

"""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,
                    "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)

    # ── 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",
            "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:
                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",
]