tests/test_auth.py
Rama master · réplica de lectura del repositorio original.
"""口令的生成、存放与取用，以及绑局域网时的启动闸门。
闸门本身（`require_auth` 三兄弟怎么判、401 有几种形态）在 `test_fastapi_api.py`；
这里只管「这台机器有没有口令、`peach serve` 该不该起得来」。
"""
from __future__ import annotations
import io
import os
import tempfile
import unittest
from contextlib import redirect_stdout
from pathlib import Path
from unittest import mock
from peach import auth, cli, settings_file
class TokenFileTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.secrets = Path(self.tmp.name).resolve() / "secrets"
def test_a_missing_file_reads_as_no_token(self):
self.assertEqual(auth.read_token(self.secrets), "")
def test_a_blank_file_reads_as_no_token(self):
self.secrets.mkdir(parents=True)
auth.token_path(self.secrets).write_text(" \n", encoding="utf-8")
self.assertEqual(auth.read_token(self.secrets), "")
def test_a_new_token_carries_at_least_256_bit(self):
token = auth.write_token(self.secrets)
# token_urlsafe(32) 出 43 个字符；断言字符数而不是熵，是因为长度是这里唯一
# 能直接观察的量，而它由 TOKEN_BYTES 唯一决定。
self.assertGreaterEqual(len(token), 43)
self.assertTrue(all(c.isalnum() or c in "-_" for c in token), token)
self.assertEqual(auth.read_token(self.secrets), token)
def test_ensure_keeps_the_token_already_on_disk(self):
first, created = auth.ensure_token(self.secrets)
self.assertTrue(created)
second, created_again = auth.ensure_token(self.secrets)
self.assertEqual(second, first)
self.assertFalse(created_again)
def test_writing_again_replaces_the_token(self):
first = auth.write_token(self.secrets)
self.assertNotEqual(auth.write_token(self.secrets), first)
@unittest.skipIf(os.name == "nt", "Windows 的 chmod 只动只读位，权限靠 Peach 数据目录本身")
def test_the_token_file_is_owner_only(self):
auth.write_token(self.secrets)
self.assertEqual(auth.token_path(self.secrets).stat().st_mode & 0o777, 0o600)
def test_resolution_order_is_flag_then_environment_then_file(self):
auth.write_token(self.secrets)
on_disk = auth.read_token(self.secrets)
with mock.patch.dict(os.environ, {"PEACH_TOKEN": "from-env"}):
self.assertEqual(auth.resolve_token("from-flag", self.secrets), "from-flag")
self.assertEqual(auth.resolve_token("", self.secrets), "from-env")
with mock.patch.dict(os.environ, {}, clear=True):
self.assertEqual(auth.resolve_token("", self.secrets), on_disk)
class ServeGateTests(unittest.TestCase):
"""绑局域网地址却没有口令时，`peach serve` 必须拒绝启动而不是告警。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve() / "peach-data"
self.config = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(self.root)})
patcher = mock.patch.object(cli.settings_file, "active", return_value=self.config)
patcher.start()
self.addCleanup(patcher.stop)
env = mock.patch.dict(os.environ, {}, clear=True)
env.start()
self.addCleanup(env.stop)
def _args(self, *extra):
return cli.build_parser().parse_args(["serve", *extra])
def _secrets(self) -> Path:
return self.config.directory("secrets")
def test_a_lan_bind_without_a_token_refuses_to_start(self):
with self.assertRaises(SystemExit) as raised:
cli._serve_token(self._args("--host", "0.0.0.0"))
message = str(raised.exception)
self.assertIn("0.0.0.0", message)
self.assertIn("peach token", message)
# 拒绝的时候不许顺手把口令生成掉：那等于把硬门槛降成一句提示。
self.assertFalse(auth.token_path(self._secrets()).exists())
def test_a_loopback_bind_runs_without_a_token(self):
self.assertEqual(cli._serve_token(self._args("--host", "127.0.0.1")), "")
self.assertEqual(cli._serve_token(self._args("--host", "localhost")), "")
def test_a_lan_bind_takes_the_token_from_the_file(self):
token = auth.write_token(self._secrets())
self.assertEqual(cli._serve_token(self._args("--host", "0.0.0.0")), token)
def test_an_explicit_flag_still_wins(self):
auth.write_token(self._secrets())
self.assertEqual(
cli._serve_token(self._args("--host", "0.0.0.0", "--token", "given")), "given")
def test_no_auth_skips_the_login_page_on_loopback_and_leaves_the_token_file_alone(self):
token = auth.write_token(self._secrets())
self.assertEqual(cli._serve_token(self._args("--host", "127.0.0.1", "--no-auth")), "")
# 口令文件是托盘那份服务的门槛，开发时跳过登录不该把它擦掉或改写。
self.assertEqual(auth.read_token(self._secrets()), token)
def test_no_auth_refuses_a_lan_bind_even_when_a_token_exists(self):
auth.write_token(self._secrets())
with self.assertRaises(SystemExit) as raised:
cli._serve_token(self._args("--host", "0.0.0.0", "--no-auth"))
self.assertIn("0.0.0.0", str(raised.exception))
class TokenCommandTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve() / "peach-data"
self.config = settings_file.load_config(environ={"PEACH_DATA_ROOT": str(self.root)})
patcher = mock.patch.object(cli.settings_file, "active", return_value=self.config)
patcher.start()
self.addCleanup(patcher.stop)
def _run(self, *extra):
output = io.StringIO()
with redirect_stdout(output):
code = cli.main(["token", *extra])
return code, output.getvalue()
def test_it_creates_and_prints_a_token(self):
code, output = self._run()
self.assertEqual(code, 0)
token = auth.read_token(self.config.directory("secrets"))
self.assertTrue(token)
self.assertIn(token, output)
def test_reading_twice_gives_the_same_token(self):
self._run()
token = auth.read_token(self.config.directory("secrets"))
_code, again = self._run()
self.assertIn(token, again)
self.assertEqual(auth.read_token(self.config.directory("secrets")), token)
def test_rotate_replaces_the_token(self):
self._run()
before = auth.read_token(self.config.directory("secrets"))
_code, output = self._run("--rotate")
after = auth.read_token(self.config.directory("secrets"))
self.assertNotEqual(after, before)
self.assertIn(after, output)
class InitProvisionsTokenTests(unittest.TestCase):
"""两条 init 路径都要留下口令文件——现有部署走的正是 `--from-existing`。"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve() / "peach-data"
patcher = mock.patch.object(cli.onboarding.certs, "bootstrap_certificates")
patcher.start()
self.addCleanup(patcher.stop)
def _init(self, *extra):
output = io.StringIO()
with redirect_stdout(output):
code = cli.main(["init", "--data-root", str(self.root), *extra])
return code, output.getvalue()
def test_a_fresh_init_leaves_a_token_behind(self):
code, output = self._init()
self.assertEqual(code, 0)
token = auth.read_token(self.root / "secrets")
self.assertTrue(token)
# 口令本身不进终端输出：那条路通向日志、截图和 issue。
self.assertNotIn(token, output)
self.assertIn("peach token", output)
def test_from_existing_leaves_a_token_behind(self):
code, _output = self._init("--from-existing")
self.assertEqual(code, 0)
self.assertTrue(auth.read_token(self.root / "secrets"))
if __name__ == "__main__":
unittest.main()