fix(security): enforce explicit exception handling

This commit is contained in:
CoreRasurae
2026-02-22 12:11:12 +00:00
parent d37cfbe769
commit 28aeab6857
8 changed files with 1313 additions and 19 deletions
+87
View File
@@ -0,0 +1,87 @@
"""ASV benchmarks for exception handling overhead.
Measures the cost of error classification, redaction,
and wrapping operations.
"""
from __future__ import annotations
from cleveragents.core.error_handling import (
classify_error,
format_error_for_cli,
redact_error_details,
redact_value,
wrap_unexpected,
)
from cleveragents.core.exceptions import (
DatabaseError,
ValidationError,
)
class ClassifyBench:
"""Benchmark error classification throughput."""
timeout = 60
def setup(self) -> None:
self.validation_err = ValidationError("field invalid")
self.db_err = DatabaseError("connection lost")
self.bare_err = Exception("unknown crash")
def time_classify_validation(self) -> None:
"""Benchmark classifying a ValidationError."""
classify_error(self.validation_err)
def time_classify_database(self) -> None:
"""Benchmark classifying a DatabaseError."""
classify_error(self.db_err)
def time_classify_bare(self) -> None:
"""Benchmark classifying a bare Exception."""
classify_error(self.bare_err)
def time_format_cli(self) -> None:
"""Benchmark formatting an error for CLI."""
info = classify_error(self.validation_err)
format_error_for_cli(info)
class RedactBench:
"""Benchmark secret redaction throughput."""
timeout = 60
def setup(self) -> None:
self.details_secret = {"api_key": "sk-abc123", "plan_id": "01PLAN"}
self.details_clean = {"plan_id": "01PLAN", "status": "ok"}
self.value_with_secret = "token sk-abcdefghijklmnopqrstuvwxyz end"
def time_redact_with_secret(self) -> None:
"""Benchmark redacting details with a secret key."""
redact_error_details(self.details_secret)
def time_redact_clean(self) -> None:
"""Benchmark redacting details without secrets."""
redact_error_details(self.details_clean)
def time_redact_value(self) -> None:
"""Benchmark redacting a string value."""
redact_value(self.value_with_secret)
class WrapBench:
"""Benchmark exception wrapping throughput."""
timeout = 60
def setup(self) -> None:
self.bare_err = Exception("crash")
def time_wrap_bare(self) -> None:
"""Benchmark wrapping a bare Exception."""
wrap_unexpected(self.bare_err)
def time_wrap_with_context(self) -> None:
"""Benchmark wrapping with context."""
wrap_unexpected(self.bare_err, context={"plan_id": "01PLAN"})
+125
View File
@@ -0,0 +1,125 @@
# Error Handling
## Overview
The `cleveragents.core.error_handling` module provides centralized
error classification, secret redaction, and exception wrapping for
consistent CLI output and safe logging.
## Error Codes
All errors are mapped to HTTP-like numeric codes:
### Client Errors (400-range)
| Code | Name | Mapped From |
|------|---------------------|-------------------------------|
| 400 | BAD_REQUEST | BusinessRuleViolation, PlanError, DomainError |
| 401 | UNAUTHORIZED | AuthenticationError |
| 403 | FORBIDDEN | AuthorizationError |
| 404 | NOT_FOUND | ResourceNotFoundError |
| 409 | CONFLICT | ResourceConflictError |
| 422 | VALIDATION_FAILED | ValidationError |
| 429 | RATE_LIMITED | RateLimitError |
### Server Errors (500-range)
| Code | Name | Mapped From |
|------|---------------------|-------------------------------|
| 500 | INTERNAL | CleverAgentsError (catch-all) |
| 503 | SERVICE_UNAVAILABLE | ExternalServiceError |
| 520 | PROVIDER_ERROR | ProviderError |
| 521 | EXECUTION_ERROR | ExecutionError |
| 522 | DATABASE_ERROR | DatabaseError |
| 523 | FILESYSTEM_ERROR | FileSystemError |
| 524 | NETWORK_ERROR | NetworkError |
| 525 | CONFIGURATION_ERROR | ConfigurationError |
| 526 | TOKEN_LIMIT | TokenLimitExceededError |
| 527 | MODEL_UNAVAILABLE | ModelNotAvailableError |
| 528 | STREAM_ERROR | StreamRoutingError |
## Secret Redaction
### Redacted Keys
The following keys are automatically redacted:
`api_key`, `apikey`, `auth_token`, `authorization`, `bearer`,
`client_secret`, `credentials`, `password`, `private_key`, `secret`,
`secret_key`, `token`
### Pattern-Based Redaction
String values are scanned for these patterns:
- OpenAI API keys (`sk-...`)
- JWT tokens (`eyJ...`)
- GitHub PATs (`ghp_...`, `ghs_...`)
- GitLab PATs (`glpat-...`)
### Usage
```python
from cleveragents.core.error_handling import redact_error_details
safe = redact_error_details({
"api_key": "sk-real-secret",
"plan_id": "01PLAN001",
})
# => {"api_key": "***REDACTED***", "plan_id": "01PLAN001"}
```
## Wrapping Unexpected Exceptions
Use `wrap_unexpected` to convert non-`CleverAgentsError` exceptions
into safe, user-facing errors:
```python
from cleveragents.core.error_handling import wrap_unexpected
try:
risky_operation()
except Exception as exc:
safe = wrap_unexpected(exc, context={"plan_id": plan_id})
raise safe from exc
```
If the exception is already a `CleverAgentsError`, it is returned
unchanged with optional context merged.
## CLI Formatting
```python
from cleveragents.core.error_handling import classify_error, format_error_for_cli
info = classify_error(exc)
print(format_error_for_cli(info))
# Error [422] VALIDATION_FAILED: name is required
```
## Exception Hierarchy
```
CleverAgentsError
├── DomainError
│ ├── ValidationError
│ ├── BusinessRuleViolation
│ ├── ResourceNotFoundError
│ ├── ResourceConflictError
│ └── PlanError
├── InfrastructureError
│ ├── DatabaseError
│ ├── NetworkError
│ └── ExternalServiceError
├── ProviderError
│ ├── RateLimitError
│ ├── TokenLimitExceededError
│ └── ModelNotAvailableError
├── AuthenticationError
├── AuthorizationError
├── ConfigurationError
│ └── MissingConfigurationError
├── FileSystemError
├── ExecutionError
└── StreamRoutingError
```
+207
View File
@@ -0,0 +1,207 @@
Feature: Explicit exception handling and error classification
As a security-conscious developer
I want structured error types with redaction and classification
So that errors are safe for users and diagnostics are preserved
Background:
Given a security exceptions test environment
# -- Error code mapping ---------------------------------------------------
Scenario: Classify a ValidationError as 422
Given a ValidationError with message "field is invalid"
When I classify the security exception
Then the error code should be 422
And the error code name should be "VALIDATION_FAILED"
And the error handling category should be "CLIENT"
Scenario: Classify a ResourceNotFoundError as 404
Given a ResourceNotFoundError for resource "plan" with id "01PLAN001"
When I classify the security exception
Then the error code should be 404
And the error code name should be "NOT_FOUND"
Scenario: Classify an AuthenticationError as 401
Given an AuthenticationError with message "invalid token"
When I classify the security exception
Then the error code should be 401
And the error code name should be "UNAUTHORIZED"
Scenario: Classify an AuthorizationError as 403
Given an AuthorizationError with message "access denied"
When I classify the security exception
Then the error code should be 403
And the error code name should be "FORBIDDEN"
Scenario: Classify a RateLimitError as 429
Given a RateLimitError with message "slow down"
When I classify the security exception
Then the error code should be 429
And the error code name should be "RATE_LIMITED"
Scenario: Classify a DatabaseError as 522
Given a DatabaseError with message "connection lost"
When I classify the security exception
Then the error code should be 522
And the error handling category should be "SERVER"
Scenario: Classify a ConfigurationError as 525
Given a ConfigurationError with message "missing key"
When I classify the security exception
Then the error code should be 525
Scenario: Classify a NetworkError as 524
Given a NetworkError with message "timeout"
When I classify the security exception
Then the error code should be 524
Scenario: Classify a FileSystemError as 523
Given a FileSystemError with message "permission denied"
When I classify the security exception
Then the error code should be 523
Scenario: Classify a ProviderError as 520
Given a ProviderError with message "model failed"
When I classify the security exception
Then the error code should be 520
Scenario: Classify a TokenLimitExceededError as 526
Given a TokenLimitExceededError with message "context too long"
When I classify the security exception
Then the error code should be 526
Scenario: Classify a BusinessRuleViolation as 400
Given a BusinessRuleViolation with message "invalid transition"
When I classify the security exception
Then the error code should be 400
Scenario: Classify a PlanError as 400
Given a PlanError with message "plan stuck"
When I classify the security exception
Then the error code should be 400
Scenario: Classify an unknown Exception as 500
Given a bare Exception with message "something broke"
When I classify the security exception
Then the error code should be 500
And the error handling category should be "SERVER"
Scenario: Classify an ExecutionError as 521
Given an ExecutionError with message "agent failed"
When I classify the security exception
Then the error code should be 521
Scenario: Classify a StreamRoutingError as 528
Given a StreamRoutingError with message "routing failed"
When I classify the security exception
Then the error code should be 528
Scenario: Classify a ModelNotAvailableError as 527
Given a ModelNotAvailableError with message "deprecated"
When I classify the security exception
Then the error code should be 527
Scenario: Classify a ResourceConflictError as 409
Given a ResourceConflictError with message "already exists"
When I classify the security exception
Then the error code should be 409
Scenario: Classify a MissingConfigurationError as 525
Given a MissingConfigurationError with message "key not set"
When I classify the security exception
Then the error code should be 525
Scenario: Classify an ExternalServiceError as 503
Given an ExternalServiceError with message "service down"
When I classify the security exception
Then the error code should be 503
# -- Secret redaction -----------------------------------------------------
Scenario: Redact API key from error details
Given redaction test details with key "api_key" and value "sk-abc123secret"
When I redact the error details
Then the redacted detail "api_key" should be "***REDACTED***"
Scenario: Redact password from error details
Given redaction test details with key "password" and value "my-secret-pass"
When I redact the error details
Then the redacted detail "password" should be "***REDACTED***"
Scenario: Redact OpenAI key pattern from string value
Given redaction test details with key "message" and value "Failed with key sk-abcdefghijklmnopqrstuvwxyz"
When I redact the error details
Then the redacted detail "message" should contain "***REDACTED***"
And the redacted detail "message" should not contain "sk-abcdefghijklmnopqrstuvwxyz"
Scenario: Redact JWT token pattern from string value
Given redaction test details with key "log" and value "auth eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9 failed"
When I redact the error details
Then the redacted detail "log" should contain "***REDACTED***"
Scenario: Redact nested dict values
Given redaction test details with nested key "config" containing secret_key "abc123"
When I redact the error details
Then the redacted nested detail "config" key "secret_key" should be "***REDACTED***"
Scenario: Non-sensitive values pass through unchanged
Given redaction test details with key "plan_id" and value "01PLAN001"
When I redact the error details
Then the redacted detail "plan_id" should be "01PLAN001"
Scenario: Redact GitHub PAT pattern from string value
Given redaction test details with key "msg" and value "token ghp_abcdefghijklmnopqrstuvwxyz0123456 expired"
When I redact the error details
Then the redacted detail "msg" should contain "***REDACTED***"
Scenario: Redact value with redact_value function
Given a redaction test string value "key=sk-abcdefghijklmnopqrstuvwxyz"
When I redact the string value
Then the redacted string should contain "***REDACTED***"
# -- Wrapping unexpected exceptions ----------------------------------------
Scenario: Wrap a bare Exception into CleverAgentsError
Given a bare Exception with message "internal crash"
When I wrap the unexpected exception
Then the wrapped error should be a CleverAgentsError
And the wrapped error message should be "An unexpected error occurred"
And the wrapped error details should have key "exception_type"
Scenario: Wrap preserves CleverAgentsError as-is
Given a CleverAgentsError with message "known issue"
When I wrap the unexpected exception
Then the wrapped error message should be "known issue"
Scenario: Wrap merges context into CleverAgentsError details
Given a CleverAgentsError with message "known issue"
And a wrapping context with key "plan_id" and value "01PLAN001"
When I wrap the unexpected exception
Then the wrapped error details should have key "plan_id"
Scenario: Wrap with custom safe message
Given a bare Exception with message "segfault"
When I wrap the unexpected exception with message "Operation failed"
Then the wrapped error message should be "Operation failed"
Scenario: Wrap redacts secrets in context
Given a bare Exception with message "oops"
And a wrapping context with key "api_key" and value "sk-secret123456789012345"
When I wrap the unexpected exception
Then the wrapped error details key "api_key" should be "***REDACTED***"
# -- CLI formatting -------------------------------------------------------
Scenario: Format error for CLI output
Given a ValidationError with message "name is required"
When I classify the security exception
And I format the error for CLI
Then the error CLI output should contain "422"
And the error CLI output should contain "VALIDATION_FAILED"
And the error CLI output should contain "name is required"
Scenario: Format error with details for CLI output
Given a detailed CleverAgentsError "db fail" with detail "table" as "plans"
When I classify the security exception
And I format the error for CLI
Then the error CLI output should contain "table: plans"
+350
View File
@@ -0,0 +1,350 @@
"""Step definitions for security exception handling tests."""
from __future__ import annotations
from behave import given, then, when
from behave.runner import Context
from cleveragents.core.error_handling import (
ErrorCategory,
classify_error,
format_error_for_cli,
redact_error_details,
redact_value,
wrap_unexpected,
)
from cleveragents.core.exceptions import (
AuthenticationError,
AuthorizationError,
BusinessRuleViolation,
CleverAgentsError,
ConfigurationError,
DatabaseError,
ExecutionError,
ExternalServiceError,
FileSystemError,
MissingConfigurationError,
ModelNotAvailableError,
NetworkError,
PlanError,
ProviderError,
RateLimitError,
ResourceConflictError,
ResourceNotFoundError,
StreamRoutingError,
TokenLimitExceededError,
ValidationError,
)
__all__: list[str] = []
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a security exceptions test environment")
def step_given_sec_exc_env(context: Context) -> None:
context.sec_exc = None # Exception
context.sec_error_info = None # ErrorInfo | None
context.sec_details = {} # dict
context.sec_redacted = {} # dict
context.sec_wrapped = None # CleverAgentsError | None
context.sec_cli_output = ""
context.sec_wrap_context = {} # dict
context.sec_string_value = ""
context.sec_redacted_string = ""
# ---------------------------------------------------------------------------
# Given — exception types
# ---------------------------------------------------------------------------
@given('a ValidationError with message "{msg}"')
def step_given_validation_error(context: Context, msg: str) -> None:
context.sec_exc = ValidationError(msg)
@given('a ResourceNotFoundError for resource "{rtype}" with id "{rid}"')
def step_given_not_found(context: Context, rtype: str, rid: str) -> None:
context.sec_exc = ResourceNotFoundError(resource_type=rtype, resource_id=rid)
@given('an AuthenticationError with message "{msg}"')
def step_given_auth_error(context: Context, msg: str) -> None:
context.sec_exc = AuthenticationError(msg)
@given('an AuthorizationError with message "{msg}"')
def step_given_authz_error(context: Context, msg: str) -> None:
context.sec_exc = AuthorizationError(msg)
@given('a RateLimitError with message "{msg}"')
def step_given_rate_limit(context: Context, msg: str) -> None:
context.sec_exc = RateLimitError(msg)
@given('a DatabaseError with message "{msg}"')
def step_given_db_error(context: Context, msg: str) -> None:
context.sec_exc = DatabaseError(msg)
@given('a ConfigurationError with message "{msg}"')
def step_given_config_error(context: Context, msg: str) -> None:
context.sec_exc = ConfigurationError(msg)
@given('a NetworkError with message "{msg}"')
def step_given_network_error(context: Context, msg: str) -> None:
context.sec_exc = NetworkError(msg)
@given('a FileSystemError with message "{msg}"')
def step_given_fs_error(context: Context, msg: str) -> None:
context.sec_exc = FileSystemError(msg)
@given('a ProviderError with message "{msg}"')
def step_given_provider_error(context: Context, msg: str) -> None:
context.sec_exc = ProviderError(msg)
@given('a TokenLimitExceededError with message "{msg}"')
def step_given_token_limit(context: Context, msg: str) -> None:
context.sec_exc = TokenLimitExceededError(msg)
@given('a BusinessRuleViolation with message "{msg}"')
def step_given_biz_rule(context: Context, msg: str) -> None:
context.sec_exc = BusinessRuleViolation(msg)
@given('a PlanError with message "{msg}"')
def step_given_plan_error(context: Context, msg: str) -> None:
context.sec_exc = PlanError(msg)
@given('a bare Exception with message "{msg}"')
def step_given_bare_exception(context: Context, msg: str) -> None:
context.sec_exc = Exception(msg)
@given('an ExecutionError with message "{msg}"')
def step_given_execution_error(context: Context, msg: str) -> None:
context.sec_exc = ExecutionError(msg)
@given('a StreamRoutingError with message "{msg}"')
def step_given_stream_error(context: Context, msg: str) -> None:
context.sec_exc = StreamRoutingError(msg)
@given('a ModelNotAvailableError with message "{msg}"')
def step_given_model_unavail(context: Context, msg: str) -> None:
context.sec_exc = ModelNotAvailableError(msg)
@given('a ResourceConflictError with message "{msg}"')
def step_given_conflict(context: Context, msg: str) -> None:
context.sec_exc = ResourceConflictError(msg)
@given('a MissingConfigurationError with message "{msg}"')
def step_given_missing_config(context: Context, msg: str) -> None:
context.sec_exc = MissingConfigurationError(msg)
@given('an ExternalServiceError with message "{msg}"')
def step_given_ext_svc_error(context: Context, msg: str) -> None:
context.sec_exc = ExternalServiceError(msg)
@given('a CleverAgentsError with message "{msg}"')
def step_given_ca_error(context: Context, msg: str) -> None:
context.sec_exc = CleverAgentsError(msg)
@given('a detailed CleverAgentsError "{msg}" with detail "{k}" as "{v}"')
def step_given_ca_error_details(context: Context, msg: str, k: str, v: str) -> None:
context.sec_exc = CleverAgentsError(msg, details={k: v})
# ---------------------------------------------------------------------------
# Given — error details for redaction
# ---------------------------------------------------------------------------
@given('redaction test details with key "{key}" and value "{value}"')
def step_given_error_details(context: Context, key: str, value: str) -> None:
context.sec_details[key] = value
@given(
'redaction test details with nested key "{outer}" containing secret_key "{value}"'
)
def step_given_nested_details(context: Context, outer: str, value: str) -> None:
context.sec_details[outer] = {"secret_key": value}
@given('a redaction test string value "{value}"')
def step_given_string_value(context: Context, value: str) -> None:
context.sec_string_value = value
# ---------------------------------------------------------------------------
# Given — wrapping context
# ---------------------------------------------------------------------------
@given('a wrapping context with key "{key}" and value "{value}"')
def step_given_wrap_context(context: Context, key: str, value: str) -> None:
context.sec_wrap_context[key] = value
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
@when("I classify the security exception")
def step_when_classify(context: Context) -> None:
context.sec_error_info = classify_error(context.sec_exc)
@when("I redact the error details")
def step_when_redact_details(context: Context) -> None:
context.sec_redacted = redact_error_details(context.sec_details)
@when("I redact the string value")
def step_when_redact_string(context: Context) -> None:
context.sec_redacted_string = redact_value(context.sec_string_value)
@when("I wrap the unexpected exception")
def step_when_wrap(context: Context) -> None:
ctx = context.sec_wrap_context or None
context.sec_wrapped = wrap_unexpected(context.sec_exc, context=ctx)
@when('I wrap the unexpected exception with message "{msg}"')
def step_when_wrap_msg(context: Context, msg: str) -> None:
context.sec_wrapped = wrap_unexpected(context.sec_exc, safe_message=msg)
@when("I format the error for CLI")
def step_when_format_cli(context: Context) -> None:
assert context.sec_error_info is not None
context.sec_cli_output = format_error_for_cli(context.sec_error_info)
# ---------------------------------------------------------------------------
# Then steps — classification
# ---------------------------------------------------------------------------
@then("the error code should be {code}")
def step_then_error_code(context: Context, code: str) -> None:
assert context.sec_error_info is not None
assert context.sec_error_info.code.value == int(code), (
f"Expected {code}, got {context.sec_error_info.code.value}"
)
@then('the error code name should be "{name}"')
def step_then_error_code_name(context: Context, name: str) -> None:
assert context.sec_error_info is not None
assert context.sec_error_info.code.name == name, (
f"Expected {name}, got {context.sec_error_info.code.name}"
)
@then('the error handling category should be "{cat}"')
def step_then_error_category(context: Context, cat: str) -> None:
assert context.sec_error_info is not None
expected = ErrorCategory.CLIENT if cat == "CLIENT" else ErrorCategory.SERVER
assert context.sec_error_info.category == expected
# ---------------------------------------------------------------------------
# Then steps — redaction
# ---------------------------------------------------------------------------
@then('the redacted detail "{key}" should be "{expected}"')
def step_then_redacted_detail(context: Context, key: str, expected: str) -> None:
assert context.sec_redacted[key] == expected, (
f"Expected {expected!r}, got {context.sec_redacted[key]!r}"
)
@then('the redacted detail "{key}" should contain "{text}"')
def step_then_redacted_contains(context: Context, key: str, text: str) -> None:
val = str(context.sec_redacted[key])
assert text in val, f"Expected '{text}' in {val!r}"
@then('the redacted detail "{key}" should not contain "{text}"')
def step_then_redacted_not_contains(context: Context, key: str, text: str) -> None:
val = str(context.sec_redacted[key])
assert text not in val, f"Did not expect '{text}' in {val!r}"
@then('the redacted nested detail "{outer}" key "{inner}" should be "{expected}"')
def step_then_nested_redacted(
context: Context, outer: str, inner: str, expected: str
) -> None:
nested = context.sec_redacted[outer]
assert nested[inner] == expected
@then('the redacted string should contain "{text}"')
def step_then_redacted_string_contains(context: Context, text: str) -> None:
assert text in context.sec_redacted_string
# ---------------------------------------------------------------------------
# Then steps — wrapping
# ---------------------------------------------------------------------------
@then("the wrapped error should be a CleverAgentsError")
def step_then_wrapped_is_ca(context: Context) -> None:
assert isinstance(context.sec_wrapped, CleverAgentsError)
@then('the wrapped error message should be "{msg}"')
def step_then_wrapped_msg(context: Context, msg: str) -> None:
assert context.sec_wrapped is not None
assert context.sec_wrapped.message == msg, (
f"Expected {msg!r}, got {context.sec_wrapped.message!r}"
)
@then('the wrapped error details should have key "{key}"')
def step_then_wrapped_details_key(context: Context, key: str) -> None:
assert context.sec_wrapped is not None
assert key in context.sec_wrapped.details, (
f"Key '{key}' not in {list(context.sec_wrapped.details.keys())}"
)
@then('the wrapped error details key "{key}" should be "{expected}"')
def step_then_wrapped_detail_value(context: Context, key: str, expected: str) -> None:
assert context.sec_wrapped is not None
assert context.sec_wrapped.details[key] == expected
# ---------------------------------------------------------------------------
# Then steps — CLI formatting
# ---------------------------------------------------------------------------
@then('the error CLI output should contain "{text}"')
def step_then_cli_contains(context: Context, text: str) -> None:
assert text in context.sec_cli_output, (
f"Expected '{text}' in:\n{context.sec_cli_output}"
)
+19 -19
View File
@@ -5396,25 +5396,25 @@ By Day 30, the system must be able to (all in LOCAL MODE, server connectivity is
- [ ] Forgejo PR [Luis]: Open PR from `feature/m4-security-template` to `master` with a suitable and thorough description
**Parallel Group SEC3: Exception Handling Audit [Luis]**
- [ ] **COMMIT (Owner: Luis | Group: SEC3.exceptions | Branch: feature/m4-security-exceptions | Planned: Day 12 | Expected: Day 26) - Commit message: "fix(security): enforce explicit exception handling"**
- [ ] Git [Luis]: `git checkout master`
- [ ] Git [Luis]: `git pull origin master`
- [ ] Git [Luis]: `git checkout -b feature/m4-security-exceptions`
- [ ] Git [Luis]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Luis]: Replace silent exception handling with explicit errors and context propagation.
- [ ] Code [Luis]: Add structured error types for config, provider, and file I/O failures and include plan_id/project_name in error details.
- [ ] Code [Luis]: Ensure unexpected exceptions are wrapped in `CleverAgentsError` with a safe, user-facing message.
- [ ] Code [Luis]: Add error code mapping table (error_code -> HTTP-like category) for CLI output consistency.
- [ ] Code [Luis]: Ensure error details are redacted for secrets before logging.
- [ ] Docs [Luis]: Document error propagation standards and logging rules.
- [ ] Tests (Behave) [Luis]: Add `features/security_exceptions.feature` scenarios.
- [ ] Tests (Robot) [Luis]: Add exception handling integration smoke tests.
- [ ] Tests (ASV) [Luis]: Add `benchmarks/security_exception_bench.py` for error path overhead baseline.
- [ ] Quality [Luis]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [ ] Quality [Luis]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
- [ ] Git [Luis]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
- [ ] Git [Luis]: `git commit -m "fix(security): enforce explicit exception handling"`
- [ ] Git [Luis]: `git push -u origin feature/m4-security-exceptions`
- [X] **COMMIT (Owner: Luis | Group: SEC3.exceptions | Branch: feature/m4-security-exceptions | Planned: Day 12 | Expected: Day 26) - Commit message: "fix(security): enforce explicit exception handling"**
- [X] Git [Luis]: `git checkout master`
- [X] Git [Luis]: `git pull origin master`
- [X] Git [Luis]: `git checkout -b feature/m4-security-exceptions`
- [X] Git [Luis]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [X] Code [Luis]: Replace silent exception handling with explicit errors and context propagation.
- [X] Code [Luis]: Add structured error types for config, provider, and file I/O failures and include plan_id/project_name in error details.
- [X] Code [Luis]: Ensure unexpected exceptions are wrapped in `CleverAgentsError` with a safe, user-facing message.
- [X] Code [Luis]: Add error code mapping table (error_code -> HTTP-like category) for CLI output consistency.
- [X] Code [Luis]: Ensure error details are redacted for secrets before logging.
- [X] Docs [Luis]: Document error propagation standards and logging rules.
- [X] Tests (Behave) [Luis]: Add `features/security_exceptions.feature` scenarios.
- [X] Tests (Robot) [Luis]: Add exception handling integration smoke tests.
- [X] Tests (ASV) [Luis]: Add `benchmarks/security_exception_bench.py` for error path overhead baseline.
- [X] Quality [Luis]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [X] Quality [Luis]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
- [X] Git [Luis]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
- [X] Git [Luis]: `git commit -m "fix(security): enforce explicit exception handling"`
- [X] Git [Luis]: `git push -u origin feature/m4-security-exceptions`
- [ ] Forgejo PR [Luis]: Open PR from `feature/m4-security-exceptions` to `master` with a suitable and thorough description
**Parallel Group SEC4: Async Lifecycle Correctness [Luis]**
+109
View File
@@ -0,0 +1,109 @@
"""Robot Framework helper for exception handling tests.
Provides a CLI-style interface for Robot to invoke error
classification, redaction, and wrapping.
Usage::
python robot/helper_security_exceptions.py classify <error_type>
python robot/helper_security_exceptions.py redact <key> <value>
python robot/helper_security_exceptions.py wrap <message>
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.core.error_handling import ( # noqa: E402
classify_error,
format_error_for_cli,
redact_error_details,
wrap_unexpected,
)
from cleveragents.core.exceptions import ( # noqa: E402
AuthenticationError,
CleverAgentsError,
ConfigurationError,
DatabaseError,
ValidationError,
)
_ERROR_TYPES: dict[str, type[Exception]] = {
"validation": ValidationError,
"auth": AuthenticationError,
"config": ConfigurationError,
"database": DatabaseError,
"bare": Exception,
}
def _cmd_classify(args: list[str]) -> int:
"""Classify an error by type name."""
if len(args) < 1:
print("Usage: classify <error_type>")
return 1
etype = args[0]
exc_cls = _ERROR_TYPES.get(etype, Exception)
exc = exc_cls("test error")
info = classify_error(exc)
output = format_error_for_cli(info)
print(f"exc-classify-ok: code={info.code.value} name={info.code.name}")
print(output)
return 0
def _cmd_redact(args: list[str]) -> int:
"""Redact a key-value pair."""
if len(args) < 2:
print("Usage: redact <key> <value>")
return 1
details = {args[0]: args[1]}
redacted = redact_error_details(details)
print(f"exc-redact-ok: {json.dumps(redacted)}")
return 0
def _cmd_wrap(args: list[str]) -> int:
"""Wrap a bare Exception."""
if len(args) < 1:
print("Usage: wrap <message>")
return 1
exc = Exception(args[0])
wrapped = wrap_unexpected(exc)
is_ca = isinstance(wrapped, CleverAgentsError)
print(f"exc-wrap-ok: is_ca={is_ca} msg={wrapped.message}")
return 0
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print("Usage: helper_security_exceptions.py <command> ...")
return 1
command = sys.argv[1]
args = sys.argv[2:]
dispatch = {
"classify": _cmd_classify,
"redact": _cmd_redact,
"wrap": _cmd_wrap,
}
handler = dispatch.get(command)
if handler is None:
print(f"Unknown command: {command}")
return 1
return handler(args)
if __name__ == "__main__":
sys.exit(main())
+42
View File
@@ -0,0 +1,42 @@
*** Settings ***
Documentation Integration tests for exception handling module
Resource common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Test Cases ***
Classify Validation Error
[Documentation] Verify ValidationError classifies as 422
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_exceptions.py
... classify validation
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} code=422
Classify Database Error
[Documentation] Verify DatabaseError classifies as 522
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_exceptions.py
... classify database
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} code=522
Redact Secret Key
[Documentation] Verify api_key is redacted
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_exceptions.py
... redact api_key sk-secret12345
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} ***REDACTED***
Wrap Bare Exception
[Documentation] Verify bare Exception is wrapped in CleverAgentsError
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_exceptions.py
... wrap unexpected-crash
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} is_ca=True
Redact Non-Secret Key
[Documentation] Verify non-secret values pass through
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_exceptions.py
... redact plan_id 01PLAN001
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} 01PLAN001
Should Not Contain ${result.stdout} REDACTED
+374
View File
@@ -0,0 +1,374 @@
"""Explicit exception handling, error codes, and secret redaction.
Provides a unified error handling layer that:
- Maps exception types to HTTP-like error categories for consistent
CLI and API output.
- Redacts sensitive values from error details before logging.
- Wraps unexpected exceptions in ``CleverAgentsError`` with a safe,
user-facing message that hides internal stack details.
Usage::
from cleveragents.core.error_handling import (
ErrorCode,
classify_error,
redact_error_details,
wrap_unexpected,
)
try:
risky_operation()
except Exception as exc:
safe = wrap_unexpected(exc, context={"plan_id": plan_id})
raise safe from exc
"""
from __future__ import annotations
import logging
import re
import traceback
from enum import IntEnum
from typing import Any
from cleveragents.core.exceptions import (
AuthenticationError,
AuthorizationError,
BusinessRuleViolation,
CleverAgentsError,
ConfigurationError,
DatabaseError,
DomainError,
ExecutionError,
ExternalServiceError,
FileSystemError,
InfrastructureError,
MissingConfigurationError,
ModelNotAvailableError,
NetworkError,
PlanError,
ProviderError,
RateLimitError,
ResourceConflictError,
ResourceNotFoundError,
StreamRoutingError,
TokenLimitExceededError,
ValidationError,
)
__all__: list[str] = [
"ErrorCategory",
"ErrorCode",
"ErrorInfo",
"classify_error",
"format_error_for_cli",
"redact_error_details",
"redact_value",
"wrap_unexpected",
]
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Error code registry
# ---------------------------------------------------------------------------
class ErrorCode(IntEnum):
"""HTTP-like error codes for CLI output consistency.
Maps to conventional HTTP status code ranges:
- 400-range: client/input errors
- 500-range: server/internal errors
"""
# 400-range: Input / domain errors
BAD_REQUEST = 400
UNAUTHORIZED = 401
FORBIDDEN = 403
NOT_FOUND = 404
CONFLICT = 409
VALIDATION_FAILED = 422
RATE_LIMITED = 429
# 500-range: Internal / infrastructure errors
INTERNAL = 500
NOT_IMPLEMENTED = 501
SERVICE_UNAVAILABLE = 503
PROVIDER_ERROR = 520
EXECUTION_ERROR = 521
DATABASE_ERROR = 522
FILESYSTEM_ERROR = 523
NETWORK_ERROR = 524
CONFIGURATION_ERROR = 525
TOKEN_LIMIT = 526
MODEL_UNAVAILABLE = 527
STREAM_ERROR = 528
class ErrorCategory(IntEnum):
"""Broad error category for grouping."""
CLIENT = 4
SERVER = 5
# ---------------------------------------------------------------------------
# Exception -> ErrorCode mapping
# ---------------------------------------------------------------------------
#: Ordered list of (exception_type, error_code) pairs.
#: More specific types appear before more general ones so the first
#: match wins.
_EXCEPTION_MAP: list[tuple[type[Exception], ErrorCode]] = [
# Domain / client errors
(ValidationError, ErrorCode.VALIDATION_FAILED),
(ResourceNotFoundError, ErrorCode.NOT_FOUND),
(ResourceConflictError, ErrorCode.CONFLICT),
(BusinessRuleViolation, ErrorCode.BAD_REQUEST),
(PlanError, ErrorCode.BAD_REQUEST),
(DomainError, ErrorCode.BAD_REQUEST),
# Auth
(AuthenticationError, ErrorCode.UNAUTHORIZED),
(AuthorizationError, ErrorCode.FORBIDDEN),
# Config
(MissingConfigurationError, ErrorCode.CONFIGURATION_ERROR),
(ConfigurationError, ErrorCode.CONFIGURATION_ERROR),
# Provider
(RateLimitError, ErrorCode.RATE_LIMITED),
(TokenLimitExceededError, ErrorCode.TOKEN_LIMIT),
(ModelNotAvailableError, ErrorCode.MODEL_UNAVAILABLE),
(ProviderError, ErrorCode.PROVIDER_ERROR),
# Infrastructure
(DatabaseError, ErrorCode.DATABASE_ERROR),
(NetworkError, ErrorCode.NETWORK_ERROR),
(ExternalServiceError, ErrorCode.SERVICE_UNAVAILABLE),
(FileSystemError, ErrorCode.FILESYSTEM_ERROR),
(InfrastructureError, ErrorCode.INTERNAL),
# Execution
(StreamRoutingError, ErrorCode.STREAM_ERROR),
(ExecutionError, ErrorCode.EXECUTION_ERROR),
# Catch-all for CleverAgentsError
(CleverAgentsError, ErrorCode.INTERNAL),
]
# ---------------------------------------------------------------------------
# Secret redaction
# ---------------------------------------------------------------------------
#: Keys whose values should always be redacted.
_REDACT_KEYS: frozenset[str] = frozenset(
{
"api_key",
"apikey",
"auth_token",
"authorization",
"bearer",
"client_secret",
"credentials",
"password",
"private_key",
"secret",
"secret_key",
"token",
}
)
#: Regex patterns that look like secrets (e.g. long hex strings, JWTs).
_SECRET_PATTERNS: list[re.Pattern[str]] = [
re.compile(r"sk-[a-zA-Z0-9]{20,}"), # OpenAI-style API keys
re.compile(r"eyJ[a-zA-Z0-9_-]{10,}"), # JWT tokens
re.compile(r"ghp_[a-zA-Z0-9]{30,}"), # GitHub PATs
re.compile(r"ghs_[a-zA-Z0-9]{30,}"), # GitHub App tokens
re.compile(r"glpat-[a-zA-Z0-9_-]{20,}"), # GitLab PATs
]
_REDACTED = "***REDACTED***"
def redact_value(value: str) -> str:
"""Redact secret patterns from a string value.
Args:
value: The string to redact.
Returns:
The string with secret patterns replaced by ``***REDACTED***``.
"""
result = value
for pattern in _SECRET_PATTERNS:
result = pattern.sub(_REDACTED, result)
return result
def redact_error_details(details: dict[str, Any]) -> dict[str, Any]:
"""Redact sensitive values from error details.
Keys matching known secret names have their values fully replaced.
String values are scanned for secret patterns.
Args:
details: The error details dict.
Returns:
A new dict with sensitive values redacted.
"""
redacted: dict[str, Any] = {}
for key, value in details.items():
key_lower = key.lower().replace("-", "_").replace(" ", "_")
if key_lower in _REDACT_KEYS:
redacted[key] = _REDACTED
elif isinstance(value, str):
redacted[key] = redact_value(value)
elif isinstance(value, dict):
redacted[key] = redact_error_details(value)
else:
redacted[key] = value
return redacted
# ---------------------------------------------------------------------------
# Error classification
# ---------------------------------------------------------------------------
class ErrorInfo:
"""Structured error information with code and redacted details.
Attributes:
code: HTTP-like error code.
category: Broad error category (client or server).
message: User-safe error message.
exception_type: The original exception class name.
details: Redacted error details dict.
"""
__slots__ = ("category", "code", "details", "exception_type", "message")
def __init__(
self,
*,
code: ErrorCode,
message: str,
exception_type: str,
details: dict[str, Any] | None = None,
) -> None:
self.code = code
self.category = ErrorCategory.CLIENT if code < 500 else ErrorCategory.SERVER
self.message = message
self.exception_type = exception_type
self.details = details or {}
def classify_error(exc: Exception) -> ErrorInfo:
"""Classify an exception into a structured :class:`ErrorInfo`.
Walks the ``_EXCEPTION_MAP`` to find the most specific match,
redacts details, and returns a safe ``ErrorInfo``.
Args:
exc: The exception to classify.
Returns:
An ``ErrorInfo`` with the error code, redacted details, and
user-safe message.
"""
code = ErrorCode.INTERNAL
for exc_type, mapped_code in _EXCEPTION_MAP:
if isinstance(exc, exc_type):
code = mapped_code
break
details: dict[str, Any] = {}
if isinstance(exc, CleverAgentsError) and exc.details:
details = redact_error_details(exc.details)
return ErrorInfo(
code=code,
message=str(exc),
exception_type=type(exc).__name__,
details=details,
)
# ---------------------------------------------------------------------------
# Wrapping unexpected exceptions
# ---------------------------------------------------------------------------
def wrap_unexpected(
exc: Exception,
*,
context: dict[str, Any] | None = None,
safe_message: str = "An unexpected error occurred",
) -> CleverAgentsError:
"""Wrap an unexpected exception in ``CleverAgentsError``.
If the exception is already a ``CleverAgentsError``, it is returned
unchanged (with optional context merged into its details).
For non-``CleverAgentsError`` exceptions, a new error is created
with the *safe_message* (hiding internal details from users).
The original exception type and a sanitised traceback snippet are
stored in the ``details`` dict for diagnostics.
Args:
exc: The exception to wrap.
context: Optional extra context (e.g. ``plan_id``,
``project_name``) to merge into the error details.
safe_message: User-facing message when wrapping unknown
exceptions.
Returns:
A ``CleverAgentsError`` instance.
"""
extra_context = redact_error_details(context) if context else {}
if isinstance(exc, CleverAgentsError):
if extra_context:
exc.details.update(extra_context)
return exc
# Build sanitised details for diagnostics.
tb_lines = traceback.format_exception(type(exc), exc, exc.__traceback__)
tb_text = "".join(tb_lines[-3:]) # last 3 lines only
tb_text = redact_value(tb_text)
details: dict[str, Any] = {
"exception_type": type(exc).__name__,
"traceback_snippet": tb_text,
}
details.update(extra_context)
logger.error(
"Wrapping unexpected exception: %s: %s",
type(exc).__name__,
redact_value(str(exc)),
)
return CleverAgentsError(safe_message, details=details)
# ---------------------------------------------------------------------------
# CLI formatting
# ---------------------------------------------------------------------------
def format_error_for_cli(info: ErrorInfo) -> str:
"""Format an ``ErrorInfo`` for human-readable CLI output.
Args:
info: The structured error info.
Returns:
Multi-line string suitable for terminal display.
"""
lines: list[str] = [
f"Error [{info.code.value}] {info.code.name}: {info.message}",
]
if info.details:
for key, value in info.details.items():
lines.append(f" {key}: {value}")
return "\n".join(lines)