cleveragents-core/features/steps/security_audit_steps.py

"""Step definitions for SEC7 - Audit logging for apply.

Tests the audit service: recording, querying, filtering, retention,
pruning, and data class serialization.
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta

from behave import given, then, when
from behave.runner import Context
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from cleveragents.application.services.audit_service import (
    _TIMESTAMP_FMT,
    AuditService,
)
from cleveragents.config.settings import Settings
from cleveragents.infrastructure.database.models import AuditLogModel, Base

# ── Helpers ───────────────────────────────────────────────────────


def _make_settings(**overrides: object) -> Settings:
    """Create a Settings instance with optional field overrides."""
    Settings._instance = None
    base = Settings()
    if overrides:
        return base.model_copy(update=overrides)
    return base


def _fresh_service(settings: Settings | None = None) -> AuditService:
    """Create an AuditService backed by an in-memory SQLite database."""
    if settings is None:
        settings = _make_settings(database_url="sqlite:///:memory:", audit_async=False)
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine)()
    return AuditService(settings=settings, session=session)


def _insert_old_entry(
    service: AuditService,
    days_ago: int,
    event_type: str = "plan_applied",
) -> None:
    """Insert an entry with a manually backdated created_at.

    Uses the same timestamp format as the service (``_TIMESTAMP_FMT``)
    so that lexicographic comparisons in ``prune()`` and ``since``
    filtering remain correct.

    Directly accesses ``service._session`` because there is no public
    API for inserting entries with an arbitrary ``created_at``.
    """
    old_time = (datetime.now(tz=UTC) - timedelta(days=days_ago)).strftime(
        _TIMESTAMP_FMT
    )
    row = AuditLogModel(
        event_type=event_type,
        details="{}",
        created_at=old_time,
    )
    service._session.add(row)
    service._session.commit()


# ── Given ─────────────────────────────────────────────────────────


@given("a fresh audit service")
def step_fresh_service(context: Context) -> None:
    context.service = _fresh_service()
    context.entry = None
    context.entries = []
    context.prune_result = None


@given("a fresh audit service with mixed events")
def step_fresh_with_mixed(context: Context) -> None:
    context.service = _fresh_service()
    svc = context.service
    svc.record(event_type="plan_applied", plan_id="plan-A", project_name="project-X")
    svc.record(event_type="plan_applied", plan_id="plan-A", project_name="project-Y")
    svc.record(event_type="plan_cancelled", plan_id="plan-B", project_name="project-X")
    svc.record(event_type="config_changed", project_name="project-Z")
    context.entry = None
    context.entries = []


@given("a fresh audit service with old and new events")
def step_fresh_with_old_new(context: Context) -> None:
    context.service = _fresh_service()
    _insert_old_entry(context.service, days_ago=100)
    context.service.record(event_type="plan_applied", plan_id="new")
    context.entry = None
    context.entries = []


@given("a fresh audit service with {count:d} events")
def step_fresh_with_count(context: Context, count: int) -> None:
    context.service = _fresh_service()
    for i in range(count):
        context.service.record(event_type="plan_applied", plan_id=f"plan-{i}")
    context.entry = None
    context.entries = []


@given("a fresh audit service with entries from {days:d} days ago")
def step_fresh_with_old(context: Context, days: int) -> None:
    context.service = _fresh_service()
    _insert_old_entry(context.service, days_ago=days)
    context.entry = None
    context.entries = []
    context.prune_result = None


@given("a fresh audit service with {count:d} entries from {days:d} days ago")
def step_fresh_count_old(context: Context, count: int, days: int) -> None:
    context.service = _fresh_service()
    for _ in range(count):
        _insert_old_entry(context.service, days_ago=days)
    context.entry = None
    context.entries = []
    context.prune_result = None


@given("default settings")
def step_default_settings(context: Context) -> None:
    context.settings = _make_settings()


@given("settings with audit_retention_days {days:d}")
def step_settings_retention(context: Context, days: int) -> None:
    context.settings = _make_settings(audit_retention_days=days)


# ── When ──────────────────────────────────────────────────────────


@when(
    'I record a "{event_type}" event with plan_id "{plan_id}" and project "{project}"'
)
def step_record_plan_project(
    context: Context, event_type: str, plan_id: str, project: str
) -> None:
    context.entry = context.service.record(
        event_type=event_type, plan_id=plan_id, project_name=project
    )


@when(
    'I record a "{event_type}" event with plan_id "{plan_id}" '
    'and details reason "{reason}"'
)
def step_record_with_reason(
    context: Context, event_type: str, plan_id: str, reason: str
) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        plan_id=plan_id,
        details={"reason": reason},
    )


@when(
    'I record a "{event_type}" event with details resource_id '
    '"{resource_id}" and tool "{tool}"'
)
def step_record_resource_modified(
    context: Context, event_type: str, resource_id: str, tool: str
) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        details={"resource_id": resource_id, "tool": tool},
    )


@when('I record a "{event_type}" event with actor "{actor}"')
def step_record_with_actor(context: Context, event_type: str, actor: str) -> None:
    context.entry = context.service.record(event_type=event_type, actor_name=actor)


@when('I record a "{event_type}" event with user "{user}"')
def step_record_with_user(context: Context, event_type: str, user: str) -> None:
    context.entry = context.service.record(event_type=event_type, user_identity=user)


@when(
    'I record a "{event_type}" event with details entity_type '
    '"{entity_type}" and entity_name "{entity_name}"'
)
def step_record_entity_deleted(
    context: Context, event_type: str, entity_type: str, entity_name: str
) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        details={"entity_type": entity_type, "entity_name": entity_name},
    )


@when('I record a "{event_type}" event with no details')
def step_record_no_details(context: Context, event_type: str) -> None:
    context.entry = context.service.record(event_type=event_type)


@when('I record {count:d} "{event_type}" events')
def step_record_multiple(context: Context, count: int, event_type: str) -> None:
    for i in range(count):
        context.entry = context.service.record(
            event_type=event_type, plan_id=f"plan-{i}"
        )


@when('I list entries with plan filter "{plan_id}"')
def step_list_by_plan(context: Context, plan_id: str) -> None:
    context.entries = context.service.list_entries(plan_id=plan_id)


@when('I list entries with project filter "{project}"')
def step_list_by_project(context: Context, project: str) -> None:
    context.entries = context.service.list_entries(project_name=project)


@when('I list entries with type filter "{event_type}"')
def step_list_by_type(context: Context, event_type: str) -> None:
    context.entries = context.service.list_entries(event_type=event_type)


@when('I list entries since "{since}"')
def step_list_since(context: Context, since: str) -> None:
    context.entries = context.service.list_entries(since=since)


@when("I list entries with limit {limit:d}")
def step_list_with_limit(context: Context, limit: int) -> None:
    context.entries = context.service.list_entries(limit=limit)


@when("I list all entries")
def step_list_all(context: Context) -> None:
    context.entries = context.service.list_entries()


@when("I fetch the entry by its ID")
def step_fetch_by_id(context: Context) -> None:
    context.fetched = context.service.get_entry(context.entry.id)


@when("I fetch entry with ID {audit_id:d}")
def step_fetch_nonexistent(context: Context, audit_id: int) -> None:
    context.fetched = context.service.get_entry(audit_id)


@when("I prune with retention {days:d} days")
def step_prune(context: Context, days: int) -> None:
    context.prune_result = context.service.prune(retention_days=days)


@when("I serialize the entry to dict")
def step_serialize(context: Context) -> None:
    context.entry_dict = context.entry.as_dict()


@when('I record a "{event_type}" event with all nullable fields as None')
def step_record_all_none(context: Context, event_type: str) -> None:
    context.entry = context.service.record(event_type=event_type)


@when('I record a "{event_type}" event with datetime in details')
def step_record_datetime_details(context: Context, event_type: str) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        details={"timestamp": datetime.now(tz=UTC)},
    )


@when(
    'I record a "{event_type}" event with details files_changed '
    "{count:d} and validation_passed true"
)
def step_record_structured_details(
    context: Context, event_type: str, count: int
) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        details={"files_changed": count, "validation_passed": True},
    )


@when('I record a "{event_type}" event with session_id "{session_id}"')
def step_record_session_created(
    context: Context, event_type: str, session_id: str
) -> None:
    context.entry = context.service.record(
        event_type=event_type,
        details={"session_id": session_id},
    )


@when('I record an event with invalid event_type "{event_type}"')
def step_record_invalid_type(context: Context, event_type: str) -> None:
    try:
        context.service.record(event_type=event_type)
        context.error = None
    except ValueError as exc:
        context.error = exc


@when("I create an AuditService with invalid settings")
def step_create_invalid(context: Context) -> None:
    try:
        AuditService(settings="not-settings")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


# ── Then ──────────────────────────────────────────────────────────


@then("the audit log should contain {count:d} entry")
@then("the audit log should contain {count:d} entries")
def step_check_count(context: Context, count: int) -> None:
    assert context.service.count() == count, (
        f"Expected {count}, got {context.service.count()}"
    )


@then('the entry event_type should be "{expected}"')
def step_check_event_type(context: Context, expected: str) -> None:
    assert context.entry.event_type == expected


@then('the entry plan_id should be "{expected}"')
def step_check_plan_id(context: Context, expected: str) -> None:
    assert context.entry.plan_id == expected


@then('the entry project_name should be "{expected}"')
def step_check_project(context: Context, expected: str) -> None:
    assert context.entry.project_name == expected


@then('the entry actor_name should be "{expected}"')
def step_check_actor(context: Context, expected: str) -> None:
    assert context.entry.actor_name == expected


@then('the entry user_identity should be "{expected}"')
def step_check_user(context: Context, expected: str) -> None:
    assert context.entry.user_identity == expected


@then('the entry details should contain key "{key}"')
def step_check_details_key(context: Context, key: str) -> None:
    assert key in context.entry.details, f"Key {key!r} not in {context.entry.details}"


@then('the entry details key "{key}" should be "{expected}"')
def step_check_details_value(context: Context, key: str, expected: str) -> None:
    assert str(context.entry.details[key]) == expected


@then("the entry details should be empty")
def step_check_details_empty(context: Context) -> None:
    assert context.entry.details == {} or context.entry.details is None


@then('all returned entries should have plan_id "{expected}"')
def step_all_plan_id(context: Context, expected: str) -> None:
    assert len(context.entries) > 0
    for e in context.entries:
        assert e.plan_id == expected, f"Entry {e.id} has plan_id={e.plan_id}"


@then('all returned entries should have project_name "{expected}"')
def step_all_project(context: Context, expected: str) -> None:
    assert len(context.entries) > 0
    for e in context.entries:
        assert e.project_name == expected


@then('all returned entries should have event_type "{expected}"')
def step_all_event_type(context: Context, expected: str) -> None:
    assert len(context.entries) > 0
    for e in context.entries:
        assert e.event_type == expected


@then("no entries should be returned")
def step_no_entries(context: Context) -> None:
    assert len(context.entries) == 0


@then("exactly {count:d} entries should be returned")
def step_exact_count(context: Context, count: int) -> None:
    assert len(context.entries) == count, (
        f"Expected {count}, got {len(context.entries)}"
    )


@then('the first entry should have plan_id "{expected}"')
def step_first_plan_id(context: Context, expected: str) -> None:
    actual = context.entries[0].plan_id
    assert actual == expected, (
        f"Expected first entry plan_id={expected!r}, got {actual!r} "
        f"(created_at={context.entries[0].created_at!r})"
    )


@then('the fetched entry should have event_type "{expected}"')
def step_fetched_event_type(context: Context, expected: str) -> None:
    assert context.fetched is not None
    assert context.fetched.event_type == expected


@then('the fetched entry should have plan_id "{expected}"')
def step_fetched_plan_id(context: Context, expected: str) -> None:
    assert context.fetched is not None
    assert context.fetched.plan_id == expected


@then("no entry should be returned")
def step_no_entry(context: Context) -> None:
    assert context.fetched is None


@then("the count should be {expected:d}")
def step_count_is(context: Context, expected: int) -> None:
    assert context.service.count() == expected


@then("the audit log should be empty")
def step_log_empty(context: Context) -> None:
    assert context.service.count() == 0


@then("the audit log should not be empty")
def step_log_not_empty(context: Context) -> None:
    assert context.service.count() > 0


@then("the prune result should be {expected:d}")
def step_prune_result(context: Context, expected: int) -> None:
    assert context.prune_result == expected


@then("audit_retention_days should be {expected:d}")
def step_retention_days(context: Context, expected: int) -> None:
    assert context.settings.audit_retention_days == expected


@then('the dict should contain keys "{k1}" "{k2}" "{k3}" "{k4}" "{k5}" "{k6}"')
def step_dict_keys(
    context: Context, k1: str, k2: str, k3: str, k4: str, k5: str, k6: str
) -> None:
    d = context.entry_dict
    for key in (k1, k2, k3, k4, k5, k6):
        assert key in d, f"Key {key!r} not in dict"


@then('the dict details key "{key}" should be {expected:d}')
def step_dict_details_int(context: Context, key: str, expected: int) -> None:
    assert context.entry_dict["details"][key] == expected


@then("the entry plan_id should be None")
def step_plan_id_none(context: Context) -> None:
    assert context.entry.plan_id is None


@then("the entry project_name should be None")
def step_project_none(context: Context) -> None:
    assert context.entry.project_name is None


@then("the entry actor_name should be None")
def step_actor_none(context: Context) -> None:
    assert context.entry.actor_name is None


@then("the entry user_identity should be None")
def step_user_none(context: Context) -> None:
    assert context.entry.user_identity is None


@then("a TypeError should be raised for audit service")
def step_type_error(context: Context) -> None:
    assert isinstance(context.error, TypeError)


@then("a ValueError should be raised for audit service")
def step_value_error(context: Context) -> None:
    assert isinstance(context.error, ValueError)


@then('the audit error message should contain "{text}"')
def step_audit_error_message_contains(context: Context, text: str) -> None:
    assert text in str(context.error), f"Expected {text!r} in {context.error!s}"


# ── Context manager / owned session steps ─────────────────────


@given("a fresh audit service with owned session")
def step_fresh_owned_session(context: Context) -> None:
    settings = _make_settings(database_url="sqlite:///:memory:", audit_async=False)
    context.service = AuditService(settings=settings)
    context.entry = None
    context.entries = []


@when('I use the service as a context manager to record "{event_type}"')
def step_context_manager_record(context: Context, event_type: str) -> None:
    with context.service as svc:
        context.entry = svc.record(event_type=event_type)


@then('the audit context manager result should have event_type "{expected}"')
def step_context_mgr_event_type(context: Context, expected: str) -> None:
    assert context.entry.event_type == expected


@when("I close the service")
def step_close_service(context: Context) -> None:
    context.error = None
    try:
        context.service.close()
    except Exception as exc:
        context.error = exc


@then("no audit error should be raised")
def step_no_audit_error(context: Context) -> None:
    assert context.error is None


@given("a fresh audit service with a corrupt details row")
def step_fresh_corrupt_details(context: Context) -> None:
    context.service = _fresh_service()
    row = AuditLogModel(
        event_type="plan_applied",
        details="NOT-VALID-JSON{{{",
        created_at=datetime.now(tz=UTC).strftime(_TIMESTAMP_FMT),
    )
    context.service._session.add(row)
    context.service._session.commit()
    context.entry = None
    context.entries = []


@then("the first listed entry details should be empty")
def step_first_listed_empty_details(context: Context) -> None:
    assert context.entries[0].details == {}


# ── CLI command steps ─────────────────────────────────────────


def _get_cli_runner_and_app(context):
    """Set up a CliRunner and patch _get_audit_service."""
    from unittest.mock import patch

    from typer.testing import CliRunner

    from cleveragents.cli.commands.audit import app as audit_app

    runner = CliRunner()

    patcher = patch(
        "cleveragents.cli.commands.audit._get_audit_service",
        return_value=context.service,
    )
    patcher.start()
    if not hasattr(context, "_cleanup_handlers"):
        context._cleanup_handlers = []
    context._cleanup_handlers.append(patcher.stop)

    return runner, audit_app


def _run_audit_cli(context, args):
    """Run an audit CLI command and set both cli_result and output."""
    runner, app = _get_cli_runner_and_app(context)
    result = runner.invoke(app, args)
    context.cli_result = result
    context.output = result.output
    return result


@when("I invoke the audit CLI list command")
def step_cli_list(context: Context) -> None:
    _run_audit_cli(context, ["list"])


@when(
    'I invoke the audit CLI list command with plan "{plan}" '
    'and type "{etype}" and project "{project}" and limit {limit:d}'
)
def step_cli_list_filters(
    context: Context, plan: str, etype: str, project: str, limit: int
) -> None:
    _run_audit_cli(
        context,
        [
            "list",
            "--plan",
            plan,
            "--type",
            etype,
            "--project",
            project,
            "--limit",
            str(limit),
        ],
    )


@when("I invoke the audit CLI show command with the entry ID")
def step_cli_show(context: Context) -> None:
    _run_audit_cli(context, ["show", str(context.entry.id)])


@when("I invoke the audit CLI show command with ID {audit_id:d}")
def step_cli_show_nonexistent(context: Context, audit_id: int) -> None:
    _run_audit_cli(context, ["show", str(audit_id)])


@when("I invoke the audit CLI count command")
def step_cli_count(context: Context) -> None:
    _run_audit_cli(context, ["count"])


@when("I invoke the audit CLI prune command with days {days:d} and yes flag")
def step_cli_prune_yes(context: Context, days: int) -> None:
    _run_audit_cli(context, ["prune", "--days", str(days), "--yes"])


@when("I invoke the audit CLI prune command with days {days:d} and no confirmation")
def step_cli_prune_no_confirm(context: Context, days: int) -> None:
    runner, app = _get_cli_runner_and_app(context)
    result = runner.invoke(app, ["prune", "--days", str(days)], input="n\n")
    context.cli_result = result
    context.output = result.output


# NOTE: "the CLI exit code should be {code:d}" reused from garbage_collection_cli_steps.py
# NOTE: 'the CLI output should contain "{text}"' reused from cli_steps.py


# ── Logging configuration steps ───────────────────────────────


@when('I configure structlog for development with level "{level}"')
def step_configure_structlog_dev(context: Context, level: str) -> None:
    from cleveragents.config.logging import configure_structlog

    context.error = None
    try:
        configure_structlog(env="development", log_level=level)
        context.structlog_configured = True
    except Exception as exc:
        context.error = exc
        context.structlog_configured = False


@when('I configure structlog for production with level "{level}"')
def step_configure_structlog_prod(context: Context, level: str) -> None:
    from cleveragents.config.logging import configure_structlog

    context.error = None
    try:
        configure_structlog(env="production", log_level=level)
        context.structlog_configured = True
    except Exception as exc:
        context.error = exc
        context.structlog_configured = False


@when('I configure structlog with invalid level "{level}"')
def step_configure_structlog_invalid(context: Context, level: str) -> None:
    from cleveragents.config.logging import configure_structlog

    try:
        configure_structlog(log_level=level)
        context.error = None
    except ValueError as exc:
        context.error = exc


@when('I call get_logger with name "{name}"')
def step_get_logger(context: Context, name: str) -> None:
    from cleveragents.config.logging import get_logger

    context.logger = get_logger(name)


@then("structlog should be configured")
def step_structlog_configured(context: Context) -> None:
    assert context.structlog_configured is True


@then("a structlog logger should be returned")
def step_structlog_logger_returned(context: Context) -> None:
    assert context.logger is not None


# ── Redaction edge case steps ─────────────────────────────────


@when("I call set_show_secrets with a non-bool value")
def step_set_show_secrets_non_bool(context: Context) -> None:
    from cleveragents.shared.redaction import set_show_secrets

    try:
        set_show_secrets("yes")  # type: ignore[arg-type]
        context.error = None
    except TypeError as exc:
        context.error = exc


@when("I check if empty string is sensitive")
def step_is_sensitive_empty(context: Context) -> None:
    from cleveragents.shared.redaction import is_sensitive_key

    context.result = is_sensitive_key("")


@then("the sensitivity result should be false")
def step_sensitivity_result_false(context: Context) -> None:
    assert context.result is False


@when("I mask an empty database URL")
def step_mask_empty_url(context: Context) -> None:
    from cleveragents.shared.redaction import mask_database_url

    context.masked_url = mask_database_url("")


@then("the audit masked URL should be empty")
def step_audit_masked_url_empty(context: Context) -> None:
    assert context.masked_url == ""


@when("I register an empty secret pattern")
def step_register_empty_pattern(context: Context) -> None:
    from cleveragents.shared.redaction import register_pattern

    try:
        register_pattern("")
        context.error = None
    except ValueError as exc:
        context.error = exc


@when("I run the structlog processor with show_secrets enabled")
def step_processor_show_secrets(context: Context) -> None:
    from cleveragents.shared.redaction import (
        secrets_masking_processor,
        set_show_secrets,
    )

    set_show_secrets(True)
    try:
        event = {"event": "test", "api_key": "sk-secret123"}
        context.processor_result = secrets_masking_processor(None, "info", event)
    finally:
        set_show_secrets(False)


@then("the event dict should be returned unchanged")
def step_event_unchanged(context: Context) -> None:
    assert context.processor_result["api_key"] == "sk-secret123"


@when("I run the structlog processor with a nested dict value")
def step_processor_nested_dict(context: Context) -> None:
    from cleveragents.shared.redaction import secrets_masking_processor

    event = {
        "event": "test",
        "config": {"api_key": "sk-secret123", "name": "test"},
    }
    context.processor_result = secrets_masking_processor(None, "info", event)


@then("the nested dict should be redacted")
def step_nested_dict_redacted(context: Context) -> None:
    config = context.processor_result["config"]
    assert config["api_key"] == "***REDACTED***"
    assert config["name"] == "test"