fix(security): invalidate profile cache when file is created/modified (#355)
* fix(security): invalidate profile cache when file is created/modified Fixes #153 The security profile cache was returning stale data even after the .auto-claude-security.json file was created or updated. This caused commands like 'dotnet' to be blocked even when present in the file. Root cause: get_security_profile() cached the profile on first call without checking if the file's mtime changed on subsequent calls. Fix: Track the security profile file's mtime and invalidate the cache when the file is created (mtime goes from None to a value) or modified (mtime changes). This also helps with issue #222 where the profile is created after the agent starts - now the agent will pick up the new profile on the next command validation. Signed-off-by: Black Circle Sentinel <mludlow000@icloud.com> * fix(security): handle deleted profile file and add cache invalidation tests * fix(security): handle deleted profile file and add cache invalidation tests * test(security): improve cache tests with mocks and unique commands * test(security): add mock-free tests for cache invalidation * test(security): fix cache invalidation tests without mocks * fix(security): address review comments and add debug logs for CI hash failure * fix(analyzer): remove debug prints * fix(lint): sort imports in profile.py * fix(security): include spec_dir in cache key to prevent stale profiles The cache key previously only included project_dir, but the profile location can depend on spec_dir. This could cause stale cached profiles to be returned if spec_dir changes between calls. Fix: Add _cached_spec_dir to the cache validation logic and reset function. --------- Signed-off-by: Black Circle Sentinel <mludlow000@icloud.com> Co-authored-by: Andy <119136210+AndyMik90@users.noreply.github.com>
This commit is contained in:
@@ -9,6 +9,7 @@ Uses project_analyzer to create dynamic security profiles based on detected stac
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from project_analyzer import (
|
from project_analyzer import (
|
||||||
|
ProjectAnalyzer,
|
||||||
SecurityProfile,
|
SecurityProfile,
|
||||||
get_or_create_profile,
|
get_or_create_profile,
|
||||||
)
|
)
|
||||||
@@ -20,6 +21,22 @@ from project_analyzer import (
|
|||||||
# Cache the security profile to avoid re-analyzing on every command
|
# Cache the security profile to avoid re-analyzing on every command
|
||||||
_cached_profile: SecurityProfile | None = None
|
_cached_profile: SecurityProfile | None = None
|
||||||
_cached_project_dir: Path | 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(
|
def get_security_profile(
|
||||||
@@ -28,6 +45,11 @@ def get_security_profile(
|
|||||||
"""
|
"""
|
||||||
Get the security profile for a project, using cache when possible.
|
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:
|
Args:
|
||||||
project_dir: Project root directory
|
project_dir: Project root directory
|
||||||
spec_dir: Optional spec directory
|
spec_dir: Optional spec directory
|
||||||
@@ -35,23 +57,41 @@ def get_security_profile(
|
|||||||
Returns:
|
Returns:
|
||||||
SecurityProfile for the project
|
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()
|
project_dir = Path(project_dir).resolve()
|
||||||
|
resolved_spec_dir = Path(spec_dir).resolve() if spec_dir else None
|
||||||
|
|
||||||
# Return cached profile if same project
|
# 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:
|
if (
|
||||||
return _cached_profile
|
_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
|
# Analyze and cache
|
||||||
_cached_profile = get_or_create_profile(project_dir, spec_dir)
|
_cached_profile = get_or_create_profile(project_dir, spec_dir)
|
||||||
_cached_project_dir = project_dir
|
_cached_project_dir = project_dir
|
||||||
|
_cached_spec_dir = resolved_spec_dir
|
||||||
|
_cached_profile_mtime = _get_profile_mtime(project_dir)
|
||||||
|
|
||||||
return _cached_profile
|
return _cached_profile
|
||||||
|
|
||||||
|
|
||||||
def reset_profile_cache() -> None:
|
def reset_profile_cache() -> None:
|
||||||
"""Reset the cached profile (useful for testing or re-analysis)."""
|
"""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_profile = None
|
||||||
_cached_project_dir = None
|
_cached_project_dir = None
|
||||||
|
_cached_spec_dir = None
|
||||||
|
_cached_profile_mtime = None
|
||||||
|
|||||||
@@ -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()
|
||||||
Reference in New Issue
Block a user