Compare commits

..
Author SHA1 Message Date
Typosaurus 3fed6463be ticket #75 attempt 1 2026-07-19 20:08:35 +00:00
8 changed files with 73 additions and 64 deletions
File diff suppressed because one or more lines are too long
+18 -18
View File
@@ -4,12 +4,13 @@ import asyncio
import fcntl
import logging
import os
import re
import time
from collections import defaultdict
from contextlib import asynccontextmanager, contextmanager
from pathlib import Path
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.responses import HTMLResponse, RedirectResponse, Response
from fastapi.staticfiles import StaticFiles
from fastapi.exceptions import RequestValidationError
from starlette.middleware.gzip import GZipMiddleware
@@ -128,9 +129,6 @@ RATE_LIMIT = int(os.environ.get("DEVPLACE_RATE_LIMIT", "60"))
RATE_WINDOW = 60
WEB_WORKERS = max(1, int(os.environ.get("DEVPLACE_WEB_WORKERS", "1")))
RATE_LIMIT_DISABLED = os.environ.get("DEVPLACE_DISABLE_RATE_LIMIT") == "1"
LOGIN_EMAIL_RATE_LIMIT = int(os.environ.get("DEVPLACE_LOGIN_EMAIL_RATE_LIMIT", "10"))
_email_rate_limit_store: dict[str, list[float]] = defaultdict(list)
HOT_SETTINGS_TTL = 2.0
_hot_settings_value: dict = {}
@@ -170,20 +168,6 @@ def _worker_rate_limit(limit: int) -> int:
return max(1, -(-limit // WEB_WORKERS))
def check_email_rate_limit(email: str) -> bool:
now = time.time()
window_start = now - RATE_WINDOW
limit = LOGIN_EMAIL_RATE_LIMIT
timestamps = [
t for t in _email_rate_limit_store.get(email, []) if t > window_start
]
if len(timestamps) >= limit:
return False
timestamps.append(now)
_email_rate_limit_store[email] = timestamps
return True
_service_lock_handle = None
@@ -628,6 +612,22 @@ async def response_timing(request: Request, call_next):
return response
_KNOWN_CRAWLERS = re.compile(
r"(google.*read.*aloud|googlebot|bingbot|slurp|duckduckbot|baiduspider|yandexbot)",
re.IGNORECASE,
)
_AUTH_GATED_PATHS = frozenset({"/messages", "/notifications", "/game", "/admin"})
@app.middleware("http")
async def crawler_detection(request: Request, call_next):
if request.method == "GET" and request.url.path in _AUTH_GATED_PATHS:
ua = request.headers.get("user-agent", "")
if _KNOWN_CRAWLERS.search(ua):
return Response(status_code=204)
return await call_next(request)
app.add_middleware(GZipMiddleware, minimum_size=512, compresslevel=5)
-17
View File
@@ -20,7 +20,6 @@ from devplacepy.responses import respond, action_result, wants_json, json_error
from devplacepy.schemas import AuthPageOut
from devplacepy.services.audit import record as audit
from devplacepy.dependencies import json_or_form
from devplacepy.main import check_email_rate_limit
logger = logging.getLogger(__name__)
router = APIRouter()
@@ -73,22 +72,6 @@ async def login(request: Request, data: Annotated[LoginForm, Depends(json_or_for
metadata={"email": email},
summary=f"failed login attempt for {email}",
)
if not check_email_rate_limit(email):
audit.record(
request,
"security.rate_limit.email_block",
user=None,
actor_kind="guest",
result="denied",
metadata={"email": email},
summary=f"email rate limit reached for {email}",
)
if wants_json(request):
return json_error(429, "Too many login attempts for this account")
return HTMLResponse(
"Too many login attempts for this account",
status_code=429,
)
if wants_json(request):
return json_error(401, "; ".join(errors), errors=errors)
seo_ctx = base_seo_context(request, title="Sign In", robots="noindex,nofollow")
+1
View File
@@ -17,6 +17,7 @@ async def robots_txt(request: Request):
Disallow: /auth/
Disallow: /messages/
Disallow: /notifications/
Disallow: /game/
Disallow: /votes/
Disallow: /avatar/
Disallow: /follow/
-24
View File
@@ -181,27 +181,3 @@ def test_rate_limit_block_recorded(monkeypatch):
event_key="security.rate_limit.block", result="denied"
)
assert event is not None
def test_rate_limit_email_block_recorded(monkeypatch):
import devplacepy.main as m
from starlette.testclient import TestClient
monkeypatch.setattr(m, "LOGIN_EMAIL_RATE_LIMIT", 2)
m._email_rate_limit_store.clear()
client = TestClient(m.app)
email = "rate-limited@test.dev"
codes = [
client.post(
"/auth/login",
data={"email": email, "password": "wrong"},
).status_code
for _ in range(3)
]
first_two = codes[:2]
assert all(c == 401 for c in first_two), first_two
assert codes[2] == 429, codes
event = get_table("audit_log").find_one(
event_key="security.rate_limit.email_block", result="denied"
)
assert event is not None
+4 -3
View File
@@ -196,7 +196,8 @@ def test_robots_txt_exists(app_server):
assert "Sitemap:" in r.text
def test_robots_disallows_admin_and_uploads(app_server):
def test_robots_disallows_auth_gated_paths(app_server):
r = requests.get(f"{BASE_URL}/robots.txt")
assert "Disallow: /admin/" in r.text
assert "Disallow: /uploads/" in r.text
required = ["/auth/", "/game/", "/messages/", "/notifications/", "/admin/", "/uploads/"]
for path in required:
assert f"Disallow: {path}" in r.text, f"Missing Disallow: {path}"
+50
View File
@@ -0,0 +1,50 @@
# retoor <retoor@molodetz.nl>
import requests
from tests.conftest import BASE_URL
def test_bot_blocked_on_auth_path(app_server):
headers = {"User-Agent": "Google-Read-Aloud"}
r = requests.get(f"{BASE_URL}/messages", headers=headers)
assert r.status_code == 204
assert r.content == b""
def test_bot_blocked_on_game_path(app_server):
headers = {"User-Agent": "Google-Read-Aloud"}
r = requests.get(f"{BASE_URL}/game", headers=headers)
assert r.status_code == 204
assert r.content == b""
def test_bot_blocked_on_admin_path(app_server):
headers = {"User-Agent": "Google-Read-Aloud"}
r = requests.get(f"{BASE_URL}/admin", headers=headers)
assert r.status_code == 204
assert r.content == b""
def test_bot_not_blocked_on_public_path(app_server):
headers = {"User-Agent": "Google-Read-Aloud"}
r = requests.get(f"{BASE_URL}/", headers=headers)
assert r.status_code == 200
def test_normal_user_not_blocked(app_server):
r = requests.get(f"{BASE_URL}/messages", allow_redirects=False)
assert r.status_code == 303
def test_known_crawler_user_agent_variants(app_server):
ua_list = [
"Googlebot",
"Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)",
"bingbot/2.0; +http://www.bing.com/bingbot.htm",
"Mozilla/5.0 (compatible; DuckDuckBot-Https/1.1; ...)",
"Mozilla/5.0 (compatible; Baiduspider/2.0; +http://www.baidu.com/search/spider.html)",
]
for ua in ua_list:
headers = {"User-Agent": ua}
r = requests.get(f"{BASE_URL}/admin", headers=headers)
assert r.status_code == 204, f"Expected 204 for UA: {ua}"
-1
View File
@@ -23,7 +23,6 @@ os.environ["DEVPLACE_DATA_DIR"] = str(_TEST_DATA_DIR)
os.environ["SECRET_KEY"] = "test-secret-key"
os.environ["DEVPLACE_DISABLE_SERVICES"] = "1"
os.environ["DEVPLACE_RATE_LIMIT"] = "1000000"
os.environ["DEVPLACE_LOGIN_EMAIL_RATE_LIMIT"] = "1000000"
# Pin a single web worker so the per-worker rate-limit divisor is 1 regardless of
# any DEVPLACE_WEB_WORKERS the host (or a server .env) exports.
os.environ["DEVPLACE_WEB_WORKERS"] = "1"