vmess-domain-rotator/tests/test_substore_operator.py

import json
import pathlib
import subprocess
import unittest

PROJECT_ROOT = pathlib.Path(__file__).resolve().parent.parent


class TestSubStoreOperatorBugs(unittest.TestCase):
    """验证 substore/operator_template.js 中的类型守卫、域名正则校验与时效降级"""

    def _eval_node(self, script):
        js_file = PROJECT_ROOT / "substore" / "operator_template.js"
        node_runner = f"""
        const fs = require('fs');
        const vm = require('vm');
        const code = fs.readFileSync({json.dumps(str(js_file))}, 'utf8');
        const sandbox = {{ console, Date, Number, String, Array, Object, RegExp, JSON, Error, setTimeout, clearTimeout }};
        vm.createContext(sandbox);
        vm.runInContext(code, sandbox);
        {script}
        """
        proc = subprocess.run(["node", "-e", node_runner], capture_output=True, text=True)
        return proc

    def test_p1_extract_value_object_stringification_hazard_prevented(self):
        """[P1 修复验证] 验证 extractValue() 拒绝非字符串对象，杜绝 '[object object]' 注入"""
        script = """
        try {
            const obj = { domain: { host: "cf.example.com" } };
            sandbox.extractValue(obj, 'server');
            console.log(JSON.stringify({ ok: true }));
        } catch (e) {
            console.log(JSON.stringify({ ok: false, error: e.message }));
        }
        """
        proc = self._eval_node(script)
        self.assertEqual(proc.returncode, 0, f"Node 执行出错: {proc.stderr}")
        res = json.loads(proc.stdout)
        self.assertFalse(res["ok"], "非字符串对象必须被拒绝")
        self.assertIn("empty value field", res["error"])

    def test_p1_extract_value_host_validation_enforced(self):
        """[P1 修复验证] 验证 extractValue() 严格校验域名与 IP 格式，拒绝非法字符与报错页面"""
        script = """
        const results = {};
        // 1. HTML 报错应被拒绝
        try {
            sandbox.extractValue({ domain: "<html>502 Bad Gateway</html>" }, 'server');
            results.html = "allowed";
        } catch (e) {
            results.html = "rejected";
        }

        // 2. 空格/非法字符应被拒绝
        try {
            sandbox.extractValue({ domain: "bad domain with spaces" }, 'server');
            results.spaces = "allowed";
        } catch (e) {
            results.spaces = "rejected";
        }

        // 3. 合法域名应正常提取
        try {
            results.valid_domain = sandbox.extractValue({ domain: "SAAS.SIN.FAN" }, 'server');
        } catch (e) {
            results.valid_domain = "failed: " + e.message;
        }

        // 4. 合法 IPv4 应正常提取
        try {
            results.valid_ip = sandbox.extractValue({ ip: "104.23.141.57" }, 'server');
        } catch (e) {
            results.valid_ip = "failed: " + e.message;
        }

        console.log(JSON.stringify(results));
        """
        proc = self._eval_node(script)
        self.assertEqual(proc.returncode, 0, f"Node 执行出错: {proc.stderr}")
        res = json.loads(proc.stdout)
        self.assertEqual(res["html"], "rejected", "HTML 标签必须被拒绝")
        self.assertEqual(res["spaces"], "rejected", "包含空格的非法 host 必须被拒绝")
        self.assertEqual(res["valid_domain"], "saas.sin.fan", "合法域名小写化正常提取")
        self.assertEqual(res["valid_ip"], "104.23.141.57", "合法 IP 正常提取")

    def test_p2_operator_template_checks_staleness(self):
        """[P2 修复验证] 验证 operator_template.js 校验 updated_at 时效，超时自动判定为 stale"""
        script = """
        vm.runInContext(`
            const now = Date.now();
            const freshDate = new Date(now - 3600 * 1000).toISOString(); // 1 小时前
            const staleDate = new Date(now - 100000 * 1000).toISOString(); // 27 小时前 (> 86400s)

            const freshResult = isRouterDataFresh({ updated_at: freshDate });
            const staleResult = isRouterDataFresh({ updated_at: staleDate });

            console.log(JSON.stringify({ freshResult, staleResult, maxAge: LOCAL_MAX_AGE_SEC }));
        `, sandbox);
        """
        proc = self._eval_node(script)
        self.assertEqual(proc.returncode, 0, f"Node 执行出错: {proc.stderr}")
        res = json.loads(proc.stdout)
        self.assertTrue(res["freshResult"], "1 小时前数据判定为 fresh")
        self.assertFalse(res["staleResult"], "27 小时前数据必须判定为 stale 并触发自动降级")
        self.assertEqual(res["maxAge"], 86400, "LOCAL_MAX_AGE_SEC 默认为 86400 秒")


if __name__ == "__main__":
    unittest.main()