feat: 用户名登录——一账号一个牧场,登录后改 URL 也进不了别人的档

多用户之前是「知道牧场名就能玩」:?farm=boss 直接进别人家里。
现在网页端用用户名登录(用户选择:不设密码、一账号一个牧场,
用户名即身份——互信小圈子模型,防的是随手改 URL,不是蓄意攻击)。

* 新增 farm/server/auth.py:HMAC-SHA256 签名的 HttpOnly Cookie
  (sf_auth,30 天,恒时比较防篡改)。密钥存 <save_dir>/secret.key,
  重启不掉线,也不用建表存 token。
* app.py 三个认 ?farm= 的地方(session_of / SSE / POST /api/farms)
  统一规则:有 Cookie 强制本人的牧场(?farm=/X-Farm-Id 被忽略),
  无 Cookie 保持旧语义——bot/TUI/CLI 脚本零改动照常工作。
* 新路由:POST /api/login(全局 token 照常把它挡在门外)、
  POST /api/logout、GET /api/me(刻意不 guard:首屏只问"登录没")。
* 网页:移除牧场名输入框,改为登录遮罩 + 用户徽章 + 退出按钮;
  首屏先 whoami() 再决定连接还是弹登录。

测试 +13:Cookie 伪造/过期/跨密钥/垃圾串/非法用户名/密钥跨重启复用/
session_of 优先级(starlette Request 用 scope dict 直接构造,不需要
httpx)9 项纯逻辑档 + 4 项 TestClient 档。528 项 OK(skip 39)。

真机验收 16/16:登录/HttpOnly/?farm= 被忽略/动作落自己牧场/篡改
退回未登录/退出/非法名 400;跨重启 Cookie 仍有效且进度保留。
This commit is contained in:
snowy-yang committed 2026-09-23 21:20:25 +08:00
1 parent d1692d7b10
commit 698a5e497a
7 files changed
+391 -215

No files matched your search

+97
View File
@@ -24,6 +24,7 @@ import urllib.request
from farm import config, net
from farm.engine import GameError
from farm.server.auth import COOKIE_NAME, LoginManager
from farm.server.daemon import DaemonManager, ensure_parent_dir, pid_alive, port_busy
from farm.server.hub import (
CLOSE_MESSAGE,
@@ -53,6 +54,7 @@ try: # pragma: no cover - 环境相关
ServerConfig,
ServerHandle,
_int,
_Services,
as_asgi,
brief,
create_app,
@@ -1111,6 +1113,28 @@ class AuthTestCase(unittest.TestCase):
response = self.client.post("/api/actions?farm=alice", json={"action": "buy_feed"})
self.assertEqual(response.status_code, 401)
def test_login_sets_cookie_and_me_reports_it(self):
response = self.client.post("/api/login?token=s3cret", json={"username": "carol"})
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["user"], "carol")
self.assertEqual(self.client.get("/api/me").json()["user"], "carol")
def test_login_needs_the_token_too(self):
# 令牌是整台服务器的门,登录不豁免
self.assertEqual(
self.client.post("/api/login", json={"username": "carol"}).status_code, 401
)
def test_logged_in_user_cannot_reach_other_farms(self):
self.client.post("/api/login?token=s3cret", json={"username": "carol"})
data = self.client.get("/api/state?token=s3cret&farm=boss").json()
self.assertEqual(data["farm"], "carol")
def test_logout_clears_the_cookie(self):
self.client.post("/api/login?token=s3cret", json={"username": "carol"})
self.client.post("/api/logout?token=s3cret")
self.assertIsNone(self.client.get("/api/me").json()["user"])
def test_remote_shutdown_can_be_switched_off(self):
app = create_app(
ServerConfig(
@@ -1122,6 +1146,79 @@ class AuthTestCase(unittest.TestCase):
self.assertEqual(response.status_code, 403)
@unittest.skipUnless(HAS_FASTAPI, "未安装 FastAPI")
class LoginCookieTestCase(unittest.TestCase):
"""登录 Cookie 的签发与校验(纯逻辑档,不需要 httpx 也能跑)。"""
def setUp(self):
from fastapi import Request
self.dir = tempfile.mkdtemp(prefix="simplefarm-login-")
self.auth = LoginManager(self.dir)
self._request_cls = Request
def request(self, cookie: str | None = None, query: bytes = b"", headers=()):
raw = [(b"host", b"test")]
if cookie:
raw.append((b"cookie", cookie.encode("ascii")))
raw.extend(headers)
return self._request_cls(
{"type": "http", "headers": raw, "query_string": query, "method": "GET"}
)
def test_issue_then_verify_roundtrips_the_username(self):
self.assertEqual(self.auth.verify(self.auth.issue("小明_2")), "小明_2")
def test_tampered_signature_is_rejected(self):
token = self.auth.issue("carol")
sig = token.rpartition(".")[2]
self.assertIsNone(self.auth.verify(token[:-1] + ("0" if sig[-1] != "0" else "1")))
def test_token_from_another_secret_is_rejected(self):
other = LoginManager(tempfile.mkdtemp(prefix="simplefarm-login2-"))
self.assertIsNone(other.verify(self.auth.issue("carol")))
def test_expired_token_is_rejected(self):
from farm.server import auth as auth_mod
saved = auth_mod.TTL_SECONDS
auth_mod.TTL_SECONDS = -10 # 签出来就是过期的
try:
token = self.auth.issue("carol")
finally:
auth_mod.TTL_SECONDS = saved
self.assertIsNone(self.auth.verify(token))
def test_garbage_tokens_are_rejected(self):
for bad in ("", "abc", "a.b", "!!!.###", self.auth.issue("carol") + "x"):
self.assertIsNone(self.auth.verify(bad))
def test_illegal_username_is_refused_at_issue(self):
with self.assertRaises(GameError):
self.auth.issue("有 空格")
with self.assertRaises(GameError):
self.auth.issue("x" * 30)
def test_secret_file_is_reused_across_restarts(self):
again = LoginManager(self.dir)
self.assertEqual(again.verify(self.auth.issue("carol")), "carol")
def test_session_of_prefers_the_cookie_over_the_url(self):
svc = _Services.create(ServerConfig(save_dir=self.dir, tick_seconds=2400.0))
cookie = f"{COOKIE_NAME}={self.auth.issue('carol')}"
self.assertEqual(
svc.session_of(self.request(cookie=cookie, query=b"farm=boss")).farm_id, "carol"
)
def test_session_of_falls_back_without_a_cookie(self):
svc = _Services.create(ServerConfig(save_dir=self.dir, tick_seconds=2400.0))
self.assertEqual(svc.session_of(self.request(query=b"farm=boss")).farm_id, "boss")
self.assertEqual(
svc.session_of(self.request(headers=[(b"x-farm-id", b"boss")])).farm_id, "boss"
)
self.assertEqual(svc.session_of(self.request()).farm_id, DEFAULT_FARM)
@unittest.skipUnless(HAS_UVICORN and HAS_FASTAPI, "未安装 uvicorn / FastAPI")
class LiveServerTestCase(unittest.TestCase):
"""真起一个服务器进程内的 uvicorn,用真 socket 说话。"""