221 lines
9.9 KiB
Python
221 lines
9.9 KiB
Python
import copy
|
|
import json
|
|
import math
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from services.config import (
|
|
DEFAULT_PROXY_RUNTIME,
|
|
ConfigStore,
|
|
_normalize_proxy_runtime_settings,
|
|
)
|
|
|
|
|
|
class ProxyRuntimeConfigTests(unittest.TestCase):
|
|
def _make_store(self, initial: dict[str, object] | None = None) -> tuple[tempfile.TemporaryDirectory[str], ConfigStore]:
|
|
tmp_dir = tempfile.TemporaryDirectory()
|
|
path = Path(tmp_dir.name) / "config.json"
|
|
data = {"auth-key": "test-auth"}
|
|
if initial:
|
|
data.update(initial)
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
return tmp_dir, ConfigStore(path)
|
|
|
|
def test_defaults_are_safe_and_included_in_public_config(self) -> None:
|
|
tmp_dir, store = self._make_store({"proxy": " http://legacy.example:8080 "})
|
|
with tmp_dir:
|
|
expected_default = copy.deepcopy(DEFAULT_PROXY_RUNTIME)
|
|
runtime = store.get_proxy_runtime_settings()
|
|
self.assertEqual(runtime, expected_default)
|
|
self.assertEqual(store.get_proxy_settings(), "http://legacy.example:8080")
|
|
|
|
expected_public = copy.deepcopy(expected_default)
|
|
expected_public["clearance"]["has_cf_cookies"] = False
|
|
expected_public["clearance"]["has_cf_clearance"] = False
|
|
public_config = store.get()
|
|
self.assertEqual(public_config["proxy"], " http://legacy.example:8080 ")
|
|
self.assertEqual(public_config["proxy_runtime"], expected_public)
|
|
self.assertNotIn("auth-key", public_config)
|
|
|
|
runtime["enabled"] = True
|
|
runtime["clearance"]["enabled"] = True
|
|
self.assertEqual(DEFAULT_PROXY_RUNTIME, expected_default)
|
|
self.assertEqual(store.get_proxy_runtime_settings(), expected_default)
|
|
|
|
def test_normalize_proxy_runtime_sanitizes_invalid_values(self) -> None:
|
|
normalized = _normalize_proxy_runtime_settings(
|
|
{
|
|
"enabled": "yes",
|
|
"egress_mode": "tor",
|
|
"proxy_url": " http://proxy.example:8080 ",
|
|
"resource_proxy_url": " socks5://resource.example:1080 ",
|
|
"skip_ssl_verify": "on",
|
|
"reset_session_status_codes": ["403", 429, 99, 600, "bad", None, True],
|
|
"clearance": {
|
|
"enabled": "0",
|
|
"mode": "manual",
|
|
"cf_cookies": " a=b; c=d ",
|
|
"cf_clearance": " token ",
|
|
"user_agent": " Custom UA ",
|
|
"browser": " firefox ",
|
|
"flaresolverr_url": " http://flare.example/ ",
|
|
"timeout_sec": 0,
|
|
"refresh_interval": 59,
|
|
"warm_up_on_start": "true",
|
|
},
|
|
}
|
|
)
|
|
|
|
self.assertTrue(normalized["enabled"])
|
|
self.assertEqual(normalized["egress_mode"], "direct")
|
|
self.assertEqual(normalized["proxy_url"], "http://proxy.example:8080")
|
|
self.assertEqual(normalized["resource_proxy_url"], "socks5://resource.example:1080")
|
|
self.assertTrue(normalized["skip_ssl_verify"])
|
|
self.assertEqual(normalized["reset_session_status_codes"], [403, 429])
|
|
|
|
clearance = normalized["clearance"]
|
|
self.assertFalse(clearance["enabled"])
|
|
self.assertEqual(clearance["mode"], "manual")
|
|
self.assertEqual(clearance["cf_cookies"], "a=b; c=d")
|
|
self.assertEqual(clearance["cf_clearance"], "token")
|
|
self.assertEqual(clearance["user_agent"], "Custom UA")
|
|
self.assertEqual(clearance["browser"], "firefox")
|
|
self.assertEqual(clearance["flaresolverr_url"], "http://flare.example/")
|
|
self.assertEqual(clearance["timeout_sec"], 1)
|
|
self.assertEqual(clearance["refresh_interval"], 60)
|
|
self.assertTrue(clearance["warm_up_on_start"])
|
|
|
|
def test_normalize_proxy_runtime_uses_defaults_for_missing_or_empty_values(self) -> None:
|
|
self.assertEqual(_normalize_proxy_runtime_settings(None), DEFAULT_PROXY_RUNTIME)
|
|
self.assertEqual(
|
|
_normalize_proxy_runtime_settings(
|
|
{
|
|
"egress_mode": "single_proxy",
|
|
"reset_session_status_codes": ["bad", 99, 600],
|
|
"clearance": {
|
|
"enabled": True,
|
|
"mode": "invalid",
|
|
"user_agent": "",
|
|
"browser": "",
|
|
"timeout_sec": "bad",
|
|
"refresh_interval": "bad",
|
|
},
|
|
}
|
|
),
|
|
{
|
|
**DEFAULT_PROXY_RUNTIME,
|
|
"egress_mode": "single_proxy",
|
|
"reset_session_status_codes": [403],
|
|
"clearance": {
|
|
**DEFAULT_PROXY_RUNTIME["clearance"],
|
|
"enabled": True,
|
|
"mode": "none",
|
|
},
|
|
},
|
|
)
|
|
|
|
def test_malformed_proxy_runtime_values_fall_back_safely(self) -> None:
|
|
normalized = _normalize_proxy_runtime_settings(
|
|
{
|
|
"enabled": "maybe",
|
|
"skip_ssl_verify": "maybe",
|
|
"reset_session_status_codes": [math.inf, "NaN", "403"],
|
|
"clearance": {
|
|
"enabled": "maybe",
|
|
"timeout_sec": math.inf,
|
|
"refresh_interval": -math.inf,
|
|
"warm_up_on_start": "maybe",
|
|
},
|
|
}
|
|
)
|
|
|
|
self.assertFalse(normalized["enabled"])
|
|
self.assertFalse(normalized["skip_ssl_verify"])
|
|
self.assertEqual(normalized["reset_session_status_codes"], [403])
|
|
clearance = normalized["clearance"]
|
|
self.assertFalse(clearance["enabled"])
|
|
self.assertEqual(clearance["timeout_sec"], DEFAULT_PROXY_RUNTIME["clearance"]["timeout_sec"])
|
|
self.assertEqual(clearance["refresh_interval"], DEFAULT_PROXY_RUNTIME["clearance"]["refresh_interval"])
|
|
self.assertFalse(clearance["warm_up_on_start"])
|
|
|
|
def test_update_normalizes_and_persists_proxy_runtime(self) -> None:
|
|
tmp_dir, store = self._make_store()
|
|
with tmp_dir:
|
|
public_config = store.update(
|
|
{
|
|
"proxy_runtime": {
|
|
"enabled": "true",
|
|
"egress_mode": "single_proxy",
|
|
"proxy_url": " http://proxy.example ",
|
|
"reset_session_status_codes": [401, "403", "nope"],
|
|
"clearance": {
|
|
"enabled": "yes",
|
|
"mode": "flaresolverr",
|
|
"flaresolverr_url": " http://localhost:8191 ",
|
|
"timeout_sec": "30",
|
|
"refresh_interval": "120",
|
|
},
|
|
}
|
|
}
|
|
)
|
|
|
|
expected = {
|
|
**DEFAULT_PROXY_RUNTIME,
|
|
"enabled": True,
|
|
"egress_mode": "single_proxy",
|
|
"proxy_url": "http://proxy.example",
|
|
"reset_session_status_codes": [401, 403],
|
|
"clearance": {
|
|
**DEFAULT_PROXY_RUNTIME["clearance"],
|
|
"enabled": True,
|
|
"mode": "flaresolverr",
|
|
"flaresolverr_url": "http://localhost:8191",
|
|
"timeout_sec": 30,
|
|
"refresh_interval": 120,
|
|
},
|
|
}
|
|
expected_public = copy.deepcopy(expected)
|
|
expected_public["clearance"]["cf_cookies"] = ""
|
|
expected_public["clearance"]["cf_clearance"] = ""
|
|
expected_public["clearance"]["has_cf_cookies"] = False
|
|
expected_public["clearance"]["has_cf_clearance"] = False
|
|
self.assertEqual(public_config["proxy_runtime"], expected_public)
|
|
|
|
raw_saved = json.loads(store.path.read_text(encoding="utf-8"))
|
|
self.assertEqual(raw_saved["proxy_runtime"], expected)
|
|
reloaded = ConfigStore(store.path)
|
|
self.assertEqual(reloaded.get_proxy_runtime_settings(), expected)
|
|
|
|
def test_public_proxy_runtime_redacts_and_preserves_existing_clearance_values(self) -> None:
|
|
existing = copy.deepcopy(DEFAULT_PROXY_RUNTIME)
|
|
existing["enabled"] = True
|
|
existing["clearance"]["enabled"] = True
|
|
existing["clearance"]["mode"] = "manual"
|
|
existing["clearance"]["cf_cookies"] = "session=secret-cookie"
|
|
existing["clearance"]["cf_clearance"] = "secret-clearance"
|
|
tmp_dir, store = self._make_store({"proxy_runtime": existing})
|
|
with tmp_dir:
|
|
public_config = store.get()
|
|
public_runtime = public_config["proxy_runtime"]
|
|
public_clearance = public_runtime["clearance"]
|
|
self.assertEqual(public_clearance["cf_cookies"], "")
|
|
self.assertEqual(public_clearance["cf_clearance"], "")
|
|
self.assertTrue(public_clearance["has_cf_cookies"])
|
|
self.assertTrue(public_clearance["has_cf_clearance"])
|
|
self.assertNotIn("secret-cookie", json.dumps(public_config))
|
|
self.assertNotIn("secret-clearance", json.dumps(public_config))
|
|
|
|
public_clearance["user_agent"] = "Updated UA"
|
|
updated_public = store.update({"proxy_runtime": public_runtime})
|
|
updated_raw = json.loads(store.path.read_text(encoding="utf-8"))["proxy_runtime"]
|
|
self.assertEqual(updated_raw["clearance"]["cf_cookies"], "session=secret-cookie")
|
|
self.assertEqual(updated_raw["clearance"]["cf_clearance"], "secret-clearance")
|
|
self.assertEqual(updated_raw["clearance"]["user_agent"], "Updated UA")
|
|
self.assertEqual(updated_public["proxy_runtime"]["clearance"]["cf_cookies"], "")
|
|
self.assertEqual(updated_public["proxy_runtime"]["clearance"]["cf_clearance"], "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|