feat: render remote images in Feishu cards

channel/feishu/feishu_static_card.py

@@ -1,8 +1,13 @@
 """Helpers for choosing the native Feishu delivery format for text replies."""
 
+import ipaddress
 import json
 import re
-from typing import Tuple
+import socket
+from typing import Callable, Optional, Tuple
+from urllib.parse import urljoin, urlparse
+
+import requests
 
 
 _BLOCK_MARKDOWN = re.compile(
@@ -10,6 +15,10 @@ _BLOCK_MARKDOWN = re.compile(
 )
 _INLINE_MARKDOWN = re.compile(r"(`[^`\n]+`|\*\*[^*\n]+\*\*|\[[^]\n]+\]\([^)\n]+\))")
 _TABLE_SEPARATOR = re.compile(r"(?m)^\s*\|?\s*:?-{3,}:?\s*(?:\|\s*:?-{3,}:?\s*)+\|?\s*$")
+_MARKDOWN_IMAGE = re.compile(r"!\[([^\]\n]*)\]\(([^)\s]+)\)")
+_REDIRECT_CODES = {301, 302, 303, 307, 308}
+_MAX_REDIRECTS = 3
+_MAX_REMOTE_IMAGE_BYTES = 10 * 1024 * 1024
 
 
 def contains_markdown(text: str) -> bool:
@@ -44,3 +53,166 @@ def build_text_delivery(text: str, enabled: bool = True) -> Tuple[str, str]:
     if enabled and contains_markdown(text):
         return "interactive", json.dumps(build_markdown_card(text), ensure_ascii=False)
     return "text", json.dumps({"text": text}, ensure_ascii=False)
+
+
+def resolve_markdown_images(
+    text: str,
+    uploader: Callable[[str], Optional[str]],
+    max_images: int = 5,
+) -> str:
+    """Replace remote Markdown image URLs with Feishu image keys."""
+    cache = {}
+    uploaded = 0
+
+    def replace(match):
+        nonlocal uploaded
+        alt = match.group(1).strip() or "image"
+        target = match.group(2).strip()
+        if target.startswith("img_"):
+            return match.group(0)
+        if urlparse(target).scheme not in ("http", "https"):
+            return match.group(0)
+
+        if target not in cache:
+            if uploaded >= max_images:
+                cache[target] = None
+            else:
+                uploaded += 1
+                try:
+                    cache[target] = uploader(target)
+                except Exception:
+                    cache[target] = None
+
+        image_key = cache[target]
+        if image_key:
+            return "![{}]({})".format(alt, image_key)
+        return "[Image unavailable: {}]".format(alt)
+
+    return _MARKDOWN_IMAGE.sub(replace, text or "")
+
+
+def validate_public_image_url(url: str) -> None:
+    """Reject non-HTTP and non-public image targets before downloading."""
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise ValueError("unsupported image URL scheme")
+    if not parsed.hostname:
+        raise ValueError("image URL has no hostname")
+
+    try:
+        literal_address = ipaddress.ip_address(parsed.hostname)
+        resolved_addresses = [literal_address]
+    except ValueError:
+        try:
+            addresses = socket.getaddrinfo(
+                parsed.hostname,
+                parsed.port,
+                socket.AF_UNSPEC,
+                socket.SOCK_STREAM,
+            )
+        except socket.gaierror as exc:
+            raise ValueError("cannot resolve image hostname") from exc
+        resolved_addresses = [ipaddress.ip_address(item[4][0]) for item in addresses]
+
+    for address in resolved_addresses:
+        if (
+            address.is_private
+            or address.is_loopback
+            or address.is_link_local
+            or address.is_reserved
+            or address.is_multicast
+            or address.is_unspecified
+        ):
+            raise ValueError("image URL resolves to a non-public address")
+
+
+def download_public_image(
+    url: str,
+    get=requests.get,
+    max_bytes: int = _MAX_REMOTE_IMAGE_BYTES,
+) -> Tuple[bytes, str]:
+    """Download a public image with redirect, type, and size checks."""
+    current = url
+    for _ in range(_MAX_REDIRECTS + 1):
+        validate_public_image_url(current)
+        response = get(
+            current,
+            headers={"User-Agent": "CowAgent/Feishu"},
+            timeout=(5, 15),
+            allow_redirects=False,
+            stream=True,
+        )
+
+        if response.status_code in _REDIRECT_CODES:
+            location = response.headers.get("Location")
+            response.close()
+            if not location:
+                raise ValueError("image redirect has no location")
+            current = urljoin(current, location)
+            continue
+
+        if response.status_code != 200:
+            response.close()
+            raise ValueError("image download returned HTTP {}".format(response.status_code))
+
+        content_type = response.headers.get("Content-Type", "").split(";", 1)[0].lower()
+        if not content_type.startswith("image/"):
+            response.close()
+            raise ValueError("remote resource is not an image")
+
+        try:
+            content_length = int(response.headers.get("Content-Length") or 0)
+        except (TypeError, ValueError):
+            content_length = 0
+        if content_length > max_bytes:
+            response.close()
+            raise ValueError("remote image is too large")
+
+        chunks = []
+        downloaded = 0
+        try:
+            for chunk in response.iter_content(chunk_size=8192):
+                if not chunk:
+                    continue
+                downloaded += len(chunk)
+                if downloaded > max_bytes:
+                    raise ValueError("remote image is too large")
+                chunks.append(chunk)
+        finally:
+            response.close()
+        return b"".join(chunks), content_type
+
+    raise ValueError("too many image redirects")
+
+
+def upload_public_image_to_feishu(
+    url: str,
+    access_token: str,
+    post=requests.post,
+) -> Optional[str]:
+    """Download a public image and upload its bytes to Feishu."""
+    payload, content_type = download_public_image(url)
+    extension = {
+        "image/jpeg": "jpg",
+        "image/png": "png",
+        "image/gif": "gif",
+        "image/webp": "webp",
+        "image/bmp": "bmp",
+    }.get(content_type, "img")
+    response = post(
+        "https://open.feishu.cn/open-apis/im/v1/images",
+        headers={"Authorization": "Bearer " + access_token},
+        data={"image_type": "message"},
+        files={
+            "image": (
+                "markdown-image.{}".format(extension),
+                payload,
+                content_type,
+            )
+        },
+        timeout=(5, 15),
+    )
+    body = response.json()
+    if body.get("code") != 0:
+        return None
+    return (body.get("data") or {}).get("image_key")