get_update_pkg.py

#!/usr/bin/env python3

# https://stackoverflow.com/questions/5574702/how-to-print-to-stderr-in-python
from __future__ import print_function
import sys

def eprint(*args, **kwargs):
    print(*args, file=sys.stderr, **kwargs)

import os
import re
import csv
import sys
import hmac
import hashlib
import struct
import urllib.request
import urllib.error
import xml.etree.ElementTree as ET
import xml.dom.minidom as minidom


# http://wololo.net/talk/viewtopic.php?f=54&t=44091
KEY = bytes.fromhex("E5E278AA1EE34082A088279C83F9BBC806821C52F2AB5D2B4ABD995450355114")


def ripemd160(value):
  h1 = hashlib.new('ripemd160')
  h = hmac.new(KEY, (value).encode("ascii"), lambda: h1)
  return h

def md5(value):
  h = hmac.new(KEY, (value).encode("ascii"), hashlib.md5)
  return h

def sha1(value):
  h = hmac.new(KEY, (value).encode("ascii"), hashlib.sha1)
  return h

def sha256(value):
  h = hmac.new(KEY, (value).encode("ascii"), hashlib.sha256)
  return h

def xmlurl(title):
  h = sha256("np_" + title)
  url = "http://gs-sec.ww.np.dl.playstation.net/pl/np/%s/%s/%s-ver.xml" % (title, h.hexdigest(), title)
  return url

def title2info(title):
  url = xmlurl(title)

  try:
    with urllib.request.urlopen(url) as f:
      data = f.read()
  except urllib.error.HTTPError as e:
    if e.code == 404:
      return None
    raise e

  if data:
    return data.decode("utf-8")

  return None


def parse_sfo(sfo):
  header, version, key_table, data_table, count = struct.unpack_from("<IIIII", sfo)
  assert header == 0x46535000
  assert version == 0x00000101

  ret = {}
  for i in range(count):
    key_offset, param_type, param_len, param_maxlen, data_offset = struct.unpack_from("<HHIII", sfo, 0x14 + 0x10 * i)

    key_end = sfo.find(b"\0", key_table + key_offset)
    key = struct.unpack_from("%ds" % (key_end - (key_table + key_offset)), sfo, key_table + key_offset)
    key = key[0].decode("ascii")

    if param_type == 0x0204:
      value = struct.unpack_from("%ds" % (param_len - 1), sfo, data_table + data_offset)
      ret[key] = value[0].decode("utf-8")
    elif param_type == 0x0404:
      assert param_len == 4
      value = struct.unpack_from("<I", sfo, data_table + data_offset)
      ret[key] = value[0]
    else:
      assert not "unknown type"

  return ret


def get_pkg_start(pkg, size):
  req = urllib.request.Request(pkg, headers={"Range": "bytes=0-%d" % size})
  with urllib.request.urlopen(req) as f:
    return f.read()


def get_info(patchurl):
  MaxVersion = "99.99"

  data = get_pkg_start(patchurl, 16*1024)

  magic1 = struct.unpack_from(">I", data)[0]
  magic2 = struct.unpack_from(">I", data, 192)[0]
  if magic1 != 0x7f504b47 or magic2 != 0x7f657874:
    eprint(patchurl)
    eprint("corrupted pkg?")
    return MaxVersion

  meta_offset, meta_count = struct.unpack_from(">II", data, 8)
  sfo_offset = None
  for i in range(meta_count):
    meta_type, meta_size = struct.unpack_from(">II", data, meta_offset)
    if meta_type == 14:
      sfo_offset, sfo_size = struct.unpack_from(">II", data, meta_offset + 8)
      break
    meta_offset += 8 + meta_size

  if sfo_offset is None:
    eprint(patchurl)
    eprint("cannot find sfo in pkg")
    return MaxVersion

  if sfo_offset + sfo_size > len(data):
    eprint(patchurl)
    eprint("sfo is not in beginning of pkg [%d..%d]" % (sfo_offset, sfo_offset + sfo_size-1))
    return MaxVersion

  sfo = parse_sfo(data[sfo_offset:sfo_offset+sfo_size])
  fw = sfo["PSP2_DISP_VER"]

  name = sfo["STITLE"] if "STITLE" in sfo else sfo["TITLE"]
  name = name.replace("\n"," ").replace("\r"," ").strip()

  m = re.match(r"(\d+\.\d\d).*", fw)
  return m.group(1), name



def get_supported(xml, supported = "03.60"):
  packages = list(ET.fromstring(xml).findall("./tag/package"))
  for p in reversed(packages):
    patchurl = p.attrib["url"]
    size = p.attrib["size"]
    version = p.attrib["version"]
    fw, name = get_info(patchurl)
    if fw <= supported:
      return patchurl, size, version.lstrip("0"), fw.lstrip("0"), name

  return None


def info(title):
  xml = title2info(title)
  if xml is None:
    eprint(title + ": None")
  else:
    xml = minidom.parseString(xml)
    xml = xml.toprettyxml(indent="   ")
    print(xml)


def patch(title):
  xml = title2info(title)
  if xml is None:
    eprint(title + ": None")
  else:
    supported = get_supported(xml)
    if supported is None:
      eprint(title + ": None")
    else:
      patchurl, size, version, fw, name = supported
      print("Name:", name)
      print("Patch:", patchurl)
      print("Version:", version)
      print("FW:", fw)
      print("Size:", size)


def latest(title):
  xml = title2info(title)
  if xml is None:
    eprint(title + ": None")
  else:
    supported = get_supported(xml, "99.99")
    if supported is None:
      eprint(title + ": None")
    else:
      patchurl, size, version, fw, name = supported
      print("Name:", name)
      print("Patch:", patchurl)
      print("Version:", version)
      print("FW:", fw)
      print("Size:", size)

def nps(title):
  url = xmlurl(title)
  xml = title2info(title)
  if xml is None:
    eprint(title + ": None")
  else:
    packages = list()
    name = ""
    for x in ET.fromstring(xml).findall("./tag/package/paramsfo/title"):
      name = x.text.replace("\n"," ").replace("\r"," ").strip()
    for x in ET.fromstring(xml).findall("./tag/package"):
      version = x.attrib["version"]
      p = None
      if x.get("type") is None or x.attrib["type"] == "cumulative":
        p = x
      else:
        for y in x.findall("./hybrid_package"):
          p = y
      if p is not None:
        patchurl = p.attrib["url"]
        size = p.attrib["size"]
        sha1 = p.attrib["sha1sum"]
        fw, name2 = get_info(patchurl)
        if name == "":
          name = name2
        print("{0}\t\t{1}\t{2}\t{3}\t{4}\t\t\t{5}\t\t{6}\t{7}".format(title, name, version.lstrip("0"), fw.lstrip("0"), patchurl, size, url, sha1))

if __name__ == "__main__":
  do_help = 0
  if len(sys.argv) != 3:
    do_help = 1
  else:
    if sys.argv[1] == "xml":
      info(sys.argv[2])
    elif sys.argv[1] == "patch":
      patch(sys.argv[2])
    elif sys.argv[1] == "latest":
      latest(sys.argv[2])
    elif sys.argv[1] == "nps":
      nps(sys.argv[2])
    elif sys.argv[1] == "sha256":
      h = sha256(sys.argv[2])
      print(h.hexdigest())
    elif sys.argv[1] == "sha1":
      h = sha1(sys.argv[2])
      print(h.hexdigest())
    elif sys.argv[1] == "md5":
      h = md5(sys.argv[2])
      print(h.hexdigest())
    elif sys.argv[1] == "ripemd160":
      h = ripemd160(sys.argv[2])
      print(h.hexdigest())
    else:
      do_help = 1

  if do_help:
    eprint("Usage: %s COMMAND TITLEID" % (sys.argv[0]))
    eprint()
    eprint("Available commands:")
    eprint(" xml TITLEID     - retrieve title patch information xml file, prints to stdout")
    eprint(" patch TITLEID   - retrieve latest usable (on 3.60) patch for title, prints to stdout")
    eprint(" latest TITLEID  - retrieve latest patch (may not work on 3.60) for title, prints to stdout")
    eprint(" nps TITLEID     - retrieve all cumulative patches (may not work on 3.60) for title, prints to stdout as TSV lines for NPS")
    eprint(" sha256 VALUE    - returns PSV SHA256 hash hex digest for string")
    eprint(" sha1 VALUE      - returns PSV SHA1 hash hex digest for string")
    eprint(" md5 VALUE       - returns PSV MD5 hash hex digest for string")
    eprint(" ripemd160 VALUE - returns PSV RIPEMD-160 hash hex digest for string")
    exit(0)