diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 00000000..6313b56c --- /dev/null +++ b/.gitattributes @@ -0,0 +1 @@ +* text=auto eol=lf diff --git a/SECURITY.md b/SECURITY.md index f4d0ce33..d737d0f7 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -1,41 +1,41 @@ -The Atsign FoundationThe Atsign Foundation - -# Atsign Foundation Open Source Security Policies and Procedures - -This document outlines security procedures and general policies for the -Atsign Foundation Open Source projects as found on -https://github.com/atsign-foundation. - - * [Reporting a Vulnerability](#reporting-a-vulnerability) - * [Disclosure Policy](#disclosure-policy) - -## Reporting a Vulnerability - -The Atsign Foundation team and community take all security vulnerabilities -seriously. Thank you for improving the security of our open source -software. We appreciate your efforts and responsible disclosure and will -make every effort to acknowledge your contributions. - -Report security vulnerabilities by emailing the Atsign security team at: - - security@atsign.com - -The lead maintainer will acknowledge your email within 24 hours, and will -send a more detailed response within 48 hours indicating the next steps in -handling your report. After the initial reply to your report, the security -team will endeavor to keep you informed of the progress towards a fix and -full announcement, and may ask for additional information or guidance. - -Please report security vulnerabilities in third-party modules to the person -or team maintaining the module. - -## Disclosure Policy - -When the security team receives a security bug report, they will assign it -to a primary handler. This person will coordinate the fix and release -process, involving the following steps: - - * Confirm the problem and determine the affected versions. - * Audit code to find any potential similar problems. - * Prepare fixes for all releases still under maintenance. These fixes +The Atsign FoundationThe Atsign Foundation + +# Atsign Foundation Open Source Security Policies and Procedures + +This document outlines security procedures and general policies for the +Atsign Foundation Open Source projects as found on +https://github.com/atsign-foundation. + + * [Reporting a Vulnerability](#reporting-a-vulnerability) + * [Disclosure Policy](#disclosure-policy) + +## Reporting a Vulnerability + +The Atsign Foundation team and community take all security vulnerabilities +seriously. Thank you for improving the security of our open source +software. We appreciate your efforts and responsible disclosure and will +make every effort to acknowledge your contributions. + +Report security vulnerabilities by emailing the Atsign security team at: + + security@atsign.com + +The lead maintainer will acknowledge your email within 24 hours, and will +send a more detailed response within 48 hours indicating the next steps in +handling your report. After the initial reply to your report, the security +team will endeavor to keep you informed of the progress towards a fix and +full announcement, and may ask for additional information or guidance. + +Please report security vulnerabilities in third-party modules to the person +or team maintaining the module. + +## Disclosure Policy + +When the security team receives a security bug report, they will assign it +to a primary handler. This person will coordinate the fix and release +process, involving the following steps: + + * Confirm the problem and determine the affected versions. + * Audit code to find any potential similar problems. + * Prepare fixes for all releases still under maintenance. These fixes will be released as fast as possible to pub.dev where applicable. \ No newline at end of file diff --git a/at_client/pom.xml b/at_client/pom.xml index 818dc167..056f1f0a 100644 --- a/at_client/pom.xml +++ b/at_client/pom.xml @@ -1,21 +1,21 @@ - 4.0.0 + 4.0.0 org.atsign at_client 0.0.1-SNAPSHOT at_client - atsign Java client SDK - https://github.com/atsign-foundation/at_java - - - 8 - 8 - true - UTF-8 + atsign Java client SDK + https://github.com/atsign-foundation/at_java + + + 8 + 8 + true + UTF-8 false ${skipTests} ${skipTests} @@ -24,79 +24,78 @@ 1.2.0 - - - BSD 3-Clause License - https://github.com/atsign-foundation/at_java/blob/trunk/LICENSE - - + + + BSD 3-Clause License + https://github.com/atsign-foundation/at_java/blob/trunk/LICENSE + + - + atsign maven@atsign.com atsign https://atsign.com - + - - scm:git:git://github.com/atsign-foundation/at_java.git - scm:git:ssh://git@github.com:atsign-foundation/at_java.git - https://github.com/atsign-foundation/at_java/tree/trunk - + + scm:git:git://github.com/atsign-foundation/at_java.git + scm:git:ssh://git@github.com:atsign-foundation/at_java.git + https://github.com/atsign-foundation/at_java/tree/trunk + - - + + - + - com.googlecode.maven-download-plugin - download-maven-plugin - 1.6.8 - - - download-at_demo_data - generate-test-resources - - wget - - - https://pub.dev/packages/at_demo_data/versions/${version.at_demo_data}.tar.gz - ${project.build.directory}/at_demo_data - at_demo_data.tar.gz - true - - - - - - - org.codehaus.mojo - build-helper-maven-plugin - 3.0.0 - - - add-integration-test-source - generate-test-sources - - add-test-source - - - - src/integration-test/java - - - - - - - maven-surefire-plugin - 3.0.0-M7 - - ${skipUnitTests} - - + com.googlecode.maven-download-plugin + download-maven-plugin + 1.6.8 + + + download-at_demo_data + generate-test-resources + + wget + + + https://pub.dev/packages/at_demo_data/versions/${version.at_demo_data}.tar.gz + ${project.build.directory}/at_demo_data + true + + + + + + + org.codehaus.mojo + build-helper-maven-plugin + 3.0.0 + + + add-integration-test-source + generate-test-sources + + add-test-source + + + + src/integration-test/java + + + + + + + maven-surefire-plugin + 3.0.0-M7 + + ${skipUnitTests} + + org.apache.maven.plugins maven-failsafe-plugin @@ -115,18 +114,18 @@ - org.apache.maven.plugins - maven-source-plugin + org.apache.maven.plugins + maven-source-plugin 3.3.1 - - - attach-sources + + + attach-sources - jar - - - - + jar + + + + org.apache.maven.plugins @@ -172,128 +171,185 @@ + + com.diffplug.spotless + spotless-maven-plugin + 2.30.0 + + + spotless-check + validate + + check + + + + + + + ${project.basedir}/../config/java-format.xml + + + + + + + **/pom.xml + + + + + true + 2 + + + + + + + + org.apache.maven.plugins + maven-checkstyle-plugin + 3.3.1 + + ../config/checkstyle.xml + true + true + false + + + + checkstyle + validate + + check + + + + + - + - - com.fasterxml.jackson.core - jackson-core - 2.15.0 - - - - com.fasterxml.jackson.core - jackson-annotations - 2.14.0 - - - - com.fasterxml.jackson.core - jackson-databind - 2.14.0 - - - - - com.fasterxml.jackson.dataformat - jackson-dataformat-yaml - 2.14.0 - - - - org.bouncycastle - bcprov-jdk15to18 - 1.78 - - - - junit - junit - 4.13.2 - test - - - org.apache.commons - commons-lang3 - 3.18.0 - - - info.picocli - picocli - 4.7.0 - - - org.mockito - mockito-inline - 4.8.0 - - - org.mockito - mockito-junit-jupiter - 4.8.0 - test - - - com.github.sarxos - webcam-capture - 0.3.12 - - - org.fusesource.jansi - jansi - 2.4.0 - - - - io.cucumber - cucumber-java - ${version.cucumber} - test - - - - io.cucumber - cucumber-junit - ${version.cucumber} - test - - - - io.cucumber - cucumber-picocontainer - ${version.cucumber} - test - - - - org.testcontainers - testcontainers - 1.21.3 - test - - - - org.slf4j - slf4j-api - 2.0.7 - test - - - - org.slf4j - slf4j-simple - 2.0.7 - test - - - - org.hamcrest - hamcrest-all - 1.3 - test - - - + + com.fasterxml.jackson.core + jackson-core + 2.15.0 + + + + com.fasterxml.jackson.core + jackson-annotations + 2.14.0 + + + + com.fasterxml.jackson.core + jackson-databind + 2.14.0 + + + + + com.fasterxml.jackson.dataformat + jackson-dataformat-yaml + 2.14.0 + + + + org.bouncycastle + bcprov-jdk15to18 + 1.78 + + + + junit + junit + 4.13.2 + test + + + org.apache.commons + commons-lang3 + 3.18.0 + + + info.picocli + picocli + 4.7.0 + + + org.mockito + mockito-inline + 4.8.0 + + + org.mockito + mockito-junit-jupiter + 4.8.0 + test + + + com.github.sarxos + webcam-capture + 0.3.12 + + + org.fusesource.jansi + jansi + 2.4.0 + + + + io.cucumber + cucumber-java + ${version.cucumber} + test + + + + io.cucumber + cucumber-junit + ${version.cucumber} + test + + + + io.cucumber + cucumber-picocontainer + ${version.cucumber} + test + + + + org.testcontainers + testcontainers + 1.21.3 + test + + + + org.slf4j + slf4j-api + 2.0.7 + test + + + + org.slf4j + slf4j-simple + 2.0.7 + test + + + + org.hamcrest + hamcrest-all + 1.3 + test + + + diff --git a/at_client/src/main/java/org/atsign/client/api/AtClient.java b/at_client/src/main/java/org/atsign/client/api/AtClient.java index 205b32cf..efef6f6a 100644 --- a/at_client/src/main/java/org/atsign/client/api/AtClient.java +++ b/at_client/src/main/java/org/atsign/client/api/AtClient.java @@ -26,151 +26,185 @@ @SuppressWarnings("unused") public interface AtClient extends Secondary, AtEvents.AtEventBus { - /** - * Standard AtClient factory - uses production @ root to look up the cloud secondary address for this atSign - * @param atSign the atsign of this client - * @return An {@link AtClient} - * @throws AtException if something goes wrong with looking up or connecting to the remote secondary - */ - static AtClient withRemoteSecondary(AtSign atSign) throws AtException { - return withRemoteSecondary("root.atsign.org:64", atSign); + /** + * Standard AtClient factory - uses production @ root to look up the cloud secondary address for + * this atSign + * + * @param atSign the atsign of this client + * @return An {@link AtClient} + * @throws AtException if something goes wrong with looking up or connecting to the remote secondary + */ + static AtClient withRemoteSecondary(AtSign atSign) throws AtException { + return withRemoteSecondary("root.atsign.org:64", atSign); + } + + /** + * Standard AtClient factory - uses production @ root to look up the cloud secondary address for + * this atSign + * + * @param atSign the atsign of this client + * @param verbose set to true for chatty logs + * @return An {@link AtClient} + * @throws AtException if something goes wrong with looking up or connecting to the remote secondary + */ + static AtClient withRemoteSecondary(AtSign atSign, boolean verbose) throws AtException { + return withRemoteSecondary("root.atsign.org:64", atSign, verbose); + } + + /** + * Factory to use when you wish to use a custom Secondary.AddressFinder + * + * @param atSign the atSign of this client + * @param secondaryAddressFinder will be used to find the Secondary.Address of the atSign + * @return An {@link AtClient} + * @throws AtException if any other exception occurs while connecting to the remote (cloud) + * secondary + */ + static AtClient withRemoteSecondary(AtSign atSign, Secondary.AddressFinder secondaryAddressFinder) + throws AtException { + Secondary.Address remoteSecondaryAddress; + try { + remoteSecondaryAddress = secondaryAddressFinder.findSecondary(atSign); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to find secondary, with IOException", e); } - /** - * Standard AtClient factory - uses production @ root to look up the cloud secondary address for this atSign - * @param atSign the atsign of this client - * @param verbose set to true for chatty logs - * @return An {@link AtClient} - * @throws AtException if something goes wrong with looking up or connecting to the remote secondary - */ - static AtClient withRemoteSecondary(AtSign atSign, boolean verbose) throws AtException { - return withRemoteSecondary("root.atsign.org:64", atSign, verbose); + return withRemoteSecondary(atSign, remoteSecondaryAddress, false); + } + + /** + * Factory - returns default AtClientImpl with a RemoteSecondary and a DefaultConnectionFactory + * + * @param rootUrl the address of the root server to use - e.g. root.atsign.org:64 for production + * at-signs + * @param atSign the atSign of the client - e.g. @alice + * @return An {@link AtClient} + * @throws AtException if anything goes wrong during construction + */ + static AtClient withRemoteSecondary(String rootUrl, AtSign atSign) throws AtException { + return withRemoteSecondary(rootUrl, atSign, false); + } + + static AtClient withRemoteSecondary(String rootUrl, AtSign atSign, boolean verbose) throws AtException { + DefaultAtConnectionFactory connectionFactory = new DefaultAtConnectionFactory(); + + Secondary.Address secondaryAddress; + try { + AtRootConnection rootConnection = connectionFactory.getRootConnection(new SimpleAtEventBus(), rootUrl, verbose); + rootConnection.connect(); + secondaryAddress = rootConnection.findSecondary(atSign); + } catch (AtSecondaryNotFoundException e) { + throw e; + } catch (Exception e) { + throw new AtSecondaryNotFoundException("Failed to lookup remote secondary", e); } - /** - * Factory to use when you wish to use a custom Secondary.AddressFinder - * @param atSign the atSign of this client - * @param secondaryAddressFinder will be used to find the Secondary.Address of the atSign - * @return An {@link AtClient} - * @throws AtException if any other exception occurs while connecting to the remote (cloud) secondary - */ - static AtClient withRemoteSecondary(AtSign atSign, Secondary.AddressFinder secondaryAddressFinder) throws AtException { - Secondary.Address remoteSecondaryAddress; - try { - remoteSecondaryAddress = secondaryAddressFinder.findSecondary(atSign); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to find secondary, with IOException", e); - } - return withRemoteSecondary(atSign, remoteSecondaryAddress, false); + return withRemoteSecondary(atSign, secondaryAddress, verbose); + } + + /** + * Factory to use when you wish to use a custom Secondary.AddressFinder + * + * @param atSign the atSign of this client + * @param verbose set to true for chatty logs + * @return An {@link AtClient} + * @throws IOException if thrown by the address finder + * @throws AtException if any other exception occurs while connecting to the remote (cloud) + * secondary + */ + static AtClient withRemoteSecondary(AtSign atSign, Secondary.AddressFinder secondaryAddressFinder, boolean verbose) + throws IOException, AtException { + Secondary.Address remoteSecondaryAddress = secondaryAddressFinder.findSecondary(atSign); + return withRemoteSecondary(atSign, remoteSecondaryAddress, verbose); + } + + /** + * Factory to use when you already know the address of the remote (cloud) secondary + * + * @param atSign the atSign of this client + * @param remoteSecondaryAddress the address of the remote secondary server + * @param verbose set to true for chatty logs + * @return An {@link AtClient} + * @throws AtException if any other exception occurs while connecting to the remote (cloud) + * secondary + */ + static AtClient withRemoteSecondary(AtSign atSign, Secondary.Address remoteSecondaryAddress, boolean verbose) + throws AtException { + DefaultAtConnectionFactory connectionFactory = new DefaultAtConnectionFactory(); + AtEvents.AtEventBus eventBus = new SimpleAtEventBus(); + + Map keys; + try { + keys = KeysUtil.loadKeys(atSign); + } catch (Exception e) { + throw new AtClientConfigException("Failed to load keys", e); } - /** - * Factory - returns default AtClientImpl with a RemoteSecondary and a DefaultConnectionFactory - * @param rootUrl the address of the root server to use - e.g. root.atsign.org:64 for production at-signs - * @param atSign the atSign of the client - e.g. @alice - * @return An {@link AtClient} - * @throws AtException if anything goes wrong during construction - */ - static AtClient withRemoteSecondary(String rootUrl, AtSign atSign) throws AtException { - return withRemoteSecondary(rootUrl, atSign, false); + RemoteSecondary secondary; + try { + secondary = new RemoteSecondary(eventBus, atSign, remoteSecondaryAddress, keys, connectionFactory, verbose); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to create RemoteSecondary", e); } - static AtClient withRemoteSecondary(String rootUrl, AtSign atSign, boolean verbose) throws AtException { - DefaultAtConnectionFactory connectionFactory = new DefaultAtConnectionFactory(); - - Secondary.Address secondaryAddress; - try { - AtRootConnection rootConnection = connectionFactory.getRootConnection(new SimpleAtEventBus(), rootUrl, verbose); - rootConnection.connect(); - secondaryAddress = rootConnection.findSecondary(atSign); - } catch (AtSecondaryNotFoundException e) { - throw e; - } catch (Exception e) { - throw new AtSecondaryNotFoundException("Failed to lookup remote secondary", e); - } - - return withRemoteSecondary(atSign, secondaryAddress, verbose); - } + return new AtClientImpl(eventBus, atSign, keys, secondary); + } - /** - * Factory to use when you wish to use a custom Secondary.AddressFinder - * @param atSign the atSign of this client - * @param verbose set to true for chatty logs - * @return An {@link AtClient} - * @throws IOException if thrown by the address finder - * @throws AtException if any other exception occurs while connecting to the remote (cloud) secondary - */ - static AtClient withRemoteSecondary(AtSign atSign, Secondary.AddressFinder secondaryAddressFinder, boolean verbose) throws IOException, AtException { - Secondary.Address remoteSecondaryAddress = secondaryAddressFinder.findSecondary(atSign); - return withRemoteSecondary(atSign, remoteSecondaryAddress, verbose); - } + /** + * Factory to use when you already know the address of the remote (cloud) secondary + * + * @param atSign the atSign of this client + * @param remoteSecondaryAddress the address of the remote secondary server + * @return An {@link AtClient} + * @throws AtException if any other exception occurs while connecting to the remote (cloud) + * secondary + */ + static AtClient withRemoteSecondary(Secondary.Address remoteSecondaryAddress, AtSign atSign) throws AtException { + return withRemoteSecondary(atSign, remoteSecondaryAddress, false); + } - /** - * Factory to use when you already know the address of the remote (cloud) secondary - * @param atSign the atSign of this client - * @param remoteSecondaryAddress the address of the remote secondary server - * @param verbose set to true for chatty logs - * @return An {@link AtClient} - * @throws AtException if any other exception occurs while connecting to the remote (cloud) secondary - */ - static AtClient withRemoteSecondary(AtSign atSign, Secondary.Address remoteSecondaryAddress, boolean verbose) throws AtException { - DefaultAtConnectionFactory connectionFactory = new DefaultAtConnectionFactory(); - AtEvents.AtEventBus eventBus = new SimpleAtEventBus(); - - Map keys; - try { - keys = KeysUtil.loadKeys(atSign); - } catch (Exception e) { - throw new AtClientConfigException("Failed to load keys", e); - } - - RemoteSecondary secondary; - try { - secondary = new RemoteSecondary(eventBus, atSign, remoteSecondaryAddress, keys, connectionFactory, verbose); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to create RemoteSecondary", e); - } - - return new AtClientImpl(eventBus, atSign, keys, secondary); - } - /** - * Factory to use when you already know the address of the remote (cloud) secondary - * @param atSign the atSign of this client - * @param remoteSecondaryAddress the address of the remote secondary server - * @return An {@link AtClient} - * @throws AtException if any other exception occurs while connecting to the remote (cloud) secondary - */ - static AtClient withRemoteSecondary(Secondary.Address remoteSecondaryAddress, AtSign atSign) throws AtException { - return withRemoteSecondary(atSign, remoteSecondaryAddress, false); - } + AtSign getAtSign(); + + Secondary getSecondary(); + + Map getEncryptionKeys(); + + CompletableFuture get(SharedKey sharedKey); + + CompletableFuture getBinary(SharedKey sharedKey); + + CompletableFuture put(SharedKey sharedKey, String value); + + CompletableFuture delete(SharedKey sharedKey); + + CompletableFuture get(SelfKey selfKey); + + CompletableFuture getBinary(SelfKey selfKey); + + CompletableFuture put(SelfKey selfKey, String value); + + CompletableFuture delete(SelfKey selfKey); + + CompletableFuture get(PublicKey publicKey); + + CompletableFuture get(PublicKey publicKey, GetRequestOptions getRequestOptions); + + CompletableFuture getBinary(PublicKey publicKey); + + CompletableFuture getBinary(PublicKey publicKey, GetRequestOptions getRequestOptions); + CompletableFuture put(PublicKey publicKey, String value); - AtSign getAtSign(); - Secondary getSecondary(); - Map getEncryptionKeys(); + CompletableFuture delete(PublicKey publicKey); - CompletableFuture get(SharedKey sharedKey); - CompletableFuture getBinary(SharedKey sharedKey); - CompletableFuture put(SharedKey sharedKey, String value); - CompletableFuture delete(SharedKey sharedKey); + CompletableFuture put(SharedKey sharedKey, byte[] value); - CompletableFuture get(SelfKey selfKey); - CompletableFuture getBinary(SelfKey selfKey); - CompletableFuture put(SelfKey selfKey, String value); - CompletableFuture delete(SelfKey selfKey); + CompletableFuture put(SelfKey selfKey, byte[] value); - CompletableFuture get(PublicKey publicKey); - CompletableFuture get(PublicKey publicKey, GetRequestOptions getRequestOptions); - CompletableFuture getBinary(PublicKey publicKey); - CompletableFuture getBinary(PublicKey publicKey, GetRequestOptions getRequestOptions); - CompletableFuture put(PublicKey publicKey, String value); - CompletableFuture delete(PublicKey publicKey); + CompletableFuture put(PublicKey publicKey, byte[] value); - CompletableFuture put(SharedKey sharedKey, byte[] value); - CompletableFuture put(SelfKey selfKey, byte[] value); - CompletableFuture put(PublicKey publicKey, byte[] value); + CompletableFuture> getAtKeys(String regex); - CompletableFuture> getAtKeys(String regex); - CompletableFuture> getAtKeys(String regex, boolean fetchMetadata); + CompletableFuture> getAtKeys(String regex, boolean fetchMetadata); } diff --git a/at_client/src/main/java/org/atsign/client/api/AtConnection.java b/at_client/src/main/java/org/atsign/client/api/AtConnection.java index a649066f..e34e51b7 100644 --- a/at_client/src/main/java/org/atsign/client/api/AtConnection.java +++ b/at_client/src/main/java/org/atsign/client/api/AtConnection.java @@ -6,33 +6,35 @@ import java.net.Socket; /** - * A simple abstraction around connections to @ platform services - e.g. the root server and secondary servers + * A simple abstraction around connections to @ platform services - e.g. the root server and + * secondary servers */ @SuppressWarnings("unused") public interface AtConnection { - String getUrl(); + String getUrl(); - String getHost(); + String getHost(); - int getPort(); + int getPort(); - Socket getSocket(); + Socket getSocket(); - boolean isConnected(); + boolean isConnected(); - boolean isAutoReconnect(); + boolean isAutoReconnect(); - boolean isVerbose(); + boolean isVerbose(); - void setVerbose(boolean verbose); + void setVerbose(boolean verbose); - void connect() throws IOException, AtException; - void disconnect(); + void connect() throws IOException, AtException; - String executeCommand(String command) throws IOException; + void disconnect(); - interface Authenticator { - void authenticate(AtConnection connection) throws AtException, IOException; - } + String executeCommand(String command) throws IOException; + + interface Authenticator { + void authenticate(AtConnection connection) throws AtException, IOException; + } } diff --git a/at_client/src/main/java/org/atsign/client/api/AtConnectionFactory.java b/at_client/src/main/java/org/atsign/client/api/AtConnectionFactory.java index fb8f88f5..8936769f 100644 --- a/at_client/src/main/java/org/atsign/client/api/AtConnectionFactory.java +++ b/at_client/src/main/java/org/atsign/client/api/AtConnectionFactory.java @@ -5,13 +5,28 @@ import org.atsign.common.AtSign; /** - * For getting a hold of AtConnections to things. - * We inject an AtConnectionFactory into AtClientImpl, primarily for testability + * For getting a hold of AtConnections to things. We inject an AtConnectionFactory into + * AtClientImpl, primarily for testability */ public interface AtConnectionFactory { - AtSecondaryConnection getSecondaryConnection(AtEvents.AtEventBus eventBus, AtSign atSign, Secondary.Address secondaryAddress, AtConnection.Authenticator authenticator); - AtSecondaryConnection getSecondaryConnection(AtEvents.AtEventBus eventBus, AtSign atSign, String secondaryUrl, AtConnection.Authenticator authenticator, boolean verbose); - AtSecondaryConnection getSecondaryConnection(AtEvents.AtEventBus eventBus, AtSign atSign, Secondary.Address secondaryAddress, AtConnection.Authenticator authenticator, boolean verbose); - AtRootConnection getRootConnection(AtEvents.AtEventBus eventBus, String rootUrl); - AtRootConnection getRootConnection(AtEvents.AtEventBus eventBus, String rootUrl, boolean verbose); + AtSecondaryConnection getSecondaryConnection(AtEvents.AtEventBus eventBus, + AtSign atSign, + Secondary.Address secondaryAddress, + AtConnection.Authenticator authenticator); + + AtSecondaryConnection getSecondaryConnection(AtEvents.AtEventBus eventBus, + AtSign atSign, + String secondaryUrl, + AtConnection.Authenticator authenticator, + boolean verbose); + + AtSecondaryConnection getSecondaryConnection(AtEvents.AtEventBus eventBus, + AtSign atSign, + Secondary.Address secondaryAddress, + AtConnection.Authenticator authenticator, + boolean verbose); + + AtRootConnection getRootConnection(AtEvents.AtEventBus eventBus, String rootUrl); + + AtRootConnection getRootConnection(AtEvents.AtEventBus eventBus, String rootUrl, boolean verbose); } diff --git a/at_client/src/main/java/org/atsign/client/api/AtEvents.java b/at_client/src/main/java/org/atsign/client/api/AtEvents.java index 530dd24d..0acdd018 100644 --- a/at_client/src/main/java/org/atsign/client/api/AtEvents.java +++ b/at_client/src/main/java/org/atsign/client/api/AtEvents.java @@ -7,42 +7,42 @@ * */ public interface AtEvents { - interface AtEventListener { - void handleEvent(AtEventType eventType, Map eventData); - } - - interface AtEventBus { - /** - * @param listener to handle various events which originate from Secondaries - * @param eventTypes the set of EventTypes that the listener is interested in - */ - void addEventListener(AtEventListener listener, Set eventTypes); - - /** - * @param listener the listener to remove - */ - void removeEventListener(AtEventListener listener); - - int publishEvent(AtEventType eventType, Map eventData); - - } - enum AtEventType { - // events which originate from the client's "Monitor" connection, which asynchronously - // receives notification events from its secondary server - // Monitor-originating events most interesting to developers: - sharedKeyNotification, // a shared key has been updated by some other atSign - updateNotification, // a value has been updated for some key - deleteNotification, // some key has been deleted - updateNotificationText, // an in-the-clear notification, no encryption - // Monitor-originating events usually only consumed within the AtClient library - statsNotification, - monitorHeartbeatAck, - monitorException, - - // Event which originates from AtClient - AtClient listens for updateNotification events, - // does decryption on the fly, then publishes a decryptedUpdateNotification event - decryptedUpdateNotification, - - userDefined - } + interface AtEventListener { + void handleEvent(AtEventType eventType, Map eventData); + } + + interface AtEventBus { + /** + * @param listener to handle various events which originate from Secondaries + * @param eventTypes the set of EventTypes that the listener is interested in + */ + void addEventListener(AtEventListener listener, Set eventTypes); + + /** + * @param listener the listener to remove + */ + void removeEventListener(AtEventListener listener); + + int publishEvent(AtEventType eventType, Map eventData); + + } + enum AtEventType { + // events which originate from the client's "Monitor" connection, which asynchronously + // receives notification events from its secondary server + // Monitor-originating events most interesting to developers: + sharedKeyNotification, // a shared key has been updated by some other atSign + updateNotification, // a value has been updated for some key + deleteNotification, // some key has been deleted + updateNotificationText, // an in-the-clear notification, no encryption + // Monitor-originating events usually only consumed within the AtClient library + statsNotification, // + monitorHeartbeatAck, // + monitorException, // + + // Event which originates from AtClient - AtClient listens for updateNotification events, + // does decryption on the fly, then publishes a decryptedUpdateNotification event + decryptedUpdateNotification, + + userDefined + } } diff --git a/at_client/src/main/java/org/atsign/client/api/Secondary.java b/at_client/src/main/java/org/atsign/client/api/Secondary.java index 29198a79..be57bc9a 100644 --- a/at_client/src/main/java/org/atsign/client/api/Secondary.java +++ b/at_client/src/main/java/org/atsign/client/api/Secondary.java @@ -8,137 +8,166 @@ /** * Clients ultimately talk to a Secondary server - usually this is a microservice which implements - * the @ protocol server spec, running somewhere in the cloud. - *
- * In the initial implementation we just have AtClientImpl talking to a RemoteSecondary which in turn - * talks, via TLS over a secure socket, to the cloud Secondary server. - *
- * As we implement client-side offline storage, performance caching etc., we can expect e.g. - *
AtClient -> FastCacheSecondary -> OfflineStorageSecondary -> RemoteSecondary
+ * the @ protocol server spec, running somewhere in the cloud.
+ * In the initial implementation we just have AtClientImpl talking to a RemoteSecondary which in + * turn talks, via TLS over a secure socket, to the cloud Secondary server.
+ * As we implement client-side offline storage, performance caching etc., we can expect e.g.
+ * AtClient -> FastCacheSecondary -> OfflineStorageSecondary -> RemoteSecondary
* where FastCacheSecondary might be an in-memory LRU cache, and OfflineStorageSecondary is a - * persistent cache of some or all of the information in the RemoteSecondary. To make this - * possible, each Secondary will need to be able to fully handle the @ protocol, thus the - * interface is effectively the same as when interacting with a cloud secondary via openssl - * from command line. + * persistent cache of some or all of the information in the RemoteSecondary. To make this possible, + * each Secondary will need to be able to fully handle the @ protocol, thus the interface is + * effectively the same as when interacting with a cloud secondary via openssl from command line. */ public interface Secondary extends AtEvents.AtEventListener { - /** - * @param command in @ protocol format - * @param throwExceptionOnErrorResponse sometimes we want to inspect an error response, - * sometimes we want to just throw an exception - * @return response in @ protocol format - * @throws AtException if there was an error response and throwExceptionOnErrorResponse is true - * @throws IOException if one is encountered - */ - Response executeCommand(String command, boolean throwExceptionOnErrorResponse) throws IOException, AtException; - - void startMonitor(); - void stopMonitor(); - boolean isMonitorRunning(); - - class Response { - private String rawDataResponse = null; - private String rawErrorResponse; - private String errorCode; - private String errorText; - - public String getRawDataResponse() { - return rawDataResponse; - } - public void setRawDataResponse(String s) { - rawDataResponse = s; - rawErrorResponse = null; - errorCode = null; - errorText = null; - } - - public String getRawErrorResponse() { - return rawErrorResponse; - } - public void setRawErrorResponse(String s) { - // In format "AT1234-meaning of error code : " - rawErrorResponse = s; - rawDataResponse = null; - - String errorCodeSegment = rawErrorResponse.substring(0, rawErrorResponse.indexOf(":")).trim(); - String[] separatedByHyphen = errorCodeSegment.split("-"); - errorCode = separatedByHyphen[0].trim(); - - errorText = rawErrorResponse.replaceFirst(errorCodeSegment + ":", "").trim(); - } - public boolean isError() {return rawErrorResponse != null;} - - public String getErrorCode() {return errorCode;} - - public String getErrorText() {return errorText;} - - @Override - public String toString() { - if (isError()) { - return "error:" + rawErrorResponse; - } else { - return "data:" + rawDataResponse; - } - } - - public AtException getException() { - if (! isError()) { - return null; - } - if ("AT0001".equals(errorCode)) return new AtServerRuntimeException(errorText); - if ("AT0003".equals(errorCode)) return new AtInvalidSyntaxException(errorText); - if ("AT0005".equals(errorCode)) return new AtBufferOverFlowException(errorText); - if ("AT0006".equals(errorCode)) return new AtOutboundConnectionLimitException(errorText); - if ("AT0007".equals(errorCode)) return new AtSecondaryNotFoundException(errorText); - if ("AT0008".equals(errorCode)) return new AtHandShakeException(errorText); - if ("AT0009".equals(errorCode)) return new AtUnauthorizedException(errorText); - if ("AT0010".equals(errorCode)) return new AtInternalServerError(errorText); - if ("AT0011".equals(errorCode)) return new AtInternalServerException(errorText); - if ("AT0012".equals(errorCode)) return new AtInboundConnectionLimitException(errorText); - if ("AT0013".equals(errorCode)) return new AtBlockedConnectionException(errorText); - if ("AT0015".equals(errorCode)) return new AtKeyNotFoundException(errorText); - if ("AT0016".equals(errorCode)) return new AtInvalidAtKeyException(errorText); - if ("AT0021".equals(errorCode)) return new AtSecondaryConnectException(errorText); - if ("AT0022".equals(errorCode)) return new AtIllegalArgumentException(errorText); - if ("AT0023".equals(errorCode)) return new AtTimeoutException(errorText); - if ("AT0024".equals(errorCode)) return new AtServerIsPausedException(errorText); - if ("AT0401".equals(errorCode)) return new AtUnauthenticatedException(errorText); - - return new AtNewErrorCodeWhoDisException(errorCode, errorText); - } + /** + * @param command in @ protocol format + * @param throwExceptionOnErrorResponse sometimes we want to inspect an error response, sometimes we + * want to just throw an exception + * @return response in @ protocol format + * @throws AtException if there was an error response and throwExceptionOnErrorResponse is true + * @throws IOException if one is encountered + */ + Response executeCommand(String command, boolean throwExceptionOnErrorResponse) throws IOException, AtException; + + void startMonitor(); + + void stopMonitor(); + + boolean isMonitorRunning(); + + class Response { + private String rawDataResponse = null; + private String rawErrorResponse; + private String errorCode; + private String errorText; + + public String getRawDataResponse() { + return rawDataResponse; } - class Address { - public final String host; - public final int port; - - public Address(String host, int port) { - this.host = host; - this.port = port; - } - - public static Address fromString(String hostAndPort) throws IllegalArgumentException { - String[] split = hostAndPort.split(":"); - if (split.length != 2) { - throw new IllegalArgumentException("Cannot construct Secondary.Address from malformed host:port string '" + hostAndPort + "'"); - } - String host = split[0]; - int port; - try { - port = Integer.parseInt(split[1]); - } catch (NumberFormatException e) { - throw new IllegalArgumentException("Cannot construct Secondary.Address from malformed host:port string '" + hostAndPort + "'"); - } - return new Address(host, port); - } - - @Override - public String toString() { - return host + ":" + port; - } + public void setRawDataResponse(String s) { + rawDataResponse = s; + rawErrorResponse = null; + errorCode = null; + errorText = null; } - interface AddressFinder { - Address findSecondary(AtSign atSign) throws IOException, AtSecondaryNotFoundException; + public String getRawErrorResponse() { + return rawErrorResponse; } + + public void setRawErrorResponse(String s) { + // In format "AT1234-meaning of error code : " + rawErrorResponse = s; + rawDataResponse = null; + + String errorCodeSegment = rawErrorResponse.substring(0, rawErrorResponse.indexOf(":")).trim(); + String[] separatedByHyphen = errorCodeSegment.split("-"); + errorCode = separatedByHyphen[0].trim(); + + errorText = rawErrorResponse.replaceFirst(errorCodeSegment + ":", "").trim(); + } + + public boolean isError() { + return rawErrorResponse != null; + } + + public String getErrorCode() { + return errorCode; + } + + public String getErrorText() { + return errorText; + } + + @Override + public String toString() { + if (isError()) { + return "error:" + rawErrorResponse; + } else { + return "data:" + rawDataResponse; + } + } + + public AtException getException() { + if (!isError()) { + return null; + } + if ("AT0001".equals(errorCode)) { + return new AtServerRuntimeException(errorText); + } else if ("AT0003".equals(errorCode)) { + return new AtInvalidSyntaxException(errorText); + } else if ("AT0005".equals(errorCode)) { + return new AtBufferOverFlowException(errorText); + } else if ("AT0006".equals(errorCode)) { + return new AtOutboundConnectionLimitException(errorText); + } else if ("AT0007".equals(errorCode)) { + return new AtSecondaryNotFoundException(errorText); + } else if ("AT0008".equals(errorCode)) { + return new AtHandShakeException(errorText); + } else if ("AT0009".equals(errorCode)) { + return new AtUnauthorizedException(errorText); + } else if ("AT0010".equals(errorCode)) { + return new AtInternalServerError(errorText); + } else if ("AT0011".equals(errorCode)) { + return new AtInternalServerException(errorText); + } else if ("AT0012".equals(errorCode)) { + return new AtInboundConnectionLimitException(errorText); + } else if ("AT0013".equals(errorCode)) { + return new AtBlockedConnectionException(errorText); + } else if ("AT0015".equals(errorCode)) { + return new AtKeyNotFoundException(errorText); + } else if ("AT0016".equals(errorCode)) { + return new AtInvalidAtKeyException(errorText); + } else if ("AT0021".equals(errorCode)) { + return new AtSecondaryConnectException(errorText); + } else if ("AT0022".equals(errorCode)) { + return new AtIllegalArgumentException(errorText); + } else if ("AT0023".equals(errorCode)) { + return new AtTimeoutException(errorText); + } else if ("AT0024".equals(errorCode)) { + return new AtServerIsPausedException(errorText); + } else if ("AT0401".equals(errorCode)) { + return new AtUnauthenticatedException(errorText); + } + + return new AtNewErrorCodeWhoDisException(errorCode, errorText); + } + } + + class Address { + public final String host; + public final int port; + + public Address(String host, int port) { + this.host = host; + this.port = port; + } + + public static Address fromString(String hostAndPort) throws IllegalArgumentException { + String[] split = hostAndPort.split(":"); + if (split.length != 2) { + throw new IllegalArgumentException( + "Cannot construct Secondary.Address from malformed host:port string '" + hostAndPort + "'"); + } + String host = split[0]; + int port; + try { + port = Integer.parseInt(split[1]); + } catch (NumberFormatException e) { + throw new IllegalArgumentException( + "Cannot construct Secondary.Address from malformed host:port string '" + hostAndPort + "'"); + } + return new Address(host, port); + } + + @Override + public String toString() { + return host + ":" + port; + } + } + + interface AddressFinder { + Address findSecondary(AtSign atSign) throws IOException, AtSecondaryNotFoundException; + } } diff --git a/at_client/src/main/java/org/atsign/client/api/impl/clients/AtClientImpl.java b/at_client/src/main/java/org/atsign/client/api/impl/clients/AtClientImpl.java index f30b643c..6ed4871f 100644 --- a/at_client/src/main/java/org/atsign/client/api/impl/clients/AtClientImpl.java +++ b/at_client/src/main/java/org/atsign/client/api/impl/clients/AtClientImpl.java @@ -1,7 +1,18 @@ package org.atsign.client.api.impl.clients; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; +import static org.atsign.client.api.AtEvents.AtEventType.decryptedUpdateNotification; + +import java.io.IOException; +import java.security.*; +import java.time.OffsetDateTime; +import java.util.*; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; + +import javax.crypto.BadPaddingException; +import javax.crypto.IllegalBlockSizeException; +import javax.crypto.NoSuchPaddingException; + import org.atsign.client.api.AtClient; import org.atsign.client.api.AtEvents.AtEventBus; import org.atsign.client.api.AtEvents.AtEventListener; @@ -19,748 +30,797 @@ import org.atsign.common.options.GetRequestOptions; import org.atsign.common.response_models.LookupResponse; -import javax.crypto.BadPaddingException; -import javax.crypto.IllegalBlockSizeException; -import javax.crypto.NoSuchPaddingException; -import java.io.IOException; -import java.security.InvalidAlgorithmParameterException; -import java.security.InvalidKeyException; -import java.security.NoSuchAlgorithmException; -import java.security.NoSuchProviderException; -import java.time.OffsetDateTime; -import java.util.*; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionException; - -import static org.atsign.client.api.AtEvents.AtEventType.decryptedUpdateNotification; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; /** * @see org.atsign.client.api.AtClient */ @SuppressWarnings({"RedundantThrows", "unused"}) public class AtClientImpl implements AtClient { - static final ObjectMapper json = new ObjectMapper(); - - // Factory method - creates an AtClientImpl with a RemoteSecondary - - private final AtSign atSign; - @Override public AtSign getAtSign() {return atSign;} - - private final Map keys; - @Override public Map getEncryptionKeys() {return keys;} - private final Secondary secondary; - @Override public Secondary getSecondary() {return secondary;} - - private final AtEventBus eventBus; - public AtClientImpl(AtEventBus eventBus, AtSign atSign, Map keys, Secondary secondary) { - this.eventBus = eventBus; - this.atSign = atSign; - this.keys = keys; - this.secondary = secondary; - - eventBus.addEventListener(this, EnumSet.allOf(AtEventType.class)); - } - - @Override public void startMonitor() {secondary.startMonitor();} - @Override public void stopMonitor() {secondary.stopMonitor();} - @Override public boolean isMonitorRunning() {return secondary.isMonitorRunning();} - - @Override - public synchronized void addEventListener(AtEventListener listener, Set eventTypes) { - eventBus.addEventListener(listener, eventTypes); - } - - @Override - public synchronized void removeEventListener(AtEventListener listener) { - eventBus.removeEventListener(listener); - } - - @Override - public int publishEvent(AtEventType eventType, Map eventData) { - return eventBus.publishEvent(eventType, eventData); - } - - @Override - public synchronized void handleEvent(AtEventType eventType, Map eventData) { - switch (eventType) { - case sharedKeyNotification: { - // We've got notification that someone has shared an encryption key with us - // If we also got a value, we can decrypt it and add it to our keys map - // Note: a value isn't supplied when the ttr on the shared key was set to 0 - if (eventData.get("value") != null) { - String sharedSharedKeyName = (String) eventData.get("key"); - String sharedSharedKeyEncryptedValue = (String) eventData.get("value"); - // decrypt it with our encryption private key - try { - String sharedKeyDecryptedValue = EncryptionUtil.rsaDecryptFromBase64(sharedSharedKeyEncryptedValue, keys.get(KeysUtil.encryptionPrivateKeyName)); - keys.put(sharedSharedKeyName, sharedKeyDecryptedValue); - } catch (Exception e) { - System.err.println(OffsetDateTime.now() + ": caught exception " + e + " while decrypting received shared key " + sharedSharedKeyName); - } - } - } - break; - case updateNotification: { - // Let's see if we can decrypt it on the fly - if (eventData.get("value") != null) { - String key = (String) eventData.get("key"); - String encryptedValue = (String) eventData.get("value"); - @SuppressWarnings("unchecked") Map metadata = (Map) eventData.get("metadata"); - String ivNonce = (String) metadata.get("ivNonce"); - - try { - // decrypt it with the symmetric key that the other atSign shared with me - String encryptionKeySharedByOther = getEncryptionKeySharedByOther(SharedKey.fromString(key)); - - String decryptedValue = EncryptionUtil.aesDecryptFromBase64(encryptedValue, encryptionKeySharedByOther, ivNonce); - HashMap newEventData = new HashMap<>(eventData); - newEventData.put("decryptedValue", decryptedValue); - eventBus.publishEvent(decryptedUpdateNotification, newEventData); - } catch (Exception e) { - System.err.println(OffsetDateTime.now() + ": caught exception " + e + " while decrypting received data with key name [" + key + "]"); - } - } - } - break; - default: - break; - } - } - - @Override - public CompletableFuture get(SharedKey sharedKey) { - return CompletableFuture.supplyAsync(() -> { - try { - return _get(sharedKey); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture getBinary(SharedKey sharedKey) { - return CompletableFuture.supplyAsync(() -> { - try { - return _getBinary(sharedKey); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture put(SharedKey sharedKey, String value) { - return CompletableFuture.supplyAsync(() -> { - try { - return _put(sharedKey, value); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture delete(SharedKey sharedKey) { - return CompletableFuture.supplyAsync(() -> { - try { - return _delete(sharedKey); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture get(SelfKey selfKey) { - return CompletableFuture.supplyAsync(() -> { - try { - return _get(selfKey); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture getBinary(SelfKey selfKey) { - return CompletableFuture.supplyAsync(() -> { - try { - return _getBinary(selfKey); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture put(SelfKey selfKey, String value) { - return CompletableFuture.supplyAsync(() -> { - try { - return _put(selfKey, value); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture delete(SelfKey selfKey) { - return CompletableFuture.supplyAsync(() -> { - try { - return _delete(selfKey); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture get(PublicKey publicKey) { - return CompletableFuture.supplyAsync(() -> { - try { - return _get(publicKey); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture get(PublicKey publicKey, GetRequestOptions getRequestOptions) { - return CompletableFuture.supplyAsync(() -> { - try { - return _get(publicKey, getRequestOptions); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture getBinary(PublicKey publicKey) { - return CompletableFuture.supplyAsync(() -> { - try { - return _getBinary(publicKey); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture getBinary(PublicKey publicKey, GetRequestOptions getRequestOptions) { - return CompletableFuture.supplyAsync(() -> { - try { - return _getBinary(publicKey, getRequestOptions); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture put(PublicKey publicKey, String value) { - return CompletableFuture.supplyAsync(() -> { - try { - return _put(publicKey, value); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture delete(PublicKey publicKey) { - return CompletableFuture.supplyAsync(() -> { - try { - return _delete(publicKey); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture put(SharedKey sharedKey, byte[] value) { - return CompletableFuture.supplyAsync(() -> { - try { - return _put(sharedKey, value); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture put(SelfKey selfKey, byte[] value) { - return CompletableFuture.supplyAsync(() -> { - try { - return _put(selfKey, value); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture put(PublicKey publicKey, byte[] value) { - return CompletableFuture.supplyAsync(() -> { - try { - return _put(publicKey, value); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - @Override - public CompletableFuture> getAtKeys(String regex) { - return getAtKeys(regex, true); - } - - @Override - public CompletableFuture> getAtKeys(String regex, boolean fetchMetadata) { - return CompletableFuture.supplyAsync(() -> { - try { - return _getAtKeys(regex, fetchMetadata); - } catch (Exception e) { - throw new CompletionException(e); - } - }); - } - - /** - * Synchronous, talks @-protocol directly to the client's Secondary server - * @param command in @ protocol format - * @param throwExceptionOnErrorResponse sometimes we want to inspect an error response, - * sometimes we want to just throw an exception - * @return a Secondary Response - * @throws AtException if the response from the Secondary starts with 'error:', or - * if there is any other exception - */ - @Override - public Response executeCommand(String command, boolean throwExceptionOnErrorResponse) throws AtException, IOException { - return secondary.executeCommand(command, throwExceptionOnErrorResponse); - } - -// ============================================================================================================================================ - // ============================================================================================================================================ - // ============================================================================================================================================ - - // - // Synchronous methods which do the actual work - // - private String _get(SharedKey sharedKey) throws AtException { - if (sharedKey.sharedBy.toString().equals(atSign.toString())) { - return _getSharedByMeWithOther(sharedKey); - } else { - return _getSharedByOtherWithMe(sharedKey); - } - } - - private String _getSharedByMeWithOther(SharedKey sharedKey) throws AtException { - String shareEncryptionKey = getEncryptionKeySharedByMe(sharedKey); - - // fetch local - e.g. if I'm @bob, I would first "llookup:@alice:some.key.name@bob" - Response rawResponse; - String command = "llookup:" + sharedKey; - try { - rawResponse = secondary.executeCommand(command, true); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + command, e); - } - - try { - return EncryptionUtil.aesDecryptFromBase64(rawResponse.getRawDataResponse(), shareEncryptionKey); - } catch (Exception e) { - throw new AtDecryptionException("Failed to decrypt value with shared encryption key", e); - } - } - - private String _getSharedByOtherWithMe(SharedKey sharedKey) throws AtException { - String what; - String shareEncryptionKey = getEncryptionKeySharedByOther(sharedKey); - - Response rawResponse; - String command = "lookup:" + sharedKey.name; - if(sharedKey.getNamespace() != null && !sharedKey.getNamespace().isEmpty()) { - command += "." + sharedKey.getNamespace(); - } - command += sharedKey.sharedBy.toString(); - try { - rawResponse = secondary.executeCommand(command, true); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + command, e); - } - - what = "decrypt value with shared encryption key"; - try { - return EncryptionUtil.aesDecryptFromBase64(rawResponse.getRawDataResponse(), shareEncryptionKey); - } catch (Exception e) { - throw new AtDecryptionException("Failed to " + what, e); - } - } - - private String _put(SharedKey sharedKey, String value) throws AtException { - if (! this.atSign.equals(sharedKey.sharedBy)) { - throw new AtIllegalArgumentException("sharedBy is [" + sharedKey.sharedBy + "] but should be this client's atSign [" + atSign + "]"); - } - String what = ""; - String cipherText; - try { - what = "fetch/create shared encryption key"; - String shareToEncryptionKey = getEncryptionKeySharedByMe(sharedKey); - - what = "encrypt value with shared encryption key"; - cipherText = EncryptionUtil.aesEncryptToBase64(value, shareToEncryptionKey); - } catch (Exception e) { - throw new AtEncryptionException("Failed to " + what, e); - } - - String command = "update" + sharedKey.metadata.toString() + ":" + sharedKey + " " + cipherText; - + static final ObjectMapper json = new ObjectMapper(); + + // Factory method - creates an AtClientImpl with a RemoteSecondary + + private final AtSign atSign; + + @Override + public AtSign getAtSign() { + return atSign; + } + + private final Map keys; + + @Override + public Map getEncryptionKeys() { + return keys; + } + + private final Secondary secondary; + + @Override + public Secondary getSecondary() { + return secondary; + } + + private final AtEventBus eventBus; + + public AtClientImpl(AtEventBus eventBus, AtSign atSign, Map keys, Secondary secondary) { + this.eventBus = eventBus; + this.atSign = atSign; + this.keys = keys; + this.secondary = secondary; + + eventBus.addEventListener(this, EnumSet.allOf(AtEventType.class)); + } + + @Override + public void startMonitor() { + secondary.startMonitor(); + } + + @Override + public void stopMonitor() { + secondary.stopMonitor(); + } + + @Override + public boolean isMonitorRunning() { + return secondary.isMonitorRunning(); + } + + @Override + public synchronized void addEventListener(AtEventListener listener, Set eventTypes) { + eventBus.addEventListener(listener, eventTypes); + } + + @Override + public synchronized void removeEventListener(AtEventListener listener) { + eventBus.removeEventListener(listener); + } + + @Override + public int publishEvent(AtEventType eventType, Map eventData) { + return eventBus.publishEvent(eventType, eventData); + } + + @Override + public synchronized void handleEvent(AtEventType eventType, Map eventData) { + switch (eventType) { + case sharedKeyNotification: + // We've got notification that someone has shared an encryption key with us + // If we also got a value, we can decrypt it and add it to our keys map + // Note: a value isn't supplied when the ttr on the shared key was set to 0 + if (eventData.get("value") != null) { + String sharedSharedKeyName = (String) eventData.get("key"); + String sharedSharedKeyEncryptedValue = (String) eventData.get("value"); + // decrypt it with our encryption private key + try { + String sharedKeyDecryptedValue = EncryptionUtil.rsaDecryptFromBase64(sharedSharedKeyEncryptedValue, + keys.get(KeysUtil.encryptionPrivateKeyName)); + keys.put(sharedSharedKeyName, sharedKeyDecryptedValue); + } catch (Exception e) { + System.err + .println(OffsetDateTime.now() + ": caught exception " + e + " while decrypting received shared key " + + sharedSharedKeyName); + } + } + break; + case updateNotification: + // Let's see if we can decrypt it on the fly + if (eventData.get("value") != null) { + String key = (String) eventData.get("key"); + String encryptedValue = (String) eventData.get("value"); + @SuppressWarnings("unchecked") + Map metadata = (Map) eventData.get("metadata"); + String ivNonce = (String) metadata.get("ivNonce"); + + try { + // decrypt it with the symmetric key that the other atSign shared with me + String encryptionKeySharedByOther = getEncryptionKeySharedByOther(SharedKey.fromString(key)); + + String decryptedValue = + EncryptionUtil.aesDecryptFromBase64(encryptedValue, encryptionKeySharedByOther, ivNonce); + HashMap newEventData = new HashMap<>(eventData); + newEventData.put("decryptedValue", decryptedValue); + eventBus.publishEvent(decryptedUpdateNotification, newEventData); + } catch (Exception e) { + System.err.println(OffsetDateTime.now() + ": caught exception " + e + + " while decrypting received data with key name [" + key + "]"); + } + } + break; + default: + break; + } + } + + + @Override + public CompletableFuture get(SharedKey sharedKey) { + return CompletableFuture.supplyAsync(() -> { + try { + return _get(sharedKey); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture getBinary(SharedKey sharedKey) { + return CompletableFuture.supplyAsync(() -> { + try { + return _getBinary(sharedKey); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture put(SharedKey sharedKey, String value) { + return CompletableFuture.supplyAsync(() -> { + try { + return _put(sharedKey, value); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture delete(SharedKey sharedKey) { + return CompletableFuture.supplyAsync(() -> { + try { + return _delete(sharedKey); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture get(SelfKey selfKey) { + return CompletableFuture.supplyAsync(() -> { + try { + return _get(selfKey); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture getBinary(SelfKey selfKey) { + return CompletableFuture.supplyAsync(() -> { + try { + return _getBinary(selfKey); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture put(SelfKey selfKey, String value) { + return CompletableFuture.supplyAsync(() -> { + try { + return _put(selfKey, value); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture delete(SelfKey selfKey) { + return CompletableFuture.supplyAsync(() -> { + try { + return _delete(selfKey); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture get(PublicKey publicKey) { + return CompletableFuture.supplyAsync(() -> { + try { + return _get(publicKey); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture get(PublicKey publicKey, GetRequestOptions getRequestOptions) { + return CompletableFuture.supplyAsync(() -> { + try { + return _get(publicKey, getRequestOptions); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture getBinary(PublicKey publicKey) { + return CompletableFuture.supplyAsync(() -> { + try { + return _getBinary(publicKey); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture getBinary(PublicKey publicKey, GetRequestOptions getRequestOptions) { + return CompletableFuture.supplyAsync(() -> { + try { + return _getBinary(publicKey, getRequestOptions); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture put(PublicKey publicKey, String value) { + return CompletableFuture.supplyAsync(() -> { + try { + return _put(publicKey, value); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture delete(PublicKey publicKey) { + return CompletableFuture.supplyAsync(() -> { + try { + return _delete(publicKey); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture put(SharedKey sharedKey, byte[] value) { + return CompletableFuture.supplyAsync(() -> { + try { + return _put(sharedKey, value); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture put(SelfKey selfKey, byte[] value) { + return CompletableFuture.supplyAsync(() -> { + try { + return _put(selfKey, value); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture put(PublicKey publicKey, byte[] value) { + return CompletableFuture.supplyAsync(() -> { + try { + return _put(publicKey, value); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + @Override + public CompletableFuture> getAtKeys(String regex) { + return getAtKeys(regex, true); + } + + @Override + public CompletableFuture> getAtKeys(String regex, boolean fetchMetadata) { + return CompletableFuture.supplyAsync(() -> { + try { + return _getAtKeys(regex, fetchMetadata); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + } + + /** + * Synchronous, talks @-protocol directly to the client's Secondary server + * + * @param command in @ protocol format + * @param throwExceptionOnErrorResponse sometimes we want to inspect an error response, sometimes we + * want to just throw an exception + * @return a Secondary Response + * @throws AtException if the response from the Secondary starts with 'error:', or if there is any + * other exception + */ + @Override + public Response executeCommand(String command, boolean throwExceptionOnErrorResponse) + throws AtException, IOException { + return secondary.executeCommand(command, throwExceptionOnErrorResponse); + } + + // ============================================================================================================================================ + // ============================================================================================================================================ + // ============================================================================================================================================ + + // + // Synchronous methods which do the actual work + // + private String _get(SharedKey sharedKey) throws AtException { + if (sharedKey.sharedBy.toString().equals(atSign.toString())) { + return _getSharedByMeWithOther(sharedKey); + } else { + return _getSharedByOtherWithMe(sharedKey); + } + } + + private String _getSharedByMeWithOther(SharedKey sharedKey) throws AtException { + String shareEncryptionKey = getEncryptionKeySharedByMe(sharedKey); + + // fetch local - e.g. if I'm @bob, I would first "llookup:@alice:some.key.name@bob" + Response rawResponse; + String command = "llookup:" + sharedKey; + try { + rawResponse = secondary.executeCommand(command, true); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + command, e); + } + + try { + return EncryptionUtil.aesDecryptFromBase64(rawResponse.getRawDataResponse(), shareEncryptionKey); + } catch (Exception e) { + throw new AtDecryptionException("Failed to decrypt value with shared encryption key", e); + } + } + + private String _getSharedByOtherWithMe(SharedKey sharedKey) throws AtException { + String what; + String shareEncryptionKey = getEncryptionKeySharedByOther(sharedKey); + + Response rawResponse; + String command = "lookup:" + sharedKey.name; + if (sharedKey.getNamespace() != null && !sharedKey.getNamespace().isEmpty()) { + command += "." + sharedKey.getNamespace(); + } + command += sharedKey.sharedBy.toString(); + try { + rawResponse = secondary.executeCommand(command, true); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + command, e); + } + + what = "decrypt value with shared encryption key"; + try { + return EncryptionUtil.aesDecryptFromBase64(rawResponse.getRawDataResponse(), shareEncryptionKey); + } catch (Exception e) { + throw new AtDecryptionException("Failed to " + what, e); + } + } + + private String _put(SharedKey sharedKey, String value) throws AtException { + if (!this.atSign.equals(sharedKey.sharedBy)) { + throw new AtIllegalArgumentException( + "sharedBy is [" + sharedKey.sharedBy + "] but should be this client's atSign [" + atSign + "]"); + } + String what = ""; + String cipherText; + try { + what = "fetch/create shared encryption key"; + String shareToEncryptionKey = getEncryptionKeySharedByMe(sharedKey); + + what = "encrypt value with shared encryption key"; + cipherText = EncryptionUtil.aesEncryptToBase64(value, shareToEncryptionKey); + } catch (Exception e) { + throw new AtEncryptionException("Failed to " + what, e); + } + + String command = "update" + sharedKey.metadata.toString() + ":" + sharedKey + " " + cipherText; + + try { + return secondary.executeCommand(command, true).toString(); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + command, e); + } + } + + private String _delete(SharedKey sharedKey) throws AtException { + String command = "delete:" + sharedKey; + try { + return secondary.executeCommand(command, true).toString(); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + command, e); + } + } + + private String _get(SelfKey key) throws AtException { + // 1. build command + String command; + LlookupVerbBuilder builder = new LlookupVerbBuilder(); + builder.with(key, LlookupVerbBuilder.Type.ALL); + command = builder.build(); + + // 2. execute command + LookupResponse fetched = getLookupResponse(command); + + // 3. decrypt the value + String decryptedValue; + String encryptedValue = fetched.data; + String selfEncryptionKey = keys.get(KeysUtil.selfEncryptionKeyName); + try { + decryptedValue = EncryptionUtil.aesDecryptFromBase64(encryptedValue, selfEncryptionKey); + } catch (NoSuchPaddingException | NoSuchAlgorithmException | InvalidAlgorithmParameterException + | InvalidKeyException | IllegalBlockSizeException | BadPaddingException | NoSuchProviderException e) { + throw new AtDecryptionException("Failed to " + command, e); + } + + // 4. update metadata. squash the fetchedMetadata with current key.metadata (fetchedMetadata has higher priority) + key.metadata = Metadata.squash(fetched.metaData, key.metadata); + + return decryptedValue; + } + + private String _put(SelfKey selfKey, String value) throws AtException { + // 1. generate dataSignature + selfKey.metadata.dataSignature = generateSignature(value); + + // 2. encrypt data with self encryption key + String cipherText; + try { + cipherText = EncryptionUtil.aesEncryptToBase64(value, keys.get(KeysUtil.selfEncryptionKeyName)); + } catch (NoSuchPaddingException | NoSuchAlgorithmException | InvalidAlgorithmParameterException + | InvalidKeyException | IllegalBlockSizeException | BadPaddingException | NoSuchProviderException e) { + throw new AtEncryptionException("Failed to encrypt value with self encryption key", e); + } + + // 3. update secondary + UpdateVerbBuilder builder = new UpdateVerbBuilder(); + builder.with(selfKey, cipherText); + String command = builder.build(); + try { + return secondary.executeCommand(command, true).toString(); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + command, e); + } + } + + private String _delete(SelfKey key) throws AtException { + // 1. build delete command + DeleteVerbBuilder builder = new DeleteVerbBuilder(); + builder.with(key); + String command = builder.build(); + + // 2. run command + try { + return secondary.executeCommand(command, true).toString(); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + command, e); + } + } + + private String _get(PublicKey key) throws AtException { + return _get(key, null); + } + + private String _get(PublicKey key, GetRequestOptions getRequestOptions) throws AtException { + // 1. build command + String command; + if (atSign.toString().equals(key.sharedBy.toString())) { + // it's a public key created by this client => llookup + LlookupVerbBuilder builder = new LlookupVerbBuilder(); + builder.with(key, LlookupVerbBuilder.Type.ALL); + command = builder.build(); + } else { + // it's a public key created by another => plookup + PlookupVerbBuilder builder = new PlookupVerbBuilder(); + builder.with(key, PlookupVerbBuilder.Type.ALL); + builder.setBypassCache(getRequestOptions != null && getRequestOptions.getBypassCache()); + command = builder.build(); + } + + // 2. run the command + LookupResponse fetched = getLookupResponse(command); + + // 4. update key object metadata + key.metadata = Metadata.squash(fetched.metaData, key.metadata); + key.metadata.isCached = fetched.key.contains("cached:"); + + // 5. return the AtValue + return fetched.data; + } + + private String _put(PublicKey publicKey, String value) throws AtException { + // 1. generate dataSignature + publicKey.metadata.dataSignature = generateSignature(value); + + // 2. build command + String command; + UpdateVerbBuilder builder = new UpdateVerbBuilder(); + builder.with(publicKey, value); + command = builder.build(); + + // 3. run command + try { + return secondary.executeCommand(command, true).toString(); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + command, e); + } + } + + private String _delete(PublicKey key) throws AtException { + // 1. build command + String command; + DeleteVerbBuilder builder = new DeleteVerbBuilder(); + builder.with(key); + command = builder.build(); + + // 2. run command + try { + return secondary.executeCommand(command, true).toString(); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + command, e); + } + } + + private byte[] _getBinary(SharedKey sharedKey) throws AtException { + throw new RuntimeException("Not Implemented"); + } + + private byte[] _getBinary(SelfKey selfKey) throws AtException { + throw new RuntimeException("Not Implemented"); + } + + private byte[] _getBinary(PublicKey publicKey) throws AtException { + throw new RuntimeException("Not Implemented"); + } + + private byte[] _getBinary(PublicKey publicKey, GetRequestOptions getRequestOptions) throws AtException { + throw new RuntimeException("Not Implemented"); + } + + private String _put(SharedKey sharedKey, byte[] value) throws AtException { + throw new RuntimeException("Not Implemented"); + } + + private String _put(SelfKey selfKey, byte[] value) throws AtException { + throw new RuntimeException("Not Implemented"); + } + + private String _put(PublicKey publicKey, byte[] value) throws AtException { + throw new RuntimeException("Not Implemented"); + } + + private List _getAtKeys(String regex, boolean fetchMetadata) throws AtException { + ScanVerbBuilder scanVerbBuilder = new ScanVerbBuilder(); + scanVerbBuilder.setRegex(regex); + scanVerbBuilder.setShowHidden(true); + String scanCommand = scanVerbBuilder.build(); + Response scanRawResponse; + try { + scanRawResponse = executeCommand(scanCommand, true); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + scanCommand, e); + } + ResponseTransformers.ScanResponseTransformer scanResponseTransformer = + new ResponseTransformers.ScanResponseTransformer(); + List rawArray = scanResponseTransformer.transform(scanRawResponse); + List atKeys = new ArrayList<>(); + for (String atKeyRaw : rawArray) { // eg atKeyRaw == @bob:phone@alice + AtKey atKey = Keys.fromString(atKeyRaw); + if (fetchMetadata) { + String llookupCommand = "llookup:meta:" + atKeyRaw; + Response llookupMetaResponse; try { - return secondary.executeCommand(command, true).toString(); + llookupMetaResponse = secondary.executeCommand(llookupCommand, true); } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + command, e); + throw new AtSecondaryConnectException("Failed to execute " + llookupCommand, e); } - } - - private String _delete(SharedKey sharedKey) throws AtException { - String command = "delete:" + sharedKey; - try { - return secondary.executeCommand(command, true).toString(); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + command, e); - } - } - - private String _get(SelfKey key) throws AtException { - // 1. build command - String command; - LlookupVerbBuilder builder = new LlookupVerbBuilder(); - builder.with(key, LlookupVerbBuilder.Type.ALL); - command = builder.build(); - - // 2. execute command - LookupResponse fetched = getLookupResponse(command); - - // 3. decrypt the value - String decryptedValue; - String encryptedValue = fetched.data; - String selfEncryptionKey = keys.get(KeysUtil.selfEncryptionKeyName); - try { - decryptedValue = EncryptionUtil.aesDecryptFromBase64(encryptedValue, selfEncryptionKey); - } catch (NoSuchPaddingException | NoSuchAlgorithmException | InvalidAlgorithmParameterException | InvalidKeyException | IllegalBlockSizeException | BadPaddingException | NoSuchProviderException e) { - throw new AtDecryptionException("Failed to " + command, e); - } - - // 4. update metadata. squash the fetchedMetadata with current key.metadata (fetchedMetadata has higher priority) - key.metadata = Metadata.squash(fetched.metaData, key.metadata); - - return decryptedValue; - } - - private String _put(SelfKey selfKey, String value) throws AtException { - // 1. generate dataSignature - selfKey.metadata.dataSignature = generateSignature(value); - - // 2. encrypt data with self encryption key - String cipherText; - try { - cipherText = EncryptionUtil.aesEncryptToBase64(value, keys.get(KeysUtil.selfEncryptionKeyName)); - } catch (NoSuchPaddingException | NoSuchAlgorithmException | InvalidAlgorithmParameterException | InvalidKeyException | IllegalBlockSizeException | BadPaddingException | NoSuchProviderException e) { - throw new AtEncryptionException("Failed to encrypt value with self encryption key", e); - } - - // 3. update secondary - UpdateVerbBuilder builder = new UpdateVerbBuilder(); - builder.with(selfKey, cipherText); - String command = builder.build(); try { - return secondary.executeCommand(command, true).toString(); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + command, e); - } - } - - private String _delete(SelfKey key) throws AtException { - // 1. build delete command - DeleteVerbBuilder builder = new DeleteVerbBuilder(); - builder.with(key); - String command = builder.build(); - - // 2. run command - try { - return secondary.executeCommand(command, true).toString(); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + command, e); - } - } - - private String _get(PublicKey key) throws AtException { - return _get(key, null); - } - - private String _get(PublicKey key, GetRequestOptions getRequestOptions) throws AtException { - // 1. build command - String command; - if(atSign.toString().equals(key.sharedBy.toString())) { - // it's a public key created by this client => llookup - LlookupVerbBuilder builder = new LlookupVerbBuilder(); - builder.with(key, LlookupVerbBuilder.Type.ALL); - command = builder.build(); - } else { - // it's a public key created by another => plookup - PlookupVerbBuilder builder = new PlookupVerbBuilder(); - builder.with(key, PlookupVerbBuilder.Type.ALL); - builder.setBypassCache(getRequestOptions != null && getRequestOptions.getBypassCache()); - command = builder.build(); - } - - // 2. run the command - LookupResponse fetched = getLookupResponse(command); - - // 4. update key object metadata - key.metadata = Metadata.squash(fetched.metaData, key.metadata); - key.metadata.isCached = fetched.key.contains("cached:"); - - // 5. return the AtValue - return fetched.data; - } - - private String _put(PublicKey publicKey, String value) throws AtException { - // 1. generate dataSignature - publicKey.metadata.dataSignature = generateSignature(value); - - // 2. build command - String command; - UpdateVerbBuilder builder = new UpdateVerbBuilder(); - builder.with(publicKey, value); - command = builder.build(); - - // 3. run command - try { - return secondary.executeCommand(command, true).toString(); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + command, e); - } - } - - private String _delete(PublicKey key) throws AtException { - // 1. build command - String command; - DeleteVerbBuilder builder = new DeleteVerbBuilder(); - builder.with(key); - command = builder.build(); - - // 2. run command - try { - return secondary.executeCommand(command, true).toString(); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + command, e); - } - } - - private byte[] _getBinary(SharedKey sharedKey) throws AtException {throw new RuntimeException("Not Implemented");} - private byte[] _getBinary(SelfKey selfKey) throws AtException {throw new RuntimeException("Not Implemented");} - private byte[] _getBinary(PublicKey publicKey) throws AtException {throw new RuntimeException("Not Implemented");} - private byte[] _getBinary(PublicKey publicKey, GetRequestOptions getRequestOptions) throws AtException {throw new RuntimeException("Not Implemented");} - - private String _put(SharedKey sharedKey, byte[] value) throws AtException {throw new RuntimeException("Not Implemented");} - private String _put(SelfKey selfKey, byte[] value) throws AtException {throw new RuntimeException("Not Implemented");} - private String _put(PublicKey publicKey, byte[] value) throws AtException {throw new RuntimeException("Not Implemented");} - - private List _getAtKeys(String regex, boolean fetchMetadata) throws AtException { - ScanVerbBuilder scanVerbBuilder = new ScanVerbBuilder(); - scanVerbBuilder.setRegex(regex); - scanVerbBuilder.setShowHidden(true); - String scanCommand = scanVerbBuilder.build(); - Response scanRawResponse; - try { - scanRawResponse = executeCommand(scanCommand, true); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + scanCommand, e); - } - ResponseTransformers.ScanResponseTransformer scanResponseTransformer = new ResponseTransformers.ScanResponseTransformer(); - List rawArray = scanResponseTransformer.transform(scanRawResponse); - List atKeys = new ArrayList<>(); - for(String atKeyRaw : rawArray) { // eg atKeyRaw == @bob:phone@alice - AtKey atKey = Keys.fromString(atKeyRaw); - if (fetchMetadata) { - String llookupCommand = "llookup:meta:" + atKeyRaw; - Response llookupMetaResponse; - try { - llookupMetaResponse = secondary.executeCommand(llookupCommand, true); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + llookupCommand, e); - } - try { - atKey.metadata = Metadata.squash(atKey.metadata, Metadata.fromJson(llookupMetaResponse.getRawDataResponse())); // atKey.metadata has priority over llookupMetaRaw.data - } catch (JsonProcessingException e) { - throw new AtResponseHandlingException("Failed to parse JSON " + llookupMetaResponse.getRawDataResponse(), e); - } - } - atKeys.add(atKey); - } - return atKeys; - } - - // ============================================================================================================================================ - // ============================================================================================================================================ - // ============================================================================================================================================ - - // - // Internal utility methods. Will move these to another class later, so that other AtClient implementations can easily use them. - // - - private LookupResponse getLookupResponse(String command) throws AtException { - Response response; - try { - response = secondary.executeCommand(command, true); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + command, e); - } - - // 3. transform the data to a LlookupAllResponse object - LookupResponse fetched; - try { - fetched = json.readValue(response.getRawDataResponse(), LookupResponse.class); + // atKey.metadata has priority over llookupMetaRaw.data + atKey.metadata = Metadata.squash(atKey.metadata, Metadata.fromJson(llookupMetaResponse.getRawDataResponse())); } catch (JsonProcessingException e) { - throw new AtResponseHandlingException("Failed to parse JSON " + response.getRawDataResponse(), e); + throw new AtResponseHandlingException("Failed to parse JSON " + llookupMetaResponse.getRawDataResponse(), e); } - return fetched; + } + atKeys.add(atKey); } + return atKeys; + } - private String getEncryptionKeySharedByMe(SharedKey key) throws AtException { - // llookup:shared_key.bob@alice - Secondary.Response rawResponse; - String toLookup = "shared_key." + key.sharedWith.withoutPrefix() + atSign; + // ============================================================================================================================================ + // ============================================================================================================================================ + // ============================================================================================================================================ - String command = "llookup:" + toLookup; - try { - rawResponse = secondary.executeCommand(command, false); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + command, e); - } - - if (rawResponse.isError()) { - if (rawResponse.getException() instanceof AtKeyNotFoundException) { - // No key found - so we should create one - return createSharedEncryptionKey(key); - } else { - throw rawResponse.getException(); - } - } + // + // Internal utility methods. Will move these to another class later, so that other AtClient + // implementations can easily use them. + // - // When we stored it, we encrypted it with our encryption public key; so we need to decrypt it now with our encryption private key - try { - return EncryptionUtil.rsaDecryptFromBase64(rawResponse.getRawDataResponse(), keys.get(KeysUtil.encryptionPrivateKeyName)); - } catch (Exception e) { - throw new AtDecryptionException("Failed to decrypt " + toLookup, e); - } + private LookupResponse getLookupResponse(String command) throws AtException { + Response response; + try { + response = secondary.executeCommand(command, true); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + command, e); } - private String getEncryptionKeySharedByOther(SharedKey sharedKey) throws AtException { - // Let's see if it's in our in-memory cache - String sharedSharedKeyName = sharedKey.getSharedSharedKeyName(); - String sharedKeyValue = keys.get(sharedSharedKeyName); - if (sharedKeyValue != null) { - return sharedKeyValue; - } + // 3. transform the data to a LlookupAllResponse object + LookupResponse fetched; + try { + fetched = json.readValue(response.getRawDataResponse(), LookupResponse.class); + } catch (JsonProcessingException e) { + throw new AtResponseHandlingException("Failed to parse JSON " + response.getRawDataResponse(), e); + } + return fetched; + } - String what = ""; + private String getEncryptionKeySharedByMe(SharedKey key) throws AtException { + // llookup:shared_key.bob@alice + Secondary.Response rawResponse; + String toLookup = "shared_key." + key.sharedWith.withoutPrefix() + atSign; - // Not in memory so now let's try to fetch from remote - e.g. if I'm @bob, lookup:shared_key@alice - String lookupCommand = "lookup:" + "shared_key" + sharedKey.sharedBy; - Response rawResponse; - try { - rawResponse = secondary.executeCommand(lookupCommand, true); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + lookupCommand, e); - } - - String sharedSharedKeyDecryptedValue; - try { - sharedSharedKeyDecryptedValue = EncryptionUtil.rsaDecryptFromBase64(rawResponse.getRawDataResponse(), keys.get(KeysUtil.encryptionPrivateKeyName)); - } catch (Exception e) { - throw new AtDecryptionException("Failed to decrypt the shared_key with our encryption private key", e); - } - keys.put(sharedSharedKeyName, sharedSharedKeyDecryptedValue); - - return sharedSharedKeyDecryptedValue; + String command = "llookup:" + toLookup; + try { + rawResponse = secondary.executeCommand(command, false); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + command, e); } - private String createSharedEncryptionKey(SharedKey sharedKey) throws AtException { - // We need their public key - String theirPublicEncryptionKey = getPublicEncryptionKey(sharedKey.sharedWith); - if (theirPublicEncryptionKey == null) { - throw new AtKeyNotFoundException(" public key " + sharedKey.sharedWith + " not found but service is running - maybe that AtSign has not yet been onboarded"); - } - - // Cut an AES key - String aesKey; - try { - aesKey = EncryptionUtil.generateAESKeyBase64(); - } catch (Exception e) { - throw new AtEncryptionException("Failed to generate AES key for sharing with " + sharedKey.sharedWith, e); - } - - String what = ""; - try { - // Encrypt key with the other at-sign's publickey and save it @bob:shared_key@alice - what = "encrypt new shared key with their public key"; - String encryptedForOther = EncryptionUtil.rsaEncryptToBase64(aesKey, theirPublicEncryptionKey); - - what = "encrypt new shared key with our public key"; - // Encrypt key with our publickey and save it shared_key.bob@alice - String encryptedForUs = EncryptionUtil.rsaEncryptToBase64(aesKey, keys.get(KeysUtil.encryptionPublicKeyName)); - - what = "save encrypted shared key for us"; - secondary.executeCommand("update:" + "shared_key." + sharedKey.sharedWith.withoutPrefix() + sharedKey.sharedBy - + " " + encryptedForUs, true); - - what = "save encrypted shared key for them"; - int ttr = 24 * 60 * 60 * 1000; - secondary.executeCommand("update:ttr:" + ttr + ":" + sharedKey.sharedWith + ":shared_key" + sharedKey.sharedBy - + " " + encryptedForOther, true); - } catch (Exception e) { - throw new AtEncryptionException("Failed to " + what, e); - } - - return aesKey; + if (rawResponse.isError()) { + if (rawResponse.getException() instanceof AtKeyNotFoundException) { + // No key found - so we should create one + return createSharedEncryptionKey(key); + } else { + throw rawResponse.getException(); + } } - private String getPublicEncryptionKey(AtSign sharedWith) throws AtException { - // plookup:publickey@alice - Secondary.Response rawResponse; + // When we stored it, we encrypted it with our encryption public key; so we need to decrypt it now with our encryption private key + try { + return EncryptionUtil.rsaDecryptFromBase64(rawResponse.getRawDataResponse(), + keys.get(KeysUtil.encryptionPrivateKeyName)); + } catch (Exception e) { + throw new AtDecryptionException("Failed to decrypt " + toLookup, e); + } + } + + private String getEncryptionKeySharedByOther(SharedKey sharedKey) throws AtException { + // Let's see if it's in our in-memory cache + String sharedSharedKeyName = sharedKey.getSharedSharedKeyName(); + + String sharedKeyValue = keys.get(sharedSharedKeyName); + if (sharedKeyValue != null) { + return sharedKeyValue; + } + + String what = ""; + + // Not in memory so now let's try to fetch from remote - e.g. if I'm @bob, lookup:shared_key@alice + String lookupCommand = "lookup:" + "shared_key" + sharedKey.sharedBy; + Response rawResponse; + try { + rawResponse = secondary.executeCommand(lookupCommand, true); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + lookupCommand, e); + } + + String sharedSharedKeyDecryptedValue; + try { + sharedSharedKeyDecryptedValue = EncryptionUtil.rsaDecryptFromBase64(rawResponse.getRawDataResponse(), + keys.get(KeysUtil.encryptionPrivateKeyName)); + } catch (Exception e) { + throw new AtDecryptionException("Failed to decrypt the shared_key with our encryption private key", e); + } + keys.put(sharedSharedKeyName, sharedSharedKeyDecryptedValue); + + return sharedSharedKeyDecryptedValue; + } + + private String createSharedEncryptionKey(SharedKey sharedKey) throws AtException { + // We need their public key + String theirPublicEncryptionKey = getPublicEncryptionKey(sharedKey.sharedWith); + if (theirPublicEncryptionKey == null) { + throw new AtKeyNotFoundException(" public key " + sharedKey.sharedWith + + " not found but service is running - maybe that AtSign has not yet been onboarded"); + } + + // Cut an AES key + String aesKey; + try { + aesKey = EncryptionUtil.generateAESKeyBase64(); + } catch (Exception e) { + throw new AtEncryptionException("Failed to generate AES key for sharing with " + sharedKey.sharedWith, e); + } + + String what = ""; + try { + // Encrypt key with the other at-sign's publickey and save it @bob:shared_key@alice + what = "encrypt new shared key with their public key"; + String encryptedForOther = EncryptionUtil.rsaEncryptToBase64(aesKey, theirPublicEncryptionKey); + + what = "encrypt new shared key with our public key"; + // Encrypt key with our publickey and save it shared_key.bob@alice + String encryptedForUs = EncryptionUtil.rsaEncryptToBase64(aesKey, keys.get(KeysUtil.encryptionPublicKeyName)); + + what = "save encrypted shared key for us"; + secondary.executeCommand("update:" + "shared_key." + sharedKey.sharedWith.withoutPrefix() + sharedKey.sharedBy + + " " + encryptedForUs, true); + + what = "save encrypted shared key for them"; + int ttr = 24 * 60 * 60 * 1000; + secondary.executeCommand("update:ttr:" + ttr + ":" + sharedKey.sharedWith + ":shared_key" + sharedKey.sharedBy + + " " + encryptedForOther, true); + } catch (Exception e) { + throw new AtEncryptionException("Failed to " + what, e); + } - String command = "plookup:publickey" + sharedWith; - try { - rawResponse = secondary.executeCommand(command, false); - } catch (IOException e) { - throw new AtSecondaryConnectException("Failed to execute " + command, e); - } + return aesKey; + } + + private String getPublicEncryptionKey(AtSign sharedWith) throws AtException { + // plookup:publickey@alice + Secondary.Response rawResponse; + + String command = "plookup:publickey" + sharedWith; + try { + rawResponse = secondary.executeCommand(command, false); + } catch (IOException e) { + throw new AtSecondaryConnectException("Failed to execute " + command, e); + } + + if (rawResponse.isError()) { + if (rawResponse.getException() instanceof AtKeyNotFoundException) { + return null; + } else { + throw rawResponse.getException(); + } + } else { + return rawResponse.getRawDataResponse(); + } + } - if (rawResponse.isError()) { - if (rawResponse.getException() instanceof AtKeyNotFoundException) { - return null; - } else { - throw rawResponse.getException(); - } - } else { - return rawResponse.getRawDataResponse(); - } - } - - private String generateSignature(String value) throws AtException { - String signature; - try { - signature = EncryptionUtil.signSHA256RSA(value, keys.get(KeysUtil.encryptionPrivateKeyName)); - } catch (Exception e) { - throw new AtEncryptionException("Failed to sign value: " + value, e); - } - return signature; + private String generateSignature(String value) throws AtException { + String signature; + try { + signature = EncryptionUtil.signSHA256RSA(value, keys.get(KeysUtil.encryptionPrivateKeyName)); + } catch (Exception e) { + throw new AtEncryptionException("Failed to sign value: " + value, e); } + return signature; + } } diff --git a/at_client/src/main/java/org/atsign/client/api/impl/connections/AtConnectionBase.java b/at_client/src/main/java/org/atsign/client/api/impl/connections/AtConnectionBase.java index ddd6cadf..3bc302bd 100644 --- a/at_client/src/main/java/org/atsign/client/api/impl/connections/AtConnectionBase.java +++ b/at_client/src/main/java/org/atsign/client/api/impl/connections/AtConnectionBase.java @@ -15,130 +15,169 @@ * @see org.atsign.client.api.AtConnection */ public abstract class AtConnectionBase implements AtConnection { - private final String url; - @Override - public String getUrl() { return url; } - - private final String host; - @Override - public String getHost() { return host; } - - private final int port; - @Override - public int getPort() { return port; } - - private Socket socket; - @Override - public Socket getSocket() { return socket; } - - private boolean connected = false; - @Override - public boolean isConnected() { return connected; } - - private final boolean autoReconnect; - @Override - public boolean isAutoReconnect() {return autoReconnect;} - - protected boolean verbose; - @Override - public boolean isVerbose() {return verbose;} - @Override - public void setVerbose(boolean verbose) {this.verbose = verbose;} - - protected final Authenticator authenticator; - public Authenticator getAuthenticator() {return authenticator;} - - protected PrintWriter socketWriter; - protected Scanner socketScanner; - - protected final AtEvents.AtEventBus eventBus; - public AtConnectionBase(AtEvents.AtEventBus eventBus, String url, AtConnection.Authenticator authenticator, boolean autoReconnect, boolean verbose) { - this.eventBus = eventBus; - this.url = url; - this.host = url.split(":")[0]; - this.port = Integer.parseInt(url.split(":")[1]); - this.autoReconnect = autoReconnect; - this.verbose = verbose; - this.authenticator = authenticator; + private final String url; + + @Override + public String getUrl() { + return url; + } + + private final String host; + + @Override + public String getHost() { + return host; + } + + private final int port; + + @Override + public int getPort() { + return port; + } + + private Socket socket; + + @Override + public Socket getSocket() { + return socket; + } + + private boolean connected = false; + + @Override + public boolean isConnected() { + return connected; + } + + private final boolean autoReconnect; + + @Override + public boolean isAutoReconnect() { + return autoReconnect; + } + + protected boolean verbose; + + @Override + public boolean isVerbose() { + return verbose; + } + + @Override + public void setVerbose(boolean verbose) { + this.verbose = verbose; + } + + protected final Authenticator authenticator; + + public Authenticator getAuthenticator() { + return authenticator; + } + + protected PrintWriter socketWriter; + protected Scanner socketScanner; + + protected final AtEvents.AtEventBus eventBus; + + public AtConnectionBase(AtEvents.AtEventBus eventBus, + String url, + AtConnection.Authenticator authenticator, + boolean autoReconnect, + boolean verbose) { + this.eventBus = eventBus; + this.url = url; + this.host = url.split(":")[0]; + this.port = Integer.parseInt(url.split(":")[1]); + this.autoReconnect = autoReconnect; + this.verbose = verbose; + this.authenticator = authenticator; + } + + @Override + public synchronized void disconnect() { + if (!isConnected()) { + return; } + connected = false; + try { + System.err.println(this.getClass().getSimpleName() + " disconnecting"); + socket.close(); + socketScanner.close(); + socketWriter.close(); + socket.shutdownInput(); + socket.shutdownOutput(); + } catch (Exception ignore) { + } + } - @Override - public synchronized void disconnect() { - if (! isConnected()) { - return; - } - connected = false; - try { - System.err.println(this.getClass().getSimpleName() + " disconnecting"); - socket.close(); - socketScanner.close(); - socketWriter.close(); - socket.shutdownInput(); - socket.shutdownOutput(); - } catch (Exception ignore) { - } + @Override + public synchronized void connect() throws IOException, AtException { + if (isConnected()) { + return; } - @Override - public synchronized void connect() throws IOException, AtException { - if (isConnected()) { - return; - } - SocketFactory sf = SSLSocketFactory.getDefault(); - this.socket = sf.createSocket(host, port); - this.socketWriter = new PrintWriter(socket.getOutputStream()); - this.socketScanner = new Scanner(socket.getInputStream()); + SocketFactory sf = SSLSocketFactory.getDefault(); + this.socket = sf.createSocket(host, port); + this.socketWriter = new PrintWriter(socket.getOutputStream()); + this.socketScanner = new Scanner(socket.getInputStream()); - if (authenticator != null) { - authenticator.authenticate(this); - } - connected = true; + if (authenticator != null) { + authenticator.authenticate(this); } + connected = true; + } + + protected abstract String parseRawResponse(String rawResponse) throws IOException; - protected abstract String parseRawResponse(String rawResponse) throws IOException; + @Override + public final synchronized String executeCommand(String command) throws IOException { + return executeCommand(command, autoReconnect, true); + } - @Override - public final synchronized String executeCommand(String command) throws IOException { - return executeCommand(command, autoReconnect, true); + protected synchronized String executeCommand(String command, boolean retryOnException, boolean readTheResponse) + throws IOException { + if (socket.isClosed()) { + throw new IOException("executeCommand failed: socket is closed"); } - protected synchronized String executeCommand(String command, boolean retryOnException, boolean readTheResponse) throws IOException { - if (socket.isClosed()) { - throw new IOException("executeCommand failed: socket is closed"); + try { + if (!command.endsWith("\n")) { + command = command + "\n"; + } + socketWriter.write(command); + socketWriter.flush(); + + if (verbose) { + System.out.println("\tSENT: " + command.trim()); + } + + if (readTheResponse) { + // Responses are always terminated by newline + String rawResponse = socketScanner.nextLine(); + if (verbose) { + System.out.println("\tRCVD: " + rawResponse); } + + return parseRawResponse(rawResponse); + } else { + return ""; + } + } catch (Exception first) { + disconnect(); + + if (retryOnException) { + System.err.println("\tCaught exception " + first + " : reconnecting"); try { - if (! command.endsWith("\n")) { - command = command + "\n"; - } - socketWriter.write(command); - socketWriter.flush(); - - if (verbose) System.out.println("\tSENT: " + command.trim()); - - if (readTheResponse) { - // Responses are always terminated by newline - String rawResponse = socketScanner.nextLine(); - if (verbose) System.out.println("\tRCVD: " + rawResponse); - - return parseRawResponse(rawResponse); - } else { - return ""; - } - } catch (Exception first) { - disconnect(); - - if (retryOnException) { - System.err.println("\tCaught exception " + first + " : reconnecting"); - try { - connect(); - return executeCommand(command, false, true); - } catch (Exception second) { - second.printStackTrace(System.err); - throw new IOException("Failed to reconnect after original exception " + first + " : ", second); - } - } else { - connected = false; - - throw new IOException(first); - } + connect(); + return executeCommand(command, false, true); + } catch (Exception second) { + second.printStackTrace(System.err); + throw new IOException("Failed to reconnect after original exception " + first + " : ", second); } + } else { + connected = false; + + throw new IOException(first); + } } + } } diff --git a/at_client/src/main/java/org/atsign/client/api/impl/connections/AtMonitorConnection.java b/at_client/src/main/java/org/atsign/client/api/impl/connections/AtMonitorConnection.java index 5ee0bd05..855c3630 100644 --- a/at_client/src/main/java/org/atsign/client/api/impl/connections/AtMonitorConnection.java +++ b/at_client/src/main/java/org/atsign/client/api/impl/connections/AtMonitorConnection.java @@ -12,222 +12,237 @@ * */ public class AtMonitorConnection extends AtSecondaryConnection implements Runnable { - private static final ObjectMapper mapper = new ObjectMapper(); - - private long _lastReceivedTime = 0; - public long getLastReceivedTime() {return _lastReceivedTime;} - public void setLastReceivedTime(long lastReceivedTime) {this._lastReceivedTime = lastReceivedTime;} - - private boolean running = false; - public boolean isRunning() { return running; } - - private boolean _shouldBeRunning = false; - private void setShouldBeRunning(boolean b) { - _shouldBeRunning = b; - } - - public boolean isShouldBeRunning() { - return _shouldBeRunning; - } - - public AtMonitorConnection( - AtEventBus eventBus, - AtSign atSign, - String secondaryUrl, - Authenticator authenticator, - boolean verbose) { - // Note that the Monitor doesn't make use of the auto-reconnect functionality, it does its own thing - super(eventBus, atSign, secondaryUrl, authenticator, false, verbose); - startHeartbeat(); - } - - private long lastHeartbeatSentTime = System.currentTimeMillis(); - private long lastHeartbeatAckTime = System.currentTimeMillis(); - private final int heartbeatIntervalMillis = 30000; - private void startHeartbeat() { - new Thread(() -> { - while (true) { - if (isShouldBeRunning()) { - if (!isRunning() || lastHeartbeatSentTime - lastHeartbeatAckTime >= heartbeatIntervalMillis) { - try { - // heartbeats have stopped being acked - System.err.println("Monitor heartbeats not being received"); - stopMonitor(); - long waitStartTime = System.currentTimeMillis(); - while (isRunning() && System.currentTimeMillis() - waitStartTime < 5000) { - // wait for monitor to stop - try { - //noinspection BusyWait - Thread.sleep(1000); - } catch (Exception ignore) {} - } - if (isRunning()) { - System.err.println("Monitor thread has not stopped, but going to start another one anyway"); - } - startMonitor(); - } catch (Exception e) { - System.err.println("Monitor restart failed " + e); - e.printStackTrace(System.err); - } - } else { - if (System.currentTimeMillis() - lastHeartbeatSentTime > heartbeatIntervalMillis) { - try { - executeCommand("noop:0", false, false); - lastHeartbeatSentTime = System.currentTimeMillis(); - } catch (Exception ignore) { - // Can't do anything, the heartbeat loop will take care of restarting the monitor connection - } - } - } - } - try { - //noinspection BusyWait - Thread.sleep(heartbeatIntervalMillis / 5); - } catch (Exception ignore) {} - } - }).start(); - } - - /** - * @return true if the monitor start request has succeeded, or if the monitor is already running. - */ - @SuppressWarnings("UnusedReturnValue") - public synchronized boolean startMonitor() { - lastHeartbeatSentTime = lastHeartbeatAckTime = System.currentTimeMillis(); - - setShouldBeRunning(true); - if (! running) { - running = true; - if (!isConnected()) { + private static final ObjectMapper mapper = new ObjectMapper(); + + private long _lastReceivedTime = 0; + + public long getLastReceivedTime() { + return _lastReceivedTime; + } + + public void setLastReceivedTime(long lastReceivedTime) { + this._lastReceivedTime = lastReceivedTime; + } + + private boolean running = false; + + public boolean isRunning() { + return running; + } + + private boolean _shouldBeRunning = false; + + private void setShouldBeRunning(boolean b) { + _shouldBeRunning = b; + } + + public boolean isShouldBeRunning() { + return _shouldBeRunning; + } + + public AtMonitorConnection(AtEventBus eventBus, + AtSign atSign, + String secondaryUrl, + Authenticator authenticator, + boolean verbose) { + // Note that the Monitor doesn't make use of the auto-reconnect functionality, it does its own thing + super(eventBus, atSign, secondaryUrl, authenticator, false, verbose); + startHeartbeat(); + } + + private long lastHeartbeatSentTime = System.currentTimeMillis(); + private long lastHeartbeatAckTime = System.currentTimeMillis(); + private final int heartbeatIntervalMillis = 30000; + + private void startHeartbeat() { + new Thread(() -> { + while (true) { + if (isShouldBeRunning()) { + if (!isRunning() || lastHeartbeatSentTime - lastHeartbeatAckTime >= heartbeatIntervalMillis) { + try { + // heartbeats have stopped being acked + System.err.println("Monitor heartbeats not being received"); + stopMonitor(); + long waitStartTime = System.currentTimeMillis(); + while (isRunning() && System.currentTimeMillis() - waitStartTime < 5000) { + // wait for monitor to stop try { - connect(); - } catch (Exception e) { - System.err.println("startMonitor failed to connect to secondary : " + e.getMessage()); - running = false; - return false; + // noinspection BusyWait + Thread.sleep(1000); + } catch (Exception ignore) { } + } + if (isRunning()) { + System.err.println("Monitor thread has not stopped, but going to start another one anyway"); + } + startMonitor(); + } catch (Exception e) { + System.err.println("Monitor restart failed " + e); + e.printStackTrace(System.err); + } + } else { + if (System.currentTimeMillis() - lastHeartbeatSentTime > heartbeatIntervalMillis) { + try { + executeCommand("noop:0", false, false); + lastHeartbeatSentTime = System.currentTimeMillis(); + } catch (Exception ignore) { + // Can't do anything, the heartbeat loop will take care of restarting the monitor connection + } } - new Thread(this).start(); + } } - return true; - } - - public synchronized void stopMonitor() { - setShouldBeRunning(false); - lastHeartbeatSentTime = lastHeartbeatAckTime = System.currentTimeMillis(); - disconnect(); - } - - /** - * Please don't call this directly. Call startMonitor() instead, which starts the monitor in its own thread - */ - @SuppressWarnings("unchecked") - @Override - public void run() { - String what = ""; - // call executeCommand("monitor: