linode-utils/utils.py

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