Count per-address limits on the hop a trusted proxy added; limit signup

The limits read the leftmost X-Forwarded-For entry, which the client writes,
and the LAN nginx appended to it. The address now comes from uvicorn's
proxy-headers middleware (rightmost hop not in FORWARDED_ALLOW_IPS), applied
in the app too so tunnel-replayed calls resolve the same way; nginx sets the
header to $remote_addr and the compose backend trusts the Docker networks.
/users/signup gets a per-address limit of 10 an hour.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013GAeg4Jo6xjZzjwy9QDWMQ
This commit is contained in:
2026-10-06 12:15:58 +02:00
co-authored by Claude Opus 5.5
parent 7f06260c1a
commit 8cfe5c4e6e
10 changed files with 82 additions and 19 deletions
+7 -3
View File
@@ -4,6 +4,7 @@ from fastapi.testclient import TestClient
from fluksio.cloud import config as cloud_config
from fluksio.core.config import settings
from fluksio.main import app
PREFIX = f"{settings.API_V1_STR}/panels"
DASHBOARDS = f"{settings.API_V1_STR}/dashboards"
@@ -680,11 +681,13 @@ def test_the_device_asking_is_named_before_anyone_approves(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
"""Approving a code adopts whatever holds it, so it is worth a look."""
started = client.post(
# Behind a trusted proxy (127.0.0.1 by default), as the tunnel replays it.
proxied = TestClient(app, client=("127.0.0.1", 50000))
started = proxied.post(
f"{PREFIX}/pair",
headers={
"user-agent": "Mozilla/5.0 (X11; CrOS aarch64)",
"x-forwarded-for": "203.0.113.7, 10.0.0.1",
"x-forwarded-for": "198.51.100.1, 203.0.113.7",
},
).json()
@@ -693,8 +696,9 @@ def test_the_device_asking_is_named_before_anyone_approves(
)
assert looked.status_code == 200
assert "CrOS" in looked.json()["device"]
# The first hop, not the proxy that relayed it.
# The hop the proxy added, not the one the device wrote in front of it.
assert "203.0.113.7" in looked.json()["device"]
assert "198.51.100.1" not in looked.json()["device"]
assert looked.json()["remote"] is False
# Nobody without an account gets to enumerate what is waiting.
+29
View File
@@ -5,8 +5,11 @@ from fastapi.testclient import TestClient
from sqlmodel import Session, select
from fluksio import crud
from fluksio.api.routes import oauth
from fluksio.api.routes.users import SIGNUP_ATTEMPTS
from fluksio.core.config import settings
from fluksio.core.security import verify_password
from fluksio.main import app
from fluksio.models import User, UserCreate
from tests.utils.user import create_random_user
from tests.utils.utils import random_email, random_lower_string
@@ -373,6 +376,32 @@ def test_register_user_already_exists_error(client: TestClient) -> None:
assert r.json()["detail"] == "The user with this email already exists in the system"
def test_signup_is_limited_per_address_behind_a_proxy() -> None:
"""The limit counts the hop the trusted proxy added, not one the client wrote.
The leftmost hop is the client's own to fill in, so keying on it let one
caller look like a new one on every attempt.
"""
proxied = TestClient(app, client=("127.0.0.1", 50000)) # a trusted proxy
url = f"{settings.API_V1_STR}/users/signup"
taken = {"email": settings.FIRST_SUPERUSER, "password": random_lower_string()}
def attempt(forwarded: str) -> int:
return proxied.post(
url, json=taken, headers={"x-forwarded-for": forwarded}
).status_code
try:
with patch("fluksio.core.config.settings.SIGNUP_ENABLED", True):
for n in range(SIGNUP_ATTEMPTS):
assert attempt(f"198.51.100.{n}, 203.0.113.7") == 400
assert attempt("198.51.100.99, 203.0.113.7") == 429
assert attempt("203.0.113.8") == 400
finally:
for address in ("203.0.113.7", "203.0.113.8"):
oauth._hits.pop(f"signup:{address}", None)
def test_update_user(
client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None: