bvandeusenandClaude Opus 5.5
753a6ccb31
CI & Build / Plugin hooks (push) Successful in 12s
CI & Build / Python lint (push) Successful in 2s
CI & Build / TypeScript typecheck (push) Successful in 32s
CI & Build / Python tests (push) Successful in 1m47s
CI & Build / integration (push) Successful in 2m1s
CI & Build / Build & push image (push) Successful in 20s
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>
71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
"""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"
|