mirror of
https://github.com/OpenHands/OpenHands.git
synced 2026-10-07 15:58:03 +08:00
feat: add validation for secret names to ensure valid env var format (#12980)
Co-authored-by: Simon Rosenberg <simon@openhands.dev> Co-authored-by: openhands <openhands@all-hands.dev> Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com> Co-authored-by: simonrosenberg <157206163+simonrosenberg@users.noreply.github.com>
This commit is contained in:
co-authored by
Simon Rosenberg
openhands
Claude Sonnet 4.6
simonrosenberg
parent
fd1c6145ad
commit
8aef55b733
@@ -30,6 +30,7 @@ from openhands.app_server.integrations.provider import (
|
||||
ProviderToken,
|
||||
)
|
||||
from openhands.app_server.integrations.service_types import ProviderType
|
||||
from openhands.app_server.utils.env_var_validation import validate_env_var_name
|
||||
|
||||
|
||||
class Secrets(BaseModel):
|
||||
@@ -178,6 +179,12 @@ class CustomSecretWithoutValue(BaseModel):
|
||||
name: str
|
||||
description: str | None = None
|
||||
|
||||
@field_validator('name')
|
||||
@classmethod
|
||||
def validate_secret_name(cls, v: str) -> str:
|
||||
validate_env_var_name(v, field_name='secret name')
|
||||
return v
|
||||
|
||||
|
||||
class CustomSecretCreate(CustomSecretWithoutValue):
|
||||
"""Custom secret model with value (for creating secrets)."""
|
||||
|
||||
@@ -219,7 +219,7 @@ async def search_custom_secrets(
|
||||
if name__contains and name__contains.lower() not in secret_name.lower():
|
||||
continue
|
||||
all_secrets.append(
|
||||
CustomSecretWithoutValue(
|
||||
CustomSecretWithoutValue.model_construct(
|
||||
name=secret_name,
|
||||
description=secret_value.description,
|
||||
)
|
||||
@@ -306,37 +306,35 @@ async def update_custom_secret(
|
||||
500: Error updating secret
|
||||
"""
|
||||
existing_secrets = await secrets_store.load()
|
||||
if existing_secrets:
|
||||
# Check if the secret to update exists
|
||||
if secret_id not in existing_secrets.custom_secrets:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f'Secret with ID {secret_id} not found',
|
||||
)
|
||||
|
||||
secret_name = incoming_secret.name
|
||||
secret_description = incoming_secret.description
|
||||
|
||||
custom_secrets = dict(existing_secrets.custom_secrets)
|
||||
existing_secret = custom_secrets.pop(secret_id)
|
||||
|
||||
if secret_name != secret_id and secret_name in custom_secrets:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f'Secret {secret_name} already exists',
|
||||
)
|
||||
|
||||
custom_secrets[secret_name] = CustomSecret(
|
||||
secret=existing_secret.secret,
|
||||
description=secret_description or '',
|
||||
if not existing_secrets or secret_id not in existing_secrets.custom_secrets:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f'Secret with ID {secret_id} not found',
|
||||
)
|
||||
|
||||
updated_secrets = Secrets(
|
||||
custom_secrets=custom_secrets, # type: ignore[arg-type]
|
||||
provider_tokens=existing_secrets.provider_tokens,
|
||||
secret_name = incoming_secret.name
|
||||
secret_description = incoming_secret.description
|
||||
|
||||
custom_secrets = dict(existing_secrets.custom_secrets)
|
||||
existing_secret = custom_secrets.pop(secret_id)
|
||||
|
||||
if secret_name != secret_id and secret_name in custom_secrets:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f'Secret {secret_name} already exists',
|
||||
)
|
||||
|
||||
await secrets_store.store(updated_secrets)
|
||||
custom_secrets[secret_name] = CustomSecret(
|
||||
secret=existing_secret.secret,
|
||||
description=secret_description or '',
|
||||
)
|
||||
|
||||
updated_secrets = Secrets(
|
||||
custom_secrets=custom_secrets, # type: ignore[arg-type]
|
||||
provider_tokens=existing_secrets.provider_tokens,
|
||||
)
|
||||
|
||||
await secrets_store.store(updated_secrets)
|
||||
|
||||
return EditResponse(
|
||||
message='Secret updated successfully',
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
"""Utilities for validating environment variable names."""
|
||||
|
||||
import re
|
||||
|
||||
# Must start with a letter or underscore, contain only alphanumeric characters and underscores.
|
||||
ENV_VAR_NAME_PATTERN = re.compile(r'[a-zA-Z_][a-zA-Z0-9_]*')
|
||||
|
||||
|
||||
def is_valid_env_var_name(name: str) -> bool:
|
||||
"""Check if a name is valid for use as an environment variable."""
|
||||
return bool(ENV_VAR_NAME_PATTERN.fullmatch(name))
|
||||
|
||||
|
||||
def validate_env_var_name(name: str, field_name: str = 'name') -> None:
|
||||
"""Validate that a name is valid for use as an environment variable.
|
||||
|
||||
Raises:
|
||||
ValueError: If the name is invalid.
|
||||
"""
|
||||
if not is_valid_env_var_name(name):
|
||||
raise ValueError(
|
||||
f"Invalid {field_name} '{name}'. Must start with a letter or underscore, "
|
||||
'and contain only alphanumeric characters and underscores.'
|
||||
)
|
||||
@@ -563,3 +563,121 @@ async def test_add_multiple_git_providers_with_hosts(test_client, file_secrets_s
|
||||
stored_secrets.provider_tokens[ProviderType.GITLAB].host
|
||||
== 'gitlab.enterprise.com'
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_secret_with_invalid_name_hyphen(test_client, file_secrets_store):
|
||||
await file_secrets_store.store(Secrets())
|
||||
response = test_client.post(
|
||||
'/secrets',
|
||||
json={'name': 'MY-INVALID-SECRET', 'value': 'secret-value'},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
assert 'MY-INVALID-SECRET' in response.text or 'Invalid' in response.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_secret_with_invalid_name_starts_with_digit(
|
||||
test_client, file_secrets_store
|
||||
):
|
||||
await file_secrets_store.store(Secrets())
|
||||
response = test_client.post(
|
||||
'/secrets',
|
||||
json={'name': '1_INVALID_SECRET', 'value': 'secret-value'},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_secret_with_invalid_name_space(test_client, file_secrets_store):
|
||||
await file_secrets_store.store(Secrets())
|
||||
response = test_client.post(
|
||||
'/secrets',
|
||||
json={'name': 'MY INVALID SECRET', 'value': 'secret-value'},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_secret_with_valid_name_underscore(
|
||||
test_client, file_secrets_store
|
||||
):
|
||||
await file_secrets_store.store(Secrets())
|
||||
response = test_client.post(
|
||||
'/secrets',
|
||||
json={'name': 'MY_VALID_SECRET', 'value': 'secret-value'},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
stored = await file_secrets_store.load()
|
||||
assert 'MY_VALID_SECRET' in stored.custom_secrets
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_secret_with_valid_name_starts_with_underscore(
|
||||
test_client, file_secrets_store
|
||||
):
|
||||
await file_secrets_store.store(Secrets())
|
||||
response = test_client.post(
|
||||
'/secrets',
|
||||
json={'name': '_PRIVATE_SECRET', 'value': 'secret-value'},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
stored = await file_secrets_store.load()
|
||||
assert '_PRIVATE_SECRET' in stored.custom_secrets
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_secret_with_invalid_name(test_client, file_secrets_store):
|
||||
custom_secrets = {'VALID_SECRET': CustomSecret(secret=SecretStr('old-value'))}
|
||||
await file_secrets_store.store(Secrets(custom_secrets=custom_secrets)) # type: ignore[arg-type]
|
||||
response = test_client.put(
|
||||
'/secrets/VALID_SECRET',
|
||||
json={'name': 'INVALID-NEW-NAME', 'description': 'Updated'},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_secret_with_valid_name(test_client, file_secrets_store):
|
||||
custom_secrets = {'OLD_NAME': CustomSecret(secret=SecretStr('secret-value'))}
|
||||
await file_secrets_store.store(Secrets(custom_secrets=custom_secrets)) # type: ignore[arg-type]
|
||||
response = test_client.put(
|
||||
'/secrets/OLD_NAME',
|
||||
json={'name': 'NEW_VALID_NAME', 'description': 'Updated'},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
stored = await file_secrets_store.load()
|
||||
assert 'OLD_NAME' not in stored.custom_secrets
|
||||
assert 'NEW_VALID_NAME' in stored.custom_secrets
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_secret_with_empty_name(test_client, file_secrets_store):
|
||||
await file_secrets_store.store(Secrets())
|
||||
response = test_client.post(
|
||||
'/secrets',
|
||||
json={'name': '', 'value': 'secret-value'},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_secret_not_found_returns_404(test_client, file_secrets_store):
|
||||
await file_secrets_store.store(Secrets())
|
||||
response = test_client.put(
|
||||
'/secrets/NONEXISTENT',
|
||||
json={'name': 'NONEXISTENT', 'description': 'Updated'},
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_secrets_tolerates_legacy_invalid_names(
|
||||
test_client, file_secrets_store
|
||||
):
|
||||
custom_secrets = {'MY-LEGACY-SECRET': CustomSecret(secret=SecretStr('value'))}
|
||||
await file_secrets_store.store(Secrets(custom_secrets=custom_secrets)) # type: ignore[arg-type]
|
||||
response = test_client.get('/secrets/search')
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert any(item['name'] == 'MY-LEGACY-SECRET' for item in data['items'])
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
"""Tests for environment variable name validation utility."""
|
||||
|
||||
import pytest
|
||||
|
||||
from openhands.app_server.utils.env_var_validation import (
|
||||
is_valid_env_var_name,
|
||||
validate_env_var_name,
|
||||
)
|
||||
|
||||
|
||||
class TestIsValidEnvVarName:
|
||||
@pytest.mark.parametrize(
|
||||
'name',
|
||||
[
|
||||
'MY_VAR',
|
||||
'my_var',
|
||||
'MyVar',
|
||||
'_PRIVATE',
|
||||
'_',
|
||||
'__',
|
||||
'A',
|
||||
'a',
|
||||
'VAR123',
|
||||
'_123',
|
||||
'API_KEY',
|
||||
'DATABASE_URL',
|
||||
'GITHUB_TOKEN',
|
||||
],
|
||||
)
|
||||
def test_valid_names(self, name: str):
|
||||
assert is_valid_env_var_name(name) is True
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'name',
|
||||
[
|
||||
'MY-VAR',
|
||||
'MY VAR',
|
||||
'MY.VAR',
|
||||
'123VAR',
|
||||
'1',
|
||||
'-VAR',
|
||||
'MY@VAR',
|
||||
'MY$VAR',
|
||||
'MY#VAR',
|
||||
'MY!VAR',
|
||||
'MY%VAR',
|
||||
'MY^VAR',
|
||||
'MY&VAR',
|
||||
'MY*VAR',
|
||||
'MY(VAR',
|
||||
'MY)VAR',
|
||||
'MY+VAR',
|
||||
'MY=VAR',
|
||||
'MY[VAR',
|
||||
'MY]VAR',
|
||||
'MY{VAR',
|
||||
'MY}VAR',
|
||||
'MY|VAR',
|
||||
'MY\\VAR',
|
||||
'MY/VAR',
|
||||
'MY?VAR',
|
||||
'MY<VAR',
|
||||
'MY>VAR',
|
||||
'MY,VAR',
|
||||
'MY:VAR',
|
||||
'MY;VAR',
|
||||
"MY'VAR",
|
||||
'MY"VAR',
|
||||
'MY`VAR',
|
||||
'MY~VAR',
|
||||
'MY_VAR\n',
|
||||
],
|
||||
)
|
||||
def test_invalid_names_special_chars(self, name: str):
|
||||
assert is_valid_env_var_name(name) is False
|
||||
|
||||
def test_empty_string(self):
|
||||
assert is_valid_env_var_name('') is False
|
||||
|
||||
|
||||
class TestValidateEnvVarName:
|
||||
def test_valid_name_passes(self):
|
||||
validate_env_var_name('MY_VAR') # should not raise
|
||||
|
||||
def test_invalid_name_raises(self):
|
||||
with pytest.raises(ValueError, match='Invalid'):
|
||||
validate_env_var_name('MY-VAR')
|
||||
|
||||
def test_custom_field_name_in_error(self):
|
||||
with pytest.raises(ValueError, match='secret name'):
|
||||
validate_env_var_name('MY-VAR', field_name='secret name')
|
||||
Reference in New Issue
Block a user