PoC: Replace aws-c-mqtt with coreMQTT via channel handler adapter - #1792
Open
AniruddhaKanhere wants to merge 2 commits into
Open
PoC: Replace aws-c-mqtt with coreMQTT via channel handler adapter#1792AniruddhaKanhere wants to merge 2 commits into
AniruddhaKanhere wants to merge 2 commits into
Conversation
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| private static final Random RANDOM = new Random(); | ||
|
|
||
| private long nativeHandle; | ||
| @Getter |
| private long nativeHandle; | ||
| @Getter | ||
| private final String clientId; | ||
| @Getter |
AniruddhaKanhere
force-pushed
the
feature/coremqtt-poc
branch
from
May 14, 2026 00:17
e6f6cbb to
6fe315e
Compare
MikeDombo
reviewed
May 14, 2026
| CompletableFuture<Integer> nativeFuture = new CompletableFuture<>(); | ||
| CoreMqttNative.unsubscribe(nativeHandle, topic, nativeFuture); | ||
|
|
||
| nativeFuture.whenComplete((rc, error) -> { |
Member
There was a problem hiding this comment.
Recommendation generated by Amazon CodeGuru Reviewer. Leave feedback on this recommendation by replying to the comment or by reacting to the comment using emoji.
This code is using non-async method whenComplete on CompletableFuture (or CompletionStage). Please be aware that task subscribed to CompletableFuture(or CompletionStage) through non-async method may be executed by the thread that completes the current CompletableFuture (or CompletionStage), or the thread that calls this non-async method to add subscription. In other words, the thread that executes the subscription is non-deterministic. If you prefer having fully control over threads, you may consider using the async variants instead. In addition, please make sure tasks subscribed to CompletableFuture (or CompletionStage) are short lived and non-blocking to avoid deadlock. Learn more: https://docs.oracle.com/en/java/javase/18/docs/api/java.base/java/util/concurrent/CompletableFuture.html
CoralClient user: Please consider doing so to align with Coral Java Documentation.Work chained to CoralClient CompletableFuture (or CompletionStage) may be inadvertently scheduled on the internal pool of the library. Since the internal pool has limited number of threads, this may cause the pool to run out of available threads and eventually lead to a deadlock.
| public CompletableFuture<PubAck> publish(Publish publish) { | ||
| CompletableFuture<PubAck> future = new CompletableFuture<>(); | ||
|
|
||
| transactionLimiter.acquire(); |
Member
There was a problem hiding this comment.
Recommendation generated by Amazon CodeGuru Reviewer. Leave feedback on this recommendation by replying to the comment or by reacting to the comment using emoji.
Lock acquisition without timeout detected. Using locks without timeouts can lead to indefinite blocking and service outages if the lock cannot be acquired. Use timeout-based methods like tryLock(timeout), tryAcquire(timeout), or await(timeout) instead. Learn more on : https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/locks/Lock.html#tryLock-long-java.util.concurrent.TimeUnit
AniruddhaKanhere
force-pushed
the
feature/coremqtt-poc
branch
3 times, most recently
from
May 14, 2026 00:31
5772aaf to
75d97e2
Compare
MikeDombo
reviewed
May 14, 2026
| class CoreMqttJniClient implements IndividualMqttClient { | ||
|
|
||
| private static final Logger logger = LogManager.getLogger(CoreMqttJniClient.class); | ||
| private static final Random RANDOM = new Random(); |
Member
There was a problem hiding this comment.
Recommendation generated by Amazon CodeGuru Reviewer. Leave feedback on this recommendation by replying to the comment or by reacting to the comment using emoji.
Using insecure random number generators can lead to predictable random values. This predictability can be exploited by attackers to guess sensitive values or break cryptographic operations. To fix this, use new SecureRandom() to instantiate a java.security.SecureRandom object instead of using non-cryptographic random number generators (like java.util.Random, Math.random(), or Apache Commons random utilities). For more information about secure random number generation, see: https://owasp.org/www-community/vulnerabilities/Insecure_Randomness
Suggested remediation:
Use new SecureRandom() and provide a cryptographically strong random number generator to make your code secure.
@@ -51,3 +51,3 @@
private static final Logger logger = LogManager.getLogger(CoreMqttJniClient.class);
- private static final Random RANDOM = new Random();
+ private static final SecureRandom RANDOM = new SecureRandom();
MikeDombo
reviewed
May 14, 2026
| class CoreMqttJniClient implements IndividualMqttClient { | ||
|
|
||
| private static final Logger logger = LogManager.getLogger(CoreMqttJniClient.class); | ||
| private static final Random RANDOM = new Random(); |
Member
There was a problem hiding this comment.
Recommendation generated by Amazon CodeGuru Reviewer. Leave feedback on this recommendation by replying to the comment or by reacting to the comment using emoji.
Using insecure random number generators can lead to predictable random values. This predictability can be exploited by attackers to guess sensitive values or break cryptographic operations. To fix this, use new SecureRandom() to instantiate a java.security.SecureRandom object instead of using non-cryptographic random number generators (like java.util.Random, Math.random(), or Apache Commons random utilities). For more information about secure random number generation, see: https://owasp.org/www-community/vulnerabilities/Insecure_Randomness
Suggested remediation:
Use new SecureRandom() and provide a cryptographically strong random number generator to make your code secure.
@@ -53,3 +53,3 @@
private static final Logger logger = LogManager.getLogger(CoreMqttJniClient.class);
- private static final Random RANDOM = new Random();
+ private static final SecureRandom RANDOM = new SecureRandom();
| callbackEventManager.runOnConnectionInterrupted(errorCode); | ||
| } | ||
|
|
||
| public void onAckReceived(int packetId, int reasonCode) { |
| callbackEventManager.runOnConnectionInterrupted(errorCode); | ||
| } | ||
|
|
||
| public void onAckReceived(int packetId, int reasonCode) { |
| CompletableFuture<SubscribeResponse> future = new CompletableFuture<>(); | ||
| inprogressSubscriptions.incrementAndGet(); | ||
|
|
||
| connect().whenComplete((connResult, connError) -> { |
| public CompletableFuture<UnsubscribeResponse> unsubscribe(String topic) { | ||
| CompletableFuture<UnsubscribeResponse> future = new CompletableFuture<>(); | ||
|
|
||
| connect().whenComplete((connResult, connError) -> { |
|
|
||
| CompletableFuture<PubAck> future = new CompletableFuture<>(); | ||
|
|
||
| connect().whenComplete((connResult, connError) -> { |
MikeDombo
reviewed
May 15, 2026
| class CoreMqttJniClient implements IndividualMqttClient { | ||
|
|
||
| private static final Logger logger = LogManager.getLogger(CoreMqttJniClient.class); | ||
| private static final Random RANDOM = new Random(); |
Member
There was a problem hiding this comment.
Recommendation generated by Amazon CodeGuru Reviewer. Leave feedback on this recommendation by replying to the comment or by reacting to the comment using emoji.
Using insecure random number generators can lead to predictable random values. This predictability can be exploited by attackers to guess sensitive values or break cryptographic operations. To fix this, use new SecureRandom() to instantiate a java.security.SecureRandom object instead of using non-cryptographic random number generators (like java.util.Random, Math.random(), or Apache Commons random utilities). For more information about secure random number generation, see: https://owasp.org/www-community/vulnerabilities/Insecure_Randomness
Suggested remediation:
Use new SecureRandom() and provide a cryptographically strong random number generator to make your code secure.
@@ -52,3 +52,3 @@
private static final Logger logger = LogManager.getLogger(CoreMqttJniClient.class);
- private static final Random RANDOM = new Random();
+ private static final SecureRandom RANDOM = new SecureRandom();
|
Unit Tests Coverage Report
Minimum allowed coverage is Generated by 🐒 cobertura-action against db860f0 |
|
Integration Tests Coverage Report
Minimum allowed coverage is Generated by 🐒 cobertura-action against db860f0 |
AniruddhaKanhere
force-pushed
the
feature/coremqtt-poc
branch
from
May 18, 2026 21:29
52e3385 to
835c37d
Compare
Replace the aws-c-mqtt MQTT protocol library with FreeRTOS coreMQTT, while keeping the rest of the CRT stack (aws-c-io, s2n-tls, event loops, socket handling) intact. - Add coremqtt_channel_handler.c: aws_channel_handler that wraps coreMQTT - Add coremqtt_jni.c: JNI entry points for Java integration - Add CoreMqttJniClient.java: IndividualMqttClient implementation - Add CoreMqttNative.java: JNI method declarations - Modified MqttClient.java: route mqtt.version=coremqtt to new client - Add CMakeLists.txt, Dockerfile, README with quick-start guide - coreMQTT added as git submodule Tested: device registers as HEALTHY, subscriptions work across multiple connections, publish QoS 0/1 with PUBACK, keep-alive, reconnection.
AniruddhaKanhere
force-pushed
the
feature/coremqtt-poc
branch
from
May 19, 2026 18:55
be32036 to
0b3a2a2
Compare
MikeDombo
reviewed
May 19, 2026
| try { | ||
| subscribe(sub).get(30, TimeUnit.SECONDS); | ||
| droppedSubscriptionTopics.remove(sub); | ||
| } catch (Exception e) { |
Member
There was a problem hiding this comment.
Recommendation generated by Amazon CodeGuru Reviewer. Leave feedback on this recommendation by replying to the comment or by reacting to the comment using emoji.
It appears that your code handles a broad swath of exceptions in the catch block, potentially trapping dissimilar issues or problems that should not be dealt with at this point in the program.
| final class CoreMqttNative { | ||
|
|
||
| static { | ||
| System.loadLibrary("coremqtt_jni"); |
Member
There was a problem hiding this comment.
Recommendation generated by Amazon CodeGuru Reviewer. Leave feedback on this recommendation by replying to the comment or by reacting to the comment using emoji.
It looks like you are using System.loadLibrary() that doesn't specify an absolute path, which could potentially lead to loading a malicious library provided by an attacker. Instead, consider using System.load() to provide the library's full path for greater security.
| "Lifecycle": { | ||
| "install": "pip3 install 'awsiotsdk==1.19.0' 2>/dev/null || true", | ||
| "run": { | ||
| "script": "python3 -c \"\nimport json, time, signal, sys\ntry:\n from awsiot.greengrasscoreipc.clientv2 import GreengrassCoreIPCClientV2\nexcept ImportError:\n print('awsiotsdk not available', flush=True)\n sys.exit(1)\n\nrunning = True\ndef handler(s,f): global running; running = False\nsignal.signal(signal.SIGTERM, handler)\n\nprint('Starting coreMQTT PoC publisher...', flush=True)\nipc = GreengrassCoreIPCClientV2()\ncount = 0\nwhile running:\n count += 1\n msg = json.dumps({'message': 'Hello from coreMQTT PoC!', 'count': count, 'timestamp': int(time.time())})\n try:\n ipc.publish_to_iot_core(topic_name='gg/coremqtt/hello', qos=1, payload=msg.encode())\n print(f'Published #{count} to gg/coremqtt/hello', flush=True)\n except Exception as e:\n print(f'Error: {e}', flush=True)\n time.sleep(10)\n\"" |
Member
There was a problem hiding this comment.
Recommendation generated by Amazon CodeGuru Reviewer. Leave feedback on this recommendation by replying to the comment or by reacting to the comment using emoji.
It appears that your JSON object is invalid. Some possible reasons for this could be: use of single quotes, trailing commas, use of mathematical expressions etc. Make sure that you correctly format the JSON object to avoid production issues later on.
| while (connected() && !droppedSubscriptionTopics.isEmpty()) { | ||
| if (delayMs > 0) { | ||
| try { | ||
| Thread.sleep(delayMs); |
Member
There was a problem hiding this comment.
Recommendation generated by Amazon CodeGuru Reviewer. Leave feedback on this recommendation by replying to the comment or by reacting to the comment using emoji.
Manual polling with Thread.sleep() is used instead of AWS SDK waiters. This approach is inefficient, error-prone, and can lead to resource wastage. Use AWS SDK waiters (e.g., AmazonEC2Waiters, AmazonS3Waiters) which provide built-in polling mechanisms with exponential backoff. Learn more https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/waiters.html.
Replace manual byte-level CONNACK parsing with coreMQTT's MQTT_DeserializeConnAck. This properly: - Extracts sessionPresent flag (was hardcoded false) - Validates CONNACK reason code (server refusal now detected) - Decodes variable-length remaining length (was assuming 1 byte) - Reads server properties (serverMaxPacketSize, serverKeepAlive) - Respects server-assigned keep-alive per MQTT 5 spec
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Issue #, if available:
Description of changes:
Why is this change necessary:
How was this change tested:
Any additional information or context required to review the change:
Documentation Checklist:
Compatibility Checklist:
any deprecated method or type.
Refer to Compatibility Guidelines for more information.
By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.