Refresh agent signing keys from repository (Refs #64)

agent/tests/test_installer.py

@@ -82,16 +82,9 @@ class InstallerTests(unittest.TestCase):
             self.preflight()
         self.assertEqual((self.root / "unrelated.txt").read_text(), "keep")
 
-    def test_trust_key_must_be_preprovisioned_and_protected(self):
+    def test_no_preprovisioned_signing_key_is_required(self):
         self.keyring.unlink()
-        with self.assertRaisesRegex(SystemExit, 'Provision the publisher public key'):
-            self.preflight()
-        self.keyring.write_bytes(b'publisher public key')
-        for override in ({'uid': 1000}, {'mode': stat.S_IFREG | 0o666}, {'mode': stat.S_IFLNK | 0o777}):
-            with self.subTest(override=override):
-                self.overrides[str(self.keyring)] = override
-                with self.assertRaisesRegex(SystemExit, 'regular, root-owned'):
-                    self.preflight()
+        self.preflight()
 
     def test_known_legacy_layout_can_upgrade_and_marker_binds_exact_enrollment(self):
         self.legacy_installation()
@@ -347,6 +340,46 @@ docker() {
         self.assertNotIn("Docker is installed but unavailable", result.stderr)
 
 
+class RepositoryTrustTests(unittest.TestCase):
+    def fetch(self, status="200", body="public key", curl_exit=0):
+        block = SOURCE[SOURCE.index("# INSTALLATION_TRUST:"):SOURCE.index("# INSTALLATION_TRUST_END")]
+        with tempfile.TemporaryDirectory() as temporary:
+            script = r'''set -euo pipefail
+curl() {
+    printf '%s\n' "$@" > "$temporary_dir/curl-args"
+    printf '%s' "$EC_TEST_BODY" > "$temporary_dir/release-signing.gpg"
+    printf '%s' "$EC_TEST_STATUS"
+    return "$EC_TEST_EXIT"
+}
+''' + block + "\nprintf 'verified-fetch'"
+            result = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
+                                    env={**os.environ, "temporary_dir": temporary, "EC_TEST_BODY": body,
+                                         "EC_TEST_STATUS": status, "EC_TEST_EXIT": str(curl_exit)})
+            args = (Path(temporary) / "curl-args").read_text().splitlines()
+            return result, args
+
+    def test_installer_fetches_fixed_repository_key_with_bounds_and_tls(self):
+        result, args = self.fetch()
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout, "verified-fetch")
+        self.assertIn("https://gitea.thewrightserver.net/josh/electric-cherry/raw/branch/main/agent/release-signing.gpg", args)
+        for option, value in (("--proto", "=https"), ("--max-filesize", "262144"),
+                              ("--max-time", "120"), ("--connect-timeout", "25"),
+                              ("--header", "Cache-Control: no-cache")):
+            self.assertEqual(args[args.index(option) + 1], value)
+        self.assertNotIn("--location", args)
+        self.assertNotIn("--insecure", args)
+        self.assertLess(SOURCE.index("# INSTALLATION_TRUST:"), SOURCE.index('$controller/api/agent/release/manifest.json'))
+
+    def test_repository_failure_redirect_or_empty_key_stops_installation(self):
+        for status, body, code in (("302", "redirect", 0), ("404", "missing", 22),
+                                    ("200", "", 0), ("200", "partial", 28), ("200", "oversized", 63)):
+            with self.subTest(status=status, body=body):
+                result, _ = self.fetch(status, body, code)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertNotIn("verified-fetch", result.stdout)
+
+
 class ReleaseTests(unittest.TestCase):
     def setUp(self):
         self.temporary = tempfile.TemporaryDirectory()
@@ -409,7 +442,14 @@ class ReleaseTests(unittest.TestCase):
             with self.subTest(protocol=protocol), self.assertRaisesRegex(SystemExit, "Unsupported agent or updater"):
                 self.verify()
 
-    def test_signature_is_verified_against_pinned_key_before_manifest_parsing(self):
+    def test_empty_or_oversized_downloaded_key_cannot_authorize_installation(self):
+        for value in (b"", b"x" * (256 * 1024 + 1)):
+            self.keyring.write_bytes(value)
+            with self.subTest(size=len(value)), self.assertRaisesRegex(SystemExit, "public key is empty or exceeds"):
+                self.verify()
+        self.gpgv.assert_not_called()
+
+    def test_signature_is_verified_against_downloaded_key_before_manifest_parsing(self):
         self.verify()
         args, kwargs = self.gpgv.call_args
         self.assertEqual(args[0][0], 'gpgv')