nullrouten0andClaude Opus 5.5
ef1e4a70cb
/api/login locked every client for 30 s after five failures from anyone. Anyone who could reach the login page could keep the owner locked out by failing on purpose. If no login succeeded after a lockout, the lock also returned from 2^31 to 2^32 ms (about 24.9 to 49.7 days) later, because the deadline was never cleared and was compared as a signed difference. Count failures per remote address instead. Five failures lock that address for 30 s, as before. A shared budget caps failures from all addresses together, so changing address does not add guesses: it holds six failures and refills one every 6 s, the old global rate, so one failing address can never empty it on its own. While it is empty every client waits, as before. Lock times are measured as elapsed time. The policy is a pure header, WebConfigLoginThrottle.h. test_webconfig_login_throttle.py runs it directly (isolation, boundaries, one-hour rate bounds, millis() half-wrap, saturation, eviction) and runs the real handleLogin with ArduinoJson under ASan/UBSan. Putting every client on one key, as the old code did, fails the handler test. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
75 lines
3.9 KiB
Python
75 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Execute actual recovery/login/logout gates with real JSON and fake sockets."""
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
from test_replay_reset_integration import extract_braced
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class WebConfigRecoveryAuthTest(unittest.TestCase):
|
|
def compile_run(self, source, expect_rejection=False):
|
|
compiler = shutil.which("g++") or shutil.which("clang++")
|
|
self.assertIsNotNone(compiler)
|
|
json_header = next((ROOT / ".pio/libdeps").glob("*/ArduinoJson/src/ArduinoJson.h"), None)
|
|
self.assertIsNotNone(json_header, "native dependency headers are required")
|
|
with tempfile.TemporaryDirectory(prefix="meshcore-webconfig-auth-") as temporary:
|
|
work = Path(temporary)
|
|
generated = work / "auth.cpp"
|
|
generated.write_text(source)
|
|
binary = work / "auth"
|
|
sanitizers = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all",
|
|
"-fno-pie", "-no-pie"] if sys.platform.startswith("linux") else [])
|
|
compiled = subprocess.run([compiler, "-std=c++17", "-Wall", "-Wextra",
|
|
*sanitizers, "-I" + str(json_header.parent),
|
|
"-I" + str(ROOT / "src"),
|
|
str(generated), "-o", str(binary)],
|
|
capture_output=True, text=True, timeout=60)
|
|
self.assertEqual(compiled.returncode, 0, compiled.stderr)
|
|
result = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
|
|
if expect_rejection:
|
|
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertIn("Assertion", result.stderr)
|
|
else:
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
|
|
def fixture(self):
|
|
source = (ROOT / "src/helpers/esp32/WebConfigServer.cpp").read_text()
|
|
methods = "\n".join(extract_braced(source, signature) for signature in (
|
|
"static bool fixedTimeEquals(", "bool WebConfigServer::checkAuth(",
|
|
"void WebConfigServer::handleLogin(", "void WebConfigServer::handleLogout("))
|
|
# The actual config POST entry gate must reject a replacement password
|
|
# before parsing or queueing it; exercise its exact production prefix.
|
|
post = extract_braced(source, "void WebConfigServer::handleConfigPost(")
|
|
gate = post[post.index("{") + 1:post.index(" if (req->contentLength()")]
|
|
methods += "\nvoid WebConfigServer::checkConfigPost(AsyncWebServerRequest* req) {" + gate + "\nconfig_accepted=true;\n}"
|
|
return (ROOT / "test/fixtures/webconfig_recovery_auth.cpp").read_text().replace("@METHODS@", methods)
|
|
|
|
def test_saved_network_fallback_requires_current_admin_session(self):
|
|
self.compile_run(self.fixture())
|
|
|
|
def test_old_interface_only_fallback_fails_executable_regression(self):
|
|
fixture = self.fixture()
|
|
old = fixture.replace("if (_initial_setup || _wifi_ssid[0] == 0) { noteNetworkActivity(); return true; }", "return true;", 1)
|
|
self.assertNotEqual(old, fixture)
|
|
self.compile_run(old, expect_rejection=True)
|
|
|
|
def test_old_unauthenticated_logout_fails_executable_regression(self):
|
|
fixture = self.fixture()
|
|
signature = "void WebConfigServer::handleLogout("
|
|
start = fixture.index(signature, fixture.index("@METHODS@") if "@METHODS@" in fixture else 0)
|
|
prefix, logout = fixture[:start], fixture[start:]
|
|
gate = ' if (!checkAuth(req)) { req->send(401, "application/json", "{\\"error\\":\\"auth\\"}"); return; }\n'
|
|
old = prefix + logout.replace(gate, "", 1)
|
|
self.assertNotEqual(old, fixture)
|
|
self.compile_run(old, expect_rejection=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|