Merge pull request #53 from dnet/main

Added options for armor-less output and file-based I/O

Author: wiktor-k

README.md

@@ -88,6 +88,41 @@ print(f"Clear signed: {clear}")
 assert "PGP SIGNED MESSAGE" in str(clear)
 ```
 
+### sign_file
+
+Signs data from a file and writes the signed output to another file:
+
+```python
+from pysequoia import sign_file, SignatureMode
+import tempfile, os
+
+s = Cert.from_file("signing-key.asc")
+
+# create a file with data to sign
+with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as inp:
+  inp.write("data to be signed".encode("utf8"))
+  input_path = inp.name
+
+with tempfile.NamedTemporaryFile(delete=False, suffix=".pgp") as out:
+  output_path = out.name
+
+sign_file(s.secrets.signer(), input_path, output_path)
+signed = open(output_path, "rb").read()
+assert b"PGP MESSAGE" in signed
+
+# detached signature to file
+with tempfile.NamedTemporaryFile(delete=False, suffix=".sig") as out:
+  detached_path = out.name
+
+sign_file(s.secrets.signer(), input_path, detached_path, mode=SignatureMode.DETACHED)
+detached = open(detached_path, "rb").read()
+assert b"PGP SIGNATURE" in detached
+
+os.unlink(input_path)
+os.unlink(output_path)
+os.unlink(detached_path)
+```
+
 ### verify
 
 Verifies signed data and returns verified data:
@@ -167,6 +202,33 @@ print(f"Encrypted data: {encrypted}")
 
 The `signer` argument is optional and when omitted the function will return an unsigned (but encrypted) message.
 
+### encrypt_file
+
+Encrypts data from a file and writes the encrypted output to another file:
+
+```python
+from pysequoia import encrypt_file
+import tempfile, os
+
+s = Cert.from_file("passwd.pgp")
+r = Cert.from_bytes(open("wiktor.asc", "rb").read())
+
+# create a file with content to encrypt
+with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as inp:
+  inp.write("content to encrypt".encode("utf8"))
+  input_path = inp.name
+
+with tempfile.NamedTemporaryFile(delete=False, suffix=".pgp") as out:
+  output_path = out.name
+
+encrypt_file(signer = s.secrets.signer("hunter22"), recipients = [r], input = input_path, output = output_path)
+encrypted = open(output_path, "rb").read()
+assert b"PGP MESSAGE" in encrypted
+
+os.unlink(input_path)
+os.unlink(output_path)
+```
+
 ### decrypt
 
 Decrypts plain data:
@@ -216,6 +278,81 @@ assert decrypted.valid_sigs[0].signing_key == sender.fingerprint
 
 Here, the same remarks as to [`verify`](#verify) also apply.
 
+### decrypt_file
+
+Decrypts data from a file and writes the decrypted output to another file:
+
+```python
+from pysequoia import decrypt_file
+import tempfile, os
+
+sender = Cert.from_file("no-passwd.pgp")
+receiver = Cert.from_file("passwd.pgp")
+
+content = "Red Green Blue"
+
+encrypted = encrypt(recipients = [receiver], bytes = content.encode("utf8"))
+
+# write encrypted data to a file
+with tempfile.NamedTemporaryFile(delete=False, suffix=".pgp") as inp:
+  inp.write(encrypted)
+  input_path = inp.name
+
+with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as out:
+  output_path = out.name
+
+decrypted = decrypt_file(decryptor = receiver.secrets.decryptor("hunter22"), input = input_path, output = output_path)
+
+# content is written to the output file, not returned in memory
+assert decrypted.bytes is None
+
+# read decrypted content from the output file
+assert open(output_path, "rb").read().decode("utf8") == content
+
+# this message did not contain any valid signatures
+assert len(decrypted.valid_sigs) == 0
+
+os.unlink(input_path)
+os.unlink(output_path)
+```
+
+Decrypt file can also verify signatures while decrypting:
+
+```python
+from pysequoia import decrypt_file
+import tempfile, os
+
+sender = Cert.from_file("no-passwd.pgp")
+receiver = Cert.from_file("passwd.pgp")
+
+content = "Red Green Blue"
+
+encrypted = encrypt(signer = sender.secrets.signer(), recipients = [receiver], bytes = content.encode("utf8"))
+
+# write encrypted data to a file
+with tempfile.NamedTemporaryFile(delete=False, suffix=".pgp") as inp:
+  inp.write(encrypted)
+  input_path = inp.name
+
+with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as out:
+  output_path = out.name
+
+def get_certs(key_ids):
+  print(f"For verification after decryption, we need these keys: {key_ids}")
+  return [sender]
+
+decrypted = decrypt_file(decryptor = receiver.secrets.decryptor("hunter22"), input = input_path, output = output_path, store = get_certs)
+
+assert open(output_path, "rb").read().decode("utf8") == content
+
+# let's check the valid signature's certificate and signing subkey fingerprints
+assert decrypted.valid_sigs[0].certificate == sender.fingerprint
+assert decrypted.valid_sigs[0].signing_key == sender.fingerprint
+
+os.unlink(input_path)
+os.unlink(output_path)
+```
+
 ## Certificates
 
 The `Cert` class represents one OpenPGP certificate (commonly called a

src/decrypt.rs

@@ -1,5 +1,7 @@
+use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 
+use anyhow::Context;
 use pyo3::prelude::*;
 use sequoia_openpgp::crypto::{Decryptor, SessionKey};
 use sequoia_openpgp::packet::{PKESK, SKESK};
@@ -63,6 +65,32 @@ pub fn decrypt(
     })
 }
 
+#[pyfunction]
+#[pyo3(signature = (decryptor, input, output, store=None))]
+pub fn decrypt_file(
+    mut decryptor: PyDecryptor,
+    input: PathBuf,
+    output: PathBuf,
+    store: Option<Py<PyAny>>,
+) -> PyResult<Decrypted> {
+    if let Some(store) = store {
+        decryptor.set_verifier(PyVerifier::from_callback(store));
+    }
+    let policy = &P::new();
+
+    let mut decryptor = DecryptorBuilder::from_file(&input)
+        .context("Failed to open input file")?
+        .with_policy(policy, None, decryptor)?;
+
+    let mut sink = std::fs::File::create(&output).context("Failed to create output file")?;
+    std::io::copy(&mut decryptor, &mut sink)?;
+    let decryptor = decryptor.into_helper();
+    Ok(Decrypted {
+        content: None,
+        valid_sigs: decryptor.valid_sigs(),
+    })
+}
+
 impl VerificationHelper for PyDecryptor {
     fn get_certs(&mut self, ids: &[KeyHandle]) -> sequoia_openpgp::Result<Vec<cert::Cert>> {
         if let Some(verifier) = &mut self.verifier {

src/encrypt.rs

@@ -1,5 +1,6 @@
 use std::borrow::Cow;
 use std::io::Write;
+use std::path::PathBuf;
 
 use anyhow::Context;
 use pyo3::prelude::*;
@@ -13,13 +14,15 @@ use sequoia_openpgp::types::KeyFlags;
 use crate::cert::Cert;
 use crate::signer::PySigner;
 
-#[pyfunction]
-#[pyo3(signature = (recipients, bytes, signer=None))]
-pub fn encrypt(
-    recipients: Vec<PyRef<Cert>>,
-    bytes: &[u8],
-    signer: Option<PySigner>,
-) -> PyResult<Cow<'static, [u8]>> {
+type RecipientKey = (
+    Option<sequoia_openpgp::types::Features>,
+    sequoia_openpgp::packet::Key<
+        sequoia_openpgp::packet::key::PublicParts,
+        sequoia_openpgp::packet::key::UnspecifiedRole,
+    >,
+);
+
+fn resolve_recipient_keys(recipients: &[PyRef<Cert>]) -> PyResult<Vec<RecipientKey>> {
     let mode = KeyFlags::empty()
         .set_storage_encryption()
         .set_transport_encryption();
@@ -62,12 +65,28 @@ pub fn encrypt(
             );
         }
     }
+    Ok(recipient_keys)
+}
+
+#[pyfunction]
+#[pyo3(signature = (recipients, bytes, signer=None, *, armor=true))]
+pub fn encrypt(
+    recipients: Vec<PyRef<Cert>>,
+    bytes: &[u8],
+    signer: Option<PySigner>,
+    armor: bool,
+) -> PyResult<Cow<'static, [u8]>> {
+    let recipient_keys = resolve_recipient_keys(&recipients)?;
 
     let mut sink = vec![];
 
     let message = Message::new(&mut sink);
 
-    let message = Armorer::new(message).build()?;
+    let message = if armor {
+        Armorer::new(message).build()?
+    } else {
+        message
+    };
 
     let mut message = Encryptor::for_recipients(
         message,
@@ -91,3 +110,48 @@ pub fn encrypt(
 
     Ok(sink.into())
 }
+
+#[pyfunction]
+#[pyo3(signature = (recipients, input, output, signer=None, *, armor=true))]
+pub fn encrypt_file(
+    recipients: Vec<PyRef<Cert>>,
+    input: PathBuf,
+    output: PathBuf,
+    signer: Option<PySigner>,
+    armor: bool,
+) -> PyResult<()> {
+    let recipient_keys = resolve_recipient_keys(&recipients)?;
+
+    let mut sink = std::fs::File::create(&output).context("Failed to create output file")?;
+
+    let message = Message::new(&mut sink);
+
+    let message = if armor {
+        Armorer::new(message).build()?
+    } else {
+        message
+    };
+
+    let mut message = Encryptor::for_recipients(
+        message,
+        recipient_keys
+            .iter()
+            .map(|(features, key)| Recipient::new(features.clone(), key.key_handle(), key)),
+    )
+    .build()
+    .context("Failed to create encryptor")?;
+
+    if let Some(signer) = signer {
+        message = Signer::new(message, signer)?.build()?;
+    }
+    let mut message = LiteralWriter::new(message)
+        .build()
+        .context("Failed to create literal writer")?;
+
+    let mut input_file = std::fs::File::open(&input).context("Failed to open input file")?;
+    std::io::copy(&mut input_file, &mut message)?;
+
+    message.finalize()?;
+
+    Ok(())
+}

src/lib.rs

@@ -98,8 +98,11 @@ fn pysequoia(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<notation::Notation>()?;
     m.add_class::<sign::SignatureMode>()?;
     m.add_function(wrap_pyfunction!(sign::sign, m)?)?;
+    m.add_function(wrap_pyfunction!(sign::sign_file, m)?)?;
     m.add_function(wrap_pyfunction!(encrypt::encrypt, m)?)?;
+    m.add_function(wrap_pyfunction!(encrypt::encrypt_file, m)?)?;
     m.add_function(wrap_pyfunction!(decrypt::decrypt, m)?)?;
+    m.add_function(wrap_pyfunction!(decrypt::decrypt_file, m)?)?;
     m.add_function(wrap_pyfunction!(verify::verify, m)?)?;
     Ok(())
 }

src/sign.rs

@@ -1,6 +1,8 @@
 use std::borrow::Cow;
 use std::io::Write;
+use std::path::PathBuf;
 
+use anyhow::Context;
 use pyo3::prelude::*;
 use sequoia_openpgp::armor;
 use sequoia_openpgp::serialize::stream::Armorer;
@@ -20,16 +22,21 @@ pub enum SignatureMode {
 }
 
 #[pyfunction]
-#[pyo3(signature = (signer, bytes, *, mode=&SignatureMode::Inline))]
-pub fn sign(signer: PySigner, bytes: &[u8], mode: &SignatureMode) -> PyResult<Cow<'static, [u8]>> {
+#[pyo3(signature = (signer, bytes, *, mode=&SignatureMode::Inline, armor=true))]
+pub fn sign(
+    signer: PySigner,
+    bytes: &[u8],
+    mode: &SignatureMode,
+    armor: bool,
+) -> PyResult<Cow<'static, [u8]>> {
     use sequoia_openpgp::serialize::stream::Signer;
 
     let mut sink = vec![];
     {
         let message = Message::new(&mut sink);
-        let message = if mode == &SignatureMode::Inline {
+        let message = if mode == &SignatureMode::Inline && armor {
             Armorer::new(message).kind(armor::Kind::Message).build()?
-        } else if mode == &SignatureMode::Detached {
+        } else if mode == &SignatureMode::Detached && armor {
             Armorer::new(message).kind(armor::Kind::Signature).build()?
         } else {
             message
@@ -48,3 +55,40 @@ pub fn sign(signer: PySigner, bytes: &[u8], mode: &SignatureMode) -> PyResult<Co
 
     Ok(sink.into())
 }
+
+#[pyfunction]
+#[pyo3(signature = (signer, input, output, *, mode=&SignatureMode::Inline, armor=true))]
+pub fn sign_file(
+    signer: PySigner,
+    input: PathBuf,
+    output: PathBuf,
+    mode: &SignatureMode,
+    armor: bool,
+) -> PyResult<()> {
+    use sequoia_openpgp::serialize::stream::Signer;
+
+    let mut sink = std::fs::File::create(&output).context("Failed to create output file")?;
+    {
+        let message = Message::new(&mut sink);
+        let message = if mode == &SignatureMode::Inline && armor {
+            Armorer::new(message).kind(armor::Kind::Message).build()?
+        } else if mode == &SignatureMode::Detached && armor {
+            Armorer::new(message).kind(armor::Kind::Signature).build()?
+        } else {
+            message
+        };
+        let message = Signer::new(message, signer)?;
+        let mut message = if mode == &SignatureMode::Inline {
+            LiteralWriter::new(message.build()?).build()?
+        } else if mode == &SignatureMode::Detached {
+            message.detached().build()?
+        } else {
+            message.cleartext().build()?
+        };
+        let mut input_file = std::fs::File::open(&input).context("Failed to open input file")?;
+        std::io::copy(&mut input_file, &mut message)?;
+        message.finalize()?;
+    }
+
+    Ok(())
+}