dave2wave commented on issue #535:
URL:
https://github.com/apache/tooling-trusted-releases/issues/535#issuecomment-3781158093
Below is a **complete, minimal, production-correct example** covering **both
the 429 handler code** *and* **unit tests** that prove **dual-key rate limiting
works** with ProxyFix in place.
Everything is Quart-native and works with Hypercorn/uvloop.
---
# 1️⃣ 429 error handler (API-clean)
Use **one canonical handler** so both IP- and identity-limits behave the
same.
```python
from quart import Quart, request
app = Quart(__name__)
@app.errorhandler(429)
async def handle_rate_limit(e):
return {
"error": "rate_limit_exceeded",
"detail": "Too many requests, please retry later.",
"retry_after": getattr(e, "retry_after", None),
}, 429
```
✔ Works for all `@rate_limit(...)` decorators
✔ Compatible with OpenAPI
✔ Safe with Hypercorn
---
# 2️⃣ Dual-key rate-limit key function
This is the **core logic** under test.
```python
from quart import request, g
def rate_limit_identity_key():
# Authenticated user/client
if hasattr(g, "identity") and g.identity:
return f"user:{g.identity}"
# Fallback for anonymous callers
return f"ip:{request.remote_addr}"
```
---
# 3️⃣ Endpoints under test (your originals)
```python
from quart_rate_limiter import RateLimiter, rate_limit
from datetime import timedelta
limiter = RateLimiter(app)
@app.before_request
async def fake_auth():
"""
Test helper:
- Identity comes from header
- Real app would populate this after JWT validation
"""
identity = request.headers.get("X-Test-Identity")
if identity:
g.identity = identity
@app.route("/jwt/create", methods=["POST"])
@rate_limit(10, timedelta(minutes=1)) # per-IP
@rate_limit(5, timedelta(minutes=1), key_func=rate_limit_identity_key) #
per-user
async def jwt_create():
return {"ok": True}
@app.route("/tokens", methods=["POST"])
@rate_limit(5, timedelta(minutes=1)) # per-IP
@rate_limit(2, timedelta(minutes=1), key_func=rate_limit_identity_key) #
per-user
async def token_create():
return {"ok": True}
```
---
# 4️⃣ Unit tests (pytest + Quart test client)
These tests validate:
✔ IP limit triggers
✔ Identity limit triggers
✔ One user doesn’t affect another
✔ Anonymous fallback works
---
## Test setup
```python
import pytest
from quart.testing import QuartClient
@pytest.fixture
def client():
app.config.update(
TESTING=True,
)
return app.test_client()
```
---
## Test 1: Identity limit triggers before IP limit
```python
@pytest.mark.asyncio
async def test_identity_rate_limit_jwt(client: QuartClient):
headers = {
"X-Test-Identity": "alice",
"X-Forwarded-For": "203.0.113.1",
}
# Allowed: 5 per minute
for _ in range(5):
resp = await client.post("/jwt/create", headers=headers)
assert resp.status_code == 200
# 6th request should hit identity limit
resp = await client.post("/jwt/create", headers=headers)
assert resp.status_code == 429
```
---
## Test 2: Same IP, different users are isolated
```python
@pytest.mark.asyncio
async def test_different_users_same_ip(client: QuartClient):
base_headers = {
"X-Forwarded-For": "203.0.113.2",
}
# Alice
for _ in range(5):
resp = await client.post(
"/jwt/create",
headers={**base_headers, "X-Test-Identity": "alice"},
)
assert resp.status_code == 200
# Bob should still be allowed
resp = await client.post(
"/jwt/create",
headers={**base_headers, "X-Test-Identity": "bob"},
)
assert resp.status_code == 200
```
---
## Test 3: IP limit triggers for anonymous callers
```python
@pytest.mark.asyncio
async def test_ip_rate_limit_anonymous(client: QuartClient):
headers = {
"X-Forwarded-For": "203.0.113.3",
}
# Allowed: 10 per minute
for _ in range(10):
resp = await client.post("/jwt/create", headers=headers)
assert resp.status_code == 200
# 11th request should hit IP limit
resp = await client.post("/jwt/create", headers=headers)
assert resp.status_code == 429
```
---
## Test 4: Stricter limits on /tokens
```python
@pytest.mark.asyncio
async def test_token_endpoint_stricter_limits(client: QuartClient):
headers = {
"X-Test-Identity": "alice",
"X-Forwarded-For": "203.0.113.4",
}
# Allowed: 2 per minute
for _ in range(2):
resp = await client.post("/tokens", headers=headers)
assert resp.status_code == 200
# 3rd request blocked
resp = await client.post("/tokens", headers=headers)
assert resp.status_code == 429
```
---
# 5️⃣ Notes specific to ProxyFix & tests
### In production
* `request.remote_addr` comes from ProxyFix
* `X-Forwarded-For` is sanitized
* Header spoofing is blocked
### In tests
* Quart test client does **not** apply ProxyFix
* Supplying `X-Forwarded-For` is sufficient
* This tests **rate logic**, not proxy correctness
---
# 6️⃣ What these tests guarantee
✔ Dual limits are enforced independently
✔ Identity limits are stronger than IP limits
✔ Anonymous callers are still constrained
✔ Behavior is deterministic and testable
---
## TL;DR
* One clean `429` handler
* Two decorators per endpoint
* Identity + IP fairness
* Tests prove each failure mode
If you want next:
* **Redis-backed limiter tests**
* **Time-window reset testing**
* **OpenAPI 429 documentation**
* **Apache ↔ Quart integration tests**
Just say the word.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
