Prototype code using nfct to do a userspace conntracker.

3 files changed, 459 insertions, 0 deletions

diff --git a/lua/echoclient b/lua/echoclient
new file mode 100755
index 0000000..556021b
--- /dev/null
+++ b/lua/echoclient
@@ -0,0 +1,74 @@
+#!/usr/bin/env lua5.1
+
+-- echoclient host ctrl_port string1 string2....
+--   request one port
+--   connect to requested port
+--   send/receive each string in turn
+--   close echo port
+--   close control port
+
+require"socket"
+
+local host = arg[1]
+local ctrl_port = arg[2]
+
+if #arg < 3 then
+    print("Usage: " .. arg[0] .. " host ctrl_port string1 [string2 string3 ...]")
+    os.exit(0)
+end
+
+function connect(host, port)
+    local s = socket.tcp()
+    assert(s:settimeout(5))
+    assert(s:connect(host, port))
+    print("Connected to:", host, port, "from", s:getsockname())
+    return s
+end
+
+local function get_echo_port(ctrl_sock)
+    print"Get port..."
+    assert(ctrl_sock:send("GET\n"))
+    local port = assert(ctrl_sock:receive())
+    print("Got port:", port)
+    return tonumber(port)
+end
+
+local function send_strings(host, port)
+    print("Connect to echo port:", host, port)
+    local s = connect(host, port)
+    for i = 3, #arg do
+        print(i, "Send:", arg[i])
+        assert(s:send(arg[i].."\n"))
+        local data, err = assert(s:receive())
+        print(i, "Recv:", data)
+    end
+    return s
+end
+
+local function pause()
+    os.execute"sleep 1" -- pause so conntrack -L can show the expectations
+end
+
+print("Connect to broker...")
+
+local ctrl_sock = connect(host, ctrl_port)
+
+echo_port = get_echo_port(ctrl_sock)
+
+ctrl_sock:close()
+
+echo_sock1 = send_strings(host, echo_port, "first")
+
+pause()
+
+echo_sock2 = send_strings(host, echo_port, "second")
+
+echo_sock2:close()
+echo_sock1:close()
+
+print("Use echo port as long as we can...")
+
+while send_strings(host, echo_port, "... as long as we can") do
+    pause()
+end
+
diff --git a/lua/echoconntracker b/lua/echoconntracker
new file mode 100755
index 0000000..df622cc
--- /dev/null
+++ b/lua/echoconntracker
@@ -0,0 +1,267 @@
+#!/usr/bin/env lua5.1
+--[[
+Use a combination of nfq and nfct to do userspace connection tracking for a sample RPC-like
+service over TCP, that uses ephemeral persistent ports.
+
+The server is the echoserver running on localhost, client is the echoclient.
+
+Start server:
+
+    ./echoserver 9999
+
+Test client:
+
+    ./echoclient localhost 9999 hello world
+
+Kill it after a few connections.
+
+Start conntracker:
+
+    sudo ./echoconntracker port=9999 verbose=y
+
+Try client again... kill conntracker...
+
+To clear the conntracker's rules:
+
+    sudo ./echoconntracker port=9999 verbose=y clear=y
+
+]]
+
+require"nfct"
+require"nfq"
+require"net"
+
+local function debug(...) end
+local function verbose(...) end
+
+-- arguments
+
+function usage(k)
+    if arg[k] then
+        return
+    end
+    print("arg '"..k.."' not provided")
+    print("usage "..arg[0].." port=num [clear=y] [verbose=y|very]")
+    os.exit(1)
+end
+
+for i,a in ipairs(arg) do
+    local s,e,k,v = a:find("^([^=]+)=(.*)$")
+    arg[k] = v
+end
+
+usage"port"
+
+if arg.verbose then verbose = print end
+if arg.verbose == "very" then
+    debug = print
+end
+
+-- iptables rule setup
+
+local function execute(cmd)
+    if type(cmd) == "table" then
+        cmd = table.concat(cmd, " ")
+    end
+    print("cmd=<"..cmd..">")
+    local status = os.execute(cmd)
+    if status == 0 then
+        return status
+    end
+    return nil, string.format("%q", cmd).." failed with "..status
+end
+
+function clear_filter(clear)
+    execute"iptables -L -n"
+    execute"iptables -P INPUT ACCEPT"
+    execute"iptables -P OUTPUT ACCEPT"
+    execute"iptables -P FORWARD ACCEPT"
+    execute"iptables -F"
+    execute"iptables -L -n"
+    if clear then
+        os.exit(0)
+    end
+end
+
+--[[
+What table means:
+
+FORWARD - packets that traverse
+OUTPUT  - packets we generate
+INPUT   - packets we receive
+
+Client/Server on localhost:
+
+./echoportbroker 9999
+while ./echoclient localhost 9999 hello world; do sleep 5; done
+sudo ./inline-listen host=127.0.0.1 port=9999 table=OUTPUT
+
+]]
+
+-- TODO this assumes host is ourselves, tables would have to be different
+-- if it was remote, or we were a firewall/forwarding traffic
+function set_filter(port)
+    local execute = function (cmd)
+        assert(execute(cmd))
+    end
+    -- default for input is to DROP
+    execute{
+        "iptables -t filter",
+        "-P INPUT DROP",
+    }
+    -- outgoing responses from server are queued
+    execute{
+        "iptables -t filter",
+        "-A OUTPUT",
+        "-p tcp",
+        "--sport "..port,
+        "-j QUEUE" -- queue 0 is implicit
+    }
+    -- incoming established connections accepted
+    execute{
+        "iptables -t filter",
+        "-A INPUT",
+        "-p tcp",
+        "-m state --state RELATED,ESTABLISHED",
+        "-j ACCEPT"
+    }
+    -- outgoing new and established connections accepted
+    execute{
+        "iptables -t filter",
+        "-A OUTPUT",
+        "-p tcp",
+        "-m state --state NEW,RELATED,ESTABLISHED",
+        "-j ACCEPT"
+    }
+    -- incoming connections to server are accepted
+    execute{
+        "iptables -t filter",
+        "-A INPUT",
+        "-p tcp",
+        "--dport "..port,
+        "-m state --state NEW",
+        "-j ACCEPT",
+    }
+
+    execute"iptables -L -n"
+end
+
+clear_filter(arg.clear)
+
+set_filter(arg.port)
+
+
+-- nfct helpers
+
+local function ctprint(ct, name, ...)
+    print("ct="..nfct.tostring(ct).." -- "..name, ...)
+end
+
+local function expprint(exp, name, ...)
+    print("exp="..nfct.exp_tostring(exp).." -- "..name, ...)
+end
+
+local function check(...)
+    if (...) then
+        return ...
+    end
+    local _, emsg, eno = ...
+    local emsg = "["..tostring(eno).."] "..tostring(emsg)
+    return assert(_, emsg)
+end
+
+local function tuple(name, src, dst, sport, dport)
+    local ct = assert(nfct.new())
+
+    nfct.set_attr_pf(ct, "l3proto",  "inet")
+    nfct.set_attr_ipv4(ct, "ipv4-src", src)
+    nfct.set_attr_ipv4(ct, "ipv4-dst", dst)
+
+    nfct.set_attr_ipproto(ct, "l4proto",  "tcp")
+
+    if sport then
+        nfct.set_attr_port(ct, "port-src", sport)
+    end
+
+    nfct.set_attr_port(ct, "port-dst", dport)
+
+    ctprint(ct, name)
+
+    return ct
+end
+
+local function expect(src, dst, sport, dport, expectport)
+    -- identify the master to which this expectation is related
+    local master = tuple("master", src, dst, sport, dport)
+    local expected = tuple("expected", src, dst, nil, expectport)
+    local mask = tuple("mask", 0xffffffff, 0xffffffff, nil, expectport)
+    local timeout = 10 -- seconds FIXME we need this to be longer than the real server's timeout
+    local exp = assert(nfct.exp_new(master, expected, mask, timeout, "permanent"))
+
+    nfct.destroy(master)
+    nfct.destroy(expected)
+    nfct.destroy(mask)
+
+    expprint(exp, "expectation")
+
+    local h = assert(nfct.open("expect"))
+
+    -- FIXME this can fail if conntrack hasn't tracked the master... but is that possible? we just
+    -- got data from nfq, the connection must exist
+    check(nfct.exp_query(h, "create", exp))
+
+    nfct.exp_destroy(exp)
+
+    nfct.close(h)
+end
+
+-- Expectation tracking
+
+local qhandle = assert(nfq.open())
+
+nfq.unbind_pf(qhandle, "inet")
+nfq.bind_pf(qhandle, "inet")
+
+local queue = assert(nfq.create_queue(qhandle, 0))
+
+assert(nfq.set_mode(queue, "packet"))
+
+local n = net.init()
+
+nfq.catch(qhandle, function (nfqdata)
+    debug("CB nfq")
+
+    local inip = assert(nfq.get_payload(nfqdata))
+
+    n:clear()
+    n:decode_ip(inip)
+
+    local _, tcp = pcall(n.get_tcp, n)
+    local _, ip = pcall(n.get_ipv4, n)
+
+    if not tcp or not ip then
+        -- not of requested protocol
+        debug("ignore protocol", n:dump())
+        return "accept"
+    end
+
+    -- Original connection was client->server, and this packet is
+    -- server->client, so reverse src and dst
+    local src, dst, sport, dport = ip.dst, ip.src, tcp.dst, tcp.src
+    local indata = tcp.payload
+
+    if indata then
+        debug("data", indata)
+        local expectport = tonumber(indata)
+        if expectport ~= nil then
+            verbose("Q", "master", src, dst, sport, dport)
+            verbose("Q", "expect", src, dst, "*", expectport)
+            expect(src, dst, sport, dport, expectport)
+        end 
+    else
+        debug("Q", "flags", string.format("%#x", tcp.flags), "(non-data)")
+    end
+
+    return "accept"
+end)
+
diff --git a/lua/echoserver b/lua/echoserver
new file mode 100755
index 0000000..5f94e28
--- /dev/null
+++ b/lua/echoserver
@@ -0,0 +1,118 @@
+#!/usr/bin/env lua5.1
+
+require"socket"
+
+--[[
+FIXME
+
+control sockets are never closed, they leak
+
+echo listen sockets are never close, they listen forever
+
+]]
+
+local TIMEOUT = 0.1
+
+local ctrl_port = arg[1]
+
+if not ctrl_port then
+    print("Usage: " .. arg[0] .. " control_listen_port")
+    os.exit(0)
+end
+
+local ctrl_listen_sock = assert(socket.bind("*", ctrl_port))
+ctrl_listen_sock:settimeout(TIMEOUT)
+print("Waiting for connection...")
+
+local mon_socks = {ctrl_listen_sock} -- all socks, starting with just server ctrl sock
+local echo_listen_socks = {} -- echo listen socks
+local echo_data_socks = {} -- echo data socks
+local ctrl_data_socks = {} -- client ctrl socks
+
+local function remove(t, item)
+    for i, v in pairs(t) do
+        if v == item then
+            table.remove(t, i)
+        end
+    end
+end
+
+local function generate_echo_listen_socks(client_ctrl)
+    while true do
+        local part, msg = client_ctrl:receive()
+        -- read data and if any obtained, create socket, send port number
+        if part then
+            local echo_sock = assert(socket.bind("*", 0))
+            assert(echo_sock:settimeout(TIMEOUT))
+            table.insert(mon_socks, echo_sock)
+            table.insert(echo_listen_socks, echo_sock)
+            local ip, port = echo_sock:getsockname()
+            print("Sending port " .. port)
+            assert(client_ctrl:send(port.."\n"))
+        elseif msg == "closed" then
+            -- client closed control connection
+            print("Removing closed control data socket")
+            remove(ctrl_data_socks, client_ctrl)
+            remove(mon_socks, client_ctrl)
+            return
+        else
+            -- client doesn't need any more echo ports
+            return
+        end
+    end
+end
+
+while true do
+    print("Waiting for activity on " .. #mon_socks .. " sockets")
+    local readsocks = socket.select(mon_socks)
+    for _, sock in ipairs(readsocks) do
+        -- control listen sock
+        if sock == ctrl_listen_sock then
+            print("Handling control listen sock activity")
+            local client_ctrl = ctrl_listen_sock:accept()
+            table.insert(mon_socks, client_ctrl)
+            table.insert(ctrl_data_socks, client_ctrl)
+            client_ctrl:settimeout(TIMEOUT)
+            print("Control connection obtained from: ")
+            print(client_ctrl:getpeername())
+            generate_echo_listen_socks(client_ctrl)
+        else
+            -- control data socks
+            for _, cc_sock in ipairs(ctrl_data_socks) do
+                if sock == cc_sock then
+                    print("Handling control data sock activity")
+                    generate_echo_listen_socks(cc_sock)
+                end
+            end
+            -- echo listen socks
+            for _, esock in ipairs(echo_listen_socks) do
+                if sock == esock then
+                    print("Handling echo listen sock activity")
+                    local client = sock:accept()
+                    client:settimeout(TIMEOUT)
+                    table.insert(mon_socks, client)
+                    table.insert(echo_data_socks, client)
+                    print("Echo connection obtained from: ")
+                    print(client:getpeername())
+                end
+            end
+            -- echo data socks
+            for i, csock in ipairs(echo_data_socks) do
+                if sock == csock then
+                    print("Handling client sock activity")
+                    local part, err = sock:receive()
+                    if part then
+                        print("Echo: " .. part)
+                        sock:send(part.."\n")
+                    elseif err == "closed" then
+                        print("Removing closed echo data socket")
+                        table.remove(echo_data_socks, i)
+                        remove(mon_socks, sock)
+                    else
+                        print("Echo data error: "..err)
+                    end
+                end
+            end
+        end
+    end
+end