diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py
index db24e3b..cc015aa 100644
--- a/repeater/web/api_endpoints.py
+++ b/repeater/web/api_endpoints.py
@@ -6,6 +6,7 @@ import secrets
 import time
 from concurrent.futures import TimeoutError as FutureTimeoutError
 from datetime import datetime, timezone
+from threading import RLock
 from typing import Callable, Optional
 
 import cherrypy
@@ -209,6 +210,11 @@ POLICY_GROUP_KINDS = {
 
 
 class APIEndpoints:
+    # Serialize both provisioning routes across mounts in this process, from
+    # authorization through persistence. Do not reuse ConfigManager's locks:
+    # imports must retain its normal persistence/live-update callbacks.
+    _provisioning_lock = RLock()
+
     # How long /api/query_neighbor_scopes holds a request open. The scope helper's
     # response window is normally its 5 s floor; a slow radio config (SF12) or a
     # duty-cycle deferral can push a single query past this, in which case the
@@ -232,6 +238,7 @@ class APIEndpoints:
         self.event_loop = event_loop
         self.daemon_instance = daemon_instance
         self._config_path = config_path or "/etc/openhop_repeater/config.yaml"
+
         self.cad_calibration = CADCalibrationEngine(daemon_instance, event_loop)
 
         # Initialize ConfigManager for centralized config management
@@ -1506,6 +1513,10 @@ class APIEndpoints:
     @cherrypy.tools.json_in()
     def setup_wizard(self):
         """Complete initial setup wizard configuration"""
+        with self._provisioning_lock:
+            return self._setup_wizard_locked()
+
+    def _setup_wizard_locked(self):
         try:
             self._require_post()
             data = cherrypy.request.json
@@ -8131,6 +8142,10 @@ class APIEndpoints:
         Returns: {"success": true, "message": "...", "restart_required": true,
                   "sections_updated": [...]}
         """
+        with self._provisioning_lock:
+            return self._config_import_locked()
+
+    def _config_import_locked(self):
         self._set_cors_headers()
         if cherrypy.request.method == "OPTIONS":
             return ""
diff --git a/tests/test_setup_authorization_hardening.py b/tests/test_setup_authorization_hardening.py
index 7eb9263..230b4dc 100644
--- a/tests/test_setup_authorization_hardening.py
+++ b/tests/test_setup_authorization_hardening.py
@@ -141,6 +141,111 @@ def test_fresh_restore_custom_password_closes_followup_restore(api_env):
     assert cherrypy.response.status == 403
 
 
+@pytest.mark.parametrize("second_endpoint", ["setup_wizard", "config_import"])
+def test_concurrent_anonymous_provisioning_has_only_one_winner(
+    tmp_path, monkeypatch, second_endpoint
+):
+    config = make_config(password="admin123")
+    path = tmp_path / "config.yaml"
+    path.write_text(yaml.safe_dump(config))
+    api = APIEndpoints(config=config, config_path=str(path))
+    monkeypatch.setattr(api.config_manager, "live_update_daemon", Mock(return_value=True))
+    request = threading.local()
+    response = threading.local()
+    monkeypatch.setattr(cherrypy, "request", request)
+    monkeypatch.setattr(cherrypy, "response", response)
+    real_start = threading.Thread.start
+
+    def safe_start(thread):
+        if getattr(thread._target, "__name__", "") != "delayed_restart":
+            real_start(thread)
+
+    monkeypatch.setattr(threading.Thread, "start", safe_start)
+    first_checked = threading.Event()
+    second_contending = threading.Event()
+    release_first = threading.Event()
+    first_finished = threading.Event()
+    original_status = api._setup_status_from_config
+
+    def gated_status(current):
+        result = original_status(current)
+        if threading.current_thread().name == "first":
+            first_checked.set()
+            assert release_first.wait(5)
+        elif not first_finished.is_set():
+            # Without serialization, capture the stale authorization decision.
+            second_contending.set()
+            assert first_finished.wait(5)
+        return result
+
+    monkeypatch.setattr(api, "_setup_status_from_config", gated_status)
+    # Observe attempted acquisition, not elapsed time, to prove overlap without
+    # deadlocking the correctly serialized implementation at a barrier.
+    if hasattr(api, "_provisioning_lock"):
+        lock = api._provisioning_lock
+
+        class ObservedLock:
+            def __enter__(self):
+                if threading.current_thread().name == "second":
+                    second_contending.set()
+                return lock.__enter__()
+
+            def __exit__(self, *args):
+                return lock.__exit__(*args)
+
+        monkeypatch.setattr(api, "_provisioning_lock", ObservedLock())
+
+    outcomes = {}
+    errors = []
+
+    def invoke(name, endpoint):
+        request.method = "POST"
+        request.user = None
+        request.headers = {}
+        request.params = {}
+        request.json = (
+            {"config": make_config(password="loser-password")}
+            if endpoint == "config_import"
+            else {
+                "node_name": name,
+                "hardware_key": "kiss",
+                "admin_password": f"{name}-password",
+                "radio_preset": {"frequency": 869.618, "bandwidth": 62.5},
+            }
+        )
+        response.status = 200
+        response.headers = {}
+        try:
+            outcomes[name] = (getattr(api, endpoint)(), response.status)
+        except BaseException as exc:  # noqa: BLE001 - propagate worker failures to pytest
+            errors.append(exc)
+        finally:
+            if name == "first":
+                first_finished.set()
+
+    first = threading.Thread(target=invoke, args=("first", "setup_wizard"), name="first")
+    second = threading.Thread(target=invoke, args=("second", second_endpoint), name="second")
+    first.start()
+    try:
+        assert first_checked.wait(5)
+        second.start()
+        assert second_contending.wait(5)
+    finally:
+        release_first.set()
+        first.join(5)
+        if second.ident is not None:
+            second.join(5)
+    assert not first.is_alive() and not second.is_alive()
+    assert not errors
+    assert outcomes["first"][0]["success"] is True
+    assert outcomes["second"][1] == 403
+    assert outcomes["second"][0]["success"] is False
+    persisted = yaml.safe_load(path.read_text())
+    assert persisted["repeater"]["security"]["admin_password"] == "first-password"
+    assert persisted["setup_completed"] is True
+    api.config_manager.live_update_daemon.assert_not_called()
+
+
 class QuietHandler(WSGIRequestHandler):
     def log_message(self, format, *args):
         pass