feat(backend): restrict MinIO public read to qrcode prefix and add open_file

Only qrcode/* is anonymously readable; feedback/suggestion objects stay
private. Add resolve_object_key/open_file helpers for the admin proxy.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/core/storage.py

@@ -1,14 +1,20 @@
 """文件存储：local | minio。upload_file 返回 {url, objectKey}。"""
 from __future__ import annotations
 
+import json
+import mimetypes
 import os
 import uuid
 from datetime import datetime, timezone
 from io import BytesIO
 from pathlib import Path
-from typing import Any, Dict
+from typing import Any, BinaryIO, Dict, Tuple, Union
+from urllib.parse import urlparse
 
 from app.core.config import get_settings
+from app.core.upload_validate import IMAGE_EXTENSIONS, file_extension
+
+StreamLike = Union[BinaryIO, Any]
 
 
 class StorageError(Exception):
@@ -60,6 +66,23 @@ def _ensure_within_base(base_dir: Path, dest: Path) -> None:
         raise StorageError("非法的文件夹路径") from exc
 
 
+def _safe_content_type(filename: str) -> str:
+    guessed = mimetypes.guess_type(filename)[0] or "application/octet-stream"
+    banned = {
+        "application/x-msdownload",
+        "application/x-msdos-program",
+        "application/x-executable",
+        "application/x-sh",
+    }
+    if guessed in banned:
+        return "application/octet-stream"
+    return guessed
+
+
+def _key_has_traversal(key: str) -> bool:
+    return any(part == ".." for part in key.split("/"))
+
+
 class LocalStorage:
     def upload_file(self, content: bytes, filename: str, folder: str = "") -> Dict[str, str]:
         settings = get_settings()
@@ -69,9 +92,22 @@ class LocalStorage:
         _ensure_within_base(base_dir, dest)
         dest.parent.mkdir(parents=True, exist_ok=True)
         dest.write_bytes(content)
+        os.chmod(dest, 0o644)
         url = f"/uploads/{object_key}"
         return {"url": url, "objectKey": object_key}
 
+    def open_file(self, object_key: str) -> Tuple[StreamLike, str, str, bool]:
+        settings = get_settings()
+        base_dir = Path(settings.LOCAL_STORAGE_DIR)
+        dest = base_dir / object_key
+        _ensure_within_base(base_dir, dest)
+        if not dest.is_file():
+            raise StorageError("文件不存在")
+        filename = dest.name
+        content_type = _safe_content_type(filename)
+        inline = file_extension(filename) in IMAGE_EXTENSIONS
+        return dest.open("rb"), content_type, filename, inline
+
 
 class MinIOStorage:
     def __init__(self) -> None:
@@ -96,18 +132,46 @@ class MinIOStorage:
     def _ensure_bucket(self) -> None:
         if not self.client.bucket_exists(self.bucket_name):
             self.client.make_bucket(self.bucket_name)
+        # 仅二维码公示图匿名可读；反馈/建言附件私有
+        self._ensure_public_read_policy()
+
+    def _ensure_public_read_policy(self) -> None:
+        policy = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Principal": {"AWS": ["*"]},
+                    "Action": ["s3:GetObject"],
+                    "Resource": [f"arn:aws:s3:::{self.bucket_name}/qrcode/*"],
+                }
+            ],
+        }
+        self.client.set_bucket_policy(self.bucket_name, json.dumps(policy))
 
     def upload_file(self, content: bytes, filename: str, folder: str = "") -> Dict[str, str]:
         object_key = _object_key(folder, filename)
+        content_type = _safe_content_type(filename)
         self.client.put_object(
             bucket_name=self.bucket_name,
             object_name=object_key,
             data=BytesIO(content),
             length=len(content),
+            content_type=content_type,
         )
         url = f"{self.base_url}/{object_key}"
         return {"url": url, "objectKey": object_key}
 
+    def open_file(self, object_key: str) -> Tuple[StreamLike, str, str, bool]:
+        filename = object_key.rsplit("/", 1)[-1] or "file"
+        content_type = _safe_content_type(filename)
+        inline = file_extension(filename) in IMAGE_EXTENSIONS
+        try:
+            resp = self.client.get_object(self.bucket_name, object_key)
+        except Exception as exc:
+            raise StorageError("文件不存在或无法读取") from exc
+        return resp, content_type, filename, inline
+
 
 def get_storage() -> Any:
     settings = get_settings()
@@ -119,3 +183,63 @@ def get_storage() -> Any:
 def upload_file(content: bytes, filename: str, folder: str = "") -> Dict[str, str]:
     """上传文件，返回 {url, objectKey}。"""
     return get_storage().upload_file(content, filename, folder)
+
+
+def resolve_object_key(url: str) -> str:
+    """从库内 url 解析 objectKey；非法则 StorageError。"""
+    raw = (url or "").strip()
+    if not raw:
+        raise StorageError("无效的文件路径")
+    if "?" in raw:
+        raw = raw.split("?", 1)[0]
+
+    if raw.startswith("/uploads/"):
+        key = raw[len("/uploads/") :].lstrip("/")
+        if not key or _key_has_traversal(key):
+            raise StorageError("非法的文件夹路径")
+        return key
+
+    if raw.startswith("http://") or raw.startswith("https://"):
+        settings = get_settings()
+        parsed = urlparse(raw)
+        path = (parsed.path or "").lstrip("/")
+        bucket = settings.MINIO_BUCKET
+        marker = f"{bucket}/"
+        idx = path.find(marker)
+        if idx < 0:
+            raise StorageError("无效的文件路径")
+        key = path[idx + len(marker) :].lstrip("/")
+        if not key or _key_has_traversal(key):
+            raise StorageError("非法的文件夹路径")
+
+        # 校验 host：须匹配 MINIO_BASE_URL 或 MINIO_ENDPOINT
+        allowed_hosts: set[str] = set()
+        if settings.MINIO_BASE_URL:
+            base = urlparse(
+                settings.MINIO_BASE_URL
+                if "://" in settings.MINIO_BASE_URL
+                else f"http://{settings.MINIO_BASE_URL}"
+            )
+            if base.hostname:
+                allowed_hosts.add(base.hostname.lower())
+        endpoint_host = settings.MINIO_ENDPOINT.split(":")[0].lower()
+        if endpoint_host:
+            allowed_hosts.add(endpoint_host)
+        host = (parsed.hostname or "").lower()
+        if host not in allowed_hosts:
+            raise StorageError("无效的文件路径")
+        return key
+
+    # 允许直接传 objectKey（无前缀）
+    key = raw.lstrip("/")
+    if not key or _key_has_traversal(key) or key.startswith(".."):
+        raise StorageError("非法的文件夹路径")
+    if "://" in key:
+        raise StorageError("无效的文件路径")
+    return key
+
+
+def open_file(url: str) -> Tuple[StreamLike, str, str, bool]:
+    """返回 (stream, content_type, download_name, inline)。"""
+    key = resolve_object_key(url)
+    return get_storage().open_file(key)

backend/tests/test_minio_public_policy.py

@@ -0,0 +1,37 @@
+"""MinIO bucket allows anonymous GetObject only under qrcode/*."""
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+def test_minio_ensure_bucket_sets_public_read_policy(monkeypatch):
+    monkeypatch.setenv("STORAGE_TYPE", "minio")
+    monkeypatch.setenv("MINIO_ENDPOINT", "127.0.0.1:9000")
+    monkeypatch.setenv("MINIO_ACCESS_KEY", "ak")
+    monkeypatch.setenv("MINIO_SECRET_KEY", "sk")
+    monkeypatch.setenv("MINIO_BUCKET", "ajyy")
+    monkeypatch.setenv("MINIO_SECURE", "false")
+    monkeypatch.setenv("MINIO_BASE_URL", "")
+
+    from app.core.config import get_settings
+
+    get_settings.cache_clear()
+
+    mock_client = MagicMock()
+    mock_client.bucket_exists.return_value = True
+
+    with patch("minio.Minio", return_value=mock_client):
+        from app.core.storage import MinIOStorage
+
+        MinIOStorage()
+
+    mock_client.set_bucket_policy.assert_called_once()
+    bucket, policy_json = mock_client.set_bucket_policy.call_args[0]
+    assert bucket == "ajyy"
+    policy = json.loads(policy_json)
+    assert policy["Statement"][0]["Action"] == ["s3:GetObject"]
+    assert policy["Statement"][0]["Principal"] == {"AWS": ["*"]}
+    assert policy["Statement"][0]["Resource"] == ["arn:aws:s3:::ajyy/qrcode/*"]
+
+    get_settings.cache_clear()

backend/tests/test_storage_open.py

@@ -0,0 +1,39 @@
+"""resolve_object_key / open_file for local storage."""
+import pytest
+
+from app.core.config import get_settings
+from app.core import storage
+from app.core.storage import StorageError
+
+
+@pytest.fixture
+def local_storage_env(tmp_path, monkeypatch):
+    monkeypatch.setenv("STORAGE_TYPE", "local")
+    monkeypatch.setenv("LOCAL_STORAGE_DIR", str(tmp_path))
+    get_settings.cache_clear()
+    yield tmp_path
+    get_settings.cache_clear()
+
+
+def test_resolve_uploads_url(local_storage_env):
+    assert storage.resolve_object_key("/uploads/feedback/a.png") == "feedback/a.png"
+
+
+def test_resolve_rejects_traversal(local_storage_env):
+    with pytest.raises(StorageError):
+        storage.resolve_object_key("/uploads/../etc/passwd")
+
+
+def test_open_local_file(local_storage_env):
+    stored = storage.upload_file(b"\x89PNG\r\n\x1a\n" + b"x" * 8, "a.png", "feedback")
+    stream, content_type, filename, inline = storage.open_file(stored["url"])
+    data = stream.read() if hasattr(stream, "read") else b"".join(stream)
+    assert data.startswith(b"\x89PNG")
+    assert "png" in content_type
+    assert inline is True
+    assert filename.endswith(".png")
+
+
+def test_open_rejects_foreign_url(local_storage_env):
+    with pytest.raises(StorageError):
+        storage.open_file("https://evil.example/x.png")