Dymas/ani-cli-web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden remote access and path handling

Browse Source
This commit is contained in:
Dymas
2026-05-26 09:20:02 +02:00
parent ff605aca03
commit 83497d8ff0
4 changed files with 545 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_app.py
+258 -18
View File
@@ -2,6 +2,7 @@
import base64
import importlib.util
import io
import json
import os
import sys
import tempfile
@@ -46,6 +47,7 @@ class DummyHandler:
self.headers = {}
if content_length is not None:
self.headers["Content-Length"] = str(content_length)
self.headers.setdefault("Content-Type", "application/json")
self.rfile = io.BytesIO(body)
self.json_payload = None
self.json_status = None
@@ -170,10 +172,31 @@ class NetworkGuardTests(unittest.TestCase):
self.assertIn("Username", body)
self.assertIn("Password", body)
def test_remote_login_page_escapes_hidden_next_value(self):
handler = DummyHandler('/?next="><script>alert(1)</script>')
handler.client_address = ("192.168.1.25", 8421)
handler.command = "GET"
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
APP.Handler.do_GET(handler)
body = handler.wfile.getvalue().decode("utf-8")
self.assertNotIn('<script>alert(1)</script>', body)
self.assertIn('value="/?next=%22%3E%3Cscript%3Ealert%281%29%3C%2Fscript%3E"', body)
def test_remote_login_post_sets_cookie_and_redirects(self):
body = b"username=demo&password=secret-pass&next=%2Fwatchlist"
handler = DummyHandler("/auth/login", body=body, content_length=len(body))
handler.client_address = ("192.168.1.25", 8421)
handler.headers["Content-Type"] = "application/x-www-form-urlencoded"
with mock.patch.dict(
os.environ,
@@ -191,6 +214,84 @@ class NetworkGuardTests(unittest.TestCase):
self.assertEqual(headers.get("Location"), "/watchlist")
self.assertIn("ani_cli_web_session=", headers.get("Set-Cookie", ""))
def test_remote_session_cookie_rejects_missing_origin_on_post(self):
body = b'{"show_id":"show-1"}'
handler = DummyHandler("/api/watchlist/remove", body=body, content_length=len(body))
handler.client_address = ("192.168.1.25", 8421)
handler.headers["Host"] = "anime.example:8421"
handler.headers["Cookie"] = (
"ani_cli_web_session="
+ http_handler.remote_access_session_value({"auth_username": "demo", "auth_password": "secret-pass"})
)
handler.handler_context = mock.Mock(remove_from_watchlist=mock.Mock(return_value={"ok": True}), http_error=APP.HttpError)
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
APP.Handler.do_POST(handler)
self.assertEqual(handler.error_status, HTTPStatus.FORBIDDEN)
self.assertIn("origin", handler.error_message.lower())
def test_remote_session_cookie_rejects_cross_origin_post(self):
body = b'{"show_id":"show-1"}'
handler = DummyHandler("/api/watchlist/remove", body=body, content_length=len(body))
handler.client_address = ("192.168.1.25", 8421)
handler.headers["Host"] = "anime.example:8421"
handler.headers["Origin"] = "https://elsewhere.example"
handler.headers["Cookie"] = (
"ani_cli_web_session="
+ http_handler.remote_access_session_value({"auth_username": "demo", "auth_password": "secret-pass"})
)
handler.handler_context = mock.Mock(remove_from_watchlist=mock.Mock(return_value={"ok": True}), http_error=APP.HttpError)
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
APP.Handler.do_POST(handler)
self.assertEqual(handler.error_status, HTTPStatus.FORBIDDEN)
self.assertIn("cross-origin", handler.error_message.lower())
def test_remote_session_cookie_accepts_same_origin_post(self):
body = b'{"show_id":"show-1"}'
handler = DummyHandler("/api/watchlist/remove", body=body, content_length=len(body))
handler.client_address = ("192.168.1.25", 8421)
handler.headers["Host"] = "anime.example:8421"
handler.headers["Origin"] = "https://anime.example:8421"
handler.headers["Cookie"] = (
"ani_cli_web_session="
+ http_handler.remote_access_session_value({"auth_username": "demo", "auth_password": "secret-pass"})
)
remove_from_watchlist = mock.Mock(return_value={"ok": True})
handler.handler_context = mock.Mock(remove_from_watchlist=remove_from_watchlist, http_error=APP.HttpError)
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
APP.Handler.do_POST(handler)
self.assertEqual(handler.json_status, HTTPStatus.OK)
remove_from_watchlist.assert_called_once_with("show-1")
def test_loopback_proxy_with_forwarded_remote_ip_requires_credentials(self):
handler = DummyHandler("/api/config")
handler.client_address = ("127.0.0.1", 8421)
@@ -588,7 +689,7 @@ class DownloadQueueWorkerFailureTests(unittest.TestCase):
"mode": "sub",
"quality": "best",
"download_dir": "/tmp/example",
"discord_webhook_url": "https://discord.example/webhook",
"discord_webhook_url": "https://discord.com/api/webhooks/123456/test-token",
"discord_webhook_events": ["download_completed"],
},
watchlist_sync_fn=lambda _job: {
@@ -1641,7 +1742,7 @@ class JellyfinSyncTests(unittest.TestCase):
"jellyfin_sync_enabled": True,
"jellyfin_tv_dir": str(jellyfin_tv),
"jellyfin_movie_dir": "",
"discord_webhook_url": "https://discord.example/webhook",
"discord_webhook_url": "https://discord.com/api/webhooks/123456/test-token",
"discord_webhook_events": ["jellyfin_sync_success"],
},
)
@@ -1664,7 +1765,7 @@ class JellyfinSyncTests(unittest.TestCase):
"jellyfin_sync_enabled": True,
"jellyfin_tv_dir": "",
"jellyfin_movie_dir": "",
"discord_webhook_url": "https://discord.example/webhook",
"discord_webhook_url": "https://discord.com/api/webhooks/123456/test-token",
"discord_webhook_events": ["jellyfin_sync_failed"],
},
)
@@ -1692,7 +1793,7 @@ class JellyfinSyncTests(unittest.TestCase):
"jellyfin_sync_enabled": True,
"jellyfin_tv_dir": str(Path(temp_root) / "jellyfin-tv"),
"jellyfin_movie_dir": "",
"discord_webhook_url": "https://discord.example/webhook",
"discord_webhook_url": "https://discord.com/api/webhooks/123456/test-token",
"discord_webhook_events": ["jellyfin_sync_failed", "jellyfin_sync_success"],
},
)
@@ -1716,7 +1817,7 @@ class JellyfinSyncTests(unittest.TestCase):
"jellyfin_sync_enabled": True,
"jellyfin_tv_dir": str(Path(temp_root) / "jellyfin-tv"),
"jellyfin_movie_dir": "",
"discord_webhook_url": "https://discord.example/webhook",
"discord_webhook_url": "https://discord.com/api/webhooks/123456/test-token",
"discord_webhook_events": ["jellyfin_sync_failed"],
},
)
@@ -1991,7 +2092,7 @@ class ConfigSnapshotTests(unittest.TestCase):
def test_normalize_config_filters_discord_webhook_events(self):
config = APP.normalize_config(
{
"discord_webhook_url": " https://discord.example/webhook ",
"discord_webhook_url": " https://discord.com/api/webhooks/123456/test-token ",
"discord_webhook_events": [
"watchlist_refresh_new_episodes",
"runtime_error",
@@ -2000,7 +2101,7 @@ class ConfigSnapshotTests(unittest.TestCase):
],
}
)
self.assertEqual(config["discord_webhook_url"], "https://discord.example/webhook")
self.assertEqual(config["discord_webhook_url"], "https://discord.com/api/webhooks/123456/test-token")
self.assertEqual(
config["discord_webhook_events"],
["watchlist_refresh_new_episodes", "runtime_error"],
@@ -2014,7 +2115,7 @@ class ConfigSnapshotTests(unittest.TestCase):
with mock.patch.object(APP, "send_discord_webhook_event") as send_webhook:
result = APP.test_discord_webhook_config(
{
"discord_webhook_url": "https://discord.example/webhook",
"discord_webhook_url": "https://discord.com/api/webhooks/123456/test-token",
"discord_webhook_events": [],
}
)
@@ -2022,7 +2123,7 @@ class ConfigSnapshotTests(unittest.TestCase):
self.assertEqual(result["message"], "Discord webhook test notification sent.")
send_webhook.assert_called_once()
config, event_name = send_webhook.call_args.args[:2]
self.assertEqual(config["discord_webhook_url"], "https://discord.example/webhook")
self.assertEqual(config["discord_webhook_url"], "https://discord.com/api/webhooks/123456/test-token")
self.assertEqual(event_name, "test")
self.assertTrue(send_webhook.call_args.kwargs["force"])
@@ -2094,7 +2195,7 @@ class WatchlistAutoRefreshSchedulerTests(unittest.TestCase):
service = APP.WatchlistRefreshJobs(
FakeStore(),
config_getter=lambda: {
"discord_webhook_url": "https://discord.example/webhook",
"discord_webhook_url": "https://discord.com/api/webhooks/123456/test-token",
"discord_webhook_events": [
"watchlist_refresh_new_episodes",
"watchlist_refresh_failed",
@@ -2121,7 +2222,7 @@ class DiscordWebhookFormattingTests(unittest.TestCase):
with mock.patch.object(APP.app_support, "_post_discord_webhook", side_effect=lambda url, payload, attachments=None, timeout=10: payloads.append(payload)):
result = APP.app_support.send_discord_webhook_event(
{
"discord_webhook_url": "https://discord.example/webhook",
"discord_webhook_url": "https://discord.com/api/webhooks/123456/test-token",
"discord_webhook_events": ["download_completed"],
},
"download_completed",
@@ -2149,7 +2250,7 @@ class DiscordWebhookFormattingTests(unittest.TestCase):
with mock.patch.object(APP.app_support, "_post_discord_webhook", side_effect=lambda url, payload, attachments=None, timeout=10: payloads.append(payload)):
result = APP.app_support.send_discord_webhook_event(
{
"discord_webhook_url": "https://discord.example/webhook",
"discord_webhook_url": "https://discord.com/api/webhooks/123456/test-token",
"discord_webhook_events": ["jellyfin_sync_success"],
},
"jellyfin_sync_success",
@@ -2455,7 +2556,7 @@ class StartupBehaviorTests(unittest.TestCase):
"jellyfin_sync_enabled": True,
"jellyfin_tv_dir": "/tmp/jellyfin-tv",
"jellyfin_movie_dir": "/tmp/jellyfin-movies",
"discord_webhook_url": "https://discord.example/webhook",
"discord_webhook_url": "https://discord.com/api/webhooks/123456/test-token",
"discord_webhook_events": ["runtime_error"],
}
)
@@ -2471,7 +2572,7 @@ class StartupBehaviorTests(unittest.TestCase):
self.assertTrue(APP.CONFIG["jellyfin_sync_enabled"])
self.assertEqual(APP.CONFIG["jellyfin_tv_dir"], "/tmp/jellyfin-tv")
self.assertEqual(APP.CONFIG["jellyfin_movie_dir"], "/tmp/jellyfin-movies")
self.assertEqual(APP.CONFIG["discord_webhook_url"], "https://discord.example/webhook")
self.assertEqual(APP.CONFIG["discord_webhook_url"], "https://discord.com/api/webhooks/123456/test-token")
self.assertEqual(APP.CONFIG["discord_webhook_events"], ["runtime_error"])
APP.WATCHLIST_AUTO_REFRESH.notify_config_changed.assert_called_once_with()
finally:
@@ -2543,17 +2644,142 @@ class HandlerRouteTests(unittest.TestCase):
self.assertEqual([entry["name"] for entry in handler.json_payload["entries"]], ["folder-a", "folder-b"])
self.assertTrue(all(entry["is_dir"] for entry in handler.json_payload["entries"]))
def test_remote_filesystem_browse_rejects_paths_outside_configured_roots(self):
with tempfile.TemporaryDirectory() as allowed_root, tempfile.TemporaryDirectory() as blocked_root:
APP.save_runtime_config({"download_dir": allowed_root, "mode": "sub", "quality": "best"})
encoded = base64.b64encode(b"demo:secret-pass").decode("ascii")
handler = DummyHandler(f"/api/fs/browse?path={APP.quote(blocked_root)}&mode=dir")
handler.client_address = ("192.168.1.25", 8421)
handler.headers["Authorization"] = f"Basic {encoded}"
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
APP.Handler.do_GET(handler)
self.assertEqual(handler.error_status, HTTPStatus.FORBIDDEN)
self.assertIn("allowed remote browse roots", handler.error_message.lower())
def test_remote_filesystem_browse_defaults_blank_path_to_allowed_root(self):
with tempfile.TemporaryDirectory() as allowed_root:
Path(allowed_root, "folder-a").mkdir()
APP.save_runtime_config({"download_dir": allowed_root, "mode": "sub", "quality": "best"})
encoded = base64.b64encode(b"demo:secret-pass").decode("ascii")
handler = DummyHandler("/api/fs/browse?path=&mode=dir")
handler.client_address = ("192.168.1.25", 8421)
handler.headers["Authorization"] = f"Basic {encoded}"
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
APP.Handler.do_GET(handler)
self.assertEqual(handler.json_status, HTTPStatus.OK)
self.assertEqual(handler.json_payload["path"], allowed_root)
self.assertEqual(handler.json_payload["home_path"], allowed_root)
def test_remote_path_roots_include_explicit_allowlist(self):
with tempfile.TemporaryDirectory() as base_root, tempfile.TemporaryDirectory() as extra_root:
APP.save_runtime_config({"download_dir": base_root, "mode": "sub", "quality": "best"})
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_REMOTE_PATH_ROOTS": extra_root,
},
clear=False,
):
roots = APP.app_support.configured_remote_path_roots(APP.get_config_snapshot())
self.assertIn(Path(base_root).resolve(), roots)
self.assertIn(Path(extra_root).resolve(), roots)
def test_config_post_accepts_discord_webhook_fields(self):
body = b'{"download_dir":"/tmp/example","mode":"sub","quality":"best","discord_webhook_url":"https://discord.example/webhook","discord_webhook_events":["runtime_error","watchlist_refresh_failed","bad"]}'
body = b'{"download_dir":"/tmp/example","mode":"sub","quality":"best","discord_webhook_url":"https://discord.com/api/webhooks/123456/test-token","discord_webhook_events":["runtime_error","watchlist_refresh_failed","bad"]}'
handler = DummyHandler("/api/config", body=body, content_length=len(body))
APP.Handler.do_POST(handler)
self.assertEqual(handler.json_status, HTTPStatus.OK)
self.assertEqual(handler.json_payload["discord_webhook_url"], "https://discord.example/webhook")
self.assertEqual(handler.json_payload["discord_webhook_url"], "https://discord.com/api/webhooks/123456/test-token")
self.assertEqual(
handler.json_payload["discord_webhook_events"],
["runtime_error", "watchlist_refresh_failed"],
)
def test_config_post_rejects_non_discord_webhook_host(self):
body = b'{"download_dir":"/tmp/example","mode":"sub","quality":"best","discord_webhook_url":"https://example.com/not-discord"}'
handler = DummyHandler("/api/config", body=body, content_length=len(body))
APP.Handler.do_POST(handler)
self.assertEqual(handler.error_status, HTTPStatus.BAD_REQUEST)
self.assertIn("official discord webhook host", handler.error_message.lower())
def test_remote_config_post_rejects_download_dir_outside_allowed_roots(self):
with tempfile.TemporaryDirectory() as allowed_root, tempfile.TemporaryDirectory() as blocked_root:
APP.save_runtime_config({"download_dir": allowed_root, "mode": "sub", "quality": "best"})
body = json.dumps(
{
"download_dir": blocked_root,
"mode": "sub",
"quality": "best",
}
).encode("utf-8")
handler = DummyHandler("/api/config", body=body, content_length=len(body))
handler.client_address = ("192.168.1.25", 8421)
handler.headers["Authorization"] = f"Basic {base64.b64encode(b'demo:secret-pass').decode('ascii')}"
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
},
clear=False,
):
APP.Handler.do_POST(handler)
self.assertEqual(handler.error_status, HTTPStatus.FORBIDDEN)
self.assertIn("remote download directory", handler.error_message.lower())
def test_remote_config_post_accepts_download_dir_inside_explicit_allowlist(self):
with tempfile.TemporaryDirectory() as current_root, tempfile.TemporaryDirectory() as allowed_root:
APP.save_runtime_config({"download_dir": current_root, "mode": "sub", "quality": "best"})
body = json.dumps(
{
"download_dir": allowed_root,
"mode": "sub",
"quality": "best",
}
).encode("utf-8")
handler = DummyHandler("/api/config", body=body, content_length=len(body))
handler.client_address = ("192.168.1.25", 8421)
handler.headers["Authorization"] = f"Basic {base64.b64encode(b'demo:secret-pass').decode('ascii')}"
with mock.patch.dict(
os.environ,
{
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
"ANI_CLI_WEB_REMOTE_PATH_ROOTS": allowed_root,
},
clear=False,
):
APP.Handler.do_POST(handler)
self.assertEqual(handler.json_status, HTTPStatus.OK)
self.assertEqual(handler.json_payload["download_dir"], allowed_root)
def test_config_routes_hide_auth_fields_and_preserve_saved_credentials(self):
APP.save_runtime_config(
{
@@ -2582,7 +2808,7 @@ class HandlerRouteTests(unittest.TestCase):
self.assertNotIn("auth_password", get_handler.json_payload)
def test_config_webhook_test_route_uses_unsaved_payload(self):
body = b'{"discord_webhook_url":"https://discord.example/webhook","discord_webhook_events":["runtime_error"]}'
body = b'{"discord_webhook_url":"https://discord.com/api/webhooks/123456/test-token","discord_webhook_events":["runtime_error"]}'
handler = DummyHandler("/api/config/webhook/test", body=body, content_length=len(body))
test_webhook = mock.Mock(return_value={"message": "ok"})
handler.handler_context = mock.Mock(test_discord_webhook_config=test_webhook)
@@ -2590,15 +2816,29 @@ class HandlerRouteTests(unittest.TestCase):
self.assertEqual(handler.json_status, HTTPStatus.OK)
self.assertEqual(handler.json_payload["message"], "ok")
test_webhook.assert_called_once_with(
{"discord_webhook_url": "https://discord.example/webhook", "discord_webhook_events": ["runtime_error"]}
{"discord_webhook_url": "https://discord.com/api/webhooks/123456/test-token", "discord_webhook_events": ["runtime_error"]}
)
def test_config_webhook_test_route_rejects_non_discord_webhook_url(self):
body = b'{"discord_webhook_url":"https://example.com/not-discord","discord_webhook_events":["runtime_error"]}'
handler = DummyHandler("/api/config/webhook/test", body=body, content_length=len(body))
APP.Handler.do_POST(handler)
self.assertEqual(handler.error_status, HTTPStatus.BAD_REQUEST)
self.assertIn("official discord webhook host", handler.error_message.lower())
def test_body_json_rejects_oversized_request(self):
handler = DummyHandler("/api/watchlist/upload-thumbnail", body=b"{}", content_length=APP.MAX_JSON_BODY_BYTES + 1)
with self.assertRaises(APP.HttpError) as ctx:
APP.Handler.body_json(handler)
self.assertEqual(ctx.exception.status, HTTPStatus.REQUEST_ENTITY_TOO_LARGE)
def test_body_json_requires_application_json_content_type(self):
handler = DummyHandler("/api/watchlist/upload-thumbnail", body=b"{}", content_length=2)
handler.headers["Content-Type"] = "text/plain"
with self.assertRaises(ValueError) as ctx:
APP.Handler.body_json(handler)
self.assertIn("content-type", str(ctx.exception).lower())
def test_body_json_rejects_non_utf8_payload(self):
handler = DummyHandler("/api/watchlist/upload-thumbnail", body=b"\xff", content_length=1)
with self.assertRaises(ValueError) as ctx: