Like most software engineers and DIY enthusiasts, I like to automate my frequent daily tasks.
One of the areas that I frequently need to automate is detecting activities in my home network.
For example, when everybody left home, I need some actions to be performed (e.g. ensure security cameras are on, turn off lights if needed, etc).
Another example is finding the IP address of a new joint device (new raspberry pi or ESP32 device or a guest mobile phone). Probably there are more than a dozen ways to find that IP address, but all need some extra checks and commands. I need to automate it, ideally, get a push notification as soon as a new device is discovered in my home network.
I also have a couple of ESP8266/ESP32 sensors collecting data around the house and I need to monitor those as well, to make sure all nodes are up and running.
From a security perspective, I want to be notified if unusual activity is happening in my network (i.e. unusual TCP connection).
But there is a problem here, those tasks could result in a big, complex project.
Let's simplify and break down the goals to keep them simple and manageable.
In essence, I want to:
- Understand what's going on in my network
- Then, check if I can get any insights from network activities
- Can I generate an event from those activities?
- Can I detect a frequent pattern from those activities?
To achieve these, I used a W5100S-EVB-Pico, which is a Raspberry Pi Pico module with a W5100S ethernet chip. The following image shows the overall idea:
- Get packets from different networks
- Process them
- Publish events (e.g. Home Assistant)
I started by connecting W5100S-EVB-Pico to my home router, installed Adafruit Circuit Python on it (as it simplifies the development by auto-reloading the code), and searched for ways of sniffing packets. W5100S datasheets say it supports Promiscuous mode. After some searches, I created the following Sniffer class:
class Sniffer:
def __init__(self, interface, debug=False):
...
def _read_s0cr(self):
while True:
s0_cr = self._the_interface._read_sncr(self._socknum)
if s0_cr[0] == 0x00:
break
time.sleep(0.00025)
def _open_socket(self):
self._the_interface._write_snmr(self._socknum, SNMR_MACRAW)
self._the_interface._write_snir(self._socknum, 0xFF)
self._the_interface._write_sncr(self._socknum, CMD_SOCK_OPEN)
self._read_s0cr()
def _listen(self):
self._the_interface._write_sncr(self._socknum, CMD_SOCK_LISTEN)
self._read_s0cr()
def start(self):
self._open_socket()
self._listen()
def next_packet(self):
return self._the_interface.socket_read(self._socknum, 65565)
The next step was to decode the packet, which can be done using struct.unpack
dst_mac, src_mac, proto = struct.unpack('!6s6sH', data[:14])
After getting packet details I passed the data to a mapper to label them based on pre-configured names for my devices.
While it is not practical for a large network to label all devices like that, for a small home network it is manageable. (Also can use DHCP hostname options, for cases like random mac address).
Those details passed to an inference engine. It is a basic rule engine that counts the packets for monitored devices. My goal is to use this rule engine to collect some data and then use those to bootstrap training data for a more sophisticated machine learning model.
A notifier will send notifications to target channels. MQTT notifier implemented as it can integrated with Home Assistant easily.
Finally, there is a web interface for visual inspections:
The project is in the early stages. Check the source code here:
https://github.com/home-events/home-events
Possible use cases:
- Detect new/unknown devices in the network and send details as push notifications
- Create events for tracked devices
- Collect packets and store them in an external home server (e.g. Raspberry PI)
Comments