✨ Pass contract details to API backend in Consumer Pull HTTP requests

fundacionctic · May 13, 2024 · fb10e7a · fb10e7a
1 parent 6279ca2
commit fb10e7a
Show file tree

Hide file tree

Showing 6 changed files with 149 additions and 23 deletions.
diff --git a/connector/openapi-connector/build.gradle.kts b/connector/openapi-connector/build.gradle.kts
@@ -30,6 +30,7 @@ dependencies {
     implementation(libs.swaggerParser)
     implementation(libs.slugify)
     implementation(libs.json)
+    implementation(libs.okhttp3.okhttp)
 
     if (
         project.hasProperty("useOauthIdentity") &&

diff --git a/...i-connector/src/main/java/eu/datacellar/connector/ContractDetailsHttpParamsDecorator.java b/...i-connector/src/main/java/eu/datacellar/connector/ContractDetailsHttpParamsDecorator.java
@@ -0,0 +1,137 @@
+package eu.datacellar.connector;
+
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import org.eclipse.edc.connector.contract.spi.negotiation.store.ContractNegotiationStore;
+import org.eclipse.edc.connector.dataplane.http.spi.HttpDataAddress;
+import org.eclipse.edc.connector.dataplane.http.spi.HttpParamsDecorator;
+import org.eclipse.edc.connector.dataplane.http.spi.HttpRequestParams.Builder;
+import org.eclipse.edc.spi.monitor.Monitor;
+import org.eclipse.edc.spi.types.domain.agreement.ContractAgreement;
+import org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage;
+
+import okhttp3.HttpUrl;
+
+/**
+ * Decorator class for adding contract details to the proxied HTTP requests.
+ */
+public class ContractDetailsHttpParamsDecorator implements HttpParamsDecorator {
+
+    private final Monitor monitor;
+    private final ContractNegotiationStore contractNegotiationStore;
+    private static final String QUERY_PARAM_CONTRACT_ID = "contractId";
+    private static final String HEADER_CONNECTOR_EXT = "X-Connector-Extension";
+    private static final String HEADER_ASSET_ID = "X-Connector-Asset-Id";
+    private static final String HEADER_CONSUMER_ID = "X-Connector-Consumer-Id";
+    private static final String HEADER_CONTRACT_SIGNING_DATE = "X-Connector-Contract-Signing-Date";
+    private static final String PROP_QUERY_PARAMS = "queryParams";
+
+    /**
+     * Constructs a new instance of the ContractDetailsHttpParamsDecorator class.
+     *
+     * @param monitor                  The monitor object used for monitoring.
+     * @param contractNegotiationStore The contract negotiation store object used
+     *                                 for storing contract details.
+     */
+    public ContractDetailsHttpParamsDecorator(Monitor monitor, ContractNegotiationStore contractNegotiationStore) {
+        this.monitor = monitor;
+        this.contractNegotiationStore = contractNegotiationStore;
+    }
+
+    private Map<String, List<String>> queryParamsStringToMap(String queryParams) {
+        Map<String, List<String>> queryMap = new HashMap<>();
+
+        HttpUrl url = HttpUrl.parse("https://example.com?%s".formatted(queryParams));
+
+        for (int i = 0, size = url.querySize(); i < size; i++) {
+            String key = url.queryParameterName(i);
+            String value = url.queryParameterValue(i);
+
+            if (queryMap.containsKey(key)) {
+                queryMap.get(key).add(value);
+            } else {
+                queryMap.put(key, List.of(value));
+            }
+        }
+
+        return queryMap;
+    }
+
+    private String mapToQueryParamsString(Map<String, List<String>> queryMap) {
+        HttpUrl.Builder urlBuilder = new HttpUrl.Builder()
+                .scheme("https")
+                .host("example.com");
+
+        for (Map.Entry<String, List<String>> entry : queryMap.entrySet()) {
+            for (String value : entry.getValue()) {
+                urlBuilder.addQueryParameter(entry.getKey(), value);
+            }
+        }
+
+        HttpUrl url = urlBuilder.build();
+
+        return url.query();
+    }
+
+    /**
+     * Add contract details to the proxied request.
+     */
+    @Override
+    public Builder decorate(DataFlowStartMessage request, HttpDataAddress address, Builder builder) {
+        Package pkg = ContractDetailsHttpParamsDecorator.class.getPackage();
+        String packageName = pkg.getName();
+        builder.header(HEADER_CONNECTOR_EXT, packageName);
+
+        String queryParams = request.getProperties().getOrDefault(PROP_QUERY_PARAMS, null);
+
+        if (queryParams == null || queryParams.isEmpty()) {
+            monitor.debug("Query parameters not found in request properties");
+            return builder;
+        }
+
+        monitor.debug("Request properties: %s".formatted(request.getProperties()));
+        Map<String, List<String>> queryMap = queryParamsStringToMap(queryParams);
+        monitor.debug("Parsed query map: %s".formatted(queryMap));
+
+        String contractId = Optional.ofNullable(queryMap.get(QUERY_PARAM_CONTRACT_ID))
+                .map(list -> list.get(0))
+                .orElse(null);
+
+        if (contractId == null) {
+            monitor.debug("Contract ID not found in query parameters");
+            return builder;
+        }
+
+        monitor.debug("Contract ID: %s".formatted(contractId));
+        ContractAgreement contractAgreement = contractNegotiationStore.findContractAgreement(contractId);
+
+        if (contractAgreement == null) {
+            monitor.debug("Contract agreement with id '%s' not found in store".formatted(contractId));
+            return builder;
+        }
+
+        monitor.debug("Contract agreement: %s".formatted(contractAgreement));
+
+        builder.header(HEADER_ASSET_ID, contractAgreement.getAssetId());
+        builder.header(HEADER_CONSUMER_ID, contractAgreement.getConsumerId());
+
+        String contractSigningDate = DateTimeFormatter.ISO_INSTANT
+                .withZone(ZoneOffset.UTC)
+                .format(Instant.ofEpochSecond(contractAgreement.getContractSigningDate()));
+
+        builder.header(HEADER_CONTRACT_SIGNING_DATE, contractSigningDate);
+
+        queryMap.remove(QUERY_PARAM_CONTRACT_ID);
+        String updatedQueryString = mapToQueryParamsString(queryMap);
+        monitor.debug("Updated query string: %s".formatted(updatedQueryString));
+        builder.queryParams(updatedQueryString);
+
+        return builder;
+    }
+}
diff --git a/connector/openapi-connector/src/main/java/eu/datacellar/connector/OpenAPICoreExtension.java b/connector/openapi-connector/src/main/java/eu/datacellar/connector/OpenAPICoreExtension.java
@@ -13,6 +13,7 @@
 
 import javax.sql.DataSource;
 
+import org.eclipse.edc.connector.contract.spi.negotiation.store.ContractNegotiationStore;
 import org.eclipse.edc.connector.contract.spi.offer.store.ContractDefinitionStore;
 import org.eclipse.edc.connector.contract.spi.types.offer.ContractDefinition;
 import org.eclipse.edc.connector.dataplane.http.spi.HttpDataAddress;
@@ -106,6 +107,9 @@ public class OpenAPICoreExtension implements ServiceExtension {
     @Inject
     private ContractDefinitionStore contractStore;
 
+    @Inject
+    private ContractNegotiationStore contractNegotiationStore;
+
     @Inject
     private DataPlaneInstanceStore dataPlaneStore;
 
@@ -404,28 +408,8 @@ public void initialize(ServiceExtensionContext context) {
             monitor.warning(String.format("OpenAPI URL (property '%s') is not set", OPENAPI_URL));
         }
 
-        Package pkg = OpenAPICoreExtension.class.getPackage();
-        String pkgVersion = pkg.getImplementationVersion();
-
-        paramsProvider.registerSourceDecorator((request, address, builder) -> {
-            if (pkgVersion != null) {
-                builder.header("X-OpenAPI-Connector-Source-Version", pkgVersion);
-            }
-
-            builder.header("X-OpenAPI-Connector", "source");
-
-            return builder;
-        });
-
-        paramsProvider.registerSinkDecorator((request, address, builder) -> {
-            if (pkgVersion != null) {
-                builder.header("X-OpenAPI-Connector-Sink-Version", pkgVersion);
-            }
-
-            builder.header("X-OpenAPI-Connector", "sink");
-
-            return builder;
-        });
+        paramsProvider
+                .registerSourceDecorator(new ContractDetailsHttpParamsDecorator(monitor, contractNegotiationStore));
 
         monitor.info(String.format("Initialized extension: %s", this.getClass().getName()));
     }

diff --git a/edcpy/edcpy/backend.py b/edcpy/edcpy/backend.py
@@ -36,6 +36,7 @@ class EndpointDataReference(BaseModel):
     authKey: str
     authCode: str
     properties: dict
+    contractId: str
 
 
 async def get_messaging_app() -> AsyncGenerator[MessagingApp, None]:
@@ -127,6 +128,7 @@ async def http_pull_endpoint(
         endpoint=item.endpoint,
         id=item.id,
         properties=item.properties,
+        contract_id=item.contractId,
     )
 
     # The provider hostname is included in the routing key to facilitate

diff --git a/edcpy/edcpy/messaging.py b/edcpy/edcpy/messaging.py
@@ -26,6 +26,7 @@ class HttpPullMessage(BaseModel):
     endpoint: str
     id: str
     properties: dict
+    contract_id: str
 
     @property
     def http_method(self) -> str:
@@ -46,6 +47,7 @@ def request_args(self) -> dict:
             "method": self.http_method,
             "url": self.endpoint,
             "headers": {self.auth_key: self.auth_code},
+            "params": {"contractId": self.contract_id},
         }
 
     @property

diff --git a/edcpy/pyproject.toml b/edcpy/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "edcpy"
-version = "0.3.0a0"
+version = "0.4.0a0"
 description = "Package that provides a series of utilities to facilitate interaction with the Management and Control APIs of an EDC connector"
 authors = ["Andres Garcia Mangas <[email protected]>"]
 license = "EUPL-1.2"