Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions core/runtime_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,10 @@
class FileTextCache:
"""Optional read-through file-text cache used by scan/review passes."""

def __init__(self) -> None:
def __init__(self, *, max_entries: int = 256) -> None:
self._enabled = False
self._values: dict[str, str | None] = {}
self.max_entries = max_entries

def enable(self) -> None:
self._enabled = True
Expand All @@ -25,14 +26,22 @@ def disable(self) -> None:

def read(self, filepath: str) -> str | None:
if self._enabled and filepath in self._values:
return self._values[filepath]
# Refresh position for LRU-like behavior
content = self._values.pop(filepath)
self._values[filepath] = content
return content

try:
content = Path(filepath).read_text(errors="replace")
except OSError:
content = None

if self._enabled:
# Enforce bounds
if len(self._values) >= self.max_entries:
self._values.pop(next(iter(self._values)))
self._values[filepath] = content

return content


Expand Down
4 changes: 4 additions & 0 deletions engine/_plan/persistence.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,10 @@ def save_plan(plan: PlanModel | dict, path: Path | None = None) -> None:

def plan_path_for_state(state_path: Path) -> Path:
"""Derive plan.json path from a state file path."""
name = state_path.name
if name.startswith("state-") and name.endswith(".json"):
lang_part = name[len("state-") : -len(".json")]
return state_path.parent / f"plan-{lang_part}.json"
return state_path.parent / "plan.json"


Expand Down
2 changes: 1 addition & 1 deletion engine/detectors/security/filters.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
)
from engine.policy.zones import FileZoneMap, Zone

_EXCLUDED_SECURITY_ZONES = (Zone.TEST, Zone.CONFIG, Zone.GENERATED, Zone.VENDOR)
_EXCLUDED_SECURITY_ZONES = (Zone.GENERATED, Zone.VENDOR)


def _should_scan_file(filepath: str, zone_map: FileZoneMap | None) -> bool:
Expand Down
18 changes: 9 additions & 9 deletions tests/detectors/security/test_security.py
Original file line number Diff line number Diff line change
Expand Up @@ -185,14 +185,14 @@ def test_hardcoded_secret_short_value_ok(self):
finally:
os.unlink(path)

def test_hardcoded_secret_in_test_zone_skipped(self):
def test_hardcoded_secret_in_test_zone_allowed(self):
content = 'password = "test_secret_value123"'
path = _write_temp_file(content)
try:
zm = _make_zone_map([path], Zone.TEST)
entries, scanned = detect_security_issues([path], zm, "python")
assert entries == []
assert scanned == 0
assert len(entries) > 0
assert scanned == 1
finally:
os.unlink(path)

Expand Down Expand Up @@ -322,25 +322,25 @@ def test_vendor_zone_skipped(self):
finally:
os.unlink(path)

def test_test_zone_skipped(self):
def test_test_zone_allowed(self):
content = 'AWS_KEY = "AKIAIOSFODNN7EXAMPLE"'
path = _write_temp_file(content)
try:
zm = _make_zone_map([path], Zone.TEST)
entries, scanned = detect_security_issues([path], zm, "python")
assert len(entries) == 0
assert scanned == 0
assert len(entries) > 0
assert scanned == 1
finally:
os.unlink(path)

def test_config_zone_skipped(self):
def test_config_zone_allowed(self):
content = 'DB_URL = "postgres://admin:s3cret@localhost:5432/mydb"'
path = _write_temp_file(content)
try:
zm = _make_zone_map([path], Zone.CONFIG)
entries, scanned = detect_security_issues([path], zm, "python")
assert len(entries) == 0
assert scanned == 0
assert len(entries) > 0
assert scanned == 1
finally:
os.unlink(path)

Expand Down
18 changes: 9 additions & 9 deletions tests/detectors/test_security_internals.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,10 +38,10 @@ def test_should_scan_file_production_zone():
assert _should_scan_file("/app/main.py", zm) is True


def test_should_scan_file_test_zone_excluded():
"""Test files are excluded from security scanning."""
def test_should_scan_file_test_zone_allowed():
"""Test files are included in security scanning."""
zm = _make_zone_map([("/tests/test_foo.py", Zone.TEST)])
assert _should_scan_file("/tests/test_foo.py", zm) is False
assert _should_scan_file("/tests/test_foo.py", zm) is True


def test_should_scan_file_vendor_zone_excluded():
Expand All @@ -56,10 +56,10 @@ def test_should_scan_file_generated_zone_excluded():
assert _should_scan_file("/generated/schema.py", zm) is False


def test_should_scan_file_config_zone_excluded():
"""Config files are excluded from security scanning."""
def test_should_scan_file_config_zone_allowed():
"""Config files are included in security scanning."""
zm = _make_zone_map([("/config.py", Zone.CONFIG)])
assert _should_scan_file("/config.py", zm) is False
assert _should_scan_file("/config.py", zm) is True


def test_should_scan_file_script_zone_allowed():
Expand Down Expand Up @@ -122,8 +122,8 @@ def test_should_skip_line_star_comment():

def test_excluded_security_zones_membership():
"""Verify the exact membership of excluded zones."""
assert Zone.TEST in _EXCLUDED_SECURITY_ZONES
assert Zone.CONFIG in _EXCLUDED_SECURITY_ZONES
assert Zone.TEST not in _EXCLUDED_SECURITY_ZONES
assert Zone.CONFIG not in _EXCLUDED_SECURITY_ZONES
assert Zone.GENERATED in _EXCLUDED_SECURITY_ZONES
assert Zone.VENDOR in _EXCLUDED_SECURITY_ZONES
assert Zone.PRODUCTION not in _EXCLUDED_SECURITY_ZONES
Expand Down Expand Up @@ -191,7 +191,7 @@ def test_detect_security_issues_skips_excluded_zone(tmp_path):
"""Files in excluded zones are not scanned."""
f = tmp_path / "test_creds.py"
f.write_text('aws_key = "AKIAIOSFODNN7EXAMPLE"\n')
zm = _make_zone_map([(str(f), Zone.TEST)])
zm = _make_zone_map([(str(f), Zone.GENERATED)])
entries, scanned = detect_security_issues([str(f)], zm, "python")
assert scanned == 0
assert entries == []
Expand Down
Loading