215 lines
8.6 KiB
Python
Executable File
215 lines
8.6 KiB
Python
Executable File
#!/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()
|