fix: normalize permissions after qb verification

tests/test_jobs.py

@@ -1,4 +1,6 @@
 import hashlib
+import os
+import stat
 import tempfile
 import threading
 import unittest
@@ -10,12 +12,13 @@ from archive_clients.bencode import encode
 from archive_clients.jobs import (
     ClientJobExecutor,
     JobExecutionError,
+    _normalize_verified_resource_permissions,
     _resource_fingerprint,
 )
 from archive_clients.syncthing import RouteSetupError, SyncthingTransferStatus
-from archive_clients.resources import normalize_resource
+from archive_clients.resources import NormalizedResource, normalize_resource
 from archive_clients.state import ClientStore
-from archive_control.v1 import control_pb2, job_pb2
+from archive_control.v1 import control_pb2, job_pb2, resource_pb2
 
 
 class CompleteSyncthing:
@@ -39,6 +42,50 @@ class SlowRescanSyncthing(CompleteSyncthing):
 
 
 class ClientJobHappyPathTests(unittest.TestCase):
+    def test_verified_resource_permissions_are_readable_by_other_apps(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory) / "qb"
+            resource_directory = root / "resource"
+            resource_directory.mkdir(parents=True)
+            completed = resource_directory / "complete.bin"
+            incomplete = resource_directory / "incomplete.bin"
+            completed.write_bytes(b"complete")
+            incomplete.write_bytes(b"incomplete")
+            os.chmod(resource_directory, 0o300)
+            os.chmod(completed, 0o200)
+            os.chmod(incomplete, 0o200)
+            resource = NormalizedResource(
+                resource_pb2.ResourceSummary(),
+                (
+                    resource_pb2.TorrentFile(
+                        file_index=0,
+                        canonical_path="resource/complete.bin",
+                        logical_bytes=len(b"complete"),
+                        completed_bytes=len(b"complete"),
+                        selected=True,
+                    ),
+                    resource_pb2.TorrentFile(
+                        file_index=1,
+                        canonical_path="resource/incomplete.bin",
+                        logical_bytes=len(b"incomplete"),
+                        completed_bytes=0,
+                        selected=True,
+                    ),
+                ),
+                Mock(),
+            )
+            _normalize_verified_resource_permissions(root, resource)
+            self.assertEqual(
+                stat.S_IMODE(resource_directory.stat().st_mode) & 0o555,
+                0o555,
+            )
+            self.assertEqual(
+                stat.S_IMODE(completed.stat().st_mode) & 0o444, 0o444
+            )
+            self.assertEqual(
+                stat.S_IMODE(incomplete.stat().st_mode), 0o200
+            )
+
     def test_syncthing_api_outage_during_transfer_is_retried(self):
         """A transient local REST outage must not terminally fail the job."""
         with tempfile.TemporaryDirectory() as directory: