gnodet-bot commented on code in PR #28631: URL: https://github.com/apache/camel/pull/28631#discussion_r4400886896
########## components/camel-mtconnect/src/main/java/org/apache/camel/component/mtconnect/MultipartReader.java: ########## @@ -0,0 +1,185 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mtconnect; + +import java.io.ByteArrayOutputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Locale; + +/** + * Reads the parts of an agent's streaming response, {@code multipart/mixed} (2.x agents) or + * {@code multipart/x-mixed-replace} (1.x agents). Part header names are compared without case, because 1.x agents write + * {@code Content-length}. A part with a Content-Length header is complete after that many bytes, so it is delivered + * before the next boundary arrives; a part without one ends at the next boundary. Agents end a stream without a closing + * boundary, so the end of the input between parts is the end of the stream. The reader reads byte by byte, so its input + * must be buffered. + */ +final class MultipartReader { + + private static final int MAX_LINE = 8192; + + private final InputStream in; + private final byte[] delimiter; + private final String boundaryLine; + private boolean atBoundary; + private boolean finished; + + MultipartReader(InputStream in, String boundary) { + this.in = in; + this.boundaryLine = "--" + boundary; + this.delimiter = ("\r\n--" + boundary).getBytes(StandardCharsets.US_ASCII); + } + + /** + * The boundary parameter of a multipart content type, or null when the content type is not multipart. + */ + static String boundary(String contentType) { + if (contentType == null) { + return null; + } + String[] parameters = contentType.split(";"); + if (!parameters[0].trim().toLowerCase(Locale.ROOT).startsWith("multipart/")) { + return null; + } + for (int i = 1; i < parameters.length; i++) { + String parameter = parameters[i].trim(); + int equals = parameter.indexOf('='); + if (equals > 0 && parameter.substring(0, equals).trim().equalsIgnoreCase("boundary")) { + String value = parameter.substring(equals + 1).trim(); + if (value.length() >= 2 && value.startsWith("\"") && value.endsWith("\"")) { + value = value.substring(1, value.length() - 1); + } + return value.isEmpty() ? null : value; + } + } + return null; + } + + /** + * The content of the next part, or null when the stream ended between parts. + * + * @throws IOException when the stream ends inside a part or the framing is invalid + */ + byte[] nextPart() throws IOException { + if (finished) { + return null; + } + if (!atBoundary && !skipToBoundary()) { + finished = true; + return null; + } + atBoundary = false; + long contentLength = -1; + while (true) { + String line = readLine(); + if (line == null) { + throw new EOFException("The stream ended inside the headers of a part"); + } + if (line.isEmpty()) { + break; + } + int colon = line.indexOf(':'); + if (colon > 0 && line.substring(0, colon).trim().equalsIgnoreCase("Content-Length")) { + try { + contentLength = Long.parseLong(line.substring(colon + 1).trim()); + } catch (NumberFormatException e) { + throw new IOException("Invalid part header: " + line, e); + } + } + } + return contentLength >= 0 ? readExactly(contentLength) : readToDelimiter(); + } + + /** Skips lines up to the next boundary line; false at the end of the stream or after the closing boundary. */ + private boolean skipToBoundary() throws IOException { + while (true) { + String line = readLine(); + if (line == null) { + return false; + } + String trimmed = line.stripTrailing(); + if (trimmed.equals(boundaryLine + "--")) { + return false; + } + if (trimmed.equals(boundaryLine)) { + return true; + } + } + } + + private byte[] readExactly(long length) throws IOException { + if (length > Integer.MAX_VALUE - 8) { + throw new IOException("Part too large: " + length + " bytes"); + } + byte[] content = in.readNBytes((int) length); Review Comment: **Large `Content-Length` allocates up to ~2 GB in a single `readNBytes` call.** The guard rejects lengths above `Integer.MAX_VALUE - 8` but anything below that — including a spoofed `Content-Length: 2000000000` from a compromised agent or a MitM on an unencrypted link — triggers a single `in.readNBytes((int) length)` allocation that will exhaust the heap or cause a long GC pause before the read times out. MTConnect agents operate on trusted plant networks, so this is low risk in practice, but a configurable maximum part size (defaulting to e.g. 64 MB) would harden against a misconfigured agent or a rogue proxy. At minimum, document the absence of a size limit in the option descriptions. ########## components/camel-mtconnect/src/main/java/org/apache/camel/component/mtconnect/AgentProtocol.java: ########## @@ -0,0 +1,453 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.mtconnect; + +import java.time.Instant; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.camel.Exchange; +import org.apache.camel.support.DefaultConsumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * The MTConnect protocol against one agent, one step at a time: probe, current, with assets=true the assets, then + * sample, either the long-lived streaming request or one sample request per step. It resynchronises from current when + * the agent's buffer no longer holds the next observation, re-reads the probe when the agent restarts, and reports an + * outage once per data item. Each observation becomes one exchange of the consumer, processed synchronously in sequence + * order. The streaming and the polling consumer drive it; they own only the threading or the scheduling. Not + * thread-safe: one step runs at a time. + */ +final class AgentProtocol { + + private static final Logger LOG = LoggerFactory.getLogger(AgentProtocol.class); + + private enum Phase { + PROBE, + CURRENT, + ASSETS, + SAMPLE + } + + private static final String ASSET_CHANGED = "AssetChanged"; + private static final String ASSET_REMOVED = "AssetRemoved"; + private static final String ASSET_NOT_FOUND = "ASSET_NOT_FOUND"; + private static final String GONE = "gone"; + + private final MTConnectEndpoint endpoint; + private final DefaultConsumer consumer; + private final AgentReader reader; + private final boolean streaming; + + private Phase phase = Phase.PROBE; + private Map<String, DataItem> dataItems = Map.of(); + private final Set<String> known = new LinkedHashSet<>(); + private final Set<String> bad = new HashSet<>(); + private Long instanceId; + private long cursor; + private long count; + private String resync; + private boolean outage; + private int delivered; + // asset tracking: the asset buffer size from the probe, the device names by uuid, the AssetChanged and + // AssetRemoved observations whose asset is still to be read, and the version last sent per asset id + private Long assetBufferSize; + private Map<String, String> deviceNames = Map.of(); + private final Map<String, Observation> pendingAssets = new LinkedHashMap<>(); + private final Map<String, String> sentAssets = new HashMap<>(); + + /** + * @param streaming whether the sample phase reads the streaming request, or one sample request per step + */ + AgentProtocol(MTConnectEndpoint endpoint, DefaultConsumer consumer, AgentReader reader, boolean streaming) { + this.endpoint = endpoint; + this.consumer = consumer; + this.reader = reader; + this.streaming = streaming; + } + + /** + * Performs the next step: probe, current, assets or sample. A failure is reported here, as an outage or to the + * exception handler. + * + * @return true when the next step may follow at once; false after a failure, or when the agent ended the sample + * stream, so that the caller waits before the next step + */ + boolean step() throws InterruptedException { + delivered = 0; + try { + switch (phase) { + case PROBE -> probe(); + case CURRENT -> current(); + case ASSETS -> assets(); + case SAMPLE -> { + if (!readPendingAssets()) { + return true; + } + return streaming ? stream() : poll(); + } + } + return true; + } catch (AgentFailure failure) { + if (consumer.isRunAllowed()) { + onFailure(failure); + } + return false; + } catch (RuntimeException e) { + consumer.getExceptionHandler().handleException("Unexpected failure reading the MTConnect agent " + + endpoint.getAgentUrl(), + e); + return false; + } + } + + /** The number of messages the last step delivered. */ + int delivered() { + return delivered; + } + + private void probe() throws AgentFailure, InterruptedException { + AgentReader.Answer answer = reader.probe(); + if (answer.document() instanceof AgentDocument.Error error) { + throw agentError(error); + } + AgentDocument.Probe probe = answer.expect(AgentDocument.Probe.class); + Long probeInstance = probe.header().instanceId(); + if (instanceId != null && !instanceId.equals(probeInstance)) { + resync = MTConnectConstants.RESYNC_INSTANCE_CHANGED; + } + instanceId = probeInstance; + dataItems = probe.dataItems(); + deviceNames = new HashMap<>(); + dataItems.values().forEach(item -> deviceNames.put(item.deviceUuid(), item.deviceName())); + assetBufferSize = probe.header().assetBufferSize(); + pendingAssets.clear(); + known.clear(); Review Comment: **`sentAssets` is not cleared when the agent restarts.** `probe()` clears `pendingAssets` (line 144) and `known` (line 145) on every re-probe (agent restart or `instanceId` change), but `sentAssets` is left intact. This means that if an agent restarts and issues different asset IDs, the old IDs remain in `sentAssets` indefinitely, and assets with matching `timestamp/removed` keys will be silently suppressed on re-delivery even though they are logically new. The bounded case: if the same asset pool rotates with new timestamps, the map grows by one entry per unique asset id × version — bounded in practice by the MTConnect asset buffer (typically 1024 entries), so this is not a memory leak in normal operation. But the semantic issue (stale asset ids preventing re-delivery after a restart with different content) is real. Fix: clear `sentAssets` in `probe()` alongside `pendingAssets`: ```suggestion pendingAssets.clear(); sentAssets.clear(); known.clear(); ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
