feat(6866): the proxy hop count is an admin setting, and the server shows what each count reads

An admin can save how many reverse proxies stand in front of Grimoire.
A saved count wins over TRUSTED_PROXY_HOPS, which stays the starting
value. Each worker loads it at startup (under a deadline) and re-reads
it every 30s; the worker that saves applies it at once.

GET/PUT /api/admin/proxy returns the count, where it came from, and a
readout of the admin's own request: the socket peer, the forwarding
chains, what counts 0-3 would read, and the count that reads
Cloudflare's visitor when Cloudflare's headers are present. Saves are
audit-logged. A "too many attempts" reply now says where to look when
everyone is being blocked.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/api-reference.md

@@ -281,6 +281,7 @@ These serve the plugin's hooks.
 | GET / PUT | `/api/admin/smtp` · POST `/api/admin/smtp/test` | SMTP config + test email |
 | GET | `/api/admin/logs` · `/api/admin/logs/stats` | Log entries + category counts |
 | GET / PUT | `/api/admin/base-url` | Get / set the public base URL |
+| GET / PUT | `/api/admin/proxy` | Get / set how many reverse proxies stand in front of Grimoire (`{"hops": n}`, or `null` for the environment's count), with what each count would read for this request |
 | GET / PUT | `/api/admin/db-maintenance` (+ `/health`, POST `/run`) | VACUUM schedule, health, manual run |
 | POST / GET / DELETE | `/api/admin/invitations` (+ `/:id`) | Create / list / revoke invite links |
 | GET / PUT | `/api/admin/backups` | Scheduled backups: destination, schedule and recent runs |

src/grimoire/app.py

@@ -85,9 +85,9 @@ def create_app() -> Quart:
 
     if os.environ.get("TRUST_PROXY_HEADERS") and not os.environ.get("TRUSTED_PROXY_HOPS"):
         logger.warning(
-            "TRUST_PROXY_HEADERS is replaced by TRUSTED_PROXY_HOPS (now %d). Set "
-            "TRUSTED_PROXY_HOPS to the number of proxies in front of Grimoire and "
-            "remove TRUST_PROXY_HEADERS.",
+            "TRUST_PROXY_HEADERS is not used (the starting proxy hop count is %d). "
+            "Remove it; set the count in Settings → Config, or TRUSTED_PROXY_HOPS "
+            "to change the starting value.",
             Config.TRUSTED_PROXY_HOPS,
         )
 
@@ -202,6 +202,14 @@ def create_app() -> Quart:
         start_notification_loop()
         start_auth_token_retention_loop()
 
+        # The proxy hop count saved in Settings decides every request's client
+        # address, so it is loaded before serving, under a deadline (a failed
+        # or slow read leaves the environment's count in force), and re-read
+        # on an interval so a save in another worker reaches this one.
+        from grimoire.services import proxy_settings
+        await proxy_settings.load_at_startup()
+        proxy_settings.start_refresh_loop()
+
         # Write down any shipped default that moved in this release.
         #
         # INLINE AND AWAITED, deliberately, against the instinct the block

src/grimoire/config.py

@@ -124,8 +124,14 @@ class Config:
     BASE_URL: str = os.environ.get("BASE_URL", "http://localhost:5000").rstrip("/")
     # How many reverse proxies stand in front of the app; the client address
     # and scheme are read that many entries in from the right of the
-    # forwarding headers (grimoire/proxy.py). 0 trusts no proxy.
+    # forwarding headers (grimoire/proxy.py). 0 trusts no proxy. This is the
+    # install's starting value: a count saved in Settings → Config wins over
+    # it (services/proxy_settings.py).
     TRUSTED_PROXY_HOPS: int = _trusted_proxy_hops()
+    TRUSTED_PROXY_HOPS_FROM_ENV: bool = bool(
+        os.environ.get("TRUSTED_PROXY_HOPS", "").strip()
+        or os.environ.get("TRUST_PROXY_HEADERS", "").strip()
+    )
 
     # Git URL of the repo that ships this Grimoire plugin. There is one correct
     # value per deployment (the repo serving the plugin), so it's a config

src/grimoire/proxy.py

@@ -7,9 +7,10 @@ X-Forwarded-Proto, and each proxy on the road APPENDS its entry, so only the
 right-hand end of either header was written by a proxy we trust. Anything to
 the left of that is whatever the client chose to send.
 
-`TRUSTED_PROXY_HOPS` says how many proxies stand in front of the app (default
-1). The entry read is that many in from the right. Zero trusts no proxy: the
-headers are ignored and the socket peer is the client.
+The hop count says how many proxies stand in front of the app. An admin sets
+it in Settings → Config; until they do, it is `TRUSTED_PROXY_HOPS` from the
+environment (default 1). The entry read is that many in from the right. Zero
+trusts no proxy: the headers are ignored and the socket peer is the client.
 
 Never read X-Forwarded-For or X-Forwarded-Proto anywhere else. The client
 address keys the rate limits and the audit log, and the scheme decides the
@@ -43,10 +44,36 @@ def trusted_entry(header: str, hops: int) -> str | None:
     return entries[-min(hops, len(entries))]
 
 
-def _forwarded(name: str) -> str | None:
+# The count saved in Settings, or None when none is saved. Held here because
+# every request reads it and the read is synchronous; services/proxy_settings
+# loads it at startup, refreshes it on an interval in every worker, and sets it
+# at once in the worker that saved a new one.
+_saved_hops: int | None = None
+
+
+def set_saved_hops(hops: int | None) -> None:
+    global _saved_hops
+    _saved_hops = hops
+
+
+def saved_hops_in_force() -> int | None:
+    """The count saved in Settings that this worker holds, or None."""
+    return _saved_hops
+
+
+def trusted_hops() -> int:
+    """The hop count in force: the one saved in Settings, else the env's."""
+    return Config.TRUSTED_PROXY_HOPS if _saved_hops is None else _saved_hops
+
+
+def _chain(name: str) -> str:
     # Joined, because a header sent twice arrives as two values, and each
     # proxy's append may land on either one.
-    return trusted_entry(",".join(request.headers.getlist(name)), Config.TRUSTED_PROXY_HOPS)
+    return ",".join(request.headers.getlist(name))
+
+
+def _forwarded(name: str) -> str | None:
+    return trusted_entry(_chain(name), trusted_hops())
 
 
 def client_ip() -> str:
@@ -78,3 +105,60 @@ def apply_https_headers(response) -> None:
     HTTPS; plain HTTP is a supported way to run it."""
     if is_https():
         response.headers.setdefault("Strict-Transport-Security", HSTS_VALUE)
+
+
+# Cloudflare writes the visitor's address here on every request it proxies.
+# Read only to SUGGEST a hop count in the readout, never to decide one: a
+# client that reaches the app without Cloudflare can send it too.
+CLOUDFLARE_CLIENT_HEADER = "CF-Connecting-IP"
+CLOUDFLARE_MARKERS = (CLOUDFLARE_CLIENT_HEADER, "CF-Ray")
+READOUT_HOPS = range(0, 4)
+MAX_HOPS = 5
+
+
+def readout(peer: str, forwarded_for: str, forwarded_proto: str,
+            cloudflare_client: str | None, scheme: str = "http") -> dict:
+    """What each hop count would make of one request, so an admin can pick
+    the count that shows their own address. Pure: the request's parts come in
+    as arguments.
+
+    `suggested` is the smallest count that reads the address Cloudflare
+    reports, when the request came through Cloudflare; otherwise None, since
+    only the admin knows which address is theirs.
+    """
+    def read(hops: int) -> dict:
+        address = trusted_entry(forwarded_for, hops) or peer
+        proto = trusted_entry(forwarded_proto, hops)
+        https = scheme == "https" or (proto or "").lower() == "https"
+        return {"hops": hops, "client": address, "https": https}
+
+    suggested = None
+    if cloudflare_client:
+        suggested = next(
+            (h for h in range(1, MAX_HOPS + 1)
+             if trusted_entry(forwarded_for, h) == cloudflare_client),
+            None,
+        )
+    return {
+        "peer": peer,
+        "forwarded_for": [p.strip() for p in forwarded_for.split(",") if p.strip()],
+        "forwarded_proto": [p.strip() for p in forwarded_proto.split(",") if p.strip()],
+        "cloudflare": cloudflare_client is not None,
+        "cloudflare_client": cloudflare_client,
+        "rows": [read(h) for h in READOUT_HOPS],
+        "suggested": suggested,
+    }
+
+
+def request_readout() -> dict:
+    """`readout` for the current request."""
+    cloudflare = request.headers.get(CLOUDFLARE_CLIENT_HEADER)
+    if cloudflare is None and request.headers.get("CF-Ray"):
+        cloudflare = ""
+    return readout(
+        request.remote_addr or "",
+        _chain("X-Forwarded-For"),
+        _chain("X-Forwarded-Proto"),
+        cloudflare.strip() if cloudflare is not None else None,
+        request.scheme,
+    )

src/grimoire/rate_limit.py

@@ -2,15 +2,23 @@
 
 IMPORTANT — deployment note:
   Rate limit counters are stored in memory and are lost on process restart.
-  The bucket key is the client address from grimoire/proxy.py, read by
-  TRUSTED_PROXY_HOPS. A hop count that is too low keys every caller on the
-  proxy's address, so they all share one bucket.
+  The bucket key is the client address from grimoire/proxy.py, read by the
+  proxy hop count (Settings → Config). A count that is too low keys every
+  caller on the proxy's address, so they all share one bucket.
 """
 
 import asyncio
 import time
 from collections import defaultdict
 
+# Said beside every "too many attempts" reply, because the one case a person
+# can't diagnose on their own is everybody being blocked at once: the proxy
+# count is too low, so every visitor reads as the proxy and shares one bucket.
+SHARED_ADDRESS_HINT = (
+    "If everyone is being blocked, an admin should check the reverse proxy count "
+    "in Settings → Config: every visitor may be reading as the proxy's address."
+)
+
 _buckets: dict[str, list[float]] = defaultdict(list)
 _lock = asyncio.Lock()
 

src/grimoire/routes/admin.py

@@ -3,8 +3,11 @@ import json
 
 from quart import Blueprint, Response, g, jsonify, request
 
+from grimoire import proxy
 from grimoire.auth import admin_required, login_required, get_current_user_id
+from grimoire.config import Config
 from grimoire.proxy import client_ip
+from grimoire.services import proxy_settings
 from grimoire.services.auth import (
     create_invitation,
     delete_user,
@@ -260,7 +263,6 @@ async def test_smtp():
 @admin_bp.route("/forge-webhook", methods=["GET"])
 @admin_required
 async def get_forge_webhook_settings():
-    from grimoire.config import Config
     from grimoire.routes.webhooks import FORGE_WEBHOOK_SECRET_KEY
 
     webhook_secret = (
@@ -346,6 +348,44 @@ async def update_base_url():
     return jsonify({"status": "ok"})
 
 
+def _proxy_payload() -> dict:
+    return {
+        "hops": proxy.trusted_hops(),
+        "source": proxy_settings.source(),
+        "env_hops": Config.TRUSTED_PROXY_HOPS,
+        "max_hops": proxy.MAX_HOPS,
+        "readout": proxy.request_readout(),
+    }
+
+
+@admin_bp.route("/proxy", methods=["GET"])
+@admin_required
+async def get_proxy_setting():
+    """The hop count in force, where it came from, and what each count would
+    make of the admin's own request."""
+    return jsonify(_proxy_payload())
+
+
+@admin_bp.route("/proxy", methods=["PUT"])
+@admin_required
+async def update_proxy_setting():
+    """Save the hop count, or `{"hops": null}` to go back to the environment's.
+    It applies to the next request."""
+    data = await request.get_json() or {}
+    raw = data.get("hops")
+    hops = None if raw is None else proxy_settings.parse_hops(raw)
+    if raw is not None and hops is None:
+        return jsonify({"error": f"The proxy count must be a whole number from 0 to {proxy.MAX_HOPS}."}), 400
+    before = proxy.trusted_hops()
+    await proxy_settings.save_hops(hops)
+    uid = get_current_user_id()
+    await log_audit(
+        "proxy_hops_config", user_id=uid, username=g.user.username, ip_address=client_ip(),
+        details={"before": before, "after": proxy.trusted_hops(), "saved": hops},
+    )
+    return jsonify(_proxy_payload())
+
+
 @admin_bp.route("/db-maintenance", methods=["GET"])
 @admin_required
 async def get_db_maintenance():

src/grimoire/routes/auth.py

@@ -9,7 +9,7 @@ from quart import Blueprint, g, jsonify, redirect, request, session
 from grimoire.auth import login_required, get_current_user_id
 from grimoire.config import Config
 from grimoire.proxy import client_ip
-from grimoire.rate_limit import is_rate_limited
+from grimoire.rate_limit import SHARED_ADDRESS_HINT, is_rate_limited
 from grimoire.services.auth import (
     authenticate,
     change_password,
@@ -49,7 +49,7 @@ async def register():
     if not Config.LOCAL_AUTH_ENABLED:
         return jsonify({"error": "Local authentication is disabled. Please use SSO."}), 403
     if await is_rate_limited(f"register:{client_ip()}", max_requests=5, window_seconds=300):
-        return jsonify({"error": "Too many registration attempts. Try again later."}), 429
+        return jsonify({"error": f"Too many registration attempts. Try again later. {SHARED_ADDRESS_HINT}"}), 429
     if not await is_registration_open():
         return jsonify({"error": "Registration is closed"}), 403
 
@@ -79,7 +79,7 @@ async def login():
     if not Config.LOCAL_AUTH_ENABLED:
         return jsonify({"error": "Local authentication is disabled. Please use SSO."}), 403
     if await is_rate_limited(f"login:{client_ip()}", max_requests=10, window_seconds=60):
-        return jsonify({"error": "Too many login attempts. Try again later."}), 429
+        return jsonify({"error": f"Too many login attempts. Try again later. {SHARED_ADDRESS_HINT}"}), 429
     data = await request.get_json()
     username = (data.get("username") or "").strip()
     password = data.get("password") or ""
@@ -235,7 +235,7 @@ async def forgot_password():
 @auth_bp.route("/reset-password", methods=["POST"])
 async def reset_password():
     if await is_rate_limited(f"reset:{client_ip()}", max_requests=10, window_seconds=60):
-        return jsonify({"error": "Too many attempts. Try again later."}), 429
+        return jsonify({"error": f"Too many attempts. Try again later. {SHARED_ADDRESS_HINT}"}), 429
     data = await request.get_json()
     token = (data.get("token") or "").strip()
     new_password = data.get("new_password") or ""

src/grimoire/services/proxy_settings.py

@@ -0,0 +1,129 @@
+"""The reverse-proxy hop count an admin saves in Settings → Config.
+
+How many proxies stand in front of the app decides which address every rate
+limit and audit line sees, and whether a request counts as HTTPS
+(grimoire/proxy.py). Behind Cloudflare's proxied DNS the right count is 2; at
+the default of 1 every visitor reads as a Cloudflare edge address and they
+share one sign-in rate limit. So the count is set where an admin can find it
+and check it against their own request, not only in the environment.
+
+Precedence: the saved count, then `TRUSTED_PROXY_HOPS` from the environment,
+then 1. Saving None removes the saved count, so the environment's applies
+again.
+
+It is one instance-wide value, kept on the first admin account like the other
+instance settings. Every worker holds it in proxy.py, loads it at startup and
+re-reads it every REFRESH_SECONDS, so a save in one worker reaches the others
+within that interval; the worker that saved applies it at once.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from sqlalchemy import delete, select
+
+from grimoire import proxy
+from grimoire.config import Config
+from grimoire.models import async_session
+from grimoire.models.setting import Setting
+from grimoire.models.user import User
+
+logger = logging.getLogger(__name__)
+
+SETTING_KEY = "trusted_proxy_hops"
+REFRESH_SECONDS = 30
+# The load at startup runs inside the lifespan hook, and a hook that does not
+# return is a worker that never serves (db_maintenance_scheduler explains the
+# outage that taught this). One indexed row answers in milliseconds, so the
+# deadline only fires when the database is already in trouble.
+STARTUP_READ_TIMEOUT = 3.0
+
+
+def parse_hops(raw) -> int | None:
+    """A hop count from 0 to proxy.MAX_HOPS, or None for anything else. A
+    bool is refused, though Python counts it as an int."""
+    if isinstance(raw, bool):
+        return None
+    if isinstance(raw, str):
+        raw = raw.strip()
+        if not raw.isdigit():
+            return None
+        raw = int(raw)
+    if not isinstance(raw, int) or not 0 <= raw <= proxy.MAX_HOPS:
+        return None
+    return raw
+
+
+def _admin_rows():
+    return (
+        select(Setting)
+        .join(User, Setting.user_id == User.id)
+        .where(User.role == "admin", Setting.key == SETTING_KEY)
+    )
+
+
+async def saved_hops() -> int | None:
+    """The count saved in Settings, or None when there is none (or what is
+    stored can't be read as one)."""
+    async with async_session() as session:
+        row = (await session.execute(_admin_rows().order_by(Setting.user_id))).scalars().first()
+    return parse_hops(row.value) if row else None
+
+
+async def save_hops(hops: int | None) -> None:
+    """Save the count, or remove it with None. Any copy on another admin's
+    account goes too, so exactly one value is ever read back."""
+    async with async_session() as session:
+        admin_ids = select(User.id).where(User.role == "admin")
+        await session.execute(
+            delete(Setting).where(Setting.key == SETTING_KEY, Setting.user_id.in_(admin_ids))
+        )
+        if hops is not None:
+            first_admin = (
+                await session.execute(select(User.id).where(User.role == "admin").order_by(User.id))
+            ).scalars().first()
+            if first_admin is None:
+                raise ValueError("there is no admin account to keep the setting on")
+            session.add(Setting(user_id=first_admin, key=SETTING_KEY, value=str(hops)))
+        await session.commit()
+    proxy.set_saved_hops(hops)
+
+
+async def refresh() -> None:
+    """Load the saved count into this worker. A failed read keeps the value
+    it already had: an unreachable database must not switch the trust rule."""
+    try:
+        proxy.set_saved_hops(await saved_hops())
+    except Exception:
+        logger.debug("could not re-read the saved proxy hop count", exc_info=True)
+
+
+async def load_at_startup() -> None:
+    """`refresh`, bounded for the lifespan hook. On a timeout the environment's
+    count stays in force until the next refresh reaches the database."""
+    try:
+        await asyncio.wait_for(refresh(), timeout=STARTUP_READ_TIMEOUT)
+    except (TimeoutError, asyncio.TimeoutError):
+        logger.warning(
+            "reading the saved proxy hop count exceeded %.1fs; starting with %d "
+            "from the environment until the database answers",
+            STARTUP_READ_TIMEOUT, Config.TRUSTED_PROXY_HOPS,
+        )
+
+
+_refresh_task = None
+
+
+def start_refresh_loop() -> None:
+    global _refresh_task
+    if _refresh_task is None or _refresh_task.done():
+        from grimoire.services.background import start_periodic
+        _refresh_task = start_periodic(REFRESH_SECONDS, refresh, label="proxy_hops_refresh")
+
+
+def source() -> str:
+    """Where the count in force came from: "setting", "env" or "default"."""
+    if proxy.saved_hops_in_force() is not None:
+        return "setting"
+    return "env" if Config.TRUSTED_PROXY_HOPS_FROM_ENV else "default"

tests/test_integration_proxy_settings.py

@@ -0,0 +1,70 @@
+"""The saved proxy hop count against a real database.
+
+What only a real database shows: the count is read back after a save, saving
+leaves exactly one copy however many admins there are, saving None removes it,
+and a saved count reaches `client_ip()` through the worker's cache.
+"""
+import pytest
+from quart import Quart
+from sqlalchemy import select
+
+from grimoire import proxy
+from grimoire.config import Config
+from grimoire.models import async_session
+from grimoire.models.setting import Setting
+from grimoire.proxy import client_ip
+from grimoire.services import proxy_settings
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+
+async def _copies() -> list[Setting]:
+    async with async_session() as s:
+        return list((await s.execute(
+            select(Setting).where(Setting.key == proxy_settings.SETTING_KEY)
+        )).scalars().all())
+
+
+@pytest.fixture
+async def two_admins():
+    async with async_session() as s:
+        first = await ensure_user(s, "proxy_hops_admin_a", role="admin")
+        second = await ensure_user(s, "proxy_hops_admin_b", role="admin")
+        await s.commit()
+        ids = (first.id, second.id)
+    yield ids
+    await proxy_settings.save_hops(None)
+    proxy.set_saved_hops(None)
+
+
+async def test_a_saved_count_reads_back_and_leaves_one_copy(two_admins):
+    # A stale copy on the second admin, as an older write path could leave.
+    async with async_session() as s:
+        s.add(Setting(user_id=two_admins[1], key=proxy_settings.SETTING_KEY, value="4"))
+        await s.commit()
+
+    await proxy_settings.save_hops(2)
+    assert await proxy_settings.saved_hops() == 2
+    copies = await _copies()
+    assert len(copies) == 1 and copies[0].value == "2"
+
+
+async def test_saving_none_removes_the_count(two_admins):
+    await proxy_settings.save_hops(3)
+    await proxy_settings.save_hops(None)
+    assert await proxy_settings.saved_hops() is None
+    assert await _copies() == []
+    assert proxy.saved_hops_in_force() is None
+
+
+async def test_a_saved_count_reaches_the_client_address(two_admins, monkeypatch):
+    monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 1)
+    await proxy_settings.save_hops(2)
+    proxy.set_saved_hops(None)          # as in another worker, before its refresh
+    await proxy_settings.refresh()
+
+    app = Quart(__name__)
+    headers = {"Remote-Addr": "10.0.0.5", "X-Forwarded-For": "203.0.113.7, 172.70.1.1"}
+    async with app.test_request_context("/", headers=headers):
+        assert client_ip() == "203.0.113.7"

tests/test_proxy_settings.py

@@ -0,0 +1,232 @@
+"""The reverse-proxy hop count an admin sets in Settings → Config.
+
+Behind Cloudflare's proxied DNS the right count is 2. At the default of 1,
+every visitor reads as a Cloudflare edge address and shares one sign-in rate
+limit, so a few bad sign-ins block everyone. The count is therefore saved where
+an admin can find it, beside a readout of what each count makes of their own
+request. Covered here:
+- the saved count wins over the environment's;
+- the readout picks out the Cloudflare road;
+- the admin route validates what it saves;
+- a "too many attempts" reply says where to look.
+"""
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from quart import Quart, g
+
+from grimoire import proxy
+from grimoire.config import Config
+from grimoire.proxy import client_ip, readout, request_readout, trusted_hops
+from grimoire.rate_limit import SHARED_ADDRESS_HINT
+from grimoire.services import proxy_settings
+
+VISITOR = "203.0.113.7"
+CLOUDFLARE_EDGE = "172.70.1.1"
+LAN_VISITOR = "192.168.1.20"
+TRAEFIK = "10.0.0.5"
+
+
+@pytest.fixture(autouse=True)
+def _no_saved_count():
+    proxy.set_saved_hops(None)
+    yield
+    proxy.set_saved_hops(None)
+
+
+def _context(headers: dict):
+    app = Quart(__name__)
+    return app.test_request_context("/", headers={"Remote-Addr": TRAEFIK, **headers})
+
+
+@pytest.mark.parametrize("raw, expected", [
+    (0, 0), (2, 2), (5, 5), ("3", 3), (" 1 ", 1),
+    (6, None), (-1, None), ("x", None), ("", None), (True, None), (1.5, None), (None, None),
+])
+def test_a_hop_count_is_a_whole_number_from_zero_to_five(raw, expected):
+    assert proxy_settings.parse_hops(raw) == expected
+
+
+def test_the_saved_count_wins_over_the_environments(monkeypatch):
+    monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 1)
+    assert trusted_hops() == 1
+    proxy.set_saved_hops(2)
+    assert trusted_hops() == 2
+    proxy.set_saved_hops(0)
+    assert trusted_hops() == 0, "a saved 0 is a count, not an absence"
+
+
+async def test_the_client_address_follows_a_saved_count(monkeypatch):
+    monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 1)
+    async with _context({"X-Forwarded-For": f"{VISITOR}, {CLOUDFLARE_EDGE}"}):
+        assert client_ip() == CLOUDFLARE_EDGE
+        proxy.set_saved_hops(2)
+        assert client_ip() == VISITOR
+
+
+@pytest.mark.parametrize("saved, from_env, expected", [
+    (2, False, "setting"), (2, True, "setting"), (None, True, "env"), (None, False, "default"),
+])
+def test_the_source_names_where_the_count_came_from(monkeypatch, saved, from_env, expected):
+    monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS_FROM_ENV", from_env)
+    proxy.set_saved_hops(saved)
+    assert proxy_settings.source() == expected
+
+
+def test_the_readout_suggests_the_count_that_reads_cloudflares_visitor():
+    out = readout(TRAEFIK, f"{VISITOR}, {CLOUDFLARE_EDGE}", "https, https", VISITOR)
+    by_hops = {row["hops"]: row for row in out["rows"]}
+    assert by_hops[0]["client"] == TRAEFIK
+    assert by_hops[1]["client"] == CLOUDFLARE_EDGE
+    assert by_hops[2]["client"] == VISITOR and by_hops[2]["https"] is True
+    assert out["cloudflare"] is True and out["suggested"] == 2
+    assert out["forwarded_for"] == [VISITOR, CLOUDFLARE_EDGE]
+
+
+def test_the_readout_suggests_nothing_off_cloudflare():
+    # Only the admin knows which address is theirs on a road with no
+    # Cloudflare header to compare against.
+    out = readout(TRAEFIK, LAN_VISITOR, "http", None)
+    assert out["cloudflare"] is False and out["suggested"] is None
+    assert [row["client"] for row in out["rows"]] == [TRAEFIK, LAN_VISITOR, LAN_VISITOR, LAN_VISITOR]
+    assert not any(row["https"] for row in out["rows"])
+
+
+def test_the_readout_rows_can_disagree():
+    # The guard above passes vacuously if every row reads the same address.
+    out = readout(TRAEFIK, f"{VISITOR}, {CLOUDFLARE_EDGE}", "", VISITOR)
+    assert len({row["client"] for row in out["rows"]}) == 3
+
+
+async def test_the_request_readout_reads_the_cloudflare_headers():
+    headers = {"X-Forwarded-For": f"{VISITOR}, {CLOUDFLARE_EDGE}", "CF-Connecting-IP": VISITOR}
+    async with _context(headers):
+        out = request_readout()
+    assert out["peer"] == TRAEFIK and out["cloudflare_client"] == VISITOR and out["suggested"] == 2
+
+    async with _context({"CF-Ray": "8a1b2c3d4e5f-LHR"}):
+        out = request_readout()
+    assert out["cloudflare"] is True and out["suggested"] is None
+
+
+async def test_a_failed_refresh_keeps_the_count_in_force():
+    proxy.set_saved_hops(2)
+    with patch.object(proxy_settings, "saved_hops", AsyncMock(side_effect=OSError("db down"))):
+        await proxy_settings.refresh()
+    assert trusted_hops() == 2
+
+
+async def test_a_hanging_startup_read_does_not_hold_up_the_boot(monkeypatch):
+    async def _never():
+        await asyncio.Event().wait()
+
+    monkeypatch.setattr(proxy_settings, "STARTUP_READ_TIMEOUT", 0.05)
+    with patch.object(proxy_settings, "refresh", _never), \
+         patch.object(proxy_settings.logger, "warning") as warn:
+        await asyncio.wait_for(proxy_settings.load_at_startup(), timeout=2)
+    assert warn.called
+
+
+def test_the_startup_read_has_a_deadline():
+    assert 0 < proxy_settings.STARTUP_READ_TIMEOUT < 60
+
+
+# ── the admin route ──────────────────────────────────────────────────────────
+
+
+async def _call(method: str, body=None):
+    from grimoire.app import create_app
+    from grimoire.routes import admin
+
+    app = create_app()
+    view = admin.get_proxy_setting if method == "GET" else admin.update_proxy_setting
+    async with app.test_request_context("/api/admin/proxy", method=method, json=body,
+                                        headers={"Remote-Addr": TRAEFIK}):
+        g.user = SimpleNamespace(id=1, username="admin", role="admin")
+        resp = await view.__wrapped__()
+    if isinstance(resp, tuple):
+        return resp[1], await resp[0].get_json()
+    return resp.status_code, await resp.get_json()
+
+
+async def _fake_save(hops):
+    proxy.set_saved_hops(hops)
+
+
+async def test_the_route_reports_the_count_its_source_and_the_readout(monkeypatch):
+    monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 1)
+    proxy.set_saved_hops(2)
+    status, body = await _call("GET")
+    assert status == 200
+    assert body["hops"] == 2 and body["source"] == "setting" and body["env_hops"] == 1
+    assert body["max_hops"] == proxy.MAX_HOPS
+    assert body["readout"]["peer"] == TRAEFIK and len(body["readout"]["rows"]) == 4
+
+
+async def test_saving_a_count_applies_it_and_audits_the_change(monkeypatch):
+    from grimoire.routes import admin
+
+    monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 1)
+    save = AsyncMock(side_effect=_fake_save)
+    audit = AsyncMock()
+    with patch.object(admin.proxy_settings, "save_hops", save), patch.object(admin, "log_audit", audit):
+        status, body = await _call("PUT", {"hops": 2})
+    assert status == 200 and body["hops"] == 2
+    save.assert_awaited_once_with(2)
+    details = audit.call_args.kwargs["details"]
+    assert details == {"before": 1, "after": 2, "saved": 2}
+
+
+async def test_saving_null_goes_back_to_the_environments_count(monkeypatch):
+    from grimoire.routes import admin
+
+    monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 1)
+    proxy.set_saved_hops(3)
+    save = AsyncMock(side_effect=_fake_save)
+    with patch.object(admin.proxy_settings, "save_hops", save), patch.object(admin, "log_audit", AsyncMock()):
+        status, body = await _call("PUT", {"hops": None})
+    assert status == 200 and body["hops"] == 1
+    save.assert_awaited_once_with(None)
+
+
+@pytest.mark.parametrize("bad", [6, -1, "two", True, 1.5])
+async def test_a_count_out_of_range_is_refused_and_nothing_is_saved(bad):
+    from grimoire.routes import admin
+
+    save = AsyncMock()
+    with patch.object(admin.proxy_settings, "save_hops", save):
+        status, body = await _call("PUT", {"hops": bad})
+    assert status == 400 and "0 to 5" in body["error"]
+    save.assert_not_awaited()
+
+
+@pytest.mark.parametrize("method", ["GET", "PUT"])
+async def test_the_route_is_for_admins_only(method):
+    from grimoire.app import create_app
+    from grimoire.routes import admin
+
+    view = admin.get_proxy_setting if method == "GET" else admin.update_proxy_setting
+    app = create_app()
+    async with app.test_request_context("/api/admin/proxy", method=method, json={"hops": 2}):
+        resp = await view()
+    assert resp[1] == 401
+
+
+# ── where a lockout shows up ─────────────────────────────────────────────────
+
+
+async def test_a_rate_limited_sign_in_says_where_an_admin_should_look(monkeypatch):
+    from grimoire.app import create_app
+    from grimoire.routes import auth
+
+    monkeypatch.setattr(Config, "LOCAL_AUTH_ENABLED", True)
+    app = create_app()
+    with patch.object(auth, "is_rate_limited", AsyncMock(return_value=True)):
+        async with app.test_request_context("/api/auth/login", method="POST", json={}):
+            resp, status = await auth.login()
+    body = await resp.get_json()
+    assert status == 429
+    assert SHARED_ADDRESS_HINT in body["error"]
+    assert "Settings → Config" in SHARED_ADDRESS_HINT