snaps/file_utils.py

   1 # Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
   2 #                    and others.  All rights reserved.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at:
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 import ssl
  16
  17 import os
  18 import logging
  19
  20 from cryptography.hazmat.primitives import serialization
  21
  22 try:
  23     import urllib.request as urllib
  24 except ImportError:
  25     import urllib2 as urllib
  26
  27 import yaml
  28
  29 __author__ = 'spisarski'
  30
  31 """
  32 Utilities for basic file handling
  33 """
  34
  35 logger = logging.getLogger('file_utils')
  36
  37
  38 def file_exists(file_path):
  39     """
  40     Returns True if the image file already exists and throws an exception if
  41     the path is a directory
  42     :return:
  43     """
  44     expanded_path = os.path.expanduser(file_path)
  45     if os.path.exists(expanded_path):
  46         if os.path.isdir(expanded_path):
  47             return False
  48         return os.path.isfile(expanded_path)
  49     return False
  50
  51
  52 def download(url, dest_path, name=None):
  53     """
  54     Download a file to a destination path given a URL
  55     :param url: the endpoint to the file to download
  56     :param dest_path: the directory to save the file
  57     :param name: the file name (optional)
  58     :rtype : File object
  59     """
  60     if not name:
  61         name = url.rsplit('/')[-1]
  62     dest = dest_path + '/' + name
  63     logger.debug('Downloading file from - ' + url)
  64     # Override proxy settings to use localhost to download file
  65     download_file = None
  66
  67     if not os.path.isdir(dest_path):
  68         try:
  69             os.mkdir(dest_path)
  70         except:
  71             raise
  72     try:
  73         with open(dest, 'wb') as download_file:
  74             logger.debug('Saving file to - %s',
  75                          os.path.abspath(download_file.name))
  76             response = __get_url_response(url)
  77             download_file.write(response.read())
  78         return download_file
  79     finally:
  80         if download_file:
  81             download_file.close()
  82
  83
  84 def save_keys_to_files(keys=None, pub_file_path=None, priv_file_path=None):
  85     """
  86     Saves the generated RSA generated keys to the filesystem
  87     :param keys: the keys to save generated by cryptography
  88     :param pub_file_path: the path to the public keys
  89     :param priv_file_path: the path to the private keys
  90     """
  91     if keys:
  92         if pub_file_path:
  93             # To support '~'
  94             pub_expand_file = os.path.expanduser(pub_file_path)
  95             pub_dir = os.path.dirname(pub_expand_file)
  96
  97             if not os.path.isdir(pub_dir):
  98                 os.mkdir(pub_dir)
  99
 100             public_handle = None
 101             try:
 102                 public_handle = open(pub_expand_file, 'wb')
 103                 public_bytes = keys.public_key().public_bytes(
 104                     serialization.Encoding.OpenSSH,
 105                     serialization.PublicFormat.OpenSSH)
 106                 public_handle.write(public_bytes)
 107             finally:
 108                 if public_handle:
 109                     public_handle.close()
 110
 111             os.chmod(pub_expand_file, 0o400)
 112             logger.info("Saved public key to - " + pub_expand_file)
 113         if priv_file_path:
 114             # To support '~'
 115             priv_expand_file = os.path.expanduser(priv_file_path)
 116             priv_dir = os.path.dirname(priv_expand_file)
 117             if not os.path.isdir(priv_dir):
 118                 os.mkdir(priv_dir)
 119
 120             private_handle = None
 121             try:
 122                 private_handle = open(priv_expand_file, 'wb')
 123                 private_handle.write(
 124                     keys.private_bytes(
 125                         encoding=serialization.Encoding.PEM,
 126                         format=serialization.PrivateFormat.TraditionalOpenSSL,
 127                         encryption_algorithm=serialization.NoEncryption()))
 128             finally:
 129                 if private_handle:
 130                     private_handle.close()
 131
 132             os.chmod(priv_expand_file, 0o400)
 133             logger.info("Saved private key to - " + priv_expand_file)
 134
 135
 136 def save_string_to_file(string, file_path, mode=None):
 137     """
 138     Stores
 139     :param string: the string contents to store
 140     :param file_path: the file path to create
 141     :param mode: the file's mode
 142     :return: the file object
 143     """
 144     save_file = open(file_path, 'w')
 145     try:
 146         save_file.write(string)
 147         if mode:
 148             os.chmod(file_path, mode)
 149         return save_file
 150     finally:
 151         save_file.close()
 152
 153
 154 def get_content_length(url):
 155     """
 156     Returns the number of bytes to be downloaded from the given URL
 157     :param url: the URL to inspect
 158     :return: the number of bytes
 159     """
 160     response = __get_url_response(url)
 161     return response.headers['Content-Length']
 162
 163
 164 def __get_url_response(url):
 165     """
 166     Returns a response object for a given URL
 167     :param url: the URL
 168     :return: the response
 169     """
 170     proxy_handler = urllib.ProxyHandler({})
 171     opener = urllib.build_opener(proxy_handler)
 172     urllib.install_opener(opener)
 173     context = ssl._create_unverified_context()
 174     return urllib.urlopen(url, context=context)
 175
 176
 177 def read_yaml(config_file_path):
 178     """
 179     Reads the yaml file and returns a dictionary object representation
 180     :param config_file_path: The file path to config
 181     :return: a dictionary
 182     """
 183     logger.debug('Attempting to load configuration file - ' + config_file_path)
 184     config_file = None
 185     try:
 186         with open(config_file_path, 'r') as config_file:
 187             config = yaml.safe_load(config_file)
 188             logger.info('Loaded configuration')
 189         return config
 190     finally:
 191         if config_file:
 192             logger.info('Closing configuration file')
 193             config_file.close()
 194
 195
 196 def persist_dict_to_yaml(the_dict, file_name):
 197     """
 198     Creates a YAML file from a dict
 199     :param the_dict: the dictionary to store
 200     :param conf_dir: the directory used to store the config file
 201     :return: the file object
 202     """
 203     logger.info('Persisting %s to [%s]', the_dict, file_name)
 204     file_path = os.path.expanduser(file_name)
 205     yaml_from_dict = yaml.dump(
 206         the_dict, default_flow_style=False, default_style='')
 207     return save_string_to_file(yaml_from_dict, file_path)
 208
 209
 210 def read_os_env_file(os_env_filename):
 211     """
 212     Reads the OS environment source file and returns a map of each key/value
 213     Will ignore lines beginning with a '#' and will replace any single or
 214     double quotes contained within the value
 215     :param os_env_filename: The name of the OS environment file to read
 216     :return: a dictionary
 217     """
 218     if os_env_filename:
 219         logger.info('Attempting to read OS environment file - %s',
 220                     os_env_filename)
 221         out = {}
 222         env_file = None
 223         try:
 224             env_file = open(os_env_filename)
 225             for line in env_file:
 226                 line = line.lstrip()
 227                 if not line.startswith('#') and line.startswith('export '):
 228                     line = line.lstrip('export ').strip()
 229                     tokens = line.split('=')
 230                     if len(tokens) > 1:
 231                         # Remove leading and trailing ' & " characters from
 232                         # value
 233                         out[tokens[0]] = tokens[1].lstrip('\'').lstrip('\"').rstrip('\'').rstrip('\"')
 234         finally:
 235             if env_file:
 236                 env_file.close()
 237         return out
 238
 239
 240 def read_file(filename):
 241     """
 242     Returns the contents of a file as a string
 243     :param filename: the name of the file
 244     :return:
 245     """
 246     out = str()
 247     the_file = None
 248     try:
 249         the_file = open(filename)
 250         for line in the_file:
 251             out += line
 252         return out
 253     finally:
 254         if the_file:
 255             the_file.close()