control.py

import sys, socks, tldextract, CloudFlare, json, loguru, os, requests, socket, subprocess, platform
from urllib.parse import urlparse
from namesilo.core import NameSilo
from typing import List, Dict, Any
from pprint import pprint
from loguru import logger as log
from config import cfg as global_cfg, get_proxy_connector
from urllib.parse import urlparse



log_levels=['TRACE', 'DEBUG', 'INFO', 'SUCCESS', 'WARNING', 'ERROR', 'CRITICAL', 'RESULT']

def set_cfg(cfg):
    global global_cfg
    global_cfg = cfg

def set_proxy(PROXY: str = None, cfg_instance: (Any, 'System var, do not set') = None):
    """
    Configures a global SOCKS/HTTP proxy for all socket connections.
    If PROXY is None, uses cfg.PROXY.
    If proxy cannot be configured or is not provided, raises a ValueError to enforce mandatory proxy.
    """
    cfg = cfg_instance if cfg_instance is not None else global_cfg
    proxy_url = PROXY if PROXY is not None else getattr(cfg, "PROXY", None)

    if not proxy_url:
        raise ValueError("No proxy URL provided or found in config. Proxy is mandatory and not set.")

    parsed = urlparse(proxy_url)
    proxy_type = None
    if parsed.scheme == "socks5":
        proxy_type = socks.SOCKS5
    elif parsed.scheme == "socks4":
        proxy_type = socks.SOCKS4
    elif parsed.scheme == "http": # Basic HTTP proxy support via socks
        proxy_type = socks.HTTP
    else:
        raise ValueError(f"Unsupported proxy scheme: {parsed.scheme}. Supported schemes: socks5, socks4, http.")

    try:
        socks.set_default_proxy(
            proxy_type,
            parsed.hostname,
            parsed.port,
            username=parsed.username,
            password=parsed.password
        )
        # Monkey-patch the default socket to use socks.socksocket
        socket.socket = socks.socksocket
        log.info(f"Global proxy set to: {proxy_url}")
    except Exception as e:
        raise ValueError(f"Failed to set global proxy using socks: {e}. Proxy is mandatory and failed to configure.")

        
def check_proxy(PROXY:str=None):
    set_proxy(PROXY=PROXY)
    try:
        s = requests.Session()
        r = s.get('http://checkip.amazonaws.com', timeout=10)
        ip = r.text.strip()
        if 7 < len(ip) < 16:
            log.info(f"Proxy IP: {ip}")
        else:
            log.error("Proxy is not working!")
            raise Exception("Invalid IP returned")
    except Exception as e:
        log.error(f"Proxy is not working! Error: {e}")
        exit()


class Cloud():
        
    def __init__(self, cfg_instance:(Any, 'System var, do not set')=None, *args, **kwargs):
        
        self.cfg = cfg_instance
        API_CF = self.cfg.API_CF

        try:
            set_proxy(PROXY=self.cfg.PROXY) # Ensure proxy is set for CloudFlare operations
        except ValueError as e:
            log.error(f"Cloud initialization failed due to proxy configuration: {e}")
            exit() # Exit if proxy cannot be set for CloudFlare

        self.cf = CloudFlare.CloudFlare(token=API_CF)
        self.result = []
        log.info(API_CF)

    def add(self, domain:str, *args, **kwargs):
        try:
            zone_info = self.cf.zones.post(data={'jump_start': False, 'name': domain})
        except CloudFlare.exceptions.CloudFlareAPIError as e:
            exit('/zones.post %s - %d %s' % (domain, e, e))
        except Exception as e:
            exit('/zones.post %s - %s' % (domain, e))

        zone_id = zone_info['id']
        if 'email' in zone_info['owner']:
            zone_owner = zone_info['owner']['email']
        else:
            zone_owner = '"' + zone_info['owner']['name'] + '"'
        zone_plan = zone_info['plan']['name']
        zone_status = zone_info['status']
        log.info('\t%s name=%s owner=%s plan=%s status=%s\n' % (
            zone_id,
            domain,
            zone_owner,
            zone_plan,
            zone_status
        ))
        
    def pagerules(self, *args, **kwargs):
        targets=[{"target":"url","constraint":{"operator":"matches","value":url_match}}]
        actions=[{"id":"forwarding_url","value":{"status_code":302,"url":url_forwarded}}]
        pagerule_for_redirection = {"status": "active","priority": 1,"actions": actions,"targets": targets}

        try:
            r = self.cf.zones.pagerules.get(zone_id, data=pagerule_for_redirection)
        except CloudFlare.exceptions.CloudFlareAPIError as e:
            exit('/zones.pagerules.get %d %s - api call failed' % (e, e))
        for rule in r:
            log.info(rule)

    def dns_del(self, zone_id:str, record_id:str, *args, **kwargs):
        r = self.cf.zones.dns_records.delete(zone_id, record_id)
        log.info(r)

    def dns_edit(
        self, 
        name:(str, 'domain name or subdomain'), 
        type:(str, 'A, MX, CNAME, TXT or other records'), 
        content:(str, 'value of record'), 
        zone_id:(str, 'id of zone'),
        proxied:(bool, 'Proxy by Cloudflare or not')=True, 
        *args, **kwargs
    ):
        ''' Put DNS records \n --type A --name www --content 1.1.1.1 --zone_id ''' 
        
        dns_records = [
            {'name': name, 'type': type, 'content': content, 'proxied': proxied},
        ]
        zone_name = ''
        [
            {'name':'foo', 'type':'AAAA', 'content':'2001:d8b::100:100'},
            {'name':'bar', 'type':'CNAME', 'content':'foo.%s' % (zone_name)},
            {'name':'shakespeare', 'type':'TXT', 'content':"What's in a name? That which we call a rose by any other name would smell as sweet."}
        ]
        print(dns_records)
        for dns_record in dns_records:
            r = self.cf.zones.dns_records.post(zone_id, data=dns_record)
            log.info(r)
    def dns_show(self, zone_id:(str, 'id of zone'), *args, **kwargs):
        ''' Show DNS records of selected zone_id '''
        
        dns_records = self.cf.zones.dns_records.get(zone_id)
        for dns_record in sorted(dns_records, key=lambda v: v['name']):
            log.result(dns_record)

    def list(self, *args, **kwargs):
        ## List all zones
        for zone in self.cf.zones.get(params={'per_page':500}):
            # log.info((
            #     zone['name'],
            #     zone['name_servers'],
            #     zone['status'],
            #     zone['original_name_servers'],
            #     zone['development_mode'],
            #     zone['original_registrar'],
            #     zone['created_on'],
            #     zone['activated_on'],
            #     zone['modified_on'],
            #     zone['meta']['phishing_detected']
            # ))
            self.result.append(zone['name'])
            log.result(json.dumps(zone))
        log.info(self.result)            
            
            

class Namesilo():
    '''Domain management with Namesilo
    Used library: https://github.com/goranvrbaski/python-namesilo/blob/develop/namesilo/core.py#L384'''
    
    # domains = pd.DataFrame(namesilo.list_domains())
    # data = {"balance": namesilo.get_account_balance(), "domains": domains}
    # print(domains)
    def __init__(self, cfg_instance:(Any, 'System var, do not set')=None,
        domain:(str, 'Domain for operation')=None, 
        *args, **kwargs):
        ''' Check, Register, Change NS and other domain operations on Namesilo '''
        self.domain = domain
        self.cfg = cfg_instance

        set_proxy(PROXY=self.cfg.PROXY)
        self.namesilo = NameSilo(token=self.cfg.API_NS, sandbox=False)
        self.result = []

    def check(self, *args, **kwargs) -> bool:
        ''' Check domain '''
        domain = getattr(self, 'domain')
        available = self.namesilo.check_domain(domain)
        if available:
            log.info(f"You can buy {domain}")
            return True
        else:
            log.info(f"{domain} is registered!")
            return False

    # print(namesilo.update_dns_records)
    def change_ns(self, domain:str, ns1:(str, 'first ns')='maeve.ns.cloudflare.com', ns2:(str, 'second ns')='trey.ns.cloudflare.com', **kwargs):
        '''Change NS names of domain'''
        domain = getattr(self, 'domain', self.domain)
        x = self.namesilo.change_domain_nameservers(domain, ns1, ns2)
        log.info(x)

    def list(self, **kwargs):
        '''List all registered domains'''
        domains = self.namesilo.list_domains()
        log.info(domains)
        for domain in domains:
            log.result(domain['#text'])

    def prices(self, **kwargs):
        '''Get the price of all possible tld zones'''
        log.result(self.namesilo.get_prices())

    def price(self, **kwargs):
        '''Get the price of a domain'''
        domain = getattr(self, 'domain')
        loc = tldextract.extract(domain)
        tld = loc.suffix
        domain = loc.domain
        prices = self.namesilo.get_prices()  
        log.info(prices[tld])

    def buy(self, *args, **kwargs):
        ''' Buy domain '''
        domain = getattr(self, 'domain')
        if self.check(domain):
            if self.namesilo.register_domain(domain, years=1, auto_renew=0, private=1):
                log.info(f"You bought {domain}")

    def pre_buy(self, **kwargs):
        '''Check everything but not buy actual domain'''
        domain = getattr(self, 'domain')
        if self.check(domain):
            self.price(domain=domain)
            log.info(f"Your balance - {self.namesilo.get_account_balance()}")

    def info(self, **kwargs):
        '''Info about domain and it's nameservers'''
        domain = getattr(self, 'domain')
        # result = self.namesilo.list_dns_records(domain)
        # log.infopp(result)
        result = self.namesilo.get_domain_info(domain)
        log.info(result.__dict__)

    def balance(self, amount:int=None, **kwargs):
        '''See Balance. Add balance is not working'''
        result = self.namesilo.get_account_balance()
        log.info(result)
        # if amount:
            # self.namesilo.add_account_funds(amount=amount, payment_id=1)

    def test(self, **kwargs):

        ## Search zone
        zone = self.cf.zones.get(params={'name': self.domain})
        log.info(zone)

        ## Add zone
        zone = self.cf.zones.post(data={'jump_start':False, 'name': self.domain})


        ## Always HTTPS
        r = self.cf.zones.settings.always_use_https.get(zone['id'])
        log.info(r)
        r = self.cf.zones.settings.always_use_https.patch(zone['id'], data={'value':'on'})
        log.info(r)
        

        ## Page rules
        ## https://developers.cloudflare.com/api/operations/available-page-rules-settings-list-available-page-rules-settings
        
        targets=[{"target":"url","constraint":{"operator":"matches","value":url_match}}]
        actions=[{"id":"forwarding_url","value":{"status_code":302,"url":url_forwarded}}]
        pagerule_for_redirection = {"status": "active","priority": 1,"actions": actions,"targets": targets}

        r = self.cf.zones.pagerules.get(zone_id, data=pagerule_for_redirection)
        for rule in r:
            if (rule['actions'] == pagerule_for_redirection["actions"] and rule["targets"] == pagerule_for_redirection["targets"]):
                print('\t', '... rule already present!')
        r = self.cf.zones.pagerules.post(zone_id, data=pagerule_for_redirection)
        
        ## Import / Export dns


        ## Export dns

        dns_records = self.cf.zones.dns_records.export.get(zone_id)

        for l in dns_records.splitlines():
            if len(l) == 0 or l[0] == ';':
                # blank line or comment line are skipped - to make example easy to see
                continue
            print(l)


class Nginx():
    # TODO: Implement Nginx class
    def __init__(self, *args, **kwargs):
        pass
        
    def auth(self, **kwargs):
        pass

    def list(self, **kwargs):
        pass
    
    def domains_edit(self, domains:(list, "List of a domains to put into sever_name"),**kwargs):
        pass
    
    def domains_show(self, **kwargs):  
        pass

if __name__ == '__main__':

    try:
        from config import main_cli_entry_point # Import the new centralized function
    except ImportError as e:
        print(f"Error importing main_cli_entry_point from config.py: {e}")
        print("Please ensure config.py is in your PYTHONPATH or accessible.")
        sys.exit(1)

    
    params_cmd = [
        (['-l', '--log-level'], {'choices': log_levels, 'default': 'INFO', 'help': 'Set the logging level (default: INFO)'}),
        (['--no-file-log'], {'action': 'store_true', 'help': 'Disable file logging (console only)'}),
    ]
    params_subcmd = [*params_cmd]

    # Call the centralized entry point
    main_cli_entry_point(
        cli_module=sys.modules[__name__],
        global_args_definitions=params_cmd,
        subcommand_args_definitions=params_subcmd
    )