fix: close the search client a handle owns, not the default alias

_ElasticsearchHandle.close() and _OpenSearchHandle.close() re-resolved
the client library's "default" connection alias at close time. That is
wrong on the SIGHUP reload path, which is not a shutdown path:
_main's reload block calls _init_output_clients() -- and therefore
set_hosts() -> connections.create_connection() -- before
_close_output_clients() on the old clients, deliberately, so a broken
new config leaves the old clients running. By then the alias names the
*new* client, so the old handle closed the new client and then dropped
the alias entirely. Every later save raised
KeyError("There is no connection with alias 'default'.") until
parsedmarc was restarted, and the original client was leaked.

Verified against the installed SDK sources rather than the docs
(AGENTS.md, "Verify bug claims against authoritative sources", point 3):

  elasticsearch/dsl/connections.py (elasticsearch 9.x) and
  opensearchpy/connection/connections.py (opensearch-py 3.x) both define

      def create_connection(self, alias="default", **kwargs):
          conn = self._conns[alias] = <Client>(**kwargs)

  -- so create_connection() overwrites whatever the alias held, and
  returns the very object it stored (elasticsearch.dsl passes it through
  _with_user_agent(), which mutates headers and returns the same object,
  so identity is preserved) --

      def remove_connection(self, alias):
          errors = 0
          for d in (self._conns, self._kwargs):
              try:
                  del d[alias]
              except KeyError:
                  errors += 1
          if errors == 2:
              raise KeyError(f"There is no connection with alias {alias!r}.")

  -- i.e. it deletes the alias outright, with no regard for which client
  currently holds it, and

      def get_connection(self, alias="default"):
          if not isinstance(alias, str):
              return alias            # (opensearch-py; elasticsearch.dsl
                                      #  returns _with_user_agent(alias))
          try:
              return self._conns[alias]
          ...
          raise KeyError(f"There is no connection with alias {alias!r}.")

  -- a str only comes *back* out of get_connection() when a str was
  passed *in* as the alias, which these call sites never do; the client
  stored under "default" is always the object create_connection() built.
  The old `isinstance(conn, str)` guard was therefore dead code and is
  removed rather than carried forward (AGENTS.md: delete unreachable
  branches, don't hide them). add_connection(alias, conn) stores an
  arbitrary object as-is, which is what the new tests use.

set_hosts() now returns the client it registered, _init_output_clients()
passes that client to the handle, and close() closes that object and
gives up the alias only when the registry's current "default" *is* that
same object (identity check). The init/close order in the reload block
is unchanged.

tests/test_cli.py gains TestSearchBackendHandles, which drives the real
connection registries (the SDK boundary) with fake client objects, so no
network is touched: per backend, a reload-shaped test asserts the old
client is closed, the new one is not, and the alias still resolves to the
new client; a shutdown-shaped test asserts the alias is released when it
is still ours; and a best-effort test asserts a raising close() neither
propagates nor blocks the alias release. Each test restores whatever the
process-wide registry held beforehand. All six fail against the unfixed
handles.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sean Whalen
2026-09-09 19:06:37 -04:00
co-authored by Claude Opus 5
parent 77045c2df0
commit 3d387c29e8
5 changed files with 247 additions and 19 deletions
+4
View File
@@ -15,6 +15,10 @@
- **`parse_report_file()` now closes the file handle it opens itself for a path input if reading it raises.** When `input_` is a path, the function opened the file, read it, and closed it with no exception handling in between; an exception raised by `read()` (e.g. an `OSError` from the underlying storage) skipped the close, so the descriptor was left to be released only when Python's garbage collector eventually finalized the object — CPython's `io.IOBase.__del__` closes an unclosed file on finalization (<https://docs.python.org/3/library/io.html>) — rather than being closed deterministically. This is the pattern CodeQL's `py/file-not-closed` query flags, found in a local code-quality scan. The path branch now opens the file with a `with` block, so the handle is closed on both the success and exception paths. A file-like object or bytes buffer supplied by the caller is unaffected: as before, it is closed only after a successful read, and left open if `read()` raises.
### Bug fixes
- **A SIGHUP configuration reload no longer breaks every subsequent save to Elasticsearch or OpenSearch.** With `[elasticsearch]` or `[opensearch]` configured in watch mode, reloading the configuration re-registered the search client under the client library's `default` connection alias and then closed the previous run's clients. The close step re-resolved that alias instead of remembering the client it was created for, so it closed the *newly built* client and deleted the `default` alias outright — after which every report save failed with `KeyError: "There is no connection with alias 'default'."` until parsedmarc was restarted, and the original client was left open. Each backend's handle now closes the exact client it was created for and gives up the alias only while the alias still points at that client. Both the Elasticsearch and OpenSearch backends were affected.
## 11.0.1
### Security
+47 -15
View File
@@ -17,7 +17,7 @@ from argparse import ArgumentParser, Namespace
from configparser import ConfigParser
from glob import escape as glob_escape, glob
from ssl import CERT_NONE, create_default_context
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
import httpx
import yaml
@@ -1516,13 +1516,25 @@ def _parse_config(config: ConfigParser, opts):
class _ElasticsearchHandle:
"""Sentinel so Elasticsearch participates in _close_output_clients."""
"""Owns the Elasticsearch client so it participates in _close_output_clients.
Holds the client that ``elastic.set_hosts()`` registered under the
``default`` connection alias, rather than re-resolving that alias when
it is closed. _close_output_clients is not only a shutdown path: the
SIGHUP reload in _main deliberately builds the replacement clients
*before* closing the old ones, and building them re-registers the
``default`` alias, so by close time the alias names the new client.
Args:
connection: The client returned by ``elastic.set_hosts()``.
"""
def __init__(self, connection: Any):
self._connection = connection
def close(self):
try:
conn = elastic.connections.get_connection()
if not isinstance(conn, str):
conn.close()
self._connection.close()
except Exception:
# Best-effort, and deliberately silent: this is the first of
# two independent teardown steps, and swallowing here is what
@@ -1531,7 +1543,13 @@ class _ElasticsearchHandle:
# raises, which it cannot while this handler swallows.
pass
try:
elastic.connections.remove_connection("default")
# Give up the alias only while it still names our own client.
# elasticsearch.dsl's Connections.remove_connection() deletes
# the alias outright, so removing it after a reload had pointed
# it at a new client would leave that client unreachable, with
# every later save raising KeyError.
if elastic.connections.get_connection("default") is self._connection:
elastic.connections.remove_connection("default")
except Exception:
# Best-effort and silent for the same reason as above: a
# failure to give up the alias is not actionable during
@@ -1540,13 +1558,22 @@ class _ElasticsearchHandle:
class _OpenSearchHandle:
"""Sentinel so OpenSearch participates in _close_output_clients."""
"""Owns the OpenSearch client so it participates in _close_output_clients.
Holds the client that ``opensearch.set_hosts()`` registered under the
``default`` connection alias; see _ElasticsearchHandle for why the
alias is not re-resolved at close time.
Args:
connection: The client returned by ``opensearch.set_hosts()``.
"""
def __init__(self, connection: Any):
self._connection = connection
def close(self):
try:
conn = opensearch.connections.get_connection()
if not isinstance(conn, str):
conn.close()
self._connection.close()
except Exception:
# Best-effort, and deliberately silent: this is the first of
# two independent teardown steps, and swallowing here is what
@@ -1555,7 +1582,10 @@ class _OpenSearchHandle:
# raises, which it cannot while this handler swallows.
pass
try:
opensearch.connections.remove_connection("default")
# Only while the alias still names our own client; see
# _ElasticsearchHandle.close().
if opensearch.connections.get_connection("default") is self._connection:
opensearch.connections.remove_connection("default")
except Exception:
# Best-effort and silent for the same reason as above: a
# failure to give up the alias is not actionable during
@@ -1889,7 +1919,7 @@ def _init_output_clients(opts, index_prefix_domain_map=None):
if opts.elasticsearch_timeout is not None
else 60.0
)
elastic.set_hosts(
elasticsearch_connection = elastic.set_hosts(
opts.elasticsearch_hosts,
use_ssl=opts.elasticsearch_ssl,
ssl_cert_path=opts.elasticsearch_ssl_cert_path,
@@ -1914,7 +1944,9 @@ def _init_output_clients(opts, index_prefix_domain_map=None):
smtp_tls_indexes=es_smtp_tls_indexes,
legacy_fo_indexes=es_legacy_fo_indexes,
)
clients["elasticsearch"] = _ElasticsearchHandle()
clients["elasticsearch"] = _ElasticsearchHandle(
elasticsearch_connection
)
except Exception as e:
raise RuntimeError(f"Elasticsearch: {e}") from e
@@ -1958,7 +1990,7 @@ def _init_output_clients(opts, index_prefix_domain_map=None):
if opts.opensearch_timeout is not None
else 60.0
)
opensearch.set_hosts(
opensearch_connection = opensearch.set_hosts(
opts.opensearch_hosts,
use_ssl=opts.opensearch_ssl,
ssl_cert_path=opts.opensearch_ssl_cert_path,
@@ -1985,7 +2017,7 @@ def _init_output_clients(opts, index_prefix_domain_map=None):
smtp_tls_indexes=os_smtp_tls_indexes,
legacy_fo_indexes=os_legacy_fo_indexes,
)
clients["opensearch"] = _OpenSearchHandle()
clients["opensearch"] = _OpenSearchHandle(opensearch_connection)
except Exception as e:
raise RuntimeError(f"OpenSearch: {e}") from e
+8 -2
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any
from elasticsearch import Elasticsearch
from elasticsearch.dsl import (
Boolean,
Date,
@@ -591,7 +592,7 @@ def set_hosts(
api_key: str | None = None,
timeout: float = 60.0,
serverless: bool = False,
):
) -> Elasticsearch:
"""
Sets the Elasticsearch hosts to use
@@ -610,6 +611,11 @@ def set_hosts(
``create_indexes`` strips ``number_of_shards`` / ``number_of_replicas``
from its settings (which Serverless rejects with HTTP 400) and passes
any other settings through unchanged.
Returns:
Elasticsearch: The client registered under the ``default`` connection
alias. Callers that need to close this exact client later (rather than
whatever holds the alias at that point) should hold on to it.
"""
# Module-global; see the _SERVERLESS comment at the top of the module.
global _SERVERLESS
@@ -630,7 +636,7 @@ def set_hosts(
conn_params["basic_auth"] = (username, password)
if api_key:
conn_params["api_key"] = api_key
connections.create_connection(**conn_params)
return connections.create_connection(**conn_params)
def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
+8 -2
View File
@@ -17,6 +17,7 @@ from opensearchpy import (
Keyword,
Nested,
Object,
OpenSearch,
Q,
RequestsHttpConnection,
Search,
@@ -508,7 +509,7 @@ def set_hosts(
auth_type: str = "basic",
aws_region: str | None = None,
aws_service: str = "es",
):
) -> OpenSearch:
"""
Sets the OpenSearch hosts to use
@@ -524,6 +525,11 @@ def set_hosts(
auth_type (str): OpenSearch auth mode: basic (default) or awssigv4
aws_region (str): AWS region for SigV4 auth (required for awssigv4)
aws_service (str): AWS service for SigV4 signing (default: es)
Returns:
OpenSearch: The client registered under the ``default`` connection
alias. Callers that need to close this exact client later (rather than
whatever holds the alias at that point) should hold on to it.
"""
if not isinstance(hosts, list):
hosts = [hosts]
@@ -561,7 +567,7 @@ def set_hosts(
f"Unsupported OpenSearch auth_type '{auth_type}'. "
"Expected 'basic' or 'awssigv4'."
)
connections.create_connection(**conn_params)
return connections.create_connection(**conn_params)
def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
+180
View File
@@ -24,6 +24,7 @@ from unittest.mock import MagicMock, patch
import httpx
from azure.core.exceptions import ClientAuthenticationError
from elasticsearch import Elasticsearch
from kiota_abstractions.api_error import APIError
from msgraph.generated.models.o_data_errors.inner_error import InnerError
from msgraph.generated.models.o_data_errors.main_error import MainError
@@ -4173,6 +4174,185 @@ watch = true
es_client.close.assert_called_once()
class _FakeSearchConnection:
"""Stand-in for an Elasticsearch/OpenSearch client held in the real
connection registry. Records close() calls and nothing else, so no
network is touched while the registry itself stays real -- it is the
SDK boundary these handles are written against."""
def __init__(self, name: str):
self.name = name
self.close_count = 0
def __repr__(self):
return f"<fake {self.name} connection, closed {self.close_count}x>"
def close(self):
self.close_count += 1
class _RaisingSearchConnection(_FakeSearchConnection):
"""A client whose close() fails, e.g. because the transport is already
broken. The handles treat closing as best-effort."""
def close(self):
super().close()
raise RuntimeError("transport already closed")
class TestSearchBackendHandles(unittest.TestCase):
"""_ElasticsearchHandle / _OpenSearchHandle must close the client they
were built for, not whatever holds the ``default`` alias at close time.
A SIGHUP reload calls _init_output_clients -- which re-registers the
``default`` alias with the replacement client -- before it calls
_close_output_clients on the old clients (the order is deliberate, so a
broken new config leaves the old clients running). A handle that
re-resolved the alias when closing therefore closed the *new* client and
then dropped the alias entirely: both SDKs' Connections.remove_connection()
deletes the alias from ``_conns`` and ``_kwargs`` outright, so every later
save raised KeyError("There is no connection with alias 'default'.") and
the original client was never closed.
"""
def _isolate_default_alias(self, connections):
"""Restore whatever the process-wide registry held under the
``default`` alias, so these tests cannot leak into other tests
regardless of ordering."""
try:
previous = connections.get_connection("default")
except KeyError:
previous = None
def restore():
try:
connections.remove_connection("default")
except KeyError:
# Already gone: the test under this cleanup removed the
# alias itself, which is the state we are restoring to.
pass
if previous is not None:
connections.add_connection("default", previous)
self.addCleanup(restore)
def testElasticsearchHandleClosesItsOwnConnectionAfterReload(self):
connections = parsedmarc.elastic.connections
self._isolate_default_alias(connections)
old_conn = _FakeSearchConnection("old")
new_conn = _FakeSearchConnection("new")
connections.add_connection("default", cast(Elasticsearch, old_conn))
handle = parsedmarc.cli._ElasticsearchHandle(
connections.get_connection("default")
)
# What a reload does before the old handle is closed: the new
# client takes over the alias.
connections.add_connection("default", cast(Elasticsearch, new_conn))
handle.close()
self.assertEqual(old_conn.close_count, 1)
self.assertEqual(new_conn.close_count, 0)
self.assertIs(connections.get_connection("default"), new_conn)
def testElasticsearchHandleReleasesAliasOnShutdown(self):
"""The normal teardown path still frees the alias it owns."""
connections = parsedmarc.elastic.connections
self._isolate_default_alias(connections)
conn = _FakeSearchConnection("only")
connections.add_connection("default", cast(Elasticsearch, conn))
handle = parsedmarc.cli._ElasticsearchHandle(
connections.get_connection("default")
)
handle.close()
self.assertEqual(conn.close_count, 1)
with self.assertRaises(KeyError):
connections.get_connection("default")
def testElasticsearchHandleCloseIsBestEffort(self):
"""A client whose close() raises must not keep the handle from
releasing its alias. A second close() still calls the client's
close() again -- which raises and is swallowed -- even though the
alias-release step is now a no-op, since get_connection() raises
KeyError with the alias already gone."""
connections = parsedmarc.elastic.connections
self._isolate_default_alias(connections)
conn = _RaisingSearchConnection("raiser")
connections.add_connection("default", cast(Elasticsearch, conn))
handle = parsedmarc.cli._ElasticsearchHandle(
connections.get_connection("default")
)
handle.close()
self.assertEqual(conn.close_count, 1)
with self.assertRaises(KeyError):
connections.get_connection("default")
handle.close()
self.assertEqual(conn.close_count, 2)
def testOpenSearchHandleClosesItsOwnConnectionAfterReload(self):
connections = opensearch_module.connections
self._isolate_default_alias(connections)
old_conn = _FakeSearchConnection("old")
new_conn = _FakeSearchConnection("new")
connections.add_connection("default", old_conn)
handle = parsedmarc.cli._OpenSearchHandle(connections.get_connection("default"))
connections.add_connection("default", new_conn)
handle.close()
self.assertEqual(old_conn.close_count, 1)
self.assertEqual(new_conn.close_count, 0)
self.assertIs(connections.get_connection("default"), new_conn)
def testOpenSearchHandleReleasesAliasOnShutdown(self):
"""The normal teardown path still frees the alias it owns."""
connections = opensearch_module.connections
self._isolate_default_alias(connections)
conn = _FakeSearchConnection("only")
connections.add_connection("default", conn)
handle = parsedmarc.cli._OpenSearchHandle(connections.get_connection("default"))
handle.close()
self.assertEqual(conn.close_count, 1)
with self.assertRaises(KeyError):
connections.get_connection("default")
def testOpenSearchHandleCloseIsBestEffort(self):
"""A client whose close() raises must not keep the handle from
releasing its alias. A second close() still calls the client's
close() again -- which raises and is swallowed -- even though the
alias-release step is now a no-op, since get_connection() raises
KeyError with the alias already gone."""
connections = opensearch_module.connections
self._isolate_default_alias(connections)
conn = _RaisingSearchConnection("raiser")
connections.add_connection("default", conn)
handle = parsedmarc.cli._OpenSearchHandle(connections.get_connection("default"))
handle.close()
self.assertEqual(conn.close_count, 1)
with self.assertRaises(KeyError):
connections.get_connection("default")
handle.close()
self.assertEqual(conn.close_count, 2)
def _domain_map_tls_reports():
"""Four SMTP TLS reports for the index_prefix_domain_map tests: two
whose policy domains fold to the mapped base domain example.com (one of