Replace token auth with single-user login
This commit is contained in:
+69
-45
@@ -1,4 +1,5 @@
|
||||
#!/usr/bin/env python3
|
||||
import base64
|
||||
import importlib.util
|
||||
import io
|
||||
import os
|
||||
@@ -101,7 +102,7 @@ class NetworkGuardTests(unittest.TestCase):
|
||||
self.assertFalse(APP.client_address_is_local("192.168.1.25"))
|
||||
self.assertFalse(APP.client_address_is_local("10.0.0.8"))
|
||||
|
||||
def test_remote_access_requires_token_for_non_local_clients(self):
|
||||
def test_remote_access_requires_credentials_for_non_local_clients(self):
|
||||
handler = DummyHandler("/api/config")
|
||||
handler.client_address = ("192.168.1.25", 8421)
|
||||
|
||||
@@ -109,68 +110,56 @@ class NetworkGuardTests(unittest.TestCase):
|
||||
with self.assertRaises(PermissionError) as ctx:
|
||||
APP.Handler.ensure_client_access(handler)
|
||||
|
||||
self.assertIn("REMOTE_TOKEN", str(ctx.exception))
|
||||
self.assertIn("AUTH_USERNAME", str(ctx.exception))
|
||||
|
||||
def test_remote_access_accepts_valid_token_for_non_local_clients(self):
|
||||
def test_remote_access_accepts_valid_basic_auth_for_non_local_clients(self):
|
||||
handler = DummyHandler("/api/config")
|
||||
handler.client_address = ("192.168.1.25", 8421)
|
||||
handler.headers["Authorization"] = "Bearer secret-token"
|
||||
encoded = base64.b64encode(b"demo:secret-pass").decode("ascii")
|
||||
handler.headers["Authorization"] = f"Basic {encoded}"
|
||||
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{"ANI_CLI_WEB_ALLOW_REMOTE": "1", "ANI_CLI_WEB_REMOTE_TOKEN": "secret-token"},
|
||||
{
|
||||
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
|
||||
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
|
||||
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
|
||||
},
|
||||
clear=False,
|
||||
):
|
||||
APP.Handler.ensure_client_access(handler)
|
||||
|
||||
def test_remote_access_accepts_valid_cookie_token_for_non_local_clients(self):
|
||||
def test_remote_access_accepts_valid_session_cookie_for_non_local_clients(self):
|
||||
handler = DummyHandler("/api/config")
|
||||
handler.client_address = ("192.168.1.25", 8421)
|
||||
handler.headers["Cookie"] = "ani_cli_web_token=secret-token"
|
||||
session_value = http_handler.remote_access_session_value(
|
||||
{"auth_username": "demo", "auth_password": "secret-pass"}
|
||||
)
|
||||
handler.headers["Cookie"] = f"ani_cli_web_session={session_value}"
|
||||
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{"ANI_CLI_WEB_ALLOW_REMOTE": "1", "ANI_CLI_WEB_REMOTE_TOKEN": "secret-token"},
|
||||
{
|
||||
"ANI_CLI_WEB_ALLOW_REMOTE": "1",
|
||||
"ANI_CLI_WEB_AUTH_USERNAME": "demo",
|
||||
"ANI_CLI_WEB_AUTH_PASSWORD": "secret-pass",
|
||||
},
|
||||
clear=False,
|
||||
):
|
||||
APP.Handler.ensure_client_access(handler)
|
||||
|
||||
def test_remote_access_accepts_valid_query_token_for_non_local_clients(self):
|
||||
handler = DummyHandler("/api/config?token=secret-token")
|
||||
handler.client_address = ("192.168.1.25", 8421)
|
||||
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{"ANI_CLI_WEB_ALLOW_REMOTE": "1", "ANI_CLI_WEB_REMOTE_TOKEN": "secret-token"},
|
||||
clear=False,
|
||||
):
|
||||
APP.Handler.ensure_client_access(handler)
|
||||
|
||||
def test_remote_get_bootstraps_cookie_and_redirects_when_query_token_valid(self):
|
||||
handler = DummyHandler("/?token=secret-token")
|
||||
handler.client_address = ("192.168.1.25", 8421)
|
||||
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{"ANI_CLI_WEB_ALLOW_REMOTE": "1", "ANI_CLI_WEB_REMOTE_TOKEN": "secret-token"},
|
||||
clear=False,
|
||||
):
|
||||
APP.Handler.do_GET(handler)
|
||||
|
||||
headers = dict(handler.response_headers)
|
||||
self.assertEqual(handler.response_status, HTTPStatus.SEE_OTHER)
|
||||
self.assertEqual(headers.get("Location"), "/")
|
||||
self.assertIn("ani_cli_web_token=secret-token", headers.get("Set-Cookie", ""))
|
||||
self.assertIn("Max-Age=", headers.get("Set-Cookie", ""))
|
||||
|
||||
def test_remote_browser_get_without_token_renders_login_page(self):
|
||||
def test_remote_browser_get_without_session_renders_login_page(self):
|
||||
handler = DummyHandler("/")
|
||||
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_REMOTE_TOKEN": "secret-token"},
|
||||
{
|
||||
"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)
|
||||
@@ -178,16 +167,21 @@ class NetworkGuardTests(unittest.TestCase):
|
||||
body = handler.wfile.getvalue().decode("utf-8")
|
||||
self.assertEqual(handler.response_status, HTTPStatus.UNAUTHORIZED)
|
||||
self.assertIn('<form method="post" action="/auth/login">', body)
|
||||
self.assertIn("Remote access token", body)
|
||||
self.assertIn("Username", body)
|
||||
self.assertIn("Password", body)
|
||||
|
||||
def test_remote_login_post_sets_cookie_and_redirects(self):
|
||||
body = b"token=secret-token&next=%2Fwatchlist"
|
||||
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)
|
||||
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{"ANI_CLI_WEB_ALLOW_REMOTE": "1", "ANI_CLI_WEB_REMOTE_TOKEN": "secret-token"},
|
||||
{
|
||||
"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)
|
||||
@@ -195,9 +189,9 @@ class NetworkGuardTests(unittest.TestCase):
|
||||
headers = dict(handler.response_headers)
|
||||
self.assertEqual(handler.response_status, HTTPStatus.SEE_OTHER)
|
||||
self.assertEqual(headers.get("Location"), "/watchlist")
|
||||
self.assertIn("ani_cli_web_token=secret-token", headers.get("Set-Cookie", ""))
|
||||
self.assertIn("ani_cli_web_session=", headers.get("Set-Cookie", ""))
|
||||
|
||||
def test_loopback_proxy_with_forwarded_remote_ip_requires_token(self):
|
||||
def test_loopback_proxy_with_forwarded_remote_ip_requires_credentials(self):
|
||||
handler = DummyHandler("/api/config")
|
||||
handler.client_address = ("127.0.0.1", 8421)
|
||||
handler.headers["X-Forwarded-For"] = "203.0.113.10"
|
||||
@@ -206,7 +200,7 @@ class NetworkGuardTests(unittest.TestCase):
|
||||
with self.assertRaises(PermissionError) as ctx:
|
||||
APP.Handler.ensure_client_access(handler)
|
||||
|
||||
self.assertIn("REMOTE_TOKEN", str(ctx.exception))
|
||||
self.assertIn("AUTH_USERNAME", str(ctx.exception))
|
||||
|
||||
|
||||
class DownloadQueueCancelTests(unittest.TestCase):
|
||||
@@ -1767,6 +1761,33 @@ class HandlerRouteTests(unittest.TestCase):
|
||||
["runtime_error", "watchlist_refresh_failed"],
|
||||
)
|
||||
|
||||
def test_config_routes_hide_auth_fields_and_preserve_saved_credentials(self):
|
||||
APP.save_runtime_config(
|
||||
{
|
||||
"download_dir": "/tmp/example",
|
||||
"mode": "sub",
|
||||
"quality": "best",
|
||||
"auth_username": "demo",
|
||||
"auth_password": "secret-pass",
|
||||
}
|
||||
)
|
||||
|
||||
post_body = b'{"download_dir":"/tmp/updated","mode":"dub","quality":"720"}'
|
||||
post_handler = DummyHandler("/api/config", body=post_body, content_length=len(post_body))
|
||||
APP.Handler.do_POST(post_handler)
|
||||
|
||||
self.assertEqual(post_handler.json_status, HTTPStatus.OK)
|
||||
self.assertNotIn("auth_username", post_handler.json_payload)
|
||||
self.assertNotIn("auth_password", post_handler.json_payload)
|
||||
self.assertEqual(APP.get_config_snapshot()["auth_username"], "demo")
|
||||
self.assertEqual(APP.get_config_snapshot()["auth_password"], "secret-pass")
|
||||
|
||||
get_handler = DummyHandler("/api/config")
|
||||
APP.Handler.do_GET(get_handler)
|
||||
self.assertEqual(get_handler.json_status, HTTPStatus.OK)
|
||||
self.assertNotIn("auth_username", get_handler.json_payload)
|
||||
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"]}'
|
||||
handler = DummyHandler("/api/config/webhook/test", body=body, content_length=len(body))
|
||||
@@ -1911,10 +1932,13 @@ class HandlerRouteTests(unittest.TestCase):
|
||||
handler = DummyHandler("/api/config")
|
||||
with mock.patch.object(http_handler, "debug_enabled", return_value=False), mock.patch.object(
|
||||
http_handler, "debug_log"
|
||||
) as debug_log, mock.patch.object(http_handler.traceback, "print_exc") as print_exc:
|
||||
) as debug_log, mock.patch.object(http_handler, "send_discord_webhook_event") as send_webhook, mock.patch.object(
|
||||
http_handler.traceback, "print_exc"
|
||||
) as print_exc:
|
||||
APP.Handler.exception(handler, RuntimeError("boom"))
|
||||
|
||||
print_exc.assert_not_called()
|
||||
send_webhook.assert_called_once()
|
||||
debug_log.assert_called_once()
|
||||
self.assertEqual(handler.error_status, HTTPStatus.INTERNAL_SERVER_ERROR)
|
||||
self.assertEqual(handler.error_message, "Internal server error.")
|
||||
|
||||
Reference in New Issue
Block a user