SwissPass phishing-kit sabotage tool

sabotage-swisskit-phishing.py

@@ -0,0 +1,214 @@
+#!/usr/bin/env python3
+"""Poison a SwissPass phishing kit with realistic fake submissions.
+
+Generates bogus identities and Luhn-valid card numbers and posts them to the
+kit's login and billing endpoints so the data it steals can't be trusted.
+
+The endpoints and field names in the KIT CONTRACT block mirror the kit's wire
+format -- change them there if the kit changes, not in the logic.
+"""
+import argparse
+import csv
+import logging
+import os
+import random
+import string
+import time
+from datetime import date, timedelta
+
+import requests
+from fake_useragent import UserAgent
+from faker import Faker
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+# --- kit contract -----------------------------------------------------------
+LOGIN_PATH = "/login/"
+BILLING_PATH = "/billing/"
+# The kit posts these exact names; only edit if the kit's form changes.
+FIELD_REF = "rv_nr"
+# --- end kit contract -------------------------------------------------------
+
+CSV_PATH = "sabotage_data.csv"
+LOG_PATH = "sabotage.log"
+
+ACCEPT_LANGS = (
+    "de-CH,de;q=0.9,en;q=0.8",
+    "de-CH,de;q=0.9",
+    "de;q=1.0,en;q=0.7",
+    "de-CH,en-US;q=0.9,en;q=0.8",
+)
+
+BANKS = (
+    "UBS", "Raiffeisen", "Zürcher Kantonalbank", "PostFinance", "Credit Suisse",
+    "Migros Bank", "Bank Cler", "Luzerner Kantonalbank", "Berner Kantonalbank",
+    "Basler Kantonalbank", "Banque Cantonale Vaudoise", "Swissquote", "unbekannt",
+)
+
+fake = Faker("de_CH")
+ua = UserAgent()
+
+
+class SwissPassPoisoner:
+    def __init__(self, base_url):
+        self.base_url = base_url.rstrip("/")
+        self.session = requests.Session()
+        self.session.headers.update({
+            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+            "Cache-Control": "max-age=0",
+            "Upgrade-Insecure-Requests": "1",
+        })
+        # Retry transient 5xx: duplicate submissions are fine here, we want volume.
+        retry = Retry(total=3, backoff_factor=0.5,
+                      status_forcelist=(500, 502, 503, 504),
+                      allowed_methods=frozenset(["POST"]),
+                      raise_on_status=False)
+        adapter = HTTPAdapter(max_retries=retry)
+        self.session.mount("https://", adapter)
+        self.session.mount("http://", adapter)
+
+    def _jitter(self, base):
+        return max(0.1, base + random.uniform(-base / 2, base / 2))
+
+    @staticmethod
+    def _luhn(prefix, length):
+        num = prefix + "".join(random.choices(string.digits, k=length - 1 - len(prefix)))
+        total = sum((d * 2 - 9 if d * 2 > 9 else d * 2) if i % 2 == 0 else d
+                    for i, d in enumerate(map(int, reversed(num))))
+        return num + str((10 - total % 10) % 10)
+
+    @staticmethod
+    def _ahv():
+        """Swiss AHV/AVS number: 756 + 9 digits + EAN-13 check digit."""
+        digits = "756" + "".join(random.choices(string.digits, k=9))
+        total = sum(int(d) * (1 if i % 2 == 0 else 3) for i, d in enumerate(digits))
+        digits += str((10 - total % 10) % 10)
+        return f"{digits[:3]}.{digits[3:7]}.{digits[7:11]}.{digits[11:]}"
+
+    def generate_card(self):
+        brand = random.choice(("visa", "mastercard")) if random.random() < 0.85 else "amex"
+        if brand == "amex":
+            raw = self._luhn("37", 15)
+            cardnum = f"{raw[:4]} {raw[4:10]} {raw[10:]}"
+            cvv = str(random.randint(1000, 9999))
+        else:
+            raw = self._luhn("4" if brand == "visa" else "5", 16)
+            cardnum = " ".join(raw[i:i + 4] for i in range(0, 16, 4))
+            cvv = str(random.randint(100, 999))
+        return {
+            "cardnum": cardnum,
+            "expiry": f"{random.randint(1, 12):02d} / {random.randint(2027, 2030)}",
+            "cvv": cvv,
+            "brand": brand,
+            "bank": random.choice(BANKS),
+        }
+
+    def generate_person(self):
+        age = random.randint(60, 85) if random.random() < 0.7 else random.randint(40, 59)
+        birth = date.today() - timedelta(days=age * 365 + random.randint(0, 364))
+        return {
+            "vorname": fake.first_name(),
+            "nachname": fake.last_name(),
+            "gebdatum": birth.strftime("%Y-%m-%d"),
+            "ref": self._ahv(),
+        }
+
+    def _post(self, path, data):
+        self.session.headers["User-Agent"] = ua.random
+        self.session.headers["Accept-Language"] = random.choice(ACCEPT_LANGS)
+        try:
+            resp = self.session.post(f"{self.base_url}{path}", data=data,
+                                     allow_redirects=False, timeout=15)
+        except requests.RequestException as exc:
+            logging.warning("POST %s failed: %s", path, exc)
+            return False
+        # ponytail: trusts status alone, not the body; tighten if the kit returns 200 on rejects.
+        return resp.status_code in (200, 302)
+
+    def run(self, count, delay):
+        logging.info("start: %d entries -> %s", count, self.base_url)
+        success = 0
+        with open(CSV_PATH, "w", newline="", encoding="utf-8") as fh:
+            writer = csv.writer(fh)
+            writer.writerow(["iteration", "ahv_nr", "vorname", "nachname", "gebdatum",
+                             "cardnum", "expiry", "cvv", "brand", "bank", "login", "billing"])
+            for i in range(1, count + 1):
+                time.sleep(self._jitter(delay))
+                person = self.generate_person()
+                card = self.generate_card()
+
+                login_ok = self._post(LOGIN_PATH, {FIELD_REF: person["ref"]})
+                billing_ok = False
+                if login_ok:
+                    time.sleep(self._jitter(delay / 2))
+                    billing_ok = self._post(BILLING_PATH, {
+                        "card_action": "1", "vorname": person["vorname"],
+                        "nachname": person["nachname"], "gebdatum": person["gebdatum"],
+                        FIELD_REF: person["ref"], **card,
+                    })
+                success += billing_ok
+
+                login = "OK" if login_ok else "FAIL"
+                billing = "OK" if billing_ok else ("FAIL" if login_ok else "SKIPPED")
+                writer.writerow([i, person["ref"], person["vorname"], person["nachname"],
+                                 person["gebdatum"], card["cardnum"], card["expiry"],
+                                 card["cvv"], card["brand"], card["bank"], login, billing])
+                fh.flush()
+                logging.info("[%d/%d] %s %s | %s | login:%s billing:%s", i, count,
+                             person["vorname"], person["nachname"], card["cardnum"],
+                             login, billing)
+
+        logging.info("done: %d ok, %d failed. data: %s", success, count - success, CSV_PATH)
+
+
+def _selftest():
+    def valid_luhn(n):
+        total = 0
+        for i, d in enumerate(map(int, reversed(n))):
+            if i % 2:
+                d = d * 2 - 9 if d * 2 > 9 else d * 2
+            total += d
+        return total % 10 == 0
+
+    for prefix, length in (("4", 16), ("5", 16), ("37", 15)):
+        for _ in range(500):
+            n = SwissPassPoisoner._luhn(prefix, length)
+            assert len(n) == length and valid_luhn(n), n
+    for _ in range(500):
+        a = SwissPassPoisoner._ahv().replace(".", "")
+        assert len(a) == 13 and a.startswith("756"), a
+        total = sum(int(x) * (1 if i % 2 == 0 else 3) for i, x in enumerate(a[:12]))
+        assert int(a[12]) == (10 - total % 10) % 10, a
+    print("selftest ok")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Poison a SwissPass phishing kit with fake data.")
+    parser.add_argument("--target", default=os.environ.get("SABOTAGE_TARGET"),
+                        help="kit base URL (or set SABOTAGE_TARGET)")
+    parser.add_argument("--count", type=int, default=10, help="entries to submit (default 10)")
+    parser.add_argument("--delay", type=float, default=2.0, help="baseline seconds between steps")
+    parser.add_argument("--selftest", action="store_true", help="check the data generators and exit")
+    args = parser.parse_args()
+    if args.selftest:
+        _selftest()
+        return
+    if not args.target:
+        parser.error("--target URL or SABOTAGE_TARGET is required")
+
+    root = logging.getLogger()
+    root.setLevel(logging.INFO)
+    file_handler = logging.FileHandler(LOG_PATH)
+    file_handler.setFormatter(logging.Formatter("%(asctime)s - %(message)s"))
+    console = logging.StreamHandler()
+    console.setFormatter(logging.Formatter("%(message)s"))
+    root.handlers[:] = [file_handler, console]
+
+    try:
+        SwissPassPoisoner(args.target).run(args.count, args.delay)
+    except KeyboardInterrupt:
+        logging.warning("interrupted - partial data flushed to %s", CSV_PATH)
+
+
+if __name__ == "__main__":
+    main()