sonic_rainboom.py

#!/usr/bin/python3

TheEnd = 'Sonic Rainboom'

import base64
import sys
from cryptography import x509
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa
from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat
try:
  from pgpy import PGPKey
  from pgpy.constants import EllipticCurveOID, PubKeyAlgorithm
  from pgpy.packet.fields import ECDSAPriv, RSAPriv
  from pgpy.packet.packets import PrivKeyV4
  from pgpy.packet.types import MPI
except ImportError:
  PGPKey = None

class PGP:
  def __init__(self, data):
    if PGPKey is None:
      raise ImportError('PGPy')
    self.id = PGPKey()
    self.id.parse(data)
    self.pub = self.id._key.keymaterial.__pubkey__()
  def __repr__(self):
    if len(self.id.userids) > 0 and self.id.userids[0].name != '':
       return self.id.userids[0].name+' <'+self.id.userids[0].email+'>'
    return self.id._key.fingerprint
  def export(self, priv):
    priv = priv.private_numbers()
    key = PGPKey()
    key._uids = self.id._uids
    key._key = PrivKeyV4()
    key._key._created = self.id._key._created
    key._key._pkalg = self.id._key._pkalg
    if key._key._pkalg in (PubKeyAlgorithm.RSAEncryptOrSign, PubKeyAlgorithm.RSAEncrypt, PubKeyAlgorithm.RSASign):
      km = RSAPriv()
      km.e = MPI(priv.public_numbers.e)
      km.n = MPI(priv.public_numbers.n)
      km.d = MPI(priv.d)
      km.p = MPI(priv.p)
      km.q = MPI(priv.q)
      km.u = MPI(rsa.rsa_crt_iqmp(priv.q, priv.p))
    elif key._key._pkalg is PubKeyAlgorithm.ECDSA:
      km = ECDSAPriv()
      km.oid = self.id._key.keymaterial.oid
      km.x = MPI(priv._public_numbers.x)
      km.y = MPI(priv._public_numbers.y)
      km.s = MPI(priv._private_value)
    else:
      raise TypeError
    key._key.keymaterial = km
    key._key.keymaterial._compute_chksum()
    key._key.update_hlen()
    return str(key)

class SSH:
  def __init__(self, data):
    self.id = data.split()
    self.pub = serialization.load_ssh_public_key(data, default_backend())
  def __repr__(self):
    if len(self.id) > 2 and self.id[2] != '':
      return self.id[2].decode()
    fp = hashes.Hash(hashes.SHA256(), backend=default_backend())
    fp.update(base64.b64decode(self.id[1]))
    return base64.b64encode(fp.finalize()).decode().rstrip('=')
  def export(self, priv):
    return priv.private_bytes(Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption()).decode()

class SSL:
  def __init__(self, data, fmt='crt'):
    self.fmt = fmt
    if fmt is 'crt':
      self.cert = x509.load_pem_x509_certificate(data, default_backend())
    elif fmt is 'csr':
      self.cert = x509.load_pem_x509_csr(data, default_backend())
    else:
      raise TypeError
    self.pub = self.cert.public_key()
  def __repr__(self):
    cn = self.cert.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)
    if len(cn) > 0:
      return cn[0].value
    org = self.cert.subject.get_attributes_for_oid(x509.NameOID.ORGANIZATION_NAME)
    if len(org) > 0:
      return org[0].value
    if self.fmt == 'crt':
      return self.cert.fingerprint(hashes.SHA256()).hex()
    return 'Unknown'
  def export(self, priv):
    return priv.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()).decode()

class ECC:
  def __init__(self, tgt):
    self.pub = tgt.pub
  def compute(self, vqc):
    pub = self.pub.public_numbers()
    p = vqc.ecpoint(pub.curve, pub.x, pub.y)
    priv = ec.EllipticCurvePrivateNumbers(p, pub)
    try:
      self.priv = default_backend().load_elliptic_curve_private_numbers(priv)
    except ValueError:
      return False
    return True
  def verify(self):
    m = TheEnd.encode()
    s = self.priv.sign(m, ec.ECDSA(hashes.SHA256()))
    try:
      self.pub.verify(s, m, ec.ECDSA(hashes.SHA256()))
    except InvalidSignature:
      return False
    return True

class RSA:
  def __init__(self, tgt):
    self.pub = tgt.pub
  def _egcd(self, a, b):
    if a == 0:
      return (b, 0, 1)
    g, y, x = self._egcd(b%a, a)
    return (g, x-y*(b//a), y)
  def _invert(self, a, m):
    g, x, y = self._egcd(a, m)
    if g != 1:
      return None
    return x%m
  def compute(self, vqc):
    pub = self.pub.public_numbers()
    p, q = vqc.factors(pub.n)
    d = self._invert(pub.e, (p-1)*(q-1))
    dmp1 = rsa.rsa_crt_dmp1(d, p)
    dmq1 = rsa.rsa_crt_dmq1(d, q)
    iqmp = rsa.rsa_crt_iqmp(p, q)
    priv = rsa.RSAPrivateNumbers(p, q, d, dmp1, dmq1, iqmp, pub)
    try:
      self.priv = default_backend().load_rsa_private_numbers(priv)
    except ValueError:
      return False
    return True
  def verify(self):
    m = TheEnd.encode()
    pad = padding.PSS(padding.MGF1(hashes.SHA256()), padding.PSS.MAX_LENGTH)
    s = self.priv.sign(m, pad, hashes.SHA256())
    try:
      self.pub.verify(s, m, pad, hashes.SHA256())
    except InvalidSignature:
      return False
    return True

class VQC:
  def factors(self, c):
    # lewd ponies
    return a, b
  def ecpoint(self, curve, x, y):
    raise FutureWarning
    return s



if __name__ == '__main__':
  f = open(sys.argv[1], 'rb')
  data = f.read()

  if data.startswith(b'-----BEGIN PGP PUBLIC KEY BLOCK-----'):
    tgt = PGP(data)
  elif data.startswith(b'-----BEGIN CERTIFICATE-----'):
    tgt = SSL(data)
  elif data.startswith(b'-----BEGIN CERTIFICATE REQUEST-----'):
    tgt = SSL(data, fmt='csr')
  elif data.startswith(b'ssh-rsa') or data.startswith(b'ecdsa'):
    tgt = SSH(data)

  print(tgt, file=sys.stderr)

  if isinstance(tgt.pub, ec.EllipticCurvePublicKey):
    key = ECC(tgt)
  elif isinstance(tgt.pub, rsa.RSAPublicKey):
    key = RSA(tgt)
  else:
    raise TypeError

  vqc = VQC()

  if not key.compute(vqc) or not key.verify():
    print('No boom make sad pony :(', file=sys.stderr)
    sys.exit(9)
  print(tgt.export(key.priv), end='')