test(api): cover the credentials upload, and stop it hoarding secrets

The endpoint that receives the user's Google OAuth credentials file had
no tests. Two bugs surfaced.

The OAuth-shape check ran inside `except Exception: pass`. A JSON
document that parses but is not an object — a bare 42, true, null, a
list — makes `'installed' not in creds_data` raise TypeError, which the
bare except swallowed, and the file was then written out as
credentials.json regardless. The check now decides the outcome instead
of being advisory, so anything not credentials-shaped is refused up
front rather than failing later inside the calendar plugin.

Every overwrite copies the old file to credentials.json.backup.<ts> and
nothing removed them, so a user who re-uploaded ten times had ten
complete sets of OAuth client credentials sitting in the plugin
directory, indefinitely. Keep the newest five. Pruning is housekeeping,
so a backup that cannot be removed logs and leaves the upload alone.

27 tests: size and extension limits, malformed JSON, the shape check,
0600 permissions on the written file, backup-on-overwrite, and pruning
including the repeated-upload case that stays bounded.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

test/test_api_v3_calendar_credentials.py

@@ -0,0 +1,214 @@
+"""
+Endpoint tests for POST /plugins/calendar/upload-credentials.
+
+The endpoint takes an uploaded Google OAuth credentials file, writes it
+into the calendar plugin's directory as credentials.json at mode 0600, and
+copies any previous file aside first. It had no tests.
+
+Regression coverage for two fixed bugs:
+- The OAuth-shape check sat inside `except Exception: pass`, so a valid
+  JSON document that is not an object — a bare `42`, a list, a string —
+  raised TypeError on the membership test, was swallowed, and got saved
+  as credentials.json anyway.
+- Each overwrite created a timestamped backup and nothing ever removed
+  them, so every re-upload left another complete copy of the user's OAuth
+  client credentials in the plugin directory, indefinitely.
+"""
+
+import io
+import json
+import os
+import stat
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+URL = "/api/v3/plugins/calendar/upload-credentials"
+
+VALID_CREDENTIALS = {
+    "installed": {
+        "client_id": "abc.apps.googleusercontent.com",
+        "client_secret": "shh",
+        "redirect_uris": ["http://localhost"],
+    }
+}
+
+
+@pytest.fixture
+def plugin_dir(tmp_path, api_v3_module):
+    directory = tmp_path / "plugins" / "calendar"
+    directory.mkdir(parents=True)
+    api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(directory)
+    return directory
+
+
+def upload(client, content, filename="credentials.json"):
+    # bytes are sent verbatim (to exercise malformed input); anything else
+    # is serialized, so None becomes the JSON literal null rather than an
+    # empty body.
+    payload = content if isinstance(content, bytes) else json.dumps(content).encode()
+    return client.post(
+        URL,
+        data={"file": (io.BytesIO(payload), filename)},
+        content_type="multipart/form-data",
+    )
+
+
+def backups(plugin_dir):
+    return sorted(plugin_dir.glob("credentials.json.backup.*"))
+
+
+class TestRequestValidation:
+    def test_no_file_part_is_a_400(self, api_v3_client, plugin_dir):
+        response = api_v3_client.post(URL, data={}, content_type="multipart/form-data")
+        assert response.status_code == 400
+        assert "No file provided" in response.get_json()["message"]
+
+    def test_empty_filename_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, VALID_CREDENTIALS, filename="")
+        assert response.status_code == 400
+
+    @pytest.mark.parametrize("filename", ["creds.txt", "creds.pem", "creds"])
+    def test_non_json_extension_is_a_400(self, api_v3_client, plugin_dir, filename):
+        response = upload(api_v3_client, VALID_CREDENTIALS, filename=filename)
+        assert response.status_code == 400
+        assert "JSON file" in response.get_json()["message"]
+
+    def test_uppercase_json_extension_accepted(self, api_v3_client, plugin_dir):
+        assert upload(api_v3_client, VALID_CREDENTIALS,
+                      filename="CREDENTIALS.JSON").status_code == 200
+
+    def test_oversized_file_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, b"x" * (1024 * 1024 + 1))
+        assert response.status_code == 400
+        assert "1MB" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+    def test_invalid_json_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, b"{not json")
+        assert response.status_code == 400
+        assert "not valid JSON" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+    def test_missing_plugin_directory_is_a_404(self, api_v3_client, api_v3_module, tmp_path):
+        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
+            tmp_path / "not-installed")
+        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 404
+
+
+class TestOAuthShapeValidation:
+    def test_installed_key_accepted(self, api_v3_client, plugin_dir):
+        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 200
+
+    def test_web_key_accepted(self, api_v3_client, plugin_dir):
+        assert upload(api_v3_client, {"web": {"client_id": "x"}}).status_code == 200
+
+    def test_object_without_oauth_keys_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, {"something": "else"})
+        assert response.status_code == 400
+        assert "valid Google OAuth" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+    @pytest.mark.parametrize("content", [42, "a string", [1, 2, 3], True, None])
+    def test_valid_json_that_is_not_an_object_is_rejected(
+            self, api_v3_client, plugin_dir, content):
+        # Regression: `'installed' not in 42` raises TypeError, which the
+        # bare `except Exception: pass` swallowed — the file was then saved
+        # as credentials.json despite being unusable as credentials.
+        response = upload(api_v3_client, content)
+        assert response.status_code == 400
+        assert "valid Google OAuth" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+
+class TestSaving:
+    def test_file_written_with_contents_intact(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, VALID_CREDENTIALS)
+        assert response.status_code == 200
+        saved = json.loads((plugin_dir / "credentials.json").read_text())
+        assert saved == VALID_CREDENTIALS
+
+    def test_response_reports_the_path(self, api_v3_client, plugin_dir):
+        body = upload(api_v3_client, VALID_CREDENTIALS).get_json()
+        assert body["path"].endswith("credentials.json")
+
+    def test_permissions_are_owner_only(self, api_v3_client, plugin_dir):
+        upload(api_v3_client, VALID_CREDENTIALS)
+        mode = stat.S_IMODE((plugin_dir / "credentials.json").stat().st_mode)
+        assert mode == 0o600
+
+    def test_first_upload_creates_no_backup(self, api_v3_client, plugin_dir):
+        upload(api_v3_client, VALID_CREDENTIALS)
+        assert backups(plugin_dir) == []
+
+    def test_overwrite_backs_up_the_previous_file(self, api_v3_client, plugin_dir):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"old": 1}}))
+        upload(api_v3_client, VALID_CREDENTIALS)
+        assert len(backups(plugin_dir)) == 1
+        assert json.loads(backups(plugin_dir)[0].read_text()) == {"installed": {"old": 1}}
+        assert json.loads((plugin_dir / "credentials.json").read_text()) == VALID_CREDENTIALS
+
+
+class TestBackupPruning:
+    def _seed(self, plugin_dir, count):
+        """Create `count` backups with distinct, increasing mtimes."""
+        now = int(time.time())
+        for i in range(count):
+            path = plugin_dir / f"credentials.json.backup.{now - (count - i) * 10}"
+            path.write_text(json.dumps({"installed": {"gen": i}}))
+            os.utime(path, (now - (count - i) * 10, now - (count - i) * 10))
+
+    def test_old_backups_are_pruned(self, api_v3_client, plugin_dir):
+        # Regression: nothing ever removed these, so a plugin directory
+        # accumulated one full copy of the user's OAuth credentials per
+        # re-upload, forever.
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 7)
+        assert len(backups(plugin_dir)) == 7
+
+        upload(api_v3_client, VALID_CREDENTIALS)
+        assert len(backups(plugin_dir)) == 5
+
+    def test_the_newest_backups_are_the_ones_kept(self, api_v3_client, plugin_dir):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 7)
+
+        upload(api_v3_client, VALID_CREDENTIALS)
+        remaining = backups(plugin_dir)
+        # The just-created backup (of "cur") plus the four newest seeds.
+        contents = [json.loads(p.read_text()) for p in remaining]
+        assert {"installed": {"cur": 1}} in contents
+        assert {"installed": {"gen": 0}} not in contents  # oldest seed gone
+
+    def test_under_the_limit_nothing_is_removed(self, api_v3_client, plugin_dir):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 2)
+        upload(api_v3_client, VALID_CREDENTIALS)
+        assert len(backups(plugin_dir)) == 3  # 2 seeded + 1 new
+
+    def test_repeated_uploads_stay_bounded(self, api_v3_client, plugin_dir):
+        for i in range(10):
+            upload(api_v3_client, {"installed": {"round": i}})
+            # Distinct mtimes so ordering is well-defined between rounds.
+            for path in backups(plugin_dir):
+                os.utime(path, (path.stat().st_mtime, path.stat().st_mtime))
+            time.sleep(0.01)
+        assert len(backups(plugin_dir)) <= 5
+
+    def test_unremovable_backup_does_not_fail_the_upload(
+            self, api_v3_client, plugin_dir, monkeypatch):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 7)
+
+        def refuse(self):
+            raise OSError("read-only filesystem")
+        monkeypatch.setattr(Path, "unlink", refuse)
+
+        # Pruning is housekeeping; failing it must not lose the upload.
+        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 200

web_interface/blueprints/api_v3.py

@@ -7239,6 +7239,29 @@ def serve_plugin_static(plugin_id, file_path):
         return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
 
 
+_MAX_CREDENTIAL_BACKUPS = 5
+
+
+def _prune_credential_backups(plugin_dir: Path) -> None:
+    """Keep only the newest _MAX_CREDENTIAL_BACKUPS credential backups.
+
+    Every re-upload copies the previous credentials.json aside. Without
+    pruning those accumulate for the life of the install — each one a
+    complete set of OAuth client credentials sitting in the plugin
+    directory.
+    """
+    backups = sorted(
+        plugin_dir.glob('credentials.json.backup.*'),
+        key=lambda p: p.stat().st_mtime,
+        reverse=True,
+    )
+    for stale in backups[_MAX_CREDENTIAL_BACKUPS:]:
+        try:
+            stale.unlink()
+        except OSError:
+            logger.warning("Could not remove old credential backup %s", stale.name)
+
+
 @api_v3.route('/plugins/calendar/upload-credentials', methods=['POST'])
 def upload_calendar_credentials():
     """Upload credentials.json file for calendar plugin"""
@@ -7270,20 +7293,25 @@ def upload_calendar_credentials():
         except json.JSONDecodeError:
             return jsonify({'status': 'error', 'message': 'File is not valid JSON'}), 400
 
-        # Validate it looks like Google OAuth credentials
+        # Validate it looks like Google OAuth credentials. The content
+        # already parsed as JSON above, so anything raising here means it is
+        # not credentials-shaped — a bare scalar, for instance, where the
+        # membership test raises TypeError. Reject rather than swallow: a
+        # file saved as credentials.json but not usable as credentials only
+        # fails later, somewhere less obvious.
         try:
             file.seek(0)
             creds_data = json.loads(file.read())
             file.seek(0)
+            is_oauth_shaped = 'installed' in creds_data or 'web' in creds_data
+        except Exception:
+            is_oauth_shaped = False
 
-            # Check for required Google OAuth fields
-            if 'installed' not in creds_data and 'web' not in creds_data:
+        if not is_oauth_shaped:
             return jsonify({
                 'status': 'error',
                 'message': 'File does not appear to be a valid Google OAuth credentials file'
             }), 400
-        except Exception:
-            pass  # Continue even if validation fails
 
         # Get plugin directory
         plugin_id = 'calendar'
@@ -7303,6 +7331,7 @@ def upload_calendar_credentials():
             backup_path = Path(plugin_dir) / f'credentials.json.backup.{int(time.time())}'
             import shutil
             shutil.copy2(credentials_path, backup_path)
+            _prune_credential_backups(Path(plugin_dir))
 
         # Save new file
         file.save(str(credentials_path))