diff --git a/apps/backend/security/profile.py b/apps/backend/security/profile.py index bb3595f4..da75cff1 100644 --- a/apps/backend/security/profile.py +++ b/apps/backend/security/profile.py @@ -9,6 +9,7 @@ Uses project_analyzer to create dynamic security profiles based on detected stac from pathlib import Path from project_analyzer import ( + ProjectAnalyzer, SecurityProfile, get_or_create_profile, ) @@ -20,6 +21,22 @@ from project_analyzer import ( # Cache the security profile to avoid re-analyzing on every command _cached_profile: SecurityProfile | None = None _cached_project_dir: Path | None = None +_cached_spec_dir: Path | None = None # Track spec directory for cache key +_cached_profile_mtime: float | None = None # Track file modification time + + +def _get_profile_path(project_dir: Path) -> Path: + """Get the security profile file path for a project.""" + return project_dir / ProjectAnalyzer.PROFILE_FILENAME + + +def _get_profile_mtime(project_dir: Path) -> float | None: + """Get the modification time of the security profile file, or None if not exists.""" + profile_path = _get_profile_path(project_dir) + try: + return profile_path.stat().st_mtime if profile_path.exists() else None + except OSError: + return None def get_security_profile( @@ -28,6 +45,11 @@ def get_security_profile( """ Get the security profile for a project, using cache when possible. + The cache is invalidated when: + - The project directory changes + - The security profile file is created (was None, now exists) + - The security profile file is modified (mtime changed) + Args: project_dir: Project root directory spec_dir: Optional spec directory @@ -35,23 +57,41 @@ def get_security_profile( Returns: SecurityProfile for the project """ - global _cached_profile, _cached_project_dir + global _cached_profile, _cached_project_dir, _cached_spec_dir, _cached_profile_mtime project_dir = Path(project_dir).resolve() + resolved_spec_dir = Path(spec_dir).resolve() if spec_dir else None - # Return cached profile if same project - if _cached_profile is not None and _cached_project_dir == project_dir: - return _cached_profile + # Check if cache is valid (both project_dir and spec_dir must match) + if ( + _cached_profile is not None + and _cached_project_dir == project_dir + and _cached_spec_dir == resolved_spec_dir + ): + # Check if file has been created or modified since caching + current_mtime = _get_profile_mtime(project_dir) + # Cache is valid if: + # - Both are None (file never existed and still doesn't) + # - Both have same mtime (file unchanged) + if current_mtime == _cached_profile_mtime: + return _cached_profile + + # File was created or modified - invalidate cache + # (This happens when analyzer creates the file after agent starts) # Analyze and cache _cached_profile = get_or_create_profile(project_dir, spec_dir) _cached_project_dir = project_dir + _cached_spec_dir = resolved_spec_dir + _cached_profile_mtime = _get_profile_mtime(project_dir) return _cached_profile def reset_profile_cache() -> None: """Reset the cached profile (useful for testing or re-analysis).""" - global _cached_profile, _cached_project_dir + global _cached_profile, _cached_project_dir, _cached_spec_dir, _cached_profile_mtime _cached_profile = None _cached_project_dir = None + _cached_spec_dir = None + _cached_profile_mtime = None diff --git a/tests/test_security_cache.py b/tests/test_security_cache.py new file mode 100644 index 00000000..ccd32b05 --- /dev/null +++ b/tests/test_security_cache.py @@ -0,0 +1,108 @@ +import pytest +import json +import time +import sys +from pathlib import Path + +# Ensure local apps/backend is in path +sys.path.insert(0, str(Path(__file__).parents[1] / "apps" / "backend")) + +from security.profile import get_security_profile, reset_profile_cache +from project.models import SecurityProfile +from project.analyzer import ProjectAnalyzer + +@pytest.fixture +def mock_project_dir(tmp_path): + project_dir = tmp_path / "project" + project_dir.mkdir() + return project_dir + +@pytest.fixture +def mock_profile_path(mock_project_dir): + return mock_project_dir / ".auto-claude-security.json" + +def create_valid_profile_json(commands, project_hash=""): + """Helper to create a valid SecurityProfile JSON structure.""" + return json.dumps({ + "base_commands": commands, + "stack_commands": [], + "script_commands": [], + "custom_commands": [], + "detected_stack": { + "languages": [], + "package_managers": [], + "frameworks": [], + "databases": [], + "infrastructure": [], + "cloud_providers": [], + "code_quality_tools": [], + "version_managers": [] + }, + "custom_scripts": { + "npm_scripts": [], + "make_targets": [], + "poetry_scripts": [], + "cargo_aliases": [], + "shell_scripts": [] + }, + "project_dir": "", + "created_at": "", + "project_hash": project_hash + }) + +def get_dir_hash(project_dir): + return ProjectAnalyzer(project_dir).compute_project_hash() + +def test_cache_invalidation_on_file_creation(mock_project_dir, mock_profile_path): + reset_profile_cache() + + # 1. First call - file doesn't exist + profile1 = get_security_profile(mock_project_dir) + assert "unique_cmd_A" not in profile1.get_all_allowed_commands() + + # 2. Create the file with valid JSON and CORRECT HASH + current_hash = get_dir_hash(mock_project_dir) + mock_profile_path.write_text(create_valid_profile_json(["unique_cmd_A"], current_hash)) + + # 3. Second call - should detect file creation and reload + profile2 = get_security_profile(mock_project_dir) + assert "unique_cmd_A" in profile2.get_all_allowed_commands() + +def test_cache_invalidation_on_file_modification(mock_project_dir, mock_profile_path): + reset_profile_cache() + + # 1. Create initial file + current_hash = get_dir_hash(mock_project_dir) + mock_profile_path.write_text(create_valid_profile_json(["unique_cmd_A"], current_hash)) + + # 2. Load initial profile + profile1 = get_security_profile(mock_project_dir) + assert "unique_cmd_A" in profile1.get_all_allowed_commands() + + # Wait to ensure mtime changes + time.sleep(0.1) + + # 3. Modify the file + mock_profile_path.write_text(create_valid_profile_json(["unique_cmd_B"], current_hash)) + + # 4. Call again - should detect modification + profile2 = get_security_profile(mock_project_dir) + assert "unique_cmd_B" in profile2.get_all_allowed_commands() + +def test_cache_invalidation_on_file_deletion(mock_project_dir, mock_profile_path): + reset_profile_cache() + + # 1. Create file + current_hash = get_dir_hash(mock_project_dir) + mock_profile_path.write_text(create_valid_profile_json(["unique_cmd_A"], current_hash)) + + # 2. Load profile + profile1 = get_security_profile(mock_project_dir) + assert "unique_cmd_A" in profile1.get_all_allowed_commands() + + # 3. Delete file + mock_profile_path.unlink() + + # 4. Call again - should handle deletion gracefully and fallback to fresh analysis + profile2 = get_security_profile(mock_project_dir) + assert "unique_cmd_A" not in profile2.get_all_allowed_commands()