diff --git a/marginaltool b/marginaltool index 9d07700..d737c1b 100755 --- a/marginaltool +++ b/marginaltool @@ -13,6 +13,7 @@ import urllib.parse # use requests instead of urllib.request for keep-alive connection import requests + def init(args): args.params = urllib.parse.parse_qs(args.url.query) args.access_token = args.params["accessToken"][0] @@ -34,35 +35,44 @@ def init(args): if not args.keyfile or not args.certfile: raise Exception('key or certificate file not specified') cert = open(args.certfile).read() - args.cert = ''.join(cert[cert.find('-----BEGIN CERTIFICATE-----' + 27):cert.find('-----END CERTIFICATE-----')]) + args.cert = ''.join( + cert.split('-----BEGIN CERTIFICATE-----', 1)[1] + .split('-----END CERTIFICATE-----', 1)[0] + ) case 'pkcs11': if not args.id: args.id = config.get(args.url, 'id', fallback=None) if not args.id: raise Exception('key ID not specified') - args.cert = base64.b64encode(subprocess.run(['pkcs11-tool', '--read-object', '--type', 'cert', '--id', args.id], capture_output=True).stdout).decode() + args.cert = base64.b64encode(subprocess.run( + ['pkcs11-tool', '--read-object', '--type', 'cert', '--id', args.id], + capture_output=True).stdout + ).decode() # read the PIN once to avoid prompting for each document - import tkinter.simpledialog # only needed for PIN entry - args.pin = tkinter.simpledialog.askstring('marginaltool', 'PIN', show="*") + import tkinter.simpledialog # only needed for PIN entry + args.pin = tkinter.simpledialog.askstring( + 'marginaltool', 'PIN', show="*") case '_': raise Exception(f'invalid engine {args.engine}') + def sign(b64data, args): match args.engine: case 'file': if not args.keyfile: raise Exception('keyfile not specified') - cmd = ['openssl', 'pkeyutl', '-sign', '-inkey', args.keyfile, '-pkeyopt', 'digest:sha256'] + cmd = ['openssl', 'pkeyutl', '-sign', '-inkey', + args.keyfile, '-pkeyopt', 'digest:sha256'] env = None data = base64.b64decode(b64data) case 'pkcs11': if not args.id: raise Exception('key ID not specified') - digest_info = { # from RFC 3447 + digest_info = { # from RFC 3447 'MD2': '3020300c06082a864886f70d020205000410', 'MD5': '3020300c06082a864886f70d020505000410', 'SHA-1': '3021300906052b0e03021a05000414', @@ -70,9 +80,11 @@ def sign(b64data, args): 'SHA-384': '3041300d060960864801650304020205000430', 'SHA-512': '3051300d060960864801650304020305000440' } - cmd = ['pkcs11-tool', '--id', args.id, '-s', '-m', 'RSA-PKCS', '-p', 'env:PIN'] + cmd = ['pkcs11-tool', '--id', args.id, + '-s', '-m', 'RSA-PKCS', '-p', 'env:PIN'] env = {'PIN': args.pin} - data = bytes.fromhex(digest_info['SHA-256']) + base64.b64decode(b64data) + data = bytes.fromhex( + digest_info['SHA-256']) + base64.b64decode(b64data) case '_': raise Exception(f'invalid engine {args.engine}') @@ -83,13 +95,19 @@ def sign(b64data, args): return base64.b64encode(p.stdout).decode() + if __name__ == '__main__': - parser = argparse.ArgumentParser(description='Fake the MargTools application.') - parser.add_argument('url', type=urllib.parse.urlparse, help='bc-digsign:// url') - parser.add_argument('-e', '--engine', choices=('file', 'pkcs11'), help='use key file or PKCS11 token?') + parser = argparse.ArgumentParser( + description='Fake the MargTools application.') + parser.add_argument('url', type=urllib.parse.urlparse, + help='bc-digsign:// url') + parser.add_argument('-e', '--engine', choices=('file', + 'pkcs11'), help='use key file or PKCS11 token?') parser.add_argument('-k', '--keyfile', type=pathlib.Path, help='key file') - parser.add_argument('-c', '--certfile', type=pathlib.Path, help='certificate file') - parser.add_argument('-i', '--id', type=int, metavar='', help='key ID on PKCS11 token') + parser.add_argument('-c', '--certfile', + type=pathlib.Path, help='certificate file') + parser.add_argument('-i', '--id', type=int, + metavar='', help='key ID on PKCS11 token') args = parser.parse_args() try: @@ -100,13 +118,15 @@ if __name__ == '__main__': headers = {'Authorization': f'Bearer {args.access_token}'} # delete old signing session - r = session.delete(f'{args.url}/signatures/{args.params["startSigningToken"][0]}', headers=headers) + r = session.delete( + f'{args.url}/signatures/{args.params["startSigningToken"][0]}', headers=headers) # register a certificate or sign a document, makes no difference to us if args.params.get('registerCertificate'): q = {'registerCertificate': 1} else: - q = {'documentId': [i for i in args.params['documentId'][0].split(',')]} + q = {'documentId': [ + i for i in args.params['documentId'][0].split(',')]} qs = urllib.parse.urlencode(q, doseq=True) r = session.post(f'{args.url}/signatures?{qs}', headers=headers) @@ -122,12 +142,14 @@ if __name__ == '__main__': request[f'Signed{name}'] = [sign(v, args) for v in values] r = session.put(f'{args.url}signatures/{request["SignatureRequestId"]}', - headers=headers | {'Content-Type': 'application/json; charset=utf-8'}, - data=json.dumps(request).encode()) + headers=headers | { + 'Content-Type': 'application/json; charset=utf-8'}, + data=json.dumps(request).encode()) if not r.text: break request |= json.loads(r.text) except Exception as ex: print(f'error: {ex}', file=sys.stderr) - input('press enter to exit') # don’t close terminal immediately on fail + # don’t close terminal immediately on fail + input('press enter to exit')