All checks were successful
continuous-integration/drone/push Build is passing
98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
|
|
class Executor:
|
|
"""This class executes commands recieved by the server"""
|
|
|
|
ABSOLUTE_PATH = os.path.abspath(os.path.dirname(sys.argv[0]))
|
|
BASE_PATH = ABSOLUTE_PATH + os.path.sep + 'home' + os.path.sep
|
|
DEFAULT_FOLDER = BASE_PATH + '0'
|
|
|
|
def __init__(self, currentDiectory: str, baseDir: str = ""):
|
|
self.currentDirectory = currentDiectory
|
|
if baseDir == "":
|
|
self.baseDir = self.DEFAULT_FOLDER
|
|
else:
|
|
self.baseDir = self.BASE_PATH + baseDir + os.path.sep
|
|
|
|
|
|
def sanitizeDirectory(self, inDirectory: str) -> str:
|
|
return re.sub('[^a-zA-Z0-9]', '', inDirectory)
|
|
|
|
def sanitizeFile(self, inFile: str) -> str:
|
|
return re.sub('[^a-zA-Z0-9].[^a-zA-Z0-9]', '', inFile)
|
|
|
|
def createDirectory(self, dirName: str) -> str:
|
|
dirName = self.sanitizeDirectory(dirName)
|
|
actualDirName: str = os.path.join(self.baseDir,self.currentDirectory,dirName)
|
|
os.mkdir(actualDirName)
|
|
return actualDirName
|
|
|
|
def removeDirectory(self, dirName: str) -> str:
|
|
dirName = self.sanitizeDirectory(dirName)
|
|
actualDirName: str = os.path.join(self.baseDir, self.currentDirectory, dirName)
|
|
if actualDirName:
|
|
os.rmdir(actualDirName)
|
|
return actualDirName
|
|
|
|
def getCurrentDirectory(self) -> str:
|
|
return self.currentDirectory
|
|
|
|
def setCurrentDirectory(self, dirName: str) -> str:
|
|
if dirName == "..":
|
|
if self.currentDirectory == self.baseDir:
|
|
return self.currentDirectory
|
|
else:
|
|
directories = self.currentDirectory.split(os.path.sep)
|
|
strdirectory = ""
|
|
for dir in directories:
|
|
strdirectory += dir + os.path.sep
|
|
strdirectory = strdirectory[:-3]
|
|
if os.path.exists(strdirectory):
|
|
self.currentDirectory = strdirectory
|
|
return self.currentDirectory
|
|
else:
|
|
dirName = self.sanitizeDirectory(dirName)
|
|
joinedDir = os.path.join(self.currentDirectory, dirName)
|
|
if os.path.exists(joinedDir):
|
|
self.currentDirectory = joinedDir
|
|
return self.currentDirectory
|
|
|
|
def listCurrentDirectoryContent(self) -> str:
|
|
contents = os.listdir(self.currentDirectory)
|
|
strdirectory = ""
|
|
for content in contents:
|
|
strdirectory += content + ", "
|
|
strdirectory = strdirectory[:-1]
|
|
return strdirectory
|
|
|
|
def putFileInCurrentDirectory(self, filename: str, content: bytes) -> str:
|
|
filename = self.sanitizeFile(filename)
|
|
currenctfile = os.path.join(self.currentDirectory, filename)
|
|
f = open(currenctfile, "wb")
|
|
f.write(content)
|
|
f.close()
|
|
return currenctfile
|
|
|
|
def getFileInCurrentDirectory(self, file: str) -> bytes:
|
|
file = self.sanitizeFile(file)
|
|
currentfile = os.path.join(self.currentDirectory, file)
|
|
if os.path.exists(currentfile):
|
|
f = open(currentfile, "rb")
|
|
content = f.read()
|
|
f.close()
|
|
return content
|
|
else:
|
|
raise Exception('File not found')
|
|
|
|
def removeFileInCurrentDirectory(self, file: str) -> str:
|
|
file = self.sanitizeFile(file)
|
|
currentfile = os.path.join(self.currentDirectory, file)
|
|
if os.path.exists(currentfile):
|
|
os.remove(currentfile)
|
|
return currentfile
|