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>
68 lines
3.2 KiB
Python
68 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Execute the WebConfig login throttle and the real /api/login handler."""
|
|
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]
|
|
CLIENT_KEY = "req->client() ? req->client()->getRemoteAddress() : 0"
|
|
|
|
|
|
class WebConfigLoginThrottleTest(unittest.TestCase):
|
|
def compile_run(self, source, includes=(), expect_rejection=False):
|
|
compiler = shutil.which("g++") or shutil.which("clang++")
|
|
self.assertIsNotNone(compiler)
|
|
with tempfile.TemporaryDirectory(prefix="meshcore-webconfig-login-") as temporary:
|
|
work = Path(temporary)
|
|
generated = work / "login.cpp"
|
|
generated.write_text(source)
|
|
binary = work / "login"
|
|
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(ROOT / "src"),
|
|
*("-I" + str(path) for path in includes),
|
|
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=30)
|
|
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 handler_fixture(self):
|
|
source = (ROOT / "src/helpers/esp32/WebConfigServer.cpp").read_text()
|
|
methods = "\n".join(extract_braced(source, signature) for signature in (
|
|
"static bool fixedTimeEquals(", "void WebConfigServer::handleLogin("))
|
|
self.assertIn(CLIENT_KEY, methods)
|
|
return (ROOT / "test/fixtures/webconfig_login_handler.cpp").read_text().replace("@METHODS@", methods)
|
|
|
|
def json_include(self):
|
|
json_header = next((ROOT / ".pio/libdeps").glob("*/ArduinoJson/src/ArduinoJson.h"), None)
|
|
self.assertIsNotNone(json_header, "native dependency headers are required")
|
|
return json_header.parent
|
|
|
|
def test_policy_isolates_clients_and_bounds_all_clients_together(self):
|
|
self.compile_run((ROOT / "test/fixtures/webconfig_login_throttle.cpp").read_text())
|
|
|
|
def test_login_handler_locks_only_the_failing_address(self):
|
|
self.compile_run(self.handler_fixture(), (self.json_include(),))
|
|
|
|
def test_old_shared_lockout_fails_executable_regression(self):
|
|
# Every client on one key is the old global lockout.
|
|
fixture = self.handler_fixture()
|
|
old = fixture.replace(CLIENT_KEY, "0", 1)
|
|
self.assertNotEqual(old, fixture)
|
|
self.compile_run(old, (self.json_include(),), expect_rejection=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|