From 03237c6c0e367c3d08a758d1aafdda1b190157a8 Mon Sep 17 00:00:00 2001 From: Mathias Brulatout Date: Fri, 3 Jul 2026 10:17:44 +0200 Subject: [PATCH 1/2] transport: add opt-in raw/binary response mode Both sync (std.py, requests-based) and async (aio.py, aiohttp-based) HTTP clients unconditionally decoded every response body as UTF-8 text, which corrupts binary payloads (e.g. the gzip archive returned by GET /v1/snapshot). Adds a `raw: bool = False` parameter to HTTPClient.get() across base/std/aio: when set, the response body is kept as raw bytes (response.content / await resp.read()) instead of being decoded. Also widens put()'s data param to str | bytes, and adds CB.binary() for endpoints that need the raw body with normal status-code error handling. Off by default, so every existing endpoint is unaffected. Verified end-to-end (not just unit-level) against a live Consul instance for both the sync and async clients ahead of the Snapshot endpoint that depends on this. --- consul/aio.py | 27 ++++++++++++++++++++++----- consul/base.py | 4 ++-- consul/callback.py | 10 ++++++++++ consul/std.py | 14 ++++++++++---- 4 files changed, 44 insertions(+), 11 deletions(-) diff --git a/consul/aio.py b/consul/aio.py index 60d88c7..1ad7454 100644 --- a/consul/aio.py +++ b/consul/aio.py @@ -34,29 +34,46 @@ def __init__(self, *args, loop=None, connections_limit=None, connections_timeout self._session = aiohttp.ClientSession(connector=connector, **session_kwargs) # type: ignore async def _request( - self, callback, method, uri, headers: dict[str, str] | None, data=None, connections_timeout=None + self, + callback, + method, + uri, + headers: dict[str, str] | None, + data=None, + connections_timeout=None, + raw: bool = False, ): session_kwargs = {} if connections_timeout: timeout = aiohttp.ClientTimeout(total=connections_timeout) session_kwargs["timeout"] = timeout resp = await self._session.request(method, uri, headers=headers, data=data, **session_kwargs) # type: ignore - body = await resp.text(encoding="utf-8") + # raw=True keeps the response as bytes (e.g. the gzip archive returned by + # GET /v1/snapshot) instead of decoding it as UTF-8 text, which would corrupt it. + body = await resp.read() if raw else await resp.text(encoding="utf-8") if resp.status == 599: raise Timeout r = base.Response(resp.status, resp.headers, body) return callback(r) - def get(self, callback, path, params=None, headers: dict[str, str] | None = None, connections_timeout=None): + def get( + self, + callback, + path, + params=None, + headers: dict[str, str] | None = None, + raw: bool = False, + connections_timeout=None, + ): uri = self.uri(path, params) - return self._request(callback, "GET", uri, headers=headers, connections_timeout=connections_timeout) + return self._request(callback, "GET", uri, headers=headers, connections_timeout=connections_timeout, raw=raw) def put( self, callback, path, params=None, - data: str = "", + data: str | bytes = "", headers: dict[str, str] | None = None, connections_timeout=None, ): diff --git a/consul/base.py b/consul/base.py index 8de02b2..b1e4ca9 100644 --- a/consul/base.py +++ b/consul/base.py @@ -56,11 +56,11 @@ def uri(self, path: str, params: list[tuple[str, Any]] | None = None): return uri @abc.abstractmethod - def get(self, callback, path, params=None, headers: dict[str, str] | None = None): + def get(self, callback, path, params=None, headers: dict[str, str] | None = None, raw: bool = False): raise NotImplementedError @abc.abstractmethod - def put(self, callback, path, params=None, data: str = "", headers: dict[str, str] | None = None): + def put(self, callback, path, params=None, data: str | bytes = "", headers: dict[str, str] | None = None): raise NotImplementedError @abc.abstractmethod diff --git a/consul/callback.py b/consul/callback.py index 8a7d19f..4ce4698 100644 --- a/consul/callback.py +++ b/consul/callback.py @@ -43,6 +43,16 @@ def cb(response): return cb + @classmethod + def binary(cls) -> Callable[[Response], bytes]: + # returns the raw response body, for binary payloads (e.g. GET /v1/snapshot) + # that a JSON/text-oriented callback would corrupt + def cb(response): + CB._status(response, allow_404=False) + return response.body + + return cb + @classmethod def json( cls, diff --git a/consul/std.py b/consul/std.py index 1a716dc..6cd749c 100644 --- a/consul/std.py +++ b/consul/std.py @@ -13,15 +13,21 @@ def __init__(self, *args, **kwargs) -> None: super().__init__(*args, **kwargs) self.session = requests.session() - def response(self, response: Response): + def response(self, response: Response, raw: bool = False): + if raw: + # e.g. the gzip archive returned by GET /v1/snapshot -- decoding it as + # UTF-8 text would corrupt it, so the raw bytes are kept as-is. + return base.Response(response.status_code, response.headers, response.content) response.encoding = "utf-8" return base.Response(response.status_code, response.headers, response.text) - def get(self, callback, path, params=None, headers: dict[str, str] | None = None): + def get(self, callback, path, params=None, headers: dict[str, str] | None = None, raw: bool = False): uri = self.uri(path, params) - return callback(self.response(self.session.get(uri, headers=headers, verify=self.verify, cert=self.cert))) + return callback( + self.response(self.session.get(uri, headers=headers, verify=self.verify, cert=self.cert), raw=raw) + ) - def put(self, callback, path, params=None, data: str = "", headers: dict[str, str] | None = None): + def put(self, callback, path, params=None, data: str | bytes = "", headers: dict[str, str] | None = None): uri = self.uri(path, params) return callback( self.response(self.session.put(uri, headers=headers, data=data, verify=self.verify, cert=self.cert)) From 650cb0e6967080746d8f5444e8923be107ded34a Mon Sep 17 00:00:00 2001 From: Mathias Brulatout Date: Fri, 3 Jul 2026 10:18:06 +0200 Subject: [PATCH 2/2] api: add Snapshot endpoint (/v1/snapshot) Adds consul.snapshot.save()/restore(), built on the raw/binary response mode added in the previous commit. Both endpoints require a management-level ACL token, not a granular ACL rule. Verified against a live Consul instance for both the sync and async clients: save a snapshot, mutate state, restore, and confirm the pre-mutation state comes back. --- ENDPOINT_STATUS.md | 35 +++++++++++++++++++++++++++ consul/api/snapshot.py | 49 ++++++++++++++++++++++++++++++++++++++ consul/base.py | 2 ++ tests/api/test_snapshot.py | 26 ++++++++++++++++++++ 4 files changed, 112 insertions(+) create mode 100644 consul/api/snapshot.py create mode 100644 tests/api/test_snapshot.py diff --git a/ENDPOINT_STATUS.md b/ENDPOINT_STATUS.md index 4707840..08b7bc9 100644 --- a/ENDPOINT_STATUS.md +++ b/ENDPOINT_STATUS.md @@ -206,3 +206,38 @@ File: `consul/api/connect.py` | `GET /v1/connect/intentions/match` | `Connect.Intentions.match` | ✅ | - | | `POST /v1/connect/intentions` (legacy, deprecated 1.9.0) | - | ❌ | Intentionally not implemented — use exact-match endpoints above | | `GET/PUT/DELETE /v1/connect/intentions/:uuid` (legacy, deprecated 1.9.0) | - | ❌ | Intentionally not implemented — use exact-match endpoints above | + +## 14. Config Entries (`/v1/config`) +File: `consul/api/config.py` + +Generic CRUD covering all config entry kinds (service-defaults, service-router, +service-splitter, service-resolver, service-intentions, ingress-gateway, +terminating-gateway, proxy-defaults, mesh, exported-services, api-gateway, +http-route, tcp-route, inline-certificate, file-system-certificate, and others) +through a single Kind/Name-based implementation. Kind-specific body fields are +not individually modeled — callers pass the kind-specific body as a dict. + +| Endpoint | Python Method | Status | Missing Parameters | +| :--- | :--- | :--- | :--- | +| `PUT /v1/config` | `Config.set` | ✅ | - | +| `GET /v1/config/:kind/:name` | `Config.get` | ✅ | - | +| `GET /v1/config/:kind` | `Config.list` | ✅ | - | +| `DELETE /v1/config/:kind/:name` | `Config.delete` | ✅ | - | + +## 15. Discovery Chain (`/v1/discovery-chain`) +File: `consul/api/discovery_chain.py` + +| Endpoint | Python Method | Status | Missing Parameters | +| :--- | :--- | :--- | :--- | +| `GET /v1/discovery-chain/:service` | `DiscoveryChain.get` | ✅ | - | +| `POST /v1/discovery-chain/:service` | `DiscoveryChain.get` | ✅ | Uses POST automatically when any override is passed | + +## 16. Snapshot (`/v1/snapshot`) +File: `consul/api/snapshot.py` + +Both endpoints require a management-level ACL token (not a granular ACL rule). + +| Endpoint | Python Method | Status | Missing Parameters | +| :--- | :--- | :--- | :--- | +| `GET /v1/snapshot` | `Snapshot.save` | ✅ | - | +| `PUT /v1/snapshot` | `Snapshot.restore` | ✅ | - | diff --git a/consul/api/snapshot.py b/consul/api/snapshot.py new file mode 100644 index 0000000..c656984 --- /dev/null +++ b/consul/api/snapshot.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from typing import Any + +from consul.callback import CB + + +class Snapshot: + """ + Saves/restores a full point-in-time snapshot of the Consul server state. + Both endpoints require a management-level ACL token -- not a granular ACL + rule -- and are primarily intended for disaster recovery. + """ + + def __init__(self, agent) -> None: + self.agent = agent + + def save(self, token: str | None = None, dc: str | None = None, stale: bool = False) -> bytes: + """ + Saves a snapshot of the current Consul server state as a gzip archive. + :param token: a management-level token + :param dc: Optional datacenter to target; defaults to the client's own dc. + :param stale: If True, allows any server (not just the leader) to service the request. + :return: the raw gzip archive bytes + """ + params: list[tuple[str, Any]] = [] + dc = dc or self.agent.dc + if dc: + params.append(("dc", dc)) + if stale: + params.append(("stale", "true")) + headers = self.agent.prepare_headers(token) + return self.agent.http.get(CB.binary(), "/v1/snapshot", params=params, headers=headers, raw=True) + + def restore(self, snapshot: bytes, token: str | None = None, dc: str | None = None) -> bool: + """ + Restores a previously-saved snapshot. This is a disaster-recovery operation; + the target cluster must run the same Consul version as the source cluster. + :param snapshot: raw gzip archive bytes, as returned by :meth:`save` + :param token: a management-level token + :param dc: Optional datacenter to target; defaults to the client's own dc. + :return: True if the restore succeeded + """ + params: list[tuple[str, Any]] = [] + dc = dc or self.agent.dc + if dc: + params.append(("dc", dc)) + headers = self.agent.prepare_headers(token) + return self.agent.http.put(CB.boolean(), "/v1/snapshot", params=params, headers=headers, data=snapshot) diff --git a/consul/base.py b/consul/base.py index b1e4ca9..4ab5909 100644 --- a/consul/base.py +++ b/consul/base.py @@ -21,6 +21,7 @@ from consul.api.operator import Operator from consul.api.query import Query from consul.api.session import Session +from consul.api.snapshot import Snapshot from consul.api.status import Status from consul.api.txn import Txn from consul.exceptions import ConsulException @@ -167,6 +168,7 @@ def __init__( self.connect = Connect(self) self.config = Config(self) self.discovery_chain = DiscoveryChain(self) + self.snapshot = Snapshot(self) def __enter__(self): return self diff --git a/tests/api/test_snapshot.py b/tests/api/test_snapshot.py new file mode 100644 index 0000000..af95a63 --- /dev/null +++ b/tests/api/test_snapshot.py @@ -0,0 +1,26 @@ +import pytest + +import consul + + +class TestSnapshot: + def test_snapshot_save_and_restore(self, acl_consul) -> None: + c, master_token, _consul_version = acl_consul + + c.kv.put("snapshot-test-key", "before-snapshot", token=master_token) + + snapshot = c.snapshot.save(token=master_token) + assert isinstance(snapshot, bytes) + assert snapshot[:2] == b"\x1f\x8b" # gzip magic number + + c.kv.put("snapshot-test-key", "after-snapshot", token=master_token) + + assert c.snapshot.restore(snapshot, token=master_token) is True + + _index, value = c.kv.get("snapshot-test-key", token=master_token) + assert value["Value"] == b"before-snapshot" + + def test_snapshot_requires_management_token(self, acl_consul) -> None: + c, _master_token, _consul_version = acl_consul + + pytest.raises(consul.ACLPermissionDenied, c.snapshot.save, token="anonymous")