Merge pull request #2 from BlueBirdBack/security-hardening-ssrf-yaml-limits

security: harden URL fetches + YAML frontmatter handling

README.md

@@ -111,11 +111,15 @@ result = run("""
 
 DeepReader now seamlessly integrates with **Google NotebookLM**. 
 
-If your message includes keywords like `notebooklm`, `audio`, or `podcast`, DeepReader will automatically:
+Use explicit flags to opt in:
+- `--notebooklm` (or `/notebooklm`) → upload to NotebookLM
+- `--audio` / `--podcast` (or `/audio`) → upload + generate Audio Overview
+
+When these flags are present, DeepReader will:
 1. Parse the requested URLs into Markdown.
 2. Create a new Notebook in your Google NotebookLM account.
-3. Upload the pristine Markdown content as a source.
-4. **(Optional)** Generate an engaging Audio Overview (podcast format) and download it directly to your agent's memory folder.
+3. Upload the Markdown content as a source.
+4. **(Optional)** Generate an Audio Overview and download it to the memory folder.
 
 **Supported NotebookLM Artifacts Generation:**
 Along with Audio Overviews, this integration can easily be extended to automatically generate and save:

deepreader_skill/__init__.py

@@ -28,7 +28,7 @@ from typing import Any
 
 from .core.router import ParserRouter
 from .core.storage import StorageManager
-from .core.utils import extract_urls
+from .core.utils import extract_urls, redact_url_for_log, validate_external_url
 
 __version__ = "1.0.0"
 __all__ = ["run"]
@@ -69,6 +69,23 @@ def _get_storage() -> StorageManager:
     return _storage
 
 
+_NOTEBOOKLM_FLAGS = ("--notebooklm", "/notebooklm", "#notebooklm")
+_AUDIO_FLAGS = ("--audio", "/audio", "--podcast", "/podcast")
+
+
+def _parse_notebooklm_flags(text: str) -> tuple[bool, bool]:
+    """Parse explicit NotebookLM/audio flags from the user message."""
+    lowered = text.lower()
+    use_notebooklm = any(flag in lowered for flag in _NOTEBOOKLM_FLAGS)
+    generate_audio = any(flag in lowered for flag in _AUDIO_FLAGS)
+
+    # Audio implies NotebookLM usage.
+    if generate_audio:
+        use_notebooklm = True
+
+    return use_notebooklm, generate_audio
+
+
 # ---------------------------------------------------------------------------
 # Public API — OpenClaw Entry Point
 # ---------------------------------------------------------------------------
@@ -97,12 +114,23 @@ def run(text: str, **kwargs: Any) -> str:
 
         router = _get_router()
         storage = _get_storage()
+        use_notebooklm, generate_audio = _parse_notebooklm_flags(text)
 
         results: list[str] = []
         errors: list[str] = []
 
         for url in urls:
-            logger.info("Processing URL: %s", url)
+            safe, reason = validate_external_url(url)
+            if not safe:
+                error_msg = (
+                    f"❌ Blocked **{url}**\n"
+                    f"   Reason: {reason}"
+                )
+                errors.append(error_msg)
+                logger.warning("Blocked URL %s: %s", redact_url_for_log(url), reason)
+                continue
+
+            logger.info("Processing URL: %s", redact_url_for_log(url))
 
             # Step 2: Route to the correct parser
             parse_result = router.route(url)
@@ -113,7 +141,7 @@ def run(text: str, **kwargs: Any) -> str:
                     f"   Reason: {parse_result.error}"
                 )
                 errors.append(error_msg)
-                logger.warning("Parse failed for %s: %s", url, parse_result.error)
+                logger.warning("Parse failed for %s: %s", redact_url_for_log(url), parse_result.error)
                 continue
 
             # Step 3: Save to memory
@@ -126,22 +154,18 @@ def run(text: str, **kwargs: Any) -> str:
                     f"   Content: {len(parse_result.content)} characters"
                 )
                 
-                # --- NotebookLM Integration ---
-                text_lower = text.lower()
-                use_notebooklm = "notebooklm" in text_lower or "audio" in text_lower or "podcast" in text_lower
-                generate_audio = "audio" in text_lower or "podcast" in text_lower
-                
+                # --- NotebookLM Integration (explicit opt-in flags only) ---
                 if use_notebooklm:
                     logger.info("NotebookLM integration triggered for %s", filepath)
                     from .integrations.notebooklm import NotebookLMIntegration
                     nl_integration = NotebookLMIntegration()
-                    
+
                     nl_result = nl_integration.run_sync(
                         filepath=filepath,
                         title=parse_result.title or "DeepReader Document",
-                        generate_audio=generate_audio
+                        generate_audio=generate_audio,
                     )
-                    
+
                     if "error" in nl_result:
                         errors.append(f"❌ NotebookLM upload failed: {nl_result['error']}")
                     else:
@@ -152,13 +176,13 @@ def run(text: str, **kwargs: Any) -> str:
 
                 results.append(success_msg)
                 logger.info("Successfully saved %s", filepath)
-            except OSError as exc:
+            except Exception as exc:  # noqa: BLE001
                 error_msg = (
-                    f"❌ Parsed **{url}** but failed to save.\n"
+                    f"❌ Parsed **{url}** but failed during save/post-processing.\n"
                     f"   Error: {exc}"
                 )
                 errors.append(error_msg)
-                logger.error("Storage error for %s: %s", url, exc)
+                logger.error("Storage/post-process error for %s: %s", redact_url_for_log(url), exc)
 
         # Step 4: Build the response
         response_parts: list[str] = []

deepreader_skill/core/storage.py

@@ -7,17 +7,41 @@ YAML frontmatter and persisting it to the agent's memory directory.
 
 from __future__ import annotations
 
+import json
 import logging
-import os
+import re
 import uuid
 from datetime import datetime, timezone
 from pathlib import Path
 
-from ..core.utils import build_filename, content_hash, generate_excerpt, get_domain, get_domain_tag
+from ..core.utils import build_filename, content_hash, generate_excerpt, get_domain_tag
 from ..parsers.base import ParseResult
 
 logger = logging.getLogger("deepreader.storage")
 
+_TAG_UNSAFE = re.compile(r"[^a-z0-9_-]+")
+
+
+def _yaml_quote(value: str) -> str:
+    """Return a YAML-safe quoted scalar using JSON string escaping."""
+    return json.dumps(value, ensure_ascii=False)
+
+
+def _sanitize_metadata_value(value: str, max_length: int = 300) -> str:
+    """Normalize untrusted metadata fields for frontmatter safety."""
+    normalized = value.replace("\x00", " ").replace("\r", " ").replace("\n", " ").strip()
+    if len(normalized) > max_length:
+        normalized = normalized[:max_length].rstrip()
+    return normalized
+
+
+def _sanitize_tag(tag: str) -> str:
+    """Normalize arbitrary tags to a strict slug-ish format."""
+    value = _sanitize_metadata_value(tag, max_length=80).lower().replace(" ", "-")
+    value = _TAG_UNSAFE.sub("-", value)
+    value = re.sub(r"-{2,}", "-", value).strip("-")
+    return value or "tag"
+
 
 class StorageManager:
     """Persist parsed content to the agent's long-term memory.
@@ -109,43 +133,45 @@ class StorageManager:
         """
         doc_uuid = str(uuid.uuid4())
         iso_date = datetime.now(timezone.utc).isoformat()
-        domain = get_domain(result.url)
         domain_tag = get_domain_tag(result.url)
 
         # Build tags list
         tags = ["imported", domain_tag]
         if result.tags:
             tags.extend(result.tags)
-        # Deduplicate while preserving order
+
+        # Deduplicate while preserving order + sanitize
         seen: set[str] = set()
         unique_tags: list[str] = []
         for tag in tags:
-            if tag not in seen:
-                seen.add(tag)
-                unique_tags.append(tag)
-        tags_str = ", ".join(unique_tags)
+            clean_tag = _sanitize_tag(str(tag))
+            if clean_tag not in seen:
+                seen.add(clean_tag)
+                unique_tags.append(clean_tag)
 
-        # Escape title for YAML (handle quotes)
-        safe_title = result.title.replace('"', '\\"') if result.title else ""
-        safe_author = result.author.replace('"', '\\"') if result.author else ""
+        safe_title = _sanitize_metadata_value(result.title) if result.title else ""
+        safe_author = _sanitize_metadata_value(result.author) if result.author else ""
 
         # Build the excerpt / summary
-        excerpt = result.excerpt or generate_excerpt(result.content)
+        excerpt = _sanitize_metadata_value(result.excerpt or generate_excerpt(result.content), max_length=600)
 
         # Assemble the document
         lines: list[str] = [
             "---",
             f"uuid: {doc_uuid}",
-            f'source: "{result.url}"',
+            f"source: {_yaml_quote(result.url)}",
             f"date: {iso_date}",
             "type: external_resource",
-            f"tags: [{tags_str}]",
+            "tags:",
         ]
 
+        for tag in unique_tags:
+            lines.append(f"  - {_yaml_quote(tag)}")
+
         if safe_title:
-            lines.append(f'title: "{safe_title}"')
+            lines.append(f"title: {_yaml_quote(safe_title)}")
         if safe_author:
-            lines.append(f'author: "{safe_author}"')
+            lines.append(f"author: {_yaml_quote(safe_author)}")
 
         c_hash = content_hash(result.content)
         lines.append(f"content_hash: {c_hash}")
@@ -153,7 +179,8 @@ class StorageManager:
         lines.append("")
 
         # Heading
-        lines.append(f"# {result.title or 'Untitled'}")
+        heading_title = safe_title or "Untitled"
+        lines.append(f"# {heading_title}")
         lines.append("")
 
         # Summary section

deepreader_skill/core/utils.py

@@ -8,10 +8,12 @@ filename sanitization, and content hashing.
 from __future__ import annotations
 
 import hashlib
+import ipaddress
 import re
+import socket
 import unicodedata
 from datetime import datetime, timezone
-from urllib.parse import urlparse
+from urllib.parse import urlparse, urlunparse
 
 import tldextract
 
@@ -40,14 +42,124 @@ def extract_urls(text: str) -> list[str]:
     return urls
 
 
+# ---------------------------------------------------------------------------
+# URL Safety Helpers
+# ---------------------------------------------------------------------------
+
+_ALLOWED_SCHEMES = {"http", "https"}
+_BLOCKED_HOSTS = {
+    "localhost",
+    "localhost.localdomain",
+}
+_BLOCKED_HOST_SUFFIXES = (
+    ".local",
+    ".internal",
+    ".lan",
+    ".home",
+)
+
+
+def _is_non_public_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
+    """Return ``True`` for loopback/private/link-local/etc. addresses."""
+    return any(
+        [
+            ip.is_private,
+            ip.is_loopback,
+            ip.is_link_local,
+            ip.is_multicast,
+            ip.is_reserved,
+            ip.is_unspecified,
+        ]
+    )
+
+
+def _resolve_ips(hostname: str) -> set[str]:
+    """Resolve hostname to IPs (both v4/v6 when available)."""
+    ips: set[str] = set()
+    for result in socket.getaddrinfo(hostname, None, proto=socket.IPPROTO_TCP):
+        sockaddr = result[4]
+        if sockaddr:
+            ips.add(sockaddr[0])
+    return ips
+
+
+def validate_external_url(url: str) -> tuple[bool, str]:
+    """Validate that *url* is safe for outbound fetches.
+
+    Blocks local/internal destinations to mitigate SSRF.
+    """
+    parsed = urlparse(url)
+
+    if parsed.scheme.lower() not in _ALLOWED_SCHEMES:
+        return False, "Only http/https URLs are allowed."
+
+    hostname = (parsed.hostname or "").strip().lower().rstrip(".")
+    if not hostname:
+        return False, "URL is missing a hostname."
+
+    if hostname in _BLOCKED_HOSTS or any(hostname.endswith(s) for s in _BLOCKED_HOST_SUFFIXES):
+        return False, "Local/internal hostnames are blocked."
+
+    # Literal IP host
+    try:
+        ip_obj = ipaddress.ip_address(hostname)
+        if _is_non_public_ip(ip_obj):
+            return False, f"Blocked non-public IP: {ip_obj}"
+        return True, ""
+    except ValueError:
+        pass
+
+    # DNS hostname
+    try:
+        resolved_ips = _resolve_ips(hostname)
+    except socket.gaierror as exc:
+        return False, f"DNS resolution failed: {exc}"
+
+    if not resolved_ips:
+        return False, "DNS resolution returned no IP addresses."
+
+    for ip_str in resolved_ips:
+        ip_obj = ipaddress.ip_address(ip_str)
+        if _is_non_public_ip(ip_obj):
+            return False, f"Hostname resolves to non-public IP: {ip_str}"
+
+    return True, ""
+
+
+def redact_url_for_log(url: str) -> str:
+    """Redact query/fragment from URLs before logging."""
+    parsed = urlparse(url)
+    if not parsed.scheme or not parsed.netloc:
+        return url
+
+    cleaned = urlunparse(parsed._replace(query="", fragment=""))
+    if parsed.query or parsed.fragment:
+        return f"{cleaned} [query redacted]"
+    return cleaned
+
+
 # ---------------------------------------------------------------------------
 # Domain Helpers
 # ---------------------------------------------------------------------------
 
 def get_domain(url: str) -> str:
-    """Return the registered domain of *url* (e.g. ``'twitter.com'``)."""
+    """Return a normalized domain label for *url*."""
+    hostname = (urlparse(url).hostname or "").lower()
+    if not hostname:
+        return "unknown"
+
+    try:
+        ipaddress.ip_address(hostname)
+        return hostname
+    except ValueError:
+        pass
+
     extracted = tldextract.extract(url)
-    return f"{extracted.domain}.{extracted.suffix}".lower()
+    if extracted.domain and extracted.suffix:
+        return f"{extracted.domain}.{extracted.suffix}".lower()
+    if extracted.domain:
+        return extracted.domain.lower()
+    return hostname
 
 
 def get_domain_tag(url: str) -> str:

deepreader_skill/parsers/generic.py

@@ -23,6 +23,13 @@ class GenericParser(BaseParser):
     """Extract main content from any web page using trafilatura."""
 
     name = "generic"
+    max_response_bytes = 5_000_000
+    _allowed_content_types = (
+        "text/html",
+        "application/xhtml+xml",
+        "application/xml",
+        "text/plain",
+    )
 
     def parse(self, url: str) -> ParseResult:
         """Fetch *url* and extract the article body via trafilatura."""
@@ -62,15 +69,39 @@ class GenericParser(BaseParser):
     # ------------------------------------------------------------------
 
     def _fetch_html(self, url: str) -> str | None:
-        """Download the raw HTML content of *url*."""
-        response = requests.get(
+        """Download HTML with redirect + size/content-type safety checks."""
+        from ..core.utils import validate_external_url
+
+        with requests.get(
             url,
             headers=self._get_headers(),
             timeout=self.timeout,
             allow_redirects=True,
-        )
-        response.raise_for_status()
-        return response.text
+            stream=True,
+        ) as response:
+            response.raise_for_status()
+
+            final_url = response.url or url
+            safe, reason = validate_external_url(final_url)
+            if not safe:
+                raise requests.RequestException(f"Blocked redirect target: {reason}")
+
+            content_type = (response.headers.get("Content-Type") or "").lower()
+            if content_type and not any(ct in content_type for ct in self._allowed_content_types):
+                raise requests.RequestException(f"Unsupported content type: {content_type}")
+
+            body = bytearray()
+            for chunk in response.iter_content(chunk_size=8192):
+                if not chunk:
+                    continue
+                body.extend(chunk)
+                if len(body) > self.max_response_bytes:
+                    raise requests.RequestException(
+                        f"Response exceeds {self.max_response_bytes} bytes limit"
+                    )
+
+            encoding = response.encoding or response.apparent_encoding or "utf-8"
+            return body.decode(encoding, errors="replace")
 
     def _extract_with_trafilatura(self, url: str, html: str) -> ParseResult | None:
         """Use trafilatura to extract structured content."""

deepreader_skill/parsers/youtube.py

@@ -23,6 +23,8 @@ class YouTubeParser(BaseParser):
 
     name = "youtube"
     timeout = 25
+    max_metadata_bytes = 2_000_000
+    _allowed_content_types = ("text/html", "application/xhtml+xml")
 
     # Preferred language codes for transcripts (in priority order).
     preferred_languages: list[str] = ["en", "zh-Hans", "zh-Hant", "zh", "ja", "ko", "de", "fr", "es"]
@@ -194,10 +196,41 @@ class YouTubeParser(BaseParser):
 
         Returns ``(title, author, description)``.
         """
+        from ..core.utils import validate_external_url
+
         try:
-            resp = requests.get(url, headers=self._get_headers(), timeout=self.timeout)
-            resp.raise_for_status()
-            soup = BeautifulSoup(resp.text, "lxml")
+            with requests.get(
+                url,
+                headers=self._get_headers(),
+                timeout=self.timeout,
+                allow_redirects=True,
+                stream=True,
+            ) as resp:
+                resp.raise_for_status()
+
+                final_url = resp.url or url
+                safe, reason = validate_external_url(final_url)
+                if not safe:
+                    raise requests.RequestException(f"Blocked redirect target: {reason}")
+
+                content_type = (resp.headers.get("Content-Type") or "").lower()
+                if content_type and not any(ct in content_type for ct in self._allowed_content_types):
+                    raise requests.RequestException(f"Unsupported content type: {content_type}")
+
+                body = bytearray()
+                for chunk in resp.iter_content(chunk_size=8192):
+                    if not chunk:
+                        continue
+                    body.extend(chunk)
+                    if len(body) > self.max_metadata_bytes:
+                        raise requests.RequestException(
+                            f"Metadata page exceeds {self.max_metadata_bytes} bytes limit"
+                        )
+
+                encoding = resp.encoding or resp.apparent_encoding or "utf-8"
+                html = body.decode(encoding, errors="replace")
+
+            soup = BeautifulSoup(html, "lxml")
 
             # Title: <meta property="og:title">
             title = ""