File size: 10,587 Bytes
27867f1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 |
"""
MIT License
Copyright (C) 2021 ROCKY4546
https://github.com/rocky4546
This file is part of Cabernet
Permission is hereby granted, free of charge, to any person obtaining a copy of this software
and associated documentation files (the "Software"), to deal in the Software without restriction,
including without limitation the rights to use, copy, modify, merge, publish, distribute,
sublicense, and/or sell copies of the Software, and to permit persons to whom the Software
is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or
substantial portions of the Software.
"""
import glob
import importlib
import importlib.resources
import json
import logging
import os
import pathlib
import re
import time
import urllib.request
import shutil
import zipfile
import lib.common.utils as utils
import lib.db.datamgmt.backups as backups
from lib.db.db_plugins import DBPlugins
from lib.common.decorators import handle_url_except
from lib.common.decorators import handle_json_except
from lib.common.tmp_mgmt import TMPMgmt
TMP_ZIPFILE = utils.CABERNET_ID + '.zip'
class CabernetUpgrade:
def __init__(self, _plugins):
self.logger = logging.getLogger(__name__)
self.version_re = re.compile(r'(\d+\.\d+)\.\d+')
self.plugins = _plugins
self.config_obj = _plugins.config_obj
self.config = _plugins.config_obj.data
self.plugin_db = DBPlugins(self.config)
self.tmp_mgmt = TMPMgmt(self.config)
def update_version_info(self):
"""
Updates the database with the latest version release data
from github for cabernet and plugins loaded
"""
manifest = self.import_manifest()
release_data_list = self.github_releases(manifest)
if release_data_list is not None:
current_version = utils.VERSION
last_version = release_data_list[0]['tag_name']
last_stable_version = release_data_list[0]['tag_name']
next_version = self.get_next_release(release_data_list)
manifest['version']['current'] = current_version
manifest['version']['next'] = next_version
manifest['version']['latest'] = last_version
manifest['version']['installed'] = True
self.save_manifest(manifest)
# need to have the task take at least 1 second to register the time
time.sleep(1)
def import_manifest(self):
"""
Loads the manifest for cabernet
"""
json_settings = self.plugin_db.get_repos(utils.CABERNET_ID)
if json_settings:
json_settings = json_settings[0]
return json_settings
def load_manifest(self):
"""
Loads the cabernet manifest from DB
"""
manifest_list = self.plugin_db.get_repos(utils.CABERNET_ID)
if manifest_list is None:
return None
else:
return manifest_list[0]
def save_manifest(self, _manifest):
"""
Saves to DB the manifest for cabernet
"""
self.plugin_db.save_repo(_manifest)
def github_releases(self, _manifest):
url = ''.join([
_manifest['dir']['github_repo_' + self.config['main']['upgrade_quality']],
'/releases'
])
return self.get_uri_data(url, 2)
@handle_json_except
@handle_url_except
def get_uri_data(self, _uri, _retries):
header = {'Content-Type': 'application/json',
'User-agent': utils.DEFAULT_USER_AGENT}
req = urllib.request.Request(_uri, headers=header)
with urllib.request.urlopen(req, timeout=10.0) as resp:
x = json.load(resp)
return x
def get_next_release(self, release_data_list):
current_version = self.config['main']['version']
cur_version_float = utils.get_version_index(current_version)
next_version_int = (int(cur_version_float/100)+2)*100
prev_version = release_data_list[0]['tag_name']
data = None
for data in release_data_list:
version_float = utils.get_version_index(data['tag_name'])
if version_float < next_version_int:
break
prev_version = data['tag_name']
return prev_version
def get_stable_release(self, release_data_list):
"""
Get the latest stable release with the format z.y.x.w without additional text...
"""
pass
def upgrade_app(self, _web_status):
"""
Initial request to perform an upgrade
"""
c_manifest = self.load_manifest()
if c_manifest is None:
self.logger.info('Cabernet manifest not found, aborting')
_web_status.data += 'Cabernet manifest not found, aborting<br>\r\n'
return False
if not c_manifest['version'].get('next'):
return False
if c_manifest['version'].get('next') == c_manifest['version'].get('current'):
self.logger.info('Cabernet is on the current version, not upgrading')
_web_status.data += 'Cabernet is on the current version, not upgrading<br>\r\n'
return False
# This checks to see if additional files or folders are in the
# basedir area. if so, abort upgrade.
# It is basically for the case where we have the wrong directory
_web_status.data += 'Checking current install area for expected files...<br>\r\n'
if not self.check_expected_files(_web_status):
return False
b = backups.Backups(self.plugins)
# recursively check all folders from the basedir to see if they are writable
_web_status.data += 'Checking write permissions...<br>\r\n'
resp = b.check_code_write_permissions()
if resp is not None:
_web_status.data += resp
return False
# simple call to run a backup of the data and source
# use a direct call to the backup methods instead of calling the scheduler
_web_status.data += 'Creating backup of code and data...<br>\r\n'
if not b.backup_all():
_web_status.data += 'Backup failed, aborting upgrade<br>\r\n'
return False
_web_status.data += 'Downloading new version from website...<br>\r\n'
if not self.download_zip('/'.join([
c_manifest['dir']['github_repo_' + self.config['main']['upgrade_quality']],
'zipball', c_manifest['version']['next']
]), 2):
_web_status.data += 'Download of the new version failed, aborting upgrade<br>\r\n'
return False
# skip integrity checks using SHA256 or SHA512 for now
# Unzips the downloaded file to a temp area and check the version
# contained in the utils.py that it is the same as expected.
_web_status.data += 'Extracting zip...<br>\r\n'
# folder is relative to tmp folder
unpacked_code = self.extract_code()
if unpacked_code is None:
_web_status.data += 'Extracting from zip failed, aborting upgrade<br>\r\n'
return False
# Deletes the non-data and non-plugin files
# maybe save the pycache folders?
# this helps in case a file has no modify permission.
# it can still be removed and added.
# *.py, *.html, *.js, *.png, ...
_web_status.data += 'Deleting old code...<br>\r\n'
if b.delete_code() is None:
_web_status.data += 'Deleting old files failed, aborting upgrade<br>\r\n'
return False
# does a move of the unzipped files to the source area
_web_status.data += 'Moving new code in place...<br>\r\n'
b.restore_code(unpacked_code)
return True
def check_expected_files(self, _web_status):
"""
Check the base directory files to see if all are expected.
"""
files_present = ['build', 'lib', 'misc',
'.dockerignore', '.gitignore', 'CHANGELOG.md', 'CONTRIBUTING.md',
'Docker_entrypoint.sh', 'Dockerfile', 'Dockerfile_tvh_crypt.alpine',
'Dockerfile_tvh_crypt.slim-buster', 'LICENSE', 'README.md',
'TVHEADEND.md', 'docker-compose.yml', 'requirements.txt', 'tvh_main.py',
'data', 'config.ini', 'is_container', '.git', 'cabernet.url', 'ffmpeg',
'README.txt', 'uninst.exe']
files_present.extend([self.config['paths']['internal_plugins_pkg'], self.config['paths']['external_plugins_pkg']])
filelist = [os.path.basename(x) for x in
glob.glob(os.path.join(self.config['paths']['main_dir'], '*'))]
response = True
for file in filelist:
if file not in files_present:
_web_status.data += '#### Extra file(s) found in install directory, aborting upgrade. FILE: {}<br>\r\n'\
.format(file)
response = False
return response
@handle_json_except
@handle_url_except
def download_zip(self, _zip_url, _retries):
buf_size = 2 * 16 * 16 * 1024
save_path = pathlib.Path(self.config['paths']['tmp_dir']).joinpath(TMP_ZIPFILE)
h = {'Content-Type': 'application/zip', 'User-agent': utils.DEFAULT_USER_AGENT}
req = urllib.request.Request(_zip_url, headers=h)
with urllib.request.urlopen(req) as resp:
with open(save_path, 'wb') as out_file:
while True:
chunk = resp.read(buf_size)
if not chunk:
break
out_file.write(chunk)
return True
def extract_code(self):
try:
file_to_extract = pathlib.Path(self.config['paths']['tmp_dir']).joinpath(TMP_ZIPFILE)
out_folder = pathlib.Path(self.config['paths']['tmp_dir']).joinpath('code')
with zipfile.ZipFile(file_to_extract, 'r') as z:
files = z.namelist()
top_folder = files[0]
z.extractall(out_folder)
return pathlib.Path('code', top_folder)
except (zipfile.BadZipFile, FileNotFoundError):
return None
def cleanup_tmp(self):
dir_ = self.config['paths']['tmp_dir']
for files in os.listdir(dir_):
path = os.path.join(dir_, files)
try:
shutil.rmtree(path)
except OSError:
os.remove(path)
|