File manager

File manager - Edit - /usr/share/lve-stats3/scripts/lvestats-cpapi-helper

Back
#!/opt/cloudlinux/venv/bin/python3 """ lvestats-cpapi-helper — Bridge between lve-stats 3 (Rust) and clcommon.cpapi. Exposes cpapi operations as CLI subcommands returning JSON to stdout. Installed to /usr/share/lve-stats3/scripts/lvestats-cpapi-helper on CloudLinux systems. Usage: lvestats-cpapi-helper user-info <uid> [<uid>...] lvestats-cpapi-helper user-info-sys <uid> [<uid>...] lvestats-cpapi-helper resellers lvestats-cpapi-helper admins lvestats-cpapi-helper admin-email lvestats-cpapi-helper login-url <domain> lvestats-cpapi-helper reseller-users <reseller> lvestats-cpapi-helper domains <username> [<username>...] lvestats-cpapi-helper panel-feature <feature_name> lvestats-cpapi-helper db-access lvestats-cpapi-helper dblogin-cplogin-pairs """ from __future__ import print_function import json import os import pwd import sys # Subcommands that return panel-wide, cross-tenant data and are only ever # invoked by root callers (the lve-stats daemon collectors/savers and the root # StatsNotifier), so unprivileged local callers must not reach them at all: # * db-access / dblogin-cplogin-pairs expose the panel's MySQL login/password # and the DB->CP login mapping (original F-07 gate); # * resellers / admins / admin-email / login-url / domains / all-user-domains # enumerate reseller lists, admin usernames/emails, per-domain login URLs, # and username->domain maps. # panel-name / panel-feature carry no tenant data and stay unrestricted. # user-info / user-info-sys and reseller-users are NOT listed here: the # unprivileged tenant CLIs (cloudlinux_top, lvechart) legitimately invoke them # for the caller's own scope, so they are constrained to that scope below # instead of being blocked outright. ROOT_ONLY_COMMANDS = frozenset(( "resellers", "admins", "admin-email", "login-url", "domains", "all-user-domains", "db-access", "dblogin-cplogin-pairs", )) def _caller_username(): """Return the login name of the process's effective UID, or None if it cannot be resolved (e.g. the euid has no passwd entry).""" try: return pwd.getpwuid(os.geteuid()).pw_name except KeyError: return None def _import_cpapi(): """Import cpapi, returning (module, None) or (None, error_string).""" try: from clcommon import cpapi return cpapi, None except ImportError as e: return None, str(e) def _caller_is_reseller(caller): """Whether ``caller`` (a login resolved from the caller's euid, never from argv) is a reseller per cpapi.resellers(). Fails closed: an unresolved caller, a panel without reseller support, or any cpapi error yields False, so a non-reseller local user is refused. Mirrors the role gate in lvestats-gov-helper's _reseller_scope.""" if caller is None: return False cpapi, _import_err = _import_cpapi() if cpapi is None: return False resellers_fn = getattr(cpapi, "resellers", None) if resellers_fn is None: return False try: return caller in set(resellers_fn()) except Exception: return False def cmd_user_info(uids, search_sys_users=False, restrict_uid=None): """Resolve UIDs to panel user information. When search_sys_users=True, cpinfo queries the sys_users table (matching Python lve-stats behaviour) which works on Plesk where regular hosting users live in sys_users, not in the clients table. When restrict_uid is not None (a non-root caller), only that UID's record is resolved; every other requested UID is reported as unauthorized rather than looked up, so an unprivileged caller cannot read other tenants' email /domain/reseller metadata. The unprivileged CLI path only ever displays the caller's own row, so this does not change its behaviour. """ cpapi, import_err = _import_cpapi() users = [] unresolved = [] for uid in uids: if restrict_uid is not None and uid != restrict_uid: unresolved.append({ "uid": uid, "reason": "not authorized: caller may only query its own uid", }) continue # Step 1: UID -> username via passwd try: pw = pwd.getpwuid(uid) username = pw.pw_name # GECOS field: "Full Name,room,work_phone,home_phone" display_name = pw.pw_gecos.split(",")[0] if pw.pw_gecos else username except KeyError: unresolved.append({"uid": uid, "reason": "no such user (uid not in passwd)"}) continue if cpapi is None: unresolved.append({ "uid": uid, "reason": "cpapi not available: %s" % import_err, }) continue # Step 2: username -> panel info via cpapi try: cp_info = cpapi.cpinfo( username, keyls=("mail", "dns", "locale", "reseller"), search_sys_users=search_sys_users, ) if not cp_info: unresolved.append({ "uid": uid, "reason": "cpinfo returned empty for user '%s'" % username, }) continue info = cp_info[0] email = info[0] or "" domain = info[1] or "" locale = info[2] or "en" reseller = info[3] or "" users.append({ "uid": uid, "username": username, "email": email, "domain": domain, "locale": locale, "reseller": reseller, "display_name": display_name, }) except (IndexError, TypeError): unresolved.append({ "uid": uid, "reason": "cpinfo returned unexpected format for user '%s'" % username, }) except Exception as e: exc_name = type(e).__name__ unresolved.append({ "uid": uid, "reason": "cpapi lookup failed: %s: %s" % (exc_name, e), }) return {"users": users, "unresolved": unresolved} def cmd_resellers(): """List all reseller usernames.""" cpapi, import_err = _import_cpapi() if cpapi is None: return [] try: return list(cpapi.resellers()) except Exception: return [] def cmd_admins(): """List all admin usernames.""" cpapi, import_err = _import_cpapi() if cpapi is None: return [] try: return list(cpapi.admins()) except Exception: return [] def cmd_admin_email(): """Get admin contact email.""" cpapi, import_err = _import_cpapi() if cpapi is None: return None try: email = cpapi.get_admin_email() return email if email else None except Exception: return None def cmd_login_url(domain): """Get panel login URL for a domain.""" cpapi, import_err = _import_cpapi() if cpapi is None: return None try: return cpapi.get_user_login_url(domain) except Exception: return None def cmd_reseller_users(reseller): """List usernames belonging to a reseller via cpapi.reseller_users().""" cpapi, import_err = _import_cpapi() if cpapi is None: return [] reseller_users_fn = getattr(cpapi, 'reseller_users', None) if reseller_users_fn is None: return [] try: return list(reseller_users_fn(reseller)) except Exception: return [] def cmd_panel_feature(feature_name): """Check if a panel feature is supported. Returns true/false.""" cpapi, import_err = _import_cpapi() if cpapi is None: return False try: from clcommon.features import Feature feature = Feature[feature_name] return cpapi.is_panel_feature_supported(feature) except (KeyError, Exception): return False def cmd_db_access(): """Get MySQL access credentials from the control panel. Returns dict with status and credentials: - {"status": "ok", "login": "...", "pass": "...", "host": "..."} - {"status": "not_supported", "error": "..."} - {"status": "no_access_data", "error": "..."} """ cpapi, import_err = _import_cpapi() if cpapi is None: return {"status": "not_supported", "error": "cpapi not available: %s" % import_err} try: from clcommon.cpapi.cpapiexceptions import NoDBAccessData from clcommon.cpapi import NotSupported except ImportError: NoDBAccessData = None NotSupported = None try: access = cpapi.db_access() return { "status": "ok", "login": access.get("login", "root"), "pass": access.get("pass", ""), "host": access.get("host", "localhost"), } except Exception as e: exc_name = type(e).__name__ if NotSupported is not None and isinstance(e, NotSupported): return {"status": "not_supported", "error": "%s: %s" % (exc_name, e)} if NoDBAccessData is not None and isinstance(e, NoDBAccessData): return {"status": "no_access_data", "error": "%s: %s" % (exc_name, e)} return {"status": "not_supported", "error": "%s: %s" % (exc_name, e)} def cmd_dblogin_cplogin_pairs(): """Get mapping of database login names to control panel login names. Returns list of [db_login, cp_login] pairs. """ cpapi, import_err = _import_cpapi() if cpapi is None: return [] try: return list(cpapi.dblogin_cplogin_pairs()) except Exception: return [] def cmd_panel_name(): """Get the control panel name (e.g. 'cPanel', 'Plesk', 'DirectAdmin').""" cpapi, import_err = _import_cpapi() if cpapi is None: return "unknown" return getattr(cpapi, "CP_NAME", "unknown") or "unknown" def cmd_domains(usernames): """Get primary domain for each username via cpapi.userdomains(). Returns dict mapping username -> domain_or_null. """ cpapi, import_err = _import_cpapi() results = {} if cpapi is None: for u in usernames: results[u] = None return results for username in usernames: try: user_domains = cpapi.userdomains(username) if user_domains: results[username] = user_domains[0][0] else: results[username] = None except Exception: results[username] = None return results def cmd_all_user_domains(usernames): """Get ALL domains (main + sub + aliases) for each username. Returns dict mapping username -> list of unique domain strings. Matches Python lve-stats get_all_user_domains(). """ cpapi, import_err = _import_cpapi() results = {} if cpapi is None: for u in usernames: results[u] = [] return results for username in usernames: domains = [] aliases = [] try: user_domains = cpapi.userdomains(username) if user_domains: domains = [d[0] for d in user_domains] except Exception: pass for domain in domains: try: user_aliases = cpapi.useraliases(username, domain) if user_aliases: aliases += user_aliases except Exception: pass results[username] = list(set(domains + aliases)) return results def main(): if len(sys.argv) < 2: print("Usage: lvestats-cpapi-helper <command> [args...]", file=sys.stderr) print("Commands: user-info, user-info-sys, resellers, admins, admin-email, login-url, reseller-users, domains, all-user-domains, panel-name, db-access, dblogin-cplogin-pairs", file=sys.stderr) sys.exit(1) command = sys.argv[1] is_root = os.geteuid() == 0 # Panel-wide metadata / credential subcommands are only ever invoked by root # callers (daemon collectors/savers, root StatsNotifier); refuse them for # unprivileged local callers so they cannot enumerate other tenants' data. if command in ROOT_ONLY_COMMANDS and not is_root: print("Permission denied: '%s' requires root privileges" % command, file=sys.stderr) sys.exit(1) # reseller-users IS reached by non-root tenants (cloudlinux_top/lvechart via # --for-reseller), but only ever for the caller's own reseller account. # Restrict a non-root caller to its OWN login AND require it to actually be # a reseller (cpapi.resellers()); fail closed to permission-denied otherwise # so a non-reseller local user cannot reach cpapi.reseller_users(). if command == "reseller-users" and not is_root: caller = _caller_username() if len(sys.argv) < 3 or sys.argv[2] != caller or not _caller_is_reseller(caller): print("Permission denied: 'reseller-users' may only query the caller's own reseller", file=sys.stderr) sys.exit(1) try: if command in ("user-info", "user-info-sys"): if len(sys.argv) < 3: print("Usage: lvestats-cpapi-helper %s <uid> [<uid>...]" % command, file=sys.stderr) sys.exit(1) uids = [] for arg in sys.argv[2:]: try: uids.append(int(arg)) except ValueError: print("Invalid UID: %s" % arg, file=sys.stderr) sys.exit(1) # A non-root caller may only resolve its own UID's metadata; other # UIDs are reported unauthorized (the tenant CLI path shows only the # caller's own row, so this preserves its behaviour). restrict_uid = None if is_root else os.geteuid() result = cmd_user_info( uids, search_sys_users=(command == "user-info-sys"), restrict_uid=restrict_uid, ) elif command == "resellers": result = cmd_resellers() elif command == "admins": result = cmd_admins() elif command == "admin-email": result = cmd_admin_email() elif command == "login-url": if len(sys.argv) < 3: print("Usage: lvestats-cpapi-helper login-url <domain>", file=sys.stderr) sys.exit(1) result = cmd_login_url(sys.argv[2]) elif command == "reseller-users": if len(sys.argv) < 3: print("Usage: lvestats-cpapi-helper reseller-users <reseller>", file=sys.stderr) sys.exit(1) result = cmd_reseller_users(sys.argv[2]) elif command == "panel-feature": if len(sys.argv) < 3: print("Usage: lvestats-cpapi-helper panel-feature <feature_name>", file=sys.stderr) sys.exit(1) result = cmd_panel_feature(sys.argv[2]) elif command == "domains": if len(sys.argv) < 3: print("Usage: lvestats-cpapi-helper domains <username> [<username>...]", file=sys.stderr) sys.exit(1) result = cmd_domains(sys.argv[2:]) elif command == "all-user-domains": if len(sys.argv) < 3: print("Usage: lvestats-cpapi-helper all-user-domains <username> [<username>...]", file=sys.stderr) sys.exit(1) result = cmd_all_user_domains(sys.argv[2:]) elif command == "panel-name": result = cmd_panel_name() elif command == "db-access": result = cmd_db_access() elif command == "dblogin-cplogin-pairs": result = cmd_dblogin_cplogin_pairs() else: print("Unknown command: %s" % command, file=sys.stderr) sys.exit(1) json.dump(result, sys.stdout) sys.stdout.write("\n") except Exception as e: print("lvestats-cpapi-helper: unexpected error: %s: %s" % (type(e).__name__, e), file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()

| ver. 1.4 | Github | . | PHP 8.1.34 | Generation time: 0.04 | proxy | phpinfo | Settings