python - HSM 与 Python 请求模块的集成

Question

所以我正在编写一个需要使用客户端证书(相互身份验证)向服务器进行身份验证的应用程序。客户端证书 key 存储在 HSM(Gemalto 的)中。我将 OpenSSL 与 Luna 客户端集成，但请求模块需要一个文件:

from requests import Session

session: Session = Session()

session.cert = (
    "/ssl/client.pem",
    "/ssl/client.key"
)
session.verify = "/ssl/server.pem"

我的问题是当私钥在 HSM 中时，我找不到绑定(bind)私钥的方法。这是我到目前为止对 pycryptoki 的尝试图书馆:

from pycryptoki.session_management import (
    c_initialize_ex,
    c_open_session_ex,
    login_ex,
    c_logout_ex,
    c_close_session_ex,
    c_finalize_ex,
)
from requests import Session

c_initialize_ex()
auth_session = c_open_session_ex(0)
login_ex(auth_session, 0, "some-pass")

session: Session = Session()

session.cert = (
    "/ssl/client.pem",
    "rsa-private-156405640312",
)
session.verify = "/ssl/server.pem"

...

c_logout_ex(auth_session)
c_close_session_ex(auth_session)
c_finalize_ex()

前一段时间我在这里打开了一个问题，我必须完成应用程序的实现，所以我把 HSM 集成放在了冰上，但我需要在投入生产之前完成这项工作:https://github.com/gemalto/pycryptoki/issues/17

我也试过使用 py-hsm但它是一个低级的 api 库，我还打开了一个 issue那里有我的代码示例:

from pyhsm.hsmclient import HsmClient
from requests import Session

c = HsmClient(pkcs11_lib="/usr/lib/libCryptoki2_64.so")
c.open_session(slot=0)
c.login(pin="some-code")

session: Session = Session()
session.cert = "/ssl/client.pem"

c.logout()
c.close_session()

任何人都可以提供一个使用 HSM 中的证书对进行相互身份验证的示例吗？如果您有 C/C++ 语言的东西，那就太好了，我可以实现我的请求函数并将其包装在我的 Python 代码中。

提前致谢!

Answer 1

我已经测试了几乎所有的 Python 包装器来做同样的事情。PyKCS11 并不是很可靠。

我建议使用以下两种可能性之一:

<强>1。 PyCurl

当您正确配置 OpenSSL 和 cURL 时，cUrl 的 Python 包装器可以执行此操作。

这是一个简单的实现:

import pycurl
from io import BytesIO
import pprint
import json


c = pycurl.Curl()
url = 'https://yourserver/endpoint'
c.setopt(pycurl.URL, url)

# set proxy-insecure
c.setopt(c.SSL_VERIFYPEER, 0)
c.setopt(c.SSL_VERIFYHOST, False)
c.setopt(c.VERBOSE, 0)

c.setopt(pycurl.SSLENGINE, 'pkcs11')
c.setopt(pycurl.SSLCERTTYPE, 'eng')
c.setopt(pycurl.SSLKEYTYPE, 'eng')

c.setopt(pycurl.SSLCERT, 'pkcs11:model=XXX;manufacturer=YYYYY;serial=ZZZZ;'
                        'token=AAAAA;id=BBBBBBBBB;'
                        'object=CCCCCC;type=cert;pin-value=pin-pin')
c.setopt(pycurl.SSLKEY, 'pkcs11:model=XXX;manufacturer=YYYYY;serial=ZZZZ;'
                        'token=AAAAA;id=BBBBBBBBB;'
                        'object=CCCCCC;type=private;pin-value=pin-pin')


# set headers
c.setopt(pycurl.HEADER, True)
# c.setopt(pycurl.USERAGENT, 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:8.0) Gecko/20100101 Firefox/8.0')
c.setopt(pycurl.HTTPHEADER, ("HEADER_TO_ADD:VALUE",))

buffer = BytesIO()

c.setopt(c.WRITEDATA, buffer)
c.perform()
# HTTP response code, e.g. 200.
print('>>> Status: %d' % c.getinfo(c.RESPONSE_CODE))
# Elapsed time for the transfer.
print('>>> Time: %f' % c.getinfo(c.TOTAL_TIME))

# getinfo must be called before close.
c.close()

body = buffer.getvalue().decode('utf-8')
print('>>> Body:\n', body)

if body.find('{') >= 0:
    body = body[body.find('{'):]

dictionary = json.loads(body)
pprint.pprint(dictionary)

请注意，pkcs#11 URI 方案必须符合 RFC7512可以使用以下命令发现它:

p11tool --provider=/usr/lib/libeTPkcs11.so --list-all

包括 pin 在内的所有字段都必须经过 url 编码。使用在线网站编码/解码字符串(pin)<->url_encoded

<强>2。 M2加密

在我看来，这是为 Python 完成的最佳 pkcs#11 实现。由于 requests.Session.mount 方法和 requests.adapters.BaseAdapter，它允许覆盖 urllib2 和请求调用。

我在这里放了一些代码以将其与 urllib2 一起使用:

from M2Crypto import m2urllib2 as urllib2
from M2Crypto import m2, SSL, Engine


# load dynamic engine
e = Engine.load_dynamic_engine("pkcs11", "/usr/lib/x86_64-linux-gnu/engines-1.1/libpkcs11.so")
pk = Engine.Engine("pkcs11")
pk.ctrl_cmd_string("MODULE_PATH", "/usr/lib/libeTPkcs11.so")

m2.engine_init(m2.engine_by_id("pkcs11"))
pk.ctrl_cmd_string("PIN", 'pin-pin')
cert = pk.load_certificate('pkcs11:model=XXX;manufacturer=YYYYY;serial=ZZZZ;'
                        'token=AAAAA;id=BBBBBBBBB;'
                        'object=CCCCCC')
key = pk.load_private_key('pkcs11:model=XXX;manufacturer=YYYYY;serial=ZZZZ;'
                        'token=AAAAA;id=BBBBBBBBB;'
                        'object=CCCCCC', pin='pin-pin')


ssl_context = SSL.Context('tls')
ssl_context.set_cipher_list('EECDH+AESGCM:EECDH+aECDSA:EECDH+aRSA:EDH+AESGCM:EDH+aECDSA:EDH+aRSA:!SHA1:!SHA256:!SHA384:!MEDIUM:!LOW:!EXP:!aNULL:!eNULL:!PSK:!SRP:@STRENGTH')
ssl_context.set_default_verify_paths()
ssl_context.set_allow_unknown_ca(True)

SSL.Connection.postConnectionCheck = None

m2.ssl_ctx_use_x509(ssl_context.ctx, cert.x509)
m2.ssl_ctx_use_pkey_privkey(ssl_context.ctx, key.pkey)

opener = urllib2.build_opener(ssl_context)
urllib2.install_opener(opener)

url = 'https://yourserver/endpoint'

content = urllib2.urlopen(url=url).read()
# content = opener.open(url)
print(content)

请注意，对于 PyCurl，我们不会在 pkcs#11 URI 中指示类型和 pin。当传递给 load_private_key 时，PIN 表示为字符串而不是 url 编码。

最后，我为 HttpAdapter 找到了一个很好的实现来挂载请求以进行调用。 Here是原始实现。

我添加了一些行来支持 PIN 码并禁用检查主机名:

M2Crypto.SSL.Connection.postConnectionCheck = None

下面是如何安装适配器:

from requests import Session
from m2requests import M2HttpsAdapter
import pprint, json


request = Session()
m2httpsadapter = M2HttpsAdapter()
# Added by me to set a pin when loading private key
m2httpsadapter.pin = 'pin-pin'
# Need this attribute set to False else we cannot use direct IP (added by me to disable checking hostname)
m2httpsadapter.check_hostname = False
request.mount("https://", m2httpsadapter)

request.cert=('pkcs11:model=XXX;manufacturer=YYYYY;serial=ZZZZ;'
                        'token=AAAAA;id=BBBBBBBBB;'
                        'object=CCCCCC',
              'pkcs11:model=XXX;manufacturer=YYYYY;serial=ZZZZ;'
                        'token=AAAAA;id=BBBBBBBBB;'
                        'object=CCCCCC')

headers = {'HEADER_TO_ADD_IF_WANTED': 'VALUE', }
r = request.get("https://yourserver/endpoint", headers=headers, verify=False)

print(r.status_code)
pprint.pprint(json.loads(r.raw.data.decode('utf-8')))