feat: add Authorization to the Public API (v2) (#3966)

* add new module v2

* removed address resolver

* wip

* add tests for public api V2

* deprecation warning

* fix test

* fix copyright

extensions/data-plane/data-plane-public-api-v2/build.gradle.kts

@@ -0,0 +1,46 @@
+/*
+ *  Copyright (c) 2022 Microsoft Corporation
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Microsoft Corporation - initial API and implementation
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - improvements
+ *       Mercedes-Benz Tech Innovation GmbH - publish public api context into dedicated swagger hub page
+ *
+ */
+
+
+plugins {
+    `java-library`
+    id("io.swagger.core.v3.swagger-gradle-plugin")
+}
+
+dependencies {
+    api(project(":spi:common:http-spi"))
+    api(project(":spi:common:web-spi"))
+    api(project(":spi:data-plane:data-plane-spi"))
+    implementation(project(":core:common:util"))
+
+    implementation(project(":core:data-plane:data-plane-util"))
+    implementation(libs.jakarta.rsApi)
+
+    testImplementation(project(":extensions:common:http"))
+    testImplementation(project(":core:common:junit"))
+    testImplementation(libs.jersey.multipart)
+    testImplementation(libs.restAssured)
+    testImplementation(libs.mockserver.netty)
+    testImplementation(libs.mockserver.client)
+    testImplementation(testFixtures(project(":extensions:common:http:jersey-core")))
+}
+edcBuild {
+    swagger {
+        apiGroup.set("public-api")
+    }
+}
+
+

extensions/data-plane/data-plane-public-api-v2/src/main/java/org/eclipse/edc/connector/dataplane/api/DataPlanePublicApiExtension.java

@@ -0,0 +1,86 @@
+/*
+ *  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.dataplane.api;
+
+import org.eclipse.edc.connector.dataplane.api.controller.DataPlanePublicApiV2Controller;
+import org.eclipse.edc.connector.dataplane.spi.iam.DataPlaneAuthorizationService;
+import org.eclipse.edc.connector.dataplane.spi.pipeline.PipelineService;
+import org.eclipse.edc.runtime.metamodel.annotation.Extension;
+import org.eclipse.edc.runtime.metamodel.annotation.Inject;
+import org.eclipse.edc.spi.system.ExecutorInstrumentation;
+import org.eclipse.edc.spi.system.ServiceExtension;
+import org.eclipse.edc.spi.system.ServiceExtensionContext;
+import org.eclipse.edc.web.spi.WebServer;
+import org.eclipse.edc.web.spi.WebService;
+import org.eclipse.edc.web.spi.configuration.WebServiceConfigurer;
+import org.eclipse.edc.web.spi.configuration.WebServiceSettings;
+
+import java.util.concurrent.Executors;
+
+/**
+ * This extension provides generic endpoints which are open to public participants of the Dataspace to execute
+ * requests on the actual data source.
+ */
+@Extension(value = DataPlanePublicApiExtension.NAME)
+public class DataPlanePublicApiExtension implements ServiceExtension {
+    public static final String NAME = "Data Plane Public API";
+    private static final int DEFAULT_PUBLIC_PORT = 8185;
+    private static final String PUBLIC_API_CONFIG = "web.http.public";
+    private static final String PUBLIC_CONTEXT_ALIAS = "public";
+    private static final String PUBLIC_CONTEXT_PATH = "/api/v2/public";
+
+    private static final int DEFAULT_THREAD_POOL = 10;
+
+    private static final WebServiceSettings PUBLIC_SETTINGS = WebServiceSettings.Builder.newInstance()
+            .apiConfigKey(PUBLIC_API_CONFIG)
+            .contextAlias(PUBLIC_CONTEXT_ALIAS)
+            .defaultPath(PUBLIC_CONTEXT_PATH)
+            .defaultPort(DEFAULT_PUBLIC_PORT)
+            .name(NAME)
+            .build();
+
+    @Inject
+    private WebServer webServer;
+
+    @Inject
+    private WebServiceConfigurer webServiceConfigurer;
+
+    @Inject
+    private PipelineService pipelineService;
+
+    @Inject
+    private WebService webService;
+
+    @Inject
+    private ExecutorInstrumentation executorInstrumentation;
+    @Inject
+    private DataPlaneAuthorizationService authorizationService;
+
+    @Override
+    public String name() {
+        return NAME;
+    }
+
+    @Override
+    public void initialize(ServiceExtensionContext context) {
+        var configuration = webServiceConfigurer.configure(context, webServer, PUBLIC_SETTINGS);
+        var executorService = executorInstrumentation.instrument(
+                Executors.newFixedThreadPool(DEFAULT_THREAD_POOL),
+                "Data plane proxy transfers"
+        );
+        var publicApiController = new DataPlanePublicApiV2Controller(pipelineService, executorService, authorizationService);
+        webService.registerResource(configuration.getContextAlias(), publicApiController);
+    }
+}

extensions/data-plane/data-plane-public-api-v2/src/main/java/org/eclipse/edc/connector/dataplane/api/controller/ContainerRequestContextApi.java

@@ -0,0 +1,68 @@
+/*
+ *  Copyright (c) 2022 Amadeus
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Amadeus - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.dataplane.api.controller;
+
+import jakarta.ws.rs.container.ContainerRequestContext;
+
+import java.util.Map;
+
+/**
+ * Wrapper around {@link ContainerRequestContext} enabling mocking.
+ */
+public interface ContainerRequestContextApi {
+
+    /**
+     * Get the request headers. Note that if more than one value is associated to a specific header,
+     * only the first one is retained.
+     *
+     * @return Headers map.
+     */
+    Map<String, String> headers();
+
+    /**
+     * Format query of the request as string, e.g. "hello=world\&amp;foo=bar".
+     *
+     * @return Query param string.
+     */
+    String queryParams();
+
+    /**
+     * Format the request body into a string.
+     *
+     * @return Request body.
+     */
+    String body();
+
+    /**
+     * Get the media type from incoming request.
+     *
+     * @return Media type.
+     */
+    String mediaType();
+
+    /**
+     * Return request path, e.g. "hello/world/foo/bar".
+     *
+     * @return Path string.
+     */
+    String path();
+
+    /**
+     * Get http method from the incoming request, e.g. "GET", "POST"...
+     *
+     * @return Http method.
+     */
+    String method();
+}

extensions/data-plane/data-plane-public-api-v2/src/main/java/org/eclipse/edc/connector/dataplane/api/controller/ContainerRequestContextApiImpl.java

@@ -0,0 +1,108 @@
+/*
+ *  Copyright (c) 2022 Amadeus
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Amadeus - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.dataplane.api.controller;
+
+import jakarta.ws.rs.container.ContainerRequestContext;
+import jakarta.ws.rs.core.MediaType;
+import org.eclipse.edc.spi.EdcException;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * This class provides a set of API wrapping a {@link ContainerRequestContext}.
+ */
+public class ContainerRequestContextApiImpl implements ContainerRequestContextApi {
+
+    private static final String QUERY_PARAM_SEPARATOR = "&";
+
+    private final ContainerRequestContext context;
+
+    public ContainerRequestContextApiImpl(ContainerRequestContext context) {
+        this.context = context;
+    }
+
+    @Override
+    public Map<String, String> headers() {
+        return context.getHeaders().entrySet()
+                .stream()
+                .filter(entry -> !entry.getValue().isEmpty())
+                .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().get(0)));
+    }
+
+    @Override
+    public String queryParams() {
+        return context.getUriInfo().getQueryParameters().entrySet()
+                .stream()
+                .map(entry -> new QueryParam(entry.getKey(), entry.getValue()))
+                .filter(QueryParam::isValid)
+                .map(QueryParam::toString)
+                .collect(Collectors.joining(QUERY_PARAM_SEPARATOR));
+    }
+
+    @Override
+    public String body() {
+        try (BufferedReader br = new BufferedReader(new InputStreamReader(context.getEntityStream()))) {
+            return br.lines().collect(Collectors.joining("\n"));
+        } catch (IOException e) {
+            throw new EdcException("Failed to read request body: " + e.getMessage());
+        }
+    }
+
+    @Override
+    public String path() {
+        var pathInfo = context.getUriInfo().getPath();
+        return pathInfo.startsWith("/") ? pathInfo.substring(1) : pathInfo;
+    }
+
+    @Override
+    public String mediaType() {
+        return Optional.ofNullable(context.getMediaType())
+                .map(MediaType::toString)
+                .orElse(null);
+    }
+
+    @Override
+    public String method() {
+        return context.getMethod();
+    }
+
+    private static final class QueryParam {
+
+        private final String key;
+        private final List<String> values;
+        private final boolean valid;
+
+        private QueryParam(String key, List<String> values) {
+            this.key = key;
+            this.values = values;
+            this.valid = key != null && values != null && !values.isEmpty();
+        }
+
+        public boolean isValid() {
+            return valid;
+        }
+
+        @Override
+        public String toString() {
+            return valid ? key + "=" + values.get(0) : "";
+        }
+    }
+}

extensions/data-plane/data-plane-public-api-v2/src/main/java/org/eclipse/edc/connector/dataplane/api/controller/DataFlowRequestSupplier.java

@@ -0,0 +1,71 @@
+/*
+ *  Copyright (c) 2022 Amadeus
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Amadeus - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.dataplane.api.controller;
+
+import org.eclipse.edc.connector.dataplane.util.sink.AsyncStreamingDataSink;
+import org.eclipse.edc.spi.types.domain.DataAddress;
+import org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.function.BiFunction;
+
+import static org.eclipse.edc.connector.dataplane.spi.schema.DataFlowRequestSchema.BODY;
+import static org.eclipse.edc.connector.dataplane.spi.schema.DataFlowRequestSchema.MEDIA_TYPE;
+import static org.eclipse.edc.connector.dataplane.spi.schema.DataFlowRequestSchema.METHOD;
+import static org.eclipse.edc.connector.dataplane.spi.schema.DataFlowRequestSchema.PATH;
+import static org.eclipse.edc.connector.dataplane.spi.schema.DataFlowRequestSchema.QUERY_PARAMS;
+
+public class DataFlowRequestSupplier implements BiFunction<ContainerRequestContextApi, DataAddress, DataFlowStartMessage> {
+
+    /**
+     * Put all properties of the incoming request (method, request body, query params...) into a map.
+     */
+    private static Map<String, String> createProps(ContainerRequestContextApi contextApi) {
+        var props = new HashMap<String, String>();
+        props.put(METHOD, contextApi.method());
+        props.put(QUERY_PARAMS, contextApi.queryParams());
+        props.put(PATH, contextApi.path());
+        Optional.ofNullable(contextApi.mediaType())
+                .ifPresent(mediaType -> {
+                    props.put(MEDIA_TYPE, mediaType);
+                    props.put(BODY, contextApi.body());
+                });
+        return props;
+    }
+
+    /**
+     * Create a {@link DataFlowStartMessage} based on incoming request and claims decoded from the access token.
+     *
+     * @param contextApi  Api for accessing request properties.
+     * @param dataAddress Source data address.
+     * @return DataFlowRequest
+     */
+    @Override
+    public DataFlowStartMessage apply(ContainerRequestContextApi contextApi, DataAddress dataAddress) {
+        var props = createProps(contextApi);
+        return DataFlowStartMessage.Builder.newInstance()
+                .processId(UUID.randomUUID().toString())
+                .sourceDataAddress(dataAddress)
+                .destinationDataAddress(DataAddress.Builder.newInstance()
+                        .type(AsyncStreamingDataSink.TYPE)
+                        .build())
+                .id(UUID.randomUUID().toString())
+                .properties(props)
+                .build();
+    }
+}