fix(security): resolve fragile filesystem string matching in blast radius#516
Conversation
📝 WalkthroughWalkthroughBlastRadiusEstimator's filesystem impact detection is refactored from substring/regex matching to a tokenizer-based approach using shlex.split, extracting rm invocations with flags and paths, normalizing paths, and matching against an expanded critical path set including root. A new robustness test validates detection across command variants. ChangesBlast Radius Filesystem Detection Refactor
Estimated code review effort: 3 (Moderate) | ~20 minutes Possibly related issues
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
🧪 PR Test Results
Python 3.12 · commit 2971b5b |
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
agentwatch/core/blast_radius.py (2)
135-137: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueMove
shlex/osimports to module scope.These are re-imported on every call to
_analyze_filesystem_impact. Minor, but hoisting to the top of the file is idiomatic and avoids repeated import machinery on a hot path.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@agentwatch/core/blast_radius.py` around lines 135 - 137, The `_analyze_filesystem_impact` path is re-importing `shlex` and `os` inside the function, which should be hoisted to module scope. Move those imports to the top-level imports in `blast_radius.py` and remove the local imports from `_analyze_filesystem_impact` so the function uses the module-level symbols instead.
139-142: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueBroad
except Exceptionaroundshlex.split.
shlex.splitraisesValueErroron unbalanced quotes; catching the narrowerValueErrorinstead of bareExceptionwould be more precise and avoid silently swallowing unrelated bugs.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@agentwatch/core/blast_radius.py` around lines 139 - 142, The exception handling around shlex.split in blast_radius.py is too broad; narrow the catch in the tokenization logic to ValueError only, since shlex.split raises that for malformed input and the current broad handler can hide unrelated bugs. Update the try/except in the code path that assigns tokens from text so the fallback to text.split() only runs for the expected parse error.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@agentwatch/core/blast_radius.py`:
- Around line 138-167: The parse_and_scan helper in blast_radius.py is missing
full-path rm executions because it only matches tokens equal to "rm" or
containing "rm " / "rm\t". Update the token scan logic in parse_and_scan to also
recognize commands where the executable token is a path to rm, such as by
checking os.path.basename(token) == "rm", while preserving the existing
flag/path collection behavior for rm invocations.
- Around line 169-193: The critical-path check in blast_radius.py is too broad
because is_sub_or_equal treats "/" as matching every path, so any recursive
delete becomes critical. Update the logic around is_sub_or_equal and the
critical_paths set to exclude "/" from the general critical-directory matching,
and handle root deletions with a dedicated exact-match check instead. Keep the
existing scoring path in the invocation loop, but ensure
radius.is_critical_resource and radius.score are only raised for real critical
directories or an explicit root delete.
In `@tests/test_blast_radius_causal.py`:
- Around line 95-136: Add a negative regression test in
test_blast_radius_filesystem_robustness for a non-critical recursive delete such
as rm -rf /tmp/scratch or rm -rf ./build, and assert
BlastRadiusEstimator.estimate does not return a critical score. Use the existing
_tool_event helper and verify radius.score is not 100 and
radius.is_critical_resource is False so the root-inclusion bug in
BlastRadiusEstimator is caught by a non-critical path case.
---
Nitpick comments:
In `@agentwatch/core/blast_radius.py`:
- Around line 135-137: The `_analyze_filesystem_impact` path is re-importing
`shlex` and `os` inside the function, which should be hoisted to module scope.
Move those imports to the top-level imports in `blast_radius.py` and remove the
local imports from `_analyze_filesystem_impact` so the function uses the
module-level symbols instead.
- Around line 139-142: The exception handling around shlex.split in
blast_radius.py is too broad; narrow the catch in the tokenization logic to
ValueError only, since shlex.split raises that for malformed input and the
current broad handler can hide unrelated bugs. Update the try/except in the code
path that assigns tokens from text so the fallback to text.split() only runs for
the expected parse error.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 1839ebf1-d1a5-4b80-ba8d-03bb6d1b01ca
📒 Files selected for processing (2)
agentwatch/core/blast_radius.pytests/test_blast_radius_causal.py
| def parse_and_scan(text: str) -> list[tuple[set[str], list[str]]]: | ||
| try: | ||
| tokens = shlex.split(text) | ||
| except Exception: | ||
| tokens = text.split() | ||
|
|
||
| invocations = [] | ||
| i = 0 | ||
| while i < len(tokens): | ||
| token = tokens[i] | ||
| if any(sub in token for sub in ("rm ", "rm\t")): | ||
| invocations.extend(parse_and_scan(token)) | ||
| elif token == "rm": | ||
| flags = set() | ||
| paths = [] | ||
| j = i + 1 | ||
| while j < len(tokens): | ||
| arg = tokens[j] | ||
| if arg.startswith("-") and len(arg) > 1: | ||
| if arg.startswith("--"): | ||
| flags.add(arg) | ||
| else: | ||
| for char in arg[1:]: | ||
| flags.add(f"-{char}") | ||
| else: | ||
| paths.append(arg) | ||
| j += 1 | ||
| invocations.append((flags, paths)) | ||
| i += 1 | ||
| return invocations |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Regression: full-path rm invocations (e.g. /bin/rm -rf /etc) are no longer detected.
parse_and_scan only extracts an invocation when a token equals exactly "rm" (Line 150). A binary invoked with a full path, e.g. /bin/rm -rf /etc or /usr/bin/rm -rf /etc, produces the token /bin/rm, which fails both the token == "rm" check and the "rm " in token substring check (no trailing space inside that single token). As a result this command produces zero invocations and is silently missed — a regression from the prior naive substring approach, which would still have matched "rm -rf" as a raw substring even inside /bin/rm -rf /etc.
Consider also matching on os.path.basename(token) == "rm" to catch fully-qualified paths.
🧰 Tools
🪛 GitHub Actions: Bandit Security Scan / 0_Bandit Scan.txt
[error] 150-150: Bandit failed on reported issue [B105:hardcoded_password_string] (Possible hardcoded REDACTED). Run: bandit -r agentwatch -b bandit-baseline.json. Location: agentwatch/core/blast_radius.py:150:30
🪛 GitHub Actions: Bandit Security Scan / Bandit Scan
[error] 150-150: Bandit found a security issue (B105: hardcoded_password_string). Possible hardcoded password string. CWE-259. Location: agentwatch/core/blast_radius.py:150:30
🪛 GitHub Actions: PR Tests / 0_Test & lint.txt
[error] 150-152: ruff check (S105): Possible hardcoded password assigned to "token". Remove or properly secure the credential.
🪛 GitHub Actions: PR Tests / Test & lint
[error] 150-150: ruff check . (S105) Possible hardcoded password assigned to: "token".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentwatch/core/blast_radius.py` around lines 138 - 167, The parse_and_scan
helper in blast_radius.py is missing full-path rm executions because it only
matches tokens equal to "rm" or containing "rm " / "rm\t". Update the token scan
logic in parse_and_scan to also recognize commands where the executable token is
a path to rm, such as by checking os.path.basename(token) == "rm", while
preserving the existing flag/path collection behavior for rm invocations.
| def is_sub_or_equal(target: str, critical: str) -> bool: | ||
| # Normalize target path to handle relative components and cross-platform slashes | ||
| normalized = os.path.normpath(target).replace("\\", "/") | ||
| if not normalized.startswith("/") and target.startswith("/"): | ||
| normalized = "/" + normalized | ||
|
|
||
| target_parts = [p for p in normalized.split("/") if p] | ||
| critical_parts = [p for p in critical.split("/") if p] | ||
|
|
||
| if len(target_parts) >= len(critical_parts): | ||
| return target_parts[:len(critical_parts)] == critical_parts | ||
| return False | ||
|
|
||
| critical_paths = {"/etc", "/var", "/boot", "/root", "/home", "/usr", "/bin", "/sbin", "/"} | ||
| invocations = parse_and_scan(raw) | ||
|
|
||
| for flags, paths in invocations: | ||
| is_recursive = any(f in flags for f in ("-r", "-R", "--recursive")) | ||
| for path in paths: | ||
| # Check for critical path deletions (recursive or not, but usually recursive is critical) | ||
| if any(is_sub_or_equal(path, cp) for cp in critical_paths): | ||
| if is_recursive or path in ("/", "/*"): | ||
| radius.is_critical_resource = True | ||
| radius.score = max(radius.score, 100) | ||
| radius.affected_file_count = 50000 |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
Critical: including "/" in critical_paths makes is_sub_or_equal always return True, flagging every recursive delete as critical.
is_sub_or_equal(target, "/") splits "/" into critical_parts = [] (since "/".split("/") filtered of empty strings is []). The check target_parts[:0] == [] == critical_parts is always True regardless of target. So any(is_sub_or_equal(path, cp) for cp in critical_paths) is unconditionally True for every path (even relative paths like ./build or /tmp/scratch), because critical_paths includes "/".
Combined with if is_recursive or path in ("/", "/*"): at Line 190, this means any rm -rf <anything> — not just deletions under /etc, /var, etc. — gets is_critical_resource = True and score = 100. This directly undermines the PR's stated goal of "more accurately detect[ing] deletions inside critical directories," and isn't caught by the current tests since they only exercise genuinely critical targets.
🐛 Proposed fix
- critical_paths = {"/etc", "/var", "/boot", "/root", "/home", "/usr", "/bin", "/sbin", "/"}
+ critical_paths = {"/etc", "/var", "/boot", "/root", "/home", "/usr", "/bin", "/sbin"}
invocations = parse_and_scan(raw)
for flags, paths in invocations:
is_recursive = any(f in flags for f in ("-r", "-R", "--recursive"))
for path in paths:
+ is_root_delete = path in ("/", "/*") or os.path.normpath(path) == "/"
# Check for critical path deletions (recursive or not, but usually recursive is critical)
- if any(is_sub_or_equal(path, cp) for cp in critical_paths):
- if is_recursive or path in ("/", "/*"):
+ if is_root_delete or any(is_sub_or_equal(path, cp) for cp in critical_paths):
+ if is_recursive or is_root_delete:
radius.is_critical_resource = True
radius.score = max(radius.score, 100)
radius.affected_file_count = 50000📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def is_sub_or_equal(target: str, critical: str) -> bool: | |
| # Normalize target path to handle relative components and cross-platform slashes | |
| normalized = os.path.normpath(target).replace("\\", "/") | |
| if not normalized.startswith("/") and target.startswith("/"): | |
| normalized = "/" + normalized | |
| target_parts = [p for p in normalized.split("/") if p] | |
| critical_parts = [p for p in critical.split("/") if p] | |
| if len(target_parts) >= len(critical_parts): | |
| return target_parts[:len(critical_parts)] == critical_parts | |
| return False | |
| critical_paths = {"/etc", "/var", "/boot", "/root", "/home", "/usr", "/bin", "/sbin", "/"} | |
| invocations = parse_and_scan(raw) | |
| for flags, paths in invocations: | |
| is_recursive = any(f in flags for f in ("-r", "-R", "--recursive")) | |
| for path in paths: | |
| # Check for critical path deletions (recursive or not, but usually recursive is critical) | |
| if any(is_sub_or_equal(path, cp) for cp in critical_paths): | |
| if is_recursive or path in ("/", "/*"): | |
| radius.is_critical_resource = True | |
| radius.score = max(radius.score, 100) | |
| radius.affected_file_count = 50000 | |
| def is_sub_or_equal(target: str, critical: str) -> bool: | |
| # Normalize target path to handle relative components and cross-platform slashes | |
| normalized = os.path.normpath(target).replace("\\", "/") | |
| if not normalized.startswith("/") and target.startswith("/"): | |
| normalized = "/" + normalized | |
| target_parts = [p for p in normalized.split("/") if p] | |
| critical_parts = [p for p in critical.split("/") if p] | |
| if len(target_parts) >= len(critical_parts): | |
| return target_parts[:len(critical_parts)] == critical_parts | |
| return False | |
| critical_paths = {"/etc", "/var", "/boot", "/root", "/home", "/usr", "/bin", "/sbin"} | |
| invocations = parse_and_scan(raw) | |
| for flags, paths in invocations: | |
| is_recursive = any(f in flags for f in ("-r", "-R", "--recursive")) | |
| for path in paths: | |
| is_root_delete = path in ("/", "/*") or os.path.normpath(path) == "/" | |
| # Check for critical path deletions (recursive or not, but usually recursive is critical) | |
| if is_root_delete or any(is_sub_or_equal(path, cp) for cp in critical_paths): | |
| if is_recursive or is_root_delete: | |
| radius.is_critical_resource = True | |
| radius.score = max(radius.score, 100) | |
| radius.affected_file_count = 50000 |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentwatch/core/blast_radius.py` around lines 169 - 193, The critical-path
check in blast_radius.py is too broad because is_sub_or_equal treats "/" as
matching every path, so any recursive delete becomes critical. Update the logic
around is_sub_or_equal and the critical_paths set to exclude "/" from the
general critical-directory matching, and handle root deletions with a dedicated
exact-match check instead. Keep the existing scoring path in the invocation
loop, but ensure radius.is_critical_resource and radius.score are only raised
for real critical directories or an explicit root delete.
| def test_blast_radius_filesystem_robustness(): | ||
| estimator = BlastRadiusEstimator() | ||
|
|
||
| # 1. Test different flags and spacing | ||
| event = _tool_event("bash", "rm -f -r /etc") | ||
| radius = estimator.estimate(event) | ||
| assert radius.score == 100 | ||
| assert radius.is_critical_resource is True | ||
|
|
||
| # 2. Test long option flags | ||
| event = _tool_event("bash", "rm --recursive --force /etc") | ||
| radius = estimator.estimate(event) | ||
| assert radius.score == 100 | ||
|
|
||
| # 3. Test quotes around target | ||
| event = _tool_event("bash", "rm -rf '/etc'") | ||
| radius = estimator.estimate(event) | ||
| assert radius.score == 100 | ||
|
|
||
| event = _tool_event("bash", 'rm -rf "/etc/"') | ||
| radius = estimator.estimate(event) | ||
| assert radius.score == 100 | ||
|
|
||
| # 4. Test subdirectories | ||
| event = _tool_event("bash", "rm -rf /etc/shadow") | ||
| radius = estimator.estimate(event) | ||
| assert radius.score == 100 | ||
|
|
||
| event = _tool_event("bash", "rm -rf /var/log/nginx/access.log") | ||
| radius = estimator.estimate(event) | ||
| assert radius.score == 100 | ||
|
|
||
| # 5. Test path traversal bypass attempts | ||
| event = _tool_event("bash", "rm -rf /tmp/../etc") | ||
| radius = estimator.estimate(event) | ||
| assert radius.score == 100 | ||
|
|
||
| # 6. Test nested commands (like in bash execution) | ||
| event = _tool_event("bash", 'bash -c "rm -rf /etc"') | ||
| radius = estimator.estimate(event) | ||
| assert radius.score == 100 | ||
|
|
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Add a negative test case for non-critical recursive deletes.
All new scenarios target /etc or /var. Given the root-inclusion bug flagged in blast_radius.py (Lines 169-193), a case like rm -rf /tmp/scratch or rm -rf ./build asserting score != 100 / is_critical_resource is False would have caught that regression and guards against future reintroduction.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/test_blast_radius_causal.py` around lines 95 - 136, Add a negative
regression test in test_blast_radius_filesystem_robustness for a non-critical
recursive delete such as rm -rf /tmp/scratch or rm -rf ./build, and assert
BlastRadiusEstimator.estimate does not return a critical score. Use the existing
_tool_event helper and verify radius.score is not 100 and
radius.is_critical_resource is False so the root-inclusion bug in
BlastRadiusEstimator is caught by a non-critical path case.
Summary
Resolves issue #356.
Refactors the filesystem impact analysis within the Blast Radius Estimator to avoid fragile string matching and support more robust command analysis.
Proposed Changes
agentwatch/core/blast_radius.py:"rm -rf"sub-string checks) with a robust command token parser usingshlex.split.-r,-R,--recursive,-f, etc.) and quoted arguments.is_sub_or_equal) to accurately flag deletions inside critical directories (like/etc,/var, etc.) and path-traversal attempts.tests/test_blast_radius_causal.py:test_blast_radius_filesystem_robustnesscovering different flag spacings, long-option flags, quoted targets, subdirectory matching, path traversal bypasses (/tmp/../etc), and nested commands.Verification
python -m pytest tests/test_blast_radius_causal.py -k test_blast_radius_filesystem_robustnesswhich passes successfully.Summary by CodeRabbit