kodi
9778dc6c33
Introduce dedicated remote file facade for /Clients paths, add agent read/download endpoints, enable remote view/properties/download/image preview in the web UI, and keep remote write operations disabled.
80 lines
3.3 KiB
Python
80 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from fastapi import HTTPException
|
|
from starlette.requests import Request
|
|
|
|
from finder_commander.app import main as agent_main
|
|
|
|
|
|
class AgentFileEndpointsTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.share_root = Path(self.temp_dir.name) / "Downloads"
|
|
self.share_root.mkdir(parents=True, exist_ok=True)
|
|
self.outside_root = Path(self.temp_dir.name) / "Outside"
|
|
self.outside_root.mkdir(parents=True, exist_ok=True)
|
|
self.config_path = Path(self.temp_dir.name) / "agent.json"
|
|
self.config_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"agent_access_token": "agent-secret",
|
|
"client_id": "client-123",
|
|
"display_name": "Jan MacBook",
|
|
"shares": {"downloads": str(self.share_root)},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
os.environ["FINDER_COMMANDER_REMOTE_AGENT_CONFIG"] = str(self.config_path)
|
|
agent_main.get_runtime_config.cache_clear()
|
|
|
|
def tearDown(self) -> None:
|
|
os.environ.pop("FINDER_COMMANDER_REMOTE_AGENT_CONFIG", None)
|
|
agent_main.get_runtime_config.cache_clear()
|
|
self.temp_dir.cleanup()
|
|
|
|
@staticmethod
|
|
def _authorized_request() -> Request:
|
|
return Request({"type": "http", "headers": [(b"authorization", b"Bearer agent-secret")]})
|
|
|
|
def test_info_read_and_download_success(self) -> None:
|
|
notes = self.share_root / "notes.md"
|
|
notes.write_text("# title\nhello\n", encoding="utf-8")
|
|
|
|
info_response = agent_main.api_info(self._authorized_request(), share="downloads", path="notes.md")
|
|
self.assertEqual(info_response["kind"], "file")
|
|
self.assertEqual(info_response["extension"], ".md")
|
|
|
|
read_response = agent_main.api_read(self._authorized_request(), share="downloads", path="notes.md", max_bytes=4)
|
|
self.assertTrue(read_response["truncated"])
|
|
self.assertEqual(read_response["content"], "# ti")
|
|
|
|
download_response = agent_main.api_download(self._authorized_request(), share="downloads", path="notes.md")
|
|
self.assertEqual(download_response.media_type, "text/markdown")
|
|
self.assertIn('attachment; filename="notes.md"', download_response.headers.get("content-disposition", ""))
|
|
|
|
def test_unknown_share_and_escape_outside_root_are_rejected(self) -> None:
|
|
outside_file = self.outside_root / "secret.txt"
|
|
outside_file.write_text("secret", encoding="utf-8")
|
|
(self.share_root / "escape.txt").symlink_to(outside_file)
|
|
|
|
with self.assertRaises(HTTPException) as unknown_share:
|
|
agent_main.api_info(self._authorized_request(), share="missing", path="notes.md")
|
|
self.assertEqual(unknown_share.exception.status_code, 404)
|
|
self.assertEqual(unknown_share.exception.detail["code"], "path_not_found")
|
|
|
|
with self.assertRaises(HTTPException) as escaped:
|
|
agent_main.api_info(self._authorized_request(), share="downloads", path="escape.txt")
|
|
self.assertEqual(escaped.exception.status_code, 403)
|
|
self.assertEqual(escaped.exception.detail["code"], "path_traversal_detected")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|