Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
XDP: drop 80/tcp and 8080/tcp
#define KBUILD_MODNAME "xdp_drop_80_8080"
#include <uapi/linux/bpf.h>
#include <linux/in.h>
#include <linux/if_ether.h>
#include <linux/if_packet.h>
#include <linux/if_vlan.h>
#include <linux/ip.h>
#include <linux/ipv6.h>
BPF_HASH(counter, uint32_t, long);
static inline int get_tcp_dest_port(void *data, u64 nh_off, void *data_end) {
struct tcphdr *tcph = data + nh_off;
if (data + nh_off + sizeof(struct tcphdr) > data_end)
return 0;
return tcph->dest;
}
static inline int parse_ipv4(void *data, u64 nh_off, void *data_end) {
struct iphdr *iph = data + nh_off;
if (data + nh_off + sizeof(struct iphdr) > data_end)
return 0;
return iph->protocol;
}
static inline int parse_ipv6(void *data, u64 nh_off, void *data_end) {
struct ipv6hdr *ip6h = data + nh_off;
if (data + nh_off + sizeof(struct ipv6hdr) > data_end)
return 0;
return ip6h->nexthdr;
}
int drop_80_8080(struct xdp_md *ctx) {
void* data_end = (void*)(long)ctx->data_end;
void* data = (void*)(long)ctx->data;
struct ethhdr *eth = data;
int rc_default = XDP_PASS;
long *value;
long zero = 0;
uint16_t h_proto;
uint64_t nh_off = 0;
uint16_t dest_port;
uint32_t index;
nh_off = sizeof(*eth);
if (data + nh_off > data_end)
return rc_default;
h_proto = eth->h_proto;
if (h_proto == htons(ETH_P_IP)) {
h_proto = parse_ipv4(data, nh_off, data_end);
nh_off += sizeof(struct iphdr);
} else if (h_proto == htons(ETH_P_IPV6)) {
h_proto = parse_ipv6(data, nh_off, data_end);
nh_off += sizeof(struct ipv6hdr);
} else {
return rc_default;
}
if (h_proto == IPPROTO_TCP) {
dest_port = ntohs(get_tcp_dest_port(data, nh_off, data_end));
if (dest_port == 80 || dest_port == 8080) {
index = (uint32_t)dest_port;
value = counter.lookup_or_init(&index, &zero);
(*value) += 1;
return XDP_DROP;
}
}
return rc_default;
}
#!/usr/bin/env python
from bcc import BPF
import time
def load_xdp(source_path, func_name, device):
bpf = BPF(src_file=source_path)
func = bpf.load_func(func_name, BPF.XDP)
bpf.attach_xdp(device, func)
return bpf
def drop_packet(device):
bpf = load_xdp("./drop_80_8080.c", "drop_80_8080", device)
print("XDP loaded.")
counter = bpf.get_table("counter")
while True:
try:
for k, v in counter.items(): # type(k) = c_uint, type(v) = c_long
print("{}: {}".format(k.value, v.value))
time.sleep(1)
except KeyboardInterrupt:
break;
bpf.remove_xdp(device)
if __name__ == '__main__':
import sys
if len(sys.argv) != 2:
print("Usage: {} DEVICE".format(sys.argv[0]))
exit(1)
drop_packet(sys.argv[1])
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment