A Python 3 libnetfilter_queue handler intended to mitigate CPU load covert channels based on ping timings

README.md

Usage

To install the dependencies, run the script install_deps.sh as root.

To set up the filter, run setup_iptables.sh as root. You may need to change the queue-num parameter if you already have some NFQUEUE-based iptables rules.

To run the filter, run python3 main.py as root. If you used a queue number other than 0, you'll need to pass it as the first parameter to main.py.

You can remove all iptables rules by running iptables --flush as root. However, note that this will remove all iptables rules.

If you're SSH-ed into a machine, make sure you don't require any intervention inbetween setting up the filter and actually running it, or you'll lose connection. After you do set it up, you need to wait until your SSH socket times out and your client opens a new one.

WARNING: This software is provided "as is", without warranty of any kind. See the top of main.py for details.

install_deps.sh

#!/bin/bash
# The author (Ethan White) of this code dedicates any and all copyright interest
# in this code to the public domain. The author makes this dedication for the
# benefit of the public at large and to the detriment of his heirs and
# successors. The author intends this dedication to be an overt act of
# relinquishment in perpetuity of all present and future rights to this code
# under copyright law.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR
# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

echo "Note: If not on a Debian-based Linux, you're on your own."
echo "Detailed instructions can be found at https://github.com/kti/python-netfilterqueue."
echo ""

if [[ ! $EUID -eq 0 ]]; then
    >&2 echo "Error: This script must be run as root"
    exit 1
fi

apt-get install build-essential python3-dev libnetfilter-queue-dev
pip3 install NetfilterQueue

main.py

# The author (Ethan White) of this code dedicates any and all copyright interest
# in this code to the public domain. The author makes this dedication for the
# benefit of the public at large and to the detriment of his heirs and
# successors. The author intends this dedication to be an overt act of
# relinquishment in perpetuity of all present and future rights to this code
# under copyright law.
#
# THIS SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR
# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

import netfilterqueue
import sys
import threading
import time
import binascii
import random

rand = random.SystemRandom()

INTERVAL_MIN = 0.075
INTERVAL_MAX = 0.2
QUEUE_NUM = 0

class PacketHandlerThread(threading.Thread):
    def __init__(self, handler):
        threading.Thread.__init__(self)
        self.handler = handler
        self.stop_requested = False
        self.on_stop = None

    def run(self):
        while True:
            t = INTERVAL_MIN + (INTERVAL_MAX - INTERVAL_MIN) * rand.random()
            print("Sleeping for %sms" % t)
            time.sleep(t)

            for packet in self.handler.packets:
                packet.accept()

            self.handler.packets = []

            if self.stop_requested:
                break

        if self.on_stop is not None:
            self.on_stop()

    def request_stop(self, on_stop):
        self.on_stop = on_stop
        self.stop_requested = True

class PacketHandler:
    def __init__(self):
        self.packets = []
        self.queue = netfilterqueue.NetfilterQueue()
        self.handler_thread = PacketHandlerThread(self)

    def run(self):
        self.queue.bind(QUEUE_NUM, handler.handle)
        self.handler_thread.start()
        self.queue.run()

    def handle(self, packet):
        self.packets.append(packet)

    def cleanup(self, on_done):
        self.queue.unbind()
        self.handler_thread.request_stop(on_done)

if len(sys.argv) > 1:
    try:
        QUEUE_NUM = int(sys.argv[1])
    except ValueError:
        print("WARN: Got invalid first argument, ignoring (expected number)")

handler = PacketHandler()

try:
    handler.run()
except KeyboardInterrupt:
    sys.stdout.write("Shutting down...")
    handler.cleanup(lambda: print("Bye!"))

setup_iptables.sh

#!/bin/bash
# The author (Ethan White) of this code dedicates any and all copyright interest
# in this code to the public domain. The author makes this dedication for the
# benefit of the public at large and to the detriment of his heirs and
# successors. The author intends this dedication to be an overt act of
# relinquishment in perpetuity of all present and future rights to this code
# under copyright law.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR
# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.

# Note: Add -s [IP ADDRESS] to limit to only a particular IP address (useful for testing)
# Example: iptables -A INPUT -s 8.8.8.8 -j NFQUEUE --queue-num 0

iptables -A INPUT -j NFQUEUE --queue-num 0
iptables -A OUTPUT -j NFQUEUE --queue-num 0

ip6tables -A INPUT -j NFQUEUE --queue-num 0
ip6tables -A OUTPUT -j NFQUEUE --queue-num 0