PS5 Payload TCP Logger, useful for payload debugging before jailbreak. Tested on YARPE - PS5 FW 12.02

payload_message_client.py

"""
Payload that sends messages to remote host via syscall lvl socket operations
Mirrors SploitCore's open/write/close pattern
"""
import struct
from sc import sc
from utils.rp import log
from constants import SYSCALL

# Configuration
SERVER_IP = "192.168.1.20"
SERVER_PORT = 9000

def _ensure_syscalls():
    """Ensure all required syscalls are registered."""
    SYSCALL.setdefault("socket", 97)
    SYSCALL.setdefault("connect", 98)
    SYSCALL.setdefault("read", 3)
    SYSCALL.setdefault("write", 4)
    SYSCALL.setdefault("close", 6)

def ip_to_bytes(ip_str):
    """Convert dotted IP to 4-byte network format."""
    parts = [int(x) for x in ip_str.split('.')]
    return struct.pack("<BBBB", *parts)


def log_tcp(message, wait_response=True, stdout=True):
    """
    Send a message to remote server using syscalls.

    Args:
        message: String message to send
        wait_response: Whether to wait for server response
    """
    # Create socket
    sockfd = sc.syscalls.socket(2, 1, 0)
    if sockfd < 0:
        log("[-] socket() failed: " + str(sockfd))
        return False

    # Prepare sockaddr_in: family(2) + port(2) + ip(4) + padding(8)
    sock_addr = struct.pack(
        ">HH4s8s",
        2,  # AF_INET (sin_family)
        SERVER_PORT,  # sin_port in network byte order
        ip_to_bytes(SERVER_IP),  # sin_addr
        b'\x00' * 8  # sin_zero padding
    )

    # Connect
    ret = sc.syscalls.connect(sockfd, sock_addr, len(sock_addr))
    if ret < 0:
        log("[-] connect() failed: " + str(ret))
        sc.syscalls.close(sockfd)
        return False

    # Send message
    msg_bytes = (message + "\n").encode("utf-8")
    sent = sc.syscalls.write(sockfd, msg_bytes, len(msg_bytes))
    if sent != len(msg_bytes):
        log("[-] write() failed: sent " + str(sent) + " of " + str(len(msg_bytes)) + " bytes")
        sc.syscalls.close(sockfd)
        return False

    log("[+] Sent: " + message)

    # Wait for response
    if wait_response:
        while True:
            response = sc.syscalls.read(sockfd, 4096)
            if not response or len(response) == 0:
                break
            try:
                resp_str = response.decode("utf-8", errors="ignore").strip()
                if resp_str:
                    log("[+] Response: " + resp_str)
            except:
                pass

    sc.syscalls.close(sockfd)
    return True


def main():
    _ensure_syscalls()
    log("[*] Message client payload started")
    log("[*] Connecting to " + SERVER_IP + ":" + str(SERVER_PORT))
    log_tcp("[*] Sending this message to server...")
    log_tcp("[*] This message isn't displayed on PS5, only sent to server", stdout=False)

main()

tcp_send_ps5.py

import struct
import traceback

from constants import SYSCALL
from sc import sc
from utils.rp import log

TCP_ENABLED = True
TCP_HOST = "192.168.1.20"
TCP_PORT = 9000
_TCP_SOCKET = None

def tcp_connect():
    global _TCP_SOCKET
    try:
        if _TCP_SOCKET is not None and _TCP_SOCKET >= 0:
            return _TCP_SOCKET

        sockfd = sc.syscalls.socket(2, 1, 0)
        if sockfd < 0:
            log("[TCP] socket() failed: " + str(sockfd))
            return None

        sock_addr = struct.pack(
            ">HH4s8s",
            2,
            TCP_PORT,
            struct.pack("<BBBB", *[int(x) for x in TCP_HOST.split('.')]),
            b'\x00' * 8
        )
        ret = sc.syscalls.connect(sockfd, sock_addr, len(sock_addr))
        if ret >= 0:
            _TCP_SOCKET = sockfd
            return sockfd
        else:
            log("[TCP] connect() failed: " + str(ret) + " to " + TCP_HOST + ":" + str(TCP_PORT))
            sc.syscalls.close(sockfd)
            return None
    except Exception as e:
        log("[TCP] connect exception: " + str(e))
        return None

def tcp_send(message):
    global _TCP_SOCKET
    if not TCP_ENABLED:
        return
    try:
        sockfd = tcp_connect()
        if sockfd is None or sockfd < 0:
            return

        msg_bytes = (message + "\n").encode("utf-8")
        sent = sc.syscalls.write(sockfd, msg_bytes, len(msg_bytes))
        if sent != len(msg_bytes):
            log("[TCP] write partial: " + str(sent) + "/" + str(len(msg_bytes)))
            _TCP_SOCKET = None
    except Exception as e:
        log("[TCP] send exception: " + str(e))
        _TCP_SOCKET = None

def tcp_close():
    global _TCP_SOCKET
    if _TCP_SOCKET is not None and _TCP_SOCKET >= 0:
        try:
            sc.syscalls.close(_TCP_SOCKET)
        except:
            pass
        _TCP_SOCKET = None

def log_tcp(msg, stdout=True):
    if stdout:
        log(msg)
    tcp_send(msg)

def _ensure_syscalls():
    SYSCALL.setdefault("socket", 0x61)
    SYSCALL.setdefault("connect", 0x62)
    SYSCALL.setdefault("write", 0x4)
    SYSCALL.setdefault("close", 0x6)
    SYSCALL.setdefault("open", 0x5)

def main():
    _ensure_syscalls()
    log_tcp("[*] TCP Message Sender")
    log_tcp("[*] Platform: " + sc.platform + " FW: " + str(sc.version))
    tcp_close()

try:
    main()
except Exception as e:
    error_msg = "EXCEPTION: " + str(e)
    log_tcp(error_msg)
    tb = traceback.format_exc()
    for line in tb.split('\n'):
        if line.strip():
            log_tcp(line)
    tcp_send("EXCEPTION_END")
    tcp_close()
except:
    log_tcp("UNKNOWN_EXCEPTION")
    tcp_send("EXCEPTION_END")
    tcp_close()