fix(chv-muse-r12fix3-2): worker socket 组权限 + allowed_uids 含 chv(P6 缺口 ②)

R12 复测 2 钉死:即使 worker active,chv 客户端也永远连不上
socket——worker root + 目录 /run/chv-gitea 0700 root + socket
0600 root + allowed_uids=(0,) 三处锁死;chv ping 一律 EACCES
(rc=2),root ping rc=0,privsep_worker_call 恒为 0,全部同步
走回落。

修法:三处配套改,只改一处仍锁死——
1. chv-gitea-worker.py:DEFAULT_ALLOWED_UIDS=(0, 1000)(root +
   chv;round-11 authd 同款;环境变量
   CHV_GITEA_WORKER_ALLOWED_UIDS 可覆盖);socket 用 group 权限
   (目录 0750、socket 0660、组为第一个非 root 放行 uid 的主组,
   _socket_group_gid/_set_group_perms 与 authd 同款实现;gid
   解析不出时回落 root-only 0700/0600,绝不放给全员)。
   拒绝 0666 探针口径。
2. build.sh tmpfiles.d:d /run/chv-gitea 改为 0750 root 1000,与
   新方案一致。gid 写数字 1000 而不用组名 chv——tmpfiles-setup
   跑在 07:23:24,chv 用户由 cloud-init 在 07:23:40 才建,组名
   当时解析不出;1000 是 chv 的 uid/gid(seed users 块未指定
   uid/gid,cloud-init 取首个空闲 1000;authd 同款假设)。
   worker 启动时再按 passwd 重新推导 gid 并重设权限,tmpfiles
   行只是开机存在性保证。审计日志仍 0600 root,不动。
3. 服务单元 ReadWritePaths 不动(/run/chv-gitea 仍在列,存在性
   由 tmpfiles 保证)。

注意:单元内 agent 同样映射为 chv uid 1000,故也能走到 socket,
但只能走类型化协议——操作白名单、仓库 confinement、安全扫描三
处仍在,worker 不持凭证(见 worker 模块注释)。

测试:WorkerSocketGroupPermsTest 新增 7 项——默认白名单含
(0,1000)、环境变量覆盖、gid 可推导时 0750/0660 + 组正确、显式
socket_group 时同样组权限、只有 root 时回落 0700/0600、gid 推导
函数四档、tmpfiles 行 0750 root 1000;WorkerUnitTmpfilesTest 的
/run/chv-gitea 期望改 0750。P6/round12/P2/P4 四模块 205 项全过。
This commit is contained in:
host committed 2026-11-08 09:48:16 +00:00
1 parent e31354dc58
commit 7cba3c2b21
4 files changed
+243 -27

No files matched your search

+141 -1
View File
@@ -844,7 +844,7 @@ class WorkerUnitTmpfilesTest(unittest.TestCase):
# /run 下的目录(tmpfs,开机不存在)须有 d 行;
# 文件路径(worker 首次写审计才建)须有 f 行。
self.assertEqual(entries.get("/run/chv-gitea"),
("d", "0700"))
("d", "0750"))
self.assertEqual(entries.get("/var/log/chv-gitea-worker.log"),
("f", "0600"))
for path in paths:
@@ -957,6 +957,146 @@ class WorkerUnitBootRaceTest(unittest.TestCase):
(REPO_ROOT / "chv-guest" / name).read_text())
# -- (7d) worker socket 组权限 ----------------------------------------------
class WorkerSocketGroupPermsTest(unittest.TestCase):
"""round-12 修复轮第三轮(缺口 ②,P6):R12 复测 2 钉死——worker
root + 目录 0700 root + socket 0600 root + allowed_uids=(0,)
三处锁死,chv 客户端永远 EACCES,全部同步走回落。
修法:三处配套改,只改一处仍锁死——目录 0750 + socket 0660
(组为第一个非 root 放行 uid 的主组,round-11 authd 同款口径,
拒绝 0666)+ allowed_uids 默认含 chv(0, 1000)。gid 解析不
出时回落 root-only(0700/0600),绝不放给全员。
"""
def _build(self, tmp, allowed_uids, socket_group=None):
mod = _load_guest_worker()
helper = os.path.join(tmp, "helper")
Path(helper).write_text("#!/bin/sh\n")
os.makedirs(os.path.join(tmp, "workspace"), exist_ok=True)
server = mod.build_server(
socket_path=os.path.join(tmp, "worker.sock"),
workspace=os.path.join(tmp, "workspace"),
log_path=os.path.join(tmp, "worker.log"),
lib_dir=str(REPO_ROOT),
allowed_uids=allowed_uids,
run_git=lambda a, r: (0, "ok\n", ""),
helper_path=helper,
socket_group=socket_group)
return mod, server
def _mode(self, path):
return oct(os.stat(path).st_mode & 0o777)
def _first_non_root_uid(self):
# 测试跑在 root 身份下(本 VM 的 unittest 即 root),
# os.getuid() 不是可推导的非 root uid——从 passwd 里动态
# 取第一个非 root 用户;没有即跳过(不在本机硬编 uid)。
import pwd
for entry in pwd.getpwall():
if entry.pw_uid != 0:
return entry.pw_uid, entry.pw_gid
return None, None
def test_default_allowed_uids_include_chv(self):
import unittest.mock as mock
mod = _load_guest_worker()
with tempfile.TemporaryDirectory() as tmp:
with mock.patch.dict(os.environ):
os.environ.pop("CHV_GITEA_WORKER_ALLOWED_UIDS", None)
_, server = self._build(tmp, None)
try:
self.assertEqual(server.allowed_uids,
frozenset({0, 1000}))
finally:
server.server_close()
def test_env_override_allowed_uids(self):
import unittest.mock as mock
mod = _load_guest_worker()
with tempfile.TemporaryDirectory() as tmp:
with mock.patch.dict(
os.environ,
{"CHV_GITEA_WORKER_ALLOWED_UIDS": "0"}):
_, server = self._build(tmp, None)
try:
self.assertEqual(server.allowed_uids, frozenset({0}))
finally:
server.server_close()
def test_group_perms_when_gid_resolvable(self):
uid, gid = self._first_non_root_uid()
if uid is None:
self.skipTest("passwd 里无非 root 用户,推导无从测试")
with tempfile.TemporaryDirectory() as tmp:
_, server = self._build(tmp, [0, uid])
try:
self.assertEqual(server.socket_gid, gid)
self.assertEqual(self._mode(tmp), "0o750")
sock = os.path.join(tmp, "worker.sock")
self.assertEqual(self._mode(sock), "0o660")
self.assertEqual(os.stat(sock).st_gid, gid)
finally:
server.server_close()
def test_group_perms_with_explicit_socket_group(self):
# socket_group 显式给组名时不走 uid 推导:allowed 只有 root
# 也能拿到组权限(0750/0660),gid 即该组。
import grp
with tempfile.TemporaryDirectory() as tmp:
_, server = self._build(tmp, [0], socket_group="root")
try:
self.assertEqual(server.socket_gid,
grp.getgrnam("root").gr_gid)
self.assertEqual(self._mode(tmp), "0o750")
self.assertEqual(
self._mode(os.path.join(tmp, "worker.sock")),
"0o660")
finally:
server.server_close()
def test_root_only_fallback_when_gid_unresolvable(self):
with tempfile.TemporaryDirectory() as tmp:
_, server = self._build(tmp, [0])
try:
self.assertIsNone(server.socket_gid)
self.assertEqual(self._mode(tmp), "0o700")
self.assertEqual(
self._mode(os.path.join(tmp, "worker.sock")),
"0o600")
finally:
server.server_close()
def test_socket_group_gid_derivation(self):
import grp
mod = _load_guest_worker()
uid, gid = self._first_non_root_uid()
if uid is None:
self.skipTest("passwd 里无非 root 用户,推导无从测试")
self.assertEqual(mod._socket_group_gid([0, uid]), gid)
self.assertEqual(
mod._socket_group_gid([0, uid], socket_group="root"),
grp.getgrnam("root").gr_gid)
self.assertIsNone(mod._socket_group_gid([0]))
self.assertIsNone(
mod._socket_group_gid([0, uid],
socket_group="no-such-group-xyz"))
def test_tmpfiles_worker_dir_group_perms(self):
# tmpfiles 行须与新方案一致:0750 root + chv 的 gid。gid 写
# 数字 1000 而不用组名——tmpfiles-setup 时 chv 用户还不存在,
# 组名解析不出;worker 启动时再按 passwd 重设。
text = (REPO_ROOT / "chv-guest" / "build.sh").read_text()
match = re.search(r"d /run/chv-gitea (\S+) (\S+) (\S+)", text)
self.assertIsNotNone(match, "chv.conf 里找不到 /run/chv-gitea 行")
self.assertEqual(match.groups(), ("0750", "root", "1000"))
class BuildInfoContentsTest(unittest.TestCase):
"""round-12 修复轮(项 2,P6):worker 三件须进 BUILDINFO
contents(无条件拷入,与 P2/P4 同款钉法)。"""