Skip to content

Commit 69054ac

Browse files
joegallojoegallo
authored
Download IPinfo ip location databases (#114847)
1 parent c401a71 commit 69054ac

File tree

8 files changed

+272
-74
lines changed

8 files changed

+272
-74
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
@@ -22,6 +22,7 @@
2222
import org.elasticsearch.action.index.IndexRequest;
2323
import org.elasticsearch.action.search.SearchRequest;
2424
import org.elasticsearch.action.search.SearchResponse;
25+
import org.elasticsearch.common.Strings;
2526
import org.elasticsearch.common.settings.MockSecureSettings;
2627
import org.elasticsearch.common.settings.Settings;
2728
import org.elasticsearch.common.util.CollectionUtils;
@@ -40,18 +41,24 @@
4041

4142
import java.io.IOException;
4243
import java.util.Collection;
44+
import java.util.List;
4345
import java.util.Map;
4446

4547
import static org.elasticsearch.ingest.EnterpriseGeoIpTask.ENTERPRISE_GEOIP_DOWNLOADER;
48+
import static org.elasticsearch.ingest.geoip.EnterpriseGeoIpDownloaderTaskExecutor.IPINFO_TOKEN_SETTING;
4649
import static org.elasticsearch.ingest.geoip.EnterpriseGeoIpDownloaderTaskExecutor.MAXMIND_LICENSE_KEY_SETTING;
4750
import static org.hamcrest.Matchers.equalTo;
4851

4952
public class EnterpriseGeoIpDownloaderIT extends ESIntegTestCase {
5053

51-
private static final String DATABASE_TYPE = "GeoIP2-City";
54+
private static final String MAXMIND_DATABASE_TYPE = "GeoIP2-City";
55+
private static final String IPINFO_DATABASE_TYPE = "asn";
5256

5357
@ClassRule
54-
public static final EnterpriseGeoIpHttpFixture fixture = new EnterpriseGeoIpHttpFixture(DATABASE_TYPE);
58+
public static final EnterpriseGeoIpHttpFixture fixture = new EnterpriseGeoIpHttpFixture(
59+
List.of(MAXMIND_DATABASE_TYPE),
60+
List.of(IPINFO_DATABASE_TYPE)
61+
);
5562

5663
protected String getEndpoint() {
5764
return fixture.getAddress();
@@ -61,6 +68,7 @@ protected String getEndpoint() {
6168
protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
6269
MockSecureSettings secureSettings = new MockSecureSettings();
6370
secureSettings.setString(MAXMIND_LICENSE_KEY_SETTING.getKey(), "license_key");
71+
secureSettings.setString(IPINFO_TOKEN_SETTING.getKey(), "token");
6472
Settings.Builder builder = Settings.builder();
6573
builder.setSecureSettings(secureSettings)
6674
.put(super.nodeSettings(nodeOrdinal, otherSettings))
@@ -87,29 +95,44 @@ public void testEnterpriseDownloaderTask() throws Exception {
8795
* Note that the "enterprise database" is actually just a geolite database being loaded by the GeoIpHttpFixture.
8896
*/
8997
EnterpriseGeoIpDownloader.DEFAULT_MAXMIND_ENDPOINT = getEndpoint();
90-
final String pipelineName = "enterprise_geoip_pipeline";
98+
EnterpriseGeoIpDownloader.DEFAULT_IPINFO_ENDPOINT = getEndpoint();
9199
final String indexName = "enterprise_geoip_test_index";
100+
final String geoipPipelineName = "enterprise_geoip_pipeline";
101+
final String iplocationPipelineName = "enterprise_iplocation_pipeline";
92102
final String sourceField = "ip";
93-
final String targetField = "ip-city";
103+
final String targetField = "ip-result";
94104

95105
startEnterpriseGeoIpDownloaderTask();
96-
configureDatabase(DATABASE_TYPE);
97-
createGeoIpPipeline(pipelineName, DATABASE_TYPE, sourceField, targetField);
106+
configureMaxmindDatabase(MAXMIND_DATABASE_TYPE);
107+
configureIpinfoDatabase(IPINFO_DATABASE_TYPE);
108+
waitAround();
109+
createPipeline(geoipPipelineName, "geoip", MAXMIND_DATABASE_TYPE, sourceField, targetField);
110+
createPipeline(iplocationPipelineName, "ip_location", IPINFO_DATABASE_TYPE, sourceField, targetField);
98111

112+
/*
113+
* We know that the databases index has been populated (because we waited around, :wink:), but we don't know for sure that
114+
* the databases have been pulled down and made available on all nodes. So we run these ingest-and-check steps in assertBusy blocks.
115+
*/
99116
assertBusy(() -> {
100-
/*
101-
* We know that the .geoip_databases index has been populated, but we don't know for sure that the database has been pulled
102-
* down and made available on all nodes. So we run this ingest-and-check step in an assertBusy.
103-
*/
104117
logger.info("Ingesting a test document");
105-
String documentId = ingestDocument(indexName, pipelineName, sourceField);
118+
String documentId = ingestDocument(indexName, geoipPipelineName, sourceField, "89.160.20.128");
106119
GetResponse getResponse = client().get(new GetRequest(indexName, documentId)).actionGet();
107120
Map<String, Object> returnedSource = getResponse.getSource();
108121
assertNotNull(returnedSource);
109122
Object targetFieldValue = returnedSource.get(targetField);
110123
assertNotNull(targetFieldValue);
111124
assertThat(((Map<String, Object>) targetFieldValue).get("organization_name"), equalTo("Bredband2 AB"));
112125
});
126+
assertBusy(() -> {
127+
logger.info("Ingesting another test document");
128+
String documentId = ingestDocument(indexName, iplocationPipelineName, sourceField, "12.10.66.1");
129+
GetResponse getResponse = client().get(new GetRequest(indexName, documentId)).actionGet();
130+
Map<String, Object> returnedSource = getResponse.getSource();
131+
assertNotNull(returnedSource);
132+
Object targetFieldValue = returnedSource.get(targetField);
133+
assertNotNull(targetFieldValue);
134+
assertThat(((Map<String, Object>) targetFieldValue).get("organization_name"), equalTo("OAKLAWN JOCKEY CLUB, INC."));
135+
});
113136
}
114137

115138
private void startEnterpriseGeoIpDownloaderTask() {
@@ -128,36 +151,53 @@ private void startEnterpriseGeoIpDownloaderTask() {
128151
);
129152
}
130153

131-
private void configureDatabase(String databaseType) throws Exception {
154+
private void configureMaxmindDatabase(String databaseType) {
132155
admin().cluster()
133156
.execute(
134157
PutDatabaseConfigurationAction.INSTANCE,
135158
new PutDatabaseConfigurationAction.Request(
136159
TimeValue.MAX_VALUE,
137160
TimeValue.MAX_VALUE,
138-
new DatabaseConfiguration("test", databaseType, new DatabaseConfiguration.Maxmind("test_account"))
161+
new DatabaseConfiguration("test-1", databaseType, new DatabaseConfiguration.Maxmind("test_account"))
139162
)
140163
)
141164
.actionGet();
165+
}
166+
167+
private void configureIpinfoDatabase(String databaseType) {
168+
admin().cluster()
169+
.execute(
170+
PutDatabaseConfigurationAction.INSTANCE,
171+
new PutDatabaseConfigurationAction.Request(
172+
TimeValue.MAX_VALUE,
173+
TimeValue.MAX_VALUE,
174+
new DatabaseConfiguration("test-2", databaseType, new DatabaseConfiguration.Ipinfo())
175+
)
176+
)
177+
.actionGet();
178+
}
179+
180+
private void waitAround() throws Exception {
142181
ensureGreen(GeoIpDownloader.DATABASES_INDEX);
143182
assertBusy(() -> {
144183
SearchResponse searchResponse = client().search(new SearchRequest(GeoIpDownloader.DATABASES_INDEX)).actionGet();
145184
try {
146-
assertThat(searchResponse.getHits().getHits().length, equalTo(1));
185+
assertThat(searchResponse.getHits().getHits().length, equalTo(2));
147186
} finally {
148187
searchResponse.decRef();
149188
}
150189
});
151190
}
152191

153-
private void createGeoIpPipeline(String pipelineName, String databaseType, String sourceField, String targetField) throws IOException {
192+
private void createPipeline(String pipelineName, String processorType, String databaseType, String sourceField, String targetField)
193+
throws IOException {
154194
putJsonPipeline(pipelineName, (builder, params) -> {
155195
builder.field("description", "test");
156196
builder.startArray("processors");
157197
{
158198
builder.startObject();
159199
{
160-
builder.startObject("geoip");
200+
builder.startObject(processorType);
161201
{
162202
builder.field("field", sourceField);
163203
builder.field("target_field", targetField);
@@ -171,11 +211,11 @@ private void createGeoIpPipeline(String pipelineName, String databaseType, Strin
171211
});
172212
}
173213

174-
private String ingestDocument(String indexName, String pipelineName, String sourceField) {
214+
private String ingestDocument(String indexName, String pipelineName, String sourceField, String value) {
175215
BulkRequest bulkRequest = new BulkRequest();
176-
bulkRequest.add(
177-
new IndexRequest(indexName).source("{\"" + sourceField + "\": \"89.160.20.128\"}", XContentType.JSON).setPipeline(pipelineName)
178-
);
216+
bulkRequest.add(new IndexRequest(indexName).source(Strings.format("""
217+
{ "%s": "%s"}
218+
""", sourceField, value), XContentType.JSON).setPipeline(pipelineName));
179219
BulkResponse response = client().bulk(bulkRequest).actionGet();
180220
BulkItemResponse[] bulkItemResponses = response.getItems();
181221
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.info("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)