""" sessions.py — cPanel/WHM/Webmail session login and account management. """ import json import socket from modules.output import good, bad, warn, info, header from modules.utils import run def _whmapi1(args): """Run whmapi1 with JSON output, return parsed dict or None on error.""" out, _, rc = run(['whmapi1', '--output=json'] + args) if rc != 0: warn("whmapi1 call failed.") return None try: return json.loads(out) except json.JSONDecodeError: warn("Could not parse whmapi1 output.") return None def _ask(prompt_str): try: return input(f" {prompt_str}").strip() except (EOFError, KeyboardInterrupt): return '' # ── Session Login Functions ─────────────────────────────────────────────────── def cpanel_login(username=None): """Create a cPanel session URL for a user.""" if not username: username = _ask("cPanel username: ") if not username: return data = _whmapi1(['create_user_session', f'user={username}', 'service=cpaneld']) if not data: return try: url = data['data']['url'] print(f"\ncPanel Login URL:\n{url}\n") except KeyError: reason = data.get('metadata', {}).get('reason', 'Unknown error') warn(f"Failed: {reason}") def whm_login(username=None): """Create a WHM/reseller session URL for a user.""" if not username: username = _ask("Username: ") if not username: return data = _whmapi1(['create_user_session', f'user={username}', 'service=whostmgrd']) if not data: return if data.get('metadata', {}).get('result') == 0: reason = data['metadata'].get('reason', 'Unknown error') warn(f"Failed: {reason}") return try: url = data['data']['url'] print(f"\nWHM Login URL:\n{url}\n") except KeyError: warn("Could not extract URL from response.") def root_whm_login(): """Create a root WHM session URL using the server's primary IP.""" try: ip = socket.gethostbyname(socket.getfqdn()) except Exception: ip = socket.getfqdn() data = _whmapi1(['create_user_session', 'user=root', 'service=whostmgrd', f'preferred_domain={ip}']) if not data: return try: url = data['data']['url'] print(f"\nRoot WHM Login URL:\n{url}\n") except KeyError: reason = data.get('metadata', {}).get('reason', 'Unknown error') warn(f"Failed: {reason}") def webmail_login(): """Guided webmail session URL creation.""" email = _ask("Webmail User (user@domain.com): ") if not email or '@' not in email: warn("Invalid email address.") return ip = _ask("Your IP Address: ") if not ip: warn("IP address required.") return login, domain = email.split('@', 1) # Find cPanel user who owns this domain out, _, rc = run(['whmapi1', '--output=json', 'domainuserdata', f'domain={domain}']) if rc != 0: warn(f"Could not look up domain: {domain}") return try: domdata = json.loads(out) cpanel_user = domdata['data']['userdata']['user'] except (json.JSONDecodeError, KeyError): warn("Could not determine cPanel user for this domain.") return out, _, rc = run(['uapi', '--output=json', f'--user={cpanel_user}', 'Session', 'create_webmail_session_for_mail_user', f'login={login}', f'remote_address={ip}', f'domain={domain}']) if rc != 0: warn("uapi call failed.") return try: result = json.loads(out) data = result['result']['data'] token = data['token'] session = data['session'] hostname = socket.getfqdn() url = f"https://{hostname}:2096{token}/login/?session={session}" print(f"\nWebmail Login URL:\n{url}\n") except (json.JSONDecodeError, KeyError) as e: warn(f"Could not parse response: {e}") # ── Shell Management ────────────────────────────────────────────────────────── def shell_enable(username=None): """Enable shell access for a cPanel user.""" if not username: username = _ask("cPanel username: ") if not username: return data = _whmapi1(['modifyacct', f'user={username}', 'HASSHELL=1']) if data and data.get('metadata', {}).get('result') == 1: good(f"Shell access enabled for {username}") else: reason = (data or {}).get('metadata', {}).get('reason', 'Unknown error') warn(f"Failed: {reason}") def shell_disable(username=None): """Disable shell access for a cPanel user.""" if not username: username = _ask("cPanel username: ") if not username: return data = _whmapi1(['modifyacct', f'user={username}', 'HASSHELL=0']) if data and data.get('metadata', {}).get('result') == 1: good(f"Shell access disabled for {username}") else: reason = (data or {}).get('metadata', {}).get('reason', 'Unknown error') warn(f"Failed: {reason}") # ── 2FA Management ─────────────────────────────────────────────────────────── def twofactor_status(): """Check server-wide 2FA policy status.""" data = _whmapi1(['twofactorauth_policy_status']) if not data: return try: enabled = data['data']['is_enabled'] if enabled: good("2FA policy is ENABLED server-wide") else: bad("2FA policy is DISABLED server-wide") except KeyError: warn("Could not parse 2FA policy status.") def twofactor_enable(): """Enable 2FA policy server-wide.""" data = _whmapi1(['twofactorauth_enable_policy']) if data and data.get('metadata', {}).get('result') == 1: good("2FA policy enabled server-wide.") else: warn("Failed to enable 2FA policy.") def twofactor_disable(): """Disable 2FA policy server-wide. WARNING: disables 2FA for all users.""" warn("WARNING: This will disable 2FA for ALL users server-wide!") confirm = _ask("Type 'yes' to confirm: ") if confirm.lower() != 'yes': info("Cancelled.") return data = _whmapi1(['twofactorauth_disable_policy']) if data and data.get('metadata', {}).get('result') == 1: good("2FA policy disabled server-wide.") else: warn("Failed to disable 2FA policy.") def twofactor_users(): """List all users and their 2FA status.""" data = _whmapi1(['twofactorauth_get_user_configs']) if not data: return # Response: data = {username: {is_enabled: 0/1, primary_domain: ...}, ...} configs = data.get('data', {}) if not configs: info("No user 2FA data returned.") return header(f"{'User':<20} {'2FA':<8} Domain") print("-" * 55) for user, cfg in sorted(configs.items()): enabled = cfg.get('is_enabled', 0) domain = cfg.get('primary_domain', '') status = "Enabled" if enabled else "Disabled" print(f"{user:<20} {status:<8} {domain}") def twofactor_remove_user(username=None): """Remove a user's 2FA config. They must reconfigure from scratch.""" if not username: username = _ask("cPanel username: ") if not username: return warn(f"WARNING: This permanently removes {username}'s 2FA. They must re-enroll from scratch.") confirm = _ask("Type 'yes' to confirm: ") if confirm.lower() != 'yes': info("Cancelled.") return data = _whmapi1(['twofactorauth_remove_user_config', f'user={username}']) if data and data.get('metadata', {}).get('result') == 1: good(f"2FA config removed for {username}.") else: reason = (data or {}).get('metadata', {}).get('reason', 'Unknown error') warn(f"Failed: {reason}")