mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-06-02 20:14:36 +00:00
Compare commits
1 Commits
fix/rag-ru
...
fix/utils-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aa4d46fd87 |
@@ -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''
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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']
|
|
||||||
15
tests/unit_tests/utils/test_funcschema.py
Normal file
15
tests/unit_tests/utils/test_funcschema.py
Normal 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'] == ''
|
||||||
Reference in New Issue
Block a user