Initial commit?

1 files changed, 522 insertions, 0 deletions

diff --git a/rebiss.py b/rebiss.py
new file mode 100644
index 0000000..28a3e44
--- /dev/null
+++ b/rebiss.py
@@ -0,0 +1,522 @@
+# SPDX-License-Identifier: GPL-3.0-only
+
+import base64
+import logging
+import getpass
+import hashlib
+import json
+import os
+import re
+import subprocess
+import unicodedata
+import time
+import datetime
+
+import OpenSSL.crypto as cr
+
+
+class HashAlg:
+    __slots__ = ('len', 'hashlib_name', 'mech')
+
+    def __init__(self, len, hashlib_name, mech):
+        self.len = len
+        self.hashlib_name = hashlib_name
+        self.mech = mech
+
+HASH_ALG = {
+    'SHA256': HashAlg(32, 'sha256', 'SHA256-RSA-PKCS'),
+    'SHA384': HashAlg(48, 'sha384', 'SHA384-RSA-PKCS'),
+    'SHA512': HashAlg(64, 'sha512', 'SHA512-RSA-PKCS'),
+}
+
+
+def pkcs11_list():
+    lines = subprocess.check_output(['pkcs11-tool', '--list-slots'], text=True).splitlines()
+    readers = []
+    for line in lines:
+        m = re.fullmatch('Slot ([0-9]+) \\(.*\\): .*', line)
+        if m is None:
+            continue
+        readers.append(m.group(1))
+
+    certs = {}
+    for reader in readers:
+        lines = subprocess.check_output(['pkcs15-tool', '--reader', reader, '-c'], text=True).splitlines()
+        for line in lines:
+            m = re.fullmatch('\tID *: *([0-9a-f]+)$', line)
+            if m is None:
+                continue
+            cert_id = m.group(1)
+            cert_data = subprocess.check_output(['pkcs15-tool', '--reader', reader, '--read-certificate', cert_id])
+            cert = cr.load_certificate(cr.FILETYPE_PEM, cert_data)
+            certs[reader,cert_id] = cert
+
+    return certs
+
+def pkcs11_sign(key, msg, hash_alg, pin):
+    hash_val = hashlib.new(hash_alg.hashlib_name, msg).digest()
+
+    reader,kid = key
+    env = os.environ.copy()
+    env['PIN'] = pin
+    proc = subprocess.run(['pkcs11-tool', '--slot', reader, '--id', kid, '-m', hash_alg.mech, '--pin', 'env:PIN', '--sign'], input=hash_val, stdout=subprocess.PIPE, env=env, check=True)
+
+    return proc.stdout
+
+
+def load_ts(ts):
+    t = time.strptime(ts.decode('ascii'), '%Y%m%d%H%M%SZ')
+    return datetime.datetime(
+        t.tm_year,
+        t.tm_mon,
+        t.tm_mday,
+        t.tm_hour,
+        t.tm_min,
+        t.tm_sec,
+        tzinfo=datetime.timezone.utc,
+    )
+
+def check_selector(cert, selector, valid_only):
+    if valid_only:
+        valid_from = load_ts(cert.get_notBefore())
+        valid_until = load_ts(cert.get_notAfter())
+        if not (valid_from <= datetime.datetime.now(datetime.timezone.utc) <= valid_until):
+            return False
+
+    vals = selector.get('issuers')
+    if vals is not None:
+        issuer_cn = [val.decode('utf-8', 'surrogateescape') for name,val in cert.get_issuer().get_components() if name == b'CN']
+        if issuer_cn:
+            issuer_cn = issuer_cn[0]
+        else:
+            issuer_cn = ''
+        if f'CN={issuer_cn}' not in vals:
+            return False
+
+    return True
+
+def describe_name(val):
+    return ', '.join(f'{k.decode("ascii")}={escape_bad(v.decode("utf-8", "surrogateescape"), ",=")}' for k,v in val.get_components())
+
+def describe_cert(cert):
+    valid_from = load_ts(cert.get_notBefore())
+    valid_until = load_ts(cert.get_notAfter())
+    return (
+        f'Issuer: {describe_name(cert.get_issuer())}\n'
+        f'Subject: {describe_name(cert.get_subject())}\n'
+        f'Valid from: {valid_from:%Y-%m-%d %H:%M:%S} UTC\n'
+        f'Valid until: {valid_until:%Y-%m-%d %H:%M:%S} UTC\n'
+    )
+
+def escape_bad(l, esc=''):
+    r = []
+    for c in l:
+        o = ord(c)
+        if 0xdc80 <= o <= 0xdd00:
+            r.append(f'\\x{ord(c)-0xdc00:02x}')
+            continue
+        if c == '\\':
+            r.append('\\\\')
+            continue
+        if o < 32 or o == 127 or (c in esc and o < 128):
+            r.append(f'\\x{ord(c):02x}')
+            continue
+        cat = unicodedata.category(c)
+        if cat in {'Cc', 'Cf', 'Cs'} or (cat.startswith('Z') and c != ' ') or c in esc:
+            if o < 0x10000:
+                r.append(f'\\u{o:04x}')
+            else:
+                r.append(f'\\U{o:08x}')
+            continue
+        r.append(c)
+    return ''.join(r)
+
+def print_blob(name, data, /):
+    if type(data) is str:
+        kind = 'text'
+    elif type(data) is bytes:
+        kind = 'blob'
+        data = '\n'.join(escape_bad(l.decode('utf-8', 'surrogateescape')) for l in data.splitlines())
+    else:
+        kind = 'repr'
+        data = repr(data)
+    print(f'{name} ({kind}):')
+
+    for line in data.splitlines():
+        print(f'  > {line!s}')
+
+def prompt(txt, /, default=False):
+    while True:
+        if default:
+            d = 'Y/n'
+        else:
+            d = 'y/N'
+        txt = input(f'{txt} [{d}]? ')
+        if txt in {'y', 'Y'}:
+            return True
+        if txt in {'n', 'N'}:
+            return False
+        if not txt:
+            return default
+
+STATUS_TEXT = {
+    200: 'OK',
+    201: 'Created',
+    202: 'Accepted',
+    203: 'Non-Authoritative Information',
+    204: 'No Content',
+    205: 'Reset Content',
+    206: 'Partial Content',
+    207: 'Multi-Status',
+    208: 'Already Reported',
+    226: 'IM Used',
+    300: 'Multiple Choices',
+    301: 'Moved Permanently',
+    302: 'Found',
+    303: 'See Other',
+    304: 'Not Modified',
+    305: 'Use Proxy',
+    306: 'Switch Proxy',
+    307: 'Temporary Redirect',
+    308: 'Permanent Redirect',
+    400: 'Bad Request',
+    401: 'Unauthorized',
+    402: 'Payment Required',
+    403: 'Forbidden',
+    404: 'Not Found',
+    405: 'Method Not Allowed',
+    406: 'Not Acceptable',
+    407: 'Proxy Authentication Required',
+    408: 'Request Timeout',
+    409: 'Conflict',
+    410: 'Gone',
+    411: 'Length Required',
+    412: 'Precondition Failed',
+    413: 'Payload Too Large',
+    414: 'URI Too Long',
+    415: 'Unsupported Media Type',
+    416: 'Range Not Satisfiable',
+    417: 'Expectation Failed',
+    418: 'I\'m a teapot',
+    421: 'Misdirected Request',
+    422: 'Unprocessable Entity',
+    423: 'Locked',
+    424: 'Failed Dependency',
+    425: 'Too Early',
+    426: 'Upgrade Required',
+    428: 'Precondition Required',
+    429: 'Too Many Requests',
+    431: 'Request Header Fields Too Large',
+    451: 'Unavailable For Legal Reasons',
+    500: 'Internal Server Error',
+    501: 'Not Implemented',
+    502: 'Bad Gateway',
+    503: 'Service Unavailable',
+    504: 'Gateway Timeout',
+    505: 'HTTP Version Not Supported',
+    506: 'Variant Also Negotiates',
+    507: 'Insufficient Storage',
+    508: 'Loop Detected',
+    510: 'Not Extended',
+    511: 'Network Authentication Required',
+}
+
+STATUS_FULL = {status: f'{status} {text}' for status,text in STATUS_TEXT.items()}
+
+class Req:
+    __slots__ = ('path', 'query_string', 'method', 'headers', 'input')
+
+    def __init__(self, env):
+        self.path = env['PATH_INFO']
+        self.query_string = env.get('QUERY_STRING')
+        self.method = env['REQUEST_METHOD']
+        self.input = env['wsgi.input']
+        headers = self.headers = {}
+        for k,v in env.items():
+            if not k.startswith('HTTP_'):
+                continue
+            k = ''.join(map(str.title, k[5:].split('_')))
+            headers[k] = v
+
+    def read_data(self, /):
+        d = self.input
+        del self.input
+        return d.read()
+
+    def resp(self, /, status, headers, data, *, content_type):
+        assert type(status) is int
+        assert type(data) is bytes
+        assert type(content_type) is str
+        headers = [*headers]
+        assert all(type(k) is str and type(v) is str for k,v in headers)
+        hdr = headers.append
+
+        hdr(('Access-Control-Allow-Methods', 'GET, POST'))
+        hdr(('Access-Control-Allow-Headers', 'Accept, Content-Type'))
+
+        try:
+            hdr(('Access-Control-Allow-Origin', self.headers['Origin']))
+        except KeyError:
+            pass
+
+        if self.method != 'OPTIONS':
+            hdr(('Content-Type', content_type))
+            hdr(('Content-Length', f'{len(data)}'))
+
+        hdr(('Cache-Control', 'no-cache'))
+
+        return (status, headers, data)
+
+    def resp_json(self, /, data, *, status=200, content_type='application/json;charset=UTF-8', headers=()):
+        return self.resp(
+            status,
+            headers,
+            json.dumps(data).encode('utf-8'),
+            content_type=content_type,
+        )
+
+    def resp_error(self, status, /):
+        try:
+            text = STATUS_TEXT[status]
+        except KeyError:
+            text = f'HTTP {status}'
+        if message is not None:
+            text = f'{text}: {message}'
+        return self.resp(
+            status,
+            (),
+            text.encode('utf-8'),
+            content_type='text/plain',
+        )
+
+    def resp_404(self, /):
+        return self.resp_error(404, **kw)
+
+    def resp_405(self, /):
+        return self.resp_error(405, **kw)
+
+    def resp_biss_error(self, /, status, message):
+        return self.resp_json({
+            'status': 'failed',
+            'reasonCode': status,
+            'reasonText': message,
+        }, status=status)
+
+    def resp_biss_ok(self, /, **kw):
+        return self.resp_json({
+            'status': 'ok',
+            'reasonCode': 200,
+            'reasonText': 'response.success',
+            **kw,
+        })
+
+
+cr.X509StoreFlags.NO_CHECK_TIME = 0x200000
+store = cr.X509Store()
+store.set_flags(cr.X509StoreFlags.CRL_CHECK_ALL | cr.X509StoreFlags.X509_STRICT | cr.X509StoreFlags.NO_CHECK_TIME)
+store.load_locations('cacerts.pem')
+
+certs = pkcs11_list()
+
+def req_options(req):
+    return req.resp(200, (), b'', content_type='')
+
+def req_get_version(req):
+    return req.resp_json({
+        'version': '2.30',
+        'httpMethods': 'GET, POST',
+        'contentTypes': 'data',
+        'signatureTypes': 'signature',
+        'selectorAvailable': True,
+        'hashAlgorithms': 'SHA256, SHA384, SHA512',
+    })
+
+def req_get_status(req):
+    return req.resp_json('')
+
+def req_post_getsigner(req):
+    data = json.loads(req.read_data().decode('utf-8'))
+
+    valid_only = data.get('showValidCerts', False)
+    # WAT
+    if valid_only == 'true':
+        valid_only = True
+    if valid_only == 'false':
+        valid_only = False
+    if type(valid_only) is not bool:
+        return req.resp_biss_error(400, 'error.request.bad-type')
+
+    selector = data.get('selector', {})
+
+    for cert in certs.values():
+        if check_selector(cert, selector, valid_only):
+            break
+    else:
+        return req.resp_biss_error(403, 'error.user-canceled')
+
+    return req.resp_biss_ok(
+        chain = [
+            base64.b64encode(cr.dump_certificate(cr.FILETYPE_ASN1, cert)).decode('ascii'),
+        ],
+    )
+
+def req_post_sign(req):
+    data = json.loads(req.read_data().decode('utf-8'))
+
+    try:
+        own_cert_data = data['signerCertificateB64']
+        own_cert_data = base64.b64decode(own_cert_data)
+    except (KeyError, IndexError):
+        return req.resp_biss_error(400, 'error.wsp-cert-not-found')
+
+    for own_key,own_cert in certs.items():
+        if cr.dump_certificate(cr.FILETYPE_ASN1, own_cert) == own_cert_data:
+            break
+    else:
+        return req.resp_biss_error(400, 'error.request.bad-type')
+
+    # BISS ignores it
+    # if data['contentType'] != 'data':
+    #     return req.resp_biss_error(400, 'error.request.bad-type')
+
+    try:
+        hash_alg = HASH_ALG[data.get('hashAlgorithm', 'SHA256')]
+    except KeyError:
+        return req.resp_biss_error(400, 'error.request.bad-type')
+
+    msgs = []
+    for i,(msg,sig,server_cert) in enumerate(zip(data['contents'], data['signedContents'], data['signedContentsCert'])):
+        msg = base64.b64decode(msg)
+        sig = base64.b64decode(sig)
+        server_cert = cr.load_certificate(cr.FILETYPE_ASN1, base64.b64decode(server_cert))
+
+        try:
+            # WAT?
+            cr.verify(server_cert, sig, hashlib.new(hash_alg.hashlib_name, msg).digest(), hash_alg.hashlib_name)
+        except cr.Error:
+            return req.resp_biss_error(400, 'error.request.signature-not-val')
+
+        verify_ctx = cr.X509StoreContext(store, server_cert, None)
+        try:
+            verify_ctx.verify_certificate()
+        except cr.X509StoreContextError as exc:
+            if not prompt(f'Server certificate invalid ({exc!s}); continue'):
+                return req.resp_biss_error(403, 'error.request.certificate-not-valid')
+
+        msgs.append((msg, server_cert))
+
+    t = data.get('confirmText')
+    if t is not None:
+        print_blob('Confirmation text', t)
+
+    t = data.get('additionalConfirmText')
+    if t is not None:
+        print_blob('Additional confirmation text', t)
+
+    for (msg,server_cert) in msgs:
+        print_blob(f'Message #{1+i}', msg)
+        print_blob('Server name', describe_cert(server_cert))
+
+    if not prompt('Sign'):
+        return req.resp_biss_error(403, 'error.user-canceled')
+
+    pin = getpass.getpass('PIN: ')
+
+    sigs = []
+    for (msg, server_cert) in msgs:
+        sigs.append(pkcs11_sign(own_key, msg, hash_alg, pin))
+
+    return req.resp_biss_ok(
+        signatures = [base64.b64encode(sig).decode('ascii') for sig in sigs],
+    )
+
+def req_post_chain_found(req):
+    # stub
+    data = json.loads(req.read_data().decode('utf-8'))
+    return req.resp_biss_ok(chainFound=True)
+
+
+def fix_methods(hnd):
+    assert type(hnd) is dict and all(type(k) is str for k in hnd)
+    hnd = hnd.copy()
+
+    if 'GET' in hnd and 'HEAD' not in hnd:
+        on_get = hnd['GET']
+        def on_head(req):
+            (status, headers, data) = on_get(req)
+            return (status, headers, None)
+        hnd['HEAD'] = on_head
+
+    hnd['OPTIONS'] = req_options
+
+    return hnd
+
+def resp_exc(env, resp, exc):
+    rdata = (
+        '<!doctype html><html><head><meta charset="utf-8" /><title>Internal Server Error</title></head><body>'
+        '<h1>Internal Server Error</h1>'
+        '</body></html>'
+    ).encode()
+    resp('500 Internal Server Error', [
+        ('Content-Type', 'text/html;charset=utf-8'),
+        ('Content-Length', f'{len(rdata)}'),
+    ])
+    return rdata
+
+
+handlers = {
+    '/version': {
+        'GET': req_get_version,
+    },
+    '/status': {
+        'GET': req_get_status,
+    },
+    '/getsigner': {
+        'POST': req_post_getsigner,
+    },
+    '/sign': {
+        'POST': req_post_sign,
+    },
+    '/chainFound': {
+        'POST': req_post_chain_found,
+    },
+}
+
+handlers = {path: fix_methods(sub) for path,sub in handlers.items()}
+
+def application(env, resp):
+    try:
+        req = Req(env)
+    except Exception as exc:
+        logging.exception('Failed to parse request')
+        return resp_exc(env, resp, exc)
+
+    try:
+        h = handlers[req.path]
+    except KeyError:
+        h = Req.resp_404
+    else:
+        try:
+            h = h[req.method]
+        except KeyError:
+            h = Req.resp_405
+
+    try:
+        rcode, rhdr, rdata = h(req)
+    except Exception as exc:
+        logging.exception('Request handler failed')
+        return resp_exc(env, resp, exc)
+
+    assert type(rcode) is int
+
+    try:
+        rcode = STATUS_FULL[rcode]
+    except KeyError:
+        rcode = f'{rcode}'
+
+    resp(rcode, rhdr)
+    return rdata
+
+tty = os.open('/dev/tty', os.O_RDWR)
+os.dup2(tty, 0)
+os.dup2(tty, 1)