69 lines
1.5 KiB
Python
69 lines
1.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Minimal NFQUEUE example script.
|
|
|
|
Expected argv:
|
|
argv[1] = queue number
|
|
Any extra args are optional.
|
|
"""
|
|
|
|
import logging
|
|
import signal
|
|
import sys
|
|
from typing import Optional
|
|
|
|
try:
|
|
from netfilterqueue import NetfilterQueue
|
|
except Exception as exc: # pragma: no cover
|
|
print(f"Failed to import netfilterqueue: {exc}", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
|
logger = logging.getLogger("hello-nfqueue")
|
|
|
|
_running = True
|
|
|
|
|
|
def _stop(_sig: int, _frame: Optional[object]) -> None:
|
|
global _running
|
|
_running = False
|
|
|
|
|
|
def _handle_packet(packet) -> None:
|
|
# This demo accepts all packets and logs basic metadata.
|
|
logger.info("packet id=%s len=%s", packet.get_id(), len(packet.get_payload()))
|
|
packet.accept()
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) < 2:
|
|
print("Usage: hello_nfqueue.py <qnum> [extra args...]", file=sys.stderr)
|
|
return 1
|
|
|
|
try:
|
|
qnum = int(sys.argv[1])
|
|
except ValueError:
|
|
print("qnum must be an integer", file=sys.stderr)
|
|
return 1
|
|
|
|
signal.signal(signal.SIGINT, _stop)
|
|
signal.signal(signal.SIGTERM, _stop)
|
|
|
|
nfq = NetfilterQueue()
|
|
logger.info("Binding to NFQUEUE %d", qnum)
|
|
nfq.bind(qnum, _handle_packet)
|
|
|
|
try:
|
|
while _running:
|
|
nfq.run(block=True)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
logger.info("Unbinding NFQUEUE %d", qnum)
|
|
nfq.unbind()
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|