Compare commits

..

1 Commits

Author SHA1 Message Date
huanghuoguoguo
aa4d46fd87 fix(utils): handle missing funcschema parameter docs 2026-05-16 10:37:01 +08:00
4 changed files with 20 additions and 86 deletions

View File

@@ -1,12 +1,8 @@
from __future__ import annotations from __future__ import annotations
import posixpath import posixpath
import re from typing import Any
from typing import TYPE_CHECKING, Any from langbot.pkg.core import app
from urllib.parse import unquote
if TYPE_CHECKING:
from langbot.pkg.core import app
class RAGRuntimeService: class RAGRuntimeService:
@@ -113,17 +109,8 @@ class RAGRuntimeService:
regardless of the underlying storage provider. regardless of the underlying storage provider.
""" """
# Validate storage_path to prevent path traversal # Validate storage_path to prevent path traversal
decoded_path = unquote(storage_path).replace('\\', '/') normalized = posixpath.normpath(storage_path)
decoded_segments = decoded_path.split('/') if normalized.startswith('/') or '..' in normalized.split('/'):
normalized = posixpath.normpath(decoded_path)
if (
not storage_path
or '\x00' in decoded_path
or normalized.startswith('/')
or '..' in decoded_segments
or '..' in normalized.split('/')
or re.match(r'^[A-Za-z]:/', normalized)
):
raise ValueError('Invalid storage path') raise ValueError('Invalid storage path')
content_bytes = await self.ap.storage_mgr.storage_provider.load(normalized) content_bytes = await self.ap.storage_mgr.storage_provider.load(normalized)
return content_bytes if content_bytes else b'' return content_bytes if content_bytes else b''

View File

@@ -83,7 +83,7 @@ def get_func_schema(function: typing.Callable) -> dict:
parameters['properties'][param.name] = { parameters['properties'][param.name] = {
'type': param_type, 'type': param_type,
'description': args_doc[param.name], 'description': args_doc.get(param.name, ''),
} }
# add schema for array # add schema for array

View File

@@ -1,68 +0,0 @@
from __future__ import annotations
from types import SimpleNamespace
import pytest
from langbot.pkg.rag.service.runtime import RAGRuntimeService
class DummyStorageProvider:
def __init__(self, content: bytes | None = b'data'):
self.content = content
self.loaded_paths: list[str] = []
async def load(self, path: str):
self.loaded_paths.append(path)
return self.content
def make_service(storage_provider: DummyStorageProvider) -> RAGRuntimeService:
return RAGRuntimeService(SimpleNamespace(storage_mgr=SimpleNamespace(storage_provider=storage_provider)))
@pytest.mark.asyncio
async def test_get_file_stream_normalizes_safe_path():
storage_provider = DummyStorageProvider()
service = make_service(storage_provider)
content = await service.get_file_stream('safe/./nested/file.pdf')
assert content == b'data'
assert storage_provider.loaded_paths == ['safe/nested/file.pdf']
@pytest.mark.asyncio
@pytest.mark.parametrize(
'storage_path',
[
'',
'../secret.txt',
'/absolute/path.txt',
'..\\secret.txt',
'nested\\..\\secret.txt',
'%2e%2e/secret.txt',
'nested/%2e%2e/secret.txt',
'C:\\secret.txt',
'safe/\x00file.txt',
],
)
async def test_get_file_stream_rejects_unsafe_paths(storage_path: str):
storage_provider = DummyStorageProvider()
service = make_service(storage_provider)
with pytest.raises(ValueError, match='Invalid storage path'):
await service.get_file_stream(storage_path)
assert storage_provider.loaded_paths == []
@pytest.mark.asyncio
async def test_get_file_stream_returns_empty_bytes_for_missing_content():
storage_provider = DummyStorageProvider(content=None)
service = make_service(storage_provider)
content = await service.get_file_stream('safe/file.pdf')
assert content == b''
assert storage_provider.loaded_paths == ['safe/file.pdf']

View File

@@ -0,0 +1,15 @@
from langbot.pkg.utils.funcschema import get_func_schema
def test_get_func_schema_uses_empty_description_for_undocumented_parameter():
def sample_function(documented: str, undocumented: int):
"""Sample function.
Args:
documented(str): documented parameter description
"""
schema = get_func_schema(sample_function)
assert schema['parameters']['properties']['documented']['description'] == 'documented parameter description'
assert schema['parameters']['properties']['undocumented']['description'] == ''