Merge pull request #7 from tiwilliam/codex/find-certificate-ingestion-method-pa06ug

Support bytes-like certificate material

Author: tiwilliam

.gitignore

@@ -1,5 +1,6 @@
 /site
 /target
 /.shadowenv.d
+.venv/
 *.so
 __pycache__/

CHANGELOG.md

@@ -2,7 +2,12 @@
 
 Released XXXX-XX-XX
 
+## 0.7.0
+
+Released 2025-10-19
+
 * [#3](https://github.com/tiwilliam/rsmime/pull/3) - Replace black with ruff for formatting.
+* [#7](https://github.com/tiwilliam/rsmime/pull/7) - Introduce ``cert_data`` and ``key_data`` options for supplying certificate material without temporary files.
 
 ## 0.6.4
 

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "rsmime"
-version = "0.6.4"
+version = "0.7.0"
 edition = "2021"
 
 [lib]

Cargo.toml

@@ -23,6 +23,29 @@ except (SignError, CertificateError) as e:
     print("Failed to sign:", e)
 ```
 
+If you already have the PEM data in memory you can also provide it directly as text or
+bytes-like objects:
+
+```python
+from pathlib import Path
+from rsmime import Rsmime
+
+certificate = Path("some.crt").read_text()
+private_key = Path("some.key").read_text()
+
+client = Rsmime(cert_data=certificate, key_data=private_key)
+
+certificate_bytes = Path("some.crt").read_bytes()
+private_key_bytes = Path("some.key").read_bytes()
+
+client = Rsmime(cert_data=certificate_bytes, key_data=private_key_bytes)
+
+client = Rsmime(
+    cert_data=bytearray(Path("some.crt").read_bytes()),
+    key_data=bytearray(Path("some.key").read_bytes()),
+)
+```
+
 ### Output
 
 ```bash

docs/getting_started.md

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "rsmime"
-version = "0.6.4"
+version = "0.7.0"
 description = "Python package for signing and verifying S/MIME messages"
 classifiers = [
     "License :: OSI Approved :: MIT License",

pyproject.toml

@@ -1,13 +1,27 @@
+from os import PathLike
+
 class Rsmime:
-    def __init__(self, cert_file: str, key_file: str) -> None:
-        """Initialize client and load certificate from disk.
+    def __init__(
+        self,
+        cert_file: str | PathLike[str] | None = ...,
+        key_file: str | PathLike[str] | None = ...,
+        *,
+        cert_data: str | bytes | bytearray | memoryview | None = ...,
+        key_data: str | bytes | bytearray | memoryview | None = ...,
+    ) -> None:
+        """Initialize client and load certificate material.
 
         Parameters:
-            cert_file: Path to certificate on disk.
-            key_file: Path to private key on disk.
+            cert_file: Path to certificate on disk. Mutually exclusive with ``cert_data``.
+            key_file: Path to private key on disk. Mutually exclusive with ``key_data``.
+            cert_data: PEM-encoded certificate contents provided as a string or bytes-like
+                object.
+            key_data: PEM-encoded private key contents provided as a string or bytes-like
+                object.
 
         Raises:
-            exceptions.CertificateError: If there is an error reading the certificate or key.
+            exceptions.CertificateError: If there is an error loading, parsing, or when
+                both a file path and in-memory value are provided for the same artifact.
         """
         ...
     def sign(self, message: bytes, *, detached: bool = False) -> bytes:

python/rsmime/__init__.pyi

@@ -1,6 +1,7 @@
 extern crate openssl;
 
 use std::io::{Error, ErrorKind};
+use std::path::PathBuf;
 
 use openssl::nid::Nid;
 use openssl::pkcs7::{Pkcs7, Pkcs7Flags};
@@ -95,6 +96,39 @@ create_exception!(exceptions, CertificateExpiredError, CertificateError);
 create_exception!(exceptions, SignError, RsmimeError);
 create_exception!(exceptions, VerifyError, RsmimeError);
 
+fn material_from_sources(
+    py: Python<'_>,
+    file: Option<PathBuf>,
+    data: Option<Py<PyAny>>,
+    file_label: &str,
+    data_label: &str,
+) -> PyResult<Vec<u8>> {
+    match (file, data) {
+        (Some(path), None) => {
+            std::fs::read(path).map_err(|err| CertificateError::new_err(err.to_string()))
+        }
+        (None, Some(obj)) => {
+            let obj = obj.as_ref(py);
+
+            if let Ok(value) = obj.extract::<&str>() {
+                Ok(value.as_bytes().to_vec())
+            } else if let Ok(value) = obj.extract::<Vec<u8>>() {
+                Ok(value)
+            } else {
+                Err(CertificateError::new_err(format!(
+                    "{data_label} must be a str or bytes-like object"
+                )))
+            }
+        }
+        (Some(_), Some(_)) => Err(CertificateError::new_err(format!(
+            "Provide either {file_label} or {data_label}, not both"
+        ))),
+        (None, None) => Err(CertificateError::new_err(format!(
+            "A value must be provided via {file_label} or {data_label}"
+        ))),
+    }
+}
+
 #[pymodule]
 fn exceptions(py: Python, m: &PyModule) -> PyResult<()> {
     m.add("RsmimeError", py.get_type::<RsmimeError>())?;
@@ -132,20 +166,23 @@ struct Rsmime {
 #[pymethods]
 impl Rsmime {
     #[new]
-    #[pyo3(signature = (cert_file, key_file))]
-    fn new(cert_file: String, key_file: String) -> PyResult<Self> {
+    #[pyo3(signature = (cert_file=None, key_file=None, *, cert_data=None, key_data=None))]
+    fn new(
+        py: Python<'_>,
+        cert_file: Option<PathBuf>,
+        key_file: Option<PathBuf>,
+        cert_data: Option<Py<PyAny>>,
+        key_data: Option<Py<PyAny>>,
+    ) -> PyResult<Self> {
         let stack = Stack::new().unwrap();
 
-        let cert_data =
-            std::fs::read(cert_file).map_err(|err| CertificateError::new_err(err.to_string()))?;
+        let cert_bytes = material_from_sources(py, cert_file, cert_data, "cert_file", "cert_data")?;
+        let key_bytes = material_from_sources(py, key_file, key_data, "key_file", "key_data")?;
 
-        let cert =
-            X509::from_pem(&cert_data).map_err(|err| CertificateError::new_err(err.to_string()))?;
-
-        let key_data =
-            std::fs::read(key_file).map_err(|err| CertificateError::new_err(err.to_string()))?;
+        let cert = X509::from_pem(&cert_bytes)
+            .map_err(|err| CertificateError::new_err(err.to_string()))?;
 
-        let rsa = Rsa::private_key_from_pem(&key_data)
+        let rsa = Rsa::private_key_from_pem(&key_bytes)
             .map_err(|err| CertificateError::new_err(err.to_string()))?;
         let pkey = PKey::from_rsa(rsa).map_err(|err| CertificateError::new_err(err.to_string()))?;
 

src/lib.rs

@@ -1,45 +1,102 @@
+from pathlib import Path
+
 import pytest
 from callee import strings
+
 from rsmime import Rsmime, exceptions
 
+
+ATTACHED_SIGNATURE_REGEX = strings.Regex(
+    b'MIME-Version: 1.0\n'
+    b'Content-Disposition: attachment; filename="smime.p7m"\n'
+    b'Content-Type: application/x-pkcs7-mime; smime-type=signed-data; name="smime.p7m"\n'
+    b'Content-Transfer-Encoding: base64\n'
+    b'\n'
+    b'MIIJwQYJKoZIhvcNAQcCoIIJsjCCCa4CAQExDzANBglghkgBZQMEAgEFADASBgkq\n'
+    b'[A-Za-z0-9/+=\n]+\n'
+    b'\n'
+)
+
+DETACHED_SIGNATURE_REGEX = strings.Regex(
+    b'MIME-Version: 1.0\n'
+    b'Content-Type: multipart/signed; protocol="application/x-pkcs7-signature"; micalg="sha-256"; boundary="----[A-Z0-9]+"\n\n'
+    b'This is an S/MIME signed message\n\n'
+    b'------[A-Z0-9]+\n'
+    b'abc\n'
+    b'------[A-Z0-9]+\n'
+    b'Content-Type: application/x-pkcs7-signature; name="smime.p7s"\n'
+    b'Content-Transfer-Encoding: base64\n'
+    b'Content-Disposition: attachment; filename="smime.p7s"\n'
+    b'\n'
+    b'MIIJugYJKoZIhvcNAQcCoIIJqzCCCacCAQExDzANBglghkgBZQMEAgEFADALBgkq\n'
+    b'[A-Za-z0-9/+=\n]+\n'
+    b'\n'
+    b'------[A-Z0-9]+--\n'
+    b'\n'
+)
+
+
+def _load_text(path: str) -> str:
+    return Path(path).read_text()
+
+
+def _load_bytes(path: str) -> bytes:
+    return Path(path).read_bytes()
+
+
 working_client = Rsmime('tests/data/certificate.crt', 'tests/data/certificate.key')
 expired_client = Rsmime('tests/data/expired.crt', 'tests/data/certificate.key')
 
 
 class TestRsmime:
     def test_sign(self):
         signed_data = working_client.sign(b'abc')
-        assert signed_data == strings.Regex(
-            b'MIME-Version: 1.0\n'
-            b'Content-Disposition: attachment; filename="smime.p7m"\n'
-            b'Content-Type: application/x-pkcs7-mime; smime-type=signed-data; name="smime.p7m"\n'
-            b'Content-Transfer-Encoding: base64\n'
-            b'\n'
-            b'MIIJwQYJKoZIhvcNAQcCoIIJsjCCCa4CAQExDzANBglghkgBZQMEAgEFADASBgkq\n'
-            b'[A-Za-z0-9/+=\n]+\n'
-            b'\n'
-        )
+        assert signed_data == ATTACHED_SIGNATURE_REGEX
 
     def test_sign_detached(self):
         signed_data = working_client.sign(b'abc', detached=True)
-        assert signed_data == strings.Regex(
-            b'MIME-Version: 1.0\n'
-            b'Content-Type: multipart/signed; protocol="application/x-pkcs7-signature"; micalg="sha-256"; boundary="----[A-Z0-9]+"\n\n'
-            b'This is an S/MIME signed message\n\n'
-            b'------[A-Z0-9]+\n'
-            b'abc\n'
-            b'------[A-Z0-9]+\n'
-            b'Content-Type: application/x-pkcs7-signature; name="smime.p7s"\n'
-            b'Content-Transfer-Encoding: base64\n'
-            b'Content-Disposition: attachment; filename="smime.p7s"\n'
-            b'\n'
-            b'MIIJugYJKoZIhvcNAQcCoIIJqzCCCacCAQExDzANBglghkgBZQMEAgEFADALBgkq\n'
-            b'[A-Za-z0-9/+=\n]+\n'
-            b'\n'
-            b'------[A-Z0-9]+--\n'
-            b'\n'
+        assert signed_data == DETACHED_SIGNATURE_REGEX
+
+    def test_sign_with_in_memory_material(self):
+        client = Rsmime(
+            cert_data=_load_text('tests/data/certificate.crt'),
+            key_data=_load_text('tests/data/certificate.key'),
+        )
+
+        signed_data = client.sign(b'abc')
+
+        assert signed_data == ATTACHED_SIGNATURE_REGEX
+
+    def test_sign_with_in_memory_bytes_material(self):
+        client = Rsmime(
+            cert_data=_load_text('tests/data/certificate.crt').encode(),
+            key_data=_load_text('tests/data/certificate.key').encode(),
         )
 
+        signed_data = client.sign(b'abc')
+
+        assert signed_data == ATTACHED_SIGNATURE_REGEX
+
+    def test_sign_with_in_memory_bytearray_material(self):
+        client = Rsmime(
+            cert_data=bytearray(_load_bytes('tests/data/certificate.crt')),
+            key_data=bytearray(_load_bytes('tests/data/certificate.key')),
+        )
+
+        signed_data = client.sign(b'abc')
+
+        assert signed_data == ATTACHED_SIGNATURE_REGEX
+
+    def test_sign_with_path_objects(self):
+        client = Rsmime(
+            Path('tests/data/certificate.crt'),
+            Path('tests/data/certificate.key'),
+        )
+
+        signed_data = client.sign(b'abc')
+
+        assert signed_data == ATTACHED_SIGNATURE_REGEX
+
     def test_sign_missing_cert(self):
         with pytest.raises(
             exceptions.CertificateError, match='No such file or directory'
@@ -64,6 +121,52 @@ def test_sign_empty_data(self):
         with pytest.raises(exceptions.SignError, match='Cannot sign empty data'):
             working_client.sign(b'')
 
+    def test_conflicting_certificate_inputs(self):
+        with pytest.raises(
+            exceptions.CertificateError,
+            match='Provide either cert_file or cert_data',
+        ):
+            Rsmime(
+                'tests/data/certificate.crt',
+                'tests/data/certificate.key',
+                cert_data=_load_text('tests/data/certificate.crt'),
+            )
+
+    def test_conflicting_key_inputs(self):
+        with pytest.raises(
+            exceptions.CertificateError,
+            match='Provide either key_file or key_data',
+        ):
+            Rsmime(
+                'tests/data/certificate.crt',
+                'tests/data/certificate.key',
+                key_data=_load_text('tests/data/certificate.key'),
+            )
+
+    def test_missing_certificate_input(self):
+        with pytest.raises(
+            exceptions.CertificateError,
+            match='cert_file or cert_data',
+        ):
+            Rsmime(key_data=_load_text('tests/data/certificate.key'))
+
+    def test_missing_key_input(self):
+        with pytest.raises(
+            exceptions.CertificateError,
+            match='key_file or key_data',
+        ):
+            Rsmime(cert_data=_load_text('tests/data/certificate.crt'))
+
+    def test_in_memory_material_requires_text_or_bytes(self):
+        with pytest.raises(
+            exceptions.CertificateError,
+            match='cert_data must be a str or bytes-like object',
+        ):
+            Rsmime(
+                cert_data=object(),
+                key_data=_load_text('tests/data/certificate.key'),
+            )
+
     def test_verify(self):
         data = b'abc'
         signed_data = working_client.sign(data)