test_github_backup_settings_secrets.py 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106
  1. """Settings secrets never leave the instance in a Git backup.
  2. A restore refuses every credential-shaped key, so the backup has no use for
  3. their values. It still records whether each one was set (the placeholder) so
  4. the restore can tell a switch whose credential stayed behind from one that
  5. never had one. A URL setting with a login in it is left out: a masked URL
  6. restored over a working one would break it.
  7. """
  8. import json
  9. import pytest
  10. from sqlalchemy import delete
  11. from backend.app.models.settings import Settings
  12. from backend.app.services.github_backup import GitHubBackupService
  13. from backend.app.services.github_restore import GitHubRestoreService, _is_blocked_setting_key
  14. from backend.app.services.secret_fields import SECRET_PLACEHOLDER
  15. SECRETS = {
  16. "mqtt_password": "value-one",
  17. "ha_token": "value-two",
  18. "prometheus_token": "value-three",
  19. "obico_ml_token": "value-four",
  20. "ldap_bind_password": "value-five",
  21. "smtp_password": "value-six",
  22. "virtual_printer_access_code": "value-seven",
  23. "bambu_cloud_token": "value-eight",
  24. "auth_secret_key": "value-nine",
  25. "manyfold_client_secret": "value-ten",
  26. "MQTT_SECRET": "value-eleven",
  27. "orca_cloud_pending_verifier": "value-twelve",
  28. }
  29. async def _collect(db_session, rows: dict[str, str]) -> dict:
  30. db_session.add_all(Settings(key=key, value=value) for key, value in rows.items())
  31. await db_session.commit()
  32. files: dict = {}
  33. await GitHubBackupService()._collect_settings(db_session, files)
  34. return files["settings/app_settings.json"]["settings"]
  35. @pytest.mark.asyncio
  36. async def test_a_set_secret_is_written_as_the_placeholder(db_session):
  37. collected = await _collect(db_session, {**SECRETS, "currency": "EUR"})
  38. assert {key: collected.get(key) for key in SECRETS} == dict.fromkeys(SECRETS, SECRET_PLACEHOLDER)
  39. assert collected["currency"] == "EUR"
  40. dumped = json.dumps(collected)
  41. assert not [value for value in SECRETS.values() if value in dumped]
  42. @pytest.mark.asyncio
  43. async def test_an_empty_secret_stays_empty(db_session):
  44. """Blank counts as empty, as it does for the restore."""
  45. collected = await _collect(db_session, {"prometheus_token": "", "mqtt_password": " "})
  46. assert collected == {"prometheus_token": "", "mqtt_password": " "}
  47. @pytest.mark.asyncio
  48. async def test_a_backed_up_secret_is_refused_on_restore(db_session):
  49. plan = await GitHubRestoreService()._plan_settings(db_session, dict(SECRETS))
  50. assert sorted(plan.blocked) == sorted(SECRETS)
  51. @pytest.mark.asyncio
  52. async def test_the_backup_hides_exactly_what_the_restore_refuses(db_session):
  53. rows = {**SECRETS, "currency": "EUR", "mqtt_enabled": "true", "theme_api": "x", "ldap_server_url": "ldap://h"}
  54. collected = await _collect(db_session, rows)
  55. for key in rows:
  56. assert (collected[key] == SECRET_PLACEHOLDER) is _is_blocked_setting_key(key), key
  57. @pytest.mark.asyncio
  58. @pytest.mark.parametrize(
  59. "url",
  60. [
  61. "http://user:value-url@spoolman.local:7912/",
  62. "https://value-url@spoolman.local",
  63. "http://:value-url@spoolman.local",
  64. "http://user:value-url@spoolman.local?x=1",
  65. ],
  66. )
  67. async def test_a_url_setting_with_a_login_is_left_out(db_session, url):
  68. collected = await _collect(db_session, {"spoolman_url": url, "ha_url": "http://ha.local:8123"})
  69. assert "spoolman_url" not in collected
  70. assert collected["ha_url"] == "http://ha.local:8123"
  71. assert "value-url" not in json.dumps(collected)
  72. @pytest.mark.asyncio
  73. async def test_a_switch_whose_credential_stayed_behind_is_still_left_off_on_restore(db_session):
  74. """Restored onto an instance without the password, MQTT must not come on."""
  75. collected = await _collect(db_session, {"mqtt_enabled": "true", "mqtt_password": "value-one"})
  76. await db_session.execute(delete(Settings))
  77. await db_session.commit()
  78. plan = await GitHubRestoreService()._plan_settings(db_session, collected)
  79. assert plan.companion == ("mqtt_enabled",)
  80. assert "mqtt_password" in plan.blocked