Skip to content

Commit e950678

Browse files
joegallojoegallo
committed
Download IPinfo ip location databases (#114847)
1 parent 52587d6 commit e950678

File tree

7 files changed

+268
-70
lines changed

7 files changed

+268
-70
lines changed

‎modules/ingest-geoip/src/internalClusterTest/java/org/elasticsearch/ingest/geoip/EnterpriseGeoIpDownloaderIT.java‎

Lines changed: 60 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.elasticsearch.action.search.SearchRequest;
2525
import org.elasticsearch.action.search.SearchResponse;
2626
import org.elasticsearch.common.bytes.BytesReference;
27+
import org.elasticsearch.common.Strings;
2728
import org.elasticsearch.common.settings.MockSecureSettings;
2829
import org.elasticsearch.common.settings.Settings;
2930
import org.elasticsearch.common.util.CollectionUtils;
@@ -44,19 +45,25 @@
4445

4546
import java.io.IOException;
4647
import java.util.Collection;
48+
import java.util.List;
4749
import java.util.Map;
4850

4951
import static org.elasticsearch.ingest.EnterpriseGeoIpTask.ENTERPRISE_GEOIP_DOWNLOADER;
52+
import static org.elasticsearch.ingest.geoip.EnterpriseGeoIpDownloaderTaskExecutor.IPINFO_TOKEN_SETTING;
5053
import static org.elasticsearch.ingest.geoip.EnterpriseGeoIpDownloaderTaskExecutor.MAXMIND_LICENSE_KEY_SETTING;
5154
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
5255
import static org.hamcrest.Matchers.equalTo;
5356

5457
public class EnterpriseGeoIpDownloaderIT extends ESIntegTestCase {
5558

56-
private static final String DATABASE_TYPE = "GeoIP2-City";
59+
private static final String MAXMIND_DATABASE_TYPE = "GeoIP2-City";
60+
private static final String IPINFO_DATABASE_TYPE = "asn";
5761

5862
@ClassRule
59-
public static final EnterpriseGeoIpHttpFixture fixture = new EnterpriseGeoIpHttpFixture(DATABASE_TYPE);
63+
public static final EnterpriseGeoIpHttpFixture fixture = new EnterpriseGeoIpHttpFixture(
64+
List.of(MAXMIND_DATABASE_TYPE),
65+
List.of(IPINFO_DATABASE_TYPE)
66+
);
6067

6168
protected String getEndpoint() {
6269
return fixture.getAddress();
@@ -66,6 +73,7 @@ protected String getEndpoint() {
6673
protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
6774
MockSecureSettings secureSettings = new MockSecureSettings();
6875
secureSettings.setString(MAXMIND_LICENSE_KEY_SETTING.getKey(), "license_key");
76+
secureSettings.setString(IPINFO_TOKEN_SETTING.getKey(), "token");
6977
Settings.Builder builder = Settings.builder();
7078
builder.setSecureSettings(secureSettings)
7179
.put(super.nodeSettings(nodeOrdinal, otherSettings))
@@ -92,29 +100,44 @@ public void testEnterpriseDownloaderTask() throws Exception {
92100
* Note that the "enterprise database" is actually just a geolite database being loaded by the GeoIpHttpFixture.
93101
*/
94102
EnterpriseGeoIpDownloader.DEFAULT_MAXMIND_ENDPOINT = getEndpoint();
95-
final String pipelineName = "enterprise_geoip_pipeline";
103+
EnterpriseGeoIpDownloader.DEFAULT_IPINFO_ENDPOINT = getEndpoint();
96104
final String indexName = "enterprise_geoip_test_index";
105+
final String geoipPipelineName = "enterprise_geoip_pipeline";
106+
final String iplocationPipelineName = "enterprise_iplocation_pipeline";
97107
final String sourceField = "ip";
98-
final String targetField = "ip-city";
108+
final String targetField = "ip-result";
99109

100110
startEnterpriseGeoIpDownloaderTask();
101-
configureDatabase(DATABASE_TYPE);
102-
createGeoIpPipeline(pipelineName, DATABASE_TYPE, sourceField, targetField);
111+
configureMaxmindDatabase(MAXMIND_DATABASE_TYPE);
112+
configureIpinfoDatabase(IPINFO_DATABASE_TYPE);
113+
waitAround();
114+
createPipeline(geoipPipelineName, "geoip", MAXMIND_DATABASE_TYPE, sourceField, targetField);
115+
createPipeline(iplocationPipelineName, "ip_location", IPINFO_DATABASE_TYPE, sourceField, targetField);
103116

117+
/*
118+
* We know that the databases index has been populated (because we waited around, :wink:), but we don't know for sure that
119+
* the databases have been pulled down and made available on all nodes. So we run these ingest-and-check steps in assertBusy blocks.
120+
*/
104121
assertBusy(() -> {
105-
/*
106-
* We know that the .geoip_databases index has been populated, but we don't know for sure that the database has been pulled
107-
* down and made available on all nodes. So we run this ingest-and-check step in an assertBusy.
108-
*/
109122
logger.info("Ingesting a test document");
110-
String documentId = ingestDocument(indexName, pipelineName, sourceField);
123+
String documentId = ingestDocument(indexName, geoipPipelineName, sourceField, "89.160.20.128");
111124
GetResponse getResponse = client().get(new GetRequest(indexName, documentId)).actionGet();
112125
Map<String, Object> returnedSource = getResponse.getSource();
113126
assertNotNull(returnedSource);
114127
Object targetFieldValue = returnedSource.get(targetField);
115128
assertNotNull(targetFieldValue);
116129
assertThat(((Map<String, Object>) targetFieldValue).get("organization_name"), equalTo("Bredband2 AB"));
117130
});
131+
assertBusy(() -> {
132+
logger.info("Ingesting another test document");
133+
String documentId = ingestDocument(indexName, iplocationPipelineName, sourceField, "12.10.66.1");
134+
GetResponse getResponse = client().get(new GetRequest(indexName, documentId)).actionGet();
135+
Map<String, Object> returnedSource = getResponse.getSource();
136+
assertNotNull(returnedSource);
137+
Object targetFieldValue = returnedSource.get(targetField);
138+
assertNotNull(targetFieldValue);
139+
assertThat(((Map<String, Object>) targetFieldValue).get("organization_name"), equalTo("OAKLAWN JOCKEY CLUB, INC."));
140+
});
118141
}
119142

120143
private void startEnterpriseGeoIpDownloaderTask() {
@@ -133,29 +156,46 @@ private void startEnterpriseGeoIpDownloaderTask() {
133156
);
134157
}
135158

136-
private void configureDatabase(String databaseType) throws Exception {
159+
private void configureMaxmindDatabase(String databaseType) {
137160
admin().cluster()
138161
.execute(
139162
PutDatabaseConfigurationAction.INSTANCE,
140163
new PutDatabaseConfigurationAction.Request(
141164
TimeValue.MAX_VALUE,
142165
TimeValue.MAX_VALUE,
143-
new DatabaseConfiguration("test", databaseType, new DatabaseConfiguration.Maxmind("test_account"))
166+
new DatabaseConfiguration("test-1", databaseType, new DatabaseConfiguration.Maxmind("test_account"))
144167
)
145168
)
146169
.actionGet();
170+
}
171+
172+
private void configureIpinfoDatabase(String databaseType) {
173+
admin().cluster()
174+
.execute(
175+
PutDatabaseConfigurationAction.INSTANCE,
176+
new PutDatabaseConfigurationAction.Request(
177+
TimeValue.MAX_VALUE,
178+
TimeValue.MAX_VALUE,
179+
new DatabaseConfiguration("test-2", databaseType, new DatabaseConfiguration.Ipinfo())
180+
)
181+
)
182+
.actionGet();
183+
}
184+
185+
private void waitAround() throws Exception {
147186
ensureGreen(GeoIpDownloader.DATABASES_INDEX);
148187
assertBusy(() -> {
149188
SearchResponse searchResponse = client().search(new SearchRequest(GeoIpDownloader.DATABASES_INDEX)).actionGet();
150189
try {
151-
assertThat(searchResponse.getHits().getHits().length, equalTo(1));
190+
assertThat(searchResponse.getHits().getHits().length, equalTo(2));
152191
} finally {
153192
searchResponse.decRef();
154193
}
155194
});
156195
}
157196

158-
private void createGeoIpPipeline(String pipelineName, String databaseType, String sourceField, String targetField) throws IOException {
197+
private void createPipeline(String pipelineName, String processorType, String databaseType,
198+
String sourceField, String targetField) throws IOException {
159199
final BytesReference bytes;
160200
try (XContentBuilder builder = JsonXContent.contentBuilder()) {
161201
builder.startObject();
@@ -165,7 +205,7 @@ private void createGeoIpPipeline(String pipelineName, String databaseType, Strin
165205
{
166206
builder.startObject();
167207
{
168-
builder.startObject("geoip");
208+
builder.startObject(processorType);
169209
{
170210
builder.field("field", sourceField);
171211
builder.field("target_field", targetField);
@@ -183,11 +223,11 @@ private void createGeoIpPipeline(String pipelineName, String databaseType, Strin
183223
assertAcked(clusterAdmin().putPipeline(new PutPipelineRequest(pipelineName, bytes, XContentType.JSON)).actionGet());
184224
}
185225

186-
private String ingestDocument(String indexName, String pipelineName, String sourceField) {
226+
private String ingestDocument(String indexName, String pipelineName, String sourceField, String value) {
187227
BulkRequest bulkRequest = new BulkRequest();
188-
bulkRequest.add(
189-
new IndexRequest(indexName).source("{\"" + sourceField + "\": \"89.160.20.128\"}", XContentType.JSON).setPipeline(pipelineName)
190-
);
228+
bulkRequest.add(new IndexRequest(indexName).source(Strings.format("""
229+
{ "%s": "%s"}
230+
""", sourceField, value), XContentType.JSON).setPipeline(pipelineName));
191231
BulkResponse response = client().bulk(bulkRequest).actionGet();
192232
BulkItemResponse[] bulkItemResponses = response.getItems();
193233
assertThat(bulkItemResponses.length, equalTo(1));

‎modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/EnterpriseGeoIpDownloader.java‎

Lines changed: 117 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@
2323
import org.elasticsearch.common.CheckedSupplier;
2424
import org.elasticsearch.common.Strings;
2525
import org.elasticsearch.common.hash.MessageDigests;
26-
import org.elasticsearch.core.Nullable;
2726
import org.elasticsearch.core.TimeValue;
2827
import org.elasticsearch.core.Tuple;
2928
import org.elasticsearch.index.query.BoolQueryBuilder;
@@ -39,6 +38,8 @@
3938
import org.elasticsearch.tasks.TaskId;
4039
import org.elasticsearch.threadpool.Scheduler;
4140
import org.elasticsearch.threadpool.ThreadPool;
41+
import org.elasticsearch.xcontent.XContentParser;
42+
import org.elasticsearch.xcontent.XContentParserConfiguration;
4243
import org.elasticsearch.xcontent.XContentType;
4344

4445
import java.io.Closeable;
@@ -57,6 +58,7 @@
5758
import java.util.regex.Pattern;
5859
import java.util.stream.Collectors;
5960

61+
import static org.elasticsearch.ingest.geoip.EnterpriseGeoIpDownloaderTaskExecutor.IPINFO_SETTINGS_PREFIX;
6062
import static org.elasticsearch.ingest.geoip.EnterpriseGeoIpDownloaderTaskExecutor.MAXMIND_SETTINGS_PREFIX;
6163

6264
/**
@@ -72,6 +74,9 @@ public class EnterpriseGeoIpDownloader extends AllocatedPersistentTask {
7274
// a sha256 checksum followed by two spaces followed by an (ignored) file name
7375
private static final Pattern SHA256_CHECKSUM_PATTERN = Pattern.compile("(\\w{64})\\s\\s(.*)");
7476

77+
// an md5 checksum
78+
private static final Pattern MD5_CHECKSUM_PATTERN = Pattern.compile("(\\w{32})");
79+
7580
// for overriding in tests
7681
static String DEFAULT_MAXMIND_ENDPOINT = System.getProperty(
7782
MAXMIND_SETTINGS_PREFIX + "endpoint.default", //
@@ -80,6 +85,14 @@ public class EnterpriseGeoIpDownloader extends AllocatedPersistentTask {
8085
// n.b. a future enhancement might be to allow for a MAXMIND_ENDPOINT_SETTING, but
8186
// at the moment this is an unsupported system property for use in tests (only)
8287

88+
// for overriding in tests
89+
static String DEFAULT_IPINFO_ENDPOINT = System.getProperty(
90+
IPINFO_SETTINGS_PREFIX + "endpoint.default", //
91+
"https://ipinfo.io/data"
92+
);
93+
// n.b. a future enhancement might be to allow for an IPINFO_ENDPOINT_SETTING, but
94+
// at the moment this is an unsupported system property for use in tests (only)
95+
8396
static final String DATABASES_INDEX = ".geoip_databases";
8497
static final int MAX_CHUNK_SIZE = 1024 * 1024;
8598

@@ -444,16 +457,15 @@ private void scheduleNextRun(TimeValue time) {
444457
}
445458
}
446459

447-
@Nullable
448460
private ProviderDownload downloaderFor(DatabaseConfiguration database) {
449-
if (database.provider() instanceof DatabaseConfiguration.Maxmind) {
450-
return new MaxmindDownload(database.name(), (DatabaseConfiguration.Maxmind) database.provider());
451-
} else if (database.provider() instanceof DatabaseConfiguration.Ipinfo) {
452-
// as a temporary implementation detail, null here means 'not actually supported *just yet*'
453-
return null;
461+
if (database.provider() instanceof DatabaseConfiguration.Maxmind maxmind) {
462+
return new MaxmindDownload(database.name(), maxmind);
463+
} else if (database.provider() instanceof DatabaseConfiguration.Ipinfo ipinfo) {
464+
return new IpinfoDownload(database.name(), ipinfo);
454465
} else {
455-
assert false : "Attempted to use database downloader with unsupported provider type [" + database.provider().getClass() + "]";
456-
return null;
466+
throw new IllegalArgumentException(
467+
Strings.format("Unexpected provider [%s] for configuration [%s]", database.provider().getClass(), database.id())
468+
);
457469
}
458470
}
459471

@@ -488,7 +500,7 @@ public HttpClient.PasswordAuthenticationHolder buildCredentials() {
488500

489501
@Override
490502
public boolean validCredentials() {
491-
return auth.get() != null;
503+
return auth != null && auth.get() != null;
492504
}
493505

494506
@Override
@@ -529,7 +541,101 @@ public CheckedSupplier<InputStream, IOException> download() {
529541

530542
@Override
531543
public void close() throws IOException {
532-
auth.close();
544+
if (auth != null) auth.close();
545+
}
546+
}
547+
548+
class IpinfoDownload implements ProviderDownload {
549+
550+
final String name;
551+
final DatabaseConfiguration.Ipinfo ipinfo;
552+
HttpClient.PasswordAuthenticationHolder auth;
553+
554+
IpinfoDownload(String name, DatabaseConfiguration.Ipinfo ipinfo) {
555+
this.name = name;
556+
this.ipinfo = ipinfo;
557+
this.auth = buildCredentials();
558+
}
559+
560+
@Override
561+
public HttpClient.PasswordAuthenticationHolder buildCredentials() {
562+
final char[] tokenChars = tokenProvider.apply("ipinfo");
563+
564+
// if the token is missing or empty, return null as 'no auth'
565+
if (tokenChars == null || tokenChars.length == 0) {
566+
return null;
567+
}
568+
569+
// ipinfo uses the token as the username component of basic auth, see https://ipinfo.io/developers#authentication
570+
return new HttpClient.PasswordAuthenticationHolder(new String(tokenChars), new char[] {});
571+
}
572+
573+
@Override
574+
public boolean validCredentials() {
575+
return auth != null && auth.get() != null;
576+
}
577+
578+
private static final Set<String> FREE_DATABASES = Set.of("asn", "country", "country_asn");
579+
580+
@Override
581+
public String url(String suffix) {
582+
// note: the 'free' databases are in the sub-path 'free/' in terms of the download endpoint
583+
final String internalName;
584+
if (FREE_DATABASES.contains(name)) {
585+
internalName = "free/" + name;
586+
} else {
587+
internalName = name;
588+
}
589+
590+
// reminder, we're passing the ipinfo token as the username part of http basic auth,
591+
// see https://ipinfo.io/developers#authentication
592+
593+
String endpointPattern = DEFAULT_IPINFO_ENDPOINT;
594+
if (endpointPattern.contains("%")) {
595+
throw new IllegalArgumentException("Invalid endpoint [" + endpointPattern + "]");
596+
}
597+
if (endpointPattern.endsWith("/") == false) {
598+
endpointPattern += "/";
599+
}
600+
endpointPattern += "%s.%s";
601+
602+
// at this point the pattern looks like this (in the default case):
603+
// https://ipinfo.io/data/%s.%s
604+
// also see https://ipinfo.io/developers/database-download,
605+
// and https://ipinfo.io/developers/database-filename-reference for more
606+
607+
return Strings.format(endpointPattern, internalName, suffix);
608+
}
609+
610+
@Override
611+
public Checksum checksum() throws IOException {
612+
final String checksumJsonUrl = this.url("mmdb/checksums"); // a minor abuse of the idea of a 'suffix', :shrug:
613+
byte[] data = httpClient.getBytes(auth.get(), checksumJsonUrl); // this throws if the auth is bad
614+
Map<String, Object> checksums;
615+
try (XContentParser parser = XContentType.JSON.xContent().createParser(XContentParserConfiguration.EMPTY, data)) {
616+
checksums = parser.map();
617+
}
618+
@SuppressWarnings("unchecked")
619+
String md5 = ((Map<String, String>) checksums.get("checksums")).get("md5");
620+
logger.trace("checksum was [{}]", md5);
621+
622+
var matcher = MD5_CHECKSUM_PATTERN.matcher(md5);
623+
boolean match = matcher.matches();
624+
if (match == false) {
625+
throw new RuntimeException("Unexpected md5 response from [" + checksumJsonUrl + "]");
626+
}
627+
return Checksum.md5(md5);
628+
}
629+
630+
@Override
631+
public CheckedSupplier<InputStream, IOException> download() {
632+
final String mmdbUrl = this.url("mmdb");
633+
return () -> httpClient.get(auth.get(), mmdbUrl);
634+
}
635+
636+
@Override
637+
public void close() throws IOException {
638+
if (auth != null) auth.close();
533639
}
534640
}
535641

‎modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/EnterpriseGeoIpDownloaderTaskExecutor.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,11 +54,15 @@ public class EnterpriseGeoIpDownloaderTaskExecutor extends PersistentTasksExecut
5454

5555
static final String MAXMIND_SETTINGS_PREFIX = "ingest.geoip.downloader.maxmind.";
5656

57+
static final String IPINFO_SETTINGS_PREFIX = "ingest.ip_location.downloader.ipinfo.";
58+
5759
public static final Setting<SecureString> MAXMIND_LICENSE_KEY_SETTING = SecureSetting.secureString(
5860
MAXMIND_SETTINGS_PREFIX + "license_key",
5961
null
6062
);
6163

64+
public static final Setting<SecureString> IPINFO_TOKEN_SETTING = SecureSetting.secureString(IPINFO_SETTINGS_PREFIX + "token", null);
65+
6266
private final Client client;
6367
private final HttpClient httpClient;
6468
private final ClusterService clusterService;
@@ -106,6 +110,10 @@ private char[] getSecureToken(final String type) {
106110
if (cachedSecureSettings.getSettingNames().contains(MAXMIND_LICENSE_KEY_SETTING.getKey())) {
107111
token = cachedSecureSettings.getString(MAXMIND_LICENSE_KEY_SETTING.getKey()).getChars();
108112
}
113+
} else if (type.equals("ipinfo")) {
114+
if (cachedSecureSettings.getSettingNames().contains(IPINFO_TOKEN_SETTING.getKey())) {
115+
token = cachedSecureSettings.getString(IPINFO_TOKEN_SETTING.getKey()).getChars();
116+
}
109117
}
110118
return token;
111119
}
@@ -166,7 +174,7 @@ public synchronized void reload(Settings settings) {
166174
// `SecureSettings` are available here! cache them as they will be needed
167175
// whenever dynamic cluster settings change and we have to rebuild the accounts
168176
try {
169-
this.cachedSecureSettings = extractSecureSettings(settings, List.of(MAXMIND_LICENSE_KEY_SETTING));
177+
this.cachedSecureSettings = extractSecureSettings(settings, List.of(MAXMIND_LICENSE_KEY_SETTING, IPINFO_TOKEN_SETTING));
170178
} catch (GeneralSecurityException e) {
171179
// rethrow as a runtime exception, there's logging higher up the call chain around ReloadablePlugin
172180
throw new ElasticsearchException("Exception while reloading enterprise geoip download task executor", e);

0 commit comments

Comments
 (0)