"""Black-box contract tests for the LAN file sharing HTTP service. The production module must expose ``create_server(root, host, port)``. It must return a bound ``http.server.ThreadingHTTPServer`` (or compatible server) that can be started with ``serve_forever``. The service receives only URL paths: ``GET /`` (or ``/list``) lists the share root, while other GET paths download a regular file or a ZIP archive for a directory. """ from __future__ import annotations from http.client import HTTPConnection from io import BytesIO import json from pathlib import Path import socket import subprocess import sys import tempfile import threading import time import unittest from urllib.parse import quote import zipfile from file_server import create_server class LanFileServerTest(unittest.TestCase): """Exercise the public HTTP contract against a real local server.""" def setUp(self) -> None: # Keep fixtures outside the production share so a live service cannot # expose temporary secrets through 0.0.0.0:18001. self.temporary_directory = tempfile.TemporaryDirectory() self.share_root = Path(self.temporary_directory.name) / "share" self.share_root.mkdir() (self.share_root / "plain.txt").write_bytes(b"plain file\n") (self.share_root / "space name.txt").write_bytes(b"space file\n") (self.share_root / "café-世界.txt").write_bytes(b"unicode file\n") (self.share_root / "empty-directory").mkdir() nested_directory = self.share_root / "nested" nested_directory.mkdir() (nested_directory / "child.txt").write_bytes(b"nested child\n") self.outside_file = Path(self.temporary_directory.name) / "outside-secret.txt" self.outside_file.write_bytes(b"DO NOT EXPOSE THIS SECRET") (self.share_root / "outside-link.txt").symlink_to(self.outside_file) self.server = create_server(self.share_root, host="127.0.0.1", port=0) self.server_thread = threading.Thread( target=self.server.serve_forever, daemon=True, ) self.server_thread.start() self.host, self.port = self.server.server_address[:2] def tearDown(self) -> None: self.server.shutdown() self.server.server_close() self.server_thread.join(timeout=2) self.temporary_directory.cleanup() def get(self, target: str) -> tuple[int, dict[str, str], bytes]: connection = HTTPConnection(self.host, self.port, timeout=2) try: connection.request("GET", target) response = connection.getresponse() return response.status, dict(response.getheaders()), response.read() finally: connection.close() def assert_not_found_without_secret(self, target: str) -> None: status, _, body = self.get(target) self.assertEqual(status, 404) self.assertNotIn(b"DO NOT EXPOSE THIS SECRET", body) def test_list_returns_only_sorted_top_level_json_metadata(self) -> None: status, headers, body = self.get("/?format=json") self.assertEqual(status, 200) self.assertTrue(headers["Content-Type"].startswith("application/json")) payload = json.loads(body) simplified = [ {"name": entry["name"], "kind": entry["kind"]} for entry in payload["entries"] ] self.assertEqual( simplified, [ {"name": "café-世界.txt", "kind": "file"}, {"name": "empty-directory", "kind": "directory"}, {"name": "nested", "kind": "directory"}, {"name": "plain.txt", "kind": "file"}, {"name": "space name.txt", "kind": "file"}, ], ) for entry in payload["entries"]: if entry["kind"] == "file": self.assertIn("size", entry) def test_list_html_page_is_copy_friendly(self) -> None: status, headers, body = self.get("/") self.assertEqual(status, 200) self.assertTrue(headers["Content-Type"].startswith("text/html")) page = body.decode("utf-8") self.assertIn("plain.txt", page) self.assertIn("复制链接", page) self.assertIn("复制 wget", page) self.assertIn("/plain.txt", page) # Directories expose a browse URL (trailing slash) and a ZIP download. self.assertIn("/nested/", page) self.assertIn("/nested.zip", page) def test_nested_directory_listing_json_returns_children_only(self) -> None: status, headers, body = self.get("/nested/?format=json") self.assertEqual(status, 200) self.assertTrue(headers["Content-Type"].startswith("application/json")) payload = json.loads(body) self.assertEqual(payload.get("path"), "nested") simplified = [ {"name": entry["name"], "kind": entry["kind"]} for entry in payload["entries"] ] self.assertEqual(simplified, [{"name": "child.txt", "kind": "file"}]) def test_nested_directory_listing_html_allows_browse_and_download(self) -> None: status, headers, body = self.get("/nested/") self.assertEqual(status, 200) self.assertTrue(headers["Content-Type"].startswith("text/html")) page = body.decode("utf-8") self.assertIn("child.txt", page) self.assertIn("/nested/child.txt", page) # Breadcrumb / parent navigation back to root. self.assertIn("根目录", page) self.assertIn("返回上级", page) self.assertIn("nested", page) self.assertRegex(page, r'href="https?://[^"]+/?"') def test_deeper_directory_listing_and_file_download(self) -> None: deeper = self.share_root / "nested" / "deeper" deeper.mkdir() (deeper / "leaf.txt").write_bytes(b"leaf data\n") status, _, body = self.get("/nested/deeper/?format=json") self.assertEqual(status, 200) payload = json.loads(body) self.assertEqual(payload.get("path"), "nested/deeper") self.assertEqual( [{"name": e["name"], "kind": e["kind"]} for e in payload["entries"]], [{"name": "leaf.txt", "kind": "file"}], ) status, _, file_body = self.get("/nested/deeper/leaf.txt") self.assertEqual(status, 200) self.assertEqual(file_body, b"leaf data\n") status, headers, zip_body = self.get("/nested/deeper.zip") self.assertEqual(status, 200) self.assertTrue(headers["Content-Type"].startswith("application/zip")) with zipfile.ZipFile(BytesIO(zip_body)) as archive: self.assertEqual(archive.read("leaf.txt"), b"leaf data\n") def test_directory_listing_rejects_traversal_and_missing_dirs(self) -> None: for target in ( "/../?format=json", "/%2e%2e/", "/missing-directory/", "/nested/missing/", ): with self.subTest(target=target): self.assert_not_found_without_secret(target) def test_regular_file_download_returns_exact_bytes(self) -> None: status, headers, body = self.get("/plain.txt") self.assertEqual(status, 200) self.assertEqual(body, b"plain file\n") self.assertIn("attachment", headers["Content-Disposition"].lower()) def test_nested_regular_file_download_returns_exact_bytes(self) -> None: status, _, body = self.get("/nested/child.txt") self.assertEqual(status, 200) self.assertEqual(body, b"nested child\n") def test_directory_download_is_a_zip_with_its_descendants(self) -> None: for target in ("/nested", "/nested.zip"): with self.subTest(target=target): status, headers, body = self.get(target) self.assertEqual(status, 200) self.assertTrue(headers["Content-Type"].startswith("application/zip")) self.assertIn("nested.zip", headers["Content-Disposition"].lower()) with zipfile.ZipFile(BytesIO(body)) as archive: self.assertEqual(archive.read("child.txt"), b"nested child\n") def test_root_path_returns_listing_not_zip(self) -> None: status, headers, body = self.get("/") self.assertEqual(status, 200) self.assertTrue(headers["Content-Type"].startswith("text/html")) page = body.decode("utf-8") self.assertIn("plain.txt", page) self.assertNotIn(b"PK", body[:2]) # not a zip archive def test_empty_directory_download_is_an_empty_zip(self) -> None: status, headers, body = self.get("/empty-directory.zip") self.assertEqual(status, 200) self.assertTrue(headers["Content-Type"].startswith("application/zip")) self.assertIn("empty-directory.zip", headers["Content-Disposition"].lower()) with zipfile.ZipFile(BytesIO(body)) as archive: self.assertEqual(archive.namelist(), []) def test_url_encoded_space_and_unicode_file_names_download(self) -> None: for file_name, expected_body in ( ("space name.txt", b"space file\n"), ("café-世界.txt", b"unicode file\n"), ): with self.subTest(file_name=file_name): status, _, body = self.get("/" + quote(file_name, safe="")) self.assertEqual(status, 200) self.assertEqual(body, expected_body) def test_missing_file_and_directory_return_not_found(self) -> None: for target in ("/missing.txt", "/missing-directory/child.txt", "/empty"): with self.subTest(target=target): self.assert_not_found_without_secret(target) def test_absolute_path_attempts_do_not_escape_the_share_root(self) -> None: absolute_target = "//" + quote(str(self.outside_file).lstrip("/"), safe="/") self.assert_not_found_without_secret(absolute_target) def test_plain_and_encoded_parent_traversal_do_not_escape_the_share_root(self) -> None: parent_directory = quote(self.outside_file.parent.name, safe="") secret_name = quote(self.outside_file.name, safe="") traversal_targets = ( f"/../{parent_directory}/{secret_name}", f"/%2e%2e/{parent_directory}/{secret_name}", f"/%2e%2e%2f{parent_directory}%2f{secret_name}", ) for target in traversal_targets: with self.subTest(target=target): self.assert_not_found_without_secret(target) def test_outside_root_symlinks_are_not_listed_or_downloadable(self) -> None: status, _, listing = self.get("/?format=json") self.assertEqual(status, 200) self.assertNotIn("outside-link.txt", json.dumps(json.loads(listing))) self.assert_not_found_without_secret("/outside-link.txt") def test_cli_stops_cleanly_after_sigterm(self) -> None: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: probe.bind(("127.0.0.1", 0)) port = probe.getsockname()[1] command = ( sys.executable, str(Path(__file__).with_name("file_server.py")), "--root", str(self.share_root), "--host", "127.0.0.1", "--port", str(port), ) process = subprocess.Popen( command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) try: deadline = time.monotonic() + 2 while time.monotonic() < deadline: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client: if client.connect_ex(("127.0.0.1", port)) == 0: break time.sleep(0.05) else: self.fail("CLI server did not start listening in time") process.terminate() self.assertEqual(process.wait(timeout=2), 0) finally: if process.poll() is None: process.kill() process.wait(timeout=2) if __name__ == "__main__": unittest.main()