mirror of
https://github.com/openhop-dev/openhop_repeater.git
synced 2026-10-23 21:43:02 +00:00
297 lines
11 KiB
Python
297 lines
11 KiB
Python
"""Provisioning must not reopen when optional node/radio settings change."""
|
|
|
|
import copy
|
|
import json
|
|
import threading
|
|
from http.client import HTTPConnection
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock
|
|
from wsgiref.simple_server import WSGIRequestHandler, make_server
|
|
|
|
import cherrypy
|
|
import pytest
|
|
import yaml
|
|
|
|
from repeater.config_manager import ConfigManager
|
|
from repeater.web.api_endpoints import APIEndpoints
|
|
|
|
|
|
def make_config(password="custom-test-password", node_name="mesh-repeater-01", radio=None):
|
|
return {
|
|
"repeater": {"node_name": node_name, "security": {"admin_password": password}},
|
|
"radio_type": radio,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def api_env(tmp_path, monkeypatch):
|
|
config = make_config()
|
|
path = tmp_path / "config.yaml"
|
|
path.write_text(yaml.safe_dump(config))
|
|
api = APIEndpoints(config=config, config_path=str(path))
|
|
api.config_manager = ConfigManager(str(path), config)
|
|
monkeypatch.setattr(api.config_manager, "live_update_daemon", Mock(return_value=True))
|
|
monkeypatch.setattr(
|
|
cherrypy,
|
|
"request",
|
|
SimpleNamespace(method="POST", json={}, user=None, headers={}, params={}),
|
|
)
|
|
monkeypatch.setattr(cherrypy, "response", SimpleNamespace(status=200, headers={}))
|
|
monkeypatch.setattr(threading.Thread, "start", Mock())
|
|
return api, path
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"radio", [None, "", "none", "null", "disabled", "off", "no_radio", "modem_usb"]
|
|
)
|
|
@pytest.mark.parametrize("node_name", ["", "mesh-repeater-01", "configured-node"])
|
|
def test_custom_password_closes_setup_independent_of_optional_settings(radio, node_name):
|
|
api = APIEndpoints.__new__(APIEndpoints)
|
|
needs_setup, reasons = api._setup_status_from_config(
|
|
make_config(node_name=node_name, radio=radio)
|
|
)
|
|
assert needs_setup is False
|
|
assert reasons["default_password"] is False
|
|
|
|
|
|
@pytest.mark.parametrize("password", [None, "", "admin123", " ", "\t\n"])
|
|
def test_legacy_fresh_config_remains_usable(password):
|
|
api = APIEndpoints.__new__(APIEndpoints)
|
|
assert api._setup_status_from_config(make_config(password=password))[0] is True
|
|
|
|
|
|
@pytest.mark.parametrize("endpoint", ["setup_wizard", "config_import"])
|
|
def test_anonymous_mutation_denied_from_persisted_custom_password(api_env, endpoint):
|
|
api, path = api_env
|
|
before = path.read_bytes()
|
|
# Persisted credentials take precedence over stale startup defaults.
|
|
api.config["repeater"]["security"]["admin_password"] = "admin123"
|
|
cherrypy.request.json = {"config": make_config(password="replacement-password")}
|
|
result = getattr(api, endpoint)()
|
|
assert cherrypy.response.status == 403
|
|
assert result["success"] is False
|
|
assert path.read_bytes() == before
|
|
api.config_manager.live_update_daemon.assert_not_called()
|
|
|
|
|
|
def test_completed_wizard_persists_marker_even_with_legacy_default_password(api_env):
|
|
api, path = api_env
|
|
api.config.update(make_config(password="admin123"))
|
|
path.write_text(yaml.safe_dump(api.config))
|
|
cherrypy.request.json = {
|
|
"node_name": "mesh-repeater-01",
|
|
"hardware_key": "kiss",
|
|
"admin_password": "admin123",
|
|
"radio_preset": {"frequency": 869.618, "bandwidth": 62.5},
|
|
}
|
|
assert api.setup_wizard()["success"] is True
|
|
persisted = yaml.safe_load(path.read_text())
|
|
assert persisted["setup_completed"] is True
|
|
assert api.config["setup_completed"] is True
|
|
persisted["radio_type"] = None
|
|
assert api._setup_status_from_config(persisted)[0] is False
|
|
assert api.needs_setup()["needs_setup"] is False
|
|
assert api.setup_wizard()["success"] is False
|
|
assert cherrypy.response.status == 403
|
|
|
|
|
|
def test_explicit_false_marker_cannot_override_custom_password():
|
|
api = APIEndpoints.__new__(APIEndpoints)
|
|
config = make_config()
|
|
config["setup_completed"] = False
|
|
assert api._setup_status_from_config(config)[0] is False
|
|
|
|
|
|
def test_authenticated_restore_of_defaults_does_not_reopen_provisioning(api_env):
|
|
api, path = api_env
|
|
cherrypy.request.user = {"username": "admin", "auth_type": "jwt"}
|
|
cherrypy.request.json = {
|
|
"config": dict(make_config(password="admin123"), setup_completed=False)
|
|
}
|
|
assert api.config_import()["success"] is True
|
|
persisted = yaml.safe_load(path.read_text())
|
|
assert persisted["setup_completed"] is True
|
|
assert api._setup_status_from_config(persisted)[0] is False
|
|
|
|
|
|
@pytest.mark.parametrize("password", [False, 12345678, [], {}, " admin123 "])
|
|
def test_malformed_or_nonempty_custom_credentials_do_not_open_setup(password):
|
|
api = APIEndpoints.__new__(APIEndpoints)
|
|
assert api._setup_status_from_config(make_config(password=password))[0] is False
|
|
|
|
|
|
@pytest.mark.parametrize("password", [None, "admin123"])
|
|
def test_fresh_partial_restore_still_allows_wizard(api_env, password):
|
|
api, path = api_env
|
|
api.config.update(make_config(password=password))
|
|
path.write_text(yaml.safe_dump(api.config))
|
|
cherrypy.request.json = {"config": {"radio_type": None}}
|
|
assert api.config_import()["success"] is True
|
|
assert api.needs_setup()["needs_setup"] is True
|
|
|
|
|
|
def test_fresh_restore_custom_password_closes_followup_restore(api_env):
|
|
api, path = api_env
|
|
api.config.update(make_config(password="admin123"))
|
|
path.write_text(yaml.safe_dump(api.config))
|
|
cherrypy.request.json = {"config": make_config()}
|
|
assert api.config_import()["success"] is True
|
|
assert api.needs_setup()["needs_setup"] is False
|
|
assert api.config_import()["success"] is False
|
|
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
|
|
|
|
|
|
@pytest.mark.parametrize("endpoint", ["setup_wizard", "config_import"])
|
|
def test_anonymous_loopback_dispatch_rejects_provisioned_null_radio(
|
|
tmp_path, endpoint, monkeypatch
|
|
):
|
|
# Older direct-handler tests leave synthetic request/response objects behind.
|
|
# A real WSGI dispatch must use CherryPy's thread-local request context.
|
|
monkeypatch.setattr(cherrypy, "request", cherrypy._ThreadLocalProxy("request"))
|
|
monkeypatch.setattr(cherrypy, "response", cherrypy._ThreadLocalProxy("response"))
|
|
config = make_config()
|
|
path = tmp_path / "config.yaml"
|
|
path.write_text(yaml.safe_dump(config))
|
|
api = APIEndpoints(config=copy.deepcopy(config), config_path=str(path))
|
|
api.config_manager = Mock()
|
|
api.config_manager.save_to_file.return_value = True
|
|
# Same public provisioning routes: authorization must happen in the handlers.
|
|
app = cherrypy.Application(
|
|
SimpleNamespace(api=api),
|
|
config={
|
|
"/api/setup_wizard": {"tools.require_auth.on": False},
|
|
"/api/config_import": {"tools.require_auth.on": False},
|
|
},
|
|
)
|
|
with make_server("127.0.0.1", 0, app, handler_class=QuietHandler) as server:
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
connection = HTTPConnection("127.0.0.1", server.server_port, timeout=5)
|
|
try:
|
|
connection.request(
|
|
"POST",
|
|
f"/api/{endpoint}",
|
|
json.dumps({"config": make_config()}),
|
|
{"Content-Type": "application/json"},
|
|
)
|
|
response = connection.getresponse()
|
|
body = json.loads(response.read())
|
|
assert response.status == 403
|
|
assert body["success"] is False
|
|
assert yaml.safe_load(path.read_text()) == config
|
|
api.config_manager.update_and_save.assert_not_called()
|
|
finally:
|
|
connection.close()
|
|
server.shutdown()
|
|
thread.join(timeout=5)
|