Scapy - who ping?

whoping.py

#! /usr/bin/python3

"""
whoping?

Shows a desktop notification using 'notify-send' (or executes 
a custom command) when ICMP packets are received on the specified 
interface. Also, saves the received packets to a pcap file if desired.

Usage:
------
sudo ./whoping.py -h
sudo ./whoping.py -i eth0 -o ./whoping.pcap -c ./custom.sh -t 10
"""

import time
from subprocess import run
from scapy.all import IP, get_if_addr, sniff
from scapy.utils import PcapWriter
from argparse import ArgumentParser

default_iface = 'eth0'
default_outpath = 'whoping.pcap'
default_timeout = 10 # seconds
default_command = 'notify-send'

def debounce_duplicate(wait, key='id'):
    def decorator(fn):
        calls = {}
        def debounced(*args, **kwargs):
            def call_fn(call_id):
                fn(*args, **kwargs)
                calls[call_id] = time.time()
            id = kwargs.get(key)
            if not calls.get(id) or (time.time() - calls.get(id) > wait):
                call_fn(id)
        return debounced
    return decorator

def setup_command(command, timeout):
    @debounce_duplicate(timeout)
    def run_command(args=[], id=''):
        run([command] + args)
    return run_command

def icmp_callback(pkt, pcap_writer=None, command_runner=None, timeout=default_timeout):
    if pcap_writer:
        pcap_writer.write(pkt)
    if command_runner:
        src = pkt[IP].src if IP in pkt else '?'
        command_runner([f'ping from {src}'], id=src)

def start_sniffer(iface=default_iface, outpath=default_outpath, save=True, command=default_command, timeout=default_timeout):
    pcap_writer = PcapWriter(outpath, append=True) if save else None
    command_runner = setup_command(command, timeout) if command else None
    packet_handler = lambda pkt: icmp_callback(pkt, pcap_writer=pcap_writer, command_runner=command_runner)
    is_incoming = lambda pkt: IP in pkt and (pkt[IP].src != get_if_addr(iface))
    # sniff incoming ICMP packets only
    sniff(prn=packet_handler, iface=iface, filter='icmp', lfilter=is_incoming)

if __name__ == '__main__':
    parser = ArgumentParser()
    parser.add_argument('-i', '--iface',
        default=default_iface,
        help='interface to listen on')
    parser.add_argument('-o', '--outpath',
        default=default_outpath,
        help='output path for pcap file')
    parser.add_argument('-n', '--nosave',
        action='store_true',
        help='do not save packets to pcap')
    parser.add_argument('-c', '--command',
        default=default_command,
        help='command to execute when someone pings')
    parser.add_argument('-t', '--timeout',
        default=default_timeout,
        help='timeout between notifications (in seconds)')
    args = parser.parse_args()

    start_sniffer(
        iface=args.iface,
        outpath=args.outpath,
        save=not args.nosave,
        command=args.command,
        timeout=args.timeout)