py-pf - Managing OpenBSD's Packet Filter with Python

In this last chapter, we will see how the py-pf module can be effectively used in real-world applications. We will first start by examining the pf.lib module, which provides some higher-level classes designed to make it easier to load rules in Packet Filter; then we will move to a concrete example by implementing a very simple load balancer based on py-pf.

8.1 Keeping things easy with pf.lib

Usually, creating a ruleset is a repetitive task that requires many identical parameters (such as address family, direction or protocol) to be passed repeatedly to all rule instances. The pf.lib module attempts to minimize this overhead by providing some higher-level classes implementing the most common objects in a PF ruleset.

A simple example will best illustrate how these classes can simplify the creation of PF rulesets; we will see how the ruleset we created in the previous chapter can be more easily written by using pf.lib:

import socket
import pf

# Interfaces
ext_if = pf.PFAddr(type=pf.PF_ADDR_DYNIFTL, ifname="sis0")
int_if = "sis1"

# Internal servers
www_srv  = pf.PFRuleAddr(pf.PFAddr("192.168.30.10"), pf.lib.TCPPort(80))
smtp_srv = pf.PFRuleAddr(pf.PFAddr("192.168.30.11"), pf.lib.TCPPort(25))

# NAT outgoing connections
# rule: match out on $ext_if inet from !($ext_if) to any nat-to ($ext_if)
r0 = pf.lib.MatchOutRule(ifname=ext_if.ifname,
                         src=PFRuleAddr(addr=ext_if, neg=True),
                         nat=pf.lib.NATPool(ext_if))

# Redirect web services (with load balancing)
# rule: match in on $ext_if inet proto tcp from any to ($ext_if) port $www_prt \
#           rdr-to $www_srv round-robin sticky-address
r1 = pf.lib.MatchInRule(ifname=ext_if.ifname,
                        dst=pf.PFRuleAddr(ext_if, www_srv.port),
                        rdr=pf.lib.RDRPool(www_srv.addr,
                             opts=pf.PF_POOL_ROUNDROBIN|pf.PF_POOL_STICKYADDR))

# Default deny
# rule: block drop all
r2 = pf.lib.BlockRule()

# Spoofed address protection
# rule: block drop in quick from urpf-failed
r3 = pf.lib.BlockInRule(quick=True,
                        src=pf.PFRuleAddr(pf.PFAddr("urpf-failed")))

# Allow traffic to web server
# rules: pass in on $ext_if inet proto tcp from any to $www_srv port $www_prt synproxy state
#        pass out on $int_if inet proto tcp from any to $www_srv port $www_prt
r4 = pf.lib.PassInRule(ifname=ext_if.ifname, dst=www_srv,
                       keep_state=pf.PF_STATE_SYNPROXY)
r5 = pf.lib.PassOutRule(ifname=int_if, dst=www_srv)

# Allow incoming "unreach code needfrag" ICMP packets and all outgoing ICMP traffic.
# rules: pass in  inet proto icmp all icmp-type unreach code needfrag
#        pass out inet proto icmp all
r6 = pf.lib.PassInRule(proto=socket.IPPROTO_ICMP, code="needfrag")
r7 = pf.lib.PassOutPFRule(proto=socket.IPPROTO_ICMP)

# Allow smtp traffic from all except for addresses in the <spammers> table
# rules: table <spammers> persist
#        pass in on $ext_if inet proto tcp from !<spammers> to $smtp_srv port $smtp_prt
#        pass out on $int_if inet proto tcp from !<spammers> to $smtp_srv port $smtp_prt
t1 = pf.PFTable("spammers", flags=pf.PFR_TFLAG_PERSIST)
r8 = pf.lib.PassInRule(ifname=ext_if.ifname,
                       src=pf.PFRuleAddr(pf.PFAddr("<{0}>".format(t1.name)), neg=True),
                       dst=smtp_srv)
r9 = pf.lib.PassOutRule(ifname=int_if,
                        src=pf.PFRuleAddr(pf.PFAddr("<{0}>".format(t1.name)), neg=True),
                        dst=smtp_srv)