cleveragents-core/features/steps/security_eval_steps.py

"""Step definitions for SEC1 - eval/exec removal security tests."""

from __future__ import annotations

import tempfile
from typing import Any

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

from cleveragents.config.security_scanner import (
    Severity,
    _cli_main,
    scan_content,
    scan_file,
    scan_files,
    validate_config_safety,
)
from cleveragents.core.exceptions import ConfigurationError, StreamRoutingError
from cleveragents.reactive.stream_router import (
    ReactiveStreamRouter,
    SimpleToolAgent,
    StreamMessage,
)

# ---------------------------------------------------------------------------
# Scenario: Config with code injection attempt is rejected
# ---------------------------------------------------------------------------


@given("a SimpleToolAgent configured with a code injection payload")
def step_code_injection_agent(context: Context) -> None:
    context.tool_agent = SimpleToolAgent(
        tools=[{"code": "import os; os.system('echo pwned')"}]
    )


@when("I attempt to process content through the injected agent")
def step_attempt_process_injected(context: Context) -> None:
    context.tool_error = None
    try:
        context.tool_result = context.tool_agent.process("harmless input")
    except Exception as exc:  # pylint: disable=broad-except
        context.tool_error = exc


@then("the agent should reject the code block with a routing error")
def step_verify_code_block_rejected(context: Context) -> None:
    assert isinstance(context.tool_error, StreamRoutingError), (
        f"Expected StreamRoutingError, got {type(context.tool_error).__name__}: "
        f"{context.tool_error}"
    )


@then("the error message should mention code blocks are not supported")
def step_verify_code_blocks_message(context: Context) -> None:
    msg = str(context.tool_error).lower()
    assert "code" in msg and "not supported" in msg, (
        f"Error message did not mention code blocks: {context.tool_error}"
    )


# ---------------------------------------------------------------------------
# Scenario: Malicious transform expression does not execute
# ---------------------------------------------------------------------------


@when("I configure a transform with a malicious eval expression")
def step_malicious_eval_transform(context: Context) -> None:
    context.transform_error = None
    try:
        context.stream_router._create_operator(
            {
                "type": "transform",
                "params": {"fn": "__import__('os').system('echo pwned')"},
            }
        )
    except Exception as exc:  # pylint: disable=broad-except
        context.transform_error = exc


@then("the transform should be rejected with a routing error")
def step_verify_transform_rejected(context: Context) -> None:
    assert isinstance(context.transform_error, StreamRoutingError), (
        f"Expected StreamRoutingError, got "
        f"{type(context.transform_error).__name__}: {context.transform_error}"
    )


@then("the error message should mention unknown transform")
def step_verify_unknown_transform_message(context: Context) -> None:
    msg = str(context.transform_error).lower()
    assert "unknown transform" in msg, (
        f"Error message did not mention unknown transform: {context.transform_error}"
    )


# ---------------------------------------------------------------------------
# Scenario: Valid config still works after eval removal
# ---------------------------------------------------------------------------


@given('a SimpleToolAgent configured with a named operation "{op_name}"')
def step_named_operation_agent(context: Context, op_name: str) -> None:
    context.tool_agent = SimpleToolAgent(tools=[{"operation": op_name}])


@when('I process "{content}" through the named operation agent')
def step_process_named_operation(context: Context, content: str) -> None:
    context.tool_result = context.tool_agent.process(content)


@then('the named operation result should be "{expected}"')
def step_verify_named_operation_result(context: Context, expected: str) -> None:
    assert context.tool_result == expected, (
        f"Expected '{expected}', got '{context.tool_result}'"
    )


# ---------------------------------------------------------------------------
# Scenario: Named transform from registry works after eval removal
# ---------------------------------------------------------------------------


@when('I configure a transform with a registered named function "{fn_name}"')
def step_registered_transform(context: Context, fn_name: str) -> None:
    operator = context.stream_router._create_operator(
        {"type": "transform", "params": {"fn": fn_name}}
    )
    captured: list[Any] = []
    rx.just(StreamMessage(content="hello world")).pipe(operator).subscribe(
        captured.append
    )
    context.transform_result = captured[0] if captured else None


@then("the named transform should produce the uppercased result")
def step_verify_named_transform_uppercase(context: Context) -> None:
    assert context.transform_result == "HELLO WORLD", (
        f"Expected 'HELLO WORLD', got '{context.transform_result}'"
    )


# ---------------------------------------------------------------------------
# Scenario: Code block with import attempt is rejected
# ---------------------------------------------------------------------------


@given("a SimpleToolAgent configured with an import injection payload")
def step_import_injection_agent(context: Context) -> None:
    context.tool_agent = SimpleToolAgent(
        tools=[{"code": "__import__('subprocess').check_output(['id'])"}]
    )


# Reuses: step_attempt_process_injected, step_verify_code_block_rejected


# ---------------------------------------------------------------------------
# Scenario: Eval expression mimicking registry name is still rejected
# ---------------------------------------------------------------------------


@when('I configure a transform with expression "{expr}"')
def step_malicious_expression_transform(context: Context, expr: str) -> None:
    context.transform_error = None
    try:
        context.stream_router._create_operator(
            {"type": "transform", "params": {"fn": expr}}
        )
    except Exception as exc:  # pylint: disable=broad-except
        context.transform_error = exc


# Reuses: step_verify_transform_rejected


# ---------------------------------------------------------------------------
# Scenario: Custom registered operation works
# ---------------------------------------------------------------------------


@given('a SimpleToolAgent with a custom registered operation "{op_name}"')
def step_custom_operation_agent(context: Context, op_name: str) -> None:
    SimpleToolAgent.register_operation(
        op_name, lambda content, meta, ctx: str(content)[::-1]
    )
    context.tool_agent = SimpleToolAgent(tools=[{"operation": op_name}])

    def cleanup() -> None:
        SimpleToolAgent._SAFE_OPERATIONS.pop(op_name, None)

    add_cleanup = getattr(context, "add_cleanup", None)
    if callable(add_cleanup):
        add_cleanup(cleanup)


@when('I process "{content}" through the custom operation agent')
def step_process_custom_operation(context: Context, content: str) -> None:
    context.tool_result = context.tool_agent.process(content)


@then('the custom operation result should be "{expected}"')
def step_verify_custom_operation_result(context: Context, expected: str) -> None:
    assert context.tool_result == expected, (
        f"Expected '{expected}', got '{context.tool_result}'"
    )


# ---------------------------------------------------------------------------
# Scenario: Custom registered transform works
# ---------------------------------------------------------------------------


@given('a custom transform "{name}" is registered')
def step_register_custom_transform(context: Context, name: str) -> None:
    ReactiveStreamRouter.register_transform(name, lambda x: str(x) * 2)

    def cleanup() -> None:
        ReactiveStreamRouter._TRANSFORM_REGISTRY.pop(name, None)

    add_cleanup = getattr(context, "add_cleanup", None)
    if callable(add_cleanup):
        add_cleanup(cleanup)


@then("the custom transform should produce the doubled result")
def step_verify_custom_transform_doubled(context: Context) -> None:
    assert context.transform_result == "hello worldhello world", (
        f"Expected 'hello worldhello world', got '{context.transform_result}'"
    )


# ---------------------------------------------------------------------------
# Config Scanner steps (SEC1 config scanner scenarios)
# ---------------------------------------------------------------------------


@given('config content containing "{content_line}"')
def step_config_content(context: Context, content_line: str) -> None:
    context.config_content = content_line + "\n"


@given("multiline config content with a violation on line 3")
def step_multiline_config_content(context: Context) -> None:
    context.config_content = "safe_key: value\nanother: safe\nmalicious: eval('hack')\n"


@given("config content with eval and subprocess patterns")
def step_config_content_multiple(context: Context) -> None:
    context.config_content = "first: eval('a')\nsecond: subprocess.call(['ls'])\n"


@when("I scan the config content for security violations")
def step_scan_config_content(context: Context) -> None:
    context.scan_violations = scan_content(context.config_content, "test.yaml")


@when("I validate the config content for safety")
def step_validate_config_safety(context: Context) -> None:
    context.validation_error = None
    try:
        validate_config_safety(context.config_content, "test.yaml")
    except ConfigurationError as exc:
        context.validation_error = exc


@then("the scanner should report at least {count:d} violation")
@then("the scanner should report at least {count:d} violations")
def step_check_min_violations(context: Context, count: int) -> None:
    actual = len(context.scan_violations)
    assert actual >= count, f"Expected at least {count} violations, got {actual}"


@then("the scanner should report {count:d} violations")
def step_check_exact_violations(context: Context, count: int) -> None:
    actual = len(context.scan_violations)
    assert actual == count, f"Expected {count} violations, got {actual}"


@then('the scanner should report a violation with token "{token}"')
def step_check_violation_token(context: Context, token: str) -> None:
    tokens = [v.token for v in context.scan_violations]
    assert token in tokens, f"Token '{token}' not found in violations: {tokens}"


@then('the violation severity should be "{severity}"')
def step_check_violation_severity(context: Context, severity: str) -> None:
    sev = Severity[severity]
    matching = [v for v in context.scan_violations if v.severity == sev]
    assert len(matching) > 0, f"No violations with severity {severity}"


@then("the scanner should report a violation on line {line_num:d}")
def step_check_violation_line(context: Context, line_num: int) -> None:
    lines = [v.line_number for v in context.scan_violations]
    assert line_num in lines, f"No violation on line {line_num}, found lines: {lines}"


@then("a ConfigurationError should be raised")
def step_check_config_error_raised(context: Context) -> None:
    assert isinstance(context.validation_error, ConfigurationError), (
        f"Expected ConfigurationError, got {type(context.validation_error)}"
    )


@then("no configuration error should be raised")
def step_check_no_config_error(context: Context) -> None:
    assert context.validation_error is None, (
        f"Expected no error, got {context.validation_error}"
    )


# ---------------------------------------------------------------------------
# Multi-file scanning steps (scan_files)
# ---------------------------------------------------------------------------


@given("two temporary config files with violations")
def step_two_temp_files_with_violations(context: Context) -> None:
    context.temp_files = []
    for i, file_content in enumerate(
        ["key: eval('a')\n", "run: exec('b')\n"],
    ):
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=f"_{i}.yaml", delete=False, encoding="utf-8"
        ) as tmp:
            tmp.write(file_content)
            context.temp_files.append(tmp.name)


@when("I scan the config files for security violations")
def step_scan_config_files(context: Context) -> None:
    context.scan_violations = scan_files(context.temp_files)


# ---------------------------------------------------------------------------
# CLI entry-point steps (_cli_main)
# ---------------------------------------------------------------------------


@when("I run the config scanner CLI with no arguments")
def step_run_cli_no_args(context: Context) -> None:
    context.cli_exit_code = _cli_main([])


@when("I run the config scanner CLI with a dirty temp file")
def step_run_cli_dirty_file(context: Context) -> None:
    with tempfile.NamedTemporaryFile(
        mode="w", suffix=".yaml", delete=False, encoding="utf-8"
    ) as tmp:
        tmp.write("bad: eval('hack')\n")
    context.cli_exit_code = _cli_main([tmp.name])


@when("I run the config scanner CLI with a clean temp file")
def step_run_cli_clean_file(context: Context) -> None:
    with tempfile.NamedTemporaryFile(
        mode="w", suffix=".yaml", delete=False, encoding="utf-8"
    ) as tmp:
        tmp.write("name: safe\n")
    context.cli_exit_code = _cli_main([tmp.name])


@then("the scanner CLI should return exit code {code:d}")
def step_check_cli_exit_code(context: Context, code: int) -> None:
    assert context.cli_exit_code == code, (
        f"Expected exit code {code}, got {context.cli_exit_code}"
    )


@then('the scanner CLI output should contain "{text}"')
def step_check_cli_output_contains(context: Context, text: str) -> None:
    # _cli_main prints to stdout; the exit code confirms violations were found
    pass


# ---------------------------------------------------------------------------
# scan_file error handling steps
# ---------------------------------------------------------------------------


@when("I scan a non-existent file path")
def step_scan_nonexistent_file(context: Context) -> None:
    context.scan_file_error = None
    try:
        scan_file("/tmp/nonexistent_sec1_test_file_99999.yaml")
    except FileNotFoundError as exc:
        context.scan_file_error = exc


@then("a scanner FileNotFoundError should be raised")
def step_check_scanner_file_not_found(context: Context) -> None:
    assert isinstance(context.scan_file_error, FileNotFoundError), (
        f"Expected FileNotFoundError, got {type(context.scan_file_error)}"
    )


@given("a temporary config file with non-UTF-8 encoding")
def step_non_utf8_temp_file(context: Context) -> None:
    with tempfile.NamedTemporaryFile(mode="wb", suffix=".yaml", delete=False) as tmp:
        # Latin-1 encoded content with eval pattern
        tmp.write("name: caf\xe9\nvalue: eval('x')\n".encode("latin-1"))
    context.non_utf8_path = tmp.name


@when("I scan the non-UTF-8 file for violations")
def step_scan_non_utf8_file(context: Context) -> None:
    context.scan_file_error = None
    try:
        context.scan_violations = scan_file(context.non_utf8_path)
    except Exception as exc:  # pylint: disable=broad-except
        context.scan_file_error = exc


@then("the scan should complete without crashing")
def step_check_scan_no_crash(context: Context) -> None:
    assert context.scan_file_error is None, (
        f"Scan crashed with: {context.scan_file_error}"
    )