mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-06-02 12:05:54 +00:00
Compare commits
1 Commits
fix/rag-ru
...
fix/utils-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aa4d46fd87 |
@@ -1,12 +1,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import posixpath
|
||||
import re
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from urllib.parse import unquote
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from langbot.pkg.core import app
|
||||
from typing import Any
|
||||
from langbot.pkg.core import app
|
||||
|
||||
|
||||
class RAGRuntimeService:
|
||||
@@ -113,17 +109,8 @@ class RAGRuntimeService:
|
||||
regardless of the underlying storage provider.
|
||||
"""
|
||||
# Validate storage_path to prevent path traversal
|
||||
decoded_path = unquote(storage_path).replace('\\', '/')
|
||||
decoded_segments = decoded_path.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)
|
||||
):
|
||||
normalized = posixpath.normpath(storage_path)
|
||||
if normalized.startswith('/') or '..' in normalized.split('/'):
|
||||
raise ValueError('Invalid storage path')
|
||||
content_bytes = await self.ap.storage_mgr.storage_provider.load(normalized)
|
||||
return content_bytes if content_bytes else b''
|
||||
|
||||
@@ -83,7 +83,7 @@ def get_func_schema(function: typing.Callable) -> dict:
|
||||
|
||||
parameters['properties'][param.name] = {
|
||||
'type': param_type,
|
||||
'description': args_doc[param.name],
|
||||
'description': args_doc.get(param.name, ''),
|
||||
}
|
||||
|
||||
# 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