electrumz_server

import asyncio
import logging
import sys
import os
import clr
import datetime
from pathlib import Path
from cryptography import x509
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
import configparser

clr.AddReference("System.IO")
import System.IO as Os

from electrumz import Controller, Env
from electrumz.lib.util import CompactFormatter


def load_config(logger):
    """Load configuration from electrumz.conf file."""
    config = configparser.ConfigParser()
    config_file_path = os.path.join(os.path.dirname(__file__), 'electrumz.conf')
    
    if not os.path.exists(config_file_path):
        logger.error(f"Configuration file not found: {config_file_path}")
        raise FileNotFoundError("Configuration file not found.")
    
    config.read(config_file_path)
    
    try:
        os.environ['DAEMON_URL'] = config.get('server', 'daemon_url')
        os.environ['REPORT_SERVICES'] = config.get('server', 'report_services')
        os.environ['CACHE_MB'] = config.get('server', 'cache_mb')
        os.environ['DONATION_ADDRESS'] = config.get('server', 'donation_address')
        
    except (configparser.NoOptionError, configparser.NoSectionError) as e:
        logger.error(f"Missing configuration option: {e}")
        raise


def set_environment_variables(logger):
    """Set required environment variables."""
    appdata_local_path = Path(os.environ.get('LOCALAPPDATA', Path.home() / 'AppData' / 'Local'))
    db_directory_path = appdata_local_path / 'BTCZCommunity' / 'ElectrumZ'
    cert_file = db_directory_path / 'electrumz.crt'
    key_file = db_directory_path / 'electrumz.key'
    csr_file = db_directory_path / 'electrumz.csr'
    
    os.environ['HOME'] = str(db_directory_path)
    os.environ['DB_DIRECTORY'] = str(db_directory_path)
    os.environ['COIN'] = 'BitcoinZ'
    os.environ['EVENT_LOOP_POLICY'] = 'winloop'
    os.environ['SERVICES'] = 'tcp://:50001,ssl://:50002,wss://:50004,rpc://0.0.0.0:8000'
    os.environ['SSL_CERTFILE'] = str(cert_file)
    os.environ['SSL_KEYFILE'] = str(key_file)
    os.environ['CSR_FILE'] = str(csr_file)
    os.environ['DB_ENGINE'] = 'leveldb'
    os.environ['INITIAL_CONCURRENT'] = '1000000'
    os.environ['COST_SOFT_LIMIT'] = '1000000'
    os.environ['COST_HARD_LIMIT'] = '1000001'
    os.environ['REQUEST_SLEEP'] = '0'



def generate_ssl_files(logger):
    """Generate SSL private key, CSR, and self-signed certificate if they do not exist."""
    cert_file = os.environ['SSL_CERTFILE']
    key_file = os.environ['SSL_KEYFILE']
    csr_file = os.environ['CSR_FILE']

    if not os.path.exists(cert_file) or not os.path.exists(key_file):
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048,
        )
        
        with open(os.environ['SSL_KEYFILE'], 'wb') as key_file:
            key_file.write(
                private_key.private_bytes(
                    encoding=serialization.Encoding.PEM,
                    format=serialization.PrivateFormat.TraditionalOpenSSL,
                    encryption_algorithm=serialization.NoEncryption()
                )
            )
        logger.info(f'Private key generated at {os.environ["SSL_KEYFILE"]}')
        
        subject = x509.Name([
            x509.NameAttribute(x509.NameOID.ORGANIZATION_NAME, u'ElectrumZ'),
        ])
        
        csr = x509.CertificateSigningRequestBuilder().subject_name(
            subject
        ).sign(private_key, hashes.SHA256())
        
        with open(os.environ['CSR_FILE'], 'wb') as csr_file:
            csr_file.write(csr.public_bytes(encoding=serialization.Encoding.PEM))
        
        logger.info(f'CSR generated at {os.environ["CSR_FILE"]}')
        
        certificate = x509.CertificateBuilder().subject_name(
            subject
        ).issuer_name(
            subject
        ).public_key(
            private_key.public_key()
        ).serial_number(
            x509.random_serial_number()
        ).not_valid_before(
            datetime.datetime.utcnow()
        ).not_valid_after(
            datetime.datetime.utcnow() + datetime.timedelta(days=365)
        ).sign(private_key, hashes.SHA256())
        
        with open(os.environ['SSL_CERTFILE'], 'wb') as cert_file:
            cert_file.write(certificate.public_bytes(encoding=serialization.Encoding.PEM))
        
        logger.info(f'Self-signed certificate generated at {os.environ["SSL_CERTFILE"]}')
        
        os.remove(os.environ['CSR_FILE'])
        logger.info(f'CSR file deleted: {os.environ["CSR_FILE"]}')



def print_ansi_art(file_path):
    try:
        with open(file_path, 'r') as file:
            ansi_art = file.read()
        print(ansi_art)
    except FileNotFoundError:
        print(f"Error: The file {file_path} was not found.")
    except IOError:
        print(f"Error: Could not read the file {file_path}.")



def is_already_running(lock_file):
    if Os.File.Exists(lock_file):
        try:
            Os.File.Delete(lock_file)
        except Os.IOException:
            return True
    return False


def create_lock_file(lock_file):
    try:
        lock_file_stream = Os.FileStream(
            lock_file,
            Os.FileMode.CreateNew,
            Os.FileAccess.ReadWrite,
            Os.FileShare(0)
        )
    except Os.IOException:
        return False
    return lock_file_stream


def remove_lock_file(lock_file):
    if Os.File.Exists(lock_file):
        Os.File.Delete(lock_file)



def main():
    """Set up logging, environment variables, and run the server."""
    ansi_file_path = os.path.join(os.path.dirname(__file__), 'icons', 'electrumz.ans')
    print_ansi_art(ansi_file_path)
    appdata_local_path = Path(os.environ.get('LOCALAPPDATA', Path.home() / 'AppData' / 'Local'))
    directory_path = appdata_local_path / 'BTCZCommunity' / 'ElectrumZ'
    lock_file = Os.Path.Combine(str(directory_path), ".lock")
    log_directory = directory_path / 'logs'
    log_directory.mkdir(parents=True, exist_ok=True)
    log_file_path = log_directory / 'electrumz.log'

    log_fmt = Env.default('LOG_FORMAT', '%(asctime)s - %(levelname)s - %(name)s - %(message)s')
    datefmt = "%Y-%m-%d %H:%M:%S"
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(CompactFormatter(log_fmt, datefmt=datefmt))
    file_handler = logging.FileHandler(log_file_path)
    file_handler.setFormatter(CompactFormatter(log_fmt, datefmt=datefmt))
    logger = logging.getLogger('electrumz')
    logger.setLevel('INFO')
    logger.addHandler(console_handler)
    logger.addHandler(file_handler)

    if is_already_running(lock_file):
        logger.error("Another instance may be running.")
        return

    lock_file_stream = create_lock_file(lock_file)
    if not lock_file_stream:
        logger.error("Failed to create lock file. Another instance may be running.")
        return

    logger.info('ElectrumZ server starting')
    try:
        if sys.version_info < (3, 8):
            raise RuntimeError('ElectrumZ requires Python 3.8 or greater')

        load_config(logger)
        set_environment_variables(logger)
        generate_ssl_files(logger)

        env = Env()
        logger.info(f'Logging level: {env.log_level}')
        logger.setLevel(env.log_level)
        controller = Controller(env)
        asyncio.run(controller.run())
    except Exception:
        logger.exception('ElectrumZ server terminated abnormally')
    else:
        logger.info('ElectrumZ server terminated normally')
    finally:
        remove_lock_file(lock_file)

if __name__ == '__main__':
    main()