Gecko [ aodai.devttt.com ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
.my.cnf.4370	0 B	-rw-r--r--
.mysql.4370	0 B	-rw-r--r--
Backupsv2.py	59.03 KB	-rw-r--r--
ClusterManager.py	17.68 KB	-rw-r--r--
CyberCPLogFileWriter.py	4.37 KB	-rw-r--r--
CyberPanelUpgrade.py	3.64 KB	-rw-r--r--
DockerSites.py	61.86 KB	-rw-r--r--
IncScheduler.py	88.07 KB	-rw-r--r--
SwitchOldAliasToNew.py	2.4 KB	-rw-r--r--
__init__.py	0 B	-rw-r--r--
acl.py	50.61 KB	-rw-r--r--
acl.py.backup_dns_fix	48.95 KB	-rw-r--r--
adminPass.py	2.41 KB	-rw-------
alias.py	647 B	-rw-r--r--
apiAccess.py	418 B	-rw-r--r--
applicationInstaller.py	352.05 KB	-rw-r--r--
backupSchedule.py	22.21 KB	-rw-r--r--
backupScheduleLocal.py	5.31 KB	-rw-r--r--
backupUtilities.py	110.15 KB	-rw-r--r--
bandwidthReset.py	7.25 KB	-rw-r--r--
cPanelImporter.py	41.33 KB	-rw-r--r--
childDomain.py	1.25 KB	-rw-r--r--
cronUtil.py	5.34 KB	-rw-r--r--
csf.py	48.63 KB	-rw-r--r--
customACME.py	51.55 KB	-rw-r--r--
dnsUtilities.py	38.7 KB	-rw-r--r--
domain.xml	0 B	-rw-r--r--
emergency_2fa_disable.py	1.24 KB	-rw-r--r--
filesPermsUtilities.py	6.17 KB	-rw-r--r--
filesPermsUtilities.py.bak	6.17 KB	-rw-r--r--
findBWUsage.py	4.6 KB	-rw-r--r--
firewallUtilities.py	8.18 KB	-rw-r--r--
ftpUtilities.py	7.56 KB	-rw-r--r--
getSystemInformation.py	5.29 KB	-rw-r--r--
hashPassword.py	1.22 KB	-rw-r--r--
httpProc.py	4.81 KB	-rw-r--r--
installUtilities.py	19.04 KB	-rw-r--r--
letsEncrypt.py	0 B	-rw-r--r--
mailUtilities.py	104.79 KB	-rw-r--r--
modSec.py	26.71 KB	-rw-r--r--
mysqlUtilities.py	41.46 KB	-rw-r--r--
phpUtilities.py	19.8 KB	-rw-r--r--
phpmyadminsignin.php	2.08 KB	-rw-r--r--
pluginManagerGlobal.py	813 B	-rw-r--r--
processUtilities.py	22.42 KB	-rw-r--r--
randomPassword.py	200 B	-rw-r--r--
rebuildQuotas.py	3.29 KB	-rw-r--r--
remoteBackup.py	14.81 KB	-rw-r--r--
remoteTransferUtilities.py	16.43 KB	-rw-r--r--
renew.py	5.64 KB	-rwxr-xr-x
restoreMeta.py	7.65 KB	-rw-r--r--
serverLogs.py	735 B	-rw-r--r--
sslUtilities.py	49.26 KB	-rw-r--r--
sslv2.py	26.18 KB	-rw-r--r--
test.py	0 B	-rw-r--r--
test1.py	0 B	-rw-r--r--
tuning.py	18.08 KB	-rw-r--r--
upgrade.py	213.62 KB	-rw-r--r--
upgrade.py.bak	97.03 KB	-rw-r--r--
upgradeCritical.py	1.15 KB	-rwx------
vhost.py	47 KB	-rw-r--r--
vhostConfs.py	14.41 KB	-rw-r--r--
virtualHostUtilities.py	92.31 KB	-rw-r--r--

Code Editor : sslUtilities.py

import requests

from plogical import CyberCPLogFileWriter as logging
import os
import shlex
import subprocess
import socket
from plogical.processUtilities import ProcessUtilities

try:
    from websiteFunctions.models import ChildDomains, Websites
except:
    pass
from plogical.acl import ACLManager

class sslUtilities:
    Server_root = "/usr/local/lsws"
    redisConf = '/usr/local/lsws/conf/dvhost_redis.conf'

@staticmethod
    def parseACMEError(error_output):
        """Parse ACME error output to extract meaningful error messages"""
        if not error_output:
            return "Unknown error occurred"

error_output = str(error_output)

# Common ACME/Let's Encrypt errors
        error_patterns = {
            r"rateLimited": "Rate limit exceeded. Too many certificates issued for this domain. Please wait before retrying.",
            r"urn:ietf:params:acme:error:rateLimited": "Rate limit exceeded. Please wait before retrying.",
            r"too many certificates": "Rate limit: Too many certificates issued recently.",
            r"DNS problem: NXDOMAIN": "DNS Error: Domain does not exist or DNS not propagated.",
            r"DNS problem": "DNS validation failed. Ensure domain points to this server.",
            r"Connection refused": "Cannot connect to ACME server. Check firewall/network settings.",
            r"Connection timeout": "Connection to ACME server timed out. Check network connectivity.",
            r"Timeout during connect": "Connection timeout. The ACME server may be unreachable.",
            r"unauthorized": "Authorization failed. Domain validation unsuccessful.",
            r"urn:ietf:params:acme:error:unauthorized": "Domain authorization failed. Verify domain ownership.",
            r"Invalid response from": "Invalid response from domain during validation.",
            r"404": "Challenge file not found. Check web server configuration.",
            r"403": "Access forbidden. Check file permissions and .htaccess rules.",
            r"CAA record": "CAA record prevents certificate issuance. Update DNS CAA records.",
            r"urn:ietf:params:acme:error:caa": "CAA record forbids issuance. Check DNS CAA settings.",
            r"Challenge failed": "ACME challenge failed. Ensure port 80 is accessible.",
            r"No valid IP addresses": "No valid IP addresses found for domain.",
            r"Could not connect to": "Cannot connect to domain for validation.",
            r"conflictingRequest": "A conflicting request exists. Previous request may still be processing.",
            r"urn:ietf:params:acme:error:malformed": "Malformed request. Check domain format.",
            r"urn:ietf:params:acme:error:serverInternal": "ACME server internal error. Try again later.",
            r"urn:ietf:params:acme:error:orderNotReady": "Order not ready. Domain validation incomplete.",
            r"badNonce": "Bad nonce error. This is usually temporary, please retry.",
            r"JWS has an invalid anti-replay nonce": "Invalid nonce. Please retry the request.",
            r"Account registration error": "Account registration failed. Check email address.",
            r"Error creating new account": "Cannot create ACME account. Check email validity.",
            r"Verify error": "Certificate verification failed.",
            r"Fetching http://": "HTTP validation failed. Ensure port 80 is open.",
            r"Fetching https://": "HTTPS validation issue detected.",
            r"Invalid email address": "Invalid email address provided for registration.",
            r"blacklisted": "Domain is blacklisted by the certificate authority.",
            r"PolicyForbids": "Certificate authority policy forbids issuance for this domain."
        }

# Check each pattern
        import re
        for pattern, message in error_patterns.items():
            if re.search(pattern, error_output, re.IGNORECASE):
                # Try to extract additional context
                lines = error_output.split('\n')
                for line in lines:
                    if 'Detail:' in line:
                        message += f" Detail: {line.split('Detail:')[1].strip()}"
                        break
                return message

# Try to extract specific error details from acme.sh output
        if "[" in error_output and "]" in error_output:
            # Extract content between brackets which often contains the error
            import re
            bracket_content = re.findall(r'\[([^\]]+)\]', error_output)
            if bracket_content:
                # Get the last bracketed content as it's usually the error
                potential_error = bracket_content[-1]
                if len(potential_error) > 10:  # Make sure it's meaningful
                    return f"SSL issuance failed: {potential_error}"

# Look for lines starting with "Error:" or containing "error:"
        lines = error_output.split('\n')
        for line in lines:
            if line.strip().startswith('Error:') or 'error:' in line.lower():
                return line.strip()

# If we can't parse a specific error, return a portion of the output
        if len(error_output) > 200:
            # Get the last 200 characters which likely contain the error
            return f"SSL issuance failed: ...{error_output[-200:]}"

return f"SSL issuance failed: {error_output}"

@staticmethod
    def checkDNSRecords(domain):
        """Check if domain has valid DNS records using external DNS query"""
        try:
            # Use dig command to check DNS records from authoritative servers
            command = f"dig +short {domain} A @8.8.8.8"
            try:
                result = subprocess.run(command, shell=True, capture_output=True, text=True)
            except TypeError:
                # Fallback for Python < 3.7
                result = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                        universal_newlines=True)

# If there's any output, the domain has A records
            if result.stdout.strip():
                return True

# Also check AAAA records
            command = f"dig +short {domain} AAAA @8.8.8.8"
            try:
                result = subprocess.run(command, shell=True, capture_output=True, text=True)
            except TypeError:
                # Fallback for Python < 3.7
                result = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                        universal_newlines=True)

if result.stdout.strip():
                return True

return False
        except:
            # Fallback to socket method if dig fails
            try:
                socket.gethostbyname(domain)
                return True
            except:
                return False

DONT_ISSUE = 0
    ISSUE_SELFSIGNED = 1
    ISSUE_SSL = 2

@staticmethod
    def getDomainsCovered(cert_path):
        try:
            from cryptography import x509
            from cryptography.hazmat.backends import default_backend
            with open(cert_path, 'rb') as cert_file:
                cert_data = cert_file.read()
                cert = x509.load_pem_x509_certificate(cert_data, default_backend())

# Check for the Subject Alternative Name (SAN) extension
                san_extension = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)

if san_extension:
                    # Extract and print the domains from SAN
                    san_domains = san_extension.value.get_values_for_type(x509.DNSName)
                    try:
                        logging.CyberCPLogFileWriter.writeToFile(f'Covered domains: {str(san_domains)}')
                    except:
                        pass
                    return 1, san_domains
                else:
                    # If SAN is not present, return the Common Name as a fallback
                    return 0, None
        except BaseException as msg:
            return 0, str(msg)

@staticmethod
    def CheckIfSSLNeedsToBeIssued(virtualHostName):
        #### if website already have an SSL, better not issue again - need to check for wild-card
        filePath = '/etc/letsencrypt/live/%s/fullchain.pem' % (virtualHostName)
        if os.path.exists(filePath):
            import OpenSSL
            x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, open(filePath, 'r').read())
            SSLProvider = x509.get_issuer().get_components()[1][1].decode('utf-8')

if os.path.exists(ProcessUtilities.debugPath):
                logging.CyberCPLogFileWriter.writeToFile(f'SSL provider for {virtualHostName} is {SSLProvider}.')

#### totally seprate check to see if both non-www and www are covered

if SSLProvider == "(STAGING) Let's Encrypt":
                return sslUtilities.ISSUE_SSL

if SSLProvider == "Let's Encrypt":
                status, domains = sslUtilities.getDomainsCovered(filePath)
                if status:
                    if len(domains) > 1:
                        ### need further checks here to see if ssl is valid for less then 15 days etc
                        logging.CyberCPLogFileWriter.writeToFile(
                            '[CheckIfSSLNeedsToBeIssued] SSL exists for %s and both versions are covered, just need to ensure if SSL is valid for less then 15 days.' % (
                                virtualHostName), 0)
                        pass
                    else:
                        return sslUtilities.ISSUE_SSL

#####

expireData = x509.get_notAfter().decode('ascii')
            from datetime import datetime
            finalDate = datetime.strptime(expireData, '%Y%m%d%H%M%SZ')
            now = datetime.now()
            diff = finalDate - now

if int(diff.days) >= 15 and SSLProvider != 'Denial':
                logging.CyberCPLogFileWriter.writeToFile(
                    '[CheckIfSSLNeedsToBeIssued] SSL exists for %s and is not ready to fetch new SSL., skipping..' % (
                        virtualHostName), 0)

return sslUtilities.DONT_ISSUE
            elif SSLProvider == 'Denial':
                logging.CyberCPLogFileWriter.writeToFile(
                    f'[CheckIfSSLNeedsToBeIssued] Self-signed SSL found, lets issue new SSL for {virtualHostName}', 0)
                return sslUtilities.ISSUE_SSL
            elif SSLProvider != "Let's Encrypt":
                logging.CyberCPLogFileWriter.writeToFile(
                    f'[CheckIfSSLNeedsToBeIssued] Custom SSL found for {virtualHostName}', 0)
                return sslUtilities.DONT_ISSUE
            else:
                logging.CyberCPLogFileWriter.writeToFile(
                    f'[CheckIfSSLNeedsToBeIssued] We will issue SSL for {virtualHostName}', 0)
                return sslUtilities.ISSUE_SSL
        else:
            logging.CyberCPLogFileWriter.writeToFile(
                f'[CheckIfSSLNeedsToBeIssued] We will issue SSL for {virtualHostName}', 0)
            return sslUtilities.ISSUE_SSL

@staticmethod
    def checkIfSSLMap(virtualHostName):
        try:
            data = open("/usr/local/lsws/conf/httpd_config.conf").readlines()

sslCheck = 0

for items in data:
                if items.find("listener") > - 1 and items.find("SSL") > -1:
                    sslCheck = 1
                    continue
                if sslCheck == 1:
                    if items.find("}") > -1:
                        return 0
                if items.find(virtualHostName) > -1 and sslCheck == 1:
                    data = [_f for _f in items.split(" ") if _f]
                    if data[1] == virtualHostName:
                        return 1

except BaseException as msg:
            logging.CyberCPLogFileWriter.writeToFile(str(msg) + " [IO Error with main config file [checkIfSSLMap]]")
            return 0

@staticmethod
    def checkSSLListener():
        try:
            data = open("/usr/local/lsws/conf/httpd_config.conf").readlines()
            for items in data:
                if items.find("listener SSL") > -1:
                    return 1

except BaseException as msg:
            logging.CyberCPLogFileWriter.writeToFile(str(msg) + " [IO Error with main config file [checkSSLListener]]")
            return str(msg)
        return 0

@staticmethod
    def checkSSLIPv6Listener():
        try:
            data = open("/usr/local/lsws/conf/httpd_config.conf").readlines()
            for items in data:
                if items.find("listener SSL IPv6") > -1:
                    return 1

except BaseException as msg:
            logging.CyberCPLogFileWriter.writeToFile(
                str(msg) + " [IO Error with main config file [checkSSLIPv6Listener]]")
            return str(msg)
        return 0

@staticmethod
    def getDNSRecords(virtualHostName):
        try:

withoutWWW = socket.gethostbyname(virtualHostName)
            withWWW = socket.gethostbyname('www.' + virtualHostName)

return [1, withWWW, withoutWWW]

except BaseException as msg:
            return [0, "347 " + str(msg) + " [issueSSLForDomain]"]

@staticmethod
    def PatchVhostConf(virtualHostName):
        """Patch the virtual host configuration to add ACME challenge support

This function adds the necessary configuration to handle ACME challenges
        for both OpenLiteSpeed (OLS) and Apache configurations. It also checks
        for potential configuration conflicts before making changes.

Args:
            virtualHostName (str): The domain name to configure

Returns:
            tuple: (status, message) where status is 1 for success, 0 for failure
        """
        try:
            # Construct paths
            confPath = os.path.join(sslUtilities.Server_root, "conf", "vhosts", virtualHostName)
            completePathToConfigFile = os.path.join(confPath, "vhost.conf")

# Check if file exists
            if not os.path.exists(completePathToConfigFile):
                logging.CyberCPLogFileWriter.writeToFile(f'Configuration file not found: {completePathToConfigFile}')
                return 0, f'Configuration file not found: {completePathToConfigFile}'

# Read current configuration
            try:
                with open(completePathToConfigFile, 'r') as f:
                    DataVhost = f.read()
            except IOError as e:
                logging.CyberCPLogFileWriter.writeToFile(f'Error reading configuration file: {str(e)}')
                return 0, f'Error reading configuration file: {str(e)}'

# Check for potential conflicts
            conflicts = []

# Check if ACME challenge is already configured
            if DataVhost.find('/.well-known/acme-challenge') != -1:
                logging.CyberCPLogFileWriter.writeToFile(f'ACME challenge already configured for {virtualHostName}')
                return 1, 'ACME challenge already configured'

# Check for conflicting rewrite rules
            if DataVhost.find('rewrite') != -1 and DataVhost.find('enable 1') != -1:
                conflicts.append('Active rewrite rules found that might interfere with ACME challenges')

# Check for conflicting location blocks
            if DataVhost.find('location /.well-known') != -1:
                conflicts.append('Existing location block for /.well-known found')

# Check for conflicting aliases
            if DataVhost.find('Alias /.well-known') != -1:
                conflicts.append('Existing alias for /.well-known found')

# Check for conflicting context blocks
            if DataVhost.find('context /.well-known') != -1:
                conflicts.append('Existing context block for /.well-known found')

# Check for conflicting access controls
            if DataVhost.find('deny from all') != -1 and DataVhost.find('location') != -1:
                conflicts.append('Global deny rules found that might block ACME challenges')

# If conflicts found, log them and return
            if conflicts:
                conflict_message = 'Configuration conflicts found: ' + '; '.join(conflicts)
                logging.CyberCPLogFileWriter.writeToFile(
                    f'Configuration conflicts for {virtualHostName}: {conflict_message}')
                return 0, conflict_message

# Create challenge directory if it doesn't exist
            challenge_dir = '/usr/local/lsws/Example/html/.well-known/acme-challenge'
            try:
                os.makedirs(challenge_dir, exist_ok=True)
                # Set proper permissions
                os.chmod(challenge_dir, 0o755)
            except OSError as e:
                logging.CyberCPLogFileWriter.writeToFile(f'Error creating challenge directory: {str(e)}')
                return 0, f'Error creating challenge directory: {str(e)}'

# Handle configuration based on server type
            if ProcessUtilities.decideServer() == ProcessUtilities.OLS:
                # OpenLiteSpeed configuration
                try:
                    with open(completePathToConfigFile, 'a') as f:
                        content = '''
context /.well-known/acme-challenge {
  location                /usr/local/lsws/Example/html/.well-known/acme-challenge
  allowBrowse             1
  rewrite  {
     enable                  0
  }
  addDefaultCharset       off
  phpIniOverride  {
  }
}
'''
                        f.write(content)
                except IOError as e:
                    logging.CyberCPLogFileWriter.writeToFile(f'Error writing OLS configuration: {str(e)}')
                    return 0, f'Error writing OLS configuration: {str(e)}'
            else:
                # Apache configuration
                try:
                    # Read current configuration
                    with open(completePathToConfigFile, 'r') as f:
                        lines = f.readlines()

# Write new configuration
                    with open(completePathToConfigFile, 'w') as f:
                        check = 0
                        for line in lines:
                            f.write(line)
                            if line.find('DocumentRoot /home/') > -1 and check == 0:
                                f.write(
                                    '    Alias /.well-known/acme-challenge /usr/local/lsws/Example/html/.well-known/acme-challenge\n')
                                check = 1
                except IOError as e:
                    logging.CyberCPLogFileWriter.writeToFile(f'Error writing Apache configuration: {str(e)}')
                    return 0, f'Error writing Apache configuration: {str(e)}'

# Restart LiteSpeed
            try:
                from plogical import installUtilities
                installUtilities.installUtilities.reStartLiteSpeed()
                logging.CyberCPLogFileWriter.writeToFile(
                    f'Successfully configured ACME challenge for {virtualHostName}')
                return 1, 'Successfully configured ACME challenge'
            except Exception as e:
                logging.CyberCPLogFileWriter.writeToFile(f'Error restarting LiteSpeed: {str(e)}')
                return 0, f'Error restarting LiteSpeed: {str(e)}'

except Exception as e:
            logging.CyberCPLogFileWriter.writeToFile(f'Unexpected error in PatchVhostConf: {str(e)}')
            return 0, f'Unexpected error: {str(e)}'

@staticmethod
    def installSSLForDomain(virtualHostName, adminEmail='domain@cyberpanel.net'):

try:
            website = Websites.objects.get(domain=virtualHostName)
            adminEmail = website.adminEmail
        except BaseException as msg:
            logging.CyberCPLogFileWriter.writeToFile('%s [installSSLForDomain:72]' % (str(msg)))

if ProcessUtilities.decideServer() == ProcessUtilities.OLS:
            confPath = sslUtilities.Server_root + "/conf/vhosts/" + virtualHostName
            completePathToConfigFile = confPath + "/vhost.conf"

try:
                map = "  map                     " + virtualHostName + " " + virtualHostName + "\n"

if sslUtilities.checkSSLListener() != 1:

writeDataToFile = open("/usr/local/lsws/conf/httpd_config.conf", 'a')

listener = "listener SSL {" + "\n"
                    address = "  address                 *:443" + "\n"
                    secure = "  secure                  1" + "\n"
                    keyFile = "  keyFile                  /etc/letsencrypt/live/" + virtualHostName + "/privkey.pem\n"
                    certFile = "  certFile                 /etc/letsencrypt/live/" + virtualHostName + "/fullchain.pem\n"
                    certChain = "  certChain               1" + "\n"
                    sslProtocol = "  sslProtocol             24" + "\n"
                    enableECDHE = "  enableECDHE             1" + "\n"
                    renegProtection = "  renegProtection         1" + "\n"
                    sslSessionCache = "  sslSessionCache         1" + "\n"
                    enableSpdy = "  enableSpdy              15" + "\n"
                    enableStapling = "  enableStapling           1" + "\n"
                    ocspRespMaxAge = "  ocspRespMaxAge           86400" + "\n"
                    map = "  map                     " + virtualHostName + " " + virtualHostName + "\n"
                    final = "}" + "\n" + "\n"

writeDataToFile.writelines("\n")
                    writeDataToFile.writelines(listener)
                    writeDataToFile.writelines(address)
                    writeDataToFile.writelines(secure)
                    writeDataToFile.writelines(keyFile)
                    writeDataToFile.writelines(certFile)
                    writeDataToFile.writelines(certChain)
                    writeDataToFile.writelines(sslProtocol)
                    writeDataToFile.writelines(enableECDHE)
                    writeDataToFile.writelines(renegProtection)
                    writeDataToFile.writelines(sslSessionCache)
                    writeDataToFile.writelines(enableSpdy)
                    writeDataToFile.writelines(enableStapling)
                    writeDataToFile.writelines(ocspRespMaxAge)
                    writeDataToFile.writelines(map)
                    writeDataToFile.writelines(final)
                    writeDataToFile.writelines("\n")
                    writeDataToFile.close()

elif sslUtilities.checkSSLIPv6Listener() != 1:

writeDataToFile = open("/usr/local/lsws/conf/httpd_config.conf", 'a')

listener = "listener SSL IPv6 {" + "\n"
                    address = "  address                 [ANY]:443" + "\n"
                    secure = "  secure                  1" + "\n"
                    keyFile = "  keyFile                  /etc/letsencrypt/live/" + virtualHostName + "/privkey.pem\n"
                    certFile = "  certFile                 /etc/letsencrypt/live/" + virtualHostName + "/fullchain.pem\n"
                    certChain = "  certChain               1" + "\n"
                    sslProtocol = "  sslProtocol             24" + "\n"
                    enableECDHE = "  enableECDHE             1" + "\n"
                    renegProtection = "  renegProtection         1" + "\n"
                    sslSessionCache = "  sslSessionCache         1" + "\n"
                    enableSpdy = "  enableSpdy              15" + "\n"
                    enableStapling = "  enableStapling           1" + "\n"
                    ocspRespMaxAge = "  ocspRespMaxAge           86400" + "\n"
                    map = "  map                     " + virtualHostName + " " + virtualHostName + "\n"
                    final = "}" + "\n" + "\n"

else:

if sslUtilities.checkIfSSLMap(virtualHostName) == 0:

data = open("/usr/local/lsws/conf/httpd_config.conf").readlines()
                        writeDataToFile = open("/usr/local/lsws/conf/httpd_config.conf", 'w')
                        sslCheck = 0

for items in data:
                            if items.find("listener") > -1 and items.find("SSL") > -1:
                                sslCheck = 1

if (sslCheck == 1):
                                writeDataToFile.writelines(items)
                                writeDataToFile.writelines(map)
                                sslCheck = 0
                            else:
                                writeDataToFile.writelines(items)
                        writeDataToFile.close()

###################### Write per host Configs for SSL ###################

data = open(completePathToConfigFile, "r").readlines()

## check if vhssl is already in vhconf file

vhsslPresense = 0

for items in data:
                        if items.find("vhssl") > -1:
                            vhsslPresense = 1

if vhsslPresense == 0:
                        writeSSLConfig = open(completePathToConfigFile, "a")

vhssl = "vhssl  {" + "\n"
                        keyFile = "  keyFile                 /etc/letsencrypt/live/" + virtualHostName + "/privkey.pem\n"
                        certFile = "  certFile                /etc/letsencrypt/live/" + virtualHostName + "/fullchain.pem\n"
                        certChain = "  certChain               1" + "\n"
                        sslProtocol = "  sslProtocol             24" + "\n"
                        enableECDHE = "  enableECDHE             1" + "\n"
                        renegProtection = "  renegProtection         1" + "\n"
                        sslSessionCache = "  sslSessionCache         1" + "\n"
                        enableSpdy = "  enableSpdy              15" + "\n"
                        enableStapling = "  enableStapling           1" + "\n"
                        ocspRespMaxAge = "  ocspRespMaxAge           86400" + "\n"
                        final = "}"

writeSSLConfig.writelines("\n")

writeSSLConfig.writelines(vhssl)
                        writeSSLConfig.writelines(keyFile)
                        writeSSLConfig.writelines(certFile)
                        writeSSLConfig.writelines(certChain)
                        writeSSLConfig.writelines(sslProtocol)
                        writeSSLConfig.writelines(enableECDHE)
                        writeSSLConfig.writelines(renegProtection)
                        writeSSLConfig.writelines(sslSessionCache)
                        writeSSLConfig.writelines(enableSpdy)
                        writeSSLConfig.writelines(enableStapling)
                        writeSSLConfig.writelines(ocspRespMaxAge)
                        writeSSLConfig.writelines(final)

writeSSLConfig.writelines("\n")

writeSSLConfig.close()

return 1
            except BaseException as msg:
                logging.CyberCPLogFileWriter.writeToFile(str(msg) + " [installSSLForDomain]]")
                return 0
        else:
            if not os.path.exists(sslUtilities.redisConf):
                confPath = sslUtilities.Server_root + "/conf/vhosts/" + virtualHostName
                completePathToConfigFile = confPath + "/vhost.conf"

## Check if SSL VirtualHost already exists

data = open(completePathToConfigFile, 'r').readlines()

for items in data:
                    if items.find('*:443') > -1:
                        return 1

try:

try:
                        chilDomain = ChildDomains.objects.get(domain=virtualHostName)
                        externalApp = chilDomain.master.externalApp
                        DocumentRoot = '    DocumentRoot ' + chilDomain.path + '\n'
                    except BaseException as msg:
                        website = Websites.objects.get(domain=virtualHostName)
                        externalApp = website.externalApp
                        docRoot = ACLManager.FindDocRootOfSite(None, virtualHostName)
                        DocumentRoot = f'    DocumentRoot {docRoot}\n'

data = open(completePathToConfigFile, 'r').readlines()
                    phpHandler = ''

for items in data:
                        if items.find('AddHandler') > -1 and items.find('php') > -1:
                            phpHandler = items
                            break

confFile = open(completePathToConfigFile, 'a')

cacheRoot = """    <IfModule LiteSpeed>
            CacheRoot lscache
            CacheLookup on
        </IfModule>
    """

VirtualHost = '\n<VirtualHost *:443>\n\n'
                    ServerName = '    ServerName ' + virtualHostName + '\n'
                    ServerAlias = '    ServerAlias www.' + virtualHostName + '\n'
                    ServerAdmin = '    ServerAdmin ' + adminEmail + '\n'
                    SeexecUserGroup = '    SuexecUserGroup ' + externalApp + ' ' + externalApp + '\n'
                    CustomLogCombined = '    CustomLog /home/' + virtualHostName + '/logs/' + virtualHostName + '.access_log combined\n'

confFile.writelines(VirtualHost)
                    confFile.writelines(ServerName)
                    confFile.writelines(ServerAlias)
                    confFile.writelines(ServerAdmin)
                    confFile.writelines(SeexecUserGroup)
                    confFile.writelines(DocumentRoot)
                    confFile.writelines(CustomLogCombined)
                    confFile.writelines(cacheRoot)

SSLEngine = '    SSLEngine on\n'
                    SSLVerifyClient = '    SSLVerifyClient none\n'
                    SSLCertificateFile = '    SSLCertificateFile /etc/letsencrypt/live/' + virtualHostName + '/fullchain.pem\n'
                    SSLCertificateKeyFile = '    SSLCertificateKeyFile /etc/letsencrypt/live/' + virtualHostName + '/privkey.pem\n'

confFile.writelines(SSLEngine)
                    confFile.writelines(SSLVerifyClient)
                    confFile.writelines(SSLCertificateFile)
                    confFile.writelines(SSLCertificateKeyFile)
                    confFile.writelines(phpHandler)

VirtualHostEnd = '</VirtualHost>\n'
                    confFile.writelines(VirtualHostEnd)
                    confFile.close()
                    return 1
                except BaseException as msg:
                    logging.CyberCPLogFileWriter.writeToFile(str(msg) + " [installSSLForDomain]")
                    return 0
            else:
                cert = open('/etc/letsencrypt/live/' + virtualHostName + '/fullchain.pem').read().rstrip('\n')
                key = open('/etc/letsencrypt/live/' + virtualHostName + '/privkey.pem', 'r').read().rstrip('\n')
                command = 'redis-cli hmset "ssl:%s" crt "%s" key "%s"' % (virtualHostName, cert, key)
                logging.CyberCPLogFileWriter.writeToFile('hello world aaa')
                logging.CyberCPLogFileWriter.writeToFile(command)
                ProcessUtilities.executioner(command)
                return 1

@staticmethod
    def obtainSSLForADomain(virtualHostName, adminEmail, sslpath, aliasDomain=None, isHostname=False):
        from plogical.acl import ACLManager
        from plogical.sslv2 import sslUtilities as sslv2
        from plogical.customACME import CustomACME
        import json
        import socket

# Replace example.org emails with domain-specific email
        if adminEmail and ('example.org' in adminEmail or 'example.com' in adminEmail):
            import re
            # Remove special characters and create domain-based email
            clean_domain = re.sub(r'[^a-zA-Z0-9]', '', virtualHostName)
            adminEmail = f'{clean_domain}@cyberpanel.net'
            logging.CyberCPLogFileWriter.writeToFile(f'Replacing invalid email with {adminEmail}')

Status = 1

if sslUtilities.CheckIfSSLNeedsToBeIssued(virtualHostName) == sslUtilities.ISSUE_SSL:
            pass
        else:
            return 1

sender_email = 'root@%s' % (socket.gethostname())

sslUtilities.PatchVhostConf(virtualHostName)

if not os.path.exists('/usr/local/lsws/Example/html/.well-known/acme-challenge'):
            command = f'mkdir -p /usr/local/lsws/Example/html/.well-known/acme-challenge'
            ProcessUtilities.normalExecutioner(command)

command = f'chmod -R 755 /usr/local/lsws/Example/html'
        ProcessUtilities.executioner(command)

# Try Let's Encrypt first
        try:
            # Start with just the main domain
            domains = [virtualHostName]

# Check if www subdomain has DNS records before adding it (skip for hostnames)
            if not isHostname and sslUtilities.checkDNSRecords(f'www.{virtualHostName}'):
                domains.append(f'www.{virtualHostName}')
                logging.CyberCPLogFileWriter.writeToFile(
                    f"www.{virtualHostName} has DNS records, including in SSL request")
            elif not isHostname:
                logging.CyberCPLogFileWriter.writeToFile(
                    f"www.{virtualHostName} has no DNS records, excluding from SSL request")

if aliasDomain:
                domains.append(aliasDomain)
                # Check if www.aliasDomain has DNS records
                if sslUtilities.checkDNSRecords(f'www.{aliasDomain}'):
                    domains.append(f'www.{aliasDomain}')
                    logging.CyberCPLogFileWriter.writeToFile(
                        f"www.{aliasDomain} has DNS records, including in SSL request")
                else:
                    logging.CyberCPLogFileWriter.writeToFile(
                        f"www.{aliasDomain} has no DNS records, excluding from SSL request")

# Check if Cloudflare is used
            use_dns = False
            try:
                website = Websites.objects.get(domain=virtualHostName)
                if website.externalApp == 'cloudflare':
                    use_dns = True
            except:
                pass

acme = CustomACME(virtualHostName, adminEmail, staging=False, provider='letsencrypt')
            if acme.issue_certificate(domains, use_dns=use_dns):
                logging.CyberCPLogFileWriter.writeToFile(
                    f"Successfully obtained SSL using Let's Encrypt for: {virtualHostName}")
                return 1
        except Exception as e:
            error_msg = str(e)
            # Try to extract more detailed error information
            if hasattr(e, '__dict__'):
                error_details = str(e.__dict__)
            else:
                error_details = error_msg

logging.CyberCPLogFileWriter.writeToFile(
                f"Let's Encrypt failed for {virtualHostName}: {error_msg}"
            )
            logging.CyberCPLogFileWriter.writeToFile(
                f"Detailed error: {error_details}. Trying ZeroSSL..."
            )

# Try ZeroSSL if Let's Encrypt fails
        try:
            # Start with just the main domain
            domains = [virtualHostName]

acme = CustomACME(virtualHostName, adminEmail, staging=False, provider='zerossl')
            if acme.issue_certificate(domains, use_dns=use_dns):
                logging.CyberCPLogFileWriter.writeToFile(
                    f"Successfully obtained SSL using ZeroSSL for: {virtualHostName}")
                return 1
        except Exception as e:
            logging.CyberCPLogFileWriter.writeToFile(
                f"ZeroSSL failed: {str(e)}. Falling back to acme.sh")

# Fallback to acme.sh if both ACME providers fail
        try:
            acmePath = '/root/.acme.sh/acme.sh'
            command = '%s --register-account -m %s' % (acmePath, adminEmail)
            subprocess.call(shlex.split(command))

command = '%s --set-default-ca --server letsencrypt' % (acmePath)
            subprocess.call(shlex.split(command))

if aliasDomain is None:
                existingCertPath = '/etc/letsencrypt/live/' + virtualHostName
                if not os.path.exists(existingCertPath):
                    command = 'mkdir -p ' + existingCertPath
                    subprocess.call(shlex.split(command))

try:
                    # Build domain list for acme.sh
                    domain_list = " -d " + virtualHostName

# Check if www subdomain has DNS records (skip for hostnames)
                    if not isHostname and sslUtilities.checkDNSRecords(f'www.{virtualHostName}'):
                        domain_list += " -d www." + virtualHostName
                        logging.CyberCPLogFileWriter.writeToFile(
                            f"www.{virtualHostName} has DNS records, including in acme.sh SSL request")
                    elif not isHostname:
                        logging.CyberCPLogFileWriter.writeToFile(
                            f"www.{virtualHostName} has no DNS records, excluding from acme.sh SSL request")

command = acmePath + " --issue" + domain_list \
                              + ' --cert-file ' + existingCertPath + '/cert.pem' + ' --key-file ' + existingCertPath + '/privkey.pem' \
                              + ' --fullchain-file ' + existingCertPath + '/fullchain.pem' + ' -w /usr/local/lsws/Example/html -k ec-256 --force --staging' \
                              + ' --webroot-path /usr/local/lsws/Example/html'

try:
                        result = subprocess.run(command, capture_output=True, universal_newlines=True, shell=True)
                    except TypeError:
                        # Fallback for Python < 3.7
                        result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                                universal_newlines=True, shell=True)

if result.returncode == 0:
                        command = acmePath + " --issue" + domain_list \
                                  + ' --cert-file ' + existingCertPath + '/cert.pem' + ' --key-file ' + existingCertPath + '/privkey.pem' \
                                  + ' --fullchain-file ' + existingCertPath + '/fullchain.pem' + ' -w /usr/local/lsws/Example/html -k ec-256 --force --server letsencrypt' \
                                  + ' --webroot-path /usr/local/lsws/Example/html'

try:
                            result = subprocess.run(command, capture_output=True, universal_newlines=True, shell=True)
                        except TypeError:
                            # Fallback for Python < 3.7
                            result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                                    universal_newlines=True, shell=True)

if result.returncode == 0:
                            logging.CyberCPLogFileWriter.writeToFile(
                                "Successfully obtained SSL for: " + virtualHostName + " and: www." + virtualHostName, 0)
                            logging.CyberCPLogFileWriter.SendEmail(sender_email, adminEmail, result.stdout,
                                                                   'SSL Notification for %s.' % (virtualHostName))
                            return 1
                    return 0
                except Exception as e:
                    logging.CyberCPLogFileWriter.writeToFile(str(e))
                    return 0
            else:
                existingCertPath = '/etc/letsencrypt/live/' + virtualHostName
                if not os.path.exists(existingCertPath):
                    command = 'mkdir -p ' + existingCertPath
                    subprocess.call(shlex.split(command))

try:
                    # Build domain list for acme.sh with alias domains
                    domain_list = " -d " + virtualHostName

# Check if www subdomain has DNS records
                    if sslUtilities.checkDNSRecords(f'www.{virtualHostName}'):
                        domain_list += " -d www." + virtualHostName

# Add alias domain
                    domain_list += " -d " + aliasDomain

# Check if www.aliasDomain has DNS records
                    if sslUtilities.checkDNSRecords(f'www.{aliasDomain}'):
                        domain_list += " -d www." + aliasDomain

if result.returncode == 0:
                        return 1
                    return 0
                except Exception as e:
                    logging.CyberCPLogFileWriter.writeToFile(str(e))
                    return 0
        except Exception as e:
            logging.CyberCPLogFileWriter.writeToFile(str(e))
            return 0

def issueSSLForDomain(domain, adminEmail, sslpath, aliasDomain=None, isHostname=False):
    try:
        # Check if certificate already exists and try to renew it first
        existingCertPath = '/etc/letsencrypt/live/' + domain + '/fullchain.pem'
        if os.path.exists(existingCertPath):
            # Check if certificate is expired
            is_expired = False
            try:
                import OpenSSL
                from datetime import datetime
                with open(existingCertPath, 'r') as cert_file:
                    x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, cert_file.read())
                expire_data = x509.get_notAfter().decode('ascii')
                final_date = datetime.strptime(expire_data, '%Y%m%d%H%M%SZ')
                now = datetime.now()
                diff = final_date - now
                is_expired = diff.days < 0
                logging.CyberCPLogFileWriter.writeToFile(f"Certificate for {domain} expires in {diff.days} days")
            except Exception as e:
                logging.CyberCPLogFileWriter.writeToFile(f"Could not check certificate expiry: {str(e)}")

logging.CyberCPLogFileWriter.writeToFile(f"Certificate exists for {domain}, attempting renewal...")

# Try to renew using acme.sh
            acmePath = '/root/.acme.sh/acme.sh'
            if os.path.exists(acmePath):
                # First set the webroot path for the domain
                command = f'{acmePath} --update-account --accountemail {adminEmail}'
                subprocess.call(command, shell=True)

# Build domain list for renewal
                renewal_domains = f'-d {domain}'
                if not isHostname and sslUtilities.checkDNSRecords(f'www.{domain}'):
                    renewal_domains += f' -d www.{domain}'

# For expired certificates, use --issue --force instead of --renew
                if is_expired:
                    logging.CyberCPLogFileWriter.writeToFile(
                        f"Certificate is expired, using --issue --force for {domain}")
                    command = f'{acmePath} --issue {renewal_domains} --webroot /usr/local/lsws/Example/html --force'
                else:
                    # Try to renew with explicit webroot
                    command = f'{acmePath} --renew {renewal_domains} --webroot /usr/local/lsws/Example/html --force'

try:
                    result = subprocess.run(command, capture_output=True, text=True, shell=True)
                except TypeError:
                    # Fallback for Python < 3.7
                    result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                            universal_newlines=True, shell=True)

if result.returncode == 0:
                    logging.CyberCPLogFileWriter.writeToFile(f"Successfully renewed SSL for {domain}")
                    if sslUtilities.installSSLForDomain(domain, adminEmail) == 1:
                        return [1, "SSL successfully renewed"]
                else:
                    # Parse ACME error details
                    error_output = result.stderr if hasattr(result, 'stderr') and result.stderr else result.stdout
                    error_details = sslUtilities.parseACMEError(error_output)
                    logging.CyberCPLogFileWriter.writeToFile(f"Renewal failed for {domain}. Error: {error_details}")
                    logging.CyberCPLogFileWriter.writeToFile(f"Full error output: {error_output}")

if sslUtilities.obtainSSLForADomain(domain, adminEmail, sslpath, aliasDomain, isHostname) == 1:
            if sslUtilities.installSSLForDomain(domain, adminEmail) == 1:
                return [1, "None"]
            else:
                return [0, "210 Failed to install SSL for domain. [issueSSLForDomain]"]
        else:

pathToStoreSSLPrivKey = "/etc/letsencrypt/live/%s/privkey.pem" % (domain)
            pathToStoreSSLFullChain = "/etc/letsencrypt/live/%s/fullchain.pem" % (domain)

#### if in any case ssl failed to obtain and CyberPanel try to issue self-signed ssl, first check if ssl already present.
            ### if so, dont issue self-signed ssl, as it may override some existing ssl

if os.path.exists(pathToStoreSSLFullChain):
                import OpenSSL
                x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
                                                       open(pathToStoreSSLFullChain, 'r').read())
                SSLProvider = x509.get_issuer().get_components()[1][1].decode('utf-8')

if SSLProvider != 'Denial':
                    if sslUtilities.installSSLForDomain(domain) == 1:
                        logging.CyberCPLogFileWriter.writeToFile(
                            "We are not able to get new SSL for " + domain + ". But there is an existing SSL, it might only be for the main domain (excluding www).")
                        return [1,
                                "We are not able to get new SSL for " + domain + ". But there is an existing SSL, it might only be for the main domain (excluding www)." + " [issueSSLForDomain]"]

command = 'openssl req -newkey rsa:2048 -new -nodes -x509 -days 3650 -subj "/C=US/ST=Denial/L=Springfield/O=Dis/CN=' + domain + '" -keyout ' + pathToStoreSSLPrivKey + ' -out ' + pathToStoreSSLFullChain
            cmd = shlex.split(command)
            subprocess.call(cmd)

if sslUtilities.installSSLForDomain(domain) == 1:
                logging.CyberCPLogFileWriter.writeToFile("Self signed SSL issued for " + domain + ".")
                return [1, "Self signed certificate was issued. [issueSSLForDomain]"]
            else:
                return [0, "210 Failed to install SSL for domain. [issueSSLForDomain]"]

except BaseException as msg:
        return [0, "347 " + str(msg) + " [issueSSLForDomain]"]

${this.title}

Code Editor : sslUtilities.py