Integrating Fail2Ban with AWS Network ACLs

Posted in AWS Blog
05/10/2018 Rutger Beyen

I was recently working on a project where I couldn’t lock down the Bastion instance security group ingress rule to only allow whitelisted IP addresses. Several coworkers work from home and use the Bastion to jump into backend servers and create SSH tunnels, while they did not have AWS Console access to whitelist themselves. The security group ended up allowing on port 22. Off course the Bastion instance would only allow ssh key based logins, but all protocols eventually have vulnerabilities. So installing multiple layers of security in depth prevents the systems from being directly affected by the latest individual vulnerabilities.

Combining some bits and pieces from Google allowed me to setup Fail2Ban on the Bastion instance, while the blocking of the IPs is done in AWS NACLs in stead of the local Iptables. The setup has been done on an AmazonLinux instance.


AWS NACLs by default only allow 20 ingress and 20 egress rules. This is a soft limit, and you can have it increased to 40 by opening a support case (40 seems to be the upper hard limit).

If fail2ban wants to add another rule while the maximum has been reached, it will block the offender in the local Iptables. Running my implementation in several environments for a few weeks now, I never had more than 4 to 5 IPs blocked at the same time. Unless you are off course the victim of a targetted ddos attack…

Setup the AWS CLI

Make sure you have a correctly installed and working AWS CLI on the instance. Also make sure the EC2 role has the necessary permissions to modify the EC2 Network ACL. Testing it out:

1. Get the current MAC address of the first interface

INTERFACE=$(curl --silent

2. Get the subnet ID from the MAC address

SUBNET_ID=$(curl --silent${INTERFACE}/subnet-id)

3. Get the current Network ACL ID

ACL_ID=$(aws ec2 describe-network-acls --filters Name=association.subnet-id,Values=$SUBNET_ID | jq '.NetworkAcls[0].Associations[0].NetworkAclId' | sed 's/"//g')

4. Test that you can add a rule to the ACL

aws ec2 create-network-acl-entry --network-acl-id $ACL_ID --ingress --rule-number 1 --protocol tcp --port-range From=0,To=65535 --cidr-block --rule-action deny

5. Verify that the above IP has been blocked

aws ec2 describe-network-acls --filters,Values=$ACL_ID

6. Remove the rule again

aws ec2 delete-network-acl-entry --network-acl-id $ACL_ID --ingress --rule-number 1

Fail2Ban AWS integration

1. Install the necessary packages (if not yet present)

pip install requests boto3 tabulate
yum install sqlite

2. Create a directory to store the AWS NACL script and cd to it

mkdir /opt/aws-nacl
cd /opt/aws-nacl

3. Place the following content in the file in the above directory

"""This script is used to block and unblock IPs on Amazon EC2
network ACLs and can be used with Fail2Ban. Since only 20
inbound rules are allowed with AWS if a 'jail' is provided
the IP will be blocked on the host iptables if full"""
import json
import sqlite3
import argparse
import os
import pprint
import socket
import logging
import logging.handlers
import requests
import boto3
import subprocess
from tabulate import tabulate
#AWS only allows 20 inbound subtract default ACL rules from 20 for max
#Set rule start, by default AWS ACL starts rules at 100
#Set range for rules: highest rule ID that can be used
def check_block(ip,acl):
   acl = get_acl(acl)
   list = acl['NetworkAcls'][0]['Entries']
   for entry in list:
      if ip in entry["CidrBlock"]:
         return True
   return False
def get_acl(acl_id):
    """This function gets the ACL given an ec2 object and ACL id"""
    ec2 = boto3.client('ec2')
    acl_response = ec2.describe_network_acls(
    return acl_response
def print_inbound_acl(acl_id):
   blocks = []
   table = {num:name[8:] for name,num in vars(socket).items() if name.startswith("IPPROTO")}
   acl = get_acl(acl_id)
   list = acl['NetworkAcls'][0]['Entries']
   for entry in list:
     if not entry["Egress"]:
         if "PortRange" in entry:
                ports = ({"To":entry["PortRange"]["To"], "From":entry["PortRange"]["From"]})
                ports = ({"To":"", "From":""})
         if entry['Protocol'] == "-1":
                proto = "all"
                proto = table[int (entry['Protocol'])]
   print "Inbound Network ACL"
   print tabulate(blocks,headers=["Rule","Protocol","CIDR","Port From","Port To","Action"])
def is_acl(acl):
    ec2 = boto3.client('ec2')
        return True
    except Exception:
        return False
def get_acl_id():
    ec2 = boto3.client('ec2')
    meta = ""
    mac = requests.get(meta).text
    subnet = requests.get(meta+mac+"/subnet-id").text
    response = ec2.describe_network_acls(
                'Name': 'association.subnet-id',
    return response['NetworkAcls'][0]['Associations'][0]['NetworkAclId']
def validate_ip(ip_address):
    ip_split = ip_address.split('.')
    if len(ip_split) != 4:
        return False
    for octet in ip_split:
        if not octet.isdigit():
            return False
        octet_int = int(octet)
        if octet_int < 0 or octet_int > 255:
            return False
        return True
    except socket.error:
        return False
def sqlite_connect(file_name):
    make_table = '''CREATE TABLE if not exists blocks (id integer PRIMARY KEY AUTOINCREMENT,
               ip text NOT NULL, acl text NOT NULL, blocked boolean NOT NULL,host boolean
               NOT NULL)'''
    if not os.path.isfile(file_name):
        dir_path = os.path.dirname(os.path.realpath(__file__))
        conn = sqlite3.connect("{}/{}".format(dir_path, file_name))
        cursor = conn.cursor()
            conn = sqlite3.connect(file_name)
            cursor = conn.cursor()
        except Exception:
            print "Datatbase File is encrypted or is not a database"
    return conn
def main():
    my_logger = logging.getLogger(__file__)'Checking arguments')
    parser = argparse.ArgumentParser(description="Script to block IPs on AWS EC2 Network ACL")
    parser.add_argument('-a', '--acl', help='ACL ID')
    parser.add_argument('-j', '--jail', help='Fail2Ban Jail')
    parser.add_argument('-d', '--db', default='aws-nacl.db', help='Database')
    parser.add_argument('-b', '--block', metavar="IP", help='Block IP address')
    parser.add_argument('-u', '--unblock', metavar="IP", help='Unblock IP address')
    parser.add_argument('-g', '--get', action='store_true', help='Get ACL')
    parser.add_argument('-v', '--verbose', action='store_true', help='Verbose logging')
    args = parser.parse_args()
    ec2_resource = boto3.resource('ec2')
    pretty_printer = pprint.PrettyPrinter(indent=4)
    if args.verbose:'Setting logging to debug')
    if (args.block and args.unblock):
        my_logger.error('Invalid arguments')
    if args.acl:'Checking if valid AWS Network ACL')
        if not is_acl(args.acl):
            print('Invalid Network ACL ID')
            my_logger.error('Invalid Network ACL')
    else:'Searching for current ACL ID')
        acl = get_acl_id()
        network_acl = ec2_resource.NetworkAcl(acl)
        my_logger.debug('Network ACL ID: {}'.format(network_acl))
    if args.get or (not args.block and not args.unblock):'Printing ACL')
        exit(0)'Configuring DB')
    conn = sqlite_connect(args.db)
    cursor = conn.cursor()
    if args.block:'Checking if valid IP')
        if not validate_ip(args.block):
            print "IP {} is invalid".format(args.ip)
            exit(1)'Searching DB for IP: {}'.format(args.block))
        cursor.execute('''select count (*) from blocks where ip=? and blocked=1''', (args.block,))
        if cursor.fetchone()[0] > 0:
            print "IP {} already blocked".format(args.block)
            exit(0)'Checking AWS block count')
        cursor.execute('''select count (*) from blocks where blocked=1 and host =0''')
        block_count = cursor.fetchone()[0]
        my_logger.debug('Currently {} IPs blocked'.format(block_count))
        if block_count <= MAX_BLOCKS:
            my_logger.debug('Current blocks less then Max: {}'.format(MAX_BLOCKS))
  'Adding block to the DB')
            cursor.execute('''insert into blocks (ip, acl, blocked,host)
                               values (?,?,?,?)''', (args.block, acl, 1, 0))
  'Caculating Rule number based on DB ID')
            cursor.execute('''select seq from sqlite_sequence where name="blocks"''')
            rule_num = cursor.fetchone()[0] % RULE_RANGE + RULE_BASE
  'Adding Network ACL')
                    'From': 0,
                    'To': 65535
            if not check_block(args.block, acl):
               my_logger.error('Failed to block IP {} in AWS ACL'.format(args.block))
               cursor.execute('''UPDATE blocks SET blocked = 0 where ip=? and
                              blocked=1''', (args.block,))
            my_logger.debug('Max blocks on AWS Network ACL, checking for IPTables')
            if  args.jail:
      'Blocking IP {} in f2b-{}'.format(args.block,args.jail))
                iptables = "/sbin/iptables -w -I {} 1 -s {} -j REJECT".format(args.jail, args.block)
                print iptables
      , shell=True)
                cursor.execute('''insert into blocks (ip, acl, blocked,host)
                                  values (?,?,?,?)''', (args.block, '', 1, 1))
                my_logger.error('No IPtables Chain set, IP will not be blocked')
    if args.unblock:'Checking if valid IP')
        if not validate_ip(args.unblock):
            my_logger.error("IP {} is invalid".format(args.unblock))
            exit(1)'Checking for IP in the DB')
        test = 'select id, host from blocks where ip="{}" and blocked=1'.format(args.unblock)
        results = cursor.fetchone()
        if results is not None:
  'Found IP, getting rule number from DB')
            if results[1] == 0:
                rule_num = results[0] % RULE_RANGE + RULE_BASE
                my_logger.debug('Rule number is {}'.format(rule_num))
      'Deleting rule from AWS Network ACL')
                response = network_acl.delete_entry(
      'Updating DB')
                cursor.execute('''UPDATE blocks SET blocked = 0 where ip=? and
                                   blocked=1''', (args.unblock,))
                if args.jail:
          'Unblocking IP {} in f2b-{}'.format(args.unblock,args.jail))
                    iptables = 'iptables -w -D {} -s {} -j REJECT'.format(args.jail, args.unblock)
          , shell=True)
                    cursor.execute('''UPDATE blocks SET blocked = 0 where ip=? and blocked=1''', (args.unblock,))
            my_logger.error("IP {} not in blocks database".format(args.unblock))
if __name__ == "__main__":

4. Test the script by calling it

python -d aws-nacl.db -b -v

5. Verify that the above IP was added to the ACL

python -g

6. Remove the IP again

python -d aws-nacl.db -u -v

7. Verify that the IP was removed again

python -g


1. Install Fail2Ban itself

yum install fail2ban

2. Place the following file under /etc/fail2ban/action.d/aws.conf

# Fail2Ban configuration file
# Author: Cyril Jaquier
# Modified by Yaroslav Halchenko for multiport banning
# Modified by Ryan for AWS Network ACL block
before = iptables-blocktype.conf
# Option:  actionstart
# Notes.:  command executed once at the start of Fail2Ban.
# Values:  CMD
actionstart = iptables -N fail2ban-<name>
              iptables -A fail2ban-<name> -j RETURN
              iptables -I <chain> -p <protocol> --dport <port> -j fail2ban-<name>
# Option:  actionstop
# Notes.:  command executed once at the end of Fail2Ban
# Values:  CMD
actionstop = iptables -D <chain> -p <protocol> --dport <port> -j fail2ban-<name>
             iptables -F fail2ban-<name>
             iptables -X fail2ban-<name>
# Option:  actioncheck
# Notes.:  command executed once before each actionban command
# Values:  CMD
actioncheck = iptables -n -L <chain> | grep -q 'fail2ban-<name>[ \t]'
# Option:  actionban
# Notes.:  command executed when banning an IP. Take care that the
#          command is executed with Fail2Ban user rights.
# Tags:    See jail.conf(5) man page
# Values:  CMD
actionban = python /opt/aws-nacl/ -d /opt/aws-nacl/aws-nacl.db -v -b <ip> -j fail2ban-<name>
# Option:  actionunban
# Notes.:  command executed when unbanning an IP. Take care that the
#          command is executed with Fail2Ban user rights.
# Tags:    See jail.conf(5) man page
# Values:  CMD
actionunban = python /opt/aws-nacl/ -d /opt/aws-nacl/aws-nacl.db -v -u <ip> -j fail2ban-<name>
# Default name of the chain
name = default
# Option:  port
# Notes.:  specifies port to monitor
# Values:  [ NUM | STRING ]  Default:
port = ssh
# Option:  protocol
# Notes.:  internally used by config reader for interpolations.
# Values:  [ tcp | udp | icmp | all ] Default: tcp
protocol = tcp
# Option:  chain
# Notes    specifies the iptables chain to which the fail2ban rules should be
#          added
# Values:  STRING  Default: INPUT
chain = INPUT

3. Modify /etc/fail2ban/fail2ban.conf and change the logfile location (it’s easier to have a separate log rather then searching through /var/log/messages)

logtarget = /var/log/fail2ban.log

4. Add a file /etc/fail2ban/jail.local with the following content. Modify the values at your own convenience

#Localhost and your office HQ range
ignoreip =

action = aws[name=SSH, port=ssh, protocol=tcp]

# "bantime" is the number of seconds that a host is banned.
bantime  = 3600

# "maxretry" is the number of failures before a host get banned.
maxretry = 2

5. Check /etc/fail2ban/filter.d/sshd.conf that the correct matching patterns for the /var/log/secure logfile are present, so that it actually looks for the loglines that you want to be considered as malicious. On an Amazon Linux, I have the following in place

failregex = ^%(__prefix_line)s(?:error: PAM: )?[aA]uthentication (?:failure|error) for .* from <HOST>( via \S+)?\s*$
            ^%(__prefix_line)s(?:error: PAM: )?User not known to the underlying authentication module for .* from <HOST>\s*$
            ^%(__prefix_line)sFailed \S+ for .* from <HOST>(?: port \d*)?(?: ssh\d*)?\s*$
            ^%(__prefix_line)sROOT LOGIN REFUSED.* FROM <HOST>\s*$
            ^%(__prefix_line)s[iI](?:llegal|nvalid) user .* from <HOST> .*$
            ^%(__prefix_line)sUser .+ from <HOST> not allowed because not listed in AllowUsers\s*$
            ^%(__prefix_line)sUser .+ from <HOST> not allowed because listed in DenyUsers\s*$
            ^%(__prefix_line)sUser .+ from <HOST> not allowed because not in any group\s*$
            ^%(__prefix_line)srefused connect from \S+ \(<HOST>\)\s*$
            ^%(__prefix_line)sUser .+ from <HOST> not allowed because a group is listed in DenyGroups\s*$
            ^%(__prefix_line)sUser .+ from <HOST> not allowed because none of user's groups are listed in AllowGroups\s*$
            ^%(__prefix_line)sReceived disconnect from <HOST> port \d*:11: Bye Bye \[preauth\]

6. Add fail2ban to the startup list and start it

chkconfig --add fail2ban
chkconfig fail2ban on
service fail2ban start

Locked myself out

Things can always go wrong, but thanks to the recently released AWS feature called ‘SSM Session Manager’ you can always get a console window on your instance to start troubleshooting. One thing to make sure is that your instance is running the latest version of the AWS SSM Agent. So it’s always a good idea to update it before closing your current SSH session:

yum install

You also need to make sure that your instance is allowed to communicate with the AWS SSM service. Easiest way is to attach the ‘AmazonEC2RoleforSSM‘ policy to your EC2 role.


, , , , , ,

Leave a Reply

Your email address will not be published.


Need a hand? Or a high five?
Feel free to visit our offices and come say hi
… or just drop us a message

We are ready when you are

Cloudar NV – Operations

Prins Boudewijnlaan 24B
2550 Kontich (Antwerp)

info @

+32 3 450 67 18

Cloudar NV – HQ

Veldkant 33A
2550 Kontich (Antwerp)

VAT BE0564 763 890

    This contact form is protected by reCAPTCHA and the Google Privacy Policy and Terms of Service apply.

    © 2022 – CLOUDAR NV

    • SHARE