fix: STREAMP-15651: use stream security for process input/output validation

.github/workflows/build.yml

@@ -5,22 +5,28 @@ on: [push, pull_request]
 jobs:
   build:
     name: Build the project
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - name: Checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v5
         with:
           fetch-depth: 2
+
       - name: Setup JDK
-        uses: actions/setup-java@v1
+        uses: actions/setup-java@v5
         with:
           java-version: 17
+          java-package: jdk
+          distribution: adopt
+          architecture: x64
+
       - name: Cache local Maven repository
-        uses: actions/cache@v2
+        uses: actions/cache@v4
         with:
           path: ~/.m2/repository
           key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}
           restore-keys: |
             ${{ runner.os }}-maven-
+
       - name: mvn build
         run: ./mvnw clean install -Djib.skip -B

.github/workflows/release-patch.yml

@@ -5,17 +5,17 @@ on: [workflow_dispatch]
 jobs:
   build:
     name: Release
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - name: Checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v5
         with:
           fetch-depth: 0
           # We need a personal access token to be able to push to a protected branch
           token: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }}
 
       - name: Setup JDK
-        uses: actions/setup-java@v2
+        uses: actions/setup-java@v5
         with:
           java-version: 17
           java-package: jdk

.github/workflows/release.yml

@@ -15,17 +15,17 @@ on:
 jobs:
   build:
     name: Release
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - name: Checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v5
         with:
           fetch-depth: 0
           # We need a personal access token to be able to push to a protected branch
           token: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }}
 
       - name: Setup JDK
-        uses: actions/setup-java@v2
+        uses: actions/setup-java@v5
         with:
           java-version: 17
           java-package: jdk

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/handlers/HandlerService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2021 Expedia, Inc.
+ * Copyright (C) 2018-2025 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,9 +24,6 @@
 import java.util.Set;
 import java.util.function.Function;
 
-import lombok.Value;
-import lombok.extern.slf4j.Slf4j;
-
 import org.springframework.stereotype.Component;
 
 import com.expediagroup.streamplatform.streamregistry.core.validators.ValidationException;
@@ -47,6 +44,9 @@
 import com.expediagroup.streamplatform.streamregistry.model.StreamBinding;
 import com.expediagroup.streamplatform.streamregistry.model.Zone;
 
+import lombok.Value;
+import lombok.extern.slf4j.Slf4j;
+
 @Component
 @Slf4j
 public class HandlerService {

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/handlers/IdentityHandler.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2021 Expedia, Inc.
+ * Copyright (C) 2018-2025 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -15,12 +15,12 @@
  */
 package com.expediagroup.streamplatform.streamregistry.core.handlers;
 
-import lombok.RequiredArgsConstructor;
-
 import com.expediagroup.streamplatform.streamregistry.handler.Handler;
 import com.expediagroup.streamplatform.streamregistry.model.Entity;
 import com.expediagroup.streamplatform.streamregistry.model.Specification;
 
+import lombok.RequiredArgsConstructor;
+
 @RequiredArgsConstructor
 public class IdentityHandler<T extends Entity> implements Handler<T> {
   public static final String DEFAULT = "default";

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/ConsumerBindingService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2024 Expedia, Inc.
+ * Copyright (C) 2018-2025 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,9 +21,6 @@
 import java.util.Optional;
 import java.util.function.Predicate;
 
-import lombok.RequiredArgsConstructor;
-import lombok.val;
-
 import org.springframework.security.access.prepost.PostAuthorize;
 import org.springframework.security.access.prepost.PostFilter;
 import org.springframework.security.access.prepost.PreAuthorize;
@@ -39,6 +36,9 @@
 import com.expediagroup.streamplatform.streamregistry.model.keys.ConsumerKey;
 import com.expediagroup.streamplatform.streamregistry.repository.ConsumerBindingRepository;
 
+import lombok.RequiredArgsConstructor;
+import lombok.val;
+
 @Component
 @RequiredArgsConstructor
 public class ConsumerBindingService {

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/ConsumerService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2024 Expedia, Inc.
+ * Copyright (C) 2018-2025 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,9 +21,6 @@
 import java.util.Optional;
 import java.util.function.Predicate;
 
-import lombok.RequiredArgsConstructor;
-import lombok.val;
-
 import org.springframework.security.access.prepost.PostAuthorize;
 import org.springframework.security.access.prepost.PostFilter;
 import org.springframework.security.access.prepost.PreAuthorize;
@@ -39,6 +36,9 @@
 import com.expediagroup.streamplatform.streamregistry.model.keys.ConsumerKey;
 import com.expediagroup.streamplatform.streamregistry.repository.ConsumerRepository;
 
+import lombok.RequiredArgsConstructor;
+import lombok.val;
+
 @Component
 @RequiredArgsConstructor
 public class ConsumerService {

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/DomainService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2024 Expedia, Inc.
+ * Copyright (C) 2018-2025 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,9 +21,6 @@
 import java.util.Optional;
 import java.util.function.Predicate;
 
-import lombok.RequiredArgsConstructor;
-import lombok.val;
-
 import org.springframework.security.access.prepost.PostAuthorize;
 import org.springframework.security.access.prepost.PostFilter;
 import org.springframework.security.access.prepost.PreAuthorize;
@@ -43,6 +40,9 @@
 import com.expediagroup.streamplatform.streamregistry.model.keys.DomainKey;
 import com.expediagroup.streamplatform.streamregistry.repository.DomainRepository;
 
+import lombok.RequiredArgsConstructor;
+import lombok.val;
+
 @Component
 @RequiredArgsConstructor
 public class DomainService {

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/InfrastructureService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2024 Expedia, Inc.
+ * Copyright (C) 2018-2025 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,9 +21,6 @@
 import java.util.Optional;
 import java.util.function.Predicate;
 
-import lombok.RequiredArgsConstructor;
-import lombok.val;
-
 import org.springframework.security.access.prepost.PostAuthorize;
 import org.springframework.security.access.prepost.PostFilter;
 import org.springframework.security.access.prepost.PreAuthorize;
@@ -43,6 +40,9 @@
 import com.expediagroup.streamplatform.streamregistry.model.keys.InfrastructureKey;
 import com.expediagroup.streamplatform.streamregistry.repository.InfrastructureRepository;
 
+import lombok.RequiredArgsConstructor;
+import lombok.val;
+
 @Component
 @RequiredArgsConstructor
 public class InfrastructureService {

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/ProcessBindingService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2024 Expedia, Inc.
+ * Copyright (C) 2018-2025 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,9 +21,6 @@
 import java.util.Optional;
 import java.util.function.Predicate;
 
-import lombok.RequiredArgsConstructor;
-import lombok.val;
-
 import org.springframework.security.access.prepost.PostAuthorize;
 import org.springframework.security.access.prepost.PostFilter;
 import org.springframework.security.access.prepost.PreAuthorize;
@@ -38,6 +35,9 @@
 import com.expediagroup.streamplatform.streamregistry.model.keys.ProcessBindingKey;
 import com.expediagroup.streamplatform.streamregistry.repository.ProcessBindingRepository;
 
+import lombok.RequiredArgsConstructor;
+import lombok.val;
+
 @Component
 @RequiredArgsConstructor
 public class ProcessBindingService {

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/ProcessService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2024 Expedia, Inc.
+ * Copyright (C) 2018-2025 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,9 +21,6 @@
 import java.util.Optional;
 import java.util.function.Predicate;
 
-import lombok.RequiredArgsConstructor;
-import lombok.val;
-
 import org.springframework.security.access.prepost.PostAuthorize;
 import org.springframework.security.access.prepost.PostFilter;
 import org.springframework.security.access.prepost.PreAuthorize;
@@ -34,18 +31,18 @@
 import com.expediagroup.streamplatform.streamregistry.core.validators.ValidationException;
 import com.expediagroup.streamplatform.streamregistry.core.views.ProcessBindingView;
 import com.expediagroup.streamplatform.streamregistry.core.views.ProcessView;
-import com.expediagroup.streamplatform.streamregistry.model.Consumer;
+import com.expediagroup.streamplatform.streamregistry.core.views.StreamView;
+import com.expediagroup.streamplatform.streamregistry.model.*;
 import com.expediagroup.streamplatform.streamregistry.model.Process;
-import com.expediagroup.streamplatform.streamregistry.model.ProcessInputStream;
-import com.expediagroup.streamplatform.streamregistry.model.ProcessOutputStream;
-import com.expediagroup.streamplatform.streamregistry.model.Producer;
-import com.expediagroup.streamplatform.streamregistry.model.Status;
 import com.expediagroup.streamplatform.streamregistry.model.keys.ConsumerKey;
 import com.expediagroup.streamplatform.streamregistry.model.keys.ProcessKey;
 import com.expediagroup.streamplatform.streamregistry.model.keys.ProducerKey;
 import com.expediagroup.streamplatform.streamregistry.model.keys.ZoneKey;
 import com.expediagroup.streamplatform.streamregistry.repository.ProcessRepository;
 
+import lombok.RequiredArgsConstructor;
+import lombok.val;
+
 @Component
 @RequiredArgsConstructor
 public class ProcessService {
@@ -57,6 +54,7 @@ public class ProcessService {
   private final ProcessView processView;
   private final ConsumerService consumerService;
   private final ProducerService producerService;
+  private final StreamView streamView;
 
   @PreAuthorize("hasPermission(#process, 'CREATE')")
   public Optional<Process> create(Process process) throws ValidationException {
@@ -78,6 +76,11 @@ public Optional<Process> create(Process process) throws ValidationException {
   }
 
   private Producer buildProducer(Process process, ZoneKey zoneKey, ProcessOutputStream output) {
+    val stream = streamView.get(output.getStream());
+    if (stream.isEmpty()) {
+      throw new ValidationException("Can't authorize output for " + output.getStream() + " because the stream doesn't exist.");
+    }
+
     return new Producer(
       new ProducerKey(
         output.getStream().getDomainKey().getName(),
@@ -86,7 +89,7 @@ private Producer buildProducer(Process process, ZoneKey zoneKey, ProcessOutputSt
         zoneKey.getName(),
         process.getKey().getName()
       ),
-      process.getSpecification(),
+      secureSpecification(process.getSpecification(), stream.get()),
       process.getStatus()
     );
   }
@@ -123,6 +126,11 @@ public Optional<Process> update(Process process) throws ValidationException {
   }
 
   private Consumer buildConsumer(Process process, ZoneKey zoneKey, ProcessInputStream input) {
+    val stream = streamView.get(input.getStream());
+    if (stream.isEmpty()) {
+      throw new ValidationException("Can't authorize input for " + input.getStream() + " because the stream doesn't exist.");
+    }
+
     return new Consumer(
       new ConsumerKey(
         input.getStream().getDomainKey().getName(),
@@ -131,11 +139,22 @@ private Consumer buildConsumer(Process process, ZoneKey zoneKey, ProcessInputStr
         zoneKey.getName(),
         process.getKey().getName()
       ),
-      process.getSpecification(),
+      secureSpecification(process.getSpecification(), stream.get()),
       process.getStatus()
     );
   }
 
+  private Specification secureSpecification(Specification specification, Stream stream) {
+    return new Specification(
+      specification.getDescription(),
+      specification.getTags(),
+      specification.getType(),
+      specification.getConfiguration(),
+      stream.getSpecification().getSecurity(),
+      specification.getFunction()
+    );
+  }
+
   @PreAuthorize("hasPermission(#process, 'UPDATE_STATUS')")
   public Optional<Process> updateStatus(Process process, Status status) {
     process.setStatus(status);

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/ProducerBindingService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2024 Expedia, Inc.
+ * Copyright (C) 2018-2025 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,9 +21,6 @@
 import java.util.Optional;
 import java.util.function.Predicate;
 
-import lombok.RequiredArgsConstructor;
-import lombok.val;
-
 import org.springframework.security.access.prepost.PostAuthorize;
 import org.springframework.security.access.prepost.PostFilter;
 import org.springframework.security.access.prepost.PreAuthorize;
@@ -39,6 +36,9 @@
 import com.expediagroup.streamplatform.streamregistry.model.keys.ProducerKey;
 import com.expediagroup.streamplatform.streamregistry.repository.ProducerBindingRepository;
 
+import lombok.RequiredArgsConstructor;
+import lombok.val;
+
 @Component
 @RequiredArgsConstructor
 public class ProducerBindingService {