cleveragents-core/features/steps/tls_certificate_check_steps.py

"""Step definitions for tls_certificate_check.feature.

Tests the ``scripts/check-tls-cert.py`` TLS certificate health-check script
using injected SSL contexts and mock socket connections so no real network
connections are made.  All scenarios are tagged ``@tdd_issue @tdd_issue_1543``
as permanent regression guards for the TLS handshake failure on
``git.dev.cleveragents.com`` (issue #1543).
"""

from __future__ import annotations

import datetime
import importlib.util as _ilu
import ssl
import sys
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock, patch

from behave import given, then, when
from behave.runner import Context

# ---------------------------------------------------------------------------
# Import the module under test
# ---------------------------------------------------------------------------

_SCRIPTS_DIR = Path(__file__).parent.parent.parent / "scripts"

# Import using importlib to handle the hyphenated filename.
# We must set __name__ and __package__ on the spec so that dataclasses
# can resolve the module's namespace correctly.
_spec = _ilu.spec_from_file_location(
    "check_tls_cert",
    _SCRIPTS_DIR / "check-tls-cert.py",
)
assert _spec is not None and _spec.loader is not None
_mod = _ilu.module_from_spec(_spec)
# Register the module in sys.modules BEFORE exec so that dataclasses
# (and other introspection tools) can look it up by __module__ name.
sys.modules["check_tls_cert"] = _mod
_spec.loader.exec_module(_mod)  # type: ignore[union-attr]

check_tls_certificate = _mod.check_tls_certificate
_hostname_matches_san = _mod._hostname_matches_san
CertCheckResult = _mod.CertCheckResult


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_cert(
    cn: str,
    sans: list[str],
    not_after: datetime.datetime,
) -> dict:
    """Build a minimal certificate dict in the format returned by ssl.getpeercert()."""
    san_tuples = tuple(("DNS", san) for san in sans)
    not_after_str = not_after.strftime("%b %d %H:%M:%S %Y GMT")
    return {
        "subject": ((("commonName", cn),),),
        "subjectAltName": san_tuples,
        "notAfter": not_after_str,
    }


def _make_mock_ssl_context(cert: dict) -> MagicMock:
    """Return a mock SSLContext whose wrap_socket yields a socket with *cert*."""
    mock_tls_sock = MagicMock()
    mock_tls_sock.getpeercert.return_value = cert
    mock_tls_sock.__enter__ = MagicMock(return_value=mock_tls_sock)
    mock_tls_sock.__exit__ = MagicMock(return_value=False)

    mock_ctx = MagicMock(spec=ssl.SSLContext)
    mock_ctx.wrap_socket.return_value = mock_tls_sock
    return mock_ctx


def _make_mock_raw_sock() -> MagicMock:
    """Return a mock raw socket for use as a context manager."""
    mock_raw = MagicMock()
    mock_raw.__enter__ = MagicMock(return_value=mock_raw)
    mock_raw.__exit__ = MagicMock(return_value=False)
    return mock_raw


# ---------------------------------------------------------------------------
# Given — certificate setup
# ---------------------------------------------------------------------------


@given('a TLS certificate for "{cn}" with SANs "{sans_str}"')
def step_cert_with_sans(context: Context, cn: str, sans_str: str) -> None:
    """Set up a mock certificate with the given CN and SANs."""
    context.cert_cn = cn
    context.cert_sans = [s.strip() for s in sans_str.split(",")]
    # Default expiry: 90 days from now (overridden by subsequent Given steps)
    context.cert_not_after = datetime.datetime.now(
        tz=datetime.UTC
    ) + datetime.timedelta(days=90)
    context.ssl_error = None
    context.timeout_error = False
    context.connection_error = None


@given("the certificate expires in {days:d} days")
def step_cert_expires_in(context: Context, days: int) -> None:
    """Set the certificate expiry to *days* days from now."""
    context.cert_not_after = datetime.datetime.now(
        tz=datetime.UTC
    ) + datetime.timedelta(days=days)


@given("the certificate expired {days:d} days ago")
def step_cert_expired_ago(context: Context, days: int) -> None:
    """Set the certificate expiry to *days* days in the past."""
    context.cert_not_after = datetime.datetime.now(
        tz=datetime.UTC
    ) - datetime.timedelta(days=days)


@given('the TLS connection raises an SSLError "{reason}"')
def step_ssl_error(context: Context, reason: str) -> None:
    """Configure the mock to raise an SSLCertVerificationError."""
    context.ssl_error = ssl.SSLCertVerificationError(reason)
    context.connection_error = None
    context.timeout_error = False


@given("the TLS connection times out")
def step_timeout_error(context: Context) -> None:
    """Configure the mock to raise a TimeoutError."""
    context.timeout_error = True
    context.ssl_error = None
    context.connection_error = None


@given("the TLS connection is refused")
def step_connection_refused(context: Context) -> None:
    """Configure the mock to raise a ConnectionRefusedError."""
    context.connection_error = ConnectionRefusedError("Connection refused")
    context.ssl_error = None
    context.timeout_error = False


# ---------------------------------------------------------------------------
# When — run the check
# ---------------------------------------------------------------------------


def _run_check(context: Context, hostname: str, warn_days: int = 30) -> None:
    """Execute check_tls_certificate with appropriate mocks."""
    ssl_error = getattr(context, "ssl_error", None)
    timeout_error = getattr(context, "timeout_error", False)
    connection_error = getattr(context, "connection_error", None)

    if ssl_error is not None:
        # Simulate SSLCertVerificationError during wrap_socket
        mock_ctx = MagicMock(spec=ssl.SSLContext)
        mock_ctx.wrap_socket.side_effect = ssl_error
        mock_raw = _make_mock_raw_sock()
        with patch("socket.create_connection", return_value=mock_raw):
            context.check_result = check_tls_certificate(
                hostname=hostname,
                warn_days=warn_days,
                ssl_context=mock_ctx,
            )
    elif timeout_error:
        # Simulate TimeoutError during create_connection
        with patch("socket.create_connection", side_effect=TimeoutError("timed out")):
            context.check_result = check_tls_certificate(
                hostname=hostname,
                warn_days=warn_days,
            )
    elif connection_error is not None:
        # Simulate OSError during create_connection
        with patch("socket.create_connection", side_effect=connection_error):
            context.check_result = check_tls_certificate(
                hostname=hostname,
                warn_days=warn_days,
            )
    else:
        # Normal path: inject a mock certificate
        cert = _make_cert(
            cn=context.cert_cn,
            sans=context.cert_sans,
            not_after=context.cert_not_after,
        )
        mock_ctx = _make_mock_ssl_context(cert)
        mock_raw = _make_mock_raw_sock()
        with patch("socket.create_connection", return_value=mock_raw):
            context.check_result = check_tls_certificate(
                hostname=hostname,
                warn_days=warn_days,
                ssl_context=mock_ctx,
            )


@when('the TLS check runs for hostname "{hostname}"')
def step_run_check(context: Context, hostname: str) -> None:
    """Run the TLS check for *hostname* with default warn-days (30)."""
    _run_check(context, hostname, warn_days=30)


@when('the TLS check runs for hostname "{hostname}" with warn-days {warn_days:d}')
def step_run_check_with_warn_days(
    context: Context, hostname: str, warn_days: int
) -> None:
    """Run the TLS check for *hostname* with a custom warn-days threshold."""
    _run_check(context, hostname, warn_days=warn_days)


@when('I check if hostname "{hostname}" matches SANs "{sans_str}"')
def step_check_san_match(context: Context, hostname: str, sans_str: str) -> None:
    """Invoke _hostname_matches_san directly."""
    sans = [s.strip() for s in sans_str.split(",")]
    context.san_match_result = _hostname_matches_san(hostname, sans)


# ---------------------------------------------------------------------------
# Then — assertions
# ---------------------------------------------------------------------------


@then("the check result should be a success")
def step_check_success(context: Context) -> None:
    """Assert the check result is OK."""
    result: Any = context.check_result
    assert result.ok, f"Expected TLS check to succeed but got errors: {result.errors!r}"


@then("the check result should be a failure")
def step_check_failure(context: Context) -> None:
    """Assert the check result is not OK."""
    result: Any = context.check_result
    assert not result.ok, (
        f"Expected TLS check to fail but it succeeded (warnings: {result.warnings!r})"
    )


@then('the check error should mention "{text}"')
def step_check_error_mentions(context: Context, text: str) -> None:
    """Assert at least one error message contains *text*."""
    result: Any = context.check_result
    combined = " ".join(result.errors)
    assert text.lower() in combined.lower(), (
        f"Expected error containing {text!r} but got errors: {result.errors!r}"
    )


@then('the check warning should mention "{text}"')
def step_check_warning_mentions(context: Context, text: str) -> None:
    """Assert at least one warning message contains *text*."""
    result: Any = context.check_result
    combined = " ".join(result.warnings)
    assert text.lower() in combined.lower(), (
        f"Expected warning containing {text!r} but got warnings: {result.warnings!r}"
    )


@then("the check has no warnings")
def step_check_no_warnings(context: Context) -> None:
    """Assert the check result has no warnings."""
    result: Any = context.check_result
    assert not result.warnings, f"Expected no warnings but got: {result.warnings!r}"


@then("the SAN match result should be True")
def step_san_match_true(context: Context) -> None:
    """Assert _hostname_matches_san returned True."""
    assert context.san_match_result is True, (
        f"Expected SAN match to be True but got {context.san_match_result!r}"
    )


@then("the SAN match result should be False")
def step_san_match_false(context: Context) -> None:
    """Assert _hostname_matches_san returned False."""
    assert context.san_match_result is False, (
        f"Expected SAN match to be False but got {context.san_match_result!r}"
    )