Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2051368b67 | |||
|
23b47a0c0e
|
@@ -0,0 +1,87 @@
|
||||
"""ASV benchmarks for secure template rendering.
|
||||
|
||||
Measures the overhead introduced by security checks in the
|
||||
template rendering pipeline.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from cleveragents.templates.secure_renderer import (
|
||||
SecureTemplateRenderer,
|
||||
TemplateConfig,
|
||||
render_template_secure,
|
||||
validate_template,
|
||||
)
|
||||
|
||||
|
||||
class RenderBench:
|
||||
"""Benchmark secure template rendering throughput."""
|
||||
|
||||
timeout = 60
|
||||
|
||||
def setup(self) -> None:
|
||||
self.renderer = SecureTemplateRenderer(
|
||||
config=TemplateConfig(
|
||||
allowed_keys=frozenset({"name", "role", "project"}),
|
||||
),
|
||||
)
|
||||
self.template = "Hello {name}, you are {role} on {project}"
|
||||
self.context = {
|
||||
"name": "Alice",
|
||||
"role": "developer",
|
||||
"project": "cleveragents",
|
||||
}
|
||||
|
||||
def time_render_simple(self) -> None:
|
||||
"""Benchmark rendering a simple template."""
|
||||
self.renderer.render(self.template, self.context)
|
||||
|
||||
def time_render_no_placeholders(self) -> None:
|
||||
"""Benchmark rendering template with no placeholders."""
|
||||
self.renderer.render("Static text with no vars", {})
|
||||
|
||||
def time_render_many_placeholders(self) -> None:
|
||||
"""Benchmark rendering template with many placeholders."""
|
||||
tpl = " ".join(f"{{k{i}}}" for i in range(50))
|
||||
ctx = {f"k{i}": f"v{i}" for i in range(50)}
|
||||
renderer = SecureTemplateRenderer(config=TemplateConfig())
|
||||
renderer.render(tpl, ctx)
|
||||
|
||||
|
||||
class ValidateBench:
|
||||
"""Benchmark template validation throughput."""
|
||||
|
||||
timeout = 60
|
||||
|
||||
def setup(self) -> None:
|
||||
self.safe_template = "Hello {name}, role={role}"
|
||||
self.unsafe_template = "{obj.__class__.__init__.__globals__}"
|
||||
|
||||
def time_validate_safe(self) -> None:
|
||||
"""Benchmark validating a safe template."""
|
||||
validate_template(self.safe_template)
|
||||
|
||||
def time_validate_unsafe(self) -> None:
|
||||
"""Benchmark validating an unsafe template."""
|
||||
validate_template(self.unsafe_template)
|
||||
|
||||
def time_validate_with_allowlist(self) -> None:
|
||||
"""Benchmark validating with an allowlist."""
|
||||
validate_template(
|
||||
self.safe_template,
|
||||
allowed_keys=frozenset({"name", "role"}),
|
||||
)
|
||||
|
||||
|
||||
class ConvenienceFunctionBench:
|
||||
"""Benchmark the convenience render_template_secure function."""
|
||||
|
||||
timeout = 60
|
||||
|
||||
def time_convenience_render(self) -> None:
|
||||
"""Benchmark the convenience function."""
|
||||
render_template_secure(
|
||||
"Hello {name}",
|
||||
{"name": "World"},
|
||||
allowed_keys=frozenset({"name"}),
|
||||
)
|
||||
@@ -0,0 +1,145 @@
|
||||
# Template Security
|
||||
|
||||
## Overview
|
||||
|
||||
The `cleveragents.templates.secure_renderer` module provides a sandboxed
|
||||
template renderer that prevents template injection attacks while
|
||||
supporting simple variable substitution.
|
||||
|
||||
## Threat Model
|
||||
|
||||
Template injection occurs when user-controlled input is embedded in
|
||||
template strings and evaluated. Python's `str.format()` is particularly
|
||||
dangerous because it allows:
|
||||
|
||||
- **Attribute access**: `{obj.__class__.__init__.__globals__}` can leak
|
||||
sensitive module-level data.
|
||||
- **Index access**: `{items[0]}` can read arbitrary collection entries.
|
||||
- **Format specs**: `{val:>10}` can cause unexpected formatting or DoS.
|
||||
- **Conversion flags**: `{val!r}` can change output representation.
|
||||
|
||||
The secure renderer rejects all of these, only allowing `{name}` style
|
||||
substitution with simple identifiers.
|
||||
|
||||
## Allowed Syntax
|
||||
|
||||
Only the following template syntax is permitted:
|
||||
|
||||
```
|
||||
{variable_name}
|
||||
```
|
||||
|
||||
Where `variable_name` matches `\w+` (letters, digits, underscore).
|
||||
|
||||
Everything else is rejected:
|
||||
|
||||
| Pattern | Risk | Result |
|
||||
|------------------|-------------------------------|----------|
|
||||
| `{x.y}` | Attribute access | Rejected |
|
||||
| `{x[0]}` | Index access | Rejected |
|
||||
| `{x:>10}` | Format spec | Rejected |
|
||||
| `{x!r}` | Conversion flag | Rejected |
|
||||
| `{fn()}` | Function call | Rejected |
|
||||
| `{{ x }}` | Jinja2 expression | Rejected |
|
||||
| `{% block %}` | Jinja2 block | Rejected |
|
||||
|
||||
## Configuration
|
||||
|
||||
Use `TemplateConfig` to configure the renderer:
|
||||
|
||||
```python
|
||||
from cleveragents.templates.secure_renderer import (
|
||||
SecureTemplateRenderer,
|
||||
TemplateConfig,
|
||||
)
|
||||
|
||||
config = TemplateConfig(
|
||||
allowed_keys=frozenset({"name", "description", "role"}),
|
||||
max_template_length=10_000, # default
|
||||
max_output_length=50_000, # default
|
||||
reject_unknown_keys=True, # default
|
||||
log_rejected=True, # default
|
||||
)
|
||||
renderer = SecureTemplateRenderer(config=config)
|
||||
```
|
||||
|
||||
### Parameters
|
||||
|
||||
- **`allowed_keys`**: Set of permitted placeholder names. Empty means
|
||||
any simple identifier is allowed (open mode).
|
||||
- **`max_template_length`**: Maximum raw template string length
|
||||
(default: 10,000). Prevents DoS via oversized templates.
|
||||
- **`max_output_length`**: Maximum rendered output length
|
||||
(default: 50,000). Prevents DoS via value expansion.
|
||||
- **`reject_unknown_keys`**: Whether to raise on keys not in the
|
||||
allowlist (default: `True`). Ignored when `allowed_keys` is empty.
|
||||
- **`log_rejected`**: Log rejected keys at WARNING level (default:
|
||||
`True`).
|
||||
|
||||
## Pre-validation
|
||||
|
||||
Use `validate_template()` at action/plan creation time to catch
|
||||
issues early:
|
||||
|
||||
```python
|
||||
from cleveragents.templates.secure_renderer import validate_template
|
||||
|
||||
errors = validate_template(
|
||||
template_text,
|
||||
allowed_keys=frozenset({"name", "description"}),
|
||||
)
|
||||
if errors:
|
||||
raise ValueError(f"Invalid template: {errors}")
|
||||
```
|
||||
|
||||
## Convenience Function
|
||||
|
||||
For one-off renders without creating a renderer instance:
|
||||
|
||||
```python
|
||||
from cleveragents.templates.secure_renderer import render_template_secure
|
||||
|
||||
result = render_template_secure(
|
||||
"Hello {name}",
|
||||
{"name": "World"},
|
||||
allowed_keys=frozenset({"name"}),
|
||||
)
|
||||
```
|
||||
|
||||
## Safe Substitution Behaviour
|
||||
|
||||
If a placeholder key is not found in the context, it is left as-is
|
||||
in the output (safe-substitute behaviour):
|
||||
|
||||
```python
|
||||
renderer.render("Hello {name}, {role}", {"name": "Alice"})
|
||||
# => "Hello Alice, {role}"
|
||||
```
|
||||
|
||||
## Legacy Compatibility
|
||||
|
||||
The existing `TemplateRenderer` in `cleveragents.templates.renderer`
|
||||
now wraps `SecureTemplateRenderer` internally. It returns the raw
|
||||
template on any rendering failure to preserve backward compatibility,
|
||||
and logs a WARNING when doing so (enable DEBUG for the full template).
|
||||
|
||||
An optional `config` parameter allows callers to pass a
|
||||
`TemplateConfig` through to the underlying secure renderer:
|
||||
|
||||
```python
|
||||
from cleveragents.templates.renderer import TemplateRenderer
|
||||
from cleveragents.templates.secure_renderer import TemplateConfig
|
||||
|
||||
renderer = TemplateRenderer(
|
||||
config=TemplateConfig(allowed_keys=frozenset({"name"})),
|
||||
)
|
||||
```
|
||||
|
||||
## Exception Hierarchy
|
||||
|
||||
```
|
||||
TemplateError (base)
|
||||
├── TemplateSecurityError — unsafe constructs
|
||||
├── TemplateSizeError — length limits exceeded
|
||||
└── TemplateValidationError — allowlist violations
|
||||
```
|
||||
@@ -0,0 +1,225 @@
|
||||
@coverage_boost
|
||||
Feature: Coverage boost for security template branch
|
||||
As a developer ensuring 97%+ test coverage
|
||||
I want to exercise uncovered code paths across the codebase
|
||||
So that the coverage threshold is comfortably met
|
||||
|
||||
# =========================================================================
|
||||
# skills/registry.py — validate_skill with tool registry, inline tools,
|
||||
# and includes (lines 200-213)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Validate skill with tool registry reports missing tool refs
|
||||
Given a skill registry with a mock tool registry
|
||||
And a skill definition with tool ref "nonexistent-tool"
|
||||
When I validate the skill definition via the registry
|
||||
Then the skill validation errors should mention "not found in tool registry"
|
||||
|
||||
Scenario: Validate skill with inline tool missing description
|
||||
Given a skill registry with no tool registry
|
||||
And a skill definition with an inline tool missing its description
|
||||
When I validate the skill definition via the registry
|
||||
Then the skill validation errors should mention "missing a description"
|
||||
|
||||
Scenario: Validate skill referencing unknown included skill
|
||||
Given a skill registry with no tool registry
|
||||
And a skill definition that includes "unknown/skill"
|
||||
When I validate the skill definition via the registry
|
||||
Then the skill validation errors should mention "is not registered"
|
||||
|
||||
Scenario: Validate skill with all valid references
|
||||
Given a skill registry with no tool registry
|
||||
And a skill definition with no tool refs or includes
|
||||
When I validate the skill definition via the registry
|
||||
Then the skill validation errors should be empty
|
||||
|
||||
# =========================================================================
|
||||
# cli/formatting.py — format_output_session, _format_table edge cases
|
||||
# (lines 121, 128, 197, 231-236)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Format output with color format type
|
||||
Given sample formatting data with key "status" and value "ok"
|
||||
When I format the data with format type "color"
|
||||
Then the formatted output should contain "status"
|
||||
|
||||
Scenario: Format output session with list of dicts
|
||||
Given sample formatting data as a list of two items
|
||||
When I format the data via format_output_session with format "plain"
|
||||
Then the formatted session output should not be empty
|
||||
|
||||
Scenario: Format output session with dict input
|
||||
Given sample formatting data with key "name" and value "test"
|
||||
When I format the data via format_output_session with format "json"
|
||||
Then the formatted session output should contain "name"
|
||||
|
||||
Scenario: Format table with rows having extra keys
|
||||
Given sample formatting data as a list with mismatched keys
|
||||
When I format the data with format type "table"
|
||||
Then the formatted output should contain "extra_col"
|
||||
|
||||
Scenario: Format table with empty list
|
||||
Given sample formatting data as an empty list
|
||||
When I format the data with format type "table"
|
||||
Then the formatted output should be "(empty)"
|
||||
|
||||
# =========================================================================
|
||||
# cli/commands/system.py — health check functions
|
||||
# (lines 108-118, 146-147, 174-175, 193-194, 258-278, 325-334)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Check config file when it exists and is readable
|
||||
Given a system health check environment
|
||||
And a config file that exists and is readable
|
||||
When I run the config file health check
|
||||
Then the health check status should be "ok"
|
||||
And the health check details should be "readable"
|
||||
|
||||
Scenario: Check data dir when it does not exist
|
||||
Given a system health check environment
|
||||
And a data dir that does not exist
|
||||
When I run the data dir health check
|
||||
Then the health check status should be "warn"
|
||||
And the health check details should contain "missing"
|
||||
|
||||
Scenario: Check disk space with plenty of space
|
||||
Given a system health check environment
|
||||
When I run the disk space health check
|
||||
Then the health check status should be "ok"
|
||||
|
||||
Scenario: Check git availability
|
||||
Given a system health check environment
|
||||
When I run the git availability health check
|
||||
Then the health check status should be "ok"
|
||||
|
||||
Scenario: Check file permissions on writable data dir
|
||||
Given a system health check environment
|
||||
And a writable data dir
|
||||
When I run the file permissions health check
|
||||
Then the health check status should be "ok"
|
||||
And the health check details should be "data dir r/w"
|
||||
|
||||
Scenario: Check file permissions on nonexistent data dir
|
||||
Given a system health check environment
|
||||
And a data dir that does not exist
|
||||
When I run the file permissions health check
|
||||
Then the health check status should be "warn"
|
||||
And the health check details should contain "does not exist"
|
||||
|
||||
Scenario: Build info data with existing database file
|
||||
Given a system health check environment
|
||||
And a database file that exists
|
||||
When I build info data
|
||||
Then the info data should contain a db_size value
|
||||
|
||||
Scenario: Build info data with log directory
|
||||
Given a system health check environment
|
||||
And a log directory with files
|
||||
When I build info data
|
||||
Then the info data should contain a logs value
|
||||
|
||||
# =========================================================================
|
||||
# cli/commands/session.py — session CLI commands
|
||||
# (lines 56-69, 75, 157-159, 267, 277-278, 320-323, 381-382, 482-485)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Get session service lazily initialises from container
|
||||
Given a session CLI test environment
|
||||
When I call get_session_service
|
||||
Then a session service should be returned
|
||||
|
||||
Scenario: Reset session service clears the cached instance
|
||||
Given a session CLI test environment
|
||||
When I call reset_session_service
|
||||
Then the session service cache should be cleared
|
||||
|
||||
Scenario: Session create with non-rich format outputs formatted data
|
||||
Given a session CLI test environment
|
||||
And a mock session service that returns a created session
|
||||
When I invoke session create with format "json"
|
||||
Then the covboost session output should contain "session_id"
|
||||
|
||||
Scenario: Session create error shows session not found message
|
||||
Given a session CLI test environment
|
||||
And a mock session service that raises SessionNotFoundError on create
|
||||
When I invoke session create expecting an error
|
||||
Then the session CLI should have exited with error
|
||||
|
||||
Scenario: Session list with no sessions shows empty message
|
||||
Given a session CLI test environment
|
||||
And a mock session service that returns no sessions
|
||||
When I invoke session list with format "rich"
|
||||
Then the covboost session output should contain "No sessions found"
|
||||
|
||||
Scenario: Session delete without confirmation aborts
|
||||
Given a session CLI test environment
|
||||
And a mock session service that returns a session
|
||||
When I invoke session delete without confirming
|
||||
Then the session CLI should have been aborted
|
||||
|
||||
Scenario: Session delete with yes flag succeeds
|
||||
Given a session CLI test environment
|
||||
And a mock session service that returns a session
|
||||
When I invoke session delete with yes flag
|
||||
Then the covboost session output should contain "deleted"
|
||||
|
||||
Scenario: Session export to stdout outputs JSON
|
||||
Given a session CLI test environment
|
||||
And a mock session service that can export
|
||||
When I invoke session export to stdout
|
||||
Then the covboost session output should contain "session_id"
|
||||
|
||||
Scenario: Session export to file that exists without force fails
|
||||
Given a session CLI test environment
|
||||
And a mock session service that can export
|
||||
And a temporary output file that already exists
|
||||
When I invoke session export to that file without force
|
||||
Then the session CLI should have exited with error
|
||||
|
||||
# =========================================================================
|
||||
# cli/commands/cleanup.py — error handling paths
|
||||
# (lines 49-58, 94, 129-131)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Active plan detection with OSError returns empty set
|
||||
Given a cleanup CLI test environment
|
||||
And a cleanup service that raises OSError on sandbox scan
|
||||
When I detect active plans
|
||||
Then the active plans set should be empty
|
||||
|
||||
Scenario: Cleanup prune command runs successfully
|
||||
Given a cleanup CLI test environment
|
||||
And a mock cleanup service
|
||||
When I invoke cleanup prune
|
||||
Then the cleanup should have run without error
|
||||
|
||||
# =========================================================================
|
||||
# cli/commands/audit.py — audit service creation
|
||||
# (lines 29-32, 110-111)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Audit service is created from settings
|
||||
Given an audit CLI test environment
|
||||
When I create the audit service
|
||||
Then an audit service instance should be returned
|
||||
|
||||
Scenario: Audit list command with no entries
|
||||
Given an audit CLI test environment
|
||||
And a mock audit service with no entries
|
||||
When I invoke audit list
|
||||
Then the covboost audit output should contain "No audit"
|
||||
|
||||
# =========================================================================
|
||||
# application/container.py — container init paths
|
||||
# (lines 66-69, 125-130)
|
||||
# =========================================================================
|
||||
|
||||
Scenario: Container initialises database from settings
|
||||
Given a container test environment
|
||||
When I get the application container
|
||||
Then the container should provide a database session
|
||||
|
||||
Scenario: Container clears its singleton on reset
|
||||
Given a container test environment
|
||||
When I reset the container singleton
|
||||
Then the container cache should be cleared
|
||||
@@ -0,0 +1,205 @@
|
||||
Feature: Secure template rendering
|
||||
As a security-conscious developer
|
||||
I want templates to be sandboxed and validated
|
||||
So that template injection attacks are prevented
|
||||
|
||||
Background:
|
||||
Given a secure template test environment
|
||||
|
||||
# -- Safe rendering -------------------------------------------------------
|
||||
|
||||
Scenario: Render a simple placeholder
|
||||
Given the secure template text "{greeting} {name}"
|
||||
And a secure template context key "greeting" with value "Hello"
|
||||
And a secure template context key "name" with value "World"
|
||||
When I render the secure template
|
||||
Then the secure template output should be "Hello World"
|
||||
|
||||
Scenario: Render with missing placeholder leaves it intact
|
||||
Given the secure template text "Hello {name}, your role is {role}"
|
||||
And a secure template context key "name" with value "Alice"
|
||||
When I render the secure template
|
||||
Then the secure template output should be "Hello Alice, your role is {role}"
|
||||
|
||||
Scenario: Render with empty context returns template as-is
|
||||
Given the secure template text "No placeholders here"
|
||||
When I render the secure template
|
||||
Then the secure template output should be "No placeholders here"
|
||||
|
||||
Scenario: Render with numeric values converts to string
|
||||
Given the secure template text "Count: {count}"
|
||||
And a secure template context key "count" with value "42"
|
||||
When I render the secure template
|
||||
Then the secure template output should be "Count: 42"
|
||||
|
||||
# -- Security rejection ---------------------------------------------------
|
||||
|
||||
Scenario: Reject attribute access in template
|
||||
Given the secure template text "{obj.__class__.__name__}"
|
||||
When I render the secure template expecting a security error
|
||||
Then the secure template error should mention "Unsafe"
|
||||
|
||||
Scenario: Reject index access in template
|
||||
Given the secure template text "{items[0]}"
|
||||
When I render the secure template expecting a security error
|
||||
Then the secure template error should mention "Unsafe"
|
||||
|
||||
Scenario: Reject format spec in template
|
||||
Given the secure template text "{value:>10}"
|
||||
When I render the secure template expecting a security error
|
||||
Then the secure template error should mention "Unsafe"
|
||||
|
||||
Scenario: Reject conversion flag in template
|
||||
Given the secure template text "{value!r}"
|
||||
When I render the secure template expecting a security error
|
||||
Then the secure template error should mention "Unsafe"
|
||||
|
||||
Scenario: Reject function call in template
|
||||
Given the secure template text "{fn()}"
|
||||
When I render the secure template expecting a security error
|
||||
Then the secure template error should mention "Unsafe"
|
||||
|
||||
Scenario: Reject Jinja2 expression delimiters
|
||||
Given the secure template text "{{ config }}"
|
||||
When I render the secure template expecting a security error
|
||||
Then the secure template error should mention "Unsafe"
|
||||
|
||||
Scenario: Reject Jinja2 block delimiters
|
||||
Given the secure template text "{% for x in items %}x{% endfor %}"
|
||||
When I render the secure template expecting a security error
|
||||
Then the secure template error should mention "Unsafe"
|
||||
|
||||
# -- Size limits -----------------------------------------------------------
|
||||
|
||||
Scenario: Reject template exceeding max length
|
||||
Given a secure template that is 11000 characters long
|
||||
When I render the secure template expecting a size error
|
||||
Then the secure template error should mention "exceeds maximum"
|
||||
|
||||
Scenario: Reject output exceeding max length
|
||||
Given the secure template text "{big}"
|
||||
And a secure template config with max output length of 20
|
||||
And a secure template context key "big" with value "This value is definitely longer than twenty characters"
|
||||
When I render the secure template expecting a size error
|
||||
Then the secure template error should mention "exceeds maximum"
|
||||
|
||||
# -- Allowlist enforcement ------------------------------------------------
|
||||
|
||||
Scenario: Reject template key not in allowlist
|
||||
Given the secure template text "{name} {secret}"
|
||||
And a secure template with allowed keys "name"
|
||||
And a secure template context key "name" with value "Alice"
|
||||
And a secure template context key "secret" with value "hidden"
|
||||
When I render the secure template expecting a validation error
|
||||
Then the secure template error should mention "disallowed"
|
||||
|
||||
Scenario: Accept template key that is in allowlist
|
||||
Given the secure template text "{name}"
|
||||
And a secure template with allowed keys "name"
|
||||
And a secure template context key "name" with value "Alice"
|
||||
When I render the secure template
|
||||
Then the secure template output should be "Alice"
|
||||
|
||||
Scenario: Open allowlist allows any simple key
|
||||
Given the secure template text "{anything}"
|
||||
And the secure template has no key restrictions
|
||||
And a secure template context key "anything" with value "works"
|
||||
When I render the secure template
|
||||
Then the secure template output should be "works"
|
||||
|
||||
# -- Pre-validation -------------------------------------------------------
|
||||
|
||||
Scenario: Validate clean template returns no errors
|
||||
Given the secure template text "{name} {role}"
|
||||
When I validate the secure template
|
||||
Then the secure template validation should be empty
|
||||
|
||||
Scenario: Validate unsafe template returns errors
|
||||
Given the secure template text "{obj.__class__}"
|
||||
When I validate the secure template
|
||||
Then the secure template validation should not be empty
|
||||
And the secure template validation should mention "Unsafe"
|
||||
|
||||
Scenario: Validate template with unknown keys returns errors
|
||||
Given the secure template text "{name} {secret}"
|
||||
And a secure template with allowed keys "name"
|
||||
When I validate the secure template
|
||||
Then the secure template validation should not be empty
|
||||
And the secure template validation should mention "Unknown"
|
||||
|
||||
Scenario: Validate oversized template returns errors
|
||||
Given a secure template that is 11000 characters long
|
||||
When I validate the secure template
|
||||
Then the secure template validation should not be empty
|
||||
And the secure template validation should mention "exceeds"
|
||||
|
||||
# -- Convenience function -------------------------------------------------
|
||||
|
||||
Scenario: render_template_secure convenience function works
|
||||
Given the secure template text "{greeting}"
|
||||
And a secure template context key "greeting" with value "Hi"
|
||||
When I render via the secure convenience function
|
||||
Then the secure template output should be "Hi"
|
||||
|
||||
Scenario: render_template_secure rejects unsafe input
|
||||
Given the secure template text "{obj.attr}"
|
||||
When I render via the secure convenience function expecting an error
|
||||
Then the secure template error should mention "Unsafe"
|
||||
|
||||
# -- Config property access ------------------------------------------------
|
||||
|
||||
Scenario: Renderer exposes active configuration
|
||||
Given a secure template with allowed keys "name"
|
||||
When I create a renderer with those settings
|
||||
Then the renderer config allowed keys should contain "name"
|
||||
|
||||
# -- Validate with matching allowlist (no unknown keys) -------------------
|
||||
|
||||
Scenario: Validate template where all keys match allowlist
|
||||
Given the secure template text "{name}"
|
||||
And a secure template with allowed keys "name"
|
||||
When I validate the secure template
|
||||
Then the secure template validation should be empty
|
||||
|
||||
# -- Rejected keys without logging ----------------------------------------
|
||||
|
||||
Scenario: Reject unknown key without logging when log_rejected is false
|
||||
Given the secure template text "{name} {secret}"
|
||||
And a secure template config with logging disabled
|
||||
And a secure template context key "name" with value "Alice"
|
||||
And a secure template context key "secret" with value "hidden"
|
||||
When I render the secure template expecting a validation error
|
||||
Then the secure template error should mention "disallowed"
|
||||
|
||||
# -- reject_unknown_keys=False path ----------------------------------------
|
||||
|
||||
Scenario: Allow unknown keys when reject_unknown_keys is disabled
|
||||
Given the secure template text "{name} {extra}"
|
||||
And a secure template config that permits unknown keys
|
||||
And a secure template context key "name" with value "Alice"
|
||||
And a secure template context key "extra" with value "data"
|
||||
When I render the secure template
|
||||
Then the secure template output should be "Alice data"
|
||||
|
||||
# -- Legacy renderer with config passthrough --------------------------------
|
||||
|
||||
Scenario: Legacy TemplateRenderer accepts config passthrough
|
||||
Given the secure template text "{name}"
|
||||
And a secure template with allowed keys "name"
|
||||
And a secure template context key "name" with value "Carol"
|
||||
When I render via the legacy TemplateRenderer with config
|
||||
Then the secure template output should be "Carol"
|
||||
|
||||
# -- Legacy renderer backward compat --------------------------------------
|
||||
|
||||
Scenario: Legacy TemplateRenderer returns raw on unsafe template
|
||||
Given the secure template text "{obj.__class__}"
|
||||
And a secure template context key "obj" with value "test"
|
||||
When I render via the legacy TemplateRenderer
|
||||
Then the secure template output should be "{obj.__class__}"
|
||||
|
||||
Scenario: Legacy TemplateRenderer renders safe templates
|
||||
Given the secure template text "{name}"
|
||||
And a secure template context key "name" with value "Bob"
|
||||
When I render via the legacy TemplateRenderer
|
||||
Then the secure template output should be "Bob"
|
||||
@@ -0,0 +1,855 @@
|
||||
"""Step definitions for security_template_coverage_boost.feature.
|
||||
|
||||
Covers uncovered paths in skills/registry.py, cli/formatting.py,
|
||||
cli/commands/system.py, cli/commands/session.py, cli/commands/cleanup.py,
|
||||
cli/commands/audit.py, and application/container.py.
|
||||
|
||||
All step patterns use unique prefixes to avoid AmbiguousStep collisions.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import os
|
||||
import tempfile
|
||||
from io import StringIO
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import typer
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
__all__: list[str] = []
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Helpers
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
def _make_mock_tool_registry(tools: dict[str, Any] | None = None) -> Any:
|
||||
"""Return a mock tool registry that resolves tool names."""
|
||||
reg = MagicMock()
|
||||
tools = tools or {}
|
||||
|
||||
def _get_tool(name: str) -> Any:
|
||||
return tools.get(name)
|
||||
|
||||
reg.get_tool = _get_tool
|
||||
return reg
|
||||
|
||||
|
||||
def _make_skill_definition(
|
||||
name: str = "test/skill",
|
||||
description: str = "A test skill",
|
||||
tool_refs: list[str] | None = None,
|
||||
inline_tools: list[Any] | None = None,
|
||||
includes: list[Any] | None = None,
|
||||
) -> Any:
|
||||
"""Build a minimal SkillDefinition for registry validation."""
|
||||
from cleveragents.domain.models.core.skill import Skill
|
||||
from cleveragents.skills.protocol import SkillDefinition, SkillMetadata
|
||||
|
||||
skill = Skill(
|
||||
name=name,
|
||||
description=description,
|
||||
tool_refs=tool_refs or [],
|
||||
anonymous_tools=inline_tools or [],
|
||||
includes=includes or [],
|
||||
)
|
||||
metadata = SkillMetadata(
|
||||
name=name,
|
||||
description=description,
|
||||
tool_count=len(tool_refs or []) + len(inline_tools or []),
|
||||
)
|
||||
return SkillDefinition(skill=skill, metadata=metadata)
|
||||
|
||||
|
||||
def _mock_settings(**overrides: Any) -> Any:
|
||||
"""Create mock settings with sensible defaults for CLI tests."""
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
s = MagicMock()
|
||||
s.database_url = overrides.get("database_url", f"sqlite:///{tmpdir}/test.db")
|
||||
s.data_dir = overrides.get("data_dir", Path(tmpdir))
|
||||
s.storage_path = overrides.get("storage_path", Path(tmpdir))
|
||||
s.config_path = overrides.get("config_path", Path(tmpdir) / "config.toml")
|
||||
s.log_dir = overrides.get("log_dir", Path(tmpdir) / "logs")
|
||||
s.default_automation_profile = "auto"
|
||||
s.has_provider_configured = MagicMock(return_value=False)
|
||||
s.configured_providers = []
|
||||
s.debug_enabled = False
|
||||
return s
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# skills/registry.py — validate_skill (lines 200-213)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
@given("a skill registry with a mock tool registry")
|
||||
def step_skill_reg_with_tools(context: Context) -> None:
|
||||
from cleveragents.skills.registry import SkillRegistry
|
||||
|
||||
context.cov_tool_registry = _make_mock_tool_registry(tools={})
|
||||
context.cov_skill_registry = SkillRegistry(
|
||||
tool_registry=context.cov_tool_registry,
|
||||
)
|
||||
context.cov_validation_errors = []
|
||||
|
||||
|
||||
@given("a skill registry with no tool registry")
|
||||
def step_skill_reg_no_tools(context: Context) -> None:
|
||||
from cleveragents.skills.registry import SkillRegistry
|
||||
|
||||
context.cov_skill_registry = SkillRegistry(tool_registry=None)
|
||||
context.cov_validation_errors = []
|
||||
|
||||
|
||||
@given('a skill definition with tool ref "{ref}"')
|
||||
def step_skill_def_tool_ref(context: Context, ref: str) -> None:
|
||||
context.cov_skill_def = _make_skill_definition(tool_refs=[ref])
|
||||
|
||||
|
||||
@given("a skill definition with an inline tool missing its description")
|
||||
def step_skill_def_inline_no_desc(context: Context) -> None:
|
||||
from cleveragents.domain.models.core.skill import SkillInlineTool
|
||||
from cleveragents.domain.models.core.tool import ToolSource
|
||||
|
||||
# Use model_construct to bypass pydantic min-length validation
|
||||
inline = SkillInlineTool.model_construct(
|
||||
description="",
|
||||
source=ToolSource.CUSTOM,
|
||||
timeout=300,
|
||||
resource_slots=[],
|
||||
)
|
||||
context.cov_skill_def = _make_skill_definition(inline_tools=[inline])
|
||||
|
||||
|
||||
@given('a skill definition that includes "{include_name}"')
|
||||
def step_skill_def_include(context: Context, include_name: str) -> None:
|
||||
from cleveragents.domain.models.core.skill import SkillInclude
|
||||
|
||||
inc = SkillInclude(name=include_name)
|
||||
context.cov_skill_def = _make_skill_definition(includes=[inc])
|
||||
|
||||
|
||||
@given("a skill definition with no tool refs or includes")
|
||||
def step_skill_def_empty(context: Context) -> None:
|
||||
context.cov_skill_def = _make_skill_definition()
|
||||
|
||||
|
||||
@when("I validate the skill definition via the registry")
|
||||
def step_validate_skill(context: Context) -> None:
|
||||
context.cov_validation_errors = context.cov_skill_registry.validate_skill(
|
||||
context.cov_skill_def,
|
||||
)
|
||||
|
||||
|
||||
@then('the skill validation errors should mention "{text}"')
|
||||
def step_skill_val_mention(context: Context, text: str) -> None:
|
||||
combined = " ".join(context.cov_validation_errors)
|
||||
assert text in combined, (
|
||||
f"Expected '{text}' in validation errors: {context.cov_validation_errors}"
|
||||
)
|
||||
|
||||
|
||||
@then("the skill validation errors should be empty")
|
||||
def step_skill_val_empty(context: Context) -> None:
|
||||
assert context.cov_validation_errors == [], (
|
||||
f"Expected no errors, got: {context.cov_validation_errors}"
|
||||
)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# cli/formatting.py — format_output, format_output_session (lines 121,128,197,231-236)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
@given('sample formatting data with key "{key}" and value "{value}"')
|
||||
def step_fmt_data_dict(context: Context, key: str, value: str) -> None:
|
||||
context.cov_format_data = {key: value}
|
||||
context.cov_format_output = ""
|
||||
context.cov_session_output = ""
|
||||
|
||||
|
||||
@given("sample formatting data as a list of two items")
|
||||
def step_fmt_data_list(context: Context) -> None:
|
||||
context.cov_format_data = [{"a": "1", "b": "2"}, {"a": "3", "b": "4"}]
|
||||
context.cov_format_output = ""
|
||||
context.cov_session_output = ""
|
||||
|
||||
|
||||
@given("sample formatting data as a list with mismatched keys")
|
||||
def step_fmt_data_mismatched(context: Context) -> None:
|
||||
context.cov_format_data = [
|
||||
{"col_a": "val1"},
|
||||
{"col_a": "val2", "extra_col": "val3"},
|
||||
]
|
||||
context.cov_format_output = ""
|
||||
context.cov_session_output = ""
|
||||
|
||||
|
||||
@given("sample formatting data as an empty list")
|
||||
def step_fmt_data_empty(context: Context) -> None:
|
||||
context.cov_format_data = []
|
||||
context.cov_format_output = ""
|
||||
context.cov_session_output = ""
|
||||
|
||||
|
||||
@when('I format the data with format type "{fmt}"')
|
||||
def step_fmt_format(context: Context, fmt: str) -> None:
|
||||
from cleveragents.cli.formatting import format_output
|
||||
|
||||
context.cov_format_output = format_output(context.cov_format_data, fmt)
|
||||
|
||||
|
||||
@when('I format the data via format_output_session with format "{fmt}"')
|
||||
def step_fmt_session(context: Context, fmt: str) -> None:
|
||||
from cleveragents.cli.formatting import format_output_session
|
||||
|
||||
context.cov_session_output = format_output_session(context.cov_format_data, fmt)
|
||||
|
||||
|
||||
@then("the formatted output should contain {text}")
|
||||
def step_fmt_output_contains(context: Context, text: str) -> None:
|
||||
text = text.strip('"')
|
||||
assert text in context.cov_format_output, (
|
||||
f"Expected '{text}' in output:\n{context.cov_format_output}"
|
||||
)
|
||||
|
||||
|
||||
@then('the formatted output should be "{expected}"')
|
||||
def step_fmt_output_exact(context: Context, expected: str) -> None:
|
||||
assert context.cov_format_output.strip() == expected, (
|
||||
f"Expected '{expected}', got: '{context.cov_format_output.strip()}'"
|
||||
)
|
||||
|
||||
|
||||
@then("the formatted session output should not be empty")
|
||||
def step_fmt_session_not_empty(context: Context) -> None:
|
||||
assert len(context.cov_session_output.strip()) > 0
|
||||
|
||||
|
||||
@then('the formatted session output should contain "{text}"')
|
||||
def step_fmt_session_contains(context: Context, text: str) -> None:
|
||||
assert text in context.cov_session_output, (
|
||||
f"Expected '{text}' in session output:\n{context.cov_session_output}"
|
||||
)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# cli/commands/system.py — health check functions (lines 108-118 etc.)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
@given("a system health check environment")
|
||||
def step_sys_health_env(context: Context) -> None:
|
||||
context.cov_health_result = {}
|
||||
context.cov_info_data = {}
|
||||
context.cov_tmpdir = tempfile.mkdtemp()
|
||||
|
||||
|
||||
@given("a config file that exists and is readable")
|
||||
def step_sys_config_exists(context: Context) -> None:
|
||||
config_path = Path(context.cov_tmpdir) / "config.toml"
|
||||
config_path.write_text("[settings]\nkey = 'value'\n")
|
||||
context.cov_config_path = config_path
|
||||
|
||||
|
||||
@given("a data dir that does not exist")
|
||||
def step_sys_data_dir_missing(context: Context) -> None:
|
||||
context.cov_data_dir = Path(context.cov_tmpdir) / "nonexistent_data_dir"
|
||||
|
||||
|
||||
@given("a writable data dir")
|
||||
def step_sys_writable_data_dir(context: Context) -> None:
|
||||
data_dir = Path(context.cov_tmpdir) / "data"
|
||||
data_dir.mkdir(parents=True, exist_ok=True)
|
||||
context.cov_data_dir = data_dir
|
||||
|
||||
|
||||
@given("a database file that exists")
|
||||
def step_sys_db_exists(context: Context) -> None:
|
||||
db_path = Path(context.cov_tmpdir) / "test.db"
|
||||
db_path.write_bytes(b"\x00" * 1024)
|
||||
context.cov_db_path = db_path
|
||||
|
||||
|
||||
@given("a log directory with files")
|
||||
def step_sys_log_dir(context: Context) -> None:
|
||||
log_dir = Path(context.cov_tmpdir) / "logs"
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
(log_dir / "test.log").write_text("log content\n")
|
||||
context.cov_log_dir = log_dir
|
||||
|
||||
|
||||
@when("I run the config file health check")
|
||||
def step_sys_check_config(context: Context) -> None:
|
||||
from cleveragents.cli.commands.system import _check_config_file
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"CLEVERAGENTS_CONFIG_PATH": str(context.cov_config_path)},
|
||||
):
|
||||
context.cov_health_result = _check_config_file()
|
||||
|
||||
|
||||
@when("I run the data dir health check")
|
||||
def step_sys_check_data_dir(context: Context) -> None:
|
||||
from cleveragents.cli.commands.system import _check_data_dir
|
||||
|
||||
ms = _mock_settings(data_dir=context.cov_data_dir)
|
||||
with patch("cleveragents.config.settings.get_settings", return_value=ms):
|
||||
context.cov_health_result = _check_data_dir()
|
||||
|
||||
|
||||
@when("I run the disk space health check")
|
||||
def step_sys_check_disk(context: Context) -> None:
|
||||
from cleveragents.cli.commands.system import _check_disk_space
|
||||
|
||||
context.cov_health_result = _check_disk_space()
|
||||
|
||||
|
||||
@when("I run the git availability health check")
|
||||
def step_sys_check_git(context: Context) -> None:
|
||||
from cleveragents.cli.commands.system import _check_git
|
||||
|
||||
context.cov_health_result = _check_git()
|
||||
|
||||
|
||||
@when("I run the file permissions health check")
|
||||
def step_sys_check_perms(context: Context) -> None:
|
||||
from cleveragents.cli.commands.system import _check_file_permissions
|
||||
|
||||
data_dir = getattr(context, "cov_data_dir", Path(context.cov_tmpdir))
|
||||
ms = _mock_settings(data_dir=data_dir)
|
||||
with patch("cleveragents.config.settings.get_settings", return_value=ms):
|
||||
context.cov_health_result = _check_file_permissions()
|
||||
|
||||
|
||||
@when("I build info data")
|
||||
def step_sys_build_info(context: Context) -> None:
|
||||
from cleveragents.cli.commands.system import build_info_data
|
||||
|
||||
db_path = getattr(context, "cov_db_path", Path(context.cov_tmpdir) / "test.db")
|
||||
log_dir = getattr(context, "cov_log_dir", Path(context.cov_tmpdir) / "logs")
|
||||
|
||||
ms = _mock_settings(
|
||||
database_url=f"sqlite:///{db_path}",
|
||||
log_dir=log_dir,
|
||||
)
|
||||
with patch("cleveragents.config.settings.get_settings", return_value=ms):
|
||||
context.cov_info_data = build_info_data()
|
||||
|
||||
|
||||
@then('the health check status should be "{expected}"')
|
||||
def step_sys_health_status(context: Context, expected: str) -> None:
|
||||
from cleveragents.cli.commands.system import CheckStatus
|
||||
|
||||
status_map = {
|
||||
"ok": CheckStatus.OK,
|
||||
"warn": CheckStatus.WARN,
|
||||
"error": CheckStatus.ERROR,
|
||||
}
|
||||
assert context.cov_health_result["status"] == status_map[expected], (
|
||||
f"Expected '{expected}', got: {context.cov_health_result['status']}"
|
||||
)
|
||||
|
||||
|
||||
@then('the health check details should be "{expected}"')
|
||||
def step_sys_health_details(context: Context, expected: str) -> None:
|
||||
assert context.cov_health_result["details"] == expected, (
|
||||
f"Expected '{expected}', got: '{context.cov_health_result['details']}'"
|
||||
)
|
||||
|
||||
|
||||
@then('the health check details should contain "{text}"')
|
||||
def step_sys_health_details_contains(context: Context, text: str) -> None:
|
||||
details = context.cov_health_result["details"]
|
||||
assert text in details, f"Expected '{text}' in: '{details}'"
|
||||
|
||||
|
||||
@then("the info data should contain a db_size value")
|
||||
def step_sys_info_db_size(context: Context) -> None:
|
||||
assert "storage" in context.cov_info_data
|
||||
assert "db_size" in context.cov_info_data["storage"]
|
||||
assert context.cov_info_data["storage"]["db_size"] != "unknown"
|
||||
|
||||
|
||||
@then("the info data should contain a logs value")
|
||||
def step_sys_info_logs(context: Context) -> None:
|
||||
assert "storage" in context.cov_info_data
|
||||
assert "logs" in context.cov_info_data["storage"]
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# cli/commands/session.py — session CLI (lines 56-69, 75, 157-159, etc.)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
@given("a session CLI test environment")
|
||||
def step_sess_env(context: Context) -> None:
|
||||
context.cov_sess_output = ""
|
||||
context.cov_sess_error = None
|
||||
context.cov_sess_aborted = False
|
||||
context.cov_sess_exit_code = 0
|
||||
context.cov_sess_tmpdir = tempfile.mkdtemp()
|
||||
|
||||
|
||||
@given("a mock session service that returns a created session")
|
||||
def step_sess_mock_create(context: Context) -> None:
|
||||
from datetime import UTC, datetime
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.session_id = "01TEST000000000000000000001"
|
||||
mock_session.actor_name = "test-actor"
|
||||
mock_session.namespace = "default"
|
||||
mock_session.created_at = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
mock_session.updated_at = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
mock_session.message_count = 0
|
||||
mock_session.messages = []
|
||||
mock_session.linked_plan_ids = []
|
||||
|
||||
svc = MagicMock()
|
||||
svc.create.return_value = mock_session
|
||||
context.cov_mock_sess_svc = svc
|
||||
|
||||
|
||||
@given("a mock session service that raises SessionNotFoundError on create")
|
||||
def step_sess_mock_create_err(context: Context) -> None:
|
||||
from cleveragents.domain.models.core.session import SessionNotFoundError
|
||||
|
||||
svc = MagicMock()
|
||||
svc.create.side_effect = SessionNotFoundError("test-id")
|
||||
context.cov_mock_sess_svc = svc
|
||||
|
||||
|
||||
@given("a mock session service that returns no sessions")
|
||||
def step_sess_mock_list_empty(context: Context) -> None:
|
||||
svc = MagicMock()
|
||||
svc.list.return_value = []
|
||||
context.cov_mock_sess_svc = svc
|
||||
|
||||
|
||||
@given("a mock session service that returns a session")
|
||||
def step_sess_mock_get(context: Context) -> None:
|
||||
from datetime import UTC, datetime
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.session_id = "01TEST000000000000000000001"
|
||||
mock_session.actor_name = "test-actor"
|
||||
mock_session.namespace = "default"
|
||||
mock_session.created_at = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
mock_session.updated_at = datetime(2026, 1, 1, tzinfo=UTC)
|
||||
mock_session.message_count = 0
|
||||
mock_session.messages = []
|
||||
mock_session.linked_plan_ids = []
|
||||
|
||||
svc = MagicMock()
|
||||
svc.get.return_value = mock_session
|
||||
svc.delete.return_value = None
|
||||
context.cov_mock_sess_svc = svc
|
||||
|
||||
|
||||
@given("a mock session service that can export")
|
||||
def step_sess_mock_export(context: Context) -> None:
|
||||
svc = MagicMock()
|
||||
svc.export_session.return_value = {
|
||||
"session_id": "01TEST000000000000000000001",
|
||||
"messages": [],
|
||||
}
|
||||
context.cov_mock_sess_svc = svc
|
||||
|
||||
|
||||
@given("a temporary output file that already exists")
|
||||
def step_sess_temp_file(context: Context) -> None:
|
||||
out_path = Path(context.cov_sess_tmpdir) / "existing_export.json"
|
||||
out_path.write_text("{}")
|
||||
context.cov_output_path = out_path
|
||||
|
||||
|
||||
@when("I call get_session_service")
|
||||
def step_sess_get_service(context: Context) -> None:
|
||||
from cleveragents.cli.commands import session as session_mod
|
||||
|
||||
mock_container = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_container.db.return_value = mock_db
|
||||
|
||||
original_service = session_mod._service
|
||||
try:
|
||||
session_mod._service = None
|
||||
with (
|
||||
patch(
|
||||
"cleveragents.application.container.get_container",
|
||||
return_value=mock_container,
|
||||
),
|
||||
patch(
|
||||
"cleveragents.application.services.session_service.PersistentSessionService",
|
||||
) as mock_cls,
|
||||
):
|
||||
mock_cls.return_value = MagicMock()
|
||||
result = session_mod._get_session_service()
|
||||
context.cov_sess_svc_result = result
|
||||
finally:
|
||||
session_mod._service = original_service
|
||||
|
||||
|
||||
@when("I call reset_session_service")
|
||||
def step_sess_reset_service(context: Context) -> None:
|
||||
from cleveragents.cli.commands import session as session_mod
|
||||
|
||||
original = session_mod._service
|
||||
session_mod._service = MagicMock() # set something
|
||||
session_mod._reset_session_service()
|
||||
context.cov_sess_cache_cleared = session_mod._service is None
|
||||
session_mod._service = original # restore
|
||||
|
||||
|
||||
@when('I invoke session create with format "{fmt}"')
|
||||
def step_sess_create_fmt(context: Context, fmt: str) -> None:
|
||||
from cleveragents.cli.commands import session as session_mod
|
||||
|
||||
buf = StringIO()
|
||||
with patch.object(
|
||||
session_mod,
|
||||
"_get_session_service",
|
||||
return_value=context.cov_mock_sess_svc,
|
||||
):
|
||||
try:
|
||||
with patch("typer.echo", side_effect=lambda x: buf.write(str(x))):
|
||||
session_mod.create(fmt=fmt)
|
||||
except SystemExit:
|
||||
pass
|
||||
context.cov_sess_output = buf.getvalue()
|
||||
|
||||
|
||||
@when("I invoke session create expecting an error")
|
||||
def step_sess_create_err(context: Context) -> None:
|
||||
from cleveragents.cli.commands import session as session_mod
|
||||
|
||||
with patch.object(
|
||||
session_mod,
|
||||
"_get_session_service",
|
||||
return_value=context.cov_mock_sess_svc,
|
||||
):
|
||||
try:
|
||||
session_mod.create(fmt="json")
|
||||
except (SystemExit, Exception) as exc:
|
||||
context.cov_sess_error = exc
|
||||
context.cov_sess_exit_code = (
|
||||
getattr(exc, "code", 1) if isinstance(exc, SystemExit) else 1
|
||||
)
|
||||
|
||||
|
||||
@when('I invoke session list with format "{fmt}"')
|
||||
def step_sess_list_fmt(context: Context, fmt: str) -> None:
|
||||
from cleveragents.cli.commands import session as session_mod
|
||||
|
||||
buf = StringIO()
|
||||
with (
|
||||
patch.object(
|
||||
session_mod,
|
||||
"_get_session_service",
|
||||
return_value=context.cov_mock_sess_svc,
|
||||
),
|
||||
patch.object(
|
||||
session_mod.console,
|
||||
"print",
|
||||
side_effect=lambda x="", **kw: buf.write(str(x) + "\n"),
|
||||
),
|
||||
):
|
||||
session_mod.list_sessions(fmt=fmt)
|
||||
context.cov_sess_output = buf.getvalue()
|
||||
|
||||
|
||||
@when("I invoke session delete without confirming")
|
||||
def step_sess_delete_no_confirm(context: Context) -> None:
|
||||
from cleveragents.cli.commands import session as session_mod
|
||||
|
||||
with patch.object(
|
||||
session_mod,
|
||||
"_get_session_service",
|
||||
return_value=context.cov_mock_sess_svc,
|
||||
):
|
||||
try:
|
||||
with patch("typer.confirm", return_value=False):
|
||||
session_mod.delete(
|
||||
session_id="01TEST000000000000000000001",
|
||||
yes=False,
|
||||
)
|
||||
except (typer.Abort, SystemExit):
|
||||
context.cov_sess_aborted = True
|
||||
|
||||
|
||||
@when("I invoke session delete with yes flag")
|
||||
def step_sess_delete_yes(context: Context) -> None:
|
||||
from cleveragents.cli.commands import session as session_mod
|
||||
|
||||
buf_lines: list[str] = []
|
||||
with (
|
||||
patch.object(
|
||||
session_mod,
|
||||
"_get_session_service",
|
||||
return_value=context.cov_mock_sess_svc,
|
||||
),
|
||||
patch.object(
|
||||
session_mod.console,
|
||||
"print",
|
||||
side_effect=lambda x="", **kw: buf_lines.append(str(x)),
|
||||
),
|
||||
):
|
||||
session_mod.delete(
|
||||
session_id="01TEST000000000000000000001",
|
||||
yes=True,
|
||||
)
|
||||
context.cov_sess_output = "\n".join(buf_lines)
|
||||
|
||||
|
||||
@when("I invoke session export to stdout")
|
||||
def step_sess_export_stdout(context: Context) -> None:
|
||||
from cleveragents.cli.commands import session as session_mod
|
||||
|
||||
buf = StringIO()
|
||||
with (
|
||||
patch.object(
|
||||
session_mod,
|
||||
"_get_session_service",
|
||||
return_value=context.cov_mock_sess_svc,
|
||||
),
|
||||
patch("typer.echo", side_effect=lambda x: buf.write(str(x))),
|
||||
):
|
||||
session_mod.export_session(
|
||||
session_id="01TEST000000000000000000001",
|
||||
output=None,
|
||||
force=False,
|
||||
)
|
||||
context.cov_sess_output = buf.getvalue()
|
||||
|
||||
|
||||
@when("I invoke session export to that file without force")
|
||||
def step_sess_export_no_force(context: Context) -> None:
|
||||
from cleveragents.cli.commands import session as session_mod
|
||||
|
||||
with patch.object(
|
||||
session_mod,
|
||||
"_get_session_service",
|
||||
return_value=context.cov_mock_sess_svc,
|
||||
):
|
||||
try:
|
||||
session_mod.export_session(
|
||||
session_id="01TEST000000000000000000001",
|
||||
output=context.cov_output_path,
|
||||
force=False,
|
||||
)
|
||||
except (SystemExit, Exception) as exc:
|
||||
context.cov_sess_exit_code = getattr(exc, "code", 1)
|
||||
context.cov_sess_error = exc
|
||||
|
||||
|
||||
@then("a session service should be returned")
|
||||
def step_sess_svc_returned(context: Context) -> None:
|
||||
assert context.cov_sess_svc_result is not None
|
||||
|
||||
|
||||
@then("the session service cache should be cleared")
|
||||
def step_sess_cache_cleared(context: Context) -> None:
|
||||
assert context.cov_sess_cache_cleared is True
|
||||
|
||||
|
||||
@then('the covboost session output should contain "{text}"')
|
||||
def step_sess_output_has(context: Context, text: str) -> None:
|
||||
assert text in context.cov_sess_output, (
|
||||
f"Expected '{text}' in:\n{context.cov_sess_output}"
|
||||
)
|
||||
|
||||
|
||||
@then("the session CLI should have exited with error")
|
||||
def step_sess_exit_err(context: Context) -> None:
|
||||
assert context.cov_sess_error is not None or context.cov_sess_exit_code != 0, (
|
||||
"Expected an error exit"
|
||||
)
|
||||
|
||||
|
||||
@then("the session CLI should have been aborted")
|
||||
def step_sess_aborted(context: Context) -> None:
|
||||
assert context.cov_sess_aborted is True, "Expected abort"
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# cli/commands/cleanup.py — error handling (lines 49-58, 94)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
@given("a cleanup CLI test environment")
|
||||
def step_clean_env(context: Context) -> None:
|
||||
context.cov_clean_result = None
|
||||
context.cov_active_plans = frozenset()
|
||||
|
||||
|
||||
@given("a cleanup service that raises OSError on sandbox scan")
|
||||
def step_clean_oserror(context: Context) -> None:
|
||||
svc = MagicMock()
|
||||
svc._get_sandbox_dirs.side_effect = OSError("Permission denied")
|
||||
context.cov_clean_svc = svc
|
||||
|
||||
|
||||
@given("a mock cleanup service")
|
||||
def step_clean_mock_svc(context: Context) -> None:
|
||||
svc = MagicMock()
|
||||
svc._get_sandbox_dirs.return_value = []
|
||||
svc.extract_plan_id_from_sandbox.return_value = None
|
||||
|
||||
report = MagicMock()
|
||||
report.stale_items = []
|
||||
report.purged_items = []
|
||||
report.total_bytes_freed = 0
|
||||
svc.scan.return_value = report
|
||||
svc.purge.return_value = report
|
||||
context.cov_clean_svc = svc
|
||||
|
||||
|
||||
@when("I detect active plans")
|
||||
def step_clean_detect(context: Context) -> None:
|
||||
from cleveragents.cli.commands.cleanup import _detect_active_plan_ids
|
||||
|
||||
context.cov_active_plans = _detect_active_plan_ids(context.cov_clean_svc)
|
||||
|
||||
|
||||
@when("I invoke cleanup prune")
|
||||
def step_clean_prune(context: Context) -> None:
|
||||
from cleveragents.cli.commands.cleanup import scan
|
||||
|
||||
with (
|
||||
patch(
|
||||
"cleveragents.cli.commands.cleanup._get_cleanup_service",
|
||||
return_value=context.cov_clean_svc,
|
||||
),
|
||||
contextlib.suppress(SystemExit),
|
||||
):
|
||||
scan()
|
||||
context.cov_clean_result = "ok"
|
||||
|
||||
|
||||
@then("the active plans set should be empty")
|
||||
def step_clean_empty(context: Context) -> None:
|
||||
assert context.cov_active_plans == frozenset(), (
|
||||
f"Expected empty, got: {context.cov_active_plans}"
|
||||
)
|
||||
|
||||
|
||||
@then("the cleanup should have run without error")
|
||||
def step_clean_ok(context: Context) -> None:
|
||||
assert context.cov_clean_result == "ok"
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# cli/commands/audit.py — audit service creation (lines 29-32, 110-111)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
@given("an audit CLI test environment")
|
||||
def step_audit_env(context: Context) -> None:
|
||||
context.cov_audit_svc = None
|
||||
context.cov_audit_output = ""
|
||||
|
||||
|
||||
@given("a mock audit service with no entries")
|
||||
def step_audit_mock_empty(context: Context) -> None:
|
||||
svc = MagicMock()
|
||||
svc.list_entries.return_value = []
|
||||
svc.__enter__ = MagicMock(return_value=svc)
|
||||
svc.__exit__ = MagicMock(return_value=False)
|
||||
context.cov_mock_audit_svc = svc
|
||||
|
||||
|
||||
@when("I create the audit service")
|
||||
def step_audit_create(context: Context) -> None:
|
||||
from cleveragents.cli.commands.audit import _get_audit_service
|
||||
|
||||
ms = _mock_settings()
|
||||
with (
|
||||
patch("cleveragents.config.settings.get_settings", return_value=ms),
|
||||
patch(
|
||||
"cleveragents.application.services.audit_service.AuditService.__init__",
|
||||
return_value=None,
|
||||
),
|
||||
):
|
||||
context.cov_audit_svc = _get_audit_service()
|
||||
|
||||
|
||||
@when("I invoke audit list")
|
||||
def step_audit_list(context: Context) -> None:
|
||||
from cleveragents.cli.commands.audit import list_entries
|
||||
|
||||
buf = StringIO()
|
||||
console_mock = MagicMock()
|
||||
console_mock.print = lambda x="", **kw: buf.write(str(x) + "\n")
|
||||
|
||||
with (
|
||||
patch(
|
||||
"cleveragents.cli.commands.audit._get_audit_service",
|
||||
return_value=context.cov_mock_audit_svc,
|
||||
),
|
||||
patch(
|
||||
"cleveragents.cli.commands.audit.get_console",
|
||||
return_value=console_mock,
|
||||
),
|
||||
):
|
||||
list_entries()
|
||||
context.cov_audit_output = buf.getvalue()
|
||||
|
||||
|
||||
@then("an audit service instance should be returned")
|
||||
def step_audit_returned(context: Context) -> None:
|
||||
assert context.cov_audit_svc is not None
|
||||
|
||||
|
||||
@then('the covboost audit output should contain "{text}"')
|
||||
def step_audit_output_has(context: Context, text: str) -> None:
|
||||
assert text in context.cov_audit_output, (
|
||||
f"Expected '{text}' in: {context.cov_audit_output}"
|
||||
)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# application/container.py — container init (lines 66-69, 125-130)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
@given("a container test environment")
|
||||
def step_ctr_env(context: Context) -> None:
|
||||
context.cov_container = None
|
||||
context.cov_ctr_cleared = False
|
||||
|
||||
|
||||
@when("I get the application container")
|
||||
def step_ctr_get(context: Context) -> None:
|
||||
from cleveragents.application.container import get_container, reset_container
|
||||
|
||||
reset_container()
|
||||
context.cov_container = get_container()
|
||||
|
||||
|
||||
@when("I reset the container singleton")
|
||||
def step_ctr_reset(context: Context) -> None:
|
||||
import cleveragents.application.container as container_mod
|
||||
from cleveragents.application.container import get_container, reset_container
|
||||
|
||||
_ = get_container()
|
||||
reset_container()
|
||||
context.cov_ctr_cleared = container_mod._container is None
|
||||
|
||||
|
||||
@then("the container should provide a database session")
|
||||
def step_ctr_has_db(context: Context) -> None:
|
||||
assert context.cov_container is not None
|
||||
# Container has a 'db' provider attribute
|
||||
assert hasattr(context.cov_container, "database_url")
|
||||
|
||||
|
||||
@then("the container cache should be cleared")
|
||||
def step_ctr_cleared(context: Context) -> None:
|
||||
assert context.cov_ctr_cleared is True
|
||||
@@ -0,0 +1,243 @@
|
||||
"""Step definitions for secure template rendering tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
from cleveragents.templates.renderer import TemplateRenderer
|
||||
from cleveragents.templates.secure_renderer import (
|
||||
SecureTemplateRenderer,
|
||||
TemplateConfig,
|
||||
TemplateError,
|
||||
TemplateSecurityError,
|
||||
TemplateSizeError,
|
||||
TemplateValidationError,
|
||||
render_template_secure,
|
||||
validate_template,
|
||||
)
|
||||
|
||||
__all__: list[str] = []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Background
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a secure template test environment")
|
||||
def step_given_secure_template_env(context: Context) -> None:
|
||||
context.sec_template = ""
|
||||
context.sec_context = {} # dict[str, str]
|
||||
context.sec_rendered = ""
|
||||
context.sec_error = None # TemplateError | None
|
||||
context.sec_validation_errors = [] # list[str]
|
||||
context.sec_config = None # TemplateConfig | None
|
||||
context.sec_allowed_keys = None # frozenset[str] | None
|
||||
context.sec_renderer = None # SecureTemplateRenderer | None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Given steps — template text
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given('the secure template text "{template}"')
|
||||
def step_given_template(context: Context, template: str) -> None:
|
||||
context.sec_template = template
|
||||
|
||||
|
||||
@given('a secure template context key "{key}" with value "{value}"')
|
||||
def step_given_context_key(context: Context, key: str, value: str) -> None:
|
||||
context.sec_context[key] = value
|
||||
|
||||
|
||||
@given('a secure template with allowed keys "{keys}"')
|
||||
def step_given_allowed_keys(context: Context, keys: str) -> None:
|
||||
context.sec_allowed_keys = frozenset(k.strip() for k in keys.split(","))
|
||||
|
||||
|
||||
@given("the secure template has no key restrictions")
|
||||
def step_given_template_open(context: Context) -> None:
|
||||
context.sec_allowed_keys = None
|
||||
|
||||
|
||||
@given("a secure template that is {n} characters long")
|
||||
def step_given_long_template(context: Context, n: str) -> None:
|
||||
context.sec_template = "x" * int(n)
|
||||
|
||||
|
||||
@given("a secure template config with max output length of {n}")
|
||||
def step_given_max_output_config(context: Context, n: str) -> None:
|
||||
context.sec_config = TemplateConfig(max_output_length=int(n))
|
||||
|
||||
|
||||
@given("a secure template config with logging disabled")
|
||||
def step_given_config_logging_disabled(context: Context) -> None:
|
||||
context.sec_config = TemplateConfig(
|
||||
allowed_keys=context.sec_allowed_keys or frozenset({"name"}),
|
||||
log_rejected=False,
|
||||
)
|
||||
|
||||
|
||||
@given("a secure template config that permits unknown keys")
|
||||
def step_given_config_permit_unknown(context: Context) -> None:
|
||||
context.sec_config = TemplateConfig(
|
||||
allowed_keys=frozenset({"name"}),
|
||||
reject_unknown_keys=False,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# When steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I render the secure template")
|
||||
def step_when_render_securely(context: Context) -> None:
|
||||
config = context.sec_config
|
||||
if config is None and context.sec_allowed_keys is not None:
|
||||
config = TemplateConfig(allowed_keys=context.sec_allowed_keys)
|
||||
renderer = SecureTemplateRenderer(config=config)
|
||||
context.sec_rendered = renderer.render(context.sec_template, context.sec_context)
|
||||
|
||||
|
||||
@when("I render the secure template expecting a security error")
|
||||
def step_when_render_security_error(context: Context) -> None:
|
||||
config = context.sec_config or TemplateConfig()
|
||||
renderer = SecureTemplateRenderer(config=config)
|
||||
try:
|
||||
renderer.render(context.sec_template, context.sec_context)
|
||||
context.sec_error = None
|
||||
except TemplateError as exc:
|
||||
assert isinstance(exc, TemplateSecurityError), (
|
||||
f"Expected TemplateSecurityError, got {type(exc).__name__}: {exc}"
|
||||
)
|
||||
context.sec_error = exc
|
||||
|
||||
|
||||
@when("I render the secure template expecting a size error")
|
||||
def step_when_render_size_error(context: Context) -> None:
|
||||
config = context.sec_config or TemplateConfig()
|
||||
renderer = SecureTemplateRenderer(config=config)
|
||||
try:
|
||||
renderer.render(context.sec_template, context.sec_context)
|
||||
context.sec_error = None
|
||||
except TemplateError as exc:
|
||||
assert isinstance(exc, TemplateSizeError), (
|
||||
f"Expected TemplateSizeError, got {type(exc).__name__}: {exc}"
|
||||
)
|
||||
context.sec_error = exc
|
||||
|
||||
|
||||
@when("I render the secure template expecting a validation error")
|
||||
def step_when_render_validation_error(context: Context) -> None:
|
||||
config = context.sec_config
|
||||
if config is None and context.sec_allowed_keys is not None:
|
||||
config = TemplateConfig(allowed_keys=context.sec_allowed_keys)
|
||||
elif config is None:
|
||||
config = TemplateConfig()
|
||||
renderer = SecureTemplateRenderer(config=config)
|
||||
try:
|
||||
renderer.render(context.sec_template, context.sec_context)
|
||||
context.sec_error = None
|
||||
except TemplateError as exc:
|
||||
assert isinstance(exc, TemplateValidationError), (
|
||||
f"Expected TemplateValidationError, got {type(exc).__name__}: {exc}"
|
||||
)
|
||||
context.sec_error = exc
|
||||
|
||||
|
||||
@when("I validate the secure template")
|
||||
def step_when_validate_template(context: Context) -> None:
|
||||
allowed = context.sec_allowed_keys or frozenset()
|
||||
context.sec_validation_errors = validate_template(
|
||||
context.sec_template, allowed_keys=allowed
|
||||
)
|
||||
|
||||
|
||||
@when("I render via the secure convenience function")
|
||||
def step_when_render_convenience(context: Context) -> None:
|
||||
context.sec_rendered = render_template_secure(
|
||||
context.sec_template, context.sec_context
|
||||
)
|
||||
|
||||
|
||||
@when("I render via the secure convenience function expecting an error")
|
||||
def step_when_render_convenience_error(context: Context) -> None:
|
||||
try:
|
||||
render_template_secure(context.sec_template, context.sec_context)
|
||||
context.sec_error = None
|
||||
except TemplateError as exc:
|
||||
context.sec_error = exc
|
||||
|
||||
|
||||
@when("I create a renderer with those settings")
|
||||
def step_when_create_renderer(context: Context) -> None:
|
||||
config = context.sec_config
|
||||
if config is None and context.sec_allowed_keys is not None:
|
||||
config = TemplateConfig(allowed_keys=context.sec_allowed_keys)
|
||||
context.sec_renderer = SecureTemplateRenderer(config=config)
|
||||
|
||||
|
||||
@when("I render via the legacy TemplateRenderer")
|
||||
def step_when_render_legacy(context: Context) -> None:
|
||||
renderer = TemplateRenderer()
|
||||
context.sec_rendered = renderer.render(context.sec_template, context.sec_context)
|
||||
|
||||
|
||||
@when("I render via the legacy TemplateRenderer with config")
|
||||
def step_when_render_legacy_with_config(context: Context) -> None:
|
||||
config = context.sec_config
|
||||
if config is None and context.sec_allowed_keys is not None:
|
||||
config = TemplateConfig(allowed_keys=context.sec_allowed_keys)
|
||||
renderer = TemplateRenderer(config=config)
|
||||
context.sec_rendered = renderer.render(context.sec_template, context.sec_context)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Then steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then('the secure template output should be "{expected}"')
|
||||
def step_then_rendered_output(context: Context, expected: str) -> None:
|
||||
assert context.sec_rendered == expected, (
|
||||
f"Expected {expected!r}, got {context.sec_rendered!r}"
|
||||
)
|
||||
|
||||
|
||||
@then('the secure template error should mention "{text}"')
|
||||
def step_then_error_mentions(context: Context, text: str) -> None:
|
||||
assert context.sec_error is not None, "Expected a template error"
|
||||
assert text in str(context.sec_error), (
|
||||
f"Expected '{text}' in error: {context.sec_error}"
|
||||
)
|
||||
|
||||
|
||||
@then("the secure template validation should be empty")
|
||||
def step_then_validation_empty(context: Context) -> None:
|
||||
assert context.sec_validation_errors == [], (
|
||||
f"Expected no errors, got: {context.sec_validation_errors}"
|
||||
)
|
||||
|
||||
|
||||
@then("the secure template validation should not be empty")
|
||||
def step_then_validation_not_empty(context: Context) -> None:
|
||||
assert len(context.sec_validation_errors) > 0, (
|
||||
"Expected validation errors but got none"
|
||||
)
|
||||
|
||||
|
||||
@then('the secure template validation should mention "{text}"')
|
||||
def step_then_validation_mentions(context: Context, text: str) -> None:
|
||||
combined = " ".join(context.sec_validation_errors)
|
||||
assert text in combined, f"Expected '{text}' in: {context.sec_validation_errors}"
|
||||
|
||||
|
||||
@then('the renderer config allowed keys should contain "{key}"')
|
||||
def step_then_config_has_key(context: Context, key: str) -> None:
|
||||
cfg = context.sec_renderer.config # exercises the .config property (line 155)
|
||||
assert key in cfg.allowed_keys, (
|
||||
f"Expected '{key}' in allowed_keys, got: {cfg.allowed_keys}"
|
||||
)
|
||||
+19
-19
@@ -5374,25 +5374,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-eval` to `master` with description "Remove eval-based config parsing and add security checks.". (Code review by Brent still pending)
|
||||
|
||||
**Parallel Group SEC2: Template Injection Prevention [Luis]**
|
||||
- [ ] **COMMIT (Owner: Luis | Group: SEC2.template | Branch: feature/m4-security-template | Planned: Day 11 | Expected: Day 26) - Commit message: "fix(security): harden template rendering"**
|
||||
- [ ] Git [Luis]: `git checkout master`
|
||||
- [ ] Git [Luis]: `git pull origin master`
|
||||
- [ ] Git [Luis]: `git checkout -b feature/m4-security-template`
|
||||
- [ ] Git [Luis]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
|
||||
- [ ] Code [Luis]: Replace unsafe template usage with a sandboxed renderer and strict token set.
|
||||
- [ ] Code [Luis]: Deny attribute access, function calls, and filters; allow only `{var}` substitution with a fixed allowlist.
|
||||
- [ ] Code [Luis]: Add max template length + max render output size checks with explicit errors.
|
||||
- [ ] Code [Luis]: Add template allowlist validation for each template context key and log rejected keys.
|
||||
- [ ] Code [Luis]: Add unit helper to pre-validate template strings at action/plan creation time.
|
||||
- [ ] Docs [Luis]: Add `docs/reference/template_security.md` with safe patterns.
|
||||
- [ ] Tests (Behave) [Luis]: Add `features/security_templates.feature` scenarios.
|
||||
- [ ] Tests (Robot) [Luis]: Add `robot/security_templates.robot` smoke tests.
|
||||
- [ ] Tests (ASV) [Luis]: Add `benchmarks/security_template_bench.py` for render 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): harden template rendering"`
|
||||
- [ ] Git [Luis]: `git push -u origin feature/m4-security-template`
|
||||
- [X] **COMMIT (Owner: Luis | Group: SEC2.template | Branch: feature/m4-security-template | Planned: Day 11 | Expected: Day 26) - Commit message: "fix(security): harden template rendering"**
|
||||
- [X] Git [Luis]: `git checkout master`
|
||||
- [X] Git [Luis]: `git pull origin master`
|
||||
- [X] Git [Luis]: `git checkout -b feature/m4-security-template`
|
||||
- [X] Git [Luis]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
|
||||
- [X] Code [Luis]: Replace unsafe template usage with a sandboxed renderer and strict token set.
|
||||
- [X] Code [Luis]: Deny attribute access, function calls, and filters; allow only `{var}` substitution with a fixed allowlist.
|
||||
- [X] Code [Luis]: Add max template length + max render output size checks with explicit errors.
|
||||
- [X] Code [Luis]: Add template allowlist validation for each template context key and log rejected keys.
|
||||
- [X] Code [Luis]: Add unit helper to pre-validate template strings at action/plan creation time.
|
||||
- [X] Docs [Luis]: Add `docs/reference/template_security.md` with safe patterns.
|
||||
- [X] Tests (Behave) [Luis]: Add `features/security_templates.feature` scenarios.
|
||||
- [X] Tests (Robot) [Luis]: Add `robot/security_templates.robot` smoke tests.
|
||||
- [X] Tests (ASV) [Luis]: Add `benchmarks/security_template_bench.py` for render 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): harden template rendering"`
|
||||
- [X] Git [Luis]: `git push -u origin feature/m4-security-template`
|
||||
- [ ] 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]**
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
"""Robot Framework helper for secure template rendering.
|
||||
|
||||
Provides a CLI-style interface for Robot to invoke the secure
|
||||
template renderer. Exit code 0 = success, 1 = failure.
|
||||
|
||||
Usage::
|
||||
|
||||
python robot/helper_security_templates.py render \\
|
||||
<template> <key1=val1> <key2=val2> ...
|
||||
python robot/helper_security_templates.py validate <template>
|
||||
python robot/helper_security_templates.py reject <template>
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
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.templates.secure_renderer import ( # noqa: E402
|
||||
SecureTemplateRenderer,
|
||||
TemplateConfig,
|
||||
TemplateError,
|
||||
validate_template,
|
||||
)
|
||||
|
||||
|
||||
def _cmd_render(args: list[str]) -> int:
|
||||
"""Render a template with key=value context pairs."""
|
||||
if len(args) < 1:
|
||||
print("Usage: render <template> [key=value ...]")
|
||||
return 1
|
||||
template = args[0]
|
||||
context: dict[str, str] = {}
|
||||
for pair in args[1:]:
|
||||
if "=" in pair:
|
||||
k, v = pair.split("=", 1)
|
||||
context[k] = v
|
||||
renderer = SecureTemplateRenderer(config=TemplateConfig())
|
||||
try:
|
||||
result = renderer.render(template, context)
|
||||
print(f"sec-render-ok: {result}")
|
||||
return 0
|
||||
except TemplateError as exc:
|
||||
print(f"sec-render-err: {exc}")
|
||||
return 1
|
||||
|
||||
|
||||
def _cmd_validate(args: list[str]) -> int:
|
||||
"""Validate a template for security issues."""
|
||||
if len(args) < 1:
|
||||
print("Usage: validate <template>")
|
||||
return 1
|
||||
template = args[0]
|
||||
errors = validate_template(template)
|
||||
if errors:
|
||||
for err in errors:
|
||||
print(f"sec-validate-err: {err}")
|
||||
return 1
|
||||
print("sec-validate-ok: clean")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_reject(args: list[str]) -> int:
|
||||
"""Verify that a template is rejected as unsafe."""
|
||||
if len(args) < 1:
|
||||
print("Usage: reject <template>")
|
||||
return 1
|
||||
template = args[0]
|
||||
renderer = SecureTemplateRenderer(config=TemplateConfig())
|
||||
try:
|
||||
renderer.render(template, {})
|
||||
print("sec-reject-fail: template was accepted")
|
||||
return 1
|
||||
except TemplateError as exc:
|
||||
print(f"sec-reject-ok: {exc}")
|
||||
return 0
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Entry point called by Robot Framework ``Run Process``."""
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: helper_security_templates.py <command> ...")
|
||||
return 1
|
||||
|
||||
command = sys.argv[1]
|
||||
args = sys.argv[2:]
|
||||
|
||||
dispatch = {
|
||||
"render": _cmd_render,
|
||||
"validate": _cmd_validate,
|
||||
"reject": _cmd_reject,
|
||||
}
|
||||
|
||||
handler = dispatch.get(command)
|
||||
if handler is None:
|
||||
print(f"Unknown command: {command}")
|
||||
return 1
|
||||
|
||||
return handler(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,82 @@
|
||||
*** Settings ***
|
||||
Documentation Integration tests for secure template rendering
|
||||
Resource common.resource
|
||||
Suite Setup Setup Test Environment
|
||||
Suite Teardown Cleanup Test Environment
|
||||
|
||||
*** Test Cases ***
|
||||
Safe Template Renders Correctly
|
||||
[Documentation] Verify a safe template renders with correct output
|
||||
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_templates.py
|
||||
... render Hello {name} name\=World
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} sec-render-ok: Hello World
|
||||
|
||||
Unsafe Template Is Rejected
|
||||
[Documentation] Verify attribute access template is rejected
|
||||
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_templates.py
|
||||
... reject {obj.__class__}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} sec-reject-ok
|
||||
|
||||
Template Validation Passes Clean Template
|
||||
[Documentation] Verify validation succeeds on a clean template
|
||||
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_templates.py
|
||||
... validate {name} is {role}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} sec-validate-ok
|
||||
|
||||
Template Validation Fails Unsafe Template
|
||||
[Documentation] Verify validation rejects unsafe template
|
||||
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_templates.py
|
||||
... validate {obj.__class__}
|
||||
Should Be Equal As Integers ${result.rc} 1
|
||||
Should Contain ${result.stdout} sec-validate-err
|
||||
|
||||
Render Rejects Index Access
|
||||
[Documentation] Verify index access in template is rejected
|
||||
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_templates.py
|
||||
... reject {items[0]}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} sec-reject-ok
|
||||
|
||||
Reject Command Fails For Safe Template
|
||||
[Documentation] Verify reject command reports failure for safe template
|
||||
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_templates.py
|
||||
... reject {name}
|
||||
Should Be Equal As Integers ${result.rc} 1
|
||||
Should Contain ${result.stdout} sec-reject-fail
|
||||
|
||||
Helper Shows Usage On No Command
|
||||
[Documentation] Verify helper prints usage when no command given
|
||||
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_templates.py
|
||||
Should Be Equal As Integers ${result.rc} 1
|
||||
Should Contain ${result.stdout} Usage
|
||||
|
||||
Helper Rejects Unknown Command
|
||||
[Documentation] Verify helper rejects an unknown sub-command
|
||||
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_templates.py
|
||||
... bogus
|
||||
Should Be Equal As Integers ${result.rc} 1
|
||||
Should Contain ${result.stdout} Unknown command
|
||||
|
||||
Render Shows Usage On No Args
|
||||
[Documentation] Verify render sub-command prints usage with no arguments
|
||||
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_templates.py
|
||||
... render
|
||||
Should Be Equal As Integers ${result.rc} 1
|
||||
Should Contain ${result.stdout} Usage
|
||||
|
||||
Validate Shows Usage On No Args
|
||||
[Documentation] Verify validate sub-command prints usage with no arguments
|
||||
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_templates.py
|
||||
... validate
|
||||
Should Be Equal As Integers ${result.rc} 1
|
||||
Should Contain ${result.stdout} Usage
|
||||
|
||||
Reject Shows Usage On No Args
|
||||
[Documentation] Verify reject sub-command prints usage with no arguments
|
||||
${result}= Run Process ${PYTHON} ${WORKSPACE}/robot/helper_security_templates.py
|
||||
... reject
|
||||
Should Be Equal As Integers ${result.rc} 1
|
||||
Should Contain ${result.stdout} Usage
|
||||
@@ -0,0 +1,28 @@
|
||||
"""Template rendering with security controls.
|
||||
|
||||
Re-exports the secure renderer for convenience.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from cleveragents.templates.secure_renderer import (
|
||||
SecureTemplateRenderer,
|
||||
TemplateConfig,
|
||||
TemplateError,
|
||||
TemplateSecurityError,
|
||||
TemplateSizeError,
|
||||
TemplateValidationError,
|
||||
render_template_secure,
|
||||
validate_template,
|
||||
)
|
||||
|
||||
__all__: list[str] = [
|
||||
"SecureTemplateRenderer",
|
||||
"TemplateConfig",
|
||||
"TemplateError",
|
||||
"TemplateSecurityError",
|
||||
"TemplateSizeError",
|
||||
"TemplateValidationError",
|
||||
"render_template_secure",
|
||||
"validate_template",
|
||||
]
|
||||
@@ -1,27 +1,61 @@
|
||||
"""Minimal template renderer shim for reactive routing.
|
||||
|
||||
This is a placeholder to satisfy imports from the reactive stack. It provides a
|
||||
basic render pass-through suitable for tests; replace with real templating if
|
||||
needed.
|
||||
Wraps :class:`~cleveragents.templates.secure_renderer.SecureTemplateRenderer`
|
||||
to satisfy the existing import contract while enforcing security controls.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from enum import Enum
|
||||
from typing import Any
|
||||
|
||||
from cleveragents.templates.secure_renderer import (
|
||||
SecureTemplateRenderer,
|
||||
TemplateConfig,
|
||||
TemplateError,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TemplateEngine(Enum):
|
||||
"""Supported template engines."""
|
||||
|
||||
SIMPLE = "simple"
|
||||
|
||||
|
||||
class TemplateRenderer:
|
||||
def __init__(self, engine: TemplateEngine = TemplateEngine.SIMPLE):
|
||||
"""Thin wrapper over :class:`SecureTemplateRenderer`.
|
||||
|
||||
The ``render`` method silently returns the raw template on
|
||||
any rendering failure to preserve backward compatibility.
|
||||
|
||||
Args:
|
||||
engine: Template engine selector (only ``SIMPLE`` is supported).
|
||||
config: Optional :class:`TemplateConfig` forwarded to the
|
||||
underlying :class:`SecureTemplateRenderer`. When ``None``
|
||||
the default (open-allowlist) config is used.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
engine: TemplateEngine = TemplateEngine.SIMPLE,
|
||||
config: TemplateConfig | None = None,
|
||||
) -> None:
|
||||
self.engine = engine
|
||||
self._secure = SecureTemplateRenderer(config=config or TemplateConfig())
|
||||
|
||||
def render(self, template: str, context: dict[str, Any] | None = None) -> str:
|
||||
"""Render *template* with *context*, returning raw on error."""
|
||||
context = context or {}
|
||||
try:
|
||||
return template.format(**context)
|
||||
except Exception:
|
||||
return self._secure.render(template, context)
|
||||
except TemplateError:
|
||||
logger.warning(
|
||||
"Template rendering failed; returning raw template "
|
||||
"(length=%d). Enable DEBUG logging for details.",
|
||||
len(template),
|
||||
)
|
||||
logger.debug("Suppressed TemplateError for template: %r", template)
|
||||
return template
|
||||
|
||||
@@ -0,0 +1,346 @@
|
||||
"""Sandboxed template renderer with strict security controls.
|
||||
|
||||
Replaces unsafe ``str.format()`` with a restricted token set that only
|
||||
allows ``{variable_name}`` substitution from a validated allowlist.
|
||||
|
||||
Security properties:
|
||||
|
||||
- **No attribute access**: ``{obj.attr}`` is rejected.
|
||||
- **No function calls**: ``{fn()}`` is rejected.
|
||||
- **No format specs**: ``{val:>10}`` is rejected.
|
||||
- **No index access**: ``{val[0]}`` is rejected.
|
||||
- **Fixed allowlist**: Only context keys present in the allowlist can
|
||||
appear in templates.
|
||||
- **Length limits**: Both the input template and rendered output are
|
||||
bounded to prevent denial-of-service attacks.
|
||||
|
||||
Usage::
|
||||
|
||||
renderer = SecureTemplateRenderer(
|
||||
config=TemplateConfig(
|
||||
allowed_keys=frozenset({"name", "description"}),
|
||||
),
|
||||
)
|
||||
result = renderer.render("Hello {name}", {"name": "world"})
|
||||
|
||||
Or use the convenience function::
|
||||
|
||||
result = render_template_secure(
|
||||
"Hello {name}",
|
||||
{"name": "world"},
|
||||
allowed_keys=frozenset({"name"}),
|
||||
)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
__all__: list[str] = [
|
||||
"SecureTemplateRenderer",
|
||||
"TemplateConfig",
|
||||
"TemplateError",
|
||||
"TemplateSecurityError",
|
||||
"TemplateSizeError",
|
||||
"TemplateValidationError",
|
||||
"render_template_secure",
|
||||
"validate_template",
|
||||
]
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Constants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
#: Default maximum template length in characters.
|
||||
DEFAULT_MAX_TEMPLATE_LENGTH: int = 10_000
|
||||
|
||||
#: Default maximum rendered output length in characters.
|
||||
DEFAULT_MAX_OUTPUT_LENGTH: int = 50_000
|
||||
|
||||
#: Regex for a safe placeholder: ``{simple_name}`` only.
|
||||
#:
|
||||
#: Note: ``\w`` matches Unicode word characters (letters, digits,
|
||||
#: underscore) — not just ASCII ``[a-zA-Z0-9_]``. This means keys
|
||||
#: like ``{café}`` or ``{名前}`` are accepted. If ASCII-only keys
|
||||
#: are required, change the pattern to ``[a-zA-Z_][a-zA-Z0-9_]*``.
|
||||
_SAFE_PLACEHOLDER_RE = re.compile(r"\{(\w+)\}")
|
||||
|
||||
#: Single combined pattern that detects **all** unsafe template
|
||||
#: constructs in one pass: attribute access, index access, conversion
|
||||
#: flags, format specs, function calls, and Jinja2 delimiters.
|
||||
_UNSAFE_RE: re.Pattern[str] = re.compile(
|
||||
r"|".join(
|
||||
[
|
||||
r"\{[^}]*\.[^}]*\}", # attribute access {x.y}
|
||||
r"\{[^}]*\[[^}]*\]\}", # index access {x[0]}
|
||||
r"\{[^}]*![^}]*\}", # conversion {x!r}
|
||||
r"\{[^}]*:[^}]+\}", # format spec {x:>10}
|
||||
r"\{[^}]*\([^}]*\)\}", # function call {fn()}
|
||||
r"\{\{", # Jinja2 start
|
||||
r"\}\}", # Jinja2 end
|
||||
r"\{%", # Jinja2 block
|
||||
r"%\}", # Jinja2 block end
|
||||
]
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Exceptions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TemplateError(Exception):
|
||||
"""Base exception for all template rendering errors."""
|
||||
|
||||
|
||||
class TemplateSecurityError(TemplateError):
|
||||
"""Raised when a template contains unsafe constructs."""
|
||||
|
||||
|
||||
class TemplateSizeError(TemplateError):
|
||||
"""Raised when a template or its output exceeds size limits."""
|
||||
|
||||
|
||||
class TemplateValidationError(TemplateError):
|
||||
"""Raised when template validation fails (e.g. unknown keys)."""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Configuration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TemplateConfig:
|
||||
"""Configuration for the secure template renderer.
|
||||
|
||||
Attributes:
|
||||
allowed_keys: The set of context key names that may appear in
|
||||
templates. If empty, **all** simple identifier keys are
|
||||
permitted (open-allowlist mode).
|
||||
max_template_length: Maximum number of characters in the raw
|
||||
template string.
|
||||
max_output_length: Maximum number of characters in the rendered
|
||||
output string.
|
||||
reject_unknown_keys: When ``True`` (default), referencing a key
|
||||
not in *allowed_keys* raises ``TemplateValidationError``.
|
||||
Ignored when *allowed_keys* is empty.
|
||||
log_rejected: When ``True``, log rejected keys at WARNING level.
|
||||
"""
|
||||
|
||||
allowed_keys: frozenset[str] = field(default_factory=frozenset)
|
||||
max_template_length: int = DEFAULT_MAX_TEMPLATE_LENGTH
|
||||
max_output_length: int = DEFAULT_MAX_OUTPUT_LENGTH
|
||||
reject_unknown_keys: bool = True
|
||||
log_rejected: bool = True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Renderer
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class SecureTemplateRenderer:
|
||||
"""Sandboxed template renderer with strict security controls.
|
||||
|
||||
Only ``{variable_name}`` substitutions are allowed. Attribute
|
||||
access, index access, format specs, conversion flags, and function
|
||||
calls are all rejected.
|
||||
"""
|
||||
|
||||
def __init__(self, config: TemplateConfig | None = None) -> None:
|
||||
self._config = config or TemplateConfig()
|
||||
|
||||
@property
|
||||
def config(self) -> TemplateConfig:
|
||||
"""Return the active configuration."""
|
||||
return self._config
|
||||
|
||||
# -- Public API ---------------------------------------------------------
|
||||
|
||||
def render(
|
||||
self,
|
||||
template: str,
|
||||
context: dict[str, Any] | None = None,
|
||||
) -> str:
|
||||
"""Render *template* with the given *context*.
|
||||
|
||||
Args:
|
||||
template: The template string containing ``{key}``
|
||||
placeholders.
|
||||
context: Mapping of placeholder names to values.
|
||||
|
||||
Returns:
|
||||
The rendered string.
|
||||
|
||||
Raises:
|
||||
TemplateSizeError: If the template or output exceeds the
|
||||
configured length limits.
|
||||
TemplateSecurityError: If the template contains unsafe
|
||||
constructs (attribute access, function calls, etc.).
|
||||
TemplateValidationError: If the template references keys
|
||||
not in the allowed set.
|
||||
"""
|
||||
context = context or {}
|
||||
self._check_template_length(template)
|
||||
self._check_security(template)
|
||||
used_keys = _extract_placeholder_names(template)
|
||||
self._check_allowed_keys(used_keys)
|
||||
rendered = self._substitute(template, context)
|
||||
self._check_output_length(rendered)
|
||||
return rendered
|
||||
|
||||
def validate(self, template: str) -> list[str]:
|
||||
"""Validate *template* without rendering.
|
||||
|
||||
Returns a list of error messages (empty = valid).
|
||||
"""
|
||||
errors: list[str] = []
|
||||
|
||||
if len(template) > self._config.max_template_length:
|
||||
errors.append(
|
||||
f"Template length {len(template)} exceeds maximum "
|
||||
f"{self._config.max_template_length}"
|
||||
)
|
||||
|
||||
for match in _UNSAFE_RE.finditer(template):
|
||||
errors.append(f"Unsafe construct detected: {match.group()!r}")
|
||||
|
||||
# Check for unknown keys when allowlist is enforced.
|
||||
if self._config.allowed_keys and self._config.reject_unknown_keys:
|
||||
used_keys = _extract_placeholder_names(template)
|
||||
unknown = used_keys - self._config.allowed_keys
|
||||
if unknown:
|
||||
errors.append(f"Unknown template keys: {', '.join(sorted(unknown))}")
|
||||
|
||||
return errors
|
||||
|
||||
# -- Internal helpers ---------------------------------------------------
|
||||
|
||||
def _check_template_length(self, template: str) -> None:
|
||||
if len(template) > self._config.max_template_length:
|
||||
raise TemplateSizeError(
|
||||
f"Template length {len(template)} exceeds maximum "
|
||||
f"{self._config.max_template_length}"
|
||||
)
|
||||
|
||||
def _check_output_length(self, rendered: str) -> None:
|
||||
if len(rendered) > self._config.max_output_length:
|
||||
raise TemplateSizeError(
|
||||
f"Rendered output length {len(rendered)} exceeds maximum "
|
||||
f"{self._config.max_output_length}"
|
||||
)
|
||||
|
||||
def _check_security(self, template: str) -> None:
|
||||
match = _UNSAFE_RE.search(template)
|
||||
if match:
|
||||
raise TemplateSecurityError(f"Unsafe template construct: {match.group()!r}")
|
||||
|
||||
def _check_allowed_keys(
|
||||
self,
|
||||
used_keys: set[str],
|
||||
) -> None:
|
||||
if not self._config.allowed_keys:
|
||||
return
|
||||
unknown = used_keys - self._config.allowed_keys
|
||||
if unknown and self._config.reject_unknown_keys:
|
||||
if self._config.log_rejected:
|
||||
logger.warning(
|
||||
"Rejected template keys: %s",
|
||||
", ".join(sorted(unknown)),
|
||||
)
|
||||
raise TemplateValidationError(
|
||||
f"Template uses disallowed keys: {', '.join(sorted(unknown))}"
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _substitute(template: str, context: dict[str, Any]) -> str:
|
||||
"""Perform safe ``{key}`` substitution.
|
||||
|
||||
Unknown placeholders are left as-is (safe-substitute behaviour).
|
||||
"""
|
||||
|
||||
def _replace(match: re.Match[str]) -> str:
|
||||
key = match.group(1)
|
||||
if key in context:
|
||||
return str(context[key])
|
||||
return match.group(0)
|
||||
|
||||
return _SAFE_PLACEHOLDER_RE.sub(_replace, template)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Module-level helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _extract_placeholder_names(template: str) -> set[str]:
|
||||
"""Extract all ``{name}`` placeholder names from *template*."""
|
||||
return set(_SAFE_PLACEHOLDER_RE.findall(template))
|
||||
|
||||
|
||||
def validate_template(
|
||||
template: str,
|
||||
*,
|
||||
allowed_keys: frozenset[str] | None = None,
|
||||
max_template_length: int = DEFAULT_MAX_TEMPLATE_LENGTH,
|
||||
) -> list[str]:
|
||||
"""Pre-validate a template string at creation time.
|
||||
|
||||
Convenience wrapper that builds a :class:`SecureTemplateRenderer`
|
||||
and calls :meth:`~SecureTemplateRenderer.validate`.
|
||||
|
||||
Args:
|
||||
template: The template string to validate.
|
||||
allowed_keys: Optional set of allowed placeholder names.
|
||||
max_template_length: Maximum template length.
|
||||
|
||||
Returns:
|
||||
A list of error messages. Empty means the template is valid.
|
||||
"""
|
||||
config = TemplateConfig(
|
||||
allowed_keys=allowed_keys or frozenset(),
|
||||
max_template_length=max_template_length,
|
||||
)
|
||||
renderer = SecureTemplateRenderer(config=config)
|
||||
return renderer.validate(template)
|
||||
|
||||
|
||||
def render_template_secure(
|
||||
template: str,
|
||||
context: dict[str, Any],
|
||||
*,
|
||||
allowed_keys: frozenset[str] | None = None,
|
||||
max_template_length: int = DEFAULT_MAX_TEMPLATE_LENGTH,
|
||||
max_output_length: int = DEFAULT_MAX_OUTPUT_LENGTH,
|
||||
) -> str:
|
||||
"""Render a template securely with one function call.
|
||||
|
||||
Convenience wrapper around :class:`SecureTemplateRenderer`.
|
||||
|
||||
Args:
|
||||
template: The template string.
|
||||
context: Mapping of key names to values.
|
||||
allowed_keys: Optional set of allowed placeholder names.
|
||||
max_template_length: Maximum template length.
|
||||
max_output_length: Maximum rendered output length.
|
||||
|
||||
Returns:
|
||||
The rendered string.
|
||||
|
||||
Raises:
|
||||
TemplateError: On any security, validation, or size error.
|
||||
"""
|
||||
config = TemplateConfig(
|
||||
allowed_keys=allowed_keys or frozenset(),
|
||||
max_template_length=max_template_length,
|
||||
max_output_length=max_output_length,
|
||||
)
|
||||
renderer = SecureTemplateRenderer(config=config)
|
||||
return renderer.render(template, context)
|
||||
Reference in New Issue
Block a user