woggioni-opencode-agent
bba22b357e
- bump kaya-core/kaya-rsgi to >= 0.0.3 and add kaya-forwarded: forwarded header handling is no longer built into core, it is opt-in via ForwardedHeadersMixin and gated on trusted proxy CIDRs - add TRUSTED_PROXY_CIDRS setting (comma-separated CIDRs, validated at startup; empty means no proxy is trusted) and wire the mixin in app.py - cover trusted/untrusted peers, all-trusted chains and the RFC 7239 Forwarded header with port in the test suite - document the new variable in README, .env.example and docker-compose.yml
216 lines
8.9 KiB
Python
216 lines
8.9 KiB
Python
"""Route tests using kaya's ASGI transport via httpx."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
|
|
from httpx import ASGITransport, AsyncClient
|
|
from pwo import async_test
|
|
|
|
from pyfconfig.app import app
|
|
|
|
# httpx's ASGITransport populates the scope with this client tuple.
|
|
CLIENT_IP = "127.0.0.1"
|
|
CLIENT_PORT = "123"
|
|
# First entry of the X-Forwarded-For header in ALL_HEADERS. With
|
|
# kaya-forwarded (see tests/__init__.py for the trusted CIDRs), the chain is
|
|
# walked right-to-left skipping trusted proxies: 10.0.0.1 is trusted, so
|
|
# ip_addr resolves to this instead of the socket peer address.
|
|
FORWARDED_IP = "203.0.113.7"
|
|
|
|
ALL_HEADERS = {
|
|
"User-Agent": "test-agent/1.0",
|
|
"Accept-Language": "en-US,en;q=0.9",
|
|
"Referer": "https://example.com/page",
|
|
"Connection": "keep-alive",
|
|
"Keep-Alive": "timeout=5",
|
|
"Accept-Encoding": "gzip, deflate",
|
|
"Accept": "text/html,application/xhtml+xml",
|
|
"Accept-Charset": "utf-8",
|
|
"Via": "1.1 proxy",
|
|
"X-Forwarded-For": "203.0.113.7, 10.0.0.1",
|
|
}
|
|
|
|
|
|
class RoutesTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.transport = ASGITransport(app=app)
|
|
|
|
def client(self) -> AsyncClient:
|
|
return AsyncClient(transport=self.transport, base_url="http://127.0.0.1")
|
|
|
|
@async_test
|
|
async def test_health(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/api/health")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertIn("ok", r.text)
|
|
|
|
@async_test
|
|
async def test_ip(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/ip")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual(CLIENT_IP, r.text.strip())
|
|
|
|
@async_test
|
|
async def test_ip_honors_x_forwarded_for(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/ip", headers={"X-Forwarded-For": "203.0.113.7"})
|
|
self.assertEqual("203.0.113.7", r.text.strip())
|
|
|
|
@async_test
|
|
async def test_ip_ignores_x_forwarded_for_from_untrusted_peer(self) -> None:
|
|
# The socket peer is not in TRUSTED_PROXY_CIDRS, so proxy headers
|
|
# are ignored and the peer address itself is reported.
|
|
transport = ASGITransport(app=app, client=("192.0.2.10", 5555))
|
|
async with AsyncClient(transport=transport, base_url="http://192.0.2.10") as client:
|
|
r = await client.get("/ip", headers={"X-Forwarded-For": "203.0.113.7"})
|
|
self.assertEqual("192.0.2.10", r.text.strip())
|
|
|
|
@async_test
|
|
async def test_ip_all_trusted_chain_uses_leftmost_entry(self) -> None:
|
|
# 10.1.2.3 is inside the trusted 10.0.0.0/8, so the whole chain is
|
|
# trusted and the leftmost entry is the original client.
|
|
async with self.client() as client:
|
|
r = await client.get("/ip", headers={"X-Forwarded-For": "10.1.2.3"})
|
|
self.assertEqual("10.1.2.3", r.text.strip())
|
|
|
|
@async_test
|
|
async def test_forwarded_header_with_port(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/all.json", headers={"Forwarded": "for=203.0.113.7:4455"})
|
|
data = json.loads(r.text)
|
|
self.assertEqual("203.0.113.7", data["ip_addr"])
|
|
self.assertEqual("4455", data["port"])
|
|
|
|
@async_test
|
|
async def test_ua(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/ua", headers={"User-Agent": "test-agent/1.0"})
|
|
self.assertEqual("test-agent/1.0", r.text.strip())
|
|
|
|
@async_test
|
|
async def test_lang(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/lang", headers={"Accept-Language": "en-US"})
|
|
self.assertEqual("en-US", r.text.strip())
|
|
|
|
@async_test
|
|
async def test_encoding(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/encoding", headers={"Accept-Encoding": "gzip"})
|
|
self.assertEqual("gzip", r.text.strip())
|
|
|
|
@async_test
|
|
async def test_mime(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/mime", headers={"Accept": "application/json"})
|
|
self.assertEqual("application/json", r.text.strip())
|
|
|
|
@async_test
|
|
async def test_charset(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/charset", headers={"Accept-Charset": "utf-8"})
|
|
self.assertEqual("utf-8", r.text.strip())
|
|
|
|
@async_test
|
|
async def test_forwarded(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/forwarded", headers={"X-Forwarded-For": "203.0.113.7"})
|
|
self.assertEqual("203.0.113.7", r.text.strip())
|
|
|
|
@async_test
|
|
async def test_forwarded_absent_is_empty(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/forwarded")
|
|
self.assertEqual("", r.text.strip())
|
|
|
|
@async_test
|
|
async def test_all_text(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/all", headers=ALL_HEADERS)
|
|
self.assertEqual(200, r.status_code)
|
|
lines = r.text.strip().splitlines()
|
|
keys = [line.split(":", 1)[0] for line in lines]
|
|
self.assertEqual(
|
|
[
|
|
"ip_addr", "remote_host", "user_agent", "port", "language",
|
|
"referer", "connection", "keep_alive", "method", "encoding",
|
|
"mime", "charset", "via", "forwarded",
|
|
],
|
|
keys,
|
|
)
|
|
values = dict(line.split(": ", 1) for line in lines)
|
|
self.assertEqual(FORWARDED_IP, values["ip_addr"])
|
|
self.assertEqual("unavailable", values["remote_host"])
|
|
self.assertEqual("test-agent/1.0", values["user_agent"])
|
|
self.assertEqual(CLIENT_PORT, values["port"])
|
|
self.assertEqual("en-US,en;q=0.9", values["language"])
|
|
self.assertEqual("https://example.com/page", values["referer"])
|
|
self.assertEqual("keep-alive", values["connection"])
|
|
self.assertEqual("timeout=5", values["keep_alive"])
|
|
self.assertEqual("GET", values["method"])
|
|
self.assertEqual("gzip, deflate", values["encoding"])
|
|
self.assertEqual("text/html,application/xhtml+xml", values["mime"])
|
|
self.assertEqual("utf-8", values["charset"])
|
|
self.assertEqual("1.1 proxy", values["via"])
|
|
self.assertEqual("203.0.113.7, 10.0.0.1", values["forwarded"])
|
|
|
|
@async_test
|
|
async def test_all_json_omits_empty_fields(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/all.json", headers=ALL_HEADERS)
|
|
self.assertEqual(200, r.status_code)
|
|
data = json.loads(r.text)
|
|
self.assertEqual(FORWARDED_IP, data["ip_addr"])
|
|
self.assertEqual("test-agent/1.0", data["user_agent"])
|
|
self.assertEqual(CLIENT_PORT, data["port"])
|
|
self.assertEqual("GET", data["method"])
|
|
# Empty fields are omitted entirely.
|
|
data_minimal = json.loads(
|
|
(await client.get("/all.json")).text
|
|
)
|
|
self.assertNotIn("remote_host", data_minimal)
|
|
self.assertNotIn("language", data_minimal)
|
|
self.assertNotIn("forwarded", data_minimal)
|
|
self.assertEqual(CLIENT_IP, data_minimal["ip_addr"])
|
|
|
|
@async_test
|
|
async def test_root_plain_text_for_cli(self) -> None:
|
|
async with self.client() as client:
|
|
# httpx sends Accept: */* by default, like curl.
|
|
r = await client.get("/")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertIn("text/plain", r.headers["content-type"])
|
|
self.assertEqual(CLIENT_IP, r.text.strip())
|
|
|
|
@async_test
|
|
async def test_root_html_for_browsers(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/", headers={"Accept": "text/html"})
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertIn("text/html", r.headers["content-type"])
|
|
self.assertIn("What Is My IP Address?", r.text)
|
|
self.assertIn(CLIENT_IP, r.text)
|
|
self.assertIn("curl pyfconfig/all.json", r.text)
|
|
|
|
@async_test
|
|
async def test_root_html_escapes_user_input(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get(
|
|
"/",
|
|
headers={
|
|
"Accept": "text/html",
|
|
"User-Agent": "<script>alert(1)</script>",
|
|
},
|
|
)
|
|
self.assertNotIn("<script>alert(1)</script>", r.text)
|
|
self.assertIn("<script>", r.text)
|
|
|
|
@async_test
|
|
async def test_unknown_path_is_404(self) -> None:
|
|
async with self.client() as client:
|
|
r = await client.get("/nope")
|
|
self.assertEqual(404, r.status_code)
|