pinchflat-mcp: redact credentials, gate admin tools, expose retention

packages/pinchflat-mcp/server.py

@@ -1,5 +1,28 @@
 #!/usr/bin/env python3
-"""Pinchflat MCP server — exposes the Pinchflat REST API as MCP tools."""
+"""Pinchflat MCP server — exposes the Pinchflat REST API as MCP tools.
+
+Tool tiers
+----------
+Two tiers of tools exist and only one is exposed by default:
+
+* ``subscriptions`` (default) — sources, media items, media profiles, tasks.
+  This is what an assistant needs to manage subscriptions and their files.
+* ``admin`` — global application state: ``get_settings``, ``get_app_info``.
+  These describe the installation rather than a subscription and are NOT
+  exposed unless explicitly enabled.
+
+Select the tier with ``PINCHFLAT_MCP_TOOLSET``:
+
+    PINCHFLAT_MCP_TOOLSET=subscriptions   # default
+    PINCHFLAT_MCP_TOOLSET=admin
+    PINCHFLAT_MCP_TOOLSET=all
+
+Responses are scrubbed of secrets on the way out (see ``_scrub``), because the
+Pinchflat settings record embeds the application's own ``api_token``. Exposing
+it to a tool caller hands out a credential that outlives the tool session, so
+secret-bearing keys are redacted at the transport boundary rather than left to
+individual tools to remember.
+"""
 
 import os
 import sys
@@ -20,13 +43,76 @@ HEADERS = {"Authorization": f"Bearer {PINCHFLAT_TOKEN}", "Content-Type": "applic
 
 mcp = FastMCP("pinchflat")
 
+# --- Tool tiers -------------------------------------------------------------
+
+_TIERS = {
+    "subscriptions": {"subscriptions"},
+    "admin": {"subscriptions", "admin"},
+    "all": {"subscriptions", "admin"},
+}
+_TOOLSET = os.environ.get("PINCHFLAT_MCP_TOOLSET", "subscriptions").strip().lower()
+ENABLED_TIERS = _TIERS.get(_TOOLSET, _TIERS["subscriptions"])
+
+
+def tool(tier: str = "subscriptions"):
+    """Register a tool only when its tier is enabled."""
+
+    def decorator(fn):
+        if tier in ENABLED_TIERS:
+            return mcp.tool()(fn)
+        return fn
+
+    return decorator
+
+
+# --- Response scrubbing -----------------------------------------------------
+
+# Keys whose values are credentials and must never cross the tool boundary.
+SECRET_KEYS = frozenset(
+    {
+        "api_token",
+        "api_key",
+        "token",
+        "secret",
+        "password",
+        "password_hash",
+        "youtube_api_key",
+        "apprise_server",
+    }
+)
+REDACTED = "[redacted]"
+
+
+def _scrub(obj: Any) -> Any:
+    """Recursively redact credential-bearing values from an API response."""
+    if isinstance(obj, dict):
+        out = {}
+        for k, v in obj.items():
+            if isinstance(k, str) and k.strip().lower() in SECRET_KEYS and v:
+                out[k] = REDACTED
+            else:
+                out[k] = _scrub(v)
+        return out
+    if isinstance(obj, list):
+        return [_scrub(v) for v in obj]
+    return obj
+
+
+# --- HTTP helpers -----------------------------------------------------------
+
+
+def _unwrap(r: httpx.Response) -> Any:
+    """Decode a response and scrub secrets before it reaches the caller."""
+    body = r.json()
+    return _scrub(body)
+
 
 def _get(path: str, params: Optional[dict] = None) -> Any:
     """Make a GET request to the Pinchflat API."""
     with httpx.Client(timeout=30) as client:
         r = client.get(f"{BASE}{path}", headers=HEADERS, params=params)
         r.raise_for_status()
-        return r.json()
+        return _unwrap(r)
 
 
 def _post(path: str, json: Optional[dict] = None) -> Any:
@@ -34,7 +120,7 @@ def _post(path: str, json: Optional[dict] = None) -> Any:
     with httpx.Client(timeout=30) as client:
         r = client.post(f"{BASE}{path}", headers=HEADERS, json=json)
         r.raise_for_status()
-        return r.json()
+        return _unwrap(r)
 
 
 def _put(path: str, json: dict) -> Any:
@@ -42,38 +128,40 @@ def _put(path: str, json: dict) -> Any:
     with httpx.Client(timeout=30) as client:
         r = client.put(f"{BASE}{path}", headers=HEADERS, json=json)
         r.raise_for_status()
-        return r.json()
+        return _unwrap(r)
 
 
-def _delete(path: str) -> Any:
+def _delete(path: str, params: Optional[dict] = None, timeout: int = 30) -> Any:
     """Make a DELETE request to the Pinchflat API."""
-    with httpx.Client(timeout=30) as client:
+    with httpx.Client(timeout=timeout) as client:
-        r = client.delete(f"{BASE}{path}", headers=HEADERS)
+        r = client.delete(f"{BASE}{path}", headers=HEADERS, params=params)
         r.raise_for_status()
-        return r.json()
+        return _unwrap(r)
 
 
 # --- Sources ---
 
-@mcp.tool()
+
+@tool()
 def list_sources() -> dict:
     """List all configured media sources (YouTube channels, playlists, etc.)."""
     return _get("/sources")
 
 
-@mcp.tool()
+@tool()
 def get_source(source_id: int) -> dict:
     """Get details of a specific source by ID, including its pending tasks."""
     return _get(f"/sources/{source_id}")
 
 
-@mcp.tool()
+@tool()
 def create_source(
     original_url: str,
     media_profile_id: int,
     collection_type: str = "channel",
     custom_name: Optional[str] = None,
     download_cutoff_date: Optional[str] = None,
+    retention_period_days: Optional[int] = None,
 ) -> dict:
     """Create a new media source (e.g. add a YouTube channel to monitor).
 
@@ -83,6 +171,8 @@ def create_source(
         collection_type: 'channel', 'playlist', or 'video'
         custom_name: Optional custom name for the source
         download_cutoff_date: Only download media after this date (YYYY-MM-DD)
+        retention_period_days: Delete downloaded media this many days after
+            publication; omit to keep everything
     """
     source = {
         "media_profile_id": media_profile_id,
@@ -93,18 +183,30 @@ def create_source(
         source["custom_name"] = custom_name
     if download_cutoff_date:
         source["download_cutoff_date"] = download_cutoff_date
+    if retention_period_days is not None:
+        source["retention_period_days"] = retention_period_days
     return _post("/sources", json={"source": source})
 
 
-@mcp.tool()
+@tool()
 def update_source(
     source_id: int,
     custom_name: Optional[str] = None,
     download_cutoff_date: Optional[str] = None,
     enabled: Optional[bool] = None,
     index_frequency_minutes: Optional[int] = None,
+    retention_period_days: Optional[int] = None,
+    prevent_culling: Optional[bool] = None,
+    download_media: Optional[bool] = None,
 ) -> dict:
-    """Update an existing source. Only provided fields will be changed."""
+    """Update an existing source. Only provided fields will be changed.
+
+    Args:
+        retention_period_days: Delete downloaded media this many days after
+            publication (0 keeps everything)
+        prevent_culling: If True, never delete this source's downloads
+        download_media: If False, index the source but download nothing
+    """
     updates = {}
     if custom_name is not None:
         updates["custom_name"] = custom_name
@@ -114,34 +216,46 @@ def update_source(
         updates["enabled"] = enabled
     if index_frequency_minutes is not None:
         updates["index_frequency_minutes"] = index_frequency_minutes
+    if retention_period_days is not None:
+        updates["retention_period_days"] = retention_period_days
+    if prevent_culling is not None:
+        updates["prevent_culling"] = prevent_culling
+    if download_media is not None:
+        updates["download_media"] = download_media
+    if not updates:
+        return {"data": {"message": "No fields to update."}}
     return _put(f"/sources/{source_id}", json={"source": updates})
 
 
-@mcp.tool()
+@tool()
 def delete_source(source_id: int) -> dict:
-    """Delete a source (marks it for deletion)."""
+    """Delete a source (marks it for deletion).
+
+    Note: this removes the source and its media records, but does NOT delete
+    the files from disk. Delete the media items first if space is the goal.
+    """
     return _delete(f"/sources/{source_id}")
 
 
-@mcp.tool()
+@tool()
 def force_download_pending(source_id: int) -> dict:
     """Force download of all pending media items for a source."""
     return _post(f"/sources/{source_id}/force_download_pending")
 
 
-@mcp.tool()
+@tool()
 def force_redownload(source_id: int) -> dict:
     """Force re-download of all existing media for a source."""
     return _post(f"/sources/{source_id}/force_redownload")
 
 
-@mcp.tool()
+@tool()
 def force_index(source_id: int) -> dict:
     """Force re-index of a source."""
     return _post(f"/sources/{source_id}/force_index")
 
 
-@mcp.tool()
+@tool()
 def force_metadata_refresh(source_id: int) -> dict:
     """Force metadata refresh for a source."""
     return _post(f"/sources/{source_id}/force_metadata_refresh")
@@ -149,7 +263,8 @@ def force_metadata_refresh(source_id: int) -> dict:
 
 # --- Media Items ---
 
-@mcp.tool()
+
+@tool()
 def list_media(
     source_id: Optional[int] = None,
     limit: int = 50,
@@ -172,25 +287,25 @@ def list_media(
     return _get("/media", params=params)
 
 
-@mcp.tool()
+@tool()
 def search_media(query: str) -> dict:
     """Search media items by title or description."""
     return _get("/media/search", params={"q": query})
 
 
-@mcp.tool()
+@tool()
 def get_media(media_id: int) -> dict:
     """Get details of a specific media item."""
     return _get(f"/media/{media_id}")
 
 
-@mcp.tool()
+@tool()
 def force_download_media(media_id: int) -> dict:
     """Force download of a specific media item."""
     return _post(f"/media/{media_id}/force_download")
 
 
-@mcp.tool()
+@tool()
 def delete_media(media_id: int, prevent_download: bool = False) -> dict:
     """Delete a media item's downloaded files from disk.
 
@@ -201,21 +316,19 @@ def delete_media(media_id: int, prevent_download: bool = False) -> dict:
     params = {}
     if prevent_download:
         params["prevent_download"] = "true"
-    with httpx.Client(timeout=120) as client:
+    return _delete(f"/media/{media_id}", params=params, timeout=120)
-        r = client.delete(f"{BASE}/media/{media_id}", headers=HEADERS, params=params)
-        r.raise_for_status()
-        return r.json()
 
 
 # --- Media Profiles ---
 
-@mcp.tool()
+
+@tool()
 def list_media_profiles() -> dict:
     """List all media profiles (download quality/container settings)."""
     return _get("/media_profiles")
 
 
-@mcp.tool()
+@tool()
 def get_media_profile(profile_id: int) -> dict:
     """Get details of a specific media profile."""
     return _get(f"/media_profiles/{profile_id}")
@@ -223,7 +336,8 @@ def get_media_profile(profile_id: int) -> dict:
 
 # --- Tasks ---
 
-@mcp.tool()
+
+@tool()
 def list_tasks(source_id: Optional[int] = None) -> dict:
     """List all tasks. Optionally filter by source_id."""
     params = {}
@@ -232,21 +346,22 @@ def list_tasks(source_id: Optional[int] = None) -> dict:
     return _get("/tasks", params=params)
 
 
-@mcp.tool()
+@tool()
 def get_task(task_id: int) -> dict:
     """Get details of a specific task."""
     return _get(f"/tasks/{task_id}")
 
 
-# --- Settings & Info ---
+# --- Settings & Info (admin tier — not exposed by default) ---
 
-@mcp.tool()
+
+@tool("admin")
 def get_settings() -> dict:
-    """Get current Pinchflat settings."""
+    """Get current Pinchflat settings (credential values are redacted)."""
     return _get("/settings")
 
 
-@mcp.tool()
+@tool("admin")
 def get_app_info() -> dict:
     """Get Pinchflat app info (version, environment, etc.)."""
     return _get("/app_info")
@@ -254,4 +369,4 @@ def get_app_info() -> dict:
 
 if __name__ == "__main__":
     transport = os.environ.get("PINCHFLAT_MCP_TRANSPORT", "stdio")
-    mcp.run(transport=transport if transport in ("stdio", "http") else "stdio")
+    mcp.run(transport=transport if transport in ("stdio", "http") else "stdio")