Integrate locations

justfile

@@ -1,2 +1,5 @@
 parse:
     uv run main.py
+
+stats:
+    uv run stats.py

locate.py

@@ -1,40 +0,0 @@
-#!/usr/bin/env python3
-import sqlite3, requests, time
-import matplotlib.pyplot as plt
-
-DB = "sshd_failures.db"
-
-# get top IPs (limit to avoid API spam)
-ips = []
-db = sqlite3.connect(DB)
-for ip, count, _ in db.execute("""
-    SELECT ip, COUNT(*), MAX(ts)
-    FROM failures
-    GROUP BY ip
-    ORDER BY COUNT(*) DESC
-    LIMIT 30
-"""):
-    ips.append((ip, count))
-db.close()
-
-lats, lons, sizes = [], [], []
-
-for ip, count in ips:
-    try:
-        r = requests.get(f"http://ip-api.com/json/{ip}", timeout=5).json()
-        if r["status"] == "success":
-            lats.append(r["lat"])
-            lons.append(r["lon"])
-            sizes.append(count * 5)
-            print(f"{ip:15} -> {r['country']}")
-        time.sleep(0.5)  # be nice to API
-    except:
-        pass
-
-# plot
-plt.figure()
-plt.scatter(lons, lats, s=sizes)
-plt.title("SSHD Attack Origins")
-plt.xlabel("Longitude")
-plt.ylabel("Latitude")
-plt.show()

main.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-import sqlite3, subprocess, re, os
+import sqlite3, subprocess, re, os, requests, time
 
 DB = "sshd_failures.db"
 STATE = "cursor.txt"
@@ -12,9 +12,25 @@ def get_cursor():
 def save_cursor(c):
     open(STATE, "w").write(c)
 
+def geo(ip):
+    r = requests.get(f"http://ip-api.com/json/{ip}", timeout=5).json()
+    if r["status"] == "success":
+        return r["country"], r["lat"], r["lon"]
+    return None, None, None
+
 def main():
     db = sqlite3.connect(DB)
-    db.execute("CREATE TABLE IF NOT EXISTS failures(ts INTEGER, ip TEXT)")
+
+    db.execute("CREATE TABLE IF NOT EXISTS failures (ts INTEGER, ip TEXT)")
+
+    db.execute("""
+    CREATE TABLE IF NOT EXISTS ip_geo (
+        ip TEXT PRIMARY KEY,
+        country TEXT,
+        lat REAL,
+        lon REAL
+    )
+    """)
 
     cmd = ["journalctl", "-u", "sshd", "-o", "short-unix", "--no-pager"]
     cur = get_cursor()
@@ -23,18 +39,10 @@ def main():
 
     out = subprocess.check_output(cmd, text=True)
 
-    total = 0
-    parsed = 0
-    inserted = 0
-
     for line in out.splitlines():
-        total += 1
-
         if not line or not line[0].isdigit():
             continue
 
-        parsed += 1
-
         ts_str, msg = line.split(" ", 1)
 
         try:
@@ -43,15 +51,31 @@ def main():
             continue
 
         m = PATTERN.search(msg)
-        if m:
-            db.execute("INSERT INTO failures VALUES (?, ?)", (ts, m.group(1)))
-            inserted += 1
+        if not m:
+            continue
+
+        ip = m.group(1)
+
+        # store raw event
+        db.execute("INSERT INTO failures VALUES (?, ?)", (ts, ip))
+
+        # check cache
+        cached = db.execute(
+            "SELECT 1 FROM ip_geo WHERE ip = ?",
+            (ip,)
+        ).fetchone()
+
+        if not cached:
+            country, lat, lon = geo(ip)
+            db.execute(
+                "INSERT OR REPLACE INTO ip_geo VALUES (?, ?, ?, ?)",
+                (ip, country, lat, lon)
+            )
+            time.sleep(0.2)
 
     db.commit()
     db.close()
 
-    print(f"lines={total} parsed={parsed} inserted={inserted}")
-
     # save cursor
     out = subprocess.check_output(
         ["journalctl", "-u", "sshd", "-n", "1", "-o", "export"],

stats.py

@@ -1,17 +1,33 @@
 #!/usr/bin/env python3
-import sqlite3, datetime
+import sqlite3
+from datetime import datetime
 
-DB = "sshd_failures.db"
+db = sqlite3.connect("sshd_failures.db")
 
-db = sqlite3.connect(DB)
-
-for ip, count, last_ts in db.execute("""
-    SELECT ip, COUNT(*), MAX(ts)
-    FROM failures
-    GROUP BY ip
-    ORDER BY COUNT(*) DESC
+for row in db.execute("""
+SELECT
+    g.country,
+    COUNT(*) AS total,
+    MAX(f.ts) AS last_seen,
+    (
+        SELECT f2.ip
+        FROM failures f2
+        JOIN ip_geo g2 ON f2.ip = g2.ip
+        WHERE g2.country = g.country
+        GROUP BY f2.ip
+        ORDER BY COUNT(*) DESC
+        LIMIT 1
+    ) AS top_ip
+FROM failures f
+JOIN ip_geo g ON f.ip = g.ip
+GROUP BY g.country
+ORDER BY total DESC
 """):
-    last = datetime.datetime.fromtimestamp(last_ts)
-    print(f"{ip:40} {count:5}  last={last}")
+    country, total, last_seen, top_ip = row
+
+    if last_seen:
+        last_seen = datetime.fromtimestamp(last_seen)
+
+    print(f"{country:25} {total:6}  top_ip={top_ip:18} last={last_seen}")
 
 db.close()