aa5eddbe33
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 1m2s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 0s
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 9s
333 lines
9.9 KiB
Python
333 lines
9.9 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import textwrap
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
SCRIPT = ROOT / "scripts" / "reboot-recovery" / "collect-windows99-vmware-verify.sh"
|
|
|
|
|
|
def _write_executable(path: Path, text: str) -> None:
|
|
path.write_text(textwrap.dedent(text).lstrip())
|
|
path.chmod(0o755)
|
|
|
|
|
|
def _run_collector(fake_bin: Path, *args: str) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
env["PATH"] = f"{fake_bin}:{env['PATH']}"
|
|
env["WINDOWS99_KNOWN_HOSTS_FILE"] = str(fake_bin / "known_hosts")
|
|
return subprocess.run(
|
|
["bash", str(SCRIPT), *args],
|
|
cwd=ROOT,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def _key_values(stdout: str) -> dict[str, str]:
|
|
values: dict[str, str] = {}
|
|
for line in stdout.splitlines():
|
|
if "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
values[key] = value
|
|
return values
|
|
|
|
|
|
def test_collector_contract_forbids_secret_and_runtime_actions() -> None:
|
|
text = SCRIPT.read_text()
|
|
|
|
assert "BatchMode=yes" in text
|
|
assert "PreferredAuthentications=publickey" in text
|
|
assert "PubkeyAuthentication=yes" in text
|
|
assert "PasswordAuthentication=no" in text
|
|
assert "KbdInteractiveAuthentication=no" in text
|
|
assert "NumberOfPasswordPrompts=0" in text
|
|
assert "ConnectionAttempts=1" in text
|
|
assert "GSSAPIAuthentication=no" in text
|
|
assert "in_memory_stdin_scriptblock" in text
|
|
assert "[Console]::In.ReadToEnd()" in text
|
|
assert ".\\\\windows99-vmware-autostart.ps1 -Mode Verify" not in text
|
|
for forbidden in [
|
|
"sshpass",
|
|
"PasswordAuthentication=yes",
|
|
"KbdInteractiveAuthentication=yes",
|
|
"net use",
|
|
"shutdown",
|
|
"Restart-Computer",
|
|
"Start-VM",
|
|
"vmrun start",
|
|
"Set-ItemProperty",
|
|
"Register-ScheduledTask",
|
|
]:
|
|
assert forbidden not in text
|
|
|
|
|
|
def test_check_mode_reports_open_ports_and_missing_publickey_auth(tmp_path: Path) -> None:
|
|
fake_bin = tmp_path / "bin"
|
|
fake_bin.mkdir()
|
|
_write_executable(
|
|
fake_bin / "nc",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
port="${!#}"
|
|
if [[ "$port" == "22" || "$port" == "3389" ]]; then
|
|
exit 0
|
|
fi
|
|
exit 1
|
|
""",
|
|
)
|
|
_write_executable(
|
|
fake_bin / "ssh",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
printf '%s\n' 'Permission denied (publickey,password,keyboard-interactive).' >&2
|
|
exit 255
|
|
""",
|
|
)
|
|
|
|
result = _run_collector(fake_bin, "--check")
|
|
|
|
assert result.returncode == 0
|
|
values = _key_values(result.stdout)
|
|
assert values["schema_version"] == "windows99_vmware_verify_collector_v1"
|
|
assert values["dry_run"] == "true"
|
|
assert values["ssh_auth_probe_user_limit"] == "2"
|
|
assert values["port_22_open"] == "1"
|
|
assert values["port_3389_open"] == "1"
|
|
assert values["port_5985_open"] == "0"
|
|
assert values["port_5986_open"] == "0"
|
|
assert values["ssh_batchmode_auth_ready"] == "0"
|
|
assert values["ssh_auth_probed_users"] == "2"
|
|
assert values["remote_verify_attempted"] == "0"
|
|
assert values["verify_collection_status"] == "blocked_ssh_publickey_auth_missing"
|
|
assert values["secret_value_read"] == "false"
|
|
assert values["password_prompt_allowed"] == "false"
|
|
assert values["remote_write_performed"] == "false"
|
|
assert values["host_reboot_performed"] == "false"
|
|
assert values["vm_power_change_performed"] == "false"
|
|
assert values["windows_update_policy_apply_performed"] == "false"
|
|
|
|
|
|
def test_check_mode_probes_all_explicit_users_without_secret_prompt(tmp_path: Path) -> None:
|
|
fake_bin = tmp_path / "bin"
|
|
fake_bin.mkdir()
|
|
_write_executable(
|
|
fake_bin / "nc",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
port="${!#}"
|
|
if [[ "$port" == "22" || "$port" == "3389" ]]; then
|
|
exit 0
|
|
fi
|
|
exit 1
|
|
""",
|
|
)
|
|
_write_executable(
|
|
fake_bin / "ssh",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
exit 255
|
|
""",
|
|
)
|
|
|
|
result = _run_collector(
|
|
fake_bin,
|
|
"--check",
|
|
"--users",
|
|
"ogt wooo ooo administrator Administrator",
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
values = _key_values(result.stdout)
|
|
assert values["ssh_auth_probe_user_limit"] == "5"
|
|
assert values["ssh_auth_probed_users"] == "5"
|
|
assert values["ssh_batchmode_auth_ready"] == "0"
|
|
assert values["verify_collection_status"] == "blocked_ssh_publickey_auth_missing"
|
|
assert values["secret_value_read"] == "false"
|
|
assert values["password_prompt_allowed"] == "false"
|
|
|
|
|
|
def test_collect_mode_blocks_without_publickey_auth(tmp_path: Path) -> None:
|
|
fake_bin = tmp_path / "bin"
|
|
fake_bin.mkdir()
|
|
_write_executable(
|
|
fake_bin / "nc",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
port="${!#}"
|
|
[[ "$port" == "22" ]]
|
|
""",
|
|
)
|
|
_write_executable(
|
|
fake_bin / "ssh",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
exit 255
|
|
""",
|
|
)
|
|
|
|
result = _run_collector(fake_bin, "--collect")
|
|
|
|
assert result.returncode == 75
|
|
values = _key_values(result.stdout)
|
|
assert values["dry_run"] == "true"
|
|
assert values["ssh_batchmode_auth_ready"] == "0"
|
|
assert values["remote_verify_attempted"] == "0"
|
|
assert values["verify_collection_status"] == "blocked_ssh_publickey_auth_missing"
|
|
|
|
|
|
def test_check_mode_auth_ready_does_not_run_remote_verify(tmp_path: Path) -> None:
|
|
fake_bin = tmp_path / "bin"
|
|
fake_bin.mkdir()
|
|
_write_executable(
|
|
fake_bin / "nc",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
exit 0
|
|
""",
|
|
)
|
|
_write_executable(
|
|
fake_bin / "ssh",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
args="$*"
|
|
if [[ "$args" == *"powershell"* ]]; then
|
|
printf '%s\n' 'unexpected remote verify' >&2
|
|
exit 44
|
|
fi
|
|
printf '%s\n' 'AWOOOI_WINDOWS99_SSH_READY'
|
|
exit 0
|
|
""",
|
|
)
|
|
|
|
result = _run_collector(fake_bin, "--check")
|
|
|
|
assert result.returncode == 0
|
|
values = _key_values(result.stdout)
|
|
assert values["ssh_batchmode_auth_ready"] == "1"
|
|
assert values["remote_verify_mode"] == "in_memory_stdin_scriptblock"
|
|
assert values["local_verify_script_present"] == "1"
|
|
assert values["remote_verify_attempted"] == "0"
|
|
assert values["verify_collection_status"] == "ready_ssh_batchmode_auth_probe_only"
|
|
assert "VMRUN_PRESENT=1" not in result.stdout
|
|
assert "remote_verify_output_begin" not in result.stdout
|
|
|
|
|
|
def test_collect_mode_runs_readonly_remote_verify_when_auth_ready(tmp_path: Path) -> None:
|
|
fake_bin = tmp_path / "bin"
|
|
fake_bin.mkdir()
|
|
_write_executable(
|
|
fake_bin / "nc",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
exit 0
|
|
""",
|
|
)
|
|
_write_executable(
|
|
fake_bin / "ssh",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
args="$*"
|
|
if [[ "$args" == *"powershell"* ]]; then
|
|
printf '%s\n' 'AWOOOI_WINDOWS99_VMWARE_AUTOSTART=1'
|
|
printf '%s\n' 'MODE=Verify'
|
|
printf '%s\n' 'VMRUN_PRESENT=1'
|
|
printf '%s\n' 'VMWARE_AUTOSTART_VERIFY_READY=1'
|
|
exit 0
|
|
fi
|
|
printf '%s\n' 'AWOOOI_WINDOWS99_SSH_READY'
|
|
exit 0
|
|
""",
|
|
)
|
|
|
|
result = _run_collector(fake_bin, "--collect")
|
|
|
|
assert result.returncode == 0
|
|
values = _key_values(result.stdout)
|
|
assert values["dry_run"] == "false"
|
|
assert values["ssh_batchmode_auth_ready"] == "1"
|
|
assert values["remote_verify_attempted"] == "1"
|
|
assert values["remote_verify_exit_status"] == "0"
|
|
assert values["verify_collection_status"] == "collected_windows99_vmware_verify_stdout"
|
|
assert "remote_verify_output_begin" in result.stdout
|
|
assert "VMRUN_PRESENT=1" in result.stdout
|
|
assert "VMWARE_AUTOSTART_VERIFY_READY=1" in result.stdout
|
|
assert "remote_verify_output_end" in result.stdout
|
|
|
|
|
|
def test_collect_mode_rejects_success_exit_without_verify_marker(tmp_path: Path) -> None:
|
|
fake_bin = tmp_path / "bin"
|
|
fake_bin.mkdir()
|
|
_write_executable(
|
|
fake_bin / "nc",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
exit 0
|
|
""",
|
|
)
|
|
_write_executable(
|
|
fake_bin / "ssh",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
args="$*"
|
|
if [[ "$args" == *"powershell"* ]]; then
|
|
printf '%s\n' 'Windows PowerShell 5.1'
|
|
printf '%s\n' "-File argument '.\\windows99-vmware-autostart.ps1' does not exist"
|
|
exit 0
|
|
fi
|
|
printf '%s\n' 'AWOOOI_WINDOWS99_SSH_READY'
|
|
exit 0
|
|
""",
|
|
)
|
|
|
|
result = _run_collector(fake_bin, "--collect")
|
|
|
|
assert result.returncode == 75
|
|
values = _key_values(result.stdout)
|
|
assert values["remote_verify_attempted"] == "1"
|
|
assert values["remote_verify_exit_status"] == "0"
|
|
assert values["verify_collection_status"] == "blocked_remote_verify_output_invalid"
|
|
assert "AWOOOI_WINDOWS99_VMWARE_AUTOSTART=1" not in result.stdout
|
|
|
|
|
|
def test_collect_mode_accepts_windows_crlf_verify_markers(tmp_path: Path) -> None:
|
|
fake_bin = tmp_path / "bin"
|
|
fake_bin.mkdir()
|
|
_write_executable(
|
|
fake_bin / "nc",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
exit 0
|
|
""",
|
|
)
|
|
_write_executable(
|
|
fake_bin / "ssh",
|
|
"""
|
|
#!/usr/bin/env bash
|
|
args="$*"
|
|
if [[ "$args" == *"powershell"* ]]; then
|
|
printf 'AWOOOI_WINDOWS99_VMWARE_AUTOSTART=1\\r\\n'
|
|
printf 'MODE=Verify\\r\\n'
|
|
printf 'VMRUN_PRESENT=1\\r\\n'
|
|
printf 'VMWARE_AUTOSTART_VERIFY_READY=0\\r\\n'
|
|
exit 0
|
|
fi
|
|
printf '%s\n' 'AWOOOI_WINDOWS99_SSH_READY'
|
|
exit 0
|
|
""",
|
|
)
|
|
|
|
result = _run_collector(fake_bin, "--collect")
|
|
|
|
assert result.returncode == 0
|
|
values = _key_values(result.stdout)
|
|
assert values["verify_collection_status"] == "collected_windows99_vmware_verify_stdout"
|
|
assert "AWOOOI_WINDOWS99_VMWARE_AUTOSTART=1\r" not in result.stdout
|
|
assert "VMWARE_AUTOSTART_VERIFY_READY=0" in result.stdout
|