fix(admin): mask all stored configuration secrets and credential URLs

deploy/test_server_agent.py

@@ -113,6 +113,52 @@ class AgentTests(unittest.TestCase):
             self.agent.save_config({"revision": result["revision"], "values": {"TRAIN_HOUR": None}})
             self.assertNotIn("TRAIN_HOUR", agent_module.read_env(self.deploy / "config.local"))
 
+    def test_secret_defaults_and_shadowed_values_never_leave_worker(self):
+        self.schema[1]["default_value"] = "default-canary"
+        (self.deploy / "config.schema.json").write_text(json.dumps(self.schema), encoding="utf-8")
+        (self.deploy / "config.local").write_text("GITEA_TOKEN=local-canary\n", encoding="utf-8")
+        config = self.agent.config()
+        encoded = json.dumps(config)
+        for value in ("top-secret", "local-canary", "default-canary"):
+            self.assertNotIn(value, encoded)
+        token = next(entry for entry in config["entries"] if entry["key"] == "GITEA_TOKEN")
+        for field in ("env", "local", "effective", "default_value"):
+            self.assertEqual("••••••••", token[field])
+
+    def test_real_catalog_masks_every_secret_and_credential_bearing_url(self):
+        schema = json.loads(Path(__file__).with_name("config.schema.json").read_text(encoding="utf-8"))
+        (self.deploy / "config.schema.json").write_text(json.dumps(schema), encoding="utf-8")
+        secrets = {item["key"] for item in schema if item["secret"]}
+        self.assertTrue({"POSTGRES_PASSWORD", "DATABASE_URL", "REDIS_URL", "JWT_PRIVATE_KEY",
+                         "JWT_PREVIOUS_KEYS", "BOOTSTRAP_ADMIN_PASSWORD", "INTERNAL_TOKEN",
+                         "STREAM_SIGNING_KEY", "GENIUS_ACCESS_TOKEN", "MEILI_MASTER_KEY", "GITEA_TOKEN"} <= secrets)
+        self.assertNotIn("ACCESS_TOKEN_TTL", secrets)
+        env = {key: f"env-canary-{key}" for key in secrets}
+        local = {key: f"local-canary-{key}" for key in secrets}
+        env.update(GITEA_URL="https://user:password@example.org", MEILI_URL="https://example.org?api_key=query-canary")
+        local.update(GITEA_URL="https://example.org#fragment-canary", MEILI_URL="https://token@example.org")
+        for name, values in ((".env", env), ("config.local", local)):
+            (self.deploy / name).write_text("\n".join(f"{key}={value}" for key, value in values.items()), encoding="utf-8")
+        config = self.agent.config()
+        for value in (*env.values(), *local.values()):
+            self.assertNotIn(value, json.dumps(config))
+        entries = {entry["key"]: entry for entry in config["entries"]}
+        for key in secrets | {"GITEA_URL", "MEILI_URL"}:
+            self.assertTrue(entries[key]["secret"])
+            self.assertEqual("••••••••", entries[key]["effective"])
+        self.assertEqual("15m", entries["ACCESS_TOKEN_TTL"]["effective"])
+
+    def test_plain_urls_remain_readable_and_local_credentials_are_masked_after_save(self):
+        self.schema.append({**self.schema[0], "key": "MEILI_URL", "default_value": "https://example.org/search"})
+        (self.deploy / "config.schema.json").write_text(json.dumps(self.schema), encoding="utf-8")
+        entry = self.agent.config()["entries"][-1]
+        self.assertFalse(entry["secret"])
+        self.assertEqual("https://example.org/search", entry["effective"])
+        with patch.object(self.agent, "command", return_value=""):
+            config = self.agent.save_config({"revision": self.agent.revision(), "values": {"MEILI_URL": "https://user:save-canary@example.org"}})
+        self.assertNotIn("save-canary", json.dumps(config))
+        self.assertTrue(config["entries"][-1]["secret"])
+
     def test_invalid_values_and_compose_failure_preserve_original(self):
         original = (self.deploy / "config.local").read_bytes()
         for values in ({"TRAIN_HOUR": "24"}, {"TRAIN_HOUR": "bad"}, {"UNKNOWN": "x"}, {"PIPU_DATA": "/tmp"}, {"LOG_LEVEL": "x\nOTHER=y"}):