diff --git a/CHANGELOG.md b/CHANGELOG.md index bbe0f9a..dbe0be1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # CHANGELOG +## 1.6.3 - 2026-04-13 + +- Support new regions af-south-1 +- Fix timeout issue + ## 1.6.2 - 2026-01-19 - Support new regions eu-west-2 and ca-central-1 diff --git a/README.md b/README.md index 2545b52..2b8e06f 100644 --- a/README.md +++ b/README.md @@ -187,7 +187,7 @@ Creates a new instance of the `AmaasClient` class, and provisions essential sett | Parameter | Description | | ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| region | The region you obtained your api key. Value provided must be one of the Vision One regions, e.g. `us-east-1`, `eu-central-1`, `ap-northeast-1`, `ap-southeast-2`, `ap-southeast-1`, `ap-south-1`, `me-central-1`,`eu-west-2`,`ca-central-1`, etc. If host is given, region will be ignored. | +| region | The region you obtained your api key. Value provided must be one of the Vision One regions, e.g. `us-east-1`, `eu-central-1`, `ap-northeast-1`, `ap-southeast-2`, `ap-southeast-1`, `ap-south-1`, `me-central-1`,`eu-west-2`,`ca-central-1`,`af-south-1`, etc. If host is given, region will be ignored. | | host | The host ip address of self hosted AMaaS scanner. Ignore if to use Trend AMaaS service | | apikey | Your own Vision One API Key. | | timeoutInSecs | Timeout to cancel the connection to server in seconds. Valid value is 0, 1, 2, ... ; default to 300 seconds. | @@ -205,7 +205,7 @@ Creates a new instance of the `AmaasClient` class, and provisions essential sett | Parameter | Description | | ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| region | The region you obtained your api key. Value provided must be one of the Vision One regions, e.g. `us-east-1`, `eu-central-1`, `ap-northeast-1`, `ap-southeast-2`, `ap-southeast-1`, `ap-south-1`, `me-central-1`,`eu-west-2`,`ca-central-1` ,etc. | +| region | The region you obtained your api key. Value provided must be one of the Vision One regions, e.g. `us-east-1`, `eu-central-1`, `ap-northeast-1`, `ap-southeast-2`, `ap-southeast-1`, `ap-south-1`, `me-central-1`,`eu-west-2`,`ca-central-1`,`af-south-1` ,etc. | | apikey | Your own Vision One API Key. | | timeoutInSecs | Timeout to cancel the connection to server in seconds. Valid value is 0, 1, 2, ... ; default to 300 seconds. | diff --git a/VERSION b/VERSION index fdd3be6..266146b 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.6.2 +1.6.3 diff --git a/pom.xml b/pom.xml index f380dbc..fc81d40 100644 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ com.trend file-security-java-sdk - 1.6.2 + 1.6.3 file-security-java-sdk https://github.com/trendmicro/tm-v1-fs-java-sdk diff --git a/protos/scan.proto b/protos/scan.proto index 561138b..547d745 100644 --- a/protos/scan.proto +++ b/protos/scan.proto @@ -15,6 +15,7 @@ enum Stage { STAGE_INIT = 0; STAGE_RUN = 1; STAGE_FINI = 2; + STAGE_HEARTBEAT = 3; } message C2S { diff --git a/src/main/java/com/trend/cloudone/amaas/AMaasClient.java b/src/main/java/com/trend/cloudone/amaas/AMaasClient.java index 5e763b4..a09b9fe 100644 --- a/src/main/java/com/trend/cloudone/amaas/AMaasClient.java +++ b/src/main/java/com/trend/cloudone/amaas/AMaasClient.java @@ -6,6 +6,8 @@ import java.nio.file.Paths; import java.util.Arrays; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.logging.Level; import java.util.logging.Logger; @@ -304,6 +306,7 @@ private static void log(final Level level, final String msg, final Object... par */ static class AMaasServerCallback implements StreamObserver { private static final int POLL_TIME_MILLIS = 200; + static final int DEFAULT_HEARTBEAT_INTERVAL_MS = 30 * 1000; private StreamObserver requestObserver; private AMaasReader reader; @@ -316,10 +319,57 @@ static class AMaasServerCallback implements StreamObserver { private boolean bulk = true; private long start = System.currentTimeMillis(); private long timeoutSecs; + private final Object streamLock = new Object(); + private volatile boolean streamClosed = false; + private ScheduledExecutorService heartbeatExecutor = null; + private int heartbeatIntervalMs = DEFAULT_HEARTBEAT_INTERVAL_MS; AMaasServerCallback() { } + void setHeartbeatIntervalMs(final int intervalMs) { + this.heartbeatIntervalMs = intervalMs; + } + + protected void startHeartbeat() { + heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(new java.util.concurrent.ThreadFactory() { + @Override + public Thread newThread(final Runnable r) { + Thread t = new Thread(r, "amaas-heartbeat"); + t.setDaemon(true); + return t; + } + }); + heartbeatExecutor.scheduleAtFixedRate(new Runnable() { + @Override + public void run() { + if (streamClosed) { + return; + } + ScanOuterClass.C2S hb = ScanOuterClass.C2S.newBuilder() + .setStage(Stage.STAGE_HEARTBEAT) + .build(); + synchronized (streamLock) { + if (!streamClosed) { + try { + requestObserver.onNext(hb); + log(Level.FINE, "Sent heartbeat"); + } catch (Exception e) { + log(Level.WARNING, "Heartbeat send failed: {0}", e.getMessage()); + } + } + } + } + }, heartbeatIntervalMs, heartbeatIntervalMs, TimeUnit.MILLISECONDS); + } + + protected void stopHeartbeat() { + if (heartbeatExecutor != null) { + heartbeatExecutor.shutdownNow(); + heartbeatExecutor = null; + } + } + private AMaasException processError() { AMaasException err = null; if (this.grpcStatus == Status.Code.UNAUTHENTICATED) { @@ -363,7 +413,13 @@ public void onNext(final ScanOuterClass.S2C s2cMsg) { case CMD_RETR: if (s2cMsg.getStage() != Stage.STAGE_RUN) { log(Level.INFO, "Received unexpected command RETR at stage {0}", s2cMsg.getStage()); - requestObserver.onError(new StatusRuntimeException(Status.ABORTED)); + synchronized (streamLock) { + if (!streamClosed) { + streamClosed = true; + requestObserver.onError(new StatusRuntimeException(Status.ABORTED)); + } + } + return; } java.util.List bulkLength; java.util.List bulkOffset; @@ -401,30 +457,55 @@ public void onNext(final ScanOuterClass.S2C s2cMsg) { if (TimeUnit.MILLISECONDS.toSeconds(duration) > this.timeoutSecs) { log(Level.INFO, "DEADLINE_EXCEEDED {0}", duration); - requestObserver.onError(new StatusRuntimeException(Status.DEADLINE_EXCEEDED)); + synchronized (streamLock) { + if (!streamClosed) { + streamClosed = true; + requestObserver.onError(new StatusRuntimeException(Status.DEADLINE_EXCEEDED)); + } + } return; } } - requestObserver.onNext(request); + synchronized (streamLock) { + if (!streamClosed) { + requestObserver.onNext(request); + } + } } catch (IOException e) { log(Level.SEVERE, "Exception when processing server message", e.getMessage()); - requestObserver.onError(new StatusRuntimeException(Status.ABORTED)); + synchronized (streamLock) { + if (!streamClosed) { + streamClosed = true; + requestObserver.onError(new StatusRuntimeException(Status.ABORTED)); + } + } } } break; case CMD_QUIT: this.scanResult = s2cMsg.getResult(); log(Level.INFO, "Scan succeeded: result={0} fetchCount={1} fetchSize={2}.", this.scanResult, this.fetchCount, this.fetchSize); - requestObserver.onCompleted(); + synchronized (streamLock) { + streamClosed = true; + requestObserver.onCompleted(); + } break; default: log(Level.WARNING, "Unknown command"); - requestObserver.onError(new StatusRuntimeException(Status.INVALID_ARGUMENT)); + synchronized (streamLock) { + if (!streamClosed) { + streamClosed = true; + requestObserver.onError(new StatusRuntimeException(Status.INVALID_ARGUMENT)); + } + } } } @Override public void onError(final Throwable t) { + synchronized (streamLock) { + streamClosed = true; + } log(Level.WARNING, "scan Failed: {0}", Status.fromThrowable(t)); this.done = true; this.grpcStatus = Status.fromThrowable(t).getCode(); @@ -433,6 +514,9 @@ public void onError(final Throwable t) { @Override public void onCompleted() { + synchronized (streamLock) { + streamClosed = true; + } log(Level.INFO, "File successfully scanned."); this.done = true; this.grpcStatus = Status.Code.OK; @@ -528,10 +612,13 @@ public String scanRun(final AMaasReader reader, final AMaasScanOptions options) ScanOuterClass.C2S request = builder.build(); requestObserver.onNext(request); + serverCallback.startHeartbeat(); - String scanResult = serverCallback.waitTilExit(); - - return scanResult; + try { + return serverCallback.waitTilExit(); + } finally { + serverCallback.stopHeartbeat(); + } } /** diff --git a/src/main/java/com/trend/cloudone/amaas/AMaasRegion.java b/src/main/java/com/trend/cloudone/amaas/AMaasRegion.java index 615764e..e9ba802 100644 --- a/src/main/java/com/trend/cloudone/amaas/AMaasRegion.java +++ b/src/main/java/com/trend/cloudone/amaas/AMaasRegion.java @@ -25,6 +25,7 @@ final class AMaasRegion { static final String AWS_TREND_REGION = "us-east-2"; static final String AWS_GB_REGION = "eu-west-2"; static final String AWS_AE_REGION = "me-central-1"; + static final String AWS_ZA_REGION = "af-south-1"; static final String C1_JP_REGION = "jp-1"; static final String C1_SG_REGION = "sg-1"; static final String C1_AU_REGION = "au-1"; @@ -35,11 +36,12 @@ final class AMaasRegion { static final String C1_TREND_REGION = "trend-us-1"; static final String C1_GB_REGION = "gb-1"; static final String C1_AE_REGION = "ae-1"; + static final String C1_ZA_REGION = "za-1"; - static final List C1_REGIONS = Arrays.asList(new String[]{C1_AU_REGION, C1_CA_REGION, C1_DE_REGION, C1_GB_REGION, C1_IN_REGION, C1_JP_REGION, C1_SG_REGION, C1_US_REGION, C1_TREND_REGION}); - static final List V1_REGIONS = Arrays.asList(new String[]{AWS_AU_REGION, AWS_CA_REGION, AWS_DE_REGION, AWS_GB_REGION, AWS_IN_REGION, AWS_JP_REGION, AWS_SG_REGION, AWS_US_REGION, AWS_AE_REGION}); - static final List SUPPORTED_V1_REGIONS = Arrays.asList(new String[]{AWS_AU_REGION, AWS_DE_REGION, AWS_JP_REGION, AWS_SG_REGION, AWS_US_REGION, AWS_IN_REGION, AWS_AE_REGION, AWS_CA_REGION, AWS_GB_REGION}); - static final List SUPPORTED_C1_REGIONS = Arrays.asList(new String[]{C1_AU_REGION, C1_CA_REGION, C1_DE_REGION, C1_GB_REGION, C1_IN_REGION, C1_JP_REGION, C1_SG_REGION, C1_US_REGION}); + static final List C1_REGIONS = Arrays.asList(new String[]{C1_AU_REGION, C1_CA_REGION, C1_DE_REGION, C1_GB_REGION, C1_IN_REGION, C1_JP_REGION, C1_SG_REGION, C1_US_REGION, C1_TREND_REGION, C1_ZA_REGION}); + static final List V1_REGIONS = Arrays.asList(new String[]{AWS_AU_REGION, AWS_CA_REGION, AWS_DE_REGION, AWS_GB_REGION, AWS_IN_REGION, AWS_JP_REGION, AWS_SG_REGION, AWS_US_REGION, AWS_AE_REGION, AWS_ZA_REGION}); + static final List SUPPORTED_V1_REGIONS = Arrays.asList(new String[]{AWS_AU_REGION, AWS_DE_REGION, AWS_JP_REGION, AWS_SG_REGION, AWS_US_REGION, AWS_IN_REGION, AWS_AE_REGION, AWS_CA_REGION, AWS_GB_REGION, AWS_ZA_REGION}); + static final List SUPPORTED_C1_REGIONS = Arrays.asList(new String[]{C1_AU_REGION, C1_CA_REGION, C1_DE_REGION, C1_GB_REGION, C1_IN_REGION, C1_JP_REGION, C1_SG_REGION, C1_US_REGION, C1_ZA_REGION}); static final Map V1_TO_C1_REGION_MAPPING = new HashMap() { { @@ -52,6 +54,7 @@ final class AMaasRegion { put(AWS_AE_REGION, C1_AE_REGION); put(AWS_CA_REGION, C1_CA_REGION); put(AWS_GB_REGION, C1_GB_REGION); + put(AWS_ZA_REGION, C1_ZA_REGION); } }; @@ -67,6 +70,7 @@ final class AMaasRegion { put(C1_CA_REGION, "antimalware.ca-1.cloudone.trendmicro.com"); put(C1_TREND_REGION, "antimalware.trend-us-1.cloudone.trendmicro.com"); put(C1_AE_REGION, "antimalware.ae-1.cloudone.trendmicro.com"); + put(C1_ZA_REGION, "antimalware.za-1.cloudone.trendmicro.com"); } }; diff --git a/src/test/java/com/trend/cloudone/amaas/AMaasClientTest.java b/src/test/java/com/trend/cloudone/amaas/AMaasClientTest.java index a76642d..b6204ee 100644 --- a/src/test/java/com/trend/cloudone/amaas/AMaasClientTest.java +++ b/src/test/java/com/trend/cloudone/amaas/AMaasClientTest.java @@ -6,6 +6,7 @@ import org.junit.Rule; import org.junit.rules.ExpectedException; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import com.google.gson.Gson; import com.google.gson.GsonBuilder; @@ -24,11 +25,22 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; public class AMaasClientTest { private static final int TIMEOUT_SEC = 5000; private static final int BUFF_LENGTH = 64; + private static final int HEARTBEAT_INTERVAL_MS = 50; + private static final int HEARTBEAT_FAST_INTERVAL_MS = 10; + private static final int HEARTBEAT_WAIT_MS = 250; + private static final int HEARTBEAT_SHORT_WAIT_MS = 80; + private static final int HEARTBEAT_STOP_WAIT_MS = 150; + private static final int HEARTBEAT_MIN_EXPECTED = 3; + private static final int DATA_MSG_COUNT = 20; + private static final int DATA_MSG_DELAY_MS = 5; + private static final byte[] TEST_DATA = new byte[]{1, 2, 3}; private final MutableHandlerRegistry serviceRegistry = new MutableHandlerRegistry(); private ScanGrpc.ScanStub client; private Gson gson; @@ -234,5 +246,112 @@ public void testGetTagListErrors() { err = AMaasClient.getTagListErrors(taglist); assertEquals(err.getErrorCode(), AMaasErrorCode.MSG_ID_ERR_LENGTH_OF_TAG); } + + // Heartbeat tests + + private StreamObserver createRecordingObserver(final List recorded) { + return new StreamObserver() { + @Override + public void onNext(final ScanOuterClass.C2S value) { + recorded.add(value); + } + + @Override + public void onError(final Throwable t) { + } + + @Override + public void onCompleted() { + } + }; + } + + @Test + public void testHeartbeatSendsHeartbeats() throws Exception { + AMaasClient.AMaasServerCallback callback = new AMaasClient.AMaasServerCallback(); + callback.setHeartbeatIntervalMs(HEARTBEAT_INTERVAL_MS); + + List recorded = new CopyOnWriteArrayList<>(); + StreamObserver mockObserver = createRecordingObserver(recorded); + + AMaasBufferReader reader = new AMaasBufferReader(TEST_DATA, "test_heartbeat", true); + callback.setContext(mockObserver, reader, false, TIMEOUT_SEC); + callback.startHeartbeat(); + + Thread.sleep(HEARTBEAT_WAIT_MS); + + callback.stopHeartbeat(); + + long heartbeatCount = recorded.stream() + .filter(msg -> msg.getStage() == Stage.STAGE_HEARTBEAT) + .count(); + + assertTrue("Expected at least 3 heartbeats, got " + heartbeatCount, + heartbeatCount >= HEARTBEAT_MIN_EXPECTED); + } + + @Test + public void testHeartbeatStopsAfterStop() throws Exception { + AMaasClient.AMaasServerCallback callback = new AMaasClient.AMaasServerCallback(); + callback.setHeartbeatIntervalMs(HEARTBEAT_INTERVAL_MS); + + List recorded = new CopyOnWriteArrayList<>(); + StreamObserver mockObserver = createRecordingObserver(recorded); + + AMaasBufferReader reader = new AMaasBufferReader(TEST_DATA, "test_heartbeat_stop", true); + callback.setContext(mockObserver, reader, false, TIMEOUT_SEC); + callback.startHeartbeat(); + + Thread.sleep(HEARTBEAT_SHORT_WAIT_MS); + + callback.stopHeartbeat(); + + long heartbeatCountBeforeStop = recorded.stream() + .filter(msg -> msg.getStage() == Stage.STAGE_HEARTBEAT) + .count(); + assertTrue("Expected at least 1 heartbeat before stop, got " + heartbeatCountBeforeStop, + heartbeatCountBeforeStop > 0); + + int sizeAfterStop = recorded.size(); + Thread.sleep(HEARTBEAT_STOP_WAIT_MS); + + assertEquals("No new messages should arrive after stopHeartbeat", + sizeAfterStop, recorded.size()); + } + + @Test + public void testHeartbeatDataAndHeartbeatCoexist() throws Exception { + AMaasClient.AMaasServerCallback callback = new AMaasClient.AMaasServerCallback(); + callback.setHeartbeatIntervalMs(HEARTBEAT_FAST_INTERVAL_MS); + + List recorded = new CopyOnWriteArrayList<>(); + StreamObserver mockObserver = createRecordingObserver(recorded); + + AMaasBufferReader reader = new AMaasBufferReader(TEST_DATA, "test_coexist", true); + callback.setContext(mockObserver, reader, false, TIMEOUT_SEC); + callback.startHeartbeat(); + + for (int i = 0; i < DATA_MSG_COUNT; i++) { + ScanOuterClass.C2S dataMsg = ScanOuterClass.C2S.newBuilder() + .setStage(Stage.STAGE_RUN) + .setOffset(i) + .build(); + mockObserver.onNext(dataMsg); + Thread.sleep(DATA_MSG_DELAY_MS); + } + + callback.stopHeartbeat(); + + long dataCount = recorded.stream() + .filter(msg -> msg.getStage() == Stage.STAGE_RUN) + .count(); + long heartbeatCount = recorded.stream() + .filter(msg -> msg.getStage() == Stage.STAGE_HEARTBEAT) + .count(); + + assertEquals(DATA_MSG_COUNT, (int) dataCount); + assertTrue("Expected at least 1 heartbeat during data sends, got " + heartbeatCount, + heartbeatCount > 0); + } } diff --git a/src/test/java/com/trend/cloudone/amaas/AMaasRegionTest.java b/src/test/java/com/trend/cloudone/amaas/AMaasRegionTest.java index 0228e66..12ceb2a 100644 --- a/src/test/java/com/trend/cloudone/amaas/AMaasRegionTest.java +++ b/src/test/java/com/trend/cloudone/amaas/AMaasRegionTest.java @@ -22,6 +22,9 @@ public void testGetHostForRegion() { region = AMaasRegion.getServiceFqdn("eu-west-2"); assertEquals("antimalware.gb-1.cloudone.trendmicro.com", region); + + region = AMaasRegion.getServiceFqdn("af-south-1"); + assertEquals("antimalware.za-1.cloudone.trendmicro.com", region); } diff --git a/src/test/java/com/trend/cloudone/amaas/MockScanServicer.java b/src/test/java/com/trend/cloudone/amaas/MockScanServicer.java index a625c0a..6eca9a5 100644 --- a/src/test/java/com/trend/cloudone/amaas/MockScanServicer.java +++ b/src/test/java/com/trend/cloudone/amaas/MockScanServicer.java @@ -133,10 +133,14 @@ public StreamObserver run(final StreamObserver responseObserver) { @Override public void onNext(final C2S request) { + if (request.getStage() == Stage.STAGE_HEARTBEAT) { + return; + } S2C resp = processRequest(request); if (resp == null) { Status status = Status.INTERNAL; responseObserver.onError(status.asRuntimeException()); + return; } responseObserver.onNext(resp); }