Make Windows backup receiver portable

infra/backup/receive_external.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import argparse
 import json
 import os
+import secrets
 import shutil
 import subprocess
 import sys
@@ -36,9 +37,9 @@ def check_private_acl(path: Path) -> None:
     command = r"""
 $ErrorActionPreference = 'Stop'
 $acl = Get-Acl -LiteralPath $env:AKYLDASH_ACL_PATH
-$drive = ([System.IO.Path]::GetPathRoot($env:AKYLDASH_ACL_PATH)).Substring(0,1)
-$volume = Get-Volume -DriveLetter $drive
-if ($volume.FileSystem -ne 'NTFS') { exit 5 }
+$driveRoot = [System.IO.Path]::GetPathRoot($env:AKYLDASH_ACL_PATH)
+$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')
 $ownerAllowed = $false
@@ -201,6 +202,7 @@ def run_release(request: dict, *, repository: Path = DEFAULT_REPOSITORY, ssh: st
         raise ValueError("OpenSSH client is unavailable")
     if not ssh_alias or any(character not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.-" for character in ssh_alias):
         raise ValueError("SSH alias must name an existing owner-controlled host entry")
+    check_repository_target(repository, "preflight-" + secrets.token_hex(16))
     process = subprocess.Popen(
         [ssh, "-T", ssh_alias, "/usr/local/sbin/akyldash-release-deploy"],
         stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
@@ -236,6 +238,7 @@ def run_rollback(*, repository: Path = DEFAULT_REPOSITORY, ssh: str = "ssh",
         raise ValueError("OpenSSH client is unavailable")
     if not ssh_alias or any(character not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.-" for character in ssh_alias):
         raise ValueError("SSH alias must name an existing owner-controlled host entry")
+    check_repository_target(repository, "preflight-" + secrets.token_hex(16))
     process = subprocess.Popen(
         [ssh, "-T", ssh_alias, "/usr/local/sbin/akyldash-release-rollback"],
         stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,

infra/backup/test_receive_external.py

@@ -1,5 +1,6 @@
 import io
 import json
+import os
 import tarfile
 import tempfile
 import unittest
@@ -55,13 +56,16 @@ class ExternalReceiverTests(unittest.TestCase):
             stream, manifest_sha, size, digest = self.source_stream()
             ack = io.BytesIO()
             with patch.object(receive_external, "check_repository_target", return_value=(root, final)), \
-                 patch.object(receive_external, "check_private_acl"):
+                 patch.object(receive_external, "check_private_acl") as acl_check:
                 result = receive_external.receive_backup(
                     stream, ack, repository=root, request=self.request
                 )
             self.assertEqual(result, {"backup_id": final.name, "archive_sha256": digest, "size": size})
             self.assertTrue((final / "backup.tar.gz").is_file())
-            self.assertEqual((final / "backup.tar.gz").stat().st_mode & 0o777, 0o600)
+            if os.name == "nt":
+                acl_check.assert_any_call(final / "backup.tar.gz")
+            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")
             ack.seek(0)
             self.assertTrue(read_ack(
@@ -114,7 +118,8 @@ class ExternalReceiverTests(unittest.TestCase):
                     "returncode": 0,
                 })(),
             ]
-            with patch.object(receive_external.subprocess, "run", side_effect=responses), \
+            with patch.object(receive_external.os, "name", "posix"), \
+                 patch.object(receive_external.subprocess, "run", side_effect=responses), \
                  patch.object(receive_external, "check_private_acl"):
                 root, final = receive_external.check_repository_target(
                     repository, "backup-id"
@@ -122,6 +127,13 @@ class ExternalReceiverTests(unittest.TestCase):
             self.assertEqual(root, repository / ".backups")
             self.assertEqual(final, root / "backup-id")
 
+    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 self.assertRaisesRegex(ValueError, r"restricted to D:\\Dev\\akyldash"):
+                    receive_external.check_repository_target(repository, "backup-id")
+
     def test_windows_acl_probe_does_not_inherit_credentials(self):
         result = type("Result", (), {"returncode": 0})()
         with patch.dict("os.environ", {"GITEA_TOKEN": "must-not-pass"}, clear=True), \
@@ -130,6 +142,21 @@ class ExternalReceiverTests(unittest.TestCase):
         child_env = run_mock.call_args.kwargs["env"]
         self.assertNotIn("GITEA_TOKEN", child_env)
         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])
+
+    def test_destination_preflight_fails_before_opening_ssh(self):
+        with tempfile.TemporaryDirectory() as directory, \
+             patch.object(receive_external.shutil, "which", return_value="ssh"), \
+             patch.object(receive_external, "check_repository_target", side_effect=ValueError("preflight failed")) as preflight, \
+             patch.object(receive_external.subprocess, "Popen") as popen:
+            with self.assertRaisesRegex(ValueError, "preflight failed"):
+                receive_external.run_release(self.request, repository=Path(directory))
+            with self.assertRaisesRegex(ValueError, "preflight failed"):
+                receive_external.run_rollback(repository=Path(directory))
+        self.assertEqual(preflight.call_count, 2)
+        self.assertTrue(all(call.args[1].startswith("preflight-") for call in preflight.call_args_list))
+        popen.assert_not_called()
 
 
 if __name__ == "__main__":