Skip to content

Commit db40972

Browse files
ndr-brtndr-brt
authored
feat: provide full mutual authentication (#58)
1 parent 5051fe6 commit db40972

22 files changed

Lines changed: 424 additions & 151 deletions

‎build.gradle.kts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ plugins {
88
}
99

1010
group = "org.eclipse.dataplane-core"
11-
version = "0.0.7-SNAPSHOT"
11+
version = "0.0.8-SNAPSHOT"
1212

1313
repositories {
1414
mavenCentral()

‎src/main/java/org/eclipse/dataplane/Dataplane.java‎

Lines changed: 29 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -39,8 +39,10 @@
3939
import org.eclipse.dataplane.port.DataPlaneRegistrationApiController;
4040
import org.eclipse.dataplane.port.DataPlaneSignalingApiController;
4141
import org.eclipse.dataplane.port.exception.AuthorizationNotSupported;
42+
import org.eclipse.dataplane.port.exception.ControlPlaneNotRegistered;
4243
import org.eclipse.dataplane.port.exception.DataFlowNotifyControlPlaneFailed;
4344
import org.eclipse.dataplane.port.exception.DataplaneNotRegistered;
45+
import org.eclipse.dataplane.port.exception.ResourceNotFoundException;
4446
import org.eclipse.dataplane.port.store.ControlPlaneStore;
4547
import org.eclipse.dataplane.port.store.DataFlowStore;
4648
import org.eclipse.dataplane.port.store.InMemoryControlPlaneStore;
@@ -57,6 +59,7 @@
5759
import java.util.UUID;
5860

5961
import static com.fasterxml.jackson.databind.DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES;
62+
import static jakarta.ws.rs.core.HttpHeaders.AUTHORIZATION;
6063
import static java.util.Collections.emptyMap;
6164

6265
public class Dataplane {
@@ -84,7 +87,7 @@ public static Builder newInstance() {
8487
}
8588

8689
public DataPlaneSignalingApiController controller() {
87-
return new DataPlaneSignalingApiController(this);
90+
return new DataPlaneSignalingApiController(this, authorizations);
8891
}
8992

9093
public DataPlaneRegistrationApiController registrationController() {
@@ -104,7 +107,14 @@ public Result<DataFlowStatusResponseMessage> status(String dataFlowId) {
104107
.map(f -> new DataFlowStatusResponseMessage(f.getId(), f.getState().name()));
105108
}
106109

107-
public Result<DataFlowResponseMessage> prepare(DataFlowPrepareMessage message) {
110+
private Result<Void> checkControlPlane(String controlplaneId) {
111+
if (controlPlaneStore.exists(controlplaneId)) {
112+
return Result.success();
113+
}
114+
return Result.failure(new ControlPlaneNotRegistered(controlplaneId));
115+
}
116+
117+
public Result<DataFlowResponseMessage> prepare(String controlplaneId, DataFlowPrepareMessage message) {
108118
var initialDataFlow = DataFlow.newInstance()
109119
.id(message.processId())
110120
.state(DataFlow.State.INITIATING)
@@ -117,9 +127,11 @@ public Result<DataFlowResponseMessage> prepare(DataFlowPrepareMessage message) {
117127
.participantId(message.participantId())
118128
.counterPartyId(message.counterPartyId())
119129
.dataspaceContext(message.dataspaceContext())
130+
.controlplaneId(controlplaneId)
120131
.build();
121132

122-
return onPrepare.action(initialDataFlow)
133+
return checkControlPlane(controlplaneId)
134+
.compose(v -> onPrepare.action(initialDataFlow))
123135
.compose(dataFlow -> {
124136
if (dataFlow.isInitiating()) {
125137
dataFlow.transitionToPrepared();
@@ -137,7 +149,7 @@ public Result<DataFlowResponseMessage> prepare(DataFlowPrepareMessage message) {
137149
}
138150

139151

140-
public Result<DataFlowResponseMessage> start(DataFlowStartMessage message) {
152+
public Result<DataFlowResponseMessage> start(String controlplaneId, DataFlowStartMessage message) {
141153
var initialDataFlow = DataFlow.newInstance()
142154
.id(message.processId())
143155
.state(DataFlow.State.INITIATING)
@@ -149,9 +161,11 @@ public Result<DataFlowResponseMessage> start(DataFlowStartMessage message) {
149161
.participantId(message.participantId())
150162
.counterPartyId(message.counterPartyId())
151163
.dataspaceContext(message.dataspaceContext())
164+
.controlplaneId(controlplaneId)
152165
.build();
153166

154-
return onStart.action(initialDataFlow)
167+
return checkControlPlane(controlplaneId)
168+
.compose(v -> onStart.action(initialDataFlow))
155169
.compose(dataFlow -> {
156170
if (dataFlow.isInitiating()) {
157171
dataFlow.transitionToStarted();
@@ -310,14 +324,16 @@ private Result<Void> notifyControlPlane(String action, DataFlow dataFlow, Object
310324
.header("content-type", "application/json")
311325
.POST(HttpRequest.BodyPublishers.ofString(body));
312326

313-
var controlPlane = controlPlaneStore.findByEndpoint(dataFlow.getCallbackAddress());
314-
if (controlPlane.succeeded()) {
315-
var authorizationProfile = controlPlane.getContent().authorization();
316-
if (authorizationProfile != null) {
317-
var authorization = authorizations.get(authorizationProfile.getType());
318-
authorization.apply(requestBuilder, authorizationProfile);
319-
}
320-
}
327+
controlPlaneStore.findById(dataFlow.getControlplaneId())
328+
.compose(controlPlane -> {
329+
var authorizationProfile = controlPlane.authorization();
330+
if (authorizationProfile != null) {
331+
var authorization = authorizations.get(authorizationProfile.getType());
332+
return authorization.authorizationHeader(authorizationProfile);
333+
}
334+
return Result.failure(new ResourceNotFoundException("ControlPlane has no authorization"));
335+
})
336+
.onSuccess(authorizationHeader -> requestBuilder.header(AUTHORIZATION, authorizationHeader));
321337

322338
return httpClient.send(requestBuilder.build(), HttpResponse.BodyHandlers.discarding());
323339
})

‎src/main/java/org/eclipse/dataplane/domain/Result.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
package org.eclipse.dataplane.domain;
1616

1717
import java.util.NoSuchElementException;
18+
import java.util.function.Consumer;
1819
import java.util.function.Function;
1920

2021
public abstract class Result<C> {
@@ -52,6 +53,10 @@ public static <R> Result<R> attempt(ExceptionThrowingSupplier<R> resultSupplier)
5253

5354
public abstract <T> Result<T> compose(ExceptionThrowingFunction<C, Result<T>> transformValue);
5455

56+
public abstract Result<C> onSuccess(Consumer<C> onSuccessDo);
57+
58+
public abstract Result<C> onFailure(Consumer<Exception> onFailureDo);
59+
5560
public boolean succeeded() {
5661
return this instanceof Result.Success<C>;
5762
}
@@ -101,6 +106,17 @@ public <T> Result<T> compose(ExceptionThrowingFunction<C, Result<T>> transformVa
101106
return Result.failure(e);
102107
}
103108
}
109+
110+
@Override
111+
public Result<C> onSuccess(Consumer<C> onSuccessDo) {
112+
onSuccessDo.accept(content);
113+
return this;
114+
}
115+
116+
@Override
117+
public Result<C> onFailure(Consumer<Exception> onFailureDo) {
118+
return this;
119+
}
104120
}
105121

106122
private static class Failure<C> extends Result<C> {
@@ -140,6 +156,17 @@ public <T> Result<T> map(ExceptionThrowingFunction<C, T> transformValue) {
140156
public <T> Result<T> compose(ExceptionThrowingFunction<C, Result<T>> transformValue) {
141157
return Result.failure(this.exception);
142158
}
159+
160+
@Override
161+
public Result<C> onSuccess(Consumer<C> onSuccessDo) {
162+
return this;
163+
}
164+
165+
@Override
166+
public Result<C> onFailure(Consumer<Exception> onFailureDo) {
167+
onFailureDo.accept(exception);
168+
return this;
169+
}
143170
}
144171

145172
@FunctionalInterface
@@ -151,4 +178,5 @@ public interface ExceptionThrowingFunction<T, R> {
151178
public interface ExceptionThrowingSupplier<T> {
152179
T get() throws Exception;
153180
}
181+
154182
}

‎src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlow.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ public class DataFlow {
3939
private List<String> labels;
4040
private Map<String, Object> metadata;
4141
private DataAddress dataAddress;
42+
private String controlplaneId;
4243

4344
public static DataFlow.Builder newInstance() {
4445
return new Builder();
@@ -158,6 +159,10 @@ public URI callbackEndpointFor(String action) {
158159
return URI.create(getCallbackAddress() + "/transfers/" + getId() + "/dataflow/" + action);
159160
}
160161

162+
public String getControlplaneId() {
163+
return controlplaneId;
164+
}
165+
161166
public static class Builder {
162167
private final DataFlow dataFlow = new DataFlow();
163168

@@ -234,6 +239,11 @@ public Builder metadata(Map<String, Object> metadata) {
234239
dataFlow.metadata = metadata;
235240
return this;
236241
}
242+
243+
public Builder controlplaneId(String controlplaneId) {
244+
dataFlow.controlplaneId = controlplaneId;
245+
return this;
246+
}
237247
}
238248

239249
public enum State {

‎src/main/java/org/eclipse/dataplane/domain/registration/Authorization.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414

1515
package org.eclipse.dataplane.domain.registration;
1616

17-
import java.net.http.HttpRequest;
17+
import org.eclipse.dataplane.domain.Result;
1818

1919
/**
2020
* Defines structure for an authorization profile.
@@ -30,6 +30,7 @@ public interface Authorization {
3030
* Function that applies the authorization profile to the request builder.
3131
* e.g. the Authorization header could be added with proper content.
3232
*/
33-
HttpRequest.Builder apply(HttpRequest.Builder requestBuilder, AuthorizationProfile profile);
33+
Result<String> authorizationHeader(AuthorizationProfile profile);
3434

35+
Result<String> extractCallerId(String authorizationHeader);
3536
}

‎src/main/java/org/eclipse/dataplane/domain/registration/Oauth2ClientCredentialsAuthorization.java‎

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,8 @@
1515
package org.eclipse.dataplane.domain.registration;
1616

1717
import com.fasterxml.jackson.databind.ObjectMapper;
18+
import com.nimbusds.jwt.SignedJWT;
19+
import org.eclipse.dataplane.domain.Result;
1820

1921
import java.net.URI;
2022
import java.net.URLEncoder;
@@ -25,7 +27,6 @@
2527
import java.util.Map;
2628
import java.util.stream.Collectors;
2729

28-
import static jakarta.ws.rs.core.HttpHeaders.AUTHORIZATION;
2930
import static jakarta.ws.rs.core.MediaType.APPLICATION_FORM_URLENCODED;
3031

3132
public class Oauth2ClientCredentialsAuthorization implements Authorization {
@@ -39,7 +40,7 @@ public String type() {
3940
}
4041

4142
@Override
42-
public HttpRequest.Builder apply(HttpRequest.Builder requestBuilder, AuthorizationProfile profile) {
43+
public Result<String> authorizationHeader(AuthorizationProfile profile) {
4344
var tokenEndpoint = profile.stringAttribute("tokenEndpoint");
4445

4546
var parameters = Map.of(
@@ -63,10 +64,25 @@ public HttpRequest.Builder apply(HttpRequest.Builder requestBuilder, Authorizati
6364
var response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
6465
var body = response.body();
6566
var accessToken = objectMapper.readValue(body, Map.class).get("access_token").toString();
66-
return requestBuilder.header(AUTHORIZATION, "Bearer " + accessToken);
67+
return Result.success("Bearer " + accessToken);
6768
} catch (Exception e) {
68-
throw new RuntimeException(e);
69+
return Result.failure(e);
6970
}
7071

7172
}
73+
74+
@Override
75+
public Result<String> extractCallerId(String authorizationHeader) {
76+
try {
77+
var token = authorizationHeader.substring("Bearer ".length());
78+
var jwt = SignedJWT.parse(token);
79+
var sub = jwt.getJWTClaimsSet().getClaims().get("sub");
80+
if (sub instanceof String callerId) {
81+
return Result.success(callerId);
82+
}
83+
return Result.failure(new RuntimeException("JWT sub claim %s is not a string".formatted(sub)));
84+
} catch (Exception e) {
85+
return Result.failure(e);
86+
}
87+
}
7288
}

‎src/main/java/org/eclipse/dataplane/port/DataPlaneSignalingApiController.java‎

Lines changed: 31 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,19 +16,26 @@
1616

1717
import jakarta.ws.rs.Consumes;
1818
import jakarta.ws.rs.GET;
19+
import jakarta.ws.rs.NotAuthorizedException;
1920
import jakarta.ws.rs.POST;
2021
import jakarta.ws.rs.Path;
2122
import jakarta.ws.rs.PathParam;
2223
import jakarta.ws.rs.Produces;
24+
import jakarta.ws.rs.container.ContainerRequestContext;
25+
import jakarta.ws.rs.core.Context;
2326
import jakarta.ws.rs.core.Response;
2427
import org.eclipse.dataplane.Dataplane;
28+
import org.eclipse.dataplane.domain.Result;
2529
import org.eclipse.dataplane.domain.dataflow.DataFlow;
2630
import org.eclipse.dataplane.domain.dataflow.DataFlowPrepareMessage;
2731
import org.eclipse.dataplane.domain.dataflow.DataFlowStartMessage;
2832
import org.eclipse.dataplane.domain.dataflow.DataFlowStartedNotificationMessage;
2933
import org.eclipse.dataplane.domain.dataflow.DataFlowStatusResponseMessage;
3034
import org.eclipse.dataplane.domain.dataflow.DataFlowSuspendMessage;
3135
import org.eclipse.dataplane.domain.dataflow.DataFlowTerminateMessage;
36+
import org.eclipse.dataplane.domain.registration.Authorization;
37+
38+
import java.util.Map;
3239

3340
import static jakarta.ws.rs.core.MediaType.APPLICATION_JSON;
3441
import static jakarta.ws.rs.core.MediaType.WILDCARD;
@@ -39,15 +46,20 @@
3946
public class DataPlaneSignalingApiController {
4047

4148
private final Dataplane dataplane;
49+
private final Map<String, Authorization> authorizations;
4250

43-
public DataPlaneSignalingApiController(Dataplane dataplane) {
51+
public DataPlaneSignalingApiController(Dataplane dataplane, Map<String, Authorization> authorizations) {
4452
this.dataplane = dataplane;
53+
this.authorizations = authorizations;
4554
}
4655

4756
@POST
4857
@Path("/prepare")
49-
public Response prepare(DataFlowPrepareMessage message) {
50-
var response = dataplane.prepare(message).orElseThrow(ExceptionMapper.MAP_TO_WSRS);
58+
public Response prepare(DataFlowPrepareMessage message, @Context ContainerRequestContext requestContext) {
59+
var response = extractControlplaneId(requestContext)
60+
.compose(controlplaneId -> dataplane.prepare(controlplaneId, message))
61+
.orElseThrow(ExceptionMapper.MAP_TO_WSRS);
62+
5163
if (response.state().equals(DataFlow.State.PREPARING.name())) {
5264
return Response.accepted(response).build();
5365
}
@@ -56,8 +68,11 @@ public Response prepare(DataFlowPrepareMessage message) {
5668

5769
@POST
5870
@Path("/start")
59-
public Response start(DataFlowStartMessage message) {
60-
var response = dataplane.start(message).orElseThrow(ExceptionMapper.MAP_TO_WSRS);
71+
public Response start(DataFlowStartMessage message, @Context ContainerRequestContext requestContext) {
72+
var response = extractControlplaneId(requestContext)
73+
.compose(controlplaneId -> dataplane.start(controlplaneId, message))
74+
.orElseThrow(ExceptionMapper.MAP_TO_WSRS);
75+
6176
if (response.state().equals(DataFlow.State.STARTING.name())) {
6277
return Response.accepted(response).build();
6378
}
@@ -99,4 +114,15 @@ public DataFlowStatusResponseMessage status(@PathParam("flowId") String flowId)
99114
return dataplane.status(flowId).orElseThrow(ExceptionMapper.MAP_TO_WSRS);
100115
}
101116

117+
private Result<String> extractControlplaneId(ContainerRequestContext requestContext) {
118+
var authorizationHeader = requestContext.getHeaderString("Authorization");
119+
if (authorizationHeader == null) {
120+
return Result.failure(new NotAuthorizedException("Authorization header missing"));
121+
}
122+
return authorizations.values().stream()
123+
.map(authorization -> authorization.extractCallerId(authorizationHeader))
124+
.filter(Result::succeeded).findFirst()
125+
.orElseGet(() -> Result.failure(new NotAuthorizedException("Authorization method not recognized")));
126+
}
127+
102128
}

‎src/main/java/org/eclipse/dataplane/port/ExceptionMapper.java‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,22 +15,32 @@
1515
package org.eclipse.dataplane.port;
1616

1717
import jakarta.ws.rs.BadRequestException;
18+
import jakarta.ws.rs.NotAuthorizedException;
1819
import jakarta.ws.rs.NotFoundException;
1920
import jakarta.ws.rs.WebApplicationException;
2021
import org.eclipse.dataplane.port.exception.AuthorizationNotSupported;
22+
import org.eclipse.dataplane.port.exception.ControlPlaneNotRegistered;
2123
import org.eclipse.dataplane.port.exception.ResourceNotFoundException;
2224

2325
import java.util.function.Function;
2426

2527
public interface ExceptionMapper {
2628

2729
Function<Exception, WebApplicationException> MAP_TO_WSRS = exception -> {
30+
if (exception instanceof WebApplicationException webApplicationException) {
31+
return webApplicationException;
32+
}
33+
2834
if (exception instanceof ResourceNotFoundException notFound) {
2935
return new NotFoundException(notFound);
3036
}
3137

32-
if (exception instanceof AuthorizationNotSupported authorizationNotSupported) {
33-
return new BadRequestException(authorizationNotSupported);
38+
if (exception instanceof ControlPlaneNotRegistered controlPlaneNotRegistered) {
39+
return new NotAuthorizedException(controlPlaneNotRegistered);
40+
}
41+
42+
if (exception instanceof AuthorizationNotSupported) {
43+
return new BadRequestException(exception);
3444
}
3545

3646
return new WebApplicationException("unexpected internal server error");

0 commit comments

Comments
 (0)