Use the S3 object monitor to retrieve Tor exit node lists.
This commit is contained in:
parent
cfa8cbedc1
commit
fbaf4a09e2
|
@ -397,7 +397,6 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
|
||||||
ExecutorService gcmSenderExecutor = environment.lifecycle().executorService(name(getClass(), "gcmSender-%d")).maxThreads(1).minThreads(1).build();
|
ExecutorService gcmSenderExecutor = environment.lifecycle().executorService(name(getClass(), "gcmSender-%d")).maxThreads(1).minThreads(1).build();
|
||||||
ExecutorService backupServiceExecutor = environment.lifecycle().executorService(name(getClass(), "backupService-%d")).maxThreads(1).minThreads(1).build();
|
ExecutorService backupServiceExecutor = environment.lifecycle().executorService(name(getClass(), "backupService-%d")).maxThreads(1).minThreads(1).build();
|
||||||
ExecutorService storageServiceExecutor = environment.lifecycle().executorService(name(getClass(), "storageService-%d")).maxThreads(1).minThreads(1).build();
|
ExecutorService storageServiceExecutor = environment.lifecycle().executorService(name(getClass(), "storageService-%d")).maxThreads(1).minThreads(1).build();
|
||||||
ExecutorService torExitNodeExecutor = environment.lifecycle().executorService(name(getClass(), "torExitNode-%d")).maxThreads(1).minThreads(1).build();
|
|
||||||
ExecutorService donationExecutor = environment.lifecycle().executorService(name(getClass(), "donation-%d")).maxThreads(1).minThreads(1).build();
|
ExecutorService donationExecutor = environment.lifecycle().executorService(name(getClass(), "donation-%d")).maxThreads(1).minThreads(1).build();
|
||||||
|
|
||||||
ExternalServiceCredentialGenerator directoryCredentialsGenerator = new ExternalServiceCredentialGenerator(config.getDirectoryConfiguration().getDirectoryClientConfiguration().getUserAuthenticationTokenSharedSecret(),
|
ExternalServiceCredentialGenerator directoryCredentialsGenerator = new ExternalServiceCredentialGenerator(config.getDirectoryConfiguration().getDirectoryClientConfiguration().getUserAuthenticationTokenSharedSecret(),
|
||||||
|
@ -437,7 +436,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
|
||||||
GCMSender gcmSender = new GCMSender(gcmSenderExecutor, accountsManager, config.getGcmConfiguration().getApiKey());
|
GCMSender gcmSender = new GCMSender(gcmSenderExecutor, accountsManager, config.getGcmConfiguration().getApiKey());
|
||||||
RateLimiters rateLimiters = new RateLimiters(config.getLimitsConfiguration(), dynamicConfigurationManager, rateLimitersCluster);
|
RateLimiters rateLimiters = new RateLimiters(config.getLimitsConfiguration(), dynamicConfigurationManager, rateLimitersCluster);
|
||||||
ProvisioningManager provisioningManager = new ProvisioningManager(pubSubManager);
|
ProvisioningManager provisioningManager = new ProvisioningManager(pubSubManager);
|
||||||
TorExitNodeManager torExitNodeManager = new TorExitNodeManager(recurringJobExecutor, torExitNodeExecutor, config.getTorExitNodeConfiguration());
|
TorExitNodeManager torExitNodeManager = new TorExitNodeManager(recurringJobExecutor, config.getTorExitNodeConfiguration());
|
||||||
|
|
||||||
AccountAuthenticator accountAuthenticator = new AccountAuthenticator(accountsManager);
|
AccountAuthenticator accountAuthenticator = new AccountAuthenticator(accountsManager);
|
||||||
DisabledPermittedAccountAuthenticator disabledPermittedAccountAuthenticator = new DisabledPermittedAccountAuthenticator(accountsManager);
|
DisabledPermittedAccountAuthenticator disabledPermittedAccountAuthenticator = new DisabledPermittedAccountAuthenticator(accountsManager);
|
||||||
|
|
|
@ -15,37 +15,37 @@ public class TorExitNodeConfiguration {
|
||||||
|
|
||||||
@JsonProperty
|
@JsonProperty
|
||||||
@NotBlank
|
@NotBlank
|
||||||
private String listUrl;
|
private String s3Region;
|
||||||
|
|
||||||
|
@JsonProperty
|
||||||
|
@NotBlank
|
||||||
|
private String s3Bucket;
|
||||||
|
|
||||||
|
@JsonProperty
|
||||||
|
@NotBlank
|
||||||
|
private String objectKey;
|
||||||
|
|
||||||
@JsonProperty
|
@JsonProperty
|
||||||
private Duration refreshInterval = Duration.ofMinutes(5);
|
private Duration refreshInterval = Duration.ofMinutes(5);
|
||||||
|
|
||||||
@JsonProperty
|
public String getS3Region() {
|
||||||
@Valid
|
return s3Region;
|
||||||
private CircuitBreakerConfiguration circuitBreakerConfiguration = new CircuitBreakerConfiguration();
|
|
||||||
|
|
||||||
@JsonProperty
|
|
||||||
@Valid
|
|
||||||
private RetryConfiguration retryConfiguration = new RetryConfiguration();
|
|
||||||
|
|
||||||
public String getListUrl() {
|
|
||||||
return listUrl;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@VisibleForTesting
|
@VisibleForTesting
|
||||||
public void setListUrl(final String listUrl) {
|
public void setS3Region(final String s3Region) {
|
||||||
this.listUrl = listUrl;
|
this.s3Region = s3Region;
|
||||||
|
}
|
||||||
|
|
||||||
|
public String getS3Bucket() {
|
||||||
|
return s3Bucket;
|
||||||
|
}
|
||||||
|
|
||||||
|
public String getObjectKey() {
|
||||||
|
return objectKey;
|
||||||
}
|
}
|
||||||
|
|
||||||
public Duration getRefreshInterval() {
|
public Duration getRefreshInterval() {
|
||||||
return refreshInterval;
|
return refreshInterval;
|
||||||
}
|
}
|
||||||
|
|
||||||
public CircuitBreakerConfiguration getCircuitBreakerConfiguration() {
|
|
||||||
return circuitBreakerConfiguration;
|
|
||||||
}
|
|
||||||
|
|
||||||
public RetryConfiguration getRetryConfiguration() {
|
|
||||||
return retryConfiguration;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -5,32 +5,25 @@
|
||||||
|
|
||||||
package org.whispersystems.textsecuregcm.util;
|
package org.whispersystems.textsecuregcm.util;
|
||||||
|
|
||||||
|
import static com.codahale.metrics.MetricRegistry.name;
|
||||||
|
|
||||||
|
import com.amazonaws.services.s3.model.S3Object;
|
||||||
import com.google.common.annotations.VisibleForTesting;
|
import com.google.common.annotations.VisibleForTesting;
|
||||||
import io.dropwizard.lifecycle.Managed;
|
import io.dropwizard.lifecycle.Managed;
|
||||||
import java.net.URI;
|
|
||||||
import java.net.http.HttpClient;
|
|
||||||
import java.net.http.HttpRequest;
|
|
||||||
import java.net.http.HttpResponse;
|
|
||||||
import java.time.Duration;
|
|
||||||
import java.util.Collections;
|
|
||||||
import java.util.Set;
|
|
||||||
import java.util.concurrent.CompletableFuture;
|
|
||||||
import java.util.concurrent.ExecutorService;
|
|
||||||
import java.util.concurrent.ScheduledExecutorService;
|
|
||||||
import java.util.concurrent.ScheduledFuture;
|
|
||||||
import java.util.concurrent.TimeUnit;
|
|
||||||
import java.util.concurrent.atomic.AtomicReference;
|
|
||||||
import java.util.stream.Collectors;
|
|
||||||
import io.micrometer.core.instrument.Counter;
|
import io.micrometer.core.instrument.Counter;
|
||||||
import io.micrometer.core.instrument.Metrics;
|
import io.micrometer.core.instrument.Metrics;
|
||||||
import io.micrometer.core.instrument.Timer;
|
import io.micrometer.core.instrument.Timer;
|
||||||
import org.apache.commons.lang3.StringUtils;
|
import java.io.BufferedReader;
|
||||||
|
import java.io.InputStream;
|
||||||
|
import java.io.InputStreamReader;
|
||||||
|
import java.util.Collections;
|
||||||
|
import java.util.Set;
|
||||||
|
import java.util.concurrent.ScheduledExecutorService;
|
||||||
|
import java.util.concurrent.atomic.AtomicReference;
|
||||||
|
import java.util.stream.Collectors;
|
||||||
import org.slf4j.Logger;
|
import org.slf4j.Logger;
|
||||||
import org.slf4j.LoggerFactory;
|
import org.slf4j.LoggerFactory;
|
||||||
import org.whispersystems.textsecuregcm.configuration.TorExitNodeConfiguration;
|
import org.whispersystems.textsecuregcm.configuration.TorExitNodeConfiguration;
|
||||||
import org.whispersystems.textsecuregcm.http.FaultTolerantHttpClient;
|
|
||||||
|
|
||||||
import static com.codahale.metrics.MetricRegistry.name;
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* A utility for checking whether IP addresses belong to Tor exit nodes using the "bulk exit list."
|
* A utility for checking whether IP addresses belong to Tor exit nodes using the "bulk exit list."
|
||||||
|
@ -39,14 +32,8 @@ import static com.codahale.metrics.MetricRegistry.name;
|
||||||
*/
|
*/
|
||||||
public class TorExitNodeManager implements Managed {
|
public class TorExitNodeManager implements Managed {
|
||||||
|
|
||||||
private final ScheduledExecutorService refreshScheduledExecutorService;
|
private final S3ObjectMonitor exitListMonitor;
|
||||||
private final Duration refreshDelay;
|
|
||||||
private ScheduledFuture<?> refreshFuture;
|
|
||||||
|
|
||||||
private final FaultTolerantHttpClient refreshClient;
|
|
||||||
private final URI exitNodeListUri;
|
|
||||||
|
|
||||||
private final AtomicReference<String> lastEtag = new AtomicReference<>(null);
|
|
||||||
private final AtomicReference<Set<String>> exitNodeAddresses = new AtomicReference<>(Collections.emptySet());
|
private final AtomicReference<Set<String>> exitNodeAddresses = new AtomicReference<>(Collections.emptySet());
|
||||||
|
|
||||||
private static final Timer REFRESH_TIMER = Metrics.timer(name(TorExitNodeManager.class, "refresh"));
|
private static final Timer REFRESH_TIMER = Metrics.timer(name(TorExitNodeManager.class, "refresh"));
|
||||||
|
@ -54,78 +41,45 @@ public class TorExitNodeManager implements Managed {
|
||||||
|
|
||||||
private static final Logger log = LoggerFactory.getLogger(TorExitNodeManager.class);
|
private static final Logger log = LoggerFactory.getLogger(TorExitNodeManager.class);
|
||||||
|
|
||||||
public TorExitNodeManager(final ScheduledExecutorService scheduledExecutorService, final ExecutorService clientExecutorService, final TorExitNodeConfiguration configuration) {
|
public TorExitNodeManager(
|
||||||
this.refreshScheduledExecutorService = scheduledExecutorService;
|
final ScheduledExecutorService scheduledExecutorService,
|
||||||
|
final TorExitNodeConfiguration configuration) {
|
||||||
|
|
||||||
this.exitNodeListUri = URI.create(configuration.getListUrl());
|
this.exitListMonitor = new S3ObjectMonitor(
|
||||||
this.refreshDelay = configuration.getRefreshInterval();
|
configuration.getS3Region(),
|
||||||
|
configuration.getS3Bucket(),
|
||||||
|
configuration.getObjectKey(),
|
||||||
|
scheduledExecutorService,
|
||||||
|
configuration.getRefreshInterval(),
|
||||||
|
this::handleExitListChanged);
|
||||||
|
}
|
||||||
|
|
||||||
refreshClient = FaultTolerantHttpClient.newBuilder()
|
@Override
|
||||||
.withCircuitBreaker(configuration.getCircuitBreakerConfiguration())
|
public synchronized void start() {
|
||||||
.withRetry(configuration.getRetryConfiguration())
|
exitListMonitor.refresh();
|
||||||
.withVersion(HttpClient.Version.HTTP_1_1)
|
exitListMonitor.start();
|
||||||
.withConnectTimeout(Duration.ofSeconds(10))
|
}
|
||||||
.withRedirect(HttpClient.Redirect.NEVER)
|
|
||||||
.withExecutor(clientExecutorService)
|
@Override
|
||||||
.withName("tor-exit-node")
|
public synchronized void stop() {
|
||||||
.withSecurityProtocol(FaultTolerantHttpClient.SECURITY_PROTOCOL_TLS_1_3)
|
exitListMonitor.stop();
|
||||||
.build();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean isTorExitNode(final String address) {
|
public boolean isTorExitNode(final String address) {
|
||||||
return exitNodeAddresses.get().contains(address);
|
return exitNodeAddresses.get().contains(address);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void handleExitListChanged(final S3Object exitList) {
|
||||||
|
REFRESH_TIMER.record(() -> handleExitListChanged(exitList.getObjectContent()));
|
||||||
|
}
|
||||||
|
|
||||||
@VisibleForTesting
|
@VisibleForTesting
|
||||||
CompletableFuture<?> refresh() {
|
void handleExitListChanged(final InputStream inputStream) {
|
||||||
final String etag = lastEtag.get();
|
try (final BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
|
||||||
|
exitNodeAddresses.set(reader.lines().collect(Collectors.toSet()));
|
||||||
final HttpRequest request;
|
} catch (final Exception e) {
|
||||||
{
|
REFRESH_ERRORS.increment();
|
||||||
final HttpRequest.Builder builder = HttpRequest.newBuilder().GET().uri(exitNodeListUri);
|
log.warn("Failed to refresh Tor exit node list", e);
|
||||||
|
|
||||||
if (StringUtils.isNotBlank(etag)) {
|
|
||||||
builder.header("If-None-Match", etag);
|
|
||||||
}
|
|
||||||
|
|
||||||
request = builder.build();
|
|
||||||
}
|
|
||||||
|
|
||||||
final long start = System.nanoTime();
|
|
||||||
|
|
||||||
return refreshClient.sendAsync(request, HttpResponse.BodyHandlers.ofString())
|
|
||||||
.whenComplete((response, cause) -> {
|
|
||||||
REFRESH_TIMER.record(System.nanoTime() - start, TimeUnit.NANOSECONDS);
|
|
||||||
|
|
||||||
if (cause != null) {
|
|
||||||
REFRESH_ERRORS.increment();
|
|
||||||
log.warn("Failed to refresh Tor exit node list", cause);
|
|
||||||
} else {
|
|
||||||
if (response.statusCode() == 200) {
|
|
||||||
exitNodeAddresses.set(response.body().lines().collect(Collectors.toSet()));
|
|
||||||
response.headers().firstValue("ETag").ifPresent(newEtag -> lastEtag.compareAndSet(etag, newEtag));
|
|
||||||
} else if (response.statusCode() != 304) {
|
|
||||||
REFRESH_ERRORS.increment();
|
|
||||||
log.warn("Failed to refresh Tor exit node list: {} ({})", response.statusCode(), response.body());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public synchronized void start() {
|
|
||||||
if (refreshFuture != null) {
|
|
||||||
refreshFuture.cancel(true);
|
|
||||||
}
|
|
||||||
|
|
||||||
refreshFuture = refreshScheduledExecutorService
|
|
||||||
.scheduleAtFixedRate(this::refresh, 0, refreshDelay.toMillis(), TimeUnit.MILLISECONDS);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public synchronized void stop() {
|
|
||||||
if (refreshFuture != null) {
|
|
||||||
refreshFuture.cancel(true);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -5,89 +5,32 @@
|
||||||
|
|
||||||
package org.whispersystems.textsecuregcm.util;
|
package org.whispersystems.textsecuregcm.util;
|
||||||
|
|
||||||
import com.github.tomakehurst.wiremock.junit.WireMockRule;
|
import static org.junit.Assert.assertFalse;
|
||||||
import com.github.tomakehurst.wiremock.matching.StringValuePattern;
|
import static org.junit.Assert.assertTrue;
|
||||||
import org.junit.After;
|
import static org.mockito.Mockito.mock;
|
||||||
import org.junit.Before;
|
|
||||||
import org.junit.Rule;
|
import java.io.ByteArrayInputStream;
|
||||||
|
import java.nio.charset.StandardCharsets;
|
||||||
|
import java.util.concurrent.ScheduledExecutorService;
|
||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
import org.whispersystems.textsecuregcm.configuration.TorExitNodeConfiguration;
|
import org.whispersystems.textsecuregcm.configuration.TorExitNodeConfiguration;
|
||||||
import org.whispersystems.textsecuregcm.redis.AbstractRedisClusterTest;
|
import org.whispersystems.textsecuregcm.redis.AbstractRedisClusterTest;
|
||||||
|
|
||||||
import java.util.UUID;
|
|
||||||
import java.util.concurrent.ExecutorService;
|
|
||||||
import java.util.concurrent.Executors;
|
|
||||||
import java.util.concurrent.ScheduledExecutorService;
|
|
||||||
import java.util.concurrent.TimeUnit;
|
|
||||||
|
|
||||||
import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
|
|
||||||
import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
|
|
||||||
import static com.github.tomakehurst.wiremock.client.WireMock.get;
|
|
||||||
import static com.github.tomakehurst.wiremock.client.WireMock.getRequestedFor;
|
|
||||||
import static com.github.tomakehurst.wiremock.client.WireMock.post;
|
|
||||||
import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
|
|
||||||
import static com.github.tomakehurst.wiremock.client.WireMock.verify;
|
|
||||||
import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.options;
|
|
||||||
import static org.junit.Assert.*;
|
|
||||||
|
|
||||||
public class TorExitNodeManagerTest extends AbstractRedisClusterTest {
|
public class TorExitNodeManagerTest extends AbstractRedisClusterTest {
|
||||||
|
|
||||||
private ScheduledExecutorService scheduledExecutorService;
|
|
||||||
private ExecutorService clientExecutorService;
|
|
||||||
|
|
||||||
private TorExitNodeManager torExitNodeManager;
|
|
||||||
|
|
||||||
private static final String LIST_PATH = "/list";
|
|
||||||
|
|
||||||
@Rule
|
|
||||||
public WireMockRule wireMockRule = new WireMockRule(options().dynamicPort().dynamicHttpsPort());
|
|
||||||
|
|
||||||
@Before
|
|
||||||
public void setUp() throws Exception {
|
|
||||||
final TorExitNodeConfiguration configuration = new TorExitNodeConfiguration();
|
|
||||||
configuration.setListUrl("http://localhost:" + wireMockRule.port() + LIST_PATH);
|
|
||||||
|
|
||||||
scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
|
|
||||||
clientExecutorService = Executors.newSingleThreadExecutor();
|
|
||||||
|
|
||||||
torExitNodeManager = new TorExitNodeManager(scheduledExecutorService, clientExecutorService, configuration);
|
|
||||||
}
|
|
||||||
|
|
||||||
@After
|
|
||||||
public void tearDown() throws Exception {
|
|
||||||
scheduledExecutorService.shutdown();
|
|
||||||
scheduledExecutorService.awaitTermination(1, TimeUnit.SECONDS);
|
|
||||||
|
|
||||||
clientExecutorService.shutdown();
|
|
||||||
clientExecutorService.awaitTermination(1, TimeUnit.SECONDS);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void testIsTorExitNode() {
|
public void testIsTorExitNode() {
|
||||||
|
final TorExitNodeConfiguration configuration = new TorExitNodeConfiguration();
|
||||||
|
configuration.setS3Region("ap-northeast-3");
|
||||||
|
|
||||||
|
final TorExitNodeManager torExitNodeManager =
|
||||||
|
new TorExitNodeManager(mock(ScheduledExecutorService.class), configuration);
|
||||||
|
|
||||||
assertFalse(torExitNodeManager.isTorExitNode("10.0.0.1"));
|
assertFalse(torExitNodeManager.isTorExitNode("10.0.0.1"));
|
||||||
assertFalse(torExitNodeManager.isTorExitNode("10.0.0.2"));
|
assertFalse(torExitNodeManager.isTorExitNode("10.0.0.2"));
|
||||||
|
|
||||||
final String etag = UUID.randomUUID().toString();
|
torExitNodeManager.handleExitListChanged(
|
||||||
|
new ByteArrayInputStream("10.0.0.1\n10.0.0.2".getBytes(StandardCharsets.UTF_8)));
|
||||||
wireMockRule.stubFor(get(urlEqualTo(LIST_PATH))
|
|
||||||
.willReturn(aResponse()
|
|
||||||
.withHeader("ETag", etag)
|
|
||||||
.withBody("10.0.0.1\n10.0.0.2")));
|
|
||||||
|
|
||||||
torExitNodeManager.refresh().join();
|
|
||||||
|
|
||||||
verify(getRequestedFor(urlEqualTo(LIST_PATH)).withoutHeader("If-None-Match"));
|
|
||||||
|
|
||||||
assertTrue(torExitNodeManager.isTorExitNode("10.0.0.1"));
|
|
||||||
assertTrue(torExitNodeManager.isTorExitNode("10.0.0.2"));
|
|
||||||
assertFalse(torExitNodeManager.isTorExitNode("10.0.0.3"));
|
|
||||||
|
|
||||||
wireMockRule.stubFor(get(urlEqualTo(LIST_PATH)).withHeader("If-None-Match", equalTo(etag))
|
|
||||||
.willReturn(aResponse().withStatus(304)));
|
|
||||||
|
|
||||||
torExitNodeManager.refresh().join();
|
|
||||||
|
|
||||||
verify(getRequestedFor(urlEqualTo(LIST_PATH)).withHeader("If-None-Match", equalTo(etag)));
|
|
||||||
|
|
||||||
assertTrue(torExitNodeManager.isTorExitNode("10.0.0.1"));
|
assertTrue(torExitNodeManager.isTorExitNode("10.0.0.1"));
|
||||||
assertTrue(torExitNodeManager.isTorExitNode("10.0.0.2"));
|
assertTrue(torExitNodeManager.isTorExitNode("10.0.0.2"));
|
||||||
|
|
Loading…
Reference in New Issue