test_secrets_upload: Don't prepend sudo inside test; Improve secret upload test
This commit is contained in:
@@ -1,21 +1,28 @@
|
||||
import tarfile
|
||||
from pathlib import Path
|
||||
from shlex import quote
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from clan_cli.cmd import Log, RunOpts
|
||||
from clan_cli.cmd import run as run_local
|
||||
from clan_cli.errors import ClanError
|
||||
from clan_cli.ssh.host import Host
|
||||
|
||||
|
||||
def upload(
|
||||
host: Host,
|
||||
local_src: Path, # must be a directory
|
||||
local_src: Path,
|
||||
remote_dest: Path, # must be a directory
|
||||
file_user: str = "root",
|
||||
file_group: str = "root",
|
||||
dir_mode: int = 0o700,
|
||||
file_mode: int = 0o400,
|
||||
) -> None:
|
||||
# Check if the remote destination is at least 3 directories deep
|
||||
if len(remote_dest.parts) < 3:
|
||||
msg = f"The remote destination must be at least 3 directories deep. Got: {remote_dest}. Reason: The directory will be deleted with 'rm -rf'."
|
||||
raise ClanError(msg)
|
||||
|
||||
# Create the tarball from the temporary directory
|
||||
with TemporaryDirectory(prefix="facts-upload-") as tardir:
|
||||
tar_path = Path(tardir) / "upload.tar.gz"
|
||||
@@ -55,64 +62,22 @@ def upload(
|
||||
with local_src.open("rb") as f:
|
||||
tar.addfile(tarinfo, f)
|
||||
|
||||
priviledge_escalation = []
|
||||
if host.user != "root":
|
||||
priviledge_escalation = ["sudo", "--"]
|
||||
sudo = ""
|
||||
if host.user != "root" and os.environ.get("IN_PYTEST") is None:
|
||||
sudo = "sudo -- "
|
||||
|
||||
if local_src.is_dir():
|
||||
cmd = [
|
||||
*host.ssh_cmd(),
|
||||
"--",
|
||||
*priviledge_escalation,
|
||||
"bash",
|
||||
"-c",
|
||||
'exec "$@"',
|
||||
"--",
|
||||
"rm",
|
||||
"-r",
|
||||
str(remote_dest),
|
||||
"mkdir",
|
||||
"-m",
|
||||
f"{dir_mode:o}",
|
||||
"-p",
|
||||
str(remote_dest),
|
||||
"&&",
|
||||
"tar",
|
||||
"-C",
|
||||
str(remote_dest),
|
||||
"-xzf",
|
||||
"-",
|
||||
]
|
||||
else:
|
||||
# For single file, extract to parent directory and ensure correct name
|
||||
cmd = [
|
||||
*host.ssh_cmd(),
|
||||
"--",
|
||||
*priviledge_escalation,
|
||||
"bash",
|
||||
"-c",
|
||||
'exec "$@"',
|
||||
"--",
|
||||
"rm",
|
||||
"-r",
|
||||
str(remote_dest),
|
||||
"mkdir",
|
||||
"-m",
|
||||
f"{dir_mode:o}",
|
||||
"-p",
|
||||
str(remote_dest.parent),
|
||||
"&&",
|
||||
"tar",
|
||||
"-C",
|
||||
str(remote_dest.parent),
|
||||
"-xzf",
|
||||
"-",
|
||||
]
|
||||
cmd = "rm -rf $0 && mkdir -m $1 -p $0 && tar -C $0 -xzf -"
|
||||
|
||||
# TODO accept `input` to be an IO object instead of bytes so that we don't have to read the tarfile into memory.
|
||||
with tar_path.open("rb") as f:
|
||||
run_local(
|
||||
cmd,
|
||||
[
|
||||
*host.ssh_cmd(),
|
||||
"--",
|
||||
f"{sudo}bash -c {quote(cmd)}",
|
||||
str(remote_dest),
|
||||
f"{dir_mode:o}",
|
||||
],
|
||||
RunOpts(
|
||||
input=f.read(),
|
||||
log=Log.BOTH,
|
||||
|
||||
Reference in New Issue
Block a user