1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
#!/usr/bin/env python3
import http.server
import os
import re
import socketserver
import subprocess
SOCK_PATH = "/var/run/pfban/ban.sock"
PF_TABLE = "blocked"
# Simple, conservative filter to avoid junk / injection
IP_RE = re.compile(r"^[0-9A-Fa-f:.]{3,}$")
def ensure_socket_dir(path: str) -> None:
os.makedirs(os.path.dirname(path), mode=0o755, exist_ok=True)
class BanHandler(http.server.BaseHTTPRequestHandler):
# Silence default logging
def log_message(self, format, *args):
return
def do_POST(self):
if self.path != "/ban":
self.send_response(404)
self.end_headers()
return
ip = self.headers.get("X-IP", "").strip()
if ip and IP_RE.match(ip):
subprocess.run(
["/sbin/pfctl", "-t", PF_TABLE, "-T", "add", ip],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
self.send_response(204)
self.end_headers()
def do_GET(self):
self.send_response(405)
self.end_headers()
class ThreadingUnixHTTPServer(
socketserver.ThreadingMixIn,
socketserver.UnixStreamServer,
):
daemon_threads = True
def main() -> None:
ensure_socket_dir(SOCK_PATH)
# Remove stale socket if present
try:
os.unlink(SOCK_PATH)
except FileNotFoundError:
pass
with ThreadingUnixHTTPServer(SOCK_PATH, BanHandler) as httpd:
# Allow nginx workers (www) to connect
os.chmod(SOCK_PATH, 0o660)
httpd.serve_forever()
if __name__ == "__main__":
main()
|
