From 2e401124e86ec13b7e40fd3699be1562439d57c9 Mon Sep 17 00:00:00 2001 From: StasonJatham Date: Thu, 21 May 2026 17:34:48 +0200 Subject: [PATCH] Honor trusted proxy client IP headers --- karldns/security.py | 8 ++++--- tests/test_security.py | 47 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 51 insertions(+), 4 deletions(-) diff --git a/karldns/security.py b/karldns/security.py index d0a333d..314fedb 100644 --- a/karldns/security.py +++ b/karldns/security.py @@ -146,9 +146,11 @@ def require_admin_csrf(settings: DdnsSettings, supplied: str) -> None: def client_ip(request: Request, settings: DdnsSettings | None = None) -> str: peer = request.client.host if request.client else "unknown" - forwarded = request.headers.get("x-forwarded-for", "") - if forwarded and settings and _trusted_proxy(peer, settings.trusted_proxy_ips): - return forwarded.split(",", 1)[0].strip() + if settings and _trusted_proxy(peer, settings.trusted_proxy_ips): + for header_name in ("cf-connecting-ip", "x-forwarded-for", "x-real-ip"): + candidate = request.headers.get(header_name, "") + if candidate: + return candidate.split(",", 1)[0].strip() return peer diff --git a/tests/test_security.py b/tests/test_security.py index b46ae35..fa62c10 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -4,12 +4,13 @@ import logging from pathlib import Path import pytest +from fastapi import Request from fastapi.testclient import TestClient import karldns.services.ddns as ddns_module from karldns import DdnsSettings, DdnsStore, DnsProviderError, DnsReconciliationError, UpdateResult, make_app from karldns.dns import validate_dns_provider_settings -from karldns.security import DASHBOARD_UNLOCK_COOKIE +from karldns.security import DASHBOARD_UNLOCK_COOKIE, client_ip def test_generated_hostname_is_not_an_update_secret(tmp_path: Path) -> None: @@ -242,6 +243,50 @@ def test_untrusted_forwarded_for_does_not_bypass_rate_limit(tmp_path: Path) -> N assert second.status_code == 429 +def test_trusted_proxy_uses_cloudflare_connecting_ip(tmp_path: Path) -> None: + settings = DdnsSettings( + database_path=tmp_path / "ddns.sqlite3", + admin_password="admin", + trusted_proxy_ips={"10.0.0.0/8"}, + ) + app = make_app(settings) + + @app.get("/debug-ip", include_in_schema=False) + def debug_ip(request: Request) -> dict[str, str]: + return {"ip": client_ip(request, settings)} + + client = TestClient(app, client=("10.1.2.3", 12345)) + response = client.get( + "/debug-ip", + headers={ + "cf-connecting-ip": "198.51.100.20", + "x-forwarded-for": "203.0.113.5", + }, + ) + + assert response.status_code == 200 + assert response.json()["ip"] == "198.51.100.20" + + +def test_untrusted_proxy_headers_are_ignored(tmp_path: Path) -> None: + settings = DdnsSettings( + database_path=tmp_path / "ddns.sqlite3", + admin_password="admin", + trusted_proxy_ips={"10.0.0.0/8"}, + ) + app = make_app(settings) + + @app.get("/debug-ip", include_in_schema=False) + def debug_ip(request: Request) -> dict[str, str]: + return {"ip": client_ip(request, settings)} + + client = TestClient(app, client=("198.51.100.1", 12345)) + response = client.get("/debug-ip", headers={"cf-connecting-ip": "198.51.100.20"}) + + assert response.status_code == 200 + assert response.json()["ip"] == "198.51.100.1" + + def test_dns_delete_failure_is_logged_without_secret_management_slug(tmp_path: Path, monkeypatch, caplog) -> None: def fail_delete(hostname, settings): raise DnsProviderError("dns down")