curo1305
6fed5ba531
Research, pattern mapping, and verification complete. Walking Skeleton mode active (MVP Phase 1). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
37 KiB
37 KiB
phase, plan, type, wave, depends_on, files_modified, autonomous, requirements, user_setup, tags, must_haves
| phase | plan | type | wave | depends_on | files_modified | autonomous | requirements | user_setup | tags | must_haves | |||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 01-infrastructure-foundation | 04 | execute | 3 |
|
|
true |
|
|
|
Purpose: This is the heart of the walking skeleton. Without these modules, no document bytes can land in MinIO and no metadata can land in PostgreSQL. Plan 05 will compose this layer into the API + lifespan + Celery flow.
Output: Three new files under backend/storage/, one fully rewritten backend/services/storage.py, and all six Plan 02 tests/test_storage.py scaffolds flipping to PASSED.
<execution_context> @$HOME/.claude/get-shit-done/workflows/execute-plan.md @$HOME/.claude/get-shit-done/templates/summary.md </execution_context>
@CLAUDE.md @.planning/PROJECT.md @.planning/ROADMAP.md @.planning/STATE.md @.planning/phases/01-infrastructure-foundation/01-CONTEXT.md @.planning/phases/01-infrastructure-foundation/01-RESEARCH.md @.planning/phases/01-infrastructure-foundation/01-PATTERNS.md @.planning/phases/01-infrastructure-foundation/SKELETON.md @.planning/phases/01-infrastructure-foundation/01-03-SUMMARY.md @backend/ai/base.py @backend/ai/__init__.py @backend/ai/openai_provider.py Project's established ABC + factory pattern (from `backend/ai/base.py` + `backend/ai/__init__.py`, lines 1-36 of each — read both files before implementing):# backend/ai/base.py — pattern to mirror
class AIProvider(ABC):
@abstractmethod
async def classify(...) -> ClassificationResult: ...
@abstractmethod
async def health_check(self) -> bool: ...
# backend/ai/__init__.py — factory pattern to mirror
def get_provider(settings: dict) -> AIProvider:
...
match active:
case "openai":
return OpenAIProvider(...)
case _:
raise ValueError(...)
Existing services/storage.py public surface that api/documents.py and api/topics.py and services/classifier.py currently call (sync):
save_upload(file_bytes, original_name, mime_type) -> dict(returns{"id", "filename", "path"})save_metadata(meta: dict) -> Noneget_metadata(doc_id: str) -> dict | Nonelist_metadata(topic: str | None = None) -> list[dict]delete_document(doc_id: str) -> boolupdate_document_topics(doc_id: str, topics: list[str]) -> dict | Noneremove_topic_from_all_documents(topic_name: str) -> intload_topics() -> list[dict]save_topics(topics: list[dict]) -> Noneget_topic(topic_id: str) -> dict | Nonecreate_topic(name: str, description: str = "", color: str = "#6366f1") -> dictupdate_topic(topic_id: str, **kwargs) -> dict | Nonedelete_topic(topic_id: str) -> str | None(returns the deleted topic name)topic_doc_counts() -> dict[str, int]load_settings() -> dictsave_settings(settings: dict) -> Nonemask_api_key(key: str) -> strsettings_masked(settings: dict) -> dict
New services/storage.py MUST keep all of those names but accept an AsyncSession parameter where DB access is needed, except mask_api_key and settings_masked (pure functions — keep sync).
Schema fields the service layer reads/writes (declared in backend/db/models.py from Plan 03):
Document.id, user_id, folder_id, filename, object_key, content_type, size_bytes, storage_backend, extracted_text, status, created_at, updated_atTopic.id, user_id (NULLABLE — Phase 1 has no users), name, description, colorDocumentTopic.document_id, topic_id(composite PK association table)
Settings persistence note: Phase 1 keeps services/storage.load_settings() / save_settings() reading the flat file SETTINGS_FILE (/app/data/settings.json) because the users.ai_provider / users.ai_model schema cannot be populated until Phase 2. Document this with a # Phase 2 will migrate this to DB-backed per-user settings (D-03 deferred to user-scoped column population) comment.
Create `backend/storage/base.py` with the exact contents from RESEARCH.md Pattern 8 (lines 612-640): `from abc import ABC, abstractmethod`, `class StorageBackend(ABC):` with five `@abstractmethod async def` methods — `put_object(self, user_id: str, document_id: str, file_bytes: bytes, extension: str, content_type: str) -> str`, `get_object(self, object_key: str) -> bytes`, `delete_object(self, object_key: str) -> None`, `presigned_get_url(self, object_key: str, expires_minutes: int = 60) -> str`, `health_check(self) -> bool`. Each abstract method has a one-line docstring per RESEARCH.md.
Create `backend/storage/minio_backend.py` implementing `MinIOBackend(StorageBackend)`. Imports: `import asyncio`, `import io`, `import uuid`, `from datetime import timedelta`, `from minio import Minio`, `from storage.base import StorageBackend`. Constructor `def __init__(self, endpoint: str, access_key: str, secret_key: str, bucket: str, secure: bool = False)` stores `self._bucket = bucket` and `self._client = Minio(endpoint=endpoint, access_key=access_key, secret_key=secret_key, secure=secure)`.
Implement `async def put_object(self, user_id: str, document_id: str, file_bytes: bytes, extension: str, content_type: str) -> str`: construct `object_key = f"{user_id}/{document_id}/{uuid.uuid4()}{extension}"`. Build `data = io.BytesIO(file_bytes)` (the constructor sets pointer at 0 — no `seek(0)` needed per Pitfall 3, but add an explicit `data.seek(0)` immediately afterward as a belt-and-braces safety). Call `await asyncio.to_thread(self._client.put_object, self._bucket, object_key, data, length=len(file_bytes), content_type=content_type)`. Return `object_key`. IMPORTANT: do NOT accept any `filename`/`original_name` parameter — STORE-02 requires the human filename never reaches this method.
Implement `async def get_object(self, object_key: str) -> bytes`: define a sync helper `def _fetch():` inside the method that calls `response = self._client.get_object(self._bucket, object_key)` then `try: return response.read() finally: response.close(); response.release_conn()`. Call `return await asyncio.to_thread(_fetch)`. This pattern correctly handles MinIO's HTTPResponse-returning `get_object`.
Implement `async def delete_object(self, object_key: str) -> None`: `await asyncio.to_thread(self._client.remove_object, self._bucket, object_key)`.
Implement `async def presigned_get_url(self, object_key: str, expires_minutes: int = 60) -> str`: `return await asyncio.to_thread(self._client.presigned_get_object, self._bucket, object_key, timedelta(minutes=expires_minutes))`.
Implement `async def health_check(self) -> bool`: wrap `await asyncio.to_thread(self._client.bucket_exists, self._bucket)` in a `try/except Exception: return False`; the truthy path returns the exact boolean from the SDK.
Finally, after writing the implementation, edit the six tests in `backend/tests/test_storage.py` to REMOVE the `@pytest.mark.xfail(strict=False, reason="implemented in plan 04")` markers (the tests should now PASS unmarked) — but keep the existing `try/except ImportError: pytest.skip(...)` blocks because they harmlessly fall through when imports succeed.
Add a module-level `_storage = None` and a `def _backend(): global _storage; _storage = _storage or get_storage_backend(); return _storage` helper so the MinIO backend is lazily instantiated once per process (matching the singleton-like behavior of the previous module-level filelock objects).
Implement every function below as `async def` (except `mask_api_key`, `settings_masked` which stay sync). The session is the first positional parameter for every DB-touching function.
1. `async def save_upload(session: AsyncSession, file_bytes: bytes, original_name: str, mime_type: str) -> dict`: generate `doc_id = uuid.uuid4()`. Extract `suffix = Path(original_name).suffix.lower()`. Build `doc = Document(id=doc_id, user_id=None, filename=original_name, content_type=mime_type, size_bytes=len(file_bytes), storage_backend="minio", status="pending", object_key="")`. `session.add(doc)`. `await session.flush()` (to materialize `doc.id` without committing). Call `object_key = await _backend().put_object(user_id="null-user", document_id=str(doc_id), file_bytes=file_bytes, extension=suffix, content_type=mime_type)` — the literal sentinel `"null-user"` is used because there is no user in Phase 1 (D-03); Phase 2 will replace this with `str(current_user.id)`. Set `doc.object_key = object_key`. `await session.commit()`. Return `{"id": str(doc_id), "filename": original_name, "path": object_key, "object_key": object_key, "user_id": None}`.
2. `async def save_metadata(session: AsyncSession, meta: dict) -> None`: look up `doc = await session.get(Document, uuid.UUID(meta["id"]))`. If not found, return. Set `doc.extracted_text = meta.get("extracted_text", "")`. If `meta.get("topics")` is a non-empty list of strings, call `await update_document_topics(session, meta["id"], meta["topics"])` (which handles association table reconciliation). Set `doc.status = "classified" if meta.get("classified_at") else "pending"`. `await session.commit()`.
3. `async def get_metadata(session: AsyncSession, doc_id: str) -> dict | None`: try `uid = uuid.UUID(doc_id)` inside try/except `ValueError: return None` (keep the legacy string-doc-id tolerance from `api/documents.py`). `doc = await session.get(Document, uid)`. If `None`, return `None`. Load topics: `topics_q = await session.execute(select(Topic.name).join(DocumentTopic, DocumentTopic.topic_id == Topic.id).where(DocumentTopic.document_id == uid))`; `topic_names = [r[0] for r in topics_q]`. Return `{"id": str(doc.id), "original_name": doc.filename, "filename": doc.filename, "mime_type": doc.content_type, "size_bytes": doc.size_bytes, "extracted_text": doc.extracted_text or "", "topics": topic_names, "created_at": doc.created_at.isoformat() if doc.created_at else None, "classified_at": doc.updated_at.isoformat() if doc.status == "classified" else None}`.
4. `async def list_metadata(session: AsyncSession, topic: str | None = None) -> list[dict]`: base query `stmt = select(Document).order_by(Document.created_at.desc())`. If `topic` is provided, join `DocumentTopic` and `Topic` and filter `Topic.name == topic`. Execute, then for each `Document` call the same dict-build code as `get_metadata` (factor it out into a private `def _doc_to_dict(doc, topic_names)` helper).
5. `async def delete_document(session: AsyncSession, doc_id: str) -> bool`: lookup as above; if not found, return `False`. Call `await _backend().delete_object(doc.object_key)` inside try/except (log to stderr on failure but still proceed with DB delete — the bytes may already be gone). `await session.delete(doc)`. `await session.commit()`. Return `True`.
6. `async def update_document_topics(session: AsyncSession, doc_id: str, topics: list[str]) -> dict | None`: lookup the doc; return `None` if missing. Delete all existing `DocumentTopic` rows for this document via `await session.execute(delete(DocumentTopic).where(DocumentTopic.document_id == uid))`. For each name in `topics` (deduped), call `topic_dict = await create_topic(session, name)` (auto-create if missing) and insert a `DocumentTopic(document_id=uid, topic_id=uuid.UUID(topic_dict["id"]))`. Set `doc.status = "classified"`. `await session.commit()`. Return `await get_metadata(session, doc_id)`.
7. `async def remove_topic_from_all_documents(session: AsyncSession, topic_name: str) -> int`: find the topic by name; if missing, return 0. Run `result = await session.execute(delete(DocumentTopic).where(DocumentTopic.topic_id == topic.id))`. `await session.commit()`. Return `result.rowcount`.
8. `async def load_topics(session: AsyncSession) -> list[dict]`: `q = await session.execute(select(Topic).order_by(Topic.name))`. Return `[{"id": str(t.id), "name": t.name, "description": t.description, "color": t.color} for t in q.scalars()]`.
9. `async def save_topics(session: AsyncSession, topics: list[dict]) -> None`: idempotent bulk replace — delete all `Topic` rows, then insert the provided list. (Phase 1 use: not directly called by any current endpoint; preserved for legacy compatibility — add a `# legacy: not used by current endpoints` comment.)
10. `async def get_topic(session: AsyncSession, topic_id: str) -> dict | None`: `t = await session.get(Topic, uuid.UUID(topic_id))`; return `None` or `{"id": str(t.id), ...}`.
11. `async def create_topic(session: AsyncSession, name: str, description: str = "", color: str = "#6366f1") -> dict`: case-insensitive deduplication — `q = await session.execute(select(Topic).where(sql_func.lower(Topic.name) == name.lower()))`. If a row exists, return its dict shape. Otherwise insert a new `Topic(name=name, description=description, color=color)`, commit, return dict.
12. `async def update_topic(session: AsyncSession, topic_id: str, name=None, description=None, color=None) -> dict | None`: lookup; if missing, `None`. Update only non-None fields. Commit. Return dict.
13. `async def delete_topic(session: AsyncSession, topic_id: str) -> str | None`: lookup; if missing, `None`. Save the name. Delete the Topic (cascade removes DocumentTopic rows via `ondelete="CASCADE"`). Commit. Return the saved name.
14. `async def topic_doc_counts(session: AsyncSession) -> dict[str, int]`: `q = await session.execute(select(Topic.name, sql_func.count(DocumentTopic.document_id)).join(DocumentTopic, DocumentTopic.topic_id == Topic.id, isouter=True).group_by(Topic.name))`. Return `{name: count for name, count in q}`.
15. `def load_settings() -> dict`: keep sync; read `SETTINGS_FILE` (still flat JSON) — if file missing, return a deep copy of `DEFAULT_SETTINGS`. Add comment `# Phase 2 will move per-user settings to users.ai_provider / users.ai_model`.
16. `def save_settings(settings: dict) -> None`: keep sync; write `SETTINGS_FILE`. No filelock — Phase 1 settings file is single-writer.
17. `def mask_api_key(key: str) -> str` and `def settings_masked(settings: dict) -> dict`: keep verbatim from current file (pure functions).
At the bottom of the file, expose every function via explicit `__all__ = ["save_upload", "save_metadata", "get_metadata", "list_metadata", "delete_document", "update_document_topics", "remove_topic_from_all_documents", "load_topics", "save_topics", "get_topic", "create_topic", "update_topic", "delete_topic", "topic_doc_counts", "load_settings", "save_settings", "mask_api_key", "settings_masked"]` so importers see the public surface.
<threat_model>
Trust Boundaries
| Boundary | Description |
|---|---|
services/storage.py → storage/minio_backend.py |
Internal Python call; bytes never cross a network boundary inside this module |
storage/minio_backend.py → MinIO server |
App connects with MINIO_ACCESS_KEY (app-level credentials, not root) over Docker internal HTTP |
services/storage.py → PostgreSQL via AsyncSession |
All queries parameterized via SQLAlchemy ORM (SEC-03) |
STRIDE Threat Register
| Threat ID | Category | Component | Disposition | Mitigation Plan |
|---|---|---|---|---|
| T-01-04-01 | Tampering | Object key prediction / path traversal | mitigate | MinIOBackend.put_object constructs keys server-side as {user_id}/{document_id}/{uuid4()}{ext} (STORE-02, D-06); no caller can inject a key. The extension is derived from Path(original_name).suffix.lower() — the only piece of user input — and is concatenated as a suffix to a freshly generated UUID, so a malicious extension cannot escape the prefix (the leading {user_id}/{document_id}/{uuid4} portion is always server-controlled). Wave 0 tests test_object_key_schema + test_filename_not_in_object_key enforce this on every CI run. |
| T-01-04-02 | Information Disclosure | Human filename leaking into the object store namespace | mitigate | MinIOBackend.put_object does NOT accept a filename parameter (only extension). The human filename is stored only in the documents.filename DB column (D-06). Test test_filename_not_in_object_key asserts this on every run. |
| T-01-04-03 | Information Disclosure | Extension as a covert channel (e.g., .html for XSS via download) |
accept | Phase 1 has no download flow exposed to the browser (no presigned URL endpoint, no proxied download endpoint). Phase 4 (DOC-02) will introduce PDF-only proxied preview and content-type validation. Documented in SKELETON.md. |
| T-01-04-04 | Tampering | SQL injection via ORM string concatenation | mitigate | All queries use select(...), delete(...), session.execute(...) with parameterized values — never f-strings or .format() for SQL. The only string interpolation is the MinIO object key, which is not SQL. SEC-03 / CLAUDE.md. |
| T-01-04-05 | Denial of Service | Unbounded list_metadata returning all documents | accept | Phase 1 has only the single test user's documents (≤ 100 MB of uploads from D-04 reset state). Phase 3 (STORE-03) introduces pagination + quota enforcement. The existing pagination in api/documents.py (page, per_page) remains in effect — Plan 05 will wire it. |
| T-01-04-SC | Tampering | npm/pip/cargo installs | N/A | No new installs in this plan; uses minio/sqlalchemy from Plan 01. RESEARCH.md Package Legitimacy Audit covers both (slopcheck OK). |
| </threat_model> |
<success_criteria>
backend/storage/base.py,backend/storage/__init__.py,backend/storage/minio_backend.pyexist; the ABC + factory pattern mirrorsbackend/ai/exactly.backend/services/storage.pyis rewritten with async signatures, ORM access, and the MinIO backend integration;filelockis gone from this module.- All six Plan 02
tests/test_storage.pytests pass (xfail markers removed). - The legacy
tests/test_documents.pysync tests continue to pass — the cutover happens in Plan 05; this plan is additive on the storage layer side. </success_criteria>