Harden Windows backup ACL preflight

CHANGELOG.md

@@ -16,7 +16,9 @@
   отдельного restore-теста и owner bootstrap. Принимаются только завершённые
   успешные Gitea runs; pending recovery journal записывается до остановки
   сервисов, повторно применённый release отклоняется до простоя, а проверка
-  `main` использует явно заданный production checkout.
+  `main` использует явно заданный production checkout. Windows preflight
+  проверяет owner ACL и требует свободное место под максимальный допустимый
+  архив.
 - Frontend `0.3.3` / Backend `0.13.2`: управление пользователями отображается
   согласно правам аккаунта, подписи ролей локализованы, повторяющиеся действия
   кабинета убраны, а знак лаборатории приведён к основному бренду.

docs/operations/BACKUP.md

@@ -20,6 +20,8 @@ verified production backup, so production deployment remains blocked.
   restrictive ACLs, and the exact `/.backups/` rule in that checkout's local
   `.git/info/exclude`. Before SSH connects, it also requires at least 100 GiB
   plus 1 MiB of free D: space, matching the protocol's maximum archive size.
+  The ACL owner and every allowed full-control entry must be the current user,
+  SYSTEM, or local Administrators; other and deny entries fail closed.
   It writes into a private staging directory, checks the
   stream size and SHA-256, verifies the archive and manifest, then publishes
   the complete backup directory without overwriting an existing ID.

docs/operations/CURRENT_STATE.md

@@ -27,8 +27,9 @@ remains blocked by the backup gate.
   `conclusion=success`, resolve `main` from `/opt/akyldash/current`, write a
   durable pending journal before stopping services, and reject an already
   applied release before downtime. The Windows client checks D: capacity,
-  NTFS/path/ACL, and the local ignore rule before SSH; it requires 100 GiB plus
-  1 MiB free. These code changes remain unverified on Windows and the VPS.
+  NTFS/path/ACL owner and allowed entries, and the local ignore rule before
+  SSH; it requires 100 GiB plus 1 MiB free. These code changes remain
+  unverified on Windows and the VPS.
 - Synthetic tests cover archive/protocol integrity, helper download rules,
   artifact/run provenance, receiver publication, release schema, and rollback
   refusal. Windows ACL, NTFS, PowerShell, SSH alias, Gitea helper against the

infra/backup/receive_external.py

@@ -20,6 +20,7 @@ from external_session import (ArchiveFrameWriter, read_challenge, read_response,
 
 DEFAULT_REPOSITORY = Path(r"D:\Dev\akyldash")
 MAX_ERROR_BYTES = 8192
+_WINDOWS = os.name == "nt"
 
 
 def _reparse(path: Path) -> bool:
@@ -42,6 +43,8 @@ $volume = [System.IO.DriveInfo]::new($driveRoot)
 if (-not $volume.IsReady -or $volume.DriveFormat -ne 'NTFS') { exit 5 }
 $owner = [System.Security.Principal.WindowsIdentity]::GetCurrent().User.Value
 $allowed = @($owner, 'S-1-5-18', 'S-1-5-32-544')
+$aclOwner = $acl.GetOwner([System.Security.Principal.SecurityIdentifier]).Value
+if ($aclOwner -notin $allowed) { exit 6 }
 $ownerAllowed = $false
 foreach ($rule in $acl.Access) {
   $sid = $rule.IdentityReference.Translate([System.Security.Principal.SecurityIdentifier]).Value
@@ -71,11 +74,11 @@ def check_capacity(free_bytes: int) -> None:
 def check_repository_target(repository: Path, backup_id: str, *, allow_existing: bool = False) -> tuple[Path, Path]:
     if not repository.is_absolute() or not repository.is_dir() or _reparse(repository):
         raise ValueError("repository must be an existing real directory on the selected Windows volume")
-    if os.name == "nt" and repository.drive.upper() != "D:":
+    if _WINDOWS and repository.drive.upper() != "D:":
         raise ValueError("external backup destination is restricted to D:\\Dev\\akyldash")
-    if os.name == "nt" and repository.resolve().as_posix().lower() != "d:/dev/akyldash":
+    if _WINDOWS and repository.resolve().as_posix().lower() != "d:/dev/akyldash":
         raise ValueError("external backup destination is restricted to D:\\Dev\\akyldash")
-    if os.name == "nt":
+    if _WINDOWS:
         check_capacity(shutil.disk_usage(repository).free)
     for parent in repository.parents:
         if _reparse(parent):
@@ -192,6 +195,8 @@ def receive_backup(reader, writer, *, repository: Path, request: dict, expected_
         os.chmod(sidecar, 0o600)
         check_private_acl(sidecar)
         os.rename(staging, final)
+        for path in (final, final / "backup.tar.gz", final / "backup.tar.gz.sha256"):
+            check_private_acl(path)
         write_ack(
             writer, session_nonce=header["nonce"], backup_id=header["backup_id"], size=size,
             archive_sha256=archive_sha256, manifest_sha256=header["manifest_sha256"], verified=True,

infra/backup/test_receive_external.py

@@ -69,7 +69,8 @@ class ExternalReceiverTests(unittest.TestCase):
             self.assertEqual(result, {"backup_id": final.name, "archive_sha256": digest, "size": size})
             self.assertTrue((final / "backup.tar.gz").is_file())
             if os.name == "nt":
-                acl_check.assert_any_call(final / "backup.tar.gz")
+                for path in (final, final / "backup.tar.gz", final / "backup.tar.gz.sha256"):
+                    acl_check.assert_any_call(path)
             else:
                 self.assertEqual((final / "backup.tar.gz").stat().st_mode & 0o777, 0o600)
             self.assertEqual((final / "backup.tar.gz.sha256").read_text(), f"{digest}  backup.tar.gz\n")
@@ -124,7 +125,7 @@ class ExternalReceiverTests(unittest.TestCase):
                     "returncode": 0,
                 })(),
             ]
-            with patch.object(receive_external.os, "name", "posix"), \
+            with patch.object(receive_external, "_WINDOWS", False), \
                  patch.object(receive_external.subprocess, "run", side_effect=responses), \
                  patch.object(receive_external, "check_private_acl"):
                 root, final = receive_external.check_repository_target(
@@ -136,7 +137,7 @@ class ExternalReceiverTests(unittest.TestCase):
     def test_windows_path_guard_rejects_a_synthetic_non_d_destination(self):
         with tempfile.TemporaryDirectory() as directory:
             repository = Path(directory)
-            with patch.object(receive_external.os, "name", "nt"):
+            with patch.object(receive_external, "_WINDOWS", True):
                 with self.assertRaisesRegex(ValueError, r"restricted to D:\\Dev\\akyldash"):
                     receive_external.check_repository_target(repository, "backup-id")
 
@@ -150,6 +151,17 @@ class ExternalReceiverTests(unittest.TestCase):
         self.assertEqual(child_env["AKYLDASH_ACL_PATH"], r"D:\Dev\akyldash\.backups")
         self.assertIn("System.IO.DriveInfo", run_mock.call_args.args[0][-1])
         self.assertNotIn("Get-Volume", run_mock.call_args.args[0][-1])
+        self.assertIn("$acl.GetOwner(", run_mock.call_args.args[0][-1])
+        self.assertIn("$aclOwner -notin $allowed", run_mock.call_args.args[0][-1])
+
+    def test_private_acl_rejects_an_unapproved_acl_owner(self):
+        result = type("Result", (), {"returncode": 6})()
+        with patch.object(receive_external.subprocess, "run", return_value=result) as run_mock:
+            with self.assertRaisesRegex(ValueError, "PowerShell status 6"):
+                receive_external.check_private_acl(Path(r"D:\Dev\akyldash\.backups"))
+        command = run_mock.call_args.args[0][-1]
+        self.assertIn("$acl.GetOwner(", command)
+        self.assertIn("if ($aclOwner -notin $allowed) { exit 6 }", command)
 
     def test_destination_preflight_fails_before_opening_ssh(self):
         with tempfile.TemporaryDirectory() as directory, \

infra/deploy/test_external_deploy.py

@@ -5,6 +5,10 @@ from contextlib import nullcontext
 from pathlib import Path
 from unittest.mock import Mock, patch
 
+import sys
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "backup"))
+
 import external_deploy