mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
R-b12 的尾巴:check_pgbouncer 判同一个主机端口时把主机名规范化——localhost、127.x.x.x、::1 都算本机回环,IP 按地址比(::ffff:10.0.0.5 和 10.0.0.5 相同),解析不成 IP 的按字面(小写)比、不查 DNS。补测试:localhost 对 127.0.0.1、127.0.0.1 对 ::1、IPv4 映射地址同端口都拒绝,本机另一个端口、名字不同的放行
This commit is contained in:
+16
-2
@@ -318,9 +318,23 @@ def check_pgbouncer() -> None:
|
||||
|
||||
|
||||
def _endpoint(url: str) -> tuple[str, int]:
|
||||
"""地址连的是哪个主机端口(端口不写按 5432)。"""
|
||||
"""地址连的是哪个主机端口(端口不写按 5432)。主机名规范化:localhost、127.x.x.x、::1 都算本机回环;IP 地址按地址比
|
||||
(::ffff:10.0.0.5 和 10.0.0.5 相同);解析不成 IP 的按字面(小写)比,不查 DNS。"""
|
||||
import ipaddress
|
||||
|
||||
parsed = make_url(url)
|
||||
return (parsed.host or "").lower(), int(parsed.port or 5432)
|
||||
host = (parsed.host or "").strip().strip("[]").lower().rstrip(".")
|
||||
try:
|
||||
addr = ipaddress.ip_address(host)
|
||||
except ValueError:
|
||||
addr = None
|
||||
if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped is not None:
|
||||
addr = addr.ipv4_mapped
|
||||
if host == "localhost" or host.endswith(".localhost") or (addr is not None and addr.is_loopback):
|
||||
host = "loopback"
|
||||
elif addr is not None:
|
||||
host = str(addr)
|
||||
return host, int(parsed.port or 5432)
|
||||
|
||||
|
||||
async def get_session() -> AsyncIterator[AsyncSession]:
|
||||
|
||||
@@ -109,3 +109,25 @@ def test_copying_checkpoints_goes_through_the_direct_url(monkeypatch, tmp_path):
|
||||
assert seen["direct_url"] == settings.database_direct_url, "目标就是应用库:断点走它的直连地址"
|
||||
assert to_postgres.main([str(tmp_path), "postgresql://u:p@elsewhere.internal:5432/x"]) == 2
|
||||
assert seen["direct_url"] is None, "目标是别的库:不套用应用库的直连地址"
|
||||
|
||||
|
||||
def test_loopback_spellings_count_as_the_same_endpoint(monkeypatch):
|
||||
"""R-b12 的尾巴:localhost、127.0.0.1、::1 是同一台机器——同一个端口时照样拒绝;IP 按地址比,解析不成 IP 的按字面比。"""
|
||||
from app.db import base
|
||||
|
||||
def norm(url: str) -> str:
|
||||
return Settings(database_url=url, _env_file=None).database_url
|
||||
|
||||
monkeypatch.setattr(settings, "db_pgbouncer", True)
|
||||
for app_url, direct in (("postgresql://u:p@localhost:6432/db", "postgresql://u:p@127.0.0.1:6432/db"),
|
||||
("postgresql://u:p@127.0.0.1:6432/db", "postgresql://u:p@[::1]:6432/db"),
|
||||
("postgresql://u:p@10.0.0.5:6432/db", "postgresql://u:p@[::ffff:10.0.0.5]:6432/db")):
|
||||
monkeypatch.setattr(settings, "database_url", norm(app_url))
|
||||
monkeypatch.setattr(settings, "database_direct_url", norm(direct))
|
||||
with pytest.raises(RuntimeError, match="指着同一个地址"):
|
||||
base.check_pgbouncer()
|
||||
monkeypatch.setattr(settings, "database_url", norm("postgresql://u:p@localhost:6432/db"))
|
||||
monkeypatch.setattr(settings, "database_direct_url", norm("postgresql://u:p@127.0.0.1:5432/db"))
|
||||
base.check_pgbouncer() # 本机另一个端口:放行
|
||||
monkeypatch.setattr(settings, "database_direct_url", norm("postgresql://u:p@db.internal:6432/db"))
|
||||
base.check_pgbouncer() # 名字不同、不查 DNS:按字面算不同
|
||||
|
||||
Reference in New Issue
Block a user