"""Regression (#1666·SEC): ``test_connection`` must scrub the structured ``detail`` payload, not just ``message``. Legacy or third-party drivers can put credential fragments into ``result.detail``: * feishu_chat_approval / feishu_contact — ``{"token_prefix": token[:8]+"..."}`` * huilianyi — token ``[:6]`` fragments * tencent_cloud — raw ``{exc}`` vendor exception reprs in message Before the fix ``test_connection`` re-emitted ``result.detail`` verbatim, so even 6–8 chars of a bearer token / app secret reached the admin-facing test payload and the audit log. ``_scrub_detail`` now drops credential-shaped keys and runs every string value through the same ``_scrub_message`` redactor. """ from __future__ import annotations import inspect import app.integrations.service as integration_service from app.integrations.service import _scrub_detail, _scrub_message def test_scrub_message_redacts_token_prefix_key_value_forms() -> None: for message in ( "OK · token_prefix=u-abc123... · probe green", "token_prefix: u-abc123...", "TOKEN_PREFIX = u-abc123...", '{"token_prefix":"u-abc123...","probe":"green"}', "{'token_prefix': 'u-abc123...', 'probe': 'green'}", ): out = _scrub_message(message) assert "u-abc123" not in out assert "" in out def test_scrub_message_uses_shared_credential_redactor_for_provider_shapes() -> None: secrets = ( "tencent-secret-id-value", "tencent-secret-key-value", "bearer-token-value-123456", "basic-token-value-123456", "amz-signature-value-123456", "oss-security-token-value-123456", ) message = ( "SecretId=tencent-secret-id-value " "SecretKey=tencent-secret-key-value " "Authorization: Bearer bearer-token-value-123456 " "Authorization: Basic basic-token-value-123456 " "https://example.invalid/object?X-Amz-Signature=" "amz-signature-value-123456&SecurityToken=token " "x-oss-security-token:oss-security-token-value-123456" ) output = _scrub_message(message) for secret in secrets: assert secret not in output assert "" in output assert "Authorization: Bearer " in output assert "Authorization: Basic " in output def test_scrub_detail_drops_token_prefix_key() -> None: out = _scrub_detail({"token_prefix": "u-abc12...", "ok": True}) assert "token_prefix" not in out assert out["ok"] is True def test_scrub_detail_drops_all_credential_keys_case_insensitive() -> None: out = _scrub_detail({ "Token": "abc", "ACCESS_TOKEN": "xyz", "app_secret": "s3cr3t", "tenant_access_token": "t-123", "note": "keep me", }) assert set(out) == {"note"} assert out["note"] == "keep me" def test_scrub_detail_drops_provider_specific_credential_keys() -> None: out = _scrub_detail({ "SecretId": "arbitrary-short-id", "SecretKey": "arbitrary-short-key", "X-Amz-Signature": "arbitrary-signature", "nested": {"authorization": "Basic arbitrary"}, "request_id": "keep-this-diagnostic", }) assert out == {"nested": {}, "request_id": "keep-this-diagnostic"} def test_integration_service_does_not_maintain_a_second_regex_set() -> None: source = inspect.getsource(integration_service) assert "from app.core.credential_redaction import" in source assert "_KEYVAL_PATTERNS" not in source assert "_BEGIN_PRIVATE_KEY_RE" not in source def test_scrub_detail_redacts_credential_shaped_values() -> None: # An AWS access key id embedded in a nested string value is redacted. out = _scrub_detail({"body": "denied for AKIAIOSFODNN7EXAMPLE on call"}) assert "AKIAIOSFODNN7EXAMPLE" not in out["body"] assert "" in out["body"] def test_scrub_detail_recurses_into_nested_structures() -> None: out = _scrub_detail({ "outer": {"token": "leak", "keep": "hi"}, "items": [{"access_token": "leak2"}, "AKIAIOSFODNN7EXAMPLE"], }) assert "token" not in out["outer"] assert out["outer"]["keep"] == "hi" assert out["items"][0] == {} assert "" in out["items"][1] def test_scrub_detail_passthrough_for_non_credential_scalars() -> None: assert _scrub_detail({"code": "TIMEOUT", "status": 500}) == { "code": "TIMEOUT", "status": 500, } assert _scrub_detail(None) is None