Fix certificate reading #1
52
marginaltool
52
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,28 +35,37 @@ 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="*")
|
||||
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)
|
||||
|
||||
|
@ -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='<KEY ID>', 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='<KEY ID>', 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,7 +142,8 @@ 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'},
|
||||
headers=headers | {
|
||||
'Content-Type': 'application/json; charset=utf-8'},
|
||||
data=json.dumps(request).encode())
|
||||
if not r.text:
|
||||
break
|
||||
|
@ -130,4 +151,5 @@ if __name__ == '__main__':
|
|||
|
||||
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')
|
||||
|
|
Loading…
Reference in a new issue