Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,38 @@
from openhands.sdk.utils.pydantic_secrets import serialize_secret, validate_secret


class _SecretProbeCipher:
"""Fake cipher that flags secret serialization without needing a real key.

Passed as ``context={"cipher": probe}`` this forces every secret field's
serializer down the "encrypted" branch (see ``resolve_expose_mode``),
reusing the real serialization pipeline to detect secret-bearing fields
instead of hand-walking the model for ``SecretStr`` instances. That
matters for fields like ``AgentContext.secrets``, whose bare-string
entries are plain ``str`` at rest and only become secret-shaped inside
their own field serializer at dump time -- a value-type walk can't see
that, but reusing the pipeline does.
"""

def __init__(self) -> None:
self.found = False

def try_decrypt_str(self, raw: str) -> str | None: # noqa: ARG002
return None

def encrypt(self, value: SecretStr) -> str:
if value.get_secret_value():
self.found = True
return ""


def _contains_secret_value(model: BaseModel) -> bool:
"""Check if serializing `model` would touch any secret-bearing field."""
probe = _SecretProbeCipher()
model.model_dump(mode="json", context={"cipher": probe})
return probe.found


class SettingsUpdatePayload(TypedDict, total=False):
"""Typed payload for PersistedSettings.update() method.

Expand Down Expand Up @@ -180,6 +212,17 @@ def llm_api_key_is_set(self) -> bool:
)
return bool(secret_value and secret_value.strip())

@property
def has_any_secret(self) -> bool:
Comment thread
simonrosenberg marked this conversation as resolved.
"""Check if these settings contain any secret value anywhere.

Broader than ``llm_api_key_is_set``: walks the whole ``agent_settings``
tree (MCP server env/headers, ``critic_api_key``, provider creds,
``agent_context.secrets``, ...) rather than checking a fixed field
list, so it stays correct as new secret-bearing fields are added.
"""
return _contains_secret_value(self.agent_settings)

def update(
self,
payload: SettingsUpdatePayload,
Expand Down Expand Up @@ -412,6 +455,11 @@ class Secrets(BaseModel):

model_config = ConfigDict(frozen=True)

@property
def has_any_secret(self) -> bool:
"""Check if these secrets contain any non-empty value."""
return _contains_secret_value(self)

def get_env_vars(self) -> dict[str, str]:
"""Get secrets as environment variables dict.

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -373,7 +373,7 @@ def save(self, settings: PersistedSettings) -> None:
else:
context = {"expose_secrets": "plaintext"}
# Warn about plaintext secret storage (only if secrets exist)
if settings.llm_api_key_is_set:
if settings.has_any_secret:
logger.warning(
"Saving settings with secrets in PLAINTEXT (no cipher configured). "
"Configure OH_SECRET_KEY for production deployments."
Expand Down Expand Up @@ -542,7 +542,7 @@ def _save_with_versions(
else:
context = {"expose_secrets": "plaintext"}
# Warn about plaintext secret storage (only if secrets exist)
if secrets.custom_secrets:
if secrets.has_any_secret:
logger.warning(
"Saving secrets in PLAINTEXT (no cipher configured). "
"Configure OH_SECRET_KEY for production deployments."
Expand Down
145 changes: 145 additions & 0 deletions tests/agent_server/test_persistence_secret_detection.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
"""Tests for has_any_secret: detect secret-bearing fields via the real
serialization pipeline, not a hardcoded field checklist.
"""

import logging
import tempfile
from base64 import urlsafe_b64encode
from pathlib import Path

import pytest
from pydantic import SecretStr

from openhands.agent_server.persistence import (
CustomSecret,
FileSecretsStore,
FileSettingsStore,
PersistedSettings,
Secrets,
)
from openhands.sdk.utils.cipher import Cipher


@pytest.fixture
def persistence_dir():
with tempfile.TemporaryDirectory() as tmpdir:
yield Path(tmpdir)


@pytest.fixture
def cipher():
return Cipher(urlsafe_b64encode(b"a" * 32).decode("ascii"))


def test_has_any_secret_detects_llm_api_key():
settings = PersistedSettings.model_validate(
{"agent_settings": {"llm": {"model": "gpt-4o", "api_key": "sk-test"}}}
)
assert settings.has_any_secret


def test_has_any_secret_detects_critic_api_key():
"""critic_api_key is a separate secret from llm.api_key."""
settings = PersistedSettings.model_validate(
{"agent_settings": {"verification": {"critic_api_key": "sk-critic"}}}
)
assert settings.has_any_secret


def test_has_any_secret_detects_mcp_server_secret():
settings = PersistedSettings.model_validate(
{
"agent_settings": {
"mcp_config": {
"mcpServers": {
"github": {
"command": "uvx",
"args": ["mcp-server-github"],
"env": {"GITHUB_TOKEN": "ghp-test"},
}
}
}
}
}
)
assert settings.has_any_secret


def test_has_any_secret_detects_agent_context_secret():
"""agent_context.secrets values are plain str at rest (only secret-shaped
at serialize time via their own field serializer), so a naive
isinstance(v, SecretStr) walk would miss this -- must still be caught."""
settings = PersistedSettings.model_validate(
{"agent_settings": {"agent_context": {"secrets": {"MY_TOKEN": "sk-ctx"}}}}
)
assert settings.has_any_secret


def test_has_any_secret_false_for_empty_settings():
assert not PersistedSettings().has_any_secret


def test_secrets_has_any_secret_detects_custom_secret():
secrets = Secrets(
custom_secrets={
"MY_SECRET": CustomSecret(name="MY_SECRET", secret=SecretStr("sk-test"))
}
)
assert secrets.has_any_secret


def test_secrets_has_any_secret_false_for_empty_secrets():
assert not Secrets().has_any_secret


def test_settings_save_warns_for_critic_key_only_without_cipher(
persistence_dir, caplog
):
"""Regression test: before has_any_secret, the plaintext warning was
gated on llm_api_key_is_set alone, so a critic_api_key-only settings
object triggered no warning at all."""
caplog.set_level(logging.WARNING)
store = FileSettingsStore(persistence_dir=persistence_dir)
settings = PersistedSettings.model_validate(
{"agent_settings": {"verification": {"critic_api_key": "sk-critic"}}}
)

store.save(settings)

assert "PLAINTEXT" in caplog.text


def test_settings_save_no_warning_when_no_secrets_present(persistence_dir, caplog):
caplog.set_level(logging.WARNING)
store = FileSettingsStore(persistence_dir=persistence_dir)

store.save(PersistedSettings())

assert "PLAINTEXT" not in caplog.text


def test_settings_save_with_cipher_round_trips_critic_key(persistence_dir, cipher):
store = FileSettingsStore(persistence_dir=persistence_dir, cipher=cipher)
settings = PersistedSettings.model_validate(
{"agent_settings": {"verification": {"critic_api_key": "sk-critic"}}}
)

store.save(settings)

reloaded = store.load()
assert reloaded is not None
assert reloaded.has_any_secret


def test_secrets_save_warns_without_cipher(persistence_dir, caplog):
caplog.set_level(logging.WARNING)
store = FileSecretsStore(persistence_dir=persistence_dir)
secrets = Secrets(
custom_secrets={
"MY_SECRET": CustomSecret(name="MY_SECRET", secret=SecretStr("sk-test"))
}
)

store.save(secrets)

assert "PLAINTEXT" in caplog.text
Loading