Netty: Added ip filter capabilties to pipeline

Used the existing infra structure to filter by ip in the netty pipeline
before any other handler is hit, in order to reject as soon as possible.
Right now the connection is simply closed.

The configuration is a simple YAML file which uses allow/deny rules

Original commit: elastic/x-pack-elasticsearch@000e44f8cc

src/main/java/org/elasticsearch/shield/authz/FileRolesStore.java

@@ -87,7 +87,7 @@ public class FileRolesStore extends AbstractComponent implements RolesStore {
             while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
                 if (token == XContentParser.Token.FIELD_NAME) {
                     currentFieldName = parser.currentName();
-                } else if (token == XContentParser.Token.START_OBJECT) {
+                } else if (token == XContentParser.Token.START_OBJECT && currentFieldName != null) {
                     String roleName = currentFieldName;
                     Permission.Compound.Builder builder = null;
                     while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {

src/main/java/org/elasticsearch/shield/n2n/IPFilteringN2NAuthenticator.java

@@ -0,0 +1,165 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.shield.n2n;
+
+import com.google.common.net.InetAddresses;
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.base.Charsets;
+import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.logging.ESLogger;
+import org.elasticsearch.common.logging.ESLoggerFactory;
+import org.elasticsearch.common.netty.handler.ipfilter.IpFilterRule;
+import org.elasticsearch.common.netty.handler.ipfilter.IpSubnetFilterRule;
+import org.elasticsearch.common.netty.handler.ipfilter.PatternRule;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.yaml.YamlXContent;
+import org.elasticsearch.env.Environment;
+import org.elasticsearch.watcher.FileChangesListener;
+import org.elasticsearch.watcher.FileWatcher;
+import org.elasticsearch.watcher.ResourceWatcherService;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.security.Principal;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+
+public class IPFilteringN2NAuthenticator extends AbstractComponent implements N2NAuthenticator {
+
+    private static final Pattern COMMA_DELIM = Pattern.compile("\\s*,\\s*");
+    private static final String DEFAULT_FILE = ".ip_filter.yml";
+    private static final IpFilterRule[] NO_RULES = new IpFilterRule[0];
+
+    private final Path file;
+    private final FileWatcher watcher;
+
+    private volatile IpFilterRule[] rules = NO_RULES;
+
+    @Inject
+    public IPFilteringN2NAuthenticator(Settings settings, Environment env, ResourceWatcherService watcherService) {
+        super(settings);
+        file = resolveFile(componentSettings, env);
+        rules = parseFile(file);
+        watcher = new FileWatcher(file.getParent().toFile());
+        watcher.addListener(new FileListener());
+        watcherService.add(watcher);
+    }
+
+    private Path resolveFile(Settings settings, Environment env) {
+        String location = settings.get("file");
+        if (location == null) {
+            return env.configFile().toPath().resolve(DEFAULT_FILE);
+        }
+        return Paths.get(location);
+    }
+
+    public static IpFilterRule[] parseFile(Path path) {
+        if (!Files.exists(path)) {
+            return NO_RULES;
+        }
+
+        List<IpFilterRule> rules = new ArrayList<>();
+
+        byte[] content;
+        try {
+            content = Files.readAllBytes(path);
+            try (XContentParser parser = YamlXContent.yamlXContent.createParser(content)) {
+                XContentParser.Token token;
+                String currentFieldName = null;
+                while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT && token != null) {
+                    if (token == XContentParser.Token.FIELD_NAME) {
+                        currentFieldName = parser.currentName();
+                        if (!"allow".equals(currentFieldName) && !"deny".equals(currentFieldName)) {
+                            throw new ElasticsearchParseException("Field name [" + currentFieldName + "] not valid. Must be [allow] or [deny]");
+                        }
+                    } else if (token == XContentParser.Token.VALUE_STRING && currentFieldName != null) {
+                        String value = parser.text();
+                        if (!Strings.hasLength(value)) {
+                            throw new ElasticsearchParseException("Field value for fieldname [" + currentFieldName + "] must not be empty");
+                        }
+
+                        boolean isAllowRule = currentFieldName.equals("allow");
+
+                        if (value.contains(",")) {
+                            for (String rule : COMMA_DELIM.split(parser.text().trim())) {
+                                rules.add(getRule(isAllowRule, rule));
+                            }
+                        } else {
+                            rules.add(getRule(isAllowRule, value));
+                        }
+
+                    }
+                }
+            }
+
+        } catch (IOException ioe) {
+            throw new ElasticsearchException("Failed to read & parse host access file [" + path.toAbsolutePath() + "]", ioe);
+        }
+
+        if (rules.size() == 0) {
+            return NO_RULES;
+        }
+        return rules.toArray(new IpFilterRule[rules.size()]);
+    }
+
+    private static IpFilterRule getRule(boolean isAllowRule, String value) throws UnknownHostException {
+        if ("all".equals(value)) {
+            return new PatternRule(isAllowRule, "n:*");
+        } else if (value.contains("/")) {
+            return new IpSubnetFilterRule(isAllowRule, value);
+        }
+
+        boolean isInetAddress = InetAddresses.isInetAddress(value);
+        String prefix = isInetAddress ? "i:" : "n:";
+        return new PatternRule(isAllowRule, prefix + value);
+    }
+
+    @Override
+    public boolean authenticate(@Nullable Principal peerPrincipal, InetAddress peerAddress, int peerPort) {
+        for (int i = 0; i < rules.length; i++) {
+            if (rules[i].contains(peerAddress)) {
+                return rules[i].isAllowRule();
+            }
+        }
+        return false;
+    }
+
+    private class FileListener extends FileChangesListener {
+        @Override
+        public void onFileCreated(File file) {
+            if (file.equals(IPFilteringN2NAuthenticator.this.file.toFile())) {
+                rules = parseFile(file.toPath());
+            }
+        }
+
+        @Override
+        public void onFileDeleted(File file) {
+            if (file.equals(IPFilteringN2NAuthenticator.this.file.toFile())) {
+                rules = NO_RULES;
+            }
+        }
+
+        @Override
+        public void onFileChanged(File file) {
+            if (file.equals(IPFilteringN2NAuthenticator.this.file.toFile())) {
+                if (file.equals(IPFilteringN2NAuthenticator.this.file.toFile())) {
+                    rules = parseFile(file.toPath());
+                }
+            }
+        }
+    }
+}

src/main/java/org/elasticsearch/shield/n2n/IPFilteringN2NAuthentricator.java

@@ -1,177 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the Elastic License;
- * you may not use this file except in compliance with the Elastic License.
- */
-package org.elasticsearch.shield.n2n;
-
-import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.common.Nullable;
-import org.elasticsearch.common.base.Charsets;
-import org.elasticsearch.common.component.AbstractComponent;
-import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.logging.ESLogger;
-import org.elasticsearch.common.netty.handler.ipfilter.IpFilterRule;
-import org.elasticsearch.common.netty.handler.ipfilter.IpSubnetFilterRule;
-import org.elasticsearch.common.netty.handler.ipfilter.PatternRule;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.env.Environment;
-import org.elasticsearch.watcher.FileChangesListener;
-import org.elasticsearch.watcher.FileWatcher;
-import org.elasticsearch.watcher.ResourceWatcherService;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.security.Principal;
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * A file based IP Filtering node to node authentication. IP filtering rules can be configured in
- * a monitored file (auto loads when changed). Each line in the file represents a filtering rule.
- * A rule is composed of an inclusion/exclusion sign and a matching rule:
- *
- *
- * <ul>
- *     <li><code>-PATTERN</code>: means any remote address that matches PATTERN will be denied (auth will fail)<li>
- *     <li><code>+PATTERN</code>: means any remote address that matches PATTERN will be allowed (auth will succeed)<li>
- * </ul>
- *
- * The following patterns are supported:
- *
- * <ul>
- *     <li><code>i:IP</code> - where IP is a specific node IP (regexp supported)</li>
- *     <li><code>c:MASK</code> - where MASK is a CIDR mask to match nodes IPs</li>
- *     <li><code>n:HOST</code> - where HOST is the hostname of the node (regexp supported)</li>
- * </ul>
- *
- * Examples:
- *
- * <ul>
- *     <li><code>-i:192.168.100.2</code></li>: deny access from node with the specified IP
- *     <li><code>+c:2001:db8::/48</code></li>: allow access from nodes with IPs that match the specified mask
- *     <li><code>-n:es-staging-.*</code></li>: deny access from any node in stanging env (matched on the hostname regexp)
- * </ul>
- *
- */
-public class IPFilteringN2NAuthentricator extends AbstractComponent implements N2NAuthenticator {
-
-    private static final String DEFAULT_FILE = ".ip_filtering";
-    private static final IpFilterRule[] NO_RULES = new IpFilterRule[0];
-
-    private final Path file;
-    private final FileWatcher watcher;
-
-    private volatile IpFilterRule[] rules;
-
-    @Inject
-    public IPFilteringN2NAuthentricator(Settings settings, Environment env, ResourceWatcherService watcherService) {
-        super(settings);
-        file = resolveFile(componentSettings, env);
-        rules = parseFile(file, logger);
-        watcher = new FileWatcher(file.getParent().toFile());
-        watcher.addListener(new FileListener());
-        watcherService.add(watcher);
-    }
-
-    public static Path resolveFile(Settings settings, Environment env) {
-        String location = settings.get("file" + DEFAULT_FILE);
-        if (location == null) {
-            return env.configFile().toPath().resolve(DEFAULT_FILE);
-        }
-        return Paths.get(location);
-    }
-
-    public static IpFilterRule[] parseFile(Path path, @Nullable ESLogger logger) {
-        if (!Files.exists(path)) {
-            return null;
-        }
-        List<String> lines = null;
-        try {
-            lines = Files.readAllLines(path, Charsets.UTF_8);
-        } catch (IOException ioe) {
-            throw new ElasticsearchException("Failed to read hosts file [" + path.toAbsolutePath() + "]", ioe);
-        }
-
-        List<IpFilterRule> rules = new ArrayList<>(lines.size());
-        for (String line : lines) {
-            rules.add(parseRule(path, line, logger));
-        }
-
-        if (rules.size() == 0) {
-            return NO_RULES;
-        }
-
-        return rules.toArray(new IpFilterRule[rules.size()]);
-    }
-
-    private static IpFilterRule parseRule(Path path, String rule, @Nullable ESLogger logger) {
-        if (rule == null || rule.length() == 0) {
-            return null;
-        }
-        if (!(rule.startsWith("+") || rule.startsWith("-"))) {
-            return null;
-        }
-
-        boolean allow = rule.startsWith("+");
-
-        if (rule.charAt(1) == 'n' || rule.charAt(1) == 'i') {
-            return new PatternRule(allow, rule.substring(1));
-        }
-
-        if (rule.charAt(1) == 'c') {
-            try {
-                return new IpSubnetFilterRule(allow, rule.substring(3));
-            } catch (UnknownHostException e) {
-                if (logger != null && logger.isErrorEnabled()) {
-                    logger.error("Skipping invalid ip filtering rule [" + rule + "] in hosts_allow file [" + path.toAbsolutePath() + "]", e);
-                }
-                return null;
-            }
-        }
-        if (logger != null && logger.isErrorEnabled()) {
-            logger.error("Skipping invalid ip filtering rule [" + rule + "] in hosts_allow file [" + path.toAbsolutePath() + "]. ':' can only appear once");
-        }
-        return null;
-    }
-
-    @Override
-    public boolean authenticate(@Nullable Principal peerPrincipal, InetAddress peerAddress, int peerPort) {
-        for (int i = 0; i < rules.length; i++) {
-            if (rules[i].contains(peerAddress)) {
-                return true;
-            }
-        }
-        return false;
-    }
-
-    private class FileListener extends FileChangesListener {
-        @Override
-        public void onFileCreated(File file) {
-            if (file.equals(IPFilteringN2NAuthentricator.this.file.toFile())) {
-                rules = parseFile(file.toPath(), logger);
-            }
-        }
-
-        @Override
-        public void onFileDeleted(File file) {
-            if (file.equals(IPFilteringN2NAuthentricator.this.file.toFile())) {
-                rules = NO_RULES;
-            }
-        }
-
-        @Override
-        public void onFileChanged(File file) {
-            if (file.equals(IPFilteringN2NAuthentricator.this.file.toFile())) {
-                if (file.equals(IPFilteringN2NAuthentricator.this.file.toFile())) {
-                    rules = parseFile(file.toPath(), logger);
-                }
-            }
-        }
-    }
-}

src/main/java/org/elasticsearch/shield/n2n/N2NAuthModule.java

@@ -21,6 +21,6 @@ public class N2NAuthModule extends AbstractModule {
 
     @Override
     protected void configure() {
-
+        bind(N2NNettyUpstreamHandler.class).asEagerSingleton();
     }
 }

src/main/java/org/elasticsearch/shield/n2n/N2NModule.java

@@ -14,6 +14,6 @@ public class N2NModule extends AbstractModule {
 
     @Override
     protected void configure() {
-        bind(IPFilteringN2NAuthentricator.class).asEagerSingleton();
+        bind(IPFilteringN2NAuthenticator.class).asEagerSingleton();
     }
 }

src/main/java/org/elasticsearch/shield/n2n/N2NNettyUpstreamHandler.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.shield.n2n;
+
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.netty.channel.ChannelEvent;
+import org.elasticsearch.common.netty.channel.ChannelHandler;
+import org.elasticsearch.common.netty.channel.ChannelHandlerContext;
+import org.elasticsearch.common.netty.handler.ipfilter.IpFilteringHandlerImpl;
+
+import java.net.InetSocketAddress;
+
+/**
+ *
+ */
+@ChannelHandler.Sharable
+public class N2NNettyUpstreamHandler extends IpFilteringHandlerImpl {
+
+    private IPFilteringN2NAuthenticator authenticator;
+
+    @Inject
+    public N2NNettyUpstreamHandler(IPFilteringN2NAuthenticator authenticator) {
+        this.authenticator = authenticator;
+    }
+
+    @Override
+    protected boolean accept(ChannelHandlerContext channelHandlerContext, ChannelEvent channelEvent, InetSocketAddress inetSocketAddress) throws Exception {
+        // at this stage no auth has happened, so we do not have any principal anyway
+        return authenticator.authenticate(null, inetSocketAddress.getAddress(), inetSocketAddress.getPort());
+    }
+
+}

src/main/java/org/elasticsearch/shield/ssl/netty/NettySSLHttpServerTransport.java

@@ -13,6 +13,7 @@ import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.http.netty.NettyHttpServerTransport;
+import org.elasticsearch.shield.n2n.N2NNettyUpstreamHandler;
 import org.elasticsearch.shield.ssl.SSLConfig;
 
 import javax.net.ssl.SSLEngine;
@@ -23,11 +24,14 @@ import javax.net.ssl.SSLEngine;
 public class NettySSLHttpServerTransport extends NettyHttpServerTransport {
 
     private final boolean ssl;
+    private final N2NNettyUpstreamHandler shieldUpstreamHandler;
 
     @Inject
-    public NettySSLHttpServerTransport(Settings settings, NetworkService networkService, BigArrays bigArrays) {
+    public NettySSLHttpServerTransport(Settings settings, NetworkService networkService, BigArrays bigArrays,
+                                       N2NNettyUpstreamHandler shieldUpstreamHandler) {
         super(settings, networkService, bigArrays);
         this.ssl = settings.getAsBoolean("shield.http.ssl", false);
+        this.shieldUpstreamHandler = shieldUpstreamHandler;
     }
 
     @Override
@@ -37,15 +41,24 @@ public class NettySSLHttpServerTransport extends NettyHttpServerTransport {
 
     private class HttpSslChannelPipelineFactory extends HttpChannelPipelineFactory {
 
+        private final SSLConfig sslConfig;
+
         public HttpSslChannelPipelineFactory(NettyHttpServerTransport transport) {
             super(transport);
+            if (ssl) {
+                sslConfig = new SSLConfig(settings.getByPrefix("shield.http.ssl."));
+                // try to create an SSL engine, so that exceptions lead to early exit
+                sslConfig.createSSLEngine();
+            } else {
+                sslConfig = null;
+            }
         }
 
         @Override
         public ChannelPipeline getPipeline() throws Exception {
             ChannelPipeline pipeline = super.getPipeline();
+            pipeline.addFirst("ipfilter", shieldUpstreamHandler);
             if (ssl) {
-                SSLConfig sslConfig = new SSLConfig(settings.getByPrefix("shield.http.ssl."));
                 SSLEngine engine = sslConfig.createSSLEngine();
                 engine.setUseClientMode(false);
                 // TODO MAKE ME CONFIGURABLE

src/main/java/org/elasticsearch/shield/ssl/netty/NettySSLTransport.java

@@ -13,6 +13,7 @@ import org.elasticsearch.common.netty.handler.ssl.SslHandler;
 import org.elasticsearch.common.network.NetworkService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.shield.n2n.N2NNettyUpstreamHandler;
 import org.elasticsearch.shield.ssl.SSLConfig;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.netty.NettyTransport;
@@ -25,10 +26,13 @@ import javax.net.ssl.SSLEngine;
 public class NettySSLTransport extends NettyTransport {
 
     private final boolean ssl;
+    private final N2NNettyUpstreamHandler shieldUpstreamHandler;
 
     @Inject
-    public NettySSLTransport(Settings settings, ThreadPool threadPool, NetworkService networkService, BigArrays bigArrays, Version version) {
+    public NettySSLTransport(Settings settings, ThreadPool threadPool, NetworkService networkService, BigArrays bigArrays, Version version,
+                             N2NNettyUpstreamHandler shieldUpstreamHandler) {
         super(settings, threadPool, networkService, bigArrays, version);
+        this.shieldUpstreamHandler = shieldUpstreamHandler;
         this.ssl = settings.getAsBoolean("shield.transport.ssl", false);
     }
 
@@ -48,14 +52,19 @@ public class NettySSLTransport extends NettyTransport {
 
         public SslServerChannelPipelineFactory(NettyTransport nettyTransport) {
             super(nettyTransport);
-            sslConfig = new SSLConfig(settings.getByPrefix("shield.transport.ssl."));
+            if (ssl) {
-            // try to create an SSL engine, so that exceptions lead to early exit
+                sslConfig = new SSLConfig(settings.getByPrefix("shield.transport.ssl."));
-            sslConfig.createSSLEngine();
+                // try to create an SSL engine, so that exceptions lead to early exit
+                sslConfig.createSSLEngine();
+            } else {
+                sslConfig = null;
+            }
         }
 
         @Override
         public ChannelPipeline getPipeline() throws Exception {
             ChannelPipeline pipeline = super.getPipeline();
+            pipeline.addFirst("ipfilter", shieldUpstreamHandler);
             if (ssl) {
                 SSLEngine serverEngine = sslConfig.createSSLEngine();
                 serverEngine.setUseClientMode(false);
@@ -73,9 +82,13 @@ public class NettySSLTransport extends NettyTransport {
 
         public SslClientChannelPipelineFactory(NettyTransport transport) {
             super(transport);
-            sslConfig = new SSLConfig(settings.getByPrefix("shield.transport.ssl."));
+            if (ssl) {
-            // try to create an SSL engine, so that exceptions lead to early exit
+                sslConfig = new SSLConfig(settings.getByPrefix("shield.transport.ssl."));
-            sslConfig.createSSLEngine();
+                // try to create an SSL engine, so that exceptions lead to early exit
+                sslConfig.createSSLEngine();
+            } else {
+               sslConfig = null;
+            }
         }
 
         @Override

src/test/java/org/elasticsearch/shield/n2n/IPFilteringN2NAuthenticatorTests.java

@@ -0,0 +1,164 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.shield.n2n;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+import com.google.common.net.InetAddresses;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.env.Environment;
+import org.elasticsearch.test.ElasticsearchTestCase;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.watcher.ResourceWatcherService;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.security.Principal;
+
+import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilder;
+import static org.hamcrest.Matchers.is;
+
+/**
+ *
+ */
+public class IPFilteringN2NAuthenticatorTests extends ElasticsearchTestCase {
+
+    public static final Principal NULL_PRINCIPAL = new Principal() {
+        @Override
+        public String getName() {
+            return "null";
+        }
+    };
+
+    @Rule
+    public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+    private ResourceWatcherService resourceWatcherService;
+    private File configFile;
+
+    @Before
+    public void init() throws Exception {
+        configFile = temporaryFolder.newFile();
+        Settings settings = settingsBuilder().put("watcher.interval.medium", TimeValue.timeValueMillis(200)).build();
+        resourceWatcherService = new ResourceWatcherService(settings, new ThreadPool("resourceWatcher")).start();
+    }
+
+    @Test
+    public void testThatIpV4AddressesCanBeProcessed() throws Exception {
+        String testData = "allow: 127.0.0.1\ndeny: 10.0.0.0/8";
+        Files.write(testData.getBytes(Charsets.UTF_8), configFile);
+
+        Settings settings = settingsBuilder().put("shield.n2n.file", configFile.getPath()).build();
+        IPFilteringN2NAuthenticator ipFilteringN2NAuthenticator = new IPFilteringN2NAuthenticator(settings, new Environment(), resourceWatcherService);
+
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("127.0.0.1"), 1024), is(true));
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("10.2.3.4"), 1024), is(false));
+    }
+
+    @Test
+    public void testThatIpV6AddressesCanBeProcessed() throws Exception {
+        String testData = "allow: 2001:0db8:1234::/48\ndeny: 1234:0db8:85a3:0000:0000:8a2e:0370:7334";
+        Files.write(testData.getBytes(Charsets.UTF_8), configFile);
+
+        Settings settings = settingsBuilder().put("shield.n2n.file", configFile.getPath()).build();
+        IPFilteringN2NAuthenticator ipFilteringN2NAuthenticator = new IPFilteringN2NAuthenticator(settings, new Environment(), resourceWatcherService);
+
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("2001:0db8:1234:0000:0000:8a2e:0370:7334"), 1024), is(true));
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("1234:0db8:85a3:0000:0000:8a2e:0370:7334"), 1024), is(false));
+    }
+
+    @Test
+    public void testThatHostnamesCanBeProcessed() throws Exception {
+        String testData = "allow: localhost\ndeny: '*.google.com'";
+        Files.write(testData.getBytes(Charsets.UTF_8), configFile);
+
+        Settings settings = settingsBuilder().put("shield.n2n.file", configFile.getPath()).build();
+        IPFilteringN2NAuthenticator ipFilteringN2NAuthenticator = new IPFilteringN2NAuthenticator(settings, new Environment(), resourceWatcherService);
+
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("127.0.0.1"), 1024), is(true));
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("173.194.70.100"), 1024), is(false));
+    }
+
+    @Test
+    public void testThatFileDeletionResultsInAllowingAll() throws Exception {
+        String testData = "allow: 127.0.0.1";
+        Files.write(testData.getBytes(Charsets.UTF_8), configFile);
+
+        Settings settings = settingsBuilder().put("shield.n2n.file", configFile.getPath()).build();
+        IPFilteringN2NAuthenticator ipFilteringN2NAuthenticator = new IPFilteringN2NAuthenticator(settings, new Environment(), resourceWatcherService);
+
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("127.0.0.1"), 1024), is(true));
+
+        configFile.delete();
+        assertThat(configFile.exists(), is(false));
+
+        sleep(250);
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("127.0.0.1"), 1024), is(false));
+    }
+
+    @Test
+    public void testThatAnAllowAllAuthenticatorWorks() throws Exception {
+        String testData = "allow: all";
+        Files.write(testData.getBytes(Charsets.UTF_8), configFile);
+
+        Settings settings = settingsBuilder().put("shield.n2n.file", configFile.getPath()).build();
+        IPFilteringN2NAuthenticator ipFilteringN2NAuthenticator = new IPFilteringN2NAuthenticator(settings, new Environment(), resourceWatcherService);
+
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("127.0.0.1"), 1024), is(true));
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("173.194.70.100"), 1024), is(true));
+    }
+
+    @Test
+    public void testThatCommaSeparatedValuesWork() throws Exception {
+        String testData = "allow: 192.168.23.0/24, localhost\ndeny: all";
+        Files.write(testData.getBytes(Charsets.UTF_8), configFile);
+
+        Settings settings = settingsBuilder().put("shield.n2n.file", configFile.getPath()).build();
+        IPFilteringN2NAuthenticator ipFilteringN2NAuthenticator = new IPFilteringN2NAuthenticator(settings, new Environment(), resourceWatcherService);
+
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("192.168.23.1"), 1024), is(true));
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("127.0.0.1"), 1024), is(true));
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("10.1.2.3"), 1024), is(false));
+    }
+
+    @Test
+    public void testThatOrderIsImportant() throws Exception {
+        String testData = "deny: localhost\nallow: localhost";
+        Files.write(testData.getBytes(Charsets.UTF_8), configFile);
+
+        Settings settings = settingsBuilder().put("shield.n2n.file", configFile.getPath()).build();
+        IPFilteringN2NAuthenticator ipFilteringN2NAuthenticator = new IPFilteringN2NAuthenticator(settings, new Environment(), resourceWatcherService);
+
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("127.0.0.1"), 1024), is(false));
+    }
+
+    @Test
+    public void testThatOrderIsImportantViceVersa() throws Exception {
+        String testData = "allow: localhost\ndeny: localhost";
+        Files.write(testData.getBytes(Charsets.UTF_8), configFile);
+
+        Settings settings = settingsBuilder().put("shield.n2n.file", configFile.getPath()).build();
+        IPFilteringN2NAuthenticator ipFilteringN2NAuthenticator = new IPFilteringN2NAuthenticator(settings, new Environment(), resourceWatcherService);
+
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("127.0.0.1"), 1024), is(true));
+    }
+
+    @Test
+    public void testThatEmptyFileDoesNotLeadIntoLoop() throws Exception {
+        String testData = "# \n\n";
+        Files.write(testData.getBytes(Charsets.UTF_8), configFile);
+
+        Settings settings = settingsBuilder().put("shield.n2n.file", configFile.getPath()).build();
+        IPFilteringN2NAuthenticator ipFilteringN2NAuthenticator = new IPFilteringN2NAuthenticator(settings, new Environment(), resourceWatcherService);
+
+        assertThat(ipFilteringN2NAuthenticator.authenticate(NULL_PRINCIPAL, InetAddresses.forString("127.0.0.1"), 1024), is(false));
+    }
+}

src/test/java/org/elasticsearch/shield/n2n/IpFilteringIntegrationTests.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.shield.n2n;
+
+import com.google.common.base.Charsets;
+import com.google.common.net.InetAddresses;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.http.HttpServerTransport;
+import org.elasticsearch.shield.plugin.SecurityPlugin;
+import org.elasticsearch.shield.ssl.netty.NettySSLHttpServerTransportModule;
+import org.elasticsearch.shield.ssl.netty.NettySSLTransportModule;
+import org.elasticsearch.test.ElasticsearchIntegrationTest;
+import org.elasticsearch.transport.Transport;
+import org.elasticsearch.transport.TransportModule;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.net.HttpURLConnection;
+import java.net.Socket;
+import java.net.SocketException;
+import java.net.URL;
+import java.util.Locale;
+
+import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilder;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
+
+/**
+ *
+ */
+@ElasticsearchIntegrationTest.ClusterScope(scope = ElasticsearchIntegrationTest.Scope.SUITE, numDataNodes = 1, transportClientRatio = 0.0, numClientNodes = 0)
+public class IpFilteringIntegrationTests extends ElasticsearchIntegrationTest {
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return settingsBuilder()
+                .put(super.nodeSettings(nodeOrdinal))
+                .put("discovery.zen.ping.multicast.ping.enabled", false)
+                .put("node.mode", "network")
+                //.put("network.host", "127.0.0.1")
+                .put("http.type", NettySSLHttpServerTransportModule.class.getName())
+                .put(TransportModule.TRANSPORT_TYPE_KEY, NettySSLTransportModule.class.getName())
+                .put("plugin.types", SecurityPlugin.class.getName())
+                //.put("shield.n2n.file", configFile.getPath())
+                .build();
+    }
+
+    @Test(expected = SocketException.class)
+    public void testThatIpFilteringIsIntegratedIntoNettyPipelineViaHttp() throws Exception {
+        TransportAddress transportAddress = internalCluster().getInstance(HttpServerTransport.class).boundAddress().boundAddress();
+        assertThat(transportAddress, is(instanceOf(InetSocketTransportAddress.class)));
+        InetSocketTransportAddress inetSocketTransportAddress = (InetSocketTransportAddress) transportAddress;
+        String url = String.format(Locale.ROOT, "http://%s:%s/", InetAddresses.toUriString(inetSocketTransportAddress.address().getAddress()), inetSocketTransportAddress.address().getPort());
+
+        logger.info("Opening connection to {}", url);
+        HttpURLConnection connection = (HttpURLConnection) new URL(url).openConnection();
+        connection.connect();
+        connection.getResponseCode();
+    }
+
+    @Ignore("Need to investigate further, why this does not fail")
+    @Test(expected = SocketException.class)
+    public void testThatIpFilteringIsIntegratedIntoNettyPipelineViaTransportClient() throws Exception {
+        InetSocketTransportAddress transportAddress = (InetSocketTransportAddress) internalCluster().getDataNodeInstance(Transport.class).boundAddress().boundAddress();
+
+        // TODO: This works and I do not understand why, telnet breaks...
+        Socket socket = new Socket(transportAddress.address().getAddress(), transportAddress.address().getPort());
+        socket.getOutputStream().write("foo".getBytes(Charsets.UTF_8));
+        socket.getOutputStream().flush();
+        socket.getInputStream().close();
+        assertThat(socket.isConnected(), is(true));
+        socket.close();
+    }
+}

src/test/java/org/elasticsearch/shield/n2n/N2NNettyUpstreamHandlerTests.java

@@ -0,0 +1,219 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.shield.n2n;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+import com.google.common.net.InetAddresses;
+import org.elasticsearch.common.netty.channel.*;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.env.Environment;
+import org.elasticsearch.test.ElasticsearchTestCase;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.watcher.ResourceWatcherService;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+
+import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilder;
+import static org.hamcrest.Matchers.is;
+
+/**
+ *
+ */
+public class N2NNettyUpstreamHandlerTests extends ElasticsearchTestCase {
+
+    @Rule
+    public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+    private N2NNettyUpstreamHandler nettyUpstreamHandler;
+
+    @Before
+    public void init() throws Exception {
+        File configFile = temporaryFolder.newFile();
+        ResourceWatcherService resourceWatcherService = new ResourceWatcherService(ImmutableSettings.EMPTY, new ThreadPool("resourceWatcher")).start();
+
+        String testData = "allow: 127.0.0.1\ndeny: 10.0.0.0/8";
+        Files.write(testData.getBytes(Charsets.UTF_8), configFile);
+
+        Settings settings = settingsBuilder().put("shield.n2n.file", configFile.getPath()).build();
+        IPFilteringN2NAuthenticator ipFilteringN2NAuthenticator = new IPFilteringN2NAuthenticator(settings, new Environment(), resourceWatcherService);
+
+        nettyUpstreamHandler = new N2NNettyUpstreamHandler(ipFilteringN2NAuthenticator);
+    }
+
+    @Test
+    public void testThatFilteringWorksByIp() throws Exception {
+        InetSocketAddress localhostAddr = new InetSocketAddress(InetAddresses.forString("127.0.0.1"), 12345);
+        assertThat(nettyUpstreamHandler.accept(new NullChannelHandlerContext(), new UpstreamMessageEvent(new NullChannel(), "my message", localhostAddr), localhostAddr), is(true));
+
+        InetSocketAddress remoteAddr = new InetSocketAddress(InetAddresses.forString("10.0.0.8"), 12345);
+        assertThat(nettyUpstreamHandler.accept(new NullChannelHandlerContext(), new UpstreamMessageEvent(new NullChannel(), "my message", remoteAddr), remoteAddr), is(false));
+    }
+
+
+    private static class NullChannelHandlerContext implements ChannelHandlerContext {
+        public boolean canHandleDownstream() {
+            return false;
+        }
+
+        public boolean canHandleUpstream() {
+            return false;
+        }
+
+        public Object getAttachment() {
+            return null;
+        }
+
+        public Channel getChannel() {
+            return null;
+        }
+
+        public ChannelHandler getHandler() {
+            return null;
+        }
+
+        public String getName() {
+            return null;
+        }
+
+        public ChannelPipeline getPipeline() {
+            return null;
+        }
+
+        public void sendDownstream(ChannelEvent e) {
+            // NOOP
+        }
+
+        public void sendUpstream(ChannelEvent e) {
+            // NOOP
+        }
+
+        public void setAttachment(Object attachment) {
+            // NOOP
+        }
+    }
+
+    private static class NullChannel implements Channel {
+        public ChannelFuture bind(SocketAddress localAddress) {
+            return null;
+        }
+
+        public ChannelFuture close() {
+            return null;
+        }
+
+        public ChannelFuture connect(SocketAddress remoteAddress) {
+            return null;
+        }
+
+        public ChannelFuture disconnect() {
+            return null;
+        }
+
+        public ChannelFuture getCloseFuture() {
+            return null;
+        }
+
+        public ChannelConfig getConfig() {
+            return null;
+        }
+
+        public ChannelFactory getFactory() {
+            return null;
+        }
+
+        public Integer getId() {
+            return null;
+        }
+
+        public int getInterestOps() {
+            return 0;
+        }
+
+        public SocketAddress getLocalAddress() {
+            return null;
+        }
+
+        public Channel getParent() {
+            return null;
+        }
+
+        public ChannelPipeline getPipeline() {
+            return null;
+        }
+
+        public SocketAddress getRemoteAddress() {
+            return null;
+        }
+
+        public boolean isBound() {
+            return false;
+        }
+
+        public boolean isConnected() {
+            return false;
+        }
+
+        public boolean isOpen() {
+            return false;
+        }
+
+        public boolean isReadable() {
+            return false;
+        }
+
+        public boolean isWritable() {
+            return false;
+        }
+
+        public ChannelFuture setInterestOps(int interestOps) {
+            return null;
+        }
+
+        public ChannelFuture setReadable(boolean readable) {
+            return null;
+        }
+
+        public ChannelFuture unbind() {
+            return null;
+        }
+
+        public ChannelFuture write(Object message) {
+            return null;
+        }
+
+        public ChannelFuture write(Object message, SocketAddress remoteAddress) {
+            return null;
+        }
+
+        public int compareTo(Channel o) {
+            return 0;
+        }
+
+        public int hashCode() {
+            return 0;
+        }
+
+        public boolean equals(Object o) {
+            return this == o;
+        }
+
+        public Object getAttachment() {
+            return null;
+        }
+
+        public void setAttachment(Object attachment) {
+            // NOOP
+        }
+    }
+}

src/test/java/org/elasticsearch/shield/ssl/SslIntegrationTests.java

@@ -6,7 +6,9 @@
 package org.elasticsearch.shield.ssl;
 
 import com.carrotsearch.ant.tasks.junit4.dependencies.com.google.common.base.Charsets;
+import com.google.common.io.Files;
 import com.google.common.net.InetAddresses;
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.client.Client;
@@ -26,12 +28,11 @@ import org.elasticsearch.test.ElasticsearchIntegrationTest;
 import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.transport.Transport;
 import org.elasticsearch.transport.TransportModule;
-import org.junit.After;
+import org.junit.*;
-import org.junit.Before;
-import org.junit.Test;
 
 import javax.net.ssl.*;
 import java.io.File;
+import java.io.IOException;
 import java.io.InputStreamReader;
 import java.net.HttpURLConnection;
 import java.net.URL;
@@ -45,6 +46,19 @@ import static org.hamcrest.Matchers.*;
 @ClusterScope(scope = Scope.SUITE, numDataNodes = 1, transportClientRatio = 0.0, numClientNodes = 0)
 public class SslIntegrationTests extends ElasticsearchIntegrationTest {
 
+    private static File ipFilterFile = null;
+
+    @BeforeClass
+    public static void writeAllowAllIpFilterFile() {
+        try {
+            ipFilterFile = File.createTempFile("elasticsearch", "ipfilter");
+            ipFilterFile.deleteOnExit();
+            Files.write("allow: all\n".getBytes(com.google.common.base.Charsets.UTF_8), ipFilterFile);
+        } catch (IOException e) {
+            throw new ElasticsearchException("error creating temp file", e);
+        }
+    }
+
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
         File testnodeStore;
@@ -74,6 +88,7 @@ public class SslIntegrationTests extends ElasticsearchIntegrationTest {
                 .put("http.type", NettySSLHttpServerTransportModule.class.getName())
                 .put(TransportModule.TRANSPORT_TYPE_KEY, NettySSLTransportModule.class.getName())
                 .put("plugin.types", SecurityPlugin.class.getName())
+                .put("shield.n2n.file", ipFilterFile.getPath())
                 .build();
     }
 
@@ -215,6 +230,7 @@ public class SslIntegrationTests extends ElasticsearchIntegrationTest {
                 .put("shield.transport.ssl.truststore_password", "testclient")
                 .put("discovery.zen.ping.multicast.ping.enabled", false)
                 .put(TransportModule.TRANSPORT_TYPE_KEY, NettySSLTransportModule.class.getName())
+                .put("shield.n2n.file", ipFilterFile.getPath())
                 //.put("plugin.types", SecurityPlugin.class.getName())
                 .put("cluster.name", internalCluster().getClusterName());
     }