feat(skills): add ReDoS skill with recheck-assisted detection

strix/core/runner.py

@@ -8,7 +8,7 @@ import io
 import json
 import logging
 import uuid
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
@@ -203,7 +203,7 @@ async def run_strix_scan(
     scan_id: str | None = None,
     image: str,
     local_sources: list[dict[str, Any]] | None = None,
-    extra_files: list[dict[str, Any]] | None = None,
+    extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None = None,
     coordinator: AgentCoordinator | None = None,
     interactive: bool = False,
     max_turns: int = DEFAULT_MAX_TURNS,

strix/report/coverage.py

@@ -124,6 +124,7 @@ _SKILL_PHRASINGS: dict[str, tuple[str, ...]] = {
     "prototype_pollution": ("prototype pollution",),
     "race_conditions": ("race condition", "toctou"),
     "rce": ("rce", "remote code execution", "code execution", "command injection"),
+    "redos": ("redos", "regex denial", "regular expression denial", "catastrophic backtracking"),
     "semantic_confusion": (
         "semantic confusion",
         "parser differential",

strix/runtime/deferred_session.py

@@ -0,0 +1,255 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from agents.sandbox.session.manager import Instrumentation
+from agents.sandbox.session.sandbox_session import SandboxSession
+from agents.sandbox.session.sandbox_session_state import SandboxSessionState
+from agents.sandbox.snapshot import NoopSnapshot
+
+
+if TYPE_CHECKING:
+    import io
+    from collections.abc import Sequence
+
+    from agents.sandbox.entries import BaseEntry
+    from agents.sandbox.manifest import Manifest
+    from agents.sandbox.materialization import MaterializedFile
+    from agents.sandbox.session.base_sandbox_session import BaseSandboxSession
+    from agents.sandbox.types import ExecResult, ExposedPortEndpoint, User
+
+
+logger = logging.getLogger(__name__)
+
+
+class _PendingSessionState(SandboxSessionState):
+    type: str = "pending"
+
+
+class DeferredSandboxSession(SandboxSession):
+    def __init__(
+        self,
+        bring_up: asyncio.Task[tuple[Any, BaseSandboxSession]],
+        *,
+        manifest: Manifest,
+        exposed_ports: tuple[int, ...] = (),
+    ) -> None:
+        self._bring_up = bring_up
+        self._pending_state: SandboxSessionState = _PendingSessionState(
+            snapshot=NoopSnapshot(id="pending"),
+            manifest=manifest,
+            exposed_ports=exposed_ports,
+        )
+        self._pending_limits: Any = None
+        self._pending_archive_limits: Any = None
+        self._resolved: BaseSandboxSession | None = None
+        self._client: Any = None
+        self._instrumentation: Instrumentation = Instrumentation()
+        self._seq = 0
+
+    @property
+    def pending(self) -> bool:
+        return self._resolved is None
+
+    @property
+    def bring_up(self) -> asyncio.Task[tuple[Any, BaseSandboxSession]]:
+        return self._bring_up
+
+    async def resolve(self) -> BaseSandboxSession:
+        if self._resolved is not None:
+            return self._resolved
+        client, session = await asyncio.shield(self._bring_up)
+        self._adopt(client, session)
+        return session
+
+    def _adopt(self, client: Any, session: BaseSandboxSession) -> None:
+        if self._resolved is not None:
+            return
+        self._client = client
+        self._resolved = session
+        self._inner = session
+        if self._pending_limits is not None:
+            session._set_concurrency_limits(self._pending_limits)
+        if self._pending_archive_limits is not None:
+            session._set_archive_limits(self._pending_archive_limits)
+        if self._pending_state.manifest != session.state.manifest:
+            session.state = session.state.model_copy(
+                update={"manifest": self._pending_state.manifest}
+            )
+
+    @property
+    def client(self) -> Any:
+        return self._client
+
+    @property
+    def state(self) -> SandboxSessionState:
+        if self._resolved is None:
+            return self._pending_state
+        return self._resolved.state
+
+    @state.setter
+    def state(self, value: SandboxSessionState) -> None:
+        if self._resolved is None:
+            self._pending_state = value
+        else:
+            self._resolved.state = value
+
+    def _set_concurrency_limits(self, limits: Any) -> None:
+        if self._resolved is None:
+            self._pending_limits = limits
+        else:
+            self._resolved._set_concurrency_limits(limits)
+
+    def _set_archive_limits(self, limits: Any) -> None:
+        if self._resolved is None:
+            self._pending_archive_limits = limits
+        else:
+            self._resolved._set_archive_limits(limits)
+
+    def _manifest_base_dir(self) -> Path:
+        if self._resolved is None:
+            return Path.cwd()
+        return self._resolved._manifest_base_dir()
+
+    def normalize_path(self, path: Path | str, *, for_write: bool = False) -> Path:
+        if self._resolved is None:
+            return Path(path)
+        return self._resolved.normalize_path(path, for_write=for_write)
+
+    def register_persist_workspace_skip_path(self, path: Path | str) -> Path:
+        if self._resolved is None:
+            return Path(path)
+        return self._resolved.register_persist_workspace_skip_path(path)
+
+    def supports_pty(self) -> bool:
+        return self._resolved.supports_pty() if self._resolved is not None else False
+
+    async def running(self) -> bool:
+        if self._resolved is None:
+            if not self._bring_up.done():
+                return True
+            return not self._bring_up.cancelled() and self._bring_up.exception() is None
+        return await self._resolved.running()
+
+    async def start(self) -> None:
+        if self._resolved is None:
+            return
+        await self._resolved.start()
+
+    async def stop(self) -> None:
+        if self._resolved is None:
+            return
+        await self._resolved.stop()
+
+    async def shutdown(self) -> None:
+        if self._resolved is None:
+            return
+        await self._resolved.shutdown()
+
+    async def aclose(self) -> None:
+        if self._resolved is None:
+            return
+        await self._resolved.aclose()
+
+    async def _validate_manifest_application(self, *, only_ephemeral: bool = False) -> None:
+        session = await self.resolve()
+        await session._validate_manifest_application(only_ephemeral=only_ephemeral)
+
+    async def apply_manifest(self, *, only_ephemeral: bool = False) -> Any:
+        session = await self.resolve()
+        return await session.apply_manifest(only_ephemeral=only_ephemeral)
+
+    async def _apply_entry_batch(
+        self,
+        entries: Sequence[tuple[Path, BaseEntry]],
+        *,
+        base_dir: Path,
+    ) -> list[MaterializedFile]:
+        del base_dir
+        session = await self.resolve()
+        return await session._apply_entry_batch(entries, base_dir=session._manifest_base_dir())
+
+    async def exec(self, *command: str | Path, **kwargs: Any) -> ExecResult:
+        await self.resolve()
+        return await super().exec(*command, **kwargs)
+
+    async def ls(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().ls(*args, **kwargs)
+
+    async def rm(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().rm(*args, **kwargs)
+
+    async def mkdir(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().mkdir(*args, **kwargs)
+
+    async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase:
+        await self.resolve()
+        return await super().read(path, user=user)
+
+    async def write(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().write(*args, **kwargs)
+
+    async def resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
+        await self.resolve()
+        return await super().resolve_exposed_port(port)
+
+    async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
+        session = await self.resolve()
+        return await session._resolve_exposed_port(port)
+
+    async def pty_exec_start(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().pty_exec_start(*args, **kwargs)
+
+    async def pty_write_stdin(self, *args: Any, **kwargs: Any) -> Any:
+        await self.resolve()
+        return await super().pty_write_stdin(*args, **kwargs)
+
+    async def pty_terminate_all(self) -> None:
+        if self._resolved is None:
+            return
+        await super().pty_terminate_all()
+
+    async def persist_workspace(self) -> io.IOBase:
+        await self.resolve()
+        return await super().persist_workspace()
+
+    async def hydrate_workspace(self, data: io.IOBase) -> None:
+        await self.resolve()
+        await super().hydrate_workspace(data)
+
+    async def teardown(self) -> None:
+        if self._resolved is None:
+            self._bring_up.cancel()
+            with contextlib.suppress(BaseException):
+                client, session = await self._bring_up
+                self._adopt(client, session)
+        if self._resolved is None:
+            return
+        await self._client.delete(self._resolved)
+
+
+class DeferredSandboxClient:
+    def __init__(self, session: DeferredSandboxSession) -> None:
+        self._session = session
+
+    def __getattr__(self, name: str) -> Any:
+        client = self._session.client
+        if client is None:
+            raise AttributeError(f"sandbox client unavailable before bring-up finishes ({name})")
+        return getattr(client, name)
+
+    async def delete(self, session: Any) -> None:
+        if isinstance(session, DeferredSandboxSession):
+            await session.teardown()
+            return
+        real = await self._session.resolve()
+        await self._session.client.delete(real if session is self._session else session)

strix/runtime/session_manager.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
+import inspect
 import io
 import logging
 import os
@@ -19,9 +21,12 @@ from strix.config import load_settings
 from strix.runtime.backends import backend_supports_bind_mounts, get_backend
 from strix.runtime.caido_bootstrap import bootstrap_caido
 from strix.runtime.caido_handle import CaidoBootstrapHandle
+from strix.runtime.deferred_session import DeferredSandboxClient, DeferredSandboxSession
 
 
 if TYPE_CHECKING:
+    from collections.abc import Awaitable
+
     from agents.sandbox.session import BaseSandboxSession
 
     from strix.runtime.status import StatusSink
@@ -255,13 +260,22 @@ def _gitdir_from_pointer(git_file: Path) -> Path | None:
     return None
 
 
+async def _resolve_extra_file_archive(
+    extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None,
+    local_sources: list[dict[str, Any]],
+) -> bytes | None:
+    files = await extra_files if inspect.isawaitable(extra_files) else extra_files
+    return build_extra_file_archive(files, local_sources) if files else None
+
+
 async def create_or_reuse(
     scan_id: str,
     *,
     image: str,
     local_sources: list[dict[str, Any]],
-    extra_files: list[dict[str, Any]] | None = None,
+    extra_files: list[dict[str, Any]] | Awaitable[list[dict[str, Any]] | None] | None = None,
     status_sink: StatusSink | None = None,
+    defer: bool = False,
 ) -> dict[str, Any]:
     """Return the existing session bundle for ``scan_id`` or create a new one.
 
@@ -271,7 +285,9 @@ async def create_or_reuse(
     Each ``extra_files`` entry (``{"workspace_path": "/workspace/<rel>",
     "content": bytes | str}``) lands as a regular, agent-writable file at its
     ``workspace_path`` on every backend: the files are uploaded as one archive
-    and unpacked inside the sandbox right after bring-up.
+    and unpacked inside the sandbox right after bring-up. ``extra_files`` may
+    also be an awaitable resolving to that list: provisioning starts at once and
+    the files are awaited only when the sandbox is ready to receive them.
     """
 
     def report(phase: str) -> None:
@@ -292,9 +308,6 @@ async def create_or_reuse(
     else:
         bind_mounts = []
         entries = build_manifest_entries(local_sources)
-    extra_file_archive = (
-        build_extra_file_archive(extra_files, local_sources) if extra_files else None
-    )
 
     # Caido runs as an in-container sidecar; HTTP(S) traffic from any
     # process started via ``session.exec`` (the SDK's Shell tool, etc.)
@@ -323,41 +336,69 @@ async def create_or_reuse(
         backend_name,
         image,
     )
-    report("Starting sandbox container")
-    client, session = await backend(
-        image=image,
-        manifest=manifest,
-        exposed_ports=(_CONTAINER_CAIDO_PORT,),
-        bind_mounts=bind_mounts,
-    )
 
-    if extra_file_archive is not None:
-        report("Placing workspace files")
+    async def _bring_up() -> tuple[Any, Any]:
+        report("Starting sandbox container")
+        client, session = await _create_or_discard(
+            backend(
+                image=image,
+                manifest=manifest,
+                exposed_ports=(_CONTAINER_CAIDO_PORT,),
+                bind_mounts=bind_mounts,
+            )
+        )
         try:
-            await stage_extra_files(session, extra_file_archive)
+            extra_file_archive = await _resolve_extra_file_archive(extra_files, local_sources)
+            if extra_file_archive is not None:
+                report("Placing workspace files")
+                await stage_extra_files(session, extra_file_archive)
         except BaseException:
             await _discard_session(client, session)
             raise
+        return client, session
 
-    report("Setting up the proxy")
-    caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
-    scheme = "https" if caido_endpoint.tls else "http"
-    host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
-    logger.debug("Caido host endpoint resolved: %s", host_caido_url)
+    async def _caido_url(session: Any) -> str:
+        report("Setting up the proxy")
+        caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
+        scheme = "https" if caido_endpoint.tls else "http"
+        host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"
+        logger.debug("Caido host endpoint resolved: %s", host_caido_url)
+        return host_caido_url
 
-    # The Caido login + project setup polls the guest for a couple of seconds
-    # and nothing needs the client before the first proxy tool call, so it
-    # runs concurrently with the rest of scan start; consumers resolve the
-    # handle at first use (see CaidoBootstrapHandle).
-    caido_client = CaidoBootstrapHandle(
-        asyncio.create_task(
-            bootstrap_caido(
-                session,
-                host_url=host_caido_url,
-                container_url=container_caido_url,
-            ),
-            name=f"caido-bootstrap-{scan_id}",
+    def _caido(session: Any, host_caido_url: str) -> Any:
+        return bootstrap_caido(
+            session,
+            host_url=host_caido_url,
+            container_url=container_caido_url,
         )
+
+    if defer:
+        bring_up = asyncio.create_task(_bring_up(), name=f"sandbox-bring-up-{scan_id}")
+        deferred = DeferredSandboxSession(
+            bring_up,
+            manifest=manifest,
+            exposed_ports=(_CONTAINER_CAIDO_PORT,),
+        )
+
+        async def _caido_when_up() -> Any:
+            session = await deferred.resolve()
+            return await _caido(session, await _caido_url(session))
+
+        bundle = {
+            "client": DeferredSandboxClient(deferred),
+            "session": deferred,
+            "caido_client": CaidoBootstrapHandle(
+                asyncio.create_task(_caido_when_up(), name=f"caido-bootstrap-{scan_id}")
+            ),
+        }
+        _SESSION_CACHE[scan_id] = bundle
+        logger.info("Sandbox session for scan %s deferred and cached", scan_id)
+        return bundle
+
+    client, session = await _bring_up()
+    host_caido_url = await _caido_url(session)
+    caido_client = CaidoBootstrapHandle(
+        asyncio.create_task(_caido(session, host_caido_url), name=f"caido-bootstrap-{scan_id}")
     )
 
     bundle = {
@@ -370,6 +411,26 @@ async def create_or_reuse(
     return bundle
 
 
+_DISCARD_TASKS: set[asyncio.Task[None]] = set()
+
+
+async def _create_or_discard(create: Awaitable[tuple[Any, Any]]) -> tuple[Any, Any]:
+    pending = asyncio.ensure_future(create)
+    try:
+        return await asyncio.shield(pending)
+    except asyncio.CancelledError:
+        discard = asyncio.create_task(_discard_when_created(pending))
+        _DISCARD_TASKS.add(discard)
+        discard.add_done_callback(_DISCARD_TASKS.discard)
+        raise
+
+
+async def _discard_when_created(create: asyncio.Future[tuple[Any, Any]]) -> None:
+    with contextlib.suppress(BaseException):
+        client, session = await create
+        await _discard_session(client, session)
+
+
 async def _discard_session(client: Any, session: Any) -> None:
     """Best-effort teardown of a session that never made it into the cache."""
     try:

strix/skills/vulnerabilities/redos.md

@@ -0,0 +1,129 @@
+---
+name: redos
+description: Regular expression denial of service (ReDoS, CWE-1333) in backtracking regex engines, covering exponential and polynomial patterns, recheck-assisted detection, growth-curve confirmation, and severity calibration
+---
+
+# ReDoS (Regular Expression Denial of Service)
+
+A backtracking regex engine can take exponential or polynomial time on a crafted input that almost matches. When attacker-controlled text reaches such a pattern, one request or one document can pin a CPU core for seconds to hours. Tracked as CWE-1333 (Inefficient Regular Expression Complexity).
+
+The textbook shape, nested quantifiers like `(a+)+`, is the rarest one in real code. Most real ReDoS is polynomial: two adjacent pieces that can match the same characters, or an unanchored pattern that fails slowly at every start position. Do not rule a pattern out because it has no nested quantifier.
+
+## Engines
+
+Backtracking (vulnerable): Python `re` and `regex`, JavaScript (V8, JavaScriptCore, SpiderMonkey), Java `java.util.regex`, .NET (default mode), PCRE/PHP, Perl, Ruby before 3.2.
+
+Linear-time (not vulnerable to ReDoS): Go `regexp`, Rust `regex`, RE2 and its bindings (`google-re2`, `re2j`), .NET `RegexOptions.NonBacktracking`. Ruby 3.2+ memoizes most patterns and has `Regexp.timeout`; treat it as lower risk, not immune.
+
+PHP is special: when `pcre.backtrack_limit` is hit, `preg_match` returns `false` instead of hanging. That turns ReDoS into a logic bypass when the caller treats `false` as "no match", e.g. a validator or WAF rule that silently passes.
+
+## Vulnerable Shapes
+
+**Nested quantifiers (exponential)**: `^(a+)+$`, `^(a*)*$`, `^(\w+\s?)+$`. About 30 matching characters plus one that fails the anchor can hang.
+
+**Overlapping alternation inside a repeat (exponential)**: `^(a|a)*$`, `^(\w|\d)+$`, `^(.*,)*$`.
+
+**Adjacent overlapping quantifiers (polynomial)**: two consecutive pieces that can both match the same character, followed by something that fails. The engine tries every way to split the run between them.
+
+~~~text
+\d+\d+x          a run of digits split between two \d+, with no x after it (cubic: also retried at every start)
+^`{3,}[^\n]*\n   a run of backticks split between the fence and [^\n]*, with no newline after it
+~~~
+
+**Unanchored pattern that fails slowly at every start (polynomial)**: `<[^>]+>` on `<<<<…` with no `>`, or `\s+$` on a long run of spaces followed by one non-space character. Each start position scans to the end, and `sub`/`findall`/`search` retry at every position.
+
+**Lazy scan with an absent terminator inside a loop**: `.*?END` or `[\s\S]*?` applied repeatedly to text that never contains `END`.
+
+**Backreferences**: `^(a+)\1$` is quadratic: the engine tries every split of the run between the group and its repeat.
+
+## Where to Look
+
+- Validators: email, URL, hostname, semver, UUID, phone, password rules
+- Parsers and sanitizers: markdown, MDX, HTML stripping, template engines, log and CSV parsers, user-agent parsing
+- Routing and matching: route patterns, CORS origin checks, WAF and content-filter rules, glob-to-regex conversions
+- Patterns built from input: `re.compile(user_value)`, search filters, MongoDB `$regex`, Elasticsearch `regexp` queries (these are regex injection, the attacker supplies the pattern itself)
+- Document and RAG loaders that clean uploaded or fetched files: one large file per request is an easy delivery path
+
+Find candidate patterns in source:
+
+```bash
+rg -n --type py "re\.(compile|match|search|fullmatch|sub|subn|findall|finditer|split)\(" .
+rg -n --type js --type ts "new RegExp\(|/[^/\n]+/[dgimsuyv]*\.(test|exec)|\.(match|matchAll|replace|replaceAll|search|split)\(/" .
+```
+
+## Detect with recheck
+
+[recheck](https://github.com/makenowjust-labs/recheck) combines static analysis and fuzzing, reports the complexity class (linear, polynomial with degree, exponential), and produces an attack string. It catches polynomial cases that nested-quantifier heuristics and some tools miss. Install a pinned version at scan time:
+
+```bash
+mkdir -p /tmp/recheck && cd /tmp/recheck && npm install --silent [email protected]
+```
+
+Put the patterns in a JSON file, one `[source, flags]` pair per entry, so no shell escaping is involved:
+
+```js
+// /tmp/recheck/run.mjs   usage: node run.mjs patterns.json
+import { check } from "recheck";
+import { readFileSync } from "node:fs";
+for (const [source, flags] of JSON.parse(readFileSync(process.argv[2], "utf8"))) {
+  const r = await check(source, flags, { timeout: 30000 });
+  console.log(JSON.stringify({ source, status: r.status, complexity: r.complexity?.summary, attack: r.attack?.pattern }));
+}
+```
+
+recheck understands JavaScript syntax only. Translate other dialects first:
+
+| Python | JavaScript |
+|--------|-----------|
+| `(?P<name>…)` | `(?<name>…)` |
+| `(?P=name)` | `\k<name>` |
+| `\Z` | `$(?![\s\S])` |
+| `\A` | `^` without the `m` flag |
+| inline `(?m)`, `(?s)`, `(?i)` | `m`, `s`, `i` in the flags string |
+| `(?>…)`, `a*+` (3.11+) | no equivalent; recheck will over-report, so confirm in Python |
+
+Treat recheck's verdict as a lead and its attack string as a hint. The attack string can rely on JavaScript-only behavior and run fast in the target language: with the `m` flag, JavaScript `^` also matches after `\r`, `
`, and `
`, while Python `^` only matches after `\n`. Rebuild the attack for the real engine. If `npm install` fails (no network), reason from the shapes above and go straight to timing.
+
+## Confirm in the Target Language
+
+One timing proves nothing. Measure growth: double or 10x the input and compare times. Linear grows about 10x per 10x, quadratic about 100x, cubic about 1000x, exponential explodes after a few dozen characters.
+
+```python
+import re, time
+pattern = re.compile(r"(?ms)^(?P<fence>`{3,}|~{3,})[^\n]*\n.*?(?:^(?P=fence)[ \t]*$|\Z)")
+for n in (1_000, 10_000, 100_000):
+    payload = "`" * n   # the repeated character, with the required terminator missing
+    t = time.perf_counter(); pattern.sub("", payload); print(n, f"{time.perf_counter() - t:.3f}s")
+```
+
+Build the payload as: a prefix that reaches the vulnerable piece, then many copies of the character both quantifiers accept, then a suffix that makes the overall match fail (or no suffix, when the failing piece needs a character that never arrives). Run the confirmation against the function the attacker reaches, not just the bare regex, so input caps and preprocessing are included.
+
+## Severity Calibration
+
+Report the time at a realistic input size, not the asymptotic class alone. Ask:
+
+- Can the attacker supply the text, and how large can it be? Body size limits, field length validation, and upload caps decide whether 100 KB (seconds for quadratic) is reachable.
+- What does one slow match block? A Node.js event loop or a single-threaded worker stalls every user (higher). A per-request thread with a timeout affects one request (lower).
+- Is the work repeatable without authentication or rate limits?
+- Diff or PR review: compare against the base code. A new slow pattern next to an existing pattern with the same complexity on the same input does not add an attack; say so and calibrate down instead of reporting a new HIGH.
+
+Exponential on short unauthenticated input in a shared event loop is high. Quadratic needing megabytes behind authentication is usually low or informational.
+
+## False Positives
+
+- Linear-time engine (Go, Rust, RE2, .NET NonBacktracking)
+- Input length capped well below where the curve hurts
+- recheck flagged a construct the target dialect handles atomically (possessive quantifiers, atomic groups)
+- Only the developer controls the input (build scripts, static config)
+
+## Remediation
+
+- Make overlapping pieces unambiguous: atomic groups (`` (?>`{3,}) ``; Python 3.11+, Java, PCRE, .NET) or possessive quantifiers (`a++`; Python 3.11+, Java, PCRE), or rewrite so adjacent pieces cannot match the same character
+- Anchor patterns and bound repeats (`{1,256}`) where the format allows
+- Cap input length before matching
+- Switch to a linear-time engine (`google-re2`, Rust/Go regex) for untrusted input
+- Use engine timeouts where available: .NET `matchTimeout`, Ruby `Regexp.timeout`, Python `regex` module `timeout=`
+
+## Summary
+
+Look for adjacent pieces that can match the same character, not just nested quantifiers. Use recheck to flag candidates and get a starting attack string, then rebuild and time the attack in the target language at 1k, 10k, and 100k characters. Calibrate severity by reachable input size, what a stalled match blocks, and, in a PR, whether the base code was already as slow.

strix/tools/reporting/tool.py

@@ -1876,10 +1876,6 @@ def _build_dependency_evidence(
         evidence += f"\n\n**Usage analysis:** {label}."
         if reachability_evidence and reachability_evidence.strip():
             evidence += f" {reachability_evidence.strip()}"
-        evidence += (
-            " This is a prioritization signal from static analysis, not a"
-            " proof of exploitability or of safety."
-        )
     return evidence
 
 

tests/test_deferred_session.py

@@ -0,0 +1,316 @@
+from __future__ import annotations
+
+import asyncio
+import io
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+import pytest
+from agents.run_config import SandboxRunConfig
+from agents.sandbox import SandboxAgent
+from agents.sandbox.capabilities import Filesystem, Shell
+from agents.sandbox.runtime_session_manager import SandboxRuntimeSessionManager
+
+from strix.runtime import session_manager
+from strix.runtime.backends import _BACKENDS, _BIND_MOUNT_BACKENDS, register_backend
+from strix.runtime.deferred_session import DeferredSandboxClient, DeferredSandboxSession
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+@dataclass
+class _RuntimeSettings:
+    backend: str
+
+
+@dataclass
+class _Settings:
+    runtime: _RuntimeSettings
+
+
+@dataclass
+class _Endpoint:
+    host: str = "127.0.0.1"
+    port: int = 8080
+    tls: bool = False
+
+
+@dataclass
+class _ExecResult:
+    exit_code: int = 0
+    stdout: bytes = b""
+    stderr: bytes = b""
+
+    def ok(self) -> bool:
+        return self.exit_code == 0
+
+
+class _State:
+    def __init__(self) -> None:
+        self.manifest: Any = None
+        self.session_id = "00000000-0000-4000-8000-000000000001"
+
+    def model_copy(self, *, update: dict[str, Any]) -> _State:
+        copy = _State()
+        copy.manifest = update.get("manifest", self.manifest)
+        return copy
+
+
+class _Session:
+    def __init__(self) -> None:
+        self.state = _State()
+        self.execs: list[tuple[str, ...]] = []
+        self.writes: list[tuple[Path, bytes]] = []
+        self.limits: list[Any] = []
+
+    def _set_concurrency_limits(self, limits: Any) -> None:
+        self.limits.append(limits)
+
+    def _set_archive_limits(self, limits: Any) -> None:
+        self.limits.append(limits)
+
+    async def running(self) -> bool:
+        return True
+
+    async def resolve_exposed_port(self, _port: int) -> _Endpoint:
+        return _Endpoint()
+
+    async def exec(self, *argv: str, timeout: float | None = None, **_kw: Any) -> _ExecResult:
+        del timeout
+        self.execs.append(argv)
+        return _ExecResult()
+
+    async def write(self, path: Path, data: io.IOBase, **_kw: Any) -> None:
+        self.writes.append((path, data.read()))
+
+
+class _Client:
+    def __init__(self) -> None:
+        self.deleted: list[Any] = []
+
+    async def delete(self, session: Any) -> None:
+        self.deleted.append(session)
+
+
+async def _no_caido(*_args: Any, **_kwargs: Any) -> None:
+    return None
+
+
+class _Backend:
+    def __init__(self, *, fail: bool = False) -> None:
+        self.gate = asyncio.Event()
+        self.calls = 0
+        self.fail = fail
+        self.client = _Client()
+        self.session = _Session()
+
+    async def __call__(self, **_kwargs: Any) -> tuple[Any, Any]:
+        self.calls += 1
+        await self.gate.wait()
+        if self.fail:
+            raise RuntimeError("bring-up exploded")
+        return self.client, self.session
+
+
+@pytest.fixture
+def backend(monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest) -> Iterator[_Backend]:
+    name = f"deferred-{request.node.name}"
+    fake = _Backend(fail="fail" in request.node.name)
+    register_backend(name, fake, supports_bind_mounts=True)
+    monkeypatch.setattr(session_manager, "bootstrap_caido", _no_caido)
+    monkeypatch.setattr(
+        session_manager, "load_settings", lambda: _Settings(runtime=_RuntimeSettings(backend=name))
+    )
+    yield fake
+    _BACKENDS.pop(name, None)
+    _BIND_MOUNT_BACKENDS.discard(name)
+
+
+async def _settle() -> None:
+    for _ in range(10):
+        await asyncio.sleep(0)
+
+
+async def _create(scan_id: str, **kwargs: Any) -> dict[str, Any]:
+    return await session_manager.create_or_reuse(
+        scan_id, image="img", local_sources=[], defer=True, **kwargs
+    )
+
+
+@pytest.mark.asyncio
+async def test_deferred_bundle_returns_before_the_backend_finishes(backend: _Backend) -> None:
+    scan_id = "deferred-returns"
+    try:
+        bundle = await asyncio.wait_for(_create(scan_id), timeout=1)
+        session = bundle["session"]
+        assert isinstance(session, DeferredSandboxSession)
+        assert isinstance(bundle["client"], DeferredSandboxClient)
+        assert session.pending
+        await _settle()
+        assert backend.calls == 1
+        assert await session.running()
+        assert session.state.manifest.environment.value["PYTHONUNBUFFERED"] == "1"
+        again = await session_manager.create_or_reuse(scan_id, image="img", local_sources=[])
+        assert again is bundle
+        assert backend.calls == 1
+    finally:
+        await session_manager.cleanup(scan_id)
+
+
+@pytest.mark.asyncio
+async def test_first_io_resolves_the_sandbox_then_delegates(backend: _Backend) -> None:
+    scan_id = "deferred-io"
+    try:
+        bundle = await _create(scan_id)
+        session = bundle["session"]
+        session._set_concurrency_limits("limits")
+        exec_task = asyncio.create_task(session.exec("echo", "hi"))
+        await asyncio.sleep(0.05)
+        assert not exec_task.done()
+        backend.gate.set()
+        result = await asyncio.wait_for(exec_task, timeout=1)
+        assert result.ok()
+        assert backend.session.execs == [("echo", "hi")]
+        assert not session.pending
+        assert backend.session.limits == ["limits"]
+        assert backend.session.state.manifest is session.state.manifest
+        await session.write(Path("/workspace/x"), io.BytesIO(b"data"))
+        assert backend.session.writes == [(Path("/workspace/x"), b"data")]
+    finally:
+        await session_manager.cleanup(scan_id)
+    assert backend.client.deleted == [backend.session]
+
+
+@pytest.mark.asyncio
+async def test_failed_bring_up_surfaces_on_first_io(backend: _Backend) -> None:
+    scan_id = "deferred-fail"
+    try:
+        bundle = await _create(scan_id)
+        session = bundle["session"]
+        backend.gate.set()
+        with pytest.raises(RuntimeError, match="bring-up exploded"):
+            await asyncio.wait_for(session.exec("true"), timeout=1)
+        assert not await session.running()
+        with pytest.raises(RuntimeError, match="bring-up exploded"):
+            await bundle["caido_client"].get()
+    finally:
+        await session_manager.cleanup(scan_id)
+    assert backend.client.deleted == []
+
+
+@pytest.mark.asyncio
+async def test_cleanup_while_pending_cancels_the_bring_up(backend: _Backend) -> None:
+    scan_id = "deferred-cancel"
+    bundle = await _create(scan_id)
+    session = bundle["session"]
+    await asyncio.wait_for(session_manager.cleanup(scan_id), timeout=1)
+    assert session.bring_up.cancelled()
+    assert backend.client.deleted == []
+    assert session_manager._SESSION_CACHE.get(scan_id) is None
+    backend.gate.set()
+    await _settle()
+    assert backend.client.deleted == [backend.session]
+
+
+@pytest.mark.asyncio
+async def test_cleanup_while_pending_ignores_a_failed_late_bring_up(backend: _Backend) -> None:
+    scan_id = "deferred-cancel-fail"
+    await _create(scan_id)
+    await asyncio.wait_for(session_manager.cleanup(scan_id), timeout=1)
+    backend.gate.set()
+    await _settle()
+    assert backend.client.deleted == []
+
+
+@pytest.mark.asyncio
+async def test_extra_files_are_staged_inside_the_deferred_bring_up(backend: _Backend) -> None:
+    scan_id = "deferred-extra"
+    try:
+        bundle = await _create(
+            scan_id,
+            extra_files=[{"workspace_path": "/workspace/.strix/notes.md", "content": "hi"}],
+        )
+        assert backend.session.writes == []
+        backend.gate.set()
+        await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
+        assert len(backend.session.writes) == 1
+        assert backend.session.execs
+    finally:
+        await session_manager.cleanup(scan_id)
+
+
+@pytest.mark.asyncio
+async def test_awaitable_extra_files_do_not_delay_the_backend_call(backend: _Backend) -> None:
+    scan_id = "deferred-extra-awaitable"
+    files: asyncio.Future[list[dict[str, Any]] | None] = asyncio.get_running_loop().create_future()
+    try:
+        bundle = await _create(scan_id, extra_files=files)
+        backend.gate.set()
+        await _settle()
+        assert backend.calls == 1
+        assert backend.session.writes == []
+        files.set_result([{"workspace_path": "/workspace/.strix/notes.md", "content": "hi"}])
+        await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
+        assert len(backend.session.writes) == 1
+    finally:
+        await session_manager.cleanup(scan_id)
+
+
+@pytest.mark.asyncio
+async def test_awaitable_extra_files_resolving_to_none_stage_nothing(backend: _Backend) -> None:
+    scan_id = "deferred-extra-none"
+
+    async def _nothing() -> list[dict[str, Any]] | None:
+        return None
+
+    try:
+        bundle = await _create(scan_id, extra_files=_nothing())
+        backend.gate.set()
+        await asyncio.wait_for(bundle["session"].resolve(), timeout=1)
+        assert backend.session.writes == []
+    finally:
+        await session_manager.cleanup(scan_id)
+
+
+@pytest.mark.asyncio
+async def test_default_path_still_blocks_until_the_sandbox_is_up(backend: _Backend) -> None:
+    scan_id = "deferred-off"
+    backend.gate.set()
+    try:
+        bundle = await session_manager.create_or_reuse(scan_id, image="img", local_sources=[])
+        assert bundle["session"] is backend.session
+        assert bundle["client"] is backend.client
+    finally:
+        await session_manager.cleanup(scan_id)
+    assert backend.client.deleted == [backend.session]
+
+
+@pytest.mark.asyncio
+async def test_sdk_session_prep_does_not_wait_for_the_bring_up(backend: _Backend) -> None:
+    scan_id = "deferred-sdk-prep"
+    try:
+        bundle = await _create(scan_id)
+        agent = SandboxAgent(
+            name="root", instructions="x", tools=[], capabilities=[Filesystem(), Shell()]
+        )
+        manager = SandboxRuntimeSessionManager(
+            starting_agent=agent,
+            sandbox_config=SandboxRunConfig(client=bundle["client"], session=bundle["session"]),
+            run_state=None,
+        )
+        session = await asyncio.wait_for(
+            manager.ensure_session(
+                agent=agent, capabilities=list(agent.capabilities), is_resumed_state=False
+            ),
+            timeout=1,
+        )
+        assert session is bundle["session"]
+        assert session.pending
+        backend.gate.set()
+        await asyncio.wait_for(session.resolve(), timeout=1)
+        assert not session.pending
+    finally:
+        await session_manager.cleanup(scan_id)

tests/test_reporting_fields.py

@@ -663,7 +663,6 @@ async def test_dependency_report_records_reachability(report_state: ReportState)
         == "src/render.ts:14 calls `_.template()`."
     )
     assert "**Usage analysis:**" in report["evidence"]
-    assert "not a proof of exploitability or of safety" in report["evidence"]
     # The level must never influence the rating — that comes from the contextual
     # breakdown, or from advisory_cvss when no breakdown applies.
     assert report["severity"] == "high"