diff --git a/routes/personal_routes.py b/routes/personal_routes.py
index b9ba0a7b9..77526c1d1 100644
--- a/routes/personal_routes.py
+++ b/routes/personal_routes.py
@@ -8,7 +8,7 @@ from fastapi import APIRouter, HTTPException, Query, Request, UploadFile, File,
 from src.request_models import DirectoryRequest
 from core.constants import BASE_DIR, PERSONAL_DIR
 from src.rag_singleton import get_rag_manager
-from src.auth_helpers import get_current_user, require_user
+from src.auth_helpers import require_privilege, require_user
 from core.middleware import require_admin
 from src.upload_handler import secure_filename
 
@@ -194,7 +194,7 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
     @router.post("/upload")
     async def upload_files_to_rag(request: Request, files: List[UploadFile] = File(...)):
         """Upload files directly into RAG. Supports text and PDF."""
-        user = get_current_user(request)
+        user = require_privilege(request, "can_use_documents")
         rag = _rag()
         if not rag:
             raise HTTPException(503, "RAG system is not available — is the embedding service running?")
diff --git a/tests/test_personal_upload_privilege.py b/tests/test_personal_upload_privilege.py
new file mode 100644
index 000000000..88d8a2f31
--- /dev/null
+++ b/tests/test_personal_upload_privilege.py
@@ -0,0 +1,98 @@
+import asyncio
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from fastapi import HTTPException
+
+from routes import personal_routes
+
+
+def _upload_endpoint():
+    router = personal_routes.setup_personal_routes(_FakePersonalDocs(), None, True)
+    for route in router.routes:
+        if getattr(route, "path", "") == "/api/personal/upload" and "POST" in getattr(route, "methods", set()):
+            return route.endpoint
+    raise AssertionError("upload endpoint not found")
+
+
+def _request(privileges):
+    class _AuthManager:
+        def get_privileges(self, user):
+            assert user == "alice"
+            return privileges
+
+    return SimpleNamespace(
+        state=SimpleNamespace(current_user="alice"),
+        app=SimpleNamespace(
+            state=SimpleNamespace(
+                auth_manager=_AuthManager(),
+            ),
+        ),
+        client=SimpleNamespace(host="203.0.113.10"),
+    )
+
+
+class _FakePersonalDocs:
+    def __init__(self):
+        self.added = []
+
+    def add_directory(self, directory, index=False):
+        self.added.append((directory, index))
+
+
+class _FakeRAG:
+    def __init__(self):
+        self.docs = []
+
+    def _split_into_chunks(self, text, chunk_size=500):
+        return [text]
+
+    def add_document(self, chunk, metadata):
+        self.docs.append((chunk, metadata))
+        return True
+
+
+class _Upload:
+    filename = "notes.txt"
+
+    async def read(self, limit):
+        return b"hello from upload"
+
+
+def test_personal_upload_requires_document_privilege(monkeypatch):
+    monkeypatch.setenv("AUTH_ENABLED", "true")
+    monkeypatch.setattr(
+        personal_routes,
+        "get_rag_manager",
+        lambda: pytest.fail("RAG must not be touched before privilege passes"),
+    )
+
+    endpoint = _upload_endpoint()
+
+    with pytest.raises(HTTPException) as exc:
+        asyncio.run(endpoint(request=_request({"can_use_documents": False}), files=[]))
+
+    assert exc.value.status_code == 403
+
+
+def test_personal_upload_indexes_with_privileged_owner(tmp_path, monkeypatch):
+    monkeypatch.setenv("AUTH_ENABLED", "true")
+    monkeypatch.setattr(personal_routes, "UPLOADS_DIR", str(tmp_path))
+    rag = _FakeRAG()
+    monkeypatch.setattr(personal_routes, "get_rag_manager", lambda: rag)
+
+    endpoint = _upload_endpoint()
+    result = asyncio.run(
+        endpoint(
+            request=_request({"can_use_documents": True}),
+            files=[_Upload()],
+        )
+    )
+
+    assert result["success"] is True
+    assert result["indexed_count"] == 1
+    assert rag.docs[0][0] == "hello from upload"
+    metadata = rag.docs[0][1]
+    assert metadata["owner"] == "alice"
+    assert Path(metadata["directory"]).name == "alice"