"""Coverage for ``redact_provider_error`` credential-pattern surface.

Pre-launch security review 2026-05-16 added two patterns to
``_CRED_PATTERNS`` (PEM private-key blocks + JWT bearer tokens) — both
shapes show up in Azure / GCP / Atlassian SDK error strings under
certain misconfig paths and were previously slipping through the helper
that backs every cloud driver's error-redaction wrap. The same review
also rewrote the substitution callback to be group-safe: the prior
form referenced ``m.group(1)`` unconditionally when the matched string
contained ``=``, which would IndexError on every PEM match because PEM
bodies legitimately contain ``=`` base64 padding and the PEM regex has
no capture groups.

These tests pin both the new pattern coverage and the substitution
safety so neither regresses silently.
"""
from __future__ import annotations

from app.domains.cmdb.providers.exceptions import (
    redact_leg_entry,
    redact_provider_error,
)

# ── PEM private-key bodies ──────────────────────────────────────


_PEM_BODY = (
    "-----BEGIN PRIVATE KEY-----\n"
    "MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDV"
    "abc123def456ghi789jkl012mno345pqr678stu901vwx234yz567=\n"
    "AnotherLineWithBase64Plus/AndSlashes+And==Padding=\n"
    "-----END PRIVATE KEY-----"
)


def test_pem_private_key_block_is_redacted() -> None:
    """A GCP SA JSON parse error that echoes the private_key body
    must NOT land in the integrations table or admin browser."""
    msg = f"failed to parse SA credential: {_PEM_BODY} (invalid key)"
    out = redact_provider_error(msg, max_len=10_000)
    assert "-----BEGIN" not in out
    assert "MIIEvQIB" not in out
    assert "<redacted>" in out


def test_pem_rsa_private_key_variant_is_redacted() -> None:
    """The ``BEGIN RSA PRIVATE KEY`` variant (PKCS#1 not PKCS#8)."""
    body = (
        "-----BEGIN RSA PRIVATE KEY-----\n"
        "MIIEowIBAAKCAQEAxx==\n"
        "-----END RSA PRIVATE KEY-----"
    )
    out = redact_provider_error(body)
    assert "BEGIN" not in out
    assert "<redacted>" in out


def test_pem_redaction_does_not_index_error_on_equals_padding() -> None:
    """Regression: pre-fix substitution lambda did
    ``m.group(1)`` when ``"=" in m.group(0)``. PEM bodies have ``=``
    padding and PEM regex has no capture groups → IndexError. The
    rewritten ``_redact_match`` uses ``m.lastindex`` to guard."""
    # Force a PEM block that contains '=' (it always does, but pin it
    # explicitly so the test fails if the body shape is ever munged).
    assert "=" in _PEM_BODY
    out = redact_provider_error(_PEM_BODY)
    # If the lambda were still broken this call would have raised
    # IndexError before reaching the assertion.
    assert "<redacted>" in out


# ── JWT bearer tokens ───────────────────────────────────────────


_JWT = (
    "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9."
    "eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ."
    "SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c"
)


def test_jwt_bearer_token_is_redacted() -> None:
    msg = f"403 forbidden: token {_JWT} expired"
    out = redact_provider_error(msg)
    assert "eyJhbGciOiJSUzI1NiI" not in out
    assert "<redacted>" in out


def test_short_jwt_lookalike_not_redacted() -> None:
    """``eyJ.x.y`` form shorter than the 8-char floor must not match —
    otherwise we'd swallow incidental dotted identifiers like
    ``image.foo.bar`` if they happened to start with eyJ. The 8-char
    floor per segment protects against this."""
    # Build a deliberately-short eyJ-prefix string that fails the
    # 8-char floor in at least one segment.
    msg = "module eyJxx.yy.zz unreachable"
    out = redact_provider_error(msg)
    assert "eyJxx.yy.zz" in out


def test_jwt_floor_exact_8_chars_per_segment_matches() -> None:
    """Pin the 8-char-per-segment floor specifically: 8 chars per
    segment is the minimum that triggers the JWT pattern."""
    # eyJ + 8 more = 11 chars (first segment); 8 chars for second and
    # third segments. ``\w-`` covers alphanumerics and dashes.
    tok = "eyJabcdefgh.abcdefgh.abcdefgh"
    out = redact_provider_error(tok)
    assert "<redacted>" in out
    assert "eyJabcdefgh" not in out


def test_jwt_floor_below_8_chars_does_not_match() -> None:
    """Pin the floor from the other side: 7 chars in any segment
    should not trigger."""
    # 7 chars in the middle segment.
    tok = "eyJabcdefgh.abcdefg.abcdefgh"
    out = redact_provider_error(tok)
    assert "eyJabcdefgh" in out


# ── Existing patterns (sanity, not exhaustive) ──────────────────


def test_akia_access_key_id_is_redacted() -> None:
    msg = "AccessDenied: AKIAEXAMPLE0123456789 lacks permission"
    out = redact_provider_error(msg)
    assert "AKIAEXAMPLE0123456789" not in out
    assert "<redacted>" in out


def test_akia_with_internal_equals_collapses_fully() -> None:
    """Regression: an AKIA-class token that contains ``=`` (some
    base64-encoded STS session tokens carry padding) previously
    rendered as ``AKIA=<redacted>`` — leaking the vendor discriminator
    and treating the body as a ``key=value`` split point. Fix:
    AKIA-family pattern is now non-capturing so ``m.lastindex is None``
    routes through the full-collapse branch."""
    msg = "session expired: AKIAEXAMPLE0123456789=AB rejected"
    out = redact_provider_error(msg)
    assert "AKIA" not in out  # vendor prefix gone too
    assert "0123456789" not in out
    assert "<redacted>" in out


def test_signature_query_param_redacted_preserves_key() -> None:
    """``key=value`` form: preserve the key, redact the value — admin
    needs to see WHICH credential field failed."""
    msg = "url: https://x.example/?X-Amz-Signature=ABCDEF0123456789abcdef0123456789 expired"
    out = redact_provider_error(msg)
    assert "ABCDEF0123456789" not in out
    assert "X-Amz-Signature=<redacted>" in out


def test_authorization_bearer_header_redacted() -> None:
    """Bearer pattern matches ``Authorization: Bearer <token>`` and
    collapses the entire match to ``<redacted>``. That's coarser than
    the ``key=`` form (which preserves the key name), but acceptable
    for header-style payloads — the admin still sees the surrounding
    context (``401: <redacted>``) and the token is gone."""
    msg = "401: Authorization: Bearer abcdef0123456789ABCDEF0123456789abcdef"
    out = redact_provider_error(msg)
    assert "abcdef0123456789ABCDEF" not in out
    assert "<redacted>" in out
    # Surrounding text outside the matched span is preserved.
    assert out.startswith("401:")


# ── Length cap + None safety ────────────────────────────────────


def test_length_truncated_with_ellipsis() -> None:
    big = "x" * 1000
    out = redact_provider_error(big, max_len=100)
    assert len(out) <= 101  # 100 + 1-char ellipsis
    assert out.endswith("…")


def test_none_input_returns_empty_string() -> None:
    assert redact_provider_error(None) == ""


def test_empty_message_exception_falls_back_to_type_name() -> None:
    """httpx timeout / connection exceptions stringify to ``""`` — their
    meaning is in the class name. The redactor must surface the type
    name so admin-facing strings never collapse to ``UCloud HTTP
    error: `` (prod 2026-06-22 cn-bj2 servers sync).
    """
    import httpx

    for exc in (
        httpx.ReadTimeout(""),
        httpx.ConnectTimeout(""),
        httpx.PoolTimeout(""),
    ):
        out = redact_provider_error(exc)
        assert out == type(exc).__name__
        # Composed into the call-site shape it now carries the cause.
        assert f"UCloud HTTP error: {out}".strip().endswith(
            type(exc).__name__
        )


def test_exception_with_message_is_unchanged() -> None:
    """The empty-message fallback must NOT clobber a real message."""
    out = redact_provider_error(RuntimeError("real cause"))
    assert out == "real cause"


def test_empty_string_input_stays_empty() -> None:
    """A genuinely-empty string / non-exception input is not coerced to a
    type name — the fallback is exception-only."""
    assert redact_provider_error("") == ""
    assert redact_provider_error("   ") == "   "


def test_non_string_input_coerced() -> None:
    """Some SDKs raise with dict args; we accept ``object`` and ``str()``."""
    out = redact_provider_error({"Signature": "ABCDEF" * 8})
    # The ``str()`` coercion produces ``"{'Signature': 'ABCDEF...'}"``;
    # the value-side gets the AKIA-style alphanumeric run, which the
    # current pattern set doesn't catch verbatim (no ``=`` prefix), so
    # this only checks the helper doesn't crash on the coercion path.
    assert isinstance(out, str)


# ── Substitution safety regression ─────────────────────────────


def test_substitution_does_not_crash_on_pattern_without_groups() -> None:
    """The two new patterns (PEM, JWT) have NO capture groups. The
    substitution lambda must not call ``m.group(1)`` unconditionally."""
    # If the regression were back, this call would raise IndexError.
    out = redact_provider_error("intro " + _PEM_BODY + " outro " + _JWT)
    assert "<redacted>" in out
    assert "-----BEGIN" not in out
    assert "eyJhbGciOi" not in out


def test_idempotent_on_already_redacted_string() -> None:
    """Calling the redactor twice produces no ``<redacted><redacted>``
    artifact — important because ``_summarize_leg_failures`` reads
    fields that already passed through ``_redact_leg``."""
    once = redact_provider_error("AKIAEXAMPLE0123456789 denied")
    twice = redact_provider_error(once)
    assert once == twice


def test_named_secret_fields_are_redacted() -> None:
    cases = [
        ("SecretKey=plain-format-secret-value", "plain-format-secret-value"),
        (
            '{"secret_id":"tenant-key-without-akid-prefix"}',
            "tenant-key-without-akid-prefix",
        ),
        (
            "client_secret: oauth-client-secret-value",
            "oauth-client-secret-value",
        ),
    ]
    for message, secret in cases:
        out = redact_provider_error(message)
        assert secret not in out
        assert "<redacted>" in out


# ── redact_leg_entry — used by both test_connection AND sync path ─


def test_redact_leg_entry_scrubs_leg_and_reason_fields() -> None:
    """Symmetric redaction wrap consumed by both
    ``_CloudDriverBase.test_connection`` and
    ``app.domains.cmdb.sync._sync_capabilities``. Pin the helper
    contract here so both paths share the same guarantee."""
    entry = {
        "leg": "region:us-east-1",
        "reason": "AccessKeyId=AKIAEXAMPLE0123456789 denied",
        "capability": "servers",
    }
    out = redact_leg_entry(entry)
    assert "AKIAEXAMPLE0123456789" not in out["reason"]
    assert "<redacted>" in out["reason"]
    # Non-cred fields pass through.
    assert out["capability"] == "servers"
    assert out["leg"] == "region:us-east-1"


def test_redact_leg_entry_handles_non_dict_input() -> None:
    """Sync path's ``isinstance(e, dict)`` filter is the first line of
    defense, but the helper itself must also be defensive."""
    assert redact_leg_entry("not a dict") == "not a dict"  # type: ignore[arg-type]
    assert redact_leg_entry(None) is None  # type: ignore[arg-type]


def test_redact_leg_entry_handles_missing_keys() -> None:
    """Future driver might only emit ``leg`` without ``reason`` or
    vice versa — helper must not KeyError."""
    assert redact_leg_entry({"leg": "x"}) == {"leg": "x"}
    assert redact_leg_entry({"reason": "x"}) == {"reason": "x"}
    assert redact_leg_entry({}) == {}


def test_redact_leg_entry_returns_copy_not_mutation() -> None:
    """Caller may iterate the original list (e.g. for counts) after
    redaction — helper must not mutate the input dict."""
    original = {"leg": "x", "reason": "AKIAEXAMPLE0123456789"}
    redact_leg_entry(original)
    assert original["reason"] == "AKIAEXAMPLE0123456789"