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 fcntl
import logging import logging
import os import os
import re
import time import time
from collections import defaultdict from collections import defaultdict
from contextlib import asynccontextmanager, contextmanager from contextlib import asynccontextmanager, contextmanager
from pathlib import Path from pathlib import Path
from fastapi import FastAPI, Request 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.staticfiles import StaticFiles
from fastapi.exceptions import RequestValidationError from fastapi.exceptions import RequestValidationError
from starlette.middleware.gzip import GZipMiddleware from starlette.middleware.gzip import GZipMiddleware
@@ -128,9 +129,6 @@ RATE_LIMIT = int(os.environ.get("DEVPLACE_RATE_LIMIT", "60"))
RATE_WINDOW = 60 RATE_WINDOW = 60
WEB_WORKERS = max(1, int(os.environ.get("DEVPLACE_WEB_WORKERS", "1"))) WEB_WORKERS = max(1, int(os.environ.get("DEVPLACE_WEB_WORKERS", "1")))
RATE_LIMIT_DISABLED = os.environ.get("DEVPLACE_DISABLE_RATE_LIMIT") == "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_TTL = 2.0
_hot_settings_value: dict = {} _hot_settings_value: dict = {}
@@ -170,20 +168,6 @@ def _worker_rate_limit(limit: int) -> int:
return max(1, -(-limit // WEB_WORKERS)) 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 _service_lock_handle = None
@@ -628,6 +612,22 @@ async def response_timing(request: Request, call_next):
return response 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) 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.schemas import AuthPageOut
from devplacepy.services.audit import record as audit from devplacepy.services.audit import record as audit
from devplacepy.dependencies import json_or_form from devplacepy.dependencies import json_or_form
from devplacepy.main import check_email_rate_limit
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
router = APIRouter() router = APIRouter()
@@ -73,22 +72,6 @@ async def login(request: Request, data: Annotated[LoginForm, Depends(json_or_for
metadata={"email": email}, metadata={"email": email},
summary=f"failed login attempt for {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): if wants_json(request):
return json_error(401, "; ".join(errors), errors=errors) return json_error(401, "; ".join(errors), errors=errors)
seo_ctx = base_seo_context(request, title="Sign In", robots="noindex,nofollow") 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: /auth/
Disallow: /messages/ Disallow: /messages/
Disallow: /notifications/ Disallow: /notifications/
Disallow: /game/
Disallow: /votes/ Disallow: /votes/
Disallow: /avatar/ Disallow: /avatar/
Disallow: /follow/ Disallow: /follow/
-24
View File
@@ -181,27 +181,3 @@ def test_rate_limit_block_recorded(monkeypatch):
event_key="security.rate_limit.block", result="denied" event_key="security.rate_limit.block", result="denied"
) )
assert event is not None 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 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") r = requests.get(f"{BASE_URL}/robots.txt")
assert "Disallow: /admin/" in r.text required = ["/auth/", "/game/", "/messages/", "/notifications/", "/admin/", "/uploads/"]
assert "Disallow: /uploads/" in r.text 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["SECRET_KEY"] = "test-secret-key"
os.environ["DEVPLACE_DISABLE_SERVICES"] = "1" os.environ["DEVPLACE_DISABLE_SERVICES"] = "1"
os.environ["DEVPLACE_RATE_LIMIT"] = "1000000" 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 # 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. # any DEVPLACE_WEB_WORKERS the host (or a server .env) exports.
os.environ["DEVPLACE_WEB_WORKERS"] = "1" os.environ["DEVPLACE_WEB_WORKERS"] = "1"