improve documentation and fixed/improved path handling and added copy files from local->pve(->lxc)

This commit is contained in:
Mathieu Broillet 2023-06-13 15:29:22 +02:00
parent 42dc7ba207
commit 00f14332e0
No known key found for this signature in database
GPG Key ID: 7D4F25BC50A0AA32
7 changed files with 426 additions and 133 deletions

View File

@ -7,7 +7,8 @@ from ..utils import proxmox_utils
class LXC: class LXC:
"""LXC object""" """LXC object"""
def __init__(self, lxc_id, lxc_hostname, os, resources, network, options, creation, deploy): def __init__(self, lxc_id: int, lxc_hostname: str, os: dict, resources: dict, network: dict, options: dict,
creation: dict, deploy: dict):
self.lxc_id = lxc_id self.lxc_id = lxc_id
self.lxc_hostname = lxc_hostname self.lxc_hostname = lxc_hostname
@ -52,9 +53,12 @@ class LXC:
return hash(self.lxc_id) return hash(self.lxc_id)
def get_id(self): def get_id(self):
""" """Get LXC ID
Get LXC ID
:return: lxc id Returns
-------
int
LXC ID
""" """
return self.lxc_id return self.lxc_id

View File

@ -1,10 +1,11 @@
from pathlib import Path from pathlib import Path
from . import LXC
from ..utils import proxmox_utils from ..utils import proxmox_utils
from ..utils.resources_utils import get_path from ..utils.resources_utils import get_path
def run_script(lxc, step): def run_script(lxc: LXC, step: dict):
"""Method use to dispatch the script to the correct method depending on the type of script, being """Method use to dispatch the script to the correct method depending on the type of script, being
local (on the machine running this program), remote (url) or in the LXC. local (on the machine running this program), remote (url) or in the LXC.
@ -13,7 +14,7 @@ def run_script(lxc, step):
lxc : LXC lxc : LXC
The LXC object used to run the script The LXC object used to run the script
step: dict step: dict
Dictionary containing the step information about the script to run Dictionary containing the step information and configuration about the script to run
Typically read from the "creation/steps/<step>" in JSON file Typically read from the "creation/steps/<step>" in JSON file
""" """
@ -23,8 +24,9 @@ def run_script(lxc, step):
install_package(lxc, "bash") install_package(lxc, "bash")
# Run local script # Run local script
if "path" in step: if "local_path" in step:
_run_local_script_on_lxc(lxc, step["path"]) path = get_path(lxc, step["local_path"])
_run_local_script_on_lxc(lxc, path)
# Run remote script # Run remote script
elif "url" in step: elif "url" in step:
@ -32,33 +34,78 @@ def run_script(lxc, step):
# Run script in LXC # Run script in LXC
elif "lxc_path" in step: elif "lxc_path" in step:
_run_script_on_lxc(lxc, step["lxc_path"]) path = get_path(lxc, step["lxc_path"])
_run_script_on_lxc(lxc, path)
def _run_script_on_lxc(lxc, path: "Path inside the LXC"): def _run_script_on_lxc(lxc: LXC, path: Path):
"""Run a script in the LXC""" """Runs a script present inside the LXC storage, without copying or downloading it
lxc.run_command(f"bash {path}") Parameters
----------
lxc: LXC
The LXC object used to run the script
path: pathlib.Path
Path to the script inside the LXC storage
"""
lxc.run_command(f"bash {str(path)}")
def _run_local_script_on_lxc(lxc, path: "Path to the script on the machine running this program"): def _run_local_script_on_lxc(lxc: LXC, path: Path):
"""Run a local script in the LXC""" """Runs a script present on the machine running this program, inside the LXC
It works by copying the script from the local machine to the PVE using SCP and
then using "pct push" to send it to the LXC.
path = get_path(lxc, path) Parameters
with open(path, "r") as file: ----------
script = file.read() lxc: LXC
lxc.run_command("bash <<EOF\n" + script + "\nEOF") The LXC object used to run the script
path: pathlib.Path
Path object to the script on the local machine (machine running this program)
"""
# Copy the script to /tmp and run it from there
proxmox_utils.copy_file_to_lxc(lxc.get_id(), str(path), "/tmp/pdj-scripts/")
lxc.run_command(f"bash /tmp/pdj-scripts/{path.name}")
# with open(path, "r") as file:
# script = file.read()
#
# # TODO : Might work with EOF instead of copying the script to /tmp but unable to make it work
# # lxc.run_command("bash <<EOF\n" + script + "\nEOF")
def _run_remote_script_on_lxc(lxc, url: "URL to the script"): def _run_remote_script_on_lxc(lxc: LXC, url: str):
"""Run a remote script in the LXC""" """Runs a script from a URL on a LXC
Usually this should not be called directly but rather through the run_script() method with the
according step configuration (local_path, url or lxc_path) which is itself usually read from the JSON file.
Parameters
----------
lxc: LXC
The LXC object used to run the script
url: str
URL of the script to run
"""
if lxc.has_program("curl"): if lxc.has_program("curl"):
lxc.run_command(f"curl -sSL {url} | bash") lxc.run_command(f"curl -sSL {url} | bash")
def create_folder_in_lxc(lxc, path, permission=755): def create_folder_in_lxc(lxc: LXC, path: str, permission=755):
"""Create a folder in the LXC""" """Create a folder in the LXC
It's possible to specify the permission of the folder, by default it uses 755
Parameters
----------
lxc: LXC
The LXC object of the LXC to create the folder in
path: str
Path to the folder to create in the LXC
permission: int, optional
Permission of the folder to create in the LXC (default is 755)
"""
lxc.run_command(f"mkdir -p {path}") lxc.run_command(f"mkdir -p {path}")
@ -66,47 +113,72 @@ def create_folder_in_lxc(lxc, path, permission=755):
lxc.run_command(f"chmod -R {permission} {path}") lxc.run_command(f"chmod -R {permission} {path}")
def create_file_in_lxc(lxc, path, permission=644): def create_file_in_lxc(lxc: LXC, path: str, permission=644):
"""Create a file in the LXC""" """Create a file in the LXC
It's possible to specify the permission of the file, by default it uses 644
Parameters
----------
lxc: LXC
The LXC object of the LXC to create the file in
path: str
Path to the file to create in the LXC
permission: int, optional
Permission of the file to create in the LXC (default is 644)
"""
lxc.run_command(f"touch {path}") lxc.run_command(f"touch {path}")
if permission != 644: if permission != 644:
lxc.run_command(f"chmod {permission} {path}") lxc.run_command(f"chmod {permission} {path}")
def copy_local_file_to_lxc(lxc, path, destination): def remove_file_in_lxc(lxc: LXC, path: str):
"""Remove a file in the LXC
Parameters
----------
lxc: LXC
The LXC object of the LXC to remove the file in
path: str
Path to the file to remove in the LXC
"""
lxc.run_command(f"rm {path}")
def copy_local_file_to_lxc(lxc: LXC, path: Path, destination: str):
"""Copy a local file to the LXC """Copy a local file to the LXC
Parameters Parameters
---------- ----------
lxc : LXC lxc : LXC
The LXC object of the LXC to copy the file to The LXC object of the LXC to copy the file to
path: str path: pathlib.Path
Path to the file on the machine running this program Path object to the file on the machine running this program
destination: str destination: str
Path to the destination in the LXC Path to the destination in the LXC
""" """
proxmox_utils.run_command_locally(f"scp {get_path(lxc, path)} {lxc.get_ssh_string()}:{destination}") proxmox_utils.run_command_locally(f"scp {str(path)} {lxc.get_ssh_string()}:{destination}")
def copy_local_folder_to_lxc(lxc, path, destination): def copy_local_folder_to_lxc(lxc: LXC, path: Path, destination: str):
"""Copy a local folder to the LXC """Copy a local folder to the LXC
Parameters Parameters
---------- ----------
lxc : LXC lxc : LXC
The LXC object of the LXC to copy the folder to The LXC object of the LXC to copy the folder to
path: str path: pathlib.Path
Path to the folder on the machine running this program Path object to the folder on the machine running this program
destination: str destination: str
Path to the destination in the LXC Path to the destination in the LXC
""" """
proxmox_utils.run_command_locally(f"scp -r {get_path(lxc, path)} {lxc.get_ssh_string()}:{destination}") proxmox_utils.run_command_locally(f"scp -r {str(path)} {lxc.get_ssh_string()}:{destination}")
def run_docker_command(lxc, container, command): def run_docker_command(lxc: LXC, container: str, command: str):
"""Run a command inside a docker container on the LXC """Run a command inside a docker container on the LXC
Parameters Parameters
@ -117,11 +189,16 @@ def run_docker_command(lxc, container, command):
Name of the docker container to run the command in Name of the docker container to run the command in
command : str command : str
Command to run in the docker container Command to run in the docker container
Examples
--------
>>> run_docker_command(lxc, "<container-name>", "<command>")
""" """
lxc.run_command(f"docker exec -it {container} {command}") lxc.run_command(f"docker exec -it {container} {command}")
def run_docker_compose_command(lxc, command, working_directory=None): def run_docker_compose_command(lxc: LXC, command: str, working_directory: str = None):
"""Run a docker-compose command in the LXC """Run a docker-compose command in the LXC
Parameters Parameters
@ -130,13 +207,14 @@ def run_docker_compose_command(lxc, command, working_directory=None):
The LXC object of the LXC to run the docker-compose command in The LXC object of the LXC to run the docker-compose command in
command: str command: str
The docker-compose command to run The docker-compose command to run
working_directory: str working_directory: str, optional
The working directory to run the command in The working directory to run the command in
Examples Examples
-------- --------
>>> run_docker_compose_command(lxc, "up -d", "/home/user/traefik") >>> run_docker_compose_command(lxc, "up -d", "/home/user/traefik")
""" """
docker_compose_exec = "docker-compose" docker_compose_exec = "docker-compose"
if not lxc.has_program(docker_compose_exec): if not lxc.has_program(docker_compose_exec):
docker_compose_exec = "docker compose" docker_compose_exec = "docker compose"
@ -147,16 +225,33 @@ def run_docker_compose_command(lxc, command, working_directory=None):
lxc.run_command(f"{docker_compose_exec} {command}") lxc.run_command(f"{docker_compose_exec} {command}")
def download_file(lxc, url, destination): def download_file(lxc: LXC, url: str, destination: str):
"""Download a file from a URL to the LXC and save it to the destination
Parameters
----------
lxc: LXC
The LXC object of the LXC to download the file to
url: str
URL of the file to download
destination: str
Path to the destination to save the file to
Needs to end with a trailing slash
Examples
--------
>>> download_file(lxc, "https://example.com/file.zip", "/home/user/")
"""
"""Download a file from a URL to the LXC""" """Download a file from a URL to the LXC"""
if type(url) is list: if type(url) is list:
for u in url: for u in url:
lxc.run_command(f"wget {u} -O {destination}") lxc.run_command(f"wget {u} --directory-prefix={destination}")
else: else:
lxc.run_command(f"wget {url} -O {destination}") lxc.run_command(f"wget {url} --directory-prefix={destination}")
def unzip_file(lxc, path, destination=None): def unzip_file(lxc: LXC, path: str, destination: str = None):
"""Unzip a file in the LXC """Unzip a file in the LXC
Parameters Parameters
@ -165,13 +260,15 @@ def unzip_file(lxc, path, destination=None):
The LXC object of the LXC to unzip the file in The LXC object of the LXC to unzip the file in
path: str path: str
Path to the file to unzip Path to the file to unzip
destination: str destination: str, optional
Path to the destination to unzip the file to Path to the destination to unzip the file to
If not specified, it will unzip the file in the same directory as the file
Needs to end with a trailing slash
Examples Examples
-------- --------
>>> unzip_file(lxc, "/home/user/file.zip", "/home/user/destination") >>> unzip_file(lxc, "/home/user/file.zip", "/home/user/extracted_files/")
>>> unzip_file(lxc, "/home/user/file.tar.gz", "/home/user/destination") >>> unzip_file(lxc, "/home/user/file.tar.gz", "/home/user/extracted_files/")
""" """
if destination is None: if destination is None:
@ -180,10 +277,10 @@ def unzip_file(lxc, path, destination=None):
if ".zip" in path: if ".zip" in path:
lxc.run_command(f"unzip {path} -d {destination}") lxc.run_command(f"unzip {path} -d {destination}")
elif ".tar.gz" in path: elif ".tar.gz" in path:
lxc.run_command(f"tar -xzf {path} -C {destination}") lxc.run_command(f"mkdir -p {destination} && tar -xzf {path} --directory {destination}")
def install_package(lxc, package): def install_package(lxc: LXC, package: str or list):
"""Install a package in the LXC """Install a package in the LXC
Parameters Parameters
@ -191,7 +288,7 @@ def install_package(lxc, package):
lxc: LXC lxc: LXC
The LXC object of the LXC to install the package in The LXC object of the LXC to install the package in
package: str or list package: str or list
Name of the package to install Name(s) of the package(s) to install
Examples Examples
-------- --------
@ -206,7 +303,7 @@ def install_package(lxc, package):
lxc.run_command(f"{proxmox_utils.get_install_package_command(lxc.get_os_name())} {package}") lxc.run_command(f"{proxmox_utils.get_install_package_command(lxc.get_os_name())} {package}")
def remove_package(lxc, package): def remove_package(lxc: LXC, package: str or list):
"""Remove a package in the LXC """Remove a package in the LXC
Parameters Parameters
@ -214,7 +311,7 @@ def remove_package(lxc, package):
lxc: LXC lxc: LXC
The LXC object of the LXC to remove the package in The LXC object of the LXC to remove the package in
package: str or list package: str or list
Name of the package to remove Name(s) of the package(s) to remove
Examples Examples
-------- --------
@ -232,20 +329,20 @@ def remove_package(lxc, package):
lxc.run_command(f"{proxmox_utils.get_remove_package_command(lxc.get_os_name())} {package}") lxc.run_command(f"{proxmox_utils.get_remove_package_command(lxc.get_os_name())} {package}")
def replace_in_files(lxc, path, search, replace, case_sensitive=False): def replace_in_files(lxc: LXC, path: str or list, search: str, replace: str, case_sensitive: bool = False):
"""Replace a string in one or multiples files in the LXC """Replace a string in one or multiples files in the LXC
Parameters Parameters
---------- ----------
lxc : LXC lxc : LXC
The LXC object of the LXC to replace the string in The LXC object of the LXC to replace the string in
path : str or list path : str or list of str
Path to the file(s) to replace the string in Path to the file(s) to replace the string in
search : str search : str
String to search for String to search for
replace : str replace : str
String to replace the search string with String to replace the search string with
case_sensitive : bool case_sensitive : bool, optional
Whether the search should be case sensitive or not Whether the search should be case sensitive or not
Examples Examples
@ -258,6 +355,6 @@ def replace_in_files(lxc, path, search, replace, case_sensitive=False):
if type(path) is list: if type(path) is list:
for p in path: for p in path:
lxc.run_command(f"sed -i {'-i' if case_sensitive else ''} 's/{search}/{replace}/g' {p}") lxc.run_command(f"sed {'-i' if case_sensitive else ''} 's/{search}/{replace}/g' {p}")
else: else:
lxc.run_command(f"sed -i {'-i' if case_sensitive else ''} 's/{search}/{replace}/g' {path}") lxc.run_command(f"sed {'-i' if case_sensitive else ''} 's/{search}/{replace}/g' {path}")

View File

@ -1,10 +1,11 @@
import logging import logging
from pathlib import Path from pathlib import Path
from . import commands_utils from . import commands_utils, LXC
from ..utils.resources_utils import get_path
def are_all_conditions_met(lxc): def are_all_conditions_met(lxc: LXC):
""" """
Check conditions for running the creation steps for an LXC. Check conditions for running the creation steps for an LXC.
@ -43,28 +44,29 @@ def are_all_conditions_met(lxc):
return all(result) return all(result)
def optional(var, placeholder): def optional(var: any, placeholder: any):
"""Return a placeholder if the given variable is None, otherwise return the variable """Return a placeholder if the given variable is None, otherwise return the variable
Parameters Parameters
---------- ----------
var var: any
Variable to check if it is None Variable to check if it is None
placeholder placeholder: any
Placeholder to return if the variable is None Placeholder to return if the variable is None
Returns Returns
------- -------
var or placeholder any[any]
The variable if it is not None, otherwise the placeholder
""" """
if var is None: if var is None:
return placeholder return placeholder
else: else:
return var return var
def run_steps(lxc): def run_steps(lxc: LXC):
"""Run the creation steps for the given LXC """Run the creation steps for the given LXC
Parameters Parameters
@ -85,18 +87,18 @@ def run_steps(lxc):
case "file_create": case "file_create":
commands_utils.create_file_in_lxc(lxc, step["path"], optional(step["permission"], 644)) commands_utils.create_file_in_lxc(lxc, step["path"], optional(step["permission"], 644))
case "file_copy": case "file_copy":
commands_utils.copy_local_file_to_lxc(lxc, step["path"], step["destination"]) commands_utils.copy_local_file_to_lxc(lxc, get_path(lxc, step["path"]), step["destination"])
case "folder": case "folder":
commands_utils.create_folder_in_lxc(lxc, step["path"], optional(step["permission"], 755)) commands_utils.create_folder_in_lxc(lxc, step["path"], optional(step["permission"], 755))
case "folder_copy": case "folder_copy":
commands_utils.copy_local_folder_to_lxc(lxc, step["path"], step["destination"]) commands_utils.copy_local_folder_to_lxc(lxc, get_path(lxc, step["path"]), step["destination"])
case "command": case "command":
lxc.run_command(command=step["command"], working_directory=optional(step["working_directory"], None)) lxc.run_command(command=step["command"], working_directory=optional(step["working_directory"], None))
case "docker": case "docker":
commands_utils.run_docker_command(lxc, step["container"], step["command"]) commands_utils.run_docker_command(lxc, step["container"], step["command"])
case "docker_compose": case "docker_compose":
commands_utils.run_docker_compose_command(lxc, step["command"], commands_utils.run_docker_compose_command(lxc, step["command"],
optional(step["working_directory"], None)) optional(step["working_directory"], None))
case "git": case "git":
lxc.run_command(command=f"git clone {step['url']} {step['destination']}") lxc.run_command(command=f"git clone {step['url']} {step['destination']}")
case "download": case "download":

View File

@ -1,4 +1,5 @@
import json import json
from pathlib import Path
from . import LXC from . import LXC
@ -17,12 +18,12 @@ def get_all_lxcs():
return lxcs return lxcs
def get_lxc(lxc_id): def get_lxc(lxc_id: int):
"""Get LXC by ID """Get LXC by ID
Parameters Parameters
---------- ----------
lxc_id : str lxc_id : int
ID of the LXC to get ID of the LXC to get
Returns Returns
@ -38,23 +39,23 @@ def get_lxc(lxc_id):
return None return None
def load_lxc(file, lxc_id): def load_lxc(lxc_id: int, filepath: Path):
"""Load LXC from JSON file """Load LXC from JSON file
Parameters Parameters
---------- ----------
file : str lxc_id : int
Path to the JSON file of the LXC to load
lxc_id : str
ID of the LXC to load ID of the LXC to load
filepath : pathlib.Path
Path object to the JSON file of the LXC to load
Examples Examples
-------- --------
>>> load_lxc("./resources/lxc/100/config.json", "100") >>> load_lxc(100, Path("<full-path>/resources/lxc/100/config.json"))
""" """
# Load JSON data # Load JSON data
data = json.loads(file) data = json.loads(filepath.read_text())
# Extract values from JSON # Extract values from JSON
lxc_id = lxc_id lxc_id = lxc_id

View File

@ -1,5 +1,5 @@
import logging import logging
import os from pathlib import Path
from . import project_path from . import project_path
from .lxc.utils import load_lxc, get_all_lxcs from .lxc.utils import load_lxc, get_all_lxcs
@ -7,19 +7,17 @@ from .lxc.utils import load_lxc, get_all_lxcs
def run(): def run():
# Read all files in the resources directory # Read all files in the resources directory
resources = os.listdir(project_path / "resources") resources = Path(project_path).joinpath("resources").glob("*")
# Go through each LXC file # Go through each LXC file
for resource in resources: for resource in resources:
if resource == "lxc": if resource == "lxc":
# Read all files in the LXC directory # Read all files in the LXC directory
lxc_folders = os.listdir(project_path / "resources" / "lxc") lxc_folders = Path(project_path).joinpath("resources", "lxc").glob("*")
for lxc_folder in lxc_folders: for lxc_folder in lxc_folders:
lxc_file = os.path.join(project_path / "resources" / "lxc", lxc_folder, "config.json") lxc_file = Path(project_path).joinpath("resources", "lxc", lxc_folder, "config.json")
# Open the file
with open(lxc_file, "r") as file: load_lxc(filepath=lxc_file, lxc_id=int(lxc_folder.name))
# Load the LXC
load_lxc(file.read(), lxc_id=lxc_folder)
for lxc in get_all_lxcs(): for lxc in get_all_lxcs():
logging.info(f"Loading LXC {lxc.lxc_id}") logging.info(f"Loading LXC {lxc.lxc_id}")

View File

@ -1,56 +1,84 @@
import json import json
import logging import logging
import os import os
import pathlib
import subprocess import subprocess
from pathlib import Path
from .. import project_path from .. import project_path
def get_pve_version(): def get_pve_version():
""" """Get PVE version
Get PVE version
:return: pve version Returns
-------
str
PVE version
""" """
return run_command_on_pve(command="pveversion", warn_exit_status=True) return run_command_on_pve(command="pveversion", warn_exit_status=True)
def get_pve_hostname(): def get_pve_hostname():
""" """Get PVE hostname
Get PVE hostname
:return: pve hostname Returns
-------
str
PVE hostname
""" """
return run_command_on_pve(command="hostname", warn_exit_status=True) return run_command_on_pve(command="hostname", warn_exit_status=True)
def does_lxc_exist(lxc_id): def does_lxc_exist(lxc_id: int):
""" """Check if an LXC exists with the given ID
Check if LXC exists
:param lxc_id: lxc id Parameters
:return: does lxc exists ----------
lxc_id: int
Returns
-------
bool
Does LXC exist?
""" """
# TODO: only check in VMID column # TODO: only check in VMID column
return lxc_id in run_command_on_pve(command=f"pct list | awk '/{lxc_id}/'") return str(lxc_id) in run_command_on_pve(command=f"pct list | awk '/{lxc_id}/'")
def does_qemu_vm_exist(vm_id): def does_qemu_vm_exist(vm_id: int):
"""Check if a QEMU VM exists with the given ID
Parameters
----------
vm_id: int
Returns
-------
bool
Does QEMU VM exist?
""" """
Check if QEMU VM exists
:param vm_id: vm id
:return: does qemu vm exists
"""
# TODO: only check in VMID column # TODO: only check in VMID column
return vm_id in run_command_on_pve(command=f"qm list {vm_id}") return str(vm_id) in run_command_on_pve(command=f"qm list {vm_id}")
def execute_tteck_script(script_url, env_variables): def execute_tteck_script(script_url: str, env_variables: list):
""" """Execute TTECK script with already filled environment variables to run silently (non-interactive)
Execute TTECK script with already filled environment variables to run silently (non-interactive)
:param script_url: script url (github or other) Parameters
:param env_variables: list of environment variables ----------
:return: status code script_url: str
script url (github or other)
env_variables: list
list of environment variables
Returns
-------
int
status code
""" """
env_variables = " ".join(env_variables) env_variables = " ".join(env_variables)
@ -59,20 +87,46 @@ def execute_tteck_script(script_url, env_variables):
def get_config(): def get_config():
"""Get the JSON content of config.json file as a dict
Returns
-------
dict
JSON content of config.json file
"""
config_file = os.path.join(project_path / "resources" / "config.json") config_file = os.path.join(project_path / "resources" / "config.json")
with open(config_file, "r") as file: with open(config_file, "r") as file:
return json.loads(file.read()) return json.loads(file.read())
def run_command_on_pve(command, warn_exit_status=False, only_code=False, local=False): def run_command_on_pve(command: str, warn_exit_status: bool = False, only_code: bool = False, local: bool = False):
""" """Run command on the Proxmox VE host
Run command on PVE
Parameters
----------
command: str
command to run
warn_exit_status: bool, optional
should an exception be thrown if the exit status is not 0
only_code: bool, optional
should it only return the exit code and not the stdout
local: bool, optional
should it be run locally not via ssh even with local mode off in the pve section of config.json
Returns
-------
str
command output by default
int
command exit code if only_code is True
Raises
------
Exception
if the exit status is not 0 and warn_exit_status is True
:param local: should be run locally not via ssh even with local mode off
:param only_code: should we only return the exit code and not the stdout
:param warn_exit_status: should an exception be thrown if the exit status is not 0
:param command: command
:return: command
""" """
# Get config # Get config
@ -117,12 +171,45 @@ def run_command_on_pve(command, warn_exit_status=False, only_code=False, local=F
return command.stdout.rstrip() return command.stdout.rstrip()
def run_command_locally(command, warn_exit_status=False, only_code=False): def run_command_locally(command: str, warn_exit_status: bool = False, only_code: bool = False):
"""Force running command locally even if local mode is off in config.json
See Also
--------
run_command_on_pve
"""
run_command_on_pve(command=command, warn_exit_status=warn_exit_status, only_code=only_code, local=True) run_command_on_pve(command=command, warn_exit_status=warn_exit_status, only_code=only_code, local=True)
def run_command_ssh(command, host, username, port=22, warn_exit_status=False, only_code=False): def run_command_ssh(command: str, host: str, username: str, port: int = 22, warn_exit_status: bool = False,
# Run command on PVE via SSH and return output only_code: bool = False):
"""Run command on a remote host via SSH
Parameters
----------
command: str
command to run
host: str
host to run the command on
username: str
username to use to connect to the host (usually root for lxc and vm)
port: int, optional
port to use to connect to the host (default: 22)
warn_exit_status: bool, optional
should an exception be thrown if the exit status is not 0
only_code: bool, optional
should it only return the exit code and not the stdout
Returns
-------
str
command output by default
int
command exit code if only_code is True
"""
logging.debug(f"Running command on host {host} (ssh): {command}") logging.debug(f"Running command on host {host} (ssh): {command}")
# catch errors code # catch errors code
@ -139,12 +226,24 @@ def run_command_ssh(command, host, username, port=22, warn_exit_status=False, on
return command.stdout.rstrip() return command.stdout.rstrip()
def get_install_package_command(distribution): def get_install_package_command(distribution: str):
""" """Retrieve the correct command to install a package based on the distribution specified
Get the install package without interaction command based on the distribution. It supports all the distribution supported by Proxmox VE (from the pct command documentation).
Debian, Ubuntu, CentOS, Fedora, Gentoo, Alpine, ArchLinux, Devuan, NixOS, OpenSUSE
:param distribution: Distribution name Parameters
:return: Install package command ----------
distribution: str
Name of the distribution as specific in the proxmox pct command documentation
See Also
--------
https://pve.proxmox.com/pve-docs/pct.1.html
Returns
-------
str
Beginning of the command to install a package, the package name should be appended to it
""" """
distribution = distribution.lower() distribution = distribution.lower()
@ -173,12 +272,24 @@ def get_install_package_command(distribution):
raise Exception(f"Unsupported distribution: {distribution}") raise Exception(f"Unsupported distribution: {distribution}")
def get_remove_package_command(distribution): def get_remove_package_command(distribution: str):
""" """Retrieve the correct command to uninstall a package based on the distribution specified
Get the remove package without interaction command based on the distribution. It supports all the distribution supported by Proxmox VE (from the pct command documentation).
Debian, Ubuntu, CentOS, Fedora, Gentoo, Alpine, ArchLinux, Devuan, NixOS, OpenSUSE
:param distribution: Distribution name Parameters
:return: Remove package command ----------
distribution: str
Name of the distribution as specific in the proxmox pct command documentation
See Also
--------
https://pve.proxmox.com/pve-docs/pct.1.html
Returns
-------
str
Beginning of the command to uninstall a package, the package name should be appended to it
""" """
distribution = distribution.lower() distribution = distribution.lower()
@ -208,8 +319,59 @@ def get_remove_package_command(distribution):
def get_ssh_public_key(): def get_ssh_public_key():
""" Retrieve the SSH public key of the machine running this script
Returns
-------
str
SSH public key
""" """
Get SSH public key
:return: ssh public key # TODO: maybe implement custom path for the key
"""
return run_command_on_pve(command="cat ~/.ssh/id_rsa.pub", warn_exit_status=True) return run_command_on_pve(command="cat ~/.ssh/id_rsa.pub", warn_exit_status=True)
def copy_file_to_pve(path: Path, destination: str):
"""Copy a local file (on the machine running this script) to the PVE
Parameters
----------
path: pathlib.Path
Path object to file on local machine
destination: str
Destination on PVE
Examples
--------
>>> copy_file_to_pve(Path("/home/user/file.txt"), "/root/file.txt")
"""
config = get_config()
run_command_locally(f"scp {str(path)} {config['pve']['root']}@{config['pve']['host']}:{destination}")
def copy_file_to_lxc(lxc_id: int, path: Path, destination: str):
"""Copy a local file (on the machine running this script) to the PVE and finally to the LXC
It works by copying the script from the local machine to the PVE using SCP and
then using "pct push" to send it to the LXC.
Sometimes SSH is not installed on freshs LXC, so by using pct push we are sure that it will work.
Parameters
----------
lxc_id: int
LXC ID
path: pathlib.Path
Path object to file on local machine
destination: str
Destination on LXC
Examples
--------
>>> copy_file_to_lxc(100, Path("/home/user/file.txt"), "/root/")
"""
file = path.name
temp_path = f"/tmp/proxmoxdeployscripts/{file}"
copy_file_to_pve(path, temp_path)
run_command_on_pve(f"pct push {lxc_id} {temp_path} {destination}")

View File

@ -1,15 +1,44 @@
import os import pathlib
from .. import project_path from .. import project_path
from ..lxc import LXC
def get_path(lxc, path): def get_path(lxc: LXC, path: str):
parent = os.path.join(project_path / "resources") """Returns the complete path to a global, protected or LXC-relative resource file.
if path.startswith("/global/"): For example, if the path is "global/test.txt", the function will return the complete path to
the file in the resources/ folder.
Another example, if the path is "protected/test.txt", the function will return the complete path to
the file in the protected_resources/ folder.
Other folders can be added like "global/scripts/test.sh" or "protected/scripts/test.sh".
If none of the above conditions are met, the function will return the path relative to the LXC folder
depending on the given LXC object.
Parameters
----------
lxc: LXC
LXC object to use to get the relative path
Can be omitted if you know what you are doing and the path is either global or protected.
path: str
Path to the resource file
Returns
-------
pathlib.Path
Complete path to the resource file
"""
parent = pathlib.Path(project_path).joinpath("resources")
if path.startswith("global/"):
# Use global folder # Use global folder
return parent + "/scripts/" + path[len("/global/"):] return parent.joinpath(path[len("global/"):])
elif path.startswith("protected/"):
# Use protected folder
return parent.parent.joinpath("protected_resources", path[len("protected/"):])
else: else:
# Use VM/LXC folder # Use VM/LXC folder
lxc_id = lxc.get_id() lxc_id = lxc.get_id()
return parent + f"/resources/lxc/{lxc_id}/{path}" return parent.joinpath("lxc", lxc_id, path)