1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
#!/usr/bin/env python3
import argparse
import base64
import configparser
import json
import os
import pathlib
import subprocess
import sys
import traceback
import urllib.parse
import getpass
# use requests instead of urllib.request for keep-alive connection
import requests
def sign(data, keyfile, unlock_key=None, engine=None):
# pkcs11-tool --id 02 -s -p $PIN -m RSA-PKCS
if engine is None:
cmd = ['openssl', 'pkeyutl', '-sign', '-inkey', keyfile, '-pkeyopt', 'digest:sha256']
raw_data = base64.b64decode(data)
env = None
elif engine == 'pkcs11':
env = {'PIN': unlock_key}
"""magic_prefix is ASN.1 DER for
DigestInfo ::= SEQUENCE {
digestAlgorithm DigestAlgorithm,
digest OCTET STRING
}
"""
magic_prefix = bytes.fromhex("3031300d060960864801650304020105000420")
raw_data = magic_prefix + base64.b64decode(data)
cmd = ['pkcs11-tool', '--id', keyfile, '-s', '-m', 'RSA-PKCS', '-p', 'env:PIN']
# cmd = ['openssl', 'pkeyutl', '-sign', '-pkeyopt', 'digest:sha256', '-engine', 'pkcs11', '-keyform', 'engine', '-inkey', keyfile]
p = subprocess.run(
cmd,
env=env,
input=raw_data,
capture_output=True)
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('-k', '--user-key', type=pathlib.Path, help='key file')
parser.add_argument('-c', '--user-cert', type=pathlib.Path, help='certificate file')
parser.add_argument('-e', '--engine', type=str, help='"pkcs11" for smart card')
args = parser.parse_args()
try:
# parse query string
params = urllib.parse.parse_qs(args.url.query)
url = params['baseUrl'][0]
token = params['accessToken'][0]
# if missing, get user key and cert from section [url] in ~/.margfools
config = configparser.ConfigParser()
config.read(os.path.expanduser('~') + '/.margfools')
if not args.user_key:
args.user_key = config.get(url, 'user-key')
if not args.user_cert:
args.user_cert = config.get(url, 'user-cert', fallback=None)
if not args.user_key:
print('user key not specified', file=sys.stderr)
sys.exit(1)
if not args.engine:
args.engine = config.get(url, 'engine')
engine = args.engine
user_keyfile = args.user_key
# base64.b64encode
unlock_key = None
if engine is None:
if not args.user_cert:
print('user cert not specified', file=sys.stderr)
sys.exit(1)
user_cert = ''.join(line.strip() for line in open(args.user_cert) if not line.startswith('-----'))
elif engine == 'pkcs11':
user_cert = base64.b64encode(subprocess.run(["pkcs11-tool", "--read-object", "--type", "cert", "--id", user_keyfile], capture_output=True).stdout).decode()
unlock_key = getpass.getpass("PIN:")
session = requests.Session()
headers={'Authorization': f'Bearer {token}'}
# delete old signing session
r = session.delete(f'{url}/signatures/{params["startSigningToken"][0]}', headers=headers)
# register a certificate or sign a document, makes no difference to us
if params.get('registerCertificate'):
q = {'registerCertificate': 1}
else:
q = {'documentId': [i for i in params['documentId'][0].split(',')]}
qs = urllib.parse.urlencode(q, doseq=True)
r = session.post(f'{url}/signatures?{qs}', headers=headers)
# get signature request and mix in my secrets and publics
request = json.loads(r.text)
request['AuthenticationToken'] = token
request['CertificatePublicKey'] = user_cert
# keep signing whatever they send us
while True:
for name in ('AttachmentHashes', 'XmlHashes'):
if request.get(name) is not None:
request[f'Signed{name}'] = [sign(e, user_keyfile, unlock_key, engine=engine) for e in request[name]]
d = json.dumps(request)
d = d.encode()
r = session.put(f'{url}signatures/{request["SignatureRequestId"]}',
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:
traceback.print_exc()
# Don't close terminal immediately on fail
input()
|
