Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/securesystemslib/signer/_hsm_signer.py: 29%

1"""Hardware Security Module (HSM) Signer 

2 

3Uses python-pkcs11 API to create signatures with HSMs (e.g. YubiKey) and to export 

4the related public keys. 

5 

6""" 

7 

8from __future__ import annotations 

9 

10import binascii 

11import hashlib 

12import os 

13from urllib import parse 

14 

15from securesystemslib.exceptions import UnsupportedLibraryError 

16from securesystemslib.signer._constants import ECDSA_SHA2_NISTP256, ECDSA_SHA2_NISTP384 

17from securesystemslib.signer._key import Key, SSlibKey 

18from securesystemslib.signer._signature import Signature 

19from securesystemslib.signer._signer import SecretsHandler, Signer 

20 

21CRYPTO_IMPORT_ERROR = None 

22try: 

23 from cryptography.hazmat.primitives.asymmetric.ec import ( 

24 SECP256R1, 

25 SECP384R1, 

26 EllipticCurvePublicKey, 

27 ) 

28 from cryptography.hazmat.primitives.serialization import load_der_public_key 

29 

30 _SCHEME_FOR_CURVE = { 

31 SECP256R1: ECDSA_SHA2_NISTP256, 

32 SECP384R1: ECDSA_SHA2_NISTP384, 

33 } 

34 

35except ImportError: 

36 CRYPTO_IMPORT_ERROR = "'cryptography' required" 

37 

38PKCS11_IMPORT_ERROR = None 

39try: 

40 import pkcs11 

41 import pkcs11.util.ec 

42 from pkcs11.exceptions import ( 

43 MultipleObjectsReturned, 

44 MultipleTokensReturned, 

45 NoSuchKey, 

46 NoSuchToken, 

47 ) 

48except ImportError: 

49 PKCS11_IMPORT_ERROR = "'python-pkcs11' required" 

50 

51 

52class HSMSigner(Signer): 

53 """Hardware Security Key Signer. 

54 

55 HSMSigner uses PKCS#11 API to sign with hardware security keys like YubiKeys. 

56 It Supports signing schemes ECDSA_SHA2_NISTP256 and ECDSA_SHA2_NISTP384. 

57 

58 The environment variable ``PYKCS11LIB`` must be set to the path of the PKCS#11 

59 shared library module for the hardware token (for example, 

60 ``/usr/lib/x86_64-linux-gnu/libykcs11.so`` or 

61 ``/usr/local/lib/libykcs11.dylib``). 

62 

63 The private key URI scheme is: ``hsm:[<KEYID>][?label=<LABEL>]`` where both KEYID 

64 and LABEL are optional. Example URIs: 

65 

66 * ``hsm:``: 

67 Sign with a key with default keyid 2 (PIV digital signature slot 9c) on the 

68 only token/smartcard available. 

69 * ``hsm:2?label=YubiKey+PIV+%2315835999``: 

70 Sign with key with keyid 2 (PIV slot 9c) on a token with label 

71 "YubiKey+PIV+%2315835999". 

72 

73 Requires environment variable ``PYKCS11LIB`` to contain path to PKCS#11 module. 

74 

75 Usage:: 

76 

77 # Store public key and URI for your HSM device for later use. By default 

78 # slot 9c is selected. 

79 uri, pubkey = HSMSigner.import_() 

80 

81 # Later, use the uri and pubkey to sign (providing PIN via SecretsHandler) 

82 def pin_handler(secret: str) -> str: 

83 return getpass(f"Enter {secret}: ") 

84 

85 signer = Signer.from_priv_key_uri(uri, pubkey, pin_handler) 

86 sig = signer.sign(b"DATA") 

87 pubkey.verify_signature(sig, b"DATA") 

88 """ 

89 

90 SCHEME_KEYID = 2 

91 SCHEME = "hsm" 

92 SECRETS_HANDLER_MSG = "pin" 

93 

94 def __init__( 

95 self, 

96 hsm_keyid: int, 

97 public_key: SSlibKey, 

98 pin_handler: SecretsHandler, 

99 token_label: str | None = None, 

100 ): 

101 if CRYPTO_IMPORT_ERROR: 

102 raise UnsupportedLibraryError(CRYPTO_IMPORT_ERROR) 

103 

104 if PKCS11_IMPORT_ERROR: 

105 raise UnsupportedLibraryError(PKCS11_IMPORT_ERROR) 

106 

107 if public_key.scheme not in [ECDSA_SHA2_NISTP256, ECDSA_SHA2_NISTP384]: 

108 raise ValueError(f"unsupported scheme {public_key.scheme}") 

109 

110 self.hsm_keyid = hsm_keyid 

111 self.token_label = token_label 

112 self._public_key = public_key 

113 self.pin_handler = pin_handler 

114 

115 @property 

116 def public_key(self) -> SSlibKey: 

117 return self._public_key 

118 

119 @staticmethod 

120 def _find_token(token_label: str | None = None) -> pkcs11.Token: 

121 """Return the PKCS#11 token that matches token_label.""" 

122 lib_path = os.environ.get("PYKCS11LIB") 

123 if not lib_path: 

124 raise ValueError("PYKCS11LIB environment variable must be set") 

125 try: 

126 return pkcs11.lib(lib_path).get_token( 

127 token_label=token_label, token_flags=pkcs11.TokenFlag.TOKEN_INITIALIZED 

128 ) 

129 except NoSuchToken: 

130 label_str = f" for label {token_label}" if token_label else "" 

131 raise ValueError(f"No PKCS#11 token found{label_str}") 

132 except MultipleTokensReturned: 

133 label_str = f" for label {token_label}" if token_label else "" 

134 raise ValueError(f"Multiple PKCS#11 tokens found{label_str}") 

135 

136 @staticmethod 

137 def _find_pub_key(session: pkcs11.Session, keyid: int) -> EllipticCurvePublicKey: 

138 """Find ecdsa public key on HSM, return corresponding `cryptography` EC key.""" 

139 id_bytes = pkcs11.util.biginteger(keyid) 

140 object_class = pkcs11.ObjectClass.PUBLIC_KEY 

141 try: 

142 pkcs11_key = session.get_key(object_class, pkcs11.KeyType.EC, id=id_bytes) 

143 except NoSuchKey: 

144 raise ValueError("could not find ECDSA key on the PKCS#11 token") 

145 except MultipleObjectsReturned: 

146 raise ValueError("found multiple ECDSA keys on the PKCS#11 token") 

147 

148 if not isinstance(pkcs11_key, pkcs11.PublicKey): 

149 raise AssertionError("PKCS key is not a public key") 

150 key = load_der_public_key(pkcs11.util.ec.encode_ec_public_key(pkcs11_key)) 

151 if not isinstance(key, EllipticCurvePublicKey): 

152 raise AssertionError("PKCS key is not an EC key") 

153 return key 

154 

155 @staticmethod 

156 def _find_signing_key(session: pkcs11.Session, keyid: int) -> pkcs11.SignMixin: 

157 """Find ecdsa signing key on HSM.""" 

158 id_bytes = pkcs11.util.biginteger(keyid) 

159 object_class = pkcs11.ObjectClass.PRIVATE_KEY 

160 try: 

161 pkcs11_key = session.get_key(object_class, pkcs11.KeyType.EC, id=id_bytes) 

162 except NoSuchKey: 

163 raise ValueError("could not find ECDSA private key on the PKCS#11 token") 

164 except MultipleObjectsReturned: 

165 raise ValueError("found multiple ECDSA private keys on the PKCS#11 token") 

166 if not isinstance(pkcs11_key, pkcs11.SignMixin): 

167 raise AssertionError("Found private key cannot be used for signing") 

168 return pkcs11_key 

169 

170 @classmethod 

171 def import_( 

172 cls, 

173 hsm_keyid: int | None = None, 

174 token_label: str | None = None, 

175 secrets_handler: SecretsHandler | None = None, 

176 ) -> tuple[str, SSlibKey]: 

177 """Import public key and signer details from HSM. 

178 

179 Either only one cryptographic token must be present when importing or a 

180 token_label that matches a single token must be provided. 

181 

182 import_() should be called once and the returned URI and public 

183 key should be stored for later use. 

184 

185 Arguments: 

186 hsm_keyid: Key identifier on the token. 

187 Default is 2 (meaning PIV key slot 9c). 

188 token_label: Token label to filter the correct cryptographic token. 

189 If no label is provided one is built from the token found. 

190 secrets_handler: Will be called if reading the public key requires PIN. 

191 

192 Raises: 

193 UnsupportedLibraryError: ``python-pkcs11`` and ``cryptography`` 

194 libraries not found. 

195 ValueError: A matching HSM device or key could not be found. 

196 """ 

197 if CRYPTO_IMPORT_ERROR: 

198 raise UnsupportedLibraryError(CRYPTO_IMPORT_ERROR) 

199 

200 if PKCS11_IMPORT_ERROR: 

201 raise UnsupportedLibraryError(PKCS11_IMPORT_ERROR) 

202 

203 hsm_keyid = hsm_keyid if hsm_keyid is not None else cls.SCHEME_KEYID 

204 token = cls._find_token(token_label) 

205 uri = f"{cls.SCHEME}:{hsm_keyid}" 

206 if token.label: 

207 uri = f"{uri}?{parse.urlencode({'label': token.label})}" 

208 

209 try: 

210 with token.open() as session: 

211 pubkey = cls._find_pub_key(session, hsm_keyid) 

212 except ValueError: 

213 # key not found while unauthenticated: it may be set to CKA_PRIVATE 

214 if secrets_handler is None: 

215 raise ValueError( 

216 "No keys found unauthenticated and no secrets handler provided" 

217 ) 

218 pin = secrets_handler(cls.SECRETS_HANDLER_MSG) 

219 with token.open(user_pin=pin) as session: 

220 pubkey = cls._find_pub_key(session, hsm_keyid) 

221 

222 if type(pubkey.curve) not in _SCHEME_FOR_CURVE: 

223 raise ValueError(f"{pubkey.curve.name} is not a supported EC curve") 

224 

225 scheme = _SCHEME_FOR_CURVE[type(pubkey.curve)] 

226 key = SSlibKey.from_crypto(pubkey, scheme=scheme) 

227 return uri, key 

228 

229 @classmethod 

230 def from_priv_key_uri( 

231 cls, 

232 priv_key_uri: str, 

233 public_key: Key, 

234 secrets_handler: SecretsHandler | None = None, 

235 ) -> HSMSigner: 

236 if not isinstance(public_key, SSlibKey): 

237 raise ValueError(f"expected SSlibKey for {priv_key_uri}") 

238 

239 uri = parse.urlparse(priv_key_uri) 

240 

241 if uri.scheme != cls.SCHEME: 

242 raise ValueError(f"HSMSigner does not support {priv_key_uri}") 

243 

244 keyid = int(uri.path) if uri.path else cls.SCHEME_KEYID 

245 token_label = dict(parse.parse_qsl(uri.query)).get("label") 

246 

247 if secrets_handler is None: 

248 raise ValueError("HSMSigner requires a secrets handler") 

249 

250 return cls(keyid, public_key, secrets_handler, token_label) 

251 

252 def sign(self, payload: bytes) -> Signature: 

253 """Signs payload with Hardware Security Module (HSM). 

254 

255 Arguments: 

256 payload: bytes to be signed. 

257 

258 Raises: 

259 ValueError: No compatible key for ``hsm_keyid`` found on HSM. 

260 

261 Returns: 

262 Signature. 

263 """ 

264 

265 hasher = hashlib.new(name=f"sha{self.public_key.scheme[-3:]}") 

266 hasher.update(payload) 

267 

268 pin = self.pin_handler(self.SECRETS_HANDLER_MSG) 

269 token = self._find_token(self.token_label) 

270 

271 with token.open(user_pin=pin) as session: 

272 key = self._find_signing_key(session, self.hsm_keyid) 

273 signature = key.sign(hasher.digest(), mechanism=pkcs11.Mechanism.ECDSA) 

274 

275 # Convert the PKCS#11 raw signature to ASN.1 DER 

276 asn_sig = pkcs11.util.ec.encode_ecdsa_signature(signature) 

277 hex_asn_sig = binascii.hexlify(asn_sig).decode("ascii") 

278 

279 return Signature(self.public_key.keyid, hex_asn_sig)