bug6059: forwarding of messages to web socket clients tested successfully

2026-09-22 21:55:39 +00:00 · 2024-12-21 19:09:19 +01:00
parent 8840cf9f43
commit 1377a13994
9 changed files with 152 additions and 15 deletions
@@ -31,7 +31,6 @@ import com.sap.sse.common.TimePoint;
 import com.sap.sse.common.TimeRange;
 import com.sap.sse.security.SecurityService;
 import com.sap.sse.security.shared.HasPermissions.DefaultActions;
-import com.sap.sse.security.shared.TypeRelativeObjectIdentifier;

@Path(RestApiApplication.API + RestApiApplication.V1 + RiotDataAccessWindowsResource.DATA_ACCESS_WINDOWS)
 public class RiotDataAccessWindowsResource extends AbstractRiotServerResource {
@@ -50,7 +49,7 @@ public class RiotDataAccessWindowsResource extends AbstractRiotServerResource {
        result.put("data_access_windows", dawsJson);
        for (final DataAccessWindow daw : riot.getDataAccessWindows()) {
            if (SecurityUtils.getSubject().isPermitted(daw.getIdentifier().getStringPermission(DefaultActions.READ))
-                    && serialNumbers.contains(daw.getDeviceSerialNumber())
+                    && (serialNumbers.isEmpty() || serialNumbers.contains(daw.getDeviceSerialNumber()))
                    && TimeRange.create(startTime == null ? null : TimePoint.of(Long.valueOf(startTime)),
                                        endTime == null ? null : TimePoint.of(Long.valueOf(endTime))).intersects(daw.getTimeRange())) {
                final JSONObject dataAccessWindowJson = new JSONObject();
@@ -70,7 +69,7 @@ public class RiotDataAccessWindowsResource extends AbstractRiotServerResource {
        final JSONObject dawJson = (JSONObject) new JSONParser().parse(new InputStreamReader(inputStream));
        final DataAccessWindow daw = new DataAccessWindowDeserializer().createDataAccessWindowFromJson(dawJson);
        securityService.setOwnershipCheckPermissionForObjectCreationAndRevertOnError(SecuredDomainType.IGTIMI_DATA_ACCESS_WINDOW,
-                new TypeRelativeObjectIdentifier(Long.toString(daw.getId())), daw.getName(),
+                daw.getIdentifier().getTypeRelativeObjectIdentifier(), daw.getName(),
                ()->riot.addDataAccessWindow(daw));
        return Response.ok().build();
    }
@@ -19,13 +19,11 @@ import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
 import org.json.simple.parser.ParseException;

-import com.sap.sailing.domain.common.security.SecuredDomainType;
 import com.sap.sailing.domain.igtimiadapter.Device;
 import com.sap.sailing.domain.igtimiadapter.impl.DeviceDeserializer;
 import com.sap.sailing.domain.igtimiadapter.impl.DeviceSerializer;
 import com.sap.sailing.domain.igtimiadapter.server.riot.RiotServer;
 import com.sap.sse.security.SecurityService;
-import com.sap.sse.security.shared.TypeRelativeObjectIdentifier;

@Path(RestApiApplication.API + RestApiApplication.V1 + RiotDevicesResource.DEVICES)
 public class RiotDevicesResource extends AbstractRiotServerResource {
@@ -55,8 +53,9 @@ public class RiotDevicesResource extends AbstractRiotServerResource {
        final InputStream inputStream = request.getInputStream();
        final JSONObject deviceJson = (JSONObject) new JSONParser().parse(new InputStreamReader(inputStream));
        final Device device = new DeviceDeserializer().createDeviceFromJson(deviceJson);
-        securityService.setOwnershipCheckPermissionForObjectCreationAndRevertOnError(SecuredDomainType.IGTIMI_DEVICE, new TypeRelativeObjectIdentifier(device.getSerialNumber()), device.getName(),
-                ()->riot.addDevice(device));
+        securityService.setOwnershipCheckPermissionForObjectCreationAndRevertOnError(
+                device.getPermissionType(), device.getIdentifier().getTypeRelativeObjectIdentifier(),
+                device.getName(), ()->riot.addDevice(device));
        return Response.ok().build();
    }
 }
@@ -48,7 +48,7 @@ public class RiotWebsocketHandlerImpl implements RiotWebsocketHandler {

    @OnWebSocketConnect
    public void onConnect(Session session) {
-        logger.info("Connect with session "+session);
+        logger.info("Connect with session from "+session.getRemote());
        this.session = session;
        final UpgradeRequest upgradeRequest = session.getUpgradeRequest();
        final HttpServletRequest requestWrapper = new UpgradeRequestAsHttpServletRequestWrapper(upgradeRequest);
@@ -69,7 +69,8 @@ public class RiotWebsocketHandlerImpl implements RiotWebsocketHandler {
        final Subject subject = SecurityUtils.getSubject();
        logger.info("Subject: "+subject+" with principal "+subject.getPrincipal());
        final Activator activator = Activator.getInstance();
-        authenticatedUser = activator.getSecurityService().getCurrentUser();
+        authenticatedUser = activator.getSecurityService().getCurrentUser() == null ?
+                activator.getSecurityService().getAllUser() : activator.getSecurityService().getCurrentUser();
        activator.getRiotServer().addWebSocketClient(this);
    }

@@ -136,4 +137,12 @@ public class RiotWebsocketHandlerImpl implements RiotWebsocketHandler {
    public void flush() throws IOException {
        session.getRemote().flush();
    }
+    
+    @Override
+    public String toString() {
+        return "Riot web socket for session from "+
+                (session==null?"unknown target ":session.getRemote())+
+                ", subcribed to devices "+getDeviceSerialNumbers()+
+                " and with authenticated user "+getAuthenticatedUser();
+    }
 }
@@ -22,7 +22,8 @@ Import-Package: com.google.gwt.user.client.rpc.core.com.sap.sse.security.shared,
 com.sap.sse.security.shared,
 com.sap.sse.shared.util,
 com.sap.sse.util,
- org.osgi.framework;version="1.3.0"
+ org.osgi.framework;version="1.3.0",
+ org.osgi.util.tracker;version="[1.5.0,2.0.0)"
 Require-Bundle: com.sap.sailing.domain.igtimiadapter,
 com.sap.sailing.domain.igtimiadapter.persistence,
 com.sap.sse.security.common
@@ -37,7 +37,14 @@ public class Activator implements BundleActivator {
    private static final String RIOT_PORT_DEFAULT = "6000";

    private static BundleContext context;
+
+    private static Activator instance;
    private RiotServer riotServer;
+    private FullyInitializedReplicableTracker<SecurityService> securityServiceServiceTracker;
+    
+    public Activator() {
+        instance = this;
+    }

    static BundleContext getContext() {
        return context;
@@ -49,6 +56,7 @@ public class Activator implements BundleActivator {
        final int riotPort = Integer.valueOf(riotPortAsString);
        final DomainObjectFactory domainObjectFactory = PersistenceFactory.INSTANCE.getDefaultDomainObjectFactory();
        final MongoObjectFactory mongoObjectFactory = PersistenceFactory.INSTANCE.getDefaultMongoObjectFactory();
+        // TODO should we really start the RiotServer if no port is specified? We could interpret no specified port as a hint to not start the server
        final RiotServerImpl riotServerImpl = new RiotServerImpl(riotPort, domainObjectFactory, mongoObjectFactory);
        riotServer = riotServerImpl;
        context.registerService(RiotServer.class, riotServer, null);
@@ -62,7 +70,7 @@ public class Activator implements BundleActivator {
            }
        }, null);
        new Thread(() -> {
-            final FullyInitializedReplicableTracker<SecurityService> securityServiceServiceTracker = FullyInitializedReplicableTracker.createAndOpen(context, SecurityService.class);
+            securityServiceServiceTracker = FullyInitializedReplicableTracker.createAndOpen(context, SecurityService.class);
            try {
                final SecurityService securityService = securityServiceServiceTracker.getInitializedService(0);
                for (Resource resource : riotServer.getResources()) {
@@ -89,4 +97,16 @@ public class Activator implements BundleActivator {
    public void stop(BundleContext bundleContext) throws Exception {
        Activator.context = null;
    }
+    
+    public static Activator getInstance() {
+        return instance;
+    }
+    
+    public RiotServer getRiotServer() {
+        return riotServer;
+    }
+    
+    public SecurityService getSecurityService() {
+        return securityServiceServiceTracker.getService();
+    }
 }
@@ -20,6 +20,9 @@ import com.sap.sse.security.shared.impl.User;
 *
 */
 public interface RiotWebsocketHandler {
+    /**
+     * Never {@code null}; may be the "<all>" user representing anonymous users
+     */
    User getAuthenticatedUser();
    
    Set<String> getDeviceSerialNumbers();
@@ -90,6 +90,8 @@ public interface RiotServer extends Replicable<ReplicableRiotServer, RiotReplica
    Iterable<Device> getDevices();
    
    Device getDeviceById(long id);
+
+    Device getDeviceBySerialNumber(String deviceSerialNumber);
    
    void addDevice(Device device);
    
@@ -148,7 +148,7 @@ public class RiotConnectionImpl implements RiotConnection {
                    nextMessageLength = 0;
                    try {
                        messageBuffer.flip();
-                        final Msg message = Msg.parseFrom(messageBuffer, protobufExtensionRegistry);       
+                        final Msg message = Msg.parseFrom(messageBuffer, protobufExtensionRegistry);
                        processMessage(message); // extract device serial number and device group token and send auth response for auth request
                        riotServer.notifyListeners(message, serialNumber);
                        storeMessage(message);
@@ -24,12 +24,19 @@ import java.util.concurrent.ConcurrentMap;
 import java.util.logging.Level;
 import java.util.logging.Logger;

+import com.igtimi.IgtimiData.DataMsg;
+import com.igtimi.IgtimiData.DataPoint;
+import com.igtimi.IgtimiDevice.DeviceManagement;
+import com.igtimi.IgtimiDevice.DeviceManagementResponse;
 import com.igtimi.IgtimiStream.Msg;
 import com.sap.sailing.domain.igtimiadapter.DataAccessWindow;
+import com.sap.sailing.domain.igtimiadapter.DataPointTimePointExtractor;
+import com.sap.sailing.domain.igtimiadapter.DataPointVisitor;
 import com.sap.sailing.domain.igtimiadapter.Device;
 import com.sap.sailing.domain.igtimiadapter.Resource;
 import com.sap.sailing.domain.igtimiadapter.persistence.DomainObjectFactory;
 import com.sap.sailing.domain.igtimiadapter.persistence.MongoObjectFactory;
+import com.sap.sailing.domain.igtimiadapter.server.Activator;
 import com.sap.sailing.domain.igtimiadapter.server.RiotWebsocketHandler;
 import com.sap.sailing.domain.igtimiadapter.server.replication.ReplicableRiotServer;
 import com.sap.sailing.domain.igtimiadapter.server.replication.RiotReplicationOperation;
@@ -37,10 +44,16 @@ import com.sap.sailing.domain.igtimiadapter.server.riot.RiotConnection;
 import com.sap.sailing.domain.igtimiadapter.server.riot.RiotMessageListener;
 import com.sap.sailing.domain.igtimiadapter.server.riot.RiotServer;
 import com.sap.sse.common.Duration;
+import com.sap.sse.common.TimePoint;
 import com.sap.sse.common.TimeRange;
 import com.sap.sse.common.Util;
 import com.sap.sse.replication.interfaces.impl.AbstractReplicableWithObjectInputStream;
+import com.sap.sse.security.SecurityService;
+import com.sap.sse.security.shared.AccessControlListAnnotation;
 import com.sap.sse.security.shared.HasPermissions.DefaultActions;
+import com.sap.sse.security.shared.OwnershipAnnotation;
+import com.sap.sse.security.shared.PermissionChecker;
+import com.sap.sse.security.shared.impl.User;
 import com.sap.sse.shared.util.Wait;
 import com.sap.sse.util.ObjectInputStreamResolvingAgainstCache;

@@ -50,6 +63,7 @@ public class RiotServerImpl extends AbstractReplicableWithObjectInputStream<Repl
    private final ConcurrentMap<Long, Resource> resources;
    private final ConcurrentMap<Long, DataAccessWindow> dataAccessWindows;
    private final ConcurrentMap<Long, Device> devices;
+    private final ConcurrentMap<String, Device> devicesBySerialNumber;
    private final Set<RiotMessageListener> listeners;
    private final Selector socketSelector;
    private final ServerSocketChannel serverSocketChannel;
@@ -88,12 +102,14 @@ public class RiotServerImpl extends AbstractReplicableWithObjectInputStream<Repl
        this.resources = new ConcurrentHashMap<>();
        this.dataAccessWindows = new ConcurrentHashMap<>();
        this.devices = new ConcurrentHashMap<>();
+        this.devicesBySerialNumber = new ConcurrentHashMap<>();
        this.connections = new ConcurrentHashMap<>();
        this.domainObjectFactory = domainObjectFactory;
        this.mongoObjectFactory = mongoObjectFactory;
-        this.liveWebSocketConnections = new HashSet<>();
+        this.liveWebSocketConnections = ConcurrentHashMap.newKeySet();
        for (final Device device : domainObjectFactory.getDevices()) {
            devices.put(device.getId(), device);
+            devicesBySerialNumber.put(device.getSerialNumber(), device);
        }
        for (final Resource resource : domainObjectFactory.getResources()) {
            resources.put(resource.getId(), resource);
@@ -201,10 +217,87 @@ public class RiotServerImpl extends AbstractReplicableWithObjectInputStream<Repl
     *            the serial number of the device from which the message originated
     */
    void notifyListeners(Msg message, String deviceSerialNumber) {
+        if (deviceSerialNumber != null) {
+            forwardMessageToEligibleWebSocketClients(message, deviceSerialNumber);
+        }
        for (final RiotMessageListener listener : listeners) {
            listener.onMessage(message);
        }
-        // TODO forward to live web socket connections
+    }
+
+    /**
+     * @param deviceSerialNumber must not be {@code null}
+     */
+    private void forwardMessageToEligibleWebSocketClients(Msg message, String deviceSerialNumber) {
+        final Device device = getDeviceBySerialNumber(deviceSerialNumber);
+        if (device == null) {
+            logger.info("Received message from unknown devices "+deviceSerialNumber);
+            // TODO should we use this as the trigger to *create* the device?
+        } else {
+            final byte[] messageAsBytes = message.toByteArray();
+            final SecurityService securityService = Activator.getInstance().getSecurityService();
+            final TimeRange messageDataTimeRange = getTimeRange(message);
+            final Iterable<DataAccessWindow> daws = getDataAccessWindows(Collections.singleton(deviceSerialNumber), messageDataTimeRange);
+            for (final RiotWebsocketHandler webSocketClient : liveWebSocketConnections) {
+                final User user = webSocketClient.getAuthenticatedUser();
+                final OwnershipAnnotation deviceOwnership = securityService.getOwnership(device.getIdentifier());
+                final AccessControlListAnnotation deviceAccessControlList = securityService.getAccessControlList(device.getIdentifier());
+                if (!Util.isEmpty(Util.filter(daws, daw->{
+                    final OwnershipAnnotation dawOownership = securityService.getOwnership(daw.getIdentifier());
+                    final AccessControlListAnnotation dawAccessControlList = securityService.getAccessControlList(daw.getIdentifier());
+                    return PermissionChecker.isPermitted(
+                            daw.getIdentifier().getPermission(DefaultActions.READ),
+                            user, securityService.getAllUser(),
+                            dawOownership==null?null:dawOownership.getAnnotation(),
+                            dawAccessControlList==null?null:dawAccessControlList.getAnnotation());
+                }))
+                && PermissionChecker.isPermitted(
+                        device.getIdentifier().getPermission(DefaultActions.READ),
+                        user, securityService.getAllUser(),
+                        deviceOwnership==null?null:deviceOwnership.getAnnotation(),
+                        deviceAccessControlList==null?null:deviceAccessControlList.getAnnotation()) ) {
+                    webSocketClient.sendBytesByFuture(ByteBuffer.wrap(messageAsBytes));
+                }
+            }
+        }
+    }
+
+    /**
+     * Returns a time range spanning the time point of the first and the last of the {@link DataPoint}s inside
+     * 
+     * @return {@code null} in case there is nothing inside the message that specifies a time point, otherwise a time
+     *         range that includes the time points of all elements from the message that have a time stamp and that
+     *         cannot be made any smaller while still containing all those elements.
+     */
+    private TimeRange getTimeRange(Msg message) {
+        TimePoint from = null;
+        TimePoint to = null;
+        if (message.hasData()) {
+            for (final DataMsg data : message.getData().getDataList()) {
+                for (final DataPoint dataPoint : data.getDataList()) {
+                    final TimePoint dataPointTimePoint = DataPointVisitor.accept(dataPoint, new DataPointTimePointExtractor());
+                    if (from==null || dataPointTimePoint.before(from)) {
+                        from = dataPointTimePoint;
+                    }
+                    if (to==null || dataPointTimePoint.after(to)) {
+                        to = dataPointTimePoint;
+                    }
+                }
+            }
+        } else if (message.hasDeviceManagement()) {
+            final DeviceManagement deviceManagement = message.getDeviceManagement();
+            if (deviceManagement.hasResponse()) {
+                final DeviceManagementResponse response = deviceManagement.getResponse();
+                final TimePoint timePoint = TimePoint.of(response.getTimestamp());
+                if (from==null || timePoint.before(from)) {
+                    from = timePoint;
+                }
+                if (to==null || timePoint.after(to)) {
+                    to = timePoint;
+                }
+            }
+        }
+        return TimeRange.create(from, to==null?null:to.plus(Duration.ONE_MILLISECOND));
    }

    @Override
@@ -216,6 +309,11 @@ public class RiotServerImpl extends AbstractReplicableWithObjectInputStream<Repl
    public Device getDeviceById(long id) {
        return devices.get(id);
    }
+    
+    @Override
+    public Device getDeviceBySerialNumber(String deviceSerialNumber) {
+        return devicesBySerialNumber.get(deviceSerialNumber);
+    }

    @Override
    public void addDevice(Device device) {
@@ -225,6 +323,7 @@ public class RiotServerImpl extends AbstractReplicableWithObjectInputStream<Repl
    @Override
    public Void internalAddDevice(Device device) {
        devices.put(device.getId(), device);
+        devicesBySerialNumber.put(device.getSerialNumber(), device);
        mongoObjectFactory.storeDevice(device);
        return null;
    }
@@ -236,7 +335,8 @@ public class RiotServerImpl extends AbstractReplicableWithObjectInputStream<Repl
    
    @Override
    public Void internalRemoveDevice(long deviceId) {
-        devices.remove(deviceId);
+        final Device device = devices.remove(deviceId);
+        devicesBySerialNumber.remove(device.getSerialNumber());
        mongoObjectFactory.removeDevice(deviceId);
        return null;
    }
@@ -329,6 +429,7 @@ public class RiotServerImpl extends AbstractReplicableWithObjectInputStream<Repl
     */
    public void clear() {
        devices.clear();
+        devicesBySerialNumber.clear();
        resources.clear();
        dataAccessWindows.clear();
        connections.clear();
@@ -352,6 +453,9 @@ public class RiotServerImpl extends AbstractReplicableWithObjectInputStream<Repl
    public void initiallyFillFromInternal(ObjectInputStream is)
            throws IOException, ClassNotFoundException, InterruptedException {
        devices.putAll((Map<Long, Device>) is.readObject());
+        for (final Device device : devices.values()) {
+            devicesBySerialNumber.put(device.getSerialNumber(), device);
+        }
        resources.putAll((Map<Long, Resource>) is.readObject());
        dataAccessWindows.putAll((Map<Long, DataAccessWindow>) is.readObject());
    }