docs: add attachment access control implementation plan

Task breakdown covering upload hardening, MinIO prefix policy,
admin JWT proxy, web-admin opener, and mobile local preview.

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

docs/superpowers/plans/2026-09-09-attachment-access-control.md

@@ -0,0 +1,599 @@
+# Attachment Access Control Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Make feedback/suggestion attachments private (JWT proxy for admin, local preview on mobile) while keeping `qrcode/*` anonymously readable, and harden uploads against executable/malware disguise.
+
+**Architecture:** Same MinIO bucket with prefix policy (`qrcode/*` public only). Store URLs unchanged. Admin downloads via authenticated streaming proxy. Mobile preview uses local blob/temp paths only. Upload validation adds magic-byte checks; public upload folders are whitelisted.
+
+**Tech Stack:** FastAPI, MinIO SDK, pytest, Vue 3 + Element Plus + axios, uni-app (mobile-wx)
+
+**Spec:** `docs/superpowers/specs/2026-09-09-attachment-access-control-design.md`
+
+## Global Constraints
+
+- Same bucket, prefix split — do **not** introduce a second bucket
+- Admin access via **JWT streaming proxy**, not presigned URLs
+- Do **not** migrate DB attachment URL strings
+- Public upload folders: only `feedback`, `suggestion`
+- No antivirus/sandbox/zip deep scan this iteration
+- Minimal diffs; match existing ajyy code style (Chinese error messages, `AppHTTPError`, `success_response`)
+
+## File Structure
+
+| File | Responsibility |
+|------|----------------|
+| `backend/app/core/upload_validate.py` | Extension whitelist + size + magic-byte checks; image-only helper for qrcode |
+| `backend/app/core/storage.py` | Prefix public policy; `resolve_object_key`; `open_file` stream; local `0o644` |
+| `backend/app/api/public.py` | Enforce public folder whitelist |
+| `backend/app/api/admin.py` | Attachment proxy endpoint; qrcode folder image-only validation |
+| `backend/tests/test_upload_validate.py` | Magic / reject exe tests |
+| `backend/tests/test_minio_public_policy.py` | Expect `qrcode/*` resource only |
+| `backend/tests/test_attachment_proxy.py` | Auth + path validation + local stream |
+| `backend/tests/test_upload.py` | Folder whitelist cases |
+| `web-admin/src/api/admin.ts` | `openAdminAttachment(url)` blob helper |
+| `web-admin/src/views/feedback/index.vue` | Use proxy opener |
+| `web-admin/src/views/suggestion/index.vue` | Show attachments + proxy opener |
+| `mobile-wx/utils/upload.js` | Return `previewUrl` + `url` |
+| `mobile-wx/pages/feedback/index.vue` | Bind `previewUrl` |
+| `mobile-wx/pages/suggest/index.vue` | Bind `previewUrl` |
+
+---
+
+### Task 1: Upload magic-byte validation
+
+**Files:**
+- Modify: `backend/app/core/upload_validate.py`
+- Create: `backend/tests/test_upload_validate.py`
+- Test: `backend/tests/test_upload_validate.py`
+
+**Interfaces:**
+- Consumes: existing `validate_upload(filename, content)`, `ALLOWED_EXTENSIONS`, `MAX_UPLOAD_BYTES`
+- Produces: `validate_upload` also rejects magic mismatches; `validate_image_upload(filename, content)` for qrcode-only image types; `IMAGE_EXTENSIONS` set
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `backend/tests/test_upload_validate.py`:
+
+```python
+import pytest
+from app.core.response import AppHTTPError
+from app.core.upload_validate import validate_upload, validate_image_upload
+
+PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16
+JPEG = b"\xff\xd8\xff" + b"\x00" * 16
+PDF = b"%PDF-1.4" + b"\x00" * 16
+EXE_MZ = b"MZ" + b"\x00" * 64
+
+
+def test_rejects_exe_extension():
+    with pytest.raises(AppHTTPError) as ei:
+        validate_upload("malware.exe", EXE_MZ)
+    assert ei.value.status_code == 400
+
+
+def test_rejects_png_extension_with_exe_magic():
+    with pytest.raises(AppHTTPError) as ei:
+        validate_upload("fake.png", EXE_MZ)
+    assert ei.value.status_code == 400
+    assert "类型" in ei.value.message or "格式" in ei.value.message or "内容" in ei.value.message
+
+
+def test_accepts_real_png():
+    validate_upload("a.png", PNG)
+
+
+def test_accepts_jpeg_and_pdf():
+    validate_upload("a.jpg", JPEG)
+    validate_upload("a.pdf", PDF)
+
+
+def test_image_upload_rejects_pdf():
+    with pytest.raises(AppHTTPError):
+        validate_image_upload("x.pdf", PDF)
+
+
+def test_image_upload_accepts_png():
+    validate_image_upload("x.png", PNG)
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run: `cd backend && . venv/bin/activate && pytest tests/test_upload_validate.py -v`  
+Expected: FAIL (magic mismatch / `validate_image_upload` missing)
+
+- [ ] **Step 3: Implement validation**
+
+In `upload_validate.py`:
+
+- Keep `ALLOWED_EXTENSIONS` as today (no exe/js/html — already excluded by whitelist)
+- Add `IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"}`
+- Add `_MAGIC_CHECKS`: map ext → list of callables or prefix bytes:
+  - png: starts with `\x89PNG\r\n\x1a\n`
+  - jpg/jpeg: starts with `\xff\xd8\xff`
+  - gif: `GIF87a` or `GIF89a`
+  - webp: `RIFF` at 0 and `WEBP` at 8
+  - bmp: `BM`
+  - pdf: `%PDF`
+- For extensions **with** a magic rule: if content does not match → `AppHTTPError(400, "文件内容与扩展名不符")`
+- For extensions **without** magic rule (video, office, zip): extension + size only (YAGNI)
+- `validate_image_upload`: require ext in `IMAGE_EXTENSIONS`, then call `validate_upload`
+
+Check how `AppHTTPError` exposes status — use the same attribute existing tests use (`status_code` or similar). If `AppHTTPError` uses different fields, match `app/core/response.py`.
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run: `pytest tests/test_upload_validate.py -v`  
+Expected: PASS
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add backend/app/core/upload_validate.py backend/tests/test_upload_validate.py
+git commit -m "feat(backend): validate upload magic bytes and image-only helper"
+```
+
+---
+
+### Task 2: MinIO qrcode-only public policy + storage open helpers
+
+**Files:**
+- Modify: `backend/app/core/storage.py`
+- Modify: `backend/tests/test_minio_public_policy.py`
+- Test: `backend/tests/test_minio_public_policy.py` (extend with resolve/open local tests in same file or `tests/test_storage_open.py`)
+
+**Interfaces:**
+- Consumes: `MinIOStorage`, `LocalStorage`, `get_storage`, `get_settings`
+- Produces:
+  - Policy Resource: `arn:aws:s3:::{bucket}/qrcode/*` only
+  - `resolve_object_key(url: str) -> str` (module-level or on storage)
+  - `open_stored_file(url: str) -> tuple[BinaryIO|Iterator, str content_type, str filename, bool inline]`  
+    Prefer module functions: `storage.resolve_object_key(url) -> str` and `storage.open_file(url) -> ...` that dispatch on `get_storage()`
+
+- [ ] **Step 1: Update failing policy test**
+
+Change `tests/test_minio_public_policy.py` expected Resource to:
+
+```python
+assert policy["Statement"][0]["Resource"] == ["arn:aws:s3:::ajyy/qrcode/*"]
+```
+
+Update docstring to say only qrcode prefix is public.
+
+Add `tests/test_storage_open.py`:
+
+```python
+from pathlib import Path
+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")
+```
+
+- [ ] **Step 2: Run tests — expect FAIL**
+
+Run: `pytest tests/test_minio_public_policy.py tests/test_storage_open.py -v`  
+Expected: policy assertion fail; `resolve_object_key` / `open_file` missing
+
+- [ ] **Step 3: Implement**
+
+In `storage.py`:
+
+1. `_ensure_public_read_policy`: Resource → `f"arn:aws:s3:::{self.bucket_name}/qrcode/*"`; update comment (only qrcode anonymous)
+2. `LocalStorage.upload_file`: after write, `os.chmod(dest, 0o644)`
+3. Add module helpers:
+
+```python
+def resolve_object_key(url: str) -> str:
+    """Extract object key from stored url; raise StorageError if illegal."""
+    ...
+
+def open_file(url: str):
+    """Return (stream, content_type, download_name, inline)."""
+    ...
+```
+
+`resolve_object_key` rules:
+
+- Strip query string
+- If starts with `/uploads/`: key = remainder; reject `..` segments
+- If http(s): must host-match configured MinIO base (endpoint or `MINIO_BASE_URL`) and path contain `/{bucket}/`; key = after bucket; reject otherwise
+- Empty → StorageError
+
+`open_file`:
+
+- key = resolve_object_key(url)
+- Guess content_type via mimetypes; force `application/octet-stream` if guessed type looks executable (`application/x-msdownload`, `application/x-msdos-program`, etc.)
+- `inline = True` iff extension in image set (same as IMAGE_EXTENSIONS) else False
+- Local: open file from `LOCAL_STORAGE_DIR / key` after `_ensure_within_base`
+- MinIO: `client.get_object(bucket, key)` → response stream
+
+Also add thin wrappers used by API later if preferred: keep logic in `storage.py`.
+
+- [ ] **Step 4: Run tests — expect PASS**
+
+Run: `pytest tests/test_minio_public_policy.py tests/test_storage_open.py -v`
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add backend/app/core/storage.py backend/tests/test_minio_public_policy.py backend/tests/test_storage_open.py
+git commit -m "feat(backend): restrict MinIO public read to qrcode prefix and add open_file"
+```
+
+**Ops note (do in this task after code works):** Restart backend against the test MinIO so policy is overwritten; or run a one-liner `set_bucket_policy` for `qrcode/*` only. Verify anonymous GET on a `feedback/` object returns 403 and `qrcode/` returns 200.
+
+---
+
+### Task 3: Public upload folder whitelist + admin qrcode image-only
+
+**Files:**
+- Modify: `backend/app/api/public.py`
+- Modify: `backend/app/api/admin.py` (upload path only)
+- Modify: `backend/tests/test_upload.py`
+
+**Interfaces:**
+- Consumes: `validate_upload`, `validate_image_upload`, `storage.upload_file`
+- Produces: public upload accepts only folders `feedback` / `suggestion` (after normalize); admin upload with `folder=qrcode` calls `validate_image_upload`
+
+- [ ] **Step 1: Write failing tests in `test_upload.py`**
+
+```python
+def test_public_upload_rejects_qrcode_folder(client, local_storage_env):
+    files = {"file": ("a.png", b"\x89PNG\r\n\x1a\n" + b"\x00" * 8, "image/png")}
+    r = client.post(
+        "/api/public/attachment/upload",
+        files=files,
+        data={"folder": "qrcode"},
+    )
+    assert r.status_code == 400
+    assert r.json()["code"] == 400
+
+
+def test_public_upload_allows_feedback_folder(client, local_storage_env):
+    files = {"file": ("a.png", b"\x89PNG\r\n\x1a\n" + b"\x00" * 8, "image/png")}
+    r = client.post(
+        "/api/public/attachment/upload",
+        files=files,
+        data={"folder": "feedback"},
+    )
+    assert r.status_code == 200
+    assert "/feedback/" in r.json()["data"]["url"] or r.json()["data"]["url"].startswith("/uploads/feedback/")
+```
+
+Existing png tests may need real PNG magic bytes if they still use `b"fake-png-bytes"` — update those fixtures to valid PNG headers so Task 1 does not break them.
+
+- [ ] **Step 2: Run — expect FAIL**
+
+Run: `pytest tests/test_upload.py -v`
+
+- [ ] **Step 3: Implement**
+
+In `public.py` before upload:
+
+```python
+PUBLIC_UPLOAD_FOLDERS = {"feedback", "suggestion"}
+# normalize via storage._normalize_folder or duplicate check
+folder_norm = ...
+if folder_norm not in PUBLIC_UPLOAD_FOLDERS:
+    raise AppHTTPError(400, "不允许的上传目录")
+```
+
+Empty folder → 400 (do not allow date-only root).
+
+In `admin.py` `admin_upload`: if normalized folder == `qrcode`, call `validate_image_upload` instead of `validate_upload`; else `validate_upload`.
+
+- [ ] **Step 4: Run — expect PASS**
+
+Run: `pytest tests/test_upload.py tests/test_upload_validate.py -v`
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add backend/app/api/public.py backend/app/api/admin.py backend/tests/test_upload.py
+git commit -m "feat(backend): whitelist public upload folders and image-only qrcode"
+```
+
+---
+
+### Task 4: Admin JWT attachment proxy
+
+**Files:**
+- Modify: `backend/app/api/admin.py`
+- Create: `backend/tests/test_attachment_proxy.py`
+
+**Interfaces:**
+- Consumes: `storage.open_file`, `storage.StorageError`, `get_current_user`
+- Produces: `GET /api/web/admin/attachment/file?url=...` → streaming file with safe headers
+
+- [ ] **Step 1: Write failing API tests**
+
+```python
+def _token(client):
+    r = client.post("/api/web/auth/login", json={"username": "admin", "password": "Admin@123"})
+    return r.json()["data"]["token"]
+
+
+def test_attachment_proxy_requires_auth(client, local_storage_env):
+    # ensure local storage for this client — may need fixture that patches env before app storage calls
+    r = client.get("/api/web/admin/attachment/file", params={"url": "/uploads/feedback/x.png"})
+    assert r.status_code == 401
+
+
+def test_attachment_proxy_streams_file(client, local_storage_env, monkeypatch):
+    from app.core import storage
+    from app.core.config import get_settings
+    get_settings.cache_clear()
+    stored = storage.upload_file(b"\x89PNG\r\n\x1a\n" + b"\x00" * 8, "a.png", "feedback")
+    token = _token(client)
+    r = client.get(
+        "/api/web/admin/attachment/file",
+        params={"url": stored["url"]},
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert r.status_code == 200
+    assert r.headers.get("x-content-type-options", "").lower() == "nosniff"
+    assert "image" in (r.headers.get("content-type") or "")
+    assert r.content.startswith(b"\x89PNG")
+
+
+def test_attachment_proxy_rejects_foreign_url(client, local_storage_env):
+    token = _token(client)
+    r = client.get(
+        "/api/web/admin/attachment/file",
+        params={"url": "https://evil.example/a.png"},
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert r.status_code == 400
+```
+
+**Note:** `client` fixture uses whatever `STORAGE_TYPE` is in env. Prefer a dedicated fixture that forces `STORAGE_TYPE=local` and `LOCAL_STORAGE_DIR` for these tests (copy pattern from `test_upload.py` `local_storage_env`), and ensure `get_settings.cache_clear()` so uploads hit local disk.
+
+- [ ] **Step 2: Run — expect FAIL (404 route)**
+
+Run: `pytest tests/test_attachment_proxy.py -v`
+
+- [ ] **Step 3: Implement endpoint**
+
+In `admin.py`:
+
+```python
+@router.get("/attachment/file")
+def download_attachment_file(
+    url: str = Query(...),
+    user: Annotated[SysUser, Depends(get_current_user)],
+):
+    try:
+        stream, content_type, filename, inline = storage.open_file(url)
+    except StorageError as exc:
+        raise AppHTTPError(400, exc.message) from exc
+    # log: user.id, object key
+    disposition = "inline" if inline else "attachment"
+    headers = {
+        "X-Content-Type-Options": "nosniff",
+        "Content-Disposition": f'{disposition}; filename="{filename}"',
+    }
+    return StreamingResponse(stream, media_type=content_type, headers=headers)
+```
+
+Import `StreamingResponse` from `starlette.responses` / `fastapi.responses`.  
+On MinIO stream, ensure body is closed after send if SDK requires it (generator with try/finally).
+
+- [ ] **Step 4: Run — expect PASS**
+
+Run: `pytest tests/test_attachment_proxy.py -v`
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add backend/app/api/admin.py backend/tests/test_attachment_proxy.py
+git commit -m "feat(backend): add JWT-authenticated attachment file proxy"
+```
+
+---
+
+### Task 5: web-admin open attachments via proxy
+
+**Files:**
+- Modify: `web-admin/src/api/admin.ts`
+- Modify: `web-admin/src/views/feedback/index.vue`
+- Modify: `web-admin/src/views/suggestion/index.vue`
+
+**Interfaces:**
+- Consumes: `GET /api/web/admin/attachment/file?url=`
+- Produces: `openAdminAttachment(url: string): Promise<void>`
+
+- [ ] **Step 1: Add API helper**
+
+In `admin.ts`:
+
+```typescript
+import http from '@/utils/http'
+import axios from 'axios'
+
+export async function openAdminAttachment(url: string) {
+  const token = localStorage.getItem('ajyy_token')
+  const base = import.meta.env.VITE_API_BASE_URL || '/api'
+  const res = await axios.get(`${base}/web/admin/attachment/file`, {
+    params: { url },
+    headers: token ? { Authorization: `Bearer ${token}` } : {},
+    responseType: 'blob'
+  })
+  const blobUrl = URL.createObjectURL(res.data)
+  window.open(blobUrl, '_blank', 'noopener,noreferrer')
+  // revoke later
+  setTimeout(() => URL.revokeObjectURL(blobUrl), 60_000)
+}
+```
+
+Prefer reusing axios instance with `responseType: 'blob'` carefully: default `http` interceptor expects JSON `code`. **Must** use raw axios (or a dedicated call that skips JSON envelope parse) because proxy returns raw bytes, not `{code:200}`.
+
+- [ ] **Step 2: Wire feedback detail**
+
+Replace bare `<a :href="u">` with:
+
+```vue
+<a href="#" @click.prevent="openAdminAttachment(u)">附件{{ i + 1 }}</a>
+```
+
+Import `openAdminAttachment` from `@/api/admin`.
+
+- [ ] **Step 3: Wire suggestion detail**
+
+In `suggestion/index.vue` descriptions, add attachments block like feedback (API already returns `attachmentUrls`). Same click handler.
+
+- [ ] **Step 4: Manual check**
+
+With backend on local storage or MinIO: login → open feedback with attachment → file opens; logged-out request to proxy fails.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add web-admin/src/api/admin.ts web-admin/src/views/feedback/index.vue web-admin/src/views/suggestion/index.vue
+git commit -m "feat(web-admin): open feedback/suggestion attachments via auth proxy"
+```
+
+---
+
+### Task 6: mobile-wx local previewUrl
+
+**Files:**
+- Modify: `mobile-wx/utils/upload.js`
+- Modify: `mobile-wx/pages/feedback/index.vue`
+- Modify: `mobile-wx/pages/suggest/index.vue`
+
+**Interfaces:**
+- Consumes: existing choose/upload flows
+- Produces: each item `{ previewUrl, url, fileName, size? }`; UI uses `previewUrl` only
+
+- [ ] **Step 1: Change `chooseAndUploadFiles` return shape**
+
+H5 path after picking `file`:
+
+```javascript
+const previewUrl = URL.createObjectURL(file)
+const data = await uploadBlobAsFile(file, folder)
+items.push({
+  previewUrl,
+  url: data.url,
+  fileName: data.fileName || file.name,
+  size: data.size
+})
+```
+
+App/`chooseFile` / `chooseImage` paths:
+
+```javascript
+items.push({
+  previewUrl: path, // tempFilePath
+  url: data.url,
+  fileName: ...
+})
+```
+
+Export a `revokePreview(item)` that if `previewUrl.startsWith('blob:')` calls `URL.revokeObjectURL`.
+
+- [ ] **Step 2: Update pages**
+
+`feedback/index.vue` / `suggest/index.vue`:
+
+```vue
+<image v-if="isImage(item)" :src="item.previewUrl" mode="aspectFill" />
+```
+
+On `removeFile(i)`: revoke then splice.  
+On successful submit / `beforeDestroy`/`unmounted`: revoke all.
+
+Submit still: `attachmentUrls: this.attachments.map((a) => a.url)`.
+
+- [ ] **Step 3: Manual check**
+
+Upload image on feedback page — thumbnail shows from blob/temp; DevTools Network must **not** need anonymous MinIO GET for that thumbnail.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add mobile-wx/utils/upload.js mobile-wx/pages/feedback/index.vue mobile-wx/pages/suggest/index.vue
+git commit -m "feat(mobile-wx): preview uploads from local paths only"
+```
+
+---
+
+### Task 7: Final verification
+
+- [ ] **Step 1: Run full backend tests**
+
+```bash
+cd backend && . venv/bin/activate && pytest -q
+```
+
+Expected: all green
+
+- [ ] **Step 2: Acceptance smoke (from spec §7)**
+
+| Check | How |
+|-------|-----|
+| Anonymous GET feedback object | `curl -sI <minio>/ajyy/feedback/...` → 403 |
+| Anonymous GET qrcode object | → 200 |
+| Proxy without token | → 401 |
+| Proxy with token | → 200 body |
+| Upload exe | → 400 |
+| Fake png (MZ bytes) | → 400 |
+| Admin UI attachment | opens when logged in |
+| Mobile preview | local only |
+| Qrcode admin preview | still works |
+
+- [ ] **Step 3: Commit any leftover doc tweaks only if needed**
+
+Optional one-liner in `docs/deploy-test.md`: note `qrcode/*` public policy. Only if you touch deploy docs.
+
+---
+
+## Spec coverage checklist
+
+| Spec item | Task |
+|-----------|------|
+| `qrcode/*` public policy | Task 2 |
+| Private feedback/suggestion objects | Task 2 + ops |
+| Public folder whitelist | Task 3 |
+| Magic bytes / non-exec | Task 1 |
+| Local `0o644` | Task 2 |
+| Admin JWT proxy + headers | Task 4 |
+| web-admin blob open | Task 5 |
+| Suggestion attachments UI | Task 5 |
+| Mobile local preview | Task 6 |
+| Qrcode image-only upload | Task 3 |
+| No DB migration / no AV | N/A (explicit non-goals) |