FluxIP/app/fleet/proxy.py

367 lines
12 KiB
Python

from __future__ import annotations
import asyncio
import base64
import ipaddress
import re
import socket
from contextlib import suppress
from dataclasses import dataclass
from urllib.parse import unquote_to_bytes, urlsplit
_SUPPORTED_PROTOCOLS = {"socks5", "http"}
_HEX_ESCAPE_RE = re.compile(r"%(?![0-9a-fA-F]{2})")
_HTTP_HEADER_LIMIT = 16 * 1024
_PROBE_TARGET_HOST = "1.1.1.1"
_PROBE_TARGET_PORT = 443
@dataclass(frozen=True, slots=True)
class ProxyEndpoint:
protocol: str
host: str
port: int
username: str | None
password: str | None
@dataclass(frozen=True, slots=True)
class _AuthorityCandidate:
host: str
port: int
username: str | None
password: str | None
reversed_format: bool
def parse_proxy_url(value: str) -> ProxyEndpoint:
"""Parse supported proxy URLs without retaining or exposing their raw value."""
if not isinstance(value, str) or not value.strip():
raise ValueError("代理链接不能为空")
if _contains_control(value):
raise ValueError("代理链接包含非法控制字符")
normalized = value.strip()
try:
parsed = urlsplit(normalized)
except ValueError as exc:
raise ValueError("代理链接格式无效") from exc
protocol = parsed.scheme.lower()
if protocol not in _SUPPORTED_PROTOCOLS:
raise ValueError("代理协议仅支持 socks5 或 http")
if not parsed.netloc:
raise ValueError("代理链接缺少主机和端口")
if parsed.path or parsed.query or parsed.fragment:
raise ValueError("代理链接不能包含路径、查询参数或片段")
if normalized.count("://") != 1 or parsed.netloc.count("@") > 1:
raise ValueError("代理链接格式无效")
candidates = _authority_candidates(parsed.netloc, protocol)
if not candidates:
raise ValueError("代理链接格式无效")
selected = _select_candidate(candidates)
return ProxyEndpoint(
protocol=protocol,
host=selected.host,
port=selected.port,
username=selected.username,
password=selected.password,
)
async def probe_proxy(
host: str,
port: int,
protocol: str,
username: str | None = None,
password: str | None = None,
timeout_seconds: float = 5,
) -> None:
"""Verify proxy authentication and an outbound CONNECT without logging credentials."""
normalized_protocol = str(protocol).strip().lower()
if normalized_protocol not in _SUPPORTED_PROTOCOLS:
raise ValueError("代理协议仅支持 socks5 或 http")
normalized_host = str(host).strip()
if (
not normalized_host
or _contains_control(normalized_host)
or any(character.isspace() for character in normalized_host)
):
raise ValueError("代理主机无效")
if isinstance(port, bool) or not isinstance(port, int) or not 1 <= port <= 65535:
raise ValueError("代理端口必须在 1 到 65535 之间")
if timeout_seconds <= 0:
raise ValueError("代理探测超时必须大于 0")
_validate_credentials(username, password, normalized_protocol)
writer: asyncio.StreamWriter | None = None
try:
async with asyncio.timeout(timeout_seconds):
reader, writer = await asyncio.open_connection(normalized_host, port)
if normalized_protocol == "socks5":
await _probe_socks5(reader, writer, username, password)
else:
await _probe_http(reader, writer, username, password)
except TimeoutError:
raise OSError("代理探测超时") from None
except asyncio.IncompleteReadError:
raise OSError("代理服务器返回了不完整响应") from None
except asyncio.LimitOverrunError:
raise OSError("代理服务器返回的响应过大") from None
finally:
if writer is not None:
writer.close()
with suppress(OSError):
await writer.wait_closed()
def _authority_candidates(authority: str, protocol: str) -> list[_AuthorityCandidate]:
if "@" not in authority:
endpoint = _parse_endpoint(authority)
return (
[
_AuthorityCandidate(
host=endpoint[0],
port=endpoint[1],
username=None,
password=None,
reversed_format=False,
)
]
if endpoint
else []
)
left, right = authority.split("@", 1)
candidates: list[_AuthorityCandidate] = []
for endpoint_text, credentials_text, reversed_format in (
(right, left, False),
(left, right, True),
):
endpoint = _parse_endpoint(endpoint_text)
credentials = _parse_credentials(credentials_text, protocol)
if endpoint is None or credentials is None:
continue
candidates.append(
_AuthorityCandidate(
host=endpoint[0],
port=endpoint[1],
username=credentials[0],
password=credentials[1],
reversed_format=reversed_format,
)
)
return candidates
def _select_candidate(candidates: list[_AuthorityCandidate]) -> _AuthorityCandidate:
if len(candidates) == 1:
return candidates[0]
standard = next(item for item in candidates if not item.reversed_format)
reversed_candidate = next(item for item in candidates if item.reversed_format)
standard_score = _host_likelihood(standard.host)
reversed_score = _host_likelihood(reversed_candidate.host)
if reversed_score > standard_score:
return reversed_candidate
return standard
def _parse_endpoint(value: str) -> tuple[str, int] | None:
if not value or _contains_control(value) or any(character.isspace() for character in value):
return None
if "%" in value or any(character in value for character in "/?#"):
return None
try:
parsed = urlsplit(f"//{value}")
host = parsed.hostname
port = parsed.port
except ValueError:
return None
if (
not host
or port is None
or not 1 <= port <= 65535
or parsed.username is not None
or parsed.password is not None
or parsed.path
or parsed.query
or parsed.fragment
):
return None
normalized_host = host.rstrip(".").lower()
if not normalized_host or any(character.isspace() for character in normalized_host):
return None
try:
normalized_host = str(ipaddress.ip_address(normalized_host))
except ValueError:
try:
normalized_host = normalized_host.encode("idna").decode("ascii")
except UnicodeError:
return None
labels = normalized_host.split(".")
if any(
not label
or len(label) > 63
or label.startswith("-")
or label.endswith("-")
or not all(character.isalnum() or character == "-" for character in label)
for label in labels
):
return None
return normalized_host, port
def _parse_credentials(value: str, protocol: str) -> tuple[str, str] | None:
raw_username, separator, raw_password = value.partition(":")
if not separator or not raw_username or not raw_password:
return None
try:
username = _decode_component(raw_username)
password = _decode_component(raw_password)
_validate_credentials(username, password, protocol)
except ValueError:
return None
return username, password
def _decode_component(value: str) -> str:
if _HEX_ESCAPE_RE.search(value):
raise ValueError("代理认证信息包含无效编码")
try:
decoded = unquote_to_bytes(value).decode("utf-8")
except UnicodeDecodeError as exc:
raise ValueError("代理认证信息编码无效") from exc
if not decoded or _contains_control(decoded):
raise ValueError("代理认证信息格式无效")
return decoded
def _validate_credentials(
username: str | None,
password: str | None,
protocol: str,
) -> None:
if (username is None) != (password is None):
raise ValueError("代理用户名和密码必须同时提供")
if username is None or password is None:
return
if not username or not password or _contains_control(username) or _contains_control(password):
raise ValueError("代理认证信息格式无效")
if protocol == "socks5" and (
len(username.encode("utf-8")) > 255 or len(password.encode("utf-8")) > 255
):
raise ValueError("SOCKS5 用户名和密码不能超过 255 字节")
def _host_likelihood(host: str) -> int:
try:
ipaddress.ip_address(host)
return 3
except ValueError:
if "." in host or host == "localhost":
return 2
return 1
async def _probe_socks5(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
username: str | None,
password: str | None,
) -> None:
method = 2 if username is not None else 0
writer.write(bytes((5, 1, method)))
await writer.drain()
version, selected_method = await reader.readexactly(2)
if version != 5:
raise OSError("目标端口不是 SOCKS5 服务")
if selected_method == 255:
raise OSError("SOCKS5 服务器拒绝了认证方式")
if selected_method != method:
raise OSError("SOCKS5 服务器未接受指定认证方式")
if method == 2:
username_bytes = str(username).encode("utf-8")
password_bytes = str(password).encode("utf-8")
writer.write(
bytes((1, len(username_bytes)))
+ username_bytes
+ bytes((len(password_bytes),))
+ password_bytes
)
await writer.drain()
auth_version, auth_status = await reader.readexactly(2)
if auth_version != 1 or auth_status != 0:
raise OSError("SOCKS5 用户名或密码认证失败")
writer.write(
b"\x05\x01\x00\x01"
+ socket.inet_aton(_PROBE_TARGET_HOST)
+ _PROBE_TARGET_PORT.to_bytes(2, "big")
)
await writer.drain()
version, reply, reserved, address_type = await reader.readexactly(4)
if version != 5 or reserved != 0:
raise OSError("SOCKS5 服务器返回了无效响应")
if reply != 0:
raise OSError("SOCKS5 服务器无法建立外部连接")
if address_type == 1:
await reader.readexactly(4)
elif address_type == 3:
address_length = (await reader.readexactly(1))[0]
await reader.readexactly(address_length)
elif address_type == 4:
await reader.readexactly(16)
else:
raise OSError("SOCKS5 服务器返回了无效地址类型")
await reader.readexactly(2)
async def _probe_http(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
username: str | None,
password: str | None,
) -> None:
headers = [
f"CONNECT {_PROBE_TARGET_HOST}:{_PROBE_TARGET_PORT} HTTP/1.1",
f"Host: {_PROBE_TARGET_HOST}:{_PROBE_TARGET_PORT}",
"Proxy-Connection: close",
]
if username is not None and password is not None:
encoded = base64.b64encode(f"{username}:{password}".encode()).decode("ascii")
headers.append(f"Proxy-Authorization: Basic {encoded}")
writer.write(("\r\n".join(headers) + "\r\n\r\n").encode("ascii"))
await writer.drain()
response_headers = await _read_http_headers(reader)
status_line = response_headers.split(b"\r\n", 1)[0]
try:
version, raw_status, _reason = status_line.decode("ascii").split(" ", 2)
status = int(raw_status)
except (UnicodeDecodeError, ValueError) as exc:
raise OSError("HTTP 代理返回了无效响应") from exc
if version not in {"HTTP/1.0", "HTTP/1.1"}:
raise OSError("HTTP 代理返回了无效响应")
if status == 407:
raise OSError("HTTP 代理用户名或密码认证失败")
if not 200 <= status < 300:
raise OSError("HTTP 代理无法建立外部连接")
async def _read_http_headers(reader: asyncio.StreamReader) -> bytes:
response = bytearray()
while b"\r\n\r\n" not in response:
chunk = await reader.read(1024)
if not chunk:
raise OSError("HTTP 代理返回了不完整响应")
response.extend(chunk)
if len(response) > _HTTP_HEADER_LIMIT:
raise OSError("HTTP 代理返回的响应过大")
return bytes(response)
def _contains_control(value: str) -> bool:
return any(ord(character) < 32 or ord(character) == 127 for character in value)