0.2.0 — a notebook in your pocket, ready to be hosted #3
+15
-6
@@ -8,6 +8,8 @@ request returns 429 before any session is opened, which is the whole point of
|
||||
checking the limit before the password. The happy path can't be reached here and is
|
||||
not pretended at.
|
||||
"""
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from thoughtsync import ratelimit
|
||||
@@ -90,8 +92,13 @@ async def test_login_starts_refusing(app):
|
||||
body = {"email": "someone@example.com", "password": "wrong-password"}
|
||||
# Pre-load the account's counter to its limit rather than posting that many
|
||||
# times: every real attempt would need a database to reach the password check.
|
||||
for i in range(ratelimit.ACCOUNT_LIMIT):
|
||||
ratelimit.sign_in_by_account.record("someone@example.com", now=float(i))
|
||||
#
|
||||
# On the REAL clock, not an injected one. The window is trailing, so hits stamped
|
||||
# at t=0..9 are fifteen minutes stale the moment the route reads
|
||||
# `time.monotonic()` and get pruned before they can refuse anything.
|
||||
now = time.monotonic()
|
||||
for _ in range(ratelimit.ACCOUNT_LIMIT):
|
||||
ratelimit.sign_in_by_account.record("someone@example.com", now=now)
|
||||
resp = await client.post("/api/auth/login", json=body)
|
||||
assert resp.status_code == 429
|
||||
assert resp.headers.get("Retry-After")
|
||||
@@ -101,8 +108,9 @@ async def test_login_starts_refusing(app):
|
||||
|
||||
async def test_device_login_shares_the_account_counter(app):
|
||||
client = app.test_client()
|
||||
for i in range(ratelimit.ACCOUNT_LIMIT):
|
||||
ratelimit.sign_in_by_account.record("someone@example.com", now=float(i))
|
||||
now = time.monotonic()
|
||||
for _ in range(ratelimit.ACCOUNT_LIMIT):
|
||||
ratelimit.sign_in_by_account.record("someone@example.com", now=now)
|
||||
resp = await client.post(
|
||||
"/api/auth/device-login",
|
||||
json={"email": "someone@example.com", "password": "wrong-password"},
|
||||
@@ -114,8 +122,9 @@ async def test_device_login_shares_the_account_counter(app):
|
||||
|
||||
async def test_register_is_throttled_by_address(app):
|
||||
client = app.test_client()
|
||||
for i in range(ratelimit.REGISTER_LIMIT):
|
||||
ratelimit.register_by_address.record("203.0.113.9", now=float(i))
|
||||
now = time.monotonic()
|
||||
for _ in range(ratelimit.REGISTER_LIMIT):
|
||||
ratelimit.register_by_address.record("203.0.113.9", now=now)
|
||||
resp = await client.post(
|
||||
"/api/auth/register",
|
||||
json={"email": "new@example.com", "password": "a-long-enough-password"},
|
||||
|
||||
Reference in New Issue
Block a user