cleveragents-core/robot/helper_security_eval_scanner.py

"""Helper utilities for SEC1 config security scanner Robot integration tests.

Covers the security_scanner module: pattern detection, line-number reporting,
validation gate, and comment skipping.

Each command prints structured output lines that Robot Framework can
independently verify:
  VIOLATIONS=<count>
  TOKEN=<token>
  SEVERITY=<severity>
  LINE=<line_number>
  STATUS=ok|fail
"""

from __future__ import annotations

import sys

from cleveragents.config.security_scanner import (
    Severity,
    scan_content,
    validate_config_safety,
)
from cleveragents.core.exceptions import ConfigurationError


def _detect_eval() -> None:
    """Scanner should detect eval() in config content."""
    content = 'transform: eval("1+1")\n'
    violations = scan_content(content, "test.yaml")
    print(f"VIOLATIONS={len(violations)}")
    for v in violations:
        print(f"TOKEN={v.token}")
        print(f"SEVERITY={v.severity.name}")
    assert any(v.token == "eval(" for v in violations), "eval( not detected"
    assert all(
        v.severity == Severity.CRITICAL for v in violations if v.token == "eval("
    )
    print("STATUS=ok")


def _detect_exec() -> None:
    """Scanner should detect exec() in config content."""
    content = 'code: exec("import os")\n'
    violations = scan_content(content, "test.yaml")
    print(f"VIOLATIONS={len(violations)}")
    for v in violations:
        print(f"TOKEN={v.token}")
    assert any(v.token == "exec(" for v in violations), "exec( not detected"
    print("STATUS=ok")


def _detect_import() -> None:
    """Scanner should detect __import__ in config content."""
    content = "loader: __import__('os')\n"
    violations = scan_content(content, "test.yaml")
    print(f"VIOLATIONS={len(violations)}")
    for v in violations:
        print(f"TOKEN={v.token}")
    assert any(v.token == "__import__" for v in violations), "__import__ not detected"
    print("STATUS=ok")


def _detect_os_system() -> None:
    """Scanner should detect os.system in config content."""
    content = "run: os.system('rm -rf /')\n"
    violations = scan_content(content, "test.yaml")
    print(f"VIOLATIONS={len(violations)}")
    for v in violations:
        print(f"TOKEN={v.token}")
    assert any(v.token == "os.system" for v in violations), "os.system not detected"
    print("STATUS=ok")


def _detect_subprocess() -> None:
    """Scanner should detect subprocess calls in config content."""
    content = "cmd: subprocess.call(['ls'])\n"
    violations = scan_content(content, "test.yaml")
    print(f"VIOLATIONS={len(violations)}")
    for v in violations:
        print(f"TOKEN={v.token}")
    assert any(v.token == "subprocess.call" for v in violations), (
        "subprocess.call not detected"
    )
    print("STATUS=ok")


def _clean_config() -> None:
    """Scanner should return zero violations for safe content."""
    content = "name: my-project\nversion: 1.0.0\ndescription: A safe config\n"
    violations = scan_content(content, "test.yaml")
    print(f"VIOLATIONS={len(violations)}")
    assert len(violations) == 0, f"Expected 0 violations, got {len(violations)}"
    print("STATUS=ok")


def _line_numbers() -> None:
    """Scanner should report correct line numbers."""
    content = "safe: true\nunsafe: eval('boom')\nother: ok\n"
    violations = scan_content(content, "test.yaml")
    eval_violations = [v for v in violations if v.token == "eval("]
    print(f"VIOLATIONS={len(eval_violations)}")
    for v in eval_violations:
        print(f"LINE={v.line_number}")
    assert len(eval_violations) == 1, (
        f"Expected 1 eval violation, got {len(eval_violations)}"
    )
    assert eval_violations[0].line_number == 2, (
        f"Expected line 2, got {eval_violations[0].line_number}"
    )
    print("STATUS=ok")


def _validate_raises() -> None:
    """validate_config_safety should raise ConfigurationError on violations."""
    content = 'action: exec("bad")\n'
    try:
        validate_config_safety(content, "test.yaml")
        print("STATUS=fail")
        print("REASON=Expected ConfigurationError")
        return
    except ConfigurationError:
        pass
    # Clean content should not raise
    validate_config_safety("name: safe\n", "test.yaml")
    print("STATUS=ok")


def _skip_comments() -> None:
    """Scanner should skip pure-comment lines."""
    content = "# eval('this is a comment')\nname: safe\n"
    violations = scan_content(content, "test.yaml")
    print(f"VIOLATIONS={len(violations)}")
    assert len(violations) == 0, f"Expected 0 violations, got {len(violations)}"
    print("STATUS=ok")


def _detect_template() -> None:
    """Scanner should detect template directives."""
    content = "template: {{ user_input }}\n"
    violations = scan_content(content, "test.yaml")
    print(f"VIOLATIONS={len(violations)}")
    for v in violations:
        print(f"TOKEN={v.token}")
    assert any(v.token == "{{" for v in violations), "{{ not detected"
    print("STATUS=ok")


def _inline_comment() -> None:
    """Scanner should skip patterns in inline comments."""
    content = "name: safe-project  # eval('just a note')\n"
    violations = scan_content(content, "test.yaml")
    print(f"VIOLATIONS={len(violations)}")
    assert len(violations) == 0, f"Expected 0 violations, got {len(violations)}"
    print("STATUS=ok")


def main() -> None:
    if len(sys.argv) < 2:
        raise SystemExit("Expected command argument")
    command = sys.argv[1]
    commands = {
        "detect-eval": _detect_eval,
        "detect-exec": _detect_exec,
        "detect-import": _detect_import,
        "detect-os-system": _detect_os_system,
        "detect-subprocess": _detect_subprocess,
        "clean-config": _clean_config,
        "line-numbers": _line_numbers,
        "validate-raises": _validate_raises,
        "skip-comments": _skip_comments,
        "detect-template": _detect_template,
        "inline-comment": _inline_comment,
    }
    if command not in commands:
        raise SystemExit(f"Unknown command: {command}")
    commands[command]()


if __name__ == "__main__":
    main()