Browse Source

Check productionpath safety, remove empty parent folders, implement delete project

fixes #3
master
Burathar 3 years ago
parent
commit
7c58c165fd
  1. 75
      biscd/biscd/models/project.py
  2. 15
      biscd/biscd/models/yaml_serializable.py
  3. 10
      biscd/biscd/routes.py

75
biscd/biscd/models/project.py

@ -4,7 +4,7 @@ from shutil import rmtree
import string import string
import random import random
from werkzeug.utils import secure_filename from werkzeug.utils import secure_filename
from git import Repo, InvalidGitRepositoryError, GitCommandError, cmd as git_cmd from git import Repo, InvalidGitRepositoryError, GitCommandError
from flask import current_app as app from flask import current_app as app
from biscd import config from biscd import config
@ -42,20 +42,28 @@ class Project(YamlSerializable):
return hash(self.name) return hash(self.name)
@property @property
def absulute_path(self): def absolute_production_path(self):
if not hasattr(self, 'relative_production_path') or not self.relative_production_path: if not hasattr(self, 'relative_production_path') or not self.relative_production_path:
return None return None
return Path(config['production_path'].get()) / self.relative_production_path production_path = Path(config['production_path'].get())
absolute_path = (production_path / self.relative_production_path).resolve()
try:
if str(absolute_path.relative_to(production_path)) != self.relative_production_path:
raise ValueError
except ValueError as v_e:
raise ValueError(f"{self.name}'s relative_production_path is not within the " \
f"application's production path! {self.relative_production_path}") from v_e
return absolute_path
def update(self): def update(self):
"""Update project's local repoistory""" """Update project's local repoistory"""
# pylint: disable=no-member # pylint: disable=no-member
self._make_sure_production_path_exists() self._make_sure_production_path_exists()
path = self.absulute_path path = self.absolute_production_path
app.logger.info(f"Updating {path}") app.logger.info("Updating project '%s' in %s", self.name, path)
try: try:
repo = Repo(path) repo = Repo(path)
app.logger.info('Already under git control') app.logger.info('%s was already under git control', self.name)
except InvalidGitRepositoryError: except InvalidGitRepositoryError:
repo = self._clone_repo(path) repo = self._clone_repo(path)
# If repo is not type Repo it must be an error and thus returned directly # If repo is not type Repo it must be an error and thus returned directly
@ -63,7 +71,7 @@ class Project(YamlSerializable):
return repo return repo
remotes = repo.remotes remotes = repo.remotes
if not any(remotes): if not any(remotes):
app.logger.error(f"Repo {path} doesn't have any remotes.") app.logger.error("%s's repo %s doesn't have any remotes.", self.name, path)
return ["Repo doesn't have any remotes. please fix manually", 'error'] return ["Repo doesn't have any remotes. please fix manually", 'error']
response = remotes[0].pull() response = remotes[0].pull()
for item in response: for item in response:
@ -95,40 +103,65 @@ class Project(YamlSerializable):
'error'] 'error']
return None return None
def delete(self):
if self.production_path_exists():
result = self.delete_files()
if result and result[1] in ['warning', 'error']:
return result
super().delete()
return [f"{self.name} has been deleted", 'success']
def delete_files(self): def delete_files(self):
if not hasattr(self, 'relative_production_path') or not self.relative_production_path:
return ['Project does not have a production path', 'warning']
if self.production_path_exists(): if self.production_path_exists():
path = self.absulute_path path = self.absolute_production_path
app.logger.info(f"Deleting {path}") app.logger.info(f"Deleting {path}")
rmtree(path) rmtree(path)
path = Path(self.relative_production_path).parent
while len(path.parts) > 0 and self._delete_dir_if_empty(path):
path = path.parent
self.relative_production_path = None self.relative_production_path = None
self.save() self.save()
@classmethod
def _delete_dir_if_empty(cls, relative_path):
production_path = config['production_path'].get()
absolute_path = (production_path / relative_path).resolve()
if absolute_path.is_dir() and not any(absolute_path.iterdir()):
if absolute_path.relative_to(production_path) == relative_path:
absolute_path.rmdir()
app.logger.debug('Deleted empty dir %s', absolute_path)
return True
raise ValueError("Trying to delete a directory outdize of production path! : "
+ str(absolute_path))
return False
def production_path_exists(self): def production_path_exists(self):
"""Return True if project's the production directory path exists""" """Return True if project's the production directory path exists"""
prod_path_dir = Path(config['production_path'].get())
if hasattr(self, 'relative_production_path') and self.relative_production_path: if hasattr(self, 'relative_production_path') and self.relative_production_path:
full_path = prod_path_dir / self.relative_production_path absolute_path = self.absolute_production_path
return full_path.is_dir() return absolute_path.is_dir()
return False return False
def _make_sure_production_path_exists(self): def _make_sure_production_path_exists(self):
prod_path_dir = Path(config['production_path'].get())
if hasattr(self, 'relative_production_path') and self.relative_production_path: if hasattr(self, 'relative_production_path') and self.relative_production_path:
full_path = prod_path_dir / self.relative_production_path absolute_path = self.absolute_production_path
if not full_path.is_dir(): if not absolute_path.is_dir():
full_path.mkdir(parents=True) absolute_path.mkdir(parents=True)
else: else:
def generate_full_path(): prod_path_dir = Path(config['production_path'].get())
def generate_absolute_path():
rand_num = ''.join(random.choices(string.digits, k=6)) rand_num = ''.join(random.choices(string.digits, k=6))
return prod_path_dir / (secure_filename(self.name) + rand_num) return prod_path_dir / (secure_filename(self.name) + rand_num)
full_path = generate_full_path() absolute_path = generate_absolute_path()
# Make sure path is unique # Make sure path is unique
while full_path.is_dir(): while absolute_path.is_dir():
full_path = generate_full_path() absolute_path = generate_absolute_path()
full_path.mkdir(parents=True) absolute_path.mkdir(parents=True)
self.relative_production_path = str(full_path.relative_to(prod_path_dir)) self.relative_production_path = str(absolute_path.relative_to(prod_path_dir))
self.save() self.save()
def user_access(self, user): def user_access(self, user):

15
biscd/biscd/models/yaml_serializable.py

@ -1,5 +1,6 @@
from abc import ABCMeta, abstractmethod from abc import ABCMeta, abstractmethod
from flask import abort from flask import abort
from git import exc
import yaml import yaml
import inspect import inspect
@ -52,6 +53,20 @@ class YamlSerializable(RecursiveProperty):
print(ymlsls) print(ymlsls)
self._save_all_to_file(ymlsls) self._save_all_to_file(ymlsls)
def delete(self):
# pylint: disable=no-member
if self.name is None:
raise TypeError("Name cannot be None")
ymlsls = self._get_all_from_file()
try:
ymlsl = next((ymlsl for ymlsl in ymlsls if [*ymlsl][0] == self.name))
ymlsls.remove(ymlsl)
#ymlsl = next((ymlsl for ymlsl in ymlsls if [*ymlsl][0] == self.name), None)
except KeyError as key_error:
raise ValueError(
f"{type(self).__name__} {self.name} does not exist") from key_error
self._save_all_to_file(ymlsls)
@classmethod @classmethod
def first_or_404(cls, **kwargs): def first_or_404(cls, **kwargs):
ymlsl = cls.first(**kwargs) ymlsl = cls.first(**kwargs)

10
biscd/biscd/routes.py

@ -122,3 +122,13 @@ def project_delete_files(project_name):
result = project.delete_files() result = project.delete_files()
flash_result(result) flash_result(result)
return redirect(url_for('project_dashboard', project_name=project.name)) return redirect(url_for('project_dashboard', project_name=project.name))
@app.route('/project/<project_name>/delete', methods=['GET'])
@login_required
def project_delete(project_name):
project = Project.first_or_404(name=project_name)
if project.user_access(current_user) != 'Owner':
abort(401)
result = project.delete()
flash_result(result)
return redirect(url_for('index'))

Loading…
Cancel
Save