2026-09-09-attachment-access-control.md 20 KB

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:

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
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:

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:

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:
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
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

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:

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
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

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:

@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
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:

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:

<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
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:

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:

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:

<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
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
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)