mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-08-11 00:30:12 -04:00
48a7024b96
While auditing real-world Bambuddy backup repos on GitHub I found
several left public. That's a serious leak: the settings backup only
filters bambu_cloud_token and auth_secret_key, so mqtt_username,
mqtt_password, ha_token, prometheus_token, bambu_cloud_email,
external_url, and the printer access codes (via K-profiles) were going
to whatever visibility the user picked.
Hard guard at every save and re-checked on every push:
- POST /github-backup/config and PATCH /github-backup/config (when URL,
token, or provider changes) run a connection test internally and
return 400 unless is_private comes back True.
- run_backup() re-checks before each scheduled or manual push, so a
repository that flipped from private to public gets a clear
"Backup aborted: the target repository is no longer private" failure.
Each provider's test_connection now returns is_private (GitHub /
Gitea / Forgejo read data.private, GitLab reads visibility=="private";
"internal" is treated as non-private). None means "couldn't determine"
and is also rejected -- safer to fail closed.
Frontend renders visibility inline on Test Connection: green check when
private, red warning panel listing every credential at risk when public,
yellow when unknown.
---
ui(github-backup): show save-failure messages inline on the card
The new "repository is not private" rejection message is ~250 characters
listing every credential the backup carries (MQTT password, HA token,
Prometheus token, Bambu Cloud email, printer access codes), which clips
badly in a toast.
Both the initial-setup save and the debounced autosave now stash the
backend's error message into a saveError state and render it as a red
inline banner above the test-result block, with whitespace-pre-wrap so
the full message stays readable. The banner clears on success, on the
next save attempt, and when the user starts editing URL / token / provider
-- the three fields whose changes invalidate the privacy check -- so it
doesn't linger after the user has already addressed the cause.
Short success toasts (Settings saved, Token updated, Backup enabled) are
unchanged.
568 lines
22 KiB
Python
568 lines
22 KiB
Python
"""Integration tests for GitHub Backup API endpoints."""
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _mock_private_repo_check():
|
|
"""Default mock: test_connection returns success + confirmed private.
|
|
|
|
POST /config and PATCH /config now refuse to save when the target repo
|
|
isn't confirmed private (Bambuddy backups carry credentials — see
|
|
`_enforce_private_repo` in github_backup.py routes). The default mock
|
|
here keeps the existing test suite green; tests that need to exercise
|
|
the public / unknown-visibility branches override this fixture inline.
|
|
"""
|
|
with patch(
|
|
"backend.app.services.github_backup.github_backup_service.test_connection",
|
|
new=AsyncMock(
|
|
return_value={
|
|
"success": True,
|
|
"message": "Connection successful",
|
|
"repo_name": "test/repo",
|
|
"permissions": {"push": True},
|
|
"is_private": True,
|
|
}
|
|
),
|
|
) as m:
|
|
yield m
|
|
|
|
|
|
class TestGitHubBackupConfigAPI:
|
|
"""Integration tests for /api/v1/github-backup endpoints."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_config_no_config(self, async_client: AsyncClient):
|
|
"""Verify getting config when none exists returns null."""
|
|
response = await async_client.get("/api/v1/github-backup/config")
|
|
assert response.status_code == 200
|
|
assert response.json() is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_config(self, async_client: AsyncClient):
|
|
"""Verify GitHub backup config can be created."""
|
|
data = {
|
|
"repository_url": "https://github.com/test/repo",
|
|
"access_token": "ghp_testtoken123",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": False,
|
|
"backup_spools": False,
|
|
"backup_archives": False,
|
|
"enabled": True,
|
|
}
|
|
response = await async_client.post("/api/v1/github-backup/config", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["repository_url"] == "https://github.com/test/repo"
|
|
assert result["branch"] == "main"
|
|
assert result["has_token"] is True
|
|
assert result["enabled"] is True
|
|
assert result["backup_spools"] is False
|
|
assert result["backup_archives"] is False
|
|
# Token should not be exposed in response
|
|
assert "access_token" not in result
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_config_after_create(self, async_client: AsyncClient):
|
|
"""Verify getting config after creation returns the config."""
|
|
# Create config first
|
|
data = {
|
|
"repository_url": "https://github.com/test/getrepo",
|
|
"access_token": "ghp_testtoken456",
|
|
"branch": "develop",
|
|
"schedule_enabled": True,
|
|
"schedule_type": "weekly",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": False,
|
|
"backup_settings": True,
|
|
"enabled": True,
|
|
}
|
|
await async_client.post("/api/v1/github-backup/config", json=data)
|
|
|
|
# Get config
|
|
response = await async_client.get("/api/v1/github-backup/config")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result is not None
|
|
assert result["repository_url"] == "https://github.com/test/getrepo"
|
|
assert result["branch"] == "develop"
|
|
assert result["schedule_type"] == "weekly"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_config_with_spools_and_archives(self, async_client: AsyncClient):
|
|
"""Verify config with spool and archive backup enabled."""
|
|
data = {
|
|
"repository_url": "https://github.com/test/spoolarchive",
|
|
"access_token": "ghp_spooltoken",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": False,
|
|
"backup_settings": False,
|
|
"backup_spools": True,
|
|
"backup_archives": True,
|
|
"enabled": True,
|
|
}
|
|
response = await async_client.post("/api/v1/github-backup/config", json=data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["backup_spools"] is True
|
|
assert result["backup_archives"] is True
|
|
assert result["backup_cloud_profiles"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_config_partial(self, async_client: AsyncClient):
|
|
"""Verify partial update of GitHub backup config."""
|
|
# Create config first
|
|
create_data = {
|
|
"repository_url": "https://github.com/test/update",
|
|
"access_token": "ghp_token",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": False,
|
|
"backup_spools": False,
|
|
"backup_archives": False,
|
|
"enabled": True,
|
|
}
|
|
await async_client.post("/api/v1/github-backup/config", json=create_data)
|
|
|
|
# Partial update
|
|
update_data = {
|
|
"branch": "develop",
|
|
"schedule_enabled": True,
|
|
}
|
|
response = await async_client.patch("/api/v1/github-backup/config", json=update_data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["branch"] == "develop"
|
|
assert result["schedule_enabled"] is True
|
|
# Original values should be preserved
|
|
assert result["repository_url"] == "https://github.com/test/update"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_config_enable_spools_and_archives(self, async_client: AsyncClient):
|
|
"""Verify partial update can enable spool and archive backup."""
|
|
# Create config first
|
|
create_data = {
|
|
"repository_url": "https://github.com/test/updatetoggle",
|
|
"access_token": "ghp_toggletoken",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": False,
|
|
"backup_spools": False,
|
|
"backup_archives": False,
|
|
"enabled": True,
|
|
}
|
|
await async_client.post("/api/v1/github-backup/config", json=create_data)
|
|
|
|
# Enable spools and archives via partial update
|
|
update_data = {
|
|
"backup_spools": True,
|
|
"backup_archives": True,
|
|
}
|
|
response = await async_client.patch("/api/v1/github-backup/config", json=update_data)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["backup_spools"] is True
|
|
assert result["backup_archives"] is True
|
|
# Other values preserved
|
|
assert result["backup_kprofiles"] is True
|
|
assert result["backup_settings"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_update_config_rejects_disabling_insecure_http_for_stored_http_url(self, async_client: AsyncClient):
|
|
"""Verify PATCH rejects leaving a stored HTTP URL without explicit insecure-HTTP allowance."""
|
|
create_data = {
|
|
"repository_url": "http://git.example.com/test/httprepo",
|
|
"access_token": "gitea_token",
|
|
"branch": "main",
|
|
"provider": "gitea",
|
|
"allow_insecure_http": True,
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": False,
|
|
"backup_spools": False,
|
|
"backup_archives": False,
|
|
"enabled": True,
|
|
}
|
|
create_response = await async_client.post("/api/v1/github-backup/config", json=create_data)
|
|
assert create_response.status_code == 200
|
|
|
|
response = await async_client.patch("/api/v1/github-backup/config", json={"allow_insecure_http": False})
|
|
|
|
assert response.status_code == 422
|
|
assert "Allow insecure HTTP" in response.json()["detail"]
|
|
|
|
stored_response = await async_client.get("/api/v1/github-backup/config")
|
|
assert stored_response.status_code == 200
|
|
stored = stored_response.json()
|
|
assert stored["repository_url"] == "http://git.example.com/test/httprepo"
|
|
assert stored["allow_insecure_http"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_config(self, async_client: AsyncClient):
|
|
"""Verify GitHub backup config can be deleted."""
|
|
# Create config first
|
|
create_data = {
|
|
"repository_url": "https://github.com/test/delete",
|
|
"access_token": "ghp_deletetoken",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": False,
|
|
"enabled": True,
|
|
}
|
|
await async_client.post("/api/v1/github-backup/config", json=create_data)
|
|
|
|
# Delete
|
|
response = await async_client.delete("/api/v1/github-backup/config")
|
|
assert response.status_code == 200
|
|
|
|
# Verify it's deleted
|
|
get_response = await async_client.get("/api/v1/github-backup/config")
|
|
assert get_response.status_code == 200
|
|
assert get_response.json() is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_config_not_found(self, async_client: AsyncClient):
|
|
"""Verify deleting non-existent config returns 404."""
|
|
# Make sure no config exists
|
|
await async_client.delete("/api/v1/github-backup/config")
|
|
|
|
# Try to delete again
|
|
response = await async_client.delete("/api/v1/github-backup/config")
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestGitHubBackupPrivateRepoGuard:
|
|
"""Refuse to save a config when the target repository is not private.
|
|
|
|
Bambuddy backups contain MQTT credentials, HA/Prometheus tokens, the
|
|
Bambu Cloud email, and printer access codes via K-profiles — they must
|
|
never be pushed to a public or internal-visibility repository.
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_config_rejects_public_repo(self, async_client: AsyncClient):
|
|
"""POST /config returns 400 when the connection test reports is_private=False."""
|
|
with patch(
|
|
"backend.app.services.github_backup.github_backup_service.test_connection",
|
|
new=AsyncMock(
|
|
return_value={
|
|
"success": True,
|
|
"message": "Connection successful",
|
|
"repo_name": "test/public-repo",
|
|
"permissions": {"push": True},
|
|
"is_private": False,
|
|
}
|
|
),
|
|
):
|
|
response = await async_client.post(
|
|
"/api/v1/github-backup/config",
|
|
json={
|
|
"repository_url": "https://github.com/test/public-repo",
|
|
"access_token": "ghp_token",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": True,
|
|
"enabled": True,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "not private" in response.json()["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_config_rejects_unknown_visibility(self, async_client: AsyncClient):
|
|
"""POST /config returns 400 when is_private cannot be determined (None)."""
|
|
with patch(
|
|
"backend.app.services.github_backup.github_backup_service.test_connection",
|
|
new=AsyncMock(
|
|
return_value={
|
|
"success": True,
|
|
"message": "Connection successful",
|
|
"repo_name": "test/repo",
|
|
"permissions": {"push": True},
|
|
"is_private": None,
|
|
}
|
|
),
|
|
):
|
|
response = await async_client.post(
|
|
"/api/v1/github-backup/config",
|
|
json={
|
|
"repository_url": "https://github.com/test/repo",
|
|
"access_token": "ghp_token",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": True,
|
|
"enabled": True,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "could not confirm" in response.json()["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_config_rejects_failed_connection(self, async_client: AsyncClient):
|
|
"""POST /config returns 400 when the connection test itself fails."""
|
|
with patch(
|
|
"backend.app.services.github_backup.github_backup_service.test_connection",
|
|
new=AsyncMock(
|
|
return_value={
|
|
"success": False,
|
|
"message": "Invalid access token",
|
|
"repo_name": None,
|
|
"permissions": None,
|
|
"is_private": None,
|
|
}
|
|
),
|
|
):
|
|
response = await async_client.post(
|
|
"/api/v1/github-backup/config",
|
|
json={
|
|
"repository_url": "https://github.com/test/repo",
|
|
"access_token": "bad-token",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": True,
|
|
"enabled": True,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "invalid access token" in response.json()["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_patch_rejects_url_change_to_public_repo(self, async_client: AsyncClient):
|
|
"""Changing the repository_url on an existing config re-checks privacy."""
|
|
# Initial create succeeds via the default autouse mock (private).
|
|
await async_client.post(
|
|
"/api/v1/github-backup/config",
|
|
json={
|
|
"repository_url": "https://github.com/test/private-repo",
|
|
"access_token": "ghp_token",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": True,
|
|
"enabled": True,
|
|
},
|
|
)
|
|
|
|
# Now try to switch to a public repo — must be rejected.
|
|
with patch(
|
|
"backend.app.services.github_backup.github_backup_service.test_connection",
|
|
new=AsyncMock(
|
|
return_value={
|
|
"success": True,
|
|
"message": "Connection successful",
|
|
"repo_name": "test/public-repo",
|
|
"permissions": {"push": True},
|
|
"is_private": False,
|
|
}
|
|
),
|
|
):
|
|
response = await async_client.patch(
|
|
"/api/v1/github-backup/config",
|
|
json={"repository_url": "https://github.com/test/public-repo"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "not private" in response.json()["detail"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_patch_skips_check_for_unrelated_fields(self, async_client: AsyncClient):
|
|
"""PATCHing a non-target field (e.g. schedule) does NOT re-run the test.
|
|
|
|
Without this, every benign toggle would trigger a live API call.
|
|
"""
|
|
await async_client.post(
|
|
"/api/v1/github-backup/config",
|
|
json={
|
|
"repository_url": "https://github.com/test/private-repo",
|
|
"access_token": "ghp_token",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": True,
|
|
"enabled": True,
|
|
},
|
|
)
|
|
|
|
# Replace the mock with one that would fail if called — proves the
|
|
# PATCH didn't hit test_connection for a schedule-only change.
|
|
mock = AsyncMock(side_effect=AssertionError("test_connection should not be called"))
|
|
with patch(
|
|
"backend.app.services.github_backup.github_backup_service.test_connection",
|
|
new=mock,
|
|
):
|
|
response = await async_client.patch(
|
|
"/api/v1/github-backup/config",
|
|
json={"schedule_enabled": True},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
mock.assert_not_called()
|
|
|
|
|
|
class TestGitHubBackupStatusAPI:
|
|
"""Integration tests for /api/v1/github-backup/status endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_status_no_config(self, async_client: AsyncClient):
|
|
"""Verify status when no config exists."""
|
|
# Ensure no config
|
|
await async_client.delete("/api/v1/github-backup/config")
|
|
|
|
response = await async_client.get("/api/v1/github-backup/status")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["configured"] is False
|
|
assert result["enabled"] is False
|
|
assert result["is_running"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_status_with_config(self, async_client: AsyncClient):
|
|
"""Verify status when config exists."""
|
|
# Create config
|
|
create_data = {
|
|
"repository_url": "https://github.com/test/status",
|
|
"access_token": "ghp_statustoken",
|
|
"branch": "main",
|
|
"schedule_enabled": True,
|
|
"schedule_type": "hourly",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": False,
|
|
"enabled": True,
|
|
}
|
|
await async_client.post("/api/v1/github-backup/config", json=create_data)
|
|
|
|
response = await async_client.get("/api/v1/github-backup/status")
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["configured"] is True
|
|
assert result["enabled"] is True
|
|
assert result["is_running"] is False
|
|
assert result["next_scheduled_run"] is not None
|
|
|
|
|
|
class TestGitHubBackupLogsAPI:
|
|
"""Integration tests for /api/v1/github-backup/logs endpoints."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_logs_no_config(self, async_client: AsyncClient):
|
|
"""Verify getting logs when no config exists returns empty list."""
|
|
# Ensure no config
|
|
await async_client.delete("/api/v1/github-backup/config")
|
|
|
|
response = await async_client.get("/api/v1/github-backup/logs")
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_logs_with_config(self, async_client: AsyncClient):
|
|
"""Verify getting logs with config."""
|
|
# Create config
|
|
create_data = {
|
|
"repository_url": "https://github.com/test/logs",
|
|
"access_token": "ghp_logstoken",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": False,
|
|
"enabled": True,
|
|
}
|
|
await async_client.post("/api/v1/github-backup/config", json=create_data)
|
|
|
|
response = await async_client.get("/api/v1/github-backup/logs")
|
|
assert response.status_code == 200
|
|
# No backups run yet, so empty list
|
|
assert response.json() == []
|
|
|
|
|
|
class TestGitHubBackupTriggerAPI:
|
|
"""Integration tests for /api/v1/github-backup/run endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_trigger_no_config(self, async_client: AsyncClient):
|
|
"""Verify triggering backup without config returns 404."""
|
|
# Ensure no config
|
|
await async_client.delete("/api/v1/github-backup/config")
|
|
|
|
response = await async_client.post("/api/v1/github-backup/run")
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_trigger_disabled_config(self, async_client: AsyncClient):
|
|
"""Verify triggering backup with disabled config returns 400."""
|
|
# Create disabled config
|
|
create_data = {
|
|
"repository_url": "https://github.com/test/trigger",
|
|
"access_token": "ghp_triggertoken",
|
|
"branch": "main",
|
|
"schedule_enabled": False,
|
|
"schedule_type": "daily",
|
|
"backup_kprofiles": True,
|
|
"backup_cloud_profiles": True,
|
|
"backup_settings": False,
|
|
"enabled": False, # Disabled
|
|
}
|
|
await async_client.post("/api/v1/github-backup/config", json=create_data)
|
|
|
|
response = await async_client.post("/api/v1/github-backup/run")
|
|
assert response.status_code == 400
|
|
assert "disabled" in response.json()["detail"].lower()
|