pact_plugin_driver/
grpc_plugin.rs

1//! gRPC plugin wrapper and process management
2
3use std::process::Command;
4use std::process::Stdio;
5
6use anyhow::anyhow;
7use async_trait::async_trait;
8use log::max_level;
9use os_info::Type;
10use prost::Message;
11use std::path::PathBuf;
12use sysinfo::{Pid, System};
13use tonic::codegen::InterceptedService;
14use tonic::metadata::{Ascii, MetadataValue};
15use tonic::service::Interceptor;
16use tonic::transport::Channel;
17use tonic::{Request, Status};
18use tracing::{debug, warn};
19use uuid::Uuid;
20
21use crate::child_process::ChildPluginProcess;
22use crate::plugin_models::{
23  PactPlugin, PactPluginManifest, PactPluginRpc, PluginInitRequest, PluginInitResponse,
24  PluginInstance, PluginInterfaceVersion,
25};
26use crate::proto::pact_plugin_client::PactPluginClient as PactPluginClientV1;
27use crate::proto::*;
28use crate::proto_v2::{self, pact_plugin_client::PactPluginClient as PactPluginClientV2};
29
30pub(crate) enum PluginClient {
31  V1(PactPluginClientV1<InterceptedService<Channel, PactPluginInterceptor>>),
32  V2(PactPluginClientV2<InterceptedService<Channel, PactPluginInterceptor>>),
33}
34
35impl PluginClient {
36  pub(crate) fn convert_message<T, U>(message: T) -> Result<U, Status>
37  where
38    T: Message,
39    U: Message + Default,
40  {
41    U::decode(message.encode_to_vec().as_slice()).map_err(|err| {
42      Status::internal(format!(
43        "Failed to convert between plugin interface message versions: {}",
44        err
45      ))
46    })
47  }
48
49  async fn init_plugin(
50    &mut self,
51    request: PluginInitRequest,
52  ) -> Result<PluginInitResponse, Status> {
53    match self {
54      PluginClient::V1(client) => client
55        .init_plugin(Request::new(InitPluginRequest {
56          implementation: request.implementation,
57          version: request.version,
58        }))
59        .await
60        .map(|response| PluginInitResponse {
61          catalogue: response.into_inner().catalogue,
62          plugin_capabilities: vec![],
63        }),
64      PluginClient::V2(client) => client
65        .init_plugin(Request::new(proto_v2::InitPluginRequest {
66          implementation: request.implementation,
67          version: request.version,
68          host_capabilities: request.host_capabilities,
69          plugin_instance_id: request.plugin_instance_id,
70        }))
71        .await
72        .and_then(|response| match response.into_inner().response {
73          Some(proto_v2::init_plugin_response::Response::Success(success)) => {
74            Ok(PluginInitResponse {
75              catalogue: success
76                .catalogue
77                .into_iter()
78                .map(Self::convert_message)
79                .collect::<Result<Vec<CatalogueEntry>, Status>>()?,
80              plugin_capabilities: success.plugin_capabilities,
81            })
82          }
83          Some(proto_v2::init_plugin_response::Response::Failure(failure)) => {
84            let mut error = failure.error;
85            if !failure.missing_host_capabilities.is_empty() {
86              error.push_str(" (missing host capabilities: ");
87              error.push_str(failure.missing_host_capabilities.join(", ").as_str());
88              error.push(')');
89            }
90            Err(Status::failed_precondition(error))
91          }
92          None => Err(Status::internal(
93            "Plugin returned an invalid V2 InitPlugin response",
94          )),
95        }),
96    }
97  }
98
99  async fn compare_contents(
100    &mut self,
101    request: CompareContentsRequest,
102  ) -> Result<CompareContentsResponse, Status> {
103    match self {
104      PluginClient::V1(client) => client
105        .compare_contents(Request::new(request))
106        .await
107        .map(|response| response.into_inner()),
108      PluginClient::V2(client) => {
109        let mut v2_req = Self::convert_message::<_, proto_v2::CompareContentsRequest>(request)?;
110        if let Some(id) = crate::test_context::current_test_run_id() {
111          let ctx = v2_req.test_context.get_or_insert_with(prost_types::Struct::default);
112          ctx.fields.entry("testRunId".to_string()).or_insert_with(|| prost_types::Value {
113            kind: Some(prost_types::value::Kind::StringValue(id)),
114          });
115        }
116        client
117          .compare_contents(Request::new(v2_req))
118          .await
119          .and_then(|response| Self::convert_message(response.into_inner()))
120      }
121    }
122  }
123
124  async fn configure_interaction(
125    &mut self,
126    request: ConfigureInteractionRequest,
127  ) -> Result<ConfigureInteractionResponse, Status> {
128    match self {
129      PluginClient::V1(client) => client
130        .configure_interaction(Request::new(request))
131        .await
132        .map(|response| response.into_inner()),
133      PluginClient::V2(client) => {
134        let mut v2_req =
135          Self::convert_message::<_, proto_v2::ConfigureInteractionRequest>(request)?;
136        if let Some(id) = crate::test_context::current_test_run_id() {
137          let ctx = v2_req.test_context.get_or_insert_with(prost_types::Struct::default);
138          ctx.fields.entry("testRunId".to_string()).or_insert_with(|| prost_types::Value {
139            kind: Some(prost_types::value::Kind::StringValue(id)),
140          });
141        }
142        client
143          .configure_interaction(Request::new(v2_req))
144          .await
145          .and_then(|response| Self::convert_message(response.into_inner()))
146      }
147    }
148  }
149
150  async fn generate_content(
151    &mut self,
152    request: GenerateContentRequest,
153  ) -> Result<GenerateContentResponse, Status> {
154    match self {
155      PluginClient::V1(client) => client
156        .generate_content(Request::new(request))
157        .await
158        .map(|response| response.into_inner()),
159      PluginClient::V2(client) => client
160        .generate_content(Request::new(Self::convert_message::<
161          _,
162          proto_v2::GenerateContentRequest,
163        >(request)?))
164        .await
165        .and_then(|response| Self::convert_message(response.into_inner())),
166    }
167  }
168
169  async fn start_mock_server(
170    &mut self,
171    request: StartMockServerRequest,
172  ) -> Result<StartMockServerResponse, Status> {
173    match self {
174      PluginClient::V1(client) => client
175        .start_mock_server(Request::new(request))
176        .await
177        .map(|response| response.into_inner()),
178      PluginClient::V2(client) => client
179        .start_mock_server(Request::new(Self::convert_message::<
180          _,
181          proto_v2::StartMockServerRequest,
182        >(request)?))
183        .await
184        .and_then(|response| Self::convert_message(response.into_inner())),
185    }
186  }
187
188  async fn shutdown_mock_server(
189    &mut self,
190    request: ShutdownMockServerRequest,
191  ) -> Result<ShutdownMockServerResponse, Status> {
192    match self {
193      PluginClient::V1(client) => client
194        .shutdown_mock_server(Request::new(request))
195        .await
196        .map(|response| response.into_inner()),
197      PluginClient::V2(client) => client
198        .shutdown_mock_server(Request::new(Self::convert_message::<
199          _,
200          proto_v2::MockServerRequest,
201        >(request)?))
202        .await
203        .and_then(|response| {
204          Self::convert_message::<_, ShutdownMockServerResponse>(response.into_inner())
205        }),
206    }
207  }
208
209  async fn start_mock_server_v2(
210    &mut self,
211    request: proto_v2::StartMockServerRequest,
212  ) -> Result<StartMockServerResponse, Status> {
213    match self {
214      PluginClient::V1(_) => Err(Status::unimplemented("V2 interface not supported on V1 plugin")),
215      PluginClient::V2(client) => client
216        .start_mock_server(Request::new(request))
217        .await
218        .and_then(|response| Self::convert_message(response.into_inner())),
219    }
220  }
221
222  async fn prepare_interaction_for_verification_v2(
223    &mut self,
224    request: proto_v2::VerificationPreparationRequest,
225  ) -> Result<VerificationPreparationResponse, Status> {
226    match self {
227      PluginClient::V1(_) => Err(Status::unimplemented("V2 interface not supported on V1 plugin")),
228      PluginClient::V2(client) => client
229        .prepare_interaction_for_verification(Request::new(request))
230        .await
231        .and_then(|response| Self::convert_message(response.into_inner())),
232    }
233  }
234
235  async fn verify_interaction_v2(
236    &mut self,
237    request: proto_v2::VerifyInteractionRequest,
238  ) -> Result<VerifyInteractionResponse, Status> {
239    match self {
240      PluginClient::V1(_) => Err(Status::unimplemented("V2 interface not supported on V1 plugin")),
241      PluginClient::V2(client) => client
242        .verify_interaction(Request::new(request))
243        .await
244        .and_then(|response| Self::convert_message(response.into_inner())),
245    }
246  }
247
248  async fn get_mock_server_results(
249    &mut self,
250    request: MockServerRequest,
251  ) -> Result<MockServerResults, Status> {
252    match self {
253      PluginClient::V1(client) => client
254        .get_mock_server_results(Request::new(request))
255        .await
256        .map(|response| response.into_inner()),
257      PluginClient::V2(client) => client
258        .get_mock_server_results(Request::new(Self::convert_message::<
259          _,
260          proto_v2::MockServerRequest,
261        >(request)?))
262        .await
263        .and_then(|response| Self::convert_message(response.into_inner())),
264    }
265  }
266
267  async fn prepare_interaction_for_verification(
268    &mut self,
269    request: VerificationPreparationRequest,
270  ) -> Result<VerificationPreparationResponse, Status> {
271    match self {
272      PluginClient::V1(client) => client
273        .prepare_interaction_for_verification(Request::new(request))
274        .await
275        .map(|response| response.into_inner()),
276      PluginClient::V2(client) => client
277        .prepare_interaction_for_verification(Request::new(Self::convert_message::<
278          _,
279          proto_v2::VerificationPreparationRequest,
280        >(request)?))
281        .await
282        .and_then(|response| Self::convert_message(response.into_inner())),
283    }
284  }
285
286  async fn verify_interaction(
287    &mut self,
288    request: VerifyInteractionRequest,
289  ) -> Result<VerifyInteractionResponse, Status> {
290    match self {
291      PluginClient::V1(client) => client
292        .verify_interaction(Request::new(request))
293        .await
294        .map(|response| response.into_inner()),
295      PluginClient::V2(client) => client
296        .verify_interaction(Request::new(Self::convert_message::<
297          _,
298          proto_v2::VerifyInteractionRequest,
299        >(request)?))
300        .await
301        .and_then(|response| Self::convert_message(response.into_inner())),
302    }
303  }
304
305  async fn update_catalogue(&mut self, request: Catalogue) -> Result<(), Status> {
306    match self {
307      PluginClient::V1(client) => client
308        .update_catalogue(Request::new(request))
309        .await
310        .map(|_| ()),
311      PluginClient::V2(client) => client
312        .update_catalogue(Request::new(
313          Self::convert_message::<_, proto_v2::Catalogue>(request)?,
314        ))
315        .await
316        .map(|_| ()),
317    }
318  }
319}
320
321/// Interceptor to inject the server key as an authorisation header
322#[derive(Clone, Debug)]
323pub(crate) struct PactPluginInterceptor {
324  /// Server key to inject
325  server_key: MetadataValue<Ascii>,
326}
327
328impl PactPluginInterceptor {
329  pub(crate) fn new(server_key: &str) -> anyhow::Result<Self> {
330    let token = MetadataValue::try_from(server_key)?;
331    Ok(PactPluginInterceptor { server_key: token })
332  }
333}
334
335impl Interceptor for PactPluginInterceptor {
336  fn call(&mut self, mut request: Request<()>) -> Result<Request<()>, Status> {
337    request
338      .metadata_mut()
339      .insert("authorization", self.server_key.clone());
340    Ok(request)
341  }
342}
343
344/// Wrapper around `PactPlugin` that provides gRPC connectivity
345#[derive(Debug, Clone)]
346pub struct GrpcPactPlugin {
347  pub plugin: PactPlugin,
348}
349
350impl GrpcPactPlugin {
351  pub fn new(plugin: PactPlugin) -> Self {
352    GrpcPactPlugin { plugin }
353  }
354
355  #[allow(deprecated)]
356  async fn connect_channel(&self) -> anyhow::Result<Channel> {
357    let port = self.plugin.child.port();
358    match Channel::from_shared(format!("http://[::1]:{}", port))?
359      .connect()
360      .await
361    {
362      Ok(channel) => Ok(channel),
363      Err(err) => {
364        debug!("IP6 connection failed, will try IP4 address - {err}");
365        Channel::from_shared(format!("http://127.0.0.1:{}", port))?
366          .connect()
367          .await
368          .map_err(|err| anyhow!(err))
369      }
370    }
371  }
372
373  #[allow(deprecated)]
374  async fn get_plugin_client(&self) -> anyhow::Result<PluginClient> {
375    let channel = self.connect_channel().await?;
376    let interceptor =
377      PactPluginInterceptor::new(self.plugin.child.plugin_info.server_key.as_str())?;
378    match self.plugin.interface_version {
379      PluginInterfaceVersion::V1 => Ok(PluginClient::V1(PactPluginClientV1::with_interceptor(
380        channel,
381        interceptor,
382      ))),
383      PluginInterfaceVersion::V2 => Ok(PluginClient::V2(PactPluginClientV2::with_interceptor(
384        channel,
385        interceptor,
386      ))),
387    }
388  }
389}
390
391#[async_trait]
392impl PactPluginRpc for GrpcPactPlugin {
393  async fn init_plugin(
394    &mut self,
395    request: PluginInitRequest,
396  ) -> anyhow::Result<PluginInitResponse> {
397    let mut client = self.get_plugin_client().await?;
398    client.init_plugin(request).await.map_err(anyhow::Error::from)
399  }
400}
401
402#[async_trait]
403impl PluginInstance for GrpcPactPlugin {
404  fn manifest(&self) -> &PactPluginManifest {
405    &self.plugin.manifest
406  }
407
408  fn instance_id(&self) -> &str {
409    &self.plugin.instance_id
410  }
411
412  fn has_capability(&self, capability: &str) -> bool {
413    self.plugin.has_capability(capability)
414  }
415
416  #[allow(deprecated)]
417  fn kill(&self) {
418    self.plugin.child.kill();
419  }
420
421  async fn compare_contents(
422    &self,
423    request: CompareContentsRequest,
424  ) -> anyhow::Result<CompareContentsResponse> {
425    let mut client = self.get_plugin_client().await?;
426    client.compare_contents(request).await.map_err(anyhow::Error::from)
427  }
428
429  async fn configure_interaction(
430    &self,
431    request: ConfigureInteractionRequest,
432  ) -> anyhow::Result<ConfigureInteractionResponse> {
433    let mut client = self.get_plugin_client().await?;
434    client.configure_interaction(request).await.map_err(anyhow::Error::from)
435  }
436
437  async fn generate_content(
438    &self,
439    request: GenerateContentRequest,
440  ) -> anyhow::Result<GenerateContentResponse> {
441    let mut client = self.get_plugin_client().await?;
442    client.generate_content(request).await.map_err(anyhow::Error::from)
443  }
444
445  async fn start_mock_server(
446    &self,
447    request: StartMockServerRequest,
448  ) -> anyhow::Result<StartMockServerResponse> {
449    let mut client = self.get_plugin_client().await?;
450    client.start_mock_server(request).await.map_err(anyhow::Error::from)
451  }
452
453  async fn start_mock_server_v2(
454    &self,
455    request: proto_v2::StartMockServerRequest,
456  ) -> anyhow::Result<StartMockServerResponse> {
457    let mut client = self.get_plugin_client().await?;
458    client.start_mock_server_v2(request).await.map_err(anyhow::Error::from)
459  }
460
461  async fn shutdown_mock_server(
462    &self,
463    request: ShutdownMockServerRequest,
464  ) -> anyhow::Result<ShutdownMockServerResponse> {
465    let mut client = self.get_plugin_client().await?;
466    client.shutdown_mock_server(request).await.map_err(anyhow::Error::from)
467  }
468
469  async fn get_mock_server_results(
470    &self,
471    request: MockServerRequest,
472  ) -> anyhow::Result<MockServerResults> {
473    let mut client = self.get_plugin_client().await?;
474    client.get_mock_server_results(request).await.map_err(anyhow::Error::from)
475  }
476
477  async fn prepare_interaction_for_verification(
478    &self,
479    request: VerificationPreparationRequest,
480  ) -> anyhow::Result<VerificationPreparationResponse> {
481    let mut client = self.get_plugin_client().await?;
482    client
483      .prepare_interaction_for_verification(request)
484      .await
485      .map_err(anyhow::Error::from)
486  }
487
488  async fn prepare_interaction_for_verification_v2(
489    &self,
490    request: proto_v2::VerificationPreparationRequest,
491  ) -> anyhow::Result<VerificationPreparationResponse> {
492    let mut client = self.get_plugin_client().await?;
493    client
494      .prepare_interaction_for_verification_v2(request)
495      .await
496      .map_err(anyhow::Error::from)
497  }
498
499  async fn verify_interaction(
500    &self,
501    request: VerifyInteractionRequest,
502  ) -> anyhow::Result<VerifyInteractionResponse> {
503    let mut client = self.get_plugin_client().await?;
504    client.verify_interaction(request).await.map_err(anyhow::Error::from)
505  }
506
507  async fn verify_interaction_v2(
508    &self,
509    request: proto_v2::VerifyInteractionRequest,
510  ) -> anyhow::Result<VerifyInteractionResponse> {
511    let mut client = self.get_plugin_client().await?;
512    client.verify_interaction_v2(request).await.map_err(anyhow::Error::from)
513  }
514
515  async fn update_catalogue(&self, request: Catalogue) -> anyhow::Result<()> {
516    let mut client = self.get_plugin_client().await?;
517    client.update_catalogue(request).await.map_err(anyhow::Error::from)
518  }
519}
520
521/// Start a plugin process and return a PactPlugin with the child process attached
522pub(crate) async fn start_plugin_process(manifest: &PactPluginManifest) -> anyhow::Result<PactPlugin> {
523  debug!("Starting plugin with manifest {:?}", manifest);
524
525  let os_info = os_info::get();
526  debug!("Detected OS: {}", os_info);
527  let mut path = if let Some(entry_point) = manifest.entry_points.get(&os_info.to_string()) {
528    PathBuf::from(entry_point)
529  } else if os_info.os_type() == Type::Windows && manifest.entry_points.contains_key("windows") {
530    PathBuf::from(manifest.entry_points.get("windows").unwrap())
531  } else {
532    PathBuf::from(&manifest.entry_point)
533  };
534  if !path.is_absolute() || !path.exists() {
535    path = PathBuf::from(manifest.plugin_dir.clone()).join(path);
536  }
537  debug!("Starting plugin using {:?}", &path);
538
539  let host_port = match crate::plugin_host::ensure_plugin_host_running().await {
540    Ok(port) => Some(port),
541    Err(err) => {
542      warn!("Could not start PluginHost server, Log RPC forwarding will be unavailable: {}", err);
543      None
544    }
545  };
546
547  let log_level = max_level();
548  let mut child_command = Command::new(path.clone());
549  let mut child_command = child_command
550    .env("LOG_LEVEL", log_level.to_string())
551    .env("RUST_LOG", log_level.to_string())
552    .current_dir(manifest.plugin_dir.clone());
553
554  let instance_id = Uuid::new_v4().to_string();
555
556  child_command = child_command.env("PACT_PLUGIN_INSTANCE_ID", &instance_id);
557  if let Some(port) = host_port {
558    child_command = child_command.env("PACT_PLUGIN_HOST", format!("127.0.0.1:{}", port));
559  }
560
561  if let Some(args) = &manifest.args {
562    child_command = child_command.args(args);
563  }
564
565  let child = child_command
566    .stdout(Stdio::piped())
567    .stderr(Stdio::piped())
568    .spawn()
569    .map_err(|err| {
570      anyhow!(
571        "Was not able to start plugin process for '{}' - {}",
572        path.to_string_lossy(),
573        err
574      )
575    })?;
576  let child_pid = child.id();
577  debug!("Plugin {} started with PID {} (instance {})", manifest.name, child_pid, instance_id);
578  crate::plugin_manager::register_plugin_instance(&instance_id, &manifest.name);
579
580  match ChildPluginProcess::new(child, manifest, instance_id.clone()).await {
581    Ok(child) => {
582      let plugin = PactPlugin::new(manifest, child)?;
583      Ok(plugin)
584    }
585    Err(err) => {
586      crate::plugin_manager::deregister_plugin_instance(&instance_id);
587      let mut s = System::new();
588      s.refresh_processes();
589      if let Some(process) = s.process(Pid::from_u32(child_pid)) {
590        #[cfg(not(windows))]
591        process.kill();
592        // revert windows specific logic once https://github.com/GuillaumeGomez/sysinfo/pull/1341/files is merged/released
593        #[cfg(windows)]
594        let _ = Command::new("taskkill.exe")
595          .arg("/PID")
596          .arg(child_pid.to_string())
597          .arg("/F")
598          .arg("/T")
599          .output();
600      } else {
601        warn!("Child process with PID {} was not found", child_pid);
602      }
603      Err(err)
604    }
605  }
606}
607
608#[cfg(test)]
609pub(crate) mod tests {
610  use tonic::Status;
611
612  use crate::plugin_models::PluginInitResponse;
613  use crate::proto::*;
614
615  use super::PluginClient;
616
617  #[test]
618  fn converts_between_v1_and_v2_messages() {
619    use std::collections::HashMap;
620    use crate::plugin_models::PluginInitRequest;
621    use crate::proto_v2;
622
623    let request = PluginInitRequest {
624      implementation: "plugin-driver-rust".to_string(),
625      version: "1.0.0-beta.1".to_string(),
626      host_capabilities: vec!["interaction/request-response".to_string()],
627      plugin_instance_id: "test-instance-id".to_string(),
628    };
629
630    let converted_request = proto_v2::InitPluginRequest {
631      implementation: request.implementation,
632      version: request.version,
633      host_capabilities: request.host_capabilities,
634      plugin_instance_id: request.plugin_instance_id,
635    };
636    assert_eq!(converted_request.implementation, "plugin-driver-rust");
637    assert_eq!(converted_request.version, "1.0.0-beta.1");
638    assert_eq!(
639      converted_request.host_capabilities,
640      vec!["interaction/request-response"]
641    );
642
643    let response = proto_v2::InitPluginResponse {
644      response: Some(proto_v2::init_plugin_response::Response::Success(
645        proto_v2::InitPluginSuccess {
646          catalogue: vec![proto_v2::CatalogueEntry {
647            r#type: proto_v2::catalogue_entry::EntryType::ContentMatcher as i32,
648            key: "test".to_string(),
649            values: HashMap::new(),
650          }],
651          plugin_capabilities: vec!["plugin/verification".to_string()],
652        },
653      )),
654    };
655
656    let converted_response = match response.response.unwrap() {
657      proto_v2::init_plugin_response::Response::Success(success) => PluginInitResponse {
658        catalogue: success
659          .catalogue
660          .into_iter()
661          .map(PluginClient::convert_message)
662          .collect::<Result<Vec<CatalogueEntry>, Status>>()
663          .unwrap(),
664        plugin_capabilities: success.plugin_capabilities,
665      },
666      _ => unreachable!(),
667    };
668    assert_eq!(converted_response.catalogue.len(), 1);
669    assert_eq!(converted_response.catalogue[0].key, "test");
670    assert_eq!(
671      converted_response.catalogue[0].r#type,
672      catalogue_entry::EntryType::ContentMatcher as i32
673    );
674    assert_eq!(converted_response.plugin_capabilities, vec!["plugin/verification"]);
675  }
676}
pact_plugin_driver/grpc_plugin.rs

pact_plugin_driver/
grpc_plugin.rs
