""" apache.py — Apache process and worker functions. """ import json import os import subprocess from modules.output import warn, info, good, bad, header from modules.utils import require_bin, run def _get_fullstatus(): """Run apachectl fullstatus and return text output, or None on failure.""" if not require_bin('apachectl', 'apache'): return None result = subprocess.run( ['apachectl', 'fullstatus'], stdout=subprocess.PIPE, stderr=subprocess.PIPE ) stdout = result.stdout.decode('utf-8', errors='replace') stderr = result.stderr.decode('utf-8', errors='replace') if 'Connection refused' in stdout or 'Connection refused' in stderr: bad("Apache is down (connection refused).") return None if result.returncode != 0 and not stdout.strip(): warn(f"apachectl fullstatus failed: {stderr.strip()}") return None return stdout def apacheprocs(): """ Show Apache status from apachectl fullstatus. Outputs the header block through the scoreboard key. """ text = _get_fullstatus() if text is None: return scoreboard_end = '"I" Idle cleanup of worker, "." Open slot with no current process' for line in text.splitlines(): print(line) if scoreboard_end in line: break def apache_stats(): """ Show Apache request table from apachectl fullstatus, sorted by VHost. Only shows complete request lines (with client/protocol/vhost/request). """ import re as _re text = _get_fullstatus() if text is None: return lines = text.splitlines() # Find the request table header: starts with 'Srv' and contains 'PID' and 'Acc' start_idx = None for i, line in enumerate(lines): if line.strip().startswith('Srv') and 'PID' in line and 'Acc' in line: start_idx = i break if start_idx is None: info("No request table found in apachectl fullstatus output.") return # Collect lines until the ════ separator sep_pattern = _re.compile(r'^[═=]{10,}') table_lines = [] legend_start = None for i in range(start_idx + 1, len(lines)): if sep_pattern.match(lines[i]): legend_start = i + 1 break table_lines.append(lines[i]) # Filter to complete request lines: start with \d+-\d+, have >=14 tokens, # and token[11] looks like an IP (contains a dot) req_pattern = _re.compile(r'^\d+-\d+\s') complete = [] for line in table_lines: if not req_pattern.match(line): continue parts = line.split() if len(parts) >= 14 and '.' in parts[11]: complete.append(line) if not complete: info("No active requests found.") return # Sort by VHost (index 13) complete.sort(key=lambda l: l.split()[13]) print(lines[start_idx]) for line in complete: print(line) # Print legend: from legend_start through 'Slot Total megabytes' if legend_start: slot_legend = 'Slot Total megabytes transferred this slot' for i in range(legend_start, len(lines)): print(lines[i]) if slot_legend in lines[i]: break def service_status(): """ Show detailed service uptime: Apache, MariaDB/MySQL, and DNS. """ _check_service_uptime('httpd', 'Apache (httpd)') _check_mysql_uptime() _check_dns_uptime() def _check_service_uptime(service, label): out, _, _ = run(['systemctl', 'status', f'{service}.service']) is_down = any(x in out for x in ('inactive', 'failed', 'dead')) if is_down: bad(f"{label} is down.") return uptime = '' for line in out.splitlines(): if 'Active:' in line and ';' in line: uptime = line.split(';')[-1].strip().rstrip(')') break good(f"{label} is up — {uptime}" if uptime else f"{label} is up.") def _check_mysql_uptime(): for svc in ('mariadb', 'mysqld'): out, _, _ = run(['systemctl', 'is-active', f'{svc}.service']) if out.strip() == 'active': label = 'MariaDB' if svc == 'mariadb' else 'MySQL' uptime_out, _, _ = run(['systemctl', 'status', f'{svc}.service']) uptime = '' for line in uptime_out.splitlines(): if 'Active:' in line and ';' in line: uptime = line.split(';')[-1].strip().rstrip(')') break good(f"{label} ({svc}) is up — {uptime}" if uptime else f"{label} ({svc}) is up.") adm_out, _, adm_rc = run(['mysqladmin', 'version']) if adm_rc == 0: for line in adm_out.splitlines(): line = line.strip() if line and not line.startswith('Copyright') and not line.startswith('mysqladmin'): print(f" {line}") return bad("MySQL/MariaDB is down.") def _check_dns_uptime(): config = '/var/cpanel/cpanel.config' dns_type = 'bind' if os.path.isfile(config): with open(config, 'r') as f: for line in f: if 'local_nameserver_type' in line: dns_type = line.split('=')[-1].strip() break if dns_type == 'powerdns': _check_service_uptime('pdns', 'POWERDNS (pdns)') else: _check_service_uptime('named', 'NAMED (named)') def domain_config(domain=None): """ Show domain configuration from whmapi1 domainuserdata. Usage: domain-config example.com """ if not domain: try: domain = input(" Domain: ").strip() except (EOFError, KeyboardInterrupt): return if not domain: warn("No domain provided.") return out, _, rc = run(['whmapi1', '--output=json', 'domainuserdata', f'domain={domain}']) if rc != 0: warn(f"Could not fetch domain data for: {domain}") return try: data = json.loads(out) if data.get('metadata', {}).get('result') != 1: reason = data.get('metadata', {}).get('reason', 'unknown error') warn(f"Domain not found on this server: {domain} ({reason})") return userdata = data['data']['userdata'] except (json.JSONDecodeError, KeyError): warn("Could not parse whmapi1 domainuserdata output.") return header(f"Domain config: {domain}") for key, val in userdata.items(): if isinstance(val, list): print(f" {key}:") for item in val: if isinstance(item, dict): for k, v in item.items(): print(f" {k}: {v}") else: print(f" {item}") else: print(f" {key}: {val}")