import json import os import re from datetime import datetime from pytz import utc from linode.api import Api _datacenter_dict = None _pricing_plan_dict = None _distribution_dict = None _stackscript_dict = None _linode_dict = None _kernel_dict = None def get_api(apiKey): return Api(key=apiKey) def get_datacenter_dict(api, reload=False): global _datacenter_dict if reload or (_datacenter_dict is None): _datacenter_dict = {dc['ABBR']: dc for dc in api.avail_datacenters()} return _datacenter_dict def get_pricing_plan_dict(api, reload=False): global _pricing_plan_dict if reload or (_pricing_plan_dict is None): _pricing_plan_dict = {plan['LABEL']: plan for plan in api.avail_linodeplans()} return _pricing_plan_dict def get_distribution_dict(api, reload=False): global _distribution_dict if reload or (_distribution_dict is None): _distribution_dict = {dist['LABEL']: dist for dist in api.avail_distributions()} return _distribution_dict def get_stackscript_dict(api, reload=False): global _stackscript_dict if reload or (_stackscript_dict is None): _stackscript_dict = {script['LABEL']: script for script in api.stackscript_list()} return _stackscript_dict def get_kernel_dict(api, reload=False): global _kernel_dict if reload or (_kernel_dict is None): _kernel_dict = {kernel['LABEL']: kernel for kernel in api.avail_kernels()} return _kernel_dict def get_latest_kernel(api, is_64_bit=True): kernels = get_kernel_dict(api) prefix = 'latest 64 bit' if is_64_bit else 'latest 32 bit' for label, kernel in kernels.items(): if label.lower().startswith(prefix): return kernel raise RuntimeError('Latest %s bit kernel not found' % ('64' if is_64_bit else '32',)) def get_linode_dict(api, reload=False): global _linode_dict if reload or (_linode_dict is None): _linode_dict = {linode['LABEL']: linode for linode in api.linode_list()} return _linode_dict def get_linode_by_label(api, linode_label, reload=False): linodes = get_linode_dict(api, reload) if linode_label in linodes: return linodes[linode_label] else: raise ValueError('Invalid Linode label: ' + linode_label) def get_linode_by_id(api, linode_id, reload=False): linodes = get_linode_dict(api, reload) for linode in linodes.values(): if linode['LINODEID'] == linode_id: return linode raise ValueError('Invalid Linode ID: ' + linode_id) def get_swap_disk(api, linode_id): for disk in api.linode_disk_list(LinodeID=linode_id): if disk['TYPE'] == 'swap': return disk return None def get_data_disk(api, linode_id): for disk in api.linode_disk_list(LinodeID=linode_id): if disk['LABEL'].lower() == 'data': return disk return None def create_data_disk(api, linode_id, data_disk_size_mb): disk = api.linode_disk_create(LinodeID=linode_id, Size=data_disk_size_mb, Label="Data", Type="ext4") return disk['DiskID'] def create_swap_disk(api, linode_id, swap_disk_size_mb): disk = api.linode_disk_create(LinodeID=linode_id, Size=swap_disk_size_mb, Label="Swap", Type="swap") return disk['DiskID'] def create_linode(api, datacenter_abbr, pricing_plan_label): dcs = get_datacenter_dict(api) if datacenter_abbr not in dcs: raise ValueError('Invalid datacenter: ' + datacenter_abbr) plans = get_pricing_plan_dict(api) if pricing_plan_label not in plans: raise ValueError('Invalid pricing plan: ' + pricing_plan_label) dc_id = dcs[datacenter_abbr]['DATACENTERID'] plan_id = plans[pricing_plan_label]['PLANID'] return api.linode_create(DatacenterID=dc_id, PlanID=plan_id) def deploy_linode_from_stackscript(api, linode_id, configuration_label, stackscript_label, stackscript_args, distribution_label, disk_label, disk_size_mb, root_password, kernel_id=None, extra_disk_ids=[]): scripts = get_stackscript_dict(api) script = scripts.get(stackscript_label) if script is None: raise ValueError('Invalid stackscript: ' + stackscript_label) dists = get_distribution_dict(api) dist = dists.get(distribution_label) if dist is None: raise ValueError('Invalid distribution: ' + distribution_label) if not str(dist['DISTRIBUTIONID']) in str(script['DISTRIBUTIONIDLIST']).split(','): raise ValueError('The specified StackScript does not support the specified distribution') if kernel_id is None: kernel = get_latest_kernel(api, dist['IS64BIT'] != 0) kernel_id = kernel['KERNELID'] disk = api.linode_disk_createfromstackscript( LinodeID=linode_id, StackScriptID=script['STACKSCRIPTID'], StackScriptUDFResponses=json.dumps(stackscript_args), DistributionID=dist['DISTRIBUTIONID'], Label=disk_label, Size=disk_size_mb, rootPass=root_password ) disk_list = ','.join(str(x) for x in [disk['DiskID']] + list(extra_disk_ids)) config = api.linode_config_create( LinodeID=linode_id, KernelID=kernel_id, Label=configuration_label, DiskList=disk_list, helper_distro=True, helper_disableUpdateDB=True, helper_depmod=True, devtmpfs_automount=True, helper_network=True ) return config def parse_asn1_utctime(value): if value == '': return None else: matches = re.match(r'^(\d\d)(\d\d)(\d\d)(\d\d)(\d\d)(\d\d)Z$', value) if matches: year = int(matches.group(1)) year = year + (2000 if year <= 50 else 1900) return datetime(year, int(matches.group(2)), int(matches.group(3)), int(matches.group(4)), int(matches.group(5)), int(matches.group(6)), tzinfo=utc) else: raise ValueError("Invalid ASN1 UTCTime value: {}".format(value)) def load_openssl_ca_index(ca_root_dir): ret = {} with open(os.path.join(ca_root_dir, 'index.txt'), 'r') as f: # Later instances of the same certificate name will replace earlier # ones. for line in f: fields = line.rstrip("\r\n").split("\t") # As per OpenSSL's apps/apps.h ret[fields[5]] = { 'type': fields[0], 'exp_date': parse_asn1_utctime(fields[1]), 'rev_date': parse_asn1_utctime(fields[2]), 'serial': fields[3], 'file': fields[4], 'name': fields[5], } return ret def find_cert_with_common_name(ca_index, common_name): suffix = '/CN=' + common_name for key, value in ca_index.items(): if key.endswith(suffix): return value return None