feat: allow custom HTTP headers in event_publishers configuration (#…

…140)

* WIP

* feat init agetn_holder

* feat: add `HolderState`

* feat: add Holder functionality to `agent_store` and `agent_api_rest`

* feat: add Holder functionality to Event Publisher

* feat: add `SendCredentialOffer` to `agent_verification`

* feat: add `/offers/send` issuance endpoint to `agent_api_rest`

* fix: remove incorrect Content Type

* feat: add `Status` enum

* feat: add REST API for Holder

* feat: add `AllOffersView`

* feat: add Holder views to `init.sql`

* fix: fix `OfferView` update

* feat: add credentials endpoint for Holder

* refactor: refactor Router

* test: refactor test framework

* refactor: deprecate `path` closure

* refactor: remove unused dependencies

* style: add clippy exception

* build: bump oid4vc dependencies

* refactor: move all `CustomQuery` logic to `agent_shared`

* fix: add Into<SubjectSyntaxType> for SupportedDidMethod

* fix: return 200 OK when list is empty

* refactor: clean up code

* fix: Fix error handling for the Offer aggregate

* fix: add error handling for to Offer aggregate

* refactor: apply clippy suggestion

* test: update Postman Collection

* feat: add Events to `config.rs`

* docs: add new Holder events  to `agent_event_publisher_http` documentation

* feat: init `agent_identity`

* style: use consistent nameing for `View` variables

* style: rename variables

* refactor: use `type` for `View`s to reduce code duplication

* refactor: use `Jwt` instead of `Value`

* feat: add error handling

* fix: remove `presentation_id` from route

* refactor: add error handling and comments

* refactor: remove unused dependencies

* build: remove unused dependencies

* feat: add `UnsupportedCredentialFormatError` error

* test: update Postman Collection

* feat: update `init.sql` file

* feat: add tests and error handling

* feat: add error handling

* test: add unit tests for `Service`, `Presentation` and received `Offer`

* feat: add `GET` method for `/v0/services` endpoint

* test: update Postman collection

* docs: add document, service and presentation events

* fix: remove unused import

* ci: add DS_Store to .gitignore file

* feat: add Document and Service to config.rs

* fix: update .env.example variables

* feat: make `/accept` endpoint respond with the Offer

* feat: add individual aggregate instance endpoints

* test: update Postman collection

* fix: add `all_authorization_requests` table

* test: update test

* feat: init identity `Connection` aggregate

* refactor: merge verification Connection into AuthorizationRequest aggregate

* feat: add `Connection` aggregate and corresponding endpoints

* feat: add `Connection` related endpoints to Postman collection

* fix: add `all_connections` table

* fix: undo openbadgesv3_credentials change

* feat: add Status field to Issuance Offer and Credential

* feat: support Headers in Event Publisher

* feat: change `openid4vci/offers` from GET to POST

* test: add `services/:service_id` endpoints to Postman collection

* fix: change method from `get` to `post`

* feat: add public `/linked-verifiable-presentations` endpoint

* feat: add `offers_params` endpoint handler

* style: make requests human-readable

* test: update path of linked VP in test

* style: use human-readable names for requests

* fix: remove duplicate functions

* feat: add GET service by ID endpoint

* docs: add aggregate fields

* docs: add README file for `AuthorizationRequest` aggregate

* fix: remove duplicate request

* test: make sure that the headers are read from the config file

* fix: change `credential_offer_endpoint` to camelCase

* fix: change request names

* feat: add optional `alias` field to `Connection` aggregate

* feat: add `Query Connection by Alias` request

* refactor: replace `http-serde-ext`  with `http-serde`

* docs: add custom HTTP header example

agent_application/example.config.yaml

@@ -43,6 +43,8 @@ event_publishers:
   http:
     enabled: false
     target_url: "https://my-domain.example.org/event-subscriber"
+    headers:
+      Authorization: Basic YWxhZGRpbjpvcGVuc2VzYW1l
     events:
       server_config: []
       credential: [UnsignedCredentialCreated, CredentialSigned]

agent_event_publisher_http/Cargo.toml

@@ -15,6 +15,7 @@ agent_verification = { path = "../agent_verification" }
 anyhow = "1.0"
 async-trait.workspace = true
 cqrs-es.workspace = true
+http-serde = "2.1"
 rustls = { version = "0.23", default-features = false, features = [
     "logging",
     "std",

agent_event_publisher_http/src/lib.rs

@@ -46,6 +46,7 @@ impl EventPublisherHttp {
         let connection = (!event_publisher_http.events.connection.is_empty()).then(|| {
             AggregateEventPublisherHttp::<Connection>::new(
                 event_publisher_http.target_url.clone(),
+                event_publisher_http.headers.clone(),
                 event_publisher_http
                     .events
                     .connection
@@ -58,6 +59,7 @@ impl EventPublisherHttp {
         let server_config = (!event_publisher_http.events.server_config.is_empty()).then(|| {
             AggregateEventPublisherHttp::<ServerConfig>::new(
                 event_publisher_http.target_url.clone(),
+                event_publisher_http.headers.clone(),
                 event_publisher_http
                     .events
                     .server_config
@@ -70,6 +72,7 @@ impl EventPublisherHttp {
         let credential = (!event_publisher_http.events.credential.is_empty()).then(|| {
             AggregateEventPublisherHttp::<Credential>::new(
                 event_publisher_http.target_url.clone(),
+                event_publisher_http.headers.clone(),
                 event_publisher_http
                     .events
                     .credential
@@ -82,6 +85,7 @@ impl EventPublisherHttp {
         let offer = (!event_publisher_http.events.offer.is_empty()).then(|| {
             AggregateEventPublisherHttp::<Offer>::new(
                 event_publisher_http.target_url.clone(),
+                event_publisher_http.headers.clone(),
                 event_publisher_http
                     .events
                     .offer
@@ -94,6 +98,7 @@ impl EventPublisherHttp {
         let holder_credential = (!event_publisher_http.events.holder_credential.is_empty()).then(|| {
             AggregateEventPublisherHttp::<agent_holder::credential::aggregate::Credential>::new(
                 event_publisher_http.target_url.clone(),
+                event_publisher_http.headers.clone(),
                 event_publisher_http
                     .events
                     .holder_credential
@@ -106,6 +111,7 @@ impl EventPublisherHttp {
         let received_offer = (!event_publisher_http.events.received_offer.is_empty()).then(|| {
             AggregateEventPublisherHttp::<agent_holder::offer::aggregate::Offer>::new(
                 event_publisher_http.target_url.clone(),
+                event_publisher_http.headers.clone(),
                 event_publisher_http
                     .events
                     .received_offer
@@ -118,6 +124,7 @@ impl EventPublisherHttp {
         let authorization_request = (!event_publisher_http.events.authorization_request.is_empty()).then(|| {
             AggregateEventPublisherHttp::<AuthorizationRequest>::new(
                 event_publisher_http.target_url.clone(),
+                event_publisher_http.headers.clone(),
                 event_publisher_http
                     .events
                     .authorization_request
@@ -195,6 +202,8 @@ where
     A: Aggregate,
 {
     pub target_url: String,
+    #[serde(with = "http_serde::option::header_map", default)]
+    pub headers: Option<reqwest::header::HeaderMap>,
     pub target_events: Vec<String>,
     #[serde(skip)]
     pub client: reqwest::Client,
@@ -206,9 +215,10 @@ impl<A> AggregateEventPublisherHttp<A>
 where
     A: Aggregate,
 {
-    pub fn new(target_url: String, target_events: Vec<String>) -> Self {
+    pub fn new(target_url: String, headers: Option<reqwest::header::HeaderMap>, target_events: Vec<String>) -> Self {
         AggregateEventPublisherHttp {
             target_url,
+            headers,
             target_events,
             client: reqwest::Client::new(),
             _marker: std::marker::PhantomData,
@@ -224,11 +234,15 @@ where
     async fn dispatch(&self, _view_id: &str, events: &[EventEnvelope<A>]) {
         for event in events {
             if self.target_events.contains(&event.payload.event_type()) {
-                let request = self.client.post(&self.target_url).json(&event.payload);
+                let mut request = self.client.post(&self.target_url).json(&event.payload);
+
+                if let Some(headers) = &self.headers {
+                    request = request.headers(headers.clone());
+                }
 
                 info!(
-                    "Dispatching event: {:?} to HTTP endpoint: {:?}",
-                    &event.payload, &self.target_url
+                    "Dispatching event: {:?} to HTTP endpoint: {:?} with headers: {:?}",
+                    event.payload, self.target_url, self.headers
                 );
 
                 // Send the request in a separate thread so that we don't have to await the response in the current thread.
@@ -301,10 +315,16 @@ mod tests {
         // Wait for the request to arrive at the mock server endpoint.
         std::thread::sleep(std::time::Duration::from_millis(100));
 
+        let received_requests = mock_server.received_requests().await;
+        let received_request = received_requests.as_ref().unwrap().first().unwrap();
+
         // Assert that the event was dispatched to the target URL.
+        assert_eq!(offer_event, serde_json::from_slice(&received_request.body).unwrap());
+
+        // Assert that the request contained the expected headers.
         assert_eq!(
-            offer_event,
-            serde_json::from_slice(&mock_server.received_requests().await.unwrap().first().unwrap().body).unwrap()
+            "Basic YWxhZGRpbjpvcGVuc2VzYW1l",
+            received_request.headers.get("Authorization").unwrap()
         );
 
         // A new event for the `Offer` aggregate that the publisher is not interested in.

agent_shared/Cargo.toml

@@ -9,6 +9,7 @@ async-trait.workspace = true
 config = { version = "0.14" }
 cqrs-es.workspace = true
 dotenvy = { version = "0.15" }
+http-serde = "2.1"
 # TODO: replace all identity_* with identity_iota?
 identity_iota.workspace = true
 jsonwebtoken.workspace = true
@@ -17,6 +18,7 @@ oid4vci.workspace = true
 oid4vp.workspace = true
 once_cell.workspace = true
 rand = "0.8"
+reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 serde.workspace = true
 serde_json.workspace = true
 serde_with = "3.0"

agent_shared/src/config.rs

@@ -111,6 +111,8 @@ pub struct EventPublishers {
 pub struct EventPublisherHttp {
     pub enabled: bool,
     pub target_url: String,
+    #[serde(with = "http_serde::option::header_map", default)]
+    pub headers: Option<reqwest::header::HeaderMap>,
     pub events: Events,
 }
 

agent_shared/tests/test-config.yaml

@@ -32,6 +32,8 @@ event_publishers:
   http:
     enabled: false
     target_url: "http://localhost"
+    headers:
+      Authorization: Basic YWxhZGRpbjpvcGVuc2VzYW1l
     events:
       server_config: []
       credential: [UnsignedCredentialCreated, CredentialSigned]

agent_verification/Cargo.toml

@@ -16,6 +16,7 @@ jsonwebtoken.workspace = true
 oid4vc-core.workspace = true
 oid4vc-manager.workspace = true
 oid4vp.workspace = true
+reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 serde.workspace = true
 serde_json.workspace = true
 siopv2.workspace = true