90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
"""Credential-transport contracts for the GCP leoclean role provisioner."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PROVISIONER = ROOT / "ops" / "provision_gcp_leoclean_runtime_role.sh"
|
|
|
|
|
|
def test_password_provisioner_launcher_keeps_admin_secret_out_of_argv_and_preserves_runtime_records(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
provisioner = PROVISIONER.read_text(encoding="utf-8")
|
|
prefix = '"$SETSID_BIN" -- "$BASH_BIN" --noprofile --norc -c \'\n'
|
|
suffix = '\n \' leoclean-runtime-role-provisioner "$PSQL_BIN"'
|
|
assert provisioner.count(prefix) == 1
|
|
assert "set +x" in provisioner
|
|
assert "IFS= read -r PGPASSWORD" in provisioner
|
|
assert 'PGPASSWORD="$administrator_password"' not in provisioner
|
|
assert provisioner.index("printf '%s\\n' \"$administrator_password\"") < provisioner.index(
|
|
"printf '%s\\n' \"$runtime_password\""
|
|
)
|
|
launcher = provisioner.split(prefix, 1)[1].split(suffix, 1)[0]
|
|
|
|
probe = tmp_path / "password_transport_probe.py"
|
|
probe.write_text(
|
|
"""import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
records = sys.stdin.buffer.read().splitlines()
|
|
payload = {
|
|
"administrator_sha256": hashlib.sha256(os.environ["PGPASSWORD"].encode()).hexdigest(),
|
|
"runtime_sha256": [hashlib.sha256(record).hexdigest() for record in records],
|
|
}
|
|
print(json.dumps(payload, sort_keys=True))
|
|
raise SystemExit(0 if len(records) == 2 else 73)
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
administrator = " admin $() * \\ ' \" ;\ttrailing "
|
|
runtime = " runtime [] {} $HOME \\ path ; ' \" \t "
|
|
command = [
|
|
"/bin/bash",
|
|
"--noprofile",
|
|
"--norc",
|
|
"-c",
|
|
launcher,
|
|
"leoclean-runtime-role-provisioner",
|
|
sys.executable,
|
|
str(probe),
|
|
]
|
|
assert administrator not in command
|
|
assert runtime not in command
|
|
|
|
completed = subprocess.run(
|
|
command,
|
|
input=f"{administrator}\n{runtime}\n{runtime}\n",
|
|
capture_output=True,
|
|
check=False,
|
|
text=True,
|
|
)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
assert completed.stderr == ""
|
|
observed = json.loads(completed.stdout)
|
|
assert observed == {
|
|
"administrator_sha256": hashlib.sha256(administrator.encode()).hexdigest(),
|
|
"runtime_sha256": [hashlib.sha256(runtime.encode()).hexdigest()] * 2,
|
|
}
|
|
assert administrator not in completed.stdout
|
|
assert runtime not in completed.stdout
|
|
|
|
for invalid_input in (administrator, f"{'a' * 4097}\n"):
|
|
denied = subprocess.run(
|
|
command,
|
|
input=invalid_input,
|
|
capture_output=True,
|
|
check=False,
|
|
text=True,
|
|
)
|
|
assert denied.returncode == 64
|
|
assert denied.stdout == ""
|
|
assert denied.stderr == ""
|