thomasciesla/geoip_shop_manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

geoip_shop_manager.py aktualisiert

Browse Source
This commit is contained in:
thomasciesla
2025-12-08 18:06:29 +01:00
parent d70317a311
commit cf0c3151b6
1 changed files with 301 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

375
geoip_shop_manager.py
View File

@@ -128,6 +128,104 @@ if (!$is_dach) {{
}}
'''
# PHP GeoIP blocking script - PHP ONLY (no CrowdSec queue)
GEOIP_SCRIPT_PHP_ONLY = '''<?php
/**
* GeoIP Blocking Script - Blocks all non-DACH IPs (PHP-only mode)
* DACH = Germany (DE), Austria (AT), Switzerland (CH)
* Logs blocked IPs (no CrowdSec synchronisation)
* Valid until: {expiry_date}
*/
// Auto-disable after 72 hours
$expiry_date = strtotime('{expiry_timestamp}');
if (time() > $expiry_date) {{
return; // Script expired, allow all traffic
}}
// Get visitor IP
$visitor_ip = $_SERVER['REMOTE_ADDR'] ?? '';
if (empty($visitor_ip)) {{
return;
}}
// Skip private IPs
if (filter_var($visitor_ip, FILTER_VALIDATE_IP, FILTER_FLAG_NO_PRIV_RANGE | FILTER_FLAG_NO_RES_RANGE) === false) {{
return;
}}
// Files
$cache_file = __DIR__ . '/{cache_file}';
$cache_duration = 86400; // 24 hours
$log_file = __DIR__ . '/{log_file}';
// Function to download DACH IP ranges (Germany, Austria, Switzerland)
function download_dach_ranges() {{
$ranges = [];
$countries = ['de', 'at', 'ch']; // Germany, Austria, Switzerland
foreach ($countries as $country) {{
$url = "https://www.ipdeny.com/ipblocks/data/aggregated/$country-aggregated.zone";
$content = @file_get_contents($url);
if ($content !== false) {{
$lines = explode("\\n", trim($content));
foreach ($lines as $line) {{
$line = trim($line);
if (!empty($line) && strpos($line, '/') !== false) {{
$ranges[] = $line;
}}
}}
}}
}}
return $ranges;
}}
// Function to check if IP is in CIDR range
function ip_in_range($ip, $cidr) {{
list($subnet, $mask) = explode('/', $cidr);
$ip_long = ip2long($ip);
$subnet_long = ip2long($subnet);
$mask_long = -1 << (32 - (int)$mask);
return ($ip_long & $mask_long) == ($subnet_long & $mask_long);
}}
// Load or download IP ranges
$dach_ranges = [];
if (file_exists($cache_file) && (time() - filemtime($cache_file)) < $cache_duration) {{
$dach_ranges = unserialize(file_get_contents($cache_file));
}} else {{
$dach_ranges = download_dach_ranges();
if (!empty($dach_ranges)) {{
@file_put_contents($cache_file, serialize($dach_ranges));
}}
}}
// Check if visitor IP is from DACH region
$is_dach = false;
foreach ($dach_ranges as $range) {{
if (ip_in_range($visitor_ip, $range)) {{
$is_dach = true;
break;
}}
}}
// Block non-DACH IPs
if (!$is_dach) {{
$timestamp = date('Y-m-d H:i:s');
$user_agent = $_SERVER['HTTP_USER_AGENT'] ?? 'Unknown';
$request_uri = $_SERVER['REQUEST_URI'] ?? '/';
// Log for humans
$log_entry = "[$timestamp] IP: $visitor_ip | UA: $user_agent | URI: $request_uri\\n";
@file_put_contents($log_file, $log_entry, FILE_APPEND | LOCK_EX);
header('HTTP/1.1 403 Forbidden');
exit;
}}
'''
# Python watcher script (runs as systemd service)
WATCHER_SCRIPT_CONTENT = '''#!/usr/bin/env python3
"""
@@ -358,8 +456,8 @@ def uninstall_watcher_service():
print(" ✅ Service deinstalliert")
def add_shop_to_active(shop):
"""Add shop to active shops tracking"""
def add_shop_to_active(shop, mode="php+crowdsec"):
"""Add shop to active shops tracking with mode"""
os.makedirs(os.path.dirname(ACTIVE_SHOPS_FILE), exist_ok=True)
shops = {}
@@ -369,13 +467,27 @@ def add_shop_to_active(shop):
shops[shop] = {
"activated": datetime.now().isoformat(),
"expiry": (datetime.now() + timedelta(hours=72)).isoformat()
"expiry": (datetime.now() + timedelta(hours=72)).isoformat(),
"mode": mode # "php+crowdsec" or "php-only"
}
with open(ACTIVE_SHOPS_FILE, 'w') as f:
json.dump(shops, f, indent=2)
def get_shop_mode(shop):
"""Get the blocking mode for a shop"""
if not os.path.isfile(ACTIVE_SHOPS_FILE):
return "php+crowdsec"
try:
with open(ACTIVE_SHOPS_FILE, 'r') as f:
shops = json.load(f)
return shops.get(shop, {}).get("mode", "php+crowdsec")
except:
return "php+crowdsec"
def remove_shop_from_active(shop):
"""Remove shop from active shops tracking"""
if not os.path.isfile(ACTIVE_SHOPS_FILE):
@@ -398,20 +510,53 @@ def cleanup_crowdsec_decisions(shop):
print(f" 🔍 Entferne CrowdSec-Decisions für {shop}...")
# Use the same reliable method as manual cleanup
# This uses xargs to properly handle all IPs
cleanup_cmd = f"cscli decisions list -o raw | grep '{shop}' | cut -d',' -f3 | cut -d':' -f2 | xargs -I {{}} cscli decisions delete --ip {{}}"
total_removed = 0
max_iterations = 50 # Safety limit
iteration = 0
code, stdout, stderr = run_command(cleanup_cmd)
while iteration < max_iterations:
iteration += 1
# Count how many were removed by checking the output
if stdout:
# Count "deleted" messages
removed = stdout.count("decision(s) deleted")
if removed > 0:
print(f" ✅ {removed} Decisions entfernt")
else:
print(" ℹ️ Keine Decisions gefunden")
# Get all decisions with --limit 0 (no pagination)
list_cmd = f"cscli decisions list -o raw --limit 0 | grep '{shop}'"
code, stdout, stderr = run_command(list_cmd)
if code != 0 or not stdout.strip():
break # No more decisions found
# Extract IPs and delete them (process in batches of 100)
lines = stdout.strip().split('\n')
batch_count = 0
for line in lines[:100]: # Process max 100 per iteration
try:
parts = line.split(',')
if len(parts) >= 3:
ip_field = parts[2].strip()
if ':' in ip_field:
ip = ip_field.split(':', 1)[1]
else:
ip = ip_field
if ip:
del_cmd = f"cscli decisions delete --ip {ip}"
del_code, _, _ = run_command(del_cmd)
if del_code == 0:
batch_count += 1
except:
continue
total_removed += batch_count
if batch_count == 0:
break # Nothing deleted in this iteration
# Small progress indicator for large cleanups
if iteration > 1:
print(f" ... {total_removed} Decisions entfernt (Durchlauf {iteration})")
if total_removed > 0:
print(f" ✅ {total_removed} Decisions entfernt")
else:
print(" ℹ️ Keine Decisions gefunden")
@@ -450,7 +595,7 @@ def get_active_shops():
return active
def activate_blocking(shop, silent=False):
def activate_blocking(shop, silent=False, mode="php+crowdsec"):
"""Activate GeoIP blocking for a single shop"""
httpdocs = os.path.join(VHOSTS_DIR, shop, 'httpdocs')
index_php = os.path.join(httpdocs, 'index.php')
@@ -470,21 +615,28 @@ def activate_blocking(shop, silent=False):
if not silent:
print(f"\n🔧 Aktiviere DACH GeoIP-Blocking für: {shop}")
print(" (Erlaubt: Deutschland, Österreich, Schweiz)")
print(f" Modus: {'PHP + CrowdSec' if mode == 'php+crowdsec' else 'Nur PHP'}")
print("=" * 60)
# Step 1: Install watcher service if not exists
active_shops = get_active_shops()
if not active_shops: # First shop
if not silent:
print("\n[1/3] Installiere CrowdSec-Watcher-Service...")
if check_crowdsec():
install_watcher_service()
# Step 1: Install watcher service if not exists (only for php+crowdsec mode)
if mode == "php+crowdsec":
active_shops = get_active_shops()
# Check if any shop uses crowdsec mode
crowdsec_shops = [s for s in active_shops if get_shop_mode(s) == "php+crowdsec"]
if not crowdsec_shops: # First shop with crowdsec
if not silent:
print("\n[1/3] Installiere CrowdSec-Watcher-Service...")
if check_crowdsec():
install_watcher_service()
else:
if not silent:
print(" ⚠️ CrowdSec nicht verfügbar - nur PHP-Blocking")
else:
if not silent:
print(" ⚠️ CrowdSec nicht verfügbar - nur PHP-Blocking")
print("\n[1/3] CrowdSec-Watcher-Service bereits aktiv")
else:
if not silent:
print("\n[1/3] CrowdSec-Watcher-Service bereits aktiv")
print("\n[1/3] CrowdSec-Synchronisation deaktiviert (nur PHP-Modus)")
# Step 2: PHP blocking
if not silent:
@@ -516,14 +668,26 @@ def activate_blocking(shop, silent=False):
print(" ✏️ index.php modifiziert")
expiry = datetime.now() + timedelta(hours=72)
geoip_content = GEOIP_SCRIPT.format(
expiry_date=expiry.strftime('%Y-%m-%d %H:%M:%S CET'),
expiry_timestamp=expiry.strftime('%Y-%m-%d %H:%M:%S'),
cache_file=CACHE_FILE,
log_file=LOG_FILE,
crowdsec_queue=CROWDSEC_QUEUE_FILE,
shop_name=shop
)
# Generate PHP script based on mode
if mode == "php+crowdsec":
geoip_content = GEOIP_SCRIPT.format(
expiry_date=expiry.strftime('%Y-%m-%d %H:%M:%S CET'),
expiry_timestamp=expiry.strftime('%Y-%m-%d %H:%M:%S'),
cache_file=CACHE_FILE,
log_file=LOG_FILE,
crowdsec_queue=CROWDSEC_QUEUE_FILE,
shop_name=shop
)
else:
# PHP-only mode: no crowdsec queue writing
geoip_content = GEOIP_SCRIPT_PHP_ONLY.format(
expiry_date=expiry.strftime('%Y-%m-%d %H:%M:%S CET'),
expiry_timestamp=expiry.strftime('%Y-%m-%d %H:%M:%S'),
cache_file=CACHE_FILE,
log_file=LOG_FILE,
shop_name=shop
)
with open(blocking_file, 'w', encoding='utf-8') as f:
f.write(geoip_content)
@@ -533,7 +697,7 @@ def activate_blocking(shop, silent=False):
# Step 3: Register shop
if not silent:
print("\n[3/3] Registriere Shop...")
add_shop_to_active(shop)
add_shop_to_active(shop, mode)
if not silent:
print(" ✅ Shop registriert")
@@ -541,10 +705,12 @@ def activate_blocking(shop, silent=False):
print("\n" + "=" * 60)
print(f"✅ DACH GeoIP-Blocking aktiviert für: {shop}")
print(f" Erlaubte Länder: 🇩🇪 DE | 🇦🇹 AT | 🇨🇭 CH")
print(f" Modus: {'PHP + CrowdSec 🛡️' if mode == 'php+crowdsec' else 'Nur PHP 📝'}")
print(f" Gültig bis: {expiry.strftime('%Y-%m-%d %H:%M:%S CET')}")
print(f" PHP-Log: {os.path.join(httpdocs, LOG_FILE)}")
print(f" CrowdSec-Queue: {os.path.join(httpdocs, CROWDSEC_QUEUE_FILE)}")
print(f"\n 🔄 Der Watcher-Service synchronisiert blockierte IPs zu CrowdSec")
if mode == "php+crowdsec":
print(f" CrowdSec-Queue: {os.path.join(httpdocs, CROWDSEC_QUEUE_FILE)}")
print(f"\n 🔄 Der Watcher-Service synchronisiert blockierte IPs zu CrowdSec")
print("=" * 60)
return True
@@ -560,8 +726,12 @@ def deactivate_blocking(shop, silent=False):
log_file = os.path.join(httpdocs, LOG_FILE)
queue_file = os.path.join(httpdocs, CROWDSEC_QUEUE_FILE)
# Get mode before removing from tracking
shop_mode = get_shop_mode(shop)
if not silent:
print(f"\n🔧 Deaktiviere DACH GeoIP-Blocking für: {shop}")
print(f" Modus war: {'PHP + CrowdSec' if shop_mode == 'php+crowdsec' else 'Nur PHP'}")
print("=" * 60)
# Step 1: Remove PHP blocking
@@ -593,23 +763,27 @@ def deactivate_blocking(shop, silent=False):
if not silent:
print(" ✅ Shop deregistriert")
# Step 3: Clean CrowdSec decisions
# Step 3: Clean CrowdSec decisions (only if mode was php+crowdsec)
if not silent:
print("\n[3/4] CrowdSec-Decisions entfernen...")
if check_crowdsec():
if shop_mode == "php+crowdsec" and check_crowdsec():
cleanup_crowdsec_decisions(shop)
else:
if not silent:
print(" ℹ️ Keine CrowdSec-Synchronisation aktiv (PHP-only Modus)")
# Step 4: Uninstall service if last shop
# Step 4: Uninstall service if no more crowdsec shops
if not silent:
print("\n[4/4] Prüfe Watcher-Service...")
remaining_shops = [s for s in get_active_shops() if s != shop]
if not remaining_shops:
remaining_shops = get_active_shops()
crowdsec_shops = [s for s in remaining_shops if get_shop_mode(s) == "php+crowdsec"]
if not crowdsec_shops:
if not silent:
print(" ℹ️ Keine aktiven Shops mehr - deinstalliere Service")
print(" ℹ️ Keine Shops mit CrowdSec-Modus mehr - deinstalliere Service")
uninstall_watcher_service()
else:
if not silent:
print(f" ℹ️ Service bleibt aktiv ({len(remaining_shops)} Shop(s) noch aktiv)")
print(f" ℹ️ Service bleibt aktiv ({len(crowdsec_shops)} Shop(s) mit CrowdSec-Modus)")
if not silent:
print("\n" + "=" * 60)
@@ -638,7 +812,21 @@ def activate_all_shops():
for shop in available_shops:
print(f" • {shop}")
# Ask for mode
print(f"\n🔧 Wähle den Blocking-Modus:")
print(f" [1] PHP + CrowdSec (IPs werden an CrowdSec gemeldet)")
print(f" [2] Nur PHP (keine CrowdSec-Synchronisation)")
mode_choice = input(f"\nModus wählen [1/2]: ").strip()
if mode_choice == "2":
mode = "php-only"
mode_display = "Nur PHP 📝"
else:
mode = "php+crowdsec"
mode_display = "PHP + CrowdSec 🛡️"
print(f"\n⚠️ Dies aktiviert den Schutz für alle oben genannten Shops!")
print(f" Modus: {mode_display}")
confirm = input(f"\nFortfahren? (ja/nein): ").strip().lower()
if confirm not in ['ja', 'j', 'yes', 'y']:
@@ -646,17 +834,19 @@ def activate_all_shops():
return
print(f"\n{'=' * 60}")
print(" Starte Aktivierung...")
print(f" Starte Aktivierung ({mode_display})...")
print(f"{'=' * 60}")
success_count = 0
failed_count = 0
failed_shops = []
# Install watcher service first if needed
if not active_shops and check_crowdsec():
print("\n📦 Installiere CrowdSec-Watcher-Service...")
install_watcher_service()
# Install watcher service first if needed (only for php+crowdsec mode)
if mode == "php+crowdsec":
crowdsec_shops = [s for s in active_shops if get_shop_mode(s) == "php+crowdsec"]
if not crowdsec_shops and check_crowdsec():
print("\n📦 Installiere CrowdSec-Watcher-Service...")
install_watcher_service()
for i, shop in enumerate(available_shops, 1):
print(f"\n[{i}/{len(available_shops)}] Aktiviere: {shop}")
@@ -702,22 +892,31 @@ def activate_all_shops():
with open(index_php, 'w', encoding='utf-8') as f:
f.write('\n'.join(lines))
# Create blocking script
# Create blocking script based on mode
expiry = datetime.now() + timedelta(hours=72)
geoip_content = GEOIP_SCRIPT.format(
expiry_date=expiry.strftime('%Y-%m-%d %H:%M:%S CET'),
expiry_timestamp=expiry.strftime('%Y-%m-%d %H:%M:%S'),
cache_file=CACHE_FILE,
log_file=LOG_FILE,
crowdsec_queue=CROWDSEC_QUEUE_FILE,
shop_name=shop
)
if mode == "php+crowdsec":
geoip_content = GEOIP_SCRIPT.format(
expiry_date=expiry.strftime('%Y-%m-%d %H:%M:%S CET'),
expiry_timestamp=expiry.strftime('%Y-%m-%d %H:%M:%S'),
cache_file=CACHE_FILE,
log_file=LOG_FILE,
crowdsec_queue=CROWDSEC_QUEUE_FILE,
shop_name=shop
)
else:
geoip_content = GEOIP_SCRIPT_PHP_ONLY.format(
expiry_date=expiry.strftime('%Y-%m-%d %H:%M:%S CET'),
expiry_timestamp=expiry.strftime('%Y-%m-%d %H:%M:%S'),
cache_file=CACHE_FILE,
log_file=LOG_FILE,
shop_name=shop
)
with open(blocking_file, 'w', encoding='utf-8') as f:
f.write(geoip_content)
# Register shop
add_shop_to_active(shop)
# Register shop with mode
add_shop_to_active(shop, mode)
print(f" ✅ Aktiviert (bis {expiry.strftime('%Y-%m-%d %H:%M')})")
success_count += 1
@@ -740,6 +939,7 @@ def activate_all_shops():
print(f" • {shop}")
print(f"\n 🇩🇪 🇦🇹 🇨🇭 Nur DACH-Traffic erlaubt")
print(f" 🔧 Modus: {mode_display}")
print(f" ⏰ Gültig für 72 Stunden")
print(f"{'=' * 60}")
@@ -869,7 +1069,7 @@ def get_crowdsec_stats_by_shop():
return {}
stats = {}
code, stdout, _ = run_command("cscli decisions list -o raw")
code, stdout, _ = run_command("cscli decisions list -o raw --limit 0")
if code == 0 and stdout:
lines = stdout.strip().split('\n')
@@ -934,8 +1134,8 @@ def show_all_logs():
# Top blocked IPs
if all_ips:
print(f"\n🔥 Top 10 blockierte IPs (alle Shops):")
sorted_ips = sorted(all_ips.items(), key=lambda x: x[1], reverse=True)[:10]
print(f"\n🔥 Top 100 blockierte IPs (alle Shops):")
sorted_ips = sorted(all_ips.items(), key=lambda x: x[1], reverse=True)[:100]
for ip, count in sorted_ips:
bar = "█" * min(count // 5, 20) if count > 0 else "█"
print(f" {ip}: {count} {bar}")
@@ -950,9 +1150,13 @@ def show_logs(shop):
"""Show logs for a single shop"""
httpdocs = os.path.join(VHOSTS_DIR, shop, 'httpdocs')
log_file = os.path.join(httpdocs, LOG_FILE)
shop_mode = get_shop_mode(shop)
mode_display = "PHP + CrowdSec 🛡️" if shop_mode == "php+crowdsec" else "Nur PHP 📝"
print(f"\n📊 Logs für {shop} [{mode_display}]")
if os.path.isfile(log_file):
print(f"\n📊 PHP-Blocks für {shop}:")
print(f"\n📝 PHP-Blocks:")
print("=" * 80)
with open(log_file, 'r') as f:
lines = f.readlines()
@@ -961,15 +1165,15 @@ def show_logs(shop):
print("=" * 80)
print(f"Gesamt: {len(lines)}")
else:
print(f"ℹ️ Keine Logs für {shop}")
print(f"ℹ️ Keine PHP-Logs für {shop}")
if check_crowdsec():
print(f"\n📊 CrowdSec Decisions für {shop}:")
# Only show CrowdSec decisions if mode is php+crowdsec
if shop_mode == "php+crowdsec" and check_crowdsec():
print(f"\n🛡️ CrowdSec Decisions:")
print("=" * 80)
# Use raw output (CSV format)
# Format: id,source,ip,reason,action,country,as,events_count,expiration,simulated,alert_id
code, stdout, _ = run_command("cscli decisions list -o raw")
# Use raw output with --limit 0 (no pagination)
code, stdout, _ = run_command("cscli decisions list -o raw --limit 0")
if code == 0 and stdout:
lines = stdout.strip().split('\n')
shop_decisions = []
@@ -1004,6 +1208,8 @@ def show_logs(shop):
print("Konnte Decisions nicht abrufen")
print("=" * 80)
elif shop_mode == "php-only":
print(f"\n📝 CrowdSec-Synchronisation ist für diesen Shop deaktiviert (PHP-only Modus)")
def main():
@@ -1058,9 +1264,23 @@ def main():
shop_idx = int(shop_choice) - 1
if 0 <= shop_idx < len(available_shops):
selected_shop = available_shops[shop_idx]
confirm = input(f"\n⚠️ DACH-Blocking aktivieren für '{selected_shop}'? (ja/nein): ").strip().lower()
# Ask for mode
print(f"\n🔧 Wähle den Blocking-Modus:")
print(f" [1] PHP + CrowdSec (IPs werden an CrowdSec gemeldet)")
print(f" [2] Nur PHP (keine CrowdSec-Synchronisation)")
mode_choice = input(f"\nModus wählen [1/2]: ").strip()
if mode_choice == "2":
mode = "php-only"
mode_display = "Nur PHP"
else:
mode = "php+crowdsec"
mode_display = "PHP + CrowdSec"
confirm = input(f"\n⚠️ DACH-Blocking ({mode_display}) aktivieren für '{selected_shop}'? (ja/nein): ").strip().lower()
if confirm in ['ja', 'j', 'yes', 'y']:
activate_blocking(selected_shop)
activate_blocking(selected_shop, mode=mode)
else:
print("❌ Ungültig")
except ValueError:
@@ -1075,7 +1295,9 @@ def main():
print("\n📋 Aktive Shops:")
for i, shop in enumerate(active_shops, 1):
print(f" [{i}] {shop}")
mode = get_shop_mode(shop)
mode_icon = "🛡️" if mode == "php+crowdsec" else "📝"
print(f" [{i}] {shop} {mode_icon}")
shop_choice = input("\nWähle einen Shop: ").strip()
try:
@@ -1099,7 +1321,9 @@ def main():
print("\n📋 Logs anzeigen für:")
print(f" [0] 📊 ALLE Shops (Zusammenfassung)")
for i, shop in enumerate(active_shops, 1):
print(f" [{i}] {shop}")
mode = get_shop_mode(shop)
mode_icon = "🛡️" if mode == "php+crowdsec" else "📝"
print(f" [{i}] {shop} {mode_icon}")
shop_choice = input("\nWähle eine Option: ").strip()
try:
@@ -1121,7 +1345,10 @@ def main():
print(f" Aktive DACH-Blockings: {len(active_shops)}")
if active_shops:
for shop in active_shops:
print(f" ✓ {shop}")
mode = get_shop_mode(shop)
mode_icon = "🛡️" if mode == "php+crowdsec" else "📝"
mode_text = "PHP+CS" if mode == "php+crowdsec" else "PHP"
print(f" ✓ {shop} [{mode_text}] {mode_icon}")
elif choice == "5":
activate_all_shops()