176 lines
9.1 KiB
Python
176 lines
9.1 KiB
Python
"""Real application HTTP checks for installation, cookies, and backup controls.
|
|
|
|
All persistence and artwork paths are isolated in temporary directories; workers,
|
|
logging file handlers, and the metrics listener are disabled for these tests.
|
|
"""
|
|
|
|
import io
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from backend.app import db, main
|
|
from backend.app.config import settings
|
|
from backend.app.services import backups
|
|
|
|
|
|
OPERATOR_TOKEN = "installation-http-operator-token-test-123456789"
|
|
OWNER_PASSWORD = "installation-http-owner-password-123456789"
|
|
BACKUP_PASSPHRASE = "installation-http-backup-passphrase-123456789"
|
|
|
|
|
|
class InstallationHttpTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = Path(self.temporary.name)
|
|
for key, value in {
|
|
"sqlite_path": str(self.root / "magent.db"),
|
|
"sqlite_journal_mode": "DELETE",
|
|
"jwt_secret": "installation-http-test-jwt-secret-1234567890",
|
|
"settings_encryption_key": None,
|
|
"admin_username": "unused-environment-admin",
|
|
"admin_password": "",
|
|
"setup_token": OPERATOR_TOKEN,
|
|
"auth_cookie_secure": True,
|
|
"auth_cookie_domain": None,
|
|
"auth_cookie_samesite": "strict",
|
|
}.items():
|
|
context = patch.object(settings, key, value)
|
|
context.start()
|
|
self.addCleanup(context.stop)
|
|
for context in (
|
|
patch.object(main, "configure_logging"),
|
|
patch.object(main, "start_metrics"),
|
|
patch.object(main, "_background_tasks", []),
|
|
patch.object(main, "_background_started", False),
|
|
patch.object(backups, "_assets_root", return_value=self.root / "assets"),
|
|
patch.dict("os.environ", {"BACKGROUND_TASKS_ENABLED": "false"}),
|
|
):
|
|
context.start()
|
|
self.addCleanup(context.stop)
|
|
self.origin = str(settings.cors_allow_origin).rstrip("/")
|
|
self.client = self.enterContext(TestClient(main.app, base_url="https://magent.test"))
|
|
self.client.headers["Origin"] = self.origin
|
|
|
|
def create_owner(self):
|
|
response = self.client.post("/setup/bootstrap", json={
|
|
"setup_token": OPERATOR_TOKEN, "username": "owner", "password": OWNER_PASSWORD,
|
|
})
|
|
self.assertEqual(response.status_code, 201, response.text)
|
|
return response
|
|
|
|
def sign_in(self):
|
|
response = self.client.post("/auth/login", data={"username": "owner", "password": OWNER_PASSWORD})
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertIn(settings.auth_cookie_name, self.client.cookies)
|
|
auth_cookie = next(value for value in response.headers.get_list("set-cookie") if value.startswith(settings.auth_cookie_name + "="))
|
|
self.assertIn("HttpOnly", auth_cookie)
|
|
self.assertIn("Secure", auth_cookie)
|
|
self.assertIn("SameSite=strict", auth_cookie)
|
|
self.assertNotIn("Authorization", self.client.headers)
|
|
|
|
def test_fresh_setup_cookie_settings_completion_and_backup_round_trip(self):
|
|
status = self.client.get("/setup/status")
|
|
self.assertEqual(status.json(), {"setup_required": True, "needs_admin": True})
|
|
self.assertEqual(status.headers["cache-control"], "no-store")
|
|
self.assertIn("default-src 'none'", status.headers["content-security-policy"])
|
|
self.assertEqual(self.client.get("/setup/state").status_code, 401)
|
|
self.assertEqual(self.client.get("/admin/backups").status_code, 401)
|
|
|
|
self.create_owner()
|
|
self.sign_in()
|
|
self.assertEqual(self.client.get("/setup/state").json()["step"], "apps")
|
|
response = self.client.put("/admin/settings", json={
|
|
"jellyfin_base_url": "http://jellyfin.test:8096",
|
|
"jellyfin_api_key": "test-integration-key-for-setup",
|
|
"site_login_message": "Welcome to this installation",
|
|
})
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertEqual(response.json()["updated"], 3)
|
|
values = {row["key"]: row for row in self.client.get("/admin/settings").json()["settings"]}
|
|
self.assertEqual(values["jellyfin_base_url"]["value"], "http://jellyfin.test:8096")
|
|
self.assertIsNone(values["jellyfin_api_key"]["value"])
|
|
self.assertTrue(values["jellyfin_api_key"]["isSet"])
|
|
response = self.client.put("/setup/state", json={"step": "review"})
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
completed = self.client.post("/setup/complete")
|
|
self.assertEqual(completed.status_code, 200, completed.text)
|
|
self.assertTrue(completed.json()["completed"])
|
|
self.assertEqual(self.client.get("/setup/status").json(), {"setup_required": False, "needs_admin": False})
|
|
self.assertEqual(main._background_tasks, [])
|
|
|
|
exported = self.client.post("/admin/backups/export", json={
|
|
"passphrase": BACKUP_PASSPHRASE, "include_cache": False,
|
|
})
|
|
self.assertEqual(exported.status_code, 200, exported.text[:100])
|
|
self.assertTrue(exported.content.startswith(backups.MAGIC))
|
|
self.assertEqual(exported.headers["cache-control"], "no-store")
|
|
self.assertNotIn(b"test-integration-key-for-setup", exported.content)
|
|
restored = self.client.post("/admin/backups/restore", files={
|
|
"file": ("restore.magent-backup", io.BytesIO(exported.content), "application/octet-stream"),
|
|
}, data={"passphrase": BACKUP_PASSPHRASE, "confirmation": "RESTORE"})
|
|
self.assertEqual(restored.status_code, 202, restored.text)
|
|
self.assertTrue(restored.json()["restart_required"])
|
|
self.assertEqual(db.get_setting("site_login_message"), "Welcome to this installation")
|
|
self.assertIsNotNone(self.client.get("/admin/backups").json()["pending_restore"])
|
|
cancelled = self.client.delete("/admin/backups/restore")
|
|
self.assertEqual(cancelled.status_code, 200, cancelled.text)
|
|
self.assertIsNone(self.client.get("/admin/backups").json()["pending_restore"])
|
|
|
|
def test_cross_origin_bootstrap_and_authenticated_changes_are_rejected(self):
|
|
response = self.client.post("/setup/bootstrap", headers={"Origin": "https://unrelated.invalid"}, json={
|
|
"setup_token": OPERATOR_TOKEN, "username": "owner", "password": OWNER_PASSWORD,
|
|
})
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertFalse(db.has_admin_user())
|
|
self.create_owner()
|
|
self.sign_in()
|
|
response = self.client.put("/setup/state", headers={"Origin": "https://unrelated.invalid"}, json={"step": "review"})
|
|
self.assertEqual(response.status_code, 403)
|
|
response = self.client.post("/admin/backups/export", headers={"Origin": "https://unrelated.invalid"}, json={"passphrase": BACKUP_PASSPHRASE})
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertEqual(self.client.get("/setup/state").json()["step"], "apps")
|
|
|
|
def test_setup_validation_errors_do_not_echo_password_or_token(self):
|
|
secret_password = "private-password-marker-" + "p" * 1024
|
|
secret_token = "private-token-marker-" + "t" * 1024
|
|
for payload, secret in (
|
|
({"setup_token": OPERATOR_TOKEN, "username": "owner", "password": secret_password}, secret_password),
|
|
({"setup_token": secret_token, "username": "owner", "password": OWNER_PASSWORD}, secret_token),
|
|
({"setup_token": OPERATOR_TOKEN, "password": OWNER_PASSWORD}, OWNER_PASSWORD),
|
|
):
|
|
with self.subTest(secret=secret[:22]):
|
|
response = self.client.post("/setup/bootstrap", json=payload)
|
|
self.assertEqual(response.status_code, 422, response.text)
|
|
self.assertNotIn(secret, response.text)
|
|
self.assertNotIn(OPERATOR_TOKEN, response.text)
|
|
for error in response.json()["detail"]:
|
|
self.assertNotIn("input", error)
|
|
|
|
def test_backup_validation_errors_do_not_echo_passphrases(self):
|
|
self.create_owner()
|
|
self.sign_in()
|
|
passphrase = "private-backup-passphrase-marker-" + "p" * 1024
|
|
response = self.client.post("/admin/backups/export", json={"passphrase": passphrase})
|
|
self.assertEqual(response.status_code, 422)
|
|
self.assertNotIn(passphrase, response.text)
|
|
response = self.client.post("/admin/backups/restore", files={"file": ("archive", b"data")}, data={
|
|
"passphrase": passphrase, "confirmation": "RESTORE",
|
|
})
|
|
self.assertEqual(response.status_code, 422)
|
|
self.assertNotIn(passphrase, response.text)
|
|
self.assertIsNone(self.client.get("/admin/backups").json()["pending_restore"])
|
|
|
|
def test_real_middleware_rejects_oversized_bootstrap_before_creation(self):
|
|
response = self.client.post("/setup/bootstrap", content=b"x" * (17 * 1024), headers={"Content-Type": "application/json"})
|
|
self.assertEqual(response.status_code, 413, response.text)
|
|
self.assertFalse(db.has_admin_user())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|