Compare commits

...
Author SHA1 Message Date
Ahmed Allam 6797c96239 chore: drop added comments and docstrings 2026-07-26 23:18:01 +00:00
Ahmed Allam 979aca7a32 fix(runtime): retry transient mid-stream provider errors instead of crashing the scan 2026-07-26 22:09:17 +00:00
3b79e97f00 feat(context): spill oversized tool output into the sandbox workspace (#882)
Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>
2026-07-26 14:42:22 -07:00
devin-ai-integration[bot]andGitHub 66a283b71b feat(context): model-aware conversation compaction for long scans (#881) 2026-07-26 14:39:41 -07:00
Ahmed AllamandAhmed Allam 74f334cb93 refactor(context): trim verbose comments 2026-07-26 14:38:12 -07:00
Ahmed AllamandAhmed Allam 8e9a6bf903 fix(context): reject tool-output byte ceilings below the notice size
A configured tool_output_max_bytes smaller than the truncation notice
itself can't fit a bounded preview, so a persisted result could exceed the
ceiling. Enforce a config floor (ge=1024) so nonsensical values are
rejected at load time instead of being worked around at runtime.
2026-07-26 14:38:12 -07:00
Ahmed AllamandAhmed Allam 0ebd3c6230 fix(context): reserve notice budget so bounded output honors max_bytes
The head+tail slices could each take half of max_bytes, then the
truncation notice and its separators were appended on top, so the value
persisted to history could exceed the configured maximum. Reserve an
upper bound for the notice (and separators) out of the byte budget before
slicing so the whole joined result stays within max_bytes.
2026-07-26 14:38:12 -07:00
Ahmed AllamandAhmed Allam 6bda366065 fix(context): bound native filesystem tool output in Responses mode
Chat-completions mode converts filesystem CustomTools to FunctionTools
(which bounds their result), but the Responses-API path kept them native
and unbounded, so a large read_file could still exhaust the context
window. Always configure the Filesystem capability to head+tail bound
tool output in both modes.
2026-07-26 14:38:12 -07:00
Ahmed AllamandAhmed Allam 1f36f5d401 fix(context): clamp shell output cap and count byte-trimmed dropped lines
Treat tool_output_max_tokens as a ceiling so an explicit model-supplied
cap can't exceed it, and derive the truncation notice's dropped-line
count from the lines actually kept after the byte-trim pass. Also cast
the pygments fallback lexer so it satisfies the resolve_lexer return
type under the pre-commit mypy hook.
2026-07-26 14:38:12 -07:00
Ahmed AllamandAhmed Allam a70a87f272 feat(context): bound per-tool output before it enters agent history
Cap the size of every tool result so a single verbose command (recursive
find, noisy scanner, full page dump) can't pin the conversation near the
model's context window for the rest of a scan.

- New ContextSettings config group with env-tunable caps.
- Default the SDK shell tools' max_output_tokens so exec_command /
  write_stdin truncate head+tail instead of returning unbounded output.
- Bound Strix's own FunctionTool/CustomTool results (line + UTF-8 byte
  head+tail preview with a truncation notice) and cap error strings.
2026-07-26 14:38:12 -07:00
devin-ai-integration[bot]andGitHub d2fbcb726d feat(reporting): add read-only list_reports + get_report tools (#889) 2026-07-26 14:05:53 -07:00
Ahmed AllamandAhmed Allam 8169e177de docs(skills): remove references to tools not installed in the sandbox
Skills and the agent system prompt referenced external CLIs that are not
present in containers/Dockerfile, which could lead the agent to invoke
missing binaries. Replace them with installed equivalents:

- asset_discovery: drop amass/cero and the projectdiscovery tools that are
  not installed (tlsx/dnsx/asnmap/mapcidr/uncover); rewrite around the
  installed subfinder/httpx/naabu plus curl+jq (crt.sh), openssl s_client,
  dig, and whois. Stop claiming the full projectdiscovery suite is available.
- subdomain_takeover: replace dnsx with dig in the pipeline example.
- weak_password_detection: drop hydra/cewl/patator; use ffuf for web logins
  and nmap NSE *-brute scripts for services; fix dead /usr/share/wordlists
  and /usr/share/seclists paths (nothing ships by default -> download to
  /home/pentester/tools/wordlists at runtime).
- system_prompt: replace msfconsole with sqlmap in the interactive-process
  example.

active_directory skill is left as-is: it already ships an explicit install
block for its tools.
2026-07-26 14:00:07 -07:00
Ahmed AllamandAhmed Allam 589bade39a fix: restore viewer-auth.json path in auth module docstring 2026-07-26 13:11:14 -07:00
Ahmed AllamandAhmed Allam d1e8225d5f refactor: move strix/viewer under strix/interface 2026-07-26 13:11:14 -07:00
118 changed files with 2936 additions and 337 deletions
+3 -3
View File
@@ -1,8 +1,8 @@
# Node / local-viewer SPA source (the built bundle in
# strix/viewer/static/ is committed and shipped; do not ignore it)
# strix/interface/viewer/static/ is committed and shipped; do not ignore it)
node_modules/
strix/viewer/frontend/node_modules/
strix/viewer/frontend/.vite/
strix/interface/viewer/frontend/node_modules/
strix/interface/viewer/frontend/.vite/
# Python
__pycache__/
+5 -5
View File
@@ -102,16 +102,16 @@ We welcome feature ideas! Please:
## 🖥️ Local viewer SPA
`strix view` serves a prebuilt web UI whose source lives in
`strix/viewer/frontend/` (a Vite + React project) and whose built output is
committed to `strix/viewer/static/` and shipped in the package. End users never
run a JS build. If you change anything under `strix/viewer/frontend/`, rebuild
`strix/interface/viewer/frontend/` (a Vite + React project) and whose built output is
committed to `strix/interface/viewer/static/` and shipped in the package. End users never
run a JS build. If you change anything under `strix/interface/viewer/frontend/`, rebuild
and commit the output:
```bash
make viewer # or: cd strix/viewer/frontend && npm ci && npm run build
make viewer # or: cd strix/interface/viewer/frontend && npm ci && npm run build
```
Commit both the source change and the regenerated `strix/viewer/static/`.
Commit both the source change and the regenerated `strix/interface/viewer/static/`.
## 🤝 Community
+2 -2
View File
@@ -69,8 +69,8 @@ clean:
viewer:
@echo "🖥️ Building the local-viewer SPA..."
cd strix/viewer/frontend && npm ci && npm run build
@echo "✅ Viewer built to strix/viewer/static/ (commit the changes)."
cd strix/interface/viewer/frontend && npm ci && npm run build
@echo "✅ Viewer built to strix/interface/viewer/static/ (commit the changes)."
dev: format lint type-check
@echo "✅ Development cycle complete!"
+6 -6
View File
@@ -79,10 +79,10 @@ build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["strix"]
# The prebuilt viewer bundle under strix/viewer/static/ ships automatically
# The prebuilt viewer bundle under strix/interface/viewer/static/ ships automatically
# (hatchling includes non-.py files under the package). The Vite SOURCE lives
# under the package dir too (strix/viewer/frontend/) but must never ship in the wheel.
exclude = ["strix/viewer/frontend", "strix/viewer/frontend/**"]
# under the package dir too (strix/interface/viewer/frontend/) but must never ship in the wheel.
exclude = ["strix/interface/viewer/frontend", "strix/interface/viewer/frontend/**"]
# ============================================================================
# Type Checking Configuration
@@ -222,10 +222,10 @@ ignore = [
"tests/test_codex_streaming.py" = ["N802"]
"tests/test_report_pdf.py" = ["S105", "S106"]
# Stdlib HTTP handler overrides (do_GET/do_POST) and lazy imports that avoid a
# circular dependency with strix.telemetry / strix.viewer.report_pdf.
"strix/viewer/server.py" = ["N802", "PLC0415"]
# circular dependency with strix.telemetry / strix.interface.viewer.report_pdf.
"strix/interface/viewer/server.py" = ["N802", "PLC0415"]
# Lazy telemetry import to avoid importing PostHog before the viewer starts.
"strix/viewer/cli.py" = ["PLC0415"]
"strix/interface/viewer/cli.py" = ["PLC0415"]
# Lazy imports inside functions to avoid circular dependency with
# strix.telemetry / strix.report.dedupe / cvss.
"strix/tools/notes/tools.py" = ["PLC0415", "TC002"]
+7 -7
View File
@@ -26,7 +26,7 @@ for tcss_file in strix_root.rglob('*.tcss'):
datas.append((str(tcss_file), str(rel_path.parent)))
# Prebuilt local-viewer SPA (served by `strix view`).
viewer_static = strix_root / 'viewer' / 'static'
viewer_static = strix_root / 'interface' / 'viewer' / 'static'
for asset in viewer_static.rglob('*'):
if asset.is_file():
rel_path = asset.relative_to(project_root)
@@ -158,12 +158,12 @@ hiddenimports = [
'strix.report.dedupe',
'strix.report.state',
'strix.report.writer',
'strix.viewer',
'strix.viewer.auth',
'strix.viewer.cli',
'strix.viewer.report_pdf',
'strix.viewer.server',
'strix.viewer.transcript',
'strix.interface.viewer',
'strix.interface.viewer.auth',
'strix.interface.viewer.cli',
'strix.interface.viewer.report_pdf',
'strix.interface.viewer.server',
'strix.interface.viewer.transcript',
# PDF report generation + encryption
'reportlab',
+91 -14
View File
@@ -16,6 +16,7 @@ from agents.tool import CustomTool, FunctionTool, Tool
from pydantic import ValidationError
from strix.agents.prompt import render_system_prompt
from strix.config import load_settings
from strix.tools.agents_graph.tools import (
agent_finish,
create_agent,
@@ -33,6 +34,7 @@ from strix.tools.notes.tools import (
list_notes,
update_note,
)
from strix.tools.output_store import bound_and_store, bound_text
from strix.tools.proxy.tools import (
list_requests,
list_sitemap,
@@ -41,7 +43,12 @@ from strix.tools.proxy.tools import (
view_request,
view_sitemap_entry,
)
from strix.tools.reporting.tool import create_dependency_report, create_vulnerability_report
from strix.tools.reporting.tool import (
create_dependency_report,
create_vulnerability_report,
get_report,
list_reports,
)
from strix.tools.thinking.tool import think
from strix.tools.todo.tools import (
create_todo,
@@ -103,8 +110,36 @@ def _extract_custom_input(tool: CustomTool, raw_input: str | dict[str, Any]) ->
return value if isinstance(value, str) else ""
def _tool_output_limits() -> tuple[int, int]:
context = load_settings().context
return context.tool_output_max_lines, context.tool_output_max_bytes
async def _bound_result(result: Any) -> Any:
if not isinstance(result, str):
return result
max_lines, max_bytes = _tool_output_limits()
return await bound_and_store(result, max_lines=max_lines, max_bytes=max_bytes)
def _format_tool_error(exc: Exception) -> str:
return str(exc) or exc.__class__.__name__
message = str(exc) or exc.__class__.__name__
max_lines, max_bytes = _tool_output_limits()
return bound_text(message, max_lines=max_lines, max_bytes=max_bytes)
def _with_bounded_result(tool: FunctionTool) -> FunctionTool:
"""Cap a tool's result size before it enters history (idempotent)."""
if getattr(tool, "_strix_bounded", False):
return tool
invoke_tool = tool.on_invoke_tool
async def invoke(ctx: Any, raw_input: str) -> Any:
return await _bound_result(await invoke_tool(ctx, raw_input))
tool.on_invoke_tool = invoke
tool._strix_bounded = True # type: ignore[attr-defined]
return tool
def _function_tool_with_error_result(tool: FunctionTool) -> FunctionTool:
@@ -112,7 +147,7 @@ def _function_tool_with_error_result(tool: FunctionTool) -> FunctionTool:
async def invoke(ctx: Any, raw_input: str) -> Any:
try:
return await invoke_tool(ctx, raw_input)
return await _bound_result(await invoke_tool(ctx, raw_input))
except Exception as exc: # noqa: BLE001 - tool errors should be model-visible results.
logger.debug("Tool %s failed; returning error as result", tool.name, exc_info=True)
return _format_tool_error(exc)
@@ -127,7 +162,7 @@ def _custom_tool_as_function_tool(tool: CustomTool) -> FunctionTool:
if not custom_input:
return f"`{_custom_tool_input_field(tool)}` must be a non-empty string."
try:
return await tool.on_invoke_tool(ctx, custom_input)
return await _bound_result(await tool.on_invoke_tool(ctx, custom_input))
except Exception as exc: # noqa: BLE001 - matches SDK CustomTool error-as-result behavior.
logger.debug("Tool %s failed; returning error as result", tool.name, exc_info=True)
return _format_tool_error(exc)
@@ -159,12 +194,35 @@ def _custom_tool_as_function_tool(tool: CustomTool) -> FunctionTool:
)
def _configure_chat_completions_filesystem_tools(toolset: Any) -> None:
def _bound_custom_tool(tool: CustomTool) -> CustomTool:
"""Bound a native ``CustomTool`` result in place (Responses path)."""
invoke_tool = tool.on_invoke_tool
async def invoke(ctx: Any, raw_input: str) -> Any:
return await _bound_result(await invoke_tool(ctx, raw_input))
tool.on_invoke_tool = invoke
return tool
def _configure_filesystem_tools(toolset: Any, *, chat_completions: bool) -> None:
for name, tool in vars(toolset).items():
if isinstance(tool, CustomTool):
setattr(toolset, name, _custom_tool_as_function_tool(tool))
if chat_completions:
if isinstance(tool, CustomTool):
setattr(toolset, name, _custom_tool_as_function_tool(tool))
elif isinstance(tool, FunctionTool):
setattr(toolset, name, _function_tool_with_error_result(tool))
elif isinstance(tool, CustomTool):
setattr(toolset, name, _bound_custom_tool(tool))
elif isinstance(tool, FunctionTool):
setattr(toolset, name, _function_tool_with_error_result(tool))
setattr(toolset, name, _with_bounded_result(tool))
def _make_filesystem_configurator(*, chat_completions: bool) -> Any:
def configure(toolset: Any) -> None:
_configure_filesystem_tools(toolset, chat_completions=chat_completions)
return configure
_CHARS_ESCAPE_RE = re.compile(r"\\(?:u[0-9a-fA-F]{4}|x[0-9a-fA-F]{2}|[0abtnvfr\\])")
@@ -205,6 +263,16 @@ def _format_validation_error(tool_name: str, exc: ValidationError) -> str:
return f"{tool_name}: invalid arguments — " + "; ".join(parts)
def _apply_shell_output_cap(parsed: dict[str, Any]) -> None:
"""Clamp the SDK shell tools' ``max_output_tokens`` to the configured
ceiling; a smaller explicit value is respected."""
ceiling = load_settings().context.tool_output_max_tokens
requested = parsed.get("max_output_tokens")
parsed["max_output_tokens"] = (
ceiling if not isinstance(requested, int) or requested > ceiling else requested
)
def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
invoke_tool = tool.on_invoke_tool
@@ -213,8 +281,10 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
parsed = json.loads(raw_input)
except (json.JSONDecodeError, TypeError):
parsed = None
if isinstance(parsed, dict) and "shell" not in parsed:
parsed["shell"] = "bash"
if isinstance(parsed, dict):
if "shell" not in parsed:
parsed["shell"] = "bash"
_apply_shell_output_cap(parsed)
raw_input = json.dumps(parsed)
try:
return await invoke_tool(ctx, raw_input)
@@ -240,8 +310,10 @@ def _wrap_write_stdin(tool: FunctionTool) -> FunctionTool:
parsed = json.loads(raw_input)
except json.JSONDecodeError:
parsed = None
if isinstance(parsed, dict) and isinstance(parsed.get("chars"), str):
parsed["chars"] = _decode_chars_escape(parsed["chars"])
if isinstance(parsed, dict):
if isinstance(parsed.get("chars"), str):
parsed["chars"] = _decode_chars_escape(parsed["chars"])
_apply_shell_output_cap(parsed)
raw_input = json.dumps(parsed)
try:
return await invoke_tool(ctx, raw_input)
@@ -343,6 +415,8 @@ _BASE_TOOLS: tuple[Tool, ...] = (
web_search,
create_vulnerability_report,
create_dependency_report,
list_reports,
get_report,
list_requests,
view_request,
repeat_request,
@@ -440,6 +514,9 @@ def build_strix_agent(
else:
tools = [*_BASE_TOOLS, *agent_tools, agent_finish]
_ensure_unique_tool_names(tools)
tools = [
_with_bounded_result(tool) if isinstance(tool, FunctionTool) else tool for tool in tools
]
logger.info(
"Built %s agent '%s' (skills=%d, tools=%d, scan_mode=%s, whitebox=%s)",
@@ -459,8 +536,8 @@ def build_strix_agent(
model=None,
capabilities=[
Filesystem(
configure_tools=(
_configure_chat_completions_filesystem_tools if chat_completions_tools else None
configure_tools=_make_filesystem_configurator(
chat_completions=chat_completions_tools,
),
),
Shell(
+2 -1
View File
@@ -188,7 +188,7 @@ EFFICIENCY TACTICS:
script fail with `ModuleNotFoundError`.
- `exec_command` runs each command in a fresh non-interactive shell (plain
pipes, no TTY). To drive an interactive or long-running process with
`write_stdin` — REPLs, `ssh`/`nc`/`ftp`, `msfconsole`, or to send Ctrl-C —
`write_stdin` — REPLs, `ssh`/`nc`/`ftp`, `sqlmap`, or to send Ctrl-C —
you MUST start it with `exec_command(cmd="...", tty=true)` and then
`write_stdin(session_id=<id>, chars="...")`. Calling `write_stdin` on a
default (non-TTY) command or on a process that has already exited fails with
@@ -215,6 +215,7 @@ VALIDATION REQUIREMENTS:
- A vulnerability is ONLY considered reported when a reporting agent uses create_vulnerability_report (or create_dependency_report for known-CVE dependency/supply-chain findings) with full details. Mentions in agent_finish, finish_scan, or generic messages are NOT sufficient
- Reporting and fixing are ONE step, not two: when source is available, the reporting agent derives the concrete fix and files it INLINE via create_vulnerability_report (`code_locations` with `fix_before`/`fix_after` + `fix_pr_body`) — the report is not complete without it. Do NOT report first and then spawn a separate downstream agent to re-derive and re-apply the same patch; that just re-does the analysis and wastes tokens. (Do not silently patch a finding WITHOUT filing a report — the report, with its embedded fix, is the deliverable.)
- DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, DO NOT attempt to re-submit the same vulnerability. Accept the rejection and move on to testing other areas. The vulnerability has already been reported by another agent
- REVIEWING FILED FINDINGS (orchestrator/root agent): use list_reports to see every vulnerability filed so far in this scan (by any agent, root or child) — metadata-first with per-severity counts — and get_report to read one finding in full by its id. These are read-only orchestration tools: the root agent uses them to track coverage, avoid dispatching work on already-covered ground, assemble the finish_scan executive summary, and reason about attack-chaining across confirmed findings. Leaf/specialist agents should NOT call them — just do your assigned testing and file findings. Each entry shows which agent filed it (agent_name), and your own entries are flagged by_you. list_notes/get_note do the same for notes.
</execution_guidelines>
<vulnerability_focus>
+2
View File
@@ -17,6 +17,7 @@ from strix.config.loader import (
persist_current,
)
from strix.config.settings import (
ContextSettings,
DedupeSettings,
IntegrationSettings,
LlmSettings,
@@ -27,6 +28,7 @@ from strix.config.settings import (
__all__ = [
"ContextSettings",
"DedupeSettings",
"IntegrationSettings",
"LlmSettings",
+21
View File
@@ -55,6 +55,26 @@ class DedupeSettings(BaseSettings):
api_base: str | None = Field(default=None, alias="DEDUPE_LLM_API_BASE")
class ContextSettings(BaseSettings):
"""Context-window management: per-tool-output caps and history compaction."""
model_config = _BASE_CONFIG
auto_compact: bool = Field(default=True, alias="STRIX_CONTEXT_AUTO_COMPACT")
compact_buffer_tokens: int = Field(default=20_000, gt=0, alias="STRIX_CONTEXT_BUFFER_TOKENS")
keep_tokens: int = Field(default=8_000, gt=0, alias="STRIX_CONTEXT_KEEP_TOKENS")
fallback_context_tokens: int = Field(
default=200_000, gt=0, alias="STRIX_CONTEXT_FALLBACK_TOKENS"
)
summary_max_tokens: int = Field(default=4_096, gt=0, alias="STRIX_CONTEXT_SUMMARY_TOKENS")
tool_output_max_tokens: int = Field(default=8_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_TOKENS")
tool_output_max_lines: int = Field(default=2_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_LINES")
# Floor above the truncation-notice size so a preview always fits.
tool_output_max_bytes: int = Field(
default=50 * 1024, ge=1024, alias="STRIX_TOOL_OUTPUT_MAX_BYTES"
)
class RuntimeSettings(BaseSettings):
model_config = _BASE_CONFIG
@@ -99,6 +119,7 @@ class Settings(BaseSettings):
llm: LlmSettings = Field(default_factory=LlmSettings)
dedupe: DedupeSettings = Field(default_factory=DedupeSettings)
runtime: RuntimeSettings = Field(default_factory=RuntimeSettings)
context: ContextSettings = Field(default_factory=ContextSettings)
telemetry: TelemetrySettings = Field(default_factory=TelemetrySettings)
integrations: IntegrationSettings = Field(default_factory=IntegrationSettings)
viewer: ViewerSettings = Field(default_factory=ViewerSettings)
+112 -1
View File
@@ -13,7 +13,13 @@ from agents import RunConfig, Runner
from agents.exceptions import AgentsException, MaxTurnsExceeded, UserError
from agents.sandbox.errors import ExecTransportError
from docker import errors as docker_errors # type: ignore[import-untyped, unused-ignore]
from openai import APIError
from openai import (
APIConnectionError,
APIError,
APIStatusError,
APITimeoutError,
RateLimitError,
)
from strix.core.hooks import BudgetExceededError
from strix.core.inputs import child_initial_input
@@ -22,6 +28,7 @@ from strix.core.sessions import (
open_agent_session,
strip_all_images_from_session,
)
from strix.llm.compaction import is_context_overflow, maybe_compact
if TYPE_CHECKING:
@@ -40,6 +47,69 @@ logger = logging.getLogger(__name__)
StreamEventSink = Callable[[str, Any], None]
_INPUT_REJECTION_CODES = frozenset({400, 404, 422})
_MAX_COMPACTIONS_PER_CYCLE = 2
def _run_config_model(run_config: RunConfig) -> str | None:
return run_config.model if isinstance(run_config.model, str) else None
def _agent_instructions(agent: Any) -> str:
instructions = getattr(agent, "instructions", None)
return instructions if isinstance(instructions, str) else ""
def _agent_tools_text(agent: Any) -> str:
parts: list[str] = []
for tool in getattr(agent, "tools", []) or []:
name = getattr(tool, "name", "")
description = getattr(tool, "description", "") or ""
schema = getattr(tool, "params_json_schema", "") or ""
parts.append(f"{name} {description} {schema}")
return "\n".join(parts)
async def _compact_session(
agent: Any, session: Session, run_config: RunConfig, *, force: bool
) -> bool:
model = _run_config_model(run_config)
if session is None or model is None:
return False
return await maybe_compact(
session,
model=model,
instructions=_agent_instructions(agent),
tools_text=_agent_tools_text(agent),
force=force,
)
_TRANSIENT_MODEL_STATUS_CODES = frozenset({408, 500, 502, 503, 504})
_MAX_TRANSIENT_MODEL_RETRIES = 4
_TRANSIENT_MODEL_RETRY_BASE_DELAY_S = 2.0
_TRANSIENT_MODEL_RETRY_MAX_DELAY_S = 30.0
def _model_error_status_code(exc: BaseException) -> int | None:
code = getattr(exc, "status_code", None)
return code if isinstance(code, int) else None
def _is_transient_model_error(exc: BaseException) -> bool:
if isinstance(exc, RateLimitError):
return False
if isinstance(exc, APITimeoutError | APIConnectionError):
return True
if isinstance(exc, APIStatusError):
return exc.status_code in _TRANSIENT_MODEL_STATUS_CODES
if isinstance(exc, APIError):
return _model_error_status_code(exc) is None
return False
def _transient_model_retry_delay(attempt: int) -> float:
delay = _TRANSIENT_MODEL_RETRY_BASE_DELAY_S * float(2 ** (attempt - 1))
return min(delay, _TRANSIENT_MODEL_RETRY_MAX_DELAY_S)
async def run_agent_loop(
@@ -350,6 +420,8 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
hooks: RunHooks[dict[str, Any]] | None,
) -> RunResultBase | None:
image_strips = 0
compactions = 0
model_retries = 0
while True:
try:
await coordinator.mark_running(agent_id)
@@ -360,6 +432,10 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
await enforce_image_budget(session, max_images)
except Exception:
logger.exception("image-budget enforcement failed for %s", agent_id)
try:
await _compact_session(agent, session, run_config, force=False)
except Exception:
logger.exception("proactive compaction failed for %s", agent_id)
stream = Runner.run_streamed(
agent,
input=input_data,
@@ -428,6 +504,41 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
)
input_data = []
continue
if (
compactions < _MAX_COMPACTIONS_PER_CYCLE
and session is not None
and is_context_overflow(exc)
):
try:
compacted = await _compact_session(agent, session, run_config, force=True)
except Exception:
logger.exception("overflow compaction recovery failed for %s", agent_id)
compacted = False
if compacted:
compactions += 1
logger.info(
"Compacted %s session after context overflow; retrying (%d)",
agent_id,
compactions,
)
input_data = []
continue
if model_retries < _MAX_TRANSIENT_MODEL_RETRIES and _is_transient_model_error(exc):
model_retries += 1
delay = _transient_model_retry_delay(model_retries)
logger.warning(
"transient model/provider error for %s; replaying turn "
"(attempt %d/%d, backoff %.1fs): %r",
agent_id,
model_retries,
_MAX_TRANSIENT_MODEL_RETRIES,
delay,
exc,
)
await asyncio.sleep(delay)
if session is not None:
input_data = []
continue
if not interactive:
raise
if isinstance(exc, MaxTurnsExceeded):
+21
View File
@@ -3,10 +3,12 @@
from __future__ import annotations
import contextlib
import io
import json
import logging
import uuid
from collections.abc import Callable
from pathlib import Path
from typing import TYPE_CHECKING, Any
from agents import RunConfig
@@ -40,6 +42,10 @@ from strix.core.paths import run_dir_for, runtime_state_dir
from strix.core.sessions import open_agent_session
from strix.runtime import session_manager
from strix.telemetry.logging import set_scan_id, setup_scan_logging
from strix.tools.output_store import (
WORKSPACE_SPILL_DIR,
configure_spill_writer,
)
if TYPE_CHECKING:
@@ -203,6 +209,20 @@ async def run_strix_scan(
)
logger.info("Sandbox ready for scan %s", scan_id)
sandbox_session = bundle["session"]
async def _spill_to_workspace(output_id: str, text: str) -> str | None:
"""Write an oversized tool result into the sandbox; return its path or None."""
path = f"{WORKSPACE_SPILL_DIR}/{output_id}.txt"
try:
await sandbox_session.write(Path(path), io.BytesIO(text.encode("utf-8")))
except Exception:
logger.exception("failed to spill tool output to sandbox workspace")
return None
return path
configure_spill_writer(_spill_to_workspace)
sessions_to_close: list[SQLiteSession] = []
try:
@@ -399,6 +419,7 @@ async def run_strix_scan(
await coordinator.set_status(root_id, "failed")
raise
finally:
configure_spill_writer(None)
for s in sessions_to_close:
with contextlib.suppress(Exception):
s.close()
+33
View File
@@ -92,6 +92,39 @@ async def _rewrite_session(
return True
async def replace_session_items(
session: Session,
new_items: list[Any],
*,
expected_len: int | None = None,
) -> bool:
"""Overwrite the session's items, restoring the originals on failure.
When ``expected_len`` is given, the rewrite is skipped if the session no
longer has that many items (a concurrent writer changed it), so a slow
compaction summary can't clobber newer turns.
"""
async with session_write_lock(session):
original = list(await session.get_items())
if expected_len is not None and len(original) != expected_len:
logger.warning(
"skipping session rewrite: expected %d items, found %d",
expected_len,
len(original),
)
return False
rebuilt = cast("list[TResponseInputItem]", new_items)
await session.clear_session()
try:
await session.add_items(rebuilt)
except Exception:
logger.exception("session rewrite failed; restoring original items")
await session.clear_session()
await session.add_items(original)
raise
return True
async def strip_all_images_from_session(session: Session) -> bool:
"""Replace every image tool output with a text placeholder (rejection recovery)."""
+1 -1
View File
@@ -952,7 +952,7 @@ def main() -> None:
# `strix view [<run>]` is a viewer-only subcommand, dispatched before the
# scan argument parser (which requires a target) and before any scan setup.
if len(sys.argv) > 1 and sys.argv[1] == "view":
from strix.viewer.cli import run_view
from strix.interface.viewer.cli import run_view
run_view(sys.argv[2:])
return
+1 -1
View File
@@ -1862,7 +1862,7 @@ class StrixTUIApp(App): # type: ignore[misc]
webbrowser.open(self._viewer_url)
return
try:
from strix.viewer.server import authorized_url, bundle_is_built, serve
from strix.interface.viewer.server import authorized_url, bundle_is_built, serve
if not bundle_is_built():
self._set_viewer_cta("[#eab308]Viewer UI not built[/]")
@@ -7,6 +7,13 @@ from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
def _author_label(note: dict[str, Any]) -> str:
if note.get("by_you"):
return "you"
agent_name = note.get("agent_name")
return str(agent_name).strip() if agent_name else ""
@register_tool_renderer
class CreateNoteRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "create_note"
@@ -123,6 +130,9 @@ class ListNotesRenderer(BaseToolRenderer):
text.append("\n - ")
text.append(title)
text.append(f" ({category})", style="dim")
author = _author_label(note)
if author:
text.append(f" by {author}", style="dim")
if note_content:
text.append("\n ")
@@ -156,6 +166,9 @@ class GetNoteRenderer(BaseToolRenderer):
text.append("\n ")
text.append(title)
text.append(f" ({category})", style="dim")
author = _author_label(note)
if author:
text.append(f" by {author}", style="dim")
if content:
text.append("\n ")
text.append(content, style="dim")
@@ -431,3 +431,117 @@ class CreateDependencyReportRenderer(BaseToolRenderer):
css_classes = cls.get_css_classes("completed")
return Static(padded, classes=css_classes)
_LIST_SEVERITY_COLORS = {
"critical": "#dc2626",
"high": "#ea580c",
"medium": "#d97706",
"low": "#65a30d",
"info": "#0284c7",
"none": "#6b7280",
}
def _severity_style(severity: Any) -> str:
return _LIST_SEVERITY_COLORS.get(str(severity or "").lower(), "#d97706")
def _author_label(report: dict[str, Any]) -> str:
if report.get("by_you"):
return "you"
agent_name = report.get("agent_name")
return str(agent_name).strip() if agent_name else ""
@register_tool_renderer
class ListReportsRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "list_reports"
css_classes: ClassVar[list[str]] = ["tool-call", "reporting-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = _coerce_dict(tool_data.get("result"))
text = Text()
text.append("", style="#ef4444")
text.append("reports", style="dim")
if isinstance(tool_data.get("result"), str) and str(tool_data["result"]).strip():
text.append("\n ")
text.append(str(tool_data["result"]).strip(), style="dim")
elif result.get("success"):
total = result.get("total_count", 0)
reports = _coerce_list_of_dicts(result.get("reports"))
counts = _coerce_dict(result.get("severity_counts"))
text.append(f" ({total})", style="dim")
for sev, count in counts.items():
text.append(" ")
text.append(f"{sev} {count}", style=_severity_style(sev))
if not reports:
text.append("\n ")
text.append("No reports filed yet", style="dim")
else:
for report in reports:
rid = str(report.get("id", "")).strip()
title = str(report.get("title", "")).strip() or "(untitled)"
severity = str(report.get("severity", "")).strip()
text.append("\n - ")
if severity:
text.append(severity.upper(), style=f"bold {_severity_style(severity)}")
text.append(" ")
if rid:
text.append(f"{rid} ", style="dim")
text.append(title)
author = _author_label(report)
if author:
text.append(f" ({author})", style="dim")
else:
text.append("\n ")
text.append("Loading...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class GetReportRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "get_report"
css_classes: ClassVar[list[str]] = ["tool-call", "reporting-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = _coerce_dict(tool_data.get("result"))
text = Text()
text.append("", style="#ef4444")
text.append("report read", style="dim")
report = _coerce_dict(result.get("report")) if result.get("success") else {}
if report:
rid = str(report.get("id", "")).strip()
title = str(report.get("title", "")).strip() or "(untitled)"
severity = str(report.get("severity", "")).strip()
text.append("\n ")
if severity:
text.append(severity.upper(), style=f"bold {_severity_style(severity)}")
text.append(" ")
if rid:
text.append(f"{rid} ", style="dim")
text.append(title)
author = _author_label(report)
if author:
text.append(f" ({author})", style="dim")
target = str(report.get("target", "")).strip()
if target:
text.append("\n ")
text.append(target, style="dim")
else:
text.append("\n ")
detail = result.get("error") if result.get("success") is False else None
text.append(str(detail) if detail else "Loading...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@@ -6,7 +6,7 @@ directly from the run's on-disk files. No cloud dependency, no file picker.
from __future__ import annotations
from strix.viewer.server import serve
from strix.interface.viewer.server import serve
__all__ = ["serve"]
@@ -155,7 +155,7 @@ def _post_json(path: str, payload: dict[str, Any], *, timeout: int) -> tuple[int
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310
with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 # nosec B310
return response.status, _parse_body(response.read())
except urllib.error.HTTPError as exc:
return exc.code, _parse_body(exc.read())
@@ -16,8 +16,8 @@ from strix.core.paths import (
run_record_path,
runs_base_dir,
)
from strix.viewer.server import authorized_url, bundle_is_built, serve
from strix.viewer.transcript import read_run_summary
from strix.interface.viewer.server import authorized_url, bundle_is_built, serve
from strix.interface.viewer.transcript import read_run_summary
if TYPE_CHECKING:
@@ -58,7 +58,7 @@ def run_view(argv: list[str]) -> None:
if not bundle_is_built():
console.print(
"[bold red]Viewer UI is not built.[/]\n"
"Build it with: [cyan]cd strix/viewer/frontend && npm ci && npm run build[/]"
"Build it with: [cyan]cd strix/interface/viewer/frontend && npm ci && npm run build[/]"
)
raise SystemExit(1)

Before

Width:  |  Height:  |  Size: 3.7 KiB

After

Width:  |  Height:  |  Size: 3.7 KiB

@@ -49,6 +49,9 @@ export default function NotesRenderer({ toolName, args, result }: ToolRendererPr
<div className="mt-1.5 text-[#999] text-[13px]">
{note.title ?? "(untitled)"}
<span className="text-[#555] ml-1">({note.category ?? "general"})</span>
{(note.by_you || note.agent_name) && (
<span className="text-[#666] ml-1 text-xs">by {note.by_you ? "you" : note.agent_name}</span>
)}
</div>
{note.content && <div className="mt-1"><Markdown text={note.content} /></div>}
</>
@@ -74,6 +77,9 @@ export default function NotesRenderer({ toolName, args, result }: ToolRendererPr
<span className="text-[#555] mr-1">-</span>
<span className="text-[#999]">{n.title ?? "(untitled)"}</span>
<span className="text-[#555] ml-1">({n.category ?? "general"})</span>
{(n.by_you || n.agent_name) && (
<span className="text-[#666] ml-1 text-xs">by {n.by_you ? "you" : n.agent_name}</span>
)}
{n.content && <div className="ml-3"><Markdown text={n.content} /></div>}
</div>
))}
@@ -0,0 +1,121 @@
"use client";
import type { ToolRendererProps } from "@/types/events";
import { TruncatedText } from "./ToolCard";
import Markdown from "./Markdown";
const SEVERITY_COLORS: Record<string, string> = {
critical: "text-red-400", high: "text-orange-400", medium: "text-yellow-400",
low: "text-blue-400", info: "text-cyan-400", none: "text-[#888]",
};
interface ReportEntry {
id?: string;
title?: string;
severity?: string;
cvss?: number;
cve?: string;
cwe?: string;
target?: string;
endpoint?: string;
method?: string;
description_preview?: string;
description?: string;
agent_name?: string;
by_you?: boolean;
}
function authorTag(r: ReportEntry) {
if (!r.agent_name && !r.by_you) return null;
const label = r.by_you ? "you" : r.agent_name;
return <span className="text-[#666] text-xs ml-1.5">({label})</span>;
}
function sevBadge(severity: string | undefined) {
const sev = String(severity ?? "").toLowerCase();
const color = SEVERITY_COLORS[sev] ?? "text-yellow-400";
return <span className={`font-semibold text-[13px] ${color}`}>{sev.toUpperCase() || "—"}</span>;
}
export default function ReportListRenderer({ toolName, result }: ToolRendererProps) {
const res = result as Record<string, unknown> | null;
const ok = res != null && typeof res === "object" && res.success === true;
if (toolName === "get_report") {
const report = ok ? (res.report as ReportEntry | undefined) : undefined;
return (
<div>
<span className="text-red-400/80 font-semibold text-sm">report</span>
{report ? (
<div className="mt-1.5 space-y-2">
<div className="flex items-center gap-2 flex-wrap">
{sevBadge(report.severity)}
{report.cvss != null && <span className="text-[#888] text-[13px]">CVSS {report.cvss}</span>}
{report.id && <span className="text-[#555] font-mono text-[13px]">{report.id}</span>}
{report.cve && <span className="text-[#888] font-mono text-[13px]">{report.cve}</span>}
{report.cwe && <span className="text-[#888] font-mono text-[13px]">{report.cwe}</span>}
{(report.agent_name || report.by_you) && (
<span className="text-[#666] text-[13px]">{report.by_you ? "you" : report.agent_name}</span>
)}
</div>
{report.title && <div className="text-[15px] text-white/80 font-semibold">{report.title}</div>}
{(report.target || report.endpoint) && (
<div className="text-[13px] text-[#888] font-mono">
{report.target}{report.endpoint ? ` ${report.method ?? ""} ${report.endpoint}` : ""}
</div>
)}
{report.description && <TruncatedText text={report.description} maxLines={20} />}
</div>
) : (
<div className="mt-1 text-[#555] text-xs">
{(res && typeof res === "object" && (res.error as string)) || "Report not found"}
</div>
)}
</div>
);
}
// list_reports
const rawReports = ok ? res.reports : null;
const reports: ReportEntry[] = Array.isArray(rawReports) ? (rawReports as ReportEntry[]) : [];
const total = ok && typeof res.total_count === "number" ? (res.total_count as number) : reports.length;
const counts = ok && res.severity_counts && typeof res.severity_counts === "object"
? (res.severity_counts as Record<string, number>)
: {};
const countEntries = Object.entries(counts);
return (
<div>
<div className="flex items-center gap-2 flex-wrap">
<span className="text-red-400/80 font-semibold text-sm">reports</span>
<span className="text-[#555] text-[13px]">({total})</span>
{countEntries.map(([sev, n]) => (
<span key={sev} className="text-[13px]">
{sevBadge(sev)}<span className="text-[#888] ml-0.5">{n}</span>
</span>
))}
</div>
{reports.length > 0 ? (
<div className="mt-1.5 space-y-1">
{reports.map((r, i) => (
<div key={r.id ?? i} className="text-[13px]">
<span className="text-[#555] mr-1">-</span>
{sevBadge(r.severity)}
{r.id && <span className="text-[#555] font-mono ml-1.5">{r.id}</span>}
<span className="text-[#999] ml-1.5">{r.title ?? "(untitled)"}</span>
{authorTag(r)}
{(r.target || r.endpoint) && (
<div className="ml-3 text-[#666] font-mono text-xs">
{r.target}{r.endpoint ? ` ${r.method ?? ""} ${r.endpoint}` : ""}
</div>
)}
{r.description_preview && (
<div className="ml-3"><Markdown text={r.description_preview} /></div>
)}
</div>
))}
</div>
) : <div className="mt-1 text-[#555] text-xs">No reports filed yet</div>}
</div>
);
}
@@ -12,6 +12,7 @@ import FileEditRenderer from "./FileEditRenderer";
import ApplyPatchRenderer from "./ApplyPatchRenderer";
import ViewImageRenderer from "./ViewImageRenderer";
import VulnReportRenderer from "./VulnReportRenderer";
import ReportListRenderer from "./ReportListRenderer";
import ProxyRenderer from "./ProxyRenderer";
import ThinkRenderer from "./ThinkRenderer";
import AgentCommsRenderer from "./AgentCommsRenderer";
@@ -101,7 +102,7 @@ const CATEGORY_TOOLS: Record<ToolCategory, readonly string[]> = {
filesystem: ["apply_patch", "view_image", "str_replace_editor", "list_files", "search_files"],
// Caido proxy tools (legacy: send_request)
proxy: ["list_requests", "view_request", "repeat_request", "list_sitemap", "view_sitemap_entry", "scope_rules", "send_request"],
reporting: ["create_vulnerability_report"],
reporting: ["create_vulnerability_report", "list_reports", "get_report"],
thinking: ["think"],
agents: ["create_agent", "agent_finish", "send_message_to_agent", "wait_for_message", "view_agent_graph", "stop_agent"],
search: ["web_search"],
@@ -128,6 +129,8 @@ const RENDERER_OVERRIDES: Partial<Record<string, ComponentType<ToolRendererProps
finish_scan: FinishRenderer,
apply_patch: ApplyPatchRenderer,
view_image: ViewImageRenderer,
list_reports: ReportListRenderer,
get_report: ReportListRenderer,
};
/**
@@ -5,7 +5,7 @@ import { fileURLToPath, URL } from "node:url";
// The viewer is served as static files by a stdlib Python server on an
// arbitrary ephemeral port, so all asset URLs must be relative (base: "./").
// The build output is committed at strix/viewer/static and shipped.
// The build output is committed at strix/interface/viewer/static and shipped.
export default defineConfig({
base: "./",
plugins: [react(), tailwindcss()],
@@ -38,7 +38,7 @@ from reportlab.platypus import (
TableStyle,
)
from strix.viewer.transcript import (
from strix.interface.viewer.transcript import (
primary_target,
read_run_summary,
read_vulnerabilities,
@@ -27,8 +27,8 @@ from typing import TYPE_CHECKING, Any
from urllib.parse import parse_qs, unquote, urlencode, urlsplit
from strix.core.paths import run_record_path
from strix.viewer import auth
from strix.viewer.transcript import (
from strix.interface.viewer import auth
from strix.interface.viewer.transcript import (
build_run_state,
primary_target,
read_report_markdown,
@@ -367,7 +367,7 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
self._send_json(HTTPStatus.CONFLICT, {"error": "run_not_finished"})
return
from strix.viewer.report_pdf import build_encrypted_report
from strix.interface.viewer.report_pdf import build_encrypted_report
pdf_bytes, password, filename = build_encrypted_report(run_dir)
run_name = str(summary.get("run_name") or run_dir.name)
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -6,8 +6,8 @@
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<meta name="color-scheme" content="dark" />
<title>Strix Results</title>
<script type="module" crossorigin src="./assets/index-Dd1cyttN.js"></script>
<link rel="stylesheet" crossorigin href="./assets/index-vV8wxCG6.css">
<script type="module" crossorigin src="./assets/index-DzvI_0HX.js"></script>
<link rel="stylesheet" crossorigin href="./assets/index-C3kQ5kk8.css">
</head>
<body>
<div id="root"></div>

Before

Width:  |  Height:  |  Size: 3.7 KiB

After

Width:  |  Height:  |  Size: 3.7 KiB

@@ -44,8 +44,8 @@ def build_run_state(run_dir: Path) -> dict[str, Any]:
Reuses the Textual-free ``TuiLiveView`` projection so the viewer and the TUI
share one parser for ``agents.json`` + ``agents.db`` and never drift.
"""
# Imported lazily so importing strix.viewer does not eagerly pull the TUI.
from strix.interface.tui.live_view import TuiLiveView # noqa: PLC0415
# Imported lazily so importing strix.interface.viewer does not eagerly pull the TUI.
from strix.interface.tui.live_view import TuiLiveView
view = TuiLiveView()
view.hydrate_from_run_dir(run_dir)
+1
View File
@@ -0,0 +1 @@
"""LLM-facing context management: model-aware budgets and history compaction."""
+354
View File
@@ -0,0 +1,354 @@
"""Provider-agnostic conversation compaction.
When an agent's session grows past the model's usable context window, older
turns are summarised into a single checkpoint while the most recent turns are
kept verbatim. This runs for every LiteLLM provider (not just OpenAI), keeps a
security-focused structured summary, and preserves tool-call/tool-result
pairing so the trimmed history is still valid provider input.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
import litellm
from litellm.exceptions import BadRequestError, ContextWindowExceededError
from strix.config import load_settings
from strix.core.sessions import replace_session_items, session_write_lock
from strix.llm.context_budget import context_window, count_tokens, output_limit
if TYPE_CHECKING:
from agents.memory import Session
logger = logging.getLogger(__name__)
_CHECKPOINT_TAG = "<conversation-checkpoint>"
_TOOL_OUTPUT_MAX_CHARS = 2_000
_MIN_ITEMS_TO_COMPACT = 6
_HEAD_TRUNCATED_MARKER = "\n\n[... older conversation omitted to fit the summary request ...]\n\n"
# Providers that don't type overflow errors (OpenRouter maps every 400 to a
# plain BadRequestError) leave only the message to go on, so we match it the way
# LiteLLM's own checker does — but with rate-limit exclusions first, so a
# throttling 429 is never mistaken for an overflow and sent into compaction.
_OVERFLOW_EXCLUSIONS = (
"rate limit",
"too many requests",
"throttling",
"service unavailable",
"quota",
)
_OVERFLOW_MARKERS = (
"context length",
"context window",
"context_length_exceeded",
"prompt is too long",
"input is too long",
"input length",
"maximum prompt length",
"reduce the length of the messages",
"too many tokens",
"token limit exceeded",
"request entity too large",
)
def is_context_overflow(exc: BaseException) -> bool:
"""Whether ``exc`` is a model context-window-overflow error.
LiteLLM types most providers' overflow as ContextWindowExceededError, but its
OpenRouter branch raises a plain BadRequestError, so for that we fall back to
matching the provider message.
"""
if isinstance(exc, ContextWindowExceededError):
return True
if isinstance(exc, BadRequestError):
msg = str(exc).lower()
if any(x in msg for x in _OVERFLOW_EXCLUSIONS):
return False
return any(x in msg for x in _OVERFLOW_MARKERS)
return False
_SUMMARY_INSTRUCTIONS = """\
You are compacting the earlier part of an autonomous security-testing agent's \
conversation so it fits the model context window. Produce a dense, factual \
record that lets the agent continue with no loss of important state.
This is a security engagement: dropped findings mean lost vulnerabilities. Be \
EXHAUSTIVE, not concise. Enumerate every distinct item as its own bullet — \
never merge, deduplicate, generalise, or omit distinct findings, credentials, \
or dead ends, even if they seem minor or repetitive. If the source mentions \
five vulnerabilities, list five. Copy exact values verbatim: URLs, endpoints, \
file paths, parameters, payloads, credentials, tokens, keys, hashes, cracked \
passwords, software versions, and error messages — never paraphrase or \
placeholder them. Do not invent anything and do not describe this compaction \
process.
Return Markdown with exactly these sections:
## Objective
The overall goal and target scope.
## Vulnerabilities & Findings
One bullet per DISTINCT vulnerability or finding (SQLi, XSS, SSRF, auth bypass, \
misconfig, etc.). For each: type, exact location (URL/endpoint/param/file), the \
verbatim payload or proof, confirmation status, and impact. List them all.
## Credentials & Secrets
One bullet per credential, secret, API key, token, hash, or cracked password, \
copied verbatim with where it applies. Write "(none)" only if truly none.
## System & Recon Details
Architecture, tech stack, versions, discovered endpoints/paths/params, and \
other weak points worth keeping.
## Work State
- Completed: what has been verified or finished.
- Active: what is in progress right now.
- Blocked: anything stuck and why.
## Failed Attempts & Dead Ends
One bullet per approach already tried that did not work (including WAF blocks, \
filtered inputs, non-exploitable leads) so they are not repeated. Write \
"(none)" only if truly none.
## Next Move
The concrete next step(s) the agent intended to take.
## Relevant Files
Files/notes/reports created or modified and their purpose."""
def _content_text(content: Any) -> str:
if isinstance(content, str):
return content
if isinstance(content, list):
parts: list[str] = []
for block in content:
if not isinstance(block, dict):
continue
text = block.get("text")
if isinstance(text, str):
parts.append(text)
elif block.get("type") in {"input_image", "image_url", "output_image"}:
parts.append("[image]")
return "\n".join(parts)
return ""
def _truncate(text: str, limit: int) -> str:
return text if len(text) <= limit else f"{text[:limit]}\n[truncated]"
def _serialize_item(item: Any) -> str:
if not isinstance(item, dict):
return str(item)
item_type = item.get("type")
role = item.get("role")
if item_type == "function_call":
args = _truncate(str(item.get("arguments", "")), _TOOL_OUTPUT_MAX_CHARS)
return f"[tool_call {item.get('name', '?')}] {args}"
if item_type == "function_call_output":
output = item.get("output")
text = output if isinstance(output, str) else _content_text(output)
return f"[tool_result] {_truncate(text, _TOOL_OUTPUT_MAX_CHARS)}"
if item_type == "reasoning":
return ""
if role or item_type == "message":
return f"[{role or 'assistant'}] {_content_text(item.get('content'))}".strip()
return ""
def _serialize_items(items: list[Any]) -> str:
return "\n".join(s for s in (_serialize_item(item) for item in items) if s)
def _is_tool_call(item: Any) -> bool:
return isinstance(item, dict) and item.get("type") == "function_call"
def _is_tool_output(item: Any) -> bool:
return isinstance(item, dict) and item.get("type") == "function_call_output"
def _open_calls_at(items: list[Any]) -> list[int]:
"""Prefix count of tool calls still awaiting their result at each index;
a split is only safe where this is zero."""
balance = [0] * (len(items) + 1)
for i, item in enumerate(items):
delta = 1 if _is_tool_call(item) else -1 if _is_tool_output(item) else 0
balance[i + 1] = max(0, balance[i] + delta)
return balance
def _select_split(model: str, items: list[Any], keep_tokens: int) -> int:
"""Index where the kept-verbatim recent tail begins: walk newest→oldest to
``keep_tokens``, then snap to a point with no tool call left open."""
total = 0
split = len(items)
for i in range(len(items) - 1, -1, -1):
total += count_tokens(model, _serialize_item(items[i]))
if total > keep_tokens:
break
split = i
open_calls = _open_calls_at(items)
while split > 0 and open_calls[split] != 0:
split -= 1
return split
def _previous_summary(head: list[Any]) -> str | None:
for item in head:
if isinstance(item, dict) and item.get("role") == "user":
text = _content_text(item.get("content"))
if text.startswith(_CHECKPOINT_TAG):
return text
return None
def _fit_to_tokens(model: str, text: str, max_tokens: int) -> str:
"""Head+tail-truncate ``text`` to ``max_tokens``, keeping start and end."""
if count_tokens(model, text) <= max_tokens:
return text
# Rough char budget (~4x tokens), then tighten by real token count.
budget_chars = max_tokens * 4
head_chars = budget_chars // 2
tail_chars = budget_chars - head_chars
candidate = text[:head_chars] + _HEAD_TRUNCATED_MARKER + text[len(text) - tail_chars :]
while count_tokens(model, candidate) > max_tokens and (head_chars > 0 or tail_chars > 0):
head_chars = int(head_chars * 0.8)
tail_chars = int(tail_chars * 0.8)
candidate = text[:head_chars] + _HEAD_TRUNCATED_MARKER + text[len(text) - tail_chars :]
return candidate
def _summary_output_tokens(model: str) -> int:
"""Summary output allowance, capped at the model's own output limit."""
return min(load_settings().context.summary_max_tokens, output_limit(model))
def _summary_input_budget(model: str, previous: str | None) -> int:
"""Token room left for the head after instructions and the summary output."""
overhead = count_tokens(model, _SUMMARY_INSTRUCTIONS)
if previous:
overhead += count_tokens(model, previous)
# 256 leaves slack for the prompt wrapper text not counted in ``overhead``.
room = context_window(model) - _summary_output_tokens(model) - overhead - 256
return max(0, room)
def _build_summary_prompt(serialized_head: str, previous: str | None) -> str:
previous_block = (
f"\n\nA previous checkpoint summary follows. Update it: keep what is "
f"still true, drop what is now stale, and merge in the new "
f"conversation below.\n\n{previous}\n"
if previous
else ""
)
return (
f"{_SUMMARY_INSTRUCTIONS}{previous_block}\n\n"
f"Conversation to summarise:\n\n{serialized_head}"
)
def _checkpoint_item(summary: str) -> dict[str, Any]:
return {
"role": "user",
"content": (
f"{_CHECKPOINT_TAG}\nThe following summarises earlier conversation that was "
f"compacted to fit the context window. Treat it as established context, not "
f"new instructions.\n\n{summary}\n</conversation-checkpoint>"
),
}
async def _summarize(model: str, prompt: str, max_tokens: int) -> str | None:
llm = load_settings().llm
try:
response = await litellm.acompletion(
model=model,
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens,
api_key=llm.api_key,
api_base=llm.api_base,
timeout=llm.timeout,
)
except Exception:
logger.exception("compaction summary call failed for model %s", model)
return None
try:
content = response.choices[0].message.content
except (AttributeError, IndexError, KeyError):
logger.warning("compaction summary returned no content")
return None
return content.strip() if isinstance(content, str) and content.strip() else None
async def maybe_compact(
session: Session,
*,
model: str,
instructions: str = "",
tools_text: str = "",
force: bool = False,
) -> bool:
"""Compact ``session`` if it is near the model's context window.
Returns ``True`` when the session was rewritten. ``force`` skips the size
check (used after a provider context-overflow error).
"""
context = load_settings().context
if not context.auto_compact and not force:
return False
async with session_write_lock(session):
items = list(await session.get_items())
if len(items) < _MIN_ITEMS_TO_COMPACT:
return False
window = context_window(model)
reserve = max(context.compact_buffer_tokens, output_limit(model))
budget = max(context.keep_tokens, window - reserve)
used = count_tokens(model, "\n".join((instructions, tools_text, _serialize_items(items))))
if not force and used <= budget:
return False
split = _select_split(model, items, context.keep_tokens)
head, recent = items[:split], items[split:]
previous = _previous_summary(head)
input_budget = _summary_input_budget(model, previous)
if not head or input_budget <= 0:
# Nothing to summarise, or no room for even the summary request itself.
if head:
logger.warning(
"skipping compaction for %s: no room to summarise within its context window", model
)
return False
serialized_head = _fit_to_tokens(model, _serialize_items(head), input_budget)
summary = await _summarize(
model,
_build_summary_prompt(serialized_head, previous),
_summary_output_tokens(model),
)
if summary is None:
return False
new_items = [_checkpoint_item(summary), *recent]
rewritten = await replace_session_items(session, new_items, expected_len=len(items))
if rewritten:
logger.info(
"compacted %s: %d items (~%d tok) -> %d items (summary + %d recent)",
model,
len(items),
used,
len(new_items),
len(recent),
)
return rewritten
+76
View File
@@ -0,0 +1,76 @@
"""Model-aware token budgets, resolved from LiteLLM model metadata with a
large configurable fallback for models LiteLLM doesn't map.
"""
from __future__ import annotations
import logging
from functools import lru_cache
from typing import Any
import litellm
from strix.config import load_settings
logger = logging.getLogger(__name__)
# LiteLLM keys models without the routing prefix users type (``openai/``,
# ``litellm/``, ``ollama/`` ...). Strip a leading provider segment on lookup.
_STRIPPABLE_PREFIXES = ("openai/", "litellm/", "any-llm/", "ollama/", "ollama_chat/")
_DEFAULT_OUTPUT_TOKENS = 8_192
def _lookup_key(model: str) -> str:
for prefix in _STRIPPABLE_PREFIXES:
if model.startswith(prefix):
return model[len(prefix) :]
return model
def _safe_get_model_info(model: str) -> dict[str, Any] | None:
try:
return dict(litellm.get_model_info(model))
except Exception: # noqa: BLE001 - unmapped models raise; caller falls back.
return None
@lru_cache(maxsize=128)
def _model_info(model: str) -> dict[str, int]:
for candidate in (model, _lookup_key(model)):
info = _safe_get_model_info(candidate)
if info is not None:
return {
"max_input_tokens": int(
info.get("max_input_tokens") or info.get("max_tokens") or 0
),
"max_output_tokens": int(info.get("max_output_tokens") or 0),
}
logger.debug("No LiteLLM model info for %r; using configured fallbacks", model)
return {"max_input_tokens": 0, "max_output_tokens": 0}
def context_window(model: str) -> int:
"""Input token capacity for ``model`` (configured fallback when unmapped)."""
resolved = _model_info(model)["max_input_tokens"]
return resolved or load_settings().context.fallback_context_tokens
def output_limit(model: str) -> int:
"""Max output tokens for ``model`` (a conservative default when unmapped)."""
return _model_info(model)["max_output_tokens"] or _DEFAULT_OUTPUT_TOKENS
def count_tokens(model: str, text: str) -> int:
"""Token count for ``text`` under ``model``.
Falls back to UTF-8 byte length (a guaranteed upper bound) when LiteLLM
can't count, so budget checks stay conservative.
"""
if not text:
return 0
try:
return int(litellm.token_counter(model=_lookup_key(model), text=text))
except Exception: # noqa: BLE001 - tokenizer may be unavailable for some models.
return len(text.encode("utf-8"))
+3 -3
View File
@@ -10,7 +10,7 @@ import re
import tempfile
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, cast
from pygments.lexers import PythonLexer, get_lexer_by_name, guess_lexer
from pygments.lexers.special import TextLexer
@@ -74,10 +74,10 @@ def resolve_lexer(language: str | None, code: str) -> Lexer:
try:
lexer = guess_lexer(code)
except ClassNotFound:
return PythonLexer()
return cast("Lexer", PythonLexer())
# ``guess_lexer`` returns the plain-text lexer when it can't detect anything.
if isinstance(lexer, TextLexer):
return PythonLexer()
return cast("Lexer", PythonLexer())
return lexer
+19 -20
View File
@@ -54,18 +54,17 @@ CT logs record nearly every publicly-trusted certificate. Query by domain (match
## Recommended Tooling
Prefer the projectdiscovery suite (already available in the sandbox and pipeline-friendly with JSON output):
These tools are available in the sandbox and are pipeline-friendly with JSON output:
- **`subfinder`** — passive subdomain aggregation across many sources incl. CT: `subfinder -d example.com -all -recursive -silent -oJ -o subs.jsonl`
- **`tlsx`** — TLS/cert data at scale; grab SANs and issuer/org to pivot: `tlsx -l hosts.txt -san -cn -tls-version -json -o tls.jsonl`
- **`uncover`** — query Shodan/Censys/Fofa/Quake/crt.sh engines from one CLI: `uncover -q 'ssl:"Example Inc"' -e shodan,censys,fofa -json`
- **`asnmap`** — org/domain/ASN → CIDR ranges: `asnmap -d example.com -json` / `asnmap -org "Example Inc"`
- **`mapcidr`** — expand/aggregate CIDRs into host lists for probing: `mapcidr -cidr 192.0.2.0/24 -o hosts.txt`
- **`dnsx`** — fast resolution, PTR, and wildcard filtering: `dnsx -l names.txt -a -aaaa -cname -ptr -resp -json -o dns.jsonl`
- **`httpx`** — live probing + cert grab in one pass (see methodology).
- **`httpx`** — live probing plus cert/SAN grab in one pass: `httpx -l hosts.txt -tls-grab -json` (see methodology).
- **`naabu`** — port sweep for non-HTTP services: `naabu -list hosts.txt -top-ports 100 -verify -silent`
- **`curl` + `jq`** — direct **crt.sh** JSON queries for CT (no key needed) and other index APIs.
- **`openssl s_client`** — active read of a live host's cert to extract SANs/CN.
- **`dig`** / **`nslookup`** — forward/reverse (PTR) resolution and CNAME chains.
- **`whois`** — ASN/netblock lookups (e.g. `whois -h whois.cymru.com`).
Also useful: **`amass`** (`amass intel`/`enum` for ASN, cert, and passive sources), **`cero`** (bulk SAN extraction from IPs/ranges), and direct **crt.sh** JSON queries when no keys are configured. Cross-source results — CT + passive DNS + `subfinder` together beat any single source.
Cross-source results — CT + passive DNS + `subfinder` together beat any single source. If you need a tool that is not installed, install it into the sandbox at runtime.
## Key Techniques
@@ -75,7 +74,7 @@ Every new name, PTR result, CNAME target, and cert SAN becomes a fresh seed. Loo
### Cert-Fingerprint Pivoting
Search Censys/Shodan (or `uncover`) by a cert's `fingerprint_sha256` to find every other host presenting the same certificate — the strongest cross-asset link for tying acquisitions and shadow infra to the target.
Search Censys/Shodan by a cert's `fingerprint_sha256` to find every other host presenting the same certificate — the strongest cross-asset link for tying acquisitions and shadow infra to the target.
### Naming-Convention Inference
@@ -83,11 +82,11 @@ Wildcard SANs and observed hostnames expose the org's naming scheme; generate ta
### IP-First Discovery
For ASN-owned ranges, sweep IPs directly with `naabu`/`httpx` and read served certs (`tlsx`) to find services that have no DNS name at all.
For ASN-owned ranges, sweep IPs directly with `naabu`/`httpx` and read served certs (`httpx -tls-grab`, or `openssl s_client`) to find services that have no DNS name at all.
## Advanced Techniques
- **Active SAN harvesting** across whole ranges with `tlsx`/`cero` recovers internal hostnames never logged to public CT.
- **Active SAN harvesting** across whole ranges with `httpx -tls-grab` (or `openssl s_client`) recovers internal hostnames never logged to public CT.
- **Favicon and response hashing** (`httpx -favicon`, hash pivots in Shodan) clusters instances of the same app across unrelated hostnames.
- **Vhost differentials**: probe a single IP with multiple `Host:` values to unmask co-located apps behind one address.
- **Historical CT/DNS diffing** highlights recently issued certs and newly appearing hosts — high-signal for fresh or misconfigured deployments.
@@ -108,11 +107,11 @@ For ASN-owned ranges, sweep IPs directly with `naabu`/`httpx` and read served ce
## Testing Methodology
1. **Seed** - domains, org/legal names, known IPs, email domains, code-host org
2. **Certificate transparency** - pull all logged certs per seed domain and org name (crt.sh, `uncover`)
3. **SAN/CN extraction** - parse every Subject CN and SAN with `tlsx`; each new name is a new seed
4. **Passive DNS** - resolve forward and reverse with `dnsx`; harvest historical records
5. **ASN/IP mapping** - `asnmap``mapcidr` to expand owned ranges, then sweep for live hosts
6. **Active TLS pivot** - `tlsx`/`cero` on live IPs/ports to grab SANs missing from public CT
2. **Certificate transparency** - pull all logged certs per seed domain and org name (crt.sh, Censys/Shodan)
3. **SAN/CN extraction** - parse every Subject CN and SAN with `httpx -tls-grab` (or `openssl s_client`); each new name is a new seed
4. **Passive DNS** - resolve forward and reverse with `dig`; harvest historical records
5. **ASN/IP mapping** - `whois` the netblock/ASN to expand owned ranges, then sweep for live hosts
6. **Active TLS pivot** - `httpx -tls-grab` on live IPs/ports to grab SANs missing from public CT
7. **Consolidate & probe** - dedupe, `httpx` probe, classify, and route to specialists
## Validation
@@ -139,13 +138,13 @@ For ASN-owned ranges, sweep IPs directly with `naabu`/`httpx` and read served ce
## Pro Tips
1. Loop the pipeline — every SAN, PTR, and CNAME target is a new seed until the set converges.
2. crt.sh is the cheapest high-yield source (no key); Censys/Shodan via `uncover` add cert-fingerprint and vhost pivoting when keys exist.
3. Always cert-grab live hosts with `tlsx` — active SANs catch internal hostnames never sent to public CT.
2. crt.sh is the cheapest high-yield source (no key); Censys/Shodan add cert-fingerprint and vhost pivoting when keys exist.
3. Always cert-grab live hosts with `httpx -tls-grab` (or `openssl s_client`) — active SANs catch internal hostnames never sent to public CT.
4. Internal-looking SANs (`*.internal`, `*.svc.cluster.local`, staging names) are the highest-signal leads.
5. Wildcard SANs reveal naming conventions — seed targeted guesses instead of blind brute force.
6. Cluster by function, not product name, so the workflow generalizes to any exposed service.
7. Keep JSON output throughout so stages chain cleanly (`subfinder``dnsx``httpx``naabu`).
7. Keep JSON output throughout so stages chain cleanly (`subfinder``dig``httpx``naabu`).
## Summary
Broad passive discovery — CT + TLS SAN pivoting + passive DNS + ASN/IP mapping, looped until convergence — finds the assets brute force misses, especially internal-named and forgotten services leaked through certificates. Build the inventory with the projectdiscovery suite, probe and classify it generically, then route each interesting asset to the specialist skill for its class.
Broad passive discovery — CT + TLS SAN pivoting + passive DNS + ASN/IP mapping, looped until convergence — finds the assets brute force misses, especially internal-named and forgotten services leaked through certificates. Build the inventory with `subfinder`, `httpx`, `naabu`, and CT/DNS/cert queries, probe and classify it generically, then route each interesting asset to the specialist skill for its class.
@@ -152,7 +152,7 @@ TLS clues: certificate CN/SAN referencing provider default host instead of the c
## Pro Tips
1. Build a pipeline: enumerate (subfinder/amass) → resolve (dnsx) → probe (httpx) → fingerprint (nuclei/custom) → verify claims
1. Build a pipeline: enumerate (subfinder) → resolve (dig) → probe (httpx) → fingerprint (nuclei/custom) → verify claims
2. Maintain a current fingerprint corpus; provider messages change frequently
3. Prefer minimal PoCs: static "ownership proof" page and, where allowed, DV cert issuance
4. Monitor CT for unexpected certs on your subdomains
@@ -1,6 +1,6 @@
---
name: weak-password-detection
description: Weak password detection, credential stuffing, and brute-force testing using common passwords, system-generated credentials, and tooling like Hydra
description: Weak password detection, credential stuffing, and brute-force testing using common passwords, system-generated credentials, and HTTP fuzzing / NSE brute-force tooling
---
# Weak Password Detection / Credential Brute-Force
@@ -98,7 +98,7 @@ Weak or default credentials remain one of the most prevalent and high-impact vul
- Season + year patterns: `Summer2025!`, `Winter2026@`
- Keyboard walks and leet speak variations
- Previously breached passwords for the target domain
- Cewl: `cewl -d 3 -m 5 -w custom.txt https://target.com` to generate from website content
- Scrape the target site to build a content-derived wordlist (e.g. a small custom Python crawler that harvests unique words)
### Credential Stuffing Workflows
@@ -123,37 +123,25 @@ Weak or default credentials remain one of the most prevalent and high-impact vul
### Service-Level Brute-Force
- SSH: `hydra -l admin -P passwords.txt ssh://target.com`
- FTP: `hydra -L users.txt -P passwords.txt ftp://target.com`
- RDP: `hydra -l administrator -P passwords.txt rdp://target.com`
- SMB: `hydra -L users.txt -P passwords.txt smb://target.com`
- Database: MySQL, PostgreSQL, MongoDB, Redis with weak credentials
- API endpoints: `ffuf` or custom scripts for HTTP-based brute-force
- HTTP login endpoints: `ffuf` or custom scripts (see Tooling)
- SSH/FTP/SMB/Telnet and other services: `nmap` NSE `*-brute` scripts, e.g. `nmap -p 22 --script ssh-brute --script-args userdb=users.txt,passdb=passwords.txt target.com`
- Databases (MySQL, PostgreSQL, MongoDB, Redis): weak/default credentials via the matching NSE brute script (`mysql-brute`, `pgsql-brute`, `mongodb-brute`, `redis-brute`) or a custom client script
- Any protocol lacking a ready script: custom Python
## Tooling
### Hydra (Primary Tool)
- HTTP POST form brute-force:
`hydra -l admin -P /usr/share/wordlists/rockyou.txt target.com http-post-form "/login:username=^USER^&password=^PASS^:Invalid credentials"`
- Basic Auth:
`hydra -L users.txt -P passwords.txt target.com http-get -s 8080 /admin`
- SSH:
`hydra -l root -P passwords.txt -t 4 ssh://target.com`
- FTP:
`hydra -L users.txt -P passwords.txt ftp://target.com`
- Custom headers and cookies:
`hydra ... http-post-form "/api/login:json={\"user\":\"^USER^\",\"pass\":\"^PASS^\"}:F=401"`
### ffuf (HTTP Fuzzing)
### ffuf (primary for web logins)
- Login brute-force with multiple users and passwords:
`ffuf -w users.txt:USER -w passwords.txt:PASS -u https://target.com/login -X POST -d "username=USER&password=PASS" -fr "Invalid"`
- JSON body / custom headers via `-H` and a JSON `-d` payload
- Filter by response size, status code, or regex to identify successes
### Patator (Versatile Brute-Force)
### nmap NSE (service brute-force)
- `patator http_fuzz url=https://target.com/login method=POST body='username=FILE0&password=FILE1' 0=user.txt 1=pass.txt -x ignore:fgrep='Invalid'`
- `*-brute` scripts cover many non-HTTP services:
`nmap -p 22 --script ssh-brute --script-args userdb=users.txt,passdb=passwords.txt target.com`
- Available scripts include `ssh-brute`, `ftp-brute`, `smb-brute`, `telnet-brute`, `mysql-brute`, `pgsql-brute`, `mongodb-brute`, `redis-brute`, `http-brute`, `http-form-brute`.
### Custom Python Scripts
@@ -163,10 +151,10 @@ Weak or default credentials remain one of the most prevalent and high-impact vul
### Wordlists
- `/usr/share/wordlists/rockyou.txt` (common passwords)
- `/usr/share/seclists/Passwords/` (organized by category)
- `/usr/share/seclists/Passwords/Default-Credentials/` (vendor defaults)
- Custom lists from Cewl, CeWL, or target-specific scraping
No password wordlists ship in the sandbox by default — download what you need into `/home/pentester/tools/wordlists` at runtime:
- Common passwords (e.g. `rockyou.txt`) from its upstream source
- SecLists `Passwords/` and `Passwords/Default-Credentials/` (vendor defaults) from https://github.com/danielmiessler/SecLists
- Custom lists from target-specific scraping
- Breach compilation subsets filtered by target relevance
## Validation
@@ -204,7 +192,7 @@ Weak or default credentials remain one of the most prevalent and high-impact vul
6. Check for concurrent session limits; successful logins may kick out legitimate users
7. GraphQL batching can test multiple credentials in a single request, bypassing per-request limits
8. Document the password policy and recommend minimum standards (length, complexity, breach checking)
9. When Hydra is unavailable, use ffuf or custom scripts with equivalent logic
9. For web logins prefer `ffuf`; for other services use `nmap` NSE `*-brute` scripts or custom scripts with equivalent logic
10. Combine with MFA testing: weak passwords plus missing MFA is a critical finding
## Summary

Some files were not shown because too many files have changed in this diff Show More