This blog post is about creating an automatic IP blocklist in OPNsense based on MISP data, including data retention.
Modern firewalling consists of threat hunting and IP blocklist often bought from Big-Tech companies. A note at the IKT security conference from CERTAINITY inspired me to use open data from the cyber security community and to get things going. I wanted to have something similar like these expensive blocklists but with open data. Here is what I did.
A MISP plattform is used to share threat intelligence in a local and federated level. They are often operated by a local CERT/CSIRT. You may ask nicely to get access to. I got access to a MISP containing data from CERTs all around the world sharing all kinds of threat intelligence including IP addresses (IPv4 and IPv6), ip address and port combinations, domains, hashes, URLs, etc.
Once you have access to such data sets, you may generate lists of relevant data. I decided, that ip addresses are sufficient for myself atm. Also, as there are around 600.000 datasets, containing multiple attributes, some data cleaning is necessary to process all the data. I decided to query the last 100 events every ten minutes, to exclude some tags (e.g. events containing Tor exit nodes), and only extract IP addresses. Next I write theses addresses into a file, if not existing in it already. The script to do this is completely AI generated:
#!/usr/bin/env python3
import argparse
import ipaddress
import sys
import logging
import logging.handlers
from pathlib import Path
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
IP_TYPES = {
"ip-src",
"ip-dst",
"ip-src|port",
"ip-dst|port",
}
def configure_logger():
"""Configure logging to the local operating system log."""
logger = logging.getLogger("misp_ip_extractor")
logger.setLevel(logging.INFO)
logger.propagate = False
if logger.handlers:
return logger
# Linux and other Unix-like systems commonly expose the local syslog
# socket at /dev/log.
if sys.platform.startswith(("linux", "darwin", "freebsd")):
syslog_address = "/var/run/syslog" if sys.platform == "darwin" else "/dev/log"
try:
handler = logging.handlers.SysLogHandler(
address=syslog_address,
facility=logging.handlers.SysLogHandler.LOG_USER,
)
except OSError:
handler = logging.handlers.SysLogHandler(
address=("localhost", 514),
facility=logging.handlers.SysLogHandler.LOG_USER,
)
else:
# Windows: send messages to the Windows Event Log.
handler = logging.handlers.NTEventLogHandler(
"MISP IP Extractor",
logtype="Application",
)
handler.setFormatter(
logging.Formatter("%(name)s[%(process)d]: %(levelname)s %(message)s")
)
logger.addHandler(handler)
return logger
logger = configure_logger()
def extract_ip(value, attribute_type):
"""Extract and validate an IPv4 or IPv6 address."""
if attribute_type in {"ip-src|port", "ip-dst|port"}:
value = value.split("|", 1)[0]
try:
return str(ipaddress.ip_address(value))
except ValueError:
return None
def event_has_excluded_tag(event, excluded_tags):
"""Return True if the event contains one of the excluded tags."""
event_tags = {
tag.get("name")
for tag in event.get("Tag", [])
if tag.get("name")
}
return bool(event_tags.intersection(excluded_tags))
def get_latest_events(
misp_url,
api_key,
limit=2,
verify_tls=True,
proxy=None,
):
url = f"{misp_url.rstrip('/')}/events/restSearch"
headers = {
"Authorization": api_key,
"Accept": "application/json",
"Content-Type": "application/json",
}
query = {
"returnFormat": "json",
"limit": limit,
"page": 1,
"order": "Event.timestamp desc",
"includeAttribute": True,
}
# Apply the same proxy to HTTP and HTTPS requests.
proxies = None
if proxy:
proxies = {
"http": proxy,
"https": proxy,
}
response = requests.post(
url,
headers=headers,
json=query,
verify=verify_tls,
timeout=30,
proxies=proxies,
)
response.raise_for_status()
result = response.json()
events = result.get("response", [])
if not events and isinstance(result, list):
events = result
return events
def read_existing_ips(output_file):
"""Read existing IP addresses from the output file."""
if not output_file.exists():
return set()
with output_file.open("r", encoding="utf-8") as file:
return {
line.strip()
for line in file
if line.strip()
}
def main():
parser = argparse.ArgumentParser(
description="Extract IP addresses from the latest MISP events."
)
parser.add_argument(
"--url",
required=True,
help="MISP URL, for example https://misp.example.com",
)
parser.add_argument(
"--api-key",
required=True,
help="MISP API key",
)
parser.add_argument(
"--output",
required=True,
help="File to which IP addresses will be appended",
)
parser.add_argument(
"--limit",
type=int,
default=2,
help="Number of latest events to retrieve. Default: 2",
)
parser.add_argument(
"--no-verify-tls",
action="store_true",
help="Disable TLS certificate verification",
)
parser.add_argument(
"--proxy",
help=(
"Proxy URL, for example "
"http://proxy.example.com:8080 or "
"http://user:password@proxy.example.com:8080"
),
)
parser.add_argument(
"--exclude-tags",
help=(
"Comma-separated list of event tags to exclude, "
"for example: tlp:red,malware"
),
)
args = parser.parse_args()
if args.limit < 1:
parser.error("--limit must be at least 1")
if args.no_verify_tls:
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
output_file = Path(args.output)
excluded_tags = set()
if args.exclude_tags:
excluded_tags = {
tag.strip()
for tag in args.exclude_tags.split(",")
if tag.strip()
}
try:
events = get_latest_events(
misp_url=args.url,
api_key=args.api_key,
limit=args.limit,
verify_tls=not args.no_verify_tls,
proxy=args.proxy,
)
existing_ips = read_existing_ips(output_file)
new_ips = set()
for item in events:
event = item.get("Event", item)
if event_has_excluded_tag(event, excluded_tags):
logger.debug(
"Skipping event %s because it has an excluded tag.",
event.get("id", "unknown"),
)
continue
for attribute in event.get("Attribute", []):
attribute_type = attribute.get("type")
value = attribute.get("value")
if attribute_type not in IP_TYPES or not value:
continue
ip_address = extract_ip(value, attribute_type)
if ip_address:
new_ips.add(ip_address)
ips_to_write = sorted(new_ips - existing_ips)
if ips_to_write:
with output_file.open("a", encoding="utf-8") as file:
for ip_address in ips_to_write:
file.write(f"{ip_address}\n")
logger.info("Found %s IP address(es).", len(new_ips))
logger.info("Added %s new IP address(es).", len(ips_to_write))
logger.info("Output file: %s", output_file)
except requests.RequestException as exc:
logger.error("MISP request failed: %s", exc)
return 1
except OSError as exc:
logger.error("File operation failed: %s", exc)
return 1
except ValueError:
logger.error("MISP returned invalid JSON.")
return 1
if __name__ == "__main__":
main()
The data is saved to a file which gets rotated monthly. This ensures that I keep data only 12 months in total.
# delete previous misp ips from january in the first minute 0 0 1 1 * root rm /PATH/ip-addresses-january.txt # get latest misp ips in january */10 * * 1 * root python3 /PATH/misp-ip-exporter.py --url URL --api-key API_KEY --limit 100 --output "/PATH/ip-addresses-january.txt" --proxy PROXY --exclude-tags EXCLUDE_TAGS # delete previous misp ips from february in the first minute 0 0 1 2 * root rm /PATH/ip-addresses-february.txt # get latest misp ips in february */10 * * 2 * root python3 /PATH/misp-ip-exporter.py --url URL --api-key API_KEY --limit 100 --output "/PATH/ip-addresses-february.txt" --proxy PROXY --exclude-tags EXCLUDE_TAGS <etc.>
The lists look the following, IPv4 and IPv6 are in a mixed list:
2.56.10.36 2.56.57.21 20.220.105.236 200.122.181.2 200.79.151.188 200.79.152.164 2001:1620:51a1::101 2001:19f0:6000:3570:5400:6ff:fec5:a756 2001:19f0:6001:11ad:5400:6ff:fec4:5dae 2001:19f0:6400:2b4e:5400:6ff:fec4:7b8 2001:19f0:6400:2b9b:5400:6ff:fec5:8398
Next, it is necessary that you deliver the list via http(s). If you want to restrict the access, you should at least permit the firewall.
OPNsense allows alias of the type URL (IPs). Create an alias per created ip list file and enter the URL of the file as content. Next, create an alias of the type hosts and insert all previous created aliases. You can use this summary alias as source and destination in two block rules.
Last update: 2026-10-02