Starting testing the tools for detecting noisy domains

dnstwister/tools/noisy_domains.py

@@ -0,0 +1,55 @@
+"""Noisy domain tracking and identification."""
+import datetime
+
+
+# Sliding window size
+WINDOW_SIZE = 30
+
+# Percent threshold of changes per day to mark as "noisy". 50% would mean the
+# domain changed or state once every two days or more.
+THRESHOLD = 50
+
+
+def initialise_record(domain, now=None):
+    """The initial values if there is no noise record for a domain."""
+    if now is None:
+        now = datetime.datetime.now()
+
+    model = {
+        'domain': domain,
+        'last_checked': now,
+        'window_start': now,
+        'deltas': 0,
+    }
+    return model
+
+
+def update(domain_stats, now=None):
+    """Update the domain window and stats."""
+    if now is None:
+        now = datetime.datetime.now()
+
+    # Always update the last checked time
+    domain_stats = dict(domain_stats)
+    domain_stats['last_checked'] = now
+
+    window_start = domain_stats['window_start']
+    window_age = now - window_start
+
+    breakpoint = datetime.timedelta(days=WINDOW_SIZE)
+    move_size = datetime.timedelta(days=WINDOW_SIZE / 2.0)
+
+    # Don't do anything if we're within the window.
+    if window_age <= breakpoint:
+        return domain_stats
+
+    # Shuffle the window half the window size if we're past the end.
+    domain_stats['window_start'] = window_start + move_size
+
+    # Work out what the deltas count would have been for this new timespan.
+    # Take into account how far beyond the window we are.
+    delta_factor = 1 - ((WINDOW_SIZE / 2.0) / window_age.days)
+
+    domain_stats['deltas'] = int(domain_stats['deltas'] * delta_factor)
+
+    return domain_stats

tests/test_tools_email.py

@@ -140,4 +140,3 @@ def test_email_renderer_domain_sorting():
             <a href="https://dnstwister.report/...">Unsubscribe</a>
         </p>
     """).strip()
-

tests/test_tools_noisy_domains.py

@@ -0,0 +1,87 @@
+"""Tests of the tools.noisy_domains module."""
+import datetime
+
+from dnstwister.tools import noisy_domains
+
+
+def test_initial_model():
+    """Test how we set up a new domain."""
+    domain = 'www.example.com'
+    now = datetime.datetime.now()
+    stats = noisy_domains.initialise_record(domain, now)
+
+    assert stats == {
+        'domain': domain,
+        'last_checked': now,
+        'window_start': now,
+        'deltas': 0,
+    }
+
+
+def test_update_when_inside_window():
+    """Test we don't move the window inside the window_size."""
+    domain = 'www.example.com'
+    now = datetime.datetime.now()
+    stats = noisy_domains.initialise_record(domain, now)
+
+    updated_stats = noisy_domains.update(stats, now + datetime.timedelta(days=25))
+
+    assert updated_stats == {
+        'domain': stats['domain'],
+        'last_checked': stats['last_checked'] + datetime.timedelta(days=25),
+        'window_start': stats['window_start'],
+        'deltas': stats['deltas'],
+    }
+
+
+def test_update_when_outside_window():
+    """Test we move the window once we cross the threshold."""
+    domain = 'www.example.com'
+    now = datetime.datetime.now()
+    stats = noisy_domains.initialise_record(domain, now)
+
+    updated_stats = noisy_domains.update(stats, now + datetime.timedelta(days=31))
+
+    assert updated_stats == {
+        'domain': stats['domain'],
+        'last_checked': stats['last_checked'] + datetime.timedelta(days=31),
+        'window_start': stats['window_start'] + datetime.timedelta(days=15),
+        'deltas': stats['deltas'],
+    }
+
+
+def test_update_when_outside_window_updates_deltas():
+    """Test we proportionally update the stats when we move a window."""
+    domain = 'www.example.com'
+    now = datetime.datetime.now()
+    stats = noisy_domains.initialise_record(domain, now)
+
+    stats['deltas'] = 10
+    updated_stats = noisy_domains.update(stats, now + datetime.timedelta(days=31))
+    assert updated_stats == {
+        'domain': stats['domain'],
+        'last_checked': stats['last_checked'] + datetime.timedelta(days=31),
+        'window_start': stats['window_start'] + datetime.timedelta(days=15),
+        'deltas': 5,
+    }
+
+    # The update to deltas is proportional to progress past the end of the
+    # window, to highlight this we're pretending the update has ran more than
+    # just 1 day after the window is crossed.
+    stats['deltas'] = 16
+    updated_stats = noisy_domains.update(stats, now + datetime.timedelta(days=34))
+    assert updated_stats == {
+        'domain': stats['domain'],
+        'last_checked': stats['last_checked'] + datetime.timedelta(days=34),
+        'window_start': stats['window_start'] + datetime.timedelta(days=15),
+        'deltas': 8,
+    }
+
+    stats['deltas'] = 24
+    updated_stats = noisy_domains.update(stats, now + datetime.timedelta(days=45))
+    assert updated_stats == {
+        'domain': stats['domain'],
+        'last_checked': stats['last_checked'] + datetime.timedelta(days=45),
+        'window_start': stats['window_start'] + datetime.timedelta(days=15),
+        'deltas': 16,
+    }