feat(security): implement Secret Masking in LLM Context Construction #656

Closed
CoreRasurae wants to merge 1 commits from feature/m4-secret-masking-llm-context into master
12 changed files with 558 additions and 23 deletions
+15
View File
@@ -2,6 +2,21 @@
## Unreleased
- Implemented secret masking in LLM context construction. Added
`redact_context_for_llm()` function and `LLM_REDACTED` constant to
`shared/redaction.py` using `[REDACTED]` replacement as specified by the
architecture spec (Secret Management, item 4). Wired redaction into both
`ACMSPipeline.assemble()` and `ContextAssemblyPipeline.assemble()` as a final
pass before the assembled context is returned. Extracted shared
`_redact_for_llm()` static method on the base pipeline class, added
`secret_masking_ms` to `StageTimings`, and optimised to skip `model_copy()`
when no secrets are detected. Added negative lookbehind anchors to the `sk-`
and `sk-ant-` regex patterns to prevent false-positive redaction inside common
words (e.g. "task-type-classification"). Includes 11 Behave BDD scenarios
(including `ContextAssemblyPipeline` integration and `show_secrets` bypass
invariant), 5 Robot Framework integration tests, and ASV benchmarks with both
secrets-present and no-secrets fast-path workloads. (#573)
- Fixed `agents project show` not finding a project immediately after creation.
Extended the `session.commit()` fix from #589 to also cover `update()` and
`delete()` in `NamespacedProjectRepository`, and updated the class docstring
+44
View File
@@ -0,0 +1,44 @@
"""ASV benchmarks for secret redaction throughput."""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.shared.redaction import redact_context_for_llm # noqa: E402
class SecretRedactionBenchmarks:
"""Benchmarks for redact_context_for_llm on varying content sizes."""
params = [100, 1000, 10000, 100000]
param_names = ["content_length"]
def setup(self, content_length: int) -> None:
self.redact = redact_context_for_llm
# Content with embedded secrets at intervals
base = "Normal text content. "
secret = "sk-proj-ABCDEFGHIJKLMNOP "
block = base * 9 + secret # Every 10th chunk is a secret
self.content = (block * (content_length // len(block) + 1))[:content_length]
# Content without any secrets — the expected hot path in production
self.clean_content = (base * (content_length // len(base) + 1))[:content_length]
def time_redact_context(self, content_length: int) -> None:
self.redact(self.content)
def time_redact_no_secrets(self, content_length: int) -> None:
"""Benchmark the fast path: no secrets in content."""
self.redact(self.clean_content)
def peakmem_redact_context(self, content_length: int) -> None:
self.redact(self.content)
+28 -4
View File
@@ -18,8 +18,8 @@ pattern-based secret detection and masking. It is integrated into:
| Pattern family | Regex summary | Example |
|---|---|---|
| OpenAI keys | `sk-(?:proj-)?[A-Za-z0-9_-]{10,}` | `sk-proj-abc123…` |
| Anthropic keys | `sk-ant-[A-Za-z0-9_-]{10,}` | `sk-ant-api03-xyz…` |
| OpenAI keys | `(?<![A-Za-z])sk-(?:proj-)?[A-Za-z0-9_-]{10,}` | `sk-proj-abc123…` |
| Anthropic keys | `(?<![A-Za-z])sk-ant-[A-Za-z0-9_-]{10,}` | `sk-ant-api03-xyz…` |
| Token IDs | `tok_[A-Za-z0-9]{10,}` | `tok_01HXYZ…` |
| Bearer tokens | `Bearer\s+[A-Za-z0-9._~+/=-]{20,}` | `Bearer eyJhb…` |
| Generic keys | `(?:key\|KEY)-[A-Za-z0-9]{20,}` | `KEY-abcdef…` |
@@ -33,9 +33,30 @@ detected by substring matching:
False-positive key names like `token_count`, `max_tokens`, etc. are
excluded.
## Replacement token
## Replacement tokens
All detected secrets are replaced with `***REDACTED***`.
All detected secrets in CLI output, logs, and error messages are
replaced with `***REDACTED***`.
### LLM context masking
Secrets detected during **LLM context assembly** are replaced with
`[REDACTED]` (the `LLM_REDACTED` constant). This is applied as a
final pass in the ACMS context assembly pipeline so that secrets
present in fragment content and preamble text are never sent to
language model providers. Resource content (e.g. file read results)
that enters the pipeline as context fragments is also covered.
```python
from cleveragents.shared.redaction import redact_context_for_llm
redact_context_for_llm("key is sk-proj-abc123def456ghi789")
# → "key is [REDACTED]"
```
The `redact_context_for_llm()` function uses the same pattern registry
as `redact_value()` but with the `[REDACTED]` replacement marker as
specified by the architecture spec (Secret Management, item 4).
## Global `--show-secrets` flag
@@ -108,5 +129,8 @@ leaks credentials.
## Testing
- **Behave**: `features/security_secrets.feature` (43 scenarios)
- **Behave**: `features/security/secret_masking_llm_context.feature` (LLM context masking)
- **Robot Framework**: `robot/security_secrets.robot` (10 smoke tests)
- **Robot Framework**: `robot/secret_masking_llm_context.robot` (LLM context integration)
- **ASV benchmarks**: `benchmarks/security_secrets_bench.py`
- **ASV benchmarks**: `benchmarks/bench_secret_redaction.py` (LLM redaction throughput)
+1 -1
View File
@@ -264,4 +264,4 @@ Feature: ACMS Pipeline Orchestrator and Phase 1 Components
Given the pipeline orchestrator modules are available
When I create a StageTimings with total_ms 42.5
Then the timings total_ms should be 42.5
And the timings should have all 10 named fields
And the timings should have all 11 named fields
@@ -0,0 +1,63 @@
@security @secret_masking @llm_context
Feature: Secret Masking in LLM Context Construction
As a security-conscious system
I want secrets to be masked before constructing LLM prompts
So that sensitive data is never exposed to language models
Scenario: OpenAI API key in text content is redacted
Given text content containing "my key is sk-proj-ABC123DEF456GHI789"
When the content is redacted for LLM context
Then the result should contain "[REDACTED]" instead of the API key
Scenario: Anthropic API key in text content is redacted
Given text content containing "anthropic key: sk-ant-api03-ABCDEFGHIJ"
When the content is redacted for LLM context
Then the result should contain "[REDACTED]" instead of the API key
Scenario: Bearer token in text content is redacted
Given text content containing "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.payload.sig"
When the content is redacted for LLM context
Then the result should contain "[REDACTED]" instead of the token
Scenario: Multiple secrets in content are all redacted
Given text content with multiple embedded secrets
When the content is redacted for LLM context
Then all secrets should be replaced with "[REDACTED]"
Scenario: Content without secrets is unchanged
Given text content containing only normal text
When the content is redacted for LLM context
Then the content should be unchanged
Scenario: Empty content returns empty string
Given empty text content
When the content is redacted for LLM context
Then the LLM redaction result should be an empty string
Scenario: LLM_REDACTED constant has correct value
Given the LLM_REDACTED constant
Then its value should be "[REDACTED]"
Scenario: Fragment content is redacted in assembled context payload
Given a context fragment with secret content "Config: sk-proj-ABCDEFGHIJKLMNOP"
When the fragment is assembled through the ACMS pipeline
Then the payload fragment content should contain "[REDACTED]"
And the payload fragment content should not contain "sk-proj-"
Scenario: Preamble is redacted in assembled context payload
Given fragments assembled with a provenance preamble generator
When the preamble contains a secret pattern
Then the payload preamble should contain "[REDACTED]"
Scenario: Fragment content is redacted in ContextAssemblyPipeline payload
Given a context fragment with secret content "Config: sk-proj-ABCDEFGHIJKLMNOP"
When the fragment is assembled through the ContextAssemblyPipeline
Then the payload fragment content should contain "[REDACTED]"
And the payload fragment content should not contain "sk-proj-"
Scenario: show_secrets flag does not bypass LLM context redaction
Given the global show_secrets flag is set to true
And text content containing "my key is sk-proj-ABC123DEF456GHI789"
When the content is redacted for LLM context
Then the result should contain "[REDACTED]" instead of the API key
And the global show_secrets flag is reset to false
@@ -833,10 +833,10 @@ def step_timings_total(context: Context, total: float) -> None:
assert context.test_timings.total_ms == total
@then("the timings should have all 10 named fields")
@then("the timings should have all 11 named fields")
def step_timings_fields(context: Context) -> None:
field_names = set(StageTimings.model_fields.keys())
assert len(field_names) == 10, f"Expected 10 fields, got {len(field_names)}"
assert len(field_names) == 11, f"Expected 11 fields, got {len(field_names)}"
expected_names = {
"strategy_selection_ms",
"budget_allocation_ms",
@@ -847,6 +847,7 @@ def step_timings_fields(context: Context) -> None:
"packing_ms",
"ordering_ms",
"preamble_generation_ms",
"secret_masking_ms",
"total_ms",
}
assert field_names == expected_names, (
@@ -0,0 +1,240 @@
"""Step definitions for features/security/secret_masking_llm_context.feature.
Tests the LLM context secret masking functionality: ``redact_context_for_llm``
and its integration with the ACMS context assembly pipeline.
"""
from __future__ import annotations
from behave import given, then, when
from behave.runner import Context
from cleveragents.application.services.acms_pipeline import ContextAssemblyPipeline
from cleveragents.application.services.acms_service import ACMSPipeline
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
ContextFragment,
FragmentProvenance,
)
from cleveragents.shared.redaction import (
LLM_REDACTED,
redact_context_for_llm,
set_show_secrets,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_DEFAULT_PROVENANCE = FragmentProvenance(resource_uri="test://secret-masking")
_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
def _make_fragment(content: str) -> ContextFragment:
"""Create a ContextFragment with given content and sensible defaults."""
return ContextFragment(
uko_node="test://default",
content=content,
token_count=len(content.split()),
provenance=_DEFAULT_PROVENANCE,
)
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given('text content containing "{text}"')
def step_given_text_content(context: Context, text: str) -> None:
context.input_text = text
@given("text content with multiple embedded secrets")
def step_given_multiple_secrets(context: Context) -> None:
context.input_text = (
"OpenAI key: sk-proj-ABC123DEF456GHI789 "
"and Anthropic key: sk-ant-api03-XYZXYZXYZX0 "
"and a bearer: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.payload.signature"
)
@given("text content containing only normal text")
def step_given_normal_text(context: Context) -> None:
context.input_text = "This is perfectly normal text with no secrets at all."
@given("empty text content")
def step_given_empty_content(context: Context) -> None:
context.input_text = ""
@given("the LLM_REDACTED constant")
def step_given_llm_redacted_constant(context: Context) -> None:
context.llm_redacted = LLM_REDACTED
@given('a context fragment with secret content "{content}"')
def step_given_fragment_with_secret(context: Context, content: str) -> None:
context.secret_fragment = _make_fragment(content)
@given("the global show_secrets flag is set to true")
def step_given_show_secrets_true(context: Context) -> None:
set_show_secrets(True)
# Register cleanup so show_secrets resets even if the scenario fails.
if not hasattr(context, "_cleanup_handlers"):
context._cleanup_handlers = [] # type: ignore[attr-defined]
context._cleanup_handlers.append(lambda: set_show_secrets(False))
@given("fragments assembled with a provenance preamble generator")
def step_given_fragments_with_preamble(context: Context) -> None:
# Store a fragment whose strategy_source contains a secret pattern
# so the provenance preamble will include it.
# NOTE: This step is coupled to ProvenancePreambleGenerator (acms_phase3.py)
# which embeds strategy_source in the preamble text. If that generator's
# output format changes, the corresponding "Then" assertion may break.
context.preamble_fragment = ContextFragment(
uko_node="test://preamble",
content="Some normal content",
token_count=10,
provenance=_DEFAULT_PROVENANCE,
strategy_source="sk-proj-LEAKEDSECRET1234",
)
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
@when("the content is redacted for LLM context")
def step_when_redact_for_llm(context: Context) -> None:
context.result = redact_context_for_llm(context.input_text)
@when("the fragment is assembled through the ACMS pipeline")
def step_when_assemble_fragment(context: Context) -> None:
pipeline = ACMSPipeline()
payload = pipeline.assemble(
plan_id=_PLAN_ID,
fragments=[context.secret_fragment],
budget=ContextBudget(max_tokens=4096),
)
context.payload = payload
@when("the fragment is assembled through the ContextAssemblyPipeline")
def step_when_assemble_fragment_cap(context: Context) -> None:
pipeline = ContextAssemblyPipeline()
payload = pipeline.assemble(
plan_id=_PLAN_ID,
fragments=[context.secret_fragment],
budget=ContextBudget(max_tokens=4096),
)
context.payload = payload
@when("the preamble contains a secret pattern")
def step_when_preamble_with_secret(context: Context) -> None:
from cleveragents.application.services.acms_phase3 import (
ProvenancePreambleGenerator,
)
pipeline = ACMSPipeline(
preamble_generator=ProvenancePreambleGenerator(),
)
payload = pipeline.assemble(
plan_id=_PLAN_ID,
fragments=[context.preamble_fragment],
budget=ContextBudget(max_tokens=4096),
)
context.payload = payload
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then('the result should contain "[REDACTED]" instead of the API key')
def step_then_contains_redacted_key(context: Context) -> None:
assert LLM_REDACTED in context.result, (
f"Expected {LLM_REDACTED!r} in result, got: {context.result!r}"
)
# Ensure the raw secret pattern is gone
assert "sk-proj-" not in context.result and "sk-ant-" not in context.result, (
f"Raw secret still present in result: {context.result!r}"
)
@then('the result should contain "[REDACTED]" instead of the token')
def step_then_contains_redacted_token(context: Context) -> None:
assert LLM_REDACTED in context.result, (
f"Expected {LLM_REDACTED!r} in result, got: {context.result!r}"
)
assert "eyJhbGci" not in context.result, (
f"Raw Bearer token still present in result: {context.result!r}"
)
@then('all secrets should be replaced with "[REDACTED]"')
def step_then_all_secrets_replaced(context: Context) -> None:
assert context.result.count(LLM_REDACTED) >= 3, (
f"Expected at least 3 [REDACTED] tokens, got "
f"{context.result.count(LLM_REDACTED)}: {context.result!r}"
)
assert "sk-proj-" not in context.result
assert "sk-ant-" not in context.result
assert "eyJhbGci" not in context.result
@then("the content should be unchanged")
def step_then_content_unchanged(context: Context) -> None:
assert context.result == context.input_text, (
f"Content was changed: {context.result!r} != {context.input_text!r}"
)
@then("the LLM redaction result should be an empty string")
def step_then_result_empty(context: Context) -> None:
assert context.result == "", f"Expected empty string, got: {context.result!r}"
@then('its value should be "[REDACTED]"')
def step_then_constant_value(context: Context) -> None:
assert context.llm_redacted == "[REDACTED]", (
f"Expected '[REDACTED]', got: {context.llm_redacted!r}"
)
@then('the payload fragment content should contain "[REDACTED]"')
def step_then_payload_fragment_redacted(context: Context) -> None:
assert len(context.payload.fragments) > 0, "No fragments in payload"
content = context.payload.fragments[0].content
assert LLM_REDACTED in content, (
f"Expected {LLM_REDACTED!r} in fragment content, got: {content!r}"
)
@then('the payload fragment content should not contain "sk-proj-"')
def step_then_payload_no_raw_secret(context: Context) -> None:
content = context.payload.fragments[0].content
assert "sk-proj-" not in content, (
f"Raw secret still in fragment content: {content!r}"
)
@then('the payload preamble should contain "[REDACTED]"')
def step_then_payload_preamble_redacted(context: Context) -> None:
preamble = context.payload.preamble
assert preamble is not None, "Preamble is None"
assert LLM_REDACTED in preamble, (
f"Expected {LLM_REDACTED!r} in preamble, got: {preamble!r}"
)
@then("the global show_secrets flag is reset to false")
def step_then_reset_show_secrets(context: Context) -> None:
set_show_secrets(False)
+47
View File
@@ -0,0 +1,47 @@
*** Settings ***
Documentation Integration test for secret masking in LLM context
Library Process
Library OperatingSystem
*** Test Cases ***
Secret In Resource Content Is Redacted Before LLM Context
[Documentation] Verify that secrets in resource content are redacted
[Tags] security secret_masking
${result}= Run Process python -c
... from cleveragents.shared.redaction import redact_context_for_llm; print(redact_context_for_llm("API_KEY\=sk-proj-ABCDEFGHIJKLMNOP"))
... env:PYTHONPATH=src
Should Contain ${result.stdout} [REDACTED]
Should Not Contain ${result.stdout} sk-proj-
LLM REDACTED Constant Has Correct Value
[Documentation] Verify LLM_REDACTED constant is [REDACTED]
[Tags] security secret_masking
${result}= Run Process python -c
... from cleveragents.shared.redaction import LLM_REDACTED; print(LLM_REDACTED)
... env:PYTHONPATH=src
Should Be Equal ${result.stdout.strip()} [REDACTED]
Bearer Token Is Redacted In LLM Context
[Documentation] Verify Bearer tokens are redacted
[Tags] security secret_masking
${result}= Run Process python -c
... from cleveragents.shared.redaction import redact_context_for_llm; print(redact_context_for_llm("Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.payload.sig"))
... env:PYTHONPATH=src
Should Contain ${result.stdout} [REDACTED]
Should Not Contain ${result.stdout} eyJhbGci
Empty Content Returns Empty
[Documentation] Verify empty content is returned as-is
[Tags] security secret_masking
${result}= Run Process python -c
... from cleveragents.shared.redaction import redact_context_for_llm; result\=redact_context_for_llm(""); print(f"EMPTY:{result!r}")
... env:PYTHONPATH=src
Should Contain ${result.stdout} EMPTY:''
Normal Content Is Not Modified
[Documentation] Verify content without secrets is unchanged
[Tags] security secret_masking
${result}= Run Process python -c
... from cleveragents.shared.redaction import redact_context_for_llm; print(redact_context_for_llm("Hello world, this is safe text"))
... env:PYTHONPATH=src
Should Be Equal ${result.stdout.strip()} Hello world, this is safe text
@@ -456,6 +456,7 @@ class StageTimings(BaseModel, frozen=True):
packing_ms: float = Field(default=0.0, ge=0.0)
ordering_ms: float = Field(default=0.0, ge=0.0)
preamble_generation_ms: float = Field(default=0.0, ge=0.0)
secret_masking_ms: float = Field(default=0.0, ge=0.0)
total_ms: float = Field(default=0.0, ge=0.0)
@@ -634,6 +635,15 @@ class ContextAssemblyPipeline(ACMSPipeline):
preamble = self._preamble_generator.generate(fused)
preamble_ms = (time.monotonic() - t0) * 1000
# Secret masking: redact secrets from fragment content and preamble
# before they enter the LLM context (spec §Secret Management, item 4).
t0 = time.monotonic()
redacted_fragments, redacted_preamble = self._redact_for_llm(
fused,
preamble,
)
secret_masking_ms = (time.monotonic() - t0) * 1000
total_ms = (time.monotonic() - pipeline_start) * 1000
self._last_timings = StageTimings(
@@ -646,21 +656,21 @@ class ContextAssemblyPipeline(ACMSPipeline):
packing_ms=round(pack_ms, 3),
ordering_ms=round(order_ms, 3),
preamble_generation_ms=round(preamble_ms, 3),
secret_masking_ms=round(secret_masking_ms, 3),
total_ms=round(total_ms, 3),
)
final_fragments = tuple(fused)
total_tokens = sum(f.token_count for f in final_fragments)
total_tokens = sum(f.token_count for f in redacted_fragments)
available = budget.available_tokens
budget_used = total_tokens / available if available > 0 else 0.0
context_hash = compute_context_hash(final_fragments)
provenance_map = build_provenance_map(final_fragments)
context_hash = compute_context_hash(redacted_fragments)
provenance_map = build_provenance_map(redacted_fragments)
self._pipeline_logger.info(
"Pipeline complete",
plan_id=plan_id,
strategy=strategy_name,
fragments_selected=len(final_fragments),
fragments_selected=len(redacted_fragments),
total_tokens=total_tokens,
budget_used=round(budget_used, 4),
total_ms=round(total_ms, 3),
@@ -668,12 +678,12 @@ class ContextAssemblyPipeline(ACMSPipeline):
return ContextPayload(
plan_id=plan_id,
fragments=final_fragments,
fragments=redacted_fragments,
total_tokens=total_tokens,
budget=budget,
budget_used=round(min(budget_used, 1.0), 4),
strategies_used=(strategy_name,),
context_hash=context_hash,
preamble=preamble,
preamble=redacted_preamble,
provenance_map=provenance_map,
)
@@ -41,6 +41,7 @@ from cleveragents.domain.models.core.context_fragment import (
build_provenance_map,
compute_context_hash,
)
from cleveragents.shared.redaction import redact_context_for_llm
logger = structlog.get_logger(__name__)
@@ -696,34 +697,81 @@ class ACMSPipeline:
# Phase 3: Context Finalization
preamble = self._preamble_generator.generate(fused)
final_fragments = tuple(fused)
total_tokens = sum(f.token_count for f in final_fragments)
# Secret masking: redact secrets from fragment content and preamble
# before they enter the LLM context (spec §Secret Management, item 4).
redacted_fragments, redacted_preamble = self._redact_for_llm(
fused,
preamble,
)
total_tokens = sum(f.token_count for f in redacted_fragments)
available = budget.available_tokens
budget_used = total_tokens / available if available > 0 else 0.0
context_hash = compute_context_hash(final_fragments)
provenance_map = build_provenance_map(final_fragments)
context_hash = compute_context_hash(redacted_fragments)
provenance_map = build_provenance_map(redacted_fragments)
self._logger.info(
"Context assembled",
plan_id=plan_id,
strategy=strategy_name,
fragments_selected=len(final_fragments),
fragments_selected=len(redacted_fragments),
total_tokens=total_tokens,
budget_used=round(budget_used, 4),
)
return ContextPayload(
plan_id=plan_id,
fragments=final_fragments,
fragments=redacted_fragments,
total_tokens=total_tokens,
budget=budget,
budget_used=round(min(budget_used, 1.0), 4),
strategies_used=(strategy_name,),
context_hash=context_hash,
preamble=preamble,
preamble=redacted_preamble,
provenance_map=provenance_map,
)
@staticmethod
def _redact_for_llm(
fragments: Sequence[ContextFragment],
preamble: str | None,
) -> tuple[tuple[ContextFragment, ...], str | None]:
"""Apply secret masking to fragment content and preamble.
Scans each fragment's ``content`` and the optional ``preamble``
for secret patterns using :func:`redact_context_for_llm` and
returns copies with secrets replaced by ``[REDACTED]``.
Fragments whose content is unchanged (no secrets detected) are
returned as-is to avoid unnecessary object allocation.
Note:
Fragment ``token_count`` values are **not** recalculated
after redaction. Because redaction only ever shortens
content (replacing long secret strings with the shorter
``[REDACTED]`` marker), the pre-redaction token counts
serve as conservative upper-bound estimates. This avoids
the need for a tokenizer dependency in the shared layer
and keeps budget checks safely conservative.
Args:
fragments: The fused fragments to redact.
preamble: The optional preamble text to redact.
Returns:
A ``(redacted_fragments, redacted_preamble)`` pair.
"""
redacted_fragments = tuple(
f
if (new := redact_context_for_llm(f.content)) == f.content
else f.model_copy(update={"content": new})
for f in fragments
)
redacted_preamble = (
redact_context_for_llm(preamble) if preamble is not None else None
)
return redacted_fragments, redacted_preamble
def register_strategy(
self,
name: str,
+4
View File
@@ -12,10 +12,12 @@ Includes:
"""
from cleveragents.shared.redaction import (
LLM_REDACTED,
REDACTED,
get_show_secrets,
is_sensitive_key,
mask_database_url,
redact_context_for_llm,
redact_dict,
redact_value,
register_pattern,
@@ -24,10 +26,12 @@ from cleveragents.shared.redaction import (
)
__all__ = [
"LLM_REDACTED",
"REDACTED",
"get_show_secrets",
"is_sensitive_key",
"mask_database_url",
"redact_context_for_llm",
"redact_dict",
"redact_value",
"register_pattern",
+41 -2
View File
@@ -21,6 +21,7 @@ from collections.abc import MutableMapping
from typing import Any
REDACTED = "***REDACTED***"
LLM_REDACTED = "[REDACTED]"
# ── Sensitive key-name substrings ──────────────────────────────
@@ -59,9 +60,11 @@ _FALSE_POSITIVE_KEYS: set[str] = {
_SECRET_PATTERNS: list[re.Pattern[str]] = [
# OpenAI keys: sk-proj-..., sk-...
re.compile(r"sk-(?:proj-)?[A-Za-z0-9_-]{10,}"),
# Negative lookbehind prevents false positives on common words
# ending in "sk" (e.g., "task-type-classification", "risk-management").
re.compile(r"(?<![A-Za-z])sk-(?:proj-)?[A-Za-z0-9_-]{10,}"),
# Anthropic keys: sk-ant-api03-...
re.compile(r"sk-ant-[A-Za-z0-9_-]{10,}"),
re.compile(r"(?<![A-Za-z])sk-ant-[A-Za-z0-9_-]{10,}"),
# Token IDs: tok_...
re.compile(r"tok_[A-Za-z0-9]{10,}"),
# Bearer tokens
@@ -229,6 +232,42 @@ def register_pattern(pattern: str) -> None:
_SECRET_PATTERNS.append(compiled)
# ── LLM context redaction ─────────────────────────────────────
def redact_context_for_llm(content: str) -> str:
"""Scan text content for secret patterns and replace with [REDACTED].
Designed for use in the LLM context assembly path to prevent
accidental exposure of secrets in LLM prompts. Scans for all
registered secret patterns (the ``_SECRET_PATTERNS`` regex list).
Unlike ``redact_value()`` which uses ``***REDACTED***``, this function
uses ``[REDACTED]`` as specified by the architecture spec for LLM
context masking (spec section: Secret Management, item 4).
Note:
This function is intentionally **not** gated by the global
``show_secrets`` flag. Secrets must always be masked before
entering LLM context regardless of CLI display preferences.
Args:
content: The text content to scan (prompt text, file contents,
action arguments, session messages, etc.).
Returns:
The content with detected secrets replaced by ``[REDACTED]``.
"""
if not content:
return content
result = content
with _patterns_lock:
patterns = list(_SECRET_PATTERNS)
for pattern in patterns:
result = pattern.sub(LLM_REDACTED, result)
return result
# ── structlog processor ───────────────────────────────────────