mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
feat: introduce BASIC_MEMORY_PROJECT_ROOT for path constraints (#334)
Signed-off-by: phernandez <paul@basicmachines.co>
This commit is contained in:
+3
-2
@@ -24,11 +24,12 @@ WORKDIR /app
|
||||
RUN uv sync --locked
|
||||
|
||||
# Create necessary directories and set ownership
|
||||
RUN mkdir -p /app/data /app/.basic-memory && \
|
||||
RUN mkdir -p /app/data/basic-memory /app/.basic-memory && \
|
||||
chown -R appuser:${GID} /app
|
||||
|
||||
# Set default data directory and add venv to PATH
|
||||
ENV BASIC_MEMORY_HOME=/app/data \
|
||||
ENV BASIC_MEMORY_HOME=/app/data/basic-memory \
|
||||
BASIC_MEMORY_PROJECT_ROOT=/app/data \
|
||||
PATH="/app/.venv/bin:$PATH"
|
||||
|
||||
# Switch to the non-root user
|
||||
|
||||
@@ -103,6 +103,12 @@ class BasicMemoryConfig(BaseSettings):
|
||||
description="Skip expensive initialization synchronization. Useful for cloud/stateless deployments where project reconciliation is not needed.",
|
||||
)
|
||||
|
||||
# Project path constraints
|
||||
project_root: Optional[str] = Field(
|
||||
default=None,
|
||||
description="If set, all projects must be created underneath this directory. Paths will be sanitized and constrained to this root. If not set, projects can be created anywhere (default behavior).",
|
||||
)
|
||||
|
||||
# API connection configuration
|
||||
api_url: Optional[str] = Field(
|
||||
default=None,
|
||||
@@ -232,6 +238,10 @@ class BasicMemoryConfig(BaseSettings):
|
||||
return Path.home() / DATA_DIR_NAME
|
||||
|
||||
|
||||
# Module-level cache for configuration
|
||||
_CONFIG_CACHE: Optional[BasicMemoryConfig] = None
|
||||
|
||||
|
||||
class ConfigManager:
|
||||
"""Manages Basic Memory configuration."""
|
||||
|
||||
@@ -253,12 +263,45 @@ class ConfigManager:
|
||||
return self.load_config()
|
||||
|
||||
def load_config(self) -> BasicMemoryConfig:
|
||||
"""Load configuration from file or create default."""
|
||||
"""Load configuration from file or create default.
|
||||
|
||||
Environment variables take precedence over file config values,
|
||||
following Pydantic Settings best practices.
|
||||
|
||||
Uses module-level cache for performance across ConfigManager instances.
|
||||
"""
|
||||
global _CONFIG_CACHE
|
||||
|
||||
# Return cached config if available
|
||||
if _CONFIG_CACHE is not None:
|
||||
return _CONFIG_CACHE
|
||||
|
||||
if self.config_file.exists():
|
||||
try:
|
||||
data = json.loads(self.config_file.read_text(encoding="utf-8"))
|
||||
return BasicMemoryConfig(**data)
|
||||
file_data = json.loads(self.config_file.read_text(encoding="utf-8"))
|
||||
|
||||
# First, create config from environment variables (Pydantic will read them)
|
||||
# Then overlay with file data for fields that aren't set via env vars
|
||||
# This ensures env vars take precedence
|
||||
|
||||
# Get env-based config fields that are actually set
|
||||
env_config = BasicMemoryConfig()
|
||||
env_dict = env_config.model_dump()
|
||||
|
||||
# Merge: file data as base, but only use it for fields not set by env
|
||||
# We detect env-set fields by comparing to default values
|
||||
merged_data = file_data.copy()
|
||||
|
||||
# For fields that have env var overrides, use those instead of file values
|
||||
# The env_prefix is "BASIC_MEMORY_" so we check those
|
||||
for field_name in BasicMemoryConfig.model_fields.keys():
|
||||
env_var_name = f"BASIC_MEMORY_{field_name.upper()}"
|
||||
if env_var_name in os.environ:
|
||||
# Environment variable is set, use it
|
||||
merged_data[field_name] = env_dict[field_name]
|
||||
|
||||
_CONFIG_CACHE = BasicMemoryConfig(**merged_data)
|
||||
return _CONFIG_CACHE
|
||||
except Exception as e: # pragma: no cover
|
||||
logger.exception(f"Failed to load config: {e}")
|
||||
raise e
|
||||
@@ -268,8 +311,11 @@ class ConfigManager:
|
||||
return config
|
||||
|
||||
def save_config(self, config: BasicMemoryConfig) -> None:
|
||||
"""Save configuration to file."""
|
||||
"""Save configuration to file and invalidate cache."""
|
||||
global _CONFIG_CACHE
|
||||
save_basic_memory_config(self.config_file, config)
|
||||
# Invalidate cache so next load_config() reads fresh data
|
||||
_CONFIG_CACHE = None
|
||||
|
||||
@property
|
||||
def projects(self) -> Dict[str, str]:
|
||||
|
||||
@@ -99,13 +99,12 @@ class ProjectService:
|
||||
Raises:
|
||||
ValueError: If the project already exists
|
||||
"""
|
||||
# in cloud mode, don't allow arbitrary paths.
|
||||
if self.config_manager.config.cloud_mode_enabled:
|
||||
basic_memory_home = os.getenv("BASIC_MEMORY_HOME")
|
||||
assert basic_memory_home is not None
|
||||
base_path = Path(basic_memory_home)
|
||||
# If project_root is set, constrain all projects to that directory
|
||||
project_root = self.config_manager.config.project_root
|
||||
if project_root:
|
||||
base_path = Path(project_root)
|
||||
|
||||
# Sanitize the input path for cloud mode
|
||||
# Sanitize the input path
|
||||
# Strip leading slashes, home directory references, and parent directory references
|
||||
clean_path = path.lstrip("/").replace("~/", "").replace("~", "")
|
||||
|
||||
@@ -116,13 +115,14 @@ class ProjectService:
|
||||
path_parts.append(part)
|
||||
clean_path = "/".join(path_parts) if path_parts else ""
|
||||
|
||||
# Construct path relative to BASIC_MEMORY_HOME
|
||||
# Construct path relative to project_root
|
||||
resolved_path = (base_path / clean_path).resolve().as_posix()
|
||||
|
||||
# Verify the resolved path is actually under BASIC_MEMORY_HOME
|
||||
# Verify the resolved path is actually under project_root
|
||||
if not resolved_path.startswith(base_path.resolve().as_posix()):
|
||||
raise ValueError(
|
||||
f"Cloud mode requires projects under {basic_memory_home}. Invalid path: {path}"
|
||||
f"BASIC_MEMORY_PROJECT_ROOT is set to {project_root}. "
|
||||
f"All projects must be created under this directory. Invalid path: {path}"
|
||||
)
|
||||
else:
|
||||
resolved_path = Path(os.path.abspath(os.path.expanduser(path))).as_posix()
|
||||
|
||||
@@ -78,6 +78,11 @@ def app_config(config_home, tmp_path, monkeypatch) -> BasicMemoryConfig:
|
||||
def config_manager(
|
||||
app_config: BasicMemoryConfig, project_config: ProjectConfig, config_home: Path, monkeypatch
|
||||
) -> ConfigManager:
|
||||
# Invalidate config cache to ensure clean state for each test
|
||||
from basic_memory import config as config_module
|
||||
|
||||
config_module._CONFIG_CACHE = None
|
||||
|
||||
# Create a new ConfigManager that uses the test home directory
|
||||
config_manager = ConfigManager()
|
||||
# Update its paths to use the test directory
|
||||
|
||||
@@ -716,48 +716,47 @@ async def test_synchronize_projects_handles_case_sensitivity_bug(
|
||||
await project_service.repository.delete(db_project.id)
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="Cloud mode only runs on POSIX systems")
|
||||
@pytest.mark.skipif(os.name == "nt", reason="Project root constraints only tested on POSIX systems")
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_project_cloud_mode_sanitizes_paths(
|
||||
async def test_add_project_with_project_root_sanitizes_paths(
|
||||
project_service: ProjectService, config_manager: ConfigManager, tmp_path, monkeypatch
|
||||
):
|
||||
"""Test that cloud mode sanitizes and validates project paths."""
|
||||
# Set up cloud mode environment
|
||||
cloud_home = tmp_path / "app" / "data" / "basic-memory"
|
||||
cloud_home.mkdir(parents=True, exist_ok=True)
|
||||
"""Test that BASIC_MEMORY_PROJECT_ROOT sanitizes and validates project paths."""
|
||||
# Set up project root environment
|
||||
project_root_path = tmp_path / "app" / "data"
|
||||
project_root_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
monkeypatch.setenv("BASIC_MEMORY_HOME", str(cloud_home))
|
||||
monkeypatch.setenv("BASIC_MEMORY_CLOUD_MODE", "true")
|
||||
monkeypatch.setenv("BASIC_MEMORY_PROJECT_ROOT", str(project_root_path))
|
||||
|
||||
# Force reload config to pick up cloud mode
|
||||
from basic_memory.services import project_service as ps_module
|
||||
# Invalidate config cache so it picks up the new env var
|
||||
from basic_memory import config as config_module
|
||||
|
||||
monkeypatch.setattr(ps_module, "config", config_manager.load_config())
|
||||
config_module._CONFIG_CACHE = None
|
||||
|
||||
test_cases = [
|
||||
# (input_path, expected_result_path, should_succeed)
|
||||
("test", str(cloud_home / "test"), True), # Simple relative path
|
||||
("~/Documents/test", str(cloud_home / "Documents" / "test"), True), # Home directory
|
||||
("test", str(project_root_path / "test"), True), # Simple relative path
|
||||
("~/Documents/test", str(project_root_path / "Documents" / "test"), True), # Home directory
|
||||
(
|
||||
"/tmp/test",
|
||||
str(cloud_home / "tmp" / "test"),
|
||||
str(project_root_path / "tmp" / "test"),
|
||||
True,
|
||||
), # Absolute path (sanitized to relative)
|
||||
(
|
||||
"../../../etc/passwd",
|
||||
str(cloud_home),
|
||||
str(project_root_path),
|
||||
True,
|
||||
), # Path traversal (all ../ removed, results in cloud_home)
|
||||
("folder/subfolder", str(cloud_home / "folder" / "subfolder"), True), # Nested path
|
||||
), # Path traversal (all ../ removed, results in project_root)
|
||||
("folder/subfolder", str(project_root_path / "folder" / "subfolder"), True), # Nested path
|
||||
(
|
||||
"~/folder/../test",
|
||||
str(cloud_home / "test"),
|
||||
str(project_root_path / "test"),
|
||||
True,
|
||||
), # Mixed patterns (sanitized to just 'test')
|
||||
]
|
||||
|
||||
for i, (input_path, expected_path, should_succeed) in enumerate(test_cases):
|
||||
test_project_name = f"cloud-test-{i}"
|
||||
test_project_name = f"project-root-test-{i}"
|
||||
|
||||
try:
|
||||
# Add the project
|
||||
@@ -768,9 +767,9 @@ async def test_add_project_cloud_mode_sanitizes_paths(
|
||||
assert test_project_name in project_service.projects
|
||||
actual_path = project_service.projects[test_project_name]
|
||||
|
||||
# The path should be under cloud_home
|
||||
assert actual_path.startswith(str(cloud_home)), (
|
||||
f"Path {actual_path} should start with {cloud_home} for input {input_path}"
|
||||
# The path should be under project_root
|
||||
assert actual_path.startswith(str(project_root_path)), (
|
||||
f"Path {actual_path} should start with {project_root_path} for input {input_path}"
|
||||
)
|
||||
|
||||
# Clean up
|
||||
@@ -784,29 +783,28 @@ async def test_add_project_cloud_mode_sanitizes_paths(
|
||||
# Expected failure - continue to next test case
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="Cloud mode only runs on POSIX systems")
|
||||
@pytest.mark.skipif(os.name == "nt", reason="Project root constraints only tested on POSIX systems")
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_project_cloud_mode_rejects_escape_attempts(
|
||||
async def test_add_project_with_project_root_rejects_escape_attempts(
|
||||
project_service: ProjectService, config_manager: ConfigManager, tmp_path, monkeypatch
|
||||
):
|
||||
"""Test that cloud mode rejects paths that try to escape cloud storage."""
|
||||
# Set up cloud mode environment
|
||||
cloud_home = tmp_path / "app" / "data" / "basic-memory"
|
||||
cloud_home.mkdir(parents=True, exist_ok=True)
|
||||
"""Test that BASIC_MEMORY_PROJECT_ROOT rejects paths that try to escape the project root."""
|
||||
# Set up project root environment
|
||||
project_root_path = tmp_path / "app" / "data"
|
||||
project_root_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create a directory outside cloud_home to verify it's not accessible
|
||||
# Create a directory outside project_root to verify it's not accessible
|
||||
outside_dir = tmp_path / "outside"
|
||||
outside_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
monkeypatch.setenv("BASIC_MEMORY_HOME", str(cloud_home))
|
||||
monkeypatch.setenv("BASIC_MEMORY_CLOUD_MODE", "true")
|
||||
monkeypatch.setenv("BASIC_MEMORY_PROJECT_ROOT", str(project_root_path))
|
||||
|
||||
# Force reload config to pick up cloud mode
|
||||
from basic_memory.services import project_service as ps_module
|
||||
# Invalidate config cache so it picks up the new env var
|
||||
from basic_memory import config as config_module
|
||||
|
||||
monkeypatch.setattr(ps_module, "config", config_manager.load_config())
|
||||
config_module._CONFIG_CACHE = None
|
||||
|
||||
# All of these should succeed by being sanitized to paths under cloud_home
|
||||
# All of these should succeed by being sanitized to paths under project_root
|
||||
# The sanitization removes dangerous patterns, so they don't escape
|
||||
safe_after_sanitization = [
|
||||
"../../../etc/passwd",
|
||||
@@ -815,16 +813,16 @@ async def test_add_project_cloud_mode_rejects_escape_attempts(
|
||||
]
|
||||
|
||||
for i, attack_path in enumerate(safe_after_sanitization):
|
||||
test_project_name = f"cloud-attack-test-{i}"
|
||||
test_project_name = f"project-root-attack-test-{i}"
|
||||
|
||||
try:
|
||||
# Add the project
|
||||
await project_service.add_project(test_project_name, attack_path)
|
||||
|
||||
# Verify it was sanitized to be under cloud_home
|
||||
# Verify it was sanitized to be under project_root
|
||||
actual_path = project_service.projects[test_project_name]
|
||||
assert actual_path.startswith(str(cloud_home)), (
|
||||
f"Sanitized path {actual_path} should be under {cloud_home}"
|
||||
assert actual_path.startswith(str(project_root_path)), (
|
||||
f"Sanitized path {actual_path} should be under {project_root_path}"
|
||||
)
|
||||
|
||||
# Clean up
|
||||
@@ -835,16 +833,17 @@ async def test_add_project_cloud_mode_rejects_escape_attempts(
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="Cloud mode only runs on POSIX systems")
|
||||
@pytest.mark.skipif(os.name == "nt", reason="Project root constraints only tested on POSIX systems")
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_project_local_mode_allows_arbitrary_paths(
|
||||
async def test_add_project_without_project_root_allows_arbitrary_paths(
|
||||
project_service: ProjectService, config_manager: ConfigManager, tmp_path, monkeypatch
|
||||
):
|
||||
"""Test that local mode (non-cloud) still allows arbitrary paths."""
|
||||
# Ensure cloud mode is disabled
|
||||
monkeypatch.setenv("BASIC_MEMORY_CLOUD_MODE", "false")
|
||||
"""Test that without BASIC_MEMORY_PROJECT_ROOT set, arbitrary paths are allowed."""
|
||||
# Ensure project_root is not set
|
||||
if "BASIC_MEMORY_PROJECT_ROOT" in os.environ:
|
||||
monkeypatch.delenv("BASIC_MEMORY_PROJECT_ROOT")
|
||||
|
||||
# Force reload config to pick up local mode
|
||||
# Force reload config without project_root
|
||||
from basic_memory.services import project_service as ps_module
|
||||
|
||||
monkeypatch.setattr(ps_module, "config", config_manager.load_config())
|
||||
@@ -853,10 +852,10 @@ async def test_add_project_local_mode_allows_arbitrary_paths(
|
||||
test_dir = tmp_path / "arbitrary-location"
|
||||
test_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
test_project_name = "local-mode-test"
|
||||
test_project_name = "no-project-root-test"
|
||||
|
||||
try:
|
||||
# In local mode, we should be able to use arbitrary absolute paths
|
||||
# Without project_root, we should be able to use arbitrary absolute paths
|
||||
await project_service.add_project(test_project_name, str(test_dir))
|
||||
|
||||
# Verify the path was accepted as-is
|
||||
|
||||
Reference in New Issue
Block a user