from concurrent.futures import ThreadPoolExecutor import os import tempfile from threading import Barrier from types import SimpleNamespace import unittest from unittest.mock import AsyncMock, patch from fastapi import FastAPI from fastapi.testclient import TestClient from backend.app import db from backend.app.config import settings from backend.app.routers import setup as setup_router from backend.app.security import create_access_token from backend.app.services import setup SETUP_TOKEN = "operator-setup-token-for-tests-only-1234567890" ADMIN_PASSWORD = "A-long-admin-password!123" class SetupTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) self.addCleanup(self.temp.cleanup) for field, value in { "sqlite_path": os.path.join(self.temp.name, "test.db"), "sqlite_journal_mode": "DELETE", "admin_username": "environment-admin", "admin_password": "", "jwt_secret": "setup-test-jwt-secret-only-1234567890", "settings_encryption_key": "bWFnZW50LXNlY3VyaXR5LXRlc3Qta2V5LTMyLWJ5dGU=", }.items(): context = patch.object(settings, field, value) context.start() self.addCleanup(context.stop) context = patch.object(setup, "settings", SimpleNamespace(setup_token=SETUP_TOKEN)) context.start() self.addCleanup(context.stop) setup.initialize_setup_state() db.init_db() self.app = FastAPI() self.app.include_router(setup_router.router) self.client = TestClient(self.app) self.addCleanup(self.client.close) def bootstrap(self, **changes): return self.client.post("/setup/bootstrap", json={ "setup_token": SETUP_TOKEN, "username": "first-admin", "password": ADMIN_PASSWORD, **changes, }) def admin_headers(self): return {"Authorization": f"Bearer {create_access_token('first-admin', 'admin')}"} def test_fresh_install_requires_setup_and_exposes_no_configuration(self): response = self.client.get("/setup/status") self.assertEqual(response.status_code, 200) self.assertEqual(response.json(), {"setup_required": True, "needs_admin": True}) self.assertEqual(response.headers["cache-control"], "no-store") self.assertEqual(self.client.get("/setup/state").status_code, 401) def test_existing_install_migrates_as_completed_without_reopening_bootstrap(self): with db._connect() as conn: conn.execute("DROP TABLE installation_setup") setup.initialize_setup_state() self.assertEqual(setup.get_public_setup_status(), {"setup_required": False, "needs_admin": False}) self.assertIsNotNone(setup.get_setup_state()["completed_at"]) self.assertEqual(self.bootstrap().status_code, 409) def test_missing_marker_fails_closed(self): with db._connect() as conn: conn.execute("DROP TABLE installation_setup") self.assertFalse(setup.is_setup_required()) self.assertEqual(self.bootstrap().status_code, 409) def test_marker_survives_restart_before_schema_initialization(self): new_path = os.path.join(self.temp.name, "interrupted.db") with patch.object(settings, "sqlite_path", new_path): setup.initialize_setup_state() setup.initialize_setup_state() db.init_db() self.assertTrue(setup.is_setup_required()) def test_empty_precreated_database_is_a_fresh_install(self): new_path = os.path.join(self.temp.name, "empty.db") with open(new_path, "wb"): pass with patch.object(settings, "sqlite_path", new_path): setup.initialize_setup_state() db.init_db() self.assertTrue(setup.is_setup_required()) def test_environment_admin_uses_wizard_without_public_bootstrap(self): with patch.object(settings, "admin_password", ADMIN_PASSWORD): db.ensure_admin_user() self.assertEqual(setup.get_public_setup_status(), {"setup_required": True, "needs_admin": False}) self.assertEqual(self.bootstrap().status_code, 409) def test_valid_token_creates_local_admin_once_and_uses_password_hash(self): response = self.bootstrap() self.assertEqual(response.status_code, 201, response.text) self.assertEqual(response.json(), {"status": "created", "username": "first-admin"}) user = db.verify_user_password("first-admin", ADMIN_PASSWORD) self.assertIsNotNone(user) self.assertEqual(user["role"], "admin") self.assertEqual(user["auth_provider"], "local") self.assertNotEqual(user["password_hash"], ADMIN_PASSWORD) self.assertEqual(setup.get_setup_state()["step"], "apps") self.assertEqual(self.bootstrap(username="second-admin").status_code, 409) self.assertEqual(len(db.get_all_users()), 1) def test_invalid_and_missing_operator_tokens_never_create_admin(self): self.assertEqual(self.bootstrap(setup_token="incorrect").status_code, 403) with patch.object(setup.settings, "setup_token", ""): self.assertEqual(self.bootstrap().status_code, 403) with patch.object(setup.settings, "setup_token", "too-short"): self.assertEqual(self.bootstrap(setup_token="too-short").status_code, 403) self.assertFalse(db.has_admin_user()) def test_non_ascii_token_fails_cleanly(self): self.assertEqual(self.bootstrap(setup_token="invalid-\N{SNOWMAN}").status_code, 403) self.assertFalse(db.has_admin_user()) def test_example_and_repeated_character_setup_tokens_are_rejected(self): for token in ( "replace-with-a-separate-random-setup-token", "CHANGE_ME_before_starting_this_installation", "your-setup-token-goes-here-at-least-32-characters", "a" * 64, "0" * 64, " " * 64, ): with self.subTest(token=token), patch.object(setup.settings, "setup_token", token): self.assertFalse(setup.setup_token_configured()) with self.assertRaises(setup.InvalidSetupTokenError): setup.bootstrap_administrator(token, "owner", ADMIN_PASSWORD) self.assertFalse(db.has_admin_user()) self.assertTrue(setup.setup_token_configured()) def test_password_policy_and_username_validation(self): for username in (" ", "admin user", "admin\x7f", "admin\nname"): with self.subTest(username=repr(username)): self.assertEqual(self.bootstrap(username=username).status_code, 400) self.assertEqual(self.bootstrap(password="short").status_code, 400) self.assertFalse(db.has_admin_user()) def test_oversized_fields_and_unexpected_privileges_are_rejected(self): self.assertEqual(self.bootstrap(password="x" * 1025).status_code, 422) self.assertEqual(self.bootstrap(username="x" * 101).status_code, 422) self.assertEqual(self.bootstrap(role="admin").status_code, 422) self.assertFalse(db.has_admin_user()) def test_existing_normalized_username_is_not_replaced(self): db.create_user("Taken", ADMIN_PASSWORD) self.assertEqual(self.bootstrap(username="taken").status_code, 409) self.assertFalse(db.has_admin_user()) self.assertEqual(len(db.get_all_users()), 1) def test_bootstrap_attempts_are_persistently_limited(self): for _ in range(setup.BOOTSTRAP_IP_ATTEMPTS): self.assertEqual(self.bootstrap(setup_token="incorrect").status_code, 403) setup.initialize_setup_state() response = self.bootstrap() self.assertEqual(response.status_code, 429) self.assertGreater(int(response.headers["retry-after"]), 0) self.assertFalse(db.has_admin_user()) with db._connect() as conn: keys = [row[0] for row in conn.execute("SELECT key_hash FROM installation_setup_attempts")] self.assertNotIn("testclient", keys) def test_rate_limit_global_cap_and_expiry(self): with patch.object(setup, "time", return_value=1000): for number in range(setup.BOOTSTRAP_GLOBAL_ATTEMPTS): self.assertIsNone(setup.consume_bootstrap_attempt(f"192.0.2.{number}")) self.assertEqual(setup.consume_bootstrap_attempt("198.51.100.1"), 900) with patch.object(setup, "time", return_value=1901): self.assertIsNone(setup.consume_bootstrap_attempt("198.51.100.1")) def test_concurrent_attempts_cannot_bypass_rate_limit(self): with ThreadPoolExecutor(max_workers=8) as executor: results = list(executor.map(lambda _: setup.consume_bootstrap_attempt("192.0.2.1"), range(12))) self.assertEqual(results.count(None), setup.BOOTSTRAP_IP_ATTEMPTS) def test_concurrent_bootstraps_create_only_one_admin(self): barrier = Barrier(4) def synchronized_hash(_): barrier.wait(timeout=10) return "test-only-precomputed-hash" def create(number): try: setup.bootstrap_administrator(SETUP_TOKEN, f"admin-{number}", ADMIN_PASSWORD) return True except setup.SetupUnavailableError: return False with patch.object(setup, "hash_password", side_effect=synchronized_hash): with ThreadPoolExecutor(max_workers=4) as executor: results = list(executor.map(create, range(4))) self.assertEqual(results.count(True), 1) self.assertEqual(len(db.get_all_users()), 1) def test_state_mutations_require_admin_and_progress_resumes(self): self.assertEqual(self.bootstrap().status_code, 201) db.create_user("viewer", ADMIN_PASSWORD) user_headers = {"Authorization": f"Bearer {create_access_token('viewer', 'user')}"} for path, method, kwargs in ( ("/setup/state", "get", {}), ("/setup/state", "put", {"json": {"step": "review"}}), ("/setup/complete", "post", {}), ): with self.subTest(path=path, method=method): call = getattr(self.client, method) self.assertEqual(call(path, **kwargs).status_code, 401) self.assertEqual(call(path, headers=user_headers, **kwargs).status_code, 403) response = self.client.put("/setup/state", json={"step": "preferences"}, headers=self.admin_headers()) self.assertEqual(response.status_code, 200) setup.initialize_setup_state() db.init_db() self.assertEqual(setup.get_setup_state()["step"], "preferences") self.assertTrue(setup.is_setup_required()) self.assertEqual(self.client.put( "/setup/state", json={"step": "invalid"}, headers=self.admin_headers() ).status_code, 422) def test_completion_invokes_worker_callback_and_cannot_reopen_bootstrap(self): self.assertEqual(self.bootstrap().status_code, 201) callback = AsyncMock() self.app.state.on_setup_complete = callback response = self.client.post("/setup/complete", headers=self.admin_headers()) self.assertEqual(response.status_code, 200, response.text) self.assertTrue(response.json()["completed"]) self.assertIsNotNone(response.json()["completed_at"]) callback.assert_awaited_once() self.assertFalse(setup.is_setup_required()) # A retry can restart an idempotent callback if the first response was # interrupted, while keeping the original completion timestamp. retry = self.client.post("/setup/complete", headers=self.admin_headers()) self.assertEqual(retry.json(), response.json()) self.assertEqual(callback.await_count, 2) self.client.put("/setup/state", json={"step": "administrator"}, headers=self.admin_headers()) with db._connect() as conn: conn.execute("DELETE FROM users") self.assertEqual(self.bootstrap().status_code, 409) self.assertEqual(setup.get_setup_state()["step"], "review") def test_completion_requires_an_administrator(self): with self.assertRaises(setup.SetupUnavailableError): setup.complete_setup() self.assertTrue(setup.is_setup_required()) def test_sync_callback_is_supported(self): self.assertEqual(self.bootstrap().status_code, 201) called = [] self.app.state.on_setup_complete = lambda: called.append(True) response = self.client.post("/setup/complete", headers=self.admin_headers()) self.assertEqual(response.status_code, 200) self.assertEqual(called, [True]) if __name__ == "__main__": unittest.main()