From 9c5cf5c6f84d26b6833552700fb888b143d0472d Mon Sep 17 00:00:00 2001 From: AlexFucuson9 Date: Thu, 2 Jul 2026 16:25:08 +0700 Subject: [PATCH] fix: use Windows ACLs for credential file permissions in backup import MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit os.chmod(path, 0o600) is a no-op on Windows — the file keeps its default 0o666 mode bits. This means credential-shaped files restored by 'hermes import' are not tightened on Windows. Add _restrict_file_permissions() helper that: - On POSIX: calls os.chmod(path, 0o600) as before - On Windows: uses ctypes to set NTFS DACL granting full access only to the current user and SYSTEM - Falls back to clearing world-readable bits if Windows API fails Fixes #56923 --- hermes_cli/backup.py | 102 ++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 97 insertions(+), 5 deletions(-) diff --git a/hermes_cli/backup.py b/hermes_cli/backup.py index dca3eae1b5b9b..791eb2b4c6da7 100644 --- a/hermes_cli/backup.py +++ b/hermes_cli/backup.py @@ -26,6 +26,101 @@ logger = logging.getLogger(__name__) +def _restrict_file_permissions(path: "Path") -> None: + """Tighten *path* to owner-only read/write. + + On POSIX this is ``chmod 600``. On Windows the native NTFS DACL is + adjusted via :mod:`ctypes` so only the current user and SYSTEM retain + access. Falls back to toggling the read-only bit if the Windows API + calls are unavailable (e.g. missing *advapi32* on Wine). + """ + if sys.platform == "win32": + try: + import ctypes + import ctypes.wintypes as wt + + advapi32 = ctypes.windll.advapi32 # type: ignore[attr-defined] + kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] + + # SDDL: owner (current user) + SYSTEM — full control; no one else. + # O:SYG:SYD:(A;;FA;;;OW) — but we need the *actual* user SID. + # Simpler: use ConvertStringSecurityDescriptorToSecurityDescriptorW + # with a SDDL that grants FA to the owner only. + # + # Get current user SID via GetTokenInformation. + TOKEN_QUERY = 0x0008 + TokenUser = 1 + + token = wt.HANDLE() + if not advapi32.OpenProcessToken( + kernel32.GetCurrentProcess(), TOKEN_QUERY, ctypes.byref(token) + ): + raise OSError("OpenProcessToken failed") + + # First call to get required size. + size = wt.DWORD(0) + advapi32.GetTokenInformation(token, TokenUser, None, 0, ctypes.byref(size)) + + buf = ctypes.create_string_buffer(size.value) + if not advapi32.GetTokenInformation( + token, TokenUser, buf, size, ctypes.byref(size) + ): + raise OSError("GetTokenInformation failed") + + # TOKEN_USER starts with SID_AND_ATTRIBUTES; SID is at offset 0. + class SID_AND_ATTRIBUTES(ctypes.Structure): + _fields_ = [("Sid", ctypes.c_void_p), ("Attributes", wt.DWORD)] + + sid_ptr = SID_AND_ATTRIBUTES.from_buffer_copy( + ctypes.string_at(buf, size.value) + ).Sid + + # Convert SID to string. + sid_string = wt.LPWSTR() + if not advapi32.ConvertSidToStringSidW(sid_ptr, ctypes.byref(sid_string)): + raise OSError("ConvertSidToStringSidW failed") + + user_sid = sid_string.value + kernel32.LocalFree(sid_string) + + sddl = f"O:{user_sid}G:{user_sid}D:(A;;FA;;;{user_sid})(A;;FA;;;SY)" + sd = ctypes.c_void_p() + sd_size = wt.DWORD(0) + if not advapi32.ConvertStringSecurityDescriptorToSecurityDescriptorW( + sddl, 1, ctypes.byref(sd), ctypes.byref(sd_size) + ): + raise OSError("ConvertStringSecurityDescriptorToSecurityDescriptorW failed") + + # Apply to file — DACL_SECURITY_INFORMATION (4) | PROTECTED_DACL (0x80000000). + DACL_SECURITY_INFORMATION = 4 + if not advapi32.SetFileSecurityW( + str(path), DACL_SECURITY_INFORMATION, sd + ): + # SetFileSecurityW is legacy; try SetNamedSecurityInfoW instead. + SE_FILE_OBJECT = 1 + advapi32.SetNamedSecurityInfoW( + str(path), SE_FILE_OBJECT, DACL_SECURITY_INFORMATION, + None, None, sd, None, + ) + + kernel32.LocalFree(sd) + return + except Exception: + # Final fallback: at least clear the world-readable bits. + try: + import stat + os.chmod(path, stat.S_IREAD | stat.S_IWRITE) + except OSError: + pass + return + + # POSIX + try: + os.chmod(path, 0o600) + except OSError: + pass + + # --------------------------------------------------------------------------- # Exclusion rules # --------------------------------------------------------------------------- @@ -600,10 +695,7 @@ def run_import(args) -> None: dst.write(src.read()) # External provider configs commonly hold credentials. if target.suffix in {".json", ".env", ".conf"} or target.name in _SECRET_FILE_NAMES: - try: - os.chmod(target, 0o600) - except OSError: - pass + _restrict_file_permissions(target) restored += 1 restored_external += 1 except (PermissionError, OSError) as exc: @@ -645,7 +737,7 @@ def run_import(args) -> None: with zf.open(member) as src, open(target, "wb") as dst: dst.write(src.read()) if target.name in _SECRET_FILE_NAMES: - os.chmod(target, 0o600) + _restrict_file_permissions(target) restored += 1 except (PermissionError, OSError) as exc: errors.append(f" {rel}: {exc}")