OpenHands/tests/unit/test_logging.py

import logging
from io import StringIO

import pytest

from openhands.core.config import AppConfig, LLMConfig
from openhands.core.logger import openhands_logger as openhands_logger


@pytest.fixture
def test_handler():
    stream = StringIO()
    handler = logging.StreamHandler(stream)
    handler.setLevel(logging.INFO)
    formatter = logging.Formatter('%(message)s')
    handler.setFormatter(formatter)
    openhands_logger.addHandler(handler)
    yield openhands_logger, stream
    openhands_logger.removeHandler(handler)


def test_openai_api_key_masking(test_handler):
    logger, stream = test_handler

    api_key = 'sk-1234567890abcdef'
    message = f"OpenAI API key: api_key='{api_key}'and there's some stuff here"
    logger.info(message)
    log_output = stream.getvalue()
    assert "api_key='******'" in log_output
    assert api_key not in log_output


def test_azure_api_key_masking(test_handler):
    logger, stream = test_handler
    api_key = '1a2b3c4d5e6f7g8h9i0j'
    message = f"Azure API key: api_key='{api_key}' and chatty chat with ' and \" and '"
    logger.info(message)
    log_output = stream.getvalue()
    assert "api_key='******'" in log_output
    assert api_key not in log_output


def test_google_vertex_api_key_masking(test_handler):
    logger, stream = test_handler
    api_key = 'AIzaSyA1B2C3D4E5F6G7H8I9J0'
    message = f"Google Vertex API key: api_key='{api_key}' or not"
    logger.info(message)
    log_output = stream.getvalue()
    assert "api_key='******'" in log_output
    assert api_key not in log_output


def test_anthropic_api_key_masking(test_handler):
    logger, stream = test_handler
    api_key = 'sk-ant-1234567890abcdef-some-more-stuff-here'
    message = f"Anthropic API key: api_key='{api_key}' and there's some 'stuff' here"
    logger.info(message)
    log_output = stream.getvalue()
    assert "api_key='******'" in log_output
    assert api_key not in log_output


def test_llm_config_attributes_masking(test_handler):
    logger, stream = test_handler
    llm_config = LLMConfig(
        api_key='sk-abc123',
        aws_access_key_id='AKIAIOSFODNN7EXAMPLE',
        aws_secret_access_key='wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY',
    )
    logger.info(f'LLM Config: {llm_config}')
    log_output = stream.getvalue()
    assert "api_key='******'" in log_output
    assert "aws_access_key_id='******'" in log_output
    assert "aws_secret_access_key='******'" in log_output
    assert 'sk-abc123' not in log_output
    assert 'AKIAIOSFODNN7EXAMPLE' not in log_output
    assert 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY' not in log_output


def test_app_config_attributes_masking(test_handler):
    logger, stream = test_handler
    app_config = AppConfig(e2b_api_key='e2b-xyz789')
    logger.info(f'App Config: {app_config}')
    log_output = stream.getvalue()
    assert "e2b_api_key='******'" in log_output
    assert 'github_token' not in log_output
    assert 'e2b-xyz789' not in log_output
    assert 'ghp_abcdefghijklmnopqrstuvwxyz' not in log_output


def test_sensitive_env_vars_masking(test_handler):
    logger, stream = test_handler
    sensitive_data = {
        'API_KEY': 'API_KEY_VALUE',
        'AWS_ACCESS_KEY_ID': 'AWS_ACCESS_KEY_ID_VALUE',
        'AWS_SECRET_ACCESS_KEY': 'AWS_SECRET_ACCESS_KEY_VALUE',
        'E2B_API_KEY': 'E2B_API_KEY_VALUE',
        'GITHUB_TOKEN': 'GITHUB_TOKEN_VALUE',
        'JWT_SECRET': 'JWT_SECRET_VALUE',
    }

    log_message = ' '.join(
        f"{attr}='{value}'" for attr, value in sensitive_data.items()
    )
    logger.info(log_message)

    log_output = stream.getvalue()
    for attr, value in sensitive_data.items():
        assert f"{attr}='******'" in log_output
        assert value not in log_output


def test_special_cases_masking(test_handler):
    logger, stream = test_handler
    sensitive_data = {
        'LLM_API_KEY': 'LLM_API_KEY_VALUE',
        'SANDBOX_ENV_GITHUB_TOKEN': 'SANDBOX_ENV_GITHUB_TOKEN_VALUE',
    }

    log_message = ' '.join(
        f"{attr}={value} with no single quotes' and something"
        for attr, value in sensitive_data.items()
    )
    logger.info(log_message)

    log_output = stream.getvalue()
    for attr, value in sensitive_data.items():
        assert f"{attr}='******'" in log_output
        assert value not in log_output