feat: add CCS security integration for CrewAI tool execution - #6820
feat: add CCS security integration for CrewAI tool execution#6820Correctover wants to merge 5 commits into
Conversation
Remove ToolCallHookContext and canonical_json from imports and __all__. These symbols are not defined in guardrail_provider.py and cause ImportError.
…results) - All 4 providers now pass expires_at to compute_decision_id (anti-tamper) - Docstring example fixed to match - CKGGuardrailProvider results: dict→list[tuple] to prevent overwrite - make_guardrail_hook: wrapper fn instead of bound method attr assignment - Brand attribution already present (CCS/Correctover)
…deRabbit) - Replace ctx.trail._decisions with ctx.trail.all_decisions()[0] - Tests should use public API, not implementation details
📝 WalkthroughWalkthroughThe change adds a runtime guardrail framework with authorization providers, integrity checks, audit trails, hooks, and configuration scanning. It also adds a CCS Security middleware example that verifies and wraps CrewAI tool calls. ChangesRuntime guardrail system
CCS Security example
Sequence Diagram(s)sequenceDiagram
participant ToolCallHook
participant GuardrailContext
participant GuardrailProvider
participant AuditTrail
ToolCallHook->>GuardrailContext: authorize tool call
GuardrailContext->>GuardrailProvider: authorize(context)
GuardrailProvider-->>GuardrailContext: GuardrailDecisionV1
GuardrailContext->>AuditTrail: record decision
GuardrailContext-->>ToolCallHook: allow or block result
sequenceDiagram
participant CrewAITool
participant CCSSecurityMiddleware
participant CCSVerifier
CrewAITool->>CCSSecurityMiddleware: run(arguments)
CCSSecurityMiddleware->>CCSVerifier: verify command
CCSVerifier-->>CCSSecurityMiddleware: allow or deny with reason
CCSSecurityMiddleware-->>CrewAITool: block or call original run
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 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 Warning |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🧹 Nitpick comments (6)
src/crewai/guardrails/guardrail_provider.py (2)
489-511: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the commented-out
after_tool_callblock.The block is dead code. It also holds the only usage of
ActionEnvelopeV1anddigest_result, so both are unused production surface right now. Delete the block and open a follow-up issue, or implement and registerafter_tool_callwith tests.Do you want me to open an issue to track post-execution envelope capture?
🤖 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 `@src/crewai/guardrails/guardrail_provider.py` around lines 489 - 511, Remove the commented-out after_tool_call block from GuardrailProvider, including its references to ActionEnvelopeV1 and digest_result; do not implement or register the future hook as part of this change.
306-323: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAlign the CKG constraint documentation with the implemented contract.
Three documentation statements do not match the code.
- Line 311 shows
CKGGuardrailProvider(constraints=[_no_shell_exec]).authorizeunpacks each entry withfor predicate_name, params in self._constraints, so a bare callable fails. Constraints must be(name, params_dict)tuples.- Line 302 describes
predicate(context, **params)as a caller-supplied callable._eval_predicateresolves builtin names only and raisesValueErrorfor anything else._eval_predicateline 366 states thatparam_matchesmatches a regex or a value; the implementation only compares equality. Line 370 states that custom callables can be registered during init; no such path exists.Update the docstrings, or add real callable support. As per coding guidelines: "Document public APIs and complex logic in Python code."
♻️ Proposed docstring fix
- Each constraint is a (predicate, params) tuple evaluated as: - predicate(context, **params) -> bool (True = constraint satisfied) + Each constraint is a (predicate_name, params) tuple. predicate_name must + name a built-in predicate (see _eval_predicate); params is a dict of + keyword arguments for that predicate. True means the constraint holds. A call is authorized iff ALL constraints are satisfied. Example:: - def _no_shell_exec(ctx, **_): - return ctx.tool_name != "run_shell_command" - - guardrail = CKGGuardrailProvider(constraints=[_no_shell_exec]) + guardrail = CKGGuardrailProvider( + constraints=[("tool_name_not_in", {"names": ["run_shell_command"]})] + ) """🤖 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 `@src/crewai/guardrails/guardrail_provider.py` around lines 306 - 323, Update the CKGGuardrailProvider docstrings to match the implemented contract: document constraints as (predicate_name, params_dict) tuples, describe predicate resolution as builtin-name-only rather than caller-supplied callables, state that param_matches uses equality comparison only, and remove the claim that custom callables can be registered during initialization. Keep the implementation unchanged unless needed to ensure the documentation accurately reflects __init__, add_constraint, authorize, and _eval_predicate.Source: Coding guidelines
tests/guardrails/test_guardrail_provider.py (3)
711-728: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRename the test and delete the unused locals.
The name
test_compute_decision_id_determinism_across_providerspromises equal IDs, but the test asserts that the IDs differ.claimsat line 713 andexpires_atat line 714 are never used. Rename to something liketest_decision_id_binds_to_provider_specific_claims, and remove the two unused locals. Add a separate test that callscompute_decision_idtwice with the same claims and the sameexpires_atto cover determinism directly.🤖 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/guardrails/test_guardrail_provider.py` around lines 711 - 728, Rename test_compute_decision_id_determinism_across_providers to reflect that provider-specific claims produce distinct decision IDs, and remove its unused claims and expires_at locals. Add a separate test that invokes compute_decision_id twice with identical claims and expires_at values and asserts the results are equal.
126-137: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAssert behavior instead of attribute absence.
assert not hasattr(env, "tool_result")locks the test to the current field layout. The behavior under test is that the envelope stores a digest and never the raw result. Assert thattool_result_digestequalsdigest_result(raw_value)for a known raw value, and that the raw value does not appear in the envelope fields.As per coding guidelines: "Write unit tests for new functionality that focus on behavior rather than implementation details."
🤖 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/guardrails/test_guardrail_provider.py` around lines 126 - 137, Update test_never_stores_raw_result to derive tool_result_digest by applying digest_result to a known raw value, then assert the envelope stores that digest and does not contain the raw value among its fields. Remove the hasattr(env, "tool_result") assertion so the test verifies behavior rather than the current attribute layout.Source: Coding guidelines
214-227: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConstrain the factory context to the actual hook context.
_make_contextreturns a bareMagicMock, so missingToolCallHookContextfields would not fail tests. If you need this factory, useMagicMock(spec=ToolCallHookContext)or a small frozen dataclass stub with the same fields.🤖 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/guardrails/test_guardrail_provider.py` around lines 214 - 227, Update the _make_context test factory to create a constrained ToolCallHookContext-shaped object instead of a bare MagicMock. Prefer MagicMock(spec=ToolCallHookContext), or use a frozen dataclass stub containing the required context fields, while preserving the existing tool_name, tool_input, agent, task, and crew setup.src/crewai/guardrails/__init__.py (1)
14-42: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMake the package exports match the module's public surface.
guardrail_provider.__all__listsmake_decision_idanddigest_result, but the package does not re-export them.digest_resultis required by any consumer that builds anActionEnvelopeV1, and the package does exportActionEnvelopeV1. Add both names, or remove them from the module's__all__.Also remove the stray comment on line 45. Commit notes belong in the commit message, not in the package initializer.
♻️ Proposed fix
make_guardrail_hook, detect_missing_guardrail, compute_decision_id, + make_decision_id, + digest_result, ) __all__ = ["detect_missing_guardrail", "compute_decision_id", + "make_decision_id", + "digest_result", ]🤖 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 `@src/crewai/guardrails/__init__.py` around lines 14 - 42, Update the guardrail package exports in the __init__.py import list and __all__ to match guardrail_provider.__all__: re-export make_decision_id and digest_result alongside ActionEnvelopeV1, or remove those names from the module’s public surface consistently. Also remove the stray comment near the end of the package initializer.
🤖 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 `@examples/ccs-security/ccs_guard.py`:
- Around line 6-10: Correct the documented import path in
examples/ccs-security/ccs_guard.py lines 6-10 by using the supported packaged or
renamed module path, or documenting the required copy/install method; apply the
identical correction in examples/ccs-security/README.md lines 7-12 so both
locations consistently describe an importable setup.
- Around line 62-64: Update verify_tool_call() and the secured_run() wrapping
flow so an unavailable ccs_verifier cannot permit original_run() to execute.
Require a successfully initialized verifier when creating the wrapper, or return
a denial result and prevent tool execution whenever get_verifier() returns None;
preserve normal verification behavior when the verifier is available.
In `@src/crewai/guardrails/guardrail_provider.py`:
- Around line 396-411: The AuditTrail currently overwrites repeated identical
records because it keys storage only by decision_id. In
src/crewai/guardrails/guardrail_provider.py lines 396-411, update AuditTrail to
append decisions and envelopes to ordered lists while retaining the dicts only
as lookup indexes; have all_decisions, all_envelopes, total_decisions, and
total_envelopes use those lists, clear both structures in clear(), and correct
the docstring’s thread-safety claim. In
tests/guardrails/test_guardrail_provider.py lines 697-709, preserve the
total_decisions == len(calls) and granted == 3 assertions and add coverage
proving two identical consecutive calls produce two recorded decisions.
- Around line 556-562: Update the tool-name extraction in the guardrail scanner
around _has_guardrail_provider so non-string tool entries are handled safely,
including BaseTool instances, without calling mapping-only .get on arbitrary
objects. Resolve the tool name via the object’s supported name attribute or
established tool interface, retain dictionary handling, and continue recording
tool:name entries for tools lacking a GuardrailProvider.
- Around line 326-352: Update the denial-reason comprehension in
CKGGuardrailProvider’s authorize flow to iterate directly over the results list
rather than calling results.items(). Preserve the existing successful reason and
ensure failed predicates are collected from each (name, status) tuple so denied
authorization returns normally.
- Around line 57-61: Update is_expired to use the provided now value whenever it
is not None, including 0 and 0.0; only call time.time() when now is explicitly
None, while preserving the existing expires_at check and comparison.
- Around line 247-273: Update ToolListGuardrailProvider.authorize so
default_block=False still permits tools in _allowed and only changes the
treatment of tools outside the allowlist; align the constructor documentation,
parameter semantics, and test_inverted_mode with this allowlist contract.
Preserve the existing default_block=True behavior for listed and unlisted tools.
- Around line 127-133: Update make_decision_id to accept and forward the
decision’s expires_at value to compute_decision_id so IDs generated for expiring
decisions are accepted by verify_integrity; alternatively remove the helper if
it is not intended to support that contract. If retained, export
make_decision_id from the package initializer and add coverage for non-None
expires_at integrity verification.
---
Nitpick comments:
In `@src/crewai/guardrails/__init__.py`:
- Around line 14-42: Update the guardrail package exports in the __init__.py
import list and __all__ to match guardrail_provider.__all__: re-export
make_decision_id and digest_result alongside ActionEnvelopeV1, or remove those
names from the module’s public surface consistently. Also remove the stray
comment near the end of the package initializer.
In `@src/crewai/guardrails/guardrail_provider.py`:
- Around line 489-511: Remove the commented-out after_tool_call block from
GuardrailProvider, including its references to ActionEnvelopeV1 and
digest_result; do not implement or register the future hook as part of this
change.
- Around line 306-323: Update the CKGGuardrailProvider docstrings to match the
implemented contract: document constraints as (predicate_name, params_dict)
tuples, describe predicate resolution as builtin-name-only rather than
caller-supplied callables, state that param_matches uses equality comparison
only, and remove the claim that custom callables can be registered during
initialization. Keep the implementation unchanged unless needed to ensure the
documentation accurately reflects __init__, add_constraint, authorize, and
_eval_predicate.
In `@tests/guardrails/test_guardrail_provider.py`:
- Around line 711-728: Rename
test_compute_decision_id_determinism_across_providers to reflect that
provider-specific claims produce distinct decision IDs, and remove its unused
claims and expires_at locals. Add a separate test that invokes
compute_decision_id twice with identical claims and expires_at values and
asserts the results are equal.
- Around line 126-137: Update test_never_stores_raw_result to derive
tool_result_digest by applying digest_result to a known raw value, then assert
the envelope stores that digest and does not contain the raw value among its
fields. Remove the hasattr(env, "tool_result") assertion so the test verifies
behavior rather than the current attribute layout.
- Around line 214-227: Update the _make_context test factory to create a
constrained ToolCallHookContext-shaped object instead of a bare MagicMock.
Prefer MagicMock(spec=ToolCallHookContext), or use a frozen dataclass stub
containing the required context fields, while preserving the existing tool_name,
tool_input, agent, task, and crew setup.
🪄 Autofix
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: 789472c7-df8f-47b7-81bb-f1e818b99c77
📒 Files selected for processing (5)
examples/ccs-security/README.mdexamples/ccs-security/ccs_guard.pysrc/crewai/guardrails/__init__.pysrc/crewai/guardrails/guardrail_provider.pytests/guardrails/test_guardrail_provider.py
| Usage: | ||
| from examples.ccs_security.ccs_guard import CCSSecurityMiddleware | ||
|
|
||
| # Wrap any CrewAI tool with CCS verification | ||
| secured_tool = CCSSecurityMiddleware.wrap_tool(my_tool) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Correct the documented Python module path.
examples.ccs_security.ccs_guard does not map to examples/ccs-security/ccs_guard.py. Python cannot import a module from a directory named ccs-security.
examples/ccs-security/ccs_guard.py#L6-L10: use the actual import path after renaming and packaging the example directory, or document the supported copy/install method.examples/ccs-security/README.md#L7-L12: use the same corrected import path.
📍 Affects 2 files
examples/ccs-security/ccs_guard.py#L6-L10(this comment)examples/ccs-security/README.md#L7-L12
🤖 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 `@examples/ccs-security/ccs_guard.py` around lines 6 - 10, Correct the
documented import path in examples/ccs-security/ccs_guard.py lines 6-10 by using
the supported packaged or renamed module path, or documenting the required
copy/install method; apply the identical correction in
examples/ccs-security/README.md lines 7-12 so both locations consistently
describe an importable setup.
| verifier = cls.get_verifier() | ||
| if verifier is None: | ||
| return True, "CCS not available" |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Locate ccs_guard.py =="
fd -a 'ccs_guard\.py$' . || true
echo "== Outline ccs_guard.py =="
ast-grep outline examples/ccs-security/ccs_guard.py --view expanded || true
echo "== Relevant ccs_guard.py lines =="
cat -n examples/ccs-security/ccs_guard.py | sed -n '1,140p'
echo "== Search verifier availability references =="
rg -n "ccs_verifier|get_verifier|secured_run|secured_tools|CCS not available" examples/ccs-security lib .github README* 2>/dev/null || true
echo "== BaseTool run snippet =="
cat -n lib/crewai/src/crewai/tools/base_tool.py | sed -n '300,340p'Repository: crewAIInc/crewAI
Length of output: 7066
Security Misconfiguration (CWE-693)
Reachability: External
Fail closed when CCS is unavailable.
When ccs_verifier cannot be imported or initialized, verify_tool_call() returns (True, "CCS not available"), and secured_run() calls original_run() without verification. Require the verifier during wrapping, or deny tool execution while CCS is unavailable.
🤖 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 `@examples/ccs-security/ccs_guard.py` around lines 62 - 64, Update
verify_tool_call() and the secured_run() wrapping flow so an unavailable
ccs_verifier cannot permit original_run() to execute. Require a successfully
initialized verifier when creating the wrapper, or return a denial result and
prevent tool execution whenever get_verifier() returns None; preserve normal
verification behavior when the verifier is available.
| def is_expired(self, now: float | None = None) -> bool: | ||
| """Check whether this decision has expired.""" | ||
| if self.expires_at is None: | ||
| return False | ||
| return (now or time.time()) > self.expires_at |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Use an explicit None check for now.
(now or time.time()) treats now=0 and now=0.0 as "not provided". A caller that passes an epoch-0 timestamp gets the wall clock instead. Compare against None.
🐛 Proposed fix
def is_expired(self, now: float | None = None) -> bool:
"""Check whether this decision has expired."""
if self.expires_at is None:
return False
- return (now or time.time()) > self.expires_at
+ current = time.time() if now is None else now
+ return current > self.expires_at📝 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_expired(self, now: float | None = None) -> bool: | |
| """Check whether this decision has expired.""" | |
| if self.expires_at is None: | |
| return False | |
| return (now or time.time()) > self.expires_at | |
| def is_expired(self, now: float | None = None) -> bool: | |
| """Check whether this decision has expired.""" | |
| if self.expires_at is None: | |
| return False | |
| current = time.time() if now is None else now | |
| return current > self.expires_at |
🤖 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 `@src/crewai/guardrails/guardrail_provider.py` around lines 57 - 61, Update
is_expired to use the provided now value whenever it is not None, including 0
and 0.0; only call time.time() when now is explicitly None, while preserving the
existing expires_at check and comparison.
| def make_decision_id(namespace: str, payload: dict[str, Any]) -> str: | ||
| """Convenience: hash an identifier namespace into the computation. | ||
|
|
||
| Namespacing prevents decision_id collisions across different tools | ||
| or agents within the same crew. | ||
| """ | ||
| return compute_decision_id({"namespace": namespace, **payload}) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
make_decision_id cannot produce an ID that verify_integrity accepts.
make_decision_id calls compute_decision_id without expires_at. A decision built with decision_id=make_decision_id(...) and a non-None expires_at always fails verify_integrity(). The function is also unexported from the package initializer and untested. Either forward expires_at or remove the helper.
♻️ Proposed fix
-def make_decision_id(namespace: str, payload: dict[str, Any]) -> str:
+def make_decision_id(
+ namespace: str,
+ payload: dict[str, Any],
+ expires_at: float | None = None,
+) -> str:
"""Convenience: hash an identifier namespace into the computation.
Namespacing prevents decision_id collisions across different tools
or agents within the same crew.
"""
- return compute_decision_id({"namespace": namespace, **payload})
+ return compute_decision_id({"namespace": namespace, **payload}, expires_at)📝 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 make_decision_id(namespace: str, payload: dict[str, Any]) -> str: | |
| """Convenience: hash an identifier namespace into the computation. | |
| Namespacing prevents decision_id collisions across different tools | |
| or agents within the same crew. | |
| """ | |
| return compute_decision_id({"namespace": namespace, **payload}) | |
| def make_decision_id( | |
| namespace: str, | |
| payload: dict[str, Any], | |
| expires_at: float | None = None, | |
| ) -> str: | |
| """Convenience: hash an identifier namespace into the computation. | |
| Namespacing prevents decision_id collisions across different tools | |
| or agents within the same crew. | |
| """ | |
| return compute_decision_id({"namespace": namespace, **payload}, expires_at) |
🤖 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 `@src/crewai/guardrails/guardrail_provider.py` around lines 127 - 133, Update
make_decision_id to accept and forward the decision’s expires_at value to
compute_decision_id so IDs generated for expiring decisions are accepted by
verify_integrity; alternatively remove the helper if it is not intended to
support that contract. If retained, export make_decision_id from the package
initializer and add coverage for non-None expires_at integrity verification.
| class ToolListGuardrailProvider(GuardrailProvider): | ||
| """Allows only an explicit allowlist of tool names. | ||
|
|
||
| Args: | ||
| allowed_tools: Set of tool names permitted for execution. | ||
| default_block: Whether to block (True) or allow (False) tools | ||
| not in the allowlist. | ||
|
|
||
| Example:: | ||
|
|
||
| guardrail = ToolListGuardrailProvider( | ||
| allowed_tools={"read_file", "search_web", "calculator"}, | ||
| default_block=True, | ||
| ) | ||
| """ | ||
|
|
||
| def __init__( | ||
| self, | ||
| allowed_tools: set[str], | ||
| default_block: bool = True, | ||
| ) -> None: | ||
| self._allowed = frozenset(allowed_tools) | ||
| self._default_block = default_block | ||
|
|
||
| def authorize(self, context: ToolCallHookContext) -> GuardrailDecisionV1: | ||
| is_allowed = context.tool_name in self._allowed | ||
| authorized = is_allowed if self._default_block else (not is_allowed) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
default_block=False contradicts the documented and named semantics.
The docstring states that default_block=False allows tools that are not in the allowlist. Line 273 instead inverts the decision, so allowed_tools becomes a denylist and every listed tool is blocked. A user who relaxes the guardrail with default_block=False blocks the tools they explicitly allowed.
Pick one contract and make the code, the docstring, the parameter name, and test_inverted_mode in tests/guardrails/test_guardrail_provider.py agree. If denylist mode is the intent, rename the constructor arguments (for example tools plus mode="allowlist" | "denylist"). If the docstring is the intent, apply this fix.
🐛 Proposed fix for allowlist semantics
def authorize(self, context: ToolCallHookContext) -> GuardrailDecisionV1:
is_allowed = context.tool_name in self._allowed
- authorized = is_allowed if self._default_block else (not is_allowed)
+ # default_block=False means unlisted tools are permitted too.
+ authorized = is_allowed or not self._default_block📝 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.
| class ToolListGuardrailProvider(GuardrailProvider): | |
| """Allows only an explicit allowlist of tool names. | |
| Args: | |
| allowed_tools: Set of tool names permitted for execution. | |
| default_block: Whether to block (True) or allow (False) tools | |
| not in the allowlist. | |
| Example:: | |
| guardrail = ToolListGuardrailProvider( | |
| allowed_tools={"read_file", "search_web", "calculator"}, | |
| default_block=True, | |
| ) | |
| """ | |
| def __init__( | |
| self, | |
| allowed_tools: set[str], | |
| default_block: bool = True, | |
| ) -> None: | |
| self._allowed = frozenset(allowed_tools) | |
| self._default_block = default_block | |
| def authorize(self, context: ToolCallHookContext) -> GuardrailDecisionV1: | |
| is_allowed = context.tool_name in self._allowed | |
| authorized = is_allowed if self._default_block else (not is_allowed) | |
| class ToolListGuardrailProvider(GuardrailProvider): | |
| """Allows only an explicit allowlist of tool names. | |
| Args: | |
| allowed_tools: Set of tool names permitted for execution. | |
| default_block: Whether to block (True) or allow (False) tools | |
| not in the allowlist. | |
| Example:: | |
| guardrail = ToolListGuardrailProvider( | |
| allowed_tools={"read_file", "search_web", "calculator"}, | |
| default_block=True, | |
| ) | |
| """ | |
| def __init__( | |
| self, | |
| allowed_tools: set[str], | |
| default_block: bool = True, | |
| ) -> None: | |
| self._allowed = frozenset(allowed_tools) | |
| self._default_block = default_block | |
| def authorize(self, context: ToolCallHookContext) -> GuardrailDecisionV1: | |
| is_allowed = context.tool_name in self._allowed | |
| # default_block=False means unlisted tools are permitted too. | |
| authorized = is_allowed or not self._default_block |
🤖 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 `@src/crewai/guardrails/guardrail_provider.py` around lines 247 - 273, Update
ToolListGuardrailProvider.authorize so default_block=False still permits tools
in _allowed and only changes the treatment of tools outside the allowlist; align
the constructor documentation, parameter semantics, and test_inverted_mode with
this allowlist contract. Preserve the existing default_block=True behavior for
listed and unlisted tools.
| results: list[tuple[str, bool]] = [] | ||
| all_satisfied = True | ||
|
|
||
| for predicate_name, params in self._constraints: | ||
| ok = self._eval_predicate(predicate_name, context, **params) | ||
| results.append((predicate_name, ok)) | ||
| if not ok: | ||
| all_satisfied = False | ||
|
|
||
| claims = { | ||
| "tool": context.tool_name, | ||
| "agent": getattr(context.agent, "role", "unknown"), | ||
| "constraint_results": results, | ||
| } | ||
| expires_at = time.time() + 60 | ||
| return GuardrailDecisionV1( | ||
| decision_id=compute_decision_id(claims, expires_at), | ||
| authorized=all_satisfied, | ||
| claims=claims, | ||
| expires_at=expires_at, | ||
| reason=( | ||
| "All constraints satisfied" | ||
| if all_satisfied | ||
| else f"Constraints failed: {[k for k, v in results.items() if not v]}" | ||
| ), | ||
| provider_name="ckg", | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Check how before-tool-call hook exceptions are handled by the dispatcher.
set -euo pipefail
fd -t f 'tool_hooks.py'
rg -n -C 8 'before_tool_call' --glob '*.py' -g '!tests/**'Repository: crewAIInc/crewAI
Length of output: 238
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== tool_hooks.py outline/sections ==="
wc -l lib/crewai/src/crewai/hooks/tool_hooks.py
sed -n '1,260p' lib/crewai/src/crewai/hooks/tool_hooks.py
echo "=== guardrail_provider relevant sections ==="
fd -t f 'guardrail_provider.py' | xargs -r -I{} sh -c 'echo "--- {}"; wc -l "{}"; sed -n "260,370p" "{}"'
echo "=== search guardrail provider uses ==="
rg -n 'CKGGuardrailProvider|guardrail_provider|before_tool_call|authorize\(' --glob '*.py' -g '!tests/**'Repository: crewAIInc/crewAI
Length of output: 18397
Fix the denial reason before returning it.
results is a list, so results.items() raises AttributeError whenever any CKG constraint evaluates to false. Build the failed predicate list from the list contents, e.g. [k for k, v in results if not v], so CKGGuardrailProvider.authorize() returns authorized=False instead of raising on the denial 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 `@src/crewai/guardrails/guardrail_provider.py` around lines 326 - 352, Update
the denial-reason comprehension in CKGGuardrailProvider’s authorize flow to
iterate directly over the results list rather than calling results.items().
Preserve the existing successful reason and ensure failed predicates are
collected from each (name, status) tuple so denied authorization returns
normally.
| class AuditTrail: | ||
| """In-memory audit trail that records every decision + envelope pair. | ||
|
|
||
| Thread-safe for single-threaded crew execution. For production, | ||
| swap the backend for a durable store (SQLite, Redis, etc.). | ||
| """ | ||
|
|
||
| def __init__(self) -> None: | ||
| self._decisions: dict[str, GuardrailDecisionV1] = {} | ||
| self._envelopes: dict[str, ActionEnvelopeV1] = {} | ||
|
|
||
| def record_decision(self, decision: GuardrailDecisionV1) -> None: | ||
| self._decisions[decision.decision_id] = decision | ||
|
|
||
| def record_envelope(self, envelope: ActionEnvelopeV1) -> None: | ||
| self._envelopes[envelope.decision_id] = envelope |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
AuditTrail keys records by content address, so repeated identical tool calls overwrite each other. decision_id is a SHA-256 of claims plus expires_at. Two identical calls inside one time.time() tick produce one key, so the trail silently drops evidence and under-counts calls. The test that repeats read_file inherits the same defect as an intermittent failure.
src/crewai/guardrails/guardrail_provider.py#L396-L411: append each record to an ordered list inrecord_decisionandrecord_envelope, keep the dicts as lookup indexes only, and return the lists fromall_decisions,all_envelopes,total_decisions, andtotal_envelopes. Clear both structures inclear(). Also correct the "Thread-safe for single-threaded crew execution" docstring, because no lock exists.tests/guardrails/test_guardrail_provider.py#L697-L709: after the trail records every call, keep thetotal_decisions == len(calls)andgranted == 3assertions, and add an assertion that two identical consecutive calls yield two recorded decisions.
📍 Affects 2 files
src/crewai/guardrails/guardrail_provider.py#L396-L411(this comment)tests/guardrails/test_guardrail_provider.py#L697-L709
🤖 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 `@src/crewai/guardrails/guardrail_provider.py` around lines 396 - 411, The
AuditTrail currently overwrites repeated identical records because it keys
storage only by decision_id. In src/crewai/guardrails/guardrail_provider.py
lines 396-411, update AuditTrail to append decisions and envelopes to ordered
lists while retaining the dicts only as lookup indexes; have all_decisions,
all_envelopes, total_decisions, and total_envelopes use those lists, clear both
structures in clear(), and correct the docstring’s thread-safety claim. In
tests/guardrails/test_guardrail_provider.py lines 697-709, preserve the
total_decisions == len(calls) and granted == 3 assertions and add coverage
proving two identical consecutive calls produce two recorded decisions.
| missing: list[str] = [] | ||
| tools = agent_config.get("tools", []) | ||
| for tool in tools if isinstance(tools, list) else []: | ||
| tool_name = tool if isinstance(tool, str) else tool.get("name", "?") | ||
| if not _has_guardrail_provider(agent_config, tool_name): | ||
| missing.append(f"tool:{tool_name} — no GuardrailProvider") | ||
| return missing |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Handle tool objects, not just strings and dicts.
Line 559 calls tool.get("name", "?") for any non-string entry. crewAI agents usually hold BaseTool instances. Such an object has no .get, so this scanner raises AttributeError on a real agent config.
🐛 Proposed fix
for tool in tools if isinstance(tools, list) else []:
- tool_name = tool if isinstance(tool, str) else tool.get("name", "?")
+ if isinstance(tool, str):
+ tool_name = tool
+ elif isinstance(tool, dict):
+ tool_name = tool.get("name", "?")
+ else:
+ tool_name = getattr(tool, "name", "?")📝 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.
| missing: list[str] = [] | |
| tools = agent_config.get("tools", []) | |
| for tool in tools if isinstance(tools, list) else []: | |
| tool_name = tool if isinstance(tool, str) else tool.get("name", "?") | |
| if not _has_guardrail_provider(agent_config, tool_name): | |
| missing.append(f"tool:{tool_name} — no GuardrailProvider") | |
| return missing | |
| missing: list[str] = [] | |
| tools = agent_config.get("tools", []) | |
| for tool in tools if isinstance(tools, list) else []: | |
| if isinstance(tool, str): | |
| tool_name = tool | |
| elif isinstance(tool, dict): | |
| tool_name = tool.get("name", "?") | |
| else: | |
| tool_name = getattr(tool, "name", "?") | |
| if not _has_guardrail_provider(agent_config, tool_name): | |
| missing.append(f"tool:{tool_name} — no GuardrailProvider") | |
| return missing |
🤖 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 `@src/crewai/guardrails/guardrail_provider.py` around lines 556 - 562, Update
the tool-name extraction in the guardrail scanner around _has_guardrail_provider
so non-string tool entries are handled safely, including BaseTool instances,
without calling mapping-only .get on arbitrary objects. Resolve the tool name
via the object’s supported name attribute or established tool interface, retain
dictionary handling, and continue recording tool:name entries for tools lacking
a GuardrailProvider.
1 participant
Summary
Integrate CCS (Cross-framework Command Security) verification into CrewAI's tool calling pipeline.
What is CCS?
CCS provides sub-millisecond in-process security checks (~7.5μs P50) for AI Agent tool execution. It uses semantic analysis to detect:
Changes
examples/ccs-security/ccs_guard.py—CCSSecurityMiddlewarewithwrap_tool()andverify_tool_call()examples/ccs-security/README.md— Integration guideUsage
References
pip install ccs-verifier