b3nw
950a8fc441
Build and Push Outline MCP Docker Image / build (push) Successful in 14s
Implement 3 new MCP tools: - attach_file_to_document(document_id, file_path) - upload_image_to_document(image_path, document_id, alt_text) - import_file_to_outline(file_path, collection_id, parent_document_id) Security: - Restrict file access to /tmp via _validate_file_path with realpath - 50MB max file size enforced client-side - Symlink traversal blocked Technical: - Extract shared _upload_attachment() helper - Stream files to presigned URLs instead of loading into memory - Add combined lifespan to close OutlineClient on shutdown - Update CI workflow with modern action versions and PR triggers Tests: - Add 28 tests covering path validation, size limits, upload flow, error handling, symlink traversal, and multipart imports
435 lines
16 KiB
Python
435 lines
16 KiB
Python
import os
|
|
import sys
|
|
import json
|
|
import pytest
|
|
import tempfile
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
# Set env vars before importing server
|
|
os.environ["OUTLINE_API_URL"] = "https://test.example.com"
|
|
os.environ["OUTLINE_API_TOKEN"] = "test-token"
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from server import (
|
|
_validate_file_path,
|
|
_get_content_type,
|
|
OutlineClient,
|
|
attach_file_to_document,
|
|
upload_image_to_document,
|
|
import_file_to_outline,
|
|
outline_client,
|
|
MAX_FILE_SIZE,
|
|
)
|
|
|
|
|
|
class TestValidateFilePath:
|
|
def test_valid_file_under_tmp(self, tmp_path):
|
|
test_file = tmp_path / "test.txt"
|
|
test_file.write_text("hello")
|
|
result = _validate_file_path(str(test_file))
|
|
assert result == str(test_file)
|
|
|
|
def test_path_outside_tmp_rejected(self):
|
|
with pytest.raises(ValueError, match="must be within"):
|
|
_validate_file_path("/etc/passwd")
|
|
|
|
def test_path_traversal_rejected(self):
|
|
with pytest.raises(ValueError, match="must be within"):
|
|
_validate_file_path("/tmp/../../etc/passwd")
|
|
|
|
def test_nonexistent_file_rejected(self, tmp_path):
|
|
with pytest.raises(ValueError, match="File not found"):
|
|
_validate_file_path(str(tmp_path / "nonexistent.txt"))
|
|
|
|
def test_directory_rejected(self, tmp_path):
|
|
with pytest.raises(ValueError, match="not a file"):
|
|
_validate_file_path(str(tmp_path))
|
|
|
|
def test_symlink_traversal_rejected(self, tmp_path):
|
|
evil_link = tmp_path / "evil"
|
|
evil_link.symlink_to("/etc/passwd")
|
|
with pytest.raises(ValueError, match="must be within"):
|
|
_validate_file_path(str(evil_link))
|
|
|
|
|
|
class TestGetContentType:
|
|
def test_known_image_types(self):
|
|
assert _get_content_type("test.png") == "image/png"
|
|
assert _get_content_type("test.jpg") == "image/jpeg"
|
|
assert _get_content_type("test.jpeg") == "image/jpeg"
|
|
|
|
def test_known_document_types(self):
|
|
assert _get_content_type("test.pdf") == "application/pdf"
|
|
assert _get_content_type("test.md") == "text/markdown"
|
|
assert _get_content_type("test.txt") == "text/plain"
|
|
|
|
def test_unknown_type_returns_none(self):
|
|
assert _get_content_type("test.unknownxyz") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestOutlineClientUpload:
|
|
async def test_upload_to_presigned_url_success(self):
|
|
client = OutlineClient("https://api.example.com", "token")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_response.status_code = 200
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_client = AsyncMock()
|
|
mock_client.put = AsyncMock(return_value=mock_response)
|
|
mock_client_class.return_value.__aenter__ = AsyncMock(
|
|
return_value=mock_client
|
|
)
|
|
mock_client_class.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".txt", delete=False
|
|
) as f:
|
|
f.write("test content")
|
|
temp_path = f.name
|
|
|
|
try:
|
|
result = await client.upload_to_presigned_url(
|
|
"https://upload.example.com", temp_path, "text/plain"
|
|
)
|
|
assert result["ok"] is True
|
|
finally:
|
|
os.unlink(temp_path)
|
|
|
|
async def test_upload_to_presigned_url_http_error(self):
|
|
client = OutlineClient("https://api.example.com", "token")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 403
|
|
mock_response.text = "Forbidden"
|
|
from httpx import HTTPStatusError
|
|
|
|
mock_response.raise_for_status.side_effect = HTTPStatusError(
|
|
"Forbidden", request=MagicMock(), response=mock_response
|
|
)
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_class:
|
|
mock_client = AsyncMock()
|
|
mock_client.put = AsyncMock(return_value=mock_response)
|
|
mock_client_class.return_value.__aenter__ = AsyncMock(
|
|
return_value=mock_client
|
|
)
|
|
mock_client_class.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".txt", delete=False
|
|
) as f:
|
|
f.write("test content")
|
|
temp_path = f.name
|
|
|
|
try:
|
|
result = await client.upload_to_presigned_url(
|
|
"https://upload.example.com", temp_path, "text/plain"
|
|
)
|
|
assert result["ok"] is False
|
|
assert result["status"] == 403
|
|
finally:
|
|
os.unlink(temp_path)
|
|
|
|
async def test_call_multipart_success(self):
|
|
client = OutlineClient("https://api.example.com", "token")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status = MagicMock()
|
|
mock_response.json = MagicMock(
|
|
return_value={"ok": True, "data": {"id": "doc-123"}}
|
|
)
|
|
mock_response.status_code = 200
|
|
|
|
client._client = MagicMock()
|
|
client._client.post = AsyncMock(return_value=mock_response)
|
|
client._client.is_closed = False
|
|
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".txt", delete=False
|
|
) as f:
|
|
f.write("test content")
|
|
temp_path = f.name
|
|
|
|
try:
|
|
result = await client.call_multipart(
|
|
"documents.import", temp_path, {"collectionId": "col-123"}
|
|
)
|
|
assert result["ok"] is True
|
|
assert result["data"]["id"] == "doc-123"
|
|
finally:
|
|
os.unlink(temp_path)
|
|
|
|
async def test_call_multipart_http_error(self):
|
|
client = OutlineClient("https://api.example.com", "token")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 400
|
|
mock_response.text = '{"message": "Bad request"}'
|
|
mock_response.json = MagicMock(return_value={"message": "Bad request"})
|
|
from httpx import HTTPStatusError
|
|
|
|
mock_response.raise_for_status.side_effect = HTTPStatusError(
|
|
"Bad request", request=MagicMock(), response=mock_response
|
|
)
|
|
|
|
client._client = MagicMock()
|
|
client._client.post = AsyncMock(return_value=mock_response)
|
|
client._client.is_closed = False
|
|
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".txt", delete=False
|
|
) as f:
|
|
f.write("test content")
|
|
temp_path = f.name
|
|
|
|
try:
|
|
result = await client.call_multipart(
|
|
"documents.import", temp_path, {"collectionId": "col-123"}
|
|
)
|
|
assert result["ok"] is False
|
|
assert result["error"] == "Bad request"
|
|
assert result["status"] == 400
|
|
finally:
|
|
os.unlink(temp_path)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestAttachFileToDocument:
|
|
async def test_path_validation_failure(self):
|
|
result = await attach_file_to_document("doc-123", "/etc/passwd")
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is False
|
|
assert "must be within" in parsed["error"]
|
|
|
|
async def test_size_limit_failure(self, tmp_path, monkeypatch):
|
|
test_file = tmp_path / "large.bin"
|
|
test_file.write_bytes(b"x" * 100)
|
|
monkeypatch.setattr("server.MAX_FILE_SIZE", 50)
|
|
|
|
result = await attach_file_to_document("doc-123", str(test_file))
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is False
|
|
assert "exceeds 50MB" in parsed["error"]
|
|
|
|
async def test_successful_attachment(self, tmp_path, monkeypatch):
|
|
test_file = tmp_path / "test.txt"
|
|
test_file.write_text("hello world")
|
|
|
|
async def mock_call(method, params):
|
|
assert method == "attachments.create"
|
|
return {
|
|
"ok": True,
|
|
"data": {
|
|
"uploadUrl": "https://presigned.example.com/upload",
|
|
"url": "https://cdn.example.com/attachment.txt",
|
|
},
|
|
}
|
|
|
|
async def mock_upload(url, path, ct):
|
|
assert url == "https://presigned.example.com/upload"
|
|
return {"ok": True}
|
|
|
|
monkeypatch.setattr(outline_client, "call", mock_call)
|
|
monkeypatch.setattr(outline_client, "upload_to_presigned_url", mock_upload)
|
|
|
|
result = await attach_file_to_document("doc-123", str(test_file))
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is True
|
|
assert parsed["name"] == "test.txt"
|
|
assert parsed["markdown"] == "[test.txt](https://cdn.example.com/attachment.txt)"
|
|
|
|
async def test_attachment_create_failure(self, tmp_path, monkeypatch):
|
|
test_file = tmp_path / "test.txt"
|
|
test_file.write_text("hello")
|
|
|
|
async def mock_call(method, params):
|
|
return {"ok": False, "error": "Document not found"}
|
|
|
|
monkeypatch.setattr(outline_client, "call", mock_call)
|
|
|
|
result = await attach_file_to_document("doc-123", str(test_file))
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is False
|
|
assert "Document not found" in parsed["error"]
|
|
|
|
async def test_upload_failure(self, tmp_path, monkeypatch):
|
|
test_file = tmp_path / "test.txt"
|
|
test_file.write_text("hello")
|
|
|
|
async def mock_call(method, params):
|
|
return {
|
|
"ok": True,
|
|
"data": {"uploadUrl": "https://presigned.example.com/upload"},
|
|
}
|
|
|
|
async def mock_upload(url, path, ct):
|
|
return {"ok": False, "error": "Upload rejected", "status": 413}
|
|
|
|
monkeypatch.setattr(outline_client, "call", mock_call)
|
|
monkeypatch.setattr(outline_client, "upload_to_presigned_url", mock_upload)
|
|
|
|
result = await attach_file_to_document("doc-123", str(test_file))
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is False
|
|
assert parsed["error"] == "Upload rejected"
|
|
|
|
async def test_missing_attachment_url(self, tmp_path, monkeypatch):
|
|
"""If Outline returns uploadUrl but no public url, we should error."""
|
|
test_file = tmp_path / "test.txt"
|
|
test_file.write_text("hello")
|
|
|
|
async def mock_call(method, params):
|
|
return {
|
|
"ok": True,
|
|
"data": {"uploadUrl": "https://presigned.example.com/upload"},
|
|
}
|
|
|
|
async def mock_upload(url, path, ct):
|
|
return {"ok": True}
|
|
|
|
monkeypatch.setattr(outline_client, "call", mock_call)
|
|
monkeypatch.setattr(outline_client, "upload_to_presigned_url", mock_upload)
|
|
|
|
result = await attach_file_to_document("doc-123", str(test_file))
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is False
|
|
assert "no public URL" in parsed["error"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUploadImageToDocument:
|
|
async def test_path_validation_failure(self):
|
|
result = await upload_image_to_document("/etc/passwd", "doc-123")
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is False
|
|
|
|
async def test_size_limit_failure(self, tmp_path, monkeypatch):
|
|
test_file = tmp_path / "large.png"
|
|
test_file.write_bytes(b"x" * 100)
|
|
monkeypatch.setattr("server.MAX_FILE_SIZE", 50)
|
|
|
|
result = await upload_image_to_document(str(test_file), "doc-123")
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is False
|
|
assert "exceeds 50MB" in parsed["error"]
|
|
|
|
async def test_successful_image_upload(self, tmp_path, monkeypatch):
|
|
test_file = tmp_path / "test.png"
|
|
test_file.write_text("fake png content")
|
|
|
|
async def mock_call(method, params):
|
|
return {
|
|
"ok": True,
|
|
"data": {
|
|
"uploadUrl": "https://presigned.example.com/upload",
|
|
"url": "https://cdn.example.com/image.png",
|
|
},
|
|
}
|
|
|
|
async def mock_upload(url, path, ct):
|
|
return {"ok": True}
|
|
|
|
monkeypatch.setattr(outline_client, "call", mock_call)
|
|
monkeypatch.setattr(outline_client, "upload_to_presigned_url", mock_upload)
|
|
|
|
result = await upload_image_to_document(str(test_file), "doc-123", "My Alt")
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is True
|
|
assert parsed["markdown"] == ""
|
|
|
|
async def test_default_alt_text(self, tmp_path, monkeypatch):
|
|
test_file = tmp_path / "screenshot.png"
|
|
test_file.write_text("fake png content")
|
|
|
|
async def mock_call(method, params):
|
|
return {
|
|
"ok": True,
|
|
"data": {
|
|
"uploadUrl": "https://presigned.example.com/upload",
|
|
"url": "https://cdn.example.com/image.png",
|
|
},
|
|
}
|
|
|
|
async def mock_upload(url, path, ct):
|
|
return {"ok": True}
|
|
|
|
monkeypatch.setattr(outline_client, "call", mock_call)
|
|
monkeypatch.setattr(outline_client, "upload_to_presigned_url", mock_upload)
|
|
|
|
result = await upload_image_to_document(str(test_file), "doc-123")
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is True
|
|
assert parsed["markdown"] == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestImportFileToOutline:
|
|
async def test_path_validation_failure(self):
|
|
result = await import_file_to_outline("/etc/passwd", "col-123")
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is False
|
|
assert "must be within" in parsed["error"]
|
|
|
|
async def test_size_limit_failure(self, tmp_path, monkeypatch):
|
|
test_file = tmp_path / "large.md"
|
|
test_file.write_bytes(b"x" * 100)
|
|
monkeypatch.setattr("server.MAX_FILE_SIZE", 50)
|
|
|
|
result = await import_file_to_outline(str(test_file), "col-123")
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is False
|
|
assert "exceeds 50MB" in parsed["error"]
|
|
|
|
async def test_successful_import(self, tmp_path, monkeypatch):
|
|
test_file = tmp_path / "notes.md"
|
|
test_file.write_text("# My Notes\n\nHello world")
|
|
|
|
async def mock_call_multipart(method, path, fields):
|
|
assert method == "documents.import"
|
|
assert fields["collectionId"] == "col-123"
|
|
return {"ok": True, "data": {"id": "doc-456", "url": "https://outline.example.com/doc/doc-456"}}
|
|
|
|
monkeypatch.setattr(outline_client, "call_multipart", mock_call_multipart)
|
|
|
|
result = await import_file_to_outline(str(test_file), "col-123")
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is True
|
|
assert parsed["data"]["id"] == "doc-456"
|
|
|
|
async def test_import_with_parent_document(self, tmp_path, monkeypatch):
|
|
test_file = tmp_path / "notes.md"
|
|
test_file.write_text("# Child Doc")
|
|
|
|
captured_fields = {}
|
|
|
|
async def mock_call_multipart(method, path, fields):
|
|
captured_fields.update(fields)
|
|
return {"ok": True, "data": {"id": "doc-789"}}
|
|
|
|
monkeypatch.setattr(outline_client, "call_multipart", mock_call_multipart)
|
|
|
|
result = await import_file_to_outline(
|
|
str(test_file), "col-123", parent_document_id="parent-999"
|
|
)
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is True
|
|
assert captured_fields["parentDocumentId"] == "parent-999"
|
|
|
|
async def test_import_failure(self, tmp_path, monkeypatch):
|
|
test_file = tmp_path / "notes.md"
|
|
test_file.write_text("# Notes")
|
|
|
|
async def mock_call_multipart(method, path, fields):
|
|
return {"ok": False, "error": "Collection not found"}
|
|
|
|
monkeypatch.setattr(outline_client, "call_multipart", mock_call_multipart)
|
|
|
|
result = await import_file_to_outline(str(test_file), "col-123")
|
|
parsed = json.loads(result)
|
|
assert parsed["ok"] is False
|
|
assert "Collection not found" in parsed["error"]
|