"""Regression: ``AliyunProvider.list_object_storage`` MUST include the STS ``SecurityToken`` in both the ``x-oss-security-token`` header AND the OSS Auth v1 ``CanonicalizedOSSHeaders`` block of the signing string. Production incident 2026-05-17: ``alicloud-intl-primary`` ``object_storage`` cron fired ``auth_error`` on 18/18 sub-accounts every tick. Root cause — the OSS list_buckets codepath was the ONLY place in the driver that bypassed the ``self._security_token`` plumbing; every other RPC call injected it via ``_common_params``/``_call``, but OSS uses REST-style header auth and had to be wired explicitly. Without the token the assumed-role temp AK is rejected as ``InvalidSecurityToken.NotFound`` regardless of the role's policy attachments. Master-tenant tokenless calls (legacy single-account use) must keep working exactly as before. These tests pin both halves of the contract: 1. master case (``self._security_token is None``) — neither the header nor a canonicalized header line is emitted, and the signature is computed over the legacy StringToSign so existing single-account integrations remain green. 2. sub-account case (``self._security_token`` set by ``AlicloudIntlProvider._impersonate``) — header IS emitted, value matches the token verbatim, AND the signature changes (proving the canonicalized header line was included in StringToSign — not just appended to the request). 3. cross-iteration token rotation — re-reading ``self._security_token`` each pagination loop iteration lets a mid-fetch STS refresh take effect without restarting the call. 4. error-body redaction — when OSS echoes its server-computed ```` in a SignatureDoesNotMatch body (which embeds the full token), the redaction helper must strip it BEFORE the error string lands in sync_jobs / Feishu alerts. """ from __future__ import annotations import base64 import hashlib import hmac from typing import Any import httpx import pytest import respx from app.domains.cmdb.providers.aliyun import ( AliyunProvider, _redact_aliyun_error, ) from app.domains.cmdb.providers.base import ProviderCredentials _OSS_LIST_BUCKETS_EMPTY_BODY = ( '\n' '\n' ' 1x\n' ' \n' '' ) def _creds() -> ProviderCredentials: return ProviderCredentials( provider_id="test", provider_type="aliyun", account_alias="t", fields={ "access_key_id": "AKID-MASTER", "access_key_secret": "SECRET-MASTER", }, ) def _capture_request_headers() -> dict[str, Any]: """Configure respx to return an empty (terminal) ListBuckets response, capturing the headers from the single GET that fires. """ captured: dict[str, Any] = {} def _record(request: httpx.Request) -> httpx.Response: captured["headers"] = dict(request.headers) captured["url"] = str(request.url) return httpx.Response(200, text=_OSS_LIST_BUCKETS_EMPTY_BODY) respx.mock.get("https://oss.aliyuncs.com/").mock(side_effect=_record) return captured @pytest.mark.asyncio @respx.mock async def test_oss_list_buckets_without_token_omits_header() -> None: """Master-tenant single-account case: token is None → no ``x-oss-security-token`` header on the wire and the StringToSign matches the legacy (pre-fix) shape ``GET\\n\\n\\n{date}\\n/``. Bit-identical regression pin — recompute the legacy signature from the captured Date and assert byte equality. Without this, a future refactor that injects a stray empty canonical-headers line (``GET\\n\\n\\n{date}\\n\\n/``) would silently mutate every master-tenant signature and the existing header-absence assertion would still pass. """ p = AliyunProvider(_creds()) assert p._security_token is None captured = _capture_request_headers() result = await p.list_object_storage() assert result == [] headers = {k.lower(): v for k, v in captured["headers"].items()} assert "x-oss-security-token" not in headers auth = headers["authorization"] assert auth.startswith("OSS AKID-MASTER:") actual_sig = auth.split(":", 1)[1] date_str = headers["date"] legacy_string_to_sign = f"GET\n\n\n{date_str}\n/" expected = base64.b64encode( hmac.new( b"SECRET-MASTER", legacy_string_to_sign.encode(), hashlib.sha1, ).digest() ).decode() assert actual_sig == expected, ( "Master-tenant StringToSign changed shape — pre-fix bit-identity " "regression. Was a stray empty canonical-headers line injected?" ) @pytest.mark.asyncio @respx.mock async def test_oss_list_buckets_rejects_token_with_newline() -> None: """Defensive guard: a SecurityToken containing ``\\n`` or ``\\r`` would (a) break OSS Auth v1 canonical-header parsing (opaque ``SignatureDoesNotMatch``) and (b) be rejected by httpx's h11 layer as ``LocalProtocolError`` — propagating out of the cron path as an uncategorized HTTPError that skips this loop's 401/403/429/ 5xx classifier. The driver pre-empts this with a clean ``ProviderAuthError`` so retry/skip behavior is uniform with other auth failure shapes. """ from app.domains.cmdb.providers.exceptions import ProviderAuthError # Both NL and CR variants — pre-empt the same way. for bad in ("good_token\nLF_injected", "good_token\rCR_injected"): p = AliyunProvider(_creds()) p._ak = "STS-AK-TEMP" p._sk = "STS-SK-TEMP" p._security_token = bad # Wire respx to ALSO not be hit — if the guard fails open, the # call reaches httpx which would have rejected with a different # exception class; either way the test fails. respx.mock.get("https://oss.aliyuncs.com/").mock( return_value=httpx.Response(200, text=_OSS_LIST_BUCKETS_EMPTY_BODY), ) with pytest.raises(ProviderAuthError) as excinfo: await p.list_object_storage() msg = str(excinfo.value) # The token value MUST NOT echo into the error string. assert "good_token" not in msg assert "LF_injected" not in msg assert "CR_injected" not in msg @pytest.mark.asyncio @respx.mock async def test_oss_list_buckets_with_token_emits_header_and_resigns() -> None: """STS-assumed sub-account case: header MUST be present AND the signature MUST be recomputed over a StringToSign that includes ``x-oss-security-token:\\n`` in the canonical-headers block. We verify the second half by re-computing the signature ourselves using the captured Date header and asserting it matches the Authorization header. If the driver had emitted the header but forgotten the canonicalized line in StringToSign, the signature would NOT match this recomputation and OSS would still 403. """ p = AliyunProvider(_creds()) p._ak = "STS-AK-TEMP" p._sk = "STS-SK-TEMP" p._security_token = "STS-TOKEN-VALUE-AbC123==" captured = _capture_request_headers() result = await p.list_object_storage() assert result == [] headers = {k.lower(): v for k, v in captured["headers"].items()} assert headers["x-oss-security-token"] == "STS-TOKEN-VALUE-AbC123==" auth = headers["authorization"] assert auth.startswith("OSS STS-AK-TEMP:") actual_sig = auth.split(":", 1)[1] # Recompute expected signature from captured Date — the only piece # of the StringToSign we couldn't construct without inspecting the # outgoing request. Every other line is deterministic. date_str = headers["date"] canonicalized = "x-oss-security-token:STS-TOKEN-VALUE-AbC123==\n" string_to_sign = f"GET\n\n\n{date_str}\n{canonicalized}/" expected = base64.b64encode( hmac.new( b"STS-SK-TEMP", string_to_sign.encode(), hashlib.sha1, ).digest() ).decode() assert actual_sig == expected, ( "OSS Auth v1 signature mismatch — header emitted but token " "missing from CanonicalizedOSSHeaders in StringToSign; OSS " "would reject with SignatureDoesNotMatch / InvalidSecurityToken." ) @pytest.mark.asyncio @respx.mock async def test_oss_list_buckets_token_rotation_mid_pagination() -> None: """``self._security_token`` is read once per loop iteration so a mid-pagination STS refresh propagates to the next signed request. """ p = AliyunProvider(_creds()) p._ak = "STS-AK-TEMP" p._sk = "STS-SK-TEMP" p._security_token = "TOKEN-1" page_responses = [ # Page 1 — truncated, instructs OSS client to paginate. httpx.Response(200, text=( '\n' '\n' ' 1x\n' ' \n' ' \n' ' b1\n' ' oss-cn-hangzhou\n' ' Standard\n' ' \n' ' \n' ' true\n' ' b1\n' '' )), # Page 2 — terminal. httpx.Response(200, text=_OSS_LIST_BUCKETS_EMPTY_BODY), ] captured_tokens: list[str | None] = [] page_idx = {"i": 0} def _serve(request: httpx.Request) -> httpx.Response: captured_tokens.append(request.headers.get("x-oss-security-token")) i = page_idx["i"] page_idx["i"] += 1 # Between page 1 and page 2, simulate an STS rotation. if i == 0: p._security_token = "TOKEN-2" return page_responses[i] respx.mock.get("https://oss.aliyuncs.com/").mock(side_effect=_serve) result = await p.list_object_storage() assert len(result) == 1 assert captured_tokens == ["TOKEN-1", "TOKEN-2"] @pytest.mark.asyncio @respx.mock async def test_oss_list_buckets_redacts_token_from_signature_mismatch_body() -> None: """OSS ``SignatureDoesNotMatch`` error bodies echo the server-computed ```` element verbatim — which (under STS creds) embeds the full SecurityToken. The redaction helper must strip the value BEFORE the exception message lands in sync_jobs / Feishu alerts. """ p = AliyunProvider(_creds()) p._ak = "STS-AK-TEMP" p._sk = "STS-SK-TEMP" p._security_token = "TOKEN-LEAK-CANARY-AAAAAAAA" # OSS error format (real shape; trimmed for the test). The # ```` contains the literal canonical header line # ``x-oss-security-token:TOKEN-LEAK-CANARY-AAAAAAAA\n``. error_body = ( '\n' '\n' ' SignatureDoesNotMatch\n' ' The request signature we calculated does not match.\n' ' ...\n' ' GET\n\n\nWed, 17 May 2026 08:30:00 GMT\n' 'x-oss-security-token:TOKEN-LEAK-CANARY-AAAAAAAA\n/\n' ' STS-AK-TEMP\n' '' ) respx.mock.get("https://oss.aliyuncs.com/").mock( return_value=httpx.Response(403, text=error_body), ) from app.domains.cmdb.providers.exceptions import ProviderAuthError with pytest.raises(ProviderAuthError) as excinfo: await p.list_object_storage() msg = str(excinfo.value) # Critical: the leak-canary value must NOT appear in the surfaced # error string — neither raw nor URL-encoded. assert "TOKEN-LEAK-CANARY-AAAAAAAA" not in msg # Marker should remain so operators can correlate. Lower-case the # surfaced message ONCE so we accept any case variant of the header # name (OSS error bodies are inconsistent — some echo # ``x-oss-security-token``, some ``X-OSS-Security-Token``). assert "x-oss-security-token:" in msg.lower() def test_redact_aliyun_error_strips_oss_canonical_header_token() -> None: """Unit-level guarantee on the redaction primitive itself — covers both the colon-separated header form (OSS StringToSign echo) and the query-parameter form (``SecurityToken=...`` for RPC echoes). """ raw = ( "GET\n\n\nWed, 17 May 2026 08:30:00 GMT\n" "x-oss-security-token:abcSECRET123def==\n/" ) out = _redact_aliyun_error(raw) assert "abcSECRET123def==" not in out assert "x-oss-security-token:" in out def test_redact_aliyun_error_still_strips_query_param_token() -> None: raw = "Error url=...&SecurityToken=foobar123&Action=AssumeRole" out = _redact_aliyun_error(raw) assert "foobar123" not in out assert "SecurityToken=" in out def test_shared_redact_provider_error_strips_oss_canonical_header_token() -> None: """The shared ``_CRED_PATTERNS``-based ``redact_provider_error`` must also strip OSS canonical-header echoes. This is the documented fallback for any future driver (Tencent COS / UCloud US3 / AliCloud-intl direct OSS call) that bypasses the per-Aliyun-driver ``_redact_aliyun_error``. Critical: the regex captures group 1 = key, value follows ``:`` — NOT ``=``. STS bearer tokens are base64-padded and routinely end in ``=`` or ``==``; if ``_redact_match`` checked the ``=`` branch first, this would emit ``x-oss-security-token=`` (wrong separator) for any token containing padding. Pin the contract. """ from app.domains.cmdb.providers.exceptions import redact_provider_error # Realistic Aliyun STS token shape — starts with ``STS.``, base64 # alphabet, ends with ``==`` padding. raw = ( "OSS SignatureDoesNotMatch — body: GET\n\n\n" "Wed, 17 May 2026 08:30:00 GMT\n" "x-oss-security-token:STS.NeXyzABCdef123GHI456jkl789MNOpqr==\n" "/" ) out = redact_provider_error(raw) assert "STS.NeXyzABCdef123GHI456jkl789MNOpqr==" not in out # The separator MUST be ``:`` (matching the captured key form), # not ``=`` — otherwise the ``=`` shape would fire first and emit # a confusing inconsistent form. assert "x-oss-security-token:" in out def test_redact_aliyun_error_handles_xml_escaped_form() -> None: """OSS sometimes XML-escapes the canonical block — the ``\\n`` becomes literal text or gets wrapped between element boundaries. Either way the value must be stripped before the next ``<`` or whitespace. """ raw = ( "GET Wed " "x-oss-security-token:TOKEN_INSIDE_XML_HEX==" ) out = _redact_aliyun_error(raw) assert "TOKEN_INSIDE_XML_HEX==" not in out # ── Transport-layer retry on OSS list_buckets (2026-05-24) ────────── # # Sibling-bug coverage: the OSS list_buckets path used a separate # httpx.AsyncClient (line ~699) with NO try/except wrap around # ``client.get(...)``. The RPC ``_call`` path already had # transport-retry — parity fix here. Same sync-engine impact as the # Azure object_storage ReadTimeout: a single transport flake mid- # pagination dropped the entire OSS leg. @pytest.mark.asyncio async def test_oss_get_with_transport_retry_recovers_from_read_timeout( monkeypatch: pytest.MonkeyPatch, ) -> None: """The retry helper recovers from a ReadTimeout on attempt 1 and succeeds on attempt 2 — same shape as the Azure helper but isolated to the OSS path.""" from app.domains.cmdb.providers import aliyun as _aliyun_mod from app.domains.cmdb.providers.aliyun import ( _oss_get_with_transport_retry, ) class _Fake: def __init__(self, items: list) -> None: self.items = list(items) self.calls = 0 async def get(self, url: str, headers=None, params=None): self.calls += 1 item = self.items.pop(0) if isinstance(item, BaseException): raise item return item # Suppress backoff sleep so the test doesn't burn wall-clock time. async def _noop_sleep(_d: float) -> None: return None monkeypatch.setattr(_aliyun_mod.asyncio, "sleep", _noop_sleep) success = httpx.Response(200, text=_OSS_LIST_BUCKETS_EMPTY_BODY) client = _Fake([httpx.ReadTimeout("flake"), success]) resp = await _oss_get_with_transport_retry( client, "https://oss.aliyuncs.com/", endpoint_label="test_oss_retry", ) assert resp.status_code == 200 assert client.calls == 2 @pytest.mark.asyncio async def test_oss_get_with_transport_retry_exhausts_budget_reraises( monkeypatch: pytest.MonkeyPatch, ) -> None: """5 attempts then re-raise the LAST transport exception verbatim — preserves the httpx exception class for failure-notifier text.""" from app.domains.cmdb.providers import aliyun as _aliyun_mod from app.domains.cmdb.providers.aliyun import ( _oss_get_with_transport_retry, ) class _Fake: def __init__(self, items: list) -> None: self.items = list(items) self.calls = 0 async def get(self, url: str, headers=None, params=None): self.calls += 1 raise self.items.pop(0) async def _noop_sleep(_d: float) -> None: return None monkeypatch.setattr(_aliyun_mod.asyncio, "sleep", _noop_sleep) client = _Fake([ httpx.ConnectError("1"), httpx.ConnectError("2"), httpx.ConnectError("3"), httpx.ConnectError("4"), httpx.ConnectError("5"), ]) with pytest.raises(httpx.ConnectError): await _oss_get_with_transport_retry( client, "https://oss.aliyuncs.com/", endpoint_label="test_oss_exhausted", ) assert client.calls == 5 @pytest.mark.asyncio async def test_oss_get_with_transport_retry_covers_pool_and_write_errors( monkeypatch: pytest.MonkeyPatch, ) -> None: """The retry exception set must include PoolTimeout / WriteError / WriteTimeout / RemoteProtocolError — same surface as the Azure helper.""" from app.domains.cmdb.providers import aliyun as _aliyun_mod from app.domains.cmdb.providers.aliyun import ( _oss_get_with_transport_retry, ) class _Fake: def __init__(self, items: list) -> None: self.items = list(items) self.calls = 0 async def get(self, url: str, headers=None, params=None): self.calls += 1 item = self.items.pop(0) if isinstance(item, BaseException): raise item return item async def _noop_sleep(_d: float) -> None: return None monkeypatch.setattr(_aliyun_mod.asyncio, "sleep", _noop_sleep) success = httpx.Response(200, text=_OSS_LIST_BUCKETS_EMPTY_BODY) client = _Fake([ httpx.PoolTimeout("pool"), httpx.WriteError("half-close"), httpx.RemoteProtocolError("framing"), success, ]) resp = await _oss_get_with_transport_retry( client, "https://oss.aliyuncs.com/", endpoint_label="test_oss_class_coverage", ) assert resp.status_code == 200 assert client.calls == 4