notebooks/MqttTest_example.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "a6540593-4c97-408b-bddb-40ef4bd70b40",
   "metadata": {},
   "outputs": [],
   "source": [
    "import paho.mqtt.client as mqtt\n",
    "import paho.mqtt.publish as publish\n",
    "\n",
    "from datetime import datetime as dt\n",
    "from datetime import timezone as tz\n",
    "\n",
    "from joserfc.jwk import ECKey\n",
    "from joserfc import jws\n",
    "import json\n",
    "import time\n",
    "import random\n",
    "\n",
    "# AggregatedObservations topic in MQTT\n",
    "eventsdn = 'events/down/pop/general' \n",
    "\n",
    "# Broker address\n",
    "broker = 'mqtt.example.com'\n",
    "\n",
    "# generate key pair for signing messages\n",
    "certfile = \".tls/signer.pem\" \n",
    "signkey = \".tls/signer-key.pem\"\n",
    "\n",
    "# Send Observation to Edge Policy Manager (POP)\n",
    "def observation_packet(domain, tag):\n",
    "    # Generate test packet\n",
    "    # Create an mqtt message with a single domain\n",
    "    # with all relevant tag bits set\n",
    "\n",
    "    ts = dt.now(tz=tz.utc).isoformat()\n",
    "\n",
    "    # Domain data\n",
    "    #  Name: the domain name to act on\n",
    "    #  TimeAdded: when the observation was made\n",
    "    #  TTL: how long POP should act on this information\n",
    "    #  TagMask: all attributes related to the domain namn\n",
    "    #\n",
    "    dom = {\n",
    "       'Name' : domain,\n",
    "       'TimeAdded': ts, \n",
    "       'TTL': 1200, \n",
    "       'TagMask': tag\n",
    "      }\n",
    "\n",
    "    # Domain data that is added to POP\n",
    "    added = [ dom ]\n",
    "\n",
    "    # Domain data that is deleted from POP\n",
    "    deleted = []\n",
    "\n",
    "    # MQTT message\n",
    "    msg = {\n",
    "       'SrcName': 'dns-tapir',\n",
    "       'MsgType': 'observation',\n",
    "       'ListType': 'greylist',\n",
    "       'Added': added,\n",
    "       'Deleted' : deleted,\n",
    "       'Msg': '',\n",
    "       'TimeStamp': ts,\n",
    "       'Creator': 'spark.example.com'\n",
    "      }\n",
    "\n",
    "    # Message signature\n",
    "    # Currently symetric, should be signed with Core signature\n",
    "    # POP only trusts data with proper signature\n",
    "    member = {\n",
    "           'protected': {'alg': 'ES256'},\n",
    "         }\n",
    "\n",
    "    with open(signkey) as keyf:\n",
    "        privatekey = ECKey.import_key(keyf.read())\n",
    "\n",
    "    payload = json.dumps(msg)\n",
    "    out = json.dumps(jws.serialize_json(member, payload, privatekey))\n",
    "\n",
    "    # Send a single MQTT event without creating a persistent client\n",
    "    publish.single(eventsdn, out, qos=2, hostname=broker, protocol=mqtt.MQTTv5)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "067dc36e-3935-47db-8f59-31e250f5f18e",
   "metadata": {},
   "source": [
    "## Send a domain with a numerical flag field to TEM\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "74e5ae38-7f82-43f3-b15a-924a4f57bf43",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Meta-tags\n",
    "# -------------------\n",
    "#\n",
    "# NewDomain\n",
    "# Resolving domain that is seen for the first time (in an x day window)\n",
    "#\n",
    "# LowRankDomain\n",
    "# Metatag that indicates that the domain is low ranked according to \n",
    "# Tranco, OpenPageRank, DNS query statistics, etc. \n",
    "#\n",
    "# PossibleMalware\n",
    "# Metatag that is the result of aggregating a number of features, \n",
    "# correlations to other sources or other indicators, and is set if there\n",
    "# is risk that the domain is related to malware. Includes all malware \n",
    "# interactions, such as download sites, call-home or C2, etc\n",
    "# \n",
    "# GlobalRampup \n",
    "# General rapid growth of queries\n",
    "#\n",
    "# LocalRampUp\n",
    "# Significant local growth of queries\n",
    "#\n",
    "# SustainedHighTraffic\n",
    "# The volume of queries is remarkably higher than baseline\n",
    "# \n",
    "\n",
    "NewDomain = 2\n",
    "LowRankDomain = 4\n",
    "PossibleMalware = 8\n",
    "GlobalRampUp = 16\n",
    "LocalRampUp = 32\n",
    "SustainedHighTraffic = 64\n",
    "\n",
    "# The domain the meesage relates to\n",
    "domain = \"example.com.\"\n",
    "\n",
    "# Set specific tag value\n",
    "# value = ( GlobalRampUp | SustainedHighTraffic | LowRankDomain )\n",
    "# observation_packet(domain, value)\n",
    "\n",
    "# Send a continuous stream of observations\n",
    "while True:\n",
    "    # Generate random tags\n",
    "    value = random.randrange(65537)\n",
    "    # Send packet\n",
    "    observation_packet(domain, value)\n",
    "    print(f\"{dt.now()} sent {value} for {domain}\")\n",
    "    # Wait a while\n",
    "    time.sleep(45)\n",
    "    \n",
    "    # Reset all tags\n",
    "    value = 0\n",
    "    # send packet\n",
    "    observation_packet(domain, value)\n",
    "    print(f\"{dt.now()} sent {value} for {domain}\")\n",
    "    # wait a while\n",
    "    time.sleep(15)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6020a657-2558-4203-95c4-43363edcf856",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.18"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}