474 lines
13 KiB
Python
474 lines
13 KiB
Python
import os
|
||
import requests
|
||
from pymongo import MongoClient
|
||
from bson.objectid import ObjectId
|
||
import datetime
|
||
|
||
_var = "/opt/legendary-octo-garbanzo"
|
||
_cmd = "manage-tenants.sh"
|
||
|
||
MONGO_URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017")
|
||
MONGO_DB_NAME = os.environ.get("MONGO_DB_NAME", "Invario_Website")
|
||
|
||
|
||
def _get_users_collection():
|
||
client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=1500)
|
||
db = client[MONGO_DB_NAME]
|
||
return client, db["packages"]
|
||
|
||
|
||
def add_dns(name: str, port: int) -> bool:
|
||
"""
|
||
Adds the subdomain to the DNS Routes
|
||
|
||
Input:
|
||
- name (Name of the subdomain) -> String
|
||
- port (Port number for the subdomain) -> int
|
||
|
||
Output:
|
||
- bool (True if adding of the Subdomain is active)
|
||
"""
|
||
# This is a placeholder function, as the actual implementation would depend on the DNS provider and how the DNS records are managed.
|
||
return False
|
||
|
||
def clear_special(var_:str) -> str:
|
||
"""
|
||
Clears the variable of any special carakters
|
||
|
||
Input:
|
||
- var -> String
|
||
|
||
Output:
|
||
- str cleared of the speacial carakters
|
||
"""
|
||
# Refactor the name varible to match the subdomain requirements
|
||
try:
|
||
var_ = var_.lower()
|
||
special_caracters1 = '^°!"§$%&/()=?\ß{[]}´`*~#,;.:<>|@€µ'
|
||
special_caracters2 = 'äüö'
|
||
for char in special_caracters1:
|
||
var_ = var_.replace(char, "")
|
||
for i in special_caracters2:
|
||
match i:
|
||
case "ü":
|
||
var_ = var_.replace(i, "ue")
|
||
case "ä":
|
||
var_ = var_.replace(i, "ae")
|
||
case "ö":
|
||
var_ = var_.replace(i, "oe")
|
||
except:
|
||
return False
|
||
|
||
def execute_script(wd_: str, file_: str, com_: str, com2_: str="None", com3_: str="None"):
|
||
"""
|
||
executes a script with the option of to extra inputs
|
||
|
||
Input:
|
||
- wd_ = working directory of the Inventorysystem -> String
|
||
- file_ = working file that youre targeting -> String
|
||
- com_ = first option -> String
|
||
- com2_ = second option (Optional if needet)-> String
|
||
- com3_ = third option (Optional if needet)-> String
|
||
|
||
Output:
|
||
- ether False if failed -> bool
|
||
- or result.stdout output of the executed process -> str
|
||
"""
|
||
import subprocess
|
||
update_path = os.path.join(wd_, file_)
|
||
if not update_path:
|
||
return False
|
||
if com2_ != "None":
|
||
cmd = f'bash "{update_path}" {com_} {com2_}'
|
||
elif com3_ != "None":
|
||
cmd = f'bash "{update_path}" {com_} {com2_} {com3_}'
|
||
else:
|
||
cmd = f'bash "{update_path}" {com_}'
|
||
try:
|
||
result = subprocess.run(
|
||
["bash", "-lc", cmd],
|
||
capture_output=True,
|
||
text=True,
|
||
cwd=wd_,
|
||
)
|
||
except:
|
||
return False
|
||
return result.stdout
|
||
|
||
class versions:
|
||
"""
|
||
This will give access to anything like:
|
||
- Version of the Inventorysystem
|
||
- Version of the Server
|
||
|
||
modules:
|
||
- inventorysystem()
|
||
- server()
|
||
"""
|
||
def inventorysystem(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return instace.edit(name, "inventarsystem")
|
||
|
||
def bibliothek(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return instace.edit(name, "buecherei")
|
||
|
||
def terminplanen(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return instace.edit(name, "terminverwaltung")
|
||
|
||
def emailversand(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return instace.edit(name, "emailversand")
|
||
|
||
def starter(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return instace.edit(name, "starter")
|
||
|
||
def advanced(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
return instace.edit(name, "advanced")
|
||
|
||
def testversion(name: str) -> str:
|
||
"""
|
||
Version of the Inventorysystem
|
||
|
||
Output:
|
||
- version -> String
|
||
"""
|
||
client = None
|
||
try:
|
||
client, packages = _get_users_collection()
|
||
packages.insert_one(
|
||
{
|
||
'client_name': name,
|
||
'inventarsystem': False,
|
||
'buecherei': False,
|
||
'terminverwaltung': False,
|
||
'emailversand': False,
|
||
'starter': False,
|
||
'advanced': False,
|
||
'registered_on': datetime.datetime.utcnow()
|
||
}
|
||
)
|
||
return True
|
||
finally:
|
||
if client:
|
||
client.close()
|
||
return instace.edit(name, "starter")
|
||
|
||
def _get_auth(name: str) -> int:
|
||
"""
|
||
Check if the if it is older than 5 days Test Version
|
||
|
||
Output:
|
||
- days since registration -> int
|
||
"""
|
||
client = None
|
||
try:
|
||
client, packages = _get_users_collection()
|
||
package = packages.find_one({'client_name': name}) or {}
|
||
for key in package:
|
||
if key == "registered_on":
|
||
if package[key]:
|
||
instance_age = (datetime.datetime.utcnow() - package[key]).days
|
||
else:
|
||
instance_age = None
|
||
finally:
|
||
if client:
|
||
client.close()
|
||
return int(instance_age) if int(instance_age) is not None and instance_age <= 5 else False
|
||
|
||
|
||
class instace:
|
||
"""
|
||
This will give access to anything like:
|
||
- Instances for Clients
|
||
- starting
|
||
- stopping
|
||
- restarting
|
||
- list all Clients
|
||
|
||
modules:
|
||
- new(name:str)
|
||
- remove(name:str)
|
||
- status(name:str)
|
||
- restart(name:str)
|
||
- list()
|
||
"""
|
||
|
||
def __init__():
|
||
return list()
|
||
|
||
def new_trial(name: str) -> bool:
|
||
"""
|
||
Generates a new trial instance with the subdomain [name].invario.eu
|
||
|
||
Input:
|
||
- name -> String
|
||
|
||
Output:
|
||
- bool if the creation of the instance works (True: creation worked; False: didnt work)
|
||
|
||
manage-tenant.sh trial <tenant_id> [port] [days]
|
||
"""
|
||
port_starter = 10002
|
||
port = port_starter
|
||
for i in instace.list():
|
||
port =+ 1
|
||
|
||
if execute_script(_var, _cmd, "trial", clear_special(name), port):
|
||
add_dns(name, port)
|
||
return int(port)
|
||
else:
|
||
return False
|
||
|
||
def new(name: str) -> int:
|
||
"""
|
||
Generates a new instance with the subdomain [name].invario.eu
|
||
|
||
Input:
|
||
- name -> String
|
||
|
||
Output:
|
||
- int (Port number for the subdomain) or False if the creation of the instance didnt work
|
||
"""
|
||
port_starter = 10002
|
||
port = port_starter
|
||
for i in instace.list():
|
||
port =+ 1
|
||
|
||
if execute_script(_var, _cmd, "add", clear_special(name), port):
|
||
add_dns(name, port)
|
||
return int(port)
|
||
else:
|
||
return False
|
||
|
||
def edit(name: str, module:str) -> bool:
|
||
"""
|
||
Edits the name of an instance with the subdomain [name].invario.eu to [new_name].invario.eu
|
||
|
||
Input:
|
||
- name -> String
|
||
- module -> String (Options: "inventarsystem", "buecherei", "terminverwaltung", "emailversand", "starter", "advanced")
|
||
|
||
Output:
|
||
- bool if the edit works (True: edit worked; False: didnt work)
|
||
"""
|
||
if module == "inventarsystem":
|
||
if execute_script(_var, _cmd, "module", clear_special(name), 'library=off inventory=on'):
|
||
return True
|
||
else:
|
||
return False
|
||
elif module == "buecherei":
|
||
if execute_script(_var, _cmd, "module", clear_special(name), 'library=on inventory=off'):
|
||
return True
|
||
else:
|
||
return False
|
||
elif module == "terminverwaltung":
|
||
if execute_script(_var, _cmd, "module", clear_special(name), 'library=off inventory=off'): # -> has to be changed when the module is ready
|
||
return True
|
||
else:
|
||
return False
|
||
elif module == "emailversand":
|
||
if execute_script(_var, _cmd, "module", clear_special(name), 'library=off inventory=on'):# -> has to be changed when the module is ready
|
||
return True
|
||
else:
|
||
return False
|
||
elif module == "starter":
|
||
if execute_script(_var, _cmd, "module", clear_special(name), 'library=on inventory=on'):
|
||
return True
|
||
else:
|
||
return False
|
||
elif module == "advanced":
|
||
if execute_script(_var, _cmd, "module", clear_special(name), 'library=on inventory=on'):# -> has to be changed when the module is ready
|
||
return True
|
||
else:
|
||
return False
|
||
else:
|
||
return False
|
||
|
||
|
||
def remove(name: str) -> bool:
|
||
"""
|
||
Removes a instance with the subdomain [name].invario.eu
|
||
|
||
Input:
|
||
- name -> String
|
||
|
||
Output:
|
||
- bool if the removal works (True: removal worked; False: didnt work)
|
||
"""
|
||
safe_name = clear_special(name)
|
||
# try to remove via management script
|
||
script_ok = execute_script(_var, _cmd, "remove", safe_name)
|
||
if not script_ok:
|
||
return False
|
||
|
||
# best-effort: remove the associated package entry from the database
|
||
client = None
|
||
try:
|
||
client, packages = _get_users_collection()
|
||
packages.delete_one({"client_name": name})
|
||
except Exception:
|
||
# don't fail the overall removal if DB cleanup fails
|
||
pass
|
||
finally:
|
||
if client:
|
||
client.close()
|
||
|
||
return True
|
||
|
||
|
||
def status(name: str) -> bool:
|
||
"""
|
||
Returns if a instance with the subdomain [name].invario.eu is up.
|
||
|
||
Input:
|
||
- name -> String
|
||
|
||
Output:
|
||
- bool if the page is online (True: Is working; False: Isnt online)
|
||
"""
|
||
name = clear_special(name)
|
||
request = str(requests.get(f"{name}.invario.eu/test_connection"))
|
||
if request == '{"message":"Connection successful","status":"success","status_code":200}':
|
||
return True
|
||
else:
|
||
return False
|
||
|
||
def restart(name: str) -> bool:
|
||
"""
|
||
Restart an instance with the subdomain [name].invario.eu
|
||
|
||
Input:
|
||
- name -> String
|
||
|
||
Output:
|
||
- bool if the restart works (True: restart worked; False: didnt work)
|
||
"""
|
||
if execute_script(_var, _cmd, "restart-tenant", clear_special(name)):
|
||
return True
|
||
else:
|
||
return False
|
||
|
||
def list() -> list:
|
||
"""
|
||
List off all tenants.
|
||
|
||
Output:
|
||
- list with all tenants ("tenant1", "tenant2")
|
||
"""
|
||
result = execute_script(_var, _cmd, "list")
|
||
if not isinstance(result, str):
|
||
return []
|
||
result = result.splitlines()
|
||
if not result:
|
||
return []
|
||
return [line.replace("- ", "") for line in result[1:]]
|
||
|
||
def backup(name: str) -> bool:
|
||
"""
|
||
Creates a backup of an instance with the subdomain [name].invario.eu
|
||
|
||
Input:
|
||
- name -> String
|
||
|
||
Output:
|
||
- bool if the backup works (True: backup worked; False: didnt work)
|
||
"""
|
||
import subprocess
|
||
try:
|
||
subprocess.run(
|
||
[
|
||
"docker",
|
||
"compose",
|
||
"-f",
|
||
"docker-compose-multitenant.yml",
|
||
"exec",
|
||
"-T",
|
||
"mongodb",
|
||
"mongodump",
|
||
f"--archive=/data/backups/inventar_{name}-$(date +%Y%m%d%H%M%S).gz",
|
||
"--gzip",
|
||
"--db",
|
||
f"inventar_{name}"
|
||
],
|
||
check=True
|
||
)
|
||
return True
|
||
except subprocess.CalledProcessError:
|
||
return False
|
||
|
||
class ussage:
|
||
"""
|
||
This will give informations about anything like:
|
||
- RAM Ussage of the server
|
||
- CPU Ussage of the server
|
||
- Strorage that is in use
|
||
|
||
modules:
|
||
- ram()
|
||
- cpu()
|
||
- storage()
|
||
"""
|
||
def ram() -> int:
|
||
"""
|
||
RAM ussage of the complete system
|
||
|
||
Output:
|
||
- ram ussage -> interger in GB
|
||
"""
|
||
#print("RAM usage (%):", ram.percent)
|
||
import psutil
|
||
ram = psutil.virtual_memory()
|
||
return int(round(ram.used / 1e9, 2))
|
||
|
||
def cpu() -> int:
|
||
"""
|
||
System cpu ussage
|
||
|
||
Output:
|
||
- cpu ussage -> integer in Percent
|
||
"""
|
||
import psutil
|
||
return int(psutil.cpu_percent(interval=1))
|
||
|
||
def storage() -> int:
|
||
"""
|
||
System storage ussage
|
||
|
||
Output:
|
||
- storager ussage -> integer in Percent
|
||
"""
|
||
pass |