mirror of
https://github.com/tiennm99/serena.git
synced 2026-10-12 18:16:43 +00:00
feat(pascal): Add auto-update for Pascal Language Server
- Automatically download pasls from zen010101/pascal-language-server releases - Check for updates every 24 hours with version comparison - SHA256 checksum verification for security - Atomic installation with rollback on failure - Tarfile path traversal protection - Handle nested archive directory structure CI improvements: - Fix FPC installation for Ubuntu/macOS/Windows - Add proper FPC verification step that fails on actual errors Tests: - 33 unit tests for version comparison, checksum, tarfile safety - 8 integration tests with actual pasls functionality
This commit is contained in:
1 parent
b5c49c0726
commit
0deaa923ec
3 files changed
+874
-21
No files matched your search
@@ -375,9 +375,15 @@ jobs:
|
||||
sudo apt-get install -y fpc fpc-source
|
||||
# Set environment variables for pasls
|
||||
echo "PP=/usr/bin/fpc" >> $GITHUB_ENV
|
||||
# Find FPC source directory (version may vary)
|
||||
FPCDIR=$(ls -d /usr/share/fpcsrc/*/ 2>/dev/null | head -1 || echo "/usr/share/fpcsrc")
|
||||
# Find FPC source directory (version may vary) - remove trailing slash
|
||||
FPCDIR=$(ls -d /usr/share/fpcsrc/*/ 2>/dev/null | head -1 | sed 's:/$::')
|
||||
if [[ -z "$FPCDIR" ]]; then
|
||||
FPCDIR="/usr/share/fpcsrc"
|
||||
fi
|
||||
echo "FPCDIR=$FPCDIR" >> $GITHUB_ENV
|
||||
echo "=== FPC source directory structure ==="
|
||||
ls -la "$FPCDIR" || echo "FPCDIR not found"
|
||||
ls -la "$FPCDIR/rtl" 2>/dev/null || echo "rtl subdirectory not found"
|
||||
elif [[ "${{ runner.os }}" == "macOS" ]]; then
|
||||
brew install fpc
|
||||
# Download FPC source from SourceForge (fpc-src-laz cask is incompatible with ARM64)
|
||||
@@ -386,9 +392,21 @@ jobs:
|
||||
mkdir -p "$HOME/fpcsrc"
|
||||
tar -xzf fpc-source.tar.gz -C "$HOME/fpcsrc"
|
||||
rm fpc-source.tar.gz
|
||||
# Set environment variables for pasls
|
||||
# Check extracted directory structure (might be nested)
|
||||
echo "=== Extracted FPC source structure ==="
|
||||
ls -la "$HOME/fpcsrc"
|
||||
# Find the actual FPC source root (contains rtl, packages, etc.)
|
||||
if [[ -d "$HOME/fpcsrc/fpc-${FPC_VERSION}/rtl" ]]; then
|
||||
FPCDIR="$HOME/fpcsrc/fpc-${FPC_VERSION}"
|
||||
elif [[ -d "$HOME/fpcsrc/fpc-${FPC_VERSION}/fpc-${FPC_VERSION}/rtl" ]]; then
|
||||
FPCDIR="$HOME/fpcsrc/fpc-${FPC_VERSION}/fpc-${FPC_VERSION}"
|
||||
else
|
||||
FPCDIR="$HOME/fpcsrc/fpc-${FPC_VERSION}"
|
||||
fi
|
||||
echo "PP=$(which fpc)" >> $GITHUB_ENV
|
||||
echo "FPCDIR=$HOME/fpcsrc/fpc-${FPC_VERSION}" >> $GITHUB_ENV
|
||||
echo "FPCDIR=$FPCDIR" >> $GITHUB_ENV
|
||||
echo "=== FPC source directory ==="
|
||||
ls -la "$FPCDIR" || echo "FPCDIR not found"
|
||||
elif [[ "${{ runner.os }}" == "Windows" ]]; then
|
||||
FPC_VERSION="3.2.2"
|
||||
# Download freepascal-ootb (includes FPC compiler)
|
||||
@@ -417,9 +435,44 @@ jobs:
|
||||
- name: Verify FPC installation
|
||||
shell: bash
|
||||
run: |
|
||||
echo "=== Environment variables ==="
|
||||
echo "PP=$PP"
|
||||
echo "FPCDIR=$FPCDIR"
|
||||
fpc -v || echo "FPC not in PATH, using PP directly"
|
||||
|
||||
# Create a simple test program
|
||||
if [[ "${{ runner.os }}" == "Windows" ]]; then
|
||||
TEST_PAS="$TEMP/fpc_test.pas"
|
||||
TEST_OUT="$TEMP/fpc_test"
|
||||
else
|
||||
TEST_PAS="/tmp/fpc_test.pas"
|
||||
TEST_OUT="/tmp/fpc_test"
|
||||
fi
|
||||
echo "program fpc_test; begin writeln('FPC works'); end." > "$TEST_PAS"
|
||||
|
||||
# Compile using PP (the compiler we actually use in tests)
|
||||
echo "=== Compiling test program with PP=$PP ==="
|
||||
if [[ -n "$PP" ]]; then
|
||||
"$PP" "$TEST_PAS" -o"$TEST_OUT" 2>&1
|
||||
else
|
||||
echo "ERROR: PP environment variable is not set"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Verify output binary exists
|
||||
if [[ -f "$TEST_OUT" ]] || [[ -f "${TEST_OUT}.exe" ]]; then
|
||||
echo "FPC compilation test PASSED"
|
||||
else
|
||||
echo "ERROR: FPC compilation failed - no output binary at $TEST_OUT"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
# Verify FPCDIR exists (required for pasls)
|
||||
if [[ -d "$FPCDIR" ]]; then
|
||||
echo "FPCDIR exists: $FPCDIR"
|
||||
else
|
||||
echo "ERROR: FPCDIR does not exist: $FPCDIR"
|
||||
exit 1
|
||||
fi
|
||||
- name: Cache language servers
|
||||
id: cache-language-servers
|
||||
uses: actions/cache@v3
|
||||
|
||||
@@ -4,22 +4,39 @@ Contains various configurations and settings specific to Pascal and Free Pascal.
|
||||
|
||||
pasls installation strategy:
|
||||
1. Use existing pasls from PATH
|
||||
2. Download prebuilt binary from GitHub releases
|
||||
2. Download prebuilt binary from GitHub releases (auto-updated)
|
||||
|
||||
Supported platforms for binary download:
|
||||
- linux-x64, linux-arm64
|
||||
- osx-x64, osx-arm64
|
||||
- win-x64
|
||||
|
||||
Auto-update features:
|
||||
- Checks for updates every 24 hours via GitHub API
|
||||
- SHA256 checksum verification before installation
|
||||
- Atomic update with rollback on failure
|
||||
- Windows file locking detection
|
||||
|
||||
You can pass the following entries in ls_specific_settings["pascal"]:
|
||||
- (reserved for future use)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import pathlib
|
||||
import platform
|
||||
import shutil
|
||||
import tarfile
|
||||
import threading
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
import uuid
|
||||
import zipfile
|
||||
|
||||
from solidlsp.language_servers.common import RuntimeDependency, RuntimeDependencyCollection, quote_windows_path
|
||||
from solidlsp.ls import SolidLanguageServer
|
||||
@@ -37,8 +54,18 @@ class PascalLanguageServer(SolidLanguageServer):
|
||||
Contains various configurations and settings specific to Free Pascal and Lazarus.
|
||||
"""
|
||||
|
||||
PASLS_VERSION = "0.1.0"
|
||||
PASLS_RELEASES_URL = "https://github.com/zen010101/pascal-language-server/releases/download"
|
||||
# URL configuration
|
||||
PASLS_RELEASES_URL = "https://github.com/zen010101/pascal-language-server/releases/latest/download"
|
||||
PASLS_API_URL = "https://api.github.com/repos/zen010101/pascal-language-server/releases/latest"
|
||||
|
||||
# Update check interval (seconds)
|
||||
UPDATE_CHECK_INTERVAL = 86400 # 24 hours
|
||||
|
||||
# Metadata directory name
|
||||
META_DIR = ".meta"
|
||||
|
||||
# Network timeout (seconds)
|
||||
NETWORK_TIMEOUT = 10
|
||||
|
||||
def __init__(self, config: LanguageServerConfig, repository_root_path: str, solidlsp_settings: SolidLSPSettings):
|
||||
"""
|
||||
@@ -56,10 +83,472 @@ class PascalLanguageServer(SolidLanguageServer):
|
||||
self.server_ready = threading.Event()
|
||||
self.completions_available_event = threading.Event()
|
||||
|
||||
# ============== Metadata Directory Management ==============
|
||||
|
||||
@classmethod
|
||||
def _meta_dir(cls, pasls_dir: str) -> str:
|
||||
"""Get metadata directory path, create if not exists."""
|
||||
meta_path = os.path.join(pasls_dir, cls.META_DIR)
|
||||
os.makedirs(meta_path, exist_ok=True)
|
||||
return meta_path
|
||||
|
||||
@classmethod
|
||||
def _meta_file(cls, pasls_dir: str, filename: str) -> str:
|
||||
"""Get metadata file path."""
|
||||
return os.path.join(cls._meta_dir(pasls_dir), filename)
|
||||
|
||||
# ============== Version Management ==============
|
||||
|
||||
@staticmethod
|
||||
def _normalize_version(version: str | None) -> str:
|
||||
"""Normalize version string by removing 'v' prefix and whitespace."""
|
||||
if not version:
|
||||
return ""
|
||||
return version.strip().lstrip("vV")
|
||||
|
||||
@classmethod
|
||||
def _is_newer_version(cls, latest: str | None, local: str | None) -> bool:
|
||||
"""Compare versions, return True if latest is newer than local."""
|
||||
if not latest:
|
||||
return False
|
||||
if not local:
|
||||
return True
|
||||
|
||||
latest_norm = cls._normalize_version(latest)
|
||||
local_norm = cls._normalize_version(local)
|
||||
|
||||
if not latest_norm:
|
||||
return False
|
||||
if not local_norm:
|
||||
return True
|
||||
|
||||
try:
|
||||
|
||||
def parse_version(v: str) -> list[int]:
|
||||
parts = []
|
||||
for part in v.split("."):
|
||||
num = ""
|
||||
for c in part:
|
||||
if c.isdigit():
|
||||
num += c
|
||||
else:
|
||||
break
|
||||
parts.append(int(num) if num else 0)
|
||||
return parts
|
||||
|
||||
latest_parts = parse_version(latest_norm)
|
||||
local_parts = parse_version(local_norm)
|
||||
|
||||
# Pad to same length
|
||||
max_len = max(len(latest_parts), len(local_parts))
|
||||
latest_parts.extend([0] * (max_len - len(latest_parts)))
|
||||
local_parts.extend([0] * (max_len - len(local_parts)))
|
||||
|
||||
return latest_parts > local_parts
|
||||
except Exception:
|
||||
log.warning(f"Failed to parse versions for comparison: {latest_norm} vs {local_norm}")
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def _get_latest_version(cls) -> str | None:
|
||||
"""Get latest version from GitHub API, return None on failure."""
|
||||
try:
|
||||
headers = {"Accept": "application/vnd.github.v3+json", "User-Agent": "Serena-LSP"}
|
||||
# Support GITHUB_TOKEN for CI environments with rate limits
|
||||
github_token = os.environ.get("GITHUB_TOKEN")
|
||||
if github_token:
|
||||
headers["Authorization"] = f"token {github_token}"
|
||||
|
||||
req = urllib.request.Request(cls.PASLS_API_URL, headers=headers)
|
||||
with urllib.request.urlopen(req, timeout=cls.NETWORK_TIMEOUT) as response:
|
||||
data = json.loads(response.read().decode())
|
||||
return data.get("tag_name")
|
||||
except Exception as e:
|
||||
log.debug(f"Failed to get latest pasls version: {type(e).__name__}: {e}")
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def _get_local_version(cls, pasls_dir: str) -> str | None:
|
||||
"""Read local version file."""
|
||||
version_file = cls._meta_file(pasls_dir, "version")
|
||||
if os.path.exists(version_file):
|
||||
try:
|
||||
with open(version_file, encoding="utf-8") as f:
|
||||
return f.read().strip()
|
||||
except OSError:
|
||||
return None
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def _save_local_version(cls, pasls_dir: str, version: str) -> None:
|
||||
"""Save version to local file."""
|
||||
version_file = cls._meta_file(pasls_dir, "version")
|
||||
try:
|
||||
with open(version_file, "w", encoding="utf-8") as f:
|
||||
f.write(version)
|
||||
except OSError as e:
|
||||
log.warning(f"Failed to save version file: {e}")
|
||||
|
||||
# ============== Update Check Timing ==============
|
||||
|
||||
@classmethod
|
||||
def _should_check_update(cls, pasls_dir: str) -> bool:
|
||||
"""Check if we should query for updates (more than 24 hours since last check)."""
|
||||
last_check_file = cls._meta_file(pasls_dir, "last_check")
|
||||
if not os.path.exists(last_check_file):
|
||||
return True
|
||||
try:
|
||||
with open(last_check_file, encoding="utf-8") as f:
|
||||
last_check = float(f.read().strip())
|
||||
return (time.time() - last_check) > cls.UPDATE_CHECK_INTERVAL
|
||||
except (OSError, ValueError):
|
||||
return True
|
||||
|
||||
@classmethod
|
||||
def _update_last_check(cls, pasls_dir: str) -> None:
|
||||
"""Update last check timestamp."""
|
||||
last_check_file = cls._meta_file(pasls_dir, "last_check")
|
||||
try:
|
||||
with open(last_check_file, "w", encoding="utf-8") as f:
|
||||
f.write(str(time.time()))
|
||||
except OSError as e:
|
||||
log.warning(f"Failed to update last check time: {e}")
|
||||
|
||||
# ============== SHA256 Checksum ==============
|
||||
|
||||
@classmethod
|
||||
def _get_checksums(cls) -> dict[str, str] | None:
|
||||
"""Download checksums file from GitHub, return {filename: sha256} dict."""
|
||||
checksums_url = f"{cls.PASLS_RELEASES_URL}/checksums.sha256"
|
||||
try:
|
||||
req = urllib.request.Request(checksums_url, headers={"User-Agent": "Serena-LSP"})
|
||||
with urllib.request.urlopen(req, timeout=cls.NETWORK_TIMEOUT) as response:
|
||||
content = response.read().decode("utf-8")
|
||||
checksums = {}
|
||||
for line in content.strip().split("\n"):
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
parts = line.split()
|
||||
if len(parts) >= 2:
|
||||
sha256 = parts[0]
|
||||
filename = parts[1].lstrip("*") # Remove possible * prefix
|
||||
checksums[filename] = sha256
|
||||
return checksums
|
||||
except Exception as e:
|
||||
log.warning(f"Failed to get checksums: {type(e).__name__}: {e}")
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _calculate_sha256(file_path: str) -> str:
|
||||
"""Calculate SHA256 checksum of a file."""
|
||||
sha256_hash = hashlib.sha256()
|
||||
with open(file_path, "rb") as f:
|
||||
for chunk in iter(lambda: f.read(8192), b""):
|
||||
sha256_hash.update(chunk)
|
||||
return sha256_hash.hexdigest()
|
||||
|
||||
@classmethod
|
||||
def _verify_checksum(cls, file_path: str, expected_sha256: str) -> bool:
|
||||
"""Verify file checksum."""
|
||||
try:
|
||||
actual_sha256 = cls._calculate_sha256(file_path)
|
||||
if actual_sha256.lower() == expected_sha256.lower():
|
||||
log.debug(f"Checksum verified: {file_path}")
|
||||
return True
|
||||
else:
|
||||
log.error(f"Checksum mismatch for {file_path}: expected {expected_sha256}, got {actual_sha256}")
|
||||
return False
|
||||
except Exception as e:
|
||||
log.error(f"Failed to verify checksum: {e}")
|
||||
return False
|
||||
|
||||
# ============== Windows File Locking ==============
|
||||
|
||||
@staticmethod
|
||||
def _is_file_locked(file_path: str) -> bool:
|
||||
"""Check if file is locked (Windows)."""
|
||||
if platform.system() != "Windows":
|
||||
return False
|
||||
|
||||
if not os.path.exists(file_path):
|
||||
return False
|
||||
|
||||
try:
|
||||
with open(file_path, "a"):
|
||||
pass
|
||||
return False
|
||||
except (OSError, PermissionError):
|
||||
return True
|
||||
|
||||
@classmethod
|
||||
def _safe_remove(cls, file_path: str) -> bool:
|
||||
"""Safely remove file, handle Windows file locking."""
|
||||
if not os.path.exists(file_path):
|
||||
return True
|
||||
|
||||
if platform.system() == "Windows" and cls._is_file_locked(file_path):
|
||||
temp_name = f"{file_path}.old.{uuid.uuid4().hex[:8]}"
|
||||
try:
|
||||
os.rename(file_path, temp_name)
|
||||
log.info(f"File locked, renamed to: {temp_name}")
|
||||
cls._mark_for_cleanup(os.path.dirname(file_path), temp_name)
|
||||
return True
|
||||
except PermissionError:
|
||||
log.warning(f"Cannot remove/rename locked file: {file_path}")
|
||||
return False
|
||||
else:
|
||||
try:
|
||||
os.remove(file_path)
|
||||
return True
|
||||
except OSError as e:
|
||||
log.warning(f"Failed to remove file {file_path}: {e}")
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def _mark_for_cleanup(cls, pasls_dir: str, file_path: str) -> None:
|
||||
"""Mark file for later cleanup."""
|
||||
cleanup_file = cls._meta_file(pasls_dir, "cleanup_list")
|
||||
try:
|
||||
with open(cleanup_file, "a", encoding="utf-8") as f:
|
||||
f.write(file_path + "\n")
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def _cleanup_old_files(cls, pasls_dir: str) -> None:
|
||||
"""Clean up old files marked for deletion."""
|
||||
cleanup_file = cls._meta_file(pasls_dir, "cleanup_list")
|
||||
if not os.path.exists(cleanup_file):
|
||||
return
|
||||
|
||||
try:
|
||||
with open(cleanup_file, encoding="utf-8") as f:
|
||||
files = [line.strip() for line in f if line.strip()]
|
||||
|
||||
remaining = []
|
||||
for file_path in files:
|
||||
if os.path.exists(file_path):
|
||||
try:
|
||||
os.remove(file_path)
|
||||
log.debug(f"Cleaned up old file: {file_path}")
|
||||
except OSError:
|
||||
remaining.append(file_path)
|
||||
|
||||
if remaining:
|
||||
with open(cleanup_file, "w", encoding="utf-8") as f:
|
||||
f.write("\n".join(remaining) + "\n")
|
||||
else:
|
||||
os.remove(cleanup_file)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
# ============== Download and Atomic Update ==============
|
||||
|
||||
@classmethod
|
||||
def _download_archive(cls, url: str, target_path: str) -> bool:
|
||||
"""Download archive to specified path."""
|
||||
try:
|
||||
os.makedirs(os.path.dirname(target_path), exist_ok=True)
|
||||
req = urllib.request.Request(url, headers={"User-Agent": "Serena-LSP"})
|
||||
with urllib.request.urlopen(req, timeout=60) as response:
|
||||
with open(target_path, "wb") as f:
|
||||
while True:
|
||||
chunk = response.read(8192)
|
||||
if not chunk:
|
||||
break
|
||||
f.write(chunk)
|
||||
return True
|
||||
except Exception as e:
|
||||
log.error(f"Failed to download {url}: {type(e).__name__}: {e}")
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def _is_safe_tar_member(cls, member: tarfile.TarInfo, target_dir: str) -> bool:
|
||||
"""Check if tar member is safe (prevent path traversal attack)."""
|
||||
# Check for .. in path components
|
||||
if ".." in member.name.split("/") or ".." in member.name.split("\\"):
|
||||
return False
|
||||
|
||||
# Check extracted path is within target directory
|
||||
abs_target = os.path.abspath(target_dir)
|
||||
abs_member = os.path.abspath(os.path.join(target_dir, member.name))
|
||||
|
||||
return abs_member.startswith(abs_target + os.sep) or abs_member == abs_target
|
||||
|
||||
@classmethod
|
||||
def _extract_archive(cls, archive_path: str, target_dir: str, archive_type: str) -> bool:
|
||||
"""Safely extract archive to specified directory."""
|
||||
try:
|
||||
os.makedirs(target_dir, exist_ok=True)
|
||||
|
||||
if archive_type == "gztar":
|
||||
with tarfile.open(archive_path, "r:gz") as tar:
|
||||
for member in tar.getmembers():
|
||||
if not cls._is_safe_tar_member(member, target_dir):
|
||||
log.error(f"Unsafe tar member detected (path traversal): {member.name}")
|
||||
return False
|
||||
tar.extractall(target_dir)
|
||||
|
||||
elif archive_type == "zip":
|
||||
with zipfile.ZipFile(archive_path, "r") as zip_ref:
|
||||
for name in zip_ref.namelist():
|
||||
if ".." in name.split("/") or ".." in name.split("\\"):
|
||||
log.error(f"Unsafe zip member detected (path traversal): {name}")
|
||||
return False
|
||||
abs_target = os.path.abspath(target_dir)
|
||||
abs_member = os.path.abspath(os.path.join(target_dir, name))
|
||||
if not (abs_member.startswith(abs_target + os.sep) or abs_member == abs_target):
|
||||
log.error(f"Unsafe zip member detected (path traversal): {name}")
|
||||
return False
|
||||
zip_ref.extractall(target_dir)
|
||||
|
||||
else:
|
||||
log.error(f"Unsupported archive type: {archive_type}")
|
||||
return False
|
||||
|
||||
# Handle nested directory: if extraction created a single subdirectory,
|
||||
# move its contents up to target_dir (common with GitHub release archives)
|
||||
cls._flatten_single_subdir(target_dir)
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
log.error(f"Failed to extract archive: {type(e).__name__}: {e}")
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def _flatten_single_subdir(cls, target_dir: str) -> None:
|
||||
"""If target_dir contains only a single subdirectory, move its contents up."""
|
||||
entries = os.listdir(target_dir)
|
||||
if len(entries) == 1:
|
||||
subdir = os.path.join(target_dir, entries[0])
|
||||
if os.path.isdir(subdir):
|
||||
# Move all contents from subdir to target_dir
|
||||
for item in os.listdir(subdir):
|
||||
src = os.path.join(subdir, item)
|
||||
dst = os.path.join(target_dir, item)
|
||||
shutil.move(src, dst)
|
||||
# Remove the now-empty subdirectory
|
||||
os.rmdir(subdir)
|
||||
|
||||
@classmethod
|
||||
def _get_archive_filename(cls, dep: RuntimeDependency) -> str:
|
||||
"""Get archive filename from URL."""
|
||||
assert dep.url is not None, "RuntimeDependency.url must be set"
|
||||
return dep.url.split("/")[-1]
|
||||
|
||||
@classmethod
|
||||
def _atomic_install(cls, pasls_dir: str, deps: RuntimeDependencyCollection, checksums: dict[str, str] | None) -> bool:
|
||||
"""Atomic update: download -> verify checksum -> extract -> replace."""
|
||||
temp_dir = pasls_dir + ".tmp"
|
||||
backup_dir = pasls_dir + ".backup"
|
||||
temp_archive_dir = os.path.join(os.path.expanduser("~"), "solidlsp_tmp")
|
||||
|
||||
try:
|
||||
dep = deps.get_single_dep_for_current_platform()
|
||||
assert dep.url is not None, "RuntimeDependency.url must be set"
|
||||
assert dep.archive_type is not None, "RuntimeDependency.archive_type must be set"
|
||||
|
||||
archive_filename = cls._get_archive_filename(dep)
|
||||
archive_path = os.path.join(temp_archive_dir, archive_filename)
|
||||
|
||||
# 1. Clean up any existing temp directory
|
||||
if os.path.exists(temp_dir):
|
||||
shutil.rmtree(temp_dir)
|
||||
os.makedirs(temp_archive_dir, exist_ok=True)
|
||||
|
||||
# 2. Download archive
|
||||
log.info(f"Downloading pasls archive: {archive_filename}")
|
||||
if not cls._download_archive(dep.url, archive_path):
|
||||
log.error("Failed to download pasls archive")
|
||||
return False
|
||||
|
||||
# 3. Verify SHA256 checksum (critical security step, before extraction)
|
||||
if checksums:
|
||||
expected_sha256 = checksums.get(archive_filename)
|
||||
if expected_sha256:
|
||||
log.info(f"Verifying SHA256 checksum for {archive_filename}...")
|
||||
if not cls._verify_checksum(archive_path, expected_sha256):
|
||||
log.error(f"SHA256 checksum verification FAILED for {archive_filename}")
|
||||
log.error("Aborting installation due to checksum mismatch - possible security issue!")
|
||||
try:
|
||||
os.remove(archive_path)
|
||||
except OSError:
|
||||
pass
|
||||
return False
|
||||
log.info("SHA256 checksum verified successfully")
|
||||
else:
|
||||
log.warning(f"No checksum found for {archive_filename} in checksums file")
|
||||
else:
|
||||
log.warning("No checksums available - skipping verification (not recommended for production)")
|
||||
|
||||
# 4. Extract to temp directory
|
||||
os.makedirs(temp_dir, exist_ok=True)
|
||||
log.info("Extracting archive to temporary directory...")
|
||||
if not cls._extract_archive(archive_path, temp_dir, dep.archive_type):
|
||||
log.error("Failed to extract archive")
|
||||
return False
|
||||
|
||||
# 5. Set execute permission
|
||||
binary_path = deps.binary_path(temp_dir)
|
||||
if os.path.exists(binary_path):
|
||||
try:
|
||||
os.chmod(binary_path, 0o755)
|
||||
except OSError:
|
||||
pass # May fail on Windows
|
||||
|
||||
# 6. Backup old version
|
||||
if os.path.exists(pasls_dir):
|
||||
if os.path.exists(backup_dir):
|
||||
shutil.rmtree(backup_dir)
|
||||
shutil.move(pasls_dir, backup_dir)
|
||||
|
||||
# 7. Replace with new version
|
||||
shutil.move(temp_dir, pasls_dir)
|
||||
|
||||
# 8. Restore meta directory from backup (preserves version info, last_check, etc.)
|
||||
if os.path.exists(backup_dir):
|
||||
backup_meta = os.path.join(backup_dir, cls.META_DIR)
|
||||
if os.path.exists(backup_meta):
|
||||
target_meta = os.path.join(pasls_dir, cls.META_DIR)
|
||||
if not os.path.exists(target_meta):
|
||||
shutil.copytree(backup_meta, target_meta)
|
||||
|
||||
# 9. Clean up downloaded archive and temp directory
|
||||
try:
|
||||
os.remove(archive_path)
|
||||
os.rmdir(temp_archive_dir)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
log.info("pasls installation completed successfully")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
log.error(f"Installation failed: {e}")
|
||||
|
||||
# Rollback
|
||||
if os.path.exists(backup_dir) and not os.path.exists(pasls_dir):
|
||||
try:
|
||||
shutil.move(backup_dir, pasls_dir)
|
||||
log.info("Rolled back to previous version")
|
||||
except Exception as rollback_error:
|
||||
log.error(f"Rollback failed: {rollback_error}")
|
||||
|
||||
# Clean up temp directory
|
||||
if os.path.exists(temp_dir):
|
||||
try:
|
||||
shutil.rmtree(temp_dir)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def _setup_runtime_dependencies(cls, solidlsp_settings: SolidLSPSettings) -> str:
|
||||
"""
|
||||
Setup runtime dependencies for Pascal Language Server (pasls).
|
||||
Automatically checks for updates every 24 hours with security verification.
|
||||
|
||||
Returns:
|
||||
str: The command to start the pasls server
|
||||
@@ -71,13 +560,21 @@ class PascalLanguageServer(SolidLanguageServer):
|
||||
log.info(f"Found pasls in PATH: {pasls_in_path}")
|
||||
return quote_windows_path(pasls_in_path)
|
||||
|
||||
# Use RuntimeDependencyCollection for download
|
||||
pasls_dir = cls.ls_resources_dir(solidlsp_settings)
|
||||
os.makedirs(pasls_dir, exist_ok=True)
|
||||
|
||||
# Clean up old files from previous sessions
|
||||
cls._cleanup_old_files(pasls_dir)
|
||||
|
||||
# Use RuntimeDependencyCollection for platform detection
|
||||
# Asset names follow zen010101/pascal-language-server release convention:
|
||||
# pasls-{cpu_arch}-{os}.{ext} where cpu_arch is x86_64/aarch64/i386
|
||||
deps = RuntimeDependencyCollection(
|
||||
[
|
||||
RuntimeDependency(
|
||||
id="PascalLanguageServer",
|
||||
description="Pascal Language Server for Linux (x64)",
|
||||
url=f"{cls.PASLS_RELEASES_URL}/v{cls.PASLS_VERSION}/pasls-linux-x64.tar.gz",
|
||||
url=f"{cls.PASLS_RELEASES_URL}/pasls-x86_64-linux.tar.gz",
|
||||
platform_id="linux-x64",
|
||||
archive_type="gztar",
|
||||
binary_name="pasls",
|
||||
@@ -85,7 +582,7 @@ class PascalLanguageServer(SolidLanguageServer):
|
||||
RuntimeDependency(
|
||||
id="PascalLanguageServer",
|
||||
description="Pascal Language Server for Linux (arm64)",
|
||||
url=f"{cls.PASLS_RELEASES_URL}/v{cls.PASLS_VERSION}/pasls-linux-arm64.tar.gz",
|
||||
url=f"{cls.PASLS_RELEASES_URL}/pasls-aarch64-linux.tar.gz",
|
||||
platform_id="linux-arm64",
|
||||
archive_type="gztar",
|
||||
binary_name="pasls",
|
||||
@@ -93,23 +590,23 @@ class PascalLanguageServer(SolidLanguageServer):
|
||||
RuntimeDependency(
|
||||
id="PascalLanguageServer",
|
||||
description="Pascal Language Server for macOS (x64)",
|
||||
url=f"{cls.PASLS_RELEASES_URL}/v{cls.PASLS_VERSION}/pasls-darwin-x64.tar.gz",
|
||||
url=f"{cls.PASLS_RELEASES_URL}/pasls-x86_64-darwin.zip",
|
||||
platform_id="osx-x64",
|
||||
archive_type="gztar",
|
||||
archive_type="zip",
|
||||
binary_name="pasls",
|
||||
),
|
||||
RuntimeDependency(
|
||||
id="PascalLanguageServer",
|
||||
description="Pascal Language Server for macOS (arm64)",
|
||||
url=f"{cls.PASLS_RELEASES_URL}/v{cls.PASLS_VERSION}/pasls-darwin-arm64.tar.gz",
|
||||
url=f"{cls.PASLS_RELEASES_URL}/pasls-aarch64-darwin.zip",
|
||||
platform_id="osx-arm64",
|
||||
archive_type="gztar",
|
||||
archive_type="zip",
|
||||
binary_name="pasls",
|
||||
),
|
||||
RuntimeDependency(
|
||||
id="PascalLanguageServer",
|
||||
description="Pascal Language Server for Windows (x64)",
|
||||
url=f"{cls.PASLS_RELEASES_URL}/v{cls.PASLS_VERSION}/pasls-win32-x64.zip",
|
||||
url=f"{cls.PASLS_RELEASES_URL}/pasls-x86_64-win64.zip",
|
||||
platform_id="win-x64",
|
||||
archive_type="zip",
|
||||
binary_name="pasls.exe",
|
||||
@@ -117,17 +614,65 @@ class PascalLanguageServer(SolidLanguageServer):
|
||||
]
|
||||
)
|
||||
|
||||
pasls_dir = cls.ls_resources_dir(solidlsp_settings)
|
||||
pasls_executable_path = deps.binary_path(pasls_dir)
|
||||
|
||||
# Determine if download is needed
|
||||
need_download = False
|
||||
latest_version = None
|
||||
checksums = None
|
||||
|
||||
if not os.path.exists(pasls_executable_path):
|
||||
log.info(f"Downloading pasls to {pasls_dir}...")
|
||||
deps.install(pasls_dir)
|
||||
# First install
|
||||
log.info("pasls not found, will download...")
|
||||
need_download = True
|
||||
latest_version = cls._get_latest_version()
|
||||
checksums = cls._get_checksums()
|
||||
elif cls._should_check_update(pasls_dir):
|
||||
# Check for updates
|
||||
log.debug("Checking for pasls updates...")
|
||||
latest_version = cls._get_latest_version()
|
||||
local_version = cls._get_local_version(pasls_dir)
|
||||
|
||||
if cls._is_newer_version(latest_version, local_version):
|
||||
log.info(f"New pasls version available: {latest_version} (current: {local_version})")
|
||||
|
||||
# Check Windows file locking
|
||||
if cls._is_file_locked(pasls_executable_path):
|
||||
log.warning("Cannot update pasls: file is in use. Will retry next time.")
|
||||
else:
|
||||
need_download = True
|
||||
checksums = cls._get_checksums()
|
||||
else:
|
||||
log.debug(f"pasls is up to date: {local_version}")
|
||||
|
||||
if need_download:
|
||||
if cls._atomic_install(pasls_dir, deps, checksums):
|
||||
# Update metadata after successful installation
|
||||
if latest_version:
|
||||
cls._save_local_version(pasls_dir, latest_version)
|
||||
else:
|
||||
# API failed but download succeeded, record placeholder version
|
||||
cls._save_local_version(pasls_dir, "unknown")
|
||||
cls._update_last_check(pasls_dir)
|
||||
else:
|
||||
# Installation failed, use existing version if available
|
||||
if not os.path.exists(pasls_executable_path):
|
||||
raise RuntimeError("Failed to install pasls and no local version available")
|
||||
log.warning("Update failed, using existing version")
|
||||
|
||||
# Update check time even if no update (avoid frequent checks)
|
||||
if not need_download and cls._should_check_update(pasls_dir):
|
||||
cls._update_last_check(pasls_dir)
|
||||
|
||||
assert os.path.exists(pasls_executable_path), f"pasls executable not found at {pasls_executable_path}"
|
||||
os.chmod(pasls_executable_path, 0o755)
|
||||
log.info(f"Using pasls at: {pasls_executable_path}")
|
||||
|
||||
# Ensure execute permission
|
||||
try:
|
||||
os.chmod(pasls_executable_path, 0o755)
|
||||
except OSError:
|
||||
pass # May fail on Windows, ignore
|
||||
|
||||
log.info(f"Using pasls at: {pasls_executable_path}")
|
||||
return quote_windows_path(pasls_executable_path)
|
||||
|
||||
@staticmethod
|
||||
|
||||
@@ -0,0 +1,255 @@
|
||||
"""
|
||||
Unit tests for the Pascal language server auto-update functionality.
|
||||
|
||||
These tests validate the version comparison, checksum verification,
|
||||
and other helper methods without requiring network access or the
|
||||
actual Pascal language server.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import tarfile
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from solidlsp.language_servers.pascal_server import PascalLanguageServer
|
||||
|
||||
pytestmark = [pytest.mark.pascal]
|
||||
|
||||
|
||||
class TestVersionNormalization:
|
||||
"""Test version string normalization."""
|
||||
|
||||
def test_normalize_version_with_v_prefix(self) -> None:
|
||||
"""Test that 'v' prefix is stripped."""
|
||||
assert PascalLanguageServer._normalize_version("v1.0.0") == "1.0.0"
|
||||
|
||||
def test_normalize_version_with_capital_v_prefix(self) -> None:
|
||||
"""Test that 'V' prefix is stripped."""
|
||||
assert PascalLanguageServer._normalize_version("V1.0.0") == "1.0.0"
|
||||
|
||||
def test_normalize_version_without_prefix(self) -> None:
|
||||
"""Test version without prefix is unchanged."""
|
||||
assert PascalLanguageServer._normalize_version("1.0.0") == "1.0.0"
|
||||
|
||||
def test_normalize_version_with_whitespace(self) -> None:
|
||||
"""Test that whitespace is stripped."""
|
||||
assert PascalLanguageServer._normalize_version(" v1.0.0 ") == "1.0.0"
|
||||
|
||||
def test_normalize_version_empty(self) -> None:
|
||||
"""Test empty version returns empty string."""
|
||||
assert PascalLanguageServer._normalize_version("") == ""
|
||||
|
||||
def test_normalize_version_none(self) -> None:
|
||||
"""Test None returns empty string."""
|
||||
assert PascalLanguageServer._normalize_version(None) == ""
|
||||
|
||||
|
||||
class TestVersionComparison:
|
||||
"""Test version comparison logic."""
|
||||
|
||||
def test_newer_version_major(self) -> None:
|
||||
"""Test detection of newer major version."""
|
||||
assert PascalLanguageServer._is_newer_version("v2.0.0", "v1.0.0") is True
|
||||
|
||||
def test_newer_version_minor(self) -> None:
|
||||
"""Test detection of newer minor version."""
|
||||
assert PascalLanguageServer._is_newer_version("v1.1.0", "v1.0.0") is True
|
||||
|
||||
def test_newer_version_patch(self) -> None:
|
||||
"""Test detection of newer patch version."""
|
||||
assert PascalLanguageServer._is_newer_version("v1.0.1", "v1.0.0") is True
|
||||
|
||||
def test_same_version(self) -> None:
|
||||
"""Test same version returns False."""
|
||||
assert PascalLanguageServer._is_newer_version("v1.0.0", "v1.0.0") is False
|
||||
|
||||
def test_older_version(self) -> None:
|
||||
"""Test older version returns False."""
|
||||
assert PascalLanguageServer._is_newer_version("v1.0.0", "v2.0.0") is False
|
||||
|
||||
def test_latest_none_returns_false(self) -> None:
|
||||
"""Test None latest version returns False."""
|
||||
assert PascalLanguageServer._is_newer_version(None, "v1.0.0") is False
|
||||
|
||||
def test_local_none_returns_true(self) -> None:
|
||||
"""Test None local version returns True (first install)."""
|
||||
assert PascalLanguageServer._is_newer_version("v1.0.0", None) is True
|
||||
|
||||
def test_both_none_returns_false(self) -> None:
|
||||
"""Test both None returns False."""
|
||||
assert PascalLanguageServer._is_newer_version(None, None) is False
|
||||
|
||||
def test_version_with_different_lengths(self) -> None:
|
||||
"""Test versions with different number of parts."""
|
||||
assert PascalLanguageServer._is_newer_version("v1.0.1", "v1.0") is True
|
||||
assert PascalLanguageServer._is_newer_version("v1.0", "v1.0.1") is False
|
||||
|
||||
def test_version_with_prerelease(self) -> None:
|
||||
"""Test versions with prerelease suffixes."""
|
||||
# Prerelease suffix is ignored, only numeric parts are compared
|
||||
assert PascalLanguageServer._is_newer_version("v1.1.0-beta", "v1.0.0") is True
|
||||
|
||||
|
||||
class TestSHA256Checksum:
|
||||
"""Test SHA256 checksum calculation and verification."""
|
||||
|
||||
def test_calculate_sha256(self) -> None:
|
||||
"""Test SHA256 calculation for a known content."""
|
||||
with tempfile.NamedTemporaryFile(mode="wb", delete=False) as f:
|
||||
f.write(b"test content")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
result = PascalLanguageServer._calculate_sha256(temp_path)
|
||||
expected = hashlib.sha256(b"test content").hexdigest()
|
||||
assert result == expected
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
def test_verify_checksum_correct(self) -> None:
|
||||
"""Test checksum verification with correct checksum."""
|
||||
with tempfile.NamedTemporaryFile(mode="wb", delete=False) as f:
|
||||
f.write(b"test content")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
expected = hashlib.sha256(b"test content").hexdigest()
|
||||
assert PascalLanguageServer._verify_checksum(temp_path, expected) is True
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
def test_verify_checksum_incorrect(self) -> None:
|
||||
"""Test checksum verification with incorrect checksum."""
|
||||
with tempfile.NamedTemporaryFile(mode="wb", delete=False) as f:
|
||||
f.write(b"test content")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
wrong_checksum = "0" * 64
|
||||
assert PascalLanguageServer._verify_checksum(temp_path, wrong_checksum) is False
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
def test_verify_checksum_case_insensitive(self) -> None:
|
||||
"""Test checksum verification is case insensitive."""
|
||||
with tempfile.NamedTemporaryFile(mode="wb", delete=False) as f:
|
||||
f.write(b"test content")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
expected = hashlib.sha256(b"test content").hexdigest().upper()
|
||||
assert PascalLanguageServer._verify_checksum(temp_path, expected) is True
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
|
||||
class TestTarfileSafety:
|
||||
"""Test tarfile path traversal protection."""
|
||||
|
||||
def test_safe_tar_member_normal_path(self) -> None:
|
||||
"""Test normal path is considered safe."""
|
||||
member = tarfile.TarInfo(name="pasls")
|
||||
assert PascalLanguageServer._is_safe_tar_member(member, "/tmp/target") is True
|
||||
|
||||
def test_safe_tar_member_nested_path(self) -> None:
|
||||
"""Test nested path is considered safe."""
|
||||
member = tarfile.TarInfo(name="subdir/pasls")
|
||||
assert PascalLanguageServer._is_safe_tar_member(member, "/tmp/target") is True
|
||||
|
||||
def test_unsafe_tar_member_path_traversal(self) -> None:
|
||||
"""Test path traversal is detected."""
|
||||
member = tarfile.TarInfo(name="../etc/passwd")
|
||||
assert PascalLanguageServer._is_safe_tar_member(member, "/tmp/target") is False
|
||||
|
||||
def test_unsafe_tar_member_hidden_traversal(self) -> None:
|
||||
"""Test hidden path traversal in nested path."""
|
||||
member = tarfile.TarInfo(name="subdir/../../etc/passwd")
|
||||
assert PascalLanguageServer._is_safe_tar_member(member, "/tmp/target") is False
|
||||
|
||||
def test_safe_tar_member_similar_name(self) -> None:
|
||||
"""Test path containing '..' in filename (not directory) is safe."""
|
||||
member = tarfile.TarInfo(name="file..name")
|
||||
assert PascalLanguageServer._is_safe_tar_member(member, "/tmp/target") is True
|
||||
|
||||
|
||||
class TestMetadataManagement:
|
||||
"""Test metadata directory and file management."""
|
||||
|
||||
def test_meta_dir_creates_directory(self) -> None:
|
||||
"""Test _meta_dir creates directory if not exists."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
meta_path = PascalLanguageServer._meta_dir(temp_dir)
|
||||
assert os.path.exists(meta_path)
|
||||
assert meta_path == os.path.join(temp_dir, PascalLanguageServer.META_DIR)
|
||||
|
||||
def test_meta_file_returns_correct_path(self) -> None:
|
||||
"""Test _meta_file returns correct path."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
meta_file = PascalLanguageServer._meta_file(temp_dir, "version")
|
||||
expected = os.path.join(temp_dir, PascalLanguageServer.META_DIR, "version")
|
||||
assert meta_file == expected
|
||||
|
||||
|
||||
class TestUpdateCheckTiming:
|
||||
"""Test update check timing logic."""
|
||||
|
||||
def test_should_check_update_no_last_check(self) -> None:
|
||||
"""Test should check when no last_check file exists."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
assert PascalLanguageServer._should_check_update(temp_dir) is True
|
||||
|
||||
def test_should_check_update_recent_check(self) -> None:
|
||||
"""Test should not check when recently checked."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
# Create meta dir and last_check file with current time
|
||||
meta_dir = PascalLanguageServer._meta_dir(temp_dir)
|
||||
last_check_file = os.path.join(meta_dir, "last_check")
|
||||
with open(last_check_file, "w") as f:
|
||||
f.write(str(time.time()))
|
||||
|
||||
assert PascalLanguageServer._should_check_update(temp_dir) is False
|
||||
|
||||
def test_should_check_update_old_check(self) -> None:
|
||||
"""Test should check when last check was > 24 hours ago."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
# Create meta dir and last_check file with old time
|
||||
meta_dir = PascalLanguageServer._meta_dir(temp_dir)
|
||||
last_check_file = os.path.join(meta_dir, "last_check")
|
||||
old_time = time.time() - (PascalLanguageServer.UPDATE_CHECK_INTERVAL + 3600)
|
||||
with open(last_check_file, "w") as f:
|
||||
f.write(str(old_time))
|
||||
|
||||
assert PascalLanguageServer._should_check_update(temp_dir) is True
|
||||
|
||||
def test_update_last_check_creates_file(self) -> None:
|
||||
"""Test _update_last_check creates timestamp file."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
PascalLanguageServer._update_last_check(temp_dir)
|
||||
last_check_file = PascalLanguageServer._meta_file(temp_dir, "last_check")
|
||||
assert os.path.exists(last_check_file)
|
||||
|
||||
with open(last_check_file) as f:
|
||||
timestamp = float(f.read().strip())
|
||||
assert abs(timestamp - time.time()) < 5 # within 5 seconds
|
||||
|
||||
|
||||
class TestVersionPersistence:
|
||||
"""Test local version persistence."""
|
||||
|
||||
def test_save_and_get_local_version(self) -> None:
|
||||
"""Test saving and retrieving local version."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
PascalLanguageServer._save_local_version(temp_dir, "v1.0.0")
|
||||
version = PascalLanguageServer._get_local_version(temp_dir)
|
||||
assert version == "v1.0.0"
|
||||
|
||||
def test_get_local_version_not_exists(self) -> None:
|
||||
"""Test getting version when file doesn't exist."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
version = PascalLanguageServer._get_local_version(temp_dir)
|
||||
assert version is None
|
||||
Reference in new issue
Block a user