211 lines
7.2 KiB
Python
211 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
||
"""Astrion Landlock 只读域 launcher（容器内运行，非后端模块）。
|
||
|
||
由后端 modules/docker_readonly_exec.py 部署到容器内（docker cp）并调用：
|
||
|
||
python3 launcher.py --ro /workspace -- <cmd> [args...]
|
||
python3 launcher.py --selftest --ro /workspace
|
||
|
||
语义（2026-09 云端实测，kernel 6.8 / Landlock ABI V4 / Docker 28 默认 seccomp 放行）：
|
||
|
||
- handled 集合只含写类操作（写文件/建删文件目录/创建设备节点/符号链接/
|
||
rename-link 跨越/截断），不授权的路径一律被内核拒绝；
|
||
- ro 路径（通常即工作区挂载点）不加任何规则 → 写类操作全拒；
|
||
- /tmp、/var/tmp、/dev/shm 显式授予写类权限 → 与纯 DAC 只读的历史行为对齐
|
||
（只读身份 HOME=/tmp，常见临时写入不受影响）；其余路径写权限收紧，这正是
|
||
要修复的 world-writable（777/o+w）绕 DAC 漏洞本身；
|
||
- 读/执行不进入 handled 集合，完全交给 DAC（600 权限敏感文件仍不可读）；
|
||
- 域随 fork/exec 继承且不可自行解除（配合 no_new_privs），子进程同受限。
|
||
|
||
注：不能采用「/ 授全量 + ro 路径授空」的交集写法——landlock_add_rule 对
|
||
allowed_access=0 的规则返回 ENOMSG（errno 42），内核拒绝空授权规则。
|
||
|
||
仅依赖 python3 标准库（ctypes 直调 syscall），x86_64/aarch64 通用。
|
||
"""
|
||
import ctypes
|
||
import os
|
||
import sys
|
||
|
||
libc = ctypes.CDLL(None, use_errno=True)
|
||
|
||
PR_SET_NO_NEW_PRIVS = 38
|
||
|
||
# x86_64 与 aarch64 编号一致
|
||
SYS_CREATE_RULESET = 444
|
||
SYS_ADD_RULE = 445
|
||
SYS_RESTRICT_SELF = 446
|
||
|
||
LANDLOCK_RULE_PATH_BENEATH = 1
|
||
LANDLOCK_CREATE_RULESET_VERSION = 1 << 0
|
||
|
||
# access_fs 位（ABI V1 全集 + V2/V3 增量）
|
||
FS_EXECUTE = 1 << 0
|
||
FS_WRITE_FILE = 1 << 1
|
||
FS_READ_FILE = 1 << 2
|
||
FS_READ_DIR = 1 << 3
|
||
FS_REMOVE_DIR = 1 << 4
|
||
FS_REMOVE_FILE = 1 << 5
|
||
FS_MAKE_CHAR = 1 << 6
|
||
FS_MAKE_DIR = 1 << 7
|
||
FS_MAKE_REG = 1 << 8
|
||
FS_MAKE_SOCK = 1 << 9
|
||
FS_MAKE_FIFO = 1 << 10
|
||
FS_MAKE_BLOCK = 1 << 11
|
||
FS_MAKE_SYM = 1 << 12
|
||
FS_REFER = 1 << 13 # ABI V2：跨目录 rename/link
|
||
FS_TRUNCATE = 1 << 14 # ABI V3：truncate(2)
|
||
|
||
# 只读域要管的写类操作（读/执行刻意排除，留给 DAC 决定）
|
||
WRITE_OPS_V1 = (FS_WRITE_FILE | FS_REMOVE_DIR | FS_REMOVE_FILE |
|
||
FS_MAKE_CHAR | FS_MAKE_DIR | FS_MAKE_REG | FS_MAKE_SOCK |
|
||
FS_MAKE_FIFO | FS_MAKE_BLOCK | FS_MAKE_SYM)
|
||
|
||
|
||
class RulesetAttr(ctypes.Structure):
|
||
_fields_ = [("handled_access_fs", ctypes.c_uint64)]
|
||
|
||
|
||
class PathBeneathAttr(ctypes.Structure):
|
||
_fields_ = [("allowed_access", ctypes.c_uint64),
|
||
("parent_fd", ctypes.c_int32),
|
||
("_pad", ctypes.c_int32)]
|
||
|
||
|
||
def probe_abi() -> int:
|
||
"""返回内核 Landlock ABI 版本（>=1），不支持/被拦返回 <1。"""
|
||
ctypes.set_errno(0)
|
||
ret = libc.syscall(SYS_CREATE_RULESET, None, 0,
|
||
LANDLOCK_CREATE_RULESET_VERSION, 0, 0, 0)
|
||
return ret
|
||
|
||
|
||
def handled_for_abi(abi: int) -> int:
|
||
handled = WRITE_OPS_V1
|
||
if abi >= 2:
|
||
handled |= FS_REFER
|
||
if abi >= 3:
|
||
handled |= FS_TRUNCATE
|
||
return handled
|
||
|
||
|
||
def install_readonly_domain(ro_paths, rw_paths=("/tmp", "/var/tmp", "/dev/shm")):
|
||
"""安装只读域。成功返回 None，失败返回错误描述字符串。
|
||
|
||
ro_paths：写类操作全拒的路径（不加规则，靠「无覆盖即拒绝」生效）。
|
||
rw_paths：显式授予写类权限的路径（对齐纯 DAC 只读下的常用可写区）。
|
||
"""
|
||
abi = probe_abi()
|
||
if abi < 1:
|
||
return f"kernel-unsupported(errno={ctypes.get_errno()})"
|
||
handled = handled_for_abi(abi)
|
||
|
||
attr = RulesetAttr(handled)
|
||
ctypes.set_errno(0)
|
||
ruleset_fd = libc.syscall(SYS_CREATE_RULESET, ctypes.byref(attr),
|
||
ctypes.sizeof(attr), 0, 0, 0)
|
||
if ruleset_fd < 0:
|
||
e = ctypes.get_errno()
|
||
return f"create_ruleset(errno={e}:{os.strerror(e)})"
|
||
|
||
# 只给存在的 rw 路径加授权规则；ro 路径刻意不加规则（无覆盖 → 拒绝）
|
||
for path in rw_paths:
|
||
if not os.path.exists(path):
|
||
continue
|
||
try:
|
||
fd = os.open(path, os.O_PATH)
|
||
except OSError as e:
|
||
return f"open({path})(errno={e.errno}:{e.strerror})"
|
||
pba = PathBeneathAttr(handled, fd, 0)
|
||
ctypes.set_errno(0)
|
||
ret = libc.syscall(SYS_ADD_RULE, ruleset_fd, LANDLOCK_RULE_PATH_BENEATH,
|
||
ctypes.byref(pba), 0, 0, 0)
|
||
os.close(fd)
|
||
if ret < 0:
|
||
e = ctypes.get_errno()
|
||
return f"add_rule({path})(errno={e}:{os.strerror(e)})"
|
||
|
||
if libc.prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0) < 0:
|
||
return "prctl(PR_SET_NO_NEW_PRIVS) failed"
|
||
|
||
ctypes.set_errno(0)
|
||
ret = libc.syscall(SYS_RESTRICT_SELF, ruleset_fd, 0, 0, 0, 0, 0)
|
||
if ret < 0:
|
||
e = ctypes.get_errno()
|
||
return f"restrict_self(errno={e}:{os.strerror(e)})"
|
||
return None
|
||
|
||
|
||
def selftest(ro_paths) -> int:
|
||
"""自检：安装域后本进程与子进程写 ro 路径都必须被拒。"""
|
||
err = install_readonly_domain(ro_paths)
|
||
if err:
|
||
print(f"SELFTEST_FAIL install: {err}", file=sys.stderr)
|
||
return 2
|
||
probe = os.path.join(ro_paths[0], ".landlock_selftest_probe")
|
||
try:
|
||
with open(probe, "w") as f:
|
||
f.write("x")
|
||
print("SELFTEST_FAIL write-not-denied", file=sys.stderr)
|
||
return 3 # 残留文件由部署方以容器 root 清理
|
||
except OSError as e:
|
||
if e.errno != 13: # EACCES
|
||
print(f"SELFTEST_FAIL unexpected errno={e.errno} ({e.strerror})",
|
||
file=sys.stderr)
|
||
return 4
|
||
import subprocess
|
||
code = f"open({probe!r},'w').write('x')"
|
||
r = subprocess.run([sys.executable, "-c", code], capture_output=True)
|
||
if r.returncode == 0:
|
||
print("SELFTEST_FAIL child-not-denied", file=sys.stderr)
|
||
return 5
|
||
# 白名单区（/tmp）必须仍可写，否则说明规则误配，会造成行为回归
|
||
try:
|
||
tmp_probe = f"/tmp/.landlock_selftest_rw.{os.getpid()}"
|
||
with open(tmp_probe, "w") as f:
|
||
f.write("x")
|
||
os.remove(tmp_probe)
|
||
except OSError as e:
|
||
print(f"SELFTEST_FAIL tmp-not-writable: errno={e.errno} ({e.strerror})",
|
||
file=sys.stderr)
|
||
return 6
|
||
print("LANDLOCK_SELFTEST_OK")
|
||
return 0
|
||
|
||
|
||
def main() -> None:
|
||
args = sys.argv[1:]
|
||
ro_paths = []
|
||
selftest_mode = False
|
||
cmd = []
|
||
i = 0
|
||
while i < len(args):
|
||
a = args[i]
|
||
if a == "--ro" and i + 1 < len(args):
|
||
ro_paths.append(os.path.normpath(args[i + 1]))
|
||
i += 2
|
||
elif a == "--selftest":
|
||
selftest_mode = True
|
||
i += 1
|
||
elif a == "--":
|
||
cmd = args[i + 1:]
|
||
break
|
||
else:
|
||
i += 1
|
||
if not ro_paths:
|
||
print("launcher: missing --ro <path>", file=sys.stderr)
|
||
sys.exit(2)
|
||
if selftest_mode:
|
||
sys.exit(selftest(ro_paths))
|
||
if not cmd:
|
||
print("launcher: missing command after --", file=sys.stderr)
|
||
sys.exit(2)
|
||
err = install_readonly_domain(ro_paths)
|
||
if err:
|
||
print(f"launcher: install readonly domain failed: {err}", file=sys.stderr)
|
||
sys.exit(2)
|
||
os.execvp(cmd[0], cmd)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|