From 2c4a1a4d4239d9e11e2de8fe355337885e72befa Mon Sep 17 00:00:00 2001 From: thinktraveller <161934609+thinktraveller@users.noreply.github.com> Date: Sun, 12 Jul 2026 11:35:48 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E6=9B=B4=E6=96=B0=E4=BA=86=E4=B8=80?= =?UTF-8?q?=E4=B8=AA=E4=BF=AE=E5=A4=8D=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- compose.yaml | 1 + patches/knowhere/subscription_service.py | 216 +++++++++++++++++++++++ 2 files changed, 217 insertions(+) create mode 100644 patches/knowhere/subscription_service.py diff --git a/compose.yaml b/compose.yaml index cab2751..e0616f3 100644 --- a/compose.yaml +++ b/compose.yaml @@ -20,6 +20,7 @@ services: - knowhere_user_data:/data/users - knowhere_model_cache:/data/models - knowhere_secrets:/data/secrets + - ./patches/knowhere/subscription_service.py:/opt/knowhere/source/api/apps/api/app/services/s3_events/subscription_service.py:ro postgres: image: postgres:15-alpine diff --git a/patches/knowhere/subscription_service.py b/patches/knowhere/subscription_service.py new file mode 100644 index 0000000..1b3df27 --- /dev/null +++ b/patches/knowhere/subscription_service.py @@ -0,0 +1,216 @@ +"""SNS subscription confirmation handling.""" +from __future__ import annotations + +import asyncio +import os +import socket +from collections.abc import Sequence + +from loguru import logger +from urllib.parse import ParseResult, parse_qs, urlparse, urlunparse + +from shared.core.config import settings +from shared.services.http.pinned_outbound import send_pinned_outbound_request +from shared.services.http.url_security import ( + HTTPURLValidationResult, + validate_http_url_and_resolve_ip_async, +) + + +SNS_SUBSCRIPTION_TIMEOUT_SECONDS = 10 +LOCALSTACK_HOSTNAMES = frozenset( + { + "localstack", + "localhost.localstack.cloud", + } +) + + +async def confirm_sns_subscription(subscribe_url: str) -> dict[str, str]: + rewritten_url = _rewrite_localstack_subscribe_url(subscribe_url) + validation = await _validate_sns_confirmation_url(rewritten_url) + if not validation.is_valid: + logger.warning( + f"SNS subscription confirmation URL failed validation: {validation.error_message}" + ) + return {"message": "SNS subscription confirmation failed"} + + if not validation.validated_ip: + logger.warning("SNS subscription confirmation URL validation returned no IP") + return {"message": "SNS subscription confirmation failed"} + + try: + response = await send_pinned_outbound_request( + method="GET", + url=rewritten_url, + pinned_ip=validation.validated_ip, + timeout_seconds=SNS_SUBSCRIPTION_TIMEOUT_SECONDS, + ) + if response.status == 200: + logger.info("SNS subscription confirmed successfully") + return {"message": "SNS subscription confirmed"} + + if 300 <= response.status < 400: + logger.warning( + f"SNS subscription confirmation redirect blocked, status={response.status}" + ) + else: + logger.error( + f"SNS subscription confirmation failed, status={response.status}" + ) + return {"message": "SNS subscription confirmation failed"} + except Exception as exc: + logger.error(f"Failed to reach the SNS confirmation URL: {exc}") + return {"message": "SNS subscription confirmation failed"} + + +async def _validate_sns_confirmation_url(url: str) -> HTTPURLValidationResult: + """Validate the SNS confirmation URL, allowing configured LocalStack endpoints.""" + if _is_configured_localstack_confirmation_url(url): + return await _resolve_configured_localstack_confirmation_url(url) + + return await validate_http_url_and_resolve_ip_async(url) + + +async def _resolve_configured_localstack_confirmation_url( + url: str, +) -> HTTPURLValidationResult: + parsed_url = urlparse(url) + hostname = parsed_url.hostname + if not hostname: + return HTTPURLValidationResult( + is_valid=False, + url=url, + error_message="URL must include a hostname", + failure_reason="missing_hostname", + ) + + try: + loop = asyncio.get_running_loop() + address_infos = await loop.getaddrinfo(hostname, None) + except socket.gaierror as exc: + return HTTPURLValidationResult( + is_valid=False, + url=url, + hostname=hostname, + error_message=f"Unable to resolve hostname {hostname}: {exc}", + failure_reason="hostname_resolution_failed", + ) + + validated_ip = _extract_first_resolved_ip(address_infos) + if not validated_ip: + return HTTPURLValidationResult( + is_valid=False, + url=url, + hostname=hostname, + error_message=f"Unable to resolve hostname {hostname}", + failure_reason="hostname_resolution_failed", + ) + + return HTTPURLValidationResult( + is_valid=True, + url=url, + hostname=hostname, + validated_ip=validated_ip, + ) + + +def _rewrite_localstack_subscribe_url(subscribe_url: str) -> str: + parsed_subscribe_url = urlparse(subscribe_url) + storage_endpoint = _parse_configured_storage_endpoint() + if not storage_endpoint: + return subscribe_url + + if not _is_confirm_subscription_action(parsed_subscribe_url): + return subscribe_url + + if not _is_localstack_endpoint(parsed_subscribe_url): + return subscribe_url + + if not _is_localstack_endpoint(storage_endpoint): + return subscribe_url + + return urlunparse( + parsed_subscribe_url._replace( + scheme=storage_endpoint.scheme, + netloc=storage_endpoint.netloc, + ) + ) + + +def _is_configured_localstack_confirmation_url(url: str) -> bool: + parsed_url = urlparse(url) + storage_endpoint = _parse_configured_storage_endpoint() + if not storage_endpoint: + return False + + if not _is_confirm_subscription_action(parsed_url): + return False + + if not _is_localstack_endpoint(parsed_url): + return False + + if not _is_localstack_endpoint(storage_endpoint): + return False + + return _endpoint_origin(parsed_url) == _endpoint_origin(storage_endpoint) + + +def _parse_configured_storage_endpoint() -> ParseResult | None: + endpoint_urls = [ + os.getenv("SELF_HOSTED_AWS_ENDPOINT_URL", ""), + os.getenv("S3_PRIVATE_DOMAIN", ""), + settings.S3_ENDPOINT_URL, + ] + + parsed_fallback: ParseResult | None = None + for endpoint_url in endpoint_urls: + endpoint_url = endpoint_url.strip() + if not endpoint_url: + continue + + parsed_endpoint = urlparse(endpoint_url) + if parsed_endpoint.scheme not in {"http", "https"}: + continue + if not parsed_endpoint.hostname: + continue + if _is_localstack_endpoint(parsed_endpoint): + return parsed_endpoint + if parsed_fallback is None: + parsed_fallback = parsed_endpoint + + return parsed_fallback + + +def _is_confirm_subscription_action(parsed_url: ParseResult) -> bool: + actions = parse_qs(parsed_url.query).get("Action", []) + return any(action.lower() == "confirmsubscription" for action in actions) + + +def _is_localstack_endpoint(parsed_url: ParseResult) -> bool: + hostname = (parsed_url.hostname or "").rstrip(".").lower() + return hostname in LOCALSTACK_HOSTNAMES + + +def _endpoint_origin(parsed_url: ParseResult) -> tuple[str, str, int | None]: + return ( + parsed_url.scheme.lower(), + (parsed_url.hostname or "").rstrip(".").lower(), + parsed_url.port, + ) + + +def _extract_first_resolved_ip(address_infos: Sequence[object]) -> str | None: + for address_info in address_infos: + if not isinstance(address_info, tuple) or len(address_info) < 5: + continue + family = address_info[0] + socket_address = address_info[4] + if family not in (socket.AF_INET, socket.AF_INET6): + continue + if not isinstance(socket_address, tuple) or not socket_address: + continue + address = socket_address[0] + if isinstance(address, str) and address: + return address + return None From d37c5519fa44110ff4ca4b80370a74b754fa9362 Mon Sep 17 00:00:00 2001 From: thinktraveller <161934609+thinktraveller@users.noreply.github.com> Date: Mon, 13 Jul 2026 16:28:13 +0800 Subject: [PATCH 2/2] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- patches/knowhere/subscription_service.py | 25 ++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/patches/knowhere/subscription_service.py b/patches/knowhere/subscription_service.py index 1b3df27..d2280c2 100644 --- a/patches/knowhere/subscription_service.py +++ b/patches/knowhere/subscription_service.py @@ -107,6 +107,31 @@ async def _resolve_configured_localstack_confirmation_url( failure_reason="hostname_resolution_failed", ) + # Avoid turning a trusted LocalStack hostname into a generic SSRF primitive if DNS is misconfigured. + import ipaddress + + try: + ip = ipaddress.ip_address(validated_ip) + except ValueError: + return HTTPURLValidationResult( + is_valid=False, + url=url, + hostname=hostname, + error_message=f"Resolved IP {validated_ip} for hostname {hostname} is invalid", + failure_reason="hostname_resolution_failed", + ) + + if not (ip.is_private or ip.is_loopback or ip.is_link_local): + return HTTPURLValidationResult( + is_valid=False, + url=url, + hostname=hostname, + error_message=( + f"Resolved IP {validated_ip} for hostname {hostname} is not a private/local address" + ), + failure_reason="disallowed_ip_address", + ) + return HTTPURLValidationResult( is_valid=True, url=url,