Mock authenticator/counterfeit detector for DualShock4 controllers. Does basic detection for banned keys through SHA256 fingerprint (fingerprint database required). Does not include CA public key for obvious reason.

ds4poke.py

#!/usr/bin/env python3
# Credits to:
#   - Eleccelerator and PS4 Developer Wiki for the HID report format used in
#     authentication procedure.
#   - Author of jedi_crypto.py who provides detailed information on the basic
#     building blocks used by the authentication scheme (and the Jedi CA
#     certificate).
#
# This tool is for education and demonstration purpose only. Use it at your own
# risk.
# Released to public domain

import sys
import os
import zlib
import io
import time
import argparse
import json
from ctypes import *

from Crypto.Hash import SHA256
from Crypto.PublicKey import RSA
from Crypto.Signature import PKCS1_PSS
from Crypto.Util.number import bytes_to_long

JEDI_CA_PUBKEY_FINGERPRINT = b'\xe5\xe0\x95\xe6C\xb5h\x8b@\x0cu{LD\xef\xac\xc2\x93aH\xe5\xce\xbdlmA\x0fT\xf1H\x7fI'

class DS4Auth(LittleEndianStructure):
    _pack_ = 1
    _fields_ = [
        ('type', c_uint8),
        ('seq', c_uint8),
        ('page', c_uint8),
        ('sbz', c_uint8),
        ('data', c_uint8 * 56),
        ('crc32', c_uint32)
    ]

class DS4AuthResult(LittleEndianStructure):
    _pack_ = 1
    _fields_ = [
        ('type', c_uint8),
        ('seq', c_uint8),
        ('status', c_uint8),
        ('padding', c_uint8 * 9),
        ('crc32', c_uint32)
    ]

class DS4AuthReset(LittleEndianStructure):
    _pack_ = 1
    _fields_ = [
        ('type', c_uint8),
        ('sbz0', c_uint8),
        ('nonce_page_size', c_uint8),
        ('resp_page_size', c_uint8),
        ('sbz1', c_uint8 * 4)
    ]

class DS4IdentityBlock(BigEndianStructure):
    _pack_ = 1
    _fields_ = [
        ('serial', c_uint8 * 0x10),
        ('modulus', c_uint8 * 0x100),
        ('exponent', c_uint8 * 0x100)
    ]

class DS4Response(LittleEndianStructure):
    _pack_ = 1
    _fields_ = [
        ('sig_nonce', c_uint8 * 0x100),
        ('identity', DS4IdentityBlock),
        ('sig_identity', c_uint8 * 0x100)
    ]

def parse_vidpid(vidpidstr):
    vids, pids = vidpidstr.split(':')[:2]
    return int(vids, 16) & 0xffff, int(pids, 16) & 0xffff

def parse_args():
    p = argparse.ArgumentParser()
    p.add_argument('-d', '--vidpid',
                   help='Specify VID:PID of target device',
                   default='054c:05c4')
    p.add_argument('-c', '--jedi-ca-pubkey',
                   help='Location of Jedi CA public key',
                   default='jedi.pub')
    p.add_argument('-f', '--force-bad-ca',
                   help='Skip Jedi CA fingerprint check. Useful for debugging',
                   action='store_true',
                   default=False)
    p.add_argument('-n', '--no-cuk-check',
                   help='Disable cert check for Controller Unique Key',
                   action='store_true',
                   default=False)
    p.add_argument('-s', '--seq',
                   help='Override sequence ID in authentication packet',
                   type=int,
                   default=1)
    p.add_argument('-r', '--revocation-list',
                   help='Path to revocation list',
                   default=None)
    p.add_argument('-R', '--with-reset',
                   help='Send GET_REPORT 0xf3 to reset the authentication '
                        'state and/or get challenge+response buffer sizes (if '
                        'non-default) before authentication happens, required '
                        'for third-party (licensed) controllers',
                   action='store_true',
                   default=False)
    p.add_argument('-T', '--use-ps4-timing', '--wtf-activision',
                   help='Further emulate the packet timing of PS4 to workaround'
                        ' firmware bugs in certain third-party controllers',
                   action='store_true',
                   default=False)
    p.add_argument('-t', '--timeout',
                   help='Manually specify timeout in seconds (default: 15)',
                   type=float,
                   default=15.0)
    p.add_argument('-H', '--hidraw',
                   help='Use hidraw backend (Linux-only)',
                   action='store_true',
                   default=False)
    p.add_argument('-v', '--verbose',
                   help='Print protocol trace',
                   action='store_true',
                   default=False)
    return p, p.parse_args()


if __name__ == '__main__':
    def _trace(msg):
        if args.verbose:
            print(msg)
    p, args = parse_args()
    if args.hidraw:
        import hidraw as hid
    else:
        import hid
    dev = hid.device()
    try:
        dev.open(*parse_vidpid(args.vidpid))
        nonce = io.BytesIO(os.urandom(256))
        nonce_page_size = resp_page_size = DS4Auth.data.size

        if args.with_reset:
            _trace('<= GET 0xf3 len=8')
            recv = dev.get_feature_report(0xf3, 8)
            _trace(f'=> GET 0xf3 payload={bytes(recv).hex()}')
            resetbuf = DS4AuthReset()
            size = min(len(recv), sizeof(resetbuf))
            memmove(addressof(resetbuf), bytes(recv), size)
            print('reset:', bytearray(recv).hex())
            print('nonce_page_size =', resetbuf.nonce_page_size)
            print('resp_page_size =', resetbuf.resp_page_size)
            assert resetbuf.sbz0 == 0 and False not in (v == 0 for v in resetbuf.sbz1), 'SBZ is not zero'
            # TODO PDP remote needs this to work, figure out what is bit 7
            resetbuf.nonce_page_size &= 0x7f
            resetbuf.resp_page_size &= 0x7f
            if resetbuf.nonce_page_size > DS4Auth.data.size:
                print('WARNING: Nonce page size > protocol maximum, use protocol maximum')
            else:
                nonce_page_size = resetbuf.nonce_page_size
            if resetbuf.resp_page_size > DS4Auth.data.size:
                print('WARNING: Response page size > protocol maximum, use protocol maximum')
            else:
                resp_page_size = resetbuf.resp_page_size
            time.sleep(1)

        # Ensure the buffer size is within range
        assert nonce_page_size <= DS4Auth.data.size, 'Oversized nonce page'
        assert resp_page_size <= DS4Auth.data.size, 'Oversized resp page'
        print('nonce =', nonce.getvalue().hex())
        challengebuf = DS4Auth()
        challengebuf.seq = args.seq
        challengebuf.type = 0xf0
        while True:
            payload = nonce.read(nonce_page_size)
            if len(payload) == 0:
                break
            memset(addressof(challengebuf.data), 0, DS4Auth.data.size)
            memmove(addressof(challengebuf.data), payload, min(nonce_page_size, len(payload)))
            print('page =', challengebuf.page, 'data =', memoryview(challengebuf.data).hex())
            challengebuf.crc32 = zlib.crc32(bytes(challengebuf)[:sizeof(DS4Auth)-sizeof(c_uint32)])
            print('crc =', challengebuf.crc32)
            _trace(f'<= SET 0xf0 payload={bytes(challengebuf).hex()}')
            dev.send_feature_report(challengebuf)
            challengebuf.page += 1
            print('sleeping')
            time.sleep(1)
            memset(challengebuf.data, 0, sizeof(challengebuf.data))
        if args.use_ps4_timing:
            print('extra sleep')
            time.sleep(1)
        begin = time.perf_counter()
        while True:
            _trace(f'<= GET 0xf2 len={sizeof(DS4AuthResult)}')
            recv = dev.get_feature_report(0xf2, sizeof(DS4AuthResult))
            _trace(f'=> GET 0xf2 payload={bytes(recv).hex()}')
            result = DS4AuthResult()
            size = min(len(recv), sizeof(result))
            memmove(addressof(result), bytes(recv), size)
            print('seq =', result.seq, 'status =', result.status)
            print('crc =', result.crc32)
            if zlib.crc32(bytes(result)[:sizeof(DS4AuthResult)-sizeof(c_uint32)]) != result.crc32:
                print('crc mismatch')
            if result.seq != challengebuf.seq:
                print('oops')
            if result.status == 0:
                break
            if time.perf_counter() - begin > args.timeout:
                print('timeout waiting for ok')
                break
            time.sleep(1)
        print('auth ok')

        if args.use_ps4_timing:
            print('extra sleep')
            time.sleep(1)

        resp = io.BytesIO()

        for i in range(-(-sizeof(DS4Response) // resp_page_size)):
            _trace(f'<= GET 0xf1 len={sizeof(DS4Auth)}')
            recv = dev.get_feature_report(0xf1, sizeof(DS4Auth))
            _trace(f'=> GET 0xf1 payload={bytes(recv).hex()}')
            respbuf = DS4Auth()
            size = min(len(recv), sizeof(respbuf))
            memmove(addressof(respbuf), bytes(recv), size)
            print('seq =', respbuf.seq, 'page =', respbuf.page)
            print('crc =', respbuf.crc32)
            # TODO crc on PDP remote is broken. Figure out why.
            if zlib.crc32(bytes(respbuf)[:sizeof(DS4Auth)-sizeof(c_uint32)]) != respbuf.crc32:
                print('crc mismatch')
            print('data =', memoryview(respbuf.data).hex())
            resp.write(memoryview(respbuf.data)[:resp_page_size])

            time.sleep(1)

        print('resp =', resp.getbuffer().hex())

        # verify
        resp.seek(0)
        resp_check = DS4Response()
        resp.readinto(resp_check)
        print('serial =', memoryview(resp_check.identity.serial).hex())

        pss_ca = None
        if os.path.isfile(args.jedi_ca_pubkey) and not args.no_cuk_check:
            with open(args.jedi_ca_pubkey, 'rb') as f:
                ca = RSA.importKey(f.read())

            if not args.force_bad_ca and SHA256.new(ca.exportKey('DER')).digest() != JEDI_CA_PUBKEY_FINGERPRINT:
                print('WARNING: Wrong fingerprint for Jedi CA, disabling authenticity check')
            else:
                pss_ca = PKCS1_PSS.new(ca)
        elif args.no_cuk_check:
            print('Authenticity check disabled by user')
        else:
            print('WARNING: Cannot open Jedi CA, disabling authenticity check')

        cuk = RSA.construct((bytes_to_long(bytes(resp_check.identity.modulus)), bytes_to_long(bytes(resp_check.identity.exponent))))
        fp_cuk = SHA256.new(cuk.exportKey('DER')).digest()
        print('fp_cuk =', fp_cuk.hex())

        sha_nonce = SHA256.new(nonce.getvalue())
        sha_identity = SHA256.new(bytes(resp_check.identity))

        pss_cuk = PKCS1_PSS.new(cuk)

        print('sig_nonce =', memoryview(resp_check.sig_nonce).hex())
        print('identity =', memoryview(resp_check.identity).hex())
        print('sig_identity =', memoryview(resp_check.sig_identity).hex())

        result_sig_nonce = pss_cuk.verify(sha_nonce, bytes(resp_check.sig_nonce))
        if pss_ca is not None:
            result_identity = pss_ca.verify(sha_identity, bytes(resp_check.sig_identity))
        else:
            result_identity = None

        if args.revocation_list is not None and not args.no_cuk_check:
            with open(args.revocation_list, 'r') as rl:
                revocation_list = json.load(rl)
            tag = revocation_list.get(fp_cuk.hex())
            if tag is not None:
                print('Revoked key {} detected'.format(tag))
                result_revocation = True
            else:
                result_revocation = False
        else:
            result_revocation = None

        if result_sig_nonce:
            print('good sig for nonce')
        else:
            print('bad sig for nonce')

        if result_identity:
            print('good sig for controller unique key')
        elif result_identity is None:
            print('cannot decide the authenticity of controller unique key')
        else:
            print('bad sig for controller unique key')

        if result_sig_nonce and result_identity and not result_revocation:
            print('the controller seems to be genuine')
        elif result_identity is None:
            print('cannot decide the authenticity of the controller')
        else:
            print('the controller may be fake')
    finally:
        dev.close()

jedi_revoke.json

{
    "a67f9f080a183e37e81329109cf12f1cdbfadb494659f28f6a1a3b88d3954e6f": "jedi_sacrifice",
    "c51e01b943797dde36727fb647e7bb762d2979d0c59acff930f781d54c960280": "bk16_4c960280",
    "cd989d1389c93d5ba53e88150a57c8e77f74102babbd0599932cb68d99ff146b": "bk17_99ff146b",
    "67781734c8f39d621887b79c0a17fa493589a661bfea8b420705322cef5f45ca": "bk17_ef5f45ca",
    "731d0d1ce05b69ebedccc413f9e81ed4085eabb1f99f442f9a7f7f104e5d3cbe": "bk18_4e5d3cbe",
    "0b8a553f955aeb62c2256512e3a96ffad8b9f7786daa1ba2a303263f01b1aa9a": "bk18_oem_01b1aa9a"
}

requirements.txt

pycryptodome
hidapi