martinzink commented on code in PR #2225:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2225#discussion_r3925405521
##########
minifi_rust/extensions/minifi_pgp/src/processors/decrypt_content.rs:
##########
@@ -0,0 +1,356 @@
+mod processor_definition;
+
+use processor_definition::*;
+
+use crate::controller_services::private_key_service::PGPPrivateKeyService;
+
+use minifi_native::macros::{ComponentIdentifier, PropertyType};
+use minifi_native::{
+ FlowFileStreamTransform, GetControllerService, GetProperty, InputStream,
Logger, MinifiError,
+ OutputStream, ProcessError, RouteErrorExt, Schedule, TransformStreamResult,
+};
+use pgp::composed::{Message, TheRing};
+use std::fmt::Debug;
+use strum_macros::{Display, EnumString, IntoStaticStr, VariantNames};
+
+#[derive(
+ Debug, Clone, Copy, PartialEq, Display, EnumString, VariantNames,
IntoStaticStr, PropertyType,
+)]
+#[strum(serialize_all = "UPPERCASE", const_into_str)]
+enum DecryptionStrategy {
+ Decrypted,
+ Packaged,
+}
+
+#[derive(Debug, ComponentIdentifier)]
+pub(crate) struct DecryptContentPGP {
+ decompress_data: bool,
+ symmetric_password: Option<pgp::types::Password>,
+}
+
+impl Schedule for DecryptContentPGP {
+ fn schedule<P: GetProperty, L>(context: &P, _logger: &L) -> Result<Self,
MinifiError>
+ where
+ Self: Sized,
+ L: Logger,
+ {
+ let decryption_strategy = context.get_property(&DECRYPTION_STRATEGY)?;
+
+ let symmetric_password = context.get_property(&SYMMETRIC_PASSWORD)?;
+ let has_context_service =
context.get_raw_property(&PRIVATE_KEY_SERVICE)?.is_some();
+ if !has_context_service && symmetric_password.is_none() {
+ Err(MinifiError::validation(
+ "Either Symmetric Password or Private Key Service must be set",
+ ))
+ } else {
+ Ok(DecryptContentPGP {
+ decompress_data: decryption_strategy ==
DecryptionStrategy::Decrypted,
+ symmetric_password,
+ })
+ }
+ }
+}
+
+impl DecryptContentPGP {
+ fn decrypt_msg<'a>(
+ &'a self,
+ msg: Message<'a>,
+ private_key_service: Option<&'a PGPPrivateKeyService>,
+ ) -> pgp::errors::Result<Message<'a>> {
+ let mut ring = if let Some(pks) = private_key_service {
+ pks.get_the_ring()
+ } else {
+ TheRing::default()
+ };
+
+ ring.decrypt_options = ring.decrypt_options.enable_gnupg_aead();
+
+ if let Some(sym_passwd) = &self.symmetric_password {
+ ring.message_password.push(sym_passwd);
+ }
+ let (decrypted_msg, _ring_result) = msg.decrypt_the_ring(ring, false)?;
+ Ok(decrypted_msg)
+ }
+
+ fn extract_attributes_from_decrypted_message(
+ decrypted_msg: &Message,
+ ) -> Vec<(&'static str, String)> {
+ let mut res = Vec::new();
+ if let Some(literal_data_header) = decrypted_msg.literal_data_header()
{
+ if let Ok(file_name) =
str::from_utf8(literal_data_header.file_name()) {
+ res.push((LITERAL_DATA_FILENAME.name, file_name.to_string()));
+ }
+ // NiFi uses ms timestamp
+ res.push((
+ LITERAL_DATA_MODIFIED.name,
+ (1000u64 * literal_data_header.created().as_secs() as
u64).to_string(),
+ ));
+ }
+ res
+ }
+}
+
+impl FlowFileStreamTransform for DecryptContentPGP {
+ fn transform<Ctx: GetProperty + GetControllerService, LoggerImpl: Logger>(
+ &self,
+ context: &Ctx,
+ input_stream: &mut dyn InputStream,
+ output_stream: &mut dyn OutputStream,
+ _logger: &LoggerImpl,
+ ) -> Result<TransformStreamResult, ProcessError> {
+ let private_key_service =
context.get_controller_service(&PRIVATE_KEY_SERVICE)?;
+
+ let msg = Message::from_reader(input_stream)
+ .map(|(msg, _header)| msg)
+ .route_err_to_failure()?;
+
+ let mut decrypted_msg = self
+ .decrypt_msg(msg, private_key_service)
+ .route_err_to_failure()?;
+
+ if self.decompress_data && decrypted_msg.is_compressed() {
+ decrypted_msg = decrypted_msg
+ .decompress()
+ .map_err(MinifiError::other)
+ .route_err_to_failure()?
+ };
+
+ let attributes_to_add =
Self::extract_attributes_from_decrypted_message(&decrypted_msg);
+ let _written_bytes =
+ std::io::copy(&mut decrypted_msg.into_inner(),
output_stream).route_err_to_failure()?;
Review Comment:
good catch and the fix is not straightforward, so we will skip this feature
for now and address it in a followup PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
