mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 09:26:39 +02:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e03133eddd | ||
|
|
0b45d223c3 |
+4
-26
@@ -6,7 +6,6 @@ import csv
|
||||
import io
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import tempfile
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
@@ -19,21 +18,6 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
_SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4}
|
||||
|
||||
_BACKTICK_RUN = re.compile(r"`+")
|
||||
|
||||
|
||||
def _safe_fence(content: str) -> str:
|
||||
"""Return a backtick fence that ``content`` cannot break out of.
|
||||
|
||||
Per CommonMark a fenced code block is closed only by a run of backticks at
|
||||
least as long as the opening fence. LLM-authored, attacker-influenced values
|
||||
(PoC scripts, code snippets) may contain their own ``` runs, so we open with
|
||||
a fence one backtick longer than the longest run inside ``content`` (never
|
||||
fewer than three). Everything in ``content`` then renders verbatim.
|
||||
"""
|
||||
longest = max((len(m.group()) for m in _BACKTICK_RUN.finditer(content)), default=0)
|
||||
return "`" * max(3, longest + 1)
|
||||
|
||||
|
||||
def read_run_record(run_dir: Path) -> dict[str, Any]:
|
||||
path = run_record_path(run_dir)
|
||||
@@ -187,11 +171,9 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
|
||||
lines.append(str(report["poc_description"]))
|
||||
lines.append("")
|
||||
if report.get("poc_script_code"):
|
||||
code = str(report["poc_script_code"])
|
||||
fence = _safe_fence(code)
|
||||
lines.append(fence)
|
||||
lines.append(code)
|
||||
lines.append(fence)
|
||||
lines.append("```")
|
||||
lines.append(str(report["poc_script_code"]))
|
||||
lines.append("```")
|
||||
lines.append("")
|
||||
|
||||
if report.get("code_locations"):
|
||||
@@ -208,11 +190,7 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
|
||||
if loc.get("label"):
|
||||
lines.append(f" {loc['label']}")
|
||||
if loc.get("snippet"):
|
||||
snippet = str(loc["snippet"])
|
||||
fence = _safe_fence(snippet)
|
||||
lines.append(f" {fence}")
|
||||
lines.extend(f" {ln}" for ln in snippet.splitlines())
|
||||
lines.append(f" {fence}")
|
||||
lines.append(f" ```\n {loc['snippet']}\n ```")
|
||||
if loc.get("fix_before") or loc.get("fix_after"):
|
||||
lines.append("\n **Suggested Fix:**")
|
||||
lines.append("```diff")
|
||||
|
||||
@@ -167,9 +167,6 @@ async def get_request_with_client(
|
||||
return await client.request.get(request_id, opts)
|
||||
|
||||
|
||||
_FRAMING_HEADERS = frozenset({"content-length", "transfer-encoding"})
|
||||
|
||||
|
||||
def build_raw_request(
|
||||
*,
|
||||
method: str,
|
||||
@@ -190,16 +187,7 @@ def build_raw_request(
|
||||
final_headers = {**headers}
|
||||
final_headers.setdefault("Host", parsed.netloc)
|
||||
final_headers.setdefault("User-Agent", "strix")
|
||||
# Framing headers inherited from the captured request describe the ORIGINAL
|
||||
# body; once the body is modified for replay they are stale. We always send a
|
||||
# plain (non-chunked) body with an explicit Content-Length, so drop any
|
||||
# inherited Content-Length AND Transfer-Encoding (case-insensitively) and
|
||||
# recompute the length from the body actually being sent. This keeps the two
|
||||
# framing mechanisms from conflicting (RFC 7230 3.3.3: a leftover
|
||||
# Transfer-Encoding would make the target ignore Content-Length and try to
|
||||
# parse the body as chunked), so the replay is never desynced.
|
||||
final_headers = {k: v for k, v in final_headers.items() if k.lower() not in _FRAMING_HEADERS}
|
||||
if body:
|
||||
if body and "Content-Length" not in {k.title() for k in final_headers}:
|
||||
final_headers["Content-Length"] = str(len(body.encode("utf-8")))
|
||||
|
||||
lines = [f"{method.upper()} {path} HTTP/1.1"]
|
||||
|
||||
@@ -422,30 +422,6 @@ async def create_vulnerability_report(
|
||||
"availability": "H"
|
||||
}
|
||||
|
||||
**CVSS calibration** — score the weakness you actually proved, not a
|
||||
hypothetical worst case. Most over-rating comes from these mistakes:
|
||||
|
||||
- **Don't presuppose a separate compromise.** If exploitation
|
||||
requires the attacker to already hold a victim secret (a stolen
|
||||
session cookie/token, a leaked one-time link, intercepted traffic),
|
||||
that acquisition is not free. Do not score it as
|
||||
``privileges_required:N`` with ``attack_complexity:L`` as if
|
||||
directly reachable, and do not rate a replay-of-captured-secret
|
||||
issue High/Critical unless the *same* finding demonstrates a
|
||||
concrete way to obtain that secret. Issues like a session that
|
||||
survives logout or a replayable link are session-management /
|
||||
defense-in-depth weaknesses — usually Low/Medium on their own.
|
||||
- **Reserve ``H`` impact for demonstrated broad impact.** ``C:H`` /
|
||||
``I:H`` require proof of wide or systemic read/write. A single
|
||||
user's data, a read-only information leak, or merely confirming
|
||||
that an account / domain / software version *exists* (enumeration)
|
||||
is ``C:L`` (often ``I:N``) — not ``C:H``.
|
||||
- **Model required position and interaction honestly.** An
|
||||
adversary-in-the-middle prerequisite (e.g. cleartext transmission)
|
||||
or a required victim action is not guaranteed — reflect it in
|
||||
``attack_complexity`` / ``user_interaction`` instead of assuming the
|
||||
ideal condition always holds.
|
||||
|
||||
**CVE / CWE rules**: pass the bare ID only (``CVE-2024-1234``,
|
||||
``CWE-89``) — no name, no parenthetical. Be 100% certain; if
|
||||
unsure, use ``web_search`` to verify the ID before passing, or omit
|
||||
|
||||
@@ -140,59 +140,6 @@ async def test_host_call_serializes_concurrent_calls() -> None:
|
||||
assert state["max"] == 1
|
||||
|
||||
|
||||
def _headers_named(raw: bytes, name: str) -> list[str]:
|
||||
head = raw.decode("utf-8").split("\r\n\r\n", 1)[0]
|
||||
return [
|
||||
line.split(":", 1)[1].strip()
|
||||
for line in head.split("\r\n")[1:]
|
||||
if line.split(":", 1)[0].strip().lower() == name.lower()
|
||||
]
|
||||
|
||||
|
||||
def test_build_raw_request_recomputes_content_length_for_modified_body() -> None:
|
||||
# The captured request declared Content-Length: 12 (original body); the
|
||||
# replayed body is longer. The emitted request must carry exactly one
|
||||
# Content-Length equal to the ACTUAL body length, or the target truncates
|
||||
# the modified payload (or the connection desyncs).
|
||||
body = '{"user":"a\' OR 1=1 -- injected long payload"}'
|
||||
_conn, raw = caido_api.build_raw_request(
|
||||
method="POST",
|
||||
url="https://example.com/login",
|
||||
headers={"content-length": "12", "Content-Type": "application/json"},
|
||||
body=body,
|
||||
)
|
||||
sent_body = raw.decode("utf-8").split("\r\n\r\n", 1)[1]
|
||||
assert sent_body == body
|
||||
assert _headers_named(raw, "Content-Length") == [str(len(body.encode("utf-8")))]
|
||||
|
||||
|
||||
def test_build_raw_request_drops_transfer_encoding_for_modified_body() -> None:
|
||||
body = '{"user":"updated"}'
|
||||
_conn, raw = caido_api.build_raw_request(
|
||||
method="POST",
|
||||
url="https://example.com/login",
|
||||
headers={
|
||||
"tRaNsFeR-EnCoDiNg": "chunked",
|
||||
"Content-Length": "7",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
body=body,
|
||||
)
|
||||
assert _headers_named(raw, "Transfer-Encoding") == []
|
||||
assert _headers_named(raw, "Content-Length") == [str(len(body.encode("utf-8")))]
|
||||
|
||||
|
||||
def test_build_raw_request_drops_stale_content_length_for_empty_body() -> None:
|
||||
# A body cleared to empty must not keep the inherited (non-zero) length.
|
||||
_conn, raw = caido_api.build_raw_request(
|
||||
method="POST",
|
||||
url="https://example.com/x",
|
||||
headers={"Content-Length": "12"},
|
||||
body="",
|
||||
)
|
||||
assert _headers_named(raw, "Content-Length") == []
|
||||
|
||||
|
||||
class _Ctx:
|
||||
def __init__(self, context: Any) -> None:
|
||||
self.context = context
|
||||
|
||||
@@ -113,28 +113,6 @@ def test_render_vulnerability_md_includes_dependency_fields() -> None:
|
||||
assert "## Assumptions" in md
|
||||
|
||||
|
||||
def test_render_vulnerability_md_poc_code_cannot_break_out_of_fence() -> None:
|
||||
# LLM/target-authored PoC content containing its own ``` must not close the
|
||||
# fence early and turn the injected markdown into live headings/images.
|
||||
injected = "curl x\n```\n\n## Injected Heading\n"
|
||||
md = render_vulnerability_md(_sample_report(poc_script_code=injected))
|
||||
lines = md.split("\n")
|
||||
fence = next(ln for ln in lines[lines.index("## Proof of Concept") + 1 :] if ln.strip())
|
||||
assert set(fence) == {"`"}
|
||||
assert len(fence) >= 4 # wider than the payload's 3-backtick run
|
||||
assert injected in md # the payload survives verbatim, inside the fence
|
||||
|
||||
|
||||
def test_render_vulnerability_md_snippet_cannot_break_out_of_fence() -> None:
|
||||
snippet = "row = q()\n```\n## Injected"
|
||||
md = render_vulnerability_md(
|
||||
_sample_report(code_locations=[{"file": "app.py", "snippet": snippet}]),
|
||||
)
|
||||
assert (
|
||||
" ````\n row = q()\n ```\n ## Injected\n ````"
|
||||
) in md # indented fence widened past the payload's ``` run
|
||||
|
||||
|
||||
def test_write_vulnerabilities_creates_markdown_csv_and_json(tmp_path: Path) -> None:
|
||||
reports = [
|
||||
_sample_report(id="vuln-0001", severity="medium", timestamp="2026-07-02 11:00:00 UTC"),
|
||||
|
||||
Reference in New Issue
Block a user