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-17 10:56:01 +01:00
parent c21d62a27a
commit 7b681aab10
1 changed files with 107 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
geoip_shop_manager.py
View File

@@ -8,6 +8,7 @@ Supports two modes:
- geoip: GeoIP blocking (only allowed regions can access)
- bot: Rate-limit bots by bot-type, shop remains globally accessible
v4.2.0: Korrekte Ownership für erstellte Dateien (nicht mehr root:root)
v4.1.0: IP-basierte Bot-Erkennung (für Bots die sich tarnen)
v4.0.0: Bot-Rate-Limiting nach Bot-Typ (nicht IP), CrowdSec entfernt
"""
@@ -23,6 +24,7 @@ import socket
import ipaddress
from datetime import datetime, timedelta
from pathlib import Path
from collections import Counter
# ANSI Color Codes
COLOR_GREEN = "\033[92m"
@@ -520,6 +522,70 @@ GENERIC_BOT_PATTERNS = [
]
# =============================================================================
# OWNERSHIP HELPER FUNCTIONS
# =============================================================================
def get_most_common_owner(httpdocs_path):
"""
Ermittelt die häufigste uid:gid-Kombination im httpdocs-Verzeichnis.
Gibt (uid, gid) zurück oder (None, None) wenn nicht ermittelbar.
"""
if not os.path.isdir(httpdocs_path):
return None, None
owner_counts = Counter()
try:
for entry in os.listdir(httpdocs_path):
entry_path = os.path.join(httpdocs_path, entry)
try:
stat_info = os.stat(entry_path)
owner_counts[(stat_info.st_uid, stat_info.st_gid)] += 1
except (OSError, IOError):
continue
except (OSError, IOError):
return None, None
if not owner_counts:
return None, None
# Häufigste Kombination zurückgeben
most_common = owner_counts.most_common(1)[0][0]
return most_common
def set_owner(path, uid, gid, recursive=False):
"""
Setzt Owner und Gruppe für eine Datei oder einen Ordner.
Optional rekursiv für Verzeichnisse.
"""
if uid is None or gid is None:
return
try:
os.chown(path, uid, gid)
if recursive and os.path.isdir(path):
for root, dirs, files in os.walk(path):
for d in dirs:
try:
os.chown(os.path.join(root, d), uid, gid)
except (OSError, IOError):
pass
for f in files:
try:
os.chown(os.path.join(root, f), uid, gid)
except (OSError, IOError):
pass
except (OSError, IOError):
pass
# =============================================================================
# EXISTING HELPER FUNCTIONS
# =============================================================================
def ip_in_cidr(ip_str, cidr_str):
"""Prüft ob eine IP in einem CIDR-Netz liegt."""
try:
@@ -624,7 +690,7 @@ def generate_php_bot_ip_ranges():
# CACHE VALIDATION
# =============================================================================
def generate_and_validate_cache(httpdocs_path, geo_region):
def generate_and_validate_cache(httpdocs_path, geo_region, uid=None, gid=None):
cache_file = os.path.join(httpdocs_path, CACHE_FILE)
region_info = get_geo_region_info(geo_region)
countries = region_info["countries"]
@@ -662,6 +728,9 @@ if (count($ranges) >= {min_expected}) {{
result = subprocess.run(['php', temp_php], capture_output=True, text=True, timeout=120)
output = result.stdout.strip()
if output.startswith('OK:'):
# Cache-Datei wurde erstellt, Ownership setzen
if os.path.isfile(cache_file):
set_owner(cache_file, uid, gid)
return True, int(output.split(':')[1]), None
elif output.startswith('FAIL:'):
return False, int(output.split(':')[1]), f"Nur {output.split(':')[1]} Ranges (min. {min_expected} erwartet)"
@@ -1227,6 +1296,18 @@ def activate_blocking(shop, silent=False, mode="geoip", geo_region="dach", rate_
print(f"❌ index.php nicht gefunden")
return False
# Ermittle den häufigsten Owner im httpdocs-Verzeichnis
uid, gid = get_most_common_owner(httpdocs)
if not silent and uid is not None:
import pwd
import grp
try:
user_name = pwd.getpwuid(uid).pw_name
group_name = grp.getgrgid(gid).gr_name
print(f"\n👤 Ownership: {user_name}:{group_name} (uid={uid}, gid={gid})")
except (KeyError, ImportError):
print(f"\n👤 Ownership: uid={uid}, gid={gid}")
if not silent:
print(f"\n🔧 Aktiviere {region_info['icon']} {region_info['name']} für: {shop}")
if bot_mode:
@@ -1242,6 +1323,7 @@ def activate_blocking(shop, silent=False, mode="geoip", geo_region="dach", rate_
print("\n[1/3] Aktiviere PHP-Blocking...")
shutil.copy2(index_php, backup_php)
set_owner(backup_php, uid, gid)
with open(index_php, 'r', encoding='utf-8') as f:
content = f.read()
@@ -1260,6 +1342,7 @@ def activate_blocking(shop, silent=False, mode="geoip", geo_region="dach", rate_
lines.insert(insert_line, require_statement)
with open(index_php, 'w', encoding='utf-8') as f:
f.write('\n'.join(lines))
set_owner(index_php, uid, gid)
expiry = datetime.now() + timedelta(hours=72)
@@ -1271,6 +1354,8 @@ def activate_blocking(shop, silent=False, mode="geoip", geo_region="dach", rate_
os.chmod(ratelimit_path, 0o777)
os.chmod(os.path.join(ratelimit_path, 'bans'), 0o777)
os.chmod(os.path.join(ratelimit_path, 'counts'), 0o777)
# Set ownership for ratelimit directories
set_owner(ratelimit_path, uid, gid, recursive=True)
if rate_limit is None:
rate_limit = DEFAULT_RATE_LIMIT
@@ -1304,6 +1389,8 @@ def activate_blocking(shop, silent=False, mode="geoip", geo_region="dach", rate_
with open(blocking_file, 'w', encoding='utf-8') as f:
f.write(geoip_content)
set_owner(blocking_file, uid, gid)
if not silent:
print(" ✅ PHP-Blocking aktiviert")
@@ -1316,7 +1403,7 @@ def activate_blocking(shop, silent=False, mode="geoip", geo_region="dach", rate_
if not silent:
print(f"\n[2/3] Generiere IP-Range-Cache ({len(region_info['countries'])} Länder)...")
success, range_count, error = generate_and_validate_cache(httpdocs, geo_region)
success, range_count, error = generate_and_validate_cache(httpdocs, geo_region, uid, gid)
if success:
if not silent:
@@ -1470,7 +1557,11 @@ def activate_all_shops():
print(f" ⚠️ Übersprungen")
continue
# Ermittle Owner für diesen Shop
uid, gid = get_most_common_owner(httpdocs)
shutil.copy2(index_php, backup_php)
set_owner(backup_php, uid, gid)
with open(index_php, 'r', encoding='utf-8') as f:
content = f.read()
@@ -1488,6 +1579,7 @@ def activate_all_shops():
lines.insert(insert_line, require_statement)
with open(index_php, 'w', encoding='utf-8') as f:
f.write('\n'.join(lines))
set_owner(index_php, uid, gid)
expiry = datetime.now() + timedelta(hours=72)
@@ -1497,6 +1589,7 @@ def activate_all_shops():
os.chmod(ratelimit_path, 0o777)
os.chmod(os.path.join(ratelimit_path, 'bans'), 0o777)
os.chmod(os.path.join(ratelimit_path, 'counts'), 0o777)
set_owner(ratelimit_path, uid, gid, recursive=True)
geoip_content = BOT_SCRIPT_TEMPLATE.format(
expiry_date=expiry.strftime('%Y-%m-%d %H:%M:%S CET'),
@@ -1524,6 +1617,7 @@ def activate_all_shops():
with open(blocking_file, 'w', encoding='utf-8') as f:
f.write(geoip_content)
set_owner(blocking_file, uid, gid)
if bot_mode:
add_shop_to_active(shop, mode, geo_region, rate_limit, ban_duration)
@@ -1531,7 +1625,7 @@ def activate_all_shops():
print(f" ✅ Aktiviert ({len(BOT_PATTERNS)} Patterns, {total_ip_ranges} IP-Ranges, {rate_limit} req/min)")
else:
print(f" ⏳ Cache generieren...")
cache_ok, count, _ = generate_and_validate_cache(httpdocs, geo_region)
cache_ok, count, _ = generate_and_validate_cache(httpdocs, geo_region, uid, gid)
add_shop_to_active(shop, mode, geo_region)
print(f" ✅ Aktiviert ({count:,} Ranges)" if cache_ok else f" ⚠️ Aktiviert (Cache-Warnung)")
@@ -1620,7 +1714,11 @@ def activate_direct_shops_only():
print(f" ⚠️ Übersprungen")
continue
# Ermittle Owner für diesen Shop
uid, gid = get_most_common_owner(httpdocs)
shutil.copy2(index_php, backup_php)
set_owner(backup_php, uid, gid)
with open(index_php, 'r', encoding='utf-8') as f:
content = f.read()
@@ -1638,6 +1736,7 @@ def activate_direct_shops_only():
lines.insert(insert_line, require_statement)
with open(index_php, 'w', encoding='utf-8') as f:
f.write('\n'.join(lines))
set_owner(index_php, uid, gid)
expiry = datetime.now() + timedelta(hours=72)
@@ -1647,6 +1746,7 @@ def activate_direct_shops_only():
os.chmod(ratelimit_path, 0o777)
os.chmod(os.path.join(ratelimit_path, 'bans'), 0o777)
os.chmod(os.path.join(ratelimit_path, 'counts'), 0o777)
set_owner(ratelimit_path, uid, gid, recursive=True)
geoip_content = BOT_SCRIPT_TEMPLATE.format(
expiry_date=expiry.strftime('%Y-%m-%d %H:%M:%S CET'),
@@ -1674,6 +1774,7 @@ def activate_direct_shops_only():
with open(blocking_file, 'w', encoding='utf-8') as f:
f.write(geoip_content)
set_owner(blocking_file, uid, gid)
if bot_mode:
add_shop_to_active(shop, mode, geo_region, rate_limit, ban_duration)
@@ -1681,7 +1782,7 @@ def activate_direct_shops_only():
print(f" ✅ Aktiviert ({len(BOT_PATTERNS)} Patterns, {total_ip_ranges} IP-Ranges, {rate_limit} req/min)")
else:
print(f" ⏳ Cache generieren...")
cache_ok, count, _ = generate_and_validate_cache(httpdocs, geo_region)
cache_ok, count, _ = generate_and_validate_cache(httpdocs, geo_region, uid, gid)
add_shop_to_active(shop, mode, geo_region)
print(f" ✅ Aktiviert ({count:,} Ranges)" if cache_ok else f" ⚠️ Aktiviert (Cache-Warnung)")
@@ -2056,11 +2157,12 @@ def main():
total_ip_ranges = sum(len(ranges) for ranges in BOT_IP_RANGES.values())
print("\n" + "=" * 60)
print(" GeoIP Shop Blocker Manager v4.1.0")
print(" GeoIP Shop Blocker Manager v4.2.0")
print(" 🇩🇪🇦🇹🇨🇭 DACH | 🇪🇺 Eurozone+GB | 🤖 Bot-Rate-Limiting")
print(" 🛡️ Mit Cache-Validierung und Fail-Open")
print(" 🚦 Rate-Limiting nach BOT-TYP (nicht IP)")
print(f" 🌐 IP-basierte Erkennung: {total_ip_ranges} Ranges für {len(BOT_IP_RANGES)} Bot(s)")
print(" 👤 Korrekte Ownership für erstellte Dateien")
print("=" * 60)
while True: