_synta/pkcs11.rs
1//! Python bindings for PKCS#11 token management.
2//!
3//! Exposes the `synta.pkcs11` submodule with:
4//! - `SlotInfo` — information about a PKCS#11 slot / token
5//! - `KeyInfo` — information about a private key stored on a token
6//! - `Pkcs11Token` — session handle for a single named token
7//! - `list_slots(module=None)` — enumerate all token slots
8
9use pyo3::exceptions::PyValueError;
10use pyo3::prelude::*;
11
12use synta_certificate::crypto::{KeySpec, TokenManager};
13use synta_certificate::pkcs11_mgmt::Pkcs11Manager;
14use synta_certificate::{merge_object_label, Pkcs11Uri, Pkcs11UriAttributes};
15
16use crate::crypto_keys::PyPrivateKey;
17
18// ── SlotInfo ──────────────────────────────────────────────────────────────────
19
20/// Information about a PKCS#11 token slot.
21///
22/// Returned by :func:`list_slots`.
23#[pyclass(frozen, name = "SlotInfo")]
24pub struct PySlotInfo {
25 pub(crate) slot_id: u64,
26 pub(crate) token_label: String,
27 pub(crate) manufacturer_id: String,
28 pub(crate) model: String,
29 pub(crate) serial_number: String,
30 pub(crate) flags: u64,
31}
32
33#[pymethods]
34impl PySlotInfo {
35 /// Numeric PKCS#11 slot ID (``CK_SLOT_ID``).
36 #[getter]
37 fn slot_id(&self) -> u64 {
38 self.slot_id
39 }
40
41 /// Token label (`CKA_LABEL` on the token object), space-padding removed.
42 #[getter]
43 fn token_label(&self) -> &str {
44 &self.token_label
45 }
46
47 /// Manufacturer ID field from ``CK_TOKEN_INFO``.
48 #[getter]
49 fn manufacturer_id(&self) -> &str {
50 &self.manufacturer_id
51 }
52
53 /// Model field from ``CK_TOKEN_INFO``.
54 #[getter]
55 fn model(&self) -> &str {
56 &self.model
57 }
58
59 /// Serial number field from ``CK_TOKEN_INFO``.
60 #[getter]
61 fn serial_number(&self) -> &str {
62 &self.serial_number
63 }
64
65 /// Subset of ``CKF_*`` flag bits from ``CK_TOKEN_INFO`` (PKCS#11 v3.0 §4.9).
66 ///
67 /// Four bits are captured:
68 ///
69 /// - ``0x0002`` (``CKF_WRITE_PROTECTED``) — token is read-only
70 /// - ``0x0004`` (``CKF_LOGIN_REQUIRED``) — a PIN must be presented before accessing private objects
71 /// - ``0x0100`` (``CKF_PROTECTED_AUTHENTICATION_PATH``) — PIN entry is via an on-device keypad
72 /// - ``0x0400`` (``CKF_TOKEN_INITIALIZED``) — token has been initialised
73 ///
74 /// All other bits are zero.
75 #[getter]
76 fn flags(&self) -> u64 {
77 self.flags
78 }
79
80 fn __repr__(&self) -> String {
81 format!(
82 "SlotInfo(slot_id={}, token_label={:?}, manufacturer_id={:?})",
83 self.slot_id, self.token_label, self.manufacturer_id
84 )
85 }
86}
87
88// ── KeyInfo ───────────────────────────────────────────────────────────────────
89
90/// Information about a private key object stored on a PKCS#11 token.
91///
92/// Returned by :meth:`Pkcs11Token.list_keys`.
93#[pyclass(frozen, name = "KeyInfo")]
94pub struct PyKeyInfo {
95 pub(crate) label: String,
96 pub(crate) id: Vec<u8>,
97 pub(crate) key_type: String,
98 pub(crate) key_bits: u32,
99}
100
101#[pymethods]
102impl PyKeyInfo {
103 /// Key label (``CKA_LABEL``).
104 #[getter]
105 fn label(&self) -> &str {
106 &self.label
107 }
108
109 /// Raw ``CKA_ID`` bytes; may be empty.
110 #[getter]
111 fn id<'py>(&self, py: Python<'py>) -> Bound<'py, pyo3::types::PyBytes> {
112 pyo3::types::PyBytes::new(py, &self.id)
113 }
114
115 /// Human-readable key type: ``"RSA"``, ``"EC"``, ``"Ed"``, ``"ML-DSA"``,
116 /// ``"ML-KEM"``, or ``"Unknown"``.
117 #[getter]
118 fn key_type(&self) -> &str {
119 &self.key_type
120 }
121
122 /// RSA modulus length in bits, derived from ``CKA_MODULUS``.
123 ///
124 /// ``0`` for all non-RSA key types (EC, Ed25519, Ed448, ML-DSA, ML-KEM) because
125 /// those algorithms do not expose ``CKA_MODULUS``.
126 #[getter]
127 fn key_bits(&self) -> u32 {
128 self.key_bits
129 }
130
131 fn __repr__(&self) -> String {
132 format!(
133 "KeyInfo(label={:?}, key_type={:?}, key_bits={})",
134 self.label, self.key_type, self.key_bits
135 )
136 }
137}
138
139// ── Pkcs11Token ───────────────────────────────────────────────────────────────
140
141/// A handle for PKCS#11 management operations on a single named token.
142///
143/// Example:
144/// >>> import synta.pkcs11
145/// >>> token = synta.pkcs11.Pkcs11Token(
146/// ... "pkcs11:token=MySoftHSM2Token0?pin-value=1234"
147/// ... )
148/// >>> token.list_keys()
149/// [KeyInfo(label='caKey', key_type='RSA', key_bits=2048)]
150#[pyclass(frozen, name = "Pkcs11Token")]
151pub struct PyPkcs11Token {
152 mgr: Pkcs11Manager,
153 /// Verbatim URI string (used for repr and for building per-operation URIs).
154 uri: String,
155 /// Decoded attributes from `uri`, cached so methods do not need to re-parse.
156 attrs: Pkcs11UriAttributes,
157}
158
159fn make_mgr(uri: &Pkcs11Uri, module: Option<&str>) -> PyResult<Pkcs11Manager> {
160 match module {
161 Some(path) => Pkcs11Manager::new(path).map_err(|e| PyValueError::new_err(format!("{e}"))),
162 None => Pkcs11Manager::from_uri(uri).map_err(|e| PyValueError::new_err(format!("{e}"))),
163 }
164}
165
166/// Redact the `pin-value=…` component from a PKCS#11 URI for safe display.
167fn redact_pin(uri: &str) -> String {
168 let body = uri.strip_prefix("pkcs11:").unwrap_or(uri);
169 let (path, query) = body.split_once('?').unwrap_or((body, ""));
170 if query.is_empty() {
171 return uri.to_owned();
172 }
173 let redacted: Vec<&str> = query
174 .split('&')
175 .map(|seg| {
176 if seg.starts_with("pin-value=") {
177 "pin-value=***"
178 } else {
179 seg
180 }
181 })
182 .collect();
183 format!("pkcs11:{path}?{}", redacted.join("&"))
184}
185
186#[pymethods]
187impl PyPkcs11Token {
188 /// Create a PKCS#11 token handle.
189 ///
190 /// ``uri`` must be a ``pkcs11:`` URI identifying the token; include a
191 /// ``?pin-value=<PIN>`` query component to authenticate automatically.
192 ///
193 /// ``module`` optionally specifies the path to the PKCS#11 shared library.
194 /// When omitted, the library is resolved from the ``PKCS11_MODULE_PATH``
195 /// environment variable or the system default (``p11-kit-proxy.so``).
196 ///
197 /// Example:
198 /// >>> token = synta.pkcs11.Pkcs11Token(
199 /// ... "pkcs11:token=TestToken?pin-value=1234",
200 /// ... module="/usr/lib64/pkcs11/libkryoptic_pkcs11.so",
201 /// ... )
202 #[new]
203 #[pyo3(signature = (uri, module = None))]
204 fn new(uri: &str, module: Option<&str>) -> PyResult<Self> {
205 let parsed = Pkcs11Uri::parse(uri).ok_or_else(|| {
206 PyValueError::new_err(format!("invalid PKCS#11 URI: {:?}", redact_pin(uri)))
207 })?;
208 let mgr = make_mgr(&parsed, module)?;
209 let attrs = parsed.attrs;
210 Ok(Self {
211 mgr,
212 uri: uri.to_owned(),
213 attrs,
214 })
215 }
216
217 /// Return ``True`` if a private key matching ``object_uri``'s label exists
218 /// on this token.
219 ///
220 /// ``object_uri`` must be a ``pkcs11:`` URI with an ``object=<label>``
221 /// component. The token name and PIN are inherited from the URI used to
222 /// construct this :class:`Pkcs11Token` when not present in ``object_uri``.
223 ///
224 /// Example:
225 /// >>> token.find_key("pkcs11:object=caKey")
226 /// True
227 fn find_key(&self, object_uri: &str) -> PyResult<bool> {
228 let mut uri = Pkcs11Uri::parse(object_uri).ok_or_else(|| {
229 PyValueError::new_err(format!("invalid PKCS#11 URI: {:?}", redact_pin(object_uri)))
230 })?;
231 // Inherit token and PIN from the construction URI when absent in object_uri.
232 if uri.attrs.token.is_none() {
233 uri.attrs.token = self.attrs.token.clone();
234 }
235 if !uri.attrs.has_pin() {
236 uri.attrs
237 .set_pin_value(self.attrs.pin_value().map(|s| s.to_owned()));
238 }
239 self.mgr
240 .find_key(&uri)
241 .map_err(|e| PyValueError::new_err(format!("{e}")))
242 }
243
244 /// List all private keys on this token.
245 ///
246 /// Returns a list of :class:`KeyInfo` objects.
247 ///
248 /// Example:
249 /// >>> keys = token.list_keys()
250 /// >>> for k in keys:
251 /// ... print(k.label, k.key_type, k.key_bits)
252 fn list_keys(&self) -> PyResult<Vec<PyKeyInfo>> {
253 let token_name = self
254 .attrs
255 .token
256 .as_deref()
257 .ok_or_else(|| PyValueError::new_err("Pkcs11Token URI must contain 'token='"))?;
258 let pin = self.attrs.pin_value();
259 let keys = self
260 .mgr
261 .list_keys(token_name, pin)
262 .map_err(|e| PyValueError::new_err(format!("{e}")))?;
263 Ok(keys
264 .into_iter()
265 .map(|k| PyKeyInfo {
266 label: k.label,
267 id: k.id,
268 key_type: k.key_type,
269 key_bits: k.key_bits,
270 })
271 .collect())
272 }
273
274 /// Destroy the private key (and matching public key) identified by
275 /// ``object_uri`` from this token.
276 ///
277 /// The token name and PIN are inherited from the URI used to construct this
278 /// :class:`Pkcs11Token` when not present in ``object_uri``.
279 ///
280 /// Example:
281 /// >>> token.delete_key("pkcs11:object=caKey")
282 fn delete_key(&self, object_uri: &str) -> PyResult<()> {
283 let mut uri = Pkcs11Uri::parse(object_uri).ok_or_else(|| {
284 PyValueError::new_err(format!("invalid PKCS#11 URI: {:?}", redact_pin(object_uri)))
285 })?;
286 // Inherit token and PIN from the construction URI when absent in object_uri.
287 if uri.attrs.token.is_none() {
288 uri.attrs.token = self.attrs.token.clone();
289 }
290 if !uri.attrs.has_pin() {
291 uri.attrs
292 .set_pin_value(self.attrs.pin_value().map(|s| s.to_owned()));
293 }
294 self.mgr
295 .delete_key(&uri)
296 .map_err(|e| PyValueError::new_err(format!("{e}")))
297 }
298
299 /// Generate a key pair on this token and return the private key.
300 ///
301 /// ``key_type`` is ``"rsa"``, ``"ec"``, ``"ed25519"``, ``"ed448"``,
302 /// ``"ml-dsa"``, or ``"ml-kem"``.
303 /// ``param`` is the RSA bit-length (``int``) for RSA keys, the curve name
304 /// (``str``) for EC keys (``"P-256"``, ``"P-384"``, ``"P-521"``), or the
305 /// parameter-set name for PQC keys (``"ML-DSA-44"``, ``"ML-DSA-65"``,
306 /// ``"ML-DSA-87"``; ``"ML-KEM-512"``, ``"ML-KEM-768"``, ``"ML-KEM-1024"``).
307 /// ``label`` is the ``CKA_LABEL`` to assign to the generated key objects.
308 /// ``extractable`` controls ``CKA_EXTRACTABLE`` on the private key; default
309 /// is ``False`` (hardware-resident key that cannot be exported). Set to
310 /// ``True`` only when key export is explicitly required.
311 ///
312 /// The token and PIN are taken from the URI used to construct this
313 /// :class:`Pkcs11Token`.
314 ///
315 /// Example:
316 /// >>> key = token.generate_key_pair("rsa", 4096, "caKey")
317 /// >>> key = token.generate_key_pair("ec", "P-256", "ecKey")
318 /// >>> key = token.generate_key_pair("ml-dsa", "ML-DSA-65", "dsaKey")
319 /// >>> key = token.generate_key_pair("ml-kem", "ML-KEM-768", "kemKey")
320 /// >>> key = token.generate_key_pair("rsa", 2048, "backupKey", extractable=True)
321 #[pyo3(signature = (key_type, param, label, extractable = false))]
322 fn generate_key_pair(
323 &self,
324 key_type: &str,
325 param: Bound<'_, PyAny>,
326 label: &str,
327 extractable: bool,
328 ) -> PyResult<PyPrivateKey> {
329 let spec = build_key_spec(key_type, ¶m)?;
330
331 // Merge the label into the base URI (preserving token= and pin-value=).
332 let full_uri_str = merge_object_label(&self.uri, label);
333 let full_uri = Pkcs11Uri::parse(&full_uri_str).ok_or_else(|| {
334 PyValueError::new_err(format!("failed to build URI with label '{label}'"))
335 })?;
336
337 let inner = self
338 .mgr
339 .generate_key_pair_in_token(&spec, &full_uri, extractable)
340 .map_err(|e| PyValueError::new_err(format!("{e}")))?;
341
342 Ok(PyPrivateKey { inner })
343 }
344
345 fn __repr__(&self) -> String {
346 format!("Pkcs11Token(uri={:?})", redact_pin(&self.uri))
347 }
348}
349
350// ── Module-level list_slots ───────────────────────────────────────────────────
351
352/// List all PKCS#11 token slots using the system module.
353///
354/// ``module`` optionally specifies the PKCS#11 shared library path. When
355/// omitted, the library is resolved from ``PKCS11_MODULE_PATH`` or the system
356/// default (``p11-kit-proxy.so``).
357///
358/// Example:
359/// >>> import synta.pkcs11
360/// >>> slots = synta.pkcs11.list_slots()
361/// >>> for s in slots:
362/// ... print(s.slot_id, s.token_label, s.manufacturer_id)
363#[pyfunction]
364#[pyo3(signature = (module = None))]
365pub fn list_slots(module: Option<&str>) -> PyResult<Vec<PySlotInfo>> {
366 let mgr = if let Some(path) = module {
367 Pkcs11Manager::new(path).map_err(|e| PyValueError::new_err(format!("{e}")))?
368 } else {
369 Pkcs11Manager::from_env().map_err(|e| PyValueError::new_err(format!("{e}")))?
370 };
371 let slots = mgr
372 .list_slots()
373 .map_err(|e| PyValueError::new_err(format!("{e}")))?;
374 Ok(slots
375 .into_iter()
376 .map(|s| PySlotInfo {
377 slot_id: s.slot_id,
378 token_label: s.token_label,
379 manufacturer_id: s.manufacturer_id,
380 model: s.model,
381 serial_number: s.serial_number,
382 flags: s.flags,
383 })
384 .collect())
385}
386
387// ── Helpers ───────────────────────────────────────────────────────────────────
388
389fn build_key_spec(key_type: &str, param: &Bound<'_, PyAny>) -> PyResult<KeySpec> {
390 match key_type.to_ascii_lowercase().as_str() {
391 "rsa" => {
392 let bits: u32 = param.extract().map_err(|e| {
393 PyValueError::new_err(format!("RSA key requires an integer bit-length as param: {e}"))
394 })?;
395 Ok(KeySpec::Rsa(bits))
396 }
397 "ec" => {
398 let curve: String = param.extract().map_err(|e| {
399 PyValueError::new_err(format!(
400 "EC key requires a curve name string as param (e.g. 'P-256'): {e}"
401 ))
402 })?;
403 Ok(KeySpec::Ec(curve))
404 }
405 "ed25519" => Ok(KeySpec::Ed25519),
406 "ed448" => Ok(KeySpec::Ed448),
407 "ml-dsa" => {
408 let ps: String = param.extract().map_err(|e| {
409 PyValueError::new_err(format!(
410 "ML-DSA key requires a parameter-set string (e.g. 'ML-DSA-65'): {e}"
411 ))
412 })?;
413 Ok(KeySpec::MlDsa(ps))
414 }
415 "ml-kem" => {
416 let ps: String = param.extract().map_err(|e| {
417 PyValueError::new_err(format!(
418 "ML-KEM key requires a parameter-set string (e.g. 'ML-KEM-768'): {e}"
419 ))
420 })?;
421 Ok(KeySpec::MlKem(ps))
422 }
423 other => Err(PyValueError::new_err(format!(
424 "unsupported key type {other:?}; expected 'rsa', 'ec', 'ed25519', 'ed448', 'ml-dsa', or 'ml-kem'"
425 ))),
426 }
427}
428
429// ── Module registration ───────────────────────────────────────────────────────
430
431pub fn register_pkcs11_module(parent: &Bound<'_, PyModule>) -> PyResult<()> {
432 let py = parent.py();
433 let m = PyModule::new(py, "pkcs11")?;
434
435 m.add_class::<PySlotInfo>()?;
436 m.add_class::<PyKeyInfo>()?;
437 m.add_class::<PyPkcs11Token>()?;
438 m.add_function(wrap_pyfunction!(list_slots, &m)?)?;
439
440 crate::install_submodule(
441 parent,
442 &m,
443 "synta.pkcs11",
444 Some("PKCS#11 token management: slot listing, key find/list/delete/generate."),
445 )
446}