#!/usr/bin/env python3 import os import sys import json import math import time import subprocess import secrets import hashlib import threading import re from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler from http import cookies from urllib.parse import urlparse, parse_qs from datetime import datetime, timedelta # --- CONSTANTS --- DB_FILE = "/etc/firewallfalcon/users.db" RESELLERS_DB = "/etc/firewallfalcon/resellers.db" BW_DIR = "/etc/firewallfalcon/bandwidth" PANEL_CONF = "/etc/firewallfalcon/panel.conf" PANEL_HTML = "/etc/firewallfalcon/panel/index.html" FF_USERS_GROUP = "firewallfalcon-users" PORT = 44380 # --- GLOBAL STATE --- SESSION_FILE = "/etc/firewallfalcon/sessions.json" sessions = {} # token -> {"username": str, "role": "admin"|"reseller", "created_at": float} def load_sessions(): global sessions try: if os.path.exists(SESSION_FILE): with open(SESSION_FILE, "r") as f: sessions = json.load(f) except Exception: sessions = {} def save_sessions(): try: os.makedirs(os.path.dirname(SESSION_FILE), exist_ok=True) with open(SESSION_FILE, "w") as f: json.dump(sessions, f) except Exception: pass load_sessions() db_lock = threading.Lock() PROTOCOLS = [ {"name": "OpenSSH", "service": "sshd", "service_alt": "ssh", "check_file": None, "port": "22"}, {"name": "BadVPN (UDPGW)", "service": "badvpn", "check_file": "/etc/systemd/system/badvpn.service", "port": "7300"}, {"name": "UDP Custom", "service": "udp-custom", "check_file": "/etc/systemd/system/udp-custom.service", "port": "36712"}, {"name": "HAProxy Edge", "service": "haproxy", "check_file": "/etc/haproxy/haproxy.cfg", "port": "80/443"}, {"name": "Nginx Proxy", "service": "nginx", "check_file": "/etc/nginx/sites-available/default", "port": "8880/8443"}, {"name": "DNSTT (SlowDNS)", "service": "dnstt", "check_file": "/etc/systemd/system/dnstt.service", "port": "53"}, {"name": "Falcon Proxy", "service": "falconproxy", "check_file": "/etc/systemd/system/falconproxy.service", "port": "8080"}, {"name": "ZiVPN", "service": "zivpn", "check_file": "/etc/systemd/system/zivpn.service", "port": "5667"}, {"name": "X-UI / 3X-UI", "service": "x-ui", "check_file": "/etc/systemd/system/x-ui.service", "port": "2053"}, ] # --- UTILS --- def run_cmd(cmd_args, ignore_errors=False): try: if isinstance(cmd_args, str): res = subprocess.run(cmd_args, shell=True, capture_output=True, text=True, timeout=10) else: res = subprocess.run(cmd_args, capture_output=True, text=True, timeout=10) if not ignore_errors and res.returncode != 0: print(f"Command error: {cmd_args} -> {res.stderr}", file=sys.stderr) return res.returncode, res.stdout.strip(), res.stderr.strip() except Exception as e: print(f"Exception running command {cmd_args}: {e}", file=sys.stderr) return -1, "", str(e) def force_delete_system_user(username): """Kill all processes and force-delete a Linux system user.""" import time as _time # Kill all processes owned by this user run_cmd(["pkill", "-9", "-u", username], ignore_errors=True) run_cmd(["killall", "-u", username, "-9"], ignore_errors=True) _time.sleep(0.5) # Force delete with retry code, _, _ = run_cmd(["userdel", "-rf", username], ignore_errors=True) if code != 0: # Retry after killing harder run_cmd(["pkill", "-9", "-u", username], ignore_errors=True) _time.sleep(1) run_cmd(["userdel", "-rf", username], ignore_errors=True) # --- USERS DB --- def parse_db_line(line): parts = line.strip().split(":") if len(parts) < 5: return None user = { "username": parts[0], "password": parts[1], "expire_date": parts[2], "conn_limit": int(parts[3]) if parts[3].isdigit() else 1, "bandwidth_gb": float(parts[4]), "daily_bandwidth_gb": 0.0, "account_type": "", "owner": "admin" } if len(parts) > 5: try: user["daily_bandwidth_gb"] = float(parts[5]) except ValueError: user["daily_bandwidth_gb"] = 0.0 if len(parts) > 6: user["account_type"] = parts[6] if len(parts) > 7: user["owner"] = parts[7] if parts[7] else "admin" return user def read_db(): users = [] if not os.path.exists(DB_FILE): return users with open(DB_FILE, "r") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue u = parse_db_line(line) if u: users.append(u) return users def _fmt_bw(v): """Format bandwidth: 0.0 -> '0', 3.5 -> '3.5', 10.0 -> '10'""" f = float(v) return str(int(f)) if f == int(f) else str(f) def format_db_line(u): bw = _fmt_bw(u.get('bandwidth_gb', 0)) dbw = _fmt_bw(u.get('daily_bandwidth_gb', 0)) owner = u.get('owner', 'admin') return f"{u['username']}:{u['password']}:{u['expire_date']}:{u['conn_limit']}:{bw}:{dbw}:{u.get('account_type','web')}:{owner}\n" # --- RESELLERS DB --- def parse_reseller_line(line): parts = line.strip().split(":") if len(parts) < 5: return None return { "username": parts[0], "password": parts[1], "expire_date": parts[2], "max_users": int(parts[3]) if parts[3].isdigit() else 10, "enabled": parts[4] == "1", "type": parts[5] if len(parts) > 5 else "quota", "credits": int(parts[6]) if len(parts) > 6 and parts[6].isdigit() else 0 } def read_resellers(): resellers = [] if not os.path.exists(RESELLERS_DB): return resellers with open(RESELLERS_DB, "r") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue r = parse_reseller_line(line) if r: resellers.append(r) return resellers def format_reseller_line(r): enabled = "1" if r.get("enabled", True) else "0" rtype = r.get("type", "quota") credits = r.get("credits", 0) return f"{r['username']}:{r['password']}:{r['expire_date']}:{r['max_users']}:{enabled}:{rtype}:{credits}\n" def write_resellers(resellers): os.makedirs(os.path.dirname(RESELLERS_DB), exist_ok=True) with open(RESELLERS_DB, "w") as f: for r in resellers: f.write(format_reseller_line(r)) def calculate_credit_cost(days, conn_limit=1): """Calculate credit cost: 1 credit per 30 days per device, rounded up.""" return math.ceil(days / 30) * max(1, conn_limit) # --- ONLINE SESSIONS --- def get_online_sessions(target_user=None): managed_users = set(u["username"] for u in read_db()) if target_user and target_user not in managed_users: return 0 user_pids = {} try: code, out, _ = run_cmd(["ps", "-C", "sshd,sshd-session", "-o", "pid=,user="], ignore_errors=True) if code == 0 and out: for line in out.strip().splitlines(): parts = line.split() if len(parts) != 2: continue pid, owner = parts if owner in ("root", "sshd", ""): continue if owner not in managed_users: continue if target_user and owner != target_user: continue if owner not in user_pids: user_pids[owner] = set() user_pids[owner].add(pid) except Exception as e: print(f"Error checking online sessions: {e}", file=sys.stderr) if target_user: return len(user_pids.get(target_user, set())) else: return sum(len(pids) for pids in user_pids.values()) def get_online_sessions_for_users(usernames): """Get online session counts for a set of usernames efficiently (single ps call).""" user_pids = {} try: code, out, _ = run_cmd(["ps", "-C", "sshd,sshd-session", "-o", "pid=,user="], ignore_errors=True) if code == 0 and out: for line in out.strip().splitlines(): parts = line.split() if len(parts) != 2: continue pid, owner = parts if owner in ("root", "sshd", ""): continue if owner not in usernames: continue if owner not in user_pids: user_pids[owner] = set() user_pids[owner].add(pid) except Exception as e: print(f"Error checking online sessions: {e}", file=sys.stderr) return user_pids def read_file_int(path, default=0): try: if os.path.exists(path): with open(path, "r") as f: return int(f.read().strip()) except Exception: pass return default def refresh_ssh_banner_config(): """Refresh dynamic SSH banner sshd config when banners are enabled.""" if not os.path.exists("/etc/firewallfalcon/banners_enabled"): return sshd_ff_config = "/etc/ssh/sshd_config.d/firewallfalcon-banners.conf" banner_dir = "/etc/firewallfalcon/banners" os.makedirs(banner_dir, exist_ok=True) lines = ["# FirewallFalcon - Dynamic per-user SSH banners\n"] users = read_db() for u in users: un = u["username"] lines.append(f"Match User {un}\n") lines.append(f" Banner {banner_dir}/{un}.txt\n") new_content = "".join(lines) # Only update if changed old_content = "" if os.path.exists(sshd_ff_config): try: with open(sshd_ff_config, "r") as f: old_content = f.read() except: pass if new_content != old_content: try: with open(sshd_ff_config, "w") as f: f.write(new_content) # Ensure Include directive exists try: with open("/etc/ssh/sshd_config", "r") as f: sshd_content = f.read() if "Include /etc/ssh/sshd_config.d/" not in sshd_content: with open("/etc/ssh/sshd_config", "a") as f: f.write("\nInclude /etc/ssh/sshd_config.d/*.conf\n") except: pass run_cmd("systemctl reload sshd 2>/dev/null || systemctl reload ssh 2>/dev/null", ignore_errors=True) except Exception as e: print(f"Error refreshing SSH banner config: {e}", file=sys.stderr) # --- PANEL CREDS --- def get_panel_creds(): creds = {"PANEL_USER": "", "PANEL_PASS_HASH": "", "PANEL_PASS_PLAIN": "", "PANEL_SECRET": "", "PANEL_NAME": "\ud835\udc06\ud835\udc01\ud835\udc0b-\ud835\udc02\ud835\udc0b\ud835\udc0e\ud835\udc14\ud835\udc03 \ud835\udfd0\ud835\udfd2\ud835\udfd1", "PANEL_LOGO": ""} if os.path.exists(PANEL_CONF): with open(PANEL_CONF, "r") as f: for line in f: line = line.strip() if "=" in line: k, v = line.split("=", 1) v = v.strip().strip('"').strip("'") if k in creds: creds[k] = v return creds def write_panel_creds(user, pass_plain, secret=None, panel_name=None, panel_logo=None): creds = get_panel_creds() creds["PANEL_USER"] = user creds["PANEL_PASS_PLAIN"] = pass_plain creds["PANEL_PASS_HASH"] = hashlib.sha256(pass_plain.encode()).hexdigest() if secret is not None: creds["PANEL_SECRET"] = secret.strip().lstrip('/') if panel_name is not None: creds["PANEL_NAME"] = panel_name.strip() or "\ud835\udc06\ud835\udc01\ud835\udc0b-\ud835\udc02\ud835\udc0b\ud835\udc0e\ud835\udc14\ud835\udc03 \ud835\udfd0\ud835\udfd2\ud835\udfd1" if panel_logo is not None: creds["PANEL_LOGO"] = panel_logo.strip() or "" os.makedirs(os.path.dirname(PANEL_CONF), exist_ok=True) with open(PANEL_CONF, "w") as f: for k, v in creds.items(): f.write(f"{k}={v}\n") # --- SESSION --- def cleanup_sessions(): now = time.time() expired = [t for t, s in sessions.items() if now - s["created_at"] > 86400] for t in expired: del sessions[t] if expired: save_sessions() def generate_password(length=8): chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" return "".join(secrets.choice(chars) for _ in range(length)) def calculate_expire_date(days): return (datetime.now() + timedelta(days=days)).strftime("%Y-%m-%d") def check_session(headers): """Returns session info dict or None if not authenticated.""" cleanup_sessions() if "Cookie" in headers: C = cookies.SimpleCookie(headers["Cookie"]) if "session" in C: token = C["session"].value if token in sessions: return sessions[token] return None # --- HTTP HANDLER --- class PanelAPIHandler(BaseHTTPRequestHandler): def log_message(self, format, *args): pass # Suppress default logging def send_json(self, status, data, extra_headers=None): self.send_response(status) self.send_header('Content-Type', 'application/json') if extra_headers: for k, v in extra_headers.items(): self.send_header(k, v) self.end_headers() self.wfile.write(json.dumps(data).encode('utf-8')) def _get_session(self): """Helper: returns session dict or sends 401 and returns None.""" s = check_session(self.headers) if not s: self.send_json(401, {"error": "Unauthorized"}) return s def _require_admin(self, session): """Helper: returns True if admin, else sends 403 and returns False.""" if session.get("role") != "admin": self.send_json(403, {"error": "Admin access required"}) return False return True def do_GET(self): try: parsed_path = urlparse(self.path) raw_path = parsed_path.path.strip() creds = get_panel_creds() secret = creds.get("PANEL_SECRET", "").strip().lstrip('/') # Check secret path matching for HTML serving is_html_request = False if not secret: if raw_path in ("/", "/index.html"): is_html_request = True else: valid_paths = (f"/{secret}", f"/{secret}/", f"/{secret}/index.html") if raw_path in valid_paths: is_html_request = True if is_html_request: if os.path.exists(PANEL_HTML): with open(PANEL_HTML, "rb") as f: content = f.read() self.send_response(200) self.send_header('Content-Type', 'text/html') self.end_headers() self.wfile.write(content) else: self.send_response(404) self.end_headers() self.wfile.write(b"HTML not found") return # Clean API path if prefixed with secret api_path = raw_path if secret and api_path.startswith(f"/{secret}/api/"): api_path = api_path[len(secret) + 1:] if not api_path.startswith("/api/"): self.send_response(404) self.end_headers() self.wfile.write(b"Not Found") return # Public endpoint: branding (no auth required) if api_path == "/api/branding": creds = get_panel_creds() self.send_json(200, { "panel_name": creds.get("PANEL_NAME", "\ud835\udc06\ud835\udc01\ud835\udc0b-\ud835\udc02\ud835\udc0b\ud835\udc0e\ud835\udc14\ud835\udc03 \ud835\udfd0\ud835\udfd2\ud835\udfd1"), "panel_logo": creds.get("PANEL_LOGO", ""), "has_custom_logo": os.path.exists("/etc/firewallfalcon/panel/logo.png") and creds.get("PANEL_LOGO") == "custom" }) return # Serve custom logo image (public, no auth) logo_path = "/etc/firewallfalcon/panel/logo.png" if api_path == "/api/logo.png": if os.path.exists(logo_path): with open(logo_path, "rb") as f: img_data = f.read() self.send_response(200) ct = 'image/png' if img_data[:3] == b'\xff\xd8\xff': ct = 'image/jpeg' elif img_data[:4] == b'RIFF' and img_data[8:12] == b'WEBP': ct = 'image/webp' elif img_data[:3] == b'GIF': ct = 'image/gif' self.send_header('Content-Type', ct) self.send_header('Cache-Control', 'no-cache') self.end_headers() self.wfile.write(img_data) else: self.send_response(404) self.end_headers() self.wfile.write(b"No custom logo") return session = self._get_session() if not session: return if api_path == "/api/me": self.handle_get_me(session) elif api_path == "/api/dashboard": self.handle_get_dashboard(session) elif api_path == "/api/users": self.handle_get_users(session) elif api_path == "/api/protocols": if not self._require_admin(session): return self.handle_get_protocols() elif api_path == "/api/settings": if not self._require_admin(session): return self.handle_get_settings() elif api_path == "/api/resellers": if not self._require_admin(session): return self.handle_get_resellers() else: self.send_json(404, {"error": "Not Found"}) except Exception as e: print(f"Error handling GET {self.path}: {e}", file=sys.stderr) self.send_json(500, {"error": str(e)}) def do_POST(self): parsed_path = urlparse(self.path) path = parsed_path.path content_length = int(self.headers.get('Content-Length', 0)) # Handle logo upload BEFORE reading body as JSON (binary upload) if path == "/api/logo/upload": session = self._get_session() if not session: return if not self._require_admin(session): return raw_data = self.rfile.read(content_length) if content_length > 0 else b'' self.handle_logo_upload(raw_data) return post_data = self.rfile.read(content_length) try: body = json.loads(post_data.decode('utf-8')) if post_data else {} except json.JSONDecodeError: body = {} if path == "/api/login": self.handle_login(body) return session = self._get_session() if not session: return if path == "/api/logout": self.handle_logout() elif path == "/api/logo/delete": if not self._require_admin(session): return self.handle_logo_delete() elif path == "/api/users": self.handle_post_users(body, session) elif path == "/api/users/bulk": self.handle_post_users_bulk(body, session) elif path == "/api/users/trial": self.handle_post_trial_user(body, session) elif path.startswith("/api/users/") and path.endswith("/lock"): user = path.split("/")[3] self.handle_user_action(user, "lock", session=session) elif path.startswith("/api/users/") and path.endswith("/unlock"): user = path.split("/")[3] self.handle_user_action(user, "unlock", session=session) elif path.startswith("/api/users/") and path.endswith("/renew"): user = path.split("/")[3] self.handle_user_action(user, "renew", body=body, session=session) elif path.startswith("/api/users/") and path.endswith("/reset-bandwidth"): user = path.split("/")[3] self.handle_user_action(user, "reset-bandwidth", session=session) elif path.startswith("/api/protocols/") and path.endswith("/restart"): if not self._require_admin(session): return service = path.split("/")[3] self.handle_protocol_restart(service) elif path == "/api/resellers": if not self._require_admin(session): return self.handle_post_reseller(body) elif path == "/api/system/reboot": if not self._require_admin(session): return self.handle_system_reboot() elif path.startswith("/api/resellers/") and path.endswith("/toggle"): if not self._require_admin(session): return reseller_name = path.split("/")[3] self.handle_toggle_reseller(reseller_name) elif path.startswith("/api/resellers/") and path.endswith("/add-credits"): if not self._require_admin(session): return username = path.split("/")[3] self.handle_add_credits(username, body) else: self.send_json(404, {"error": "Not Found"}) def do_PUT(self): parsed_path = urlparse(self.path) path = parsed_path.path session = self._get_session() if not session: return content_length = int(self.headers.get('Content-Length', 0)) post_data = self.rfile.read(content_length) try: body = json.loads(post_data.decode('utf-8')) if post_data else {} except json.JSONDecodeError: body = {} if path.startswith("/api/users/"): user = path.split("/")[3] self.handle_put_user(user, body, session) elif path == "/api/settings": if not self._require_admin(session): return self.handle_put_settings(body) elif path.startswith("/api/resellers/"): if not self._require_admin(session): return reseller_name = path.split("/")[3] self.handle_put_reseller(reseller_name, body) else: self.send_json(404, {"error": "Not Found"}) def do_DELETE(self): parsed_path = urlparse(self.path) path = parsed_path.path session = self._get_session() if not session: return if path.startswith("/api/resellers/"): if not self._require_admin(session): return reseller_name = path.split("/")[3] # Check query params for delete_users flag qs = parse_qs(urlparse(self.path).query) delete_users = qs.get("delete_users", ["0"])[0] == "1" self.handle_delete_reseller(reseller_name, delete_users) elif path.startswith("/api/users/"): user = path.split("/")[3] self.handle_delete_user(user, session) else: self.send_json(404, {"error": "Not Found"}) # --- OWNERSHIP CHECK --- def _check_user_ownership(self, username, session): """Check if the session owner can manage this user. Returns True if allowed.""" if session.get("role") == "admin": return True # Reseller can only manage their own users users = read_db() user = next((u for u in users if u["username"] == username), None) if not user: return False return user.get("owner", "admin") == session.get("username") # --- HANDLERS --- def handle_login(self, body): user = body.get("username", "") pwd = body.get("password", "") pwd_hash = hashlib.sha256(pwd.encode()).hexdigest() # Check admin credentials first creds = get_panel_creds() if user == creds.get("PANEL_USER") and pwd_hash == creds.get("PANEL_PASS_HASH"): token = secrets.token_hex(32) sessions[token] = {"username": user, "role": "admin", "created_at": time.time()} save_sessions() cookie_str = f"session={token}; Path=/; HttpOnly; Max-Age=86400" self.send_json(200, {"success": True, "role": "admin"}, {"Set-Cookie": cookie_str}) return # Check reseller credentials resellers = read_resellers() for r in resellers: if r["username"] == user and r["password"] == pwd: if not r["enabled"]: return self.send_json(401, {"error": "Account is disabled"}) # Check reseller expiry try: exp = datetime.strptime(r["expire_date"], "%Y-%m-%d") if exp < datetime.now(): return self.send_json(401, {"error": "Account has expired"}) except ValueError: pass token = secrets.token_hex(32) sessions[token] = {"username": user, "role": "reseller", "created_at": time.time()} save_sessions() cookie_str = f"session={token}; Path=/; HttpOnly; Max-Age=86400" self.send_json(200, {"success": True, "role": "reseller"}, {"Set-Cookie": cookie_str}) return self.send_json(401, {"error": "Invalid credentials"}) def handle_logout(self): if "Cookie" in self.headers: C = cookies.SimpleCookie(self.headers["Cookie"]) if "session" in C: token = C["session"].value if token in sessions: del sessions[token] save_sessions() cookie_str = f"session=; Path=/; HttpOnly; Max-Age=0" self.send_json(200, {"success": True}, {"Set-Cookie": cookie_str}) def handle_get_me(self, session): creds = get_panel_creds() data = { "role": session.get("role", "admin"), "username": session.get("username", ""), "panel_name": creds.get("PANEL_NAME", "\ud835\udc06\ud835\udc01\ud835\udc0b-\ud835\udc02\ud835\udc0b\ud835\udc0e\ud835\udc14\ud835\udc03 \ud835\udfd0\ud835\udfd2\ud835\udfd1"), "panel_logo": creds.get("PANEL_LOGO", ""), "has_custom_logo": os.path.exists("/etc/firewallfalcon/panel/logo.png") and creds.get("PANEL_LOGO") == "custom" } if session.get("role") == "reseller": resellers = read_resellers() r = next((x for x in resellers if x["username"] == session["username"]), None) if r: data["expire_date"] = r["expire_date"] data["max_users"] = r["max_users"] data["reseller_type"] = r.get("type", "quota") data["credits"] = r.get("credits", 0) users = read_db() owned = [u for u in users if u.get("owner") == session["username"]] data["created_users"] = len(owned) self.send_json(200, data) def handle_get_dashboard(self, session): users = read_db() if session.get("role") == "reseller": # Reseller sees only their stats owned_users = [u for u in users if u.get("owner") == session["username"]] owned_usernames = set(u["username"] for u in owned_users) online_pids = get_online_sessions_for_users(owned_usernames) online_total = sum(len(pids) for pids in online_pids.values()) resellers = read_resellers() r = next((x for x in resellers if x["username"] == session["username"]), None) max_users = r["max_users"] if r else 0 expire_date = r["expire_date"] if r else "N/A" self.send_json(200, { "user_count": len(owned_users), "online_sessions": online_total, "protocols": [], "reseller_info": { "max_users": max_users, "created_users": len(owned_users), "expire_date": expire_date, "type": r.get("type", "quota") if r else "quota", "credits": r.get("credits", 0) if r else 0 } }) return # --- ADMIN BRANCH --- _, ip, _ = run_cmd("curl -s -4 --max-time 3 icanhazip.com") os_name = "Unknown OS" try: with open("/etc/os-release") as f: for line in f: if line.startswith("PRETTY_NAME="): os_name = line.split("=")[1].strip().strip('"') break except Exception: pass _, uptime_str, _ = run_cmd("uptime -p") if uptime_str.startswith("up "): uptime_str = uptime_str[3:] ram_total = 0 ram_available = 0 try: with open("/proc/meminfo") as f: for line in f: if line.startswith("MemTotal:"): ram_total = int(line.split()[1]) // 1024 elif line.startswith("MemAvailable:"): ram_available = int(line.split()[1]) // 1024 except Exception: pass ram_used = ram_total - ram_available if ram_total > 0 else 0 ram_percent = round((ram_used / ram_total) * 100, 1) if ram_total > 0 else 0.0 cpu_load = 0.0 try: with open("/proc/loadavg") as f: cpu_load = float(f.read().split()[0]) except Exception: pass # Admin sees everything online_sessions = get_online_sessions() procs = self.get_protocols_status() self.send_json(200, { "server_ip": ip, "os_name": os_name, "uptime": uptime_str, "ram_percent": ram_percent, "ram_used_mb": ram_used, "ram_total_mb": ram_total, "cpu_load_1m": cpu_load, "user_count": len(users), "online_sessions": online_sessions, "protocols": procs }) def get_protocols_status(self): procs = [] for p in PROTOCOLS: installed = False if p["check_file"] is None: installed = True else: installed = os.path.exists(p["check_file"]) running = False if installed: code, _, _ = run_cmd(["systemctl", "is-active", p["service"]]) if code == 0: running = True elif "service_alt" in p: code2, _, _ = run_cmd(["systemctl", "is-active", p["service_alt"]]) if code2 == 0: running = True procs.append({ "name": p["name"], "service": p["service"], "installed": installed, "running": running, "port": p["port"] }) return procs def handle_get_users(self, session): users = read_db() # Scope by role if session.get("role") == "reseller": users = [u for u in users if u.get("owner") == session["username"]] # Efficient batch online session lookup all_usernames = set(u["username"] for u in users) online_pids = get_online_sessions_for_users(all_usernames) result = [] for u in users: un = u["username"] total_bw = read_file_int(f"{BW_DIR}/{un}.usage") daily_bw = read_file_int(f"{BW_DIR}/{un}.daily_usage") code, out, _ = run_cmd(["passwd", "-S", un]) is_locked = False if code == 0 and len(out.split()) >= 2: is_locked = (out.split()[1] == "L") code2, _, _ = run_cmd(["id", un]) exists_on_system = (code2 == 0) is_expired = False try: if u["expire_date"] != "Never" and u["expire_date"]: exp_date = datetime.strptime(u["expire_date"], "%Y-%m-%d") is_expired = exp_date < datetime.now() except ValueError: pass online_count = len(online_pids.get(un, set())) u_ext = dict(u) u_ext["total_used_bytes"] = total_bw u_ext["daily_used_bytes"] = daily_bw u_ext["is_locked"] = is_locked u_ext["is_expired"] = is_expired u_ext["is_online"] = online_count > 0 u_ext["online_sessions"] = online_count u_ext["exists_on_system"] = exists_on_system result.append(u_ext) self.send_json(200, {"users": result}) def _create_user(self, un, pwd, days, conn, bw, dbw, acct_type="web", owner="admin"): if not re.match(r'^[a-zA-Z0-9_]{3,32}$', un): raise ValueError("Invalid username") if any(u["username"] == un for u in read_db()): raise ValueError("User already exists in DB") code, _, _ = run_cmd(["id", un]) if code == 0: raise ValueError("User already exists on system") run_cmd(["useradd", "-m", "-s", "/usr/sbin/nologin", un]) run_cmd(["usermod", "-aG", FF_USERS_GROUP, un], ignore_errors=True) run_cmd(f"echo '{un}:{pwd}' | chpasswd") exp_date_str = calculate_expire_date(days) run_cmd(["chage", "-E", exp_date_str, un]) new_u = { "username": un, "password": pwd, "expire_date": exp_date_str, "conn_limit": conn, "bandwidth_gb": bw, "daily_bandwidth_gb": dbw, "account_type": acct_type, "owner": owner } with db_lock: os.makedirs(os.path.dirname(DB_FILE), exist_ok=True) with open(DB_FILE, "a") as f: f.write(format_db_line(new_u)) # Initialize bandwidth usage files to prevent immediate locking os.makedirs(BW_DIR, exist_ok=True) with open(f"{BW_DIR}/{un}.usage", "w") as f: f.write("0") with open(f"{BW_DIR}/{un}.daily_usage", "w") as f: f.write("0") refresh_ssh_banner_config() return new_u def handle_post_users(self, body, session): try: owner = session.get("username", "admin") if session.get("role") == "reseller" else "admin" # Reseller quota check if session.get("role") == "reseller": resellers = read_resellers() r = next((x for x in resellers if x["username"] == session["username"]), None) if not r: return self.send_json(403, {"error": "Reseller account not found"}) if r.get("type") == "credits": # Credit-based: check credit balance days = int(body.get("days", 30)) conn = int(body.get("conn_limit", 1)) cost = calculate_credit_cost(days, conn) if r.get("credits", 0) < cost: return self.send_json(403, {"error": f"Not enough credits. Need {cost}, have {r.get('credits', 0)}"}) else: # Quota-based: check user count users = read_db() owned = [u for u in users if u.get("owner") == session["username"]] if len(owned) >= r["max_users"]: return self.send_json(403, {"error": f"User limit reached ({r['max_users']})"}) un = body.get("username", "") pwd = body.get("password", "") or generate_password() days = int(body.get("days", 30)) conn = int(body.get("conn_limit", 1)) bw = float(body.get("bandwidth_gb", 0)) dbw = float(body.get("daily_bandwidth_gb", 0)) new_u = self._create_user(un, pwd, days, conn, bw, dbw, owner=owner) # Deduct credits if credit-based reseller if session.get("role") == "reseller": resellers = read_resellers() r = next((x for x in resellers if x["username"] == session["username"]), None) if r and r.get("type") == "credits": cost = calculate_credit_cost(days, conn) r["credits"] = max(0, r.get("credits", 0) - cost) write_resellers(resellers) self.send_json(200, new_u) except Exception as e: self.send_json(400, {"error": str(e)}) def handle_post_users_bulk(self, body, session): try: owner = session.get("username", "admin") if session.get("role") == "reseller" else "admin" # Reseller quota check remaining_quota = float('inf') if session.get("role") == "reseller": resellers = read_resellers() r = next((x for x in resellers if x["username"] == session["username"]), None) if not r: return self.send_json(403, {"error": "Reseller account not found"}) if r.get("type") == "credits": days_val = int(body.get("days", 30)) conn_val = int(body.get("conn_limit", 1)) cost_per = calculate_credit_cost(days_val, conn_val) count_requested = int(body.get("count", 1)) max_affordable = r.get("credits", 0) // cost_per if cost_per > 0 else 0 remaining_quota = max_affordable if remaining_quota <= 0: return self.send_json(403, {"error": f"Not enough credits. Need {cost_per} per user, have {r.get('credits', 0)}"}) else: users = read_db() owned = [u for u in users if u.get("owner") == session["username"]] remaining_quota = r["max_users"] - len(owned) if remaining_quota <= 0: return self.send_json(403, {"error": f"User limit reached ({r['max_users']})"}) prefix = body.get("prefix", "user") count = min(int(body.get("count", 1)), int(remaining_quota)) days = int(body.get("days", 30)) conn = int(body.get("conn_limit", 1)) bw = float(body.get("bandwidth_gb", 0)) dbw = float(body.get("daily_bandwidth_gb", 0)) created = [] existing_users = set(u["username"] for u in read_db()) idx = 1 for _ in range(count): while f"{prefix}{idx}" in existing_users: idx += 1 un = f"{prefix}{idx}" pwd = generate_password() try: u = self._create_user(un, pwd, days, conn, bw, dbw, "bulk", owner=owner) created.append(u) existing_users.add(un) except Exception as e: pass # skip failures in bulk idx += 1 # Deduct credits if credit-based reseller if session.get("role") == "reseller" and len(created) > 0: resellers = read_resellers() r = next((x for x in resellers if x["username"] == session["username"]), None) if r and r.get("type") == "credits": days_val = int(body.get("days", 30)) conn_val = int(body.get("conn_limit", 1)) total_cost = len(created) * calculate_credit_cost(days_val, conn_val) r["credits"] = max(0, r.get("credits", 0) - total_cost) write_resellers(resellers) self.send_json(200, {"users": created}) except Exception as e: self.send_json(400, {"error": str(e)}) def handle_post_trial_user(self, body, session): try: owner = session.get("username", "admin") if session.get("role") == "reseller" else "admin" is_reseller = session.get("role") == "reseller" if is_reseller: resellers = read_resellers() r = next((x for x in resellers if x["username"] == session["username"]), None) if not r: return self.send_json(403, {"error": "Reseller account not found"}) # Reseller trials: max 5 active trial accounts, 10 min lifetime, no credit cost users = read_db() owned_trials = [u for u in users if u.get("owner") == session["username"] and u.get("account_type") == "trial"] if len(owned_trials) >= 5: return self.send_json(403, {"error": "Trial limit reached (max 5 active trials)"}) # Force 10-minute trial for resellers total_minutes = 10 conn = 1 bw = 0 else: # Admin: accept minutes directly from frontend total_minutes = int(body.get("minutes", 60)) conn = int(body.get("conn_limit", 1)) bw = float(body.get("bandwidth_gb", 0)) un = body.get("username", "") if not un: import random, string un = "trial_" + "".join(random.choices(string.ascii_lowercase + string.digits, k=5)) pwd = body.get("password", "") or generate_password() # Always set days=1 minimum for chage (system expiry), real cleanup is via at/limiter days = max(1, total_minutes // 1440) new_u = self._create_user(un, pwd, days, conn, bw, 0, acct_type="trial", owner=owner) # Store trial expiry timestamp for backup cleanup by limiter from datetime import datetime, timedelta expiry_dt = datetime.now() + timedelta(minutes=total_minutes) expiry_ts = int(expiry_dt.timestamp()) trial_expiry_file = f"/etc/firewallfalcon/bandwidth/{un}.trial_expiry" os.makedirs("/etc/firewallfalcon/bandwidth", exist_ok=True) with open(trial_expiry_file, "w") as f: f.write(str(expiry_ts)) # Schedule auto-cleanup via 'at' daemon (primary method) cleanup_script = "/usr/local/bin/firewallfalcon-trial-cleanup.sh" if os.path.exists(cleanup_script): run_cmd(f"echo '{cleanup_script} {un}' | at now + {total_minutes} minutes", ignore_errors=True) expiry_time = expiry_dt.strftime("%Y-%m-%d %H:%M:%S") new_u["expiry_time"] = expiry_time new_u["minutes"] = total_minutes # No credit deduction for reseller trials self.send_json(200, new_u) except Exception as e: self.send_json(400, {"error": str(e)}) def handle_put_user(self, username, body, session): if not self._check_user_ownership(username, session): return self.send_json(403, {"error": "Access denied"}) with db_lock: users = read_db() idx = next((i for i, u in enumerate(users) if u["username"] == username), -1) if idx == -1: return self.send_json(404, {"error": "User not found"}) u = users[idx] if "password" in body: u["password"] = body["password"] run_cmd(f"echo '{username}:{u['password']}' | chpasswd") if "days" in body: u["expire_date"] = calculate_expire_date(int(body["days"])) run_cmd(["chage", "-E", u["expire_date"], username]) if "conn_limit" in body: u["conn_limit"] = int(body["conn_limit"]) if "bandwidth_gb" in body: u["bandwidth_gb"] = float(body["bandwidth_gb"]) if "daily_bandwidth_gb" in body: u["daily_bandwidth_gb"] = float(body["daily_bandwidth_gb"]) lines = [] with open(DB_FILE, "r") as f: lines = f.readlines() with open(DB_FILE, "w") as f: for line in lines: if line.startswith(f"{username}:"): f.write(format_db_line(u)) else: f.write(line) self.send_json(200, u) def handle_delete_user(self, username, session): if not self._check_user_ownership(username, session): return self.send_json(403, {"error": "Access denied"}) force_delete_system_user(username) with db_lock: lines = [] if os.path.exists(DB_FILE): with open(DB_FILE, "r") as f: lines = f.readlines() with open(DB_FILE, "w") as f: for line in lines: if not line.startswith(f"{username}:"): f.write(line) run_cmd(f"rm -f {BW_DIR}/{username}.*", ignore_errors=True) run_cmd(f"rm -f /etc/firewallfalcon/banners/{username}.txt", ignore_errors=True) refresh_ssh_banner_config() self.send_json(200, {"success": True}) def handle_user_action(self, username, action, body=None, session=None): if not self._check_user_ownership(username, session): return self.send_json(403, {"error": "Access denied"}) if action == "lock": run_cmd(["usermod", "-L", username]) run_cmd(["killall", "-u", username, "-9"], ignore_errors=True) self.send_json(200, {"success": True}) elif action == "unlock": run_cmd(["usermod", "-U", username]) run_cmd(f"rm -f {BW_DIR}/{username}.conn_locked", ignore_errors=True) run_cmd(f"rm -f {BW_DIR}/{username}.daily_locked", ignore_errors=True) self.send_json(200, {"success": True}) elif action == "renew": days = int(body.get("days", 30)) if body else 30 # Deduct credits for credit-based reseller renewals if session and session.get("role") == "reseller": resellers = read_resellers() r = next((x for x in resellers if x["username"] == session["username"]), None) if r and r.get("type") == "credits": # Look up the user's conn_limit for renewal cost users = read_db() u = next((x for x in users if x["username"] == username), None) user_conn = u.get("conn_limit", 1) if u else 1 cost = calculate_credit_cost(days, user_conn) if r.get("credits", 0) < cost: return self.send_json(403, {"error": f"Not enough credits. Need {cost}, have {r.get('credits', 0)}"}) r["credits"] = max(0, r.get("credits", 0) - cost) write_resellers(resellers) with db_lock: users = read_db() u = next((x for x in users if x["username"] == username), None) if not u: return self.send_json(404, {"error": "User not found"}) u["expire_date"] = calculate_expire_date(days) run_cmd(["chage", "-E", u["expire_date"], username]) lines = [] with open(DB_FILE, "r") as f: lines = f.readlines() with open(DB_FILE, "w") as f: for line in lines: if line.startswith(f"{username}:"): f.write(format_db_line(u)) else: f.write(line) self.send_json(200, {"success": True, "expire_date": u["expire_date"]}) elif action == "reset-bandwidth": os.makedirs(BW_DIR, exist_ok=True) with open(f"{BW_DIR}/{username}.usage", "w") as f: f.write("0") with open(f"{BW_DIR}/{username}.daily_usage", "w") as f: f.write("0") run_cmd(f"rm -f {BW_DIR}/{username}.conn_locked", ignore_errors=True) run_cmd(f"rm -f {BW_DIR}/{username}.daily_locked", ignore_errors=True) run_cmd(["usermod", "-U", username]) self.send_json(200, {"success": True}) else: self.send_json(400, {"error": "Unknown action"}) def handle_get_protocols(self): self.send_json(200, self.get_protocols_status()) def handle_protocol_restart(self, service): code, out, err = run_cmd(["systemctl", "restart", service]) if code == 0: self.send_json(200, {"success": True}) else: self.send_json(500, {"success": False, "error": err}) def handle_system_reboot(self): self.send_json(200, {"success": True, "message": "Rebooting system..."}) # Schedule reboot in background so we can return response first threading.Timer(2.0, lambda: subprocess.run(["reboot"])).start() def handle_logo_upload(self, raw_data): try: if len(raw_data) > 2097152: # 2MB max return self.send_json(400, {"error": "File too large (max 2MB)"}) if len(raw_data) == 0: return self.send_json(400, {"error": "No file data"}) logo_path = "/etc/firewallfalcon/panel/logo.png" os.makedirs(os.path.dirname(logo_path), exist_ok=True) with open(logo_path, "wb") as f: f.write(raw_data) # Set PANEL_LOGO to 'custom' to signal frontend to use image creds = get_panel_creds() creds["PANEL_LOGO"] = "custom" os.makedirs(os.path.dirname(PANEL_CONF), exist_ok=True) with open(PANEL_CONF, "w") as f: for k, v in creds.items(): f.write(f"{k}={v}\n") self.send_json(200, {"success": True, "panel_logo": "custom"}) except Exception as e: self.send_json(500, {"error": str(e)}) def handle_logo_delete(self): logo_path = "/etc/firewallfalcon/panel/logo.png" try: if os.path.exists(logo_path): os.remove(logo_path) creds = get_panel_creds() creds["PANEL_LOGO"] = "" os.makedirs(os.path.dirname(PANEL_CONF), exist_ok=True) with open(PANEL_CONF, "w") as f: for k, v in creds.items(): f.write(f"{k}={v}\n") self.send_json(200, {"success": True, "panel_logo": ""}) except Exception as e: self.send_json(500, {"error": str(e)}) def handle_get_settings(self): creds = get_panel_creds() self.send_json(200, { "username": creds.get("PANEL_USER", ""), "secret": creds.get("PANEL_SECRET", ""), "panel_name": creds.get("PANEL_NAME", "\ud835\udc06\ud835\udc01\ud835\udc0b-\ud835\udc02\ud835\udc0b\ud835\udc0e\ud835\udc14\ud835\udc03 \ud835\udfd0\ud835\udfd2\ud835\udfd1"), "panel_logo": creds.get("PANEL_LOGO", ""), "has_custom_logo": os.path.exists("/etc/firewallfalcon/panel/logo.png") and creds.get("PANEL_LOGO") == "custom" }) def handle_put_settings(self, body): creds = get_panel_creds() curr_pwd = body.get("current_password", "") new_user = body.get("new_username", "").strip() or creds.get("PANEL_USER", "") new_pwd = body.get("new_password", "").strip() or creds.get("PANEL_PASS_PLAIN", "") new_secret = body.get("new_secret", "").strip().lstrip('/') new_name = body.get("panel_name", "").strip() new_logo = body.get("panel_logo", "").strip() curr_hash = hashlib.sha256(curr_pwd.encode()).hexdigest() if curr_hash != creds.get("PANEL_PASS_HASH"): return self.send_json(401, {"error": "Invalid current password"}) write_panel_creds(new_user, new_pwd, secret=new_secret, panel_name=new_name or None, panel_logo=new_logo or None) self.send_json(200, {"success": True, "secret": new_secret, "panel_name": new_name or creds.get("PANEL_NAME", "\ud835\udc06\ud835\udc01\ud835\udc0b-\ud835\udc02\ud835\udc0b\ud835\udc0e\ud835\udc14\ud835\udc03 \ud835\udfd0\ud835\udfd2\ud835\udfd1"), "panel_logo": new_logo or creds.get("PANEL_LOGO", "")}) # --- RESELLER HANDLERS (admin only) --- def handle_get_resellers(self): resellers = read_resellers() users = read_db() result = [] for r in resellers: owned = [u for u in users if u.get("owner") == r["username"]] is_expired = False try: exp = datetime.strptime(r["expire_date"], "%Y-%m-%d") is_expired = exp < datetime.now() except ValueError: pass result.append({ "username": r["username"], "password": r["password"], "expire_date": r["expire_date"], "max_users": r["max_users"], "created_users": len(owned), "enabled": r["enabled"], "is_expired": is_expired, "type": r.get("type", "quota"), "credits": r.get("credits", 0) }) self.send_json(200, {"resellers": result}) def handle_post_reseller(self, body): try: un = body.get("username", "").strip() pwd = body.get("password", "") or generate_password() days = int(body.get("days", 30)) max_users = int(body.get("max_users", 10)) if not re.match(r'^[a-zA-Z0-9_]{3,32}$', un): return self.send_json(400, {"error": "Invalid username (3-32 chars, alphanumeric + underscore)"}) # Check conflicts with admin username creds = get_panel_creds() if un == creds.get("PANEL_USER"): return self.send_json(400, {"error": "Username conflicts with admin"}) resellers = read_resellers() if any(r["username"] == un for r in resellers): return self.send_json(400, {"error": "Reseller already exists"}) rtype = body.get("type", "quota") if rtype == "credits": credits_amount = int(body.get("credits", 0)) credit_days = int(body.get("days", 0)) new_r = { "username": un, "password": pwd, "expire_date": calculate_expire_date(credit_days) if credit_days > 0 else "Never", "max_users": 0, "enabled": True, "type": "credits", "credits": credits_amount } else: new_r = { "username": un, "password": pwd, "expire_date": calculate_expire_date(days), "max_users": max_users, "enabled": True, "type": "quota", "credits": 0 } resellers.append(new_r) write_resellers(resellers) self.send_json(200, new_r) except Exception as e: self.send_json(400, {"error": str(e)}) def handle_put_reseller(self, username, body): resellers = read_resellers() idx = next((i for i, r in enumerate(resellers) if r["username"] == username), -1) if idx == -1: return self.send_json(404, {"error": "Reseller not found"}) r = resellers[idx] if "password" in body and body["password"]: r["password"] = body["password"] if "days" in body: r["expire_date"] = calculate_expire_date(int(body["days"])) if "max_users" in body: r["max_users"] = int(body["max_users"]) if "add_credits" in body: r["credits"] = r.get("credits", 0) + int(body["add_credits"]) if "set_credits" in body: r["credits"] = int(body["set_credits"]) resellers[idx] = r write_resellers(resellers) self.send_json(200, r) def handle_toggle_reseller(self, username): resellers = read_resellers() idx = next((i for i, r in enumerate(resellers) if r["username"] == username), -1) if idx == -1: return self.send_json(404, {"error": "Reseller not found"}) resellers[idx]["enabled"] = not resellers[idx]["enabled"] now_enabled = resellers[idx]["enabled"] write_resellers(resellers) # Lock/unlock all SSH users owned by this reseller users = read_db() owned = [u for u in users if u.get("owner") == username] for u in owned: un = u["username"] if now_enabled: run_cmd(f"usermod -U {un}", ignore_errors=True) else: run_cmd(f"usermod -L {un}", ignore_errors=True) run_cmd(f"killall -u {un} -9", ignore_errors=True) # Invalidate reseller's panel sessions when disabling if not now_enabled: tokens_to_remove = [t for t, s in sessions.items() if s.get("username") == username and s.get("role") == "reseller"] for t in tokens_to_remove: del sessions[t] if tokens_to_remove: save_sessions() self.send_json(200, {"success": True, "enabled": now_enabled, "affected_users": len(owned)}) def handle_add_credits(self, username, body): resellers = read_resellers() idx = next((i for i, r in enumerate(resellers) if r["username"] == username), -1) if idx == -1: return self.send_json(404, {"error": "Reseller not found"}) amount = int(body.get("credits", 0)) if amount <= 0: return self.send_json(400, {"error": "Credits must be positive"}) resellers[idx]["credits"] = resellers[idx].get("credits", 0) + amount write_resellers(resellers) self.send_json(200, {"success": True, "credits": resellers[idx]["credits"]}) def handle_delete_reseller(self, username, delete_users=False): resellers = read_resellers() idx = next((i for i, r in enumerate(resellers) if r["username"] == username), -1) if idx == -1: return self.send_json(404, {"error": "Reseller not found"}) if delete_users: users = read_db() owned = [u for u in users if u.get("owner") == username] for u in owned: un = u["username"] force_delete_system_user(un) run_cmd(f"rm -f {BW_DIR}/{un}.*", ignore_errors=True) run_cmd(f"rm -f /etc/firewallfalcon/banners/{un}.txt", ignore_errors=True) with db_lock: if os.path.exists(DB_FILE): with open(DB_FILE, "r") as f: lines = f.readlines() owned_names = set(u["username"] for u in owned) with open(DB_FILE, "w") as f: for line in lines: parts = line.strip().split(":") if parts and parts[0] not in owned_names: f.write(line) del resellers[idx] write_resellers(resellers) # Invalidate reseller's sessions tokens_to_remove = [t for t, s in sessions.items() if s.get("username") == username and s.get("role") == "reseller"] for t in tokens_to_remove: del sessions[t] if tokens_to_remove: save_sessions() self.send_json(200, {"success": True}) def main(): os.makedirs(os.path.dirname(DB_FILE), exist_ok=True) server_address = ('0.0.0.0', PORT) httpd = ThreadingHTTPServer(server_address, PanelAPIHandler) print(f"Starting server on port {PORT}...", flush=True) try: httpd.serve_forever() except KeyboardInterrupt: pass httpd.server_close() print("Server stopped.") if __name__ == '__main__': main()