feat: add secure GCP operator reauth helper (#90)
Some checks are pending
CI / lint-and-test (push) Waiting to run
Some checks are pending
CI / lint-and-test (push) Waiting to run
* feat: add secure GCP operator reauth helper * fix: keep secure GCP prompt open on empty input * fix: enable paste in secure GCP password dialog
This commit is contained in:
parent
62bfd429f2
commit
cf796fea76
3 changed files with 690 additions and 0 deletions
37
docs/gcp-operator-reauth.md
Normal file
37
docs/gcp-operator-reauth.md
Normal file
|
|
@ -0,0 +1,37 @@
|
|||
# GCP Operator Reauthentication
|
||||
|
||||
From the `teleo-infrastructure` repository root, inspect the current local
|
||||
operator state without opening a dialog:
|
||||
|
||||
```bash
|
||||
scripts/gcp-operator-reauth.sh --status
|
||||
```
|
||||
|
||||
Store or update the current Google password through a native secure prompt:
|
||||
|
||||
```bash
|
||||
scripts/gcp-operator-reauth.sh --store-password
|
||||
```
|
||||
|
||||
With no option, the script asks before opening that prompt. It uses
|
||||
`pinentry-mac` when installed. Otherwise it compiles a temporary AppKit helper
|
||||
that uses `NSAlert` and `NSSecureTextField`. The helper stores the value as a
|
||||
non-synchronizing generic password in this Mac's encrypted Keychain. The
|
||||
password is never printed, copied to the clipboard, submitted to Google, or
|
||||
placed in a process argument or plaintext file.
|
||||
|
||||
The stored password is only an emergency operator convenience. A Google
|
||||
password cannot refresh or satisfy gcloud OAuth by itself, and this script does
|
||||
not attempt to automate Google login. When OAuth is stale, follow the one
|
||||
`clear_CTA` printed by the script and rerun `--status`.
|
||||
|
||||
`--status` verifies the selected account and project, checks token refresh
|
||||
without printing the token, reads the expected VM identity, and asks gcloud to
|
||||
construct the IAP SSH command with `--dry-run`. It does not create a tunnel or
|
||||
start SSH. `iap_ssh_preflight=ready_dry_run_only` must not be reported as a live
|
||||
IAP connection.
|
||||
|
||||
The durable unattended operator is
|
||||
`.github/workflows/gcp-iap-operator.yml`. It uses GitHub OIDC, Workload Identity
|
||||
Federation, fixed reviewed operations, and IAP after the one-time authenticated
|
||||
bootstrap. No stored Google password is an alternative to that route.
|
||||
403
scripts/gcp-operator-reauth.sh
Executable file
403
scripts/gcp-operator-reauth.sh
Executable file
|
|
@ -0,0 +1,403 @@
|
|||
#!/usr/bin/env bash
|
||||
set +x
|
||||
set -Eeuo pipefail
|
||||
umask 077
|
||||
|
||||
readonly EXPECTED_ACCOUNT="billy@livingip.xyz"
|
||||
readonly PROJECT_ID="teleo-501523"
|
||||
readonly INSTANCE="teleo-prod-1"
|
||||
readonly ZONE="europe-west6-a"
|
||||
readonly EXPECTED_INTERNAL_IP="10.60.0.3"
|
||||
readonly KEYCHAIN_SERVICE="com.livingip.teleo.gcp-operator-reauth"
|
||||
readonly KEYCHAIN_LABEL="Teleo GCP operator emergency password"
|
||||
readonly WORKFLOW_PATH=".github/workflows/gcp-iap-operator.yml"
|
||||
readonly RERUN_COMMAND="scripts/gcp-operator-reauth.sh --status"
|
||||
|
||||
HELPER_DIR=""
|
||||
KEYCHAIN_HELPER=""
|
||||
|
||||
usage() {
|
||||
cat <<'EOF'
|
||||
Usage: scripts/gcp-operator-reauth.sh [--status | --store-password]
|
||||
|
||||
--status Report Keychain presence and local gcloud/IAP prerequisites.
|
||||
--store-password Securely store or update the emergency Google password, then report status.
|
||||
|
||||
With no option, the script asks before opening the secure password prompt.
|
||||
EOF
|
||||
}
|
||||
|
||||
die() {
|
||||
printf 'gcp-operator-reauth: %s\n' "$1" >&2
|
||||
exit 2
|
||||
}
|
||||
|
||||
cleanup() {
|
||||
if [[ -n "$HELPER_DIR" && -d "$HELPER_DIR" ]]; then
|
||||
rm -rf -- "$HELPER_DIR"
|
||||
fi
|
||||
}
|
||||
trap cleanup EXIT
|
||||
|
||||
validate_test_overrides() {
|
||||
if [[ -n "${TELEO_GCP_REAUTH_KEYCHAIN_HELPER:-}${TELEO_GCP_REAUTH_PINENTRY:-}" && \
|
||||
"${TELEO_GCP_REAUTH_TESTING:-0}" != "1" ]]; then
|
||||
die "helper overrides are accepted only when TELEO_GCP_REAUTH_TESTING=1"
|
||||
fi
|
||||
}
|
||||
|
||||
build_keychain_helper() {
|
||||
if [[ -n "${TELEO_GCP_REAUTH_KEYCHAIN_HELPER:-}" ]]; then
|
||||
[[ -x "$TELEO_GCP_REAUTH_KEYCHAIN_HELPER" ]] || die "test Keychain helper is not executable"
|
||||
KEYCHAIN_HELPER="$TELEO_GCP_REAUTH_KEYCHAIN_HELPER"
|
||||
return
|
||||
fi
|
||||
|
||||
command -v swiftc >/dev/null 2>&1 || die "swiftc is required for secure macOS Keychain access"
|
||||
HELPER_DIR="$(mktemp -d "${TMPDIR:-/tmp}/teleo-gcp-keychain.XXXXXX")"
|
||||
chmod 0700 "$HELPER_DIR"
|
||||
local source="$HELPER_DIR/GcpOperatorKeychain.swift"
|
||||
KEYCHAIN_HELPER="$HELPER_DIR/gcp-operator-keychain"
|
||||
|
||||
cat >"$source" <<'SWIFT'
|
||||
import AppKit
|
||||
import Foundation
|
||||
import Security
|
||||
|
||||
func fail(_ message: String, _ code: Int32 = 2) -> Never {
|
||||
FileHandle.standardError.write(Data("gcp-operator-keychain: \(message)\n".utf8))
|
||||
exit(code)
|
||||
}
|
||||
|
||||
guard CommandLine.arguments.count == 5 else {
|
||||
fail("expected action, service, account, and label")
|
||||
}
|
||||
|
||||
let action = CommandLine.arguments[1]
|
||||
let service = CommandLine.arguments[2]
|
||||
let account = CommandLine.arguments[3]
|
||||
let label = CommandLine.arguments[4]
|
||||
|
||||
func itemQuery() -> [String: Any] {
|
||||
[
|
||||
kSecClass as String: kSecClassGenericPassword,
|
||||
kSecAttrService as String: service,
|
||||
kSecAttrAccount as String: account,
|
||||
kSecAttrSynchronizable as String: kCFBooleanFalse as Any,
|
||||
]
|
||||
}
|
||||
|
||||
func itemExists() -> Bool {
|
||||
var query = itemQuery()
|
||||
query[kSecMatchLimit as String] = kSecMatchLimitOne
|
||||
let status = SecItemCopyMatching(query as CFDictionary, nil)
|
||||
if status == errSecSuccess {
|
||||
return true
|
||||
}
|
||||
if status == errSecItemNotFound {
|
||||
return false
|
||||
}
|
||||
fail("Keychain status failed with OSStatus \(status)")
|
||||
}
|
||||
|
||||
func store(_ password: Data) -> String {
|
||||
guard !password.isEmpty else {
|
||||
fail("empty password was not stored")
|
||||
}
|
||||
guard password.count <= 4096 else {
|
||||
fail("password exceeds the 4096-byte limit")
|
||||
}
|
||||
|
||||
let query = itemQuery()
|
||||
if itemExists() {
|
||||
let attributes: [String: Any] = [
|
||||
kSecValueData as String: password,
|
||||
kSecAttrLabel as String: label,
|
||||
kSecAttrAccessible as String: kSecAttrAccessibleWhenUnlockedThisDeviceOnly,
|
||||
]
|
||||
let status = SecItemUpdate(query as CFDictionary, attributes as CFDictionary)
|
||||
guard status == errSecSuccess else {
|
||||
fail("Keychain update failed with OSStatus \(status)")
|
||||
}
|
||||
return "updated"
|
||||
}
|
||||
|
||||
var item = query
|
||||
item[kSecValueData as String] = password
|
||||
item[kSecAttrLabel as String] = label
|
||||
item[kSecAttrAccessible as String] = kSecAttrAccessibleWhenUnlockedThisDeviceOnly
|
||||
let status = SecItemAdd(item as CFDictionary, nil)
|
||||
guard status == errSecSuccess else {
|
||||
fail("Keychain add failed with OSStatus \(status)")
|
||||
}
|
||||
return "stored"
|
||||
}
|
||||
|
||||
func hexValue(_ byte: UInt8) -> UInt8? {
|
||||
switch byte {
|
||||
case 48...57: return byte - 48
|
||||
case 65...70: return byte - 55
|
||||
case 97...102: return byte - 87
|
||||
default: return nil
|
||||
}
|
||||
}
|
||||
|
||||
func decodeAssuan(_ value: Substring) -> Data? {
|
||||
let input = Array(value.utf8)
|
||||
var output = Data()
|
||||
var index = 0
|
||||
while index < input.count {
|
||||
if input[index] == 37 {
|
||||
guard index + 2 < input.count,
|
||||
let high = hexValue(input[index + 1]),
|
||||
let low = hexValue(input[index + 2]) else {
|
||||
return nil
|
||||
}
|
||||
output.append((high << 4) | low)
|
||||
index += 3
|
||||
} else {
|
||||
output.append(input[index])
|
||||
index += 1
|
||||
}
|
||||
}
|
||||
return output
|
||||
}
|
||||
|
||||
func passwordFromPinentry(_ response: Data) -> Data {
|
||||
guard let text = String(data: response, encoding: .utf8) else {
|
||||
fail("pinentry returned non-UTF-8 protocol data")
|
||||
}
|
||||
for line in text.split(separator: "\n", omittingEmptySubsequences: false) {
|
||||
if line.hasPrefix("D ") {
|
||||
guard let decoded = decodeAssuan(line.dropFirst(2)) else {
|
||||
fail("pinentry returned invalid escaped data")
|
||||
}
|
||||
return decoded
|
||||
}
|
||||
}
|
||||
fail("pinentry did not return a password")
|
||||
}
|
||||
|
||||
switch action {
|
||||
case "status":
|
||||
print(itemExists() ? "present" : "absent")
|
||||
case "store-pinentry":
|
||||
let response = FileHandle.standardInput.readDataToEndOfFile()
|
||||
print(store(passwordFromPinentry(response)))
|
||||
case "prompt-store":
|
||||
let app = NSApplication.shared
|
||||
app.setActivationPolicy(.accessory)
|
||||
app.activate(ignoringOtherApps: true)
|
||||
let alert = NSAlert()
|
||||
alert.alertStyle = .informational
|
||||
alert.messageText = "Store Teleo GCP emergency password"
|
||||
alert.informativeText = "This password is stored only in this Mac's encrypted Keychain. It cannot refresh gcloud OAuth by itself."
|
||||
alert.addButton(withTitle: "Store in Keychain")
|
||||
alert.addButton(withTitle: "Cancel")
|
||||
let field = NSSecureTextField(frame: NSRect(x: 0, y: 0, width: 360, height: 24))
|
||||
field.placeholderString = "Current Google password"
|
||||
let mainMenu = NSMenu()
|
||||
let editMenuItem = NSMenuItem(title: "Edit", action: nil, keyEquivalent: "")
|
||||
let editMenu = NSMenu(title: "Edit")
|
||||
editMenu.addItem(withTitle: "Paste", action: #selector(NSText.paste(_:)), keyEquivalent: "v")
|
||||
editMenuItem.submenu = editMenu
|
||||
mainMenu.addItem(editMenuItem)
|
||||
app.mainMenu = mainMenu
|
||||
let fieldMenu = NSMenu(title: "Edit")
|
||||
fieldMenu.addItem(withTitle: "Paste", action: #selector(NSText.paste(_:)), keyEquivalent: "")
|
||||
field.menu = fieldMenu
|
||||
alert.accessoryView = field
|
||||
alert.window.initialFirstResponder = field
|
||||
while true {
|
||||
app.activate(ignoringOtherApps: true)
|
||||
guard alert.runModal() == .alertFirstButtonReturn else {
|
||||
fail("password entry cancelled", 3)
|
||||
}
|
||||
if !field.stringValue.isEmpty {
|
||||
print(store(Data(field.stringValue.utf8)))
|
||||
break
|
||||
}
|
||||
NSSound.beep()
|
||||
alert.informativeText = "Enter the current Google password. Empty values are never stored."
|
||||
alert.window.initialFirstResponder = field
|
||||
}
|
||||
default:
|
||||
fail("unsupported action")
|
||||
}
|
||||
SWIFT
|
||||
|
||||
swiftc -O -framework AppKit -framework Security "$source" -o "$KEYCHAIN_HELPER"
|
||||
chmod 0700 "$KEYCHAIN_HELPER"
|
||||
}
|
||||
|
||||
find_pinentry() {
|
||||
if [[ -n "${TELEO_GCP_REAUTH_PINENTRY:-}" ]]; then
|
||||
[[ -x "$TELEO_GCP_REAUTH_PINENTRY" ]] || die "test pinentry helper is not executable"
|
||||
printf '%s\n' "$TELEO_GCP_REAUTH_PINENTRY"
|
||||
return
|
||||
fi
|
||||
if command -v pinentry-mac >/dev/null 2>&1; then
|
||||
command -v pinentry-mac
|
||||
return
|
||||
fi
|
||||
for candidate in /opt/homebrew/bin/pinentry-mac /usr/local/bin/pinentry-mac; do
|
||||
if [[ -x "$candidate" ]]; then
|
||||
printf '%s\n' "$candidate"
|
||||
return
|
||||
fi
|
||||
done
|
||||
return 1
|
||||
}
|
||||
|
||||
store_password() {
|
||||
local pinentry result
|
||||
if pinentry="$(find_pinentry)"; then
|
||||
result="$({
|
||||
printf 'SETTITLE Teleo GCP operator\n'
|
||||
printf 'SETDESC Store the current Google password in macOS Keychain. This does not authenticate gcloud.\n'
|
||||
printf 'SETPROMPT Google password:\n'
|
||||
printf 'GETPIN\n'
|
||||
printf 'BYE\n'
|
||||
} | "$pinentry" | "$KEYCHAIN_HELPER" store-pinentry "$KEYCHAIN_SERVICE" "$EXPECTED_ACCOUNT" "$KEYCHAIN_LABEL")" || {
|
||||
die "secure password entry was cancelled or failed"
|
||||
}
|
||||
printf 'secure_prompt=pinentry-mac\n'
|
||||
else
|
||||
result="$("$KEYCHAIN_HELPER" prompt-store "$KEYCHAIN_SERVICE" "$EXPECTED_ACCOUNT" "$KEYCHAIN_LABEL")" || {
|
||||
die "secure password entry was cancelled or failed"
|
||||
}
|
||||
printf 'secure_prompt=AppKit NSSecureTextField\n'
|
||||
fi
|
||||
case "$result" in
|
||||
stored | updated) printf 'keychain_password=%s\n' "$result" ;;
|
||||
*) die "Keychain helper returned an invalid result" ;;
|
||||
esac
|
||||
}
|
||||
|
||||
emit_security_boundary() {
|
||||
printf '%s\n' \
|
||||
'security_truth=The stored Google password is emergency convenience only; it cannot satisfy or automate gcloud OAuth.' \
|
||||
'password_handling=The script never prints, copies, or submits the password and never places it in a command argument.' \
|
||||
"durable_unattended_route=$WORKFLOW_PATH uses GitHub OIDC/Workload Identity Federation and IAP after one-time bootstrap."
|
||||
}
|
||||
|
||||
emit_reauth_cta() {
|
||||
printf '%s\n' \
|
||||
'oauth_state=stale_or_unavailable' \
|
||||
"clear_CTA=Run 'gcloud auth login $EXPECTED_ACCOUNT --force --no-launch-browser' in a terminal, paste the displayed URL into the dedicated Chrome GCP session, complete reauthentication, return the authorization code to gcloud, then rerun '$RERUN_COMMAND'."
|
||||
}
|
||||
|
||||
report_status() {
|
||||
local keychain_state account project instance_state gcloud_bin
|
||||
keychain_state="$("$KEYCHAIN_HELPER" status "$KEYCHAIN_SERVICE" "$EXPECTED_ACCOUNT" "$KEYCHAIN_LABEL")"
|
||||
case "$keychain_state" in
|
||||
present | absent) ;;
|
||||
*) die "Keychain helper returned an invalid status" ;;
|
||||
esac
|
||||
|
||||
emit_security_boundary
|
||||
printf 'keychain_password=%s\n' "$keychain_state"
|
||||
|
||||
if ! gcloud_bin="$(command -v gcloud)"; then
|
||||
printf '%s\n' 'gcloud_state=missing' 'operator_readiness=not_ready'
|
||||
return 1
|
||||
fi
|
||||
printf 'gcloud_binary=%s\n' "$gcloud_bin"
|
||||
|
||||
account="$(gcloud config get-value account --quiet 2>/dev/null || true)"
|
||||
project="$(gcloud config get-value project --quiet 2>/dev/null || true)"
|
||||
printf 'gcloud_account=%s\n' "${account:-unset}"
|
||||
printf 'gcloud_project=%s\n' "${project:-unset}"
|
||||
if [[ "$account" != "$EXPECTED_ACCOUNT" || "$project" != "$PROJECT_ID" ]]; then
|
||||
printf '%s\n' \
|
||||
'gcloud_config_state=mismatch' \
|
||||
"clear_CTA=Run 'gcloud config set account $EXPECTED_ACCOUNT && gcloud config set project $PROJECT_ID', then rerun '$RERUN_COMMAND'." \
|
||||
'operator_readiness=not_ready'
|
||||
return 1
|
||||
fi
|
||||
printf 'gcloud_config_state=ready\n'
|
||||
|
||||
if ! gcloud auth print-access-token --account="$EXPECTED_ACCOUNT" >/dev/null 2>&1; then
|
||||
emit_reauth_cta
|
||||
printf 'operator_readiness=not_ready\n'
|
||||
return 1
|
||||
fi
|
||||
printf 'oauth_state=ready\n'
|
||||
|
||||
if ! command -v ssh >/dev/null 2>&1; then
|
||||
printf '%s\n' 'ssh_binary=missing' 'operator_readiness=not_ready'
|
||||
return 1
|
||||
fi
|
||||
printf 'ssh_binary=ready\n'
|
||||
|
||||
instance_state="$(
|
||||
gcloud compute instances describe "$INSTANCE" \
|
||||
--project="$PROJECT_ID" \
|
||||
--zone="$ZONE" \
|
||||
--format='value(name,status,networkInterfaces[0].networkIP)' \
|
||||
--quiet 2>/dev/null || true
|
||||
)"
|
||||
if [[ "$instance_state" != "$INSTANCE"$'\t'"RUNNING"$'\t'"$EXPECTED_INTERNAL_IP" ]]; then
|
||||
printf '%s\n' \
|
||||
'instance_readback=not_ready' \
|
||||
"clear_CTA=Confirm $INSTANCE is RUNNING at $EXPECTED_INTERNAL_IP in $PROJECT_ID/$ZONE, then rerun '$RERUN_COMMAND'." \
|
||||
'operator_readiness=not_ready'
|
||||
return 1
|
||||
fi
|
||||
printf 'instance_readback=%s RUNNING %s\n' "$INSTANCE" "$EXPECTED_INTERNAL_IP"
|
||||
|
||||
if ! gcloud compute ssh "$INSTANCE" \
|
||||
--project="$PROJECT_ID" \
|
||||
--zone="$ZONE" \
|
||||
--tunnel-through-iap \
|
||||
--ssh-key-expire-after=5m \
|
||||
--dry-run \
|
||||
--quiet >/dev/null 2>&1; then
|
||||
printf '%s\n' \
|
||||
'iap_ssh_preflight=not_ready' \
|
||||
"clear_CTA=Repair the local gcloud IAP/SSH prerequisite reported by 'gcloud compute ssh $INSTANCE --project=$PROJECT_ID --zone=$ZONE --tunnel-through-iap --ssh-key-expire-after=5m --dry-run', then rerun '$RERUN_COMMAND'." \
|
||||
'operator_readiness=not_ready'
|
||||
return 1
|
||||
fi
|
||||
printf '%s\n' \
|
||||
'iap_ssh_preflight=ready_dry_run_only' \
|
||||
'iap_connection=not_attempted' \
|
||||
'operator_readiness=ready_for_explicit_operator_action'
|
||||
}
|
||||
|
||||
main() {
|
||||
local mode="interactive" answer
|
||||
[[ $# -le 1 ]] || {
|
||||
usage >&2
|
||||
exit 2
|
||||
}
|
||||
if [[ $# -eq 1 ]]; then
|
||||
case "$1" in
|
||||
--status) mode="status" ;;
|
||||
--store-password) mode="store" ;;
|
||||
-h | --help) usage; exit 0 ;;
|
||||
*) usage >&2; exit 2 ;;
|
||||
esac
|
||||
fi
|
||||
|
||||
validate_test_overrides
|
||||
build_keychain_helper
|
||||
|
||||
if [[ "$mode" == "interactive" ]]; then
|
||||
if [[ ! -t 0 && "${TELEO_GCP_REAUTH_TESTING:-0}" != "1" ]]; then
|
||||
die "interactive mode requires a terminal; use --status or --store-password"
|
||||
fi
|
||||
printf 'Store or update the emergency Google password in macOS Keychain? [y/N] ' >&2
|
||||
IFS= read -r answer || answer=""
|
||||
case "$answer" in
|
||||
y | Y | yes | YES) store_password ;;
|
||||
*) printf 'keychain_password=unchanged\n' ;;
|
||||
esac
|
||||
elif [[ "$mode" == "store" ]]; then
|
||||
store_password
|
||||
fi
|
||||
|
||||
report_status
|
||||
}
|
||||
|
||||
main "$@"
|
||||
250
tests/test_gcp_operator_reauth.py
Normal file
250
tests/test_gcp_operator_reauth.py
Normal file
|
|
@ -0,0 +1,250 @@
|
|||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = REPO_ROOT / "scripts" / "gcp-operator-reauth.sh"
|
||||
DOC = REPO_ROOT / "docs" / "gcp-operator-reauth.md"
|
||||
FAKE_SECRET = "fake password%value"
|
||||
|
||||
|
||||
def write_executable(path: Path, source: str) -> None:
|
||||
path.write_text(source, encoding="utf-8")
|
||||
path.chmod(0o755)
|
||||
|
||||
|
||||
def fake_operator_env(tmp_path: Path, *, token_ready: bool = True) -> tuple[dict[str, str], Path, Path, Path]:
|
||||
bin_dir = tmp_path / "bin"
|
||||
bin_dir.mkdir()
|
||||
calls = tmp_path / "gcloud-calls.txt"
|
||||
helper_calls = tmp_path / "keychain-helper-calls.txt"
|
||||
keychain_state = tmp_path / "keychain-state"
|
||||
|
||||
write_executable(
|
||||
bin_dir / "gcloud",
|
||||
"""#!/usr/bin/env bash
|
||||
set -eu
|
||||
printf '%s\\n' "$*" >> "$FAKE_GCLOUD_CALLS"
|
||||
case "$1:$2:${3:-}" in
|
||||
config:get-value:account) printf '%s\\n' 'billy@livingip.xyz' ;;
|
||||
config:get-value:project) printf '%s\\n' 'teleo-501523' ;;
|
||||
auth:print-access-token:*)
|
||||
[[ "${FAKE_GCLOUD_TOKEN_READY:-0}" == "1" ]] || exit 1
|
||||
printf '%s\\n' 'fake-token-that-must-be-discarded'
|
||||
;;
|
||||
compute:instances:describe) printf 'teleo-prod-1\\tRUNNING\\t10.60.0.3\\n' ;;
|
||||
compute:ssh:teleo-prod-1) exit 0 ;;
|
||||
*) exit 90 ;;
|
||||
esac
|
||||
""",
|
||||
)
|
||||
write_executable(bin_dir / "ssh", "#!/usr/bin/env bash\nexit 0\n")
|
||||
write_executable(
|
||||
bin_dir / "pinentry-mac",
|
||||
"""#!/usr/bin/env bash
|
||||
set -eu
|
||||
cat >/dev/null
|
||||
printf '%s\\n' 'OK fake-pinentry' 'D fake%20password%25value' 'OK'
|
||||
""",
|
||||
)
|
||||
write_executable(
|
||||
bin_dir / "keychain-helper",
|
||||
"""#!/usr/bin/env bash
|
||||
set -eu
|
||||
printf '%s\\n' "$*" >> "$FAKE_KEYCHAIN_HELPER_CALLS"
|
||||
action="$1"
|
||||
case "$action" in
|
||||
status)
|
||||
if [[ -s "$FAKE_KEYCHAIN_STATE" ]]; then printf 'present\\n'; else printf 'absent\\n'; fi
|
||||
;;
|
||||
store-pinentry)
|
||||
response="$(cat)"
|
||||
[[ "$response" == *'D fake%20password%25value'* ]]
|
||||
printf '%s' 'fake password%value' > "$FAKE_KEYCHAIN_STATE"
|
||||
printf 'stored\\n'
|
||||
;;
|
||||
prompt-store)
|
||||
printf '%s' 'fake AppKit password' > "$FAKE_KEYCHAIN_STATE"
|
||||
printf 'stored\\n'
|
||||
;;
|
||||
*) exit 91 ;;
|
||||
esac
|
||||
""",
|
||||
)
|
||||
|
||||
env = {
|
||||
**os.environ,
|
||||
"PATH": f"{bin_dir}:{os.environ['PATH']}",
|
||||
"FAKE_GCLOUD_CALLS": str(calls),
|
||||
"FAKE_GCLOUD_TOKEN_READY": "1" if token_ready else "0",
|
||||
"FAKE_KEYCHAIN_HELPER_CALLS": str(helper_calls),
|
||||
"FAKE_KEYCHAIN_STATE": str(keychain_state),
|
||||
"TELEO_GCP_REAUTH_TESTING": "1",
|
||||
"TELEO_GCP_REAUTH_KEYCHAIN_HELPER": str(bin_dir / "keychain-helper"),
|
||||
"TELEO_GCP_REAUTH_PINENTRY": str(bin_dir / "pinentry-mac"),
|
||||
}
|
||||
return env, calls, helper_calls, keychain_state
|
||||
|
||||
|
||||
def run_script(env: dict[str, str], *args: str, input_text: str | None = None) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
["bash", str(SCRIPT), *args],
|
||||
cwd=REPO_ROOT,
|
||||
env=env,
|
||||
input=input_text,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
def test_fake_pinentry_to_keychain_to_ready_status_lifecycle(tmp_path: Path) -> None:
|
||||
env, calls_path, helper_calls_path, state_path = fake_operator_env(tmp_path)
|
||||
|
||||
completed = run_script(env, "--store-password")
|
||||
|
||||
assert completed.returncode == 0, completed.stderr
|
||||
assert completed.stdout.count("security_truth=") == 1
|
||||
assert "secure_prompt=pinentry-mac" in completed.stdout
|
||||
assert "keychain_password=stored" in completed.stdout
|
||||
assert "keychain_password=present" in completed.stdout
|
||||
assert "oauth_state=ready" in completed.stdout
|
||||
assert "instance_readback=teleo-prod-1 RUNNING 10.60.0.3" in completed.stdout
|
||||
assert "iap_ssh_preflight=ready_dry_run_only" in completed.stdout
|
||||
assert "iap_connection=not_attempted" in completed.stdout
|
||||
assert "operator_readiness=ready_for_explicit_operator_action" in completed.stdout
|
||||
assert state_path.read_text(encoding="utf-8") == FAKE_SECRET
|
||||
|
||||
combined_output = completed.stdout + completed.stderr
|
||||
calls = calls_path.read_text(encoding="utf-8")
|
||||
helper_calls = helper_calls_path.read_text(encoding="utf-8")
|
||||
assert FAKE_SECRET not in combined_output
|
||||
assert "fake-token-that-must-be-discarded" not in combined_output
|
||||
assert FAKE_SECRET not in calls
|
||||
assert FAKE_SECRET not in helper_calls
|
||||
assert "auth print-access-token --account=billy@livingip.xyz" in calls
|
||||
assert "compute instances describe teleo-prod-1" in calls
|
||||
assert "compute ssh teleo-prod-1" in calls
|
||||
assert "--tunnel-through-iap" in calls
|
||||
assert "--ssh-key-expire-after=5m" in calls
|
||||
assert "--dry-run" in calls
|
||||
|
||||
|
||||
def test_default_interactive_mode_uses_the_secure_popup_path(tmp_path: Path) -> None:
|
||||
env, _, _, state_path = fake_operator_env(tmp_path)
|
||||
|
||||
completed = run_script(env, input_text="yes\n")
|
||||
|
||||
assert completed.returncode == 0, completed.stderr
|
||||
assert "Store or update" in completed.stderr
|
||||
assert "secure_prompt=pinentry-mac" in completed.stdout
|
||||
assert state_path.read_text(encoding="utf-8") == FAKE_SECRET
|
||||
assert FAKE_SECRET not in completed.stdout + completed.stderr
|
||||
|
||||
|
||||
def test_appkit_secure_field_fallback_runs_when_pinentry_is_absent(tmp_path: Path) -> None:
|
||||
env, _, _, state_path = fake_operator_env(tmp_path)
|
||||
pinentry = Path(env.pop("TELEO_GCP_REAUTH_PINENTRY"))
|
||||
pinentry.unlink()
|
||||
env["PATH"] = f"{pinentry.parent}:/usr/bin:/bin"
|
||||
|
||||
completed = run_script(env, "--store-password")
|
||||
|
||||
assert completed.returncode == 0, completed.stderr
|
||||
assert "secure_prompt=AppKit NSSecureTextField" in completed.stdout
|
||||
assert state_path.read_text(encoding="utf-8") == "fake AppKit password"
|
||||
assert "fake AppKit password" not in completed.stdout + completed.stderr
|
||||
|
||||
|
||||
def test_status_does_not_open_a_popup_or_read_a_password(tmp_path: Path) -> None:
|
||||
env, _, helper_calls_path, state_path = fake_operator_env(tmp_path)
|
||||
state_path.write_text(FAKE_SECRET, encoding="utf-8")
|
||||
|
||||
completed = run_script(env, "--status")
|
||||
|
||||
assert completed.returncode == 0, completed.stderr
|
||||
assert "keychain_password=present" in completed.stdout
|
||||
assert "secure_prompt=" not in completed.stdout
|
||||
assert helper_calls_path.read_text(encoding="utf-8").splitlines() == [
|
||||
"status com.livingip.teleo.gcp-operator-reauth billy@livingip.xyz Teleo GCP operator emergency password"
|
||||
]
|
||||
assert FAKE_SECRET not in completed.stdout + completed.stderr
|
||||
|
||||
|
||||
def test_stale_oauth_emits_one_exact_nonlaunching_cta_and_stops(tmp_path: Path) -> None:
|
||||
env, calls_path, _, _ = fake_operator_env(tmp_path, token_ready=False)
|
||||
|
||||
completed = run_script(env, "--status")
|
||||
|
||||
assert completed.returncode == 1
|
||||
assert "oauth_state=stale_or_unavailable" in completed.stdout
|
||||
assert completed.stdout.count("clear_CTA=") == 1
|
||||
assert "gcloud auth login billy@livingip.xyz --force --no-launch-browser" in completed.stdout
|
||||
assert "dedicated Chrome GCP session" in completed.stdout
|
||||
assert "scripts/gcp-operator-reauth.sh --status" in completed.stdout
|
||||
assert "operator_readiness=not_ready" in completed.stdout
|
||||
calls = calls_path.read_text(encoding="utf-8")
|
||||
assert "compute instances describe" not in calls
|
||||
assert "compute ssh" not in calls
|
||||
|
||||
|
||||
def test_source_enforces_secret_and_gui_boundaries() -> None:
|
||||
source = SCRIPT.read_text(encoding="utf-8")
|
||||
|
||||
assert "NSAlert" in source
|
||||
assert "NSSecureTextField" in source
|
||||
assert "app.activate(ignoringOtherApps: true)" in source
|
||||
assert 'action: #selector(NSText.paste(_:)), keyEquivalent: "v"' in source
|
||||
assert "field.menu = fieldMenu" in source
|
||||
assert "Empty values are never stored." in source
|
||||
assert "SecItemAdd" in source
|
||||
assert "SecItemUpdate" in source
|
||||
assert "kSecAttrAccessibleWhenUnlockedThisDeviceOnly" in source
|
||||
assert "kSecAttrSynchronizable" in source
|
||||
assert "store-pinentry" in source
|
||||
assert "FileHandle.standardInput.readDataToEndOfFile()" in source
|
||||
assert "cannot satisfy or automate gcloud OAuth" in source
|
||||
assert "ready_dry_run_only" in source
|
||||
assert "iap_connection=not_attempted" in source
|
||||
for forbidden in (
|
||||
"osascript",
|
||||
"AppleScript",
|
||||
"System Events",
|
||||
"NSPasteboard",
|
||||
"pbcopy",
|
||||
"open -a",
|
||||
"open -n",
|
||||
"security add-generic-password",
|
||||
):
|
||||
assert forbidden not in source
|
||||
assert not re.search(r"gcloud auth login.*(?:^|\n)\s*gcloud auth login", source)
|
||||
|
||||
|
||||
def test_shell_syntax_and_operator_documentation() -> None:
|
||||
subprocess.run(["bash", "-n", str(SCRIPT)], cwd=REPO_ROOT, check=True)
|
||||
documentation = DOC.read_text(encoding="utf-8")
|
||||
assert "scripts/gcp-operator-reauth.sh --status" in documentation
|
||||
assert "scripts/gcp-operator-reauth.sh --store-password" in documentation
|
||||
assert "cannot refresh or satisfy gcloud OAuth by itself" in documentation
|
||||
assert ".github/workflows/gcp-iap-operator.yml" in documentation
|
||||
assert "ready_dry_run_only" in documentation
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform != "darwin" or shutil.which("swiftc") is None, reason="macOS Swift toolchain required")
|
||||
def test_embedded_swift_helper_typechecks(tmp_path: Path) -> None:
|
||||
source = SCRIPT.read_text(encoding="utf-8")
|
||||
match = re.search(r"<<'SWIFT'\n(?P<swift>.*?)\nSWIFT", source, flags=re.DOTALL)
|
||||
assert match is not None
|
||||
swift_source = tmp_path / "GcpOperatorKeychain.swift"
|
||||
swift_source.write_text(match.group("swift"), encoding="utf-8")
|
||||
|
||||
subprocess.run(
|
||||
["swiftc", "-typecheck", "-framework", "AppKit", "-framework", "Security", str(swift_source)],
|
||||
cwd=REPO_ROOT,
|
||||
check=True,
|
||||
)
|
||||
Loading…
Reference in a new issue