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